diff --git a/doc/idle-join-request-retention.md b/doc/idle-join-request-retention.md new file mode 100644 index 00000000000..04eff96f057 --- /dev/null +++ b/doc/idle-join-request-retention.md @@ -0,0 +1,99 @@ +# Join request retention while idle + +Pending join requests already have durable activity timestamps. Retention now +uses their next 90-day archive deadline instead of including them in the main +maintenance tick's attention-feed scan. The regular attention UI still includes +join requests. This changes automatic retention discovery, not access approval. + +## Owner and eligibility + +`join-request-retention.ts` owns the candidate query, bounded scan, and deadline. +A candidate is a pending_approval request belonging to an active company, with +an invite whose company matches the request. Kept and already archived items do +not schedule work. Invite expiration or revocation does not remove the existing +attention item and therefore does not disable its retention. No provider policy +is interpreted by the worker or shared scheduler. + +The next deadline is the earliest eligible updatedAt plus 90 days. The partial +pending index orders updated_at and id. Resolved request history is excluded by +the index predicate. Each pass attempts at most 100 due rows. The private cursor +preserves database timestamp precision so a failed page cannot starve later work. +Errors keep the shared 30-second retry; successful full pages continue until +work is exhausted. No eligible rows means no scheduled SQL or network scan. + +Each candidate is re-read under its request row lock before retention changes. +This preserves a concurrent request update or status transition. The shared +retention owner still controls keep, archive/revive, version checks, activity +logging, and notification creation. Archiving does not approve or reject a join. +Join requests have no origin agent, so their archival creates no agent delivery. +Company and invite joins preserve existing visibility policy; this does not add +cross-table serializability beyond the existing retention contract. + +## Producer audit + +| Producer | Notification boundary | +| --- | --- | +| Invite acceptance creates a pending request | Existing acceptance transaction, before the request insert | +| Supported gateway invite replay changes activity | Small writeJoinRequestChange transaction around the existing SQL update | +| Automatic human approval after acceptance | Same helper around the status update | +| Explicit approval and rejection | Same helper around the status update | +| Slack self-service access request | Existing platform transaction, before the request insert | +| Company pause, resume, archive, or deletion | Company-level retention notification inside the existing transaction | +| Budget-driven company pause or resume | Same company notification in the accounting transaction | +| Keep/unkeep, manual archive/revive, accepted archive proposal | Existing typed retention notification now routes join_request to its worker | + +Provider calls, membership grants, and unrelated route behavior stay outside the +small SQL mutation helper. Nested callers retain the existing outer transaction +settlement and uncertain-outcome protocol. The access router itself requires a +root database connection; it is not wrapped in a caller transaction. + +API-key consumption only permits an already-approved request. It cannot create +pending retention work and does not acquire a new wrapper or signal. Invite +acceptance that reuses a human request does not update that request. Revocation, +expiry, and acceptedAt changes do not change the feed predicate. Invite/company +identity is immutable in supported routes; company deletion is the invite delete +owner. A future producer that changes these facts must join the same contract. +Process-local signals do not cover independent SQL writers or another instance. + +## Lifecycle and limits + +Startup reconstructs pending deadlines. The worker is registered only when the +existing heartbeat maintenance switch is enabled. Writes during drain retain a +wake without passive SQL. Failed work retains its retry across drain. Shutdown +joins an admitted archive and stops before subsequent candidates; restart picks +up the durable remainder. There is no new queue table or permanent DB listener. +The in-process deadline is not a Cloud alarm. Existing unsupported-future-work +sleep blockers remain in place. + +## Verification + +Seventeen new real-PostgreSQL cases pass. They cover actual invite acceptance, +rejection, human approval, gateway replay, and Slack access requests; empty and +resolved-history quiet; saved deadlines; startup reconstruction; keep and manual +archival controls; company pause/resume; unchanged expired/revoked invite policy; +normal versus maintenance feeds; nested commit/rollback; first writes and failed +work during drain; concurrent source updates; 101-row microsecond paging with a +full failed page; and shutdown/restart. + +Another 139 consumer cases pass: 71 invitation/startup cases, 21 approval worker +cases, and 47 shared retention/attention/budget cases. This is 156 selected tests, +not a full local suite claim. The first consumer command used a nonexistent +retention test filename and ran only the approval suite; the corrected selection +ran the actual retention suite against its dedicated external database. + +Initial test failures exposed fixture assumptions: the access router cannot run +on a transaction handle; unresolved recovery cannot be expected to become idle; +Slack links require actual endpoint/principal ownership rows; and shutdown tests +must establish admission and deterministic candidate order. Those fixtures were +corrected without weakening assertions or production checks. The first replay +fixture lacked required gateway credentials. The final fixture supplies test-only +credentials. A reused transaction spy retained admission calls during the quiet +assertion; restoring it before the quiet measurement fixed the measurement. + +Migration 0387 applies, the full migration set replays, and the concurrent index +statement replays idempotently. Migration safety and numbering checks pass; the +five newest schema snapshots are retained. The actual next-deadline query with +40,000 resolved requests and one pending request uses the pending index and +executes in 0.017 ms locally. The probe rolls back its data. This is query-plan +evidence, not production latency or measured savings. Workspace typechecking and the production build pass. Fresh PR CI/review remain +required before readiness. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index e1b421845f3..10b50209fb6 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -866,3 +866,25 @@ preserves cursor precision in the owning query. All 87 workspace service/worker cases and server typechecking pass. See [workspace pagination](../idle-terminal-workspace-pagination.md). Publication and fresh CI/review are next. Join retention remains the next new migration duty. + +## Join request retention implementation + +Terminal workspace pagination is [#17505](https://github.com/paperclipai/paperclip/pull/17505), +head `28a197cdfd`, stacked on #17497. It has fresh 5/5 review without unresolved +findings. The full CI matrix is running. Workspace typecheck/build and all 87 +selected service/worker tests passed locally. + +Join retention is implemented on `codex/sleep-15-join-retention`, based on #17505. +The worker preserves pending request/invite/company eligibility, manual controls, +transaction ownership, microsecond paging, and bounded recovery. All supported +writers are audited; approved-only API-key consumption is deliberately exempt. +See [join retention](../idle-join-request-retention.md). Seventeen new real-DB +lifecycle cases and 139 consumers pass (156 total). Initial fixture and command +failures are recorded in that document. Index generation, apply, replay, safety, +numbering, and an actual-query plan with 40,000 resolved requests pass. Workspace +typecheck/build and remote publication/CI/review remain in progress. No slice is +claimed ready based only on local tests, and no merge is authorized. + +After join retention, migrate the remaining attention source owners separately: +decisions, interactions, recovery actions, reviews, and other source kinds. Keep +the full original server inventory and later aggregate-tick retirement in scope. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 2afeb7d9f78..6a0102a04b0 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -512,6 +512,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0384_many_reaper.sql", "0385_normal_blindfold.sql", "0386_omniscient_luke_cage.sql", + "0387_huge_skrulls.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/migrations/0387_huge_skrulls.sql b/packages/db/src/migrations/0387_huge_skrulls.sql new file mode 100644 index 00000000000..dd3fa35bc3c --- /dev/null +++ b/packages/db/src/migrations/0387_huge_skrulls.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "join_requests_pending_retention_idx" ON "join_requests" USING btree ("updated_at","id") WHERE "join_requests"."status" = 'pending_approval'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0382_snapshot.json b/packages/db/src/migrations/meta/0387_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0382_snapshot.json rename to packages/db/src/migrations/meta/0387_snapshot.json index 7f91405ae3e..7a8769624cb 100644 --- a/packages/db/src/migrations/meta/0382_snapshot.json +++ b/packages/db/src/migrations/meta/0387_snapshot.json @@ -1,6 +1,6 @@ { - "id": "5e8d042e-d697-4753-9582-7d0c042ef893", - "prevId": "afdc5e70-19b3-416d-917a-beb9fdc1e2e1", + "id": "4c86af49-02a0-496f-b028-b468d9e4e960", + "prevId": "02c5ad8a-cdb9-41ad-acf1-5ab4f7d5bd67", "version": "7", "dialect": "postgresql", "tables": { @@ -5721,6 +5721,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": [ @@ -24030,6 +24052,50 @@ "method": "btree", "with": {} }, + "decisions_running_recovery_idx": { + "name": "decisions_running_recovery_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" = 'running'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "decisions_open_expiry_idx": { + "name": "decisions_open_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"status\" = 'open'", + "concurrently": false, + "method": "btree", + "with": {} + }, "decisions_pending_continuation_idx": { "name": "decisions_pending_continuation_idx", "columns": [ @@ -32035,6 +32101,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": [ @@ -41829,6 +41917,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": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 8b1b2657d6f..6979bbe3e1d 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2696,6 +2696,13 @@ "when": 1792877330859, "tag": "0386_omniscient_luke_cage", "breakpoints": true + }, + { + "idx": 387, + "version": "7", + "when": 1792880971454, + "tag": "0387_huge_skrulls", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/join_requests.ts b/packages/db/src/schema/join_requests.ts index 3813c27db78..43a58726d45 100644 --- a/packages/db/src/schema/join_requests.ts +++ b/packages/db/src/schema/join_requests.ts @@ -31,6 +31,8 @@ export const joinRequests = pgTable( updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, (table) => ({ + pendingRetentionIdx: index("join_requests_pending_retention_idx").on(table.updatedAt, table.id) + .where(sql`${table.status} = 'pending_approval'`), inviteUniqueIdx: uniqueIndex("join_requests_invite_unique_idx").on(table.inviteId), companyStatusTypeCreatedIdx: index("join_requests_company_status_type_created_idx").on( table.companyId, diff --git a/server/src/__tests__/join-request-retention-work.test.ts b/server/src/__tests__/join-request-retention-work.test.ts new file mode 100644 index 00000000000..e4c9018b4c2 --- /dev/null +++ b/server/src/__tests__/join-request-retention-work.test.ts @@ -0,0 +1,308 @@ +import { createSlackIdentityLinks } from "../services/chat-channels/slack/identity-links.js"; +import { createHash, randomUUID } from "node:crypto"; +import express from "express"; +import request from "supertest"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, authUsers, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, companies, companyMemberships, companySecretProviderConfigs, companySecrets, companySecretVersions, createDb, decisionArchiveNotificationOutbox, + decisionRetention, invites, joinRequests, principalPermissionGrants, toolApplications, toolConnections, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "../services/join-request-retention.js"; +import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js"; +import { attentionService } from "../services/attention.js"; +import { companyService } from "../services/companies.js"; +import { writeJoinRequestChange } from "../services/retention-work-notifications.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_JOIN_RETENTION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; +(support.supported ? describe : describe.skip)("join request retention worker", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string, userId: string; + let coordinator: ReturnType | undefined; + let draining: boolean; + let accessRoutes: typeof import("../routes/access.js").accessRoutes; + const errors: unknown[] = []; + beforeAll(async () => { + ({ accessRoutes } = await import("../routes/access.js")); + if (!externalUrl) temporary = await startEmbeddedPostgresTestDatabase("join-retention-"); + db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 30_000); + beforeEach(async () => { + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "0123456789abcdef0123456789abcdef"); + companyId = randomUUID(); userId = randomUUID(); draining = false; errors.length = 0; + await db.insert(companies).values({ id: companyId, name: "Join retention", issuePrefix: `J${companyId.slice(0, 6)}` }); + await db.insert(authUsers).values({ id: userId, name: "Owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner", status: "active" }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); vi.unstubAllEnvs(); + await db.execute(sql`drop trigger if exists join_retention_fail on decision_retention`); + await db.execute(sql`drop function if exists join_retention_fail()`); + await db.delete(decisionArchiveNotificationOutbox); await db.delete(decisionRetention); await db.delete(activityLog); + await db.delete(chatIdentityLinks); await db.delete(chatExternalPrincipals); await db.delete(chatEndpoints); + await db.delete(toolConnections); await db.delete(toolApplications); await db.delete(agents); + await db.delete(joinRequests); await db.delete(invites); await db.delete(principalPermissionGrants); + await db.delete(companySecretVersions); await db.delete(companySecrets); await db.delete(companySecretProviderConfigs); + await db.delete(companyMemberships); await db.delete(authUsers); await db.delete(companies); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + const old = () => new Date(Date.now() - delay - 60_000); + const state = (id: string) => decisionRetentionService(db).getState(companyId, "join_request", id); + const authActor = () => ({ type: "board" as const, source: "local_implicit" as const, userId, companyIds: [companyId], isInstanceAdmin: true, + memberships: [{ companyId, membershipRole: "owner", status: "active" }] }); + const mutation = (id: string) => ({ companyId, sourceKind: "join_request" as const, sourceId: id, authActor: authActor(), + actor: { actorType: "user" as const, actorId: userId, agentId: null, userId, runId: null, agentApiKeyId: null, responsibleUserId: userId } }); + async function invite(database: Db = db) { + const token = randomUUID(); + const [row] = await database.insert(invites).values({ companyId, tokenHash: createHash("sha256").update(token).digest("hex"), + allowedJoinTypes: "both", expiresAt: new Date(Date.now() + 86_400_000), invitedByUserId: userId }).returning(); + return { ...row, token }; + } + async function create(at = new Date(), database: Db = db) { + const invitation = await invite(database); + const [row] = await writeJoinRequestChange(database, tx => tx.insert(joinRequests).values({ companyId, inviteId: invitation.id, + requestType: "agent", requestIp: "127.0.0.1", agentName: "Pending agent", createdAt: at, updatedAt: at }).returning()); + return row; + } + function app(database: Db = db) { + const server = express(); server.use(express.json()); + server.use((req, _res, next) => { req.actor = authActor(); next(); }); + server.use("/api", accessRoutes(database, { deploymentMode: "authenticated", deploymentExposure: "private", + bindHost: "127.0.0.1", allowedHostnames: [] })); + server.use((error: { status?: number; message?: string }, _req: express.Request, res: express.Response, _next: express.NextFunction) => { + res.status(error.status ?? 500).json({ error: error.message }); + }); + return server; + } + async function accept(token: string, database: Db = db) { + const response = await request(app(database)).post(`/api/invites/${token}/accept`) + .send({ requestType: "agent", agentName: "New agent", adapterType: "claude_local" }); + expect(response.status, JSON.stringify(response.body)).toBe(202); + const [row] = await database.select().from(joinRequests).where(eq(joinRequests.companyId, companyId)); + return row; + } + async function start(service = joinRequestRetentionService(db)) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); + await registerJoinRequestRetentionWork(coordinator, service, () => !draining).ready; + return service; + } + async function settle() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 }); + } + async function quiet() { + await settle(); expect(coordinator!.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(300_000); + expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); executes.mockRestore(); writes.mockRestore(); + } + async function archived(id: string) { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect((await state(id))?.archivedAt).toBeInstanceOf(Date); }, { timeout: 5000 }); + } + + it("wakes from real invite acceptance, waits for the saved deadline, and disarms on rejection", async () => { + await start(); await quiet(); + const invitation = await invite(); const row = await accept(invitation.token); + await settle(); expect(coordinator!.nextWakeAt()).toBe(row.updatedAt.getTime() + delay); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(300_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + const response = await request(app()).post(`/api/companies/${companyId}/join-requests/${row.id}/reject`).send({}); + expect(response.status, JSON.stringify(response.body)).toBe(200); await quiet(); + expect(await state(row.id)).toBeNull(); + }); + + it("archives unseen pending requests at the actual deadline without changing approval status", async () => { + await start(); await quiet(); const deadline = Date.now() + 600_000; + const row = await create(new Date(deadline - delay)); await settle(); + expect(coordinator!.nextWakeAt()).toBe(deadline); expect(await state(row.id)).toBeNull(); + const window = deadline - Date.now() - 30_000; expect(window).toBeGreaterThan(0); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(window); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await vi.advanceTimersByTimeAsync(60_000); await archived(row.id); await quiet(); + expect((await db.select().from(joinRequests))[0].status).toBe("pending_approval"); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + }); + + it("reconstructs pending requests at startup and ignores resolved history", async () => { + const due = await create(old()), resolved = await create(old()); + await db.update(joinRequests).set({ status: "approved" }).where(eq(joinRequests.id, resolved.id)); + await start(); await archived(due.id); await quiet(); expect(await state(resolved.id)).toBeNull(); + }); + + it("keeps revoked and expired invite requests visible under the existing retention policy", async () => { + const row = await create(old()); await db.update(invites).set({ expiresAt: old(), revokedAt: old() }); + const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); + expect(feed.items.some(item => item.subject.id === row.id)).toBe(true); + await start(); await archived(row.id); await quiet(); + }); + + it("waits for company activation and disarms when the company pauses", async () => { + const row = await create(old()); await companyService(db).update(companyId, { status: "paused" }); + await start(); await quiet(); expect(await state(row.id)).toBeNull(); + await companyService(db).update(companyId, { status: "active" }); await archived(row.id); await quiet(); + await create(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await companyService(db).update(companyId, { status: "paused" }); await quiet(); + }); + + it("honors keep and schedules an old request again when unkept", async () => { + const row = await create(old()); await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); + await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: true }); + await start(); await quiet(); + await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: false }); await archived(row.id); await quiet(); + }); + + it("honors manual archival and reactivates on revive", async () => { + const row = await create(old()); await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); + await decisionRetentionService(db).archive(mutation(row.id)); await start(); await quiet(); + await decisionRetentionService(db).revive(mutation(row.id)); await archived(row.id); await quiet(); + }); + + it("preserves normal attention reads but omits join requests from maintenance", async () => { + const row = await create(old()); + const excluded = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, excludeSources: ["join_request"] }); + expect(excluded.items.some(item => item.sourceKind === "join_request")).toBe(false); expect(await state(row.id)).toBeNull(); + const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); + expect(feed.items.some(item => item.subject.id === row.id)).toBe(true); + }); + + it("defers nested request writes until outer commit and releases rolled-back writes", async () => { + await start(); await quiet(); + await expect(db.transaction(async tx => { + await create(new Date(), tx as unknown as Db); + await vi.advanceTimersByTimeAsync(120_000); expect(coordinator!.nextWakeAt()).toBeNull(); throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(); expect(await db.select().from(joinRequests)).toHaveLength(0); + let row: typeof joinRequests.$inferSelect; + await db.transaction(async tx => { + row = await create(new Date(), tx as unknown as Db); + await vi.advanceTimersByTimeAsync(120_000); expect(coordinator!.nextWakeAt()).toBeNull(); + }); + await settle(); expect(coordinator!.nextWakeAt()).toBe(row!.updatedAt.getTime() + delay); + }); + + it("retains a first write during drain without passive SQL", async () => { + await start(); await quiet(); draining = true; const row = await create(old()); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + draining = false; await archived(row.id); await quiet(); + }); + + it("retains a failed archival through drain until the database recovers", async () => { + const row = await create(old()); + await db.execute(sql`create function join_retention_fail() returns trigger language plpgsql as $$ begin + raise exception 'retention unavailable'; end $$`); + await db.execute(sql`create trigger join_retention_fail before insert or update on decision_retention for each row execute function join_retention_fail()`); + const service = joinRequestRetentionService(db); const sweep = vi.spyOn(service, "sweep"); + await start(service); await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); draining = true; + await Promise.allSettled(sweep.mock.results.map(result => result.value)); + await vi.advanceTimersByTimeAsync(30_000); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await db.execute(sql`drop trigger join_retention_fail on decision_retention`); draining = false; + await archived(row.id); await quiet(); + for (const error of errors) expect(error).toMatchObject({ cause: { message: "retention unavailable" } }); errors.length = 0; + }); + + it("wakes from the Slack access-request producer and deduplicates repeated requests", async () => { + await start(); await quiet(); const token = randomUUID(), endpointId = randomUUID(); + const [agent] = await db.insert(agents).values({ companyId, name: "Chat agent", role: "engineer", status: "active", adapterType: "codex_local" }).returning(); + const [application] = await db.insert(toolApplications).values({ companyId, name: "Slack", type: "chat" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: application.id, name: "Slack", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning(); + await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId: connection.id, provider: "slack", publicId: randomUUID(), assignedAgentId: agent.id, status: "active" }); + const [principal] = await db.insert(chatExternalPrincipals).values({ companyId, provider: "slack", providerAccountId: "team", externalId: randomUUID() }).returning(); + await db.insert(chatIdentityLinks).values({ companyId, endpointId, principalId: principal.id, + confirmationTokenHash: createHash("sha256").update(token).digest("hex"), expiresAt: new Date(Date.now() + 600_000) }); + const links = createSlackIdentityLinks(db, { + previewIdentityLink: async () => ({ selfService: true, canConfirm: false, companyId, endpointId }), + stageProviderEffect: vi.fn(async () => null), + }); + expect(await links.requestIdentityAccess(token, userId, "127.0.0.1")).toEqual({ status: "pending_approval" }); + await settle(); const [row] = await db.select().from(joinRequests); + expect(coordinator!.nextWakeAt()).toBe(row.updatedAt.getTime() + delay); + await links.requestIdentityAccess(token, userId, "127.0.0.1"); await settle(); + expect(await db.select().from(joinRequests)).toHaveLength(1); + }); + + it("disarms after the real approval route resolves a human request", async () => { + const requester = randomUUID(); + await db.insert(authUsers).values({ id: requester, name: "Joiner", email: `${requester}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + const row = await create(); + await db.update(joinRequests).set({ requestType: "human", requestingUserId: requester }).where(eq(joinRequests.id, row.id)); + await start(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + const response = await request(app()).post(`/api/companies/${companyId}/join-requests/${row.id}/approve`).send({}); + expect(response.status, JSON.stringify(response.body)).toBe(200); await quiet(); + expect((await db.select().from(joinRequests))[0].status).toBe("approved"); + }); + + it("rechecks activity after waiting for a concurrent request writer", async () => { + const row = await create(old()); let release!: () => void, locked!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const admitted = new Promise(resolve => { locked = resolve; }); + const write = db.transaction(async tx => { + await tx.execute(sql`select id from join_requests where id = ${row.id} for update`); locked(); await gate; + await writeJoinRequestChange(tx as unknown as Db, owner => owner.update(joinRequests).set({ updatedAt: new Date() }).where(eq(joinRequests.id, row.id))); + }); + await admitted; const sweep = joinRequestRetentionService(db).sweep(); + await vi.advanceTimersByTimeAsync(100); release(); await write; await sweep; + expect(await state(row.id)).toBeNull(); + }); + + it("pages past failed requests with database timestamp precision and retries them later", async () => { + const ids: string[] = []; + for (let i = 0; i < 101; i++) ids.push((await create(old())).id); + ids.sort(); const last = ids.at(-1)!; + await db.execute(sql`update join_requests set updated_at = date_trunc('milliseconds', now() - interval '91 days') + interval '456 microseconds'`); + await db.execute(sql`create function join_retention_fail() returns trigger language plpgsql as $$ begin + if new.archived_at is not null and new.source_id <> TG_ARGV[0] then raise exception 'retention unavailable'; end if; + return new; end $$`); + await db.execute(sql`create trigger join_retention_fail before update on decision_retention for each row execute function join_retention_fail(${sql.raw("'" + last + "'")})`); + const service = joinRequestRetentionService(db); + await expect(service.sweep()).rejects.toThrow("Join request retention failed"); + expect(await service.sweep()).toEqual({ archived: 1 }); expect((await state(last))?.archivedAt).toBeInstanceOf(Date); + await db.execute(sql`drop trigger join_retention_fail on decision_retention`); + expect(await service.sweep()).toEqual({ archived: 100 }); expect(await service.sweep()).toEqual({ archived: 0 }); + expect(await service.next()).toBeNull(); + }); + + it("joins an admitted archive on shutdown and leaves later candidates for restart", async () => { + const first = await create(new Date(old().getTime() - 1000)), second = await create(old()); + let release!: () => void, locked!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const acquired = new Promise(resolve => { locked = resolve; }); + const transaction = db.transaction.bind(db); + const transactionSpy = vi.spyOn(db, "transaction").mockImplementationOnce((write, options) => transaction(async tx => { + locked(); await gate; return write(tx); + }, options)); + const starting = start(); await acquired; + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + try { await vi.advanceTimersByTimeAsync(500); expect(stopped).toBe(false); } finally { release(); } + await starting; await stopping; coordinator = undefined; transactionSpy.mockRestore(); + expect((await state(first.id))?.archivedAt).toBeInstanceOf(Date); expect(await state(second.id)).toBeNull(); + await start(); await archived(second.id); await quiet(); + }); + + + it("moves the saved deadline after a supported gateway invite replay", async () => { + const invitation = await invite(); await start(); await quiet(); + const body = { requestType: "agent", agentName: "Gateway", adapterType: "openclaw_gateway", + agentDefaultsPayload: { url: "ws://127.0.0.1:18789", disableDeviceAuth: true, headers: { "x-openclaw-token": "fixture-gateway-token-1234567890" } } }; + const first = await request(app()).post(`/api/invites/${invitation.token}/accept`).send(body); + expect(first.status, JSON.stringify(first.body)).toBe(202); await settle(); + const initial = coordinator!.nextWakeAt(); + await vi.advanceTimersByTimeAsync(60_000); + const replay = await request(app()).post(`/api/invites/${invitation.token}/accept`).send({ ...body, agentName: "Renamed gateway" }); + expect(replay.status, JSON.stringify(replay.body)).toBe(202); await settle(); + const rows = await db.select().from(joinRequests); expect(rows).toHaveLength(1); + expect(coordinator!.nextWakeAt()).toBe(rows[0].updatedAt.getTime() + delay); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(initial!); + }); + +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 3a58ab926f6..c3fc88e4d12 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1,3 +1,4 @@ +import { joinRequestRetentionService } from "../services/join-request-retention.js"; import { approvalRetentionService } from "../services/approval-retention.js"; import { logger } from "../middleware/logger.js"; import { getStartupRecoveryState } from "../startup-recovery-state.js"; @@ -322,6 +323,11 @@ vi.mock("../realtime/live-events-ws.js", () => ({ setupLiveEventsWebSocketServer: vi.fn(), })); +vi.mock("../services/join-request-retention.js", async importOriginal => ({ + ...await importOriginal(), + joinRequestRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })), +})); + vi.mock("../services/approval-retention.js", async importOriginal => ({ ...await importOriginal(), approvalRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })), @@ -537,6 +543,18 @@ describe("startServer feedback export wiring", () => { } }); + it.each([false, true])("registers join request retention only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.joinRequestRetention); + expect(Boolean(registration)).toBe(enabled); + if (enabled) { + const service = vi.mocked(joinRequestRetentionService).mock.results.at(-1)!.value; + await registration![1].run(new AbortController().signal); + expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled(); + } + }); + it("registers durable case handoffs independently of the heartbeat scheduler", async () => { await startServer(); expect(caseHandoffDeliveryServiceFactoryMock).toHaveBeenCalledWith(createDbMock.mock.results[0]!.value, { heartbeat: heartbeatServiceMock }); diff --git a/server/src/index.ts b/server/src/index.ts index fc27cd291a0..323bf6f84b3 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "./services/join-request-retention.js"; import { approvalRetentionService, registerApprovalRetentionWork } from "./services/approval-retention.js"; import { registerDecisionTargetWork } from "./services/decision-target-work.js"; import { registerDecisionEffectRecoveryWork } from "./services/decision-effect-recovery-work.js"; @@ -1776,7 +1777,7 @@ async function startServerWithDatabaseTeardown( // per company per tick. const page = await attentionService(db as any).list(company.id, { includeDismissed: true, - excludeSources: ["approval"], + excludeSources: ["approval", "join_request"], all: true, allowUnscopedAll: true, }); @@ -1826,6 +1827,8 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("approval_retention", () => registerApprovalRetentionWork(deliveryWork, approvalRetentionService(db as any), () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); + await step("join_request_retention", () => registerJoinRequestRetentionWork(deliveryWork, joinRequestRetentionService(db as any), + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("retention_sweep", runRetentionSweep); }; diff --git a/server/src/routes/access.ts b/server/src/routes/access.ts index 526bd1a3b42..e1fe2566c3f 100644 --- a/server/src/routes/access.ts +++ b/server/src/routes/access.ts @@ -1,3 +1,4 @@ +import { notifyRetentionWork, writeJoinRequestChange } from "../services/retention-work-notifications.js"; import { createAgentLifecycle } from "../services/agent-lifecycle.js"; import { createHash, @@ -3136,7 +3137,7 @@ export function accessRoutes( const approvedAt = new Date(); const approvedByUserId = input.invite.invitedByUserId ?? (isLocalImplicit(input.req) ? "local-board" : null); - const approved = await db + const approved = await writeJoinRequestChange(db, tx => tx .update(joinRequests) .set({ status: "approved", @@ -3146,7 +3147,7 @@ export function accessRoutes( }) .where(eq(joinRequests.id, input.joinRequest.id)) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); await logActivity(db, { companyId: input.companyId, @@ -3882,6 +3883,7 @@ export function accessRoutes( ) ); + await notifyRetentionWork(tx, "join_request"); const row = await tx .insert(joinRequests) .values({ @@ -3912,7 +3914,7 @@ export function accessRoutes( .then((rows) => rows[0]); return row; }) - : await db + : await writeJoinRequestChange(db, tx => tx .update(joinRequests) .set({ requestIp: requestIp(req), @@ -3935,7 +3937,7 @@ export function accessRoutes( }) .where(eq(joinRequests.id, replayJoinRequestId as string)) .returning() - .then((rows) => rows[0]); + .then((rows) => rows[0])); if (!created) { throw conflict("Join request not found"); @@ -4279,7 +4281,7 @@ export function accessRoutes( ); } - const approved = await db + const approved = await writeJoinRequestChange(db, tx => tx .update(joinRequests) .set({ status: "approved", @@ -4291,7 +4293,7 @@ export function accessRoutes( }) .where(eq(joinRequests.id, requestId)) .returning() - .then((rows) => rows[0]); + .then((rows) => rows[0])); await logActivity(db, { companyId, @@ -4338,7 +4340,7 @@ export function accessRoutes( if (existing.status !== "pending_approval") throw conflict("Join request is not pending"); - const rejected = await db + const rejected = await writeJoinRequestChange(db, tx => tx .update(joinRequests) .set({ status: "rejected", @@ -4349,7 +4351,7 @@ export function accessRoutes( }) .where(eq(joinRequests.id, requestId)) .returning() - .then((rows) => rows[0]); + .then((rows) => rows[0])); await logActivity(db, { companyId, diff --git a/server/src/services/attention.ts b/server/src/services/attention.ts index b268ac98dc9..cae1ca417a9 100644 --- a/server/src/services/attention.ts +++ b/server/src/services/attention.ts @@ -1375,6 +1375,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions })); } + if (!options.excludeSources?.includes("join_request")) { const pendingJoins = await db .select({ id: joinRequests.id, @@ -1436,6 +1437,8 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions })); } + } + const recoveryRows = await db .select() .from(issueRecoveryActions) diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 84e4ca1e6dc..aba74d6e0c6 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,4 +1,4 @@ -import { notifyRetentionWork } from "./retention-work-notifications.js"; +import { notifyCompanyRetentionWork, notifyRetentionWork } from "./retention-work-notifications.js"; import { addCents, compareCents, normalizeCents } from "@paperclipai/shared"; import { and, desc, eq, inArray, sql, or, isNull } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -297,7 +297,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; - await notifyRetentionWork(db, "approval"); + await notifyCompanyRetentionWork(db); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); await db @@ -354,7 +354,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; - await notifyRetentionWork(db, "approval"); + await notifyCompanyRetentionWork(db); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await db .update(companies) diff --git a/server/src/services/chat-channels/slack/identity-links.ts b/server/src/services/chat-channels/slack/identity-links.ts index 1a37771fa01..6ce66825fb1 100644 --- a/server/src/services/chat-channels/slack/identity-links.ts +++ b/server/src/services/chat-channels/slack/identity-links.ts @@ -1,3 +1,4 @@ +import { notifyRetentionWork } from "../../retention-work-notifications.js"; import { createHash, randomBytes } from "node:crypto"; import { and, desc, eq, gt, or, sql } from "drizzle-orm"; @@ -97,6 +98,7 @@ export function createSlackIdentityLinks(db: Db, { previewIdentityLink, stagePro inviteType: "company_join", allowedJoinTypes: "human", acceptedAt: now, expiresAt: now, defaultsPayload: { human: { role: "operator" }, source: "slack_identity_link" }, }).returning({ id: invites.id }); + await notifyRetentionWork(tx, "join_request"); const [request] = await tx.insert(joinRequests).values({ inviteId: invite.id, companyId: preview.companyId, requestType: "human", status: "pending_approval", requestIp, requestingUserId: userId, requestEmailSnapshot: user.email, diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index f6fc7ed5de2..4f17b854339 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,4 +1,4 @@ -import { notifyRetentionWork } from "./retention-work-notifications.js"; +import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { isRunnerApiResponseAsset } from "./asset-content-access.js"; import { deleteCompany } from "./company-deletion.js"; @@ -331,7 +331,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); - await notifyRetentionWork(tx, "approval"); + await notifyCompanyRetentionWork(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); } const updated = await tx @@ -483,7 +483,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); - await notifyRetentionWork(tx, "approval"); + await notifyCompanyRetentionWork(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); await tx .update(companies) diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 63361d81f8e..44eed1a462e 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,4 +1,4 @@ -import { notifyRetentionWork } from "./retention-work-notifications.js"; +import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { notifyDecisionTargetWork } from "./decision-target-work.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { agentObservabilityService } from "./agent-observability.js"; @@ -95,7 +95,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(agentRuntimeState).where(eq(agentRuntimeState.companyId, id)); await tx.delete(issueComments).where(eq(issueComments.companyId, id)); await tx.delete(approvalComments).where(eq(approvalComments.companyId, id)); - await notifyRetentionWork(tx, "approval"); + await notifyCompanyRetentionWork(tx); await tx.delete(approvals).where(eq(approvals.companyId, id)); await tx.delete(companySecrets).where(eq(companySecrets.companyId, id)); await tx.delete(joinRequests).where(eq(joinRequests.companyId, id)); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 3e5fa30f09e..7355c5d3a6c 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = { executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", approvalRetention: "approval-retention", + joinRequestRetention: "join-request-retention", decisionTargets: "decision-targets", decisionEffectRecovery: "decision-effect-recovery", decisionExpiry: "decision-expiry", diff --git a/server/src/services/join-request-retention.ts b/server/src/services/join-request-retention.ts new file mode 100644 index 00000000000..e88f4792c12 --- /dev/null +++ b/server/src/services/join-request-retention.ts @@ -0,0 +1,68 @@ +import { and, asc, eq, isNull, lte, or, sql, type SQL } from "drizzle-orm"; +import { companies, decisionRetention, invites, joinRequests, type Db } from "@paperclipai/db"; +import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; + +const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; + +export function joinRequestRetentionService(db: Db) { + let cursor: { at: string; id: string } | undefined; + function candidates(database: Db, condition?: SQL) { + return database.select({ id: joinRequests.id, companyId: joinRequests.companyId, + at: joinRequests.updatedAt, cursorAt: sql`${joinRequests.updatedAt}::text` }).from(joinRequests) + .innerJoin(companies, eq(companies.id, joinRequests.companyId)) + .innerJoin(invites, and(eq(invites.id, joinRequests.inviteId), eq(invites.companyId, joinRequests.companyId))) + .leftJoin(decisionRetention, and(eq(decisionRetention.companyId, joinRequests.companyId), + eq(decisionRetention.sourceKind, "join_request"), sql`${decisionRetention.sourceId} = ${joinRequests.id}::text`)) + .where(and(sql`${joinRequests.status} = 'pending_approval'`, eq(companies.status, "active"), + or(isNull(decisionRetention.id), eq(decisionRetention.keep, false)), isNull(decisionRetention.archivedAt), condition)) + .orderBy(asc(joinRequests.updatedAt), asc(joinRequests.id)); + } + async function sweep(now = new Date(), signal?: AbortSignal) { + const cutoff = new Date(now.getTime() - archiveDelay); + const rows = await candidates(db, and(lte(joinRequests.updatedAt, cutoff), cursor + ? sql`(${joinRequests.updatedAt}, ${joinRequests.id}) > (${cursor.at}::timestamptz, ${cursor.id}::uuid)` : undefined)).limit(100); + let archived = 0; + const errors: unknown[] = []; + for (const candidate of rows) { + if (signal?.aborted) break; + try { + archived += await db.transaction(async tx => { + const database = tx as unknown as Db; + const [current] = await candidates(database, and(eq(joinRequests.id, candidate.id), lte(joinRequests.updatedAt, cutoff))) + .for("update", { of: joinRequests }); + if (!current) return 0; + const retention = decisionRetentionService(database); + const source = { sourceKind: "join_request" as const, subject: { id: current.id }, activityAt: current.at.toISOString() }; + const states = await retention.syncItems(current.companyId, [source]); + const state = states.get(`join_request:${current.id}`)!; + return retention.autoArchive({ companyId: current.companyId, now, items: [{ ...source, + keep: state.keep, archivedAt: state.archivedAt?.toISOString() ?? null, retentionVersion: state.version, + }] }); + }); + } catch (error) { errors.push(error); } + cursor = { at: candidate.cursorAt, id: candidate.id }; + } + if (rows.length < 100) cursor = undefined; + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, "Join request retention failed"); + return { archived }; + } + async function next() { + const [row] = await candidates(db).limit(1); + return row ? row.at.getTime() + archiveDelay : null; + } + return { sweep, next }; +} + +export function registerJoinRequestRetentionWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.joinRequestRetention, { + run: signal => service.sweep(new Date(), signal), next: service.next, canRun, retryMs: 30_000, + }); +} diff --git a/server/src/services/retention-work-notifications.ts b/server/src/services/retention-work-notifications.ts index 3464c624e89..a52ce2de883 100644 --- a/server/src/services/retention-work-notifications.ts +++ b/server/src/services/retention-work-notifications.ts @@ -4,6 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export async function notifyRetentionWork(transaction: object, kind: AttentionSourceKind) { if (kind === "approval") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention); + if (kind === "join_request") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.joinRequestRetention); } export function writeApprovalChange(db: Db, write: (transaction: Db) => Promise): Promise { @@ -12,3 +13,15 @@ export function writeApprovalChange(db: Db, write: (transaction: Db) => Promi return write(tx as unknown as Db); }); } + +export async function notifyCompanyRetentionWork(transaction: object) { + await notifyRetentionWork(transaction, "approval"); + await notifyRetentionWork(transaction, "join_request"); +} + +export function writeJoinRequestChange(db: Db, write: (transaction: Db) => Promise): Promise { + return db.transaction(async tx => { + await notifyRetentionWork(tx, "join_request"); + return write(tx as unknown as Db); + }); +}