feat: add is_pushed columns to tables for reconciliation

.gitignore

@@ -1,3 +1,4 @@
 .bin
 *.out.json
 ginkgo.report
+.vscode

context/properties.go

@@ -34,10 +34,9 @@ func (p Properties) Duration(key string, def time.Duration) time.Duration {
 	} else if dur, err := time.ParseDuration(d); err != nil {
 		logger.Warnf("property[%s] invalid duration %s", key, d)
 		return def
-	} else if err == nil {
+	} else {
 		return dur
 	}
-	return def
 }
 
 func (p Properties) Int(key string, def int) int {
@@ -46,10 +45,9 @@ func (p Properties) Int(key string, def int) int {
 	} else if i, err := strconv.Atoi(d); err != nil {
 		logger.Warnf("property[%s] invalid int %s", key, d)
 		return def
-	} else if err == nil {
+	} else {
 		return i
 	}
-	return def
 }
 
 func (p Properties) Off(key string) bool {
@@ -84,7 +82,11 @@ func (k Context) Properties() Properties {
 	return props
 }
 
-func SetLocalProperty(ctx Context, property, value string) {
+func SetLocalProperty(property, value string) {
+	if Local == nil {
+		Local = make(map[string]string)
+	}
+
 	Local[property] = value
 }
 

models/artifacts.go

@@ -4,6 +4,8 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/samber/lo"
+	"gorm.io/gorm"
 )
 
 // Artifact represents the artifacts table
@@ -25,3 +27,17 @@ type Artifact struct {
 	DeletedAt           *time.Time `json:"deleted_at,omitempty" yaml:"deleted_at,omitempty" time_format:"postgres_timestamp"`
 	ExpiresAt           *time.Time `json:"expires_at,omitempty" yaml:"expires_at,omitempty" time_format:"postgres_timestamp"`
 }
+
+func (t Artifact) TableName() string {
+	return "artifacts"
+}
+
+func (t Artifact) PK() string {
+	return t.ID.String()
+}
+
+func (t Artifact) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []Artifact
+	err := db.Where("is_pushed IS FALSE").Find(&items).Error
+	return lo.Map(items, func(i Artifact, _ int) DBTable { return i }), err
+}

models/canary.go

@@ -5,6 +5,8 @@ import (
 
 	"github.com/flanksource/duty/types"
 	"github.com/google/uuid"
+	"github.com/samber/lo"
+	"gorm.io/gorm"
 )
 
 type Canary struct {
@@ -22,10 +24,20 @@ type Canary struct {
 	DeletedAt   *time.Time          `json:"deleted_at,omitempty" yaml:"deleted_at,omitempty" time_format:"postgres_timestamp"`
 }
 
+func (t Canary) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []Canary
+	err := db.Where("is_pushed IS FALSE").Find(&items).Error
+	return lo.Map(items, func(i Canary, _ int) DBTable { return i }), err
+}
+
 func (c Canary) GetCheckID(checkName string) string {
 	return c.Checks[checkName]
 }
 
+func (c Canary) PK() string {
+	return c.ID.String()
+}
+
 func (c Canary) TableName() string {
 	return "canaries"
 }

models/checks.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/flanksource/duty/types"
 	"github.com/google/uuid"
+	"github.com/samber/lo"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
 )
@@ -59,6 +60,16 @@ type Check struct {
 	TotalRuns       int        `json:"totalRuns,omitempty" gorm:"-"`
 }
 
+func (t Check) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []Check
+	err := db.Where("is_pushed IS FALSE").Find(&items).Error
+	return lo.Map(items, func(i Check, _ int) DBTable { return i }), err
+}
+
+func (c Check) PK() string {
+	return c.ID.String()
+}
+
 func (c Check) TableName() string {
 	return "checks"
 }
@@ -112,6 +123,29 @@ type CheckStatus struct {
 	IsPushed bool `json:"is_pushed,omitempty"`
 }
 
+func (s CheckStatus) UpdateIsPushed(db *gorm.DB, items []DBTable) error {
+	ids := lo.Map(items, func(a DBTable, _ int) []any {
+		c := any(a).(CheckStatus)
+		return []any{c.CheckID, c.Time}
+	})
+
+	return db.Model(&CheckStatus{}).Where("(check_id, time) IN ?", ids).Update("is_pushed", true).Error
+}
+
+func (s CheckStatus) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []CheckStatus
+	err := db.Select("check_statuses.*").
+		Joins("LEFT JOIN checks ON checks.id = check_statuses.check_id").
+		Where("checks.agent_id = ?", uuid.Nil).
+		Where("check_statuses.is_pushed IS FALSE").
+		Find(&items).Error
+	return lo.Map(items, func(i CheckStatus, _ int) DBTable { return i }), err
+}
+
+func (s CheckStatus) PK() string {
+	return s.CheckID.String() + s.Time
+}
+
 func (s CheckStatus) GetTime() (time.Time, error) {
 	return time.Parse(time.DateTime, s.Time)
 }
