fix(purchase): save purchase history once per commitment

internal/config/errors.go

@@ -7,6 +7,11 @@ var ErrAzurePricingNotReady = errors.New("azure pricing migration not ready")
 // ErrNotFound is returned when a requested config-store row does not exist.
 var ErrNotFound = errors.New("not found")
 
+// ErrEmptyPurchaseID is returned by SavePurchaseHistory for a record with no
+// purchase_id. History is saved once per (provider, account_id, purchase_id),
+// so an empty ID would be silently dropped after the first one (#774).
+var ErrEmptyPurchaseID = errors.New("purchase history record has no purchase id")
+
 // ErrPurchasePlanConflict means a plan changed after the caller read it.
 var ErrPurchasePlanConflict = errors.New("purchase plan changed")
 

internal/config/store_postgres.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
+	"github.com/LeanerCloud/cloud-commitments-go/pkg/logging"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/database"
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
@@ -2067,18 +2068,32 @@ func (s *PostgresStore) CleanupOldExecutions(ctx context.Context, retentionDays
 // PURCHASE HISTORY
 // ==========================================
 
-// SavePurchaseHistory saves a purchase history record.
+// SavePurchaseHistory saves a purchase history record. It is idempotent per
+// (provider, account_id, purchase_id): a re-drive or root retry that adopts a
+// commitment the first attempt already bought saves the same key again, and a
+// second row would double-count spend and savings (#704). The NOT EXISTS
+// covers rows that predate the unique index from migration 000108; the index
+// and ON CONFLICT cover two concurrent saves. An existing row is left as is.
 func (s *PostgresStore) SavePurchaseHistory(ctx context.Context, record *PurchaseHistoryRecord) error {
+	if err := record.Validate(); err != nil {
+		return err
+	}
 	query := `
 		INSERT INTO purchase_history (
 			account_id, purchase_id, timestamp, provider, service, region,
 			resource_type, count, term, payment, upfront_cost, monthly_cost,
 			estimated_savings, plan_id, plan_name, ramp_step, cloud_account_id,
 			source, revocation_window_closes_at, offering_class
-		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20)
+		)
+		SELECT $1::varchar, $2::varchar, $3, $4::varchar, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20
+		WHERE NOT EXISTS (
+			SELECT 1 FROM purchase_history
+			WHERE provider = $4 AND account_id = $1 AND purchase_id = $2
+		)
+		ON CONFLICT DO NOTHING
 	`
 
-	_, err := s.db.Exec(ctx, query,
+	tag, err := s.db.Exec(ctx, query,
 		record.AccountID,
 		record.PurchaseID,
 		record.Timestamp,
@@ -2104,6 +2119,9 @@ func (s *PostgresStore) SavePurchaseHistory(ctx context.Context, record *Purchas
 	if err != nil {
 		return fmt.Errorf("failed to save purchase history: %w", err)
 	}
+	if tag.RowsAffected() == 0 {
+		logging.Infof("purchase history for %s/%s/%s already recorded; keeping the existing row", record.Provider, record.AccountID, record.PurchaseID)
+	}
 
 	return nil
 }

internal/config/store_postgres_history_dedupe_integration_test.go

@@ -0,0 +1,158 @@
+//go:build integration
+
+package config
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func historyFixture(accountID, purchaseID string) *PurchaseHistoryRecord {
+	cost := 100.0
+	return &PurchaseHistoryRecord{
+		AccountID: accountID, PurchaseID: purchaseID, Timestamp: time.Now().Truncate(time.Microsecond),
+		Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.large",
+		Count: 1, Term: 1, Payment: "all-upfront", UpfrontCost: &cost,
+	}
+}
+
+// TestSavePurchaseHistory_RedriveDoesNotDuplicate is MON-06 (#704): a re-drive
+// or root retry that adopts an existing commitment saves the same
+// (provider, account_id, purchase_id) again. Two rows would double-count spend
+// and savings everywhere purchase_history is summed.
+func TestSavePurchaseHistory_RedriveDoesNotDuplicate(t *testing.T) {
+	conn := setupTestContainerDB(t)
+	cleanupTestData(t, conn)
+	store := NewPostgresStore(conn)
+	ctx := context.Background()
+
+	count := func(purchaseID string) int {
+		var n int
+		require.NoError(t, conn.Pool().QueryRow(ctx, `SELECT count(*) FROM purchase_history WHERE purchase_id = $1`, purchaseID).Scan(&n))
+		return n
+	}
+
+	require.NoError(t, store.SavePurchaseHistory(ctx, historyFixture("123456789012", "ri-0001")))
+	require.NoError(t, store.SavePurchaseHistory(ctx, historyFixture("123456789012", "ri-0001")), "a repeated save is a no-op, not an error")
+	assert.Equal(t, 1, count("ri-0001"), "re-drive must not add a second history row")
+
+	t.Run("same purchase id in another account is a different purchase", func(t *testing.T) {
+		require.NoError(t, store.SavePurchaseHistory(ctx, historyFixture("999999999999", "ri-0001")))
+		assert.Equal(t, 2, count("ri-0001"))
+	})
+
+	t.Run("same purchase id under another provider is a different purchase", func(t *testing.T) {
+		other := historyFixture("123456789012", "ri-0001")
+		other.Provider = "azure"
+		require.NoError(t, store.SavePurchaseHistory(ctx, other))
+		assert.Equal(t, 3, count("ri-0001"))
+	})
+
+	t.Run("a row that predates the dedupe index still blocks a repeat", func(t *testing.T) {
+		_, err := conn.Pool().Exec(ctx, `
+			INSERT INTO purchase_history (account_id, purchase_id, timestamp, provider, service, region, resource_type, term, payment, created_at)
+			VALUES ('123456789012', 'ri-old', now(), 'aws', 'ec2', 'us-east-1', 'm5.large', 1, 'all-upfront', now() - interval '30 days')`)
+		require.NoError(t, err)
+		require.NoError(t, store.SavePurchaseHistory(ctx, historyFixture("123456789012", "ri-old")))
+		assert.Equal(t, 1, count("ri-old"))
+	})
+}
+
+// TestSavePurchaseHistory_CreatedAtIgnoresPurchaseTimestamp: the unique index
+// only covers rows whose created_at is after the migration. created_at must
+// therefore come from the column default (now()) and never from the record's
+// Timestamp, which for a re-drive is the original purchase time.
+func TestSavePurchaseHistory_CreatedAtIgnoresPurchaseTimestamp(t *testing.T) {
+	conn := setupTestContainerDB(t)
+	cleanupTestData(t, conn)
+	store := NewPostgresStore(conn)
+	ctx := context.Background()
+
+	rec := historyFixture("123456789012", "ri-backdated")
+	rec.Timestamp = time.Now().AddDate(-1, 0, 0)
+	require.NoError(t, store.SavePurchaseHistory(ctx, rec))
+
+	var recent bool
+	require.NoError(t, conn.Pool().QueryRow(ctx,
+		`SELECT created_at > now() - interval '1 minute' FROM purchase_history WHERE purchase_id = 'ri-backdated'`).Scan(&recent))
+	assert.True(t, recent, "created_at must be the insert time, not the purchase timestamp")
+}
+
+// TestSavePurchaseHistory_ConcurrentSaveWaitsForUncommittedTwin forces the
+// interleaving the NOT EXISTS check cannot see: another transaction has
+// inserted the same key but not committed. The save must wait on the unique
+// index and, once the twin commits, do nothing.
+func TestSavePurchaseHistory_ConcurrentSaveWaitsForUncommittedTwin(t *testing.T) {
+	conn := setupTestContainerDB(t)
+	cleanupTestData(t, conn)
+	store := NewPostgresStore(conn)
+	ctx := context.Background()
+
+	tx, err := conn.Pool().Begin(ctx)
+	require.NoError(t, err)
+	_, err = tx.Exec(ctx, `
+		INSERT INTO purchase_history (account_id, purchase_id, timestamp, provider, service, region, resource_type, term, payment)
+		VALUES ('123456789012', 'ri-race', now(), 'aws', 'ec2', 'us-east-1', 'm5.large', 1, 'all-upfront')`)
+	require.NoError(t, err)
+
+	done := make(chan error, 1)
+	go func() { done <- store.SavePurchaseHistory(ctx, historyFixture("123456789012", "ri-race")) }()
+
+	select {
+	case err := <-done:
+		_ = tx.Rollback(ctx)
+		t.Fatalf("save returned (%v) while a twin was uncommitted; it must wait on the unique index", err)
+	case <-time.After(500 * time.Millisecond):
+	}
+	require.NoError(t, tx.Commit(ctx))
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("save did not finish after the twin committed")
+	}
+
+	var n int
+	require.NoError(t, conn.Pool().QueryRow(ctx, `SELECT count(*) FROM purchase_history WHERE purchase_id = 'ri-race'`).Scan(&n))
+	assert.Equal(t, 1, n)
+}
+
+// TestMigration108_AppliesOverExistingDuplicates: duplicates that predate the
+// migration must not make it fail (no rows are deleted), and the index it
+// creates must still reject a duplicate among new rows.
+func TestMigration108_AppliesOverExistingDuplicates(t *testing.T) {
+	conn := setupTestContainerDB(t)
+	cleanupTestData(t, conn)
+	ctx := context.Background()
+	pool := conn.Pool()
+
+	_, err := pool.Exec(ctx, `DROP INDEX uq_purchase_history_purchase`)
+	require.NoError(t, err)
+	for range 2 {
+		_, err = pool.Exec(ctx, `
+			INSERT INTO purchase_history (account_id, purchase_id, timestamp, provider, service, region, resource_type, term, payment, created_at)
+			VALUES ('123456789012', 'ri-dup', now(), 'aws', 'ec2', 'us-east-1', 'm5.large', 1, 'all-upfront', now() - interval '1 day')`)
+		require.NoError(t, err)
+	}
+	up, err := os.ReadFile(filepath.Join(getTestMigrationsPath(), "000108_purchase_history_unique_purchase.up.sql"))
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, string(up))
+	require.NoError(t, err, "pre-existing duplicates must not block the migration")
+
+	var n int
+	require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM purchase_history WHERE purchase_id = 'ri-dup'`).Scan(&n))
+	assert.Equal(t, 2, n, "the migration deletes nothing")
+
+	insertNew := `INSERT INTO purchase_history (account_id, purchase_id, timestamp, provider, service, region, resource_type, term, payment)
+		VALUES ('123456789012', 'ri-new', now(), 'aws', 'ec2', 'us-east-1', 'm5.large', 1, 'all-upfront')`
+	_, err = pool.Exec(ctx, insertNew)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, insertNew)
+	require.Error(t, err, "the index rejects a duplicate among rows created after the migration")
+}

internal/config/store_postgres_history_validate_test.go

@@ -0,0 +1,21 @@
+package config
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// An empty purchase id would be dropped silently after the first one (#774),
+// so SavePurchaseHistory must refuse it before touching the database. The
+// store has no connection here: reaching the insert would panic.
+func TestSavePurchaseHistory_RejectsEmptyPurchaseID(t *testing.T) {
+	store := &PostgresStore{}
+	for _, id := range []string{"", "   "} {
+		err := store.SavePurchaseHistory(context.Background(), &PurchaseHistoryRecord{AccountID: "123456789012", PurchaseID: id, Provider: "aws"})
+		require.ErrorIs(t, err, ErrEmptyPurchaseID, "id %q", id)
+	}
+	assert.NoError(t, (&PurchaseHistoryRecord{PurchaseID: "ri-1"}).Validate())
+}

internal/config/types.go

@@ -931,6 +931,15 @@ type PurchaseHistoryRecord struct {
 	ListingState string `json:"listing_state,omitempty" dynamodbav:"listing_state,omitempty"`
 }
 
+// Validate rejects a record SavePurchaseHistory cannot store without losing
+// it: one with no purchase id (see ErrEmptyPurchaseID).
+func (r *PurchaseHistoryRecord) Validate() error {
+	if strings.TrimSpace(r.PurchaseID) == "" {
+		return ErrEmptyPurchaseID
+	}
+	return nil
+}
+
 // MarketplaceListingClaim is what ClaimMarketplaceListingSlot reads back from
 // the row it just claimed, so the handler works from the row as it is at claim
 // time instead of a copy read before the claim (issue #525).

internal/database/postgres/migrations/000108_purchase_history_unique_purchase.down.sql

@@ -0,0 +1 @@
+DROP INDEX IF EXISTS uq_purchase_history_purchase;

internal/database/postgres/migrations/000108_purchase_history_unique_purchase.up.sql

@@ -0,0 +1,22 @@
+-- Migration 000108: one purchase_history row per (provider, account_id, purchase_id)
+-- from now on (#704, MON-06).
+--
+-- A re-drive or root retry that adopts an existing commitment used to insert a
+-- second row for it, double-counting spend and savings. SavePurchaseHistory
+-- now skips a key that already exists; this index makes that hold under
+-- concurrent saves.
+--
+-- Rows that exist today may already contain duplicates, and removing them
+-- deletes financial history, which is an owner decision made separately. So
+-- the index is partial: it covers only rows created after this migration runs
+-- (the cutoff is evaluated once, here, and baked into the index predicate).
+-- Existing rows, duplicates included, are untouched. SavePurchaseHistory's
+-- NOT EXISTS check still stops a new save from duplicating an older row.
+DO $$
+BEGIN
+    EXECUTE format(
+        'CREATE UNIQUE INDEX IF NOT EXISTS uq_purchase_history_purchase '
+        'ON purchase_history (provider, account_id, purchase_id) '
+        'WHERE created_at > %L::timestamptz',
+        now());
+END $$;

internal/purchase/execution_test.go

@@ -1821,6 +1821,66 @@ func TestManager_ExecuteAndFinalize_HistorySaveFailure_StaysVisible(t *testing.T
 	mockFactory.AssertExpectations(t)
 }
 
+// historyGuardStore applies the real purchase_history validation in front of
+// the mock, so a test sees the error the Postgres store would return.
+type historyGuardStore struct{ *MockConfigStore }
+
+func (s historyGuardStore) SavePurchaseHistory(ctx context.Context, record *config.PurchaseHistoryRecord) error {
+	if err := record.Validate(); err != nil {
+		return err
+	}
+	return s.MockConfigStore.SavePurchaseHistory(ctx, record)
+}
+
+// TestManager_ExecuteAndFinalize_EmptyCommitmentID_CompletesWithAuditGap
+// (#774): a successful purchase whose result carries no CommitmentID cannot be
+// stored in purchase_history. The execution must still complete, with the
+// audit-gap note, never fail and invite a re-approval of a purchase that fired.
+func TestManager_ExecuteAndFinalize_EmptyCommitmentID_CompletesWithAuditGap(t *testing.T) {
+	ctx := context.Background()
+	mockStore := new(MockConfigStore)
+	mockEmail := new(MockEmailSender)
+	mockSTS := new(MockSTSClient)
+	mockFactory := new(MockProviderFactory)
+	mockProviderInst := new(MockProvider)
+	mockServiceClient := new(MockServiceClient)
+
+	exec := &config.PurchaseExecution{
+		ExecutionID: "exec-empty-id",
+		StepNumber:  1,
+		Recommendations: []config.RecommendationRecord{
+			{Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 500.0, Selected: true},
+		},
+	}
+	// SavePurchaseHistory is deliberately not expected on the mock: the guard
+	// must reject the record before it reaches the store.
+	mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil)
+	mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
+	mockSTS.On("GetCallerIdentity", ctx, mock.AnythingOfType("*sts.GetCallerIdentityInput")).Return(&sts.GetCallerIdentityOutput{
+		Account: aws.String("123456789012"),
+	}, nil)
+	mockFactory.On("CreateAndValidateProvider", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), "aws", mock.Anything).Return(mockProviderInst, nil)
+	mockProviderInst.On("GetServiceClient", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), common.ServiceEC2, "us-east-1").Return(mockServiceClient, nil)
+	mockServiceClient.On("PurchaseCommitment", mock.MatchedBy(hasPerRecDeadline(30*time.Second)), mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")).Return(common.PurchaseResult{
+		Success:      true,
+		CommitmentID: "",
+	}, nil)
+
+	manager := &Manager{
+		config:          historyGuardStore{mockStore},
+		email:           mockEmail,
+		stsClient:       mockSTS,
+		providerFactory: mockFactory,
+		dashboardURL:    "https://dashboard.example.com",
+	}
+
+	require.NoError(t, manager.executeAndFinalize(ctx, exec))
+
+	assert.Equal(t, "completed", exec.Status, "the purchase fired, so the execution stays completed")
+	assert.Contains(t, exec.Error, "history record failed to save", "the missing history row must be flagged on the execution")
+	mockStore.AssertNotCalled(t, "SavePurchaseHistory", mock.Anything, mock.Anything)
+}
+
 // TestManager_ExecuteAndFinalize_SingleAccount_PartialSuccess is the issue #642
 // regression guard for the single-account path. A two-rec direct purchase where
 // rec A succeeds and rec B fails must: