mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 21:49:18 +01:00
fix(server): schedule Teams consent expiry from saved deadlines
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
9a06e4a6ba
commit
5ddb4543ed
12 files changed
+457
-68
No files matched your search
@@ -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
|
Next owner audit splits Teams unused-consent expiry from abandoned send/upload
|
||||||
attempt recovery, preserving one reason per PR. Company deletion cleanup policy
|
attempt recovery, preserving one reason per PR. Company deletion cleanup policy
|
||||||
and the earlier E2E failure remain unresolved completion duties.
|
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.
|
Baseline: voice call-report PR #17586 at 250fca6cff.
|
||||||
Branch: codex/sleep-26-teams-consent-expiry.
|
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
|
## 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.
|
#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
|
#17580's company deletion policy finding and #17490's E2E cancellation failure are
|
||||||
still unresolved. No original recurring-server obligation is removed from scope.
|
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.reconcileProviderRuntimes()");
|
||||||
expect(maintenance).not.toContain("chatChannels.voice.");
|
expect(maintenance).not.toContain("chatChannels.voice.");
|
||||||
expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,");
|
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", () => {
|
it("separates live voice recovery from the remaining periodic maintenance", () => {
|
||||||
|
|||||||
@@ -1203,6 +1203,40 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
|||||||
return { companyId, assignedAgentId, replacementAgentId };
|
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 () => {
|
it("retries a colliding company fixture prefix and creates a distinct complete fixture", async () => {
|
||||||
const existing = await seedCompany();
|
const existing = await seedCompany();
|
||||||
const [original] = await db.select().from(companies).where(eq(companies.id, existing.companyId));
|
const [original] = await db.select().from(companies).where(eq(companies.id, existing.companyId));
|
||||||
|
|||||||
@@ -1301,6 +1301,7 @@ export async function createApp(
|
|||||||
chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||||
chatChannels.voice.inbound.registerRetentionWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
chatChannels.voice.inbound.registerRetentionWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||||
chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||||
|
chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||||
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||||
emailChannels.start();
|
emailChannels.start();
|
||||||
const reconcileChatPublicationMaintenance = async () => {
|
const reconcileChatPublicationMaintenance = async () => {
|
||||||
|
|||||||
@@ -1255,7 +1255,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
|||||||
stageTerminalReceiptReactionRemovals, runtimeContextForRecord, scheduleMessageProcessing, processReceiptReaction,
|
stageTerminalReceiptReactionRemovals, runtimeContextForRecord, scheduleMessageProcessing, processReceiptReaction,
|
||||||
});
|
});
|
||||||
const { stageTeamsBoardFileIntent, deriveTeamsFileTransferAuthority } = createTeamsFileAuthority({ runtimeGeneration, credentialFingerprint, runtime, runtimeContexts, localRuntimeEpoch, runtimeVersions, lockCurrentPrincipalAuthorization, lifecycleRuntimeFence, assertFailedRetryReceipt, inboundWakeScope, failedChatRetrySource, isExplicitOperatorPublication, retryCommentIds });
|
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,
|
endpointRecord, runtimeContextForRecord, runtimeFor, runtimeContexts, localRuntimeEpoch, runtime,
|
||||||
deriveTeamsFileTransferAuthority, publicationFiles, scheduleMessageProcessing, processPendingPublications,
|
deriveTeamsFileTransferAuthority, publicationFiles, scheduleMessageProcessing, processPendingPublications,
|
||||||
withCredentialMutationLease, teamsFileRetryAfter, isShuttingDown: () => shuttingDown,
|
withCredentialMutationLease, teamsFileRetryAfter, isShuttingDown: () => shuttingDown,
|
||||||
@@ -3501,6 +3501,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
|||||||
const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor });
|
const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor });
|
||||||
const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
|
const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
|
||||||
return {
|
return {
|
||||||
|
registerTeamsConsentExpiryWork,
|
||||||
registerCommandWork,
|
registerCommandWork,
|
||||||
registerReceiverWork,
|
registerReceiverWork,
|
||||||
voice,
|
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 { createOutboundAttachments } from "../outbound-attachments.js";
|
||||||
|
|
||||||
import { createTeamsFileAuthority } from "./file-transfer-authority.js";
|
import { createTeamsFileAuthority } from "./file-transfer-authority.js";
|
||||||
@@ -14,7 +16,6 @@ import {
|
|||||||
eq,
|
eq,
|
||||||
inArray,
|
inArray,
|
||||||
lte,
|
lte,
|
||||||
or,
|
|
||||||
} from "drizzle-orm";
|
} from "drizzle-orm";
|
||||||
|
|
||||||
import type { Db } from "@paperclipai/db";
|
import type { Db } from "@paperclipai/db";
|
||||||
@@ -392,24 +393,11 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption
|
|||||||
.selectDistinct({ companyId: chatTeamsFileTransfers.companyId })
|
.selectDistinct({ companyId: chatTeamsFileTransfers.companyId })
|
||||||
.from(chatTeamsFileTransfers)
|
.from(chatTeamsFileTransfers)
|
||||||
.where(
|
.where(
|
||||||
or(
|
and(
|
||||||
and(
|
inArray(chatTeamsFileTransfers.phase, [
|
||||||
inArray(chatTeamsFileTransfers.phase, [
|
"consent_sending", "uploading", "file_info_sending", "conflict",
|
||||||
"consent_pending",
|
]),
|
||||||
"awaiting_consent",
|
lte(chatTeamsFileTransfers.attemptExpiresAt, new Date()),
|
||||||
"upload_pending",
|
|
||||||
]),
|
|
||||||
lte(chatTeamsFileTransfers.expiresAt, new Date()),
|
|
||||||
),
|
|
||||||
and(
|
|
||||||
inArray(chatTeamsFileTransfers.phase, [
|
|
||||||
"consent_sending",
|
|
||||||
"uploading",
|
|
||||||
"file_info_sending",
|
|
||||||
"conflict",
|
|
||||||
]),
|
|
||||||
lte(chatTeamsFileTransfers.attemptExpiresAt, new Date()),
|
|
||||||
),
|
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
const protocol = teamsTransferProtocol();
|
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.
|
// its row count and lock waits and performs no provider I/O.
|
||||||
for (const scope of scopes) {
|
for (const scope of scopes) {
|
||||||
if (isShuttingDown()) break;
|
if (isShuttingDown()) break;
|
||||||
await protocol.expireAndRecover(scope.companyId, 100);
|
await protocol.recoverAbandonedAttempts(scope.companyId, 100);
|
||||||
}
|
}
|
||||||
} catch {
|
} catch {
|
||||||
// Selection errors are closed here too; private transfer envelopes and
|
// 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 { createHash, randomUUID } from "node:crypto";
|
||||||
import {
|
import {
|
||||||
afterAll,
|
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 () => {
|
it("uses the exact attachment-derived filename for an unnamed source across issuance and restart", async () => {
|
||||||
const f = await fixture(60_000, null);
|
const f = await fixture(60_000, null);
|
||||||
const expected = `attachment-${f.a.attachmentId}`;
|
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 { createHash, randomUUID } from "node:crypto";
|
||||||
import { and, eq, gt, inArray, lte, or, sql } from "drizzle-orm";
|
import { and, eq, gt, inArray, lte, or, sql } from "drizzle-orm";
|
||||||
import { z } from "zod";
|
import { z } from "zod";
|
||||||
@@ -388,6 +389,7 @@ export function teamsFileTransferService(
|
|||||||
row: Row,
|
row: Row,
|
||||||
patch: Partial<typeof chatTeamsFileTransfers.$inferInsert>,
|
patch: Partial<typeof chatTeamsFileTransfers.$inferInsert>,
|
||||||
): Promise<Row> {
|
): Promise<Row> {
|
||||||
|
await notifyTeamsConsentExpiry(tx, row.id);
|
||||||
const [next] = await tx
|
const [next] = await tx
|
||||||
.update(chatTeamsFileTransfers)
|
.update(chatTeamsFileTransfers)
|
||||||
.set({ ...patch, version: row.version + 1, updatedAt: now() })
|
.set({ ...patch, version: row.version + 1, updatedAt: now() })
|
||||||
@@ -533,6 +535,7 @@ export function teamsFileTransferService(
|
|||||||
binding,
|
binding,
|
||||||
),
|
),
|
||||||
};
|
};
|
||||||
|
await notifyTeamsConsentExpiry(tx, id);
|
||||||
const [row] = await tx
|
const [row] = await tx
|
||||||
.insert(chatTeamsFileTransfers)
|
.insert(chatTeamsFileTransfers)
|
||||||
.values({
|
.values({
|
||||||
@@ -859,9 +862,32 @@ export function teamsFileTransferService(
|
|||||||
if (!row) fail();
|
if (!row) fail();
|
||||||
return summary(row);
|
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(
|
async function expireAndRecover(
|
||||||
companyId: string,
|
companyId: string,
|
||||||
limit = 25,
|
limit = 25,
|
||||||
|
reason: "all" | "attempts" = "all",
|
||||||
): Promise<TeamsFileRecoverySweepSummary> {
|
): Promise<TeamsFileRecoverySweepSummary> {
|
||||||
if (!z.uuid().safeParse(companyId).success) fail();
|
if (!z.uuid().safeParse(companyId).success) fail();
|
||||||
const size = Number.isSafeInteger(limit)
|
const size = Number.isSafeInteger(limit)
|
||||||
@@ -872,7 +898,8 @@ export function teamsFileTransferService(
|
|||||||
cursors = new Map();
|
cursors = new Map();
|
||||||
expiryRecoveryCursors.set(db, cursors);
|
expiryRecoveryCursors.set(db, cursors);
|
||||||
}
|
}
|
||||||
const cursor = cursors.get(companyId);
|
const cursorKey = `${companyId}:${reason}`;
|
||||||
|
const cursor = cursors.get(cursorKey);
|
||||||
const cutoff = now();
|
const cutoff = now();
|
||||||
const eligible = and(
|
const eligible = and(
|
||||||
eq(chatTeamsFileTransfers.companyId, companyId),
|
eq(chatTeamsFileTransfers.companyId, companyId),
|
||||||
@@ -881,14 +908,10 @@ export function teamsFileTransferService(
|
|||||||
inArray(chatTeamsFileTransfers.phase, [...sending, "conflict"]),
|
inArray(chatTeamsFileTransfers.phase, [...sending, "conflict"]),
|
||||||
lte(chatTeamsFileTransfers.attemptExpiresAt, cutoff),
|
lte(chatTeamsFileTransfers.attemptExpiresAt, cutoff),
|
||||||
),
|
),
|
||||||
and(
|
reason === "all" ? and(
|
||||||
inArray(chatTeamsFileTransfers.phase, [
|
inArray(chatTeamsFileTransfers.phase, unusedTeamsConsentPhases),
|
||||||
"consent_pending",
|
|
||||||
"awaiting_consent",
|
|
||||||
"upload_pending",
|
|
||||||
]),
|
|
||||||
lte(chatTeamsFileTransfers.expiresAt, cutoff),
|
lte(chatTeamsFileTransfers.expiresAt, cutoff),
|
||||||
),
|
) : undefined,
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
let rows: Array<{ id: string }>;
|
let rows: Array<{ id: string }>;
|
||||||
@@ -927,45 +950,11 @@ export function teamsFileTransferService(
|
|||||||
for (const candidate of rows) {
|
for (const candidate of rows) {
|
||||||
// Progress over rejected rows too. Neither corrupt evidence nor a held
|
// Progress over rejected rows too. Neither corrupt evidence nor a held
|
||||||
// publication lock can pin every later scan to the same oldest page.
|
// 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.cursor = candidate.id;
|
||||||
result.scanned++;
|
result.scanned++;
|
||||||
try {
|
try {
|
||||||
const recovered = await db.transaction(async (tx) => {
|
const recovered = await recoverTransfer(companyId, candidate.id, reason);
|
||||||
// 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;
|
|
||||||
});
|
|
||||||
if (recovered) result.recovered++;
|
if (recovered) result.recovered++;
|
||||||
} catch {
|
} catch {
|
||||||
// Keep the exact evidence and rollback any paired projection. Only
|
// Keep the exact evidence and rollback any paired projection. Only
|
||||||
@@ -1168,7 +1157,9 @@ export function teamsFileTransferService(
|
|||||||
recordConsent,
|
recordConsent,
|
||||||
process,
|
process,
|
||||||
get,
|
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,
|
cancel,
|
||||||
canCancelConflict,
|
canCancelConflict,
|
||||||
resolveInTransaction,
|
resolveInTransaction,
|
||||||
|
|||||||
@@ -46,6 +46,7 @@ export const DELIVERY_QUEUES = {
|
|||||||
voiceInboundRetention: "voice-inbound-retention",
|
voiceInboundRetention: "voice-inbound-retention",
|
||||||
voiceInboundCleanup: "voice-inbound-cleanup",
|
voiceInboundCleanup: "voice-inbound-cleanup",
|
||||||
voiceReplies: "voice-reply-delivery",
|
voiceReplies: "voice-reply-delivery",
|
||||||
|
chatTeamsConsentExpiry: "chat-teams-consent-expiry",
|
||||||
chatCommandRegistration: "chat-command-registration",
|
chatCommandRegistration: "chat-command-registration",
|
||||||
chatTelegramSubscriptions: "chat-telegram-subscriptions",
|
chatTelegramSubscriptions: "chat-telegram-subscriptions",
|
||||||
chatSessionSyncs: "chat-session-syncs",
|
chatSessionSyncs: "chat-session-syncs",
|
||||||
|
|||||||
Reference in new issue
Block a user