From fece36833640ac7903a480dfa379723d565f6826 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Wed, 9 Sep 2026 01:16:07 -0700 Subject: [PATCH 1/3] FEATURE: Add a Queues binding, job schema, and consumer facade Declares the adrive-jobs queue (producer JOBS plus consumer with a dead-letter queue) at the top level and in env.production, keeps the producer bindings covered by the drift check, and regenerates the Worker types. Adds the shared Job discriminated union and JobSchema, a JobQueue service (with a null layer) in the request layer, and a consumer that decodes each message, acks invalid bodies, and retries failed jobs. The Worker facade forwards queue batches to /api/internal/jobs in-process under an HMAC signed over the batch body, then acks or retries per message from the endpoint's decisions. No behaviour moves onto the queue yet; runJob only logs until D2/D3. Co-Authored-By: Claude Fable 5.1 --- apps/web/scripts/cloudflare-adapter.mjs | 53 +++++++- apps/web/src/lib/server/cron-auth.ts | 40 ++++++ apps/web/src/lib/server/edge.ts | 4 +- apps/web/src/lib/server/jobs/consumer.ts | 117 ++++++++++++++++++ apps/web/src/lib/server/layer.ts | 7 +- apps/web/src/lib/server/services/bindings.ts | 5 + apps/web/src/lib/server/services/jobs.ts | 42 +++++++ .../src/routes/api/internal/jobs/+server.ts | 74 +++++++++++ apps/web/worker-configuration.d.ts | 4 +- apps/web/wrangler.jsonc | 39 +++++- docs/release.md | 29 +++++ packages/shared/src/index.ts | 26 ++++ scripts/check-wrangler-drift.mjs | 6 + 13 files changed, 430 insertions(+), 16 deletions(-) create mode 100644 apps/web/src/lib/server/jobs/consumer.ts create mode 100644 apps/web/src/lib/server/services/jobs.ts create mode 100644 apps/web/src/routes/api/internal/jobs/+server.ts diff --git a/apps/web/scripts/cloudflare-adapter.mjs b/apps/web/scripts/cloudflare-adapter.mjs index 50b0fc1..e4824f9 100644 --- a/apps/web/scripts/cloudflare-adapter.mjs +++ b/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(); diff --git a/apps/web/src/lib/server/cron-auth.ts b/apps/web/src/lib/server/cron-auth.ts index 7cea608..6e92920 100644 --- a/apps/web/src/lib/server/cron-auth.ts +++ b/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) + ); +}; diff --git a/apps/web/src/lib/server/edge.ts b/apps/web/src/lib/server/edge.ts index 9a1d216..2161055 100644 --- a/apps/web/src/lib/server/edge.ts +++ b/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 || diff --git a/apps/web/src/lib/server/jobs/consumer.ts b/apps/web/src/lib/server/jobs/consumer.ts new file mode 100644 index 0000000..d57487b --- /dev/null +++ b/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; +} + +export type JobAction = 'ack' | 'retry'; + +export interface JobDecision { + readonly id: string; + readonly action: JobAction; +} + +const log = (entry: Record) => + 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 => { + 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 = ( + queue: string, + message: JobMessage, + run: (job: Job) => Effect.Effect +) => + 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 = ( + batch: JobBatch, + run: (job: Job) => Effect.Effect +) => + 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)); diff --git a/apps/web/src/lib/server/layer.ts b/apps/web/src/lib/server/layer.ts index 0bc1c66..5a33859 100644 --- a/apps/web/src/lib/server/layer.ts +++ b/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( diff --git a/apps/web/src/lib/server/services/bindings.ts b/apps/web/src/lib/server/services/bindings.ts index ee7a95b..9d146e7 100644 --- a/apps/web/src/lib/server/services/bindings.ts +++ b/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>()('app/Jobs') {} diff --git a/apps/web/src/lib/server/services/jobs.ts b/apps/web/src/lib/server/services/jobs.ts new file mode 100644 index 0000000..7d4bc8a --- /dev/null +++ b/apps/web/src/lib/server/services/jobs.ts @@ -0,0 +1,42 @@ +import type { Job } from '@adrive/shared'; +import { Context, Effect, Layer } from 'effect'; +import { StorageError } from '../errors'; +import { Jobs } from './bindings'; + +export interface JobSendOptions { + readonly delaySeconds?: number; +} + +export interface JobQueueShape { + readonly send: ( + job: Job, + options?: JobSendOptions + ) => Effect.Effect; +} + +export class JobQueue extends Context.Service()( + 'app/JobQueue' +) {} + +const makeJobQueue = Effect.gen(function* () { + const queue = yield* Jobs; + + return JobQueue.of({ + send: Effect.fn('JobQueue.send')(function* (job, options) { + yield* Effect.tryPromise({ + try: () => queue.send(job, { contentType: 'json', ...options }), + catch: (cause) => + new StorageError({ operation: `enqueue ${job.kind} job`, cause }) + }); + }) + }); +}); + +export const JobQueueLive = Layer.effect(JobQueue, makeJobQueue); + +// For tests and environments without the JOBS binding: sends succeed and +// go nowhere. +export const JobQueueNull = Layer.succeed( + JobQueue, + JobQueue.of({ send: () => Effect.void }) +); diff --git a/apps/web/src/routes/api/internal/jobs/+server.ts b/apps/web/src/routes/api/internal/jobs/+server.ts new file mode 100644 index 0000000..43c7ccb --- /dev/null +++ b/apps/web/src/routes/api/internal/jobs/+server.ts @@ -0,0 +1,74 @@ +import type { RequestHandler } from './$types'; +import { Effect, Schema } from 'effect'; +import { AppConfig } from '$lib/server/config'; +import { verifyJobsRequest } from '$lib/server/cron-auth'; +import { runEdge } from '$lib/server/edge'; +import { InvalidRequest, Unauthorized } from '$lib/server/errors'; +import { consumeBatch, runJob } from '$lib/server/jobs/consumer'; + +// The Worker facade (scripts/cloudflare-adapter.mjs) receives queue +// batches and forwards them here in-process, because the consumer has to +// run inside the SvelteKit bundle to reach $lib. The response tells the +// facade which messages to ack and which to retry. +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 unauthorized = () => + new Unauthorized({ message: 'Queue request is unauthorized' }); + +export const POST: RequestHandler = ({ request }) => + runEdge( + Effect.gen(function* () { + const config = yield* AppConfig; + const text = yield* Effect.tryPromise({ + try: () => request.text(), + catch: () => + new InvalidRequest({ status: 400, message: 'Queue batch is invalid' }) + }); + if (text.length > MAX_BATCH_BYTES) { + return yield* new InvalidRequest({ + status: 413, + message: 'Queue batch is too large' + }); + } + const authorized = yield* Effect.tryPromise({ + try: () => + verifyJobsRequest( + config.passcode, + request.headers.get('x-adrive-jobs-time'), + text, + request.headers.get('x-adrive-jobs-signature') + ), + catch: unauthorized + }); + if (!authorized) return yield* unauthorized(); + + const batch = yield* Effect.try({ + try: (): unknown => JSON.parse(text), + catch: () => + new InvalidRequest({ status: 400, message: 'Queue batch is invalid' }) + }).pipe( + Effect.flatMap(Schema.decodeUnknownEffect(JobBatchBody)), + Effect.mapError((cause) => + cause instanceof InvalidRequest + ? cause + : new InvalidRequest({ + status: 400, + message: 'Queue batch is invalid' + }) + ) + ); + const decisions = yield* consumeBatch(batch, runJob); + return Response.json({ decisions }); + }) + ); diff --git a/apps/web/worker-configuration.d.ts b/apps/web/worker-configuration.d.ts index e024ca9..a2b6197 100644 --- a/apps/web/worker-configuration.d.ts +++ b/apps/web/worker-configuration.d.ts @@ -1,10 +1,11 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: caead1cb99063067c29ee1f0ece3f1ab) +// Generated by Wrangler by running `wrangler types` (hash: 6c913b5c836dfb6c0be90009a078b25d) // Runtime types generated with workerd@1.20260722.1 2026-07-27 global_fetch_strictly_public,nodejs_compat interface __BaseEnv_Env { AUTH_GUARD: KVNamespace; BUCKET: R2Bucket; HYPERDRIVE: Hyperdrive; + JOBS: Queue; BROWSER: BrowserRun; AI?: Ai; ASSETS: Fetcher; @@ -24,6 +25,7 @@ declare namespace Cloudflare { AUTH_GUARD: KVNamespace; BUCKET: R2Bucket; HYPERDRIVE: Hyperdrive; + JOBS: Queue; BROWSER: BrowserRun; AI: Ai; ASSETS: Fetcher; diff --git a/apps/web/wrangler.jsonc b/apps/web/wrangler.jsonc index ca7d9b3..238b651 100644 --- a/apps/web/wrangler.jsonc +++ b/apps/web/wrangler.jsonc @@ -45,6 +45,19 @@ "id": "419a52adcd9649e0a24e5c47c2748477" } ], + "queues": { + "producers": [{ "queue": "adrive-jobs", "binding": "JOBS" }], + "consumers": [ + { + "queue": "adrive-jobs", + "max_batch_size": 5, + "max_batch_timeout": 5, + "max_retries": 5, + "max_concurrency": 10, + "dead_letter_queue": "adrive-jobs-dlq" + } + ] + }, "observability": { "enabled": true, "logs": { @@ -58,13 +71,14 @@ // Production deploys with `wrangler deploy --env production`. Wrangler // environments do not inherit bindings, so everything is redeclared. // Before the first deploy (see docs/release.md): - // 1. Create the production D1 database and paste its id below. - // 1b. Create a Hyperdrive config pointing at PlanetScale Postgres - // (direct port 5432, caching disabled) and paste its id below. + // 1. Create a Hyperdrive config pointing at PlanetScale Postgres + // (direct port 5432, caching disabled) and paste its id below. // 2. Create the production R2 bucket and KV namespace; paste the KV id. - // 3. Point DNS for both hostnames at the Worker (routes below). - // 4. `wrangler secret put PASSCODE --env production` - // 5. `DATABASE_URL=postgres://... bun db:pg:migrate` (see docs/release.md) + // 3. Create the production job queue and its dead-letter queue + // (docs/release.md, "Queues"). + // 4. Point DNS for both hostnames at the Worker (routes below). + // 5. `wrangler secret put PASSCODE --env production` + // 6. `DATABASE_URL=postgres://... bun db:pg:migrate` (see docs/release.md) "env": { "production": { // Place the Worker near D1 for dashboard/search queries. Content @@ -114,6 +128,19 @@ "id": "09b884a95e5e40389d8554b3cd690d3f" } ], + "queues": { + "producers": [{ "queue": "adrive-jobs-production", "binding": "JOBS" }], + "consumers": [ + { + "queue": "adrive-jobs-production", + "max_batch_size": 5, + "max_batch_timeout": 5, + "max_retries": 5, + "max_concurrency": 10, + "dead_letter_queue": "adrive-jobs-production-dlq" + } + ] + }, "observability": { "enabled": true, "logs": { diff --git a/docs/release.md b/docs/release.md index 180f8a3..2320799 100644 --- a/docs/release.md +++ b/docs/release.md @@ -28,6 +28,14 @@ the repository root. loudly if the `AI` binding is missing. Embeddings run within the Workers Paid plan's included neuron allocation at personal scale. +4b. From `apps/web`: create the job queue and its dead-letter queue (see +"Queues" below): + +``` +wrangler queues create adrive-jobs-production +wrangler queues create adrive-jobs-production-dlq +``` + 5. In the Cloudflare dashboard, open **Images → Transformations**, select the zone that owns `CONTENT_ORIGIN` (`davis7.space` for `files.davis7.space`), and enable transformations. Dashboard thumbnails @@ -55,6 +63,27 @@ Semantic search notes for the first deploy: with the database. Files whose embeddings are missing after a partial restore regenerate on reindex. +## Queues + +Background jobs (indexing, purges, site cleanup) travel over one Cloudflare +Queue per environment. `wrangler.jsonc` binds it as `JOBS` and declares +the Worker as its consumer; the queue itself is created once: + +``` +wrangler queues create adrive-jobs-production +wrangler queues create adrive-jobs-production-dlq +``` + +- The consumer retries a failed message up to `max_retries` (5) times, + then moves it to `adrive-jobs-production-dlq`. Messages whose body does + not decode as a job are acked and logged, never retried. +- Inspect the dead-letter queue with + `wrangler queues consumer` tooling or the dashboard; nothing drains it + automatically. Re-send a message from the DLQ only after fixing the + cause, since the same job will otherwise fail again. +- Local development uses the `adrive-jobs` / `adrive-jobs-dlq` names and + needs no provisioning; `wrangler dev` simulates the queue. + ## Releasing ``` diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index ace89a7..17768d2 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -378,3 +378,29 @@ export const ErrorResponseSchema = Schema.Struct({ }); export const API_KEY_PATTERN = /^adr_([A-Za-z0-9]{8})_([A-Za-z0-9_-]{24,})$/; + +// Background work carried over the Cloudflare Queue. Tenancy adds orgId +// to every variant later; keep the union closed so the consumer's switch +// stays exhaustive. +export const JobSchema = Schema.Union([ + Schema.Struct({ + kind: Schema.Literal('index'), + fileId: Schema.String, + version: Schema.Int + }), + Schema.Struct({ + kind: Schema.Literal('purge'), + fileId: Schema.String + }), + Schema.Struct({ + kind: Schema.Literal('scan'), + fileId: Schema.String, + version: Schema.Int + }), + Schema.Struct({ + kind: Schema.Literal('site-cleanup'), + sessionId: Schema.String + }) +]); + +export type Job = typeof JobSchema.Type; diff --git a/scripts/check-wrangler-drift.mjs b/scripts/check-wrangler-drift.mjs index e011145..9954b8b 100644 --- a/scripts/check-wrangler-drift.mjs +++ b/scripts/check-wrangler-drift.mjs @@ -181,6 +181,12 @@ const main = () => { ); } } + const queueProducers = (block) => bindingNames(block.queues?.producers); + if (!sameJson(queueProducers(config), queueProducers(prod))) { + drift.push( + `queues.producers bindings: local=${JSON.stringify(queueProducers(config))} production=${JSON.stringify(queueProducers(prod))}` + ); + } } if (drift.length > 0) { From 4a243700f4f425c81c9c9f2f36695805a9dd98b6 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Wed, 9 Sep 2026 02:43:12 -0700 Subject: [PATCH 2/3] CHORE: Drop stale D1 comments from the wrangler config Co-Authored-By: Claude Fable 5.1 --- apps/web/wrangler.jsonc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/web/wrangler.jsonc b/apps/web/wrangler.jsonc index 238b651..d4c9827 100644 --- a/apps/web/wrangler.jsonc +++ b/apps/web/wrangler.jsonc @@ -81,8 +81,8 @@ // 6. `DATABASE_URL=postgres://... bun db:pg:migrate` (see docs/release.md) "env": { "production": { - // Place the Worker near D1 for dashboard/search queries. Content - // routes still check visibility in D1, then serve from R2 or the + // Place the Worker near the database for dashboard/search queries. Content + // routes still check visibility in Postgres, then serve from R2 or the // Cache API. "placement": { "mode": "smart" }, "routes": [ From 83a750a57ff2c6fa7abf0eecf5c85c25373695e1 Mon Sep 17 00:00:00 2001 From: Ben Davis <45952064+bmdavis419@users.noreply.github.com> Date: Thu, 10 Sep 2026 22:22:33 -0700 Subject: [PATCH 3/3] BUGFIX: Bound queue requests before signature verification --- apps/web/src/lib/server/request-json.ts | 39 +++++++++++++------ .../src/routes/api/internal/jobs/+server.ts | 15 +++---- docs/release.md | 29 +++++++------- 3 files changed, 47 insertions(+), 36 deletions(-) diff --git a/apps/web/src/lib/server/request-json.ts b/apps/web/src/lib/server/request-json.ts index 656156b..ba2492b 100644 --- a/apps/web/src/lib/server/request-json.ts +++ b/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 = ( diff --git a/apps/web/src/routes/api/internal/jobs/+server.ts b/apps/web/src/routes/api/internal/jobs/+server.ts index 43c7ccb..48569f3 100644 --- a/apps/web/src/routes/api/internal/jobs/+server.ts +++ b/apps/web/src/routes/api/internal/jobs/+server.ts @@ -5,6 +5,7 @@ import { verifyJobsRequest } from '$lib/server/cron-auth'; import { runEdge } from '$lib/server/edge'; import { InvalidRequest, Unauthorized } from '$lib/server/errors'; import { consumeBatch, runJob } from '$lib/server/jobs/consumer'; +import { readBoundedText } from '$lib/server/request-json'; // The Worker facade (scripts/cloudflare-adapter.mjs) receives queue // batches and forwards them here in-process, because the consumer has to @@ -30,17 +31,11 @@ export const POST: RequestHandler = ({ request }) => runEdge( Effect.gen(function* () { const config = yield* AppConfig; - const text = yield* Effect.tryPromise({ - try: () => request.text(), - catch: () => - new InvalidRequest({ status: 400, message: 'Queue batch is invalid' }) + const text = yield* readBoundedText(request, { + maxBytes: MAX_BATCH_BYTES, + invalidLengthMessage: 'Queue batch is too large', + invalidTextMessage: 'Queue batch is invalid' }); - if (text.length > MAX_BATCH_BYTES) { - return yield* new InvalidRequest({ - status: 413, - message: 'Queue batch is too large' - }); - } const authorized = yield* Effect.tryPromise({ try: () => verifyJobsRequest( diff --git a/docs/release.md b/docs/release.md index 2320799..0ee0510 100644 --- a/docs/release.md +++ b/docs/release.md @@ -28,30 +28,29 @@ the repository root. loudly if the `AI` binding is missing. Embeddings run within the Workers Paid plan's included neuron allocation at personal scale. -4b. From `apps/web`: create the job queue and its dead-letter queue (see -"Queues" below): +5. From `apps/web`: create the job queue and its dead-letter queue: -``` -wrangler queues create adrive-jobs-production -wrangler queues create adrive-jobs-production-dlq -``` + ``` + wrangler queues create adrive-jobs-production + wrangler queues create adrive-jobs-production-dlq + ``` -5. In the Cloudflare dashboard, open **Images → Transformations**, select +6. In the Cloudflare dashboard, open **Images → Transformations**, select the zone that owns `CONTENT_ORIGIN` (`davis7.space` for `files.davis7.space`), and enable transformations. Dashboard thumbnails require this zone-level setting. -6. From `apps/web`: `wrangler secret put PASSCODE --env production` +7. From `apps/web`: `wrangler secret put PASSCODE --env production` (12+ characters). -7. The `davis7.space` zone must be active in Cloudflare. Remove existing +8. The `davis7.space` zone must be active in Cloudflare. Remove existing CNAME records for `drive.davis7.space`, `files.davis7.space`, and `adrive.davis7.space` before deployment; the custom-domain routes in the `wrangler.jsonc` files create the required DNS records automatically. `adrive.davis7.space` serves the static landing page (`apps/site`, an assets-only Worker with no build step). -8. From the repo root: `bun release` -9. From the repo root: set up backups on your backup host - (`scripts/backup/install-backup-host.sh`) and complete the restore drill - in `docs/backup-restore.md`. +9. From the repo root: `bun release` +10. From the repo root: set up backups on your backup host + (`scripts/backup/install-backup-host.sh`) and complete the restore drill + in `docs/backup-restore.md`. Semantic search notes for the first deploy: @@ -65,8 +64,8 @@ Semantic search notes for the first deploy: ## Queues -Background jobs (indexing, purges, site cleanup) travel over one Cloudflare -Queue per environment. `wrangler.jsonc` binds it as `JOBS` and declares +The `JOBS` binding provides one Cloudflare Queue per environment for +background indexing, purges, and site cleanup. `wrangler.jsonc` declares the Worker as its consumer; the queue itself is created once: ```