FEATURE: Add the Cloudflare Queues worker boundary

apps/web/scripts/cloudflare-adapter.mjs

@@ -15,7 +15,7 @@ export * from ${JSON.stringify(`./${svelteKitWorker}`)};
 const toHex = (bytes) =>
 	Array.from(new Uint8Array(bytes), (byte) => byte.toString(16).padStart(2, '0')).join('');
 
-const scheduledSignature = async (passcode, scheduledTime, cron) => {
+const hmacSign = async (passcode, message) => {
 	const key = await crypto.subtle.importKey(
 		'raw',
 		new TextEncoder().encode(passcode),
@@ -24,14 +24,17 @@ const scheduledSignature = async (passcode, scheduledTime, cron) => {
 		['sign']
 	);
 	return toHex(
-		await crypto.subtle.sign(
-			'HMAC',
-			key,
-			new TextEncoder().encode(\`\${scheduledTime}\\n\${cron}\`)
-		)
+		await crypto.subtle.sign('HMAC', key, new TextEncoder().encode(message))
 	);
 };
 
+const scheduledSignature = (passcode, scheduledTime, cron) =>
+	hmacSign(passcode, \`\${scheduledTime}\\n\${cron}\`);
+
+// Mirrors verifyJobsRequest in src/lib/server/cron-auth.ts.
+const jobsSignature = (passcode, timestamp, body) =>
+	hmacSign(passcode, \`jobs\\n\${timestamp}\\n\${body}\`);
+
 export default {
 	fetch(request, env, ctx) {
 		return sveltekit.fetch(request, env, ctx);
@@ -62,6 +65,44 @@ 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.
+	async queue(batch, env, ctx) {
+		const timestamp = String(Date.now());
+		const body = JSON.stringify({
+			queue: batch.queue,
+			messages: batch.messages.map((message) => ({
+				id: message.id,
+				attempts: message.attempts,
+				body: message.body
+			}))
+		});
+		const signature = await jobsSignature(env.PASSCODE, timestamp, body);
+		const response = await sveltekit.fetch(
+			new Request(new URL('/api/internal/jobs', env.DASHBOARD_ORIGIN), {
+				method: 'POST',
+				headers: {
+					'Content-Type': 'application/json',
+					'X-Adrive-Jobs-Time': timestamp,
+					'X-Adrive-Jobs-Signature': signature
+				},
+				body
+			}),
+			env,
+			ctx
+		);
+		if (!response.ok) {
+			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]));
+		for (const message of batch.messages) {
+			if (actions.get(message.id) === 'ack') message.ack();
+			else message.retry();
+		}
 	}
 };
 `.trimStart();

apps/web/src/lib/server/cron-auth.ts

@@ -58,3 +58,43 @@ export const verifyScheduledRequest = async (
 		scheduledMessage(scheduledTime, cron)
 	);
 };
+
+// Queue deliveries from the Worker facade are signed over the exact batch
+// JSON so a forged or altered body is rejected, not just a stale one.
+const jobsMessage = (timestamp: string, body: string) =>
+	new TextEncoder().encode(`jobs\n${timestamp}\n${body}`);
+
+export const signJobsRequest = async (
+	passcode: string,
+	timestamp: string,
+	body: string
+) => {
+	const key = await hmacKey(passcode, 'sign');
+	return bytesToHex(
+		await crypto.subtle.sign('HMAC', key, jobsMessage(timestamp, body))
+	);
+};
+
+export const verifyJobsRequest = async (
+	passcode: string,
+	timestamp: string | null,
+	body: string,
+	signature: string | null,
+	now = Date.now()
+) => {
+	if (!timestamp || !signature) return false;
+	const sentAt = Number(timestamp);
+	if (
+		!Number.isSafeInteger(sentAt) ||
+		Math.abs(now - sentAt) > SIGNATURE_WINDOW_MS
+	) {
+		return false;
+	}
+	const key = await hmacKey(passcode, 'verify');
+	return crypto.subtle.verify(
+		'HMAC',
+		key,
+		hexToBytes(signature),
+		jobsMessage(timestamp, body)
+	);
+};

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

@@ -30,6 +30,7 @@ import type { Embedder, VectorIndex } from './services/semantic';
 import type { Indexing } from './services/indexing';
 import type { Lifecycle } from './services/lifecycle';
 import type { GrantSecrets } from './services/grant-secrets';
+import type { JobQueue } from './services/jobs';
 
 export type AppServices =
 	| PgSql
@@ -45,7 +46,8 @@ export type AppServices =
 	| VectorIndex
 	| Indexing
 	| Lifecycle
-	| GrantSecrets;
+	| GrantSecrets
+	| JobQueue;
 
 export const isAppError = (failure: unknown): failure is AppError =>
 	failure instanceof InvalidRequest ||

apps/web/src/lib/server/jobs/consumer.ts

@@ -0,0 +1,117 @@
+import { JobSchema, type Job } from '@adrive/shared';
+import { Effect, Schema } from 'effect';
+import { runWorkerProgram, type AppServices } from '../edge';
+import type { StorageError } from '../errors';
+
+// The subset of a Cloudflare MessageBatch the consumer needs. A real
+// MessageBatch satisfies it, and so does the JSON the Worker facade posts
+// to /api/internal/jobs.
+export interface JobMessage {
+	readonly id: string;
+	readonly body: unknown;
+	readonly attempts: number;
+}
+
+export interface JobBatch {
+	readonly queue: string;
+	readonly messages: ReadonlyArray<JobMessage>;
+}
+
+export type JobAction = 'ack' | 'retry';
+
+export interface JobDecision {
+	readonly id: string;
+	readonly action: JobAction;
+}
+
+const log = (entry: Record<string, unknown>) =>
+	Effect.sync(() => {
+		console.log(JSON.stringify(entry));
+	});
+
+// Dispatch by job kind. Every branch only logs for now; the queue is wired
+// end to end before any behaviour moves onto it.
+export const runJob = (
+	job: Job
+): Effect.Effect<void, StorageError, AppServices> => {
+	switch (job.kind) {
+		case 'index':
+			// TODO(D2): indexing.runOne(job); ack when the version is stale.
+			return log({
+				message: 'job received',
+				kind: job.kind,
+				fileId: job.fileId,
+				version: job.version
+			});
+		case 'scan':
+			// TODO(D2): content scanning runs beside indexing.
+			return log({
+				message: 'job received',
+				kind: job.kind,
+				fileId: job.fileId,
+				version: job.version
+			});
+		case 'purge':
+			// TODO(D3): files.purgeOne(job.fileId) after the retention delay.
+			return log({
+				message: 'job received',
+				kind: job.kind,
+				fileId: job.fileId
+			});
+		case 'site-cleanup':
+			// TODO(D3): sites.cleanupSession(job.sessionId).
+			return log({
+				message: 'job received',
+				kind: job.kind,
+				sessionId: job.sessionId
+			});
+	}
+};
+
+const decodeJob = Schema.decodeUnknownEffect(JobSchema);
+
+// Invalid bodies are acked: retrying can never make them decode. Failed
+// jobs are retried; the queue's max_retries and dead-letter queue bound it.
+const consumeMessage = <R>(
+	queue: string,
+	message: JobMessage,
+	run: (job: Job) => Effect.Effect<void, StorageError, R>
+) =>
+	Effect.gen(function* () {
+		const decoded = yield* Effect.result(decodeJob(message.body));
+		if (decoded._tag === 'Failure') {
+			yield* log({
+				message: 'job message is invalid',
+				queue,
+				id: message.id,
+				cause: String(decoded.failure)
+			});
+			return 'ack' as const;
+		}
+		const outcome = yield* Effect.result(run(decoded.success));
+		if (outcome._tag === 'Failure') {
+			yield* log({
+				message: 'job failed',
+				queue,
+				id: message.id,
+				kind: decoded.success.kind,
+				attempts: message.attempts,
+				cause: String(outcome.failure.cause)
+			});
+			return 'retry' as const;
+		}
+		return 'ack' as const;
+	});
+
+export const consumeBatch = <R>(
+	batch: JobBatch,
+	run: (job: Job) => Effect.Effect<void, StorageError, R>
+) =>
+	Effect.forEach(batch.messages, (message) =>
+		consumeMessage(batch.queue, message, run).pipe(
+			Effect.map((action): JobDecision => ({ id: message.id, action }))
+		)
+	);
+
+export const handleJobBatch = (env: Env, batch: JobBatch) =>
+	runWorkerProgram(env, consumeBatch(batch, runJob));

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

@@ -2,7 +2,7 @@ import { Effect, Layer } from 'effect';
 import { ConfigLive } from './config';
 import { AuthGuardLive } from './services/auth-guard';
 import { AuthLive } from './services/auth';
-import { AuthGuardStore, Bucket, Pg } from './services/bindings';
+import { AuthGuardStore, Bucket, Jobs, Pg } from './services/bindings';
 import { pgLayer } from './pg';
 import { BlobsLive } from './services/blobs';
 import { FilesLive } from './services/files';
@@ -13,6 +13,7 @@ import { SemanticBindingsLive } from './services/semantic';
 import { IndexingLive } from './services/indexing';
 import { LifecycleLive } from './services/lifecycle';
 import { GrantSecretsLive } from './services/grant-secrets';
+import { JobQueueLive } from './services/jobs';
 
 export const PgLive = Layer.unwrap(
 	Effect.map(Pg, (hyperdrive) => pgLayer(hyperdrive))
@@ -23,11 +24,13 @@ export const requestLayer = (env: Env) => {
 		Layer.succeed(Pg, env.HYPERDRIVE),
 		Layer.succeed(Bucket, env.BUCKET),
 		Layer.succeed(AuthGuardStore, env.AUTH_GUARD),
+		Layer.succeed(Jobs, env.JOBS),
 		ConfigLive(env)
 	);
 	const pg = PgLive.pipe(Layer.provide(bindings));
 	const blobs = BlobsLive.pipe(Layer.provide(bindings));
-	const infrastructure = Layer.mergeAll(bindings, pg, blobs);
+	const jobQueue = JobQueueLive.pipe(Layer.provide(bindings));
+	const infrastructure = Layer.mergeAll(bindings, pg, blobs, jobQueue);
 	// The vector index reads and writes file_chunks, so it sits on Postgres
 	// like every other service; only the embedder still binds Workers AI.
 	const semantic = SemanticBindingsLive(env).pipe(

apps/web/src/lib/server/request-json.ts

@@ -7,6 +7,12 @@ interface BoundedJsonOptions {
 	readonly invalidJsonMessage: string;
 }
 
+interface BoundedTextOptions {
+	readonly maxBytes: number;
+	readonly invalidLengthMessage: string;
+	readonly invalidTextMessage: string;
+}
+
 type BodyReadResult =
 	| { readonly _tag: 'Success'; readonly bytes: Uint8Array }
 	| { readonly _tag: 'TooLarge' }
@@ -64,9 +70,9 @@ const readBodyAtMost = async (
 	return { _tag: 'Success', bytes };
 };
 
-export const readBoundedJson = (
+export const readBoundedText = (
 	request: Request,
-	{ maxBytes, invalidLengthMessage, invalidJsonMessage }: BoundedJsonOptions
+	{ maxBytes, invalidLengthMessage, invalidTextMessage }: BoundedTextOptions
 ) =>
 	Effect.gen(function* () {
 		// A declared length is validated up front so oversized requests fail
@@ -97,20 +103,31 @@ export const readBoundedJson = (
 			case 'ReadFailure':
 				return yield* new InvalidRequest({
 					status: 400,
-					message: invalidJsonMessage
+					message: invalidTextMessage
 				});
 			case 'Success':
-				return yield* Effect.try({
-					try: (): unknown => JSON.parse(new TextDecoder().decode(body.bytes)),
-					catch: () =>
-						new InvalidRequest({
-							status: 400,
-							message: invalidJsonMessage
-						})
-				});
+				return new TextDecoder().decode(body.bytes);
 		}
 	});
 
+export const readBoundedJson = (
+	request: Request,
+	{ maxBytes, invalidLengthMessage, invalidJsonMessage }: BoundedJsonOptions
+) =>
+	readBoundedText(request, {
+		maxBytes,
+		invalidLengthMessage,
+		invalidTextMessage: invalidJsonMessage
+	}).pipe(
+		Effect.flatMap((text) =>
+			Effect.try({
+				try: (): unknown => JSON.parse(text),
+				catch: () =>
+					new InvalidRequest({ status: 400, message: invalidJsonMessage })
+			})
+		)
+	);
+
 export const DEFAULT_JSON_BODY_LIMIT = 16 * 1024;
 
 export const decodeJson = <A, I>(

apps/web/src/lib/server/services/bindings.ts

@@ -1,3 +1,4 @@
+import type { Job } from '@adrive/shared';
 import { Context } from 'effect';
 
 // Hyperdrive binding. Only the connection string is used; the pool itself
@@ -20,3 +21,7 @@ export class AuthGuardStore extends Context.Service<
 	AuthGuardStore,
 	AuthGuardStoreShape
 >()('app/AuthGuardStore') {}
+
+// Producer side of the adrive-jobs queue. The consumer entry point lives
+// in lib/server/jobs/consumer.ts.
+export class Jobs extends Context.Service<Jobs, Queue<Job>>()('app/Jobs') {}