refactor: add temp cache for config changes

We don't want to create new config changes with a ON CONFLICT DO NOTHING
clause because now we don't want the same config change to take up
multiple quotas from the rate limiter.

i.e. if an aws scraper is run @every 5m, we'll be trying to insert the
same config changes generated by the cloudtrail scraper again & again on
every run. The same change will take up one more quota from the rate
limiter on every run.

By knowing that the change already exist, we can avoid inserting that
change in the first place and the rate limiter will be happy about it.

Author: adityathebe

api/cache.go

@@ -16,6 +16,8 @@ type TempCache struct {
 	ctx     context.Context
 	items   map[string]models.ConfigItem
 	aliases map[string]string
+
+	changes map[string]struct{}
 }
 
 func (t TempCache) FindExternal(ext v1.ExternalID) (*models.ConfigItem, error) {
@@ -76,6 +78,38 @@ func (t TempCache) Insert(item models.ConfigItem) {
 	t.items[strings.ToLower(item.ID)] = item
 }
 
+func (t TempCache) IsChangePersisted(configID, externalChangeID string) (bool, error) {
+	if configID == "" || externalChangeID == "" {
+		return false, nil
+	}
+
+	configID = strings.ToLower(configID)
+	externalChangeID = strings.ToLower(externalChangeID)
+
+	if t.changes == nil {
+		t.changes = make(map[string]struct{})
+	}
+
+	if _, ok := t.changes[configID+externalChangeID]; ok {
+		return true, nil
+	}
+
+	var result models.ConfigChange
+	if err := t.ctx.DB().Select("id").Where("config_id = ?", configID).
+		Where("external_change_id = ?", externalChangeID).
+		Limit(1).
+		Find(&result).Error; err != nil {
+		return false, err
+	}
+
+	if result.ID != "" {
+		t.changes[configID+externalChangeID] = struct{}{}
+		return true, nil
+	}
+
+	return false, nil
+}
+
 func (t TempCache) Get(id string) (*models.ConfigItem, error) {
 	id = strings.ToLower(id)
 	if id == "" {

db/models/config_change.go

@@ -7,7 +7,6 @@ import (
 	v1 "github.com/flanksource/config-db/api/v1"
 	"github.com/google/uuid"
 	"gorm.io/gorm"
-	"gorm.io/gorm/clause"
 )
 
 // ConfigChange represents the config change database table
@@ -66,6 +65,5 @@ func (c *ConfigChange) BeforeCreate(tx *gorm.DB) (err error) {
 		c.ID = uuid.New().String()
 	}
 
-	tx.Statement.AddClause(clause.OnConflict{DoNothing: true})
 	return
 }

db/update.go

@@ -269,7 +269,11 @@ func extractChanges(ctx api.ScrapeContext, result *v1.ScrapeResult, ci *models.C
 		if changeResult.UpdateExisting {
 			updates = append(updates, change)
 		} else {
-			newOnes = append(newOnes, change)
+			if ok, err := ctx.TempCache().IsChangePersisted(change.ConfigID, change.ExternalChangeId); err != nil {
+				return nil, nil, fmt.Errorf("failed to check if change is persisted: %w", err)
+			} else if !ok {
+				newOnes = append(newOnes, change)
+			}
 		}
 	}