@@ -195,13 +229,36 @@ type CheckConfigRelationship struct {
 	DeletedAt  *time.Time `json:"deleted_at,omitempty"`
 }
 
+func (s CheckConfigRelationship) UpdateIsPushed(db *gorm.DB, items []DBTable) error {
+	ids := lo.Map(items, func(a DBTable, _ int) []string {
+		c := any(a).(CheckConfigRelationship)
+		return []string{c.ConfigID.String(), c.CheckID.String(), c.CanaryID.String(), c.SelectorID}
+	})
+
+	return db.Model(&CheckConfigRelationship{}).Where("(config_id, check_id, canary_id, selector_id) IN ?", ids).Update("is_pushed", true).Error
+}
+
+func (c CheckConfigRelationship) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []CheckConfigRelationship
+	err := db.Select("check_config_relationships.*").
+		Joins("LEFT JOIN config_items ci ON check_config_relationships.config_id = ci.id").
+		Where("ci.agent_id = ?", uuid.Nil).
+		Where("check_config_relationships.is_pushed IS FALSE").
+		Find(&items).Error
+	return lo.Map(items, func(i CheckConfigRelationship, _ int) DBTable { return i }), err
+}
+
 func (c *CheckConfigRelationship) Save(db *gorm.DB) error {
 	return db.Clauses(clause.OnConflict{
 		Columns:   []clause.Column{{Name: "canary_id"}, {Name: "check_id"}, {Name: "config_id"}, {Name: "selector_id"}},
 		UpdateAll: true,
 	}).Create(c).Error
 }
 
+func (c CheckConfigRelationship) PK() string {
+	return c.ConfigID.String() + "," + c.CheckID.String() + "," + c.CanaryID.String() + "," + c.SelectorID
+}
+
 func (CheckConfigRelationship) TableName() string {
 	return "check_config_relationships"
 }

models/common.go

@@ -26,3 +26,8 @@ func asMap(t any, removeFields ...string) map[string]any {
 
 	return m
 }
+
+type DBTable interface {
+	PK() string
+	TableName() string
+}

models/components.go

@@ -12,6 +12,7 @@ import (
 	"github.com/flanksource/commons/logger"
 	"github.com/flanksource/duty/types"
 	"github.com/google/uuid"
+	"github.com/samber/lo"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
 	"gorm.io/gorm/schema"
@@ -72,6 +73,12 @@ type Component struct {
 	NodeProcessed bool `json:"-" gorm:"-"`
 }
 
+func (t Component) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []Component
+	err := db.Where("is_pushed IS FALSE").Find(&items).Error
+	return lo.Map(items, func(i Component, _ int) DBTable { return i }), err
+}
+
 func (c Component) PK() string {
 	return c.ID.String()
 }
@@ -449,7 +456,31 @@ type ComponentRelationship struct {
 	DeletedAt        *time.Time `json:"deleted_at,omitempty"`
 }
 
-func (cr ComponentRelationship) TableName() string {
+func (s ComponentRelationship) UpdateIsPushed(db *gorm.DB, items []DBTable) error {
+	ids := lo.Map(items, func(a DBTable, _ int) []string {
+		c := any(a).(ComponentRelationship)
+		return []string{c.ComponentID.String(), c.RelationshipID.String(), c.SelectorID}
+	})
+
+	return db.Model(&ComponentRelationship{}).Where("(component_id, relationship_id, selector_id) IN ?", ids).Update("is_pushed", true).Error
+}
+
+func (cr ComponentRelationship) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []ComponentRelationship
+	err := db.Select("component_relationships.*").
+		Joins("LEFT JOIN components c ON component_relationships.component_id = c.id").
+		Joins("LEFT JOIN components rel ON component_relationships.relationship_id = rel.id").
+		Where("c.agent_id = ? AND rel.agent_id = ?", uuid.Nil, uuid.Nil).
+		Where("component_relationships.is_pushed IS FALSE").
+		Find(&items).Error
+	return lo.Map(items, func(i ComponentRelationship, _ int) DBTable { return i }), err
+}
+
+func (cr ComponentRelationship) PK() string {
+	return cr.ComponentID.String() + "," + cr.RelationshipID.String() + "," + cr.SelectorID
+}
+
+func (ComponentRelationship) TableName() string {
 	return "component_relationships"
 }
 
