fix(purchase): persist the Redshift first-dispatch time and pass it as RequestedAt

internal/config/interfaces.go

@@ -792,6 +792,11 @@ type StoreInterface interface {
 	SaveExecutionIfStatusTx(ctx context.Context, tx pgx.Tx, execution *PurchaseExecution, expected []string) error
 	RecoverReapedExecutionTx(ctx context.Context, tx pgx.Tx, execution *PurchaseExecution) error
 
+	// StampFirstDispatchTx sets the lineage's first_dispatch_at on the
+	// running row (kept once set; minted only per the mint rule) and reports
+	// whether it is within horizon on the database clock.
+	StampFirstDispatchTx(ctx context.Context, tx pgx.Tx, executionID string, horizon time.Duration) (StampResult, error)
+
 	// ListRecentSubmitsTx takes a per-creator transaction-scoped lock and
 	// returns the creator's web-submitted executions scheduled at or after
 	// since that are pending, in flight or bought. executePurchase's

internal/config/store_postgres.go

@@ -1178,8 +1178,8 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 			created_by_user_id, retry_execution_id, retry_attempt_n,
 			approval_token_expires_at,
 			executed_by_user_id, executed_at, pre_approval_skip_reason,
-			idempotency_key, scheduled_execution_at, no_submission, created_by_system
-		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)
+			idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
+		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30)
 		ON CONFLICT (execution_id) DO UPDATE SET
 			status = $3,
 			notification_sent = $6,
@@ -1216,7 +1216,7 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 		planIDArg = execution.PlanID
 	}
 
-	// created_by_user_id, created_by_system and retry_attempt_n are INSERT-only: the original
+	// created_by_user_id, created_by_system, first_dispatch_at and retry_attempt_n are INSERT-only: the original
 	// creator and the retry-chain position must not be rewritten by an
 	// ON CONFLICT update (e.g. the scheduler upserting status transitions).
 	// They are omitted from the DO UPDATE SET clause above.
@@ -1259,6 +1259,7 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 		execution.ScheduledExecutionAt,
 		execution.NoSubmission,
 		execution.CreatedBySystem,
