diff --git a/internal/api/handler_history.go b/internal/api/handler_history.go index a05c2214d..8453b2a87 100644 --- a/internal/api/handler_history.go +++ b/internal/api/handler_history.go @@ -62,9 +62,10 @@ func (h *Handler) getHistory(ctx context.Context, req *events.LambdaFunctionURLR // page size, and later pages leave it out, so paging can never push a // pending approval off the first page. A failure to list them must not // hide completed history — log, skip, continue, and flag the response. - // The same filter set is applied here (in-memory against the synthesized - // row's recommendations and scheduled_date) so the two halves of the - // merged response are consistently scoped (issue #701). + // The same filter set and the account scope are pushed into the SQL read + // so the class cap counts only rows the caller can be shown (issue #632), + // and re-checked in memory against the synthesized row so the two halves + // of the merged response are consistently scoped (issue #701). // // Stale pending/notified executions are expired AFTER the response is // assembled so the GET handler is a pure read (issue #1032): the caller @@ -80,7 +81,7 @@ func (h *Handler) getHistory(ctx context.Context, req *events.LambdaFunctionURLR var staleExecs []config.PurchaseExecution actionableIncomplete := false if filters.Cursor == nil { - extra, staleExecs, actionableIncomplete = h.fetchExecutionsAsHistory(ctx, filters) + extra, staleExecs, actionableIncomplete = h.fetchExecutionsAsHistory(ctx, filters, scope) extra = scope.filter(extra) } defer h.expireStaleExecutions(staleExecs) @@ -185,32 +186,29 @@ const approvalExpiryWindow = 7 * 24 * time.Hour // history must still render. Terminal executions are read by the paged // stream instead (fetchHistoryPage). // -// The filter set (issue #701) is applied in Go against the synthetic row: -// provider via the recs' collapsed provider, account via CloudAccountID, -// date via ScheduledDate. +// The filter set (issue #701) and the caller's account scope are pushed into +// the SQL read as a superset (issue #632), so the MaxListLimit cap counts only +// executions the caller could be shown and other tenants' or filtered-out rows +// cannot crowd a pending approval out. The exact checks still run in Go +// against the synthetic row: provider via the recs' providers, account via +// CloudAccountID, plan via PlanID, date via ScheduledDate here, and the scope +// in the caller. // // The third return value reports that actionable executions may be missing: -// the class reached its MaxListLimit cap, so older executions may have been -// dropped before the filters ran, or its listing failed (issue #686). -func (h *Handler) fetchExecutionsAsHistory(ctx context.Context, filters historyFilters) ([]config.PurchaseHistoryRecord, []config.PurchaseExecution, bool) { - executions, ok := h.fetchExecutionsByClass(ctx, actionableExecutionStatuses, filters.PlanID, config.MaxListLimit) +// the class reached its MaxListLimit cap, or its listing failed (issue #686). +func (h *Handler) fetchExecutionsAsHistory(ctx context.Context, filters historyFilters, scope historyScope) ([]config.PurchaseHistoryRecord, []config.PurchaseExecution, bool) { + q := actionableExecutionsQuery(&filters, &scope) + executions, ok := h.fetchExecutionsByClass(ctx, q) incomplete := !ok || len(executions) >= config.MaxListLimit + staleExecs := h.staleExecutionsToSweep(ctx, &q, executions) if len(executions) == 0 { - return nil, nil, incomplete + return nil, staleExecs, incomplete } approver := h.resolvePendingApproverEmail(ctx) userEmailCache := h.resolveUserEmails(ctx, executions) out := make([]config.PurchaseHistoryRecord, 0, len(executions)) - var staleExecs []config.PurchaseExecution for _rvc := range executions { exec := executions[_rvc] - // Collect stale pending/notified executions for the post-assembly - // expire sweep. We do NOT mutate status here to keep the GET - // read-only: the response reflects current DB state; the sweep - // fires the transition so the next request sees "expired". - if isStaleExecution(exec) { - staleExecs = append(staleExecs, exec) - } if !filters.matchesExecution(exec) { continue } @@ -224,18 +222,13 @@ func (h *Handler) fetchExecutionsAsHistory(ctx context.Context, filters historyF // render; the caller flags the response as incomplete so the hidden rows are // not silent (issue #686). // -// A non-empty planID selects only that plan's executions in SQL, so a plan view -// is not crowded out by other plans' rows under the limit. -func (h *Handler) fetchExecutionsByClass(ctx context.Context, statuses []string, planID string, limit int) ([]config.PurchaseExecution, bool) { - var rows []config.PurchaseExecution - var err error - if planID != "" { - rows, err = h.config.ListHistoryExecutionsPage(ctx, config.HistoryExecutionsPageQuery{Statuses: statuses, PlanID: planID, Limit: limit}) - } else { - rows, err = h.config.GetExecutionsByStatuses(ctx, statuses, limit) - } +// q carries the request filters, plan_id included, and the caller's scope, so +// a plan view or a scoped caller is not crowded out by other rows under the +// limit (issues #632, #1404). +func (h *Handler) fetchExecutionsByClass(ctx context.Context, q config.ExecutionsByStatusesQuery) ([]config.PurchaseExecution, bool) { + rows, err := h.config.GetExecutionsByStatusesFiltered(ctx, q) if err != nil { - logging.Warnf("history: failed to load executions with statuses %v: %v", statuses, err) + logging.Warnf("history: failed to load executions with statuses %v: %v", q.Statuses, err) return nil, false } return rows, true diff --git a/internal/api/handler_history_actionable.go b/internal/api/handler_history_actionable.go new file mode 100644 index 000000000..c838769f9 --- /dev/null +++ b/internal/api/handler_history_actionable.go @@ -0,0 +1,103 @@ +package api + +import ( + "context" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// actionableExecutionsQuery is the SQL read for the History view's actionable +// class (issue #632). It carries the request filters and the caller's account +// scope so the store applies them before the MaxListLimit cap. Each predicate +// is a superset of the Go check the handler still runs (matchesExecution and +// historyScope.allows), so pushing it down can only drop rows Go would drop. +func actionableExecutionsQuery(filters *historyFilters, scope *historyScope) config.ExecutionsByStatusesQuery { + q := config.ExecutionsByStatusesQuery{ + Statuses: actionableExecutionStatuses, + Limit: config.MaxListLimit, + Provider: filters.Provider, + PlanID: filters.PlanID, + Scope: scope.executionScope(), + } + if filters.HasDate { + start, end := filters.Start, filters.End + q.Start, q.End = &start, &end + } + if len(filters.AccountIDs) > 0 || len(filters.ExternalIDsByProvider) > 0 { + q.AccountIDs = executionFilterAccountValues(filters) + } + return q +} + +// executionFilterAccountValues lists every account value accountMatchesFilters +// can match: the UUIDs and the external numbers of every provider. The +// provider pairing of an external number is left to the Go check, so this is +// a superset of it. +func executionFilterAccountValues(filters *historyFilters) []string { + values := append([]string{}, filters.AccountIDs...) + for _, ids := range filters.ExternalIDsByProvider { + values = appendMissing(values, ids...) + } + return values +} + +// executionScope returns the SQL form of the scope, or nil for an +// unrestricted caller. It lists every account value allows can accept: the +// scope's literal entries (allows compares the row's account against them +// directly) plus the UUID and external number of each registered account the +// scope allows, since allows also matches a row by the name of the account +// its UUID or external number resolves to. An empty list keeps only the +// caller's own unattributed rows, as allows does. +func (s *historyScope) executionScope() *config.ExecutionScope { + if s.allowed.AllowsAll() { + return nil + } + values := appendMissing(nil, s.allowed.Accounts...) + for i := range s.accounts { + a := &s.accounts[i] + if s.allowed.Allows(a.ID, a.Name) { + values = appendMissing(values, a.ID, a.ExternalID) + } + } + return &config.ExecutionScope{AccountIDs: values, UserID: s.userID} +} + +// isNarrowed reports whether q drops any row the unfiltered class read would +// return. Every filter field of ExecutionsByStatusesQuery must be listed here, +// or the stale sweep would only see the narrowed rows. +func isNarrowed(q *config.ExecutionsByStatusesQuery) bool { + return q.Start != nil || q.End != nil || q.Provider != "" || q.PlanID != "" || + len(q.AccountIDs) > 0 || q.Scope != nil +} + +// staleExecutionsToSweep returns the stale pending/notified executions the GET +// expires after it responds (see expireStaleExecutions). The sweep is not +// limited to the caller's filters or scope: it never shows a row, and the +// History load is what expires stale approvals and releases their +// suppressions. So when q is narrowed, the stale rows are read with a +// separate unfiltered query; otherwise they are taken from the class already +// read. A read failure is logged and skips the sweep for this request. +func (h *Handler) staleExecutionsToSweep(ctx context.Context, q *config.ExecutionsByStatusesQuery, executions []config.PurchaseExecution) []config.PurchaseExecution { + if isNarrowed(q) { + cutoff := time.Now().Add(-approvalExpiryWindow) + rows, err := h.config.GetExecutionsByStatusesFiltered(ctx, config.ExecutionsByStatusesQuery{ + Statuses: []string{"pending", "notified"}, + Limit: config.DefaultListLimit, + End: &cutoff, + }) + if err != nil { + logging.Warnf("history: failed to load stale executions to expire: %v", err) + return nil + } + executions = rows + } + var stale []config.PurchaseExecution + for i := range executions { + if isStaleExecution(executions[i]) { + stale = append(stale, executions[i]) + } + } + return stale +} diff --git a/internal/api/handler_history_actionable_test.go b/internal/api/handler_history_actionable_test.go new file mode 100644 index 000000000..84d08b708 --- /dev/null +++ b/internal/api/handler_history_actionable_test.go @@ -0,0 +1,257 @@ +package api + +import ( + "context" + "encoding/json" + "slices" + "testing" + "time" + + "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/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +const ( + actionableDevUUID = "11111111-1111-4111-8111-111111111111" + actionableOtherUUID = "22222222-2222-4222-8222-222222222222" +) + +var actionableTestAccounts = []config.CloudAccount{ + {ID: actionableDevUUID, Name: "dev", Provider: "aws", ExternalID: "111111111111"}, + {ID: actionableOtherUUID, Name: "other", Provider: "aws", ExternalID: "222222222222"}, +} + +// actionableHistoryRequest drives GET /api/history through HandleRequest as +// user u-1 with the given allow-list and query string. +func actionableHistoryRequest(t *testing.T, store *MockConfigStore, scope []string, params map[string]string) HistoryResponse { + t.Helper() + ctx := context.Background() + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", mock.Anything, "tok").Return(&Session{UserID: "u-1", Email: "u@example.com"}, nil) + mockAuth.On("HasPermissionAPI", mock.Anything, "u-1", mock.Anything, mock.Anything).Return(true, nil) + mockAuth.On("GetAllowedAccountsAPI", mock.Anything, "u-1").Return(scope, nil) + store.On("ListPurchaseHistoryPage", mock.Anything, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) + store.On("ListHistoryExecutionsPage", mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) + store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil).Maybe() + store.ListCloudAccountsFn = func(context.Context, config.CloudAccountFilter) ([]config.CloudAccount, error) { + return actionableTestAccounts, nil + } + + h := &Handler{auth: mockAuth, config: store} + resp, err := h.HandleRequest(ctx, &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer tok"}, + QueryStringParameters: params, + RequestContext: events.LambdaFunctionURLRequestContext{ + HTTP: events.LambdaFunctionURLRequestContextHTTPDescription{Method: "GET", Path: "/api/history"}, + }, + }) + require.NoError(t, err) + require.Equal(t, 200, resp.StatusCode, "body: %s", resp.Body) + var body HistoryResponse + require.NoError(t, json.Unmarshal([]byte(resp.Body), &body)) + return body +} + +func isActionableQuery(q config.ExecutionsByStatusesQuery) bool { + return slices.Equal(q.Statuses, actionableExecutionStatuses) +} + +func isStaleSweepQuery(q config.ExecutionsByStatusesQuery) bool { + return slices.Equal(q.Statuses, []string{"pending", "notified"}) +} + +func pendingForAccount(id, account string, scheduled time.Time) config.PurchaseExecution { + return config.PurchaseExecution{ + ExecutionID: id, Status: "pending", ScheduledDate: scheduled, + CloudAccountID: &account, + Recommendations: []config.RecommendationRecord{{Provider: "aws", Service: "ec2", CloudAccountID: &account, Count: 1}}, + } +} + +// Issue #632: a scoped user's pending approval must not be hidden because +// newer actionable rows of accounts outside the scope fill the class cap. The +// unfiltered read returns MaxListLimit of those rows; the scoped read the +// handler now issues returns the user's own row. +func TestHandleRequest_History_ScopedPendingNotCrowdedOutByOtherAccounts(t *testing.T) { + now := time.Now() + others := make([]config.PurchaseExecution, config.MaxListLimit) + for i := range others { + others[i] = pendingForAccount("other", actionableOtherUUID, now) + } + mine := pendingForAccount("mine", actionableDevUUID, now.Add(-time.Hour)) + + store := new(MockConfigStore) + store.On("GetExecutionsByStatuses", mock.Anything, mock.Anything, mock.Anything).Return(others, nil) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(func(q config.ExecutionsByStatusesQuery) bool { + return isActionableQuery(q) && q.Scope != nil && slices.Contains(q.Scope.AccountIDs, actionableDevUUID) && + !slices.Contains(q.Scope.AccountIDs, actionableOtherUUID) + })).Return([]config.PurchaseExecution{mine}, nil) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isStaleSweepQuery)).Return([]config.PurchaseExecution{}, nil) + + body := actionableHistoryRequest(t, store, []string{"dev"}, map[string]string{}) + + ids := make([]string, 0, len(body.Purchases)) + for _, p := range body.Purchases { + ids = append(ids, p.PurchaseID) + } + assert.Equal(t, []string{"mine"}, ids) + assert.False(t, body.Truncated, "the caller's class is not capped") +} + +// The request's provider and date filters reach the actionable SQL read, and +// an unrestricted caller sends no scope. +func TestHandleRequest_History_ActionableQueryCarriesFilters(t *testing.T) { + var got config.ExecutionsByStatusesQuery + store := new(MockConfigStore) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isActionableQuery)). + Run(func(args mock.Arguments) { got = args.Get(1).(config.ExecutionsByStatusesQuery) }). + Return([]config.PurchaseExecution{}, nil).Once() + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isStaleSweepQuery)).Return([]config.PurchaseExecution{}, nil).Once() + + actionableHistoryRequest(t, store, []string{"*"}, map[string]string{"provider": "aws", "start": "2026-01-01", "end": "2026-01-31"}) + + assert.Equal(t, "aws", got.Provider) + assert.Nil(t, got.Scope, "an unrestricted caller is not scoped in SQL") + assert.Equal(t, config.MaxListLimit, got.Limit) + require.NotNil(t, got.Start) + require.NotNil(t, got.End) + assert.Equal(t, time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), *got.Start) + assert.Equal(t, time.Date(2026, 1, 31, 23, 59, 59, 0, time.UTC), *got.End) + assert.Empty(t, got.AccountIDs) + store.AssertExpectations(t) +} + +// A narrowed actionable read must not narrow the stale-approval sweep: the +// History load is what expires stale approvals, whoever loads it. The sweep +// reads pending/notified rows past the expiry window with no filter or scope. +// An unfiltered, unrestricted load reuses the class it already read. +func TestHandleRequest_History_StaleSweepIgnoresFiltersAndScope(t *testing.T) { + t.Run("narrowed", func(t *testing.T) { + var sweep config.ExecutionsByStatusesQuery + store := new(MockConfigStore) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isActionableQuery)).Return([]config.PurchaseExecution{}, nil) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isStaleSweepQuery)). + Run(func(args mock.Arguments) { sweep = args.Get(1).(config.ExecutionsByStatusesQuery) }). + Return([]config.PurchaseExecution{}, nil).Once() + + before := time.Now() + actionableHistoryRequest(t, store, []string{"dev"}, map[string]string{"provider": "azure"}) + + store.AssertExpectations(t) + assert.Nil(t, sweep.Scope) + assert.Empty(t, sweep.Provider) + assert.Empty(t, sweep.AccountIDs) + assert.Nil(t, sweep.Start) + require.NotNil(t, sweep.End) + assert.WithinDuration(t, before.Add(-approvalExpiryWindow), *sweep.End, time.Minute) + }) + t.Run("not narrowed", func(t *testing.T) { + store := new(MockConfigStore) + store.On("GetExecutionsByStatuses", mock.Anything, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) + + actionableHistoryRequest(t, store, []string{"*"}, map[string]string{}) + + store.AssertNumberOfCalls(t, "GetExecutionsByStatuses", 1) + }) +} + +// actionablePlanUUID is the plan the plan-history view filters on. +const actionablePlanUUID = "33333333-3333-4333-8333-333333333333" + +func isPlanQuery(planID string) func(config.ExecutionsByStatusesQuery) bool { + return func(q config.ExecutionsByStatusesQuery) bool { return isActionableQuery(q) && q.PlanID == planID } +} + +// Issues #632 and #1404: the plan-history view's plan_id is part of the +// actionable SQL read, so MaxListLimit newer pending rows of other plans do +// not hide the plan's own pending row. The plan-less read returns the other +// plans' rows; the read carrying the plan returns the plan's row. +func TestHandleRequest_History_PlanPendingNotCrowdedOutByOtherPlans(t *testing.T) { + now := time.Now() + others := make([]config.PurchaseExecution, config.MaxListLimit) + for i := range others { + others[i] = pendingForAccount("other", actionableDevUUID, now) + others[i].PlanID = "44444444-4444-4444-8444-444444444444" + } + mine := pendingForAccount("mine", actionableDevUUID, now.Add(-time.Hour)) + mine.PlanID = actionablePlanUUID + + store := new(MockConfigStore) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isPlanQuery(""))).Return(others, nil).Maybe() + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isPlanQuery(actionablePlanUUID))). + Return([]config.PurchaseExecution{mine}, nil) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isStaleSweepQuery)).Return([]config.PurchaseExecution{}, nil).Maybe() + + body := actionableHistoryRequest(t, store, []string{"*"}, map[string]string{"plan_id": actionablePlanUUID}) + + ids := make([]string, 0, len(body.Purchases)) + for _, p := range body.Purchases { + ids = append(ids, p.PurchaseID) + } + assert.Equal(t, []string{"mine"}, ids) + assert.False(t, body.Truncated, "the plan's class is not capped") +} + +// A plan view narrows the actionable read, so it must still issue the +// separate unfiltered stale sweep: other plans' stale approvals are expired by +// any History load, plan view or not. +func TestHandleRequest_History_PlanViewStillSweepsAllStaleApprovals(t *testing.T) { + var sweep config.ExecutionsByStatusesQuery + store := new(MockConfigStore) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isActionableQuery)).Return([]config.PurchaseExecution{}, nil) + store.On("GetExecutionsByStatusesFiltered", mock.Anything, mock.MatchedBy(isStaleSweepQuery)). + Run(func(args mock.Arguments) { sweep = args.Get(1).(config.ExecutionsByStatusesQuery) }). + Return([]config.PurchaseExecution{}, nil).Once() + + actionableHistoryRequest(t, store, []string{"*"}, map[string]string{"plan_id": actionablePlanUUID}) + + store.AssertExpectations(t) + assert.Empty(t, sweep.PlanID, "the sweep is not limited to the viewed plan") + assert.Nil(t, sweep.Scope) + assert.Empty(t, sweep.Provider) + assert.Empty(t, sweep.AccountIDs) +} + +func TestIsNarrowed_PlanID(t *testing.T) { + assert.False(t, isNarrowed(&config.ExecutionsByStatusesQuery{Statuses: actionableExecutionStatuses})) + assert.True(t, isNarrowed(&config.ExecutionsByStatusesQuery{Statuses: actionableExecutionStatuses, PlanID: actionablePlanUUID})) +} + +func TestActionableExecutionsQuery_AccountValues(t *testing.T) { + f := historyFilters{ + AccountIDs: []string{actionableDevUUID}, + ExternalIDsByProvider: map[string][]string{"aws": {"111111111111"}, "": {"333333333333"}}, + } + q := actionableExecutionsQuery(&f, &historyScope{allowed: auth.UnrestrictedScope()}) + assert.ElementsMatch(t, []string{actionableDevUUID, "111111111111", "333333333333"}, q.AccountIDs) + assert.Nil(t, q.Start) + assert.Nil(t, q.End) +} + +func TestHistoryScope_ExecutionScope(t *testing.T) { + tests := []struct { + name string + scope auth.AccountScope + want *config.ExecutionScope + }{ + {"unrestricted", auth.UnrestrictedScope(), nil}, + {"empty denies attributed rows", auth.ScopeForAccounts(nil), &config.ExecutionScope{UserID: "u-1"}}, + {"by id", auth.ScopeForAccounts([]string{actionableDevUUID}), + &config.ExecutionScope{UserID: "u-1", AccountIDs: []string{actionableDevUUID, "111111111111"}}}, + {"by unique name", auth.ScopeForAccounts([]string{"dev"}).WithRegisteredNames([]string{"dev", "other"}), + &config.ExecutionScope{UserID: "u-1", AccountIDs: []string{"dev", actionableDevUUID, "111111111111"}}}, + {"literal external number", auth.ScopeForAccounts([]string{"222222222222"}), + &config.ExecutionScope{UserID: "u-1", AccountIDs: []string{"222222222222"}}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := historyScope{allowed: tt.scope, accounts: actionableTestAccounts, userID: "u-1"} + assert.Equal(t, tt.want, s.executionScope()) + }) + } +} diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 2eb1e0fc6..857dfcd0d 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -155,6 +155,10 @@ type StoreInterface interface { // this method's status filter) to avoid accidental double-processing of // failed / expired rows. GetExecutionsByStatuses(ctx context.Context, statuses []string, limit int) ([]PurchaseExecution, error) + // GetExecutionsByStatusesFiltered is GetExecutionsByStatuses with the + // History filters and account scope in q applied in SQL before the limit, + // so the cap counts only rows the caller can be shown (issue #632). + GetExecutionsByStatusesFiltered(ctx context.Context, q ExecutionsByStatusesQuery) ([]PurchaseExecution, error) // ListHistoryExecutionsPage and ListPurchaseHistoryPage read one keyset // page each of the History view's paged stream (issue #674): rows strictly // after q.After / after (nil starts at the top) in the stream order of diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index e31f7c9c8..8bb3eee12 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -1571,15 +1571,23 @@ func (s *PostgresStore) CancelScheduledExecutionAtomic(ctx context.Context, tx p // already represented by their purchase_history rows, so returning them would // only consume the cap (the History handler skips them anyway). func (s *PostgresStore) GetExecutionsByStatuses(ctx context.Context, statuses []string, limit int) ([]PurchaseExecution, error) { - if len(statuses) == 0 { + return s.GetExecutionsByStatusesFiltered(ctx, ExecutionsByStatusesQuery{Statuses: statuses, Limit: limit}) +} + +// GetExecutionsByStatusesFiltered is GetExecutionsByStatuses with the History +// filters and account scope in q applied in SQL, before the limit (issue +// #632). Order and the clean-completed exclusion are GetExecutionsByStatuses'. +func (s *PostgresStore) GetExecutionsByStatusesFiltered(ctx context.Context, q ExecutionsByStatusesQuery) ([]PurchaseExecution, error) { + if len(q.Statuses) == 0 { return nil, nil } + limit := q.Limit if limit <= 0 { limit = DefaultListLimit } - if limit > MaxListLimit { - limit = MaxListLimit - } + limit = min(limit, MaxListLimit) + conds, args := executionsByStatusesWhere(&q) + args = append(args, limit) query := ` SELECT plan_id, execution_id, status, step_number, scheduled_date, notification_sent, approval_token_hash, recommendations, @@ -1590,13 +1598,12 @@ func (s *PostgresStore) GetExecutionsByStatuses(ctx context.Context, statuses [] executed_by_user_id, executed_at, pre_approval_skip_reason, idempotency_key, scheduled_execution_at, no_submission, created_by_system FROM purchase_executions - WHERE ` + historyExecutionsPredicate + ` + WHERE ` + strings.Join(conds, "\n\t\t AND ") + ` ORDER BY (status = 'pending' AND scheduled_date > now()) ASC, CASE WHEN status = 'pending' AND scheduled_date > now() THEN scheduled_date END ASC, scheduled_date DESC - LIMIT $2 - ` - return s.queryExecutions(ctx, query, statuses, limit) + LIMIT $` + fmt.Sprint(len(args)) + return s.queryExecutions(ctx, query, args...) } // CountExecutionsByPlanAndStatus returns, keyed by plan ID, the exact number diff --git a/internal/config/store_postgres_executions_filtered.go b/internal/config/store_postgres_executions_filtered.go new file mode 100644 index 000000000..776e9603e --- /dev/null +++ b/internal/config/store_postgres_executions_filtered.go @@ -0,0 +1,105 @@ +package config + +import ( + "fmt" + "time" +) + +// ExecutionsByStatusesQuery selects executions for the History view's +// actionable class (issue #632). Statuses and Limit mean what they mean for +// GetExecutionsByStatuses. The other fields push the History filters and the +// caller's account scope into SQL, so the Limit cap applies to rows the caller +// can be shown rather than to every tenant's rows. +// +// Every predicate is a superset of the History handler's Go check +// (matchesExecution and the account scope), which still runs on the result: +// SQL may keep a row Go then drops, never the reverse. +type ExecutionsByStatusesQuery struct { + // Start and End bound scheduled_date, inclusive. nil leaves a side open. + Start *time.Time + End *time.Time + // Scope, when non-nil, keeps only executions the account scope can allow. + // nil is an unrestricted caller. + Scope *ExecutionScope + // Provider keeps executions with at least one recommendation for this + // provider. "" applies no provider filter. + Provider string + // PlanID matches purchase_executions.plan_id exactly, as + // HistoryExecutionsPageQuery.PlanID does. "" applies no plan filter. + PlanID string + Statuses []string + // AccountIDs keeps executions whose cloud_account_id is one of these + // values, or, when cloud_account_id is NULL, that have a recommendation + // whose cloud_account_id is one of them. Empty applies no account filter. + AccountIDs []string + Limit int +} + +// ExecutionScope is the SQL form of a restricted account scope. +type ExecutionScope struct { + // UserID is the caller. An execution with no cloud_account_id is kept when + // this user created it, because a row that cannot be attributed to one + // account is shown to its creator only. "" matches a NULL creator. + UserID string + // AccountIDs lists every account value the scope can allow: the scope's + // literal entries plus the UUID and external number of each registered + // account it allows. Empty keeps no attributed execution (fail closed). + AccountIDs []string +} + +// executionRecsArray is the recommendations column as a JSON array. A +// non-array value (json.Marshal of a nil slice stores 'null') reads as empty, +// as encoding/json decodes it to an empty slice. +const executionRecsArray = `(CASE WHEN jsonb_typeof(recommendations) = 'array' THEN recommendations ELSE '[]'::jsonb END)` + +// executionRecHasValue is true when a recommendation's JSON key (the lowercase +// tag json.Marshal writes for RecommendationRecord) equals one of the values +// bound at the given placeholder. +func executionRecHasValue(key string, arg int) string { + return fmt.Sprintf(`EXISTS (SELECT 1 FROM jsonb_array_elements(%s) AS r WHERE r.value->>'%s' = ANY($%d))`, + executionRecsArray, key, arg) +} + +// executionAccountIn is true when the execution's account is one of the values +// at the given placeholder: cloud_account_id when set, else any +// recommendation's cloud_account_id. The handler's effective account is +// cloud_account_id, else the account every recommendation shares, so this is +// a superset of it. +func executionAccountIn(arg int) string { + return fmt.Sprintf(`(cloud_account_id::text = ANY($%d) OR (cloud_account_id IS NULL AND %s))`, + arg, executionRecHasValue("cloud_account_id", arg)) +} + +// executionsByStatusesWhere builds the WHERE conditions and bind args for q. +// $1 is the status list, as historyExecutionsPredicate expects. +func executionsByStatusesWhere(q *ExecutionsByStatusesQuery) (conds []string, args []any) { + args = []any{q.Statuses} + conds = []string{historyExecutionsPredicate} + bind := func(v any) int { + args = append(args, v) + return len(args) + } + if q.Start != nil { + conds = append(conds, fmt.Sprintf("scheduled_date >= $%d", bind(*q.Start))) + } + if q.End != nil { + conds = append(conds, fmt.Sprintf("scheduled_date <= $%d", bind(*q.End))) + } + if q.PlanID != "" { + conds = append(conds, fmt.Sprintf("plan_id = $%d", bind(q.PlanID))) + } + if q.Provider != "" { + conds = append(conds, executionRecHasValue("provider", bind([]string{q.Provider}))) + } + if len(q.AccountIDs) > 0 { + conds = append(conds, executionAccountIn(bind(q.AccountIDs))) + } + if q.Scope != nil { + accounts := bind(append([]string{}, q.Scope.AccountIDs...)) + user := bind(q.Scope.UserID) + conds = append(conds, fmt.Sprintf( + `(cloud_account_id::text = ANY($%d) OR (cloud_account_id IS NULL AND (COALESCE(created_by_user_id::text, '') = $%d OR %s)))`, + accounts, user, executionRecHasValue("cloud_account_id", accounts))) + } + return conds, args +} diff --git a/internal/config/store_postgres_executions_filtered_integration_test.go b/internal/config/store_postgres_executions_filtered_integration_test.go new file mode 100644 index 000000000..d24b9339c --- /dev/null +++ b/internal/config/store_postgres_executions_filtered_integration_test.go @@ -0,0 +1,145 @@ +//go:build integration + +package config + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestIntegration_GetExecutionsByStatusesFiltered runs the issue #632 SQL +// pushdown over every account shape an execution can have: cloud_account_id +// set, NULL with the account only on the recommendations (one account, +// several, an external number, none), and a non-array recommendations value. +func TestIntegration_GetExecutionsByStatusesFiltered(t *testing.T) { + ctx := context.Background() + f := newHistorySummaryFixture(ctx, t) + a, b, user := f.accountA, f.accountB, f.userID + legacy := "ext-legacy" + day := func(n int) time.Time { return f.base.Add(time.Duration(n) * 24 * time.Hour) } + recA := []RecommendationRecord{{Provider: "aws", CloudAccountID: &a}} + recB := []RecommendationRecord{{Provider: "azure", CloudAccountID: &b}} + mixed := []RecommendationRecord{{Provider: "aws", CloudAccountID: &a}, {Provider: "azure", CloudAccountID: &b}} + + ids := map[string]string{ + "onA": f.saveExecution(ctx, t, day(1), "pending", "", &a, nil, recA), + "onB": f.saveExecution(ctx, t, day(2), "pending", "", &b, nil, recB), + "recsA": f.saveExecution(ctx, t, day(3), "pending", "", nil, nil, recA), + "mixedMy": f.saveExecution(ctx, t, day(4), "pending", "", nil, &user, mixed), + "mixedAny": f.saveExecution(ctx, t, day(5), "pending", "", nil, nil, mixed), + "noRecs": f.saveExecution(ctx, t, day(6), "pending", "", nil, nil, nil), + "external": f.saveExecution(ctx, t, day(7), "pending", "", nil, nil, []RecommendationRecord{{Provider: "aws", CloudAccountID: &legacy}}), + } + f.saveExecution(ctx, t, day(8), "failed", "boom", &a, nil, recA) // outside the requested statuses + + names := func(q ExecutionsByStatusesQuery) []string { + t.Helper() + q.Statuses = []string{"pending"} + if q.Limit == 0 { + q.Limit = 100 + } + got, err := f.store.GetExecutionsByStatusesFiltered(ctx, q) + require.NoError(t, err) + out := make([]string, 0, len(got)) + for i := range got { + for name, id := range ids { + if id == got[i].ExecutionID { + out = append(out, name) + } + } + } + require.Len(t, out, len(got), "every row is a seeded execution") + return out + } + start, end := day(2), day(4) + + tests := []struct { + name string + q ExecutionsByStatusesQuery + want []string + }{ + {"no filter", ExecutionsByStatusesQuery{}, []string{"external", "noRecs", "mixedAny", "mixedMy", "recsA", "onB", "onA"}}, + {"provider aws", ExecutionsByStatusesQuery{Provider: "aws"}, []string{"external", "mixedAny", "mixedMy", "recsA", "onA"}}, + {"provider azure", ExecutionsByStatusesQuery{Provider: "azure"}, []string{"mixedAny", "mixedMy", "onB"}}, + {"account A", ExecutionsByStatusesQuery{AccountIDs: []string{a}}, []string{"mixedAny", "mixedMy", "recsA", "onA"}}, + {"external number on recs", ExecutionsByStatusesQuery{AccountIDs: []string{legacy}}, []string{"external"}}, + {"date bounds inclusive", ExecutionsByStatusesQuery{Start: &start, End: &end}, []string{"mixedMy", "recsA", "onB"}}, + {"scope A", ExecutionsByStatusesQuery{Scope: &ExecutionScope{AccountIDs: []string{a}, UserID: user}}, + []string{"mixedAny", "mixedMy", "recsA", "onA"}}, + {"scope nothing keeps own unattributed rows", ExecutionsByStatusesQuery{Scope: &ExecutionScope{UserID: user}}, []string{"mixedMy"}}, + {"scope with no user matches a NULL creator", ExecutionsByStatusesQuery{Scope: &ExecutionScope{}}, + []string{"external", "noRecs", "mixedAny", "recsA"}}, + {"filters and scope combine", ExecutionsByStatusesQuery{Provider: "azure", Scope: &ExecutionScope{AccountIDs: []string{b}, UserID: user}}, + []string{"mixedAny", "mixedMy", "onB"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, names(tt.q)) + }) + } + + // The cap applies after the scope: the newer rows outside scope B do not + // push B's own row out of a three-row read, as they do unscoped. + t.Run("limit counts only rows in scope", func(t *testing.T) { + assert.Equal(t, []string{"external", "noRecs", "mixedAny"}, names(ExecutionsByStatusesQuery{Limit: 3})) + assert.Equal(t, []string{"mixedAny", "mixedMy", "onB"}, names(ExecutionsByStatusesQuery{Limit: 3, + Scope: &ExecutionScope{AccountIDs: []string{b}, UserID: "00000000-0000-4000-8000-000000000000"}})) + }) + + t.Run("GetExecutionsByStatuses is the unfiltered query", func(t *testing.T) { + got, err := f.store.GetExecutionsByStatuses(ctx, []string{"pending"}, 100) + require.NoError(t, err) + assert.Len(t, got, len(ids)) + }) +} + +// TestIntegration_GetExecutionsByStatusesFiltered_PlanID pins the plan-history +// view's plan_id on the actionable read (issues #632 and #1404): newer pending +// rows of another plan, or of no plan, must not push the plan's own row out of +// the limit, and the plan is ANDed with the scope rather than widening it. +func TestIntegration_GetExecutionsByStatusesFiltered_PlanID(t *testing.T) { + ctx := context.Background() + f := newHistorySummaryFixture(ctx, t) + planA := createHistoryTestPlan(ctx, t, f.store, "actionable-plan-a") + planB := createHistoryTestPlan(ctx, t, f.store, "actionable-plan-b") + a, b := f.accountA, f.accountB + save := func(n int, planID string, account *string) string { + exec := &PurchaseExecution{ + PlanID: planID, ExecutionID: uuid.NewString(), IdempotencyKey: uuid.NewString(), Status: "pending", + ScheduledDate: f.base.Add(time.Duration(n) * time.Hour), CloudAccountID: account, + } + require.NoError(t, f.store.SavePurchaseExecution(ctx, exec)) + return exec.ExecutionID + } + mine := save(1, planA, &a) + save(2, planB, &a) + save(3, planB, &b) + save(4, "", &a) // not tied to any plan + + ids := func(q ExecutionsByStatusesQuery) []string { + t.Helper() + q.Statuses = []string{"pending"} + got, err := f.store.GetExecutionsByStatusesFiltered(ctx, q) + require.NoError(t, err) + out := make([]string, 0, len(got)) + for i := range got { + out = append(out, got[i].ExecutionID) + } + return out + } + + assert.Equal(t, []string{mine}, ids(ExecutionsByStatusesQuery{PlanID: planA, Limit: 100})) + assert.Equal(t, []string{mine}, ids(ExecutionsByStatusesQuery{PlanID: planA, Limit: 1}), + "the limit counts only the plan's rows") + assert.NotEqual(t, []string{mine}, ids(ExecutionsByStatusesQuery{Limit: 1}), + "control: unfiltered, newer rows of other plans fill a one-row read") + assert.Len(t, ids(ExecutionsByStatusesQuery{PlanID: planB, Limit: 100}), 2) + assert.Empty(t, ids(ExecutionsByStatusesQuery{PlanID: planA, Limit: 100, + Scope: &ExecutionScope{AccountIDs: []string{b}, UserID: f.userID}}), "the plan narrows the scope, never widens it") + assert.Empty(t, ids(ExecutionsByStatusesQuery{PlanID: uuid.NewString(), Limit: 100})) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index c4c46b586..a348fbc52 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1907,6 +1907,27 @@ func (m *MockConfigStore) WithTx(ctx context.Context, fn func(tx pgx.Tx) error) return fn(nil) } +// GetExecutionsByStatusesFiltered mocks the GetExecutionsByStatusesFiltered +// operation. Without an expectation for it, it answers from the +// GetExecutionsByStatuses expectation: the unfiltered rows are a superset of +// the filtered ones, and the History handler re-applies every filter in Go, so +// tests written against GetExecutionsByStatuses keep working. +func (m *MockConfigStore) GetExecutionsByStatusesFiltered(ctx context.Context, q config.ExecutionsByStatusesQuery) ([]config.PurchaseExecution, error) { + if !isExpected(&m.Mock, "GetExecutionsByStatusesFiltered") { + return m.GetExecutionsByStatuses(ctx, q.Statuses, q.Limit) + } + m.record("GetExecutionsByStatusesFiltered", ctx, q) + args := m.Called(ctx, q) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).([]config.PurchaseExecution) + if !ok { + m.reporter.wrongReturnType("GetExecutionsByStatusesFiltered", "[]config.PurchaseExecution", args.Get(0)) + } + return filterExecutionsLikeStore(v, q.Statuses, q.Limit), args.Error(1) +} + // GetExecutionsByStatuses mocks the GetExecutionsByStatuses operation. func (m *MockConfigStore) GetExecutionsByStatuses(ctx context.Context, statuses []string, limit int) ([]config.PurchaseExecution, error) { m.record("GetExecutionsByStatuses", ctx, statuses, limit)