Simplify published job execution

packages/worker/migrations/0028-published-bundle-artifacts-and-archived-jobs.sql

@@ -0,0 +1,45 @@
+CREATE TABLE IF NOT EXISTS published_bundle_artifacts (
+	id TEXT PRIMARY KEY,
+	user_id TEXT NOT NULL,
+	source_id TEXT NOT NULL,
+	published_commit TEXT NOT NULL,
+	artifact_kind TEXT NOT NULL,
+	artifact_name TEXT,
+	entry_point TEXT NOT NULL,
+	kv_key TEXT NOT NULL,
+	dependencies_json TEXT NOT NULL DEFAULT '[]',
+	created_at TEXT NOT NULL,
+	updated_at TEXT NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_published_bundle_artifacts_source_identity
+ON published_bundle_artifacts(
+	user_id,
+	source_id,
+	artifact_kind,
+	COALESCE(artifact_name, ''),
+	entry_point
+);
+
+CREATE INDEX IF NOT EXISTS idx_published_bundle_artifacts_user_id
+ON published_bundle_artifacts(user_id);
+
+CREATE INDEX IF NOT EXISTS idx_published_bundle_artifacts_source_id
+ON published_bundle_artifacts(source_id);
+
+CREATE TABLE IF NOT EXISTS archived_job_artifacts (
+	job_id TEXT PRIMARY KEY,
+	user_id TEXT NOT NULL,
+	source_id TEXT NOT NULL,
+	published_commit TEXT NOT NULL,
+	storage_id TEXT NOT NULL,
+	retain_until TEXT NOT NULL,
+	created_at TEXT NOT NULL,
+	updated_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_archived_job_artifacts_user_id
+ON archived_job_artifacts(user_id);
+
+CREATE INDEX IF NOT EXISTS idx_archived_job_artifacts_retain_until
+ON archived_job_artifacts(retain_until);

packages/worker/src/env-schema.ts

@@ -165,6 +165,15 @@ export const EnvSchema = object({
 		'COOKIE_SECRET must be at least 32 characters for session signing.',
 	),
 	APP_DB: d1DatabaseSchema,
+	BUNDLE_ARTIFACTS_KV: createSchema<unknown, KVNamespace>((value, context) => {
+		if (value) {
+			return { value: value as KVNamespace }
+		}
+		return fail(
+			'Missing BUNDLE_ARTIFACTS_KV binding for published runtime artifacts.',
+			context.path,
+		)
+	}),
 	JOB_MANAGER: createSchema<unknown, DurableObjectNamespace>(
 		(value, context) => {
 			if (value) {

packages/worker/src/jobs/archived-artifacts-repo.ts

@@ -0,0 +1,100 @@
+export type ArchivedJobArtifactRecord = {
+	id: string
+	jobId: string
+	userId: string
+	sourceId: string
+	publishedCommit: string
+	storageId: string
+	retainUntil: string
+	createdAt: string
+	updatedAt: string
+}
+
+export async function upsertArchivedJobArtifact(input: {
+	db: D1Database
+	jobId: string
+	userId: string
+	sourceId: string
+	publishedCommit: string
+	storageId: string
+	retainUntil: string
+}) {
+	const now = new Date().toISOString()
+	const existing = await input.db
+		.prepare(
+			`SELECT id FROM archived_job_artifacts WHERE job_id = ? AND user_id = ? LIMIT 1`,
+		)
+		.bind(input.jobId, input.userId)
+		.first<{ id: string }>()
+	if (existing?.id) {
+		await input.db
+			.prepare(
+				`UPDATE archived_job_artifacts
+				SET source_id = ?, published_commit = ?, storage_id = ?, retain_until = ?, updated_at = ?
+				WHERE id = ?`,
+			)
+			.bind(
+				input.sourceId,
+				input.publishedCommit,
+				input.storageId,
+				input.retainUntil,
+				now,
+				existing.id,
+			)
+			.run()
+		return existing.id
+	}
+	const id = crypto.randomUUID()
+	await input.db
+		.prepare(
+			`INSERT INTO archived_job_artifacts (
+				id, job_id, user_id, source_id, published_commit, storage_id, retain_until, created_at, updated_at
+			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		)
+		.bind(
+			id,
+			input.jobId,
+			input.userId,
+			input.sourceId,
+			input.publishedCommit,
+			input.storageId,
+			input.retainUntil,
+			now,
+			now,
+		)
+		.run()
+	return id
+}
+
+export async function listArchivedJobArtifactsDueBefore(
+	db: D1Database,
+	retainUntil: string,
+): Promise<Array<ArchivedJobArtifactRecord>> {
+	const { results } = await db
+		.prepare(
+			`SELECT id, job_id, user_id, source_id, published_commit, storage_id, retain_until, created_at, updated_at
+			FROM archived_job_artifacts
+			WHERE retain_until <= ?
+			ORDER BY retain_until ASC`,
+		)
+		.bind(retainUntil)
+		.all<Record<string, unknown>>()
+	return (results ?? []).map((row) => ({
+		id: String(row['id']),
+		jobId: String(row['job_id']),
+		userId: String(row['user_id']),
+		sourceId: String(row['source_id']),
+		publishedCommit: String(row['published_commit']),
+		storageId: String(row['storage_id']),
+		retainUntil: String(row['retain_until']),
+		createdAt: String(row['created_at']),
+		updatedAt: String(row['updated_at']),
+	}))
+}
+
+export async function deleteArchivedJobArtifact(db: D1Database, id: string) {
+	await db
+		.prepare(`DELETE FROM archived_job_artifacts WHERE id = ?`)
+		.bind(id)
+		.run()
+}

packages/worker/src/jobs/scheduler-logging.ts

@@ -22,7 +22,18 @@ type JobSchedulerLogPayload = {
 	userId?: string
 	jobId?: string | null
 	scheduleType?: JobSchedule['type']
+	sourceId?: string | null
 	source?: SchedulerLogSource
+	artifactKind?: 'app' | 'job' | 'module' | null
+	artifactName?: string | null
+	artifactEntryPoint?: string | null
+	artifactCacheHit?: boolean
+	artifactPublishedCommit?: string | null
+	snapshotSource?: 'kv' | 'artifacts' | null
+	buildDurationMs?: number
+	dependencyCount?: number
+	cleanupDeletedSourceId?: string | null
+	cleanupDeletedStorageId?: string | null
 	nextJobId?: string | null
 	nextRunAt?: string | null
 	currentAlarmAt?: string | null