diff --git a/doc/DATABASE.md b/doc/DATABASE.md index d588e082360..b3d09d77103 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -2,6 +2,32 @@ Paperclip uses PostgreSQL via [Drizzle ORM](https://orm.drizzle.team/). There are three ways to run the database, from simplest to most production-ready. +### Task query performance + +Run-context lookups use expression indexes for both `context_snapshot.issueId` +and `context_snapshot.paperclipIssue.id`. Task creation provenance has its own +company-scoped expression index that follows the query's native issue ID and +legacy `issueId`/`taskId`/`taskKey` precedence. Migration 0345 builds the two new +indexes concurrently and can resume an interrupted build without blocking run +writes for the full build. The indexes add storage and run-write overhead; they +do not change task visibility or secret redaction. + +The chat wake-notice sweep compares UUID columns directly against guarded, +canonical lowercase UUID payload values. Malformed and noncanonical payloads +still match nothing. The query can use existing primary keys without adding a +wakeup-history index. + +To compare all three queries on the same disposable local PostgreSQL fixture: + +```sh +pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-slow-queries.ts --baseline-ref=origin/master +``` + +Use a revision before these fixes for `--baseline-ref` after they merge. The +benchmark loads that revision's chat query, uses synthetic data only, compares +complete returned rows, and prints timings and `EXPLAIN (ANALYZE, BUFFERS)` plans. +Local results measure query execution, not staging request latency. + ## 1. Embedded PostgreSQL — zero config If you don't set `DATABASE_URL`, the server automatically starts an embedded PostgreSQL instance and manages a local data directory. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 199a28ebe43..1f4a9257a39 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -473,6 +473,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0313_private_task_access.sql", "0314_private_task_draft_assets.sql", "0332_little_night_nurse.sql", + "0345_luxuriant_carmella_unuscione.sql", ]); async function applyPendingMigrationsManually( diff --git a/packages/db/src/heartbeat-context-snapshot-index-migration.test.ts b/packages/db/src/heartbeat-context-snapshot-index-migration.test.ts index 36a4ea6a768..d1dafde28b0 100644 --- a/packages/db/src/heartbeat-context-snapshot-index-migration.test.ts +++ b/packages/db/src/heartbeat-context-snapshot-index-migration.test.ts @@ -27,6 +27,8 @@ d("heartbeat context_snapshot expression index migration", () => { expect(names).toContain("heartbeat_runs_company_ctx_issue_created_idx"); expect(names).toContain("heartbeat_runs_company_ctx_task_created_idx"); expect(names).toContain("heartbeat_runs_company_ctx_taskkey_created_idx"); + expect(names).toContain("heartbeat_runs_company_ctx_paperclip_issue_created_idx"); + expect(names).toContain("heartbeat_runs_company_origin_issue_idx"); expect(names).toContain("agent_wakeup_requests_company_payload_issue_idx"); await sql.unsafe("SET enable_seqscan = off"); @@ -52,6 +54,20 @@ d("heartbeat context_snapshot expression index migration", () => { const taskKeyText = taskKeyPlan.map((r) => Object.values(r)[0]).join("\n"); expect(taskKeyText).toContain("heartbeat_runs_company_ctx_taskkey_created_idx"); + // An empty table can use any company-prefix index for the OR. Check the + // new branch directly here; the populated benchmark proves BitmapOr use. + const redactionPlan = await sql.unsafe( + "EXPLAIN SELECT id FROM heartbeat_runs WHERE company_id = '00000000-0000-0000-0000-000000000001' AND context_snapshot -> 'paperclipIssue' ->> 'id' = 'x'", + ); + const redactionText = redactionPlan.map((r) => Object.values(r)[0]).join("\n"); + expect(redactionText).toContain("heartbeat_runs_company_ctx_paperclip_issue_created_idx"); + + const originPlan = await sql.unsafe( + "EXPLAIN SELECT id FROM heartbeat_runs WHERE company_id = '00000000-0000-0000-0000-000000000001' AND coalesce(native_issue_id::text, nullif(context_snapshot->>'issueId', ''), nullif(context_snapshot->>'taskId', ''), nullif(context_snapshot->>'taskKey', '')) IN ('x', 'SRC-1')", + ); + expect(originPlan.map((r) => Object.values(r)[0]).join("\n")) + .toContain("heartbeat_runs_company_origin_issue_idx"); + const wakePlan = await sql.unsafe( "EXPLAIN SELECT id FROM agent_wakeup_requests WHERE company_id = '00000000-0000-0000-0000-000000000001' AND status = 'deferred_issue_execution' AND payload ->> 'issueId' = 'x' LIMIT 1", ); @@ -63,6 +79,7 @@ d("heartbeat context_snapshot expression index migration", () => { for (const migration of [ "./migrations/0209_heartbeat_context_snapshot_indexes.sql", "./migrations/0210_heartbeat_context_taskkey_index.sql", + "./migrations/0345_luxuriant_carmella_unuscione.sql", ]) { const migrationSql = await readFile( fileURLToPath(new URL(migration, import.meta.url)), diff --git a/packages/db/src/migrations/0345_luxuriant_carmella_unuscione.sql b/packages/db/src/migrations/0345_luxuriant_carmella_unuscione.sql new file mode 100644 index 00000000000..88c08e1c1a9 --- /dev/null +++ b/packages/db/src/migrations/0345_luxuriant_carmella_unuscione.sql @@ -0,0 +1,9 @@ +-- Keep both JSON branches indexable without blocking run writes during the build. +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_company_ctx_paperclip_issue_created_idx" ON "heartbeat_runs" USING btree ("company_id",("context_snapshot" -> 'paperclipIssue' ->> 'id'),"created_at" DESC NULLS LAST);--> statement-breakpoint +-- Match the creation-source query's exact native/legacy precedence. +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_company_origin_issue_idx" ON "heartbeat_runs" USING btree ("company_id",coalesce( + "native_issue_id"::text, + nullif("context_snapshot"->>'issueId', ''), + nullif("context_snapshot"->>'taskId', ''), + nullif("context_snapshot"->>'taskKey', '') + )); diff --git a/packages/db/src/migrations/meta/0340_snapshot.json b/packages/db/src/migrations/meta/0345_snapshot.json similarity index 98% rename from packages/db/src/migrations/meta/0340_snapshot.json rename to packages/db/src/migrations/meta/0345_snapshot.json index 188aeb107d1..ddfc238b06d 100644 --- a/packages/db/src/migrations/meta/0340_snapshot.json +++ b/packages/db/src/migrations/meta/0345_snapshot.json @@ -1,6 +1,6 @@ { - "id": "b32fa655-79c7-4793-a021-22258b6c369c", - "prevId": "045fe4f7-1636-4e1b-b304-11b6128fefcf", + "id": "3bd7b1df-1434-4fac-87d6-53b90a9eb9e1", + "prevId": "aaf08b19-e3cc-4889-8e9d-5081e2e3cf8e", "version": "7", "dialect": "postgresql", "tables": { @@ -4098,6 +4098,12 @@ "primaryKey": false, "notNull": false }, + "created_by_user_id": { + "name": "created_by_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "created_at": { "name": "created_at", "type": "timestamp with time zone", @@ -16852,6 +16858,53 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.company_skill_import_budgets": { + "name": "company_skill_import_budgets", + "schema": "", + "columns": { + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "available_bytes": { + "name": "available_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 536870912 + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "company_skill_import_budgets_company_id_companies_id_fk": { + "name": "company_skill_import_budgets_company_id_companies_id_fk", + "tableFrom": "company_skill_import_budgets", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.company_skill_source_entries": { "name": "company_skill_source_entries", "schema": "", @@ -17091,6 +17144,72 @@ "notNull": true, "default": true }, + "removed_at": { + "name": "removed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "automatic_updates": { + "name": "automatic_updates", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "import_new_skills": { + "name": "import_new_skills", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "access_scope": { + "name": "access_scope", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'company'" + }, + "refresh_user_id": { + "name": "refresh_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "refresh_agent_id": { + "name": "refresh_agent_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "refresh_local_trusted": { + "name": "refresh_local_trusted", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "refresh_grant_id": { + "name": "refresh_grant_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "next_sync_at": { + "name": "next_sync_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "failure_count": { + "name": "failure_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, "revision": { "name": "revision", "type": "integer", @@ -17116,6 +17235,26 @@ "primaryKey": false, "notNull": false }, + "last_scan_tree_sha": { + "name": "last_scan_tree_sha", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "automatic_budget_bytes": { + "name": "automatic_budget_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 536870912 + }, + "automatic_budget_at": { + "name": "automatic_budget_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, "last_error": { "name": "last_error", "type": "text", @@ -17166,6 +17305,7 @@ } ], "isUnique": true, + "where": "\"company_skill_sources\".\"removed_at\" is null", "concurrently": false, "method": "btree", "with": {} @@ -17193,6 +17333,7 @@ } ], "isUnique": true, + "where": "\"company_skill_sources\".\"removed_at\" is null", "concurrently": false, "method": "btree", "with": {} @@ -17211,13 +17352,275 @@ ], "onDelete": "cascade", "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.project_skill_sources": { + "name": "project_skill_sources", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" }, - "company_skill_sources_connection_id_tool_connections_id_fk": { - "name": "company_skill_sources_connection_id_tool_connections_id_fk", - "tableFrom": "company_skill_sources", - "tableTo": "tool_connections", + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "project_id": { + "name": "project_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "workspace_id": { + "name": "workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "source_id": { + "name": "source_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "repository_url": { + "name": "repository_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "tracking_ref": { + "name": "tracking_ref", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "active": { + "name": "active", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "refresh_user_id": { + "name": "refresh_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "refresh_agent_id": { + "name": "refresh_agent_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "refresh_local_trusted": { + "name": "refresh_local_trusted", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "connection_id": { + "name": "connection_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "refresh_grant_id": { + "name": "refresh_grant_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "next_sync_at": { + "name": "next_sync_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "failure_count": { + "name": "failure_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "last_error": { + "name": "last_error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "lease_token": { + "name": "lease_token", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "lease_expires_at": { + "name": "lease_expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "project_skill_sources_workspace_repo_ref_idx": { + "name": "project_skill_sources_workspace_repo_ref_idx", + "columns": [ + { + "expression": "workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "repository_url", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "tracking_ref", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "project_skill_sources_source_project_idx": { + "name": "project_skill_sources_source_project_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "source_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "project_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "project_skill_sources_sync_idx": { + "name": "project_skill_sources_sync_idx", + "columns": [ + { + "expression": "active", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "next_sync_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "project_skill_sources_company_id_companies_id_fk": { + "name": "project_skill_sources_company_id_companies_id_fk", + "tableFrom": "project_skill_sources", + "tableTo": "companies", "columnsFrom": [ - "connection_id" + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "project_skill_sources_project_id_projects_id_fk": { + "name": "project_skill_sources_project_id_projects_id_fk", + "tableFrom": "project_skill_sources", + "tableTo": "projects", + "columnsFrom": [ + "project_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "project_skill_sources_workspace_id_project_workspaces_id_fk": { + "name": "project_skill_sources_workspace_id_project_workspaces_id_fk", + "tableFrom": "project_skill_sources", + "tableTo": "project_workspaces", + "columnsFrom": [ + "workspace_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, + "project_skill_sources_source_id_company_skill_sources_id_fk": { + "name": "project_skill_sources_source_id_company_skill_sources_id_fk", + "tableFrom": "project_skill_sources", + "tableTo": "company_skill_sources", + "columnsFrom": [ + "source_id" ], "columnsTo": [ "id" @@ -18677,6 +19080,12 @@ "notNull": true, "default": "'company'" }, + "access_source_id": { + "name": "access_source_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, "public_share_token": { "name": "public_share_token", "type": "text", @@ -24571,6 +24980,13 @@ "primaryKey": false, "notNull": false }, + "provider_kind": { + "name": "provider_kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'openai_dot'" + }, "generation": { "name": "generation", "type": "integer", @@ -25198,6 +25614,18 @@ "primaryKey": false, "notNull": false }, + "pending_returned_at": { + "name": "pending_returned_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "observed_at": { + "name": "observed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "source_event_id": { "name": "source_event_id", "type": "text", @@ -30282,6 +30710,474 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.grok_bot_dispatches": { + "name": "grok_bot_dispatches", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "routine_id": { + "name": "routine_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "binding_id": { + "name": "binding_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "binding_generation": { + "name": "binding_generation", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "grant_id": { + "name": "grant_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "routine_revision": { + "name": "routine_revision", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "assignment_id": { + "name": "assignment_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "source_event_id": { + "name": "source_event_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "payload": { + "name": "payload", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "challenge_hash": { + "name": "challenge_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'prepared'" + }, + "http_status": { + "name": "http_status", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "sent_at": { + "name": "sent_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "acknowledged_at": { + "name": "acknowledged_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "grok_dispatch_event_uq": { + "name": "grok_dispatch_event_uq", + "columns": [ + { + "expression": "binding_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "binding_generation", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "routine_revision", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "source_event_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "grok_dispatch_company_status_idx": { + "name": "grok_dispatch_company_status_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "grok_bot_dispatches_company_id_companies_id_fk": { + "name": "grok_bot_dispatches_company_id_companies_id_fk", + "tableFrom": "grok_bot_dispatches", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "grok_bot_dispatches_routine_id_grok_bot_routines_id_fk": { + "name": "grok_bot_dispatches_routine_id_grok_bot_routines_id_fk", + "tableFrom": "grok_bot_dispatches", + "tableTo": "grok_bot_routines", + "columnsFrom": [ + "routine_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "grok_bot_dispatches_binding_id_dot_agent_bindings_id_fk": { + "name": "grok_bot_dispatches_binding_id_dot_agent_bindings_id_fk", + "tableFrom": "grok_bot_dispatches", + "tableTo": "dot_agent_bindings", + "columnsFrom": [ + "binding_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "grok_bot_dispatches_grant_id_mcp_oauth_grants_id_fk": { + "name": "grok_bot_dispatches_grant_id_mcp_oauth_grants_id_fk", + "tableFrom": "grok_bot_dispatches", + "tableTo": "mcp_oauth_grants", + "columnsFrom": [ + "grant_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "grok_bot_dispatches_assignment_id_dot_runner_assignments_id_fk": { + "name": "grok_bot_dispatches_assignment_id_dot_runner_assignments_id_fk", + "tableFrom": "grok_bot_dispatches", + "tableTo": "dot_runner_assignments", + "columnsFrom": [ + "assignment_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.grok_bot_routines": { + "name": "grok_bot_routines", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "agent_id": { + "name": "agent_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "binding_id": { + "name": "binding_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "binding_generation": { + "name": "binding_generation", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "bot_name": { + "name": "bot_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "secret_id": { + "name": "secret_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "secret_version": { + "name": "secret_version", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 1 + }, + "verified_grant_id": { + "name": "verified_grant_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "verified_at": { + "name": "verified_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "last_contact_at": { + "name": "last_contact_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "recovery_required_at": { + "name": "recovery_required_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "recovery_reason": { + "name": "recovery_reason", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "grok_routines_binding_uq": { + "name": "grok_routines_binding_uq", + "columns": [ + { + "expression": "binding_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "grok_routines_company_agent_idx": { + "name": "grok_routines_company_agent_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "agent_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "grok_bot_routines_company_id_companies_id_fk": { + "name": "grok_bot_routines_company_id_companies_id_fk", + "tableFrom": "grok_bot_routines", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "grok_bot_routines_agent_id_agents_id_fk": { + "name": "grok_bot_routines_agent_id_agents_id_fk", + "tableFrom": "grok_bot_routines", + "tableTo": "agents", + "columnsFrom": [ + "agent_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "grok_bot_routines_binding_id_dot_agent_bindings_id_fk": { + "name": "grok_bot_routines_binding_id_dot_agent_bindings_id_fk", + "tableFrom": "grok_bot_routines", + "tableTo": "dot_agent_bindings", + "columnsFrom": [ + "binding_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "grok_bot_routines_secret_id_company_secrets_id_fk": { + "name": "grok_bot_routines_secret_id_company_secrets_id_fk", + "tableFrom": "grok_bot_routines", + "tableTo": "company_secrets", + "columnsFrom": [ + "secret_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "grok_bot_routines_verified_grant_id_mcp_oauth_grants_id_fk": { + "name": "grok_bot_routines_verified_grant_id_mcp_oauth_grants_id_fk", + "tableFrom": "grok_bot_routines", + "tableTo": "mcp_oauth_grants", + "columnsFrom": [ + "verified_grant_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.heartbeat_run_events": { "name": "heartbeat_run_events", "schema": "", @@ -31611,6 +32507,54 @@ "concurrently": false, "method": "btree", "with": {} + }, + "heartbeat_runs_company_ctx_paperclip_issue_created_idx": { + "name": "heartbeat_runs_company_ctx_paperclip_issue_created_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"context_snapshot\" -> 'paperclipIssue' ->> 'id')", + "asc": true, + "isExpression": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": true, + "method": "btree", + "with": {} + }, + "heartbeat_runs_company_origin_issue_idx": { + "name": "heartbeat_runs_company_origin_issue_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "coalesce(\n \"native_issue_id\"::text,\n nullif(\"context_snapshot\"->>'issueId', ''),\n nullif(\"context_snapshot\"->>'taskId', ''),\n nullif(\"context_snapshot\"->>'taskKey', '')\n )", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": true, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -58029,6 +58973,18 @@ "name": "mcp_attachment_uploads", "schema": "", "columns": { + "grant_id": { + "name": "grant_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "work_session_id": { + "name": "work_session_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, "id": { "name": "id", "type": "uuid", @@ -58173,6 +59129,19 @@ } }, "foreignKeys": { + "mcp_attachment_uploads_grant_id_mcp_oauth_grants_id_fk": { + "name": "mcp_attachment_uploads_grant_id_mcp_oauth_grants_id_fk", + "tableFrom": "mcp_attachment_uploads", + "tableTo": "mcp_oauth_grants", + "columnsFrom": [ + "grant_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, "mcp_attachment_uploads_company_id_companies_id_fk": { "name": "mcp_attachment_uploads_company_id_companies_id_fk", "tableFrom": "mcp_attachment_uploads", @@ -58649,6 +59618,12 @@ "name": "mcp_file_tickets", "schema": "", "columns": { + "work_session_id": { + "name": "work_session_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, "token_hash": { "name": "token_hash", "type": "text", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 0b72e6b0405..1fd5c5fb5a8 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2402,6 +2402,13 @@ "when": 1791930226792, "tag": "0344_wide_blue_marvel", "breakpoints": true + }, + { + "idx": 345, + "version": "7", + "when": 1791978393174, + "tag": "0345_luxuriant_carmella_unuscione", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 8d0ac21e6ee..39af2c33a31 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -189,5 +189,17 @@ export const heartbeatRuns = pgTable( sql`(${table.contextSnapshot} ->> 'taskKey')`, table.createdAt.desc(), ), + companyCtxPaperclipIssueCreatedIdx: index("heartbeat_runs_company_ctx_paperclip_issue_created_idx") + .on(table.companyId, sql`(${table.contextSnapshot} -> 'paperclipIssue' ->> 'id')`, table.createdAt.desc()) + .concurrently(), + // Match creation provenance precedence, including empty legacy IDs. + companyOriginIssueIdx: index("heartbeat_runs_company_origin_issue_idx") + .on(table.companyId, sql`coalesce( + ${table.nativeIssueId}::text, + nullif(${table.contextSnapshot}->>'issueId', ''), + nullif(${table.contextSnapshot}->>'taskId', ''), + nullif(${table.contextSnapshot}->>'taskKey', '') + )`) + .concurrently(), }), ); diff --git a/scripts/benchmark-slow-queries.ts b/scripts/benchmark-slow-queries.ts new file mode 100644 index 00000000000..a2f68837bff --- /dev/null +++ b/scripts/benchmark-slow-queries.ts @@ -0,0 +1,184 @@ +/** Disposable PostgreSQL repro. Run: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-slow-queries.ts */ +import assert from "node:assert/strict"; +import { randomBytes, randomUUID } from "node:crypto"; +import { execFileSync } from "node:child_process"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { createRequire } from "node:module"; +const { and, eq, or, sql } = createRequire(new URL("../server/package.json", import.meta.url))("drizzle-orm"); +import { + agents, agentWakeupRequests, chatActions, chatConversations, chatEndpoints, + companies, createDb, heartbeatRuns, issues, startEmbeddedPostgresTestDatabase, + toolApplications, toolConnections, type Db, +} from "../packages/db/src/index.js"; +import { issueService } from "../server/src/services/issues.js"; +import { issueReadSqlCondition } from "../server/src/services/authorization.js"; +import { createInboundWakeNotices } from "../server/src/services/chat-channels/inbound/wakeup-notices.js"; + +console.error("[benchmark] imports complete"); + +const migration = await fs.readFile(new URL("../packages/db/src/migrations/0345_luxuriant_carmella_unuscione.sql", import.meta.url), "utf8"); +const indexes = migration.split("--> statement-breakpoint").map((statement) => statement.trim()).filter(Boolean); +const baselineRef = process.argv.find((arg) => arg.startsWith("--baseline-ref="))?.slice("--baseline-ref=".length) ?? "origin/master"; +const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8" }).trim(); +// The database helper uses unreferenced retry/cleanup timers. A standalone +// benchmark must stay alive until startup succeeds or reports its real error. +const keepAlive = setInterval(() => {}, 1000); +const database = await startEmbeddedPostgresTestDatabase("slow-query-benchmark-"); +console.error("[benchmark] database ready"); +const db = createDb(database.connectionString); +let baselineModuleDir: string | undefined; +type Query = { sql: string; params: unknown[] }; + +// Capture the real service SQL before execution; never dispatch notices or runs. +async function captureQuery(invoke: (captureDb: Db) => Promise): Promise { + let captured: Query | undefined; + const stop = new Error("query captured"); + const captureDb = new Proxy(db, { + get(target, key) { + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = db.select(...fields); + const from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + const query = from(...args); + query.execute = async () => { captured = query.toSQL(); throw stop; }; + return query; + }) as typeof builder.from; + return builder; + }; + }, + }); + try { await invoke(captureDb); } catch (error) { if (error !== stop) throw error; } + assert.ok(captured, "service must produce a SQL query"); + return captured; +} + +async function measure(query: Query) { + const samples: number[] = []; + let rows: unknown[] = []; + for (let sample = 0; sample < 7; sample++) { + const start = performance.now(); + // Positional rows retain every column, including unnamed SQL projections + // whose duplicate names would overwrite one another in object results. + rows = await db.$client.unsafe(query.sql, query.params as never[]).values(); + if (sample > 1) samples.push(performance.now() - start); + } + const [explain] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]); + samples.sort((a, b) => a - b); + return { medianMs: samples[2], rows, explain: explain["QUERY PLAN"] }; +} + +try { + // Load only the old query module against current dependencies. Keep it in a + // temporary server folder so bare package imports resolve within the worktree. + const noticeModulePath = "server/src/services/chat-channels/inbound/wakeup-notices.ts"; + const noticeModuleUrl = new URL(`../${noticeModulePath}`, import.meta.url); + const oldModule = execFileSync("git", ["show", `${baselineRevision}:${noticeModulePath}`], { encoding: "utf8" }); + const rewritten = oldModule.replace(/from "(\.{1,2}\/[^\"]+)"/g, + (_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, noticeModuleUrl)))}`); + baselineModuleDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".slow-query-baseline-")); + const baselineModuleFile = path.join(baselineModuleDir, "wakeup-notices.ts"); + await fs.writeFile(baselineModuleFile, rewritten); + const { createInboundWakeNotices: baselineNotices } = await import(pathToFileURL(baselineModuleFile).href) as typeof import("../server/src/services/chat-channels/inbound/wakeup-notices.js"); + console.error("[benchmark] baseline module ready"); + const companyId = randomUUID(); + const agentId = randomUUID(); + const sourceId = randomUUID(); + const endpointId = randomUUID(); + const conversationId = randomUUID(); + const applicationId = randomUUID(); + const connectionId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Query benchmark", issuePrefix: "BENCH" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", role: "engineer", adapterType: "codex_local" }); + await db.insert(issues).values({ id: sourceId, companyId, title: "Source", identifier: "BENCH-1", assigneeAgentId: agentId }); + await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Fixture", type: "native" }); + await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Fixture", uid: connectionId, transport: "chat_sdk", connectionPurpose: "channel" }); + await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, provider: "slack", publicId: endpointId, assignedAgentId: agentId, status: "active" }); + await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId: sourceId, externalConversationId: "fixture", externalThreadId: "fixture", externalLabel: "Fixture", state: "waiting" }); + + // Similar row counts to the investigation. Every context has 192 KiB of + // incompressible synthetic prompt text, so JSON filtering must visit TOAST. + const prompt = randomBytes(144 * 1024).toString("base64"); + await db.$client.unsafe(`INSERT INTO heartbeat_runs (company_id, agent_id, status, native_issue_id, created_at, context_snapshot) + SELECT $1::uuid, $2::uuid, 'succeeded', CASE WHEN n = 3 THEN $3::uuid END, now() + n * interval '1 millisecond', + jsonb_build_object('prompt', $4::text) || CASE + WHEN n = 1 THEN jsonb_build_object('issueId', $3::text, 'paperclipSecretRedactions', jsonb_build_array(jsonb_build_object('fixture', 'top-level'))) + WHEN n = 2 THEN jsonb_build_object('paperclipIssue', jsonb_build_object('id', $3::text), 'paperclipSecretRedactions', jsonb_build_array(jsonb_build_object('fixture', 'nested'))) + WHEN n = 3 THEN jsonb_build_object('issueId', gen_random_uuid()::text) + WHEN n = 4 THEN jsonb_build_object('taskId', $3::text) + WHEN n = 5 THEN jsonb_build_object('taskKey', 'BENCH-1') + ELSE jsonb_build_object('issueId', gen_random_uuid()::text) END + FROM generate_series(1, 1400) n`, [companyId, agentId, sourceId, prompt]); + const originatingRuns = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).limit(5); + await db.insert(issues).values(Array.from({ length: 334 }, (_, n) => ({ + companyId, title: `Candidate ${n}`, createdByUserId: "fixture", + originRunId: originatingRuns[n % originatingRuns.length].id, + }))); + const wakes = Array.from({ length: 13000 }, (_, n) => ({ + id: randomUUID(), companyId, agentId, source: "chat", status: n < 50 ? "deferred_issue_execution" : "succeeded", + })); + for (let offset = 0; offset < wakes.length; offset += 500) await db.insert(agentWakeupRequests).values(wakes.slice(offset, offset + 500)); + await db.insert(chatActions).values(wakes.slice(0, 377).map((wake, n) => ({ + id: wake.id, companyId, endpointId, conversationId, kind: "inbound_wakeup", providerActionId: wake.id, + status: "processed", payload: n % 3 === 0 ? { coalesced: true } : {}, + }))); + // Cover coalesced ownership as well as ordinary receipts. + for (let n = 1; n < 50; n += 2) await db.update(agentWakeupRequests) + .set({ payload: { coalescedIntoWakeupRequestId: wakes[n - 1].id } }).where(eq(agentWakeupRequests.id, wakes[n].id)); + for (const [n, ownerId] of [[51, "not-a-uuid"], [52, wakes[0].id.toUpperCase()], [53, ""], [54, null]] as const) { + await db.update(agentWakeupRequests).set({ payload: { coalescedIntoWakeupRequestId: ownerId } }).where(eq(agentWakeupRequests.id, wakes[n].id)); + } + const noticeQuery = (factory: typeof createInboundWakeNotices) => captureQuery(async (captureDb) => factory(captureDb, { + isShuttingDown: () => false, enqueueFailedChatRetryPublications: async () => 0, + } as never).enqueueInboundWakeupPublications(200)); + const notice = await noticeQuery(baselineNotices); + const fixedNotice = await noticeQuery(createInboundWakeNotices); + const list = await captureQuery(async (captureDb) => issueService(captureDb).list(companyId, { + createdFromIssueId: sourceId, sortField: "id", sortDir: "asc", limit: 200, + readCondition: await issueReadSqlCondition(db, { type: "board", userId: "fixture", source: "cloud_tenant" }), + })); + const redactions = db.select({ contextSnapshot: sql`jsonb_build_object('paperclipSecretRedactions', ${heartbeatRuns.contextSnapshot}->'paperclipSecretRedactions')` }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), or( + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${sourceId}`, + sql`${heartbeatRuns.contextSnapshot}->'paperclipIssue'->>'id' = ${sourceId}`, + ))).toSQL(); + + // The summary history query projects these eight context fields. Keep its + // projection as a control: these indexes do not remove per-row JSON work. + const runHistory = db.select({ id: heartbeatRuns.id, + contextIssueId: sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`, + ...Object.fromEntries(["taskId", "taskKey", "commentId", "wakeCommentId", "wakeReason", "wakeSource", "wakeTriggerDetail"] + .map((key) => [key, sql`${heartbeatRuns.contextSnapshot}->>${key}`])), + }).from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)).orderBy(sql`${heartbeatRuns.createdAt} desc`).limit(200).toSQL(); + const queries = { chatNotice: notice, createdFromIssueList: list, issueRedactions: redactions, runHistory }; + // Remove only these candidate indexes from this disposable fixture. This + // also permits repeatable comparisons after the migration is installed. + for (const definition of indexes) await db.$client.unsafe(`DROP INDEX IF EXISTS ${definition.match(/"([^"]+)"/)![0]}`); + await db.$client.unsafe("ANALYZE"); + const baseline: Record>> = {}; + for (const [name, query] of Object.entries(queries)) baseline[name] = await measure(query); + for (const definition of indexes) await db.$client.unsafe(definition); + await db.$client.unsafe("ANALYZE"); + const results = []; + for (const [name, query] of Object.entries(queries)) { + const treatment = await measure(name === "chatNotice" ? fixedNotice : query); + // Unordered SQL returns a set, not a guaranteed order. Compare full rows. + const normalize = (rows: unknown[]) => rows.map((row) => JSON.stringify(row)).sort(); + assert.deepEqual(normalize(treatment.rows), normalize(baseline[name].rows)); + results.push({ name, baselineMs: baseline[name].medianMs, treatmentMs: treatment.medianMs, + speedup: baseline[name].medianMs / treatment.medianMs, equalRows: treatment.rows.length, + baselinePlan: baseline[name].explain, treatmentPlan: treatment.explain }); + } + console.log(JSON.stringify({ baselineRevision, postgres: (await db.$client`SELECT version()`)[0].version, + fixture: { issues: 335, runs: 1400, wakes: 13000, actions: 377, promptBytes: Buffer.byteLength(prompt) }, results }, null, 2)); +} finally { + try { + await db.$client.end({ timeout: 0 }); + await database.cleanup(); + if (baselineModuleDir) await fs.rm(baselineModuleDir, { recursive: true, force: true }); + } finally { + clearInterval(keepAlive); + } +} diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index e08849a8676..901ef5c54ac 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -69275,6 +69275,72 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); + it.each(["not-a-uuid", "", "uppercase"])("skips a pre-run FIFO candidate with a noncanonical coalesced owner id (%s)", async (ownerId) => { + const context = await deferredChatQueueFixture(); + try { + const [receipt] = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, context.action.id)); + await db + .update(agentWakeupRequests) + .set({ + payload: { + ...receipt.payload, + coalescedIntoWakeupRequestId: ownerId === "uppercase" ? context.action.id.toUpperCase() : ownerId, + }, + }) + .where(eq(agentWakeupRequests.id, context.action.id)); + await expect( + context.service.enqueueInboundWakeupPublications(), + ).resolves.toBe(0); + await expect( + db + .select() + .from(chatPublications) + .where(eq(chatPublications.endpointId, context.endpoint.id)), + ).resolves.toEqual([]); + } finally { + await context.service.shutdown(); + } + }); + + it("keeps the pre-run FIFO sweep running when a valid owner's action has comment ids that are not uuids", async () => { + const context = await deferredChatQueueFixture(); + try { + // Publish the queued notice first. The removed-comment and queued-notice + // checks then have real rows to compare the malformed ids against. + await context.service.processPendingPublications(); + await expect( + db + .select({ state: chatPublications.state }) + .from(chatPublications) + .where(eq(chatPublications.endpointId, context.endpoint.id)), + ).resolves.toEqual([{ state: "published" }]); + await db + .update(chatActions) + .set({ + payload: { + ...context.action.payload, + issueId: "not-a-uuid", + commentId: "not-a-uuid", + }, + }) + .where(eq(chatActions.id, context.action.id)); + await expect( + context.service.enqueueInboundWakeupPublications(), + ).resolves.toBe(0); + await expect( + db + .select({ state: chatPublications.state }) + .from(chatPublications) + .where(eq(chatPublications.endpointId, context.endpoint.id)), + ).resolves.toEqual([{ state: "published" }]); + } finally { + await context.service.shutdown(); + } + }); + it("never promises a queued follow-up before durable admission succeeds", async () => { const context = await deferredChatQueueFixture(true); try { diff --git a/server/src/services/chat-channels/inbound/wakeup-notices.ts b/server/src/services/chat-channels/inbound/wakeup-notices.ts index b0133766d7c..3ded0ab4d71 100644 --- a/server/src/services/chat-channels/inbound/wakeup-notices.ts +++ b/server/src/services/chat-channels/inbound/wakeup-notices.ts @@ -1,6 +1,6 @@ import type { DbOrTransaction } from "../runtime-types.js"; import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; -import { and, asc, eq, inArray, isNull, or, sql } from "drizzle-orm"; +import { and, asc, eq, inArray, isNull, or, sql, type SQL } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; import { agentWakeupRequests, chatActions, chatDeliveries, chatMessageLinks, chatPublications, issueComments } from "@paperclipai/db"; @@ -135,10 +135,17 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { const retryInserted = await enqueueFailedChatRetryPublications(limit); const owner = alias(agentWakeupRequests, "chat_notice_owner"); const cursor = inboundWakeNoticeCursor; + // Keep UUID columns indexable. Match only canonical lowercase UUID text, + // preserving the old comparison while malformed payloads match nothing. + const payloadUuid = (value: SQL) => + sql`(case when ${value} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then (${value})::uuid end)`; + const payloadIssueId = payloadUuid(sql`${chatActions.payload}->>'issueId'`); + const payloadCommentId = payloadUuid(sql`${chatActions.payload}->>'commentId'`); + const coalescedOwnerId = sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId'`; const sourceRemoved = sql`exists (select 1 from issue_comments removed_comment where removed_comment.company_id = ${chatActions.companyId} - and removed_comment.issue_id::text = ${chatActions.payload}->>'issueId' - and removed_comment.id::text = ${chatActions.payload}->>'commentId' + and removed_comment.issue_id = ${payloadIssueId} + and removed_comment.id = ${payloadCommentId} and removed_comment.deleted_at is not null)`; const hasVisibleQueue = sql`exists (select 1 from chat_publications queued_notice join chat_message_links visible_queue on visible_queue.publication_id = queued_notice.id @@ -148,7 +155,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { and visible_queue.provider_message_id = queued_notice.provider_message_id and visible_queue.direction = 'outbound' and visible_queue.comment_id is null where queued_notice.company_id = ${chatActions.companyId} - and queued_notice.comment_id::text = ${chatActions.payload}->>'commentId' + and queued_notice.comment_id = ${payloadCommentId} and queued_notice.state = 'published' and queued_notice.idempotency_key = 'wake:' || ${owner.id}::text || ':queued:' || ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`; @@ -163,7 +170,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { ) .innerJoin( owner, - sql`${owner.id}::text = coalesce(${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId', ${agentWakeupRequests.id}::text)`, + sql`${owner.id} = case when ${coalescedOwnerId} is null then ${agentWakeupRequests.id} else ${payloadUuid(coalescedOwnerId)} end`, ) .where( and(