FEATURE: Queue indexing and storage cleanup

apps/web/migrations-pg/0006_failed_jobs.sql

@@ -0,0 +1,20 @@
+-- migrate:up
+-- Jobs the queue gave up on. The Worker consumes the dead-letter queue
+-- into this table so an operator (and the admin UI) can see what failed
+-- and why, then re-send once the cause is fixed. Platform table: rows
+-- are looked up by org, never joined into tenant queries, so no RLS.
+CREATE TABLE failed_jobs (
+	id text PRIMARY KEY,
+	org_id text,
+	kind text NOT NULL,
+	payload jsonb NOT NULL,
+	error text NOT NULL,
+	attempts integer NOT NULL,
+	failed_at timestamptz NOT NULL DEFAULT now(),
+	resolved_at timestamptz
+);
+
+CREATE INDEX failed_jobs_org_idx ON failed_jobs (org_id, failed_at DESC);
+
+-- migrate:down
+DROP TABLE IF EXISTS failed_jobs;

apps/web/scripts/cloudflare-adapter.mjs

@@ -67,10 +67,16 @@ export default {
 		);
 	},
 	// Queue batches are forwarded to the SvelteKit bundle in-process (the
-	// consumer lives under $lib, unreachable from this facade). The endpoint
-	// returns one ack/retry decision per message id; anything it did not
-	// decide on is retried so a crash never silently drops work.
+	// consumer lives under $lib, unreachable from this facade). The main
+	// queue goes to the jobs endpoint, its dead-letter queue (named with a
+	// -dlq suffix in both environments) to the dead endpoint that records
+	// the job. Either returns one decision per message id, \`{ ack: true }\`
+	// or \`{ retry: true, delaySeconds }\`; anything it did not decide on is
+	// retried so a crash never silently drops work.
 	async queue(batch, env, ctx) {
+		const path = batch.queue.endsWith('-dlq')
+			? '/api/internal/jobs/dead'
+			: '/api/internal/jobs';
 		const timestamp = String(Date.now());
 		const body = JSON.stringify({
 			queue: batch.queue,
@@ -82,7 +88,7 @@ export default {
 		});
 		const signature = await jobsSignature(env.MAINTENANCE_SECRET, timestamp, body);
 		const response = await sveltekit.fetch(
-			new Request(new URL('/api/internal/jobs', env.DASHBOARD_ORIGIN), {
+			new Request(new URL(path, env.DASHBOARD_ORIGIN), {
 				method: 'POST',
 				headers: {
 					'Content-Type': 'application/json',
@@ -98,10 +104,13 @@ export default {
 			throw new Error(\`Queue consumer failed with status \${response.status}\`);
 		}
 		const { decisions } = await response.json();
-		const actions = new Map(decisions.map((decision) => [decision.id, decision.action]));
+		const byId = new Map(decisions.map((decision) => [decision.id, decision]));
 		for (const message of batch.messages) {
-			if (actions.get(message.id) === 'ack') message.ack();
-			else message.retry();
+			const decision = byId.get(message.id);
+			if (decision?.ack === true) message.ack();
+			else if (Number.isFinite(decision?.delaySeconds) && decision.delaySeconds > 0) {
+				message.retry({ delaySeconds: decision.delaySeconds });
+			} else message.retry();
 		}
 	}
 };

apps/web/src/lib/server/errors.ts

@@ -22,6 +22,12 @@ export class StorageError extends Data.TaggedError('StorageError')<{
 	readonly cause: unknown;
 }> {}
 
+// A program was asked to run for an org that no longer exists. Background
+// jobs treat this as done: there is nothing left to act on.
+export class OrgMissing extends Data.TaggedError('OrgMissing')<{
+	readonly orgId: string;
+}> {}
+
 export type AppError =
 	InvalidRequest | MisdirectedRequest | Unauthorized | NotFound | StorageError;
 

apps/web/src/lib/server/failed-jobs.ts

@@ -0,0 +1,127 @@
+import { JobSchema, type FailedJob } from '@adrive/shared';
+import type { PgClient } from '@effect/sql-pg';
+import { Effect, Schema } from 'effect';
+import { StorageError } from './errors';
+
+export interface DeadLetter {
+	readonly id: string;
+	readonly body: unknown;
+	readonly attempts: number;
+	readonly error: string;
+}
+
+const decodeJob = Schema.decodeUnknownOption(JobSchema);
+
+const FailedJobRow = Schema.Struct({
+	id: Schema.String,
+	kind: Schema.String,
+	payload: Schema.Unknown,
+	error: Schema.String,
+	attempts: Schema.Int,
+	failed_at: Schema.String,
+	resolved_at: Schema.NullOr(Schema.String)
+});
+
+const decodeRows = Schema.decodeUnknownOption(Schema.Array(FailedJobRow));
+
+// One row per dead-lettered message, keyed by the message id so the
+// at-least-once DLQ delivery cannot double-record. A body that no longer
+// decodes as a job is kept too (kind 'invalid', no org) so nothing the
+// queue rejected disappears.
+export const recordFailedJob = (sql: PgClient.PgClient, letter: DeadLetter) =>
+	Effect.gen(function* () {
+		const job = decodeJob(letter.body);
+		const orgId = job._tag === 'Some' ? job.value.orgId : null;
+		const kind = job._tag === 'Some' ? job.value.kind : 'invalid';
+		const rows = yield* sql<{ id: string }>`
+			INSERT INTO failed_jobs (id, org_id, kind, payload, error, attempts)
+			VALUES (
+				${letter.id}, ${orgId}, ${kind}, ${JSON.stringify(letter.body ?? null)}::jsonb,
+				${letter.error}, ${letter.attempts}
+			)
+			ON CONFLICT (id) DO NOTHING
+			RETURNING id`;
+		return { orgId, kind, recorded: rows.length === 1 };
+	}).pipe(
+		Effect.mapError(
+			(cause) => new StorageError({ operation: 'record failed job', cause })
+		)
+	);
+
+export const listFailedJobs = (
+	sql: PgClient.PgClient,
+	orgId: string,
+	limit = 100
+) =>
+	sql`
+		SELECT id, kind, payload, error, attempts, failed_at, resolved_at
+		FROM failed_jobs
+		WHERE org_id = ${orgId}
+		ORDER BY failed_at DESC, id
+		LIMIT ${Math.max(1, Math.min(limit, 500))}`.pipe(
+		Effect.map((rows): ReadonlyArray<FailedJob> => {
+			const decoded = decodeRows(rows);
+			return decoded._tag === 'Some'
+				? decoded.value.map((row) => ({
+						id: row.id,
+						kind: row.kind,
+						payload: row.payload,
+						error: row.error,
+						attempts: row.attempts,
+						failedAt: row.failed_at,
+						resolvedAt: row.resolved_at
+					}))
+				: [];
+		}),
+		Effect.mapError(
+			(cause) => new StorageError({ operation: 'list failed jobs', cause })
+		)
+	);
+
+// Optional operator alert. The URL is a secret rather than a var so it
+// stays out of wrangler.jsonc; absent means no alert. Delivery is best
+// effort: the rows are already committed and the batch is acked either
+// way, so a webhook outage cannot dead-letter the dead letters.
+export const alertWebhookUrl = (env: Env) => {
+	const value: unknown = Reflect.get(env, 'ALERT_WEBHOOK_URL');
+	return typeof value === 'string' && value.length > 0 ? value : null;
+};
+
+export const postFailedJobAlert = (
+	url: string,
+	summary: {
+		readonly queue: string;
+		readonly recorded: number;
+		readonly kinds: ReadonlyArray<string>;
+		readonly orgIds: ReadonlyArray<string>;
+	}
+) =>
+	Effect.tryPromise(async () => {
+		const response = await fetch(url, {
+			method: 'POST',
+			signal: AbortSignal.timeout(5_000),
+			headers: { 'content-type': 'application/json' },
+			body: JSON.stringify({
+				text: `adrive: ${summary.recorded} job(s) dead-lettered on ${summary.queue}`,
+				...summary
+			})
+		});
+		await response.body?.cancel();
+		return response;
+	}).pipe(
+		Effect.flatMap((response) =>
+			response.ok
+				? Effect.void
+				: Effect.fail(`Alert webhook returned HTTP ${response.status}`)
+		),
+		Effect.catchCause((cause) =>
+			Effect.sync(() => {
+				console.error(
+					JSON.stringify({
+						message: 'failed job alert could not be delivered',
+						cause: String(cause)
+					})
+				);
+			})
+		)
+	);

apps/web/src/lib/server/job-policy.ts

@@ -0,0 +1,27 @@
+// Timing rules for queue-driven work. Pure so the consumer and the
+// services that send jobs can be tested without a queue.
+
+// Cloudflare Queues cap a message delay at 12 hours; anything later is
+// re-sent by the consumer with the remainder when it arrives.
+export const MAX_JOB_DELAY_SECONDS = 12 * 60 * 60;
+
+// A job whose row is still pending or leased this long after it should
+// have run is presumed lost and re-sent by the cron reconciliation.
+export const STUCK_JOB_MS = 15 * 60 * 1_000;
+
+export const delaySecondsUntil = (at: string | number, now = Date.now()) => {
+	const target = typeof at === 'number' ? at : new Date(at).getTime();
+	if (!Number.isFinite(target)) return 0;
+	return Math.max(
+		0,
+		Math.min(Math.ceil((target - now) / 1_000), MAX_JOB_DELAY_SECONDS)
+	);
+};
+
+// Transient failures back off by delivery count (message.attempts is 1
+// on the first delivery): a minute, then doubling up to an hour.
+export const retryDelaySeconds = (attempts: number) =>
+	Math.min(30 * 2 ** Math.max(0, attempts), 3_600);
+
+export const stuckBefore = (now = Date.now()) =>
+	new Date(now - STUCK_JOB_MS).toISOString();

apps/web/src/lib/server/jobs/batch-request.ts

@@ -0,0 +1,59 @@
+import { Effect, Schema } from 'effect';
+import { AppConfig } from '../config';
+import { verifyJobsRequest } from '../cron-auth';
+import { InvalidRequest, Unauthorized } from '../errors';
+import { readBoundedText } from '../request-json';
+
+// The batch JSON the Worker facade posts for either queue: verified
+// against the HMAC over the exact body, then decoded.
+const JobBatchBody = Schema.Struct({
+	queue: Schema.String,
+	messages: Schema.Array(
+		Schema.Struct({
+			id: Schema.String,
+			attempts: Schema.Int,
+			body: Schema.Unknown
+		})
+	)
+});
+
+const MAX_BATCH_BYTES = 1024 * 1024;
+
+const invalid = (status: 400 | 413, message: string) =>
+	new InvalidRequest({ status, message });
+
+const unauthorized = () =>
+	new Unauthorized({ message: 'Queue request is unauthorized' });
+
+export const readSignedBatch = (request: Request) =>
+	Effect.gen(function* () {
+		const config = yield* AppConfig;
+		const text = yield* readBoundedText(request, {
+			maxBytes: MAX_BATCH_BYTES,
+			invalidLengthMessage: 'Queue batch is too large',
+			invalidTextMessage: 'Queue batch is invalid'
+		});
+		const authorized = yield* Effect.tryPromise({
+			try: () =>
+				verifyJobsRequest(
+					config.maintenanceSecret,
+					request.headers.get('x-adrive-jobs-time'),
+					text,
+					request.headers.get('x-adrive-jobs-signature')
+				),
+			catch: unauthorized
+		});
+		if (!authorized) return yield* unauthorized();
+
+		return yield* Effect.try({
+			try: (): unknown => JSON.parse(text),
+			catch: () => invalid(400, 'Queue batch is invalid')
+		}).pipe(
+			Effect.flatMap(Schema.decodeUnknownEffect(JobBatchBody)),
+			Effect.mapError((cause) =>
+				cause instanceof InvalidRequest
+					? cause
+					: invalid(400, 'Queue batch is invalid')
+			)
+		);
+	});