+		execution.FirstDispatchAt,
 	)
 
 	if err != nil {
@@ -1418,7 +1419,7 @@ func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, q executi
 		          created_by_user_id, retry_execution_id, retry_attempt_n,
 		          approval_token_expires_at,
 		          executed_by_user_id, executed_at, pre_approval_skip_reason,
-		          idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		          idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 	`
 
 	records, err := queryExecutionsOn(ctx, q, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty, tokenHash)
@@ -1682,7 +1683,7 @@ func (s *PostgresStore) GetExecutionsByStatusesFiltered(ctx context.Context, q E
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE ` + strings.Join(conds, "\n\t\t  AND ") + `
 		ORDER BY (status = 'pending' AND scheduled_date > now()) ASC,
@@ -1794,7 +1795,7 @@ func (s *PostgresStore) GetPlannedExecutions(ctx context.Context, statuses []str
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE status = ANY($1)
 		ORDER BY scheduled_date ASC NULLS LAST, id ASC
@@ -1819,7 +1820,7 @@ func (s *PostgresStore) GetStaleApprovedExecutions(ctx context.Context, olderTha
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE status = 'approved' AND updated_at < NOW() - $1::interval
 	`
@@ -1861,7 +1862,7 @@ func (s *PostgresStore) ListStuckExecutions(ctx context.Context, statuses []stri
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE status = ANY($1)
 		  AND updated_at < NOW() - $2::interval
@@ -1882,7 +1883,7 @@ func (s *PostgresStore) GetPendingExecutions(ctx context.Context) ([]PurchaseExe
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE status IN ('pending', 'notified')
 		  AND (expires_at IS NULL OR expires_at > NOW())
@@ -1964,7 +1965,7 @@ func (s *PostgresStore) ListRecentSubmitsTx(ctx context.Context, tx pgx.Tx, crea
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE source = $3
 		  AND status IN (` + submitDedupeStatuses + `)
@@ -1999,7 +2000,7 @@ func getExecutionByIDOn(ctx context.Context, q executionQuerier, executionID str
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE execution_id = $1
 	`
@@ -2045,7 +2046,7 @@ func (s *PostgresStore) GetExecutionByPlanAndDate(ctx context.Context, planID st
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE plan_id = $1 AND scheduled_date = $2
 	`
@@ -2249,6 +2250,7 @@ func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) {
 			&scheduledExecutionAt,
 			&exec.NoSubmission,
 			&exec.CreatedBySystem,
+			&exec.FirstDispatchAt,
 		)
 		if err != nil {
 			return nil, fmt.Errorf("failed to scan execution: %w", err)
@@ -2290,7 +2292,7 @@ func (s *PostgresStore) GetScheduledExecutionsDue(ctx context.Context) ([]Purcha
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
 		       executed_by_user_id, executed_at, pre_approval_skip_reason,
-		       idempotency_key, scheduled_execution_at, no_submission, created_by_system
+		       idempotency_key, scheduled_execution_at, no_submission, created_by_system, first_dispatch_at
 		FROM purchase_executions
 		WHERE status = 'scheduled'
 		  AND scheduled_execution_at IS NOT NULL

internal/config/store_postgres_first_dispatch.go

@@ -0,0 +1,50 @@
+package config
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// StampResult is the outcome of StampFirstDispatchTx. At is nil when no stamp
+// exists and none may be minted; WithinHorizon is judged on the database
+// clock and is false when At is nil.
+type StampResult struct {
+	At            *time.Time
+	WithinHorizon bool
+}
+
+// StampFirstDispatchTx stamps a lineage's first dispatch time on the claimed
+// running row, with the database clock, and returns it. An existing stamp is
+// kept. A stamp is minted only at attempt 0, or when the predecessor is a
+// no_submission attempt 0 (the whole lineage never submitted); otherwise a
+// stamp-less successor gets At == nil and the caller must refuse it. A row
+// that is not running returns ErrExecutionNotInExpectedStatus.
+//
+// It takes a tx so the dispatch gate can stamp and check conflicts atomically.
+func (s *PostgresStore) StampFirstDispatchTx(ctx context.Context, tx pgx.Tx, executionID string, horizon time.Duration) (StampResult, error) {
+	var res StampResult
+	var within *bool
+	err := tx.QueryRow(ctx, `
+		UPDATE purchase_executions
+		SET first_dispatch_at = COALESCE(first_dispatch_at, CASE
+			WHEN retry_attempt_n = 0 OR EXISTS (
+				SELECT 1 FROM purchase_executions p
+				WHERE p.retry_execution_id = purchase_executions.execution_id
+				  AND p.no_submission AND p.retry_attempt_n = 0
+			) THEN now() END)
+		WHERE execution_id = $1 AND status = 'running'
+		RETURNING first_dispatch_at, first_dispatch_at > now() - make_interval(secs => $2)`,
+		executionID, horizon.Seconds()).Scan(&res.At, &within)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return StampResult{}, fmt.Errorf("%w: execution %s is not running", ErrExecutionNotInExpectedStatus, executionID)
+	}
+	if err != nil {
+		return StampResult{}, fmt.Errorf("stamp first dispatch of execution %s: %w", executionID, err)
+	}
+	res.WithinHorizon = within != nil && *within
+	return res, nil
+}

internal/config/store_postgres_first_dispatch_at_integration_test.go

@@ -0,0 +1,55 @@
+//go:build integration
+// +build integration
+
+package config
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// first_dispatch_at is written only by the INSERT; a stale struct upserted
+// later (nil or a different value) must not move or clear it.
+func TestSavePurchaseExecution_FirstDispatchAtIsInsertOnly(t *testing.T) {
+	ctx := context.Background()
+	store := setupRampStepStore(ctx, t)
+
+	stamp := time.Now().Add(-time.Hour).UTC().Truncate(time.Microsecond)
+	exec := &PurchaseExecution{
+		ExecutionID: uuid.New().String(), IdempotencyKey: uuid.New().String(),
+		Status: "pending", StepNumber: 1, ScheduledDate: time.Now(), FirstDispatchAt: &stamp,
+	}
+	require.NoError(t, store.SavePurchaseExecution(ctx, exec))
+
+	got, err := store.GetExecutionByID(ctx, exec.ExecutionID)
+	require.NoError(t, err)
+	require.NotNil(t, got.FirstDispatchAt, "insert persists the stamp")
+	assert.True(t, stamp.Equal(*got.FirstDispatchAt))
+
+	other := stamp.Add(30 * time.Minute)
+	for _, stale := range []*time.Time{nil, &other} {
+		exec.FirstDispatchAt = stale
+		exec.Status = "running"
+		require.NoError(t, store.SavePurchaseExecution(ctx, exec))
+		got, err = store.GetExecutionByID(ctx, exec.ExecutionID)
+		require.NoError(t, err)
+		assert.Equal(t, "running", got.Status, "the upsert still updates status")
+		require.NotNil(t, got.FirstDispatchAt, "upsert must not clear the stamp")
+		assert.True(t, stamp.Equal(*got.FirstDispatchAt), "upsert must not move the stamp")
+	}
+
+	// An unstamped row stays unstamped when a later upsert carries a value.
+	plain := &PurchaseExecution{ExecutionID: uuid.New().String(), IdempotencyKey: uuid.New().String(),
+		Status: "pending", StepNumber: 1, ScheduledDate: time.Now()}
+	require.NoError(t, store.SavePurchaseExecution(ctx, plain))
+	plain.FirstDispatchAt = &other
+	require.NoError(t, store.SavePurchaseExecution(ctx, plain))
+	got, err = store.GetExecutionByID(ctx, plain.ExecutionID)
+	require.NoError(t, err)
+	assert.Nil(t, got.FirstDispatchAt, "upsert must not set the stamp")
+}