Change caching mechanism (#59)

* change caching mevhanism

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

* clean up

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

* use v2 path

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

* handle orphand field

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

* clean up

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

* code clean up

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

* test changes

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

* handle csv issue

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

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

Author: khalifapro

.circleci/config.yml

@@ -91,7 +91,7 @@ workflows:
             - build
           filters:
             branches:
-              only: main
+              only: change-caching-mechanism #main
             tags:
               ignore: /.*/
       - approve_prod:

cmd/git/git.go

@@ -2,7 +2,10 @@ package main
 
 import (
 	"bufio"
+	"bytes"
 	"context"
+	"crypto/sha256"
+	"encoding/csv"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -513,6 +516,8 @@ var (
 	// max upstream date
 	gMaxUpstreamDt    time.Time
 	gMaxUpstreamDtMtx = &sync.Mutex{}
+	cachedCommits     = make(map[string]CommitCache)
+	commitsCacheFile  = "commits-cache.csv"
 )
 
 // Publisher - for streaming data to Kinesis
@@ -1766,25 +1771,23 @@ func (j *DSGit) GitEnrichItems(ctx *shared.Ctx, thrN int, items []interface{}, d
 			data := j.GetModelData(ctx, *docs)
 			if j.Publisher != nil {
 				formattedData := make([]interface{}, 0)
-				vals := make([]map[string]interface{}, 0)
+				commits := make([]CommitCache, 0)
 				for _, d := range data {
-					isCreated, err := j.cacheProvider.IsKeyCreated(j.endpoint, d.Payload.ID)
-					if err != nil {
-						j.log.WithFields(logrus.Fields{"operation": "GitEnrichItems"}).Errorf("error creating cache for commit %s, error %v", d.Payload.SHA, err)
-						continue
-					}
+					isCreated := isKeyCreated(d.Payload.ID)
 					if !isCreated {
 						formattedData = append(formattedData, d)
-						b, err := json.Marshal(d)
-						if err != nil {
+						b, er := json.Marshal(d)
+						if er != nil {
 							j.log.WithFields(logrus.Fields{"operation": "GitEnrichItems"}).Errorf("error marshall data for commit %s, error %v", d.Payload.SHA, err)
 							continue
 						}
-						vals = append(vals, map[string]interface{}{
-							"id": d.Payload.ID,
-							"data": map[string]interface{}{
-								"content": b,
-							},
+						tStamp := d.Payload.SyncTimestamp.Unix()
+						contentHash := fmt.Sprintf("%x", sha256.Sum256(b))
+						commits = append(commits, CommitCache{
+							Timestamp:      fmt.Sprintf("%v", tStamp),
+							EntityID:       d.Payload.ID,
+							SourceEntityID: d.Payload.SHA,
+							Hash:           contentHash,
 						})
 					}
 				}
@@ -1796,13 +1799,10 @@ func (j *DSGit) GitEnrichItems(ctx *shared.Ctx, thrN int, items []interface{}, d
 						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)
+				if err = j.createCacheFile(commits, path); err != nil {
+					return
 				}
+
 			} else {
 				var jsonBytes []byte
 				jsonBytes, err = jsoniter.Marshal(data)
@@ -2396,6 +2396,35 @@ func (j *DSGit) Sync(ctx *shared.Ctx) (err error) {
 			j.log.WithFields(logrus.Fields{"operation": "Sync"}).Infof("%s resuming from %v (%d threads)", j.URL, ctx.DateFrom, thrN)
 		}
 	}
+	comB, err := j.cacheProvider.GetFileByKey(j.endpoint, commitsCacheFile)
+	if err != nil {
+		return
+	}
+	reader := csv.NewReader(bytes.NewBuffer(comB))
+	records, err := reader.ReadAll()
+	if err != nil {
+		return
+	}
+	for i, record := range records {
+		if i == 0 {
+			continue
+		}
+		orphaned, err := strconv.ParseBool(record[5])
+		if err != nil {
+			orphaned = false
+		}
+		if lastSync != "" {
+			orphaned = true
+		}
+		cachedCommits[record[1]] = CommitCache{
+			Timestamp:      record[0],
+			EntityID:       record[1],
+			SourceEntityID: record[2],
+			FileLocation:   record[3],
+			Hash:           record[4],
+			Orphaned:       orphaned,
+		}
+	}
 	if ctx.DateTo != nil {
 		j.log.WithFields(logrus.Fields{"operation": "Sync"}).Infof("%s fetching till %v (%d threads)", j.URL, ctx.DateTo, thrN)
 	}
@@ -2740,7 +2769,69 @@ func (j *DSGit) initStructuredLogger() {
 
 // AddCacheProvider - adds cache provider
 func (j *DSGit) AddCacheProvider() {
-	cacheProvider := cache.NewManager(GitDataSource, os.Getenv("STAGE"))
+	cacheProvider := cache.NewManager(fmt.Sprintf("v2/%s", GitDataSource), os.Getenv("STAGE"))
 	j.cacheProvider = *cacheProvider
 	j.endpoint = strings.ReplaceAll(strings.TrimPrefix(strings.TrimPrefix(strings.TrimPrefix(j.URL, "https://"), "git://"), "http://"), "/", "-")
 }
+
+func (j *DSGit) createCacheFile(cache []CommitCache, path string) error {
+	for _, comm := range cache {
+		comm.FileLocation = path
+		cachedCommits[comm.EntityID] = comm
+	}
+	records := [][]string{
+		{"timestamp", "entity_id", "source_entity_id", "file_location", "hash", "orphaned"},
+	}
+	for _, c := range cachedCommits {
+		records = append(records, []string{c.Timestamp, c.EntityID, c.SourceEntityID, c.FileLocation, c.Hash, strconv.FormatBool(c.Orphaned)})
+	}
+
+	csvFile, err := os.Create(commitsCacheFile)
+	if err != nil {
+		return err
+	}
+
+	w := csv.NewWriter(csvFile)
+	err = w.WriteAll(records)
+	if err != nil {
+		return err
+	}
+	err = csvFile.Close()
+	if err != nil {
+		return err
+	}
+	file, err := os.ReadFile(commitsCacheFile)
+	if err != nil {
+		return err
+	}
+	err = os.Remove(commitsCacheFile)
+	if err != nil {
+		return err
+	}
+	err = j.cacheProvider.UpdateFileByKey(j.endpoint, commitsCacheFile, file)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func isKeyCreated(id string) bool {
+	c, ok := cachedCommits[id]
+	if ok {
+		c.Orphaned = false
+		cachedCommits[id] = c
+		return true
+	}
+	return false
+}
+
+// CommitCache single commit cache schema
+type CommitCache struct {
+	Timestamp      string `json:"timestamp"`
+	EntityID       string `json:"entity_id"`
+	SourceEntityID string `json:"source_entity_id"`
+	FileLocation   string `json:"file_location"`
+	Hash           string `json:"hash"`
+	Orphaned       bool   `json:"orphaned"`
+}