Refactor access review campaign source API

Expose campaign sources as first-class nodes, paginate fetch attempts
instead of denormalized status fields, and bind entries to their
campaign snapshot. Update GraphQL, MCP, CLI, console, and e2e coverage
to match.

Signed-off-by: Bryan Frimin <bryan@probo.com>

pkg/coredata/access_review_campaign_source_fetch_attempt.go

@@ -24,13 +24,14 @@ import (
 	"github.com/jackc/pgx/v5"
 	"go.gearno.de/kit/pg"
 	"go.probo.inc/probo/pkg/gid"
+	"go.probo.inc/probo/pkg/page"
 )
 
 type (
 	// AccessReviewCampaignSourceFetchAttempt is a single, append-only fetch run for a
 	// campaign source snapshot. Each retry produces a new row, so the error of
 	// every attempt is retained. The current state of a snapshot is the latest
-	// attempt (highest attempt_number). Terminal rows (SUCCESS / FAILED) are
+	// attempt (most recently created). Terminal rows (SUCCESS / FAILED) are
 	// immutable; only the in-flight attempt is updated.
 	//
 	// TenantID is retained on the struct because the background worker claims
@@ -40,7 +41,6 @@ type (
 		ID                           gid.GID                               `db:"id"`
 		TenantID                     gid.TenantID                          `db:"tenant_id"`
 		AccessReviewCampaignSourceID gid.GID                               `db:"access_review_campaign_source_id"`
-		AttemptNumber                int                                   `db:"attempt_number"`
 		Status                       AccessReviewCampaignSourceFetchStatus `db:"status"`
 		FetchedAccountsCount         int                                   `db:"fetched_accounts_count"`
 		Error                        *string                               `db:"error"`
@@ -48,11 +48,23 @@ type (
 		CompletedAt                  *time.Time                            `db:"completed_at"`
 		CreatedAt                    time.Time                             `db:"created_at"`
 		UpdatedAt                    time.Time                             `db:"updated_at"`
+		AttemptNumber                int                                   `db:"attempt_number"`
 	}
 
 	AccessReviewCampaignSourceFetchAttempts []*AccessReviewCampaignSourceFetchAttempt
 )
 
+func (a AccessReviewCampaignSourceFetchAttempt) CursorKey(
+	orderBy AccessReviewCampaignSourceFetchAttemptOrderField,
+) page.CursorKey {
+	switch orderBy {
+	case AccessReviewCampaignSourceFetchAttemptOrderFieldCreatedAt:
+		return page.NewCursorKey(a.ID, a.CreatedAt)
+	}
+
+	panic(fmt.Sprintf("unsupported order by: %s", orderBy))
+}
+
 var (
 	ErrNoAccessReviewCampaignSourceFetchAttemptAvailable = errors.New("no access review source fetch attempt available")
 )
@@ -185,7 +197,7 @@ SELECT
 	updated_at
 FROM access_review_campaign_source_fetch_attempts
 WHERE status = @status
-ORDER BY created_at ASC
+ORDER BY created_at ASC, id ASC
 LIMIT 1
 FOR UPDATE SKIP LOCKED
 `
@@ -263,13 +275,76 @@ ORDER BY access_review_campaign_source_id, attempt_number DESC
 	return nil
 }
 
-// LoadByCampaignSourceID returns the full attempt history for a snapshot,
-// newest first.
+// LoadByCampaignSourceID returns a page of fetch attempts for a snapshot.
 func (attempts *AccessReviewCampaignSourceFetchAttempts) LoadByCampaignSourceID(
 	ctx context.Context,
 	conn pg.Querier,
 	scope Scoper,
 	campaignSourceID gid.GID,
+	cursor *page.Cursor[AccessReviewCampaignSourceFetchAttemptOrderField],
+) error {
+	q := `
+SELECT
+	id,
+	tenant_id,
+	access_review_campaign_source_id,
+	status,
+	fetched_accounts_count,
+	error,
+	started_at,
+	completed_at,
+	created_at,
+	updated_at,
+	attempt_number
+FROM (
+	SELECT
+		id,
+		tenant_id,
+		access_review_campaign_source_id,
+		status,
+		fetched_accounts_count,
+		error,
+		started_at,
+		completed_at,
+		created_at,
+		updated_at,
+		ROW_NUMBER() OVER (ORDER BY created_at ASC, id ASC)::int AS attempt_number
+	FROM access_review_campaign_source_fetch_attempts
+	WHERE
+		%s
+		AND access_review_campaign_source_id = @access_review_campaign_source_id
+) fetch_attempts
+WHERE
+	%s
+`
+	q = fmt.Sprintf(q, scope.SQLFragment(), cursor.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"access_review_campaign_source_id": campaignSourceID}
+	maps.Copy(args, scope.SQLArguments())
+	maps.Copy(args, cursor.SQLArguments())
+
+	rows, err := conn.Query(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot query fetch attempts: %w", err)
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByName[AccessReviewCampaignSourceFetchAttempt])
+	if err != nil {
+		return fmt.Errorf("cannot collect fetch attempts: %w", err)
+	}
+
+	*attempts = result
+
+	return nil
+}
+
+// LoadAllByCampaignSourceID returns the full attempt history for a snapshot,
+// newest first.
+func (attempts *AccessReviewCampaignSourceFetchAttempts) LoadAllByCampaignSourceID(
+	ctx context.Context,
+	conn pg.Querier,
+	scope Scoper,
+	campaignSourceID gid.GID,
 ) error {
 	q := `
 SELECT
@@ -310,6 +385,33 @@ ORDER BY attempt_number DESC
 	return nil
 }
 
