feat(purchase): record Redshift purchase attempts in a durable journal

internal/config/store_postgres_attempts.go

@@ -0,0 +1,86 @@
+package config
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/jackc/pgx/v5"
+)
+
+// PostgresStore is the platform's common.AttemptJournal (cloud-commitments-go#107).
+var _ common.AttemptJournal = (*PostgresStore)(nil)
+
+// recordAttemptSQL is one autocommit upsert. The WHERE clause is the
+// transition table and mirrors validTransition in cloud-commitments-go's
+// providers/aws/services/redshift journal test file (a cross-language mirror;
+// change both together):
+//
+//	none        -> sending | purchased      (the INSERT path; the SELECT's
+//	                                         WHERE refuses none -> outcome_unknown)
+//	sending     -> purchased | outcome_unknown
+//	outcome_unknown -> purchased
+//	purchased   -> purchased, same reservation_id only
+//
+// Anything else (including identical sending or outcome_unknown repeats)
+// updates no row and RecordAttempt reports an error. offering_id is
+// write-once: the first non-NULL value survives every later write.
+const recordAttemptSQL = `
+INSERT INTO purchase_attempts (idempotency_token, execution_id, state, reservation_id, offering_id)
+SELECT $1::text, $2::text, $3::text, NULLIF($4::text, ''), NULLIF($5::text, '')
+WHERE $3::text <> 'outcome_unknown'
+   OR EXISTS (SELECT 1 FROM purchase_attempts p WHERE p.idempotency_token = $1::text AND p.execution_id = $2::text)
+ON CONFLICT (idempotency_token, execution_id) DO UPDATE SET
+    state = EXCLUDED.state,
+    reservation_id = EXCLUDED.reservation_id,
+    offering_id = COALESCE(purchase_attempts.offering_id, EXCLUDED.offering_id),
+    updated_at = now()
+WHERE (purchase_attempts.state = 'sending' AND EXCLUDED.state IN ('purchased', 'outcome_unknown'))
+   OR (purchase_attempts.state = 'outcome_unknown' AND EXCLUDED.state = 'purchased')
+   OR (purchase_attempts.state = 'purchased' AND EXCLUDED.state = 'purchased'
+       AND purchase_attempts.reservation_id = EXCLUDED.reservation_id)`
+
+// RecordAttempt durably records one purchase attempt. It validates the record,
+// then upserts in its own autocommit statement. It must not be called inside a
+// caller's transaction: the record has to be committed before it returns.
+func (s *PostgresStore) RecordAttempt(ctx context.Context, rec common.AttemptRecord) error {
+	if err := rec.Validate(); err != nil {
+		return err
+	}
+	tag, err := s.db.Exec(ctx, recordAttemptSQL,
+		rec.IdempotencyToken, rec.ExecutionID, string(rec.State), rec.ReservationID, rec.OfferingID)
+	if err != nil {
+		return fmt.Errorf("record purchase attempt %s: %w", common.MaskToken(rec.IdempotencyToken), err)
+	}
+	if tag.RowsAffected() == 0 {
+		return fmt.Errorf("record purchase attempt %s: transition to %q refused by the existing row",
+			common.MaskToken(rec.IdempotencyToken), rec.State)
+	}
+	return nil
+}
+
+// PriorAttempts returns every attempt of the token across executions. An empty
+// result is not an error. Like RecordAttempt it runs on its own connection.
+func (s *PostgresStore) PriorAttempts(ctx context.Context, token string) ([]common.AttemptRecord, error) {
+	rows, err := s.db.Query(ctx, `
+		SELECT idempotency_token, execution_id, state, COALESCE(reservation_id, ''), COALESCE(offering_id, '')
+		FROM purchase_attempts
+		WHERE idempotency_token = $1
+		ORDER BY created_at, execution_id`, token)
+	if err != nil {
+		return nil, fmt.Errorf("list purchase attempts %s: %w", common.MaskToken(token), err)
+	}
+	recs, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (common.AttemptRecord, error) {
+		var r common.AttemptRecord
+		var state string
+		if scanErr := row.Scan(&r.IdempotencyToken, &r.ExecutionID, &state, &r.ReservationID, &r.OfferingID); scanErr != nil {
+			return r, scanErr
+		}
+		r.State = common.AttemptState(state)
+		return r, nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("scan purchase attempts %s: %w", common.MaskToken(token), err)
+	}
+	return recs, nil
+}

