feat(registry): write SKILL.md bodies to R2 by content hash

docs/arch/cron.md

@@ -2,7 +2,7 @@
 
 Generated from literal `defineScheduledTask` declarations. Run `pnpm cron:docs` after schedule changes.
 
-Tasks: 25. Unique Cloudflare triggers: 16.
+Tasks: 26. Unique Cloudflare triggers: 16.
 
 | Task | Cron | Source |
 | --- | --- | --- |
@@ -11,6 +11,7 @@ Tasks: 25. Unique Cloudflare triggers: 16.
 | `ai-generate-submit` | `15 * * * *` | `layers/registry/server/tasks/ai-generate-submit.ts` |
 | `ai-ready:cron` | `*/5 * * * *` | `server/tasks/ai-ready-cron.ts` |
 | `backfill-skill-assets` | `*/5 * * * *` | `layers/registry/server/tasks/backfill-skill-assets.ts` |
+| `backfill-skill-sources` | `*/5 * * * *` | `layers/registry/server/tasks/backfill-skill-sources.ts` |
 | `classify-repository-purpose` | `0 * * * *` | `layers/registry/server/tasks/classify-repository-purpose.ts` |
 | `cleanup-auto-index-rate-limits` | `0 * * * *` | `layers/registry/server/tasks/cleanup-auto-index-rate-limits.ts` |
 | `detect-star-surges` | `30 4 * * *` | `layers/registry/server/tasks/detect-star-surges.ts` |

layers/registry/server/tasks/backfill-skill-sources.ts

@@ -0,0 +1,56 @@
+/// <reference types="@cloudflare/workers-types" />
+
+import { resolveCloudflareBindings } from '@harlan-zw/nuxt-cloudflare/bindings'
+import { runObservedScheduledTask } from '~~/server/utils/scheduled-run'
+import { reportJobRun } from '~~/server/utils/sync-job-reporter'
+import { observedSchedulePolicy } from '#shared/schedule-policy'
+import { backfillSkillSources } from '../utils/skill-source-backfill'
+
+const CRON = '*/5 * * * *'
+
+/**
+ * Scheduled task: copy every stored SKILL.md body to R2 (ADR-0019).
+ *
+ * The sync writes new bodies to R2 as it goes. This task walks the rows that
+ * predate that, 200 a run by rowid, so every row's hash names an object
+ * before the D1 column is nulled. Once the walk reaches the end of the table
+ * each run is one indexed read and no puts.
+ */
+export default defineScheduledTask({
+  name: 'backfill-skill-sources',
+  cron: '*/5 * * * *',
+  description: 'Copy stored SKILL.md bodies to R2 by content hash',
+  async run({ context }) {
+    const env = resolveCloudflareBindings<Cloudflare.Env>(context)
+    const db = env?.DB as D1Database | undefined
+    const bucket = env?.PUBLIC_ARTIFACTS
+    if (!env || !db || !bucket) {
+      emitOperationalEvent(createWideEvent({ operation: 'backfill-skill-sources', outcome: 'binding-missing' }))
+      return { result: { error: 'no-db' } }
+    }
+
+    return await runObservedScheduledTask({
+      db,
+      env,
+      context,
+      policy: observedSchedulePolicy('backfill-skill-sources'),
+    }, async () => {
+      const startedAt = Date.now()
+      const report = await backfillSkillSources(db, bucket, { now: Math.floor(startedAt / 1000) })
+      emitOperationalEvent(createWideEvent({
+        'operation': 'backfill-skill-sources',
+        'outcome': report.failed > 0 ? 'partial' : 'completed',
+        'scanned.count': report.scanned,
+        'success.count': report.stored + report.existing,
+        'failed.count': report.failed,
+      }))
+      await reportJobRun(db, 'backfill-skill-sources', {
+        cron: CRON,
+        status: report.failed > 0 ? 'partial' : 'ok',
+        durationMs: Date.now() - startedAt,
+        error: report.failed > 0 ? `${report.failed} R2 put failed; cursor held at ${report.position}` : null,
+      })
+      return { result: report }
+    })
+  },
+})