@@ -462,6 +493,34 @@ type ConfigComponentRelationship struct {
 	DeletedAt   *time.Time `json:"deleted_at,omitempty"`
 }
 
+func (s ConfigComponentRelationship) UpdateIsPushed(db *gorm.DB, items []DBTable) error {
+	ids := lo.Map(items, func(a DBTable, _ int) []string {
+		c := any(a).(ConfigComponentRelationship)
+		return []string{c.ComponentID.String(), c.ConfigID.String()}
+	})
+
+	return db.Model(&ConfigComponentRelationship{}).Where("(component_id, config_id) IN ?", ids).Update("is_pushed", true).Error
+}
+
+func (t ConfigComponentRelationship) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []ConfigComponentRelationship
+	err := db.Select("config_component_relationships.*").
+		Joins("LEFT JOIN components c ON config_component_relationships.component_id = c.id").
+		Joins("LEFT JOIN config_items ci ON config_component_relationships.config_id = ci.id").
+		Where("c.agent_id = ? AND ci.agent_id = ?", uuid.Nil, uuid.Nil).
+		Where("config_component_relationships.is_pushed IS FALSE").
+		Find(&items).Error
+	return lo.Map(items, func(i ConfigComponentRelationship, _ int) DBTable { return i }), err
+}
+
+func (t ConfigComponentRelationship) PK() string {
+	return t.ComponentID.String() + "," + t.ConfigID.String()
+}
+
+func (ConfigComponentRelationship) TableName() string {
+	return "config_component_relationships"
+}
+
 var ConfigID = func(c ConfigComponentRelationship, i int) string {
 	return c.ConfigID.String()
 }
@@ -470,10 +529,6 @@ var ConfigSelectorID = func(c ConfigComponentRelationship, i int) string {
 	return c.SelectorID
 }
 
-func (cr ConfigComponentRelationship) TableName() string {
-	return "config_component_relationships"
-}
-
 type CheckComponentRelationship struct {
 	ComponentID uuid.UUID  `json:"component_id,omitempty"`
 	CheckID     uuid.UUID  `json:"check_id,omitempty"`
@@ -484,13 +539,37 @@ type CheckComponentRelationship struct {
 	DeletedAt   *time.Time `json:"deleted_at,omitempty"`
 }
 
+func (s CheckComponentRelationship) UpdateIsPushed(db *gorm.DB, items []DBTable) error {
+	ids := lo.Map(items, func(a DBTable, _ int) []string {
+		c := any(a).(CheckComponentRelationship)
+		return []string{c.ComponentID.String(), c.CheckID.String(), c.CanaryID.String(), c.SelectorID}
+	})
+
+	return db.Model(&CheckComponentRelationship{}).Where("(component_id, check_id, canary_id, selector_id) IN ?", ids).Update("is_pushed", true).Error
+}
+
+func (t CheckComponentRelationship) GetUnpushed(db *gorm.DB) ([]DBTable, error) {
+	var items []CheckComponentRelationship
+	err := db.Select("check_component_relationships.*").
+		Joins("LEFT JOIN components c ON check_component_relationships.component_id = c.id").
+		Joins("LEFT JOIN canaries ON check_component_relationships.canary_id = canaries.id").
+		Where("c.agent_id = ? AND canaries.agent_id = ?", uuid.Nil, uuid.Nil).
+		Where("check_component_relationships.is_pushed IS FALSE").
+		Find(&items).Error
+	return lo.Map(items, func(i CheckComponentRelationship, _ int) DBTable { return i }), err
+}
+
 func (c *CheckComponentRelationship) Save(db *gorm.DB) error {
 	return db.Clauses(clause.OnConflict{
 		Columns:   []clause.Column{{Name: "canary_id"}, {Name: "check_id"}, {Name: "component_id"}, {Name: "selector_id"}},
 		UpdateAll: true,
 	}).Create(c).Error
 }
 
+func (c CheckComponentRelationship) PK() string {
+	return c.ComponentID.String() + "," + c.CheckID.String() + "," + c.CanaryID.String() + "," + c.SelectorID
+}
+
 func (CheckComponentRelationship) TableName() string {
 	return "check_component_relationships"
 }