From 5ddb4543ed90b1e2b9f6bfdefec5fbdfdc3fb142 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 06:01:32 -0500 Subject: [PATCH] fix(server): schedule Teams consent expiry from saved deadlines Co-Authored-By: Paperclip --- doc/idle-teams-consent-expiry.md | 61 ++++++ ...2026-10-24-recurring-server-sleep-stack.md | 13 ++ ...-10-25-teams-consent-expiry-sleep-audit.md | 53 ++++- .../app-private-hostname-gate.test.ts | 6 + .../chat-channels.integration.test.ts | 34 ++++ server/src/app.ts | 1 + server/src/services/chat-channels.ts | 3 +- .../microsoft-teams/consent-expiry-work.ts | 53 +++++ .../microsoft-teams/file-delivery.ts | 34 ++-- .../chat-teams-file-transfers.test.ts | 185 ++++++++++++++++++ .../src/services/chat-teams-file-transfers.ts | 81 ++++---- .../services/delivery-work-notifications.ts | 1 + 12 files changed, 457 insertions(+), 68 deletions(-) create mode 100644 doc/idle-teams-consent-expiry.md create mode 100644 server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts diff --git a/doc/idle-teams-consent-expiry.md b/doc/idle-teams-consent-expiry.md new file mode 100644 index 00000000000..6220c74ab20 --- /dev/null +++ b/doc/idle-teams-consent-expiry.md @@ -0,0 +1,61 @@ +# Teams consent expiry and idle work + +Unused Teams file consent now uses the shared delivery coordinator and scheduler. +Existing `chat_teams_file_transfers` rows own its state. There is no new table, +independent timer, provider request or external-wake capability. + +## Obligation and trigger + +Only `consent_pending`, `awaiting_consent` and `upload_pending` rows are eligible. +Each uses its saved `expiresAt`. Startup reconstructs these rows in 100-row UUID +pages. Committed writes reload the affected transfer ID. The service registers +intent before insertion and its central versioned update, within the existing +transaction. Nested savepoints wait for outer settlement. The shared work-signals +protocol retains ambiguous outcomes; callers do not manage recovery wrappers. + +Configured endpoints, completed history, unknown outcomes and active transport +attempts do not keep this worker scanning. Pending consent retains its actual +future deadline. PostgreSQL submillisecond timestamps round up to the next +millisecond; checking a deadline does not extend it. + +## Expiry and recovery + +At the deadline, the worker reloads the current row and delegates to the existing +Teams protocol owner. That owner locks publication then transfer, verifies the +current authority digest, rechecks phase and expiry, and updates the transfer +and publication projection in the same transaction. It does not send a consent +card, upload data or send a file-info card. Lock waits remain bounded. A changed +phase is re-evaluated; a deleted or settled row removes the cached obligation. + +A failed load, lock or projection retains retry work. Corrupt rows do not starve +other transfers or companies. Errors emitted by this worker are generic and do +not contain SQL parameters, private transfer envelopes or provider capabilities. +The shared worker owns coalescing, batching, dirty hints, drain/resume and shutdown. +Shutdown joins an in-flight expiry; restart reconstructs the saved deadline. + +The explicit `expireAndRecover` diagnostic method retains combined behavior. +Production periodic maintenance now selects and executes abandoned attempts only. +Attempt recovery is a separate remaining migration. Unknown provider outcomes +remain quarantined; this change does not authorize a resend. + +## Ownership and limitations + +Provider phase selection and expiry live in the Teams module and existing Teams +transfer service. Shared app wiring only registers the owner and lifecycle gate. +The queue topic is a notification namespace, not a new durable queue. Other +providers and the shared scheduler receive no Teams protocol policy. + +Supported producers use the process's owning DB client. Independent SQL writers +or another process require their own wake contract. On-use authorization remains +with the protocol. Cloud eligibility and external alarm coverage are unchanged: +a future in-process deadline cannot wake stopped compute. This proves the changed +worker can become quiet, not fleet savings or complete instance suspension. + +## Verification + +Real PostgreSQL protocol cases cover all three phases, producer activation, +rollback and held outer commit, restart, pre-deadline quiet, timestamp precision, +projection rollback, fairness, drain/resume and shutdown. Production integration +registers the worker before Board publication and proves expiry cancels the +publication without another provider call. Existing authority, publication and +consent suites remain covered. See the [audit and failure ledger](plans/2026-10-25-teams-consent-expiry-sleep-audit.md). 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 bb09a6b000e..968f73c1f38 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1358,3 +1358,16 @@ cleanup and retention workers had 37 passing cases. CI/reviews remain required. Next owner audit splits Teams unused-consent expiry from abandoned send/upload attempt recovery, preserving one reason per PR. Company deletion cleanup policy and the earlier E2E failure remain unresolved completion duties. + + +## Slice 26: Teams unused-consent expiry + +The Teams owner now selects existing unused-consent phases and their saved expiry +through scoped deadline work. Production maintenance retains only abandoned +attempts; these are the next distinct reason. No new table or Cloud alarm policy +is included. The [audit](2026-10-25-teams-consent-expiry-sleep-audit.md) records +222 unique selected passing cases, workspace types/build, and initial fixture +failures. Publication and fresh remote checks/review remain required. +Parent #17586 Actions/security/Greptile pass at 250fca6cff; Paperclip Review is +still running. The original full inventory, older E2E failure, deletion-policy +finding and master reconciliation remain outstanding. No stack PR is merged. diff --git a/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md b/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md index cfb2f005c3b..2b4a055871c 100644 --- a/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md +++ b/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md @@ -2,7 +2,7 @@ Baseline: voice call-report PR #17586 at 250fca6cff. Branch: codex/sleep-26-teams-consent-expiry. -Status: audit; implementation, local acceptance, publication and remote gates pending. +Status: implemented and locally verified; publication and remote gates pending. ## Separate reasons @@ -68,3 +68,54 @@ cannot expire consent. Full stack checks and master reconciliation remain requir #17586 is published, attached, and locally verified; remote CI/reviews are pending. #17580's company deletion policy finding and #17490's E2E cancellation failure are still unresolved. No original recurring-server obligation is removed from scope. + +## Implementation and acceptance + +The Teams owner now registers a scoped deadline worker. The existing service +signals before issue/update writes, including caller-owned resolution transactions. +Startup pages eligible rows; transfer hints reload only that primary key. The +owner rechecks expiry with the original publication/transfer lock order and +atomic projection. The old periodic selection and execution now cover attempts +only. No schema or external wake contract changes are included. + +Local verification at the slice-26 implementation: + +- 65 transfer protocol cases pass on a fresh external PostgreSQL database. + These include 102-entry paging/fairness with one corrupt row, shutdown joining + a held projection, the three consent phases, precision, rollback, restart, + drain/resume, private-error redaction and the remaining maintenance boundary. +- The earlier five-suite run passed 206 cases: then-63 transfer cases and 143 + publication, consent, consent-runtime and app-wiring cases. The final transfer + run adds two cases. The latest app-wiring assertions pass all 16 cases again. +- 14 selected real production integration cases pass. This includes registration + before an actual Board publication, post-commit wake, saved deadline, pre-expiry + SQL quiet, atomic expiry/publication cancellation, no extra provider requests, + and the existing Teams current-authority/Board-transfer group. The other 1,411 + integration cases were skipped, not passed. +- Together these are 222 unique selected cases. Workspace `pnpm -r typecheck` + and `pnpm build` pass. Server typechecking excludes test files; it is not proof + that the test files were statically checked. +- Local full `pnpm test:run` remains unclaimed because this host's embedded + PostgreSQL setup failed earlier in the stack. Full remote gates are required. + No live Teams tenant or deployed host-suspension qualification is claimed. + +Failure ledger: the first protocol run had two invalid fixtures. A direct phase +mutation set attempt expiry without its required paired attempt ID; another +assertion treated the public ISO expiry string as a Date. Both were corrected, +then all five suites passed on a fresh database. The first production expiry +assertion expected a failed publication; the existing correct expiry projection +is cancelled. The corrected test and the authority group pass. These were test +expectation/setup errors, not weakened production contracts or hidden retries. + +Evidence logs: `/tmp/sleep26-teams-tests.log` (initial two fixture failures), +`/tmp/sleep26-teams-fixed-tests.log` (206 pass), +`/tmp/sleep26-teams-final-tests.log` (65 pass), +`/tmp/sleep26-producer-tests.log` (initial projection-state assertion), +`/tmp/sleep26-production-acceptance.log` (14 pass), +`/tmp/sleep26-final-wiring.log` (16 pass), and +`/tmp/sleep26-workspace-{types,build}.log` (pass). + +Parent #17586 at 250fca6cff has passing Actions, security and Greptile checks; +Paperclip Review is still running at the latest inspection. This does not clear +older stack findings or imply merge authority. Refreshed master is b7dc594bfa; +final whole-stack reconciliation is still required. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index bc4f3461e03..4c798afba0a 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -79,6 +79,12 @@ describe("createChatReconciliationCoordinator", () => { expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()"); expect(maintenance).not.toContain("chatChannels.voice."); expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); + expect(source).toContain("chatChannels.registerTeamsConsentExpiryWork(deliveryWork,"); + const teams = readFileSync(new URL("../services/chat-channels/microsoft-teams/file-delivery.ts", import.meta.url), "utf8"); + const remainingMaintenance = teams.slice(teams.indexOf("function scheduleTeamsFileMaintenance"), teams.indexOf("function registerTeamsConsentExpiryWork")); + expect(remainingMaintenance).toContain("protocol.recoverAbandonedAttempts("); + expect(remainingMaintenance).not.toContain("protocol.expireAndRecover("); + expect(remainingMaintenance).not.toContain("chatTeamsFileTransfers.expiresAt"); }); it("separates live voice recovery from the remaining periodic maintenance", () => { diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index ce85806daf8..c3658146162 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1203,6 +1203,40 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { companyId, assignedAgentId, replacementAgentId }; } + it("expires Teams consent from the committed Board transfer without periodic maintenance", async () => { + const context = await teamsFileAuthorityFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const settle = () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + try { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + // Register before the ordinary Board producer, with no transfer rows yet. + await context.service.registerTeamsConsentExpiryWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + const attachment = await context.createFile(); + const publication = await context.service.publishBoardMessage(context.endpoint.id, context.conversation.id, + "A file awaiting consent", `teams-expiry-${randomUUID()}`, "owner-user", [attachment.id]); + const [transfer] = await db.select().from(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.publicationId, publication.id)); + expect(transfer.phase).toBe("awaiting_consent"); expect(context.consentCards).toHaveLength(1); + await settle(); expect(coordinator.nextWakeAt()).toBe(transfer.expiresAt.getTime()); + const reads = vi.spyOn(db, "select"); + try { await vi.advanceTimersByTimeAsync(10_000); expect(reads).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); } + await vi.advanceTimersByTimeAsync(transfer.expiresAt.getTime() - Date.now() + 1000); await settle(); + const [expired] = await db.select().from(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.id, transfer.id)); + const [projected] = await db.select().from(chatPublications).where(eq(chatPublications.id, publication.id)); + expect(expired.phase).toBe("expired"); expect(projected.state).toBe("cancelled"); + expect(coordinator.nextWakeAt()).toBeNull(); + expect(context.consentCards).toHaveLength(1); expect(context.fileCards).toEqual([]); + expect(context.uploadRequest).not.toHaveBeenCalled(); expect(errors).toEqual([]); + } finally { + await coordinator.stop(); vi.useRealTimers(); + await retirePublicationFixture(context.service, context.endpoint.id); + } + }); + it("retries a colliding company fixture prefix and creates a distinct complete fixture", async () => { const existing = await seedCompany(); const [original] = await db.select().from(companies).where(eq(companies.id, existing.companyId)); diff --git a/server/src/app.ts b/server/src/app.ts index 62af153dd0a..2ac96e978c7 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1301,6 +1301,7 @@ export async function createApp( chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.voice.inbound.registerRetentionWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); const reconcileChatPublicationMaintenance = async () => { diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index e193967a58f..fad48a02cf1 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1255,7 +1255,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { stageTerminalReceiptReactionRemovals, runtimeContextForRecord, scheduleMessageProcessing, processReceiptReaction, }); const { stageTeamsBoardFileIntent, deriveTeamsFileTransferAuthority } = createTeamsFileAuthority({ runtimeGeneration, credentialFingerprint, runtime, runtimeContexts, localRuntimeEpoch, runtimeVersions, lockCurrentPrincipalAuthorization, lifecycleRuntimeFence, assertFailedRetryReceipt, inboundWakeScope, failedChatRetrySource, isExplicitOperatorPublication, retryCommentIds }); - const { teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance } = createTeamsFileDelivery(db, { + const { registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance } = createTeamsFileDelivery(db, { endpointRecord, runtimeContextForRecord, runtimeFor, runtimeContexts, localRuntimeEpoch, runtime, deriveTeamsFileTransferAuthority, publicationFiles, scheduleMessageProcessing, processPendingPublications, withCredentialMutationLease, teamsFileRetryAfter, isShuttingDown: () => shuttingDown, @@ -3501,6 +3501,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor }); const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { + registerTeamsConsentExpiryWork, registerCommandWork, registerReceiverWork, voice, diff --git a/server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts b/server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts new file mode 100644 index 00000000000..e4dd2044531 --- /dev/null +++ b/server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts @@ -0,0 +1,53 @@ +import { and, eq, gt, inArray, sql } from "drizzle-orm"; +import { chatTeamsFileTransfers, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; + +export const unusedTeamsConsentPhases = ["consent_pending", "awaiting_consent", "upload_pending"]; +type Entry = { id: string; companyId: string; dueAt: number }; + +export function notifyTeamsConsentExpiry(tx: object, id: string) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTeamsConsentExpiry, id); +} + +/** Owns only unused consent deadlines, not transport attempts or unknown outcomes. */ +export function teamsConsentExpiryWork(db: Db, expire: (companyId: string, id: string) => Promise) { + async function load(id?: string, signal?: AbortSignal): Promise { + try { + const entries: Entry[] = []; let cursor: string | undefined; + for (;;) { + signal?.throwIfAborted(); + const rows = await db.select({ id: chatTeamsFileTransfers.id, companyId: chatTeamsFileTransfers.companyId, + dueAt: sql`ceil(extract(epoch from ${chatTeamsFileTransfers.expiresAt}) * 1000)`, + }).from(chatTeamsFileTransfers).where(and(inArray(chatTeamsFileTransfers.phase, unusedTeamsConsentPhases), + id ? eq(chatTeamsFileTransfers.id, id) : undefined, + cursor ? gt(chatTeamsFileTransfers.id, cursor) : undefined)).orderBy(chatTeamsFileTransfers.id).limit(100); + entries.push(...rows.map(row => ({ ...row, dueAt: Number(row.dueAt) }))); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + } catch { + // Never expose private transfer envelopes or SQL parameter errors. + throw new Error("Teams consent expiry selection is temporarily unavailable"); + } + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + const current = (await load(entry.id, signal))[0]; + if (!current) return null; + if (current.dueAt > Date.now()) return current; + signal?.throwIfAborted(); + try { await expire(current.companyId, current.id); } + catch { throw new Error("Teams consent expiry could not settle"); } + return (await load(entry.id, signal))[0] ?? null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatTeamsConsentExpiry, + decodeScope: (hint: string) => hint === "*" ? undefined : hint, + matches: (entry: Entry, id?: string) => !id || entry.id === id, + load, check, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/chat-channels/microsoft-teams/file-delivery.ts b/server/src/services/chat-channels/microsoft-teams/file-delivery.ts index 9848222c9b9..396662a4e49 100644 --- a/server/src/services/chat-channels/microsoft-teams/file-delivery.ts +++ b/server/src/services/chat-channels/microsoft-teams/file-delivery.ts @@ -1,3 +1,5 @@ +import { teamsConsentExpiryWork } from "./consent-expiry-work.js"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; import { createOutboundAttachments } from "../outbound-attachments.js"; import { createTeamsFileAuthority } from "./file-transfer-authority.js"; @@ -14,7 +16,6 @@ import { eq, inArray, lte, - or, } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -392,24 +393,11 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption .selectDistinct({ companyId: chatTeamsFileTransfers.companyId }) .from(chatTeamsFileTransfers) .where( - or( - and( - inArray(chatTeamsFileTransfers.phase, [ - "consent_pending", - "awaiting_consent", - "upload_pending", - ]), - lte(chatTeamsFileTransfers.expiresAt, new Date()), - ), - and( - inArray(chatTeamsFileTransfers.phase, [ - "consent_sending", - "uploading", - "file_info_sending", - "conflict", - ]), - lte(chatTeamsFileTransfers.attemptExpiresAt, new Date()), - ), + and( + inArray(chatTeamsFileTransfers.phase, [ + "consent_sending", "uploading", "file_info_sending", "conflict", + ]), + lte(chatTeamsFileTransfers.attemptExpiresAt, new Date()), ), ); const protocol = teamsTransferProtocol(); @@ -417,7 +405,7 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption // its row count and lock waits and performs no provider I/O. for (const scope of scopes) { if (isShuttingDown()) break; - await protocol.expireAndRecover(scope.companyId, 100); + await protocol.recoverAbandonedAttempts(scope.companyId, 100); } } catch { // Selection errors are closed here too; private transfer envelopes and @@ -430,5 +418,9 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption } }); } - return { teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance }; + function registerTeamsConsentExpiryWork(coordinator: ReturnType, canRun: () => boolean) { + return teamsConsentExpiryWork(db, (companyId, id) => teamsTransferProtocol().expireUnusedConsent(companyId, id)) + .register(coordinator, () => !isShuttingDown() && canRun()); + } + return { registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance }; } diff --git a/server/src/services/chat-teams-file-transfers.test.ts b/server/src/services/chat-teams-file-transfers.test.ts index ad6b5084c17..504f921f276 100644 --- a/server/src/services/chat-teams-file-transfers.test.ts +++ b/server/src/services/chat-teams-file-transfers.test.ts @@ -1,3 +1,6 @@ +import { teamsConsentExpiryWork } from "./chat-channels/microsoft-teams/consent-expiry-work.js"; +import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "./task-admission.js"; import { createHash, randomUUID } from "node:crypto"; import { afterAll, @@ -329,6 +332,188 @@ suite( }; } + describe("scheduled unused consent expiry", () => { + let coordinator: ReturnType | undefined; + const transferIds: string[] = [], errors: unknown[] = []; + let enabled = true; + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + if (transferIds.length) await db.delete(chatTeamsFileTransfers).where(inArray(chatTeamsFileTransfers.id, transferIds.splice(0))); + errors.length = 0; enabled = true; + }); + async function transfer() { + const f = await fixture(); transferIds.push(f.created.id); + f.opts.now = () => new Date(); + return f; + } + function worker(expire: (companyId: string, id: string) => Promise) { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + const work = teamsConsentExpiryWork(db, expire); + return { work, ready: work.register(coordinator, () => enabled).ready }; + } + async function settled() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 }); + } + async function quiet(ms = 10_000) { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute"); + try { await vi.advanceTimersByTimeAsync(ms); expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); } + finally { select.mockRestore(); transaction.mockRestore(); execute.mockRestore(); } + } + it("is quiet with configured endpoints and cancelled history", async () => { + const f = await transfer(); await f.service().cancel(f.a.companyId, f.created.id); + const expire = vi.fn(f.service().expireUnusedConsent); + await worker(expire).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect(expire).not.toHaveBeenCalled(); + }); + it.each(["consent_pending", "awaiting_consent", "upload_pending"])("restores the saved %s expiry without an early scan or provider effect", async phase => { + const f = await transfer(); + if (phase !== "consent_pending") await f.service().process(f.a.companyId, f.created.id); + if (phase === "upload_pending") await f.service().recordConsent(await f.event()); + expect((await f.read()).phase).toBe(phase); + const providerCalls = vi.mocked(f.opts.postConsent).mock.calls.length; + const due = (await f.read()).expiresAt.getTime(); + await worker(f.service().expireUnusedConsent).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBe(due); await quiet(); + await coordinator!.stop(); + await worker(f.service().expireUnusedConsent).ready; + expect(coordinator!.nextWakeAt()).toBe(due); await quiet(); + await vi.advanceTimersByTimeAsync(due - Date.now() + 1000); await settled(); + expect((await f.read()).phase).toBe("expired"); expect(coordinator!.nextWakeAt()).toBeNull(); + await quiet(180_000); expect(f.opts.postConsent).toHaveBeenCalledTimes(providerCalls); + expect(f.opts.uploadRequest).not.toHaveBeenCalled(); expect(f.opts.postFileInfo).not.toHaveBeenCalled(); expect(errors).toEqual([]); + }); + it("reactivates from real issuance and disarms on cancellation", async () => { + let f: Awaited>; + await worker((companyId, id) => f.service().expireUnusedConsent(companyId, id)).ready; + await settled(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + f = await transfer(); await settled(); + expect(coordinator!.nextWakeAt()).toBe((await f.read()).expiresAt.getTime()); + await f.service().cancel(f.a.companyId, f.created.id); await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect(f.opts.postConsent).not.toHaveBeenCalled(); + }); + it("keeps first committed work through drain without passive SQL", async () => { + enabled = false; + let f: Awaited>; + await worker((companyId, id) => f.service().expireUnusedConsent(companyId, id)).ready; + f = await transfer(); await quiet(90_000); + expect((await f.read()).phase).toBe("consent_pending"); + enabled = true; await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect((await f.read()).phase).toBe("expired"); expect(coordinator!.nextWakeAt()).toBeNull(); + await quiet(180_000); expect(errors).toEqual([]); + }); + it("retains expiry after atomic projection failure and redacts its private error", async () => { + const f = await transfer(); let fail = true; + f.opts.project = async tx => { + await tx.update(chatPublications).set({ state: "cancelled" }).where(eq(chatPublications.id, f.a.publicationId)); + if (fail) throw new Error("PRIVATE-TEAMS-UPLOAD-CANARY"); + }; + await worker(f.service().expireUnusedConsent).ready; + await vi.advanceTimersByTimeAsync((await f.read()).expiresAt.getTime() - Date.now() + 1000); await settled(); + expect((await f.read()).phase).toBe("consent_pending"); + expect((await db.select().from(chatPublications).where(eq(chatPublications.id, f.a.publicationId)))[0].state).toBe("pending"); + expect(errors).toHaveLength(1); + expect(String(errors[0])).not.toContain("CANARY"); + expect(String((errors[0] as AggregateError).errors[0])).not.toContain("CANARY"); + fail = false; await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect((await f.read()).phase).toBe("expired"); + expect((await db.select().from(chatPublications).where(eq(chatPublications.id, f.a.publicationId)))[0].state).toBe("cancelled"); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("waits for outer commit and ignores nested rollback", async () => { + const f = await transfer(); await db.delete(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.id, f.created.id)); + await worker(f.service().expireUnusedConsent).ready; await settled(); + const due = new Date(Date.now() + 300_000); + await expect(db.transaction(async tx => { + await tx.transaction(async nested => teamsFileTransferService(nested as unknown as typeof db, f.opts).issue(f.a, due)); + await vi.advanceTimersByTimeAsync(10_000); expect(coordinator!.nextWakeAt()).not.toBe(due.getTime()); + throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await settled(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + let release!: () => void, inserted!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { inserted = resolve; }); + const write = db.transaction(async tx => { + const row = await tx.transaction(async nested => teamsFileTransferService(nested as unknown as typeof db, f.opts).issue(f.a, due)); + inserted(); await hold; return row; + }); + try { + await ready; await vi.advanceTimersByTimeAsync(10_000); + expect(coordinator!.nextWakeAt()).not.toBe(due.getTime()); + } finally { release(); } + const row = await write; transferIds.push(row.id); await settled(); + expect(coordinator!.nextWakeAt()).toBe(due.getTime()); await quiet(); + }); + it("rounds PostgreSQL expiry up and leaves active or unknown attempts alone", async () => { + const f = await transfer(); + const whole = Date.now() + 60_000; + await db.update(chatTeamsFileTransfers).set({ expiresAt: sql`${new Date(whole).toISOString()}::timestamptz + interval '0.0005 second'` }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + vi.useFakeTimers({ toFake: ["Date"] }); + const work = teamsConsentExpiryWork(db, f.service().expireUnusedConsent); + const [entry] = await work.load(); expect(entry.dueAt).toBe(whole + 1); + vi.setSystemTime(whole); await work.check(entry); + expect((await f.read()).phase).toBe("consent_pending"); + vi.setSystemTime(whole + 1); expect(await work.check(entry)).toBeNull(); + expect((await f.read()).phase).toBe("expired"); + for (const phase of ["consent_sending", "uploading", "file_info_sending", "consent_unknown", "upload_unknown", "file_info_unknown", "conflict"]) { + await db.update(chatTeamsFileTransfers).set({ phase, attemptId: randomUUID(), attemptExpiresAt: new Date(0) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + const work = teamsConsentExpiryWork(db, f.service().expireUnusedConsent); + expect(await work.load()).toEqual([]); + } + }); + it("pages and drains more than 100 obligations without starving another company behind a corrupt row", async () => { + const f = await transfer(), other = await transfer(); + const due = new Date(Date.now() + 60_000); + for (let index = 0; index < 100; index++) { + const publicationId = randomUUID(); + await db.insert(chatPublications).values({ id: publicationId, companyId: f.a.companyId, + endpointId: f.a.endpointId, conversationId: f.a.conversationId, issueId: f.a.issueId, + commentId: f.a.commentId, idempotencyKey: `expiry-page:${publicationId}`, + payload: { text: "Same authorized file", attachmentIds: [f.a.attachmentId] } }); + const a = { ...f.a, publicationId }; + const created = await teamsFileTransferService(db, { ...f.opts, authorize: async () => a }).issue(a, due); + transferIds.push(created.id); + } + await db.update(chatTeamsFileTransfers).set({ authorityDigest: "0".repeat(64) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + const { work, ready } = worker(f.service().expireUnusedConsent); await ready; + expect(await work.load()).toHaveLength(102); + await vi.advanceTimersByTimeAsync(due.getTime() - Date.now() + 1000); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect((await other.read()).phase).toBe("expired"); + const remaining = await work.load(); expect(remaining.map(row => row.id)).toEqual([f.created.id]); + }, { timeout: 10_000 }); + expect((await f.read()).phase).toBe("consent_pending"); + expect(errors.length).toBeGreaterThan(0); expect(f.opts.postConsent).not.toHaveBeenCalled(); + }, 30_000); + it("joins an in-flight expiry projection during shutdown", async () => { + const f = await transfer(); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + f.opts.project = async () => { entered(); await hold; }; + await worker(f.service().expireUnusedConsent).ready; + try { + await vi.advanceTimersByTimeAsync((await f.read()).expiresAt.getTime() - Date.now() + 1000); + await ready; + let stopped = false; const stop = coordinator!.stop().then(() => { stopped = true; }); + await Promise.resolve(); expect(stopped).toBe(false); + release(); await stop; + expect((await f.read()).phase).toBe("expired"); + } finally { release(); } + }); + it("keeps the compatibility attempt sweep from expiring unused consent", async () => { + const f = await transfer(); + // Restore fixture time so the explicit maintenance operation sees it due. + f.opts.now = () => new Date(Date.parse(f.created.expiresAt) + 1); + expect(await f.service().recoverAbandonedAttempts(f.a.companyId)).toMatchObject({ scanned: 0, recovered: 0 }); + expect((await f.read()).phase).toBe("consent_pending"); + expect(await f.service().expireUnusedConsent(f.a.companyId, f.created.id)).toBe(true); + expect((await f.read()).phase).toBe("expired"); + }); + }); + it("uses the exact attachment-derived filename for an unnamed source across issuance and restart", async () => { const f = await fixture(60_000, null); const expected = `attachment-${f.a.attachmentId}`; diff --git a/server/src/services/chat-teams-file-transfers.ts b/server/src/services/chat-teams-file-transfers.ts index 4f88fb88e74..0f82870da06 100644 --- a/server/src/services/chat-teams-file-transfers.ts +++ b/server/src/services/chat-teams-file-transfers.ts @@ -1,3 +1,4 @@ +import { notifyTeamsConsentExpiry, unusedTeamsConsentPhases } from "./chat-channels/microsoft-teams/consent-expiry-work.js"; import { createHash, randomUUID } from "node:crypto"; import { and, eq, gt, inArray, lte, or, sql } from "drizzle-orm"; import { z } from "zod"; @@ -388,6 +389,7 @@ export function teamsFileTransferService( row: Row, patch: Partial, ): Promise { + await notifyTeamsConsentExpiry(tx, row.id); const [next] = await tx .update(chatTeamsFileTransfers) .set({ ...patch, version: row.version + 1, updatedAt: now() }) @@ -533,6 +535,7 @@ export function teamsFileTransferService( binding, ), }; + await notifyTeamsConsentExpiry(tx, id); const [row] = await tx .insert(chatTeamsFileTransfers) .values({ @@ -859,9 +862,32 @@ export function teamsFileTransferService( if (!row) fail(); return summary(row); } + async function recoverTransfer(companyId: string, id: string, reason: "all" | "consent" | "attempts") { + return db.transaction(async (tx) => { + // Bound maintenance waits, including locks taken by atomic projection. + await tx.execute(sql`set local lock_timeout = '250ms'`); + const row = await locked(tx, companyId, id, true); + if (reason !== "consent" && (sending.includes(row.phase) || row.phase === "conflict") && + row.attemptExpiresAt && row.attemptExpiresAt <= now()) { + await update(tx, row, { + phase: row.phase === "conflict" ? "conflict" : unknownPhase[row.phase], + attemptId: null, + attemptExpiresAt: null, + reason: row.phase === "conflict" ? row.reason : "worker_outcome_unknown", + }); + return true; + } + if (reason !== "attempts" && unusedTeamsConsentPhases.includes(row.phase) && row.expiresAt <= now()) { + await update(tx, row, { phase: "expired", reason: "unused_consent_expired" }); + return true; + } + return false; + }); + } async function expireAndRecover( companyId: string, limit = 25, + reason: "all" | "attempts" = "all", ): Promise { if (!z.uuid().safeParse(companyId).success) fail(); const size = Number.isSafeInteger(limit) @@ -872,7 +898,8 @@ export function teamsFileTransferService( cursors = new Map(); expiryRecoveryCursors.set(db, cursors); } - const cursor = cursors.get(companyId); + const cursorKey = `${companyId}:${reason}`; + const cursor = cursors.get(cursorKey); const cutoff = now(); const eligible = and( eq(chatTeamsFileTransfers.companyId, companyId), @@ -881,14 +908,10 @@ export function teamsFileTransferService( inArray(chatTeamsFileTransfers.phase, [...sending, "conflict"]), lte(chatTeamsFileTransfers.attemptExpiresAt, cutoff), ), - and( - inArray(chatTeamsFileTransfers.phase, [ - "consent_pending", - "awaiting_consent", - "upload_pending", - ]), + reason === "all" ? and( + inArray(chatTeamsFileTransfers.phase, unusedTeamsConsentPhases), lte(chatTeamsFileTransfers.expiresAt, cutoff), - ), + ) : undefined, ), ); let rows: Array<{ id: string }>; @@ -927,45 +950,11 @@ export function teamsFileTransferService( for (const candidate of rows) { // Progress over rejected rows too. Neither corrupt evidence nor a held // publication lock can pin every later scan to the same oldest page. - cursors.set(companyId, candidate.id); + cursors.set(cursorKey, candidate.id); result.cursor = candidate.id; result.scanned++; try { - const recovered = await db.transaction(async (tx) => { - // The projection may need endpoint/conversation rows too. Bound - // only maintenance lock waits, including locks inside that callback. - await tx.execute(sql`set local lock_timeout = '250ms'`); - const row = await locked(tx, companyId, candidate.id, true); - if ( - (sending.includes(row.phase) || row.phase === "conflict") && - row.attemptExpiresAt && - row.attemptExpiresAt <= now() - ) { - await update(tx, row, { - phase: - row.phase === "conflict" ? "conflict" : unknownPhase[row.phase], - attemptId: null, - attemptExpiresAt: null, - reason: - row.phase === "conflict" - ? row.reason - : "worker_outcome_unknown", - }); - return true; - } else if ( - ["consent_pending", "awaiting_consent", "upload_pending"].includes( - row.phase, - ) && - row.expiresAt <= now() - ) { - await update(tx, row, { - phase: "expired", - reason: "unused_consent_expired", - }); - return true; - } - return false; - }); + const recovered = await recoverTransfer(companyId, candidate.id, reason); if (recovered) result.recovered++; } catch { // Keep the exact evidence and rollback any paired projection. Only @@ -1168,7 +1157,9 @@ export function teamsFileTransferService( recordConsent, process, get, - expireAndRecover, + expireAndRecover: (companyId: string, limit = 25) => expireAndRecover(companyId, limit), + recoverAbandonedAttempts: (companyId: string, limit = 25) => expireAndRecover(companyId, limit, "attempts"), + expireUnusedConsent: (companyId: string, id: string) => recoverTransfer(companyId, id, "consent"), cancel, canCancelConflict, resolveInTransaction, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 05467292463..699cb7ecfb3 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -46,6 +46,7 @@ export const DELIVERY_QUEUES = { voiceInboundRetention: "voice-inbound-retention", voiceInboundCleanup: "voice-inbound-cleanup", voiceReplies: "voice-reply-delivery", + chatTeamsConsentExpiry: "chat-teams-consent-expiry", chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs",