mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-29 04:25:52 +01:00
perf: avoid full JIT and activity scans in Mine unread badges (#17973)
## Thinking Path > - Paperclip helps people manage AI agents and their work. > - Mine includes tasks the person created, was assigned, or participated in. > - Its unread badge ranks the visible personal top 100 before checking unread state. > - Correlated participation and privacy predicates can trigger expensive PostgreSQL JIT work. > - This change joins native participant IDs before ranking a narrow candidate projection. > - A partial actor index avoids scanning company activity history for each badge. > - Local representative scoped reads improve while dense unrestricted overhead remains explicit. ## Linked Issues or Issue Description **What existing behavior does this improve?** The Mine unread badge in the company sidebar. **Current behavior:** A fresh five-minute window recorded 402 ranking calls and 31.7 seconds of SQL time. Correlated predicates can spend more time compiling than executing. **Proposed behavior:** Build the same company-scoped participant set, use native issue keys, apply fresh authority before the top-100 limit, then retain the original canonical rank and second unread query. **Reason and benefit:** Reduce full JIT compilation and avoid an unindexed company-history activity scan. Realistic activity-only participation is included in the measurement. **Breaking changes:** No API or policy change. This PR adds one concurrent partial index migration. Related Mine rules in #16356 and #17756 are preserved. ## What Changed - Join distinct native participant IDs; guard exact legacy text before UUID conversion. - Skip redundant activity participation already covered by creation/assignment. Sort a narrow eligible candidate projection before canonical activity ranking. - Share existing pending-request and user-activity predicates with ordinary Mine reads. - Add partial `activity_log(company_id, actor_id, entity_id)` for user-authored issue activity. Register its idempotent concurrent migration with the existing top-level/invalid-index-repair runner. - Add real PostgreSQL coverage for visible top 100, unread/archive/source semantics, exact IDs, company scope and fresh grant/private-project access. - Add a pinned service benchmark with separate pre-index/indexed databases, identical data, full positional equality, ordinary-call timings, plans and real migration/cancellation proof. ## Verification - Fourteen focused PostgreSQL 17 tests pass (16.37 s), including assembled badge routes and grant revocation. Full workspace typecheck and build pass on the final production sources. - The earlier full local suite is incomplete: native PostgreSQL startup exhausted system semaphores, independently reproduced with empty initdb. All sixteen failed cases passed with task-owned Docker PostgreSQL 17. This is not counted as a full local-suite pass. Full Linux CI passes on final head `f44e46ce580e8345209d7165f57bd8bf696c517c`. Greptile and Paperclip Review are both 5/5 with zero unresolved comments. - Baseline `70c864633744cd96c0dc7394d631fc7119cd17d2`; source and DDL hashes retained. Two warmups and five alternating samples per arm/cell. Capture and EXPLAIN occur outside timed ordinary service calls. - All twelve 300k-history cells match complete positional rows in both SQL phases and final returned IDs. Half the personal tasks qualify through activity alone; the actor index contains matching and other-human rows. | Local service case | Before median | After median | Before / after | | --- | ---: | ---: | ---: | | dense / unrestricted / JIT on | 18.33 ms | 25.06 ms | 0.73x | | dense / scoped / JIT on | 909.30 ms | 221.10 ms | 4.11x | | sparse / unrestricted / JIT on | 13.64 ms | 5.84 ms | 2.34x | | sparse / scoped / JIT on | 719.07 ms | 10.79 ms | 66.64x | Small paths add at most 0.6 ms at the median. Dense unrestricted paths regress by 6.7–7.7 ms; dense scoped JIT-off is approximately neutral (218.34→223.24 ms). Sparse JIT-off paths improve too. This is not a uniform speedup. The actual baseline fails the compiler-work assertion. The final query avoids full JIT optimization/inlining and exceeds 1.5x where the old plan triggers those passes. Small/sparse controls meet the 25% or 2 ms allowance. Dense no-JIT cases are explicit diagnostics. The optional 60 ms absolute goal still fails at 221.10 ms on dense scoped reads: substantial privacy-predicate execution remains after compilation is removed. The failure is retained rather than counted as green. The actual normal migration runner upgraded 300k activity rows in 668.79 ms including inspection, index build/validation and journal work. Rerun took 21.83 ms; a deliberately cancelled build left an invalid index, and retry repaired it in 187.71 ms. Concurrent reads/writes completed; the final index was 188416 bytes. No migration was run on the hosted database. The million-row mixed-participation sparse control also passes all four visibility/JIT pairs with complete equality: unrestricted JIT-on 13.44→5.76 ms; scoped JIT-on 807.76→11.59 ms. The actor branch visits 36 indexed rows rather than scanning company history. Actual migration+runner took 255.06 ms; rerun 20.32 ms; cancellation repair/retry 241.73 ms. Concurrent reads and writes completed without failures. Reproduce: ```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=<directory> ``` Add `--fixture=sparse --history-decoys=940000` for the million-row control. The default harness starts disposable embedded PostgreSQL and ignores DATABASE_URL. Recorded evidence used PostgreSQL 17.11 through a scratch constructor provider because the host native runtime has semaphore limits. All normal database allocation/migration/client paths are retained. Source-only EXISTS/lateral alternatives were measured and rejected after company scans or sparse regressions. A time bound would lose historical participation, and existing indexes cannot select a user's activity. This justifies the exceptional partial index migration. The exact sample ranges, plans, failures, source hashes, migration footprint and feedback-loop timing are in `doc/plans/2026-10-27-mine-unread-badge-query.md`. ## Risks - Dense unrestricted reads are slightly slower, and dense scoped reads retain substantial authority work. The PR removes a compilation/lookup bottleneck rather than all badge latency. - The concurrent build scans activity history and can wait for long-lived transactions. Local build/cancellation timings do not guarantee the duration on larger or slower hosted databases. User-authored issue activity gains one narrow index to maintain. - Participation, visibility, ranking and legacy ID semantics must remain exact; the real database regressions exercise those boundaries. - Hosted improvement is unverified. The affected ranking SQL averaged 78.8 ms while HTTP badge requests averaged 752 ms without correlated traces. No merge or deployment is included. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, tool use and local code execution. The runtime does not expose the exact model ID or context window to this agent. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
a91f78e7e0
commit
7d8bab773d
11 files changed
+857
-29
No files matched your search
@@ -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=<directory>
|
||||
```
|
||||
|
||||
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=<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=<directory>` 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.
|
||||
@@ -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",
|
||||
|
||||
@@ -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';
|
||||
+189
-2
@@ -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": "",
|
||||
+7
@@ -2752,6 +2752,13 @@
|
||||
"when": 1793152860397,
|
||||
"tag": "0394_conscious_whizzer",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 395,
|
||||
"version": "7",
|
||||
"when": 1793159398533,
|
||||
"tag": "0395_tranquil_firestar",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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'`),
|
||||
}),
|
||||
);
|
||||
@@ -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=<commit>");
|
||||
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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<boolean>`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<string, { companyId: string; ids: string[]; payload: string }>();
|
||||
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<ReturnType<typeof seed>>, readCondition: Awaited<ReturnType<typeof issueReadSqlCondition>>, 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<ReturnType<typeof seed>>) {
|
||||
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<boolean>`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);
|
||||
}
|
||||
@@ -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<boolean>`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);
|
||||
|
||||
@@ -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<boolean>`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<boolean>`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]);
|
||||
});
|
||||
|
||||
});
|
||||
@@ -104,10 +104,8 @@ export function directHumanCommentCondition() {
|
||||
)`;
|
||||
}
|
||||
|
||||
export function pendingUserRequestCondition(companyId: string, userId: string, after?: SQL) {
|
||||
return sql<boolean>`EXISTS (
|
||||
SELECT 1 FROM ${issueThreadInteractions}
|
||||
WHERE ${issueThreadInteractions.companyId} = ${companyId}
|
||||
export function pendingUserRequestRowCondition(companyId: string, userId: string, after?: SQL) {
|
||||
return sql<boolean>`(${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<boolean>`EXISTS (
|
||||
SELECT 1 FROM ${issueThreadInteractions}
|
||||
WHERE ${pendingUserRequestRowCondition(companyId, userId, after)}
|
||||
)`;
|
||||
}
|
||||
|
||||
export function userParticipationActivityCondition(companyId: string, userId: string) {
|
||||
return sql<boolean>`(
|
||||
${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<boolean>`
|
||||
(
|
||||
@@ -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
|
||||
|
||||
@@ -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<boolean>) => {
|
||||
// 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<string | null>`${issues.id}`.as("id") }).from(issues).where(and(
|
||||
eq(issues.companyId, companyId),
|
||||
directlyParticipated,
|
||||
)),
|
||||
db.select({ id: sql<string | null>`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<string>`${issues.id}::text` }).from(issues)
|
||||
.where(and(eq(issues.companyId, companyId), directlyParticipated))),
|
||||
)),
|
||||
db.select({ id: sql<string | null>`${issueComments.issueId}`.as("id") }).from(issueComments).where(and(
|
||||
eq(issueComments.companyId, companyId), eq(issueComments.authorUserId, userId), directHumanCommentCondition(),
|
||||
)),
|
||||
db.select({ id: sql<string | null>`${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.
|
||||
|
||||
Reference in new issue
Block a user