fix: add messageId, rudderId and type based on new schema

gateway/gateway_test.go

@@ -1715,7 +1715,7 @@ var _ = Describe("Gateway", func() {
 			)
 			internalBatchPayload := fmt.Sprintf(`[{
 			"properties": {
-				"messageID": "messageID",
+				"requestType": "dummyRequestType",
 				"routingKey": "anonymousId_header<<>>anonymousId_1<<>>identified_user_id",
 				"workspaceID": %q,
 				"userID": %q,
@@ -1739,7 +1739,7 @@ var _ = Describe("Gateway", func() {
 			internalBatchPayload := func() string {
 				return fmt.Sprintf(`{
 			"properties": {
-				"messageID": %q,
+				"requestType": "dummyRequestType",
 				"routingKey": "anonymousId_header<<>>anonymousId_1<<>>identified_user_id",
 				"workspaceID": %q,
 				"userID": %q,
@@ -1751,7 +1751,7 @@ var _ = Describe("Gateway", func() {
 				"traceID": "traceID"
 			},
 			"payload": %s
-			}`, uuid.NewString(), workspaceID, userID, sourceID, validData)
+			}`, workspaceID, userID, sourceID, validData)
 			}
 			return []byte(fmt.Sprintf(`[%s,%s]`, internalBatchPayload(), internalBatchPayload()))
 		}
@@ -1999,30 +1999,6 @@ var _ = Describe("Gateway", func() {
 			Expect(failedEventStat).To(BeNil())
 		})
 
-		It("request failed message validation error", func() {
-			req, err := http.NewRequest(http.MethodPost, internalBatchEndpoint, bytes.NewBuffer([]byte(`[{}]`)))
-			Expect(err).To(BeNil())
-			resp, err := client.Do(req)
-			Expect(err).To(BeNil())
-			Expect(http.StatusBadRequest, resp.StatusCode)
-			respData, err := io.ReadAll(resp.Body)
-			defer httputil.CloseResponse(resp)
-			Expect(err).To(BeNil())
-			Expect(string(respData)).Should(ContainSubstring(response.InvalidStreamMessage))
-			failedRequestStat := statStore.Get("gateway.write_key_failed_requests", map[string]string{
-				"writeKey":    "",
-				"reqType":     "internalBatch",
-				"reason":      response.InvalidStreamMessage,
-				"workspaceId": "",
-				"sourceID":    "",
-				"sourceType":  "",
-				"sdkVersion":  "",
-				"source":      "",
-			})
-			Expect(failedRequestStat).To(Not(BeNil()))
-			Expect(failedRequestStat.Values()).To(Equal([]float64{1}))
-		})
-
 		It("request success - suppressed user", func() {
 			payload := createInternalBatchPayload(SuppressedUserID, SourceIDEnabled)
 			req, err := http.NewRequest(http.MethodPost, internalBatchEndpoint, bytes.NewBuffer(payload))
@@ -2150,7 +2126,7 @@ var _ = Describe("Gateway", func() {
 
 		It("doesn't override if receivedAt or request_ip already exists in payload", func() {
 			properties := stream.MessageProperties{
-				MessageID:     "messageID",
+				RequestType:   "dummyRequestType",
 				RoutingKey:    "anonymousId_header<<>>anonymousId_1<<>>identified_user_id",
 				WorkspaceID:   "workspaceID",
 				SourceID:      "sourceID",
@@ -2199,7 +2175,7 @@ var _ = Describe("Gateway", func() {
 
 		It("adds receivedAt and request_ip in the request payload if it's not already present", func() {
 			properties := stream.MessageProperties{
-				MessageID:     "messageID",
+				RequestType:   "dummyRequestType",
 				RoutingKey:    "anonymousId_header<<>>anonymousId_1<<>>identified_user_id",
 				WorkspaceID:   "workspaceID",
 				SourceID:      "sourceID",
@@ -2241,6 +2217,187 @@ var _ = Describe("Gateway", func() {
 			Expect(job.Batch[0].ReceivedAt).To(ContainSubstring("2024-01-01T01:01:01.000Z"))
 			Expect(job.Batch[0].RequestIP).To(ContainSubstring("dummyIP"))
 		})
+
+		It("adds messageID, rudderId in the request payload if it's not already present", func() {
+			properties := stream.MessageProperties{
+				RequestType:   "dummyRequestType",
+				RoutingKey:    "anonymousId_header<<>>anonymousId_1<<>>identified_user_id",
+				WorkspaceID:   "workspaceID",
+				SourceID:      "sourceID",
+				ReceivedAt:    time.Date(2024, 1, 1, 1, 1, 1, 1, time.UTC),
+				RequestIP:     "dummyIP",
+				DestinationID: "destinationID",
+			}
+			msg := stream.Message{
+				Properties: properties,
+				Payload:    []byte(`{}`),
+			}
+			messages := []stream.Message{msg}
+			payload, err := json.Marshal(messages)
+			Expect(err).To(BeNil())
+			req := &webRequestT{
+				reqType:        "batch",
+				authContext:    rCtxEnabled,
+				done:           make(chan<- string),
+				requestPayload: payload,
+			}
+			jobsWithStats, err := gateway.extractJobsFromInternalBatchPayload("batch", req.requestPayload)
+			Expect(err).To(BeNil())
+			Expect(jobsWithStats).To(HaveLen(1))
+			Expect(jobsWithStats[0].stat).To(Equal(gwstats.SourceStat{
+				SourceID:    "sourceID",
+				WorkspaceID: "workspaceID",
+				ReqType:     "batch",
+			}))
+
+			var job struct {
+				Batch []struct {
+					MessageID string `json:"messageID"`
+					RudderID  string `json:"rudderId"`
+				} `json:"batch"`
+			}
+			err = json.Unmarshal(jobsWithStats[0].job.EventPayload, &job)
+			Expect(err).To(BeNil())
+			Expect(job.Batch).To(HaveLen(1))
+			Expect(job.Batch[0].MessageID).To(Not(BeEmpty()))
+			Expect(job.Batch[0].RudderID).To(Not(BeEmpty()))
+		})
+
+		It("doesn't override if messageID already exists in payload", func() {
+			properties := stream.MessageProperties{
+				RequestType:   "dummyRequestType",
+				RoutingKey:    "anonymousId_header<<>>anonymousId_1<<>>identified_user_id",
+				WorkspaceID:   "workspaceID",
+				SourceID:      "sourceID",
+				ReceivedAt:    time.Date(2024, 1, 1, 1, 1, 1, 1, time.UTC),
+				RequestIP:     "dummyIP",
+				DestinationID: "destinationID",
+			}
+			msg := stream.Message{
+				Properties: properties,
+				Payload:    []byte(`{"messageId": "dummyMessageID"}`),
+			}
+			messages := []stream.Message{msg}
+			payload, err := json.Marshal(messages)
+			Expect(err).To(BeNil())
+
+			req := &webRequestT{
+				reqType:        "batch",
+				authContext:    rCtxEnabled,
+				done:           make(chan<- string),
+				requestPayload: payload,
+			}
+			jobsWithStats, err := gateway.extractJobsFromInternalBatchPayload("batch", req.requestPayload)
+			Expect(err).To(BeNil())
+			Expect(jobsWithStats).To(HaveLen(1))
+			Expect(jobsWithStats[0].stat).To(Equal(
+				gwstats.SourceStat{
+					SourceID:    "sourceID",
+					WorkspaceID: "workspaceID",
+					ReqType:     "batch",
+				},
+			))
+
+			var job struct {
+				Batch []struct {
+					MessageID string `json:"messageId"`
+				} `json:"batch"`
+			}
+			jobForm := jobsWithStats[0].job
+			err = json.Unmarshal(jobForm.EventPayload, &job)
+			Expect(err).To(BeNil())
+			Expect(job.Batch).To(HaveLen(1))
+			Expect(job.Batch[0].MessageID).To(ContainSubstring("dummyMessageID"))
+		})
+
+		It("adds type in the request payload if RequestType Property is not one of batch, replay, retl, import", func() {
+			properties := stream.MessageProperties{
+				RequestType:   "dummyRequestType",
+				RoutingKey:    "anonymousId_header<<>>anonymousId_1<<>>identified_user_id",
+				WorkspaceID:   "workspaceID",
+				SourceID:      "sourceID",
+				ReceivedAt:    time.Date(2024, 1, 1, 1, 1, 1, 1, time.UTC),
+				RequestIP:     "dummyIP",
+				DestinationID: "destinationID",
+			}
+			msg := stream.Message{
+				Properties: properties,
+				Payload:    []byte(`{}`),
+			}
+			messages := []stream.Message{msg}
+			payload, err := json.Marshal(messages)
+			Expect(err).To(BeNil())
+			req := &webRequestT{
+				reqType:        "batch",
+				authContext:    rCtxEnabled,
+				done:           make(chan<- string),
+				requestPayload: payload,
+			}
+			jobsWithStats, err := gateway.extractJobsFromInternalBatchPayload("batch", req.requestPayload)
+			Expect(err).To(BeNil())
+			Expect(jobsWithStats).To(HaveLen(1))
+			Expect(jobsWithStats[0].stat).To(Equal(gwstats.SourceStat{
+				SourceID:    "sourceID",
+				WorkspaceID: "workspaceID",
+				ReqType:     "batch",
+			}))
+
+			var job struct {
+				Batch []struct {
+					Type string `json:"type"`
+				} `json:"batch"`
+			}
+			err = json.Unmarshal(jobsWithStats[0].job.EventPayload, &job)
+			Expect(err).To(BeNil())
+			Expect(job.Batch).To(HaveLen(1))
+			Expect(job.Batch[0].Type).To(ContainSubstring("dummyRequestType"))
+		})
+
+		It("does not change type if  RequestType Property is batch, replay, retl, import", func() {
+			requestTypes := []string{"batch", "replay", "retl", "import"}
+			for _, reqType := range requestTypes {
+				properties := stream.MessageProperties{
+					RequestType:   reqType,
+					RoutingKey:    "anonymousId_header<<>>anonymousId_1<<>>identified_user_id",
+					WorkspaceID:   "workspaceID",
+					SourceID:      "sourceID",
+					ReceivedAt:    time.Date(2024, 1, 1, 1, 1, 1, 1, time.UTC),
+					RequestIP:     "dummyIP",
+					DestinationID: "destinationID",
+				}
+				msg := stream.Message{
+					Properties: properties,
+					Payload:    []byte(`{"type": "dummyType"}`),
+				}
+				messages := []stream.Message{msg}
+				payload, err := json.Marshal(messages)
+				Expect(err).To(BeNil())
+				req := &webRequestT{
+					reqType:        "batch",
+					authContext:    rCtxEnabled,
+					done:           make(chan<- string),
+					requestPayload: payload,
+				}
+				jobsWithStats, err := gateway.extractJobsFromInternalBatchPayload("batch", req.requestPayload)
+				Expect(err).To(BeNil())
+				Expect(jobsWithStats).To(HaveLen(1))
+				Expect(jobsWithStats[0].stat).To(Equal(gwstats.SourceStat{
+					SourceID:    "sourceID",
+					WorkspaceID: "workspaceID",
+					ReqType:     "batch",
+				}))
+
+				var job struct {
+					Batch []struct {
+						Type string `json:"type"`
+					} `json:"batch"`
+				}
+				err = json.Unmarshal(jobsWithStats[0].job.EventPayload, &job)
+				Expect(err).To(BeNil())
+				Expect(job.Batch).To(HaveLen(1))
+				Expect(job.Batch[0].Type).To(ContainSubstring("dummyType"))
+			}
+		})
 	})
 })
 

gateway/handle.go

@@ -401,11 +401,9 @@
 		}
 
 		// hashing combination of userIDFromReq + anonIDFromReq, using colon as a delimiter
-		var rudderId uuid.UUID
-		rudderId, err = kituuid.GetMD5UUID(userIDFromReq + ":" + anonIDFromReq)
+		rudderId, err := getRudderId(userIDFromReq, anonIDFromReq)
 		if err != nil {
-			err = errors.New(response.NonIdentifiableRequest)
-			return
+			return jobData, err
 		}
 		toSet["rudderId"] = rudderId
 		if _, ok := toSet["receivedAt"]; !ok {
@@ -786,11 +784,48 @@
 			stat.Report(gw.stats)
 			return nil, errors.New(response.InvalidStreamMessage)
 		}
+		// TODO: get rid of this check
+		if msg.Properties.RequestType != "" {
+			switch msg.Properties.RequestType {
+			case "batch", "replay", "retl", "import":
+			default:
+				msg.Payload, err = sjson.SetBytes(msg.Payload, "type", msg.Properties.RequestType)
+				if err != nil {
+					stat.RequestEventsFailed(1, response.NotRudderEvent)
+					stat.Report(gw.stats)
+					return nil, errors.New(response.NotRudderEvent)
+				}
+			}
+		}
+
+		anonIDFromReq := sanitizeAndTrim(getJSONValueBytes(msg.Payload, "anonymousId"))
+		userIDFromReq := sanitizeAndTrim(getJSONValueBytes(msg.Payload, "userId"))
+		messageID, changed := getMessageID(msg.Payload)
+		if changed {
+			msg.Payload, err = sjson.SetBytes(msg.Payload, "messageId", messageID)
+			if err != nil {
+				stat.RequestFailed(response.NotRudderEvent)
+				stat.Report(gw.stats)
+				return nil, errors.New(response.NotRudderEvent)
+			}
+		}
+		rudderId, err := getRudderId(userIDFromReq, anonIDFromReq)
+		if err != nil {
+			stat.RequestFailed(response.NotRudderEvent)
+			stat.Report(gw.stats)
+			return nil, errors.New(response.NotRudderEvent)
+		}
+		msg.Payload, err = sjson.SetBytes(msg.Payload, "rudderId", rudderId.String())
+		if err != nil {
+			stat.RequestFailed(response.NotRudderEvent)
+			stat.Report(gw.stats)
+			return nil, errors.New(response.NotRudderEvent)
+		}
 		writeKey, ok := gw.getWriteKeyFromSourceID(msg.Properties.SourceID)
 		if !ok {
 			// only live-events will not work if writeKey is not found
 			gw.logger.Errorn("unable to get writeKey for job",
-				logger.NewStringField("messageId", msg.Properties.MessageID),
+				logger.NewStringField("messageId", messageID),
 				obskit.SourceID(msg.Properties.SourceID))
 		}
 		stat.SourceID = msg.Properties.SourceID
@@ -817,7 +852,7 @@
 		}).Since(msg.Properties.ReceivedAt)
 
 		jobsDBParams := params{
-			MessageID:       msg.Properties.MessageID,
+			MessageID:       messageID,
 			SourceID:        msg.Properties.SourceID,
 			SourceJobRunID:  msg.Properties.SourceJobRunID,
 			SourceTaskRunID: msg.Properties.SourceTaskRunID,
@@ -888,6 +923,35 @@
 	return res, nil
 }
 
+// getMessageID returns the messageID from the event payload.
+// If the messageID is not present, it generates a new one.
+// It also returns a boolean indicating if the messageID was changed.
+func getMessageID(event []byte) (string, bool) {
+	messageID := getJSONValueBytes(event, "messageId")
+	sanitizedMessageID := sanitizeAndTrim(messageID)
+	if sanitizedMessageID == "" {
+		return uuid.New().String(), true
+	}
+	return sanitizedMessageID, messageID != sanitizedMessageID
+}
+
+func getRudderId(userIDFromReq, anonIDFromReq string) (uuid.UUID, error) {
+	rudderId, err := kituuid.GetMD5UUID(userIDFromReq + ":" + anonIDFromReq)
+	if err != nil {
+		err = errors.New(response.NonIdentifiableRequest)
+		return rudderId, err
+	}
+	return rudderId, nil
+}
+
+func getJSONValueBytes(payload []byte, key string) string {
+	return gjson.GetBytes(payload, key).String()
+}
+
+func sanitizeAndTrim(str string) string {
+	return strings.TrimSpace(sanitize.Unicode(str))
+}
+
 func fillReceivedAt(event []byte, receivedAt time.Time) ([]byte, error) {
 	if !gjson.GetBytes(event, "receivedAt").Exists() {
 		return sjson.SetBytes(event, "receivedAt", receivedAt.Format(misc.RFC3339Milli))

go.mod

@@ -80,7 +80,7 @@ require (
 	github.com/rudderlabs/compose-test v0.1.3
 	github.com/rudderlabs/rudder-go-kit v0.43.0
 	github.com/rudderlabs/rudder-observability-kit v0.0.3
-	github.com/rudderlabs/rudder-schemas v0.5.2
+	github.com/rudderlabs/rudder-schemas v0.5.3-0.20240927013310-2c5ae88df954
 	github.com/rudderlabs/rudder-transformer/go v0.0.0-20240910055720-f77d2ab4125a
 	github.com/rudderlabs/sql-tunnels v0.1.7
 	github.com/rudderlabs/sqlconnect-go v1.10.0

go.sum

@@ -1147,8 +1147,8 @@ github.com/rudderlabs/rudder-go-kit v0.43.0 h1:N6CAvQdjufitdiUl424+AcMebEmieB0TO
 github.com/rudderlabs/rudder-go-kit v0.43.0/go.mod h1:NrHCi0KSzHSMFXQu0t2kgJcE4ClAKklVXfb2glADvQ4=
 github.com/rudderlabs/rudder-observability-kit v0.0.3 h1:vZtuZRkGX+6rjaeKtxxFE2YYP6QlmAcVcgecTOjvz+Q=
 github.com/rudderlabs/rudder-observability-kit v0.0.3/go.mod h1:6UjAh3H6rkE0fFLh7z8ZGQEQbKtUkRfhWOf/OUhfqW8=
-github.com/rudderlabs/rudder-schemas v0.5.2 h1:qmbQrd+/2au5X+04vX8Qj4BGL7aK+TedeMdcyQyrLtY=
-github.com/rudderlabs/rudder-schemas v0.5.2/go.mod h1:iUpjG/Zb+ioZcNLvXNYXSKQ2LpPlsIDBfxfCDH9ue/E=
+github.com/rudderlabs/rudder-schemas v0.5.3-0.20240927013310-2c5ae88df954 h1:bRI7BddwtSjwMcHkQxO6l+hwXKJI3QazfKqW27JtVVE=
+github.com/rudderlabs/rudder-schemas v0.5.3-0.20240927013310-2c5ae88df954/go.mod h1:iUpjG/Zb+ioZcNLvXNYXSKQ2LpPlsIDBfxfCDH9ue/E=
 github.com/rudderlabs/rudder-transformer/go v0.0.0-20240910055720-f77d2ab4125a h1:OZcvpApxEYNkB9UNXrKDUBufQ24Lsr2Cs0pw70tzXBw=
 github.com/rudderlabs/rudder-transformer/go v0.0.0-20240910055720-f77d2ab4125a/go.mod h1:3NGitPz4pYRRZ6Xt09S+8hb0tHK/9pZcKJ3OgOTaSmE=
 github.com/rudderlabs/sql-tunnels v0.1.7 h1:wDCRl6zY4M5gfWazf7XkSTGQS3yjBzUiUgEMBIfHNDA=