fix(recommendations): key stored recs, rec ID and purchase identity on RDS az_config (#408)

internal/api/az_config_gate.go

@@ -0,0 +1,45 @@
+package api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+)
+
+// azConfigSchemaChecker is implemented by the Postgres store. Stores without a
+// schema (test doubles) have nothing to check.
+type azConfigSchemaChecker interface {
+	CheckAZConfigMigration(ctx context.Context) error
+}
+
+// azConfigNotReadyError maps ErrAZConfigMigrationNotReady to the 503 every
+// az_config-dependent path returns, and passes any other error through.
+func azConfigNotReadyError(err error) error {
+	if !errors.Is(err, config.ErrAZConfigMigrationNotReady) {
+		return err
+	}
+	return NewClientErrorWithDetails(503,
+		fmt.Sprintf("database migrations %06d and %06d are not applied; recommendations and purchases are unavailable until they are",
+			config.AZConfigRecommendationsMigration, config.AZConfigSuppressionsMigration),
+		map[string]any{})
+}
+
+// requireAZConfigSchema refuses a purchase submission before its transaction
+// opens when the az_config expand migrations are missing: the suppression
+// insert names az_config, so it would fail mid-transaction.
+func (h *Handler) requireAZConfigSchema(ctx context.Context) error {
+	checker, ok := h.config.(azConfigSchemaChecker)
+	if !ok {
+		return nil
+	}
+	err := checker.CheckAZConfigMigration(ctx)
+	if err == nil {
+		return nil
+	}
+	if errors.Is(err, config.ErrAZConfigMigrationNotReady) {
+		return azConfigNotReadyError(err)
+	}
+	return fmt.Errorf("check az_config migration: %w", err)
+}

internal/api/az_config_gate_test.go

@@ -0,0 +1,47 @@
+package api
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/aws/aws-lambda-go/events"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+)
+
+// A recommendations list below the az_config expand migrations answers the same
+// 503 as the purchase paths, not a generic 500.
+func TestGetRecommendations_MapsMissingAZConfigMigrationTo503(t *testing.T) {
+	ctx := context.Background()
+
+	mockAuth := new(MockAuthService)
+	t.Cleanup(func() { mockAuth.AssertExpectations(t) })
+	mockAuth.On("HasAPIKeyPermissionAPI", ctx, "cudly-user-key", "view", "recommendations").
+		Return("user-123", "key-abc", true, nil)
+	// Account scoping resolves against the key's owning user.
+	mockAuth.On("GetAllowedAccountsAPI", ctx, "user-123").
+		Return([]string(nil), nil)
+	// The unrestricted read probes whether it may kick a collection.
+	mockAuth.On("HasAPIKeyPermissionForConstraintsAPI", ctx, "key-abc", "user-123", "view", "recommendations", mock.Anything).
+		Return(true, nil)
+
+	mockScheduler := new(MockScheduler)
+	t.Cleanup(func() { mockScheduler.AssertExpectations(t) })
+	mockScheduler.On("ListRecommendations", mock.Anything, mock.Anything).
+		Return([]config.RecommendationRecord(nil), fmt.Errorf("suppression listing skipped: %w", config.ErrAZConfigMigrationNotReady))
+
+	handler := &Handler{auth: mockAuth, scheduler: mockScheduler}
+	req := &events.LambdaFunctionURLRequest{
+		Headers: map[string]string{"x-api-key": "cudly-user-key"},
+	}
+
+	_, err := handler.getRecommendations(ctx, req, map[string]string{})
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "got %v", err)
+	assert.Equal(t, 503, ce.code)
+	assert.Contains(t, ce.message, "000158 and 000159")
+}

internal/api/azconfig_expand_order_integration_test.go

