fix(azure): invalidate stale recommendation prices before purchase

internal/api/handler_purchases.go

@@ -364,11 +364,7 @@ func (h *Handler) runPlannedPurchase(ctx context.Context, req *events.LambdaFunc
 	}, nil
 }
 
-// executeFailureError maps a failure of the shared approve-and-execute funnel
-// (run-now, session approve, direct execute) to its HTTP error. action names
-// what the caller tried ("started", "approved", "direct-executed"). A non-nil
-// final means the purchase ran and money may have moved, so it is never a 409.
-// ErrNotFound and ErrExecutionNotInExpectedStatus are 409 only before a claim.
+// Stale Azure prices are refused before purchase, even after a successful claim.
 func executeFailureError(executionID, action string, final *config.PurchaseExecution, err error) error {
 	switch {
 	case errors.Is(err, purchase.ErrFourEyesDenied):
@@ -379,6 +375,8 @@ func executeFailureError(executionID, action string, final *config.PurchaseExecu
 		return fmt.Errorf("execution %s could not be %s: %w", executionID, action, err)
 	case errors.Is(err, config.ErrAuditLoss):
 		return NewClientError(500, fmt.Sprintf("execution %s ran but its final status could not be saved: %v", executionID, err))
+	case errors.Is(err, purchase.ErrStaleAzurePricing):
+		return NewClientError(409, err.Error())
 	default:
 		return NewClientErrorWithDetails(502, fmt.Sprintf("execution %s failed: %v", executionID, err),
 			map[string]any{"execution_id": executionID, "status": final.Status})
@@ -687,6 +685,10 @@ func (h *Handler) requireSessionBeforeLookup(ctx context.Context, req *events.La
 // whose status no longer allows the action is 409. Anything else stays a server error.
 func tokenActionError(err error) error {
 	switch {
+	case errors.Is(err, config.ErrAuditLoss):
+		return err
+	case errors.Is(err, purchase.ErrStaleAzurePricing):
+		return NewClientError(409, err.Error())
 	case errors.Is(err, purchase.ErrInvalidApprovalToken):
 		return NewClientError(403, "invalid approval token")
 	case errors.Is(err, purchase.ErrApprovalTokenExpired):

internal/api/handler_purchases_retry_redrive_test.go

@@ -124,6 +124,12 @@ func executeRedriveSuccessor(t *testing.T, successor *config.PurchaseExecution)
 	running.Status = "running"
 
 	store := new(MockConfigStore)
+	for _, rec := range approved.Recommendations {
+		if rec.Provider == "azure" {
+			store.On("ListStoredRecommendations", mock.Anything, config.RecommendationFilter{Provider: "azure", RequireAzurePricingMigration: true}).Return(approved.Recommendations, nil).Once()
+			break
+		}
+	}
 	t.Cleanup(func() { store.AssertExpectations(t) })
 
 	store.On("GetExecutionByID", mock.Anything, approved.ExecutionID).Return(&approved, nil).Once()
@@ -287,6 +293,7 @@ func TestRetryOfUnknownProviderRowIsRefused(t *testing.T) {
 // first attempt had in fact landed.
 func TestRetryOfFailedAzureReservationStillPurchasesOnce(t *testing.T) {
 	failed := redriveFailedRow(redriveAzureRIExecID, "azure", "compute")
+	failed.Recommendations[0].ID = "current-azure"
 
 	tokens, err := purchasesFiredByRetry(t, failed, sessionRetryReq())
 

internal/api/purchase_pricing_test.go

@@ -4,11 +4,13 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/auth"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/purchase"
 	"github.com/aws/aws-lambda-go/events"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -1148,3 +1150,55 @@ func TestRouter_PurchasesExecute_EmptyIDMismatchedTenancyRefused(t *testing.T) {
 	assert.Equal(t, 409, ce.code)
 	mockStore.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything)
 }
+
+func TestAzureCommitmentUnitsRegression(t *testing.T) {
+	upfront := config.RecommendationRecord{Provider: "azure", Term: 3, Count: 1, UpfrontCost: 3600}
+	monthly := config.RecommendationRecord{Provider: "azure", Term: 3, Count: 1, MonthlyCost: float64Ptr(100)}
+	assert.Equal(t, recTotalCommitment(&upfront), recTotalCommitment(&monthly))
+	assert.Equal(t, 3600.0, recTotalCommitment(&monthly))
+	assert.Equal(t, 3600.0, buildRecommendationsResponse([]config.RecommendationRecord{upfront, monthly}).Summary.TotalUpfrontCost)
+	stored := upfront
+	stored.MonthlyCost = float64Ptr(10)
+	stored.OnDemandCost = float64Ptr(150)
+	stored.Savings = 50
+	req := stored
+	req.Count = 2
+	priced, err := priceFromStored(&req, &stored, 0)
+	require.NoError(t, err)
+	assert.Equal(t, 7200.0, priced.UpfrontCost)
+	assert.Equal(t, 20.0, *priced.MonthlyCost)
+	assert.Equal(t, 300.0, *priced.OnDemandCost)
+	assert.Equal(t, 100.0, priced.Savings)
+}
+
+func TestAzureCorrectedPriceExceedsOldUnitsCap(t *testing.T) {
+	ctx := context.Background()
+	store := new(MockConfigStore)
+	authService := new(MockAuthService)
+	authService.On("ValidateSession", ctx, "azure-cap").Return(&Session{UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", Email: "capped@example.com"}, nil)
+	authService.grantPermissions([]auth.Permission{{Action: auth.ActionExecute, Resource: auth.ResourcePurchases, Constraints: &auth.PermissionConstraints{MaxPurchaseAmount: 1000}}})
+	store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil).Maybe()
+	store.On("GetPendingExecutions", mock.Anything).Return([]config.PurchaseExecution{}, nil).Maybe()
+	expectStoredRecs(store, config.RecommendationRecord{ID: "azure-corrected", Provider: "azure", Service: "compute", Region: "westeurope", ResourceType: "Standard_D2s_v3", Count: 1, Term: 3, Payment: "upfront", UpfrontCost: 3600, Savings: 50})
+	h := &Handler{config: store, auth: authService}
+	_, err := h.executePurchase(ctx, &events.LambdaFunctionURLRequest{Headers: map[string]string{"Authorization": "Bearer azure-cap"}, Body: `{"recommendations":[{"id":"azure-corrected","provider":"azure","service":"compute","region":"westeurope","resource_type":"Standard_D2s_v3","count":1,"term":3,"payment":"all-upfront","upfront_cost":70,"savings":50}]}`})
+	ce, ok := IsClientError(err)
+	require.True(t, ok)
+	assert.Equal(t, 403, ce.code)
+	assert.Contains(t, err.Error(), "exceeds the constraints configured on your execute permission for purchases")
+	store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything)
+}
+
+func TestStaleAzurePricingErrorMapping(t *testing.T) {
+	stale := fmt.Errorf("%w: %w", purchase.ErrStaleAzurePricing, config.ErrAzurePricingNotReady)
+	audit := fmt.Errorf("%w: %w", config.ErrAuditLoss, stale)
+	for _, mapped := range []error{executeFailureError("id", "approved", &config.PurchaseExecution{Status: "failed"}, stale), tokenActionError(stale)} {
+		ce, ok := IsClientError(mapped)
+		require.True(t, ok)
+		assert.Equal(t, 409, ce.code)
+	}
+	ce, ok := IsClientError(executeFailureError("id", "approved", &config.PurchaseExecution{Status: "failed"}, audit))
+	require.True(t, ok)
+	assert.Equal(t, 500, ce.code)
+	assert.ErrorIs(t, tokenActionError(audit), config.ErrAuditLoss)
+}

internal/config/errors.go

@@ -2,6 +2,8 @@ package config
 
 import "errors"
 
+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")
 

internal/config/store_postgres_azure_pricing_test.go

@@ -0,0 +1,67 @@
+package config
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/pashagolub/pgxmock/v4"
+	"github.com/stretchr/testify/require"
+)
+
+func TestListStoredRecommendationsAzurePricingReadiness(t *testing.T) {
+	queryErr := errors.New("migration query failed")
+	for _, tc := range []struct {
+		name                  string
+		version               int
+		dirty, missing, ready bool
+		err                   error
+	}{
+		{name: "old", version: 105},
+		{name: "dirty", version: 106, dirty: true},
+		{name: "missing", missing: true},
+		{name: "query error", err: queryErr},
+		{name: "ready", version: 106, ready: true},
+		{name: "later", version: 107, ready: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			pool := newMock(t)
+			defer pool.Close()
+			query := pool.ExpectQuery("SELECT version, dirty FROM schema_migrations")
+			if tc.err != nil {
+				query.WillReturnError(tc.err)
+			} else {
+				rows := pgxmock.NewRows([]string{"version", "dirty"})
+				if !tc.missing {
+					rows.AddRow(tc.version, tc.dirty)
+				}
+				query.WillReturnRows(rows)
+			}
+			if tc.ready {
+				pool.ExpectQuery("SELECT payload FROM recommendations WHERE provider = \\$1").WithArgs("azure").WillReturnRows(pgxmock.NewRows([]string{"payload"}))
+			}
+			_, err := storeWith(pool).ListStoredRecommendations(context.Background(), RecommendationFilter{Provider: "azure", RequireAzurePricingMigration: true})
+			switch {
+			case tc.ready:
+				require.NoError(t, err)
+			case tc.err != nil:
+				require.ErrorIs(t, err, queryErr)
+				require.NotErrorIs(t, err, ErrAzurePricingNotReady)
+			default:
+				require.ErrorIs(t, err, ErrAzurePricingNotReady)
+			}
+			require.NoError(t, pool.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestListStoredRecommendationsDisplayDoesNotRequireAzureMigration(t *testing.T) {
+	t.Parallel()
+	pool := newMock(t)
+	defer pool.Close()
+	pool.ExpectQuery("SELECT payload FROM recommendations WHERE provider = \\$1").WithArgs("azure").WillReturnRows(pgxmock.NewRows([]string{"payload"}))
+	_, err := storeWith(pool).ListStoredRecommendations(context.Background(), RecommendationFilter{Provider: "azure"})
+	require.NoError(t, err)
+	require.NoError(t, pool.ExpectationsWereMet())
+}

internal/config/store_postgres_recommendations.go

@@ -3,6 +3,7 @@ package config
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math"
 	"strings"
@@ -19,6 +20,8 @@ import (
 // gives 4500 params which stays conservative and leaves headroom).
 const recommendationsBatchSize = 500
 
+const azurePricingMigrationVersion = 106
+
 // ReplaceRecommendations wipes the recommendations table and reinserts the
 // full snapshot inside a single transaction. Used for a force-full-resync
 // path; the steady-state write path (see commit 6) is UpsertRecommendations.
@@ -364,12 +367,34 @@ func recOnDemandBaseline(rec *RecommendationRecord) (float64, bool) {
 	return *rec.MonthlyCost + rec.Savings + amortized, true
 }
 
+func (s *PostgresStore) checkAzurePricingMigration(ctx context.Context, required bool) error {
+	if !required {
+		return nil
+	}
+	var version int
+	var dirty bool
+	err := s.db.QueryRow(ctx, `SELECT version, dirty FROM schema_migrations`).Scan(&version, &dirty)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return ErrAzurePricingNotReady
+	}
+	if err != nil {
+		return fmt.Errorf("check Azure pricing migration: %w", err)
+	}
+	if version < azurePricingMigrationVersion || dirty {
+		return ErrAzurePricingNotReady
+	}
+	return nil
+}
+
 // ListStoredRecommendations reads recommendations matching the filter.
 // SQL-pushed conditions (Provider, Service, Region, AccountIDs,
 // MinSavingsUSD) are applied in SQL so Postgres prunes the rows; the
 // MinSavingsPct filter is applied in-process because the on-demand
 // baseline lives inside the JSONB payload (not a native column).
 func (s *PostgresStore) ListStoredRecommendations(ctx context.Context, filter RecommendationFilter) ([]RecommendationRecord, error) {
+	if err := s.checkAzurePricingMigration(ctx, filter.RequireAzurePricingMigration); err != nil {
+		return nil, err
+	}
 	whereClause, args := buildRecommendationFilter(filter)
 	rows, err := s.db.Query(ctx, `SELECT payload FROM recommendations`+whereClause, args...)
 	if err != nil {

internal/config/store_postgres_recommendations_test.go

@@ -509,3 +509,23 @@ func TestPostgresStore_SetRecommendationsCollectionError_DoesNotClearOwnerMarker
 }
 
 func float64Ptr(f float64) *float64 { return &f }
+
+func TestPostgresStore_UpsertRecommendations_ReplacesAzureWrongUnits(t *testing.T) {
+	ctx := context.Background()
+	store, cleanup := setupRecommendationsStore(ctx, t)
+	defer cleanup()
+	rec := config.RecommendationRecord{ID: "old-units", Provider: "azure", Service: "compute", Region: "westeurope", ResourceType: "Standard_D2s_v3", Count: 1, Term: 3, Payment: "all-upfront", UpfrontCost: 70, MonthlyCost: float64Ptr(10), Savings: 50}
+	collects := []config.SuccessfulCollect{{Provider: "azure"}}
+	now := time.Now().UTC()
+	require.NoError(t, store.UpsertRecommendations(ctx, now, []config.RecommendationRecord{rec}, collects))
+	rec.ID = "corrected"
+	rec.UpfrontCost = 3600
+	rec.MonthlyCost = nil
+	require.NoError(t, store.UpsertRecommendations(ctx, now.Add(time.Minute), []config.RecommendationRecord{rec}, collects))
+	got, err := store.ListStoredRecommendations(ctx, config.RecommendationFilter{Provider: "azure"})
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "corrected", got[0].ID)
+	assert.Equal(t, 3600.0, got[0].UpfrontCost)
+	assert.Nil(t, got[0].MonthlyCost)
+}

internal/config/types.go

@@ -659,13 +659,14 @@ type PurchaseSuppression struct {
 // query rather than in SQL (avoids a computed column). These two filters are
 // independent and can be combined.
 type RecommendationFilter struct {
-	Provider      string   // "aws" / "azure" / "gcp" / "" (all)
-	Service       string   // "" = all services
-	Region        string   // "" = all regions
-	AccountIDs    []string // nil/empty = all accounts
-	MinSavingsUSD float64  // 0 = no floor on monthly savings dollar amount
-	MinSavingsPct float64  // 0 = no floor on savings percentage (0–100 scale)
-	ID            string   // "" = all ids; non-empty = exact match on the id column
+	RequireAzurePricingMigration bool
+	Provider                     string   // "aws" / "azure" / "gcp" / "" (all)
+	Service                      string   // "" = all services
+	Region                       string   // "" = all regions
+	AccountIDs                   []string // nil/empty = all accounts
+	MinSavingsUSD                float64  // 0 = no floor on monthly savings dollar amount
+	MinSavingsPct                float64  // 0 = no floor on savings percentage (0–100 scale)
+	ID                           string   // "" = all ids; non-empty = exact match on the id column
 }
 
 // PurchasePlanFilter parameterises ListPurchasePlans. Zero-value means "no

internal/database/postgres/migrations/000106_invalidate_azure_recs_lookback_units.down.sql

@@ -0,0 +1 @@
+-- Wrong-unit prices cannot be restored; a successful collection supplies corrected values.

internal/database/postgres/migrations/000106_invalidate_azure_recs_lookback_units.up.sql

@@ -0,0 +1,6 @@
+-- Quiesce old Azure collectors and deploy the fixed collector with this migration.
+-- Rows collected before migration, even with the fix, need a successful recollection.
+UPDATE recommendations
+   SET payload = jsonb_set(jsonb_set(payload, '{upfront_cost}', 'null'::jsonb),
+                           '{monthly_cost}', 'null'::jsonb)
+ WHERE payload->>'provider' = 'azure';

internal/database/postgres/migrations/000106_invalidate_azure_recs_lookback_units_test.go

@@ -0,0 +1,66 @@
+//go:build integration
+
+package migrations_test
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"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"
+)
+
+func TestMigration_InvalidateAzureRecsLookbackUnits(t *testing.T) {
+	ctx := context.Background()
+	path := getMigrationsPath()
+	container, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	defer container.Cleanup(ctx)
+	pool := container.DB.Pool()
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 105))
+	fixtures := []string{
+		`{"provider":"azure","payment":"upfront","id":"azure-upfront","term":3,"count":1,"upfront_cost":38.5,"monthly_cost":12.1,"savings":40,"resource_type":"Standard_D4s_v5"}`,
+		`{"provider":"azure","payment":"monthly","id":"azure-monthly","term":3,"count":1,"upfront_cost":4.2,"monthly_cost":55,"savings":40,"resource_type":"Standard_D4s_v5"}`,
+		`{"provider":"aws","payment":"upfront","id":"aws-upfront","term":3,"count":1,"upfront_cost":500,"monthly_cost":0,"savings":25,"resource_type":"Standard_D4s_v5","on_demand_cost":90}`,
+	}
+	expected := make([]map[string]any, len(fixtures))
+	for i, payload := range fixtures {
+		require.NoError(t, json.Unmarshal([]byte(payload), &expected[i]))
+		_, err = pool.Exec(ctx, `INSERT INTO recommendations
+   (collected_at,provider,service,region,resource_type,engine,payload,upfront_cost,monthly_savings,term,payment_option)
+   VALUES (NOW(),$1,'compute','westeurope','Standard_D4s_v5','',$2::jsonb,0,0,3,$3)`,
+			expected[i]["provider"], payload, expected[i]["payment"])
+		require.NoError(t, err)
+		if expected[i]["provider"] == "azure" {
+			expected[i]["upfront_cost"] = nil
+			expected[i]["monthly_cost"] = nil
+		}
+	}
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 106))
+	sql, err := os.ReadFile(filepath.Join(path, "000106_invalidate_azure_recs_lookback_units.up.sql"))
+	require.NoError(t, err)
+	for pass := 0; pass < 3; pass++ {
+		if pass == 1 {
+			_, err = pool.Exec(ctx, string(sql))
+			require.NoError(t, err)
+		}
+		if pass == 2 {
+			require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 105))
+		}
+		for _, payload := range expected {
+			var got string
+			require.NoError(t, pool.QueryRow(ctx, `SELECT payload::text FROM recommendations WHERE provider=$1 AND payment_option=$2`, payload["provider"], payload["payment"]).Scan(&got))
+			want, err := json.Marshal(payload)
+			require.NoError(t, err)
+			assert.JSONEq(t, string(want), got)
+		}
+		var count int
+		require.NoError(t, pool.QueryRow(ctx, `SELECT COUNT(*) FROM recommendations`).Scan(&count))
+		assert.Equal(t, 3, count)
+	}
+}

internal/purchase/armed_redrive_test.go

@@ -202,7 +202,9 @@ func TestFireScheduledDelayedPurchasesRefusesArmedAzureSavingsPlanRetry(t *testi
 // above.
 func TestFirstAzureSavingsPlanPurchaseStillExecutes(t *testing.T) {
 	exec := armedExecution("azure", "savingsplans", 0, "approved")
+	exec.Recommendations[0].ID = "current-azure"
 	mgr, store, rec := armedHarness(t, common.ServiceSavingsPlansAll)
+	store.On("ListStoredRecommendations", mock.Anything, config.RecommendationFilter{Provider: "azure", RequireAzurePricingMigration: true}).Return(exec.Recommendations, nil).Once()
 	store.On("GetExecutionByID", mock.Anything, exec.ExecutionID).Return(exec, nil).Maybe()
 	expectClaim(store, exec, []string{"approved", "pending", "notified"}, "running")