220 lines
10 KiB
TypeScript
220 lines
10 KiB
TypeScript
import { HttpException, Logger } from '@nestjs/common';
|
|
import { Prisma, SmsAttemptCompletionWork } from '@prisma/client';
|
|
import { createHash, randomUUID } from 'node:crypto';
|
|
import { PrismaService } from '../prisma/prisma.service';
|
|
import { completionContext, CompletionRouteRequired } from './completion-context';
|
|
import type { RoutedChannel } from './send-chain.contracts';
|
|
import { countCompletion, observeCompletion } from './completion-metrics';
|
|
|
|
export type CompletionEventKind = 'receipt' | 'submit' | 'segment' | 'timeout' | 'rejection';
|
|
export class AttemptCompletion {
|
|
private readonly logger = new Logger(AttemptCompletion.name);
|
|
private timer?: ReturnType<typeof setInterval>;
|
|
private running = false;
|
|
constructor(
|
|
private readonly prisma: PrismaService,
|
|
private readonly execute: (kind: CompletionEventKind, payload: Prisma.JsonValue) => Promise<unknown>,
|
|
private readonly waitForRoute: (route: RoutedChannel) => Promise<void>,
|
|
) {}
|
|
|
|
start() {
|
|
this.timer = setInterval(() => void this.scan(), 5_000);
|
|
this.timer.unref();
|
|
void this.scan();
|
|
}
|
|
stop() {
|
|
if (this.timer) clearInterval(this.timer);
|
|
}
|
|
|
|
async enqueue(
|
|
messageRecordId: string,
|
|
sourceSubmitRecordId: string | undefined,
|
|
kind: CompletionEventKind,
|
|
payload: unknown,
|
|
identity?: string,
|
|
) {
|
|
const json = JSON.parse(JSON.stringify(payload)) as Prisma.InputJsonValue;
|
|
const workKey = sourceSubmitRecordId ? `attempt:${sourceSubmitRecordId}` : `message:${messageRecordId}`;
|
|
const eventKey = createHash('sha256')
|
|
.update(`${workKey}:${kind}:${identity ?? JSON.stringify(json)}`)
|
|
.digest('hex');
|
|
const work = await this.prisma.$transaction(async (tx) => {
|
|
const message = await tx.smsMessageRecord.findUniqueOrThrow({
|
|
where: { id: messageRecordId },
|
|
select: { tenantId: true },
|
|
});
|
|
if (sourceSubmitRecordId) {
|
|
const source = await tx.smsSubmitRecord.findUniqueOrThrow({ where: { id: sourceSubmitRecordId } });
|
|
if (source.messageRecordId !== messageRecordId || source.tenantId !== message.tenantId)
|
|
throw new Error('completion_source_mismatch');
|
|
}
|
|
await tx.smsAttemptCompletionWork.createMany({
|
|
data: [{ workKey, messageRecordId, sourceSubmitRecordId, tenantId: message.tenantId }],
|
|
skipDuplicates: true,
|
|
});
|
|
const current = await tx.smsAttemptCompletionWork.findUniqueOrThrow({ where: { workKey } });
|
|
if (current.messageRecordId !== messageRecordId) throw new Error('completion_work_mismatch');
|
|
await tx.$queryRaw`SELECT id FROM "SmsAttemptCompletionWork" WHERE id=${current.id} FOR UPDATE`;
|
|
const inserted = await tx.smsCompletionEvent.createMany({
|
|
data: [{ workId: current.id, eventKey, kind, payload: json }],
|
|
skipDuplicates: true,
|
|
});
|
|
if (inserted.count)
|
|
await tx.$executeRaw`
|
|
UPDATE "SmsAttemptCompletionWork" SET revision=revision+1,
|
|
state=CASE WHEN state IN ('processing', 'needs_review') THEN state ELSE 'pending' END,
|
|
"nextAttemptAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC'), "updatedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC') WHERE id=${current.id}`;
|
|
return current;
|
|
});
|
|
await this.process(work.id);
|
|
return this.prisma.smsMessageRecord.findUnique({ where: { id: messageRecordId } });
|
|
}
|
|
|
|
async scan() {
|
|
if (this.running) return;
|
|
this.running = true;
|
|
try {
|
|
observeCompletion(
|
|
await this.prisma.$queryRaw<Array<{ state: string; count: number; age: number }>>`
|
|
SELECT w.state, COUNT(*)::int AS count,
|
|
EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP AT TIME ZONE 'UTC') - MIN(COALESCE(e."createdAt", w."updatedAt")))::float AS age
|
|
FROM "SmsAttemptCompletionWork" w
|
|
LEFT JOIN LATERAL (SELECT MIN("createdAt") AS "createdAt" FROM "SmsCompletionEvent" WHERE "workId"=w.id AND "processedAt" IS NULL) e ON true
|
|
WHERE w.state IN ('pending','processing','retry_wait','needs_review') GROUP BY w.state`,
|
|
);
|
|
const rows = await this.prisma.$queryRaw<Array<{ id: string }>>`
|
|
SELECT id FROM "SmsAttemptCompletionWork"
|
|
WHERE (state IN ('pending','retry_wait') AND "nextAttemptAt" <= (CURRENT_TIMESTAMP AT TIME ZONE 'UTC'))
|
|
OR (state='processing' AND "leaseUntil" < (CURRENT_TIMESTAMP AT TIME ZONE 'UTC'))
|
|
ORDER BY "nextAttemptAt", id LIMIT 32`;
|
|
for (const row of rows) await this.process(row.id);
|
|
} catch {
|
|
this.logger.error('completion_scan_failed');
|
|
} finally {
|
|
this.running = false;
|
|
}
|
|
}
|
|
|
|
async process(id: string) {
|
|
const owner = randomUUID();
|
|
const claims = await this.prisma.$queryRaw<SmsAttemptCompletionWork[]>`
|
|
UPDATE "SmsAttemptCompletionWork" SET state='processing', "leaseOwner"=${owner},
|
|
"leaseUntil"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')+INTERVAL '60 seconds', "fenceVersion"="fenceVersion"+1,
|
|
attempts=attempts+1, "updatedAt"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')
|
|
WHERE id=${id} AND ((state IN ('pending','retry_wait') AND "nextAttemptAt" <= (CURRENT_TIMESTAMP AT TIME ZONE 'UTC'))
|
|
OR (state='processing' AND "leaseUntil" < (CURRENT_TIMESTAMP AT TIME ZONE 'UTC'))) RETURNING *`;
|
|
const claim = claims[0];
|
|
if (!claim) {
|
|
countCompletion('not_claimed');
|
|
return;
|
|
}
|
|
countCompletion('claimed');
|
|
if (claim.attempts > 1) countCompletion('recovered');
|
|
const renew = setInterval(() => {
|
|
void this.prisma
|
|
.$executeRaw`UPDATE "SmsAttemptCompletionWork" SET "leaseUntil"=(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')+INTERVAL '60 seconds'
|
|
WHERE id=${id} AND state='processing' AND "leaseOwner"=${owner} AND "fenceVersion"=${claim.fenceVersion}
|
|
AND "leaseUntil">(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')`.catch(() =>
|
|
this.logger.warn('completion_lease_renew_failed'),
|
|
);
|
|
}, 20_000);
|
|
renew.unref();
|
|
let route: RoutedChannel | undefined;
|
|
let routePlanned = false;
|
|
let planRevision: number | undefined;
|
|
try {
|
|
for (let pass = 0; pass < 64; pass++) {
|
|
try {
|
|
const done = await this.prisma.$transaction(
|
|
async (tx) => {
|
|
const rows = await tx.$queryRaw<SmsAttemptCompletionWork[]>`
|
|
SELECT * FROM "SmsAttemptCompletionWork" WHERE id=${id} AND state='processing'
|
|
AND "leaseOwner"=${owner} AND "fenceVersion"=${claim.fenceVersion} AND "leaseUntil">(CURRENT_TIMESTAMP AT TIME ZONE 'UTC') FOR UPDATE`;
|
|
const work = rows[0];
|
|
if (!work) {
|
|
countCompletion('fence_rejected');
|
|
throw new Error('completion_fence_rejected');
|
|
}
|
|
await tx.$queryRaw`SELECT id FROM "SmsMessageRecord" WHERE id=${work.messageRecordId} FOR UPDATE`;
|
|
if (planRevision !== work.revision) {
|
|
route = undefined;
|
|
routePlanned = false;
|
|
}
|
|
planRevision = work.revision;
|
|
const event = await tx.smsCompletionEvent.findFirst({
|
|
where: { workId: id, processedAt: null },
|
|
orderBy: [{ createdAt: 'asc' }, { id: 'asc' }],
|
|
});
|
|
if (event) {
|
|
await completionContext.run({ tx, messageRecordId: work.messageRecordId, route, routePlanned }, () =>
|
|
this.execute(event.kind as CompletionEventKind, event.payload),
|
|
);
|
|
await tx.smsCompletionEvent.update({ where: { id: event.id }, data: { processedAt: new Date() } });
|
|
}
|
|
const remaining = await tx.smsCompletionEvent.count({ where: { workId: id, processedAt: null } });
|
|
const message = await tx.smsMessageRecord.findUniqueOrThrow({ where: { id: work.messageRecordId } });
|
|
const retry = work.sourceSubmitRecordId
|
|
? await tx.smsSubmitRecord.findUnique({
|
|
where: { retryOfSubmitRecordId: work.sourceSubmitRecordId },
|
|
select: { id: true },
|
|
})
|
|
: null;
|
|
await tx.smsAttemptCompletionWork.update({
|
|
where: { id },
|
|
data: {
|
|
processedRevision: work.revision - remaining,
|
|
decision: message.status,
|
|
retrySubmitRecordId: retry?.id,
|
|
state: remaining ? 'processing' : 'idle',
|
|
lastError: null,
|
|
...(!remaining ? { leaseOwner: null, leaseUntil: null, attempts: 0 } : {}),
|
|
},
|
|
});
|
|
return remaining === 0;
|
|
},
|
|
{ timeout: 20_000, maxWait: 5_000 },
|
|
);
|
|
route = undefined;
|
|
routePlanned = false;
|
|
countCompletion('event_committed');
|
|
if (done) return;
|
|
} catch (error) {
|
|
if (!(error instanceof CompletionRouteRequired)) throw error;
|
|
// The probe transaction rolls back. Routing and rate limiting occur without locks.
|
|
try {
|
|
route = await error.select();
|
|
await this.waitForRoute(route);
|
|
} catch (selectionError) {
|
|
if (!(selectionError instanceof HttpException) || selectionError.getStatus() >= 500) throw selectionError;
|
|
route = undefined;
|
|
}
|
|
routePlanned = true;
|
|
}
|
|
}
|
|
throw new Error('completion_batch_budget_exhausted');
|
|
} catch (error) {
|
|
const code =
|
|
error instanceof Prisma.PrismaClientKnownRequestError
|
|
? error.code
|
|
: error instanceof Error && error.message.startsWith('completion_')
|
|
? error.message
|
|
: 'completion_processing_failed';
|
|
const exhausted = claim.attempts >= 12;
|
|
countCompletion(exhausted ? 'needs_review' : 'retry_wait');
|
|
await this.prisma.smsAttemptCompletionWork.updateMany({
|
|
where: { id, leaseOwner: owner, fenceVersion: claim.fenceVersion, state: 'processing' },
|
|
data: {
|
|
state: exhausted ? 'needs_review' : 'retry_wait',
|
|
leaseOwner: null,
|
|
leaseUntil: null,
|
|
lastError: code,
|
|
nextAttemptAt: new Date(Date.now() + Math.min(300_000, 1000 * 2 ** Math.min(claim.attempts, 8))),
|
|
},
|
|
});
|
|
this.logger.error(`${exhausted ? 'completion_needs_review' : 'completion_retry_wait'}:${code}`);
|
|
} finally {
|
|
clearInterval(renew);
|
|
}
|
|
}
|
|
}
|