diff --git a/doc/idle-teams-consent-expiry.md b/doc/idle-teams-consent-expiry.md new file mode 100644 index 00000000000..6faa09fca42 --- /dev/null +++ b/doc/idle-teams-consent-expiry.md @@ -0,0 +1,72 @@ +# 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`. Registered startup reconstruction yields after +one 100-row UUID page per scope refresh and resumes through the shared scheduler. +Registration resets the owner’s cursors so a partial scan restarts safely. The +explicit one-shot inventory method still returns the complete inventory. 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). + + +Current local acceptance covers 223 distinct cases: 209 across five complete +protocol/publication/consent/runtime/wiring suites, plus 14 selected real Board +producer and Teams-authority integration cases. The other 1,580 integration +cases were skipped. A real 105-row regression proves bounded startup, same-owner +restart, complete saved-deadline expiry, and quiet SQL. A guarded missing-reset +mutation fails that regression. See the audit for exact revisions, earlier +fixture failures, current-parent integration status and deployment limits. 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 e51747d7775..9f5cdbb750b 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1809,3 +1809,16 @@ notifications and startup; those saves select zero rows and preserve the deadlin The [report notification ledger](2026-10-25-voice-call-report-sleep-audit.md#configuration-scope-review-repair-october-28) records 303 passing cases, types/build, the narrower contract and limits. Fresh exact-head review and CI are required after the repair. + + +## 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 new file mode 100644 index 00000000000..5a1e969678a --- /dev/null +++ b/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md @@ -0,0 +1,219 @@ +# Teams unused-consent expiry sleep audit + +Baseline: voice call-report PR #17586 at 250fca6cff. +Branch: codex/sleep-26-teams-consent-expiry. +Status: implemented and locally verified; publication and remote gates pending. + +## Separate reasons + +The Teams maintenance callback in microsoft-teams/file-delivery.ts runs from +processPublicationMaintenance and repeats a database selection every 30 seconds. +It combines two obligations. This slice owns only unused consent expiry: +consent_pending, awaiting_consent and upload_pending rows become expired at their +existing expiresAt. Abandoned consent_sending/uploading/file_info_sending attempts, +and conflict rows with a retained attempt expiry, belong to the next slice. +Unknown outcomes without a live deadline are passive operator-review evidence; +expiry must not convert them into retries or manufacture delivery receipts. + +## Owner and persistence + +chatTeamsFileTransfers already owns the phase, expiry, company, endpoint, +publication, version, authority digest and encrypted private state. No schema or +new queue table is needed. teamsFileTransferService is the only production insert +and update owner found by the initial writer search. Its issue() insertion and +central update() helper are the two notification boundaries. Caller-owned +resolveInTransaction uses that helper; its signal must wait for outer commit. +Deletion should simply remove a cached entry on the next owned deadline; it must +not retain a perpetual scan. + +The existing lock order is publication -> transfer -> projection policy/source +locks. Expiry rechecks eligibility after those locks and performs no provider +I/O. Integrity checks and projection in the same transaction must remain intact. +Maintenance lock waits are bounded to 250ms, with no-wait publication/transfer +locking. A bad or locked row must not starve other companies or later rows. +Never log private envelopes, SQL parameter errors, provider capabilities or tokens. + +## Intended migration + +- Reuse scoped deadline work, with paged startup reconstruction and transfer-ID + hints from committed writes. Register the worker through the Teams owner; + app wiring only supplies shared lifecycle and admission gates. +- Load only the three unused phases, with ceil(expiresAt milliseconds) so a + PostgreSQL submillisecond expiry cannot be executed early or spin. +- Recheck and expire one transfer through its owning protocol method, preserving + atomic publication projection and no-provider-I/O semantics. Result loading + removes settled/changed rows; failed reconstruction or expiry retains retry. +- Keep explicit manual expireAndRecover compatibility for tests/diagnostics. + Production aggregate maintenance must select and execute only abandoned + attempts after this slice. It must not keep expiring unused consent as a hidden + second owner. That remaining attempt scan is intentionally still pending. +- Preserve configured idle, completed history, unknown outcome and active-attempt + quiet for this worker. Real outstanding consent remains a future obligation. +- Do not change authority-on-use checks, transport execution, Cloud wake coverage + or whole-instance sleep eligibility. + +## Verification plan + +Use the existing real-PostgreSQL Teams protocol fixture, and exercise real issue, +process/consent and cancellation/resolution writers. Cover saved expiry and +restart, first committed transfer, nested commit/rollback, no early reads, +changed phase at claim, provider-call absence, atomic projection rollback, +private-error redaction, lock contention/fair backlog, drain/resume, shutdown, +and submillisecond deadlines. Existing protocol, publication and consent tests +must continue to pass. Exercise production registration and ensure old maintenance +cannot expire consent. Full stack checks and master reconciliation remain required. + +## Other outstanding gates + +#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. + + +## Bounded reconstruction and current-parent acceptance — October 28 + +The published Teams head `aa082e4679` was inspected after the report parent was +prepared on landed master. Its worker's registered load still looped through +all pages in one scheduler pass. This acceptance repairs that existing slice; +it does not introduce another queue, scheduler, database change or duty. + +A local normal merge prepares the slice on report candidate `c28de398`. The +explicit-base merge had one stack-plan append conflict. The complete report +plan and complete Teams appendix are both preserved; all source blobs match +the clean explicit-base composition. An initial literal patch comparison +rejected shifted queue line numbers (47 to 49) and blob IDs, then an overly +broad merge whitespace check flagged an inherited runner report-lock patch. +The actual patch content and owned-scope whitespace check pass; inherited +upstream files were not altered. Actual landed-parent reconciliation and fresh +remote gates are required before publication and merge. + +The new 105-transfer PostgreSQL regression failed against the original reader: +startup selected 105 rows instead of yielding after 100. The worker now splits +one-page selection from complete explicit inventory reads. Registered refresh +returns the existing shared `{rows, complete}` contract and retains per-scope +cursors; registration clears them. Restarting the same service halfway through +a scan reconstructs all 105 saved deadlines, expires them at the original time, +and leaves no recurring SQL before the deadline or after settlement. No consent +card or upload is sent. Removing only cursor reset makes the restart read five +rows instead of 100; original source was restored before final acceptance. + +The first regression fixture faked all timers before establishing its first +database connection and timed out, including teardown. That was not the intended +row-budget failure and is not accepted as regression evidence. Seeding now uses +real network timers with only Date frozen, then controls scheduler timers after +seeding. The corrected unchanged-reader run fails at 105 versus 100, and the +fixed worker passes. Both the failed setup and actual regression remain recorded. + +Current acceptance: all 209 cases in the complete transfer (66), publication, +consent, consent-runtime and app-wiring suites pass on a new migrated fixture; +14 selected real Board-producer and Teams-authority integration cases pass on a +separate new migrated fixture (1,580 others skipped). Total 223 distinct cases; +the targeted red/green case is not added again. Server production typecheck, +build and module-boundary checks pass. The new source remains within the +existing Teams owner and existing test file. No schema, shared-worker contract, +workflow or lockfile change is introduced. The old 222-case evidence remains +historical. This does not claim a complete local repository suite, a live Teams +tenant, external Cloud wake, physical suspension or cost savings. + +Logs: `/tmp/sleep26-bounded-reconstruction-red.log` (invalid timer setup), +`/tmp/sleep26-bounded-reconstruction-actual-red.log` (105 vs 100), +`/tmp/sleep26-bounded-reconstruction-green.log`, +`/tmp/sleep26-bounded-cursor-mutant-red.log` (5 vs 100 after restart), +`/tmp/sleep26-bounded-final-protocols.log`, +`/tmp/sleep26-bounded-final-producers.log`, +`/tmp/sleep26-bounded-final-types.log`, `/tmp/sleep26-bounded-final-build.log`, +and `/tmp/sleep26-bounded-final-boundaries.log`. + +Current stack state supersedes the historical outstanding-gates paragraph: +cleanup and retention have landed, including the repaired legacy ownership case +and the explicitly accepted company-deletion ownership policy. The report parent +has a documentation-only follow-up after `c28de398` and still needs its own latest +CI/reviews. This local Teams preparation must preserve that actual landed parent +before publication; it is not authority to merge into an unmerged feature branch. + + +## Inherited retention CI fairness repair — October 28 + +Current Teams head `36f43922` received exact-head Greptile 5/5 and Ponytail 5/5, +but CI run `40708566195`, server shard 6/12 job `122887395220`, failed the existing +voice-retention case `pages through more than one batch of history across +companies`. The shard had 2,291 passed tests, one failed and one skipped. Its +initial assertion expected the other company's row already deleted after one +startup batch, even when its random ID put it beyond the first 100-row inventory +page. This CI failure is not accepted as green evidence. + +Placing the other company at a fixed final UUID reproduces the exact assertion. +The fixture now uses a supplied ID for both insertion and its transaction hint. +The test controls scheduler time, requires that row to remain before discovery, +then requires its deletion in the next fair batch after page-two discovery. It +also checks the first 25 deletions leave exactly 80 eligible rows, drains all +remaining history, and verifies SQL quiet. The existing five-second assertion +wait budget and global test timeout are unchanged. This corrects the acceptance +expectation without changing retention, Teams, or shared scheduler behavior. + +The complete 13-case retention suite passes. Its run spent 96.66 seconds in +module transforms but all actual cases completed successfully; no timeout was +increased and this was not counted as a failed test or silently retried. A +guarded mutation bypassing company rotation fails the new post-discovery fair +batch assertion. The original shared helper was restored byte-for-byte. The +previous 223-case Teams acceptance remains evidence for unchanged Teams source; +these 13 retention cases are a separate CI-repair acceptance group. Earlier +current-head Greptile/Ponytail scores become stale after publishing this test-only +follow-up, and fresh CI must pass independently. No old-job retry or waiver is +claimed. + +Logs: `/tmp/sleep26-36f-server6-job.log`, +`/tmp/sleep26-retention-fairness-red.log`, +`/tmp/sleep26-retention-fairness-green.log`, and +`/tmp/sleep26-retention-fairness-mutant-red.log`. 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 ceecf407f5e..5e8c692955b 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1218,6 +1218,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/__tests__/voice-inbound-retention.test.ts b/server/src/__tests__/voice-inbound-retention.test.ts index 306e43e9408..6bdedd256d5 100644 --- a/server/src/__tests__/voice-inbound-retention.test.ts +++ b/server/src/__tests__/voice-inbound-retention.test.ts @@ -56,7 +56,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu }); afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); async function admission(patch: Partial = {}, writer = db) { - const id = randomUUID(); + const id = patch.id ?? randomUUID(); return writer.transaction(async tx => { await notifyVoiceInbound(tx, { kind: "admission", id }); return (await tx.insert(chatVoiceInboundCalls).values({ id, companyId, endpointId, providerSessionId: randomUUID(), state: "ended", cleanupConfirmedAt: new Date(), updatedAt: new Date(Date.now() - week - 1000), approvalCode: "123456", generation: 1, credentialFingerprint: "test", toolTokenHash: "a".repeat(64), requestFingerprint: id, expiresAt: new Date(Date.now() - week - 1000), ...patch }).returning())[0]; @@ -211,12 +211,22 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); }); it("pages through more than one batch of history across companies", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); for (let i = 0; i < 104; i++) await admission(); - const other = await fixture(); const last = await admission({ companyId: other.companyId, endpointId: other.endpointId }); + const other = await fixture(); const last = await admission({ id: "ffffffff-ffff-4fff-bfff-ffffffffffff", companyId: other.companyId, endpointId: other.endpointId }); const service = await start(); - expect(await row(last.id)).toBeUndefined(); - await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(await service.load()).toHaveLength(0); }, { timeout: 5000 }); - await flush(); await quiet(); + expect(await row(last.id)).toBeDefined(); + expect(await service.load()).toHaveLength(80); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(async () => expect(await row(last.id)).toBeUndefined(), { interval: 0, timeout: 5000 }); + for (let pass = 0; pass < 20; pass++) { + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt() !== null || idleWorkSnapshot().active === 0).toBe(true), { interval: 0, timeout: 5000 }); + } + await flush(); + expect(await service.load()).toHaveLength(0); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, companyId))).toHaveLength(0); + expect(await row(last.id)).toBeUndefined(); await quiet(); }); it("retains failed deletion for a bounded retry and then becomes quiet", async () => { const call = await admission(); diff --git a/server/src/app.ts b/server/src/app.ts index 169cc9634f4..849985244af 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 ab9dd9263e3..ef96c0788df 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1251,7 +1251,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, @@ -3498,6 +3498,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..7a2a9f8e97d --- /dev/null +++ b/server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts @@ -0,0 +1,64 @@ +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) { + const cursors = new Map(); + async function page(id?: string, cursor?: string, signal?: AbortSignal): Promise { + try { + 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); + signal?.throwIfAborted(); + return rows.map(row => ({ ...row, dueAt: Number(row.dueAt) })); + } catch { + throw new Error("Teams consent expiry selection is temporarily unavailable"); + } + } + async function load(id?: string, signal?: AbortSignal): Promise { + const entries: Entry[] = []; let cursor: string | undefined; + for (;;) { + const rows = await page(id, cursor, signal); + entries.push(...rows); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + } + async function refresh(id?: string, signal?: AbortSignal) { + const key = id ?? "*", rows = await page(id, cursors.get(key), signal), complete = rows.length < 100; + if (complete) cursors.delete(key); else cursors.set(key, rows[rows.length - 1].id); + return { rows, complete }; + } + 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) { + cursors.clear(); + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatTeamsConsentExpiry, + decodeScope: (hint: string) => hint === "*" ? undefined : hint, + matches: (entry: Entry, id?: string) => !id || entry.id === id, + load: refresh, 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..0dd72d0f387 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,241 @@ 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("bounds reconstruction and resets its cursor on same-owner restart", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + const f = await transfer(), due = Date.now() + 300_000; + await db.update(chatTeamsFileTransfers).set({ expiresAt: new Date(due) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + for (let index = 0; index < 104; 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-restart:${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, new Date(due)); + transferIds.push(created.id); + } + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + const counts: number[] = [], select = db.select.bind(db); + const reads = vi.spyOn(db, "select").mockImplementation((...args) => { + const selection = select(...args), from = selection.from.bind(selection); + vi.spyOn(selection, "from").mockImplementation(table => { + const query = from(table); + if (table === chatTeamsFileTransfers) { + const execute = query.execute.bind(query); + vi.spyOn(query, "execute").mockImplementation(async () => { + const rows = await execute(); counts.push(rows.length); return rows; + }); + } + return query; + }); + return selection; + }); + const expire = vi.fn(f.service().expireUnusedConsent), work = teamsConsentExpiryWork(db, expire); + const start = () => { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + return work.register(coordinator, () => enabled).ready; + }; + try { + await start(); expect(counts.reduce((sum, count) => sum + count, 0)).toBe(100); + await coordinator!.stop(); coordinator = undefined; + const before = counts.length; + await start(); expect(counts.slice(before).reduce((sum, count) => sum + count, 0)).toBe(100); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(due), { interval: 0 }); + expect(Math.max(...counts)).toBeLessThanOrEqual(100); reads.mockRestore(); + expect(await work.load()).toHaveLength(105); + await quiet(1000); expect(expire).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(expire).toHaveBeenCalledTimes(105); + expect(await db.select().from(chatTeamsFileTransfers).where(inArray(chatTeamsFileTransfers.id, transferIds))) + .toEqual(expect.arrayContaining([expect.objectContaining({ id: f.created.id, phase: "expired" })])); + expect(await work.load()).toEqual([]); expect(errors).toEqual([]); + await quiet(); expect(f.opts.postConsent).not.toHaveBeenCalled(); expect(f.opts.uploadRequest).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); } + }, 30_000); + 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 99b6f396067..4bc877804d3 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -49,6 +49,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",