diff --git a/scripts/benchmark-inbox-mine.ts b/scripts/benchmark-inbox-mine.ts new file mode 100644 index 00000000000..9939d8b1f83 --- /dev/null +++ b/scripts/benchmark-inbox-mine.ts @@ -0,0 +1,223 @@ +// Local SQL benchmark for the Mine inbox list (GET /api/companies/:id/inbox/mine). +// +// pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-inbox-mine.ts \ +// --baseline-ref= [--issues=30000] [--activity-per-issue=12] +// [--comments-per-issue=6] [--touch-density=1] [--samples=7] [--evidence-dir=] +// +// Seeds one disposable embedded-PostgreSQL company with tens of thousands of +// issues (most of them `done`), comments, activity, archives, read states, +// pending interactions and live runs, plus a user who touched a few hundred. +// It then runs the baseline (from --baseline-ref) and the working-tree +// `issueService.list` with the exact Mine filters, asserts the complete row +// payloads are identical, and reports median timings and per-query +// EXPLAIN (ANALYZE, BUFFERS) ranked by execution time. +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { createDb, applyPendingMigrations, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import { INBOX_MINE_ISSUE_STATUS_FILTER } from "../packages/shared/src/index.js"; +import { issueService, type IssueFilters } from "../server/src/services/issues.js"; +import { issueReadSqlCondition } from "../server/src/services/authorization.js"; + +const repositoryRoot = fileURLToPath(new URL("../", import.meta.url)); +const git = (args: string[]) => execFileSync("git", args, { cwd: repositoryRoot, encoding: "utf8" }); +const arg = (name: string) => process.argv.find((value) => value.startsWith(`--${name}=`))?.slice(name.length + 3); +const baselineRef = arg("baseline-ref"); +assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref="); +const baselineRevision = git(["rev-parse", `${baselineRef}^{commit}`]).trim(); +const issueCount = Number(arg("issues") ?? 30000); +const samples = Number(arg("samples") ?? 7); +const activityPerIssue = Number(arg("activity-per-issue") ?? 12); +const commentsPerIssue = Number(arg("comments-per-issue") ?? 6); +// Participation density: 1 touches ~1.4k of 30k issues; 10 touches ~11k. +const density = Number(arg("touch-density") ?? 1); +const every = (modulus: number) => Math.max(2, Math.round(modulus / density)); +const evidenceDir = arg("evidence-dir"); + +// Baseline modules: the Mine list SQL lives in these two files. Every other +// dependency is held constant at the working tree. +const modulePaths = ["server/src/services/issues.ts", "server/src/services/issue-participation.ts"]; +const baselineDir = await fs.mkdtemp(path.join(repositoryRoot, "server/scripts/.inbox-mine-baseline-")); +for (const modulePath of modulePaths) { + const source = git(["show", `${baselineRevision}:${modulePath}`]).replace(/from "(\.{1,2}\/[^"]+)"/g, (_m, specifier: string) => { + const resolved = fileURLToPath(new URL(specifier, pathToFileURL(path.join(repositoryRoot, modulePath)))); + return `from ${JSON.stringify(resolved.endsWith("/services/issue-participation.js") ? path.join(baselineDir, "issue-participation.ts") : resolved)}`; + }); + await fs.writeFile(path.join(baselineDir, path.basename(modulePath)), source); +} +const baseline = await import(pathToFileURL(path.join(baselineDir, "issues.ts")).href) as typeof import("../server/src/services/issues.js"); + +type Captured = { sql: string; params: unknown[] }; +let database: Awaited> | undefined; +let db: Db | undefined; +const keepAlive = setInterval(() => {}, 1000); +try { + database = await startEmbeddedPostgresTestDatabase("inbox-mine-benchmark-"); + await applyPendingMigrations(database.connectionString); + // A small pool, like the server's: the list issues enrichment reads in parallel. + db = createDb(database.connectionString, { maxConnections: 8 }); + const client = db.$client as any; + let capture: Captured[] | null = null; + const unsafe = client.unsafe.bind(client); + client.unsafe = (query: string, params?: unknown[], options?: unknown) => { + if (capture && /^\s*(select|with)\b/i.test(query)) capture.push({ sql: query, params: params ?? [] }); + return unsafe(query, params, options); + }; + await client.unsafe("set client_min_messages = warning"); + const [settings] = await client.unsafe("select version(), current_setting('shared_buffers') shared_buffers, current_setting('work_mem') work_mem, current_setting('jit') jit"); + + const companyId = randomUUID(), userId = "synthetic-mine-user", otherUserId = "synthetic-other-human"; + const seedStart = performance.now(); + await client.unsafe(`insert into companies(id,name,issue_prefix) values($1,'Synthetic Mine','MINE')`, [companyId]); + const agentIds = [randomUUID(), randomUUID(), randomUUID()]; + for (const agentId of agentIds) await client.unsafe(`insert into agents(id,company_id,name,adapter_type,status) values($1,$2,$3,'process','active')`, [agentId, companyId, `Agent ${agentId.slice(0, 8)}`]); + // h(n, salt) is a deterministic 0..99 bucket so the fixture is reproducible. + await client.unsafe(`create or replace function bench_h(n int, salt text) returns int language sql immutable as $$ select abs(hashtext(n::text || salt)) % 100 $$`); + await client.unsafe(`insert into issues(id,company_id,title,status,created_by_user_id,created_by_agent_id,assignee_user_id,origin_kind,hidden_at,harness_kind,completed_at,created_at,updated_at) + select gen_random_uuid(),$1::uuid,'Synthetic '||n, + case when bench_h(n,'s')<55 then 'done' when bench_h(n,'s')<60 then 'cancelled' when bench_h(n,'s')<70 then 'backlog' + when bench_h(n,'s')<82 then 'todo' when bench_h(n,'s')<90 then 'in_progress' when bench_h(n,'s')<95 then 'in_review' else 'blocked' end, + case when n%${every(75)}=0 then $2 when bench_h(n,'c')<30 then $3 else null end, + case when n%${every(75)}<>0 and bench_h(n,'c')>=30 then ($4::uuid[])[1+n%3] else null end, + case when n%${every(311)}=0 then $2 else null end, + case when n%997=0 then 'plugin:demo:operation' else 'manual' end, + case when n%1499=0 then '2026-01-01'::timestamptz else null end, + case when n%1777=0 then 'synthetic' else null end, + case when bench_h(n,'s')<55 then '2026-01-01'::timestamptz+n*interval '1 second' else null end, + '2026-01-01'::timestamptz,'2026-01-01'::timestamptz+n*interval '1 second' + from generate_series(1,$5::int) n`, [companyId, userId, otherUserId, agentIds, issueCount]); + await client.unsafe(`create table bench_fixture_issues as select id, row_number() over (order by updated_at)::int n, status from issues where company_id=$1`, [companyId]); + await client.unsafe(`update issues i set parent_id=p.id from bench_fixture_issues c join bench_fixture_issues p on p.n=c.n/2 where i.id=c.id and c.n>100 and bench_h(c.n,'p')<35`); + // Agent comments on every issue, an outside human on some, and the user's own direct comments. + await client.unsafe(`insert into issue_comments(company_id,issue_id,author_agent_id,body,created_at) + select $1::uuid,f.id,($2::uuid[])[1+k%3],'Agent progress '||f.n||'/'||k,'2026-02-01'::timestamptz+(f.n*$3::int+k)*interval '1 second' + from bench_fixture_issues f cross join generate_series(0,$3::int-1) k`, [companyId, agentIds, commentsPerIssue]); + await client.unsafe(`insert into issue_comments(company_id,issue_id,author_user_id,body,created_at) + select $1::uuid,id,$2,'Outside human note','2026-03-01'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%7=0`, [companyId, otherUserId]); + await client.unsafe(`insert into issue_comments(company_id,issue_id,author_user_id,body,created_at) + select $1::uuid,id,$2,'My note','2026-01-15'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%${every(97)}=0`, [companyId, userId]); + // Mention-only: the user is named in a comment but never participated. + await client.unsafe(`insert into issue_comments(company_id,issue_id,author_user_id,body,created_at) + select $1::uuid,id,$2,'@synthetic-mine-user can you look?','2026-03-02'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%113=0`, [companyId, otherUserId]); + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'issue',f.id::text,'agent',($2::uuid[])[1+k%3]::text, + case when k%2=0 then 'issue.updated' else 'issue.comment_added' end, + jsonb_build_object('status',f.status,'_previous',jsonb_build_object('status','todo')), + '2026-01-20'::timestamptz+(f.n*$3::int+k)*interval '1 second' + from bench_fixture_issues f cross join generate_series(0,$3::int-1) k`, [companyId, agentIds, activityPerIssue]); + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'issue',id::text,'user',$2,'issue.updated','{}'::jsonb,'2026-01-25'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%${every(89)}=0`, [companyId, userId]); + // Reads are audited but are not participation. + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'issue',id::text,'user',$2,'issue.read_marked','{}'::jsonb,'2026-01-26'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%13=0`, [companyId, userId]); + await client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'heartbeat_run',gen_random_uuid()::text,'agent',($2::uuid[])[1],'heartbeat.run.created','{}'::jsonb,'2026-01-01'::timestamptz+n*interval '1 second' from generate_series(1,$3::int) n`, [companyId, agentIds, issueCount * 2]); + // Archives: some stay archived, some are reactivated by later outside comments (n%7=0). + await client.unsafe(`insert into issue_inbox_archives(company_id,issue_id,user_id,archived_at) + select $1::uuid,id,$2,'2026-02-15'::timestamptz from bench_fixture_issues where n%3=0 and (n%${every(75)}=0 or n%${every(89)}=0 or n%${every(97)}=0 or n%${every(311)}=0)`, [companyId, userId]); + await client.unsafe(`insert into issue_read_states(company_id,issue_id,user_id,last_read_at) + select $1::uuid,id,$2,'2026-02-20'::timestamptz from bench_fixture_issues where n%40=0`, [companyId, userId]); + await client.unsafe(`insert into issue_thread_interactions(company_id,issue_id,kind,status,addressee_user_id,payload,created_at) + select $1::uuid,id,'request_confirmation','pending',$2,'{"version":1,"prompt":"Confirm?"}'::jsonb,'2026-03-05'::timestamptz+n*interval '1 second' from bench_fixture_issues where n%503=0`, [companyId, userId]); + await client.unsafe(`insert into heartbeat_runs(id,company_id,agent_id,status,context_snapshot) + select gen_random_uuid(),$1::uuid,($2::uuid[])[1+n%3],case when n%2=0 then 'running' else 'queued' end, + jsonb_build_object('issueId',(select id from bench_fixture_issues where bench_fixture_issues.n=g.n*97)) + from generate_series(1,150) g(n)`, [companyId, agentIds]); + await client.unsafe(`insert into heartbeat_runs(company_id,agent_id,status) select $1::uuid,($2::uuid[])[1],'succeeded' from generate_series(1,$3::int)`, [companyId, agentIds, issueCount]); + await client.unsafe("ANALYZE"); + const [counts] = await client.unsafe(`select + (select count(*)::int from issues where company_id=$1) issues, + (select count(*)::int from issues where company_id=$1 and status='done') done, + (select count(*)::int from issue_comments where company_id=$1) comments, + (select count(*)::int from activity_log where company_id=$1) activity, + (select count(*)::int from issue_inbox_archives where company_id=$1) archives, + (select count(*)::int from issue_read_states where company_id=$1) read_states, + (select count(*)::int from issue_thread_interactions where company_id=$1) interactions, + (select count(*)::int from heartbeat_runs where company_id=$1 and status in ('queued','running')) live_runs`, [companyId]); + const fixtureMs = performance.now() - seedStart; + + const readCondition = await issueReadSqlCondition(db, { type: "board", source: "session", userId, authorizingUserId: userId } as never); + const filters: IssueFilters = { + // The route's exact filters. Baselines that predate touchedCandidates ignore it. + readCondition, touchedByUserId: userId, touchedCandidates: "participation", inboxArchivedByUserId: userId, + status: INBOX_MINE_ISSUE_STATUS_FILTER, listProjection: "index", + sortField: "updated", sortDir: "desc", limit: 100, + includeRoutineExecutions: true, includeLiveDescendantSummary: true, + }; + const arms = { baseline: baseline.issueService(db), treatment: issueService(db) }; + const times = { baseline: [] as number[], treatment: [] as number[] }; + let reference: unknown; + for (let iteration = -2; iteration < samples; iteration++) { + for (const arm of (iteration % 2 === 0 ? ["baseline", "treatment"] : ["treatment", "baseline"]) as Array) { + const started = performance.now(); + const rows = await arms[arm].list(companyId, filters); + const elapsed = performance.now() - started; + if (iteration >= 0) times[arm].push(elapsed); + const serialized = JSON.parse(JSON.stringify(rows)); + if (reference === undefined) reference = serialized; + else assert.deepEqual(serialized, reference, `${arm}: complete Mine row payloads must match`); + } + } + const resultIds = (reference as Array<{ id: string }>).map((row) => row.id); + + // Per-phase timing from the list's own phase hook (absent in old baselines). + const phaseTimes: Record> = { baseline: {}, treatment: {} }; + for (let iteration = 0; iteration < samples; iteration++) { + for (const arm of ["baseline", "treatment"] as const) { + await arms[arm].list(companyId, filters, { + timePhase: async (phase, read) => { + const started = performance.now(); + try { return await read(); } finally { (phaseTimes[arm]![phase] ??= []).push(performance.now() - started); } + }, + } as never); + } + } + assert.ok(resultIds.length === 100, `Fixture must fill the 100-row page (got ${resultIds.length})`); + + const plans: Record = {}; + for (const arm of ["baseline", "treatment"] as const) { + capture = []; + await arms[arm].list(companyId, filters); + const captured = capture; + capture = null; + const ranked = []; + for (const query of captured) { + const [row] = await client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params); + const plan = row["QUERY PLAN"][0]; + ranked.push({ executionMs: plan["Execution Time"], planningMs: plan["Planning Time"], + sharedHit: plan.Plan["Shared Hit Blocks"], sharedRead: plan.Plan["Shared Read Blocks"], + sql: query.sql, params: query.params, plan }); + } + ranked.sort((a, b) => b.executionMs - a.executionMs); + plans[arm] = ranked; + } + const stats = (values: number[]) => { + const sorted = [...values].sort((a, b) => a - b); + return { medianMs: sorted[Math.floor(sorted.length / 2)]!, minMs: sorted[0]!, maxMs: sorted[sorted.length - 1]!, samplesMs: values }; + }; + const summary = (arm: "baseline" | "treatment") => (plans[arm] as Array<{ executionMs: number; sql: string; sharedHit: number; sharedRead: number }>) + .map((query) => ({ executionMs: Number(query.executionMs.toFixed(2)), buffers: query.sharedHit + query.sharedRead, sql: query.sql.replace(/\s+/g, " ").slice(0, 160) })); + const report = { + baselineRevision, treatmentRevision: git(["rev-parse", "HEAD"]).trim(), treatmentDirty: git(["status", "--porcelain", "--", ...modulePaths]).trim() !== "", + settings, fixture: { ...counts, fixtureMs, resultRows: resultIds.length }, + baseline: stats(times.baseline), treatment: stats(times.treatment), + phases: Object.fromEntries(Object.entries(phaseTimes).map(([arm, phases]) => [arm, + Object.fromEntries(Object.entries(phases).map(([phase, values]) => [phase, Number(stats(values).medianMs.toFixed(2))]))])), + speedup: stats(times.baseline).medianMs / stats(times.treatment).medianMs, + queries: { baseline: summary("baseline"), treatment: summary("treatment") }, + }; + if (evidenceDir) { + await fs.mkdir(evidenceDir, { recursive: true }); + await fs.writeFile(path.join(evidenceDir, "summary.json"), JSON.stringify(report, null, 2)); + await fs.writeFile(path.join(evidenceDir, "plans.json"), JSON.stringify(plans, null, 2)); + } + console.log(JSON.stringify(report, null, 2)); +} finally { + await db?.$client.end({ timeout: 0 }); + await database?.cleanup(); + await fs.rm(baselineDir, { recursive: true, force: true }); + clearInterval(keepAlive); +} diff --git a/server/src/__tests__/inbox-mine-candidates.test.ts b/server/src/__tests__/inbox-mine-candidates.test.ts new file mode 100644 index 00000000000..6bcc51e48dc --- /dev/null +++ b/server/src/__tests__/inbox-mine-candidates.test.ts @@ -0,0 +1,139 @@ +import { randomUUID } from "node:crypto"; +import { sql } from "drizzle-orm"; +import { + activityLog, agents, companies, heartbeatRuns, issueComments, issueInboxArchives, issueReadStates, issues, issueThreadInteractions, +} from "@paperclipai/db"; +import { INBOX_MINE_ISSUE_STATUS_FILTER } from "@paperclipai/shared"; +import { expect, it } from "vitest"; +import { issueReadSqlCondition } from "../services/authorization.js"; +import { issueService, type IssueFilters } from "../services/issues.js"; +import { describeEmbeddedPostgres, useEmbeddedPostgres } from "./helpers/route-test-harness.js"; + +// The Mine list ranks candidates from the participation UNION +// (`touchedCandidates: "participation"`). These tests run the original +// `touchedByUserCondition` scan beside it and require identical ordered rows. +describeEmbeddedPostgres("Mine inbox participation candidates", () => { + const ctx = useEmbeddedPostgres("inbox-mine-candidates-"); + const at = (minute: number) => new Date(Date.UTC(2026, 9, 1, 0, minute)); + + async function mineFilters(userId: string, overrides: Partial = {}): Promise { + return { + readCondition: await issueReadSqlCondition(ctx.db, { type: "board", source: "session", userId, authorizingUserId: userId } as never), + touchedByUserId: userId, inboxArchivedByUserId: userId, + status: INBOX_MINE_ISSUE_STATUS_FILTER, listProjection: "index", + sortField: "updated", sortDir: "desc", limit: 100, + includeRoutineExecutions: true, includeLiveDescendantSummary: true, + ...overrides, + }; + } + + async function compare(companyId: string, filters: IssueFilters) { + const svc = issueService(ctx.db); + const scan = await svc.list(companyId, filters); + const ranked = await svc.list(companyId, { ...filters, touchedCandidates: "participation" }); + expect(ranked.map((row) => row.id)).toEqual(scan.map((row) => row.id)); + expect(JSON.parse(JSON.stringify(ranked))).toEqual(JSON.parse(JSON.stringify(scan))); + return scan.map((row) => row.id); + } + + it("selects the same ordered rows and payloads as the touched scan across participation edge cases", async () => { + const companyId = randomUUID(), foreignCompanyId = randomUUID(); + const userId = `mine-${randomUUID()}`, otherUserId = `other-${randomUUID()}`; + await ctx.db.insert(companies).values([ + { id: companyId, name: "Mine candidates", issuePrefix: `M${companyId.slice(0, 6).toUpperCase()}` }, + { id: foreignCompanyId, name: "Foreign", issuePrefix: `F${foreignCompanyId.slice(0, 6).toUpperCase()}` }, + ]); + const [agent] = await ctx.db.insert(agents).values({ companyId, name: "Worker", adapterType: "process", status: "active" }).returning(); + const named = new Map(); + const add = async (name: string, values: Partial = {}, minute = named.size) => { + const [row] = await ctx.db.insert(issues).values({ + companyId, title: name, status: "todo", createdByUserId: otherUserId, + createdAt: at(0), updatedAt: at(minute), ...values, + }).returning(); + named.set(name, row!.id); + return row!.id; + }; + const activity = (issueId: string, values: Partial) => ctx.db.insert(activityLog).values({ + companyId, entityType: "issue", entityId: issueId, actorType: "user", actorId: userId, action: "issue.updated", createdAt: at(500), ...values, + }); + const comment = (issueId: string, values: Partial) => ctx.db.insert(issueComments).values({ + companyId, issueId, body: "note", authorUserId: userId, createdAt: at(400), ...values, + }); + + // Included: every participation source, done work, and re-activated archives. + await add("created by me", { createdByUserId: userId }); + await add("assigned to me", { assigneeUserId: userId, createdByUserId: null, createdByAgentId: agent!.id }); + await activity(await add("updated by me"), {}); + await comment(await add("commented by me"), {}); + await add("done by me", { createdByUserId: userId, status: "done", completedAt: at(30) }); + const pending = await add("pending request", { createdByUserId: null, createdByAgentId: agent!.id }); + await ctx.db.insert(issueThreadInteractions).values({ + companyId, issueId: pending, kind: "request_confirmation", status: "pending", addresseeUserId: userId, + payload: { version: 1, prompt: "Confirm?" } as never, createdAt: at(600), + }); + const reactivated = await add("archived then re-activated", { createdByUserId: userId }); + await ctx.db.insert(issueInboxArchives).values({ companyId, issueId: reactivated, userId, archivedAt: at(700) }); + await comment(reactivated, { authorUserId: otherUserId, createdAt: at(800) }); + const readIssue = await add("read with later agent comment", { createdByUserId: userId }); + await ctx.db.insert(issueReadStates).values({ companyId, issueId: readIssue, userId, lastReadAt: at(900) }); + await comment(readIssue, { authorUserId: null, authorAgentId: agent!.id, createdAt: at(950) }); + const liveChild = await add("live child of mine", { createdByUserId: userId, parentId: named.get("created by me")! }); + const [run] = await ctx.db.insert(heartbeatRuns).values({ companyId, agentId: agent!.id, status: "running", contextSnapshot: { issueId: liveChild } }).returning(); + await ctx.db.update(issues).set({ executionRunId: run!.id }).where(sql`${issues.id} = ${liveChild}`); + + // Excluded by both selections. + const archived = await add("archived without new activity", { createdByUserId: userId }); + await ctx.db.insert(issueInboxArchives).values({ companyId, issueId: archived, userId, archivedAt: at(990) }); + await comment(await add("mention only"), { authorUserId: otherUserId, body: `@${userId} please look` }); + await activity(await add("read audit only"), { action: "issue.read_marked" }); + const upper = await add("uppercase entity id"); + await activity(upper, { entityId: upper.toUpperCase() }); + await comment(await add("channel comment"), { metadata: { sourceChannel: "slack" } as never }); + await add("agent created with my user id", { createdByUserId: userId, createdByAgentId: agent!.id }); + await add("cancelled by me", { createdByUserId: userId, status: "cancelled" }); + await add("hidden by me", { createdByUserId: userId, hiddenAt: at(1) }); + await add("harness by me", { createdByUserId: userId, harnessKind: "synthetic" }); + await add("plugin operation by me", { createdByUserId: userId, originKind: "plugin:demo:operation" }); + const privateId = randomUUID(); + await add("private to someone else", { + id: privateId, visibility: "private", privacyRootIssueId: privateId, responsibleUserId: otherUserId, + }); + await activity(privateId, {}); + const answered = await add("answered request", { createdByUserId: null, createdByAgentId: agent!.id }); + await ctx.db.insert(issueThreadInteractions).values({ + companyId, issueId: answered, kind: "request_confirmation", status: "answered", addresseeUserId: userId, + payload: { version: 1, prompt: "Confirm?" } as never, + }); + await ctx.db.insert(issues).values({ companyId: foreignCompanyId, title: "foreign", createdByUserId: userId }); + + // Page pressure: more than 100 touched issues, many with tied activity times. + for (let index = 0; index < 130; index++) { + await add(`bulk ${index}`, { createdByUserId: userId, status: index % 3 === 0 ? "done" : "in_progress" }, index % 4); + } + for (let index = 0; index < 40; index++) await add(`untouched ${index}`, {}, 1000 + index); + + const ids = await compare(companyId, await mineFilters(userId)); + expect(ids).toHaveLength(100); + const all = await compare(companyId, await mineFilters(userId, { limit: 1000 })); + for (const name of [ + "created by me", "assigned to me", "updated by me", "commented by me", "done by me", "pending request", + "archived then re-activated", "read with later agent comment", "live child of mine", + ]) expect(all, name).toContain(named.get(name)); + for (const name of [ + "archived without new activity", "mention only", "read audit only", "uppercase entity id", "channel comment", + "agent created with my user id", "cancelled by me", "hidden by me", "harness by me", "plugin operation by me", + "private to someone else", "answered request", "untouched 0", + ]) expect(all, name).not.toContain(named.get(name)); + expect(all.indexOf(named.get("archived then re-activated")!)).toBeLessThan(all.indexOf(named.get("created by me")!)); + + // Other selection shapes stay equivalent: no archive filter, other statuses, + // offsets, ascending order, unrestricted read access, and another user. + await compare(companyId, await mineFilters(userId, { inboxArchivedByUserId: undefined })); + await compare(companyId, await mineFilters(userId, { status: "done", listProjection: undefined, includeLiveDescendantSummary: false })); + await compare(companyId, await mineFilters(userId, { offset: 50, limit: 30 })); + await compare(companyId, await mineFilters(userId, { sortDir: "asc" })); + await compare(companyId, { ...await mineFilters(userId), readCondition: sql`true` }); + expect(await compare(companyId, await mineFilters(otherUserId, { limit: 1000 }))).toContain(named.get("archived then re-activated")); + expect(await compare(companyId, await mineFilters(`nobody-${randomUUID()}`))).toEqual([]); + }); +}); diff --git a/server/src/__tests__/inbox-mine-routes.test.ts b/server/src/__tests__/inbox-mine-routes.test.ts index 28cf3be36b3..485abe3977f 100644 --- a/server/src/__tests__/inbox-mine-routes.test.ts +++ b/server/src/__tests__/inbox-mine-routes.test.ts @@ -112,9 +112,12 @@ describeEmbeddedPostgres("Mine inbox snapshot routes", () => { })); for (const phase of [ "paperclip_inbox", "inbox_access", "inbox_issues", "inbox_parents", "inbox_failures", "inbox_failure_context", - "inbox_issues_query", "inbox_issues_labels", "inbox_issues_user_stats", "inbox_issues_read_stats", + "inbox_issues_candidates", "inbox_issues_query", "inbox_issues_labels", "inbox_issues_user_stats", "inbox_issues_read_stats", "inbox_issues_last_activity", "inbox_issues_archives", "inbox_issues_live_descendants", ]) expect(timingPhases.get(phase), phase).toBeGreaterThanOrEqual(0); + // Attribution: list sub-phases run inside inbox_issues, and every phase inside the request. + expect(timingPhases.get("inbox_issues")!).toBeGreaterThanOrEqual(timingPhases.get("inbox_issues_candidates")! + timingPhases.get("inbox_issues_query")! - 0.2); + expect(timingPhases.get("paperclip_inbox")!).toBeGreaterThanOrEqual(timingPhases.get("inbox_issues")! - 0.1); expect(snapshot).not.toHaveProperty("failureError"); expect(snapshot).not.toHaveProperty("contextError"); }); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 14cad9981a9..8928c30958c 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -8831,7 +8831,7 @@ export function issueRoutes( if (!userId) throw new HttpError(401, "A signed-in board user is required"); const readCondition = await timing.time("access", () => issueReadSqlCondition(db, req.actor)); const listFilters: IssueFilters = { - readCondition, touchedByUserId: userId, inboxArchivedByUserId: userId, + readCondition, touchedByUserId: userId, touchedCandidates: "participation", inboxArchivedByUserId: userId, status: INBOX_MINE_ISSUE_STATUS_FILTER, listProjection: "index", sortField: "updated", sortDir: "desc", limit: 100, includeRoutineExecutions: true, includeLiveDescendantSummary: true, diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index cec7d7a5090..3ad5ebad329 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1867,7 +1867,7 @@ export function parseStatusFilter( /** Closed, content-free phases of `issueService.list`, for response timing headers. */ export type IssueListPhase = - | "query" | "labels" | "active_runs" | "user_stats" | "read_stats" | "last_activity" + | "candidates" | "query" | "labels" | "active_runs" | "user_stats" | "read_stats" | "last_activity" | "archives" | "blocked_by" | "live_descendants" | "attention"; export interface IssueListOptions { @@ -1898,6 +1898,14 @@ export interface IssueFilters { oneChildPerParent?: boolean; assigneeUserId?: string; touchedByUserId?: string; + /** + * `participation` ranks touched issues from the indexed participation + * UNION (see `mineParticipationIssueIds`) and loads full rows only for the + * selected page, instead of scanning every status-matching issue with the + * five-way `touchedByUserCondition`. Same rows and order. Requires + * `touchedByUserId`; ignored with search or `oneChildPerParent`. + */ + touchedCandidates?: "participation"; inboxArchivedByUserId?: string; unreadForUserId?: string; projectId?: string; @@ -3028,6 +3036,36 @@ function unreadForUserCondition(companyId: string, userId: string) { `; } +/** + * Ids of every issue the user participated in, built from indexed sources. + * Each UNION member is one branch of `touchedByUserCondition`, so membership + * equals that condition. The result is aliased `mine_participants` (`id`). + */ +function mineParticipationIssueIds(db: Db, companyId: string, userId: string) { + const directlyParticipated = or( + and(eq(issues.createdByUserId, userId), humanCreatedIssueCondition()), + eq(issues.assigneeUserId, userId), + ); + const participationSources = unionAll( + db.select({ id: sql`${issues.id}`.as("id") }).from(issues).where(and( + eq(issues.companyId, companyId), + directlyParticipated, + )), + db.select({ id: sql`case when length(${activityLog.entityId}) = 36 and ${activityLog.entityId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then ${activityLog.entityId}::uuid end`.as("id") }).from(activityLog).where(and( + userParticipationActivityCondition(companyId, userId), + notInArray(activityLog.entityId, db.select({ id: sql`${issues.id}::text` }).from(issues) + .where(and(eq(issues.companyId, companyId), directlyParticipated))), + )), + db.select({ id: sql`${issueComments.issueId}`.as("id") }).from(issueComments).where(and( + eq(issueComments.companyId, companyId), eq(issueComments.authorUserId, userId), directHumanCommentCondition(), + )), + db.select({ id: sql`${issues.id}`.as("id") }).from(issueThreadInteractions) + .innerJoin(issues, and(eq(issues.id, issueThreadInteractions.issueId), eq(issues.companyId, companyId))) + .where(pendingUserRequestRowCondition(companyId, userId)), + ).as("mine_participation_sources"); + return db.selectDistinct({ id: participationSources.id }).from(participationSources).as("mine_participants"); +} + function inboxVisibleForUserCondition(companyId: string, userId: string) { return sql` NOT EXISTS ( @@ -8239,28 +8277,7 @@ export function issueService(db: Db) { unreadInboxIssueIds: async (companyId: string, userId: string, readCondition: SQL) => { // Rank the visible Mine set before counting unread, just like the inbox. // This projection never reads descriptions, run payloads, or recovery details. - const directlyParticipated = or( - and(eq(issues.createdByUserId, userId), humanCreatedIssueCondition()), - eq(issues.assigneeUserId, userId), - ); - const participationSources = unionAll( - db.select({ id: sql`${issues.id}`.as("id") }).from(issues).where(and( - eq(issues.companyId, companyId), - directlyParticipated, - )), - db.select({ id: sql`case when length(${activityLog.entityId}) = 36 and ${activityLog.entityId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then ${activityLog.entityId}::uuid end`.as("id") }).from(activityLog).where(and( - userParticipationActivityCondition(companyId, userId), - notInArray(activityLog.entityId, db.select({ id: sql`${issues.id}::text` }).from(issues) - .where(and(eq(issues.companyId, companyId), directlyParticipated))), - )), - db.select({ id: sql`${issueComments.issueId}`.as("id") }).from(issueComments).where(and( - eq(issueComments.companyId, companyId), eq(issueComments.authorUserId, userId), directHumanCommentCondition(), - )), - db.select({ id: sql`${issues.id}`.as("id") }).from(issueThreadInteractions) - .innerJoin(issues, and(eq(issues.id, issueThreadInteractions.issueId), eq(issues.companyId, companyId))) - .where(pendingUserRequestRowCondition(companyId, userId)), - ).as("mine_participation_sources"); - const participated = db.selectDistinct({ id: participationSources.id }).from(participationSources).as("mine_participants"); + const participated = mineParticipationIssueIds(db, companyId, userId); const eligible = db.select({ id: issues.id, updatedAt: issues.updatedAt }) .from(participated).innerJoin(issues, eq(issues.id, sql`"mine_participants"."id"`)).where(and( eq(issues.companyId, companyId), @@ -8354,6 +8371,10 @@ export function issueService(db: Db) { const rawSearch = filters?.q?.trim() ?? ""; const hasSearch = rawSearch.length > 0; const taskSearch = parseTaskSearch(rawSearch); + // Rank page ids from the user's participation set, then load full rows + // only for that page (see IssueFilters.touchedCandidates). + const participationCandidates = filters?.touchedCandidates === "participation" + && touchedByUserId !== undefined && !hasSearch && !filters?.oneChildPerParent; if (filters?.createdFromIssueId) { conditions.push(createdFromIssueCondition(companyId, filters.createdFromIssueId)); } @@ -8402,7 +8423,9 @@ export function issueService(db: Db) { conditions.push(eq(issues.assigneeUserId, filters.assigneeUserId)); } if (touchedByUserId) { - conditions.push(touchedByUserCondition(companyId, touchedByUserId)); + conditions.push(participationCandidates + ? sql`${issues.id} IN (SELECT "mine_participants"."id" FROM ${mineParticipationIssueIds(db, companyId, touchedByUserId)})` + : touchedByUserCondition(companyId, touchedByUserId)); } if (inboxArchivedByUserId) { conditions.push( @@ -8483,6 +8506,28 @@ export function issueService(db: Db) { description: sql`NULL`, descriptionTruncated: sql`false`, } : issueListSelect; + const orderBy = [ + ...(filters?.oneChildPerParent ? [asc(issues.parentId)] : []), + ...issueListOrderBy(companyId, { + hasSearch, + priorityOrder, + searchOrder, + sortField: filters?.sortField, + sortDir: filters?.sortDir, + }), + ]; + const paginate = (query: Q) => + offset > 0 + ? limit === undefined ? query.offset(offset) : query.limit(limit).offset(offset) + : limit === undefined ? query : query.limit(limit); + // Phase 1 of the participation path: the page's ids in final order. The + // select-list subqueries below then run only for these rows. + const pageIds = participationCandidates + ? (await timePhase("candidates", async () => paginate( + db.select({ id: issues.id }).from(issues).where(and(...conditions)).orderBy(...orderBy).$dynamic(), + ))).map((row) => row.id) + : null; + if (pageIds?.length === 0) return []; const issueSource = (filters?.oneChildPerParent ? db.selectDistinctOn([issues.parentId], selectedFields) : db.select(selectedFields)).from(issues); @@ -8492,27 +8537,22 @@ export function issueService(db: Db) { SELECT m.id, ${taskSearchScore(taskSearch)} AS score FROM matched m ) task_search`, sql`task_search.id = ${issues.id}`) : issueSource; - const baseQuery = searchedSource - .where(and(...conditions)) - .orderBy( - ...(filters?.oneChildPerParent ? [asc(issues.parentId)] : []), - ...issueListOrderBy(companyId, { - hasSearch, - priorityOrder, - searchOrder, - sortField: filters?.sortField, - sortDir: filters?.sortDir, - }), - ); - const pageQuery = - offset > 0 - ? limit === undefined - ? baseQuery.offset(offset) - : baseQuery.limit(limit).offset(offset) - : limit === undefined - ? baseQuery - : baseQuery.limit(limit); - const rows = (await timePhase("query", async () => pageQuery)).map((row) => ({ + const pageQuery = pageIds + // Re-check company, visibility and read access for the selected ids; + // the rows are put back into the phase 1 order below. + ? searchedSource.where(and( + eq(issues.companyId, companyId), + inArray(issues.id, pageIds), + visibleIssueCondition(), + filters?.readCondition, + )) + : paginate(searchedSource.where(and(...conditions)).orderBy(...orderBy).$dynamic()); + const pageRows = await timePhase("query", async () => pageQuery); + if (pageIds) { + const position = new Map(pageIds.map((id, index) => [id, index])); + pageRows.sort((a, b) => position.get(a.id)! - position.get(b.id)!); + } + const rows = pageRows.map((row) => ({ ...row, description: decodeDatabaseTextPreview( row.description,