layers/registry/server/utils/skill-source-backfill.spec.ts

@@ -0,0 +1,108 @@
+import type { SkillSourceBucket } from './skill-source-store'
+import Database from 'better-sqlite3'
+import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { backfillSkillSources, SKILL_SOURCE_BACKFILL_CURSOR } from './skill-source-backfill'
+import { skillSourceR2Key } from './skill-source-store'
+
+describe('backfillSkillSources', () => {
+  let sqlite: Database.Database
+
+  beforeEach(() => {
+    sqlite = new Database(':memory:')
+    sqlite.exec(`
+      CREATE TABLE skills (
+        owner TEXT NOT NULL, repo TEXT NOT NULL, name TEXT NOT NULL,
+        rendered_raw TEXT, rendered_raw_sha256 TEXT,
+        PRIMARY KEY (owner, repo, name)
+      );
+      CREATE TABLE registry_cursors (
+        name TEXT PRIMARY KEY, position INTEGER NOT NULL DEFAULT 0, updated_at INTEGER NOT NULL
+      );
+    `)
+    const insert = sqlite.prepare(`INSERT INTO skills (owner, repo, name, rendered_raw, rendered_raw_sha256) VALUES ('acme', 'skills', ?, ?, ?)`)
+    insert.run('one', '# One', 'a'.repeat(64))
+    insert.run('unrendered', null, null)
+    insert.run('two', '# Two', 'b'.repeat(64))
+    insert.run('three', '# Three', 'c'.repeat(64))
+  })
+
+  afterEach(() => sqlite.close())
+
+  it('walks the table by rowid across runs and stores each body once', async () => {
+    const bucket = memoryBucket()
+
+    const first = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 1, limit: 2 })
+    const second = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 2, limit: 2 })
+    const third = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 3, limit: 2 })
+
+    expect(first).toEqual({ scanned: 2, stored: 2, existing: 0, failed: 0, position: 3 })
+    expect(second).toEqual({ scanned: 1, stored: 1, existing: 0, failed: 0, position: 4 })
+    expect(third).toEqual({ scanned: 0, stored: 0, existing: 0, failed: 0, position: 4 })
+    expect([...bucket.objects.keys()].sort()).toEqual(
+      ['a', 'b', 'c'].map(hex => skillSourceR2Key(hex.repeat(64))).sort(),
+    )
+    expect(cursor()).toBe(4)
+  })
+
+  it('holds the cursor before a failed put so the next run retries it', async () => {
+    const bucket = memoryBucket(new Set([skillSourceR2Key('b'.repeat(64))]))
+
+    const report = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 1, limit: 10 })
+
+    expect(report).toEqual({ scanned: 2, stored: 1, existing: 0, failed: 1, position: 1 })
+    expect(cursor()).toBe(1)
+  })
+
+  it('reports an object that already exists without storing it again', async () => {
+    const bucket = memoryBucket()
+    bucket.objects.set(skillSourceR2Key('a'.repeat(64)), new Uint8Array())
+
+    const report = await backfillSkillSources(wrapSqlite(sqlite), bucket, { now: 1, limit: 1 })
+
+    expect(report).toEqual({ scanned: 1, stored: 0, existing: 1, failed: 0, position: 1 })
+  })
+
+  function cursor(): number | undefined {
+    return sqlite.prepare(`SELECT position FROM registry_cursors WHERE name = ?`).pluck().get(SKILL_SOURCE_BACKFILL_CURSOR) as number | undefined
+  }
+})
+
+function memoryBucket(failKeys = new Set<string>()): SkillSourceBucket & { objects: Map<string, Uint8Array> } {
+  const objects = new Map<string, Uint8Array>()
+  return {
+    objects,
+    put: (async (key: string, value: Uint8Array, options: { onlyIf?: { etagDoesNotMatch?: string } }) => {
+      if (failKeys.has(key))
+        throw new Error('r2 unavailable')
+      if (options.onlyIf?.etagDoesNotMatch === '*' && objects.has(key))
+        return null
+      objects.set(key, value)
+      return { key }
+    }) as unknown as SkillSourceBucket['put'],
+    get: (async () => null) as unknown as SkillSourceBucket['get'],
+    head: (async () => null) as unknown as SkillSourceBucket['head'],
+  }
+}
+
+function wrapSqlite(sqlite: Database.Database): D1Database {
+  return {
+    prepare(sql: string) {
+      return {
+        bind(...params: unknown[]) {
+          return {
+            async all<T>() {
+              return { results: sqlite.prepare(sql).all(...params) as T[] }
+            },
+            async first<T>() {
+              return (sqlite.prepare(sql).get(...params) as T | undefined) ?? null
+            },
+            async run() {
+              const info = sqlite.prepare(sql).run(...params)
+              return { meta: { changes: info.changes } }
+            },
+          }
+        },
+      }
+    },
+  } as unknown as D1Database
+}