@@ -0,0 +1,113 @@
+//go:build integration
+
+package api
+
+import (
+	"context"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+)
+
+// contractMigrationVersion is the version of the migration that drops the
+// 11-column recommendations key (#408). It can only run after the release that
+// upserts on the 12-column key, so every expand step the new code needs must
+// sort BEFORE it.
+func contractMigrationVersion(t *testing.T) uint {
+	t.Helper()
+	paths, err := filepath.Glob(filepath.Join(getMigrationsPath(), "*_drop_recommendations_11col_natural_key.up.sql"))
+	require.NoError(t, err)
+	require.Len(t, paths, 1)
+	n, err := strconv.ParseUint(strings.SplitN(filepath.Base(paths[0]), "_", 2)[0], 10, 64)
+	require.NoError(t, err)
+	return uint(n)
+}
+
+// previousVersion returns the highest migration version below version, so the
+// test stops at the last migration a deploy applies before the contract step.
+func previousVersion(t *testing.T, version uint) uint {
+	t.Helper()
+	paths, err := filepath.Glob(filepath.Join(getMigrationsPath(), "*.up.sql"))
+	require.NoError(t, err)
+	var best uint
+	for _, p := range paths {
+		n, err := strconv.ParseUint(strings.SplitN(filepath.Base(p), "_", 2)[0], 10, 64)
+		require.NoError(t, err)
+		if uint(n) < version && uint(n) > best {
+			best = uint(n)
+		}
+	}
+	require.NotZero(t, best)
+	return best
+}
+
+func expandOrderRDSRec(azConfig string, upfront float64) config.RecommendationRecord {
+	rec := rdsAZConfigRec(azConfig, upfront)
+	rec.ID = ""
+	return rec
+}
+
+// exerciseNewCode runs the four paths the #408 code touches against whatever
+// schema the database currently has.
+// Before the contract step the 11-column index is still in place, so a
+// Single-AZ and Multi-AZ pair of one class cannot coexist yet (the documented
+// limitation the contract step resolves, and the same failure the previous
+// release had); only after it may both variants be upserted in one batch.
+func exerciseNewCode(ctx context.Context, t *testing.T, store *config.PostgresStore, recs []config.RecommendationRecord, priced string, wantUpfront float64) {
+	t.Helper()
+	single := expandOrderRDSRec("single-az", 100)
+
+	require.NoError(t, store.UpsertRecommendations(ctx, time.Now(), recs, []config.SuccessfulCollect{{Provider: "aws"}}),
+		"upsert of RDS variants")
+
+	purchase := single
+	purchase.Count = 2
+	sups := config.BuildSuppressions([]config.RecommendationRecord{purchase}, uuid.NewString(), &config.GlobalConfig{}, time.Now())
+	require.Len(t, sups, 1)
+	require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { return store.CreateSuppressionTx(ctx, tx, &sups[0]) }),
+		"suppression write inside the purchase transaction")
+
+	active, err := store.ListActiveSuppressions(ctx)
+	require.NoError(t, err, "ListActiveSuppressions")
+	assert.NotEmpty(t, active)
+
+	h := &Handler{config: store}
+	req := []config.RecommendationRecord{{
+		Provider: "aws", Service: "rds", Region: "us-east-1", ResourceType: "db.r5.large", Engine: "mysql",
+		Term: 1, Payment: "all-upfront", Count: 1, Details: rdsDetails(priced),
+	}}
+	require.NoError(t, h.priceRecommendationsFromStore(ctx, req), "purchase pricing index lookup")
+	assert.Equal(t, wantUpfront, req[0].UpfrontCost)
+}
+
+// Issue #408 deploy order: the new code ships after the expand migrations and
+// before the contract one. A database stopped just below the contract step
+// must already serve every new-code path, and still do so once the contract
+// step has run.
+func TestAZConfigNewCodeWorksBeforeAndAfterContractMigration(t *testing.T) {
+	ctx := context.Background()
+	pg := testhelpers.RequireEmptyDatabase(ctx, t)
+	pool, path := pg.DB.Pool(), getMigrationsPath()
+	contract := contractMigrationVersion(t)
+	store := config.NewPostgresStore(pg.DB)
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, previousVersion(t, contract)))
+	single, multi := expandOrderRDSRec("single-az", 100), expandOrderRDSRec("multi-az", 300)
+	exerciseNewCode(ctx, t, store, []config.RecommendationRecord{single}, "single-az", 100)
+
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, contract))
+	_, err := pool.Exec(ctx, `TRUNCATE recommendations, purchase_suppressions`)
+	require.NoError(t, err)
+	exerciseNewCode(ctx, t, store, []config.RecommendationRecord{single, multi}, "multi-az", 300)
+}

internal/api/azconfig_gate_integration_test.go