internal/config/store_postgres_attempts_integration_test.go

@@ -0,0 +1,146 @@
+//go:build integration
+// +build integration
+
+package config
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const attemptToken = "tok-0123456789abcdef"
+
+func attemptRec(exec string, state common.AttemptState, id, offering string) common.AttemptRecord {
+	return common.AttemptRecord{IdempotencyToken: attemptToken, ExecutionID: exec, State: state, ReservationID: id, OfferingID: offering}
+}
+
+// TestRecordAttempt_TransitionTable pins the 16 cases (4 sources x 4 targets)
+// of go's validTransition: sources none, sending, outcome_unknown, purchased
+// (rn-1); targets sending, outcome_unknown, purchased rn-1, purchased rn-2.
+func TestRecordAttempt_TransitionTable(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+
+	sources := map[string][]common.AttemptRecord{
+		"none":            nil,
+		"sending":         {attemptRec("", common.AttemptSending, "", "")},
+		"outcome_unknown": {attemptRec("", common.AttemptSending, "", ""), attemptRec("", common.AttemptOutcomeUnknown, "", "")},
+		"purchased":       {attemptRec("", common.AttemptPurchased, "rn-1", "")},
+	}
+	targets := []struct {
+		name string
+		rec  func(exec string) common.AttemptRecord
+		// allowed is keyed by source name.
+		allowed map[string]bool
+	}{
+		{"sending", func(e string) common.AttemptRecord { return attemptRec(e, common.AttemptSending, "", "") },
+			map[string]bool{"none": true}},
+		{"outcome_unknown", func(e string) common.AttemptRecord { return attemptRec(e, common.AttemptOutcomeUnknown, "", "") },
+			map[string]bool{"sending": true}},
+		{"purchased_same_id", func(e string) common.AttemptRecord { return attemptRec(e, common.AttemptPurchased, "rn-1", "") },
+			map[string]bool{"none": true, "sending": true, "outcome_unknown": true, "purchased": true}},
+		{"purchased_other_id", func(e string) common.AttemptRecord { return attemptRec(e, common.AttemptPurchased, "rn-2", "") },
+			map[string]bool{"none": true, "sending": true, "outcome_unknown": true}},
+	}
+
+	n := 0
+	for srcName, setup := range sources {
+		for _, tgt := range targets {
+			n++
+			exec := fmt.Sprintf("exec-%d", n)
+			t.Run(srcName+"_to_"+tgt.name, func(t *testing.T) {
+				for _, s := range setup {
+					s.ExecutionID = exec
+					require.NoError(t, store.RecordAttempt(ctx, s))
+				}
+				err := store.RecordAttempt(ctx, tgt.rec(exec))
+				if tgt.allowed[srcName] {
+					require.NoError(t, err)
+				} else {
+					require.Error(t, err)
+				}
+			})
+		}
+	}
+	assert.Equal(t, 16, n)
+}
+
+// execCounter counts statements sent to the database, so a test can tell a
+// record rejected by Validate from one rejected by a table CHECK.
+type execCounter struct {
+	dbConn
+	execs int
+}
+
+func (c *execCounter) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
+	c.execs++
+	return c.dbConn.Exec(ctx, sql, args...)
+}
+
+func TestRecordAttempt_InvalidRecordRejectedBeforeUpsert(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+	counter := &execCounter{dbConn: store.db}
+	store.db = counter
+
+	require.Error(t, store.RecordAttempt(ctx, attemptRec("exec-1", common.AttemptPurchased, "", "")), "purchased without an ID")
+	require.Error(t, store.RecordAttempt(ctx, attemptRec("exec-1", common.AttemptSending, "rn-1", "")), "ID on a non-purchased record")
+	require.Error(t, store.RecordAttempt(ctx, attemptRec("exec-1", common.AttemptState("bogus"), "", "")), "unknown state")
+	require.Error(t, store.RecordAttempt(ctx, attemptRec("", common.AttemptSending, "", "")), "empty execution")
+
+	assert.Zero(t, counter.execs, "no invalid record reached the database")
+	got, err := store.PriorAttempts(ctx, attemptToken)
+	require.NoError(t, err)
+	assert.Empty(t, got, "no invalid record reached the table")
+}
+
+func TestPriorAttempts_AcrossExecutions(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+
+	got, err := store.PriorAttempts(ctx, attemptToken)
+	require.NoError(t, err)
+	assert.Empty(t, got, "no rows is not an error")
+
+	require.NoError(t, store.RecordAttempt(ctx, attemptRec("exec-a", common.AttemptSending, "", "off-1")))
+	require.NoError(t, store.RecordAttempt(ctx, attemptRec("exec-b", common.AttemptPurchased, "rn-9", "off-1")))
+	other := attemptRec("exec-c", common.AttemptSending, "", "")
+	other.IdempotencyToken = "other-token-0123456"
+	require.NoError(t, store.RecordAttempt(ctx, other))
+
+	got, err = store.PriorAttempts(ctx, attemptToken)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []common.AttemptRecord{
+		attemptRec("exec-a", common.AttemptSending, "", "off-1"),
+		attemptRec("exec-b", common.AttemptPurchased, "rn-9", "off-1"),
+	}, got)
+}
+
+func TestRecordAttempt_OfferingIDWriteOnce(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+
+	require.NoError(t, store.RecordAttempt(ctx, attemptRec("exec-1", common.AttemptSending, "", "off-first")))
+	require.NoError(t, store.RecordAttempt(ctx, attemptRec("exec-1", common.AttemptPurchased, "rn-1", "off-second")))
+	got, err := store.PriorAttempts(ctx, attemptToken)
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "off-first", got[0].OfferingID, "first offering survives a later write")
+
+	// A row first written without an offering takes the first one offered.
+	require.NoError(t, store.RecordAttempt(ctx, attemptRec("exec-2", common.AttemptSending, "", "")))
+	require.NoError(t, store.RecordAttempt(ctx, attemptRec("exec-2", common.AttemptOutcomeUnknown, "", "off-late")))
+	got, err = store.PriorAttempts(ctx, attemptToken)
+	require.NoError(t, err)
+	for _, r := range got {
+		if r.ExecutionID == "exec-2" {
+			assert.Equal(t, "off-late", r.OfferingID)
+		}
+	}
+}