layers/registry/server/utils/skill-source-backfill.ts

@@ -0,0 +1,87 @@
+/// <reference types="@cloudflare/workers-types" />
+
+import type { SkillSourceBucket } from './skill-source-store'
+import { putSkillSource } from './skill-source-store'
+
+export const SKILL_SOURCE_BACKFILL_CURSOR = 'skill-sources-r2-v1'
+
+/**
+ * Rows one run copies. Each is one R2 put, and a put that already exists is
+ * one conditional request. 200 keeps a run under the subrequest budget with
+ * room for its D1 reads, and copies the 25,145 rows of 2026-11-12 in about
+ * ten hours of five-minute runs.
+ */
+export const SKILL_SOURCE_BACKFILL_BATCH = 200
+
+interface SourceRow {
+  rowid: number
+  rendered_raw: string
+  rendered_raw_sha256: string
+}
+
+export interface SkillSourceBackfillReport {
+  /** Rows read past the cursor this run. Zero means the walk is complete. */
+  scanned: number
+  stored: number
+  existing: number
+  failed: number
+  /** The rowid the next run starts after. */
+  position: number
+}
+
+/**
+ * Copy the next batch of stored SKILL.md bodies to R2, by rowid.
+ *
+ * The cursor only advances past rows whose put succeeded or found the object
+ * already there. A failed put holds the cursor at the row before it, so the
+ * next run retries from that row instead of leaving a hole the readers would
+ * fall back to D1 for forever.
+ *
+ * A row the sync rewrites after this walk passes it is covered by the sync's
+ * own dual write, so one pass is enough.
+ */
+export async function backfillSkillSources(
+  db: D1Database,
+  bucket: SkillSourceBucket,
+  input: { now: number, limit?: number },
+): Promise<SkillSourceBackfillReport> {
+  const limit = input.limit ?? SKILL_SOURCE_BACKFILL_BATCH
+  const cursor = await db.prepare(`SELECT position FROM registry_cursors WHERE name = ?`)
+    .bind(SKILL_SOURCE_BACKFILL_CURSOR)
+    .first<{ position: number }>()
+  const start = cursor?.position ?? 0
+
+  const rows = await db.prepare(
+    `SELECT rowid, rendered_raw, rendered_raw_sha256
+     FROM skills
+     WHERE rowid > ?
+       AND rendered_raw IS NOT NULL
+       AND rendered_raw_sha256 IS NOT NULL
+     ORDER BY rowid
+     LIMIT ?`,
+  ).bind(start, limit).all<SourceRow>()
+
+  const report: SkillSourceBackfillReport = { scanned: 0, stored: 0, existing: 0, failed: 0, position: start }
+  for (const row of rows.results ?? []) {
+    report.scanned += 1
+    const result = await putSkillSource(bucket, { body: row.rendered_raw, contentSha256: row.rendered_raw_sha256 })
+    if (result._tag === 'failed') {
+      report.failed += 1
+      console.warn(`[backfill-skill-sources] put failed at rowid ${row.rowid}: ${result.reason}`)
+      break
+    }
+    if (result._tag === 'stored')
+      report.stored += 1
+    else
+      report.existing += 1
+    report.position = row.rowid
+  }
+
+  if (report.position !== start) {
+    await db.prepare(
+      `INSERT INTO registry_cursors (name, position, updated_at) VALUES (?, ?, ?)
+       ON CONFLICT(name) DO UPDATE SET position = excluded.position, updated_at = excluded.updated_at`,
+    ).bind(SKILL_SOURCE_BACKFILL_CURSOR, report.position, input.now).run()
+  }
+  return report
+}

