diff --git a/.golangci.yml b/.golangci.yml index ac5a3bcc..2db43f74 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -131,7 +131,7 @@ linters: # existing UK convention. Allow both spellings in the files listed below. - linters: - misspell - path: (internal/config/(store_postgres|store_postgres_pgxmock_test|store_postgres_ladder|types|interfaces)|pkg/ladder/(store|types_test)|internal/api/(handler_history|handler_ri_exchange(_test)?|router_handlers_test|coverage_extras_test)|internal/mocks/stores|internal/purchase/scheduled_fire_test)\.go + path: (internal/config/(store_postgres|store_postgres_history_page|store_postgres_pgxmock_test|store_postgres_ladder|types|interfaces)|pkg/ladder/(store|types_test)|internal/api/(handler_history|handler_ri_exchange(_test)?|router_handlers_test|coverage_extras_test)|internal/mocks/stores|internal/purchase/scheduled_fire_test)\.go text: "cancelled|initialised" paths: - third_party$ diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 76ec0b71..bad9c2de 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -216,6 +216,14 @@ func (m *mockConfigStore) GetPlannedExecutions(ctx context.Context, statuses []s return nil, nil } +func (m *mockConfigStore) ListHistoryExecutionsPage(ctx context.Context, q config.HistoryExecutionsPageQuery) ([]config.PurchaseExecution, error) { + return nil, nil +} + +func (m *mockConfigStore) ListPurchaseHistoryPage(ctx context.Context, filter config.PurchaseHistoryFilter, after *config.HistoryPageKey) ([]config.PurchaseHistoryPageRow, error) { + return nil, nil +} + func (m *mockConfigStore) GetStaleApprovedExecutions(ctx context.Context, olderThan time.Duration) ([]config.PurchaseExecution, error) { return nil, nil } diff --git a/internal/api/handler_history.go b/internal/api/handler_history.go index 40e9fb67..e79525ee 100644 --- a/internal/api/handler_history.go +++ b/internal/api/handler_history.go @@ -4,7 +4,6 @@ package api import ( "context" "fmt" - "slices" "sort" "strconv" "time" @@ -38,24 +37,25 @@ func (h *Handler) getHistory(ctx context.Context, req *events.LambdaFunctionURLR // folded in here so fetchPurchaseHistory sees a single unified account set. h.resolveHistoryAccountFilter(ctx, &filters) - completed, err := h.fetchPurchaseHistory(ctx, filters) + // The paged stream: completed purchases and terminal executions, newest + // first, filters.Limit rows the caller may see, starting after + // filters.Cursor (issue #674, see handler_history_page.go). + page, err := h.fetchHistoryPage(ctx, session, &filters) if err != nil { return nil, err } - for i := range completed { - completed[i].Status = "completed" - } - // Pull non-completed executions so the user can see (and cancel) their - // own in-flight approvals without waiting for the email, AND see the - // ones that never made it out (failed) or that sat unapproved past the - // 7-day cutoff (expired). Executions live in a separate table - // (purchase_executions) from the completed purchase_history rows, so we - // merge after the fact. A failure to list executions must not hide - // completed history — log, skip, continue. 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). + // Pull the actionable executions so the user can see (and cancel) their + // own in-flight approvals without waiting for the email. Executions live + // in a separate table (purchase_executions) from the completed + // purchase_history rows, so we merge after the fact. They are not paged: + // the first page lists the whole class (up to its own cap) whatever the + // 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). // // Stale pending/notified executions are expired AFTER the response is // assembled so the GET handler is a pure read (issue #1032): the caller @@ -63,39 +63,49 @@ func (h *Handler) getHistory(ctx context.Context, req *events.LambdaFunctionURLR // status. On servers the transitions fire in a goroutine; on Lambda they // run synchronously before returning because the execution environment // freezes once the response is out (issue #1170). - extra, staleExecs, executionsCapped := h.fetchExecutionsAsHistory(ctx, filters) - h.expireStaleExecutions(staleExecs) + var extra []config.PurchaseHistoryRecord + actionableIncomplete := false + if filters.Cursor == nil { + var staleExecs []config.PurchaseExecution + extra, staleExecs, actionableIncomplete = h.fetchExecutionsAsHistory(ctx, filters) + h.expireStaleExecutions(staleExecs) + extra, err = h.filterPurchaseHistoryByAllowedAccounts(ctx, session, extra) + if err != nil { + return nil, err + } + } - all := make([]config.PurchaseHistoryRecord, 0, len(completed)+len(extra)) - all = append(all, completed...) + all := make([]config.PurchaseHistoryRecord, 0, len(page.rows)+len(extra)) + all = append(all, page.rows...) all = append(all, extra...) - all, err = h.filterPurchaseHistoryByAllowedAccounts(ctx, session, all) - if err != nil { - return nil, err - } - // Newest first across both sources so pending items don't get buried at // the bottom just because their synthetic records were appended last. + // The stream rows are already in order and the sort is stable, so their + // relative order (and the cursor's position) is unchanged. sort.SliceStable(all, func(i, j int) bool { return all[i].Timestamp.After(all[j].Timestamp) }) - // Truncation is judged on the raw fetches (before the in-memory and scope - // filters) because the cap drops rows before they run. It is a single - // boolean: a count would reveal other tenants' row volume to a scoped user. + // Truncated is a single boolean: a count would reveal other tenants' row + // volume to a scoped user. has_more is judged on the rows the caller may + // see, so it is never set by rows outside the caller's scope. return HistoryResponse{ - Summary: summarizePurchaseHistory(all), - Purchases: all, - Truncated: len(completed) >= filters.Limit || executionsCapped, - Limit: filters.Limit, - ExecutionsLimit: config.DefaultListLimit, + Summary: summarizePurchaseHistory(all), + Purchases: all, + Truncated: page.nextCursor != "" || page.incomplete || actionableIncomplete, + Limit: filters.Limit, + ExecutionsLimit: filters.Limit, + ActionableExecutionsLimit: config.MaxListLimit, + HasMore: page.nextCursor != "", + NextCursor: page.nextCursor, }, nil } // actionableExecutionStatuses are the PurchaseExecution statuses a user can still // act on or that are in flight. They are fetched separately from the terminal -// class so newer terminal rows cannot push them out of the capped listing. +// class so newer terminal rows cannot push them out of the capped listing, and +// they are not paged: the first page lists them all (issue #674). // // Issue #372 assumed approval executes synchronously (approved -> completed/failed inside one // HTTP request) and so excluded "approved"; issue #621 showed that @@ -114,9 +124,9 @@ func (h *Handler) getHistory(ctx context.Context, req *events.LambdaFunctionURLR var actionableExecutionStatuses = []string{"pending", "notified", "scheduled", "approved", "running", "paused"} // terminalExecutionStatuses are the finished PurchaseExecution statuses the -// History view loads alongside completed purchases. +// History view pages through alongside completed purchases (fetchHistoryPage). // -// "completed" is also loaded, but fetchExecutionsAsHistory synthesizes a +// "completed" is also loaded, but the stream synthesizes a // row for it ONLY when the execution carries a non-empty Error — the // audit-gap case where the purchase succeeded but its purchase_history // write failed (issue #621 secondary path). A normal completed execution @@ -144,7 +154,7 @@ var terminalExecutionStatuses = []string{"completed", "partially_completed", "fa // approval link is stale. const approvalExpiryWindow = 7 * 24 * time.Hour -// fetchExecutionsAsHistory returns non-completed PurchaseExecutions as +// fetchExecutionsAsHistory returns the actionable PurchaseExecutions as // synthetic PurchaseHistoryRecord entries so the /api/history response is a // single flat list. Execution-level aggregates (TotalUpfrontCost / // EstimatedSavings) become the row's cost fields; provider/service collapse @@ -155,43 +165,28 @@ const approvalExpiryWindow = 7 * 24 * time.Hour // call returns (see expireStaleExecutions) so the GET is a pure read: // the response rows carry the pre-transition status and the next request // sees the updated state. A listing error is logged and skipped — completed -// history must still render. +// 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 third return value reports that either class reached its cap (actionable -// at MaxListLimit, terminal at DefaultListLimit), so older executions may have -// been dropped before the filters ran. +// 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) { - actionable := h.fetchExecutionsByClass(ctx, actionableExecutionStatuses, config.MaxListLimit) - terminal := h.fetchExecutionsByClass(ctx, terminalExecutionStatuses, config.DefaultListLimit) - executions := make([]config.PurchaseExecution, 0, len(actionable)+len(terminal)) - executions = append(executions, actionable...) - executions = append(executions, terminal...) + executions, ok := h.fetchExecutionsByClass(ctx, actionableExecutionStatuses, config.MaxListLimit) + incomplete := !ok || len(executions) >= config.MaxListLimit if len(executions) == 0 { - return nil, nil, false + return nil, nil, incomplete } approver := h.resolvePendingApproverEmail(ctx) userEmailCache := h.resolveUserEmails(ctx, executions) out := make([]config.PurchaseHistoryRecord, 0, len(executions)) var staleExecs []config.PurchaseExecution - // Terminal-class rows surfaced; drives the terminal cap signal. - terminalSurfaced := 0 for _rvc := range executions { exec := executions[_rvc] - // Dedup: a normal completed execution is already represented by its - // purchase_history rows. Skip it here so it shows exactly once. Only - // completed executions carrying an audit-gap Error (history write - // failed after a successful purchase, issue #621) are synthesized — - // those have no purchase_history row to collide with. - if exec.Status == "completed" && exec.Error == "" { - continue - } - if isTerminalExecutionStatus(exec.Status) { - terminalSurfaced++ - } // 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 @@ -202,29 +197,22 @@ func (h *Handler) fetchExecutionsAsHistory(ctx context.Context, filters historyF if !filters.matchesExecution(exec) { continue } - var createdByEmail string - if exec.CreatedByUserID != nil { - createdByEmail = userEmailCache[*exec.CreatedByUserID] - } - out = append(out, executionToHistoryRow(exec, approver, createdByEmail)) + out = append(out, executionToHistoryRow(exec, approver, createdByEmail(userEmailCache, &exec))) } - capped := len(actionable) >= config.MaxListLimit || terminalSurfaced >= config.DefaultListLimit - return out, staleExecs, capped + return out, staleExecs, incomplete } // fetchExecutionsByClass lists one status class. A listing error is logged and -// the class skipped so the other class and completed history still render. -func (h *Handler) fetchExecutionsByClass(ctx context.Context, statuses []string, limit int) []config.PurchaseExecution { +// the class skipped (ok=false) so the other class and completed history still +// render; the caller flags the response as incomplete so the hidden rows are +// not silent (issue #686). +func (h *Handler) fetchExecutionsByClass(ctx context.Context, statuses []string, limit int) ([]config.PurchaseExecution, bool) { rows, err := h.config.GetExecutionsByStatuses(ctx, statuses, limit) if err != nil { logging.Warnf("history: failed to load executions with statuses %v: %v", statuses, err) - return nil + return nil, false } - return rows -} - -func isTerminalExecutionStatus(status string) bool { - return slices.Contains(terminalExecutionStatuses, status) + return rows, true } // isStaleExecution reports whether the execution is a pending/notified @@ -720,6 +708,9 @@ type historyFilters struct { AccountIDs []string Limit int HasDate bool + // Cursor is the stream position the page starts after (issue #674); nil + // for the first page. + Cursor *config.HistoryPageKey } // parseHistoryFilters validates and normalises the /api/history query string. @@ -769,6 +760,12 @@ func parseHistoryFilters(params map[string]string) (historyFilters, error) { } f.Limit = limit + cursor, pagingErr := parseHistoryPaging(params) + if pagingErr != nil { + return f, pagingErr + } + f.Cursor = cursor + return f, nil } @@ -971,32 +968,28 @@ func stringInSlice(needle string, haystack []string) bool { // direct-execute/ambient row. Grouping the external ids by provider keeps a // reused external number across providers from leaking the wrong rows. // -// The no-filter case still uses GetAllPurchaseHistory as a fast path. The -// legacy singular account_id (LegacyAccountID) is resolved to the same dual- -// column inputs by the caller (getHistory) and arrives here folded into -// AccountIDs/ExternalIDsByProvider, so there is no longer a separate -// single-column legacy path to coerce identifiers through. -func (h *Handler) fetchPurchaseHistory(ctx context.Context, filters historyFilters) ([]config.PurchaseHistoryRecord, error) { - noFilters := !filters.HasDate && filters.Provider == "" && - len(filters.AccountIDs) == 0 && len(filters.ExternalIDsByProvider) == 0 - if noFilters { - return h.config.GetAllPurchaseHistory(ctx, filters.Limit) - } - +// It reads one keyset batch of the History stream: up to limit rows after +// `after` (nil starts at the newest row), in stream order (issue #674). An +// empty filter set selects every row. The legacy singular account_id +// (LegacyAccountID) is resolved to the same dual-column inputs by the caller +// (getHistory) and arrives here folded into AccountIDs/ExternalIDsByProvider, +// so there is no longer a separate single-column legacy path to coerce +// identifiers through. +func (h *Handler) fetchPurchaseHistory(ctx context.Context, filters *historyFilters, after *config.HistoryPageKey, limit int) ([]config.PurchaseHistoryPageRow, error) { var startPtr, endPtr *time.Time if filters.HasDate { s, e := filters.Start, filters.End startPtr = &s endPtr = &e } - return h.config.GetPurchaseHistoryFiltered(ctx, config.PurchaseHistoryFilter{ + return h.config.ListPurchaseHistoryPage(ctx, config.PurchaseHistoryFilter{ Provider: filters.Provider, AccountIDs: filters.AccountIDs, ExternalIDsByProvider: filters.ExternalIDsByProvider, Start: startPtr, End: endPtr, - Limit: filters.Limit, - }) + Limit: limit, + }, after) } // resolveHistoryAccountFilter populates filters.AccountIDs / diff --git a/internal/api/handler_history_actionable_starvation_test.go b/internal/api/handler_history_actionable_starvation_test.go index 29befc3b..5a277b6a 100644 --- a/internal/api/handler_history_actionable_starvation_test.go +++ b/internal/api/handler_history_actionable_starvation_test.go @@ -30,13 +30,13 @@ func isActionableCall(statuses []string) bool { return slices.Equal(statuses, actionableExecutionStatuses) } -func isTerminalCall(statuses []string) bool { - return slices.Equal(statuses, terminalExecutionStatuses) +func isTerminalPage(q config.HistoryExecutionsPageQuery) bool { + return slices.Equal(q.Statuses, terminalExecutionStatuses) } func newStarvationStore(ctx context.Context) *MockConfigStore { store := new(MockConfigStore) - store.On("GetAllPurchaseHistory", ctx, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) + store.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) store.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{}, nil).Maybe() return store } @@ -57,6 +57,7 @@ func TestHandler_getHistory_PendingNotStarvedByNewerTerminalRows(t *testing.T) { pending := executionsWithStatus(1, "pending", now.Add(-time.Hour), "pending") store := newStarvationStore(ctx) store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(append(execs, pending...), nil) + store.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(append(execs, pending...), nil) body := runStarvationHistory(t, ctx, store) @@ -65,6 +66,8 @@ func TestHandler_getHistory_PendingNotStarvedByNewerTerminalRows(t *testing.T) { ids[p.(map[string]any)["purchase_id"].(string)] = true } assert.True(t, ids["pending-0"], "older pending row must survive 150 newer failed rows") + assert.Len(t, ids, config.DefaultListLimit+1, "a page of failed rows plus the pending row") + assert.Equal(t, true, body["has_more"], "the older failed rows are on the next page") summary := body["summary"].(map[string]any) assert.EqualValues(t, 1, summary["total_pending"]) } @@ -87,11 +90,14 @@ func TestHandler_getHistory_FetchesDisjointClassesWithTheirLimits(t *testing.T) ctx := context.Background() store := newStarvationStore(ctx) store.On("GetExecutionsByStatuses", ctx, mock.MatchedBy(isActionableCall), config.MaxListLimit).Return([]config.PurchaseExecution{}, nil).Once() - store.On("GetExecutionsByStatuses", ctx, mock.MatchedBy(isTerminalCall), config.DefaultListLimit).Return([]config.PurchaseExecution{}, nil).Once() + store.On("ListHistoryExecutionsPage", ctx, mock.MatchedBy(func(q config.HistoryExecutionsPageQuery) bool { + return isTerminalPage(q) && q.Limit == historyStreamBatch(config.DefaultListLimit) && q.After == nil + })).Return([]config.PurchaseExecution{}, nil).Once() runStarvationHistory(t, ctx, store) - store.AssertNumberOfCalls(t, "GetExecutionsByStatuses", 2) + store.AssertNumberOfCalls(t, "GetExecutionsByStatuses", 1) + store.AssertNumberOfCalls(t, "ListHistoryExecutionsPage", 1) for _, s := range actionableExecutionStatuses { assert.NotContains(t, terminalExecutionStatuses, s, "classes must be disjoint") } @@ -119,17 +125,47 @@ func TestHandler_getHistory_ClassListingErrorKeepsOtherClass(t *testing.T) { terminalRet = mock.Arguments{nil, errors.New("terminal listing failed")} } store.On("GetExecutionsByStatuses", ctx, mock.MatchedBy(isActionableCall), mock.Anything).Return(actionableRet...) - store.On("GetExecutionsByStatuses", ctx, mock.MatchedBy(isTerminalCall), mock.Anything).Return(terminalRet...) + store.On("ListHistoryExecutionsPage", ctx, mock.MatchedBy(isTerminalPage)).Return(terminalRet...) body := runStarvationHistory(t, ctx, store) purchases := body["purchases"].([]any) require.Len(t, purchases, 1) assert.Equal(t, tt.wantID, purchases[0].(map[string]any)["purchase_id"]) + assert.Equal(t, true, body["truncated"], "a failed class listing hides rows, so the response must flag incomplete results") }) } } +// TestHandler_getHistory_BothClassListingsFailSetTruncated covers the +// no-executions early return: when both class listings fail the response +// carries only completed history and must still flag incomplete results. +func TestHandler_getHistory_BothClassListingsFailSetTruncated(t *testing.T) { + ctx := context.Background() + store := newStarvationStore(ctx) + store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(nil, errors.New("listing failed")) + store.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(nil, errors.New("listing failed")) + + body := runStarvationHistory(t, ctx, store) + + assert.Empty(t, body["purchases"]) + assert.Equal(t, true, body["truncated"]) +} + +// TestHandler_getHistory_EmptyClassListingsNotTruncated guards the other side +// of the listing-error signal: two successful but empty listings are a +// complete result and must not raise the banner. +func TestHandler_getHistory_EmptyClassListingsNotTruncated(t *testing.T) { + ctx := context.Background() + store := newStarvationStore(ctx) + store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) + store.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return([]config.PurchaseExecution{}, nil) + + body := runStarvationHistory(t, ctx, store) + + assert.Equal(t, false, body["truncated"]) +} + func TestHandler_getHistory_TruncatedPerClass(t *testing.T) { now := time.Now() tests := []struct { @@ -138,7 +174,8 @@ func TestHandler_getHistory_TruncatedPerClass(t *testing.T) { terminal int wantTruncated bool }{ - {"terminal class at cap", 0, config.DefaultListLimit, true}, + {"terminal class past the page", 0, config.DefaultListLimit + 1, true}, + {"terminal class exactly fills the page", 0, config.DefaultListLimit, false}, {"both under cap", config.DefaultListLimit, config.DefaultListLimit - 1, false}, {"actionable class at cap", config.MaxListLimit, 0, true}, } @@ -149,6 +186,7 @@ func TestHandler_getHistory_TruncatedPerClass(t *testing.T) { executionsWithStatus(tt.terminal, "failed", now, "failed")...) store := newStarvationStore(ctx) store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(rows, nil) + store.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(rows, nil) body := runStarvationHistory(t, ctx, store) diff --git a/internal/api/handler_history_page.go b/internal/api/handler_history_page.go new file mode 100644 index 00000000..81240560 --- /dev/null +++ b/internal/api/handler_history_page.go @@ -0,0 +1,385 @@ +package api + +import ( + "context" + "encoding/base64" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/google/uuid" +) + +// History paging (issue #674). +// +// GET /api/history lists two kinds of rows: +// +// - the actionable executions (pending approvals and in-flight purchases), +// listed in full on the first page only and never paged, so paging can +// never push a pending approval off the first page or hide it on a later +// one; +// - the paged stream: completed purchase_history rows and terminal +// executions merged newest first, `limit` rows per page. +// +// The stream is paged with a keyset cursor, not an offset. Its order is +// config.HistoryPageKey's (timestamp DESC, source DESC, row id DESC), which is +// unique, and next_cursor is the key of the last row on the page. A later page +// resumes strictly after it, so rows written between two page loads (new +// purchases land at the top) never shift, repeat or skip a row on the later +// page, and a page costs the same however deep it is. An offset would do +// neither: the account scope and the execution filters run in Go after the +// SQL read, so a SQL offset counts rows the caller never sees, and every +// insert ahead of the offset moves the page. +// +// The cursor names only a row the caller was shown, so it carries nothing +// about rows outside the caller's scope, and the scope is applied again on +// every page: a crafted cursor still only returns rows the caller may see. + +const ( + historyCursorVersion = "v1" + // maxHistoryCursorLen bounds the cursor before it is decoded. A valid one + // is about 80 characters. + maxHistoryCursorLen = 256 +) + +var errInvalidHistoryCursor = errors.New("invalid cursor") + +// encodeHistoryCursor renders a stream position as an opaque URL-safe token. +// The timestamp is kept to the microsecond, PostgreSQL's precision, so the +// key read back from the database round-trips exactly. +func encodeHistoryCursor(k config.HistoryPageKey) string { + raw := fmt.Sprintf("%s.%d.%d.%s", historyCursorVersion, k.Source, k.Timestamp.UnixMicro(), k.ID) + return base64.RawURLEncoding.EncodeToString([]byte(raw)) +} + +// decodeHistoryCursor parses a token from encodeHistoryCursor. It rejects +// anything else, including an ID that is not a canonical lowercase UUID: the +// store compares it with PostgreSQL's uuid ordering. +func decodeHistoryCursor(s string) (*config.HistoryPageKey, error) { + if len(s) > maxHistoryCursorLen { + return nil, errInvalidHistoryCursor + } + raw, err := base64.RawURLEncoding.DecodeString(s) + if err != nil { + return nil, errInvalidHistoryCursor + } + parts := strings.Split(string(raw), ".") + if len(parts) != 4 || parts[0] != historyCursorVersion { + return nil, errInvalidHistoryCursor + } + src, err := parseHistoryCursorSource(parts[1]) + if err != nil { + return nil, err + } + ts, err := parseHistoryCursorTime(parts[2]) + if err != nil { + return nil, err + } + if id, err := uuid.Parse(parts[3]); err != nil || id.String() != parts[3] { + return nil, errInvalidHistoryCursor + } + return &config.HistoryPageKey{Timestamp: ts, Source: src, ID: parts[3]}, nil +} + +func parseHistoryCursorSource(s string) (config.HistorySource, error) { + n, err := strconv.Atoi(s) + src := config.HistorySource(n) + if err != nil || (src != config.HistorySourceExecution && src != config.HistorySourcePurchase) { + return 0, errInvalidHistoryCursor + } + return src, nil +} + +// parseHistoryCursorTime reads the Unix-microsecond timestamp, limited to +// years PostgreSQL's timestamptz can compare against. +func parseHistoryCursorTime(s string) (time.Time, error) { + micros, err := strconv.ParseInt(s, 10, 64) + if err != nil { + return time.Time{}, errInvalidHistoryCursor + } + ts := time.UnixMicro(micros).UTC() + if ts.Year() < 1 || ts.Year() > 9999 { + return time.Time{}, errInvalidHistoryCursor + } + return ts, nil +} + +// parseHistoryPaging reads the paging parameters. An offset is refused rather +// than ignored: a client written against an offset API would otherwise get +// the first page back for every offset and never stop asking for more. +func parseHistoryPaging(params map[string]string) (*config.HistoryPageKey, error) { + if params["offset"] != "" { + return nil, NewClientError(400, "offset is not supported: page with cursor, passing the next_cursor value from the previous response") + } + s := params["cursor"] + if s == "" { + return nil, nil + } + cursor, err := decodeHistoryCursor(s) + if err != nil { + return nil, NewClientError(400, "invalid cursor: pass the next_cursor value from a previous /api/history response unchanged") + } + return cursor, nil +} + +// historyStreamBatch is how many rows each stream source reads per query: at +// least one more than the page so has_more needs no extra query when nothing +// is filtered out, and at least DefaultListLimit so a scoped caller whose +// rows are sparse needs fewer round trips. +func historyStreamBatch(limit int) int { + return min(max(limit+1, config.DefaultListLimit), config.MaxListLimit) +} + +// historyStreamRow is one row of the paged stream: its stream position, its +// History projection, and whether the request's filters and the caller's +// account scope let the caller see it. Hidden rows still advance the stream. +type historyStreamRow struct { + key config.HistoryPageKey + row config.PurchaseHistoryRecord + visible bool +} + +// historyStreamSource reads one table's half of the stream in keyset batches. +type historyStreamSource struct { + fetch func(ctx context.Context, after *config.HistoryPageKey) ([]historyStreamRow, error) + buf []historyStreamRow + after *config.HistoryPageKey + batch int + done bool +} + +// head returns the source's next row without consuming it, reading the next +// batch when the buffer is empty. nil means the source is exhausted. +func (s *historyStreamSource) head(ctx context.Context) (*historyStreamRow, error) { + if len(s.buf) == 0 && !s.done { + rows, err := s.fetch(ctx, s.after) + if err != nil { + s.done = true + return nil, err + } + if len(rows) > 0 { + last := rows[len(rows)-1].key + if s.after != nil && !s.after.SortsBefore(rows[0].key) { + s.done = true + return nil, fmt.Errorf("history stream: keyset read did not advance past %v", *s.after) + } + s.after = &last + } + s.buf = rows + s.done = len(rows) < s.batch + } + if len(s.buf) == 0 { + return nil, nil + } + return &s.buf[0], nil +} + +// historyStream merges the purchase_history and terminal-execution sources in +// stream order. +type historyStream struct { + h *Handler + session *Session + filters *historyFilters + purchases historyStreamSource + executions historyStreamSource + approver string + approverResolved bool + // executionsFailed reports that a terminal-execution read failed. The + // stream then continues with purchase_history rows only, as History did + // when a class listing failed (issue #686), and the response is flagged + // truncated. + executionsFailed bool +} + +func newHistoryStream(h *Handler, session *Session, filters *historyFilters) *historyStream { + batch := historyStreamBatch(filters.Limit) + st := &historyStream{h: h, session: session, filters: filters} + st.purchases = historyStreamSource{fetch: st.fetchPurchases, after: filters.Cursor, batch: batch} + st.executions = historyStreamSource{fetch: st.fetchTerminalExecutions, after: filters.Cursor, batch: batch} + return st +} + +// next consumes and returns the next row of the merged stream, or nil at the +// end. A purchase_history read error fails the request, as it did before +// paging; a terminal-execution read error is logged and that source dropped. +func (st *historyStream) next(ctx context.Context) (*historyStreamRow, error) { + p, err := st.purchases.head(ctx) + if err != nil { + return nil, err + } + e, err := st.executions.head(ctx) + if err != nil { + logging.Warnf("history: failed to load terminal executions: %v", err) + st.executionsFailed = true + } + var src *historyStreamSource + switch { + case p != nil && (e == nil || p.key.SortsBefore(e.key)): + src = &st.purchases + case e != nil: + src = &st.executions + default: + return nil, nil + } + row := src.buf[0] + src.buf = src.buf[1:] + return &row, nil +} + +func (st *historyStream) fetchPurchases(ctx context.Context, after *config.HistoryPageKey) ([]historyStreamRow, error) { + recs, err := st.h.fetchPurchaseHistory(ctx, st.filters, after, st.purchases.batch) + if err != nil { + return nil, err + } + rows := make([]historyStreamRow, len(recs)) + for i := range recs { + rows[i] = historyStreamRow{key: recs[i].Key(), row: recs[i].PurchaseHistoryRecord, visible: true} + rows[i].row.Status = "completed" + } + return rows, st.h.markOutOfScope(ctx, st.session, rows) +} + +// fetchTerminalExecutions reads a batch of terminal executions and projects +// the ones matching the request's filters into History rows. The date bound +// is pushed into SQL as well as checked by matchesExecution; provider and +// account are checked in Go only, like the actionable class. +func (st *historyStream) fetchTerminalExecutions(ctx context.Context, after *config.HistoryPageKey) ([]historyStreamRow, error) { + q := config.HistoryExecutionsPageQuery{Statuses: terminalExecutionStatuses, After: after, Limit: st.executions.batch} + if st.filters.HasDate { + start, end := st.filters.Start, st.filters.End + q.Start, q.End = &start, &end + } + execs, err := st.h.config.ListHistoryExecutionsPage(ctx, q) + if err != nil { + return nil, err + } + rows := make([]historyStreamRow, len(execs)) + matching := make([]config.PurchaseExecution, 0, len(execs)) + for i := range execs { + rows[i].key = config.HistoryPageKey{Timestamp: execs[i].ScheduledDate, Source: config.HistorySourceExecution, ID: execs[i].ExecutionID} + // Clean completed executions are excluded by the store; checked here + // too so one can never show twice (it renders via purchase_history). + rows[i].visible = (execs[i].Status != "completed" || execs[i].Error != "") && st.filters.matchesExecution(execs[i]) + if rows[i].visible { + matching = append(matching, execs[i]) + } + } + if len(matching) == 0 { + return rows, nil + } + approver := st.approverEmail(ctx) + emails := st.h.resolveUserEmails(ctx, matching) + for i := range execs { + if rows[i].visible { + rows[i].row = executionToHistoryRow(execs[i], approver, createdByEmail(emails, &execs[i])) + } + } + return rows, st.h.markOutOfScope(ctx, st.session, rows) +} + +// approverEmail resolves the approver address once per request. +func (st *historyStream) approverEmail(ctx context.Context) string { + if !st.approverResolved { + st.approver = st.h.resolvePendingApproverEmail(ctx) + st.approverResolved = true + } + return st.approver +} + +func createdByEmail(emails map[string]string, exec *config.PurchaseExecution) string { + if exec.CreatedByUserID == nil { + return "" + } + return emails[*exec.CreatedByUserID] +} + +// markOutOfScope hides every visible row the caller's account scope does not +// allow. The scope rule has one home, filterPurchaseHistoryByAllowedAccounts; +// this runs the batch's visible rows through it and matches the rows it keeps +// back to the batch in order. The filter keeps an order-preserving subset and +// decides each row from that row alone, and sameHistoryRow compares every +// field it reads plus the row's identity, so the match is exact. +func (h *Handler) markOutOfScope(ctx context.Context, session *Session, rows []historyStreamRow) error { + candidates := make([]config.PurchaseHistoryRecord, 0, len(rows)) + index := make([]int, 0, len(rows)) + for i := range rows { + if rows[i].visible { + candidates = append(candidates, rows[i].row) + index = append(index, i) + } + } + if len(candidates) == 0 { + return nil + } + kept, err := h.filterPurchaseHistoryByAllowedAccounts(ctx, session, candidates) + if err != nil { + return err + } + j := 0 + for n, i := range index { + if j < len(kept) && sameHistoryRow(&kept[j], &candidates[n]) { + j++ + continue + } + rows[i].visible = false + } + return nil +} + +// sameHistoryRow compares the fields the account-scope check reads (account, +// cloud account, provider, creator) and the row's identity. +func sameHistoryRow(a, b *config.PurchaseHistoryRecord) bool { + return a.PurchaseID == b.PurchaseID && a.Timestamp.Equal(b.Timestamp) && + a.AccountID == b.AccountID && a.Provider == b.Provider && + a.CreatedByUserID == b.CreatedByUserID && sameOptionalString(a.CloudAccountID, b.CloudAccountID) +} + +func sameOptionalString(a, b *string) bool { + if a == nil || b == nil { + return a == nil && b == nil + } + return *a == *b +} + +// historyPage is one page of the stream. +type historyPage struct { + rows []config.PurchaseHistoryRecord + // nextCursor is set when more visible rows follow the page. + nextCursor string + // incomplete reports that terminal executions could not be read. + incomplete bool +} + +// fetchHistoryPage returns the next filters.Limit rows of the stream the +// caller may see, starting after filters.Cursor. It reads one row past the +// page so has_more is exact: has_more is never set when the rows after the +// page are all outside the caller's scope or filters. +func (h *Handler) fetchHistoryPage(ctx context.Context, session *Session, filters *historyFilters) (historyPage, error) { + st := newHistoryStream(h, session, filters) + var page historyPage + var last config.HistoryPageKey + for { + r, err := st.next(ctx) + if err != nil { + return historyPage{}, err + } + if r == nil { + break + } + if !r.visible { + continue + } + if len(page.rows) == filters.Limit { + page.nextCursor = encodeHistoryCursor(last) + break + } + page.rows = append(page.rows, r.row) + last = r.key + } + page.incomplete = st.executionsFailed + return page, nil +} diff --git a/internal/api/handler_history_page_integration_test.go b/internal/api/handler_history_page_integration_test.go new file mode 100644 index 00000000..ca960c71 --- /dev/null +++ b/internal/api/handler_history_page_integration_test.go @@ -0,0 +1,330 @@ +//go:build integration + +package api + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "sort" + "strconv" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/aws/aws-lambda-go/events" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// historyPagingEnv is two tenants' History rows in a real database, with the +// stream key of every row so a test can compute the expected page order. +type historyPagingEnv struct { + pg *testhelpers.PostgresContainer + store *config.PostgresStore + tenantA *config.CloudAccount + tenantB *config.CloudAccount + // stream maps purchase_id (a purchase_history row) or execution_id (a + // terminal execution) to its stream key, per tenant ("" for ambient rows). + stream map[string]config.HistoryPageKey + tenant map[string]string + owner map[string]string + // actionable are the pending executions' ids, per tenant. + actionable map[string]string +} + +func setupHistoryPagingEnv(ctx context.Context, t *testing.T) *historyPagingEnv { + t.Helper() + pg := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath()) + env := &historyPagingEnv{ + pg: pg, store: config.NewPostgresStore(pg.DB), + stream: map[string]config.HistoryPageKey{}, tenant: map[string]string{}, + owner: map[string]string{}, actionable: map[string]string{}, + } + env.tenantA = &config.CloudAccount{Name: "Paging Tenant A", Provider: "aws", ExternalID: "ext-paging-a", Enabled: true} + env.tenantB = &config.CloudAccount{Name: "Paging Tenant B", Provider: "aws", ExternalID: "ext-paging-b", Enabled: true} + require.NoError(t, env.store.CreateCloudAccount(ctx, env.tenantA)) + require.NoError(t, env.store.CreateCloudAccount(ctx, env.tenantB)) + return env +} + +// addUser inserts a user so executions can reference it as their creator. +func (e *historyPagingEnv) addUser(ctx context.Context, t *testing.T, id string) { + t.Helper() + _, err := e.pg.DB.Pool().Exec(ctx, `INSERT INTO users (id, email, password_hash, group_ids) + SELECT $1, $2, 'x', ARRAY[id] FROM groups ORDER BY name LIMIT 1`, id, fmt.Sprintf("paging-%s@example.com", id)) + require.NoError(t, err) +} + +// getHistoryPage calls getHistory as the token's user and round-trips the +// response through JSON, the shape a client sees. +func getHistoryPage(ctx context.Context, t *testing.T, h *Handler, token string, params map[string]string) HistoryResponse { + t.Helper() + req := &events.LambdaFunctionURLRequest{ + Headers: map[string]string{"Authorization": "Bearer " + token}, + QueryStringParameters: params, + } + out, err := h.getHistory(ctx, req, params) + require.NoError(t, err) + raw, err := json.Marshal(out) + require.NoError(t, err) + var resp HistoryResponse + require.NoError(t, json.Unmarshal(raw, &resp)) + return resp +} + +func (e *historyPagingEnv) account(tenant string) *config.CloudAccount { + if tenant == "a" { + return e.tenantA + } + return e.tenantB +} + +// addPurchase saves a purchase_history row; legacy rows carry only the +// external account number, as direct-execute rows do. +func (e *historyPagingEnv) addPurchase(ctx context.Context, t *testing.T, tenant string, at time.Time, legacy bool) string { + t.Helper() + acct := e.account(tenant) + up := 10.0 + rec := &config.PurchaseHistoryRecord{ + AccountID: acct.ExternalID, PurchaseID: "ph-" + uuid.NewString(), Timestamp: at, Provider: "aws", + Service: "ec2", Region: "us-east-1", ResourceType: "t3.micro", Count: 1, Term: 1, + Payment: "no-upfront", UpfrontCost: &up, EstimatedSavings: 1, + } + if !legacy { + rec.CloudAccountID = &acct.ID + } + require.NoError(t, e.store.SavePurchaseHistory(ctx, rec)) + var rowID string + require.NoError(t, e.pg.DB.Pool().QueryRow(ctx, + `SELECT id::text FROM purchase_history WHERE purchase_id = $1`, rec.PurchaseID).Scan(&rowID)) + e.stream[rec.PurchaseID] = config.HistoryPageKey{Timestamp: at, Source: config.HistorySourcePurchase, ID: rowID} + e.tenant[rec.PurchaseID] = tenant + return rec.PurchaseID +} + +// addExecution saves an execution. tenant "" makes it ambient (no account on +// it or its recommendation), created by owner. +func (e *historyPagingEnv) addExecution(ctx context.Context, t *testing.T, tenant, owner, status string, at time.Time) string { + t.Helper() + exec := &config.PurchaseExecution{ + ExecutionID: uuid.NewString(), IdempotencyKey: uuid.NewString(), Status: status, Error: "boom", + ScheduledDate: at, + } + rec := config.RecommendationRecord{Provider: "aws", Service: "ec2", Region: "us-east-1"} + if tenant != "" { + id := e.account(tenant).ID + exec.CloudAccountID = &id + rec.CloudAccountID = &id + } + if owner != "" { + exec.CreatedByUserID = &owner + e.owner[exec.ExecutionID] = owner + } + exec.Recommendations = []config.RecommendationRecord{rec} + require.NoError(t, e.store.SavePurchaseExecution(ctx, exec)) + if slices.Contains(actionableExecutionStatuses, status) { + e.actionable[exec.ExecutionID] = tenant + return exec.ExecutionID + } + e.stream[exec.ExecutionID] = config.HistoryPageKey{Timestamp: at, Source: config.HistorySourceExecution, ID: exec.ExecutionID} + e.tenant[exec.ExecutionID] = tenant + return exec.ExecutionID +} + +// seed writes a tenant's rows: purchases and terminal executions sharing +// timestamps within and across the two tables, plus one pending approval +// older than all of them. +func (e *historyPagingEnv) seed(ctx context.Context, t *testing.T, tenant string, base time.Time) { + t.Helper() + for i := 0; i < 14; i++ { + e.addPurchase(ctx, t, tenant, base.Add(-time.Duration(i/3)*time.Hour), i%4 == 0) + } + statuses := []string{"failed", "expired", config.StatusCanceled, "partially_completed", "completed"} + for i := 0; i < 10; i++ { + e.addExecution(ctx, t, tenant, "", statuses[i%len(statuses)], base.Add(-time.Duration(i/3)*time.Hour)) + } + e.addExecution(ctx, t, tenant, "", "pending", base.Add(-48*time.Hour)) +} + +// expected returns the ids of the stream rows include keeps, in stream order. +func (e *historyPagingEnv) expected(include func(id string) bool) []string { + var ids []string + for id := range e.stream { + if include(id) { + ids = append(ids, id) + } + } + sort.Slice(ids, func(i, j int) bool { return e.stream[ids[i]].SortsBefore(e.stream[ids[j]]) }) + return ids +} + +// walk pages through /api/history and returns the stream rows' ids in the +// order received and the actionable rows seen per page. between, when set, +// runs after each page. +func (e *historyPagingEnv) walk(ctx context.Context, t *testing.T, h *Handler, token string, limit int, between func(page int)) (stream []string, actionable []map[string]bool, cursors []string) { + t.Helper() + cursor := "" + for page := 0; ; page++ { + require.Less(t, page, 200, "paging must terminate") + params := map[string]string{"limit": strconv.Itoa(limit)} + if cursor != "" { + params["cursor"] = cursor + } + resp := getHistoryPage(ctx, t, h, token, params) + seen := map[string]bool{} + streamRows := 0 + for _, row := range resp.Purchases { + if _, ok := e.actionable[row.PurchaseID]; ok { + seen[row.PurchaseID] = true + continue + } + stream = append(stream, row.PurchaseID) + streamRows++ + } + actionable = append(actionable, seen) + assert.LessOrEqual(t, streamRows, limit) + assert.Equal(t, resp.HasMore, resp.NextCursor != "") + if resp.HasMore { + assert.Equal(t, limit, streamRows, "only the last page may be short") + } + if between != nil { + between(page) + } + if !resp.HasMore { + return stream, actionable, cursors + } + cursor = resp.NextCursor + cursors = append(cursors, cursor) + } +} + +func historyPagingHandlers(env *historyPagingEnv) (admin, scoped *Handler) { + scopedAuth := new(MockAuthService) + scopedAuth.On("ValidateSession", mock.Anything, scopedToken).Return(&Session{UserID: scopedUserID}, nil).Maybe() + scopedAuth.On("GetUser", mock.Anything, mock.Anything).Return(&User{Email: "creator@example.com"}, nil).Maybe() + scopedAuth.grantScoped(env.tenantA.Name) + adminAuth := new(MockAuthService) + adminAuth.On("ValidateSession", mock.Anything, "admin-token").Return(&Session{UserID: "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"}, nil).Maybe() + adminAuth.On("GetUser", mock.Anything, mock.Anything).Return(&User{Email: "creator@example.com"}, nil).Maybe() + adminAuth.grantAdmin() + return &Handler{config: env.store, auth: adminAuth}, &Handler{config: env.store, auth: scopedAuth} +} + +// TestIntegration_HistoryPaging_EveryRowOnceInOrder is the issue #674 +// regression on real Postgres. Two tenants hold far more History rows than +// the page size, with completed purchases and terminal executions sharing +// timestamps within and across tables. Paging with the cursor must return +// every row exactly once, in stream order, with both pending approvals (older +// than every other row) on the first page and on no other. Rows written at the +// top between two page loads must not shift, repeat or skip any row. +func TestIntegration_HistoryPaging_EveryRowOnceInOrder(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + env := setupHistoryPagingEnv(ctx, t) + base := time.Now().UTC().Add(-24 * time.Hour).Truncate(time.Microsecond) + env.seed(ctx, t, "a", base) + env.seed(ctx, t, "b", base.Add(-30*time.Minute)) + admin, _ := historyPagingHandlers(env) + want := env.expected(func(id string) bool { + k := env.stream[id] + // A clean completed execution is listed through purchase_history only. + return k.Source == config.HistorySourcePurchase || !isCleanCompleted(ctx, t, env, id) + }) + require.Greater(t, len(want), 4*7, "the fixture must span several pages") + + var inserted []string + got, actionable, _ := env.walk(ctx, t, admin, "admin-token", 7, func(page int) { + if page == 0 { + // New rows land at the top, ahead of the cursor. + inserted = append(inserted, + env.addPurchase(ctx, t, "a", time.Now().UTC().Truncate(time.Microsecond), false), + env.addExecution(ctx, t, "b", "", "failed", time.Now().UTC().Truncate(time.Microsecond))) + } + }) + + assert.Equal(t, want, got, "every row exactly once, in stream order") + for _, id := range inserted { + assert.NotContains(t, got, id, "a row written above the cursor belongs to a fresh first page") + } + require.NotEmpty(t, actionable) + for id := range env.actionable { + assert.True(t, actionable[0][id], "pending approval %s on the first page", id) + for n := 1; n < len(actionable); n++ { + assert.False(t, actionable[n][id], "pending approval %s repeated on page %d", id, n) + } + } +} + +// TestIntegration_HistoryPaging_ScopedUserNeverPagesIntoAnotherTenant pages as +// a user scoped to tenant A while tenant B's rows are newer, interleaved and +// older. The user must get exactly tenant A's rows and their own ambient +// execution, each once and in order, never a tenant-B row or another user's +// ambient row; no cursor may name a hidden row; and has_more must be false on +// the last page although tenant-B rows follow it. +func TestIntegration_HistoryPaging_ScopedUserNeverPagesIntoAnotherTenant(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + env := setupHistoryPagingEnv(ctx, t) + base := time.Now().UTC().Add(-24 * time.Hour).Truncate(time.Microsecond) + for i := 0; i < 40; i++ { + env.addPurchase(ctx, t, "b", base.Add(time.Duration(i+1)*time.Minute), i%2 == 0) + } + env.seed(ctx, t, "a", base) + env.seed(ctx, t, "b", base) + for i := 0; i < 25; i++ { + env.addExecution(ctx, t, "b", "", "failed", base.Add(-time.Duration(100+i)*time.Hour)) + } + other := uuid.NewString() + env.addUser(ctx, t, scopedUserID) + env.addUser(ctx, t, other) + env.addExecution(ctx, t, "", scopedUserID, "expired", base.Add(-5*time.Hour)) + env.addExecution(ctx, t, "", other, "expired", base.Add(-5*time.Hour)) + _, scoped := historyPagingHandlers(env) + + visible := func(id string) bool { + if isCleanCompleted(ctx, t, env, id) { + return false + } + return env.tenant[id] == "a" || (env.tenant[id] == "" && env.owner[id] == scopedUserID) + } + want := env.expected(visible) + require.Greater(t, len(want), 3*5, "the fixture must span several pages") + + got, actionable, cursors := env.walk(ctx, t, scoped, scopedToken, 5, nil) + + assert.Equal(t, want, got, "exactly the in-scope rows, each once, in stream order") + for _, id := range got { + assert.True(t, visible(id), "row %s is outside the caller's scope", id) + } + byCursor := map[string]string{} + for id, k := range env.stream { + byCursor[encodeHistoryCursor(k)] = id + } + require.NotEmpty(t, cursors) + for _, c := range cursors { + id, ok := byCursor[c] + require.True(t, ok, "cursor must name a seeded row") + assert.True(t, visible(id), "cursor names hidden row %s", id) + } + for id, tenant := range env.actionable { + assert.Equal(t, tenant == "a", actionable[0][id], "pending approval %s (tenant %q) on the first page iff in scope", id, tenant) + } +} + +// isCleanCompleted reports whether id is a completed execution with no error. +// The fixture gives every execution an error, so none is; the check keeps the +// expectation honest if that changes. +func isCleanCompleted(ctx context.Context, t *testing.T, env *historyPagingEnv, id string) bool { + t.Helper() + if env.stream[id].Source != config.HistorySourceExecution { + return false + } + exec, err := env.store.GetExecutionByID(ctx, id) + require.NoError(t, err) + return exec.Status == "completed" && exec.Error == "" +} diff --git a/internal/api/handler_history_page_test.go b/internal/api/handler_history_page_test.go new file mode 100644 index 00000000..2733275a --- /dev/null +++ b/internal/api/handler_history_page_test.go @@ -0,0 +1,396 @@ +package api + +import ( + "context" + "encoding/base64" + "errors" + "fmt" + "slices" + "sort" + "strings" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// noCursor is the after argument of a first-page ListPurchaseHistoryPage read. +var noCursor = (*config.HistoryPageKey)(nil) + +func b64url(s string) string { return base64.RawURLEncoding.EncodeToString([]byte(s)) } + +// pageUUID returns a distinct canonical UUID per (group, i). +func pageUUID(group, i int) string { + return fmt.Sprintf("%08x-0000-4000-8000-%012x", group, i) +} + +// streamFixture is one paged-stream row of a test fixture: its key and the +// purchase_id the response carries for it. +type streamFixture struct { + key config.HistoryPageKey + id string +} + +func purchaseFixture(id, account string, ts time.Time, rowID string) (config.PurchaseHistoryPageRow, streamFixture) { + row := config.PurchaseHistoryPageRow{ + PurchaseHistoryRecord: config.PurchaseHistoryRecord{PurchaseID: id, AccountID: account, Provider: "aws", Timestamp: ts}, + RowID: rowID, + } + return row, streamFixture{key: row.Key(), id: id} +} + +func executionFixture(execID, status string, account *string, ts time.Time) (config.PurchaseExecution, streamFixture) { + exec := config.PurchaseExecution{ExecutionID: execID, Status: status, ScheduledDate: ts, CloudAccountID: account, Error: "boom"} + key := config.HistoryPageKey{Timestamp: ts, Source: config.HistorySourceExecution, ID: execID} + return exec, streamFixture{key: key, id: execID} +} + +func sortedStreamIDs(rows []streamFixture) []string { + sorted := slices.Clone(rows) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].key.SortsBefore(sorted[j].key) }) + ids := make([]string, len(sorted)) + for i := range sorted { + ids[i] = sorted[i].id + } + return ids +} + +// walkHistoryPages requests pages until has_more is false and returns them. +func walkHistoryPages(t *testing.T, h *Handler, req *events.LambdaFunctionURLRequest, params map[string]string) []HistoryResponse { + t.Helper() + var pages []HistoryResponse + cursor := "" + for { + p := map[string]string{} + for k, v := range params { + p[k] = v + } + if cursor != "" { + p["cursor"] = cursor + } + out, err := h.getHistory(context.Background(), req, p) + require.NoError(t, err) + resp := out.(HistoryResponse) + pages = append(pages, resp) + assert.Equal(t, resp.HasMore, resp.NextCursor != "", "next_cursor is set exactly when has_more is") + if !resp.HasMore { + return pages + } + require.Less(t, len(pages), 100, "paging must terminate") + cursor = resp.NextCursor + } +} + +// TestHandler_getHistory_PagesThroughEveryRowOnce pages a stream of +// completed purchases and terminal executions that share timestamps within +// and across the two sources. Every row must come back exactly once, in +// stream order, and the pending approval, older than every other row, must be +// on the first page (an offset over one merged list would put it last). +func TestHandler_getHistory_PagesThroughEveryRowOnce(t *testing.T) { + ctx := context.Background() + // Recent, so the pending row is not stale and no expiry sweep runs. + base := time.Now().UTC().Truncate(time.Second).Add(-time.Hour) + var purchases []config.PurchaseHistoryPageRow + var execs []config.PurchaseExecution + var stream []streamFixture + for i := 0; i < 9; i++ { + row, f := purchaseFixture(fmt.Sprintf("p-%d", i), "acct", base.Add(-time.Duration(i/3)*time.Hour), pageUUID(0xa, i)) + purchases = append(purchases, row) + stream = append(stream, f) + } + for i := 0; i < 8; i++ { + exec, f := executionFixture(pageUUID(0xe, i), "failed", nil, base.Add(-time.Duration(i/3)*time.Hour)) + execs = append(execs, exec) + stream = append(stream, f) + } + pending := config.PurchaseExecution{ExecutionID: pageUUID(0x9, 0), Status: "pending", ScheduledDate: base.Add(-24 * time.Hour)} + execs = append(execs, pending) + + store := new(MockConfigStore) + store.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(purchases, nil) + store.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(execs, nil) + store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(execs, nil) + mockAuth, req := adminHistoryReq(ctx) + h := &Handler{auth: mockAuth, config: store} + + pages := walkHistoryPages(t, h, req, map[string]string{"limit": "4"}) + + require.Len(t, pages, 5, "17 stream rows at 4 per page") + var got []string + for n, page := range pages { + pendingSeen := false + var streamRows int + for _, row := range page.Purchases { + if row.PurchaseID == pending.ExecutionID { + pendingSeen = true + continue + } + got = append(got, row.PurchaseID) + streamRows++ + } + assert.Equal(t, n == 0, pendingSeen, "page %d: the pending approval is listed on the first page only", n) + assert.Equal(t, page.HasMore, page.Truncated, "page %d: nothing failed, so truncated only means more pages", n) + if page.HasMore { + assert.Equal(t, 4, streamRows, "page %d is full", n) + } + } + assert.Equal(t, sortedStreamIDs(stream), got, "every row exactly once, in stream order") +} + +// TestHandler_getHistory_ScopedPagingStaysInScope pages as a user scoped to +// account A through a stream where account B's rows are newest, interleaved +// and last. The user must get every account-A row and their own ambient row +// exactly once and nothing else; no cursor may name a hidden row; and the +// last page must say has_more=false although account-B rows follow it. +func TestHandler_getHistory_ScopedPagingStaysInScope(t *testing.T) { + ctx := context.Background() + base := time.Date(2026, 9, 1, 12, 0, 0, 0, time.UTC) + accA, accB := permsAccA, permsAccB + own, other := permsScopedUserID, "dddddddd-dddd-4ddd-dddd-dddddddddddd" + var purchases []config.PurchaseHistoryPageRow + var execs []config.PurchaseExecution + var visible []streamFixture + hidden := map[string]bool{} + addPurchase := func(id, account string, ts time.Time, n int) { + row, f := purchaseFixture(id, account, ts, pageUUID(0xa, n)) + purchases = append(purchases, row) + if account == accA { + visible = append(visible, f) + } else { + hidden[f.key.ID] = true + } + } + // 150 newer account-B rows: the first page needs two store batches. + for i := 0; i < 150; i++ { + addPurchase(fmt.Sprintf("b-new-%d", i), accB, base.Add(time.Duration(150-i)*time.Minute), i) + } + // 8 purchases + 3 executions + 1 ambient row = 12 visible rows: the last + // page is full and only hidden rows follow it. + for i := 0; i < 8; i++ { + addPurchase(fmt.Sprintf("a-%d", i), accA, base.Add(-time.Duration(i)*time.Hour), 200+i) + addPurchase(fmt.Sprintf("b-mid-%d", i), accB, base.Add(-time.Duration(i)*time.Hour), 300+i) + } + for i := 0; i < 30; i++ { + addPurchase(fmt.Sprintf("b-old-%d", i), accB, base.Add(-time.Duration(100+i)*time.Hour), 400+i) + } + for i := 0; i < 3; i++ { + exec, f := executionFixture(pageUUID(0xe, i), "failed", &accA, base.Add(-time.Duration(2*i)*time.Hour)) + execs = append(execs, exec) + visible = append(visible, f) + execB, fB := executionFixture(pageUUID(0xf, i), "failed", &accB, base.Add(-time.Duration(2*i)*time.Hour)) + execs = append(execs, execB) + hidden[fB.key.ID] = true + } + ownAmbient, fOwn := executionFixture(pageUUID(0x1, 0), "expired", nil, base.Add(-90*time.Hour)) + ownAmbient.CreatedByUserID = &own + otherAmbient, fOther := executionFixture(pageUUID(0x1, 1), "expired", nil, base.Add(-91*time.Hour)) + otherAmbient.CreatedByUserID = &other + execs = append(execs, ownAmbient, otherAmbient) + visible = append(visible, fOwn) + hidden[fOther.key.ID] = true + + store := new(MockConfigStore) + store.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(purchases, nil) + store.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(execs, nil) + store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) + store.ListCloudAccountsFn = func(context.Context, config.CloudAccountFilter) ([]config.CloudAccount, error) { + return permsAccountList(), nil + } + auth := scopedAuthMock(ctx) + auth.On("GetUser", ctx, mock.Anything).Return(&User{Email: "creator@example.com"}, nil).Maybe() + h := &Handler{auth: auth, config: store} + + pages := walkHistoryPages(t, h, scopedReq(), map[string]string{"limit": "3"}) + + var got []string + for n, page := range pages { + for _, row := range page.Purchases { + got = append(got, row.PurchaseID) + } + if page.HasMore { + key, err := decodeHistoryCursor(page.NextCursor) + require.NoError(t, err) + assert.False(t, hidden[key.ID], "page %d: next_cursor names a row the caller cannot see", n) + } + } + assert.Equal(t, sortedStreamIDs(visible), got, "every in-scope row exactly once, in order, and no other row") + require.Len(t, visible, 12) + assert.Len(t, pages, 4, "no trailing empty page") + last := pages[len(pages)-1] + assert.False(t, last.HasMore, "out-of-scope rows after the last visible row must not set has_more") + assert.False(t, last.Truncated) +} + +// TestHandler_getHistory_CursorPageOmitsActionableRows checks that a page +// fetched with a cursor neither lists nor sweeps actionable executions: they +// were all on the first page. +func TestHandler_getHistory_CursorPageOmitsActionableRows(t *testing.T) { + ctx := context.Background() + base := time.Date(2026, 9, 1, 12, 0, 0, 0, time.UTC) + var purchases []config.PurchaseHistoryPageRow + for i := 0; i < 3; i++ { + row, _ := purchaseFixture(fmt.Sprintf("p-%d", i), "acct", base.Add(-time.Duration(i)*time.Hour), pageUUID(0xa, i)) + purchases = append(purchases, row) + } + store := new(MockConfigStore) + store.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(purchases, nil) + mockAuth, req := adminHistoryReq(ctx) + h := &Handler{auth: mockAuth, config: store, isLambda: true} + + cursor := encodeHistoryCursor(purchases[0].Key()) + out, err := h.getHistory(ctx, req, map[string]string{"cursor": cursor, "limit": "1"}) + require.NoError(t, err) + resp := out.(HistoryResponse) + + require.Len(t, resp.Purchases, 1) + assert.Equal(t, "p-1", resp.Purchases[0].PurchaseID) + assert.True(t, resp.HasMore) + store.AssertNotCalled(t, "GetExecutionsByStatuses", mock.Anything, mock.Anything, mock.Anything) + store.AssertNotCalled(t, "ExpireExecutionAtomic", mock.Anything, mock.Anything, mock.Anything) + store.AssertCalled(t, "ListPurchaseHistoryPage", ctx, mock.Anything, mock.MatchedBy(func(after *config.HistoryPageKey) bool { + return after != nil && *after == purchases[0].Key() + })) +} + +// TestHandler_getHistory_TerminalExecutionsFailureFlagsTruncated keeps the +// issue #686 signal on paged reads: when terminal executions cannot be read +// the purchases still page, and the response says rows are missing. +func TestHandler_getHistory_TerminalExecutionsFailureFlagsTruncated(t *testing.T) { + ctx := context.Background() + base := time.Date(2026, 9, 1, 12, 0, 0, 0, time.UTC) + row, _ := purchaseFixture("p-0", "acct", base, pageUUID(0xa, 0)) + store := new(MockConfigStore) + store.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryPageRow{row}, nil) + store.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(nil, errors.New("db down")) + mockAuth, req := adminHistoryReq(ctx) + h := &Handler{auth: mockAuth, config: store} + + cursor := encodeHistoryCursor(config.HistoryPageKey{Timestamp: base.Add(time.Hour), Source: config.HistorySourcePurchase, ID: pageUUID(0xa, 9)}) + out, err := h.getHistory(ctx, req, map[string]string{"cursor": cursor}) + require.NoError(t, err) + resp := out.(HistoryResponse) + require.Len(t, resp.Purchases, 1) + assert.False(t, resp.HasMore) + assert.True(t, resp.Truncated) +} + +// TestHandler_getHistory_PurchaseHistoryFailureFailsRequest keeps the +// pre-paging contract: a purchase_history read error is a server error. +func TestHandler_getHistory_PurchaseHistoryFailureFailsRequest(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + store.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(nil, errors.New("db down")) + mockAuth, req := adminHistoryReq(ctx) + h := &Handler{auth: mockAuth, config: store} + + _, err := h.getHistory(ctx, req, map[string]string{}) + require.Error(t, err) + _, isClient := IsClientError(err) + assert.False(t, isClient) +} + +func TestHistoryStreamSource_RefusesAKeysetReadThatDoesNotAdvance(t *testing.T) { + cursor := config.HistoryPageKey{Timestamp: time.Unix(100, 0), Source: config.HistorySourcePurchase, ID: pageUUID(1, 1)} + src := historyStreamSource{ + after: &cursor, + batch: 2, + fetch: func(context.Context, *config.HistoryPageKey) ([]historyStreamRow, error) { + return []historyStreamRow{{key: cursor}, {key: cursor}}, nil + }, + } + _, err := src.head(context.Background()) + require.Error(t, err, "a store that repeats rows must not loop the request forever") +} + +func TestHistoryCursor_RoundTrip(t *testing.T) { + for _, key := range []config.HistoryPageKey{ + {Timestamp: time.Date(2026, 9, 1, 12, 0, 0, 123456000, time.UTC), Source: config.HistorySourcePurchase, ID: pageUUID(0xa, 1)}, + {Timestamp: time.Date(1999, 1, 1, 0, 0, 0, 0, time.UTC), Source: config.HistorySourceExecution, ID: pageUUID(0xe, 2)}, + } { + got, err := decodeHistoryCursor(encodeHistoryCursor(key)) + require.NoError(t, err) + assert.True(t, key.Timestamp.Equal(got.Timestamp)) + assert.Equal(t, key.Source, got.Source) + assert.Equal(t, key.ID, got.ID) + } +} + +// TestParseHistoryFilters_PagingParams covers the paging parameters' +// validation: anything but an unchanged next_cursor value is a 400, and so is +// an offset. +func TestParseHistoryFilters_PagingParams(t *testing.T) { + enc := func(raw string) string { return b64url(raw) } + id := pageUUID(0xa, 1) + valid := encodeHistoryCursor(config.HistoryPageKey{Timestamp: time.Unix(1700000000, 0), Source: config.HistorySourcePurchase, ID: id}) + tests := []struct { + name string + params map[string]string + wantErr string + }{ + {"no paging params", map[string]string{}, ""}, + {"valid cursor", map[string]string{"cursor": valid}, ""}, + {"offset", map[string]string{"offset": "100"}, "offset is not supported"}, + {"offset zero", map[string]string{"offset": "0"}, "offset is not supported"}, + {"not base64", map[string]string{"cursor": "!!!"}, "invalid cursor"}, + {"wrong version", map[string]string{"cursor": enc("v2.2.1700000000000000." + id)}, "invalid cursor"}, + {"too few fields", map[string]string{"cursor": enc("v1.2.1700000000000000")}, "invalid cursor"}, + {"too many fields", map[string]string{"cursor": enc("v1.2.1700000000000000." + id + ".x")}, "invalid cursor"}, + {"unknown source", map[string]string{"cursor": enc("v1.3.1700000000000000." + id)}, "invalid cursor"}, + {"zero source", map[string]string{"cursor": enc("v1.0.1700000000000000." + id)}, "invalid cursor"}, + {"non-numeric timestamp", map[string]string{"cursor": enc("v1.2.soon." + id)}, "invalid cursor"}, + {"timestamp out of range", map[string]string{"cursor": enc("v1.2.9223372036854775807." + id)}, "invalid cursor"}, + {"id not a uuid", map[string]string{"cursor": enc("v1.2.1700000000000000.p-1")}, "invalid cursor"}, + {"id not canonical", map[string]string{"cursor": enc("v1.2.1700000000000000." + strings.ToUpper(id))}, "invalid cursor"}, + {"id injection", map[string]string{"cursor": enc("v1.2.1700000000000000.' OR 1=1 --")}, "invalid cursor"}, + {"overlong", map[string]string{"cursor": strings.Repeat("A", maxHistoryCursorLen+1)}, "invalid cursor"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f, err := parseHistoryFilters(tt.params) + if tt.wantErr == "" { + require.NoError(t, err) + if tt.params["cursor"] != "" { + require.NotNil(t, f.Cursor) + assert.Equal(t, id, f.Cursor.ID) + } else { + assert.Nil(t, f.Cursor) + } + return + } + require.Error(t, err) + ce, ok := IsClientError(err) + require.True(t, ok, "got %T: %v", err, err) + assert.Equal(t, 400, ce.code) + assert.Contains(t, err.Error(), tt.wantErr) + }) + } +} + +// TestHandler_getHistory_InvalidCursorReadsNothing checks the 400 happens +// before any store read. +func TestHandler_getHistory_InvalidCursorReadsNothing(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + mockAuth, req := adminHistoryReq(ctx) + h := &Handler{auth: mockAuth, config: store} + + _, err := h.getHistory(ctx, req, map[string]string{"cursor": "garbage"}) + require.Error(t, err) + ce, ok := IsClientError(err) + require.True(t, ok) + assert.Equal(t, 400, ce.code) + store.AssertNotCalled(t, "ListPurchaseHistoryPage", mock.Anything, mock.Anything, mock.Anything) + store.AssertNotCalled(t, "ListHistoryExecutionsPage", mock.Anything, mock.Anything) + store.AssertNotCalled(t, "GetExecutionsByStatuses", mock.Anything, mock.Anything, mock.Anything) +} + +func TestHistoryStreamBatch(t *testing.T) { + assert.Equal(t, config.DefaultListLimit, historyStreamBatch(1)) + assert.Equal(t, config.DefaultListLimit+1, historyStreamBatch(config.DefaultListLimit)) + assert.Equal(t, 501, historyStreamBatch(500)) + assert.Equal(t, config.MaxListLimit, historyStreamBatch(config.MaxListLimit)) +} diff --git a/internal/api/handler_history_scope_provider_test.go b/internal/api/handler_history_scope_provider_test.go index 8ed8d668..c2982425 100644 --- a/internal/api/handler_history_scope_provider_test.go +++ b/internal/api/handler_history_scope_provider_test.go @@ -25,7 +25,7 @@ func historyScopeIDs(t *testing.T, scope []string, accounts []config.CloudAccoun mockAuth.On("GetAllowedAccountsAPI", mock.Anything, "u-1").Return(scope, nil) store := new(MockConfigStore) - store.On("GetAllPurchaseHistory", mock.Anything, mock.Anything).Return(append([]config.PurchaseHistoryRecord(nil), rows...), nil) + store.On("ListPurchaseHistoryPage", mock.Anything, mock.Anything, mock.Anything).Return(append([]config.PurchaseHistoryRecord(nil), rows...), nil) store.On("GetExecutionsByStatuses", mock.Anything, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) store.ListCloudAccountsFn = func(_ context.Context, _ config.CloudAccountFilter) ([]config.CloudAccount, error) { return accounts, nil diff --git a/internal/api/handler_history_test.go b/internal/api/handler_history_test.go index 61747740..32760ef8 100644 --- a/internal/api/handler_history_test.go +++ b/internal/api/handler_history_test.go @@ -40,8 +40,8 @@ func TestHandler_getHistory(t *testing.T) { // account_id "123456789012" is not a known cloud_accounts UUID, so it is // folded into the dual-column filter as an external account number and the - // request routes through GetPurchaseHistoryFiltered (issue #701/#498/#866). - mockStore.On("GetPurchaseHistoryFiltered", ctx, config.PurchaseHistoryFilter{ExternalIDsByProvider: map[string][]string{"": {"123456789012"}}, Limit: 100}).Return(history, nil) + // request routes through the filtered SQL predicate (issue #701/#498/#866). + mockStore.On("ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{ExternalIDsByProvider: map[string][]string{"": {"123456789012"}}, Limit: historyStreamBatch(100)}, noCursor).Return(history, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) mockAuth, req := adminHistoryReq(ctx) @@ -74,7 +74,7 @@ func TestHandler_getHistory_AllAccounts(t *testing.T) { {AccountID: "222222222222", PurchaseID: "purchase-2", UpfrontCost: new(float64(200.0)), EstimatedSavings: 20.0}, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return(history, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(history, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) mockAuth, req := adminHistoryReq(ctx) @@ -122,7 +122,7 @@ func TestHandler_getHistory_IncludesPending(t *testing.T) { } approverEmail := "ops@example.com" - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return(completed, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(completed, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(pending, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -161,7 +161,7 @@ func TestHandler_getHistory_CustomLimit(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) - mockStore.On("GetAllPurchaseHistory", ctx, 50).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) mockAuth, req := adminHistoryReq(ctx) @@ -171,8 +171,10 @@ func TestHandler_getHistory_CustomLimit(t *testing.T) { "limit": "50", } - _, err := handler.getHistory(ctx, req, params) + result, err := handler.getHistory(ctx, req, params) require.NoError(t, err) + assert.Equal(t, 50, result.(HistoryResponse).Limit) + mockStore.AssertCalled(t, "ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{Limit: historyStreamBatch(50)}, noCursor) } // TestHandler_getHistory_ExpireIfStale exercises the async approval-window @@ -238,7 +240,7 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) { expired.Status = "expired" done := make(chan struct{}) - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{freshExec(), staleExec("pending")}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -298,7 +300,7 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) { expired.Status = "expired" done := make(chan struct{}) - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{staleExec("notified")}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -335,7 +337,7 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) { mockStore := new(MockConfigStore) done := make(chan struct{}) - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{staleExec("pending")}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -372,7 +374,7 @@ func TestHandler_getHistory_ExpireIfStale(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{freshExec()}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -430,7 +432,7 @@ func TestHandler_getHistory_GetIsReadOnly(t *testing.T) { // immediately and the goroutine blocks on the gate independently. gate := make(chan struct{}) transitionCalled := make(chan struct{}) - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{stale}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, "stale-ro-exec"). @@ -505,7 +507,7 @@ func TestHandler_getHistory_ScopedUserSeesEmptyAccountRows(t *testing.T) { mockStore := new(MockConfigStore) approverEmail := "ops@example.com" - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{ambientExec}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) // resolveAccountNamesByID must not error — return an empty accounts list @@ -579,7 +581,7 @@ func TestHandler_getHistory_ScopedUserCannotSeeOtherUsersEmptyAccountRows(t *tes mockStore := new(MockConfigStore) approverEmail := "ops@example.com" - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{otherUsersExec}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) mockStore.ListCloudAccountsFn = func(_ context.Context, _ config.CloudAccountFilter) ([]config.CloudAccount, error) { @@ -626,7 +628,7 @@ func TestHandler_expireStaleExecutionsAsync_SystemActorIsNil(t *testing.T) { staleID := "actor-nil-stale-exec" done := make(chan struct{}) - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{{ ExecutionID: staleID, @@ -670,7 +672,7 @@ func TestHandler_getHistory_PermissionDenied(t *testing.T) { require.Error(t, err) assert.Contains(t, err.Error(), "permission denied") mockStore.AssertNotCalled(t, "GetPurchaseHistory") - mockStore.AssertNotCalled(t, "GetAllPurchaseHistory") + mockStore.AssertNotCalled(t, "ListPurchaseHistoryPage") } // TestHandler_getHistory_IncludesInFlightApprovals is the issue #621 primary- @@ -707,7 +709,7 @@ func TestHandler_getHistory_IncludesInFlightApprovals(t *testing.T) { }, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return(completed, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(completed, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(inFlight, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approver}, nil) @@ -779,7 +781,7 @@ func TestHandler_getHistory_InProgressRowMapsRecFields(t *testing.T) { }, } approver := "ops@example.com" - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(inProgress, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approver}, nil) @@ -831,8 +833,9 @@ func TestHandler_getHistory_AuditGapCompletedVisible(t *testing.T) { }, } approver := "ops@example.com" - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) - mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(auditGap, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) + mockStore.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(auditGap, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approver}, nil) mockAuth, req := adminHistoryReq(ctx) @@ -880,8 +883,9 @@ func TestHandler_getHistory_PartiallyCompletedVisible(t *testing.T) { }, } approver := "ops@example.com" - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) - mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(partial, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) + mockStore.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(partial, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approver}, nil) mockAuth, req := adminHistoryReq(ctx) @@ -927,7 +931,7 @@ func TestHandler_getHistory_CompletedDBRowWithDescriptionStillCounts(t *testing. EstimatedSavings: 70.0, }, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return(completed, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(completed, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) approver := "ops@example.com" mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approver}, nil) @@ -981,12 +985,12 @@ func TestHandler_getHistory_FilterParams(t *testing.T) { t.Run("provider filter is pushed to SQL and prunes executions", func(t *testing.T) { mockStore, handler, req, ctx := newHandler(t) - // SQL path: must be called via GetPurchaseHistoryFiltered with + // SQL path: must be called via ListPurchaseHistoryPage with // provider="aws" and no other filters set. Return only the aws row. filtered := []config.PurchaseHistoryRecord{ {AccountID: "acc-aws", PurchaseID: "p-aws", Provider: "aws", UpfrontCost: new(float64(100.0))}, } - mockStore.On("GetPurchaseHistoryFiltered", ctx, config.PurchaseHistoryFilter{Provider: "aws", Limit: config.DefaultListLimit}). + mockStore.On("ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{Provider: "aws", Limit: historyStreamBatch(config.DefaultListLimit)}, noCursor). Return(filtered, nil).Once() // Executions path: two execs, one aws-rec and one azure-rec. Only the @@ -1016,18 +1020,17 @@ func TestHandler_getHistory_FilterParams(t *testing.T) { for _, p := range resp.Purchases { assert.NotEqual(t, "exec-azure", p.PurchaseID, "azure-only execution must be filtered out when provider=aws") } - mockStore.AssertCalled(t, "GetPurchaseHistoryFiltered", ctx, config.PurchaseHistoryFilter{Provider: "aws", Limit: config.DefaultListLimit}) + mockStore.AssertCalled(t, "ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{Provider: "aws", Limit: historyStreamBatch(config.DefaultListLimit)}, noCursor) }) t.Run("provider=all is treated as no filter (legacy SQL path)", func(t *testing.T) { mockStore, handler, req, ctx := newHandler(t) - mockStore.On("GetAllPurchaseHistory", ctx, config.DefaultListLimit).Return([]config.PurchaseHistoryRecord{}, nil).Once() + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil).Once() mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) _, err := handler.getHistory(ctx, req, map[string]string{"provider": "all"}) require.NoError(t, err) - mockStore.AssertCalled(t, "GetAllPurchaseHistory", ctx, config.DefaultListLimit) - mockStore.AssertNotCalled(t, "GetPurchaseHistoryFiltered", mock.Anything, mock.Anything) + mockStore.AssertCalled(t, "ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{Limit: historyStreamBatch(config.DefaultListLimit)}, noCursor) }) t.Run("account_ids filter is pushed to SQL and prunes executions", func(t *testing.T) { @@ -1036,7 +1039,7 @@ func TestHandler_getHistory_FilterParams(t *testing.T) { uuidA := "11111111-1111-1111-1111-111111111111" uuidB := "22222222-2222-2222-2222-222222222222" - mockStore.On("GetPurchaseHistoryFiltered", ctx, config.PurchaseHistoryFilter{AccountIDs: []string{uuidA, uuidB}, Limit: config.DefaultListLimit}). + mockStore.On("ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{AccountIDs: []string{uuidA, uuidB}, Limit: historyStreamBatch(config.DefaultListLimit)}, noCursor). Return([]config.PurchaseHistoryRecord{{AccountID: "acc-A", PurchaseID: "p-A"}}, nil).Once() execs := []config.PurchaseExecution{ @@ -1088,7 +1091,7 @@ func TestHandler_getHistory_FilterParams(t *testing.T) { // dropped when combined with provider/account_ids (issue #701/#498/#866). mockStore, handler, req, ctx := newHandler(t) - mockStore.On("GetPurchaseHistoryFiltered", ctx, config.PurchaseHistoryFilter{Provider: "aws", ExternalIDsByProvider: map[string][]string{"": {"123456789012"}}, Limit: config.DefaultListLimit}). + mockStore.On("ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{Provider: "aws", ExternalIDsByProvider: map[string][]string{"": {"123456789012"}}, Limit: historyStreamBatch(config.DefaultListLimit)}, noCursor). Return([]config.PurchaseHistoryRecord{}, nil).Once() mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) @@ -1103,7 +1106,7 @@ func TestHandler_getHistory_FilterParams(t *testing.T) { // dual-column filter rather than the single-column GetPurchaseHistory // fast path, so its rows surface regardless of cloud_account_id. mockStore, handler, req, ctx := newHandler(t) - mockStore.On("GetPurchaseHistoryFiltered", ctx, config.PurchaseHistoryFilter{ExternalIDsByProvider: map[string][]string{"": {"123456789012"}}, Limit: config.DefaultListLimit}). + mockStore.On("ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{ExternalIDsByProvider: map[string][]string{"": {"123456789012"}}, Limit: historyStreamBatch(config.DefaultListLimit)}, noCursor). Return([]config.PurchaseHistoryRecord{{AccountID: "123456789012", PurchaseID: "p-legacy"}}, nil).Once() mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) @@ -1126,13 +1129,13 @@ func TestHandler_getHistory_FilterParams(t *testing.T) { endDay := now.Format("2006-01-02") outOfRangeDay := now.AddDate(0, 0, -10) // older than the window AND outside the requested range - mockStore.On("GetPurchaseHistoryFiltered", ctx, + mockStore.On("ListPurchaseHistoryPage", ctx, mock.MatchedBy(func(f config.PurchaseHistoryFilter) bool { return f.Provider == "" && len(f.AccountIDs) == 0 && len(f.ExternalIDsByProvider) == 0 && f.Start != nil && f.Start.Format("2006-01-02") == startDay && f.End != nil && f.End.Format("2006-01-02") == endDay && - f.Limit == config.DefaultListLimit - }), + f.Limit == historyStreamBatch(config.DefaultListLimit) + }), noCursor, ).Return([]config.PurchaseHistoryRecord{{AccountID: "in-range", PurchaseID: "p-in-range"}}, nil).Once() execs := []config.PurchaseExecution{ @@ -1165,11 +1168,11 @@ func TestHandler_getHistory_FilterParams(t *testing.T) { mockStore, handler, req, ctx := newHandler(t) uuidA := "55555555-5555-5555-5555-555555555555" - mockStore.On("GetPurchaseHistoryFiltered", ctx, + mockStore.On("ListPurchaseHistoryPage", ctx, mock.MatchedBy(func(f config.PurchaseHistoryFilter) bool { return f.Provider == "aws" && len(f.AccountIDs) == 1 && f.AccountIDs[0] == uuidA && - f.Start != nil && f.End != nil && f.Limit == config.DefaultListLimit - }), + f.Start != nil && f.End != nil && f.Limit == historyStreamBatch(config.DefaultListLimit) + }), noCursor, ).Return([]config.PurchaseHistoryRecord{{AccountID: "match", PurchaseID: "p-match"}}, nil).Once() now := time.Now().UTC() @@ -1289,8 +1292,7 @@ func TestHandler_getHistory_FilterValidation(t *testing.T) { // Sanity guards: no store calls leaked through on a 400. mockStore.AssertNotCalled(t, "GetPurchaseHistory", mock.Anything, mock.Anything, mock.Anything) - mockStore.AssertNotCalled(t, "GetAllPurchaseHistory", mock.Anything, mock.Anything) - mockStore.AssertNotCalled(t, "GetPurchaseHistoryFiltered", mock.Anything, mock.Anything) + mockStore.AssertNotCalled(t, "ListPurchaseHistoryPage", mock.Anything, mock.Anything, mock.Anything) mockStore.AssertNotCalled(t, "GetExecutionsByStatuses", mock.Anything, mock.Anything, mock.Anything) }) } @@ -1306,7 +1308,7 @@ func TestHandler_getHistory_DateRangeBoundary(t *testing.T) { mockStore := new(MockConfigStore) mockAuth, req := adminHistoryReq(ctx) handler := &Handler{auth: mockAuth, config: mockStore} - mockStore.On("GetPurchaseHistoryFiltered", ctx, mock.Anything). + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) @@ -1396,7 +1398,7 @@ func TestHandler_getHistory_CreatedByUserEmailResolved(t *testing.T) { {Provider: "aws", Service: "ec2", Region: "us-east-1"}, }, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{exec}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -1427,7 +1429,7 @@ func TestHandler_getHistory_CreatedByUserEmailResolved(t *testing.T) { {Provider: "aws", Service: "ec2", Region: "us-east-1"}, }, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{exec}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -1494,7 +1496,7 @@ func TestHandler_getHistory_ApprovalQueueColumnsPopulated(t *testing.T) { }, }, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(pending, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -1532,7 +1534,7 @@ func TestHandler_getHistory_ApprovalQueueColumnsPopulated(t *testing.T) { }, }, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(pending, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -1565,7 +1567,7 @@ func TestHandler_getHistory_ApprovalQueueColumnsPopulated(t *testing.T) { }, }, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(pending, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -1618,10 +1620,10 @@ func TestHandler_getHistory_ApprovalQueueColumnsPopulated(t *testing.T) { }, } mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(pending, nil) - // account_ids is set, so fetchPurchaseHistory routes to GetPurchaseHistoryFiltered - // (not GetAllPurchaseHistory). No DB rows needed; we only care about the + // account_ids is set, so fetchPurchaseHistory passes it to + // ListPurchaseHistoryPage. No DB rows needed; we only care about the // execution path. - mockStore.On("GetPurchaseHistoryFiltered", ctx, config.PurchaseHistoryFilter{AccountIDs: []string{recAccountID}, Limit: 100}). + mockStore.On("ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{AccountIDs: []string{recAccountID}, Limit: historyStreamBatch(100)}, noCursor). Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -1668,11 +1670,11 @@ func TestHandler_getHistory_ExternalIDOnlyAccount(t *testing.T) { bRow := []config.PurchaseHistoryRecord{ {AccountID: accountBExternal, PurchaseID: "p-B", Provider: "aws", Service: "ec2", UpfrontCost: new(float64(100.0)), EstimatedSavings: 25.0}, } - mockStore.On("GetPurchaseHistoryFiltered", ctx, config.PurchaseHistoryFilter{ + mockStore.On("ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{ AccountIDs: []string{accountBUUID}, ExternalIDsByProvider: map[string][]string{"aws": {accountBExternal}}, - Limit: config.DefaultListLimit, - }).Return(bRow, nil).Once() + Limit: historyStreamBatch(config.DefaultListLimit), + }, noCursor).Return(bRow, nil).Once() mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseExecution{}, nil) t.Cleanup(func() { mockStore.AssertExpectations(t) }) @@ -1742,8 +1744,9 @@ func TestHandler_getHistory_CompletedExecutionNotDuplicated(t *testing.T) { }, } approver := "ops@example.com" - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return(completed, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(completed, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(cleanCompletedExec, nil) + mockStore.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(cleanCompletedExec, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approver}, nil) mockAuth, req := adminHistoryReq(ctx) @@ -1893,18 +1896,18 @@ func TestHandler_getHistory_LimitParsing(t *testing.T) { ce, ok := IsClientError(err) require.True(t, ok, "error must be a ClientError, got %T: %v", err, err) assert.Equal(t, tc.wantCode, ce.code) - mockStore.AssertNotCalled(t, "GetAllPurchaseHistory", mock.Anything, mock.Anything) - mockStore.AssertNotCalled(t, "GetPurchaseHistoryFiltered", mock.Anything, mock.Anything) + mockStore.AssertNotCalled(t, "ListPurchaseHistoryPage", mock.Anything, mock.Anything, mock.Anything) } else { // Success path: store receives the clamped limit. - mockStore.On("GetAllPurchaseHistory", ctx, tc.wantLimit). + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{}, nil) - _, err := handler.getHistory(ctx, req, map[string]string{"limit": tc.limitParam}) + result, err := handler.getHistory(ctx, req, map[string]string{"limit": tc.limitParam}) require.NoError(t, err) - mockStore.AssertCalled(t, "GetAllPurchaseHistory", ctx, tc.wantLimit) + assert.Equal(t, tc.wantLimit, result.(HistoryResponse).Limit, "page size") + mockStore.AssertCalled(t, "ListPurchaseHistoryPage", ctx, config.PurchaseHistoryFilter{Limit: historyStreamBatch(tc.wantLimit)}, noCursor) } }) } @@ -1974,7 +1977,7 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) { expired := staleExec() expired.Status = "expired" - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{staleExec()}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -2013,7 +2016,7 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) { // would fire. release := make(chan struct{}) swept := make(chan struct{}) - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{staleExec()}, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) @@ -2176,7 +2179,7 @@ func TestHandler_getHistory_ProjectsCreatedBySystem(t *testing.T) { {ExecutionID: "sys", Status: "pending", ScheduledDate: time.Now(), CreatedBySystem: true, Recommendations: recs}, {ExecutionID: "human", Status: "pending", ScheduledDate: time.Now(), Recommendations: recs}, } - mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(execs, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) mockAuth, req := adminHistoryReq(ctx) diff --git a/internal/api/handler_history_truncation_test.go b/internal/api/handler_history_truncation_test.go index 5e022713..5e58bc43 100644 --- a/internal/api/handler_history_truncation_test.go +++ b/internal/api/handler_history_truncation_test.go @@ -49,8 +49,9 @@ func runAdminHistory(t *testing.T, completed []config.PurchaseHistoryRecord, exe t.Helper() ctx := context.Background() mockStore := new(MockConfigStore) - mockStore.On("GetAllPurchaseHistory", ctx, mock.Anything).Return(completed, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return(completed, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(execs, nil) + mockStore.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(execs, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{}, nil).Maybe() mockAuth, req := adminHistoryReq(ctx) handler := &Handler{auth: mockAuth, config: mockStore} @@ -60,21 +61,32 @@ func runAdminHistory(t *testing.T, completed []config.PurchaseHistoryRecord, exe return historyJSON(t, result) } -func TestHandler_getHistory_TruncatedWhenCompletedAtCap(t *testing.T) { - body := runAdminHistory(t, historyRows(config.DefaultListLimit), nil, map[string]string{}) - assert.Equal(t, true, body["truncated"], "a full page of completed rows may hide older ones") +func TestHandler_getHistory_TruncatedWhenCompletedPastLimit(t *testing.T) { + body := runAdminHistory(t, historyRows(config.DefaultListLimit+1), nil, map[string]string{}) + assert.Equal(t, true, body["truncated"], "an older completed row is not on this page") + assert.Equal(t, true, body["has_more"]) + assert.Len(t, body["purchases"], config.DefaultListLimit) assert.EqualValues(t, config.DefaultListLimit, body["limit"]) } -func TestHandler_getHistory_TruncatedWhenExecutionsAtCap(t *testing.T) { - // The execution fetch is capped before the in-memory filters run, so a - // full page means pending approvals may have been dropped. - body := runAdminHistory(t, nil, failedExecutionsAtCap(), map[string]string{}) +// Paging reads one row past the page (issue #674), so exactly a page's worth +// of rows is no longer a false positive. +func TestHandler_getHistory_NotTruncatedWhenCompletedExactlyAtLimit(t *testing.T) { + body := runAdminHistory(t, historyRows(config.DefaultListLimit), nil, map[string]string{}) + assert.Equal(t, false, body["truncated"]) + assert.Equal(t, false, body["has_more"]) + assert.Len(t, body["purchases"], config.DefaultListLimit) +} + +func TestHandler_getHistory_TruncatedWhenExecutionsPastLimit(t *testing.T) { + // Terminal executions share the page with completed purchases. + body := runAdminHistory(t, nil, failedExecutionsPastLimit(), map[string]string{}) assert.Equal(t, true, body["truncated"]) + assert.Equal(t, true, body["has_more"]) } -func failedExecutionsAtCap() []config.PurchaseExecution { - execs := pendingExecutions(config.DefaultListLimit) +func failedExecutionsPastLimit() []config.PurchaseExecution { + execs := pendingExecutions(config.DefaultListLimit + 1) for i := range execs { execs[i].Status = "failed" } @@ -97,30 +109,40 @@ func TestHandler_getHistory_NotTruncatedByCleanCompletedExecutions(t *testing.T) assert.Len(t, body["purchases"], 3) } -func TestHandler_getHistory_TruncatedByFailedExecutions(t *testing.T) { - body := runAdminHistory(t, nil, failedExecutionsAtCap(), map[string]string{}) - assert.Equal(t, true, body["truncated"]) -} - func TestHandler_getHistory_AuditGapCompletedExecutionsCountTowardCap(t *testing.T) { - execs := cleanCompletedExecutions(config.DefaultListLimit) + execs := cleanCompletedExecutions(config.DefaultListLimit + 1) for i := range execs { execs[i].Error = "history write failed" } body := runAdminHistory(t, nil, execs, map[string]string{}) assert.Equal(t, true, body["truncated"]) + assert.Len(t, body["purchases"], config.DefaultListLimit) } -func TestHandler_getHistory_ExecutionsLimitIsFixed(t *testing.T) { - body := runAdminHistory(t, historyRows(2), failedExecutionsAtCap(), map[string]string{"limit": "500"}) +// Terminal executions share the page with completed purchases (issue #674), +// so their cap follows the limit parameter. +func TestHandler_getHistory_ExecutionsLimitFollowsLimit(t *testing.T) { + body := runAdminHistory(t, historyRows(2), failedExecutionsPastLimit(), map[string]string{"limit": "500"}) assert.EqualValues(t, 500, body["limit"]) - assert.EqualValues(t, config.DefaultListLimit, body["executions_limit"]) + assert.EqualValues(t, 500, body["executions_limit"]) + assert.Len(t, body["purchases"], 2+config.DefaultListLimit+1) + assert.Equal(t, false, body["truncated"]) +} + +// The actionable class (pending, approved, ...) has its own, larger cap. The +// response must expose it so the truncation banner can quote the cap that was +// actually hit instead of the terminal-class cap (issue #686). +func TestHandler_getHistory_ActionableExecutionsLimitExposed(t *testing.T) { + body := runAdminHistory(t, nil, pendingExecutions(config.MaxListLimit), map[string]string{}) assert.Equal(t, true, body["truncated"]) + assert.EqualValues(t, config.MaxListLimit, body["actionable_executions_limit"]) + assert.EqualValues(t, config.DefaultListLimit, body["executions_limit"]) } func TestHandler_getHistory_TruncatedHonorsCustomLimit(t *testing.T) { - body := runAdminHistory(t, historyRows(5), nil, map[string]string{"limit": "5"}) + body := runAdminHistory(t, historyRows(6), nil, map[string]string{"limit": "5"}) assert.Equal(t, true, body["truncated"]) + assert.Len(t, body["purchases"], 5) assert.EqualValues(t, 5, body["limit"]) } @@ -130,19 +152,21 @@ func TestHandler_getHistory_NotTruncatedUnderLimit(t *testing.T) { assert.EqualValues(t, config.DefaultListLimit, body["limit"]) } -// A scoped user's visible rows can be few while the fetch hit the cap: the -// flag is derived from the pre-scope fetch (the cap drops rows before scope -// filtering) but must expose only a boolean, never other tenants' counts. +// A scoped user's visible rows can be few while other tenants' rows fill the +// page many times over. The page is filled with rows the caller may see and +// has_more is judged on those too (issue #674), so other tenants' volume +// changes neither flag, and no count is exposed. func TestHandler_getHistory_ScopedUserTruncationLeaksNoCounts(t *testing.T) { ctx := context.Background() const scopedUserID = "scoped-user-id" own := scopedUserID - execs := failedExecutionsAtCap() - execs[0].CreatedByUserID = &own + execs := failedExecutionsPastLimit() + execs[len(execs)-1].CreatedByUserID = &own mockStore := new(MockConfigStore) - mockStore.On("GetAllPurchaseHistory", ctx, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything).Return([]config.PurchaseHistoryRecord{}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(execs, nil) + mockStore.On("ListHistoryExecutionsPage", ctx, mock.Anything).Return(execs, nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{}, nil).Maybe() mockStore.ListCloudAccountsFn = func(_ context.Context, _ config.CloudAccountFilter) ([]config.CloudAccount, error) { return nil, nil @@ -160,7 +184,9 @@ func TestHandler_getHistory_ScopedUserTruncationLeaksNoCounts(t *testing.T) { body := historyJSON(t, result) assert.Len(t, body["purchases"], 1, "scoped user sees only their own row") - assert.Equal(t, true, body["truncated"]) + assert.Equal(t, false, body["truncated"], "other tenants' rows are not missing rows") + assert.Equal(t, false, body["has_more"]) + assert.NotContains(t, body, "next_cursor") assert.NotContains(t, body, "total", "no count of rows outside the caller's scope may be exposed") summary, ok := body["summary"].(map[string]any) require.True(t, ok) diff --git a/internal/api/handler_per_account_perms_test.go b/internal/api/handler_per_account_perms_test.go index 8fd82772..54d7a6b5 100644 --- a/internal/api/handler_per_account_perms_test.go +++ b/internal/api/handler_per_account_perms_test.go @@ -300,7 +300,7 @@ func TestPerAccountPerms_History_ListFilter(t *testing.T) { } mockStore := new(MockConfigStore) - mockStore.On("GetAllPurchaseHistory", ctx, mock.Anything). + mockStore.On("ListPurchaseHistoryPage", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseHistoryRecord{rowA, rowB}, nil) mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). Return([]config.PurchaseExecution{}, nil) diff --git a/internal/api/handler_test.go b/internal/api/handler_test.go index 73d253f5..9905c130 100644 --- a/internal/api/handler_test.go +++ b/internal/api/handler_test.go @@ -1056,7 +1056,7 @@ func TestHandler_HandleRequest_GetHistory(t *testing.T) { mockStore := new(MockConfigStore) history := []config.PurchaseHistoryRecord{{PurchaseID: "purchase-1"}} - mockStore.On("GetAllPurchaseHistory", mock.Anything, 100).Return(history, nil) + mockStore.On("ListPurchaseHistoryPage", mock.Anything, mock.Anything, mock.Anything).Return(history, nil) // getHistory now also lists pending executions to include them in the // merged response — mock a no-op return so this integration-style test // that only cares about the completed-row path still exercises the diff --git a/internal/api/main_integration_test.go b/internal/api/main_integration_test.go new file mode 100644 index 00000000..0c2940c6 --- /dev/null +++ b/internal/api/main_integration_test.go @@ -0,0 +1,22 @@ +//go:build integration + +package api + +import ( + "log" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" +) + +// Integration tests that take a database from testhelpers.RequireMigratedDatabase +// share one PostgreSQL container per test binary (issue #982); stop it once +// every test has finished. +func init() { + afterAllTests = append(afterAllTests, func() { + if err := testhelpers.TerminateSharedPostgres(); err != nil { + // Ryuk reaps the container anyway, so a teardown hiccup does not + // overrule the tests' own verdict. + log.Printf("terminate shared postgres container: %v", err) + } + }) +} diff --git a/internal/api/main_nonet_test.go b/internal/api/main_nonet_test.go index 52d5e4d6..01c5ff68 100644 --- a/internal/api/main_nonet_test.go +++ b/internal/api/main_nonet_test.go @@ -19,6 +19,11 @@ import ( // netGuard is the no-network guard TestMain installs for this package (#855). var netGuard *noNetworkGuard +// afterAllTests runs after every test in the package has finished. The +// integration build registers the shared PostgreSQL container's teardown here +// (main_integration_test.go). +var afterAllTests []func() + // TestMain keeps this package's tests hermetic (#855). Before any test runs it: // // - strips every AWS_* variable and points the shared config and @@ -45,6 +50,9 @@ func TestMain(m *testing.M) { } netGuard = g code := m.Run() + for _, f := range afterAllTests { + f() + } if blocked := g.stop(); len(blocked) > 0 { fmt.Fprintf(os.Stderr, "FAIL: internal/api tests attempted %d outbound network call(s); stub the client instead (#855):\n %s\n", len(blocked), strings.Join(blocked, "\n ")) diff --git a/internal/api/openapi.yaml b/internal/api/openapi.yaml index ba01d7f2..ead8cb7d 100644 --- a/internal/api/openapi.yaml +++ b/internal/api/openapi.yaml @@ -1720,6 +1720,16 @@ paths: operationId: getHistory tags: [History] summary: Get purchase history + description: >- + Lists completed purchases and terminal executions (failed, expired, + canceled, ...) newest first, `limit` rows per page, plus the actionable + executions (pending approvals and in-flight purchases). Actionable + executions are not paged: the first page lists all of them, up to + `actionable_executions_limit`, and pages fetched with `cursor` leave + them out. To fetch the next page, repeat the request with the same + filters and `cursor` set to the previous response's `next_cursor`. + Rows written after the first page was loaded never shift, repeat or + skip a row on a later page. parameters: - $ref: '#/components/parameters/AccountID' - $ref: '#/components/parameters/StartDate' @@ -1729,6 +1739,25 @@ paths: schema: type: string enum: [hourly, daily, weekly, monthly] + - name: limit + in: query + description: >- + Page size for completed purchases and terminal executions together. + Values below 1 use the default; values above 1000 are capped at 1000. + schema: + type: integer + default: 100 + maximum: 1000 + - name: cursor + in: query + description: >- + Opaque position to continue from: the `next_cursor` value of the + previous page, passed unchanged. Omit it for the first page. Any + other value is rejected with 400. There is no `offset` parameter; + sending one is rejected with 400. + schema: + type: string + maxLength: 256 responses: '200': description: Purchase history with summary @@ -4027,16 +4056,42 @@ components: pending approvals) may be missing from purchases and summary. May be a false positive when exactly `limit` or `executions_limit` rows exist. Also true when the actionable executions class (pending, approved, - running, ...) reached its own cap of 1000 rows. + running, ...) reached its own cap (`actionable_executions_limit`), + and when listing either execution class failed, so its rows are + missing. limit: type: integer - description: Row cap on completed purchase history rows for this request. + description: >- + Page size: the most completed purchases and terminal executions + together that this response lists. executions_limit: type: integer description: >- - Row cap on terminal executions (failed, expired, canceled, ...), - fixed and ignoring the limit parameter. Actionable executions use a - separate cap of 1000. + Row cap on terminal executions (failed, expired, canceled, ...). + They share the page with completed purchases, so it equals + `limit`. Actionable executions use the separate + `actionable_executions_limit` cap. + actionable_executions_limit: + type: integer + description: >- + Row cap on actionable executions (pending, notified, scheduled, + approved, running, paused), fixed at 1000 and ignoring the limit + parameter. Actionable executions are listed on the first page only. + has_more: + type: boolean + description: >- + True when more completed purchases or terminal executions the + caller may see follow this page; fetch them with `next_cursor`. + `truncated` is also true then. When `truncated` is true and + `has_more` is false, the missing rows cannot be paged to: the + actionable executions reached their cap, or an execution listing + failed. + next_cursor: + type: string + description: >- + Opaque position after the last paged row of this response; pass it + as the `cursor` parameter to fetch the next page. Present only when + `has_more` is true. HistorySummary: type: object diff --git a/internal/api/types.go b/internal/api/types.go index 9ff2a91f..aa36b2ec 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -959,15 +959,31 @@ type HistoryResponse struct { Summary HistorySummary `json:"summary"` // Truncated is true when a fetch reached its row cap, so older rows (and // possibly pending approvals) may be missing from Purchases and Summary. - // It also fires when the actionable execution class hits its MaxListLimit cap. + // It also fires when the actionable execution class hits its MaxListLimit cap, + // and when an execution class listing failed, so its rows are missing. // It can be a false positive when exactly a cap's worth of rows exist. Truncated bool `json:"truncated"` - // Limit is the row cap on completed purchase_history rows for this request. + // Limit is the page size: the most completed purchase_history rows and + // terminal executions (failed, expired, canceled, ...) together that one + // response lists (issue #674). Limit int `json:"limit"` - // ExecutionsLimit is the row cap on terminal executions (failed, expired, - // canceled, ...), fixed regardless of the request's limit parameter. - // Actionable executions (pending, approved, ...) use config.MaxListLimit. + // ExecutionsLimit is the row cap on terminal executions. They share the + // page with completed purchases, so it equals Limit. + // Actionable executions (pending, approved, ...) use ActionableExecutionsLimit. ExecutionsLimit int `json:"executions_limit"` + // ActionableExecutionsLimit is the row cap on actionable executions + // (pending, notified, scheduled, approved, running, paused), fixed at + // config.MaxListLimit regardless of the request's limit parameter. + // Actionable executions are not paged: the first page lists them all, up + // to this cap, and pages fetched with a cursor leave them out. + ActionableExecutionsLimit int `json:"actionable_executions_limit"` + // HasMore is true when more completed purchases or terminal executions the + // caller may see follow this page; pass NextCursor as the cursor parameter + // to fetch them (issue #674). Truncated is also true then. + HasMore bool `json:"has_more"` + // NextCursor is the opaque position after the last paged row of this + // response, set only when HasMore is true. + NextCursor string `json:"next_cursor,omitempty"` } // HistorySummary provides aggregate statistics for purchase history. diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 9a078709..c8ff3bd6 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -155,6 +155,16 @@ 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) + // 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 + // HistoryPageKey, capped at the limit. The executions half excludes clean + // completed rows like GetExecutionsByStatuses and bounds scheduled_date + // inclusively; the purchase_history half applies GetPurchaseHistoryFiltered's + // predicate. Rows inserted or changed between two page reads never shift + // the rows a later page returns. + ListHistoryExecutionsPage(ctx context.Context, q HistoryExecutionsPageQuery) ([]PurchaseExecution, error) + ListPurchaseHistoryPage(ctx context.Context, filter PurchaseHistoryFilter, after *HistoryPageKey) ([]PurchaseHistoryPageRow, error) // CountExecutionsByPlanAndStatus returns, keyed by plan ID, the exact // number of executions in each of the supplied statuses that last // changed state (updated_at) at or after `since` -- not those merely diff --git a/internal/config/store_postgres_history_page.go b/internal/config/store_postgres_history_page.go new file mode 100644 index 00000000..d1e86695 --- /dev/null +++ b/internal/config/store_postgres_history_page.go @@ -0,0 +1,232 @@ +package config + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5" +) + +// HistorySource identifies the table a row of the History view's paged stream +// comes from. Its value is part of the stream order (see HistoryPageKey), so +// the constants must keep their relative order. +type HistorySource int + +const ( + // HistorySourceExecution is a terminal purchase_executions row (failed, + // expired, canceled, audit-gap completed, partially completed). + HistorySourceExecution HistorySource = 1 + // HistorySourcePurchase is a purchase_history row. + HistorySourcePurchase HistorySource = 2 +) + +// HistoryPageKey is a position in the History view's paged stream (issue +// #674): purchase_history rows and terminal executions merged in one order, +// timestamp DESC, then Source DESC, then ID DESC. Timestamp is +// purchase_history.timestamp or purchase_executions.scheduled_date, and ID is +// purchase_history.id or purchase_executions.execution_id (both UUIDs, unique +// per table), so the key is unique across the stream and a keyset page never +// repeats or skips a row. +type HistoryPageKey struct { + Timestamp time.Time + Source HistorySource + ID string +} + +// SortsBefore reports whether k comes before o in the stream order. IDs are +// compared as canonical lowercase UUID strings, which order the same way as +// PostgreSQL's uuid type. +func (k HistoryPageKey) SortsBefore(o HistoryPageKey) bool { + if !k.Timestamp.Equal(o.Timestamp) { + return k.Timestamp.After(o.Timestamp) + } + if k.Source != o.Source { + return k.Source > o.Source + } + return k.ID > o.ID +} + +// PurchaseHistoryPageRow is a purchase_history row read by +// ListPurchaseHistoryPage, with the row's primary key: the keyset tiebreaker +// for rows that share a timestamp, which purchase_id cannot be because legacy +// rows may repeat it (migration 000108). +type PurchaseHistoryPageRow struct { + PurchaseHistoryRecord + RowID string +} + +// Key returns the row's position in the History stream. +func (r *PurchaseHistoryPageRow) Key() HistoryPageKey { + return HistoryPageKey{Timestamp: r.Timestamp, Source: HistorySourcePurchase, ID: r.RowID} +} + +// HistoryExecutionsPageQuery selects one keyset page of executions for the +// History stream. +type HistoryExecutionsPageQuery struct { + // Statuses the rows must have. Clean completed rows (status completed, + // empty error) are always excluded, as in GetExecutionsByStatuses. + Statuses []string + // Start and End bound scheduled_date, inclusive; nil leaves a side open. + Start *time.Time + End *time.Time + // After is the stream position to continue from; nil starts at the top. + After *HistoryPageKey + // Limit caps the row count; clamped to [1, MaxListLimit] with a + // DefaultListLimit fallback when <= 0. + Limit int +} + +// clampHistoryPageLimit applies the list limit bounds every History page +// query shares. +func clampHistoryPageLimit(limit int) int { + if limit <= 0 { + return DefaultListLimit + } + return min(limit, MaxListLimit) +} + +// historyKeysetCondition returns the predicate selecting the rows of source +// src that come after `after` in the stream order, and the bind args with its +// parameters appended. At an equal timestamp a lower source sorts later, so +// when src differs from the cursor's source the timestamp comparison alone +// decides; within the cursor's own source the ID breaks the tie. +func historyKeysetCondition(tsCol, idCol string, src HistorySource, after *HistoryPageKey, args []any) (cond string, outArgs []any) { + args = append(args, after.Timestamp) + ts := len(args) + switch { + case src < after.Source: + return fmt.Sprintf("%s <= $%d", tsCol, ts), args + case src > after.Source: + return fmt.Sprintf("%s < $%d", tsCol, ts), args + } + args = append(args, after.ID) + return fmt.Sprintf("(%s < $%d OR (%s = $%d AND %s < $%d::uuid))", tsCol, ts, tsCol, ts, idCol, len(args)), args +} + +// purchaseHistoryPageConds builds the WHERE conditions for a +// PurchaseHistoryFilter. They are the same conditions GetPurchaseHistoryFiltered +// applies (TestIntegration_ListPurchaseHistoryPage_MatchesFilteredFetch guards +// that), so a page never lists a row the filtered fetch would not. +func purchaseHistoryPageConds(filter *PurchaseHistoryFilter) (conds []string, args []any) { + if filter.Provider != "" { + args = append(args, filter.Provider) + conds = append(conds, fmt.Sprintf("provider = $%d", len(args))) + } + conds, args = appendAccountPredicate(conds, args, filter.AccountIDs, filter.ExternalIDsByProvider) + if filter.Start != nil { + args = append(args, *filter.Start) + conds = append(conds, fmt.Sprintf("timestamp >= $%d", len(args))) + } + if filter.End != nil { + args = append(args, *filter.End) + conds = append(conds, fmt.Sprintf("timestamp <= $%d", len(args))) + } + return conds, args +} + +// whereClause joins conditions into a WHERE clause with a leading space, or +// "" when there are none. +func whereClause(conds []string) string { + if len(conds) == 0 { + return "" + } + return " WHERE " + strings.Join(conds, " AND ") +} + +// ListPurchaseHistoryPage returns up to filter.Limit purchase_history rows +// matching filter, in History stream order (timestamp DESC, id DESC), +// starting after `after` (nil starts at the newest row). See +// StoreInterface.ListPurchaseHistoryPage. +func (s *PostgresStore) ListPurchaseHistoryPage(ctx context.Context, filter PurchaseHistoryFilter, after *HistoryPageKey) ([]PurchaseHistoryPageRow, error) { + conds, args := purchaseHistoryPageConds(&filter) + if after != nil { + var cond string + cond, args = historyKeysetCondition("timestamp", "id", HistorySourcePurchase, after, args) + conds = append(conds, cond) + } + args = append(args, clampHistoryPageLimit(filter.Limit)) + query := fmt.Sprintf(` + SELECT id::text, 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, + revocation_window_closes_at, revoked_at, revoked_via, support_case_id, + offering_class, listing_id, listing_state + FROM purchase_history%s + ORDER BY timestamp DESC, id DESC + LIMIT $%d + `, whereClause(conds), len(args)) + + rows, err := s.db.Query(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("failed to query purchase history page: %w", err) + } + defer rows.Close() + + out := make([]PurchaseHistoryPageRow, 0) + for rows.Next() { + var row PurchaseHistoryPageRow + record, err := scanPurchaseHistoryRow(rowIDPrefixedRows{Rows: rows, id: &row.RowID}) + if err != nil { + return nil, err + } + row.PurchaseHistoryRecord = record + out = append(out, row) + } + return out, rows.Err() +} + +// rowIDPrefixedRows lets scanPurchaseHistoryRow, the one scanner for the +// purchase_history column list, read a result set whose first column is the +// row's id: Scan reads that column into id and hands the rest to the caller's +// destinations. +type rowIDPrefixedRows struct { + pgx.Rows + id *string +} + +func (r rowIDPrefixedRows) Scan(dest ...any) error { + return r.Rows.Scan(append([]any{r.id}, dest...)...) +} + +// ListHistoryExecutionsPage returns up to q.Limit executions in q.Statuses, +// clean completed rows excluded, in History stream order (scheduled_date DESC, +// execution_id DESC), starting after q.After. See +// StoreInterface.ListHistoryExecutionsPage. +func (s *PostgresStore) ListHistoryExecutionsPage(ctx context.Context, q HistoryExecutionsPageQuery) ([]PurchaseExecution, error) { + if len(q.Statuses) == 0 { + return nil, nil + } + args := []any{q.Statuses} + conds := []string{"status = ANY($1)", "NOT (status = 'completed' AND COALESCE(error, '') = '')"} + if q.Start != nil { + args = append(args, *q.Start) + conds = append(conds, fmt.Sprintf("scheduled_date >= $%d", len(args))) + } + if q.End != nil { + args = append(args, *q.End) + conds = append(conds, fmt.Sprintf("scheduled_date <= $%d", len(args))) + } + if q.After != nil { + var cond string + cond, args = historyKeysetCondition("scheduled_date", "execution_id", HistorySourceExecution, q.After, args) + conds = append(conds, cond) + } + args = append(args, clampHistoryPageLimit(q.Limit)) + // cancelled_by is the legacy column name (migration 000035, rename tracked in #1277). + query := fmt.Sprintf(` + SELECT plan_id, execution_id, status, step_number, scheduled_date, + notification_sent, approval_token_hash, recommendations, + total_upfront_cost, estimated_savings, completed_at, error, expires_at, + cloud_account_id, source, approved_by, COALESCE(canceled_by, cancelled_by) AS cancelled_by, capacity_percent, + created_by_user_id, retry_execution_id, retry_attempt_n, + approval_token_expires_at, + executed_by_user_id, executed_at, pre_approval_skip_reason, + idempotency_key, scheduled_execution_at, no_submission, created_by_system + FROM purchase_executions%s + ORDER BY scheduled_date DESC, execution_id DESC + LIMIT $%d + `, whereClause(conds), len(args)) + return s.queryExecutions(ctx, query, args...) +} diff --git a/internal/config/store_postgres_history_page_integration_test.go b/internal/config/store_postgres_history_page_integration_test.go new file mode 100644 index 00000000..94c70bf9 --- /dev/null +++ b/internal/config/store_postgres_history_page_integration_test.go @@ -0,0 +1,270 @@ +//go:build integration +// +build integration + +package config + +import ( + "context" + "sort" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// historyPageFixture is a store with purchase_history rows and executions +// whose timestamps collide within and across the two tables. +type historyPageFixture struct { + store *PostgresStore + keys map[string]HistoryPageKey // purchase_id or execution_id -> stream key +} + +func newHistoryPageFixture(ctx context.Context, t *testing.T) *historyPageFixture { + t.Helper() + return &historyPageFixture{store: setupRampStepStore(ctx, t), keys: map[string]HistoryPageKey{}} +} + +func (f *historyPageFixture) addPurchase(ctx context.Context, t *testing.T, provider, account string, cloudAccountID *string, at time.Time) string { + t.Helper() + rec := &PurchaseHistoryRecord{ + AccountID: account, PurchaseID: "ph-" + uuid.NewString(), Timestamp: at, Provider: provider, + Service: "ec2", Region: "us-east-1", ResourceType: "t3.micro", Count: 1, Term: 1, Payment: "no-upfront", + CloudAccountID: cloudAccountID, + } + require.NoError(t, f.store.SavePurchaseHistory(ctx, rec)) + var id string + require.NoError(t, f.store.db.QueryRow(ctx, `SELECT id::text FROM purchase_history WHERE purchase_id = $1`, rec.PurchaseID).Scan(&id)) + f.keys[rec.PurchaseID] = HistoryPageKey{Timestamp: at, Source: HistorySourcePurchase, ID: id} + return rec.PurchaseID +} + +func (f *historyPageFixture) addExecution(ctx context.Context, t *testing.T, status, execErr string, at time.Time) string { + t.Helper() + exec := &PurchaseExecution{ + ExecutionID: uuid.NewString(), IdempotencyKey: uuid.NewString(), Status: status, Error: execErr, ScheduledDate: at, + } + require.NoError(t, f.store.SavePurchaseExecution(ctx, exec)) + f.keys[exec.ExecutionID] = HistoryPageKey{Timestamp: at, Source: HistorySourceExecution, ID: exec.ExecutionID} + return exec.ExecutionID +} + +// sortedIDs orders ids by their stream key. +func (f *historyPageFixture) sortedIDs(ids []string) []string { + out := append([]string(nil), ids...) + sort.Slice(out, func(i, j int) bool { return f.keys[out[i]].SortsBefore(f.keys[out[j]]) }) + return out +} + +// walkPurchases pages ListPurchaseHistoryPage to the end. between, when set, +// runs after each page. +func walkPurchases(ctx context.Context, t *testing.T, s *PostgresStore, filter PurchaseHistoryFilter, between func(int)) []string { + t.Helper() + var got []string + var after *HistoryPageKey + for page := 0; ; page++ { + require.Less(t, page, 500) + rows, err := s.ListPurchaseHistoryPage(ctx, filter, after) + require.NoError(t, err) + for i := range rows { + got = append(got, rows[i].PurchaseID) + } + if between != nil { + between(page) + } + if len(rows) < filter.Limit { + return got + } + k := rows[len(rows)-1].Key() + after = &k + } +} + +func walkExecutions(ctx context.Context, t *testing.T, s *PostgresStore, q HistoryExecutionsPageQuery) []string { + t.Helper() + var got []string + for page := 0; ; page++ { + require.Less(t, page, 500) + rows, err := s.ListHistoryExecutionsPage(ctx, q) + require.NoError(t, err) + for i := range rows { + got = append(got, rows[i].ExecutionID) + } + if len(rows) < q.Limit { + return got + } + last := rows[len(rows)-1] + q.After = &HistoryPageKey{Timestamp: last.ScheduledDate, Source: HistorySourceExecution, ID: last.ExecutionID} + } +} + +// TestIntegration_ListPurchaseHistoryPage_WalksEveryRowOnce pages through +// purchase_history rows in groups that share a timestamp, with pages ending +// inside a group. Every row comes back once, in (timestamp DESC, id DESC) +// order, and rows inserted above the cursor between pages do not shift the +// walk. +func TestIntegration_ListPurchaseHistoryPage_WalksEveryRowOnce(t *testing.T) { + ctx := context.Background() + f := newHistoryPageFixture(ctx, t) + base := time.Now().UTC().Add(-time.Hour).Truncate(time.Microsecond) + var ids []string + for i := 0; i < 37; i++ { + ids = append(ids, f.addPurchase(ctx, t, "aws", "111111111111", nil, base.Add(-time.Duration(i/4)*time.Minute))) + } + + var inserted []string + got := walkPurchases(ctx, t, f.store, PurchaseHistoryFilter{Limit: 5}, func(page int) { + if page == 1 { + inserted = append(inserted, f.addPurchase(ctx, t, "aws", "111111111111", nil, time.Now().UTC().Truncate(time.Microsecond))) + } + }) + + assert.Equal(t, f.sortedIDs(ids), got) + assert.NotContains(t, got, inserted[0]) +} + +// TestIntegration_ListHistoryExecutionsPage_WalksEveryRowOnce is the same walk +// over executions, with rows the query must exclude mixed in: a clean +// completed execution and a status outside the requested set. +func TestIntegration_ListHistoryExecutionsPage_WalksEveryRowOnce(t *testing.T) { + ctx := context.Background() + f := newHistoryPageFixture(ctx, t) + base := time.Now().UTC().Add(-time.Hour).Truncate(time.Microsecond) + statuses := []string{"failed", "expired", StatusCanceled, "partially_completed"} + var ids []string + for i := 0; i < 29; i++ { + ids = append(ids, f.addExecution(ctx, t, statuses[i%len(statuses)], "boom", base.Add(-time.Duration(i/3)*time.Minute))) + } + ids = append(ids, f.addExecution(ctx, t, "completed", "history write failed", base)) + f.addExecution(ctx, t, "completed", "", base) + f.addExecution(ctx, t, "pending", "", base) + + got := walkExecutions(ctx, t, f.store, HistoryExecutionsPageQuery{ + Statuses: append(statuses, "completed"), Limit: 4, + }) + + assert.Equal(t, f.sortedIDs(ids), got) +} + +// TestIntegration_HistoryKeyset_CrossSourceTies pins the keyset predicate +// where the two tables meet. At an equal timestamp purchase_history rows sort +// before executions, so a cursor on a purchase row still returns executions at +// that timestamp, and a cursor on an execution no longer returns purchase rows +// at that timestamp. +func TestIntegration_HistoryKeyset_CrossSourceTies(t *testing.T) { + ctx := context.Background() + f := newHistoryPageFixture(ctx, t) + at := time.Now().UTC().Add(-time.Hour).Truncate(time.Microsecond) + p1 := f.addPurchase(ctx, t, "aws", "111111111111", nil, at) + p2 := f.addPurchase(ctx, t, "aws", "111111111111", nil, at) + e1 := f.addExecution(ctx, t, "failed", "boom", at) + e2 := f.addExecution(ctx, t, "failed", "boom", at) + older := f.addPurchase(ctx, t, "aws", "111111111111", nil, at.Add(-time.Microsecond)) + olderExec := f.addExecution(ctx, t, "failed", "boom", at.Add(-time.Microsecond)) + order := f.sortedIDs([]string{p1, p2, e1, e2, older, olderExec}) + require.Equal(t, HistorySourcePurchase, f.keys[order[0]].Source) + require.Equal(t, HistorySourcePurchase, f.keys[order[1]].Source) + + purchasesAfter := func(k HistoryPageKey) []string { + rows, err := f.store.ListPurchaseHistoryPage(ctx, PurchaseHistoryFilter{Limit: 10}, &k) + require.NoError(t, err) + ids := make([]string, len(rows)) + for i := range rows { + ids[i] = rows[i].PurchaseID + } + return ids + } + execsAfter := func(k HistoryPageKey) []string { + rows, err := f.store.ListHistoryExecutionsPage(ctx, HistoryExecutionsPageQuery{Statuses: []string{"failed"}, Limit: 10, After: &k}) + require.NoError(t, err) + ids := make([]string, len(rows)) + for i := range rows { + ids[i] = rows[i].ExecutionID + } + return ids + } + + // Cursor on the first purchase row at `at`. + assert.Equal(t, []string{order[1], older}, purchasesAfter(f.keys[order[0]])) + assert.Equal(t, []string{order[2], order[3], olderExec}, execsAfter(f.keys[order[0]])) + // Cursor on the first execution at `at`: no purchase row at `at` is left. + assert.Equal(t, []string{older}, purchasesAfter(f.keys[order[2]])) + assert.Equal(t, []string{order[3], olderExec}, execsAfter(f.keys[order[2]])) +} + +// TestIntegration_ListPurchaseHistoryPage_MatchesFilteredFetch guards the +// page query's predicate against GetPurchaseHistoryFiltered's: with no cursor +// and a limit above the row count both must select the same rows for every +// filter shape (provider, both account columns, the date bounds). +func TestIntegration_ListPurchaseHistoryPage_MatchesFilteredFetch(t *testing.T) { + ctx := context.Background() + f := newHistoryPageFixture(ctx, t) + acct := &CloudAccount{Name: "Paging drift", Provider: "aws", ExternalID: "222222222222", Enabled: true} + require.NoError(t, f.store.CreateCloudAccount(ctx, acct)) + start := time.Now().UTC().Add(-48 * time.Hour).Truncate(time.Microsecond) + end := start.Add(24 * time.Hour) + for _, at := range []time.Time{start.Add(-time.Microsecond), start, start.Add(time.Hour), end, end.Add(time.Microsecond)} { + f.addPurchase(ctx, t, "aws", acct.ExternalID, &acct.ID, at) + f.addPurchase(ctx, t, "aws", acct.ExternalID, nil, at) + f.addPurchase(ctx, t, "AWS", acct.ExternalID, nil, at) + f.addPurchase(ctx, t, "", acct.ExternalID, nil, at) + f.addPurchase(ctx, t, "azure", acct.ExternalID, nil, at) + f.addPurchase(ctx, t, "aws", "333333333333", nil, at) + } + filters := map[string]PurchaseHistoryFilter{ + "none": {}, + "provider": {Provider: "aws"}, + "cloud account": {AccountIDs: []string{acct.ID}}, + "external by aws": {ExternalIDsByProvider: map[string][]string{"aws": {acct.ExternalID}}}, + "external unknown": {ExternalIDsByProvider: map[string][]string{"": {acct.ExternalID}}}, + "both account cols": {AccountIDs: []string{acct.ID}, ExternalIDsByProvider: map[string][]string{"aws": {acct.ExternalID}}}, + "date window": {Start: &start, End: &end}, + "everything": {Provider: "aws", AccountIDs: []string{acct.ID}, ExternalIDsByProvider: map[string][]string{"aws": {acct.ExternalID}}, Start: &start, End: &end}, + } + for name, filter := range filters { + t.Run(name, func(t *testing.T) { + filter.Limit = MaxListLimit + want, err := f.store.GetPurchaseHistoryFiltered(ctx, filter) + require.NoError(t, err) + page, err := f.store.ListPurchaseHistoryPage(ctx, filter, nil) + require.NoError(t, err) + wantIDs := make([]string, len(want)) + for i := range want { + wantIDs[i] = want[i].PurchaseID + } + gotIDs := make([]string, len(page)) + for i := range page { + gotIDs[i] = page[i].PurchaseID + assert.Equal(t, f.keys[page[i].PurchaseID].ID, page[i].RowID, "row id scanned") + } + require.NotEmpty(t, wantIDs, "a vacuous comparison proves nothing") + assert.ElementsMatch(t, wantIDs, gotIDs) + assert.Equal(t, f.sortedIDs(gotIDs), gotIDs, "stream order") + }) + } +} + +// TestIntegration_ListHistoryExecutionsPage_DateBoundsInclusive checks the +// scheduled_date bounds include both edges, like the History handler's +// in-memory date check, and nothing a microsecond outside them. +func TestIntegration_ListHistoryExecutionsPage_DateBoundsInclusive(t *testing.T) { + ctx := context.Background() + f := newHistoryPageFixture(ctx, t) + start := time.Now().UTC().Add(-48 * time.Hour).Truncate(time.Microsecond) + end := start.Add(24 * time.Hour) + f.addExecution(ctx, t, "failed", "boom", start.Add(-time.Microsecond)) + onStart := f.addExecution(ctx, t, "failed", "boom", start) + onEnd := f.addExecution(ctx, t, "failed", "boom", end) + f.addExecution(ctx, t, "failed", "boom", end.Add(time.Microsecond)) + + rows, err := f.store.ListHistoryExecutionsPage(ctx, HistoryExecutionsPageQuery{ + Statuses: []string{"failed"}, Start: &start, End: &end, Limit: 10, + }) + require.NoError(t, err) + ids := make([]string, len(rows)) + for i := range rows { + ids[i] = rows[i].ExecutionID + } + assert.Equal(t, []string{onEnd, onStart}, ids) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index e946ed5f..049ae8be 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1900,6 +1900,114 @@ func (m *MockConfigStore) GetExecutionsByStatuses(ctx context.Context, statuses return filterExecutionsLikeStore(v, statuses, limit), args.Error(1) } +// ListHistoryExecutionsPage mocks the ListHistoryExecutionsPage operation. +// With no expectation registered it returns no rows. A registered +// expectation's rows go through the store contract (see +// pageExecutionsLikeStore), so a test can return one unordered fixture and +// page through it. +func (m *MockConfigStore) ListHistoryExecutionsPage(ctx context.Context, q config.HistoryExecutionsPageQuery) ([]config.PurchaseExecution, error) { + m.record("ListHistoryExecutionsPage", ctx, q) + if !isExpected(&m.Mock, "ListHistoryExecutionsPage") { + return nil, nil + } + args := m.Called(ctx, q) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).([]config.PurchaseExecution) + if !ok { + panic(fmt.Sprintf("mock: expected []config.PurchaseExecution, got %T", args.Get(0))) + } + return pageExecutionsLikeStore(v, &q), args.Error(1) +} + +// ListPurchaseHistoryPage mocks the ListPurchaseHistoryPage operation. With +// no expectation registered it returns no rows. A registered expectation's +// rows are ordered, resumed after `after` and capped like the store does; the +// filter's predicates are not applied (match on the filter argument instead). +// The expectation may return []config.PurchaseHistoryPageRow, or plain +// []config.PurchaseHistoryRecord fixtures, which get row IDs from their +// position in the fixture (see PurchaseHistoryFixtureRowID). +func (m *MockConfigStore) ListPurchaseHistoryPage(ctx context.Context, filter config.PurchaseHistoryFilter, after *config.HistoryPageKey) ([]config.PurchaseHistoryPageRow, error) { + m.record("ListPurchaseHistoryPage", ctx, filter, after) + if !isExpected(&m.Mock, "ListPurchaseHistoryPage") { + return nil, nil + } + args := m.Called(ctx, filter, after) + if args.Get(0) == nil { + return nil, args.Error(1) + } + var v []config.PurchaseHistoryPageRow + switch rows := args.Get(0).(type) { + case []config.PurchaseHistoryPageRow: + v = rows + case []config.PurchaseHistoryRecord: + v = make([]config.PurchaseHistoryPageRow, len(rows)) + for i := range rows { + v[i] = config.PurchaseHistoryPageRow{PurchaseHistoryRecord: rows[i], RowID: PurchaseHistoryFixtureRowID(i)} + } + default: + panic(fmt.Sprintf("mock: expected []config.PurchaseHistoryPageRow or []config.PurchaseHistoryRecord, got %T", args.Get(0))) + } + out := make([]config.PurchaseHistoryPageRow, 0, len(v)) + for i := range v { + if after == nil || after.SortsBefore(v[i].Key()) { + out = append(out, v[i]) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].Key().SortsBefore(out[j].Key()) }) + return capHistoryPage(out, filter.Limit), args.Error(1) +} + +// pageExecutionsLikeStore applies ListHistoryExecutionsPage's contract to +// canned rows: status filter, clean completed rows dropped, inclusive +// scheduled_date bounds, rows after q.After only, stream order, then the cap. +func pageExecutionsLikeStore(rows []config.PurchaseExecution, q *config.HistoryExecutionsPageQuery) []config.PurchaseExecution { + var out []config.PurchaseExecution + for i := range rows { + if executionMatchesPageQuery(&rows[i], q) { + out = append(out, rows[i]) + } + } + sort.SliceStable(out, func(i, j int) bool { + return executionPageKey(&out[i]).SortsBefore(executionPageKey(&out[j])) + }) + return capHistoryPage(out, q.Limit) +} + +// PurchaseHistoryFixtureRowID is the row ID ListPurchaseHistoryPage gives the +// i-th row of a []config.PurchaseHistoryRecord fixture: a valid UUID, so a +// cursor built from it decodes, and later fixture rows sort lower. +func PurchaseHistoryFixtureRowID(i int) string { + return fmt.Sprintf("00000000-0000-4000-8000-%012x", 0xffffffffffff-i) +} + +func executionMatchesPageQuery(r *config.PurchaseExecution, q *config.HistoryExecutionsPageQuery) bool { + if !slices.Contains(q.Statuses, r.Status) || (r.Status == "completed" && r.Error == "") { + return false + } + if (q.Start != nil && r.ScheduledDate.Before(*q.Start)) || (q.End != nil && r.ScheduledDate.After(*q.End)) { + return false + } + return q.After == nil || q.After.SortsBefore(executionPageKey(r)) +} + +func executionPageKey(e *config.PurchaseExecution) config.HistoryPageKey { + return config.HistoryPageKey{Timestamp: e.ScheduledDate, Source: config.HistorySourceExecution, ID: e.ExecutionID} +} + +// capHistoryPage applies the History page limit bounds the store uses. +func capHistoryPage[T any](rows []T, limit int) []T { + if limit <= 0 { + limit = config.DefaultListLimit + } + limit = min(limit, config.MaxListLimit) + if len(rows) > limit { + rows = rows[:limit] + } + return rows +} + // filterExecutionsLikeStore applies the store contract to canned rows: filter by // status, drop clean completed rows, order by ScheduledDate DESC with future // pending rows last and soonest first (issue #680), then cap. diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 845618a1..6437da6c 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -106,6 +106,14 @@ func (m *mockConfigStoreForHealth) GetPlannedExecutions(ctx context.Context, sta return nil, nil } +func (m *mockConfigStoreForHealth) ListHistoryExecutionsPage(ctx context.Context, q config.HistoryExecutionsPageQuery) ([]config.PurchaseExecution, error) { + return nil, nil +} + +func (m *mockConfigStoreForHealth) ListPurchaseHistoryPage(ctx context.Context, filter config.PurchaseHistoryFilter, after *config.HistoryPageKey) ([]config.PurchaseHistoryPageRow, error) { + return nil, nil +} + func (m *mockConfigStoreForHealth) GetStaleApprovedExecutions(ctx context.Context, olderThan time.Duration) ([]config.PurchaseExecution, error) { return nil, nil }