internal/database/postgres/migrations/000161_purchase_attempts.down.sql

@@ -0,0 +1,4 @@
+-- Reverse 000161: drop the purchase attempt journal (cloud-commitments-go#107).
+-- Lossy by design: recorded attempts are dropped, so a re-dispatch falls back
+-- to the provider's own window dedupe.
+DROP TABLE IF EXISTS purchase_attempts;

internal/database/postgres/migrations/000161_purchase_attempts.up.sql

@@ -0,0 +1,30 @@
+-- Migration 000161: durable purchase attempt journal (cloud-commitments-go#107).
+--
+-- One row per (idempotency token, execution): the Redshift provider records
+-- "sending" immediately before the AWS call and "purchased" or
+-- "outcome_unknown" after it, so a later attempt of the same token can adopt a
+-- recorded reservation instead of buying twice. The store
+-- (RecordAttempt in internal/config/store_postgres_attempts.go) enforces the
+-- transition table in its upsert; this table only pins the value set.
+--
+--   state           MUST match common.AttemptSending / AttemptPurchased /
+--                   AttemptOutcomeUnknown in cloud-commitments-go
+--                   pkg/common/attempt_journal.go byte for byte.
+--   reservation_id  the provider commitment ID; NULL unless state = 'purchased'.
+--   offering_id     the offering the attempt was sent under; NULL when unknown,
+--                   written once (the upsert keeps the first non-NULL value).
+CREATE TABLE IF NOT EXISTS purchase_attempts (
+    idempotency_token TEXT NOT NULL CHECK (idempotency_token <> ''),
+    execution_id TEXT NOT NULL CHECK (execution_id <> ''),
+    state TEXT NOT NULL CHECK (state IN ('sending', 'purchased', 'outcome_unknown')),
+    reservation_id TEXT NULL,
+    offering_id TEXT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (idempotency_token, execution_id),
+    CONSTRAINT purchase_attempts_reservation_iff_purchased
+        CHECK ((state = 'purchased') = (reservation_id IS NOT NULL))
+);
+
+CREATE INDEX IF NOT EXISTS idx_purchase_attempts_token
+    ON purchase_attempts (idempotency_token);

internal/database/postgres/migrations/000161_purchase_attempts_test.go

@@ -0,0 +1,50 @@
+//go:build integration
+
+package migrations_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Migration 000161 creates purchase_attempts; the down migration drops it.
+func TestMigration_0161_PurchaseAttempts(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	migrationsPath := getMigrationsPath()
+
+	container := testhelpers.RequireEmptyDatabase(ctx, t)
+	pool := container.DB.Pool()
+
+	tableExists := func() bool {
+		var exists bool
+		require.NoError(t, pool.QueryRow(ctx, `SELECT EXISTS (
+			SELECT 1 FROM information_schema.tables WHERE table_name = 'purchase_attempts')`).Scan(&exists))
+		return exists
+	}
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 161)))
+	assert.False(t, tableExists(), "table absent before 000161")
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 161))
+	assert.True(t, tableExists(), "table present after 000161")
+
+	_, err := pool.Exec(ctx, `INSERT INTO purchase_attempts (idempotency_token, execution_id, state) VALUES ('t', 'e', 'bogus')`)
+	require.Error(t, err, "state CHECK rejects unknown states")
+	_, err = pool.Exec(ctx, `INSERT INTO purchase_attempts (idempotency_token, execution_id, state, reservation_id) VALUES ('t', 'e', 'sending', 'rn-1')`)
+	require.Error(t, err, "reservation_id is only valid when purchased")
+	_, err = pool.Exec(ctx, `INSERT INTO purchase_attempts (idempotency_token, execution_id, state) VALUES ('t', 'e', 'purchased')`)
+	require.Error(t, err, "purchased requires a reservation_id")
+	_, err = pool.Exec(ctx, `INSERT INTO purchase_attempts (idempotency_token, execution_id, state) VALUES ('t', 'e', 'sending')`)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, `INSERT INTO purchase_attempts (idempotency_token, execution_id, state) VALUES ('t', 'e', 'sending')`)
+	require.Error(t, err, "primary key (token, execution)")
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 161)))
+	assert.False(t, tableExists(), "the down migration drops the table")
+}

