diff --git a/.changeset/contrail-dev-command.md b/.changeset/contrail-dev-command.md new file mode 100644 index 0000000..5e5e715 --- /dev/null +++ b/.changeset/contrail-dev-command.md @@ -0,0 +1,19 @@ +--- +"@atmo-dev/contrail": minor +--- + +add `contrail dev` — local dev wrapper for cloudflare workers deployments. + +replaces `wrangler dev --test-scheduled` + a separate cron-trigger script with one command. on start it: + +1. connects to your local D1 via wrangler's `getPlatformProxy`, inspects state +2. prompts to run `backfillAll` if no completed backfills exist yet +3. prompts to run `refresh` if the ingest cursor is older than 60 minutes (configurable with `--stale-after`) +4. spawns `wrangler dev --test-scheduled` +5. fires `GET /__scheduled?cron=...` every 60 seconds so the cron actually runs in local dev (wrangler's scheduler only works in deployed production) + +flags: `--cron ` (default `"*/1 * * * *"`), `--stale-after ` (default 60), `--yes` to auto-accept prompts, plus the standard `--config` / `--root` / `--binding`. + +prompts are skipped in non-TTY environments (default-declined). + +also adds `--yes` to the CLI-wide arg parser. diff --git a/README.md b/README.md index 84a3ffb..a360ae0 100644 --- a/README.md +++ b/README.md @@ -96,7 +96,6 @@ returns every `community.lexicon.calendar.event` record published anywhere on at - [Spaces](./docs/05-spaces.md) — permissioned records stored by the appview - [Communities](./docs/06-communities.md) — group-controlled atproto DIDs - [Sync](./docs/07-sync.md) — reactive client-side store over `watchRecords` -- [Examples](./docs/08-examples.md) — reference deployments in the repo - Frameworks: [SvelteKit + Cloudflare](./docs/frameworks/sveltekit-cloudflare.md) ## Packages diff --git a/apps/cloudflare-workers/README.md b/apps/cloudflare-workers/README.md index 7790687..1145387 100644 --- a/apps/cloudflare-workers/README.md +++ b/apps/cloudflare-workers/README.md @@ -17,8 +17,8 @@ backfills run via the `contrail` cli from the library (see `package.json` script ```bash pnpm install npx wrangler d1 create contrail # copy database_id into wrangler.jsonc -pnpm deploy # deploy the worker pnpm contrail backfill --remote # discover + backfill historical events +pnpm deploy # deploy the worker ``` then hit: @@ -30,10 +30,12 @@ GET https://.workers.dev/xrpc/com.example.event.listRecords?startsA ## local dev ```bash -pnpm dev # wrangler dev, cron fires every minute +pnpm dev # wraps wrangler dev + auto-fires cron + prompts for backfill/refresh if needed pnpm contrail backfill # backfill against the local D1 created by wrangler ``` +`pnpm dev` runs `contrail dev` under the hood. On start it inspects the local D1 and prompts to run backfill (if nothing's indexed yet) or refresh (if the ingest cursor is >1h old), then runs `wrangler dev --test-scheduled` with a 60s timer hitting `/__scheduled` so the cron actually fires locally. + ## extending - **add a collection:** append to `collections` in `src/contrail.config.ts`; redeploy; `pnpm contrail backfill --remote` to backfill the new one. diff --git a/apps/cloudflare-workers/lexicons/generated/com/example/event/getRecord.json b/apps/cloudflare-workers/lexicons/generated/com/example/event/getRecord.json index 3d73338..47c29e1 100644 --- a/apps/cloudflare-workers/lexicons/generated/com/example/event/getRecord.json +++ b/apps/cloudflare-workers/lexicons/generated/com/example/event/getRecord.json @@ -28,7 +28,11 @@ "type": "object", "required": [ "uri", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/cloudflare-workers/lexicons/generated/com/example/event/listRecords.json b/apps/cloudflare-workers/lexicons/generated/com/example/event/listRecords.json index cbfe275..3191ac3 100644 --- a/apps/cloudflare-workers/lexicons/generated/com/example/event/listRecords.json +++ b/apps/cloudflare-workers/lexicons/generated/com/example/event/listRecords.json @@ -89,7 +89,11 @@ "required": [ "uri", "cid", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/cloudflare-workers/package.json b/apps/cloudflare-workers/package.json index 48b55a0..99d2a6b 100644 --- a/apps/cloudflare-workers/package.json +++ b/apps/cloudflare-workers/package.json @@ -4,7 +4,7 @@ "private": true, "type": "module", "scripts": { - "dev": "wrangler dev --test-scheduled", + "dev": "contrail dev", "deploy": "wrangler deploy" }, "dependencies": { diff --git a/apps/group-chat/lexicons/generated/tools/atmo/chat/channel/getRecord.json b/apps/group-chat/lexicons/generated/tools/atmo/chat/channel/getRecord.json index 3d6c40f..ee2685a 100644 --- a/apps/group-chat/lexicons/generated/tools/atmo/chat/channel/getRecord.json +++ b/apps/group-chat/lexicons/generated/tools/atmo/chat/channel/getRecord.json @@ -37,7 +37,11 @@ "type": "object", "required": [ "uri", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/group-chat/lexicons/generated/tools/atmo/chat/channel/listRecords.json b/apps/group-chat/lexicons/generated/tools/atmo/chat/channel/listRecords.json index 903635a..a752e86 100644 --- a/apps/group-chat/lexicons/generated/tools/atmo/chat/channel/listRecords.json +++ b/apps/group-chat/lexicons/generated/tools/atmo/chat/channel/listRecords.json @@ -100,7 +100,11 @@ "required": [ "uri", "cid", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/group-chat/lexicons/generated/tools/atmo/chat/message/getRecord.json b/apps/group-chat/lexicons/generated/tools/atmo/chat/message/getRecord.json index 2206480..33f728e 100644 --- a/apps/group-chat/lexicons/generated/tools/atmo/chat/message/getRecord.json +++ b/apps/group-chat/lexicons/generated/tools/atmo/chat/message/getRecord.json @@ -37,7 +37,11 @@ "type": "object", "required": [ "uri", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/group-chat/lexicons/generated/tools/atmo/chat/message/listRecords.json b/apps/group-chat/lexicons/generated/tools/atmo/chat/message/listRecords.json index c956d72..c1ddc12 100644 --- a/apps/group-chat/lexicons/generated/tools/atmo/chat/message/listRecords.json +++ b/apps/group-chat/lexicons/generated/tools/atmo/chat/message/listRecords.json @@ -99,7 +99,11 @@ "required": [ "uri", "cid", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/group-chat/lexicons/generated/tools/atmo/chat/server/getRecord.json b/apps/group-chat/lexicons/generated/tools/atmo/chat/server/getRecord.json index 1a15c83..dcfc3ef 100644 --- a/apps/group-chat/lexicons/generated/tools/atmo/chat/server/getRecord.json +++ b/apps/group-chat/lexicons/generated/tools/atmo/chat/server/getRecord.json @@ -37,7 +37,11 @@ "type": "object", "required": [ "uri", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/group-chat/lexicons/generated/tools/atmo/chat/server/listRecords.json b/apps/group-chat/lexicons/generated/tools/atmo/chat/server/listRecords.json index aa145f6..dd2e09d 100644 --- a/apps/group-chat/lexicons/generated/tools/atmo/chat/server/listRecords.json +++ b/apps/group-chat/lexicons/generated/tools/atmo/chat/server/listRecords.json @@ -95,7 +95,11 @@ "required": [ "uri", "cid", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/sveltekit-cloudflare-workers/lexicons/generated/statusphere/app/status/getRecord.json b/apps/sveltekit-cloudflare-workers/lexicons/generated/statusphere/app/status/getRecord.json index 8cb1446..c866e5e 100644 --- a/apps/sveltekit-cloudflare-workers/lexicons/generated/statusphere/app/status/getRecord.json +++ b/apps/sveltekit-cloudflare-workers/lexicons/generated/statusphere/app/status/getRecord.json @@ -28,7 +28,11 @@ "type": "object", "required": [ "uri", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/sveltekit-cloudflare-workers/lexicons/generated/statusphere/app/status/listRecords.json b/apps/sveltekit-cloudflare-workers/lexicons/generated/statusphere/app/status/listRecords.json index 9fea6dd..762f83f 100644 --- a/apps/sveltekit-cloudflare-workers/lexicons/generated/statusphere/app/status/listRecords.json +++ b/apps/sveltekit-cloudflare-workers/lexicons/generated/statusphere/app/status/listRecords.json @@ -90,7 +90,11 @@ "required": [ "uri", "cid", - "value" + "value", + "did", + "collection", + "rkey", + "time_us" ], "properties": { "uri": { diff --git a/apps/sveltekit-cloudflare-workers/package.json b/apps/sveltekit-cloudflare-workers/package.json index 939aba9..b9924c5 100644 --- a/apps/sveltekit-cloudflare-workers/package.json +++ b/apps/sveltekit-cloudflare-workers/package.json @@ -5,11 +5,13 @@ "type": "module", "scripts": { "dev": "vite dev", - "build": "tsx scripts/generate.ts && lex-cli generate && vite build && tsx scripts/append-scheduled.ts", - "generate": "tsx scripts/generate.ts", - "generate:pull": "tsx scripts/generate.ts && lex-cli pull && tsx scripts/generate.ts && lex-cli pull && lex-cli generate", - "sync": "tsx scripts/sync.ts", - "sync:remote": "tsx scripts/sync.ts --remote", + "build": "contrail-lex generate && contrail-lex types && vite build && tsx scripts/append-scheduled.ts", + "generate": "contrail-lex generate", + "generate:pull": "contrail-lex all", + "backfill": "contrail backfill", + "backfill:remote": "contrail backfill --remote", + "refresh": "contrail refresh", + "refresh:remote": "contrail refresh --remote", "preview": "vite preview", "prepare": "svelte-kit sync || echo ''", "check": "svelte-kit sync && svelte-check --tsconfig ./tsconfig.json", diff --git a/apps/sveltekit-cloudflare-workers/scripts/generate.ts b/apps/sveltekit-cloudflare-workers/scripts/generate.ts deleted file mode 100644 index b951669..0000000 --- a/apps/sveltekit-cloudflare-workers/scripts/generate.ts +++ /dev/null @@ -1,13 +0,0 @@ -import { join, dirname } from 'path'; -import { fileURLToPath } from 'url'; -import { config } from '../src/lib/contrail.config'; -import { generateLexicons } from '@atmo-dev/contrail-lexicons'; - -const ROOT_DIR = join(dirname(fileURLToPath(import.meta.url)), '..'); - -generateLexicons({ - config, - rootDir: ROOT_DIR, - outputDir: join(ROOT_DIR, 'lexicons', 'generated'), - writeRuntimeFiles: true -}); diff --git a/apps/sveltekit-cloudflare-workers/scripts/sync.ts b/apps/sveltekit-cloudflare-workers/scripts/sync.ts deleted file mode 100644 index c1f672f..0000000 --- a/apps/sveltekit-cloudflare-workers/scripts/sync.ts +++ /dev/null @@ -1,68 +0,0 @@ -/** - * Discover users from relays and backfill their records from PDS. - * - * Usage: - * pnpm sync # local D1 - * pnpm sync:remote # prod D1 - */ -import { Contrail } from '@atmo-dev/contrail'; -import { config } from '../src/lib/contrail.config'; -import { getPlatformProxy } from 'wrangler'; - -function elapsed(start: number): string { - const ms = Date.now() - start; - if (ms < 1000) return `${ms}ms`; - if (ms < 60_000) return `${(ms / 1000).toFixed(1)}s`; - const mins = Math.floor(ms / 60_000); - const secs = ((ms % 60_000) / 1000).toFixed(0); - return `${mins}m ${secs}s`; -} - -async function main() { - const remote = process.argv.includes('--remote'); - const syncStart = Date.now(); - - console.log(`=== Sync (${remote ? 'remote/prod' : 'local'} D1) ===\n`); - - const { env, dispose } = await getPlatformProxy<{ DB: D1Database }>({ - environment: remote ? 'production' : undefined - }); - - const contrail = new Contrail({ ...config, db: env.DB }); - - try { - await contrail.init(); - - console.log('--- Discovery ---'); - const discoveryStart = Date.now(); - const discovered = await contrail.discover(); - console.log(` Done: ${discovered.length} users in ${elapsed(discoveryStart)}\n`); - - console.log('--- Backfill ---'); - const backfillStart = Date.now(); - const total = await contrail.backfill({ - concurrency: 100, - onProgress: ({ records, usersComplete, usersTotal, usersFailed }) => { - const secs = (Date.now() - backfillStart) / 1000; - const rate = secs > 0 ? Math.round(records / secs) : 0; - const failStr = usersFailed > 0 ? ` | ${usersFailed} failed` : ''; - process.stdout.write( - `\r ${records} records | ${usersComplete}/${usersTotal} users | ${rate}/s | ${elapsed(backfillStart)}${failStr} ` - ); - } - }); - process.stdout.write('\n'); - console.log(` Done: ${total} records in ${elapsed(backfillStart)}\n`); - - console.log(`=== Finished in ${elapsed(syncStart)} ===`); - console.log(` Discovered: ${discovered.length} users`); - console.log(` Backfilled: ${total} records`); - } finally { - await dispose(); - } -} - -main().catch((err) => { - console.error(err); - process.exit(1); -}); diff --git a/apps/sveltekit-cloudflare-workers/src/lib/atproto/server/repo.remote.ts b/apps/sveltekit-cloudflare-workers/src/lib/atproto/server/repo.remote.ts index a9db6fb..fc7f4d8 100644 --- a/apps/sveltekit-cloudflare-workers/src/lib/atproto/server/repo.remote.ts +++ b/apps/sveltekit-cloudflare-workers/src/lib/atproto/server/repo.remote.ts @@ -33,6 +33,8 @@ export const putRecord = command( } }); + if (!response.ok) error(502, 'putRecord failed'); + // Immediately index the new/updated record in contrail const { platform } = getRequestEvent(); const db = platform?.env?.DB; diff --git a/apps/sveltekit-cloudflare-workers/src/routes/+layout.server.ts b/apps/sveltekit-cloudflare-workers/src/routes/+layout.server.ts index 756dc10..72b6c58 100644 --- a/apps/sveltekit-cloudflare-workers/src/routes/+layout.server.ts +++ b/apps/sveltekit-cloudflare-workers/src/routes/+layout.server.ts @@ -14,10 +14,11 @@ export const load: LayoutServerLoad = async ({ locals, platform }) => { }); if (!res.ok) return { did: locals.did, profile: null }; + const entry = res.data.profiles?.[0]; return { did: locals.did, - profile: extractProfile(res.data) + profile: entry ? extractProfile(entry) : null }; } catch { return { did: locals.did, profile: null }; diff --git a/apps/sveltekit-cloudflare-workers/src/routes/+page.svelte b/apps/sveltekit-cloudflare-workers/src/routes/+page.svelte index 0225014..2129ed5 100644 --- a/apps/sveltekit-cloudflare-workers/src/routes/+page.svelte +++ b/apps/sveltekit-cloudflare-workers/src/routes/+page.svelte @@ -39,9 +39,11 @@ if (profiles[did]) return; try { const res = await contrailClient.get('statusphere.app.getProfile', { - params: { actor: did } + params: { actor: did as `did:${string}:${string}` } }); - profiles[did] = extractProfile(res.data); + if (!res.ok) return; + const entry = res.data.profiles?.[0]; + if (entry) profiles[did] = extractProfile(entry); } catch { // ignore fetch errors } diff --git a/docs/01-indexing.md b/docs/01-indexing.md index 7baeeaa..0b8d70d 100644 --- a/docs/01-indexing.md +++ b/docs/01-indexing.md @@ -97,6 +97,8 @@ async scheduled(_ev, env, ctx) { `ingest()` connects to Jetstream, streams events since the saved cursor, stops when caught up. Running every minute is fine — the next fire resumes where this one left off. Each cycle is bounded, so it can't blow past the Worker time limit. +**Local dev:** wrangler's cron scheduler only runs in deployed production. For local dev use `pnpm contrail dev` — it runs `wrangler dev --test-scheduled`, fires `/__scheduled` on your configured cron interval, and prompts you to run backfill or refresh if the local DB looks stale on start. + ### Persistent (node / any long-lived server) If your runtime can keep a socket open, skip the cron entirely: diff --git a/docs/08-examples.md b/docs/08-examples.md deleted file mode 100644 index 4445cd7..0000000 --- a/docs/08-examples.md +++ /dev/null @@ -1,65 +0,0 @@ -# Examples - -Every example lives in [`apps/`](https://github.com/flo-bit/contrail/tree/main/apps) and pins contrail as `workspace:*`. Clone the repo, `pnpm install`, and each one runs. - -## `rsvp-atmo` — the reference deployment - -[`apps/rsvp-atmo`](https://github.com/flo-bit/contrail/tree/main/apps/rsvp-atmo) - -Cloudflare Workers + D1. Indexes `community.lexicon.calendar.event` and `rsvp`. Exposes the full spaces + community + realtime surface. Cron-driven Jetstream ingestion every minute. - -Use this if you're building on Workers and want a starting point that already has deploy config wired up. - -```bash -pnpm --filter rsvp-atmo dev # local wrangler + auto-cron -pnpm --filter rsvp-atmo deploy # requires D1 database created -pnpm --filter rsvp-atmo sync # discover + backfill against D1 -``` - -## `group-chat` — full app showcase - -[`apps/group-chat`](https://github.com/flo-bit/contrail/tree/main/apps/group-chat) - -SvelteKit + Cloudflare Workers. The one that exercises everything: permissioned spaces for private rooms, community-controlled DIDs for groups, client-side `contrail-sync` for reactive messages, Durable Object-hibernated WebSockets for realtime delivery, OAuth-based login. - -This is the canonical "what can contrail do" demo. If you're trying to understand how the pieces fit together end to end, read this app's code before anything else. - -```bash -pnpm --filter sveltekit-group-chat dev -``` - -## `postgres` — Node + PG minimal - -[`apps/postgres`](https://github.com/flo-bit/contrail/tree/main/apps/postgres) - -The smallest possible Node deployment. Docker Compose for Postgres, three scripts: `sync` (discover + backfill), `ingest` (persistent Jetstream), `serve` (HTTP handler). Skips spaces/communities/realtime. - -Use this as a template if you're running on a normal server and don't need Cloudflare's bells. - -```bash -cd apps/postgres -docker compose up -d -pnpm sync -pnpm serve -``` - -## `cloudflare-workers` — minimal Workers - -[`apps/cloudflare-workers`](https://github.com/flo-bit/contrail/tree/main/apps/cloudflare-workers) - -The simplest working Worker. One collection (events), one HTTP handler, cron-driven ingest. No spaces, no communities. Good for reading top-to-bottom in one sitting to see what contrail does at minimum. - -## `sveltekit-cloudflare-workers` — SvelteKit Statusphere - -[`apps/sveltekit-cloudflare-workers`](https://github.com/flo-bit/contrail/tree/main/apps/sveltekit-cloudflare-workers) - -A Statusphere-style SvelteKit app with OAuth login, contrail-indexed public records, and Cloudflare adapter. No spaces/communities — think "atproto blog or status post UI." Useful as a scaffold for public-only apps. - -## Choosing a starting point - -| Need | Start from | -|---|---| -| "Just index some records" | `cloudflare-workers` or `postgres` | -| "Index + SvelteKit UI, public only" | `sveltekit-cloudflare-workers` | -| "Private rooms / group chat / full stack" | `group-chat` | -| "Calendar-ish domain, Workers deploy" | `rsvp-atmo` | diff --git a/packages/contrail/src/cli.ts b/packages/contrail/src/cli.ts index 4f98dfb..9246dae 100644 --- a/packages/contrail/src/cli.ts +++ b/packages/contrail/src/cli.ts @@ -5,21 +5,28 @@ * Usage: * contrail backfill [--remote] [--binding DB] [--config path] * contrail refresh [--remote] [--ignore-window 60s] [--by-collection] + * contrail dev [--binding DB] [--cron '*\/1 * * * *'] * * Config auto-detects at contrail.config.ts, src/contrail.config.ts, * src/lib/contrail.config.ts, or app/contrail.config.ts (first match wins). */ +import { spawn } from "node:child_process"; +import { createInterface } from "node:readline/promises"; +import { stdin as input, stdout as output } from "node:process"; import { findConfigFile, loadConfig, CONFIG_CANDIDATES_MESSAGE } from "./cli-config.js"; import { backfillAll, refresh } from "./workers/backfill.js"; +import { Contrail } from "./contrail.js"; import type { CollectionStats, RefreshResult } from "./core/refresh.js"; +import type { Database } from "./core/types.js"; -type Subcommand = "backfill" | "refresh" | "help"; +type Subcommand = "backfill" | "refresh" | "dev" | "help"; const USAGE = `contrail [options] Subcommands: backfill One-time bulk load from each known DID's PDS (resumable) refresh Fresh sweep: reconcile PDS vs DB, report missing + stale + dev Local wrangler dev + auto-trigger cron + backfill/refresh prompts help Print this message Options (backfill): @@ -37,6 +44,14 @@ Options (refresh): --concurrency Passed to contrail.refresh(). Default: 50. --ignore-window Seconds of grace for stale-update detection. Default: 60. --by-collection Print per-collection stats, not just totals. + +Options (dev): + --config Path to Contrail config file (TS or JS). + --root Project root for auto-detection. Default: CWD. + --binding D1 binding name in wrangler.jsonc. Default: "DB". + --cron Cron expression to fire against /__scheduled. Default: "*/1 * * * *". + --stale-after Prompt to run refresh if the ingest cursor is older than this. Default: 60. + --yes Accept all prompts without asking (CI-friendly). `; interface Args { @@ -48,6 +63,9 @@ interface Args { concurrency?: number; ignoreWindowMs?: number; byCollection: boolean; + cron?: string; + staleAfterMin?: number; + yes: boolean; } function parseArgs(argv: string[]): Args { @@ -60,6 +78,9 @@ function parseArgs(argv: string[]): Args { let concurrency: number | undefined; let ignoreWindowMs: number | undefined; let byCollection = false; + let cron: string | undefined; + let staleAfterMin: number | undefined; + let yes = false; for (let i = 0; i < args.length; i++) { const a = args[i]; if (a === "--config") config = args[++i]; @@ -69,6 +90,9 @@ function parseArgs(argv: string[]): Args { else if (a === "--concurrency") concurrency = parseInt(args[++i], 10); else if (a === "--ignore-window") ignoreWindowMs = parseInt(args[++i], 10) * 1000; else if (a === "--by-collection") byCollection = true; + else if (a === "--cron") cron = args[++i]; + else if (a === "--stale-after") staleAfterMin = parseInt(args[++i], 10); + else if (a === "--yes" || a === "-y") yes = true; else if (a === "-h" || a === "--help") return { cmd: "help", @@ -79,9 +103,12 @@ function parseArgs(argv: string[]): Args { concurrency, ignoreWindowMs, byCollection, + cron, + staleAfterMin, + yes, }; } - return { cmd, config, root, remote, binding, concurrency, ignoreWindowMs, byCollection }; + return { cmd, config, root, remote, binding, concurrency, ignoreWindowMs, byCollection, cron, staleAfterMin, yes }; } function resolveConfigPath(opts: Args): string | null { @@ -95,6 +122,21 @@ function resolveConfigPath(opts: Args): string | null { return p; } +async function promptYesNo(question: string, defaultYes: boolean, autoYes: boolean): Promise { + if (autoYes) return true; + // Non-TTY → default-decline; the user isn't there to answer. + if (!input.isTTY) return false; + const rl = createInterface({ input, output }); + try { + const hint = defaultYes ? "[Y/n]" : "[y/N]"; + const ans = (await rl.question(`${question} ${hint} `)).trim().toLowerCase(); + if (ans === "") return defaultYes; + return ans === "y" || ans === "yes"; + } finally { + rl.close(); + } +} + function formatStats(s: CollectionStats): string { return `${s.missing} missing, ${s.staleUpdates} stale updates, ${s.inSync} in sync`; } @@ -121,6 +163,94 @@ function printRefreshReport(result: RefreshResult, byCollection: boolean): void console.log(` (ignore window: ${(result.ignoreWindowMs / 1000).toFixed(0)}s)`); } +async function cmdDev(opts: Args): Promise { + const configPath = resolveConfigPath(opts); + if (!configPath) return 1; + const config = await loadConfig(configPath); + + // Pre-flight: connect to local D1 via wrangler's platform proxy, inspect + // state, prompt. Then dispose before starting wrangler dev so the two + // miniflare processes don't fight over the sqlite file. + const { getPlatformProxy } = await import("wrangler"); + const { env, dispose } = await getPlatformProxy(); + const db = (env as Record)[opts.binding] as Database | undefined; + + if (db) { + const contrail = new Contrail(config); + await contrail.init(db); + + const hasBackfilled = await db + .prepare("SELECT 1 FROM backfills WHERE completed = 1 LIMIT 1") + .first(); + + if (!hasBackfilled) { + console.log("no backfilled users in the local DB yet."); + if (await promptYesNo("run backfill now? (takes a few minutes)", true, opts.yes)) { + await contrail.backfillAll({ concurrency: opts.concurrency ?? 100 }, db); + } + } else { + const staleAfterMs = (opts.staleAfterMin ?? 60) * 60_000; + const row = await db + .prepare("SELECT time_us FROM cursor WHERE id = 1") + .first<{ time_us: number }>(); + if (row?.time_us) { + const ageMs = Date.now() - Math.floor(row.time_us / 1000); + if (ageMs > staleAfterMs) { + const hrs = (ageMs / 3_600_000).toFixed(1); + console.log(`ingest cursor is ${hrs}h old — you may have missed events.`); + if (await promptYesNo("run refresh first?", true, opts.yes)) { + await contrail.refresh({ ignoreWindowMs: opts.ignoreWindowMs }, db); + } + } + } + } + } + + await dispose(); + + // Start wrangler dev + fire /__scheduled on a loop so the cron actually + // runs in local dev (wrangler's cron scheduler only works in deployed + // production; --test-scheduled enables the manual-trigger endpoint). + const cron = opts.cron ?? "*/1 * * * *"; + const cronUrl = `http://localhost:8787/__scheduled?cron=${encodeURIComponent(cron)}`; + + const wrangler = spawn("npx", ["wrangler", "dev", "--test-scheduled"], { + stdio: "inherit", + shell: process.platform === "win32", + cwd: opts.root, + }); + + // Give wrangler a few seconds to bind the port, then start the cron loop. + const kickoff = setTimeout(() => { + fetch(cronUrl).catch(() => {}); // fire-and-forget; first run + console.log(`\nauto-ingest: firing ${cronUrl} every 60s\n`); + }, 3_000); + + const interval = setInterval(() => { + fetch(cronUrl).catch(() => {}); // wrangler may be restarting; swallow + }, 60_000); + + const cleanup = () => { + clearTimeout(kickoff); + clearInterval(interval); + try { + wrangler.kill("SIGINT"); + } catch { + /* ignore */ + } + }; + process.on("SIGINT", cleanup); + process.on("SIGTERM", cleanup); + + return new Promise((resolve) => { + wrangler.on("exit", (code) => { + clearTimeout(kickoff); + clearInterval(interval); + resolve(code ?? 0); + }); + }); +} + async function main(): Promise { const opts = parseArgs(process.argv); @@ -157,6 +287,8 @@ async function main(): Promise { return 0; } + if (opts.cmd === "dev") return cmdDev(opts); + console.error(USAGE); return 1; } diff --git a/packages/lexicons/src/generate.ts b/packages/lexicons/src/generate.ts index 9e2c745..bf57b6a 100644 --- a/packages/lexicons/src/generate.ts +++ b/packages/lexicons/src/generate.ts @@ -247,7 +247,15 @@ export function generateLexicons(options: GenerateOptions): Record { @@ -738,7 +746,7 @@ export function generateLexicons(options: GenerateOptions): Record