+func (attempts *AccessReviewCampaignSourceFetchAttempts) CountByCampaignSourceID(
+	ctx context.Context,
+	conn pg.Querier,
+	scope Scoper,
+	campaignSourceID gid.GID,
+) (int, error) {
+	q := `
+SELECT COUNT(*)
+FROM access_review_campaign_source_fetch_attempts
+WHERE
+	%s
+	AND access_review_campaign_source_id = @access_review_campaign_source_id
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+
+	args := pgx.StrictNamedArgs{"access_review_campaign_source_id": campaignSourceID}
+	maps.Copy(args, scope.SQLArguments())
+
+	var count int
+
+	if err := conn.QueryRow(ctx, q, args).Scan(&count); err != nil {
+		return 0, fmt.Errorf("cannot count fetch attempts: %w", err)
+	}
+
+	return count, nil
+}
+
 // RecoverStale fails attempts stuck in FETCHING past the threshold and queues a
 // fresh retry attempt for each, preserving the stale attempt's history. It is
 // intentionally cross-tenant. Returns the number of recovered attempts.

pkg/coredata/access_review_campaign_source_fetch_attempt_order_field.go

@@ -0,0 +1,80 @@
+// Copyright (c) 2026 Probo Inc <hello@probo.com>.
+//
+// Permission to use, copy, modify, and/or distribute this software for any
+// purpose with or without fee is hereby granted, provided that the above
+// copyright notice and this permission notice appear in all copies.
+//
+// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
+// REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
+// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
+// LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
+// OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
+// PERFORMANCE OF THIS SOFTWARE.
+
+package coredata
+
+import (
+	"encoding"
+	"fmt"
+
+	"go.probo.inc/probo/pkg/page"
+)
+
+type (
+	AccessReviewCampaignSourceFetchAttemptOrderField string
+)
+
+const (
+	AccessReviewCampaignSourceFetchAttemptOrderFieldCreatedAt AccessReviewCampaignSourceFetchAttemptOrderField = "CREATED_AT"
+)
+
+var (
+	_ page.OrderField          = AccessReviewCampaignSourceFetchAttemptOrderField("")
+	_ fmt.Stringer             = AccessReviewCampaignSourceFetchAttemptOrderField("")
+	_ encoding.TextMarshaler   = AccessReviewCampaignSourceFetchAttemptOrderField("")
+	_ encoding.TextUnmarshaler = (*AccessReviewCampaignSourceFetchAttemptOrderField)(nil)
+)
+
+func AccessReviewCampaignSourceFetchAttemptOrderFields() []AccessReviewCampaignSourceFetchAttemptOrderField {
+	return []AccessReviewCampaignSourceFetchAttemptOrderField{
+		AccessReviewCampaignSourceFetchAttemptOrderFieldCreatedAt,
+	}
+}
+
+func (v AccessReviewCampaignSourceFetchAttemptOrderField) IsValid() bool {
+	switch v {
+	case AccessReviewCampaignSourceFetchAttemptOrderFieldCreatedAt:
+		return true
+	}
+
+	return false
+}
+
+func (v AccessReviewCampaignSourceFetchAttemptOrderField) String() string {
+	return string(v)
+}
+
+func (v AccessReviewCampaignSourceFetchAttemptOrderField) MarshalText() ([]byte, error) {
+	return []byte(v.String()), nil
+}
+
+func (v *AccessReviewCampaignSourceFetchAttemptOrderField) UnmarshalText(text []byte) error {
+	val := AccessReviewCampaignSourceFetchAttemptOrderField(text)
+	if !val.IsValid() {
+		return fmt.Errorf("invalid AccessReviewCampaignSourceFetchAttemptOrderField value: %q", string(text))
+	}
+
+	*v = val
+
+	return nil
+}
+
+func (p AccessReviewCampaignSourceFetchAttemptOrderField) Column() string {
+	switch p {
+	case AccessReviewCampaignSourceFetchAttemptOrderFieldCreatedAt:
+		return "created_at"
+	}
+
+	panic(fmt.Sprintf("unsupported order by: %s", p))
+}

pkg/coredata/access_review_campaign_source_test.go

@@ -119,7 +119,6 @@ func TestSourceFetchAttempts_AppendOnly(t *testing.T) {
 		if err := first.Insert(ctx, tx, fx.scope); err != nil {
 			return err
 		}
-		require.Equal(t, 1, first.AttemptNumber)
 
 		second := &coredata.AccessReviewCampaignSourceFetchAttempt{
 			ID:                           gid.New(tenantID, coredata.AccessReviewCampaignSourceFetchAttemptEntityType),
@@ -133,17 +132,16 @@ func TestSourceFetchAttempts_AppendOnly(t *testing.T) {
 		if err := second.Insert(ctx, tx, fx.scope); err != nil {
 			return err
 		}
-		require.Equal(t, 2, second.AttemptNumber)
 
 		return nil
 	}))
 
 	var history coredata.AccessReviewCampaignSourceFetchAttempts
 	require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
-		return history.LoadByCampaignSourceID(ctx, conn, fx.scope, fx.campaignSourceID)
+		return history.LoadAllByCampaignSourceID(ctx, conn, fx.scope, fx.campaignSourceID)
 	}))
 	require.Len(t, history, 2, "both attempts must be retained")
-	assert.Equal(t, 2, history[0].AttemptNumber, "history is newest first")
+	assert.Equal(t, coredata.AccessReviewCampaignSourceFetchStatusSuccess, history[0].Status, "history is newest first")
 	assert.Equal(t, coredata.AccessReviewCampaignSourceFetchStatusFailed, history[1].Status)
 	require.NotNil(t, history[1].Error)
 	assert.Equal(t, failureMsg, *history[1].Error, "the failed attempt's error is retained")