diff --git a/doc/plans/2026-10-27-mine-unread-badge-query.md b/doc/plans/2026-10-27-mine-unread-badge-query.md new file mode 100644 index 00000000000..c87409ba194 --- /dev/null +++ b/doc/plans/2026-10-27-mine-unread-badge-query.md @@ -0,0 +1,88 @@ +# Mine unread badge query + +The Mine badge ranks the user's visible personal top 100, then checks unread state. Correlated participation and privacy predicates can trigger expensive PostgreSQL JIT compilation. The final change builds the same participant set using native UUID keys, applies fresh visibility before the limit, and ranks a narrow ordered candidate projection with the existing canonical activity expression. + +A partial activity index supports the new user lookup: `(company_id, actor_id, entity_id) WHERE actor_type = 'user' AND entity_type = 'issue'`. Activity already covered by creation/assignment is excluded from that branch before deduplication. Legacy text passes an exact lowercase 36-character UUID guard inside CASE. Malformed, uppercase, padded, newline, braced and dashless values still match nothing; the all-zero UUID remains supported. Canonical ordering, ties, archive/resurfacing and the second fresh unread query remain unchanged. + +## Final PostgreSQL 17 results + +These are ordinary warm local service calls, not hosted request timings. SQL capture and EXPLAIN happen outside the timed calls. Each arm has two warmups and five alternating samples. Complete positional rows from both SQL phases and final returned ID order match in all twelve cells. + +| Fixture / visibility / JIT | Baseline ms (range) | Treatment ms (range) | Before / after | +| --- | --- | --- | --- | +| small / unrestricted / on | 4.22 (3.88–4.50) | 4.77 (4.24–5.53) | 0.88x | +| small / unrestricted / off | 3.76 (3.47–4.08) | 4.36 (4.26–4.97) | 0.86x | +| small / scoped / on | 4.87 (4.58–4.98) | 5.06 (4.93–5.37) | 0.96x | +| small / scoped / off | 4.59 (4.33–5.10) | 5.12 (4.93–5.22) | 0.90x | +| dense / unrestricted / on | 18.33 (17.92–18.90) | 25.06 (24.44–28.48) | 0.73x | +| dense / unrestricted / off | 20.99 (20.49–34.83) | 28.73 (27.73–30.88) | 0.73x | +| dense / scoped / on | 909.30 (887.26–915.52) | 221.10 (219.58–224.21) | 4.11x | +| dense / scoped / off | 218.34 (215.13–222.99) | 223.24 (219.44–231.43) | 0.98x | +| sparse / unrestricted / on | 13.64 (13.11–14.10) | 5.84 (5.34–6.45) | 2.34x | +| sparse / unrestricted / off | 12.41 (12.27–13.00) | 4.90 (4.51–6.02) | 2.53x | +| sparse / scoped / on | 719.07 (711.22–761.60) | 10.79 (10.18–11.40) | 66.64x | +| sparse / scoped / off | 23.15 (22.64–27.02) | 12.90 (12.33–13.86) | 1.79x | + +The dense scoped path improves 4.11x; sparse paths improve 1.79–66.64x. Dense unrestricted paths remain slower by about 6.7–7.7 ms; dense scoped JIT-off is approximately neutral (218.34 to 223.24 ms). The result is strongest when the old query triggers full JIT or when a small personal set sits among company history. It is not a uniform execution-time improvement. + +The baseline-red work assertion fails on the actual captured full-JIT plan. The treatment-green assertion requires at least 1.5x improvement where the baseline performs full JIT optimization/inlining, no such passes afterward, and at most 25% or 2 ms overhead in small/sparse controls. Dense no-JIT cases are reported diagnostics; a stricter 25% dense guard failed, so that regression is explicitly retained. The original optional 60 ms cap still fails on dense scoped reads (221.10 ms). The captured ranking execution retains about 200 ms of privacy-predicate work when half the personal tasks participate through activity alone; the rewrite removes compiler passes rather than that authority work. The failed cap remains in raw evidence. It is not counted as a pass. + +## Fixture and isolation + +The pinned baseline is `70c864633744cd96c0dc7394d631fc7119cd17d2`. Both owning pre-fix service blobs match the initial baseline. The harness copies the old issue service and participation helper and holds other query dependencies constant. Separate logical databases contain identical IDs, timestamps and payloads. The baseline lacks only the candidate index and its journal row; catalog assertions verify all other index definitions match. The final report hashes both query sources, the schema, migration and runner registration. + +- Small: 12 personal issues, 48 activity rows and 24 comments. +- Dense: 1200 personal issues, 60000 issue activities, 240000 unrelated run activities and 24000 comments. +- Sparse: 12 personal issues among 1200 company issues, with the same history. +- Issue activity includes matching and other-human authors; the actor index is populated. The target user has 3600 activity rows in dense and 36 in sparse. Half the personal tasks are created by the user; the other half participate through user activity alone. Fixture assertions verify the complete declared personal set before timing. +- Base activity/comment payloads are 4096 incompressible bytes. Additional run rows carry 256 bytes. History appends across tasks in timestamp order. +- The reported workspace's fresh read-only catalog estimates were about 292k activity rows, 1198 issues and 7632 comments. Estimates are not exact counts. The large fixtures match issue/history scale and exceed its comment history. + +Environment: PostgreSQL 17.11 ARM64 Linux Docker, Node 26.4.0, postgres.js 3.4.9, Drizzle 0.45.2, shared_buffers 128 MiB, work_mem 4 MiB, jit on, jit_above_cost 100000 and full optimization/inlining thresholds 500000. Hosted PostgreSQL version/architecture and planner settings match; compute, storage and network differ. + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-mine-unread-badge.ts --baseline-ref=70c864633744cd96c0dc7394d631fc7119cd17d2 --history-decoys=240000 --verify-migration --assert-performance --evidence-dir= +``` + +The default harness starts disposable embedded PostgreSQL and ignores DATABASE_URL. Recorded runs used a scratch constructor provider for Linux PostgreSQL 17 because the host native runtime can hit semaphore limits. Database creation, schema migration, clients and cleanup remain the normal repository path. A saved report can be checked with `--validate-report=`; source hashes must still match. Add `--validate-baseline` for the original compiler-work red. `--diagnostic` runs one pair per visibility/JIT setting with a plain plan first. `--diagnostic-dense` selects dense instead of sparse. `--fixture=sparse --history-decoys=940000` selects a million-row control. `--hold-after-report=` retains one selected disposable fixture for bounded query iteration, with a private ready file and explicit release. + +## Why an exceptional migration is necessary + +The source-only participant set had no actor index and scanned company history. A plain EXISTS rewrite was decorrelated into that scan. A forced existing-entity-index lateral lookup avoided it but regressed the sparse unrestricted case from 56.64 to 66.47 ms with JIT and 12.74 to 18.68 ms without JIT. Both alternatives are rejected. A time cutoff would change historical participation; existing indexes cannot selectively find a user's issue activity without walking tasks/history. The partial index supplies that missing access path and excludes agent/system/non-issue activity from its storage and write maintenance. + +Migration `0395_tranquil_firestar.sql` uses the existing top-level concurrent-index runner and its invalid-index repair. No new startup script or live migration is used. The actual normal runner is measured against the pre-index schema with synthetic history. Concurrent reads/writes continue while it builds; a deliberate cancellation leaves an invalid index without a journal entry, then normal retry repairs it. + +For the 300k-row upgrade, total first-run wall time was 668.79 ms, rerun 21.83 ms and interrupted-build retry 187.71 ms. The migration has one DDL statement; totals include inspection, lock acquisition, concurrent build/validation, journal write and final verification. Concurrent reads completed in 0.51–6.69 ms and writes in 0.75–3.45 ms. The final partial index was 188416 bytes. Local duration does not guarantee a hosted duration, especially for larger tenants or long-lived transactions. + +## Feedback loop and rejected attempts + +Initial dependency install took 21.2 s. Frozen pnpm failed on upstream patch/lock metadata; local non-frozen resolution was retained privately and the tracked lock restored. The first service import lacked SDK dist; a narrow SDK dependency build fixed it. + +The initial PostgreSQL 18 clustered-history diagnostic took 35.4 s through useful evidence, including 29.3 s fixture setup. Ranking was 96.7 ms with 75.5 ms JIT. Interleaved history and PostgreSQL 17 made the later fixture more representative. An early text-key/grouped-history rewrite won unrestricted but regressed scoped sparse to 392 ms with JIT disabled; it was rejected. A native-key draft exposed an ambiguous Drizzle id alias; an empty-database parse smoke now runs before seeding. + +The original query-only 60k-history matrix improved target JIT cases but regressed dense JIT-off by about 17 ms and missed the 60 ms cap. Review exposed the missing actor index. Fresh catalog estimates sized the 300k and million-row controls. The first indexed draft still regressed dense execution. A narrow ordered candidate phase and skipping already-owned activity reduced that overhead. Retaining the seeded fixture shortened the next diagnostic to about ten seconds of actual calls. Repeated SQL capture had inflated timed service work; capture now occurs outside the sample interval in both arms. + +Scratch 512 MiB and eager dual-fixture 2 GiB runs hit ENOSPC; those are setup failures, not performance red. Fixtures now seed/measure sequentially. Matched pre-index/indexed databases use an owned 4 GiB tmpfs ceiling. The final full matrix took 113.26 s including both databases and fixture loading. Raw failures, plans and samples remain in private evidence. + +## Correctness and deployment limit + +Fourteen real PostgreSQL tests exercise ranking before unread, visibility before the top-100 limit, canonical ties, native assignee and exact legacy IDs, archive resurfacing/deleted comments/status changes, request audience/not_creator/dismissal/completion, provider/relay comment exclusions, company boundaries, inherited grants, private project membership, revocation and fresh badge routes. The final production source and strengthened fixtures passed all fourteen focused PostgreSQL 17 tests in 16.37 s. Full workspace typecheck and build passed. Latest-head CI remains required before handoff. + +The earlier full local suite remained incomplete after native PostgreSQL startup failures. An independent empty initdb reproduced system semaphore exhaustion at semget. All sixteen failed cases passed through task-owned Docker PostgreSQL 17; that is not a full local-suite pass. Full final-head CI is the broad gate. + +Fresh hosted ranking variants averaged 78.8 ms SQL per call. That is the affected serial SQL-time budget, not a promised HTTP saving. Badge HTTP requests averaged 752 ms (p95 1521 ms) without correlated SQL traces. A deployed revision check and comparable fresh query/request window are still necessary after rollout. No merge or deployment is part of this PR. + +## Million-row history control + +The same mixed-participation sparse fixture also contains 940000 unrelated run activities, for one million initial activity rows. Both arms again have two warmups and five alternating samples, and both complete SQL phases/final ID order match. + +| Visibility / JIT | Baseline ms (range) | Treatment ms (range) | Before / after | +| --- | --- | --- | --- | +| unrestricted / on | 13.44 (13.10–14.10) | 5.76 (5.51–6.17) | 2.33x | +| unrestricted / off | 13.45 (13.07–14.34) | 5.77 (5.11–6.84) | 2.33x | +| scoped / on | 807.76 (741.30–825.57) | 11.59 (10.39–12.67) | 69.68x | +| scoped / off | 19.69 (18.78–20.76) | 10.90 (10.06–11.73) | 1.81x | + +The activity participant branch uses the partial actor index and visits 36 matching actor rows (18 returned, 18 already-owned rows filtered), rather than scanning the million-row company history. Canonical activity and authority work remain separate. The activity relation is about 439 MB heap / 1.05 GB including indexes and TOAST, exceeding the observed hosted relation's 160 MB / 381 MB. The physical fixture remains synthetic. + +The actual normal migration runner completed first run in 255.06 ms, no-op rerun in 20.32 ms and interrupted-index repair/retry in 241.73 ms. The final index is 188416 bytes. Concurrent reads took 0.65–7.83 ms and writes 0.71–2.45 ms, without failures. The full command took 66.90 s including both database fixtures. This validates the missing actor access path above the observed target scale, not an unbounded fleet-wide migration duration. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 92993ec6b4b..0fb05bc4aa0 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -546,6 +546,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0391_third_roulette.sql", "0392_small_frightful_four.sql", "0393_open_grandmaster.sql", + "0395_tranquil_firestar.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0395_tranquil_firestar.sql b/packages/db/src/migrations/0395_tranquil_firestar.sql new file mode 100644 index 00000000000..8dff708db69 --- /dev/null +++ b/packages/db/src/migrations/0395_tranquil_firestar.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "activity_log_company_user_issue_actor_idx" ON "activity_log" USING btree ("company_id","actor_id","entity_id") WHERE "activity_log"."actor_type" = 'user' AND "activity_log"."entity_type" = 'issue'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0390_snapshot.json b/packages/db/src/migrations/meta/0395_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0390_snapshot.json rename to packages/db/src/migrations/meta/0395_snapshot.json index 346524fc38b..02a4b17b992 100644 --- a/packages/db/src/migrations/meta/0390_snapshot.json +++ b/packages/db/src/migrations/meta/0395_snapshot.json @@ -1,6 +1,6 @@ { - "id": "f38ffbf3-c0ce-435e-aad8-9e5afa6b9417", - "prevId": "394e85a6-7fc2-4ab1-80a3-33bb4c7003aa", + "id": "66fe02b3-6672-4c19-b447-57acad43bf16", + "prevId": "633afb6b-4b89-44ea-81d5-a477a3696eb5", "version": "7", "dialect": "postgresql", "tables": { @@ -1350,6 +1350,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "activity_log_company_user_issue_actor_idx": { + "name": "activity_log_company_user_issue_actor_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "actor_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "entity_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"activity_log\".\"actor_type\" = 'user' AND \"activity_log\".\"entity_type\" = 'issue'", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -5721,6 +5749,28 @@ } }, "indexes": { + "approvals_pending_retention_idx": { + "name": "approvals_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"approvals\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, "approvals_company_status_type_idx": { "name": "approvals_company_status_type_idx", "columns": [ @@ -32107,6 +32157,28 @@ "method": "btree", "with": {} }, + "grok_dispatch_prepared_idx": { + "name": "grok_dispatch_prepared_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"grok_bot_dispatches\".\"status\" = 'prepared'", + "concurrently": false, + "method": "btree", + "with": {} + }, "grok_dispatch_company_status_idx": { "name": "grok_dispatch_company_status_idx", "columns": [ @@ -41901,6 +41973,28 @@ } }, "indexes": { + "join_requests_pending_retention_idx": { + "name": "join_requests_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"join_requests\".\"status\" = 'pending_approval'", + "concurrently": false, + "method": "btree", + "with": {} + }, "join_requests_invite_unique_idx": { "name": "join_requests_invite_unique_idx", "columns": [ @@ -42921,6 +43015,99 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.native_runner_checkpoints": { + "name": "native_runner_checkpoints", + "schema": "", + "columns": { + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "issue_id": { + "name": "issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "part": { + "name": "part", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "material": { + "name": "material", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "native_runner_checkpoints_headers_idx": { + "name": "native_runner_checkpoints_headers_idx", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_runner_checkpoints\".\"part\" = 'header'", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk": { + "name": "native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk", + "tableFrom": "native_runner_checkpoints", + "tableTo": "heartbeat_runs", + "columnsFrom": [ + "company_id", + "issue_id", + "run_id" + ], + "columnsTo": [ + "company_id", + "native_issue_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "native_runner_checkpoints_run_id_part_pk": { + "name": "native_runner_checkpoints_run_id_part_pk", + "columns": [ + "run_id", + "part" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.pipeline_automation_executions": { "name": "pipeline_automation_executions", "schema": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 607851503cf..535eedc41bc 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2752,6 +2752,13 @@ "when": 1793152860397, "tag": "0394_conscious_whizzer", "breakpoints": true + }, + { + "idx": 395, + "version": "7", + "when": 1793159398533, + "tag": "0395_tranquil_firestar", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/activity_log.ts b/packages/db/src/schema/activity_log.ts index 5633c7b6ad1..03896f7c702 100644 --- a/packages/db/src/schema/activity_log.ts +++ b/packages/db/src/schema/activity_log.ts @@ -1,3 +1,4 @@ +import { sql } from "drizzle-orm"; import { pgTable, uuid, text, timestamp, jsonb, index } from "drizzle-orm/pg-core"; import { companies } from "./companies.js"; import { agents } from "./agents.js"; @@ -33,5 +34,8 @@ export const activityLog = pgTable( ), runIdIdx: index("activity_log_run_id_idx").on(table.runId), entityIdx: index("activity_log_entity_type_id_idx").on(table.entityType, table.entityId), + companyUserIssueActorIdx: index("activity_log_company_user_issue_actor_idx") + .on(table.companyId, table.actorId, table.entityId) + .where(sql`${table.actorType} = 'user' AND ${table.entityType} = 'issue'`), }), ); diff --git a/scripts/benchmark-mine-unread-badge.ts b/scripts/benchmark-mine-unread-badge.ts new file mode 100644 index 00000000000..1a2702a4ae2 --- /dev/null +++ b/scripts/benchmark-mine-unread-badge.ts @@ -0,0 +1,384 @@ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { sql } from "../server/node_modules/drizzle-orm/index.js"; +import { companies, createDb, applyPendingMigrations, inspectMigrations, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js"; +import { issueService } from "../server/src/services/issues.js"; +import { issueReadSqlCondition } from "../server/src/services/authorization.js"; + +const repositoryRoot = fileURLToPath(new URL("../", import.meta.url)); +const repositoryPath = (relative: string) => path.join(repositoryRoot, relative); +const gitOutput = (args: string[]) => execFileSync("git", args, { cwd: repositoryRoot, encoding: "utf8" }); +const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15); +assert.ok(baselineRef, "Pass a pinned pre-fix --baseline-ref="); +const baselineRevision = gitOutput(["rev-parse", `${baselineRef}^{commit}`]).trim(); +const diagnostic = process.argv.includes("--diagnostic"); +const diagnosticDense = process.argv.includes("--diagnostic-dense"); +const fixtureFilter = process.argv.find(arg => arg.startsWith("--fixture="))?.slice(10); +assert.ok(fixtureFilter === undefined || ["small", "dense", "sparse"].includes(fixtureFilter), "--fixture must be small, dense or sparse"); +const verifyMigration = process.argv.includes("--verify-migration"); +const migrationName = "0395_tranquil_firestar.sql"; +const indexName = "activity_log_company_user_issue_actor_idx"; +const migrationSql = await fs.readFile(repositoryPath(`packages/db/src/migrations/${migrationName}`), "utf8"); +const migrationHash = createHash("sha256").update(migrationSql).digest("hex"); +const holdAfterReport = process.argv.find(arg => arg.startsWith("--hold-after-report="))?.slice(20); +const historyDecoys = Number(process.argv.find(arg => arg.startsWith("--history-decoys="))?.slice(17) ?? 0); +assert.ok(Number.isSafeInteger(historyDecoys) && historyDecoys >= 0, "--history-decoys must be a nonnegative integer"); +const treatmentIsBaseline = process.argv.includes("--treatment-is-baseline"); +const enforceBudget = process.argv.includes("--assert-performance"); +const maxDenseArgument = process.argv.find(arg => arg.startsWith("--max-dense-ms="))?.slice(15); +const maxDenseMs = maxDenseArgument === undefined ? Infinity : Number(maxDenseArgument); +const validateBaseline = process.argv.includes("--validate-baseline"); +const validateReport = process.argv.find(arg => arg.startsWith("--validate-report="))?.slice(18); +const sampleGateDir = process.argv.find(arg => arg.startsWith("--sample-gate-dir="))?.slice(18); +const evidenceDir = process.argv.find(arg => arg.startsWith("--evidence-dir="))?.slice(15); +const started = performance.now(); +const modulePaths = ["server/src/services/issues.ts", "server/src/services/issue-participation.ts"]; +const proofPaths = [...modulePaths, "packages/db/src/client.ts", "packages/db/src/schema/activity_log.ts", `packages/db/src/migrations/${migrationName}`]; +const changedDependencies = gitOutput(["diff", "--name-only", baselineRevision, "--", "server/src/services", "packages/db/src", "packages/shared/src"]).trim().split("\n").filter(Boolean); +assert.deepEqual(changedDependencies.filter(file => !modulePaths.includes(file) && file !== "packages/db/src/client.ts" && file !== "packages/db/src/schema/activity_log.ts" && !file.startsWith("packages/db/src/migrations/")), [], "Baseline must use unchanged service dependencies"); +type PerformanceCase = { + jit: string; readMode: string; baseline: { medianMs: number }; treatment: { medianMs: number }; speedup: number; + plans: Array<{ arm: string; queries: Array<{ plan: Array<{ JIT?: { Options?: { Inlining?: boolean; Optimization?: boolean } } }> }> }>; +}; +type PerformanceResult = { fixture: { name: string }; cases: PerformanceCase[] }; +function assertPerformance(results: PerformanceResult[], arm: "baseline" | "treatment" = "treatment") { + const comparisons = results.flatMap(result => result.cases.map(control => ({ fixture: result.fixture.name, ...control }))); + const jitOptions = (result: PerformanceCase, selected: "baseline" | "treatment") => result.plans.find(plan => plan.arm === selected)!.queries[0]!.plan[0]!.JIT?.Options; + const isCompilationTarget = (result: PerformanceCase) => result.jit === "on" && (jitOptions(result, "baseline")?.Inlining || jitOptions(result, "baseline")?.Optimization); + const targets = comparisons.filter(isCompilationTarget); + assert.ok(targets.length > 0, "Fixture must reproduce expensive baseline JIT compilation"); + for (const result of targets) { + const options = jitOptions(result, arm); + assert.ok(!options?.Inlining && !options?.Optimization, `${arm} must avoid full JIT compilation passes`); + if (arm === "treatment") assert.ok(result.speedup >= 1.5, `Default-JIT target requires at least 1.5x improvement, observed ${result.speedup.toFixed(2)}x`); + } + if (arm === "baseline") return; + for (const result of comparisons) { + if (result.fixture === "dense" && result.jit === "on" && Number.isFinite(maxDenseMs)) assert.ok(result.treatment.medianMs <= maxDenseMs, `Dense default-JIT fixture exceeds ${maxDenseMs} ms local budget: ${result.treatment.medianMs.toFixed(2)} ms`); + if (result.fixture !== "dense" && !isCompilationTarget(result)) assert.ok(result.treatment.medianMs <= Math.max(result.baseline.medianMs * 1.25, result.baseline.medianMs + 2), `Control regressed beyond 25% and 2 ms allowance: ${result.fixture}/${result.readMode}/${result.jit}`); + } +} + +if (validateReport) { + const report = JSON.parse(await fs.readFile(validateReport, "utf8")); + assert.equal(report.baselineRevision, baselineRevision); + assert.deepEqual(Object.keys(report.sourceHashes).sort(), [...proofPaths].sort(), "Report must include the measured index and runner sources"); + for (const [file, hash] of Object.entries(report.sourceHashes)) assert.equal(createHash("sha256").update(await fs.readFile(repositoryPath(file))).digest("hex"), hash, `Source changed since measured report: ${file}`); + assertPerformance(report.results, validateBaseline ? "baseline" : "treatment"); + console.log(JSON.stringify({ status: "performance budget passed", report: validateReport, baselineRevision, sourceHashes: report.sourceHashes })); + process.exit(0); +} +const baselineDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".mine-unread-baseline-")); +const rewrite = (source: string, originalPath: string) => source.replace(/from "(\.{1,2}\/[^\"]+)"/g, (_match, specifier: string) => { + const resolved = fileURLToPath(new URL(specifier, pathToFileURL(repositoryPath(originalPath)))); + const target = resolved.endsWith("/services/issue-participation.js") + ? path.join(baselineDir, "issue-participation.ts") : resolved; + return `from ${JSON.stringify(target)}`; +}); +for (const modulePath of modulePaths) { + const source = gitOutput(["show", `${baselineRevision}:${modulePath}`]); + await fs.writeFile(path.join(baselineDir, path.basename(modulePath)), rewrite(source, modulePath)); +} +const baseline = await import(pathToFileURL(path.join(baselineDir, "issues.ts")).href) as typeof import("../server/src/services/issues.js"); +type Query = { sql: string; params: unknown[]; rows: unknown[]; elapsedMs: number }; +let database: Awaited> | undefined; +let db: Db | undefined; +let baselineDb: Db | undefined; +let baselineConnectionString: string; +const keepAlive = setInterval(() => {}, 1000); +try { + database = await startEmbeddedPostgresTestDatabase("mine-unread-benchmark-"); + db = createDb(database.connectionString, { maxConnections: 1 }); + await db.$client.unsafe('CREATE DATABASE "mine_unread_baseline"'); + const baselineUrl = new URL(database.connectionString); + baselineUrl.pathname = "/mine_unread_baseline"; + baselineConnectionString = baselineUrl.href; + await applyPendingMigrations(baselineConnectionString); + baselineDb = createDb(baselineConnectionString, { maxConnections: 1 }); + await baselineDb.$client.unsafe("set client_min_messages = warning"); + await baselineDb.$client.unsafe(`DROP INDEX "${indexName}"`); + await baselineDb.$client.unsafe("DELETE FROM drizzle.__drizzle_migrations WHERE hash=$1", [migrationHash]); + const indexDefinitions = async (source: Db) => source.$client.unsafe("SELECT indexname,indexdef FROM pg_indexes WHERE schemaname='public' ORDER BY indexname"); + const baselineIndexes = await indexDefinitions(baselineDb), treatmentIndexes = await indexDefinitions(db); + assert.deepEqual(treatmentIndexes.filter(index => index.indexname !== indexName), [...baselineIndexes]); + await db.$client.unsafe("set client_min_messages = warning"); + const [settings] = await db.$client.unsafe("select version(), current_setting('shared_buffers') shared_buffers, current_setting('work_mem') work_mem, current_setting('jit') jit, current_setting('jit_above_cost') jit_above_cost"); + await issueService(db).unreadInboxIssueIds(randomUUID(), "parse-smoke", sql`true`); + const setupMs = performance.now() - started; + const captured = new Error("Ranking query captured before execution"); + function captureDb(queries: Query[], captureOnly = false, sourceDb = db!): Db { + function captureBuilder(builder: any) { + const from = builder.from.bind(builder); + builder.from = (...args: any[]) => { + const query = from(...args), execute = query.execute.bind(query); + query.execute = async () => { + const statement = query.toSQL(), started = performance.now(); + if (captureOnly) { queries.push({ ...statement, rows: [], elapsedMs: 0 }); throw captured; } + const rows = await execute(); + queries.push({ ...statement, rows, elapsedMs: performance.now() - started }); + return rows; + }; + return query; + }; + return builder; + } + return new Proxy(sourceDb, { get(target, key) { + if (key === "select") return (...fields: any[]) => captureBuilder((target.select as any)(...fields)); + if (key === "with") return (...ctes: any[]) => { + const builder = (target.with as any)(...ctes), select = builder.select.bind(builder); + builder.select = (...fields: any[]) => captureBuilder(select(...fields)); + return builder; + }; + return Reflect.get(target, key); + } }); + } + const fixtureSeeds = new Map(); + async function seed(name: string, count: number, participatedCount: number, logsPerIssue: number, commentsPerIssue: number, target = db!) { + const fixtureStart = performance.now(); + const inputs = fixtureSeeds.get(name) ?? { companyId: randomUUID(), ids: Array.from({ length: count }, () => randomUUID()), payload: randomBytes(3072).toString("base64") }; + fixtureSeeds.set(name, inputs); + const { companyId, ids, payload } = inputs, userId = `synthetic-${name}`; + await target.insert(companies).values({ id: companyId, name: `Synthetic Mine ${name}`, issuePrefix: name.toUpperCase() }); + await target.$client.unsafe(`insert into issues(id,company_id,title,status,created_by_user_id,created_at,updated_at) + select id,$2::uuid,'Synthetic '||n,'todo',case when n <= $4 and n%2=0 then $3 else 'unrelated-human' end, + '2026-01-01'::timestamptz,'2026-01-01'::timestamptz+n*interval '1 second' + from unnest($1::uuid[]) with ordinality as f(id,n)`, [ids, companyId, userId, participatedCount]); + await target.$client.unsafe(`insert into activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action,details,created_at) + select $1::uuid,'issue',i.id::text,case when n=1 or n%10=0 then 'user' else 'agent' end, + case when (n=1 or n%20=0) and i.id=ANY($5::uuid[]) then $4 when n=1 or n%10=0 then 'other-human' else 'synthetic-worker' end, + case when n=1 or n%10=0 then 'issue.updated' when n%5=0 then 'issue.read_marked' else 'issue.updated' end, + jsonb_build_object('padding',$2::text),'2026-01-01'::timestamptz+n*interval '1 second' + from issues i cross join generate_series(1,$3::int)n where company_id=$1::uuid order by n,i.id`, [companyId, payload, logsPerIssue, userId, ids.slice(0, participatedCount)]); + await target.$client.unsafe(`insert into issue_comments(company_id,issue_id,author_user_id,body,created_at) + select $1::uuid,id,null,$2,'2026-01-02'::timestamptz+n*interval '1 second' + from issues cross join generate_series(1,$3::int)n where company_id=$1::uuid order by n,id`, [companyId, payload, commentsPerIssue]); + if (historyDecoys > 0 && name !== "small") await target.$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','synthetic-worker','heartbeat.run.created', + jsonb_build_object('padding',left($2,256)),'2026-01-01'::timestamptz+n*interval '1 second' + from generate_series(1,$3::int)n`, [companyId, payload, historyDecoys]); + await target.$client.unsafe("ANALYZE"); + const [counts] = await target.$client.unsafe(`SELECT count(*)::int participants FROM issues i WHERE company_id=$1 AND + (created_by_user_id=$2 OR EXISTS(SELECT 1 FROM activity_log a WHERE a.company_id=$1 AND a.entity_type='issue' AND a.entity_id=i.id::text AND a.actor_type='user' AND a.actor_id=$2 AND a.action='issue.updated'))`, [companyId, userId]); + assert.equal(counts!.participants, participatedCount, "Fixture must include its complete declared personal set"); + const [actorCounts] = await target.$client.unsafe("SELECT count(*)::int rows FROM activity_log WHERE company_id=$1 AND actor_type='user' AND actor_id=$2 AND entity_type='issue'", [companyId, userId]); + return { name, companyId, userId, issues: count, participatedCount, activityRows: count * logsPerIssue, + commentRows: count * commentsPerIssue, targetUserActivityRows: actorCounts!.rows, createdParticipation: Math.floor(participatedCount / 2), activityParticipation: Math.ceil(participatedCount / 2), historyDecoys: name === "small" ? 0 : historyDecoys, payloadBytes: Buffer.byteLength(payload), fixtureMs: performance.now() - fixtureStart }; + } + async function run(factory: typeof issueService, fixture: Awaited>, readCondition: Awaited>, sourceDb: Db) { + const queries: Query[] = [], started = performance.now(); + const result = await factory(captureDb(queries, false, sourceDb)).unreadInboxIssueIds(fixture.companyId, fixture.userId, readCondition); + const elapsedMs = performance.now() - started; + return { result, queries, elapsedMs, serialSqlMs: queries.reduce((total, query) => total + query.elapsedMs, 0) }; + } + const stats = (values: number[]) => ({ samplesMs: values, medianMs: [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)], minMs: Math.min(...values), maxMs: Math.max(...values) }); + let migrationProof: unknown; + async function measureMigration(fixture: Awaited>) { + const pending = await inspectMigrations(baselineConnectionString); + assert.deepEqual(pending.pendingMigrations, [migrationName]); + const monitor = createDb(baselineConnectionString, { maxConnections: 2 }); + await monitor.$client.unsafe("SET lock_timeout='1s'"); + await monitor.$client.unsafe("SET statement_timeout='5s'"); + const size = await monitor.$client.unsafe("SELECT reltuples::bigint estimated_rows,pg_relation_size(oid)::text heap_bytes,pg_total_relation_size(oid)::text total_bytes FROM pg_class WHERE oid='activity_log'::regclass"); + let completed = false; + const reads: number[] = [], writes: number[] = [], progress: unknown[] = []; + const load = (async () => { + while (!completed) { + let start = performance.now(); + await monitor.$client.unsafe("INSERT INTO activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action) VALUES($1,'issue',$2,'user','synthetic-migration-writer','issue.updated')", [fixture.companyId, fixtureSeeds.get(fixture.name)!.ids[0]!]); + writes.push(performance.now() - start); + start = performance.now(); + await monitor.$client.unsafe("SELECT id FROM issues WHERE company_id=$1 LIMIT 1", [fixture.companyId]); + reads.push(performance.now() - start); + await new Promise(resolve => setTimeout(resolve, 10)); + } + })(); + const observe = (async () => { + while (!completed) { + const rows = await monitor.$client.unsafe("SELECT phase,blocks_done,blocks_total FROM pg_stat_progress_create_index WHERE datid=(SELECT oid FROM pg_database WHERE datname=current_database())"); + progress.push(...rows); + await new Promise(resolve => setTimeout(resolve, 10)); + } + })(); + let firstRunMs: number; + try { + const start = performance.now(); + await applyPendingMigrations(baselineConnectionString); + firstRunMs = performance.now() - start; + } finally { + completed = true; + await Promise.all([load, observe]); + } + assert.ok(reads.length > 0 && writes.length > 0, "Migration proof must overlap real reads and writes"); + const validity = async () => (await monitor.$client.unsafe("SELECT i.indisvalid valid,pg_relation_size(i.indexrelid)::text bytes FROM pg_index i JOIN pg_class c ON c.oid=i.indexrelid WHERE c.relname=$1", [indexName]))[0]; + assert.equal((await validity())?.valid, true); + const rerunStart = performance.now(); + await applyPendingMigrations(baselineConnectionString); + const rerunMs = performance.now() - rerunStart; + await monitor.$client.unsafe(`DROP INDEX CONCURRENTLY "${indexName}"`); + await monitor.$client.unsafe("DELETE FROM drizzle.__drizzle_migrations WHERE hash=$1", [migrationHash]); + const blocker = await monitor.$client.reserve(); + let interruptionCode: string | undefined; + try { + await blocker.unsafe("BEGIN"); + await blocker.unsafe("INSERT INTO activity_log(company_id,entity_type,entity_id,actor_type,actor_id,action) VALUES($1,'issue',$2,'agent','synthetic-blocker','issue.updated')", [fixture.companyId, fixtureSeeds.get(fixture.name)!.ids[0]!]); + const interrupted = applyPendingMigrations(baselineConnectionString).catch((error: { code?: string }) => { interruptionCode = error.code; }); + const deadline = Date.now() + 10_000; + let cancelled = false; + while (Date.now() < deadline) { + const active = await monitor.$client.unsafe("SELECT p.pid FROM pg_stat_progress_create_index p JOIN pg_class c ON c.oid=p.index_relid WHERE p.datid=(SELECT oid FROM pg_database WHERE datname=current_database()) AND c.relname=$1 AND p.phase LIKE 'waiting for writers%'", [indexName]); + if (active.length) { await monitor.$client.unsafe("SELECT pg_cancel_backend($1)", [active[0]!.pid]); cancelled = true; break; } + await new Promise(resolve => setTimeout(resolve, 25)); + } + await blocker.unsafe("ROLLBACK"); + await interrupted; + assert.ok(cancelled, "Observe and cancel only this owned migration's blocked index build"); + } finally { + await blocker.unsafe("ROLLBACK"); + blocker.release(); + } + assert.equal(interruptionCode, "57014"); + const interruptedIndex = await validity(); + assert.equal(interruptedIndex?.valid, false); + assert.deepEqual((await inspectMigrations(baselineConnectionString)).pendingMigrations, [migrationName]); + const retryStart = performance.now(); + await applyPendingMigrations(baselineConnectionString); + const retryMs = performance.now() - retryStart; + const finalIndex = await validity(); + assert.equal(finalIndex?.valid, true); + assert.equal((await inspectMigrations(baselineConnectionString)).status, "upToDate"); + await monitor.$client.end({ timeout: 0 }); + return { migrationName, migrationSha256: createHash("sha256").update(migrationSql).digest("hex"), statementCount: 1, firstRunMs, rerunMs, retryMs, size, reads: stats(reads), writes: stats(writes), progress, interruptionCode, interruptedIndex, finalIndex }; + } + const results = []; + const shapes = diagnostic ? [[diagnosticDense ? "dense" : "sparse", 1200, diagnosticDense ? 1200 : 12, 50, 20] as const] : [ + ["small", 12, 12, 4, 2], + ["dense", 1200, 1200, 50, 20], + ["sparse", 1200, 12, 50, 20], + ] as const; + const selectedShapes = shapes.filter(shape => fixtureFilter === undefined || shape[0] === fixtureFilter); + assert.ok(selectedShapes.length, "Selected fixture must exist in this diagnostic mode"); + if (holdAfterReport) assert.equal(selectedShapes.length, 1, "Retain only one explicitly selected diagnostic fixture"); + for (const shape of selectedShapes) { + const fixture = { ...await seed(...shape), baselineFixtureMs: (await seed(...shape, baselineDb)).fixtureMs }; + if (sampleGateDir) { + await fs.mkdir(sampleGateDir, { recursive: true }); + const release = path.join(sampleGateDir, `${fixture.name}.release`); + const scoped = await issueReadSqlCondition(db, { type: "board", source: "session", userId: fixture.userId, authorizingUserId: fixture.userId }); + const estimatedPlans = []; + await db.$client.unsafe("set jit = on"); + await baselineDb.$client.unsafe("set jit = on"); + for (const [arm, factory] of [["baseline", baseline.issueService], ["treatment", issueService]] as const) { + const queries: Query[] = []; + try { await factory(captureDb(queries, true, arm === "baseline" ? baselineDb : db)).unreadInboxIssueIds(fixture.companyId, fixture.userId, scoped); } + catch (error) { if (error !== captured) throw error; } + assert.equal(queries.length, 1); + const query = queries[0]!; + const [plan] = await (arm === "baseline" ? baselineDb : db).$client.unsafe(`EXPLAIN (FORMAT JSON) ${query.sql}`, query.params as never[]); + estimatedPlans.push({ arm, plan: plan["QUERY PLAN"] }); + } + await fs.writeFile(path.join(sampleGateDir, `${fixture.name}.ready.json`), JSON.stringify({ fixture, settings, estimatedPlans })); + console.error(`Ready for paired samples: ${fixture.name}`); + const deadline = Date.now() + 600_000; + while (true) { + try { await fs.access(release); break; } catch {} + assert.ok(Date.now() < deadline, `Timed out waiting for ${release}`); + await new Promise(resolve => setTimeout(resolve, 250)); + } + await fs.rm(release); + } + const cases = []; + for (const readMode of ["unrestricted", "scoped"]) { + const readCondition = readMode === "unrestricted" ? sql`true` : await issueReadSqlCondition(db, { + type: "board", source: "session", userId: fixture.userId, authorizingUserId: fixture.userId, + }); + for (const jit of ["on", "off"]) { + await db.$client.unsafe(`set jit = ${jit}`); + await baselineDb.$client.unsafe(`set jit = ${jit}`); + const estimatedPlans = []; + if (diagnostic) { + for (const [arm, factory] of [["baseline", baseline.issueService], ["treatment", issueService]] as const) { + const queries: Query[] = []; + try { await factory(captureDb(queries, true, arm === "baseline" ? baselineDb : db)).unreadInboxIssueIds(fixture.companyId, fixture.userId, readCondition); } + catch (error) { if (error !== captured) throw error; } + assert.equal(queries.length, 1); + const query = queries[0]!; + const [plan] = await (arm === "baseline" ? baselineDb : db).$client.unsafe(`EXPLAIN (FORMAT JSON) ${query.sql}`, query.params as never[]); + estimatedPlans.push({ arm, plan: plan["QUERY PLAN"] }); + console.error(JSON.stringify({ diagnostic: fixture.name, jit, arm, estimatedCost: plan["QUERY PLAN"][0].Plan["Total Cost"] })); + } + } + const beforeTimes: number[] = [], afterTimes: number[] = []; + let beforeResult: string[] | undefined, afterResult: string[] | undefined; + for (let iteration = diagnostic ? 0 : -2; iteration < (diagnostic ? 1 : 5); iteration++) { + const order = iteration % 2 === 0 ? ["before", "after"] : ["after", "before"]; + for (const arm of order) { + const sourceDb = arm === "before" ? baselineDb : db; + const factory = arm === "before" || treatmentIsBaseline ? baseline.issueService : issueService; + const start = performance.now(); + const current = await factory(sourceDb).unreadInboxIssueIds(fixture.companyId, fixture.userId, readCondition); + const elapsedMs = performance.now() - start; + if (arm === "before") { beforeResult = current; if (iteration >= 0) beforeTimes.push(elapsedMs); } + else { afterResult = current; if (iteration >= 0) afterTimes.push(elapsedMs); } + } + assert.deepEqual(afterResult, beforeResult, `${fixture.name}/${jit}: complete returned ID order`); + } + const before = await run(baseline.issueService, fixture, readCondition, baselineDb); + const after = await run(treatmentIsBaseline ? baseline.issueService : issueService, fixture, readCondition, db); + assert.deepEqual(before.result, beforeResult, "SQL capture must preserve the measured baseline result"); + assert.deepEqual(after.result, afterResult, "SQL capture must preserve the measured treatment result"); + assert.equal(before!.queries.length, 2); + assert.equal(after!.queries.length, 2); + for (let index = 0; index < 2; index++) { + const oldQuery = before!.queries[index]!, newQuery = after!.queries[index]!; + const oldRows = await baselineDb.$client.unsafe(oldQuery.sql, oldQuery.params as never[]).values(); + const newRows = await db.$client.unsafe(newQuery.sql, newQuery.params as never[]).values(); + assert.deepEqual(newRows, oldRows, `${fixture.name}/${jit}: full positional SQL results phase ${index + 1}`); + } + const plans = []; + for (const [arm, current] of [["baseline", before!], ["treatment", after!]] as const) { + const queries = []; + for (const query of current.queries) { + const [plan] = await (arm === "baseline" ? baselineDb : db).$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + queries.push({ sql: query.sql, params: query.params, rows: query.rows.length, plan: plan["QUERY PLAN"] }); + } + plans.push({ arm, queries }); + } + const baselineStats = stats(beforeTimes), treatmentStats = stats(afterTimes); + cases.push({ readMode, jit, baseline: baselineStats, treatment: treatmentStats, speedup: baselineStats.medianMs / treatmentStats.medianMs, equalIds: after!.result.length, estimatedPlans, plans }); + } + } + results.push({ fixture, cases }); + if (sampleGateDir) await fs.writeFile(path.join(sampleGateDir, `${fixture.name}.done.json`), JSON.stringify({ fixture, cases })); + if (verifyMigration && shape === selectedShapes[selectedShapes.length - 1]) migrationProof = await measureMigration(fixture); + if (!holdAfterReport) await db.$client.unsafe("TRUNCATE activity_log, issues, companies CASCADE"); + if (!holdAfterReport) await baselineDb.$client.unsafe("TRUNCATE activity_log, issues, companies CASCADE"); + } + const report = { baselineRevision, treatmentRevision: gitOutput(["rev-parse", "HEAD"]).trim(), + treatmentIsBaseline, diagnostic, migrationProof, baselineIndexes, treatmentIndexes, sourceHashes: Object.fromEntries(await Promise.all(proofPaths.map(async file => [file, createHash("sha256").update(await fs.readFile(repositoryPath(file))).digest("hex")]))), + checkedInLockSha256: createHash("sha256").update(await fs.readFile(repositoryPath("pnpm-lock.yaml"))).digest("hex"), nodeVersion: process.version, dependencyVersions: Object.fromEntries(await Promise.all(["postgres", "drizzle-orm"].map(async name => [name, JSON.parse(await fs.readFile(repositoryPath(`packages/db/node_modules/${name}/package.json`), "utf8")).version]))), settings, setupMs, totalMs: performance.now() - started, results }; + const json = JSON.stringify(report, null, 2); + if (evidenceDir) { await fs.mkdir(evidenceDir, { recursive: true }); await fs.writeFile(path.join(evidenceDir, "benchmark.json"), json); } + console.log(json); + if (holdAfterReport) { + await fs.mkdir(holdAfterReport, { recursive: true }); + await fs.writeFile(path.join(holdAfterReport, "ready.json"), JSON.stringify({ baselineConnectionString, treatmentConnectionString: database.connectionString, baselineDir, fixture: results[0]!.fixture }), { mode: 0o600 }); + const deadline = Date.now() + 600_000; + while (true) { + try { await fs.access(path.join(holdAfterReport, "release")); await fs.rm(path.join(holdAfterReport, "release")); break; } catch {} + assert.ok(Date.now() < deadline, "Timed out waiting for retained diagnostic release"); + await new Promise(resolve => setTimeout(resolve, 250)); + } + } + if (enforceBudget && !diagnostic) assertPerformance(results); +} finally { + if (holdAfterReport) await fs.rm(path.join(holdAfterReport, "ready.json"), { force: true }); + await baselineDb?.$client.end({ timeout: 0 }); + 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__/issues-creation-visibility.test.ts b/server/src/__tests__/issues-creation-visibility.test.ts index 798a3114f05..0cee99a5ee1 100644 --- a/server/src/__tests__/issues-creation-visibility.test.ts +++ b/server/src/__tests__/issues-creation-visibility.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { eq } from "drizzle-orm"; +import { eq, sql } from "drizzle-orm"; import request from "supertest"; import { expect, it } from "vitest"; import { activityLog, agents, authUsers, chatConversations, chatEndpoints, chatMessageLinks, issueComments, issueReadStates, issues, issueThreadInteractions, toolApplications, toolConnections } from "@paperclipai/db"; @@ -133,6 +133,7 @@ describeEmbeddedPostgres("issue creation visibility", () => { } await ctx.db.delete(issues).where(eq(issues.id, ids.private!)); expect(await svc.countUnreadTouchedByUser(companyId, userId)).toBe(3); + expect((await svc.unreadInboxIssueIds(companyId, userId, sql`true`)).sort()).toEqual(expected); await ctx.db.insert(issueComments).values({ companyId, issueId: ids.github!, authorUserId: userId, body: "Direct API reply" }); const beforeReply = await request(app).get(`${path}?includeRoutineExecutions=true&inboxArchivedByUserId=me`).expect(200); const imported = beforeReply.body.find((issue: { id: string }) => issue.id === ids.slack); diff --git a/server/src/__tests__/issues-unread-inbox.test.ts b/server/src/__tests__/issues-unread-inbox.test.ts new file mode 100644 index 00000000000..627577b90b1 --- /dev/null +++ b/server/src/__tests__/issues-unread-inbox.test.ts @@ -0,0 +1,123 @@ +import { randomUUID } from "node:crypto"; +import { eq, inArray, sql } from "drizzle-orm"; +import request from "supertest"; +import { expect, it } from "vitest"; +import { activityLog, issueComments, issueInboxArchives, issueReadStates, issues, issueThreadInteractions, issueAccessGrants, projectAccessMembers, projects } from "@paperclipai/db"; +import { sidebarBadgeRoutes } from "../routes/sidebar-badges.js"; +import { issueReadSqlCondition } from "../services/authorization.js"; +import { issueService } from "../services/issues.js"; +import { describeEmbeddedPostgres, routeApp, seedCompanyWithBoardAccess, useEmbeddedPostgres } from "./helpers/route-test-harness.js"; + +const old = new Date("2026-01-01T00:00:00Z"); +const commentAt = new Date("2026-01-02T00:00:00Z"); +const newer = new Date("2026-01-03T00:00:00Z"); +const readable = sql`true`; + +describeEmbeddedPostgres("Mine unread badge", () => { + const ctx = useEmbeddedPostgres("paperclip-mine-unread-"); + + it("ranks all visible Mine tasks before unread with canonical activity and exact ID ties", async () => { + const { companyId, userId } = await seedCompanyWithBoardAccess(ctx.db, "Mine top100"); + const ids = Array.from({ length: 130 }, (_, index) => `10000000-0000-4000-8000-${String(index).padStart(12, "0")}`); + await ctx.db.insert(issues).values(ids.map(id => ({ id, companyId, title: id, status: "todo", createdByUserId: userId, createdAt: old, updatedAt: commentAt }))); + await ctx.db.insert(issueComments).values(ids.map(issueId => ({ companyId, issueId, body: "External", createdAt: commentAt }))); + await ctx.db.insert(issueReadStates).values(ids.slice(30).map(issueId => ({ companyId, issueId, userId, lastReadAt: newer }))); + const svc = issueService(ctx.db); + expect(await svc.unreadInboxIssueIds(companyId, userId, readable)).toEqual([]); + expect((await svc.unreadInboxIssueIds(companyId, userId, inArray(issues.id, ids.slice(0, 30)))).sort()).toEqual(ids.slice(0, 30)); + await ctx.db.insert(activityLog).values({ companyId, entityType: "issue", entityId: ids[0]!, actorType: "agent", actorId: "worker", action: "issue.updated", createdAt: newer }); + expect(await svc.unreadInboxIssueIds(companyId, userId, readable)).toEqual([ids[0]]); + await ctx.db.insert(activityLog).values([ + { companyId, entityType: "issue", entityId: ids[1]!, actorType: "user", actorId: userId, action: "issue.read_marked", createdAt: new Date("2026-01-04") }, + { companyId, entityType: "issue", entityId: ids[2]!, actorType: "user", actorId: userId, action: "issue.recovery_action_resolved", details: { source: "source_revalidation", trigger: "read_projection" }, createdAt: new Date("2026-01-04") }, + ]); + expect(await svc.unreadInboxIssueIds(companyId, userId, readable)).toEqual([ids[0]]); + }); + + it("keeps text activity IDs exact and company/user scoped", async () => { + const { companyId, userId } = await seedCompanyWithBoardAccess(ctx.db, "Mine IDs"); + const other = await seedCompanyWithBoardAccess(ctx.db, "Other company"); + const ids = ["a0000000-0000-4000-8000-000000000001", "b0000000-0000-4000-8000-000000000002", randomUUID(), randomUUID(), randomUUID(), "00000000-0000-0000-0000-000000000000", randomUUID()]; + await ctx.db.insert(issues).values(ids.map(id => ({ id, companyId, title: id, originKind: "routine_execution", assigneeUserId: id === ids[6] ? userId : null, createdAt: old, updatedAt: old }))); + await ctx.db.insert(issueComments).values(ids.map(issueId => ({ companyId, issueId, body: "External", createdAt: commentAt }))); + await ctx.db.insert(activityLog).values([ + { companyId, entityType: "issue", entityId: ids[0]!.toUpperCase(), actorType: "user", actorId: userId, action: "issue.updated" }, + ...[ids[0] + "\n", ids[0] + " ", ids[0]!.replaceAll("-", ""), "{" + ids[0] + "}"].map(entityId => ({ companyId, entityType: "issue", entityId, actorType: "user", actorId: userId, action: "issue.updated" })), + { companyId, entityType: "issue", entityId: ids[5]!, actorType: "user", actorId: userId, action: "issue.updated" }, + { companyId, entityType: "issue", entityId: "not-a-uuid", actorType: "user", actorId: userId, action: "issue.updated" }, + { companyId: other.companyId, entityType: "issue", entityId: ids[1]!, actorType: "user", actorId: userId, action: "issue.updated" }, + { companyId, entityType: "issue", entityId: ids[2]!, actorType: "user", actorId: other.userId, action: "issue.updated" }, + { companyId, entityType: "project", entityId: ids[3]!, actorType: "user", actorId: userId, action: "issue.updated" }, + { companyId, entityType: "issue", entityId: ids[4]!, actorType: "user", actorId: userId, action: "issue.updated" }, + ]); + expect((await issueService(ctx.db).unreadInboxIssueIds(companyId, userId, readable)).sort()).toEqual([ids[4], ids[5], ids[6]].sort()); + }); + + it("preserves pending request audience, self-creator, dismissal and completion rules", async () => { + const { companyId, userId } = await seedCompanyWithBoardAccess(ctx.db, "Mine requests"); + const names = ["addressed", "selfCreator", "otherUser", "dismissed", "done", "cancelled"]; + const ids = Object.fromEntries(names.map(name => [name, randomUUID()])); + await ctx.db.insert(issues).values(names.map(name => ({ id: ids[name]!, companyId, title: name, originKind: "routine_execution", status: name === "done" || name === "cancelled" ? name : "todo", createdAt: old, updatedAt: old }))); + await ctx.db.insert(issueThreadInteractions).values(names.map(name => ({ companyId, issueId: ids[name]!, kind: "request_confirmation", status: "pending", addresseeUserId: name === "otherUser" ? "other-user" : userId, + createdByUserId: name === "selfCreator" ? userId : null, effectiveResolverPolicy: name === "selfCreator" ? "not_creator" as const : "human_only" as const, + effectiveResolverPolicySource: "requested" as const, dismissedAt: name === "dismissed" ? newer : null, payload: { version: 1 as const, prompt: "Accept?" }, createdAt: commentAt }))); + const svc = issueService(ctx.db); + expect(await svc.unreadInboxIssueIds(companyId, userId, readable)).toEqual([ids.addressed]); + await ctx.db.insert(issueReadStates).values({ companyId, issueId: ids.addressed!, userId, lastReadAt: newer }); + expect(await svc.unreadInboxIssueIds(companyId, userId, readable)).toEqual([]); + }); + + it("keeps archive resurfacing and deleted-comment rules", async () => { + const { companyId, userId } = await seedCompanyWithBoardAccess(ctx.db, "Mine archives"); + const names = ["archived", "humanReply", "deletedReply", "statusChange", "doneAlreadyArchived"]; + const ids = Object.fromEntries(names.map(name => [name, randomUUID()])); + await ctx.db.insert(issues).values(names.map(name => ({ id: ids[name]!, companyId, title: name, createdByUserId: userId, status: name === "doneAlreadyArchived" ? "done" : "todo", completedAt: name === "doneAlreadyArchived" ? old : null, createdAt: old, updatedAt: old }))); + await ctx.db.insert(issueInboxArchives).values(names.map(name => ({ companyId, issueId: ids[name]!, userId, archivedAt: commentAt }))); + await ctx.db.insert(issueComments).values([...names.map(name => ({ companyId, issueId: ids[name]!, body: "Old unread", createdAt: commentAt })), + { companyId, issueId: ids.humanReply!, authorUserId: "other-user", body: "Resurface", createdAt: newer }, + { companyId, issueId: ids.deletedReply!, authorUserId: "other-user", body: "Deleted", deletedAt: newer, createdAt: newer }]); + await ctx.db.insert(activityLog).values(["statusChange", "doneAlreadyArchived"].map(name => ({ companyId, entityType: "issue", entityId: ids[name]!, actorType: "agent", actorId: "worker", action: "issue.updated", createdAt: newer, details: { status: name === "statusChange" ? "blocked" : "done", _previous: { status: "in_progress" } } }))); + expect((await issueService(ctx.db).unreadInboxIssueIds(companyId, userId, readable)).sort()).toEqual([ids.humanReply, ids.statusChange].sort()); + }); + + it("applies read restrictions and returns fresh unread state through the badge route", async () => { + const { companyId, userId, actor } = await seedCompanyWithBoardAccess(ctx.db, "Mine access"); + const [visible, denied] = [randomUUID(), randomUUID()]; + await ctx.db.insert(issues).values([{ id: visible, companyId, title: "Visible", createdByUserId: userId, createdAt: old, updatedAt: old }, { id: denied, companyId, title: "Denied", createdByUserId: userId, createdAt: old, updatedAt: newer }]); + await ctx.db.insert(issueComments).values([visible, denied].map(issueId => ({ companyId, issueId, body: "External", createdAt: commentAt }))); + const svc = issueService(ctx.db); + expect(await svc.unreadInboxIssueIds(companyId, userId, eq(issues.id, visible))).toEqual([visible]); + expect(await svc.unreadInboxIssueIds(companyId, userId, sql`false`)).toEqual([]); + const app = routeApp(ctx.db, actor, sidebarBadgeRoutes), url = `/api/companies/${companyId}/sidebar-badges`; + expect((await request(app).get(url).expect(200)).body.unreadIssueIds.sort()).toEqual([visible, denied].sort()); + await ctx.db.insert(issueReadStates).values({ companyId, issueId: visible, userId, lastReadAt: newer }); + expect((await request(app).get(url).expect(200)).body.unreadIssueIds).toEqual([denied]); + await request(app).get(`/api/companies/${randomUUID()}/sidebar-badges`).expect(403); + }); + + it("checks inherited task grants and private project access on each badge read", async () => { + const { companyId, userId, actor } = await seedCompanyWithBoardAccess(ctx.db, "Mine privacy"); + const [parent, child, sibling, projectIssue] = [randomUUID(), randomUUID(), randomUUID(), randomUUID()]; + const [project] = await ctx.db.insert(projects).values({ companyId, name: "Private project", visibility: "private" }).returning(); + await ctx.db.insert(issues).values({ id: parent!, companyId, title: "Private parent", visibility: "private", createdByUserId: "other-user", createdAt: old, updatedAt: old }); + await ctx.db.insert(issues).values([ + { id: child!, companyId, title: "Child", privacyParentIssueId: parent!, createdByUserId: "other-user", createdAt: old, updatedAt: old }, + { id: sibling!, companyId, title: "Sibling", privacyParentIssueId: parent!, createdByUserId: "other-user", createdAt: old, updatedAt: old }, + { id: projectIssue!, companyId, title: "Project task", projectId: project!.id, createdByUserId: "other-user", createdAt: old, updatedAt: old }, + ]); + await ctx.db.insert(issueComments).values([parent!, child!, sibling!, projectIssue!].flatMap(issueId => [ + { companyId, issueId, authorUserId: userId, body: "Direct reply", createdAt: old }, + { companyId, issueId, authorUserId: "other-user", body: "External result", createdAt: commentAt }, + ])); + const readCondition = await issueReadSqlCondition(ctx.db, { ...actor, source: "session" }); + const svc = issueService(ctx.db); + expect(await svc.unreadInboxIssueIds(companyId, userId, readCondition)).toEqual([]); + const [grant] = await ctx.db.insert(issueAccessGrants).values({ issueId: child!, subjectType: "user", subjectId: userId, source: "explicit" }).returning(); + expect(await svc.unreadInboxIssueIds(companyId, userId, readCondition)).toEqual([child]); + await ctx.db.update(issueAccessGrants).set({ revokedAt: newer }).where(eq(issueAccessGrants.id, grant!.id)); + expect(await svc.unreadInboxIssueIds(companyId, userId, readCondition)).toEqual([]); + await ctx.db.insert(projectAccessMembers).values({ companyId, projectId: project!.id, subjectType: "user", subjectId: userId }); + expect(await svc.unreadInboxIssueIds(companyId, userId, readCondition)).toEqual([projectIssue]); + }); + +}); diff --git a/server/src/services/issue-participation.ts b/server/src/services/issue-participation.ts index f615ec7a201..1ea1798ffcd 100644 --- a/server/src/services/issue-participation.ts +++ b/server/src/services/issue-participation.ts @@ -104,10 +104,8 @@ export function directHumanCommentCondition() { )`; } -export function pendingUserRequestCondition(companyId: string, userId: string, after?: SQL) { - return sql`EXISTS ( - SELECT 1 FROM ${issueThreadInteractions} - WHERE ${issueThreadInteractions.companyId} = ${companyId} +export function pendingUserRequestRowCondition(companyId: string, userId: string, after?: SQL) { + return sql`(${issueThreadInteractions.companyId} = ${companyId} AND ${issueThreadInteractions.issueId} = ${issues.id} AND ${issueThreadInteractions.status} = 'pending' AND ${issues.status} NOT IN ('done', 'cancelled') @@ -122,6 +120,31 @@ export function pendingUserRequestCondition(companyId: string, userId: string, a )`; } +export function pendingUserRequestCondition(companyId: string, userId: string, after?: SQL) { + return sql`EXISTS ( + SELECT 1 FROM ${issueThreadInteractions} + WHERE ${pendingUserRequestRowCondition(companyId, userId, after)} + )`; +} + +export function userParticipationActivityCondition(companyId: string, userId: string) { + return sql`( + ${activityLog.entityType} = 'issue' + AND ${activityLog.companyId} = ${companyId} + AND ${activityLog.actorType} = 'user' + AND ${activityLog.actorId} = ${userId} + AND NOT ${automaticReadRecoveryCleanupCondition()} + AND NOT ( + ${activityLog.action} = 'issue.inbox_touched' + AND COALESCE(${activityLog.details}->>'source', '') = 'manual_routine_run' + ) + AND ${activityLog.action} IN (${sql.join( + ISSUE_USER_PARTICIPATION_ACTIVITY_ACTIONS.map((action) => sql`${action}`), + sql`, `, + )}) + )`; +} + export function touchedByUserCondition(companyId: string, userId: string) { return sql` ( @@ -131,22 +154,8 @@ export function touchedByUserCondition(companyId: string, userId: string) { OR EXISTS ( SELECT 1 FROM ${activityLog} - WHERE ${activityLog.entityType} = 'issue' - AND ${activityLog.entityId} = ${issues.id}::text - AND ${activityLog.companyId} = ${companyId} - AND ${activityLog.actorType} = 'user' - AND ${activityLog.actorId} = ${userId} - AND NOT ${automaticReadRecoveryCleanupCondition()} - AND NOT ( - ${activityLog.action} = 'issue.inbox_touched' - AND COALESCE(${activityLog.details}->>'source', '') = 'manual_routine_run' - ) - AND ${activityLog.action} IN (${sql.join( - ISSUE_USER_PARTICIPATION_ACTIVITY_ACTIONS.map( - (action) => sql`${action}`, - ), - sql`, `, - )}) + WHERE ${activityLog.entityId} = ${issues.id}::text + AND ${userParticipationActivityCondition(companyId, userId)} ) OR EXISTS ( SELECT 1 diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 5c559559a70..26ed76d447b 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -9,7 +9,7 @@ import { readNativeMonitorSource } from "./native-runtime/native-monitor-deliver import { readNativeContinuationSource } from "./native-runtime/native-continuation-delivery.js"; import { assertNoCaseHandoffMutation, cancelCaseHandoffsForIssue, completeTerminalCaseHandoffsForIssue } from "./heartbeat/case-handoff-lifecycle.js"; import { measureRunPhase } from "./execution-preparation-timing.js"; -import { automaticReadRecoveryCleanupCondition, directHumanCommentCondition, humanCreatedIssueCondition, pendingUserRequestCondition, touchedByUserCondition } from "./issue-participation.js"; +import { automaticReadRecoveryCleanupCondition, directHumanCommentCondition, humanCreatedIssueCondition, pendingUserRequestCondition, pendingUserRequestRowCondition, touchedByUserCondition, userParticipationActivityCondition } from "./issue-participation.js"; import { resolveProjectDefaultWorkspaceId } from "./project-default-workspace.js"; import { type AuthorizationActor, executionWorkspaceReadSqlCondition, projectReadSqlCondition } from "./authorization.js"; import { assertExecutionWorkspaceRecordAccess, assertTaskWorkspaceSourceProjectAccess } from "./task-workspace-access.js"; @@ -57,6 +57,7 @@ import { sql, type SQL, } from "drizzle-orm"; +import { unionAll } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; import { activityLog, @@ -8055,18 +8056,40 @@ 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 rows = await db.select({ - id: issues.id, - }).from(issues).where(and( + 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 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), visibleIssueCondition(), isNull(issues.conversationAgentId), nonPluginOperationIssueCondition(), ne(issues.status, "cancelled"), - touchedByUserCondition(companyId, userId), inboxVisibleForUserCondition(companyId, userId), readCondition, - )).orderBy(desc(issueCanonicalLastActivityAtExpr(companyId)), desc(issues.updatedAt), desc(issues.id)).limit(100); + )).orderBy(asc(issues.id)).as("issues"); + const rows = await db.select({ id: eligible.id }).from(eligible) + .orderBy(desc(issueCanonicalLastActivityAtExpr(companyId)), desc(eligible.updatedAt), desc(eligible.id)).limit(100); if (rows.length === 0) return []; // Keep correlated predicates in WHERE (as in the list's user stats). // A single-table Drizzle SELECT projection unqualifies nested columns.