Improved datastore speed and deduplication mechanism to not overload db writes

Author: frikky

db-connector.go

@@ -12301,7 +12301,6 @@ func SetDatastoreKeyBulk(ctx context.Context, allKeys []CacheKeyData) ([]Datasto
 	nameKey := "org_cache"
 	timeNow := int64(time.Now().Unix())
 
-	newArray := []CacheKeyData{}
 	dbKeys := []*datastore.Key{}
 
 	existingInfo := []DatastoreKeyMini{}
@@ -12348,6 +12347,8 @@ func SetDatastoreKeyBulk(ctx context.Context, allKeys []CacheKeyData) ([]Datasto
 		go func(cacheData CacheKeyData, index int) {
 			defer wg.Done()
 
+			cacheData.Existed = false
+			cacheData.Changed = false 
 			cacheData.Created = timeNow
 			cacheData.Edited = timeNow
 
@@ -12368,8 +12369,9 @@ func SetDatastoreKeyBulk(ctx context.Context, allKeys []CacheKeyData) ([]Datasto
 				cacheData.PublicAuthorization = config.PublicAuthorization
 
 				cacheData.Existed = true
-			}
+			} 
 
+			cacheData.Changed = true 
 			sameValue := false
 			if getCacheError == nil && config.Value == cacheData.Value {
 				sameValue = true
@@ -12429,7 +12431,8 @@ func SetDatastoreKeyBulk(ctx context.Context, allKeys []CacheKeyData) ([]Datasto
 
 			if sameValue { 
 				if debug { 
-					log.Printf("[DEBUG] SAME VALUE FOR KEY %s in category %s. SHOULD skip datastore write.", cacheData.Key, cacheData.Category)
+					cacheData.Changed = false 
+					//log.Printf("[DEBUG] SAME VALUE FOR KEY %s in category %s. SHOULD skip datastore write.", cacheData.Key, cacheData.Category)
 				}
 
 				// FIXME: Should NOT be returning keys
@@ -12453,19 +12456,26 @@ func SetDatastoreKeyBulk(ctx context.Context, allKeys []CacheKeyData) ([]Datasto
 	close(datastoreKeys)
 
 	// Ensures no duplicates
+	newArray := []CacheKeyData{}
 	handledKeys := []string{}
+
+	skippedKeys := []string{}
 	for key := range cacheKeys {
 		if key.Key == "" {
 			continue
 		}
-		
+
+		// Assumes duplicates 
 		checkKey := fmt.Sprintf("%s_%s", key.Key, key.Category)
 		if ArrayContains(handledKeys, checkKey) {
+			if debug { 
+				log.Printf("[DEBUG] Skipping duplicate key %s in category %s", key.Key, key.Category)
+			}
+
+			handledKeys = append(handledKeys, checkKey)
 			continue
 		}
 
-		handledKeys = append(handledKeys, checkKey)
-
 		// Details to help with filtering old vs new
 		// Built for the "is_in_datastore" shuffle tools action
 		minKey := DatastoreKeyMini{
@@ -12474,8 +12484,15 @@ func SetDatastoreKeyBulk(ctx context.Context, allKeys []CacheKeyData) ([]Datasto
 		}
 
 		existingInfo = append(existingInfo, minKey)		
+		if !key.Changed {
+			parsedKey := fmt.Sprintf("%s_%s_%s", key.OrgId, key.Key, key.Category)
+			skippedKeys = append(skippedKeys, parsedKey)
+			//log.Printf("[DEBUG] Key %s did NOT change, skipping database", parsedKey)
+			continue
+		}
 
 		key.Existed = false 
+		key.Changed = false
 		newArray = append(newArray, key)
 
 	}
@@ -12486,22 +12503,67 @@ func SetDatastoreKeyBulk(ctx context.Context, allKeys []CacheKeyData) ([]Datasto
 			continue
 		}
 
+		// Duplicate handler
 		if ArrayContains(handledKeys, key.Name) {
 			continue
 		}
 
-		// Look for empty keys and continue if so:
-		//datastoreKeys <- *datastore.NameKey("", "", nil)
-		//cacheKeys <- CacheKeyData{}
+		if ArrayContains(skippedKeys, key.Name) {
+			continue
+		}
 
+		// Look for empty keys and continue if so:
 		handledKeys = append(handledKeys, key.Name)
 		dbKeys = append(dbKeys, &key)
 	}
 
+	// Autofixer on the fly
+	if len(newArray) != len(dbKeys) {
+		dbKeys = []*datastore.Key{}
+
+		// FIXME: newArray backwards to ALWAYS have latest key? Is latest last
+		// or first in the array? :thinking:
+		handledKeys := []string{}
+		skippedIndexes := []int{}
+		for skipIndex, cacheData:= range newArray {
+			datastoreId := fmt.Sprintf("%s_%s", cacheData.OrgId, cacheData.Key)
+			if len(cacheData.Category) > 0 && cacheData.Category != "default" {
+				// Adds category on the end
+				datastoreId = fmt.Sprintf("%s_%s", datastoreId, cacheData.Category)
+			}
+
+			if ArrayContains(handledKeys, datastoreId) {
+				skippedIndexes = append(skippedIndexes, skipIndex)
+				continue
+			}
+
+			handledKeys = append(handledKeys, datastoreId)
+
+			dbKeys = append(dbKeys, datastore.NameKey(nameKey, strings.ToLower(datastoreId), nil) )
+		}
+
+		// Cleanup newArray again due to transactional handler
+		// Example where problems show up are nested items:
+		// 		Multiple emails in the same thread
+		if len(skippedIndexes) > 0 {
+			newDeduped := []CacheKeyData{}
+			for index, val := range newArray {
+				if ArrayContainsInt(skippedIndexes, index) {
+					continue
+				}
+
+				newDeduped = append(newDeduped, val)
+			}
+
+			newArray = newDeduped
+		}
+	}
+
 	// New struct, to not add body, author etc
 	if project.DbType == "opensearch" {
 		var buf bytes.Buffer
 
+		// Bulk encoding them
 		for _, cacheData := range newArray {
 			cacheId := fmt.Sprintf("%s_%s", cacheData.OrgId, cacheData.Key)
 			if len(cacheData.Category) > 0 && cacheData.Category != "default" {
@@ -13864,6 +13926,8 @@ func GetAllCacheKeys(ctx context.Context, orgId string, category string, max int
 		query := datastore.NewQuery(nameKey).Filter("OrgId =", orgId).Order("-Edited")
 		if len(category) > 0 {
 			query = query.Filter("category =", category)
+		} else {
+			query = query.Filter("category =", "")
 		}
 
 		query = query.Limit(max)

shared.go

@@ -3605,6 +3605,19 @@ func ArrayContainsLower(visited []string, id string) bool {
 	for _, item := range visited {
 		if strings.ToLower(item) == strings.ToLower(id) {
 			found = true
+			break
+		}
+	}
+
+	return found
+}
+
+func ArrayContainsInt(visited []int, id int) bool {
+	found := false
+	for _, item := range visited {
+		if item == id {
+			found = true
+			break
 		}
 	}
 
@@ -3616,6 +3629,7 @@ func ArrayContains(visited []string, id string) bool {
 	for _, item := range visited {
 		if item == id {
 			found = true
+			break
 		}
 	}
 
@@ -19435,9 +19449,20 @@ func HandleSetDatastoreKey(resp http.ResponseWriter, request *http.Request) {
 		KeysExisted []DatastoreKeyMini `json:"keys_existed"`
 	}
 
+	/*
+	// For testing deduplication
 	if debug { 
-		log.Printf("[DEBUG] EXISTINGINFO: %#v", existingInfo)
+		found := []string{}
+		for _, existing := range existingInfo {
+			if ArrayContains(found, existing.Key) {
+				log.Printf("[DEBUG] Key %s already found in existing info", existing.Key)
+				continue
+			}
+
+			found = append(found, existing.Key)
+		}
 	}
+	*/
 
 	returnData := returnStruct{
 		Success: true,

structs.go

@@ -1144,6 +1144,7 @@ type CacheKeyData struct {
 	Edited  int64 `json:"edited" datastore:"Edited"`
 
 	Existed bool `json:"existed,omitempty" datastore:"Existed"` // If the key existed before the update. Should always be set back to false.
+	Changed bool `json:"changed,omitempty" datastore:"Changed"` // If the value was changed. Should always be set back to false.
 	Encrypted bool `json:"encrypted" datastore:"Encrypted"`
 	FormattedKey        string   `json:"formatted_key,omitempty" datastore:"FormattedKey"`
 	PublicAuthorization string   `json:"public_authorization,omitempty" datastore:"PublicAuthorization"` // Used for public authorization