From db8383cc2aa264e46ae09ef33ee2b22c4f8c2fff Mon Sep 17 00:00:00 2001 From: Raul Ernesto Guillen Hernandez <37725305+RauPro@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:37:33 -0600 Subject: [PATCH 1/2] perf(identity): index pending outbound counts by agent Add a partial agent_id index matching ListAgentsByUser's all-time outbound pending_review count. Build it concurrently through the existing non-transactional migration runner. Verify index validity and idempotent reapplication, plus per-agent and per-owner counts across directions, statuses and resolved holds. Refs #129 (pending-count index; DNS handling is covered by #1032). Co-authored-by: Codex Signed-off-by: Raul Ernesto Guillen Hernandez <37725305+RauPro@users.noreply.github.com> --- internal/identity/pending_count_index_test.go | 126 ++++++++++++++++++ migrations/132_messages_agent_pending_idx.sql | 17 +++ 2 files changed, 143 insertions(+) create mode 100644 internal/identity/pending_count_index_test.go create mode 100644 migrations/132_messages_agent_pending_idx.sql diff --git a/internal/identity/pending_count_index_test.go b/internal/identity/pending_count_index_test.go new file mode 100644 index 000000000..bd35294d1 --- /dev/null +++ b/internal/identity/pending_count_index_test.go @@ -0,0 +1,126 @@ +package identity_test + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/tokencanopy/e2a/internal/identity" + "github.com/tokencanopy/e2a/internal/testutil" + "github.com/tokencanopy/e2a/migrations" +) + +func TestPendingCountIndexMigration(t *testing.T) { + pool := testutil.TestDB(t) + ctx := context.Background() + + // TestDB applies the embedded migrations through the real runner. Reapply + // the index statement directly to also exercise its IF NOT EXISTS retry. + sql, err := migrations.FS.ReadFile("132_messages_agent_pending_idx.sql") + if err != nil { + t.Fatalf("read pending-count migration: %v", err) + } + if _, err := pool.Exec(ctx, string(sql)); err != nil { + t.Fatalf("reapply pending-count migration: %v", err) + } + + var definition string + var valid, ready bool + if err := pool.QueryRow(ctx, + `SELECT pg_get_indexdef(indexrelid), indisvalid, indisready + FROM pg_index + WHERE indexrelid = 'idx_messages_agent_pending'::regclass + AND indrelid = 'messages'::regclass`, + ).Scan(&definition, &valid, &ready); err != nil { + t.Fatalf("read pending-count index: %v", err) + } + if !valid || !ready { + t.Fatalf("pending-count index valid=%t ready=%t, want both true", valid, ready) + } + for _, fragment := range []string{ + "USING btree (agent_id) WHERE", + "status = 'pending_review'", + "direction = 'outbound'", + } { + if !strings.Contains(definition, fragment) { + t.Fatalf("pending-count index definition %q missing %q", definition, fragment) + } + } +} + +func TestListAgentsByUserPendingCountScopesOutboundHolds(t *testing.T) { + pool := testutil.TestDB(t) + store := identity.NewStore(pool) + ctx := context.Background() + + user, first := setupPendingAgent(t, store, "pending-count") + second, err := store.CreateAgent(ctx, "second@pending-count.example.com", first.Domain, "", "", "", user.ID) + if err != nil { + t.Fatalf("create second agent: %v", err) + } + _, otherOwner := setupPendingAgent(t, store, "pending-count-other") + + // Counts are all-time, per-agent outbound holds, not all review items or + // the seven-day activity totals returned by the neighboring subqueries. + rows := []struct { + id string + agentID string + direction string + status string + ageDays int + }{ + {"msg_pending_count_recent", first.ID, "outbound", "pending_review", 0}, + {"msg_pending_count_old", first.ID, "outbound", "pending_review", 14}, + {"msg_pending_count_inbound", first.ID, "inbound", "pending_review", 0}, + {"msg_pending_count_sent", first.ID, "outbound", "sent", 0}, + {"msg_pending_count_approved", first.ID, "inbound", "review_approved", 0}, + {"msg_pending_count_rejected", first.ID, "outbound", "review_rejected", 0}, + {"msg_pending_count_expired_approved", first.ID, "outbound", "review_expired_approved", 0}, + {"msg_pending_count_expired_rejected", first.ID, "outbound", "review_expired_rejected", 0}, + {"msg_pending_count_second", second.ID, "outbound", "pending_review", 0}, + {"msg_pending_count_other_owner", otherOwner.ID, "outbound", "pending_review", 0}, + } + for _, row := range rows { + if _, err := pool.Exec(ctx, + `INSERT INTO messages (id, agent_id, direction, status, created_at) + VALUES ($1, $2, $3, $4, now() - make_interval(days => $5))`, + row.id, row.agentID, row.direction, row.status, row.ageDays, + ); err != nil { + t.Fatalf("seed %s: %v", row.id, err) + } + } + + assertCounts := func(wantFirst int) { + t.Helper() + agents, err := store.ListAgentsByUser(ctx, user.ID, 0, time.Time{}, "") + if err != nil { + t.Fatalf("ListAgentsByUser: %v", err) + } + if len(agents) != 2 { + t.Fatalf("agent count = %d, want 2 for the requested owner", len(agents)) + } + want := map[string]int{first.ID: wantFirst, second.ID: 1} + for _, agent := range agents { + count, ok := want[agent.ID] + if !ok { + t.Fatalf("unexpected agent %q in requested owner's list", agent.ID) + } + if agent.PendingCount != count { + t.Errorf("agent %q pending count = %d, want %d", agent.ID, agent.PendingCount, count) + } + delete(want, agent.ID) + } + if len(want) != 0 { + t.Fatalf("missing agents: %v", want) + } + } + assertCounts(2) + + if _, err := pool.Exec(ctx, + `UPDATE messages SET status = 'sent' WHERE id = 'msg_pending_count_recent'`, + ); err != nil { + t.Fatalf("resolve pending hold: %v", err) + } + assertCounts(1) +} diff --git a/migrations/132_messages_agent_pending_idx.sql b/migrations/132_messages_agent_pending_idx.sql new file mode 100644 index 000000000..0e33e7aae --- /dev/null +++ b/migrations/132_messages_agent_pending_idx.sql @@ -0,0 +1,17 @@ +-- 132_messages_agent_pending_idx.sql +-- e2a:no-transaction +-- +-- Supports ListAgentsByUser's pending_count: count outbound review holds by +-- agent without scanning that agent's message history or every agent's holds. +-- The existing idx_messages_pending_review leads with approval_expires_at for +-- global expiry sweeps; it cannot directly seek to one agent's holds. +-- Match the current query's all-time predicate (pending_approval was retired +-- in 044); do not add a date or deleted_at filter absent from the query. +-- +-- Build concurrently because messages is a hot table. If an interrupted build +-- leaves an invalid index, the migration runner refuses to mark this applied. +-- Recovery: DROP INDEX CONCURRENTLY IF EXISTS idx_messages_agent_pending; +-- then retry the migration. +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_messages_agent_pending + ON messages (agent_id) + WHERE status = 'pending_review' AND direction = 'outbound'; From d7b62ccd4baa9f4424cfcfdd4622fd7802537abe Mon Sep 17 00:00:00 2001 From: Josh Zhang <39790535+jiashuoz@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:13:43 +0900 Subject: [PATCH 2/2] chore(migrations): renumber pending-count index migration to 133 Slot 132 was taken on main by 132_owner_email_proof_oidc_source.sql (#1076). Move this migration to 133 in the filename, its header line, and the test that reads it from the embedded FS. Co-Authored-By: Claude Fable 5.1 --- internal/identity/pending_count_index_test.go | 2 +- ...agent_pending_idx.sql => 133_messages_agent_pending_idx.sql} | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) rename migrations/{132_messages_agent_pending_idx.sql => 133_messages_agent_pending_idx.sql} (95%) diff --git a/internal/identity/pending_count_index_test.go b/internal/identity/pending_count_index_test.go index bd35294d1..ea6914d37 100644 --- a/internal/identity/pending_count_index_test.go +++ b/internal/identity/pending_count_index_test.go @@ -17,7 +17,7 @@ func TestPendingCountIndexMigration(t *testing.T) { // TestDB applies the embedded migrations through the real runner. Reapply // the index statement directly to also exercise its IF NOT EXISTS retry. - sql, err := migrations.FS.ReadFile("132_messages_agent_pending_idx.sql") + sql, err := migrations.FS.ReadFile("133_messages_agent_pending_idx.sql") if err != nil { t.Fatalf("read pending-count migration: %v", err) } diff --git a/migrations/132_messages_agent_pending_idx.sql b/migrations/133_messages_agent_pending_idx.sql similarity index 95% rename from migrations/132_messages_agent_pending_idx.sql rename to migrations/133_messages_agent_pending_idx.sql index 0e33e7aae..83d6c68f6 100644 --- a/migrations/132_messages_agent_pending_idx.sql +++ b/migrations/133_messages_agent_pending_idx.sql @@ -1,4 +1,4 @@ --- 132_messages_agent_pending_idx.sql +-- 133_messages_agent_pending_idx.sql -- e2a:no-transaction -- -- Supports ListAgentsByUser's pending_count: count outbound review holds by