fix: add source type to internalBatch request stats (#6030)

# Description

- Add sourceType to batch request statistics for internalBatch requests
- Update getSourceConfigFromSourceID method to return source type
- Modify handle logic to populate sourceType in batch request stats

## Linear Ticket

pipe-2160

## Security

- [x] The code changed/added as part of this pull request won't create
any security issues with how the software is being used.

Author: mihir20

gateway/gateway_test.go

@@ -1865,7 +1865,7 @@ var _ = Describe("Gateway", func() {
 						"reqType":       "internalBatch",
 						"workspaceId":   WorkspaceID,
 						"sourceID":      SourceIDEnabled,
-						"sourceType":    "",
+						"sourceType":    sourceType2,
 						"sdkVersion":    "",
 						"sourceDefName": SourceIDEnabled,
 					},
@@ -1881,7 +1881,7 @@ var _ = Describe("Gateway", func() {
 						"reqType":       "internalBatch",
 						"workspaceId":   WorkspaceID,
 						"sourceID":      SourceIDEnabled,
-						"sourceType":    "",
+						"sourceType":    sourceType2,
 						"sdkVersion":    "",
 						"sourceDefName": SourceIDEnabled,
 					},
@@ -1945,7 +1945,7 @@ var _ = Describe("Gateway", func() {
 						"reqType":       "internalBatch",
 						"workspaceId":   WorkspaceID,
 						"sourceID":      SourceIDEnabled,
-						"sourceType":    "",
+						"sourceType":    sourceType2,
 						"sdkVersion":    "",
 						"sourceDefName": SourceIDEnabled,
 					},
@@ -2073,7 +2073,7 @@ var _ = Describe("Gateway", func() {
 				"reqType":       "internalBatch",
 				"workspaceId":   WorkspaceID,
 				"sourceID":      SourceIDEnabled,
-				"sourceType":    "",
+				"sourceType":    sourceType2,
 				"sdkVersion":    "",
 				"source":        "",
 				"sourceDefName": SourceIDEnabled,
@@ -2085,7 +2085,7 @@ var _ = Describe("Gateway", func() {
 				"reqType":       "internalBatch",
 				"workspaceId":   WorkspaceID,
 				"sourceID":      SourceIDEnabled,
-				"sourceType":    "",
+				"sourceType":    sourceType2,
 				"sdkVersion":    "",
 				"source":        "",
 				"sourceDefName": SourceIDEnabled,
@@ -2120,7 +2120,7 @@ var _ = Describe("Gateway", func() {
 				"reqType":       "internalBatch",
 				"workspaceId":   WorkspaceID,
 				"sourceID":      SourceIDEnabled,
-				"sourceType":    "",
+				"sourceType":    sourceType2,
 				"sdkVersion":    "",
 				"source":        "",
 				"reason":        "storeFailed",
@@ -2133,7 +2133,7 @@ var _ = Describe("Gateway", func() {
 				"reqType":       "internalBatch",
 				"workspaceId":   WorkspaceID,
 				"sourceID":      SourceIDEnabled,
-				"sourceType":    "",
+				"sourceType":    sourceType2,
 				"sdkVersion":    "",
 				"source":        "",
 				"sourceDefName": SourceIDEnabled,

gateway/handle.go

@@ -907,17 +907,21 @@ func (gw *Handle) extractJobsFromInternalBatchPayload(reqType string, body []byt
 			}
 		}
 
-		writeKey, sourceDefName, sourceName, ok := gw.getSourceConfigFromSourceID(msg.Properties.SourceID)
+		writeKey, sourceDefName, sourceName, sourceType := "", "", "", ""
+		src, ok := gw.getSourceConfigFromSourceID(msg.Properties.SourceID)
 		if !ok {
 			// only live-events will not work if writeKey is not found
 			gw.logger.Errorn("unable to get source details from sourceID",
 				logger.NewStringField("messageId", messageID),
 				obskit.SourceID(msg.Properties.SourceID))
+		} else {
+			writeKey, sourceDefName, sourceName, sourceType = src.WriteKey, src.SourceDefinition.Name, src.Name, src.SourceDefinition.Category
 		}
 		stat.SourceID = msg.Properties.SourceID
 		stat.WorkspaceID = msg.Properties.WorkspaceID
 		stat.WriteKey = writeKey
 		stat.SourceDefName = sourceDefName
+		stat.SourceType = sourceType
 
 		if isUserSuppressed(msg.Properties.WorkspaceID, msg.Properties.UserID, msg.Properties.SourceID) {
 			gw.logger.Infon("suppressed event",
@@ -1042,13 +1046,13 @@ func fillRequestIP(event []byte, ip string) ([]byte, error) {
 	return event, nil
 }
 
-func (gw *Handle) getSourceConfigFromSourceID(sourceID string) (writeKey, sourceDefName, sourceName string, ok bool) {
+func (gw *Handle) getSourceConfigFromSourceID(sourceID string) (src backendconfig.SourceT, ok bool) {
 	gw.configSubscriberLock.RLock()
 	defer gw.configSubscriberLock.RUnlock()
 	if s, ok := gw.sourceIDSourceMap[sourceID]; ok {
-		return s.WriteKey, s.SourceDefinition.Name, s.Name, true
+		return s, true
 	}
-	return "", "", "", false
+	return backendconfig.SourceT{}, false
 }
 
 func (gw *Handle) storeJobs(ctx context.Context, jobs []*jobsdb.JobT) error {