diff --git a/internal/identity/pending_count_index_test.go b/internal/identity/pending_count_index_test.go new file mode 100644 index 000000000..ea6914d37 --- /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("133_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/133_messages_agent_pending_idx.sql b/migrations/133_messages_agent_pending_idx.sql new file mode 100644 index 000000000..83d6c68f6 --- /dev/null +++ b/migrations/133_messages_agent_pending_idx.sql @@ -0,0 +1,17 @@ +-- 133_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';