Add datalake file path to cache (#51)

* add datalake file path to cache

Signed-off-by: Ayman <[email protected]>

* clean up

Signed-off-by: Ayman <[email protected]>

Signed-off-by: Ayman <[email protected]>
Co-authored-by: Ayman <[email protected]>

Author: khalifapro

cmd/git/git.go

@@ -515,7 +515,7 @@ var (
 
 // Publisher - for streaming data to Kinesis
 type Publisher interface {
-	PushEvents(action, source, eventType, subEventType, env string, data []interface{}) error
+	PushEvents(action, source, eventType, subEventType, env string, data []interface{}) (string, error)
 }
 
 // RawPLS - programming language summary (all fields as strings)
@@ -1778,19 +1778,28 @@ func (j *DSGit) GitEnrichItems(ctx *shared.Ctx, thrN int, items []interface{}, d
 							continue
 						}
 						vals = append(vals, map[string]interface{}{
-							"id":   d.Payload.ID,
-							"data": b,
+							"id": d.Payload.ID,
+							"data": map[string]interface{}{
+								"content": b,
+							},
 						})
 					}
 				}
-				err = j.cacheProvider.Create(j.endpoint, vals)
+				path := ""
 				if len(formattedData) > 0 {
-					err = j.Publisher.PushEvents(CommitCreated, "insights", GitDataSource, "commits", os.Getenv("STAGE"), formattedData)
+					path, err = j.Publisher.PushEvents(CommitCreated, "insights", GitDataSource, "commits", os.Getenv("STAGE"), formattedData)
 					if err != nil {
 						j.log.WithFields(logrus.Fields{"operation": "GitEnrichItems"}).Errorf("Error: %+v", err)
 						return
 					}
 				}
+				if len(vals) > 0 {
+					for i, o := range vals {
+						o["data"].(map[string]interface{})["path"] = path
+						vals[i] = o
+					}
+					err = j.cacheProvider.Create(j.endpoint, vals)
+				}
 			} else {
 				var jsonBytes []byte
 				jsonBytes, err = jsoniter.Marshal(data)

go.mod

@@ -4,7 +4,7 @@ go 1.17
 
 require (
 	github.com/LF-Engineering/insights-datasource-shared v1.5.13-0.20220823221255-792e9bbc15c7
-	github.com/LF-Engineering/lfx-event-schema v0.1.30
+	github.com/LF-Engineering/lfx-event-schema v0.1.33
 	github.com/aws/aws-lambda-go v1.27.1
 	github.com/aws/aws-sdk-go v1.42.25
 	github.com/json-iterator/go v1.1.11

go.sum

@@ -2,8 +2,8 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03
 github.com/LF-Engineering/insights-datasource-shared v1.5.13-0.20220823221255-792e9bbc15c7 h1:/hH1GY0sZ537pch+ttGuaAfTtXHmeKwQKj1FyN2TxEM=
 github.com/LF-Engineering/insights-datasource-shared v1.5.13-0.20220823221255-792e9bbc15c7/go.mod h1:ZQCRAJNyizhBemQDSjqm14G2MPkOJbRzCPfjwFhYKy4=
 github.com/LF-Engineering/lfx-event-schema v0.1.14/go.mod h1:CfFIZ4mwzo88umf5+KxDQEzqlVkPG7Vx8eLK2oDfWIs=
-github.com/LF-Engineering/lfx-event-schema v0.1.30 h1:s20N1/m0lzLDdHhFjho0hWn/os2aY0ewtgxsiJpS7Hs=
-github.com/LF-Engineering/lfx-event-schema v0.1.30/go.mod h1:CfFIZ4mwzo88umf5+KxDQEzqlVkPG7Vx8eLK2oDfWIs=
+github.com/LF-Engineering/lfx-event-schema v0.1.33 h1:5Yk20tPkPbyOKEckcEF87qEQtl1H8l6TyA9wjKi+cl0=
+github.com/LF-Engineering/lfx-event-schema v0.1.33/go.mod h1:CfFIZ4mwzo88umf5+KxDQEzqlVkPG7Vx8eLK2oDfWIs=
 github.com/alecthomas/jsonschema v0.0.0-20210920000243-787cd8204a0d/go.mod h1:/n6+1/DWPltRLWL/VKyUxg6tzsl5kHUCcraimt4vr60=
 github.com/avast/retry-go v3.0.0+incompatible h1:4SOWQ7Qs+oroOTQOYnAHqelpCO0biHSxpiH9JdtuBj0=
 github.com/avast/retry-go v3.0.0+incompatible/go.mod h1:XtSnn+n/sHqQIpZ10K1qAevBhOOCWBLXXy3hyiqqBrY=