fix(server): schedule Teams consent expiry from saved deadlines

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 06:01:32 -05:00
1 parent 9a06e4a6ba
commit 5ddb4543ed
12 files changed
+457 -68

No files matched your search

+61
View File
@@ -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).
@@ -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.
@@ -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.
@@ -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", () => {
@@ -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));
+1
View File
@@ -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 () => {
+2 -1
View File
@@ -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,
@@ -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<boolean>) {
async function load(id?: string, signal?: AbortSignal): Promise<Entry[]> {
try {
const entries: Entry[] = []; let cursor: string | undefined;
for (;;) {
signal?.throwIfAborted();
const rows = await db.select({ id: chatTeamsFileTransfers.id, companyId: chatTeamsFileTransfers.companyId,
dueAt: sql<string>`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<Entry | null> {
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<typeof createDeliveryWorkCoordinator>, 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 };
}
@@ -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<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
return teamsConsentExpiryWork(db, (companyId, id) => teamsTransferProtocol().expireUnusedConsent(companyId, id))
.register(coordinator, () => !isShuttingDown() && canRun());
}
return { registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance };
}
@@ -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<typeof createDeliveryWorkCoordinator> | 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<boolean>) {
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<ReturnType<typeof transfer>>;
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<ReturnType<typeof transfer>>;
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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; });
const ready = new Promise<void>(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}`;
@@ -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<typeof chatTeamsFileTransfers.$inferInsert>,
): Promise<Row> {
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<TeamsFileRecoverySweepSummary> {
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,
@@ -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",