chore: insert requstIP and receivedAt into the payload [PIPE-1134][PI…

…PE-1135] (#4736)

gateway/gateway_test.go

@@ -602,6 +602,8 @@ var _ = Describe("Gateway", func() {
 			strippedPayload, _ := sjson.Delete(payload.String(), "messageId")
 			strippedPayload, _ = sjson.Delete(strippedPayload, "rudderId")
 			strippedPayload, _ = sjson.Delete(strippedPayload, "type")
+			strippedPayload, _ = sjson.Delete(strippedPayload, "receivedAt")
+			strippedPayload, _ = sjson.Delete(strippedPayload, "requestIP")
 
 			return strippedPayload
 		}
@@ -1551,6 +1553,53 @@ var _ = Describe("Gateway", func() {
 			Expect(job.Batch[0].MessageID).To(Equal("-a-random-string"))
 		})
 
+		It("doesn't override if receivedAt or requestIP already exists in payload", func() {
+			req := &webRequestT{
+				reqType:        "batch",
+				authContext:    rCtxEnabled,
+				done:           make(chan<- string),
+				userIDHeader:   userIDHeader,
+				requestPayload: []byte(`{"batch": [{"type": "extract", "receivedAt": "2024-01-01T01:01:01.000000001Z", "requestIP": "dummyIPFromPayload"}]}`),
+			}
+			jobForm, err := gateway.getJobDataFromRequest(req)
+			Expect(err).To(BeNil())
+
+			var job struct {
+				Batch []struct {
+					ReceivedAt string `json:"receivedAt"`
+					RequestIP  string `json:"requestIP"`
+				} `json:"batch"`
+			}
+			err = json.Unmarshal(jobForm.jobs[0].EventPayload, &job)
+			Expect(err).To(BeNil())
+			Expect(job.Batch[0].ReceivedAt).To(ContainSubstring("2024-01-01T01:01:01.000000001Z"))
+			Expect(job.Batch[0].RequestIP).To(ContainSubstring("dummyIPFromPayload"))
+		})
+
+		It("adds receivedAt and requestIP in the request payload if it's not already present", func() {
+			req := &webRequestT{
+				reqType:        "batch",
+				authContext:    rCtxEnabled,
+				done:           make(chan<- string),
+				userIDHeader:   userIDHeader,
+				requestPayload: []byte(`{"batch": [{"type": "extract"}]}`),
+			}
+			req.ipAddr = "dummyIP"
+			jobForm, err := gateway.getJobDataFromRequest(req)
+			Expect(err).To(BeNil())
+
+			var job struct {
+				Batch []struct {
+					ReceivedAt string `json:"receivedAt"`
+					RequestIP  string `json:"requestIP"`
+				} `json:"batch"`
+			}
+			err = json.Unmarshal(jobForm.jobs[0].EventPayload, &job)
+			Expect(err).To(BeNil())
+			Expect(job.Batch[0].ReceivedAt).To(Not(BeEmpty()))
+			Expect(job.Batch[0].RequestIP).To(ContainSubstring("dummyIP"))
+		})
+
 		It("allows extract events even if userID and anonID are not present in the request payload", func() {
 			req := &webRequestT{
 				reqType:        "batch",

gateway/handle.go

@@ -406,6 +406,19 @@ func (gw *Handle) getJobDataFromRequest(req *webRequestT) (jobData *jobFromReq,
 			return
 		}
 		toSet["rudderId"] = rudderId
+		if _, ok := toSet["receivedAt"]; !ok {
+			toSet["receivedAt"] = time.Now().Format(misc.RFC3339Milli)
+		}
+		if _, ok := toSet["requestIP"]; ok {
+			var tcOk bool
+			ipAddr, tcOk = toSet["requestIP"].(string)
+			if !tcOk {
+				gw.logger.Warnf("requestIP is not a string: %v", toSet["requestIP"])
+			}
+
+		} else {
+			toSet["requestIP"] = ipAddr
+		}
 		fillMessageID(toSet)
 		if eventTypeFromReq == "audiencelist" {
 			containsAudienceList = true