layers/registry/server/utils/skill-source-store.spec.ts

@@ -0,0 +1,73 @@
+import type { SkillSourceBucket } from './skill-source-store'
+import { describe, expect, it } from 'vitest'
+import { skillContentSha256 } from './skill-content-hash'
+import { hasSkillSource, putSkillSource, readSkillSource, skillSourceR2Key } from './skill-source-store'
+
+interface StoredObject {
+  bytes: Uint8Array
+  sha256: unknown
+  customMetadata: unknown
+}
+
+function memoryBucket(objects = new Map<string, StoredObject>(), failPut = false): SkillSourceBucket & { objects: Map<string, StoredObject> } {
+  return {
+    objects,
+    put: (async (key: string, value: Uint8Array, options: { onlyIf?: { etagDoesNotMatch?: string }, sha256?: unknown, customMetadata?: unknown }) => {
+      if (failPut)
+        throw new Error('r2 unavailable')
+      if (options.onlyIf?.etagDoesNotMatch === '*' && objects.has(key))
+        return null
+      objects.set(key, { bytes: value, sha256: options.sha256, customMetadata: options.customMetadata })
+      return { key }
+    }) as unknown as SkillSourceBucket['put'],
+    get: (async (key: string) => {
+      const object = objects.get(key)
+      return object ? { text: async () => new TextDecoder().decode(object.bytes) } : null
+    }) as unknown as SkillSourceBucket['get'],
+    head: (async (key: string) => (objects.has(key) ? { key } : null)) as unknown as SkillSourceBucket['head'],
+  }
+}
+
+describe('skill source store', () => {
+  it('stores a body under its hash and reports a repeat as existing', async () => {
+    const bucket = memoryBucket()
+    const body = '---\nname: one\n---\n# One'
+    const contentSha256 = await skillContentSha256(body)
+
+    const first = await putSkillSource(bucket, { body, contentSha256 })
+    const second = await putSkillSource(bucket, { body, contentSha256 })
+
+    expect(first).toEqual({ _tag: 'stored', key: skillSourceR2Key(contentSha256) })
+    expect(second).toEqual({ _tag: 'existing', key: skillSourceR2Key(contentSha256) })
+    expect(bucket.objects.get(skillSourceR2Key(contentSha256))?.sha256).toBe(contentSha256)
+    expect(await hasSkillSource(bucket, contentSha256)).toBe(true)
+  })
+
+  it('returns a failure value instead of throwing when the put fails', async () => {
+    const bucket = memoryBucket(new Map(), true)
+
+    const result = await putSkillSource(bucket, { body: 'x', contentSha256: 'ab'.repeat(32) })
+
+    expect(result).toMatchObject({ _tag: 'failed', reason: 'r2 unavailable' })
+  })
+
+  it('reads R2 first and falls back to the stored D1 body', async () => {
+    const bucket = memoryBucket()
+    const body = '# From R2'
+    const contentSha256 = await skillContentSha256(body)
+    await putSkillSource(bucket, { body, contentSha256 })
+
+    await expect(readSkillSource(bucket, { contentSha256, storedBody: '# From D1' }))
+      .resolves
+      .toEqual({ _tag: 'ok', body, from: 'r2' })
+    await expect(readSkillSource(bucket, { contentSha256: 'cd'.repeat(32), storedBody: '# From D1' }))
+      .resolves
+      .toEqual({ _tag: 'ok', body: '# From D1', from: 'd1' })
+    await expect(readSkillSource(null, { contentSha256, storedBody: '# From D1' }))
+      .resolves
+      .toEqual({ _tag: 'ok', body: '# From D1', from: 'd1' })
+    await expect(readSkillSource(bucket, { contentSha256: null, storedBody: null }))
+      .resolves
+      .toEqual({ _tag: 'missing' })
+  })
+})