chore(purchase): remove the dead server-side SQS purchase path

CHANGELOG.md

@@ -72,6 +72,11 @@ are the full record until the next release.
 
 ### Removed
 
+- The async SQS purchase-execution path (issue #399). Nothing produced SQS
+  messages, so the Lambda no longer accepts an `aws:sqs` event (it now fails
+  with "unrecognized Lambda event shape"), and the
+  `SQS_REPORT_BATCH_ITEM_FAILURES` environment variable is no longer read.
+
 - The CloudFormation install stack (`cloudformation/stacks/CUDly`). It still
   created DynamoDB tables and set no `DB_HOST`, so the Postgres-only server it
   deployed failed at startup and the stack could not serve any request or

cmd/lambda/main_test.go

@@ -124,16 +124,16 @@ func TestHandler_ScheduledEvent(t *testing.T) {
 	assert.NotNil(t, result)
 }
 
-func TestHandler_SQSEvent(t *testing.T) {
+func TestHandler_SQSEventIsRejected(t *testing.T) {
 	origApp := app
 	defer func() { app = origApp }()
 
 	app = createTestApp()
 
 	rawEvent := json.RawMessage(`{"Records":[{"eventSource":"aws:sqs","messageId":"msg-1","body":"{}"}]}`)
-	result, err := Handler(context.Background(), rawEvent)
-	require.NoError(t, err)
-	assert.NotNil(t, result)
+	_, err := Handler(context.Background(), rawEvent)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "unrecognized")
 }
 
 func TestHandler_HTTPEvent(t *testing.T) {

internal/api/handler_history.go

@@ -515,7 +515,7 @@ func annotateInFlightOrAuditGapRow(row *config.PurchaseHistoryRecord, exec confi
 //     exact session-authed click attribution.
 //  2. approver — the notification inbox that received the cancel token;
 //     authoritative accountable party but not necessarily the clicker.
-//     Used on legacy token-only paths (async workers, old email clicks).
+//     Used on legacy token-only paths (old email clicks).
 func annotateCancelled(row *config.PurchaseHistoryRecord, exec config.PurchaseExecution, approver string) {
 	if exec.CancelledBy != nil && *exec.CancelledBy != "" {
 		row.Approver = *exec.CancelledBy

internal/api/handler_purchases_pause_claim_integration_test.go

@@ -4,7 +4,6 @@ package api
 
 import (
 	"context"
-	"encoding/json"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -63,20 +62,21 @@ type pauseClaimFixture struct {
 	manager   *purchase.Manager
 	execution *config.PurchaseExecution
 	request   *events.LambdaFunctionURLRequest
-	body      string
 	workers   sync.WaitGroup
 }
 
 // pauseClaimCap is a plan cap far above the fixture recommendations.
 var pauseClaimCap = 1e9
 
 // markNotified moves the row to "notified", the only state in which the
-// scheduler and SQS paths will auto-execute a recommendation-bearing plan row
-// (the notification email is the veto window). pause only applies to "pending",
-// so tests call this after pausing and resuming and before the worker runs.
+// scheduler sweep will auto-execute a recommendation-bearing plan row
+// (the notification email is the veto window) and due. Resume pushes a past
+// scheduled date forward, which the sweep honors, so this also makes the row
+// due again. pause only applies to "pending", so tests call this after pausing
+// and resuming and before the worker runs.
 func (f *pauseClaimFixture) markNotified(t *testing.T) {
 	t.Helper()
-	_, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET status = 'notified' WHERE execution_id = $1`, f.execution.ExecutionID)
+	_, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET status = 'notified', scheduled_date = now() - interval '1 minute' WHERE execution_id = $1`, f.execution.ExecutionID)
 	require.NoError(t, err)
 }
 
@@ -133,20 +133,29 @@ func newPauseClaimFixture(t *testing.T) *pauseClaimFixture {
 	require.Equal(t, f.execution.CloudAccountID, row.CloudAccountID)
 	require.NotEqual(t, common.PurchaseSourceWeb, row.Source)
 	f.request = &events.LambdaFunctionURLRequest{Headers: map[string]string{"Authorization": "Bearer admin-token"}}
-	body, err := json.Marshal(purchase.AsyncMessage{Type: purchase.MessageTypeExecutePurchase, ExecutionID: row.ExecutionID})
-	require.NoError(t, err)
-	f.body = string(body)
 	t.Cleanup(func() { f.service.unblock(); cancel(); f.workers.Wait() })
 	return f
 }
 
 func (f *pauseClaimFixture) startWorker() <-chan error {
 	done := make(chan error, 1)
 	f.workers.Add(1)
-	go func() { defer f.workers.Done(); done <- f.manager.ProcessMessage(f.ctx, f.body) }()
+	go func() {
+		defer f.workers.Done()
+		_, err := f.manager.ProcessScheduledPurchases(f.ctx)
+		done <- err
+	}()
 	return done
 }
 
+// sweep runs one scheduler tick, the production worker for a due plan row.
+func (f *pauseClaimFixture) sweep(t *testing.T) *purchase.ProcessResult {
+	t.Helper()
+	result, err := f.manager.ProcessScheduledPurchases(f.ctx)
+	require.NoError(t, err)
+	return result
+}
+
 func (f *pauseClaimFixture) waitWorker(t *testing.T, done <-chan error) {
 	t.Helper()
 	select {
@@ -219,11 +228,11 @@ func TestPauseClaim_RunningWorkerCannotBeReclaimed(t *testing.T) {
 	before := f.state(t)
 	require.Equal(t, "running", before.Status)
 	require.Nil(t, before.Actor)
-	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
+	f.sweep(t)
 	require.EqualValues(t, 1, f.service.calls.Load())
 	pauseErr, resumeErr := f.pause(), f.resume()
 	// Reach the purchase counter even when the old handler incorrectly accepts both actions.
-	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
+	f.sweep(t)
 	assert.EqualValues(t, 1, f.service.calls.Load(), "pause/resume must not allow a second provider entry")
 	assertPauseConflict(t, pauseErr)
 	assertPauseConflict(t, resumeErr)
@@ -240,19 +249,19 @@ func TestPauseClaim_PausedQueueResumesExactlyOnce(t *testing.T) {
 	require.Equal(t, "paused", paused.Status)
 	require.Equal(t, f.execution.CreatedByUserID, paused.Actor)
 	require.NotNil(t, paused.TransitionedAt)
-	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
+	f.sweep(t)
 	require.Zero(t, f.service.calls.Load())
 	require.Equal(t, paused, f.state(t))
 	require.NoError(t, f.resume())
 	require.Equal(t, "pending", f.state(t).Status)
-	// A resumed row is pending again, so auto-execution still refuses it until
-	// a notification has gone out.
-	require.Error(t, f.manager.ProcessMessage(f.ctx, f.body))
+	// A resumed row is pending again, so the sweep still skips it until a
+	// notification has gone out.
+	require.Zero(t, f.sweep(t).Executed)
 	require.Zero(t, f.service.calls.Load())
 	f.markNotified(t)
 	f.service.unblock()
-	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
-	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
+	f.sweep(t)
+	f.sweep(t)
 	f.assertCompleted(t)
 }
 
@@ -277,7 +286,12 @@ func TestPauseClaim_ConcurrentPauseAndWorker(t *testing.T) {
 	worker := make(chan error, 1)
 	f.workers.Add(2)
 	go func() { defer f.workers.Done(); <-start; paused <- f.pause() }()
-	go func() { defer f.workers.Done(); <-start; worker <- f.manager.ProcessMessage(f.ctx, f.body) }()
+	go func() {
+		defer f.workers.Done()
+		<-start
+		_, err := f.manager.ProcessScheduledPurchases(f.ctx)
+		worker <- err
+	}()
 	close(start)
 	select {
 	case err := <-paused:
@@ -295,6 +309,6 @@ func TestPauseClaim_ConcurrentPauseAndWorker(t *testing.T) {
 	require.NoError(t, f.resume())
 	f.markNotified(t)
 	f.service.unblock()
-	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
+	f.sweep(t)
 	f.assertCompleted(t)
 }

internal/api/handler_purchases_retry_fanout_test.go

@@ -2,7 +2,6 @@ package api
 
 import (
 	"context"
-	"encoding/json"
 	"sort"
 	"sync"
 	"testing"
@@ -167,27 +166,27 @@ func (f *fanoutProviderFactory) CreateAndValidateProvider(_ context.Context, _ s
 }
 
 // executeSuccessorForReal runs the retry successor through the REAL
-// purchase.Manager the way production does: the approved row is picked up from
-// the execute_purchase queue message, claimed, and executed. It returns the
+// purchase.Manager the way production does: the pending row is approved by
+// an operator, claimed, and executed. It returns the
 // sorted idempotency tokens of every commitment purchase that reached the
 // cloud — one entry per real purchase.
 func executeSuccessorForReal(t *testing.T, successor *config.PurchaseExecution) []string {
 	t.Helper()
 	ctx := context.Background()
 
-	// The row as the DB holds it once the operator has clicked the approval
-	// link: same fields the retry handler persisted, status "approved".
-	approved := *successor
-	approved.Status = "approved"
-	running := approved
+	// The row as the retry handler persisted it (status pending), awaiting the
+	// operator's approval.
+	pending := *successor
+	running := pending
 	running.Status = "running"
 
 	store := new(MockConfigStore)
 	t.Cleanup(func() { store.AssertExpectations(t) })
 
-	store.On("GetExecutionByID", mock.Anything, approved.ExecutionID).Return(&approved, nil).Once()
-	store.On("TransitionExecutionStatus", mock.Anything, approved.ExecutionID,
-		[]string{"approved", "pending", "notified"}, "running", (*string)(nil)).Return(&running, nil).Once()
+	store.On("GetExecutionByID", mock.Anything, pending.ExecutionID).Return(&pending, nil).Maybe()
+	store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil).Maybe()
+	store.On("TransitionExecutionStatus", mock.Anything, pending.ExecutionID,
+		[]string{"pending", "notified"}, "running", (*string)(nil)).Return(&running, nil).Once()
 	store.On("SavePurchaseHistory", mock.Anything, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil)
 	// Plan progress advances only on a fully clean run; .Maybe() so a run that
 	// ends partial/failed does not turn into a mock-expectation failure that
@@ -223,12 +222,9 @@ func executeSuccessorForReal(t *testing.T, successor *config.PurchaseExecution)
 		DashboardURL:    "https://dashboard.example.com",
 	})
 
-	body, err := json.Marshal(purchase.AsyncMessage{
-		Type:        purchase.MessageTypeExecutePurchase,
-		ExecutionID: approved.ExecutionID,
-	})
-	require.NoError(t, err)
-	require.NoError(t, mgr.ProcessMessage(ctx, string(body)),
+	// Human approval of the persisted pending successor, as the retry flow does.
+	_, _, err := mgr.ApproveAndExecute(ctx, pending.ExecutionID, "", nil, nil)
+	require.NoError(t, err,
 		"the successor must execute cleanly; a non-nil error would make the purchase tally meaningless")
 
 	return svc.purchasedTokens()

internal/api/handler_purchases_retry_redrive_test.go

@@ -2,7 +2,6 @@ package api
 
 import (
 	"context"
-	"encoding/json"
 	"testing"
 
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/common"
@@ -113,30 +112,29 @@ func purchasesFiredByRetry(t *testing.T, failed *config.PurchaseExecution, req *
 }
 
 // executeRedriveSuccessor runs a retry successor through the real
-// purchase.Manager the way production does: the row is picked up from the
-// execute_purchase queue message once the operator has approved it, claimed,
-// and executed. Returns the idempotency token of every purchase that reached
+// purchase.Manager the way production does: the persisted pending row is
+// approved by an operator, claimed, and executed. Returns the idempotency token of every purchase that reached
 // the cloud.
 func executeRedriveSuccessor(t *testing.T, successor *config.PurchaseExecution) []string {
 	t.Helper()
 
-	approved := *successor
-	approved.Status = "approved"
-	running := approved
+	pending := *successor
+	running := pending
 	running.Status = "running"
 
 	store := new(MockConfigStore)
-	for _, rec := range approved.Recommendations {
+	for _, rec := range pending.Recommendations {
 		if rec.Provider == "azure" {
-			store.On("ListStoredRecommendations", mock.Anything, config.RecommendationFilter{Provider: "azure", RequireAzurePricingMigration: true}).Return(approved.Recommendations, nil).Once()
+			store.On("ListStoredRecommendations", mock.Anything, config.RecommendationFilter{Provider: "azure", RequireAzurePricingMigration: true}).Return(pending.Recommendations, nil).Once()
 			break
 		}
 	}
 	t.Cleanup(func() { store.AssertExpectations(t) })
 
-	store.On("GetExecutionByID", mock.Anything, approved.ExecutionID).Return(&approved, nil).Once()
-	store.On("TransitionExecutionStatus", mock.Anything, approved.ExecutionID,
-		[]string{"approved", "pending", "notified"}, "running", (*string)(nil)).Return(&running, nil).Once()
+	store.On("GetExecutionByID", mock.Anything, pending.ExecutionID).Return(&pending, nil).Maybe()
+	store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil).Maybe()
+	store.On("TransitionExecutionStatus", mock.Anything, pending.ExecutionID,
+		[]string{"pending", "notified"}, "running", (*string)(nil)).Return(&running, nil).Once()
 	store.On("SavePurchaseHistory", mock.Anything, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil)
 	store.SavePurchaseExecutionFn = func(_ context.Context, _ *config.PurchaseExecution) error { return nil }
 
@@ -149,12 +147,9 @@ func executeRedriveSuccessor(t *testing.T, successor *config.PurchaseExecution)
 		DashboardURL:    "https://dashboard.example.com",
 	})
 
-	body, err := json.Marshal(purchase.AsyncMessage{
-		Type:        purchase.MessageTypeExecutePurchase,
-		ExecutionID: approved.ExecutionID,
-	})
-	require.NoError(t, err)
-	require.NoError(t, mgr.ProcessMessage(context.Background(), string(body)),
+	// Human approval of the persisted pending successor, as the retry flow does.
+	_, _, err := mgr.ApproveAndExecute(context.Background(), pending.ExecutionID, "", nil, nil)
+	require.NoError(t, err,
 		"the successor must execute cleanly; a non-nil error would make the purchase tally meaningless")
 
 	return svc.purchasedTokens()

internal/api/handler_purchases_test.go

@@ -7068,3 +7068,33 @@ func TestHandler_authorizeApprovalAction_NonASCIILettersNotFolded(t *testing.T)
 		})
 	}
 }
+
+// The approver comparison ignores letter case and surrounding whitespace on
+// either side (issue #307), so a session email that differs from the stored
+// contact_email only by case is the authorized approver.
+func TestHandler_authorizeApprovalAction_ApproverMatchIgnoresEmailCase(t *testing.T) {
+	cases := []struct{ name, contact, session string }{
+		{"mixed_case_contact_lower_session", "  Owner@Example.COM ", "owner@example.com"},
+		{"lower_contact_mixed_case_session", "owner@example.com", "Owner@Example.COM"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			execID := "12345678-1234-1234-1234-123456789abc"
+
+			mockConfig := new(MockConfigStore)
+			exec := approvalTestExec(execID, tc.contact, mockConfig)
+			mockConfig.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{}, nil)
+
+			mockAuth := new(MockAuthService)
+			mockAuth.On("ValidateSession", ctx, "sess-tok").Return(&Session{Email: tc.session}, nil)
+
+			handler := &Handler{config: mockConfig, auth: mockAuth}
+			req := &events.LambdaFunctionURLRequest{Headers: map[string]string{"authorization": "Bearer sess-tok"}}
+
+			actor, err := handler.authorizeApprovalAction(ctx, req, exec)
+			require.NoError(t, err)
+			assert.Equal(t, tc.session, actor)
+		})
+	}
+}

internal/api/types.go

@@ -141,7 +141,7 @@ type BreakdownValue struct {
 //
 // `transitionedBy` (ApproveAndExecute) is the session user's UUID stamped
 // onto purchase_executions.transitioned_by for human-initiated approvals;
-// pass nil for token/SQS/system flows so transitioned_by = NULL (issue #1009).
+// pass nil for token/system flows so transitioned_by = NULL (issue #1009).
 //
 // ApproveExecution, ApproveAndExecute and RunPlannedPurchaseNow return the
 // raw revocation token minted on a successful execute (issue #103: only its

internal/config/store_postgres_no_submission_integration_test.go

@@ -104,7 +104,7 @@ func TestNoSubmission_NonFailedWriteClearsProof(t *testing.T) {
 
 // Replays the approve-path sequence at the store level (approvals.go:392-421):
 // the approve path keeps a stale copy of the approved row; a concurrent
-// claimAndExecute fails the row before submission (proof true); the approve
+// claimAndExecuteWithin fails the row before submission (proof true); the approve
 // path's audit upsert puts the row back to approved; the approve path then
 // calls the provider and the row ends failed with outcome unknown (flag
 // false). The proof must not survive on a row that reached the provider.
@@ -114,7 +114,7 @@ func TestNoSubmission_StaleApproveUpsertDoesNotKeepProofOnSubmittedRow(t *testin
 	exec := newNoSubmissionExec(ctx, t, store, "approved")
 	staleApproved := *exec
 
-	exec.Status = "failed" // claimAndExecute: refused before submission
+	exec.Status = "failed" // claimAndExecuteWithin: refused before submission
 	exec.NoSubmission = true
 	require.NoError(t, store.SavePurchaseExecution(ctx, exec))