internal/email/purchase_outcome_unknown.go

@@ -13,8 +13,8 @@ const purchaseOutcomeUnknownSubject = "CUDly - Purchase Outcome Unknown - Verify
 const purchaseOutcomeUnknownTemplate = `CUDly - Purchase Outcome Unknown
 ================================
 
-The purchase run for execution {{.ExecutionID}}{{if .PlanID}} (plan {{.PlanID}}){{end}} was interrupted
-before it recorded a result, and a recovery sweep marked it failed. CUDly
+The purchase run for execution {{.ExecutionID}}{{if .PlanID}} (plan {{.PlanID}}){{end}} {{if .UnrecordedCommitment}}ended because the provider returned a commitment that CUDly could not record.{{else}}was interrupted
+before it recorded a result, and a recovery sweep marked it failed.{{end}} CUDly
 cannot tell whether the commitments below were bought: the cloud provider
 may already have created them.
 
@@ -24,7 +24,8 @@ Buying it again without checking can create a duplicate commitment.
 Outcome unknown:
 {{range .Recommendations}}
 - {{.Count}}x {{.ResourceType}}{{if .Engine}} ({{.Engine}}){{end}} in {{.Region}}
-  Service: {{.Service}}
+  Service: {{.Service}}{{if .UnrecordedCommitmentID}}
+  {{if eq .Service "redshift"}}Redshift reserved node{{else}}Commitment{{end}} {{.UnrecordedCommitmentID}} was bought but CUDly could not record it; verify it in the cloud provider's console before buying again.{{end}}
 {{end}}
 
 Execution details:
@@ -37,8 +38,15 @@ This is an automated message from CUDly.
 // purchase-failed subject, or the outcome-unknown subject when data is for a
 // row whose purchase may already exist.
 func purchaseFailedSubject(data NotificationData, failedSubject string) string {
-	if data.OutcomeUnknown {
+	if data.outcomeUnknownVariant() {
 		return purchaseOutcomeUnknownSubject
 	}
 	return failedSubject
 }
+
+// outcomeUnknownVariant reports whether the purchase-failed email is the
+// outcome-unknown one. An unrecorded commitment implies it, so a caller that
+// later assigns OutcomeUnknown cannot switch the email back to "retry".
+func (d NotificationData) outcomeUnknownVariant() bool {
+	return d.OutcomeUnknown || d.UnrecordedCommitment
+}