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/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/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..48569f3 --- /dev/null +++ b/apps/web/src/routes/api/internal/jobs/+server.ts @@ -0,0 +1,69 @@ +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'; +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 +// 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* 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.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..d4c9827 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,17 +71,18 @@ // 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 - // 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": [ @@ -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..0ee0510 100644 --- a/docs/release.md +++ b/docs/release.md @@ -28,22 +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. -5. In the Cloudflare dashboard, open **Images → Transformations**, select +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 + ``` + +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: @@ -55,6 +62,27 @@ Semantic search notes for the first deploy: with the database. Files whose embeddings are missing after a partial restore regenerate on reindex. +## Queues + +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: + +``` +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) {