@@ -0,0 +1,75 @@
+//go:build integration
+
+package api
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+)
+
+// txCountingStore counts the transactions the purchase path opens.
+type txCountingStore struct {
+	*config.PostgresStore
+	txs int
+}
+
+func (s *txCountingStore) WithTx(ctx context.Context, fn func(pgx.Tx) error) error {
+	s.txs++
+	return s.PostgresStore.WithTx(ctx, fn)
+}
+
+// Without the az_config expand migrations the new code must refuse loudly:
+// the purchase path before it opens a transaction, collection and suppression
+// listing with the typed error, not a SQL failure.
+func TestAZConfigGate_RefusesBeforeExpandMigrations(t *testing.T) {
+	ctx := context.Background()
+	pg := testhelpers.RequireEmptyDatabase(ctx, t)
+	pool, path := pg.DB.Pool(), getMigrationsPath()
+	store := &txCountingStore{PostgresStore: config.NewPostgresStore(pg.DB)}
+	h := &Handler{config: store}
+	exec := &config.PurchaseExecution{ExecutionID: uuid.NewString()}
+
+	for _, version := range []uint{157, 158} { // before both expands, then after only the first
+		require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version))
+
+		_, err := h.persistExecutionAndSuppressions(ctx, exec, nil, "creator", "key")
+		ce, ok := IsClientError(err)
+		require.True(t, ok, "version %d: %v", version, err)
+		assert.Equal(t, 503, ce.code)
+		assert.Contains(t, ce.message, "000158 and 000159")
+		assert.Zero(t, store.txs, "no transaction may open before the gate")
+
+		rec := expandOrderRDSRec("single-az", 100)
+		err = store.UpsertRecommendations(ctx, time.Now(), []config.RecommendationRecord{rec}, []config.SuccessfulCollect{{Provider: "aws"}})
+		assert.True(t, errors.Is(err, config.ErrAZConfigMigrationNotReady), "collection: %v", err)
+
+		_, err = store.ListActiveSuppressions(ctx)
+		assert.True(t, errors.Is(err, config.ErrAZConfigMigrationNotReady), "suppression listing: %v", err)
+
+		sup := &config.PurchaseSuppression{ExecutionID: uuid.NewString(), Provider: "aws", Service: "rds", Region: "us-east-1",
+			ResourceType: "db.r5.large", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)}
+		err = store.CreateSuppression(ctx, sup)
+		assert.True(t, errors.Is(err, config.ErrAZConfigMigrationNotReady), "CreateSuppression: %v", err)
+		err = store.WithTx(ctx, func(tx pgx.Tx) error { return store.CreateSuppressionTx(ctx, tx, sup) })
+		assert.True(t, errors.Is(err, config.ErrAZConfigMigrationNotReady), "CreateSuppressionTx: %v", err)
+		store.txs = 0 // the direct WithTx above is not the purchase path
+	}
+
+	for _, version := range []uint{159, 160} {
+		require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, version))
+		require.NoError(t, h.requireAZConfigSchema(ctx), "version %d", version)
+		_, err := store.ListActiveSuppressions(ctx)
+		require.NoError(t, err, "version %d", version)
+	}
+}

internal/api/handler_purchases.go

@@ -2821,6 +2821,9 @@ func (h *Handler) persistRetryExecution(ctx context.Context, failedExec *config.
 // The successor is a web-source pending row for the retrying user, so it
 // also goes through the submit dedupe (issue #760): see writeRetrySuccessorTx.
 func (h *Handler) persistRetrySuccessorTx(ctx context.Context, failedExec, newExecution *config.PurchaseExecution, suppressions []config.PurchaseSuppression) error {
+	if err := h.requireAZConfigSchema(ctx); err != nil {
+		return err
+	}
 	err := h.config.WithTx(ctx, func(tx pgx.Tx) error {
 		return h.writeRetrySuccessorTx(ctx, tx, failedExec, newExecution, suppressions)
 	})
@@ -3666,6 +3669,9 @@ func (h *Handler) persistExecutionAndSuppressions(
 	suppressions []config.PurchaseSuppression,
 	creatorID, idempotencyKey string,
 ) (dup *config.PurchaseExecution, err error) {
+	if err := h.requireAZConfigSchema(ctx); err != nil {
+		return nil, err
+	}
 	txErr := h.config.WithTx(ctx, func(tx pgx.Tx) error {
 		// Duplicate check inside the tx, after taking the creator's submit
 		// lock: a concurrent identical submit waits for this tx to commit,
@@ -3724,11 +3730,17 @@ func purchaseIdempotencyKey(creatorID string, recs []config.RecommendationRecord
 		if err != nil {
 			return "", err
 		}
-		tuples = append(tuples, fmt.Sprintf("%s|%s|%s|%s|%s|%s|%d|%d|%s|%s|%s|%s",
+		// Single-AZ and Multi-AZ RDS buys of the same class are different
+		// purchases and must not dedupe into one (issue #408).
+		azConfig, err := r.IdentityAZConfig()
+		if err != nil {
+			return "", err
+		}
+		tuples = append(tuples, fmt.Sprintf("%s|%s|%s|%s|%s|%s|%d|%d|%s|%s|%s|%s|%s",
 			strings.ToLower(strings.TrimSpace(r.Provider)),
 			r.Service, r.Region, r.ResourceType, r.Engine, acct,
 			r.Count, r.Term, strings.ToLower(strings.TrimSpace(r.Payment)), az,
-			ci.Platform, ci.Tenancy))
+			ci.Platform, ci.Tenancy, azConfig))
 	}
 	sort.Strings(tuples)
 	h := sha256.New()

internal/api/handler_recommendations.go

@@ -85,6 +85,11 @@ func parseRecommendationFilter(params map[string]string) (config.RecommendationF
 
 // Recommendations handlers.
 func (h *Handler) getRecommendations(ctx context.Context, req *events.LambdaFunctionURLRequest, params map[string]string) (*RecommendationsResponse, error) {
+	resp, err := h.listRecommendations(ctx, req, params)
+	return resp, azConfigNotReadyError(err)
+}
+
+func (h *Handler) listRecommendations(ctx context.Context, req *events.LambdaFunctionURLRequest, params map[string]string) (*RecommendationsResponse, error) {
 	// Require view:recommendations permission
 	session, err := h.requirePermission(ctx, req, "view", "recommendations")
 	if err != nil {