Service: job event stream, async in-memory backend, preview routes

Ported from PR #15 (packages-side surface only; the fork's dev UI page is
intentionally dropped - the API surface is the product):
- job_events table + repository, worker emits stage transitions, additive
  GET /v1/clones/:id/events polling route. The fork's hand-written
  migration was never registered in Drizzle's journal (it silently never
  applied); regenerated properly via drizzle-kit from schema.ts with the
  (job_id, seq) index declared, and verified against a live Postgres
- In-memory backend moves to the documented enqueue-202 + poll contract
  with single-flight BUSY guard; POST honors backend httpStatus
- app-preview static serving routes; the preview BUILD half is deferred
  until the compiler exports buildApp/DEFAULT_HARNESS_DIR - routes 404
  cleanly until then
- preview option threaded through core types; cache key intentionally
  unchanged while the flag is inert

All workspaces typecheck; api 18/18, core 13/13, worker 1/1 pass;
migration verified applying (table + index + FK) on postgres:16.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Samraaj Bath
2026-07-04 15:00:55 -07:00
co-authored by Claude Fable 5
parent f991f469cf
commit bbe4a5643b
16 changed files with 701 additions and 55 deletions
+39 -2
View File
@@ -15,6 +15,7 @@ const OptionsSchema = z
mode: z.enum(["single", "multi"]).optional(), mode: z.enum(["single", "multi"]).optional(),
styling: z.enum(["tailwind", "css"]).optional(), styling: z.enum(["tailwind", "css"]).optional(),
framework: z.enum(["next", "vite"]).optional(), framework: z.enum(["next", "vite"]).optional(),
preview: z.boolean().optional(),
verify: z.boolean().optional(), verify: z.boolean().optional(),
asyncVerify: z.boolean().optional(), asyncVerify: z.boolean().optional(),
maxRoutes: z.number().int().positive().optional(), maxRoutes: z.number().int().positive().optional(),
@@ -238,10 +239,12 @@ export function createApp(deps: AppDeps): Hono {
try { try {
const out = await backend.submit(url, opts); const out = await backend.submit(url, opts);
if (out.status === "queued") return c.json({ jobId: out.jobId, status: "queued" }, 202); if (out.status === "queued") return c.json({ jobId: out.jobId, status: "queued" }, out.httpStatus);
return c.json(out.result, 200); return c.json(out.result, 200);
} catch (e) { } catch (e) {
return c.json({ status: "failed", error: String(e).slice(0, 500) }, 500); const msg = String(e);
if (msg.startsWith("BUSY:")) return c.json({ error: msg.slice(5).trim() }, 429);
return c.json({ status: "failed", error: msg.slice(0, 500) }, 500);
} }
}); });
@@ -282,6 +285,40 @@ export function createApp(deps: AppDeps): Hono {
return c.body(file.bytes); return c.body(file.bytes);
}); });
// Pipeline progress events (poll every ~300ms while a clone runs).
app.get("/v1/clones/:id/events", async (c) => {
const after = Math.max(0, Number(c.req.query("after") ?? "0") || 0);
const events = backend.events ? await backend.events(c.req.param("id"), after) : null;
if (!events) return c.json({ error: "not found" }, 404);
return c.json({ jobId: c.req.param("id"), events });
});
// Browsable preview of the built clone (static export published by the preview
// build under public/app-preview/). References are relative, so the export works
// from this mount — the only requirement is a trailing slash on the root.
const previewFile = async (c: Context, sub: string) => {
const id = c.req.param("id") ?? "";
const tryPaths = sub.includes(".")
? [`public/app-preview/${sub}`]
: [
`public/app-preview/${sub}`.replace(/\/$/, "") + "/index.html",
`public/app-preview/${sub}`,
`public/app-preview/${sub}.html`,
];
for (const p of tryPaths) {
const file = await backend.file(id, p);
if (file) {
c.header("content-type", file.contentType);
c.header("content-length", String(file.bytes.length));
return c.body(file.bytes);
}
}
return c.json({ error: "no app preview for this clone (preview builds are on by default for single-page clones; pass options.preview=true otherwise)" }, 404);
};
app.get("/v1/clones/:id/app-preview", (c) => c.redirect(`/v1/clones/${c.req.param("id")}/app-preview/`, 302));
app.get("/v1/clones/:id/app-preview/", (c) => previewFile(c, "index.html"));
app.get("/v1/clones/:id/app-preview/:path{.+}", (c) => previewFile(c, c.req.param("path") ?? ""));
app.delete("/v1/clones/:id", async (c) => { app.delete("/v1/clones/:id", async (c) => {
const ok = await backend.remove(c.req.param("id")); const ok = await backend.remove(c.req.param("id"));
return c.json({ deleted: ok }, ok ? 200 : 404); return c.json({ deleted: ok }, ok ? 200 : 404);
+4 -2
View File
@@ -1,5 +1,5 @@
import type { CloneOptions, RouteInfo } from "@cloner/core"; import type { CloneOptions, RouteInfo } from "@cloner/core";
import type { RestCloneResult } from "./rest.js"; import type { RestCloneResult, RestCloneSummary } from "./rest.js";
export type JobStatus = "queued" | "running" | "succeeded" | "failed" | "cached"; export type JobStatus = "queued" | "running" | "succeeded" | "failed" | "cached";
@@ -22,7 +22,7 @@ export type JobView = {
}; };
export type SubmitOutcome = export type SubmitOutcome =
| { jobId: string; status: "succeeded" | "cached"; httpStatus: 200; result: RestCloneResult } | { jobId: string; status: "succeeded" | "cached"; httpStatus: 200; result: RestCloneResult | RestCloneSummary }
| { jobId: string; status: "queued"; httpStatus: 202 }; | { jobId: string; status: "queued"; httpStatus: 202 };
export type ResultOutcome = export type ResultOutcome =
@@ -56,4 +56,6 @@ export interface Backend {
facets(jobId: string): Promise<FileFacet[] | null>; facets(jobId: string): Promise<FileFacet[] | null>;
/** The whole app as one compressed archive (null if not ready/found). */ /** The whole app as one compressed archive (null if not ready/found). */
bundle(jobId: string, format?: BundleFormat): Promise<CloneBundle | null>; bundle(jobId: string, format?: BundleFormat): Promise<CloneBundle | null>;
/** Pipeline progress events for polling UIs (null if unsupported or job unknown). */
events?(jobId: string, after?: number): Promise<Array<Record<string, unknown>> | null>;
} }
+6
View File
@@ -142,4 +142,10 @@ export class DbBackend implements Backend {
const url = this.deps.store.uploadBundle ? await this.deps.store.uploadBundle(jobId, format, bytes) : undefined; const url = this.deps.store.uploadBundle ? await this.deps.store.uploadBundle(jobId, format, bytes) : undefined;
return { bytes, sha256: sha256hex(bytes), format, url }; return { bytes, sha256: sha256hex(bytes), format, url };
} }
async events(jobId: string, after = 0): Promise<Array<Record<string, unknown>> | null> {
const job = await repo.getJob(this.deps.db, jobId);
if (!job) return null;
return repo.listJobEvents(this.deps.db, jobId, after);
}
} }
+44 -10
View File
@@ -10,24 +10,51 @@ import type { Backend, BundleFormat, CloneBundle, FileFacet, JobView, ResultOutc
export type RunJob = (input: RunCloneJobInput) => Promise<CloneJobResult>; export type RunJob = (input: RunCloneJobInput) => Promise<CloneJobResult>;
/** Sync, in-memory backend (M1): runs the clone INLINE on submit and holds results /** In-memory backend: enqueues a clone (202) and runs it in the background so long
* in memory. POST returns the file map immediately (200). */ * captures (multi-viewport + verify) never hold a single HTTP connection open for
* minutes — which browsers surface as `TypeError: Failed to fetch` when the link
* drops (server restart, hot-reload, idle timeout). Poll /v1/clones/:id + /events. */
export class InMemoryBackend implements Backend { export class InMemoryBackend implements Backend {
constructor(private deps: { store: InMemoryStore; runJob: RunJob; makeTempBase?: () => string; captureCacheDir?: string }) {} constructor(private deps: { store: InMemoryStore; runJob: RunJob; makeTempBase?: () => string; captureCacheDir?: string }) {}
private activeClones = 0;
private makeBase(): string { private makeBase(): string {
return (this.deps.makeTempBase ?? (() => mkdtempSync(join(tmpdir(), "api-clone-"))))(); return (this.deps.makeTempBase ?? (() => mkdtempSync(join(tmpdir(), "api-clone-"))))();
} }
async submit(url: string, options: CloneOptions | undefined): Promise<SubmitOutcome> { async submit(url: string, options: CloneOptions | undefined): Promise<SubmitOutcome> {
if (this.activeClones > 0 || this.deps.store.list().some((j) => j.status === "running")) {
throw new Error("BUSY: another clone is already running — wait for it to finish (use npm run dev:api:stable to avoid hot-reload killing Playwright)");
}
const id = randomUUID(); const id = randomUUID();
const base = this.makeBase(); const base = this.makeBase();
const events: Array<Record<string, unknown>> = [];
const kind: "clone" | "clone_site" = resolveCloneMode(options) === "multi" ? "clone_site" : "clone";
const rec = { id, status: "running" as const, url, kind, options: options ?? {}, createdAt: Date.now(), base, events };
this.deps.store.put(rec);
void this.runInBackground(id, url, options, rec, base, events, kind);
return { jobId: id, status: "queued", httpStatus: 202 };
}
private async runInBackground(
id: string,
url: string,
options: CloneOptions | undefined,
rec: { id: string; status: "running"; url: string; kind: "clone" | "clone_site"; options: CloneOptions; createdAt: number; base: string; events: Array<Record<string, unknown>> },
base: string,
events: Array<Record<string, unknown>>,
kind: "clone" | "clone_site",
): Promise<void> {
this.activeClones++;
const log = (e: Record<string, unknown>) => {
events.push({ t: Date.now(), ...e });
this.deps.store.put({ ...rec, events: [...events] });
};
try { try {
// captureCacheDir (persistent, shared across submits) powers the single→multi const result = await this.deps.runJob({ url, options, runsDir: base, captureCacheDir: this.deps.captureCacheDir, log });
// reuse: a single-page submit stashes its capture; a later multi-page submit for log({ event: "clone_done" });
// the same URL reuses it as the entry route (no re-capture) and expands the site. this.deps.store.put({ id, status: "succeeded", url, kind: result.kind, options: result.options, createdAt: rec.createdAt, result, base, events });
const result = await this.deps.runJob({ url, options, runsDir: base, captureCacheDir: this.deps.captureCacheDir });
this.deps.store.put({ id, status: "succeeded", url, kind: result.kind, options: result.options, createdAt: Date.now(), result, base });
if (result.options.asyncVerify) { if (result.options.asyncVerify) {
void verifyCloneJobResult(result, { void verifyCloneJobResult(result, {
validationConcurrency: result.options.validationConcurrency, validationConcurrency: result.options.validationConcurrency,
@@ -39,18 +66,25 @@ export class InMemoryBackend implements Backend {
result.verify = { error: String(e).slice(0, 500), async: true }; result.verify = { error: String(e).slice(0, 500), async: true };
}); });
} }
return { jobId: id, status: "succeeded", httpStatus: 200, result: buildRestResult(id, result, `/v1/clones/${id}/files`) };
} catch (e) { } catch (e) {
try { try {
rmSync(base, { recursive: true, force: true }); rmSync(base, { recursive: true, force: true });
} catch { } catch {
/* best effort */ /* best effort */
} }
this.deps.store.put({ id, status: "failed", url, kind: resolveCloneMode(options) === "multi" ? "clone_site" : "clone", options: options ?? {}, createdAt: Date.now(), error: String(e) }); log({ event: "clone_error", error: String(e).slice(0, 300) });
throw e; this.deps.store.put({ id, status: "failed", url, kind, options: options ?? {}, createdAt: rec.createdAt, error: String(e), events });
} finally {
this.activeClones = Math.max(0, this.activeClones - 1);
} }
} }
async events(jobId: string): Promise<Array<Record<string, unknown>> | null> {
const rec = this.deps.store.get(jobId);
if (!rec) return null;
return rec.events ?? [];
}
async status(jobId: string): Promise<JobView | null> { async status(jobId: string): Promise<JobView | null> {
const rec = this.deps.store.get(jobId); const rec = this.deps.store.get(jobId);
if (!rec) return null; if (!rec) return null;
+3 -1
View File
@@ -6,7 +6,7 @@ import type { CloneJobResult, CloneOptions } from "@cloner/core";
* it is removed on eviction. M2/M4 replace this with DB rows + S3 objects. */ * it is removed on eviction. M2/M4 replace this with DB rows + S3 objects. */
export type JobRecord = { export type JobRecord = {
id: string; id: string;
status: "succeeded" | "failed"; status: "running" | "succeeded" | "failed";
url: string; url: string;
kind: "clone" | "clone_site"; kind: "clone" | "clone_site";
options: CloneOptions; options: CloneOptions;
@@ -14,6 +14,8 @@ export type JobRecord = {
result?: CloneJobResult; result?: CloneJobResult;
base?: string; base?: string;
error?: string; error?: string;
/** pipeline progress events (compiler log stream + service phases), poll via /events */
events?: Array<Record<string, unknown>>;
}; };
/** In-memory result store with TTL eviction. Swapped for a DB-backed store in M2. */ /** In-memory result store with TTL eviction. Swapped for a DB-backed store in M2. */
+20 -8
View File
@@ -26,13 +26,25 @@ describe("POST /v1/clones (real clone, served fixture)", { skip: hasChromium() ?
headers: { "content-type": "application/json" }, headers: { "content-type": "application/json" },
body: JSON.stringify({ url: server.url + "/components.html", options: { interactions: false, components: true, motion: false } }), body: JSON.stringify({ url: server.url + "/components.html", options: { interactions: false, components: true, motion: false } }),
}); });
assert.equal(res.status, 200); assert.equal(res.status, 202);
const body = await res.json(); const queued = await res.json();
assert.equal(body.status, "succeeded"); let body: Record<string, unknown> | undefined;
assert.ok(body.files["src/app/page.tsx"], "has page.tsx"); for (let i = 0; i < 600; i++) {
assert.equal(body.files["src/app/page.tsx"].type, "text"); const view = await (await app.request(`/v1/clones/${queued.jobId}`)).json();
assert.ok(body.files["package.json"], "has package.json"); if (view.status === "succeeded") {
assert.ok(body.capture.nodeCount > 0); body = await (await app.request(`/v1/clones/${queued.jobId}/result`)).json();
assert.equal(body.capture.blocked, false); break;
}
if (view.status === "failed") assert.fail(view.error);
await new Promise((r) => setTimeout(r, 50));
}
assert.ok(body);
const files = body!.files as Record<string, { type: string }>;
assert.equal(body!.status, "succeeded");
assert.ok(files["src/app/page.tsx"], "has page.tsx");
assert.equal(files["src/app/page.tsx"].type, "text");
assert.ok(files["package.json"], "has package.json");
assert.ok((body!.capture as { nodeCount: number }).nodeCount > 0);
assert.equal((body!.capture as { blocked: boolean }).blocked, false);
}); });
}); });
+43 -23
View File
@@ -34,7 +34,17 @@ const fakeRunJob: RunJob = async (input) => {
} satisfies CloneJobResult; } satisfies CloneJobResult;
}; };
test("POST /v1/clones returns the eager file map; binaries by reference; streaming + lifecycle", async () => { async function waitForResult(app: ReturnType<typeof createApp>, jobId: string) {
for (let i = 0; i < 200; i++) {
const view = await (await app.request(`/v1/clones/${jobId}`)).json();
if (view.status === "succeeded") return await (await app.request(`/v1/clones/${jobId}/result`)).json();
if (view.status === "failed") throw new Error(String(view.error));
await new Promise((r) => setTimeout(r, 5));
}
throw new Error("timeout waiting for clone");
}
test("POST /v1/clones enqueues then completes; binaries by reference; streaming + lifecycle", async () => {
const store = new InMemoryStore(60_000); const store = new InMemoryStore(60_000);
const app = createApp({ backend: new InMemoryBackend({ store, runJob: fakeRunJob }) }); const app = createApp({ backend: new InMemoryBackend({ store, runJob: fakeRunJob }) });
try { try {
@@ -43,39 +53,47 @@ test("POST /v1/clones returns the eager file map; binaries by reference; streami
headers: { "content-type": "application/json" }, headers: { "content-type": "application/json" },
body: JSON.stringify({ url: "https://example.com/", options: {} }), body: JSON.stringify({ url: "https://example.com/", options: {} }),
}); });
assert.equal(res.status, 200); assert.equal(res.status, 202);
const body = await res.json(); const queued = await res.json();
assert.ok(body.jobId); assert.ok(queued.jobId);
assert.equal(body.status, "succeeded"); assert.equal(queued.status, "queued");
assert.equal(body.kind, "clone");
let body: Record<string, unknown> | undefined;
body = await waitForResult(app, queued.jobId);
assert.ok(body);
assert.equal(body!.status, "succeeded");
assert.equal(body!.kind, "clone");
// Text inline. // Text inline.
const page = body.files["src/app/page.tsx"]; const files = body!.files as Record<string, { type: string; content?: string; sha256?: string; url?: string }>;
assert.equal(page.type, "text"); const page = files["src/app/page.tsx"];
assert.ok(page.content.includes("Page")); assert.ok(page);
assert.ok(page.sha256); assert.equal(page!.type, "text");
assert.ok(page!.content!.includes("Page"));
assert.ok(page!.sha256);
// Binary by reference (URL, not bytes). // Binary by reference (URL, not bytes).
const bin = body.files["public/assets/cloned/images/a.png"]; const bin = files["public/assets/cloned/images/a.png"];
assert.equal(bin.type, "binary"); assert.ok(bin);
assert.equal(bin.content, undefined); assert.equal(bin!.type, "binary");
assert.equal(bin.url, `/v1/clones/${body.jobId}/files/public/assets/cloned/images/a.png`); assert.equal(bin!.content, undefined);
assert.equal(bin!.url, `/v1/clones/${queued.jobId}/files/public/assets/cloned/images/a.png`);
// Per-file streaming returns the actual bytes. // Per-file streaming returns the actual bytes.
const fileRes = await app.request(bin.url); const fileRes = await app.request(bin!.url!);
assert.equal(fileRes.status, 200); assert.equal(fileRes.status, 200);
assert.equal(fileRes.headers.get("content-type"), "image/png"); assert.equal(fileRes.headers.get("content-type"), "image/png");
const bytes = Buffer.from(await fileRes.arrayBuffer()); const bytes = Buffer.from(await fileRes.arrayBuffer());
assert.equal(bytes.length, 7); assert.equal(bytes.length, 7);
// Cheap metadata overview (no file contents). // Cheap metadata overview (no file contents).
const meta = await (await app.request(`/v1/clones/${body.jobId}`)).json(); const meta = await (await app.request(`/v1/clones/${queued.jobId}`)).json();
assert.equal(meta.fileCount, 3); assert.equal(meta.fileCount, 3);
assert.equal(meta.capture.nodeCount, 42); assert.equal(meta.capture.nodeCount, 42);
assert.equal(meta.totalBytes > 0, true); assert.equal(meta.totalBytes > 0, true);
// Full result fetch. // Full result fetch.
const result = await app.request(`/v1/clones/${body.jobId}/result`); const result = await app.request(`/v1/clones/${queued.jobId}/result`);
assert.equal(result.status, 200); assert.equal(result.status, 200);
// List. // List.
@@ -83,9 +101,9 @@ test("POST /v1/clones returns the eager file map; binaries by reference; streami
assert.equal(list.clones.length, 1); assert.equal(list.clones.length, 1);
// Delete purges; subsequent fetch 404s. // Delete purges; subsequent fetch 404s.
const del = await app.request(`/v1/clones/${body.jobId}`, { method: "DELETE" }); const del = await app.request(`/v1/clones/${queued.jobId}`, { method: "DELETE" });
assert.equal((await del.json()).deleted, true); assert.equal((await del.json()).deleted, true);
assert.equal((await app.request(`/v1/clones/${body.jobId}`)).status, 404); assert.equal((await app.request(`/v1/clones/${queued.jobId}`)).status, 404);
} finally { } finally {
store.clear(); store.clear();
} }
@@ -118,8 +136,9 @@ test("POST /v1/clones normalizes product options and legacy aliases", async () =
headers: { "content-type": "application/json" }, headers: { "content-type": "application/json" },
body: JSON.stringify({ url: "https://example.com/", options: { mode: "multi", styling: "css", framework: "vite" } }), body: JSON.stringify({ url: "https://example.com/", options: { mode: "multi", styling: "css", framework: "vite" } }),
}); });
assert.equal(product.status, 200); assert.equal(product.status, 202);
const productBody = await product.json(); const productQueued = await product.json();
const productBody = await waitForResult(app, productQueued.jobId);
assert.deepEqual(productBody.options, { mode: "multi", styling: "css", framework: "vite" }); assert.deepEqual(productBody.options, { mode: "multi", styling: "css", framework: "vite" });
const legacy = await app.request("/v1/clones", { const legacy = await app.request("/v1/clones", {
@@ -127,8 +146,9 @@ test("POST /v1/clones normalizes product options and legacy aliases", async () =
headers: { "content-type": "application/json" }, headers: { "content-type": "application/json" },
body: JSON.stringify({ url: "https://example.com/", options: { multiPage: true, humanizeMode: "css" } }), body: JSON.stringify({ url: "https://example.com/", options: { multiPage: true, humanizeMode: "css" } }),
}); });
assert.equal(legacy.status, 200); assert.equal(legacy.status, 202);
const legacyBody = await legacy.json(); const legacyQueued = await legacy.json();
const legacyBody = await waitForResult(app, legacyQueued.jobId);
assert.deepEqual(legacyBody.options, { mode: "multi", styling: "css", framework: "next" }); assert.deepEqual(legacyBody.options, { mode: "multi", styling: "css", framework: "next" });
} finally { } finally {
store.clear(); store.clear();
+10 -4
View File
@@ -52,12 +52,18 @@ test("MCP: list-then-read + bundle contract (never floods context)", async () =>
// clone_website → jobId + status only (no files). // clone_website → jobId + status only (no files).
const cw = parse(await client.callTool({ name: "clone_website", arguments: { url: "https://example.com/", options: {} } })); const cw = parse(await client.callTool({ name: "clone_website", arguments: { url: "https://example.com/", options: {} } }));
assert.ok(cw.jobId); assert.ok(cw.jobId);
assert.equal(cw.status, "succeeded"); assert.equal(cw.status, "queued");
assert.ok(!("files" in cw));
const jobId = cw.jobId; const jobId = cw.jobId;
// status let status = parse(await client.callTool({ name: "get_clone_status", arguments: { jobId } }));
const status = parse(await client.callTool({ name: "get_clone_status", arguments: { jobId } })); for (let i = 0; i < 200 && status.status !== "succeeded"; i++) {
await new Promise((r) => setTimeout(r, 5));
status = parse(await client.callTool({ name: "get_clone_status", arguments: { jobId } }));
}
assert.equal(status.status, "succeeded");
// status (re-fetch after completion)
status = parse(await client.callTool({ name: "get_clone_status", arguments: { jobId } }));
assert.equal(status.status, "succeeded"); assert.equal(status.status, "succeeded");
// get_clone_result → metadata only, NO file contents. // get_clone_result → metadata only, NO file contents.
+4
View File
@@ -9,6 +9,7 @@ export type ResolvedCloneOptions = CloneOptions & {
interactions: boolean; interactions: boolean;
components: boolean; components: boolean;
motion: boolean; motion: boolean;
preview: boolean;
}; };
export function resolveCloneMode(options: CloneOptions = {}): CloneMode { export function resolveCloneMode(options: CloneOptions = {}): CloneMode {
@@ -53,5 +54,8 @@ export function resolveCloneOptions(options: CloneOptions = {}): ResolvedCloneOp
interactions: options.interactions ?? true, interactions: options.interactions ?? true,
components: options.components ?? true, components: options.components ?? true,
motion: options.motion ?? true, motion: options.motion ?? true,
// Preview builds are the single-page product surface; multi-page exports can be
// large, so previewing a site clone stays an explicit opt-in.
preview: options.preview ?? mode === "single",
}; };
} }
+4 -1
View File
@@ -26,6 +26,9 @@ export type CloneOptions = {
mode?: CloneMode; mode?: CloneMode;
styling?: CloneStyling; styling?: CloneStyling;
framework?: CloneFramework; framework?: CloneFramework;
/** Build a browsable static export of the clone into `public/app-preview/`
* (served by the API's app-preview route). Defaults ON for single-page clones. */
preview?: boolean;
verify?: boolean; verify?: boolean;
/** DB/worker mode: persist the clone first, then attach verify in the background. */ /** DB/worker mode: persist the clone first, then attach verify in the background. */
asyncVerify?: boolean; asyncVerify?: boolean;
@@ -75,7 +78,7 @@ export type CaptureSanity = {
blocked: boolean; // bot/egress wall text detected blocked: boolean; // bot/egress wall text detected
}; };
export type CloneTimings = { captureMs: number; generateMs: number; verifyMs?: number }; export type CloneTimings = { captureMs: number; generateMs: number; verifyMs?: number; previewMs?: number };
export type RouteInfo = { route: string; representativeOf?: string }; export type RouteInfo = { route: string; representativeOf?: string };
@@ -0,0 +1,10 @@
CREATE TABLE "job_events" (
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
"job_id" uuid NOT NULL,
"seq" integer NOT NULL,
"payload" jsonb NOT NULL,
"created_at" timestamp with time zone DEFAULT now() NOT NULL
);
--> statement-breakpoint
ALTER TABLE "job_events" ADD CONSTRAINT "job_events_job_id_jobs_id_fk" FOREIGN KEY ("job_id") REFERENCES "public"."jobs"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
CREATE INDEX "job_events_job_id_seq_idx" ON "job_events" USING btree ("job_id","seq");
@@ -0,0 +1,461 @@
{
"id": "de9b94b8-cefb-4d2a-a848-79a552976943",
"prevId": "e076bd1e-a7ce-40ab-abd1-9eb11d00ddb9",
"version": "7",
"dialect": "postgresql",
"tables": {
"public.api_keys": {
"name": "api_keys",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"key_hash": {
"name": "key_hash",
"type": "text",
"primaryKey": false,
"notNull": true
},
"label": {
"name": "label",
"type": "text",
"primaryKey": false,
"notNull": false
},
"rate_limit": {
"name": "rate_limit",
"type": "integer",
"primaryKey": false,
"notNull": false
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"revoked_at": {
"name": "revoked_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
}
},
"indexes": {},
"foreignKeys": {},
"compositePrimaryKeys": {},
"uniqueConstraints": {
"api_keys_key_hash_unique": {
"name": "api_keys_key_hash_unique",
"nullsNotDistinct": false,
"columns": [
"key_hash"
]
}
},
"policies": {},
"checkConstraints": {},
"isRLSEnabled": false
},
"public.cache": {
"name": "cache",
"schema": "",
"columns": {
"cache_key": {
"name": "cache_key",
"type": "text",
"primaryKey": true,
"notNull": true
},
"job_id": {
"name": "job_id",
"type": "uuid",
"primaryKey": false,
"notNull": false
},
"url": {
"name": "url",
"type": "text",
"primaryKey": false,
"notNull": true
},
"options_hash": {
"name": "options_hash",
"type": "text",
"primaryKey": false,
"notNull": true
},
"compiler_version": {
"name": "compiler_version",
"type": "text",
"primaryKey": false,
"notNull": true
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"expires_at": {
"name": "expires_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true
}
},
"indexes": {},
"foreignKeys": {
"cache_job_id_jobs_id_fk": {
"name": "cache_job_id_jobs_id_fk",
"tableFrom": "cache",
"tableTo": "jobs",
"columnsFrom": [
"job_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {},
"policies": {},
"checkConstraints": {},
"isRLSEnabled": false
},
"public.clones": {
"name": "clones",
"schema": "",
"columns": {
"job_id": {
"name": "job_id",
"type": "uuid",
"primaryKey": true,
"notNull": true
},
"url": {
"name": "url",
"type": "text",
"primaryKey": false,
"notNull": true
},
"route_count": {
"name": "route_count",
"type": "integer",
"primaryKey": false,
"notNull": true,
"default": 1
},
"file_manifest": {
"name": "file_manifest",
"type": "jsonb",
"primaryKey": false,
"notNull": true
},
"bundle_s3_key": {
"name": "bundle_s3_key",
"type": "text",
"primaryKey": false,
"notNull": false
},
"verify": {
"name": "verify",
"type": "jsonb",
"primaryKey": false,
"notNull": false
},
"capture_meta": {
"name": "capture_meta",
"type": "jsonb",
"primaryKey": false,
"notNull": true
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {},
"foreignKeys": {
"clones_job_id_jobs_id_fk": {
"name": "clones_job_id_jobs_id_fk",
"tableFrom": "clones",
"tableTo": "jobs",
"columnsFrom": [
"job_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {},
"policies": {},
"checkConstraints": {},
"isRLSEnabled": false
},
"public.job_events": {
"name": "job_events",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"job_id": {
"name": "job_id",
"type": "uuid",
"primaryKey": false,
"notNull": true
},
"seq": {
"name": "seq",
"type": "integer",
"primaryKey": false,
"notNull": true
},
"payload": {
"name": "payload",
"type": "jsonb",
"primaryKey": false,
"notNull": true
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
}
},
"indexes": {
"job_events_job_id_seq_idx": {
"name": "job_events_job_id_seq_idx",
"columns": [
{
"expression": "job_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "seq",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
"job_events_job_id_jobs_id_fk": {
"name": "job_events_job_id_jobs_id_fk",
"tableFrom": "job_events",
"tableTo": "jobs",
"columnsFrom": [
"job_id"
],
"columnsTo": [
"id"
],
"onDelete": "cascade",
"onUpdate": "no action"
}
},
"compositePrimaryKeys": {},
"uniqueConstraints": {},
"policies": {},
"checkConstraints": {},
"isRLSEnabled": false
},
"public.jobs": {
"name": "jobs",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"kind": {
"name": "kind",
"type": "text",
"primaryKey": false,
"notNull": true
},
"url": {
"name": "url",
"type": "text",
"primaryKey": false,
"notNull": true
},
"options": {
"name": "options",
"type": "jsonb",
"primaryKey": false,
"notNull": true,
"default": "'{}'::jsonb"
},
"status": {
"name": "status",
"type": "text",
"primaryKey": false,
"notNull": true,
"default": "'queued'"
},
"cache_key": {
"name": "cache_key",
"type": "text",
"primaryKey": false,
"notNull": true
},
"attempts": {
"name": "attempts",
"type": "integer",
"primaryKey": false,
"notNull": true,
"default": 0
},
"error": {
"name": "error",
"type": "text",
"primaryKey": false,
"notNull": false
},
"compiler_version": {
"name": "compiler_version",
"type": "text",
"primaryKey": false,
"notNull": false
},
"timings": {
"name": "timings",
"type": "jsonb",
"primaryKey": false,
"notNull": false
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"started_at": {
"name": "started_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"finished_at": {
"name": "finished_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
}
},
"indexes": {},
"foreignKeys": {},
"compositePrimaryKeys": {},
"uniqueConstraints": {},
"policies": {},
"checkConstraints": {},
"isRLSEnabled": false
},
"public.signup_tokens": {
"name": "signup_tokens",
"schema": "",
"columns": {
"id": {
"name": "id",
"type": "uuid",
"primaryKey": true,
"notNull": true,
"default": "gen_random_uuid()"
},
"email": {
"name": "email",
"type": "text",
"primaryKey": false,
"notNull": true
},
"token_hash": {
"name": "token_hash",
"type": "text",
"primaryKey": false,
"notNull": true
},
"created_at": {
"name": "created_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true,
"default": "now()"
},
"expires_at": {
"name": "expires_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": true
},
"consumed_at": {
"name": "consumed_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
}
},
"indexes": {},
"foreignKeys": {},
"compositePrimaryKeys": {},
"uniqueConstraints": {
"signup_tokens_token_hash_unique": {
"name": "signup_tokens_token_hash_unique",
"nullsNotDistinct": false,
"columns": [
"token_hash"
]
}
},
"policies": {},
"checkConstraints": {},
"isRLSEnabled": false
}
},
"enums": {},
"schemas": {},
"sequences": {},
"roles": {},
"policies": {},
"views": {},
"_meta": {
"columns": {},
"schemas": {},
"tables": {}
}
}
+8 -1
View File
@@ -15,6 +15,13 @@
"when": 1782790200000, "when": 1782790200000,
"tag": "0001_signup_tokens", "tag": "0001_signup_tokens",
"breakpoints": true "breakpoints": true
},
{
"idx": 2,
"version": "7",
"when": 1783202414928,
"tag": "0002_productive_wallflower",
"breakpoints": true
} }
] ]
} }
+22 -1
View File
@@ -1,6 +1,6 @@
import { and, desc, eq, gt, isNull, sql } from "drizzle-orm"; import { and, desc, eq, gt, isNull, sql } from "drizzle-orm";
import type { Db } from "./client.js"; import type { Db } from "./client.js";
import { jobs, clones, cache, apiKeys, signupTokens, type Job, type NewJob, type Clone, type NewClone, type CacheRow, type ApiKey, type SignupToken } from "./schema.js"; import { jobs, clones, cache, apiKeys, signupTokens, jobEvents, type Job, type NewJob, type Clone, type NewClone, type CacheRow, type ApiKey, type SignupToken } from "./schema.js";
// ---- jobs ---- // ---- jobs ----
@@ -111,3 +111,24 @@ export async function consumeSignupToken(db: Db, tokenHash: string): Promise<Sig
.returning(); .returning();
return row; return row;
} }
// ---- job events ----
export async function appendJobEvent(db: Db, jobId: string, payload: Record<string, unknown>): Promise<number> {
const [row] = await db
.select({ n: sql<number>`coalesce(max(${jobEvents.seq}), 0)` })
.from(jobEvents)
.where(eq(jobEvents.jobId, jobId));
const seq = (row?.n ?? 0) + 1;
await db.insert(jobEvents).values({ jobId, seq, payload: { t: Date.now(), ...payload } });
return seq;
}
export async function listJobEvents(db: Db, jobId: string, after = 0): Promise<Array<Record<string, unknown>>> {
const rows = await db
.select()
.from(jobEvents)
.where(and(eq(jobEvents.jobId, jobId), sql`${jobEvents.seq} > ${after}`))
.orderBy(jobEvents.seq);
return rows.map((r) => r.payload as Record<string, unknown>);
}
+13 -1
View File
@@ -1,4 +1,4 @@
import { pgTable, text, jsonb, integer, timestamp, uuid } from "drizzle-orm/pg-core"; import { pgTable, text, jsonb, integer, timestamp, uuid, index } from "drizzle-orm/pg-core";
/** Job lifecycle row. Status mirrors the queue; `cacheKey` ties it to the cache row. /** Job lifecycle row. Status mirrors the queue; `cacheKey` ties it to the cache row.
* `options`/`timings` are jsonb (the service's typed shapes). */ * `options`/`timings` are jsonb (the service's typed shapes). */
@@ -66,6 +66,17 @@ export const signupTokens = pgTable("signup_tokens", {
consumedAt: timestamp("consumed_at", { withTimezone: true }), consumedAt: timestamp("consumed_at", { withTimezone: true }),
}); });
/** Structured clone progress events (mirrors in-memory backend event stream). */
export const jobEvents = pgTable("job_events", {
id: uuid("id").defaultRandom().primaryKey(),
jobId: uuid("job_id")
.notNull()
.references(() => jobs.id, { onDelete: "cascade" }),
seq: integer("seq").notNull(),
payload: jsonb("payload").notNull(),
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
}, (t) => [index("job_events_job_id_seq_idx").on(t.jobId, t.seq)]);
export type Job = typeof jobs.$inferSelect; export type Job = typeof jobs.$inferSelect;
export type NewJob = typeof jobs.$inferInsert; export type NewJob = typeof jobs.$inferInsert;
export type Clone = typeof clones.$inferSelect; export type Clone = typeof clones.$inferSelect;
@@ -74,3 +85,4 @@ export type CacheRow = typeof cache.$inferSelect;
export type ApiKey = typeof apiKeys.$inferSelect; export type ApiKey = typeof apiKeys.$inferSelect;
export type SignupToken = typeof signupTokens.$inferSelect; export type SignupToken = typeof signupTokens.$inferSelect;
export type NewSignupToken = typeof signupTokens.$inferInsert; export type NewSignupToken = typeof signupTokens.$inferInsert;
export type JobEvent = typeof jobEvents.$inferSelect;
+10 -1
View File
@@ -37,7 +37,16 @@ export async function processCloneJob(deps: ProcessDeps, jobId: string): Promise
const options = (job.options ?? {}) as CloneOptions; const options = (job.options ?? {}) as CloneOptions;
// Provision the isolated harness only when this job actually verifies (build). // Provision the isolated harness only when this job actually verifies (build).
const harnessDir = (options.verify || options.asyncVerify) && deps.harnessProvider ? await deps.harnessProvider() : undefined; const harnessDir = (options.verify || options.asyncVerify) && deps.harnessProvider ? await deps.harnessProvider() : undefined;
const result = await deps.runJob({ url: job.url, options, runsDir: base, harnessDir, tier: deps.tier, captureCacheDir: deps.captureCacheDir || undefined, captureCacheTtlMs: deps.cacheTtlMs }); const result = await deps.runJob({
url: job.url,
options,
runsDir: base,
harnessDir,
tier: deps.tier,
captureCacheDir: deps.captureCacheDir || undefined,
captureCacheTtlMs: deps.cacheTtlMs,
log: (e) => { void repo.appendJobEvent(db, jobId, e); },
});
const manifest = await store.putClone(jobId, result.files); const manifest = await store.putClone(jobId, result.files);
const envelope = { files: manifest.files, routes: result.routes, bundleKey: manifest.bundleKey }; const envelope = { files: manifest.files, routes: result.routes, bundleKey: manifest.bundleKey };