chore: store relation between commit hash and transaction timestamp (#…

…1946)

We need to store the relation between commit hash (repo state) and
cd-service transaction timestamp (database state), so that we have a 1
to 1 temporal relation between database state and repository state.

Ref: SRX-R7B9VQ

database/migrations/postgres/1726064901745858_transaction_timestamps.up.sql

@@ -0,0 +1,8 @@
+-- Table that stores which cd-service transaction timestamp corresponds to a certain commit hash
+-- on the manifest repo
+CREATE TABLE IF NOT EXISTS commit_transaction_timestamps
+(
+    commitHash VARCHAR NOT NULL,
+    transactionTimestamp TIMESTAMP NOT NULL,
+    PRIMARY KEY(commitHash)
+);

database/migrations/sqlite/1726064901745858_transaction_timestamps.up.sql

@@ -0,0 +1 @@
+../postgres/1726064901745858_transaction_timestamps.up.sql

pkg/db/db.go

@@ -4970,3 +4970,30 @@ func (h *DBHandler) processAllDeploymentRow(ctx context.Context, err error, rows
 	}
 	return deployments, nil
 }
+
+func (h *DBHandler) DBWriteCommitTransactionTimestamp(ctx context.Context, tx *sql.Tx, commitHash string, timestamp time.Time) error {
+	span, _ := tracer.StartSpanFromContext(ctx, "DBWriteCommitTransactionTimestamp")
+	defer span.Finish()
+
+	if h == nil {
+		return nil
+	}
+	if tx == nil {
+		return fmt.Errorf("attempting to write to the commit_transaction_timestamps table without a transaction")
+	}
+
+	insertQuery := h.AdaptQuery(
+		"INSERT INTO commit_transaction_timestamps (commitHash, transactionTimestamp) VALUES (?, ?);",
+	)
+
+	span.SetTag("query", insertQuery)
+	_, err := tx.Exec(
+		insertQuery,
+		commitHash,
+		timestamp,
+	)
+	if err != nil {
+		return fmt.Errorf("DBWriteCommitTransactionTimestamp error executing query: %w", err)
+	}
+	return nil
+}

services/manifest-repo-export-service/pkg/cmd/server.go

@@ -322,10 +322,17 @@ func processEsls(ctx context.Context, repo repository.Repository, dbHandler *db.
 					logger.FromContext(ctx).Sugar().Warnf("error pushing, will try again in %v", d)
 					measurePushes(ddMetrics, log, true)
 					time.Sleep(d)
+					return err2
 				} else {
 					measurePushes(ddMetrics, log, false)
 				}
-				return err2
+
+				//Get latest commit. Write esl timestamp and commit hash.
+				commit, err := repo.GetHeadCommit()
+				if err != nil {
+					return err
+				}
+				return dbHandler.DBWriteCommitTransactionTimestamp(ctx, transaction, commit.Id().String(), esl.Created)
 			})
 			if err != nil {
 				err3 := repo.FetchAndReset(ctx)

services/manifest-repo-export-service/pkg/repository/repository.go

@@ -62,6 +62,7 @@ type Repository interface {
 	StateAt(oid *git.Oid) (*State, error)
 	FetchAndReset(ctx context.Context) error
 	PushRepo(ctx context.Context) error
+	GetHeadCommit() (*git.Commit, error)
 }
 
 type TransformerBatchApplyError struct {
@@ -417,6 +418,19 @@ func (r *repository) PushRepo(ctx context.Context) error {
 	return nil
 }
 
+func (r *repository) GetHeadCommit() (*git.Commit, error) {
+	ref, err := r.repository.Head()
+	if err != nil {
+		return nil, fmt.Errorf("Error fetching HEAD: %v", err)
+	}
+	commit, err := r.repository.LookupCommit(ref.Target())
+	if err != nil {
+		return nil, fmt.Errorf("Error transalting into commit: %v", err)
+	}
+	return commit, nil
+
+}
+
 func (r *repository) ApplyTransformersInternal(ctx context.Context, transaction *sql.Tx, transformer Transformer) ([]string, *State, []*TransformerResult, *TransformerBatchApplyError) {
 	span, ctx := tracer.StartSpanFromContext(ctx, "ApplyTransformersInternal")
 	defer span.Finish()