perf(d1): bound scoring and discovery query work

layers/registry/server/api/tags/[slug].get.ts

@@ -122,34 +122,40 @@ async function buildTagProfile(event: H3Event, slug: string): Promise<TagProfile
   // FTS post-0049 indexes `repo` too, so the tuple match is precise across
   // same-(owner,name) collisions.
   //
-  // The repo Skill count is added outside the LIMIT. In the inner select list
-  // it ran for every match before the sort, reading every Skill of each
-  // match's repo: about 144K rows for `testing`.
+  // UNION collects identity candidates before reading Skills. The previous
+  // OR walked Skills and expanded generated tag JSON once per Skill.
+  // Limit narrow identities before hydrating columns, authors, and repo counts.
   const skillsRes = await db
     .prepare(
-      `SELECT tagged.*,
+      `WITH candidates AS (
+         SELECT owner, repo, name FROM skills_fts WHERE skills_fts MATCH ?
+         UNION
+         SELECT owner, repo, name FROM skills WHERE owner = ?
+         UNION
+         SELECT sg.owner, sg.repo, sg.name
+         FROM skill_generated sg, json_each(sg.payload, '$.tags') je
+         WHERE sg.kind = 'tags' AND je.value = ?
+       ), tagged AS (
+         SELECT s.owner, s.repo, s.name, r.stars
+         FROM candidates c
+         JOIN skills s ON s.owner = c.owner AND s.repo = c.repo AND s.name = c.name
+         JOIN repos r ON r.owner = s.owner AND r.repo = s.repo
+         WHERE s.source_resolved = 1 AND ${NOT_BROKEN_SQL}
+         ORDER BY r.stars DESC, s.name ASC
+         LIMIT 200
+       )
+       SELECT s.name, s.owner, s.repo, s.display_name, s.slug, tagged.stars, s.like_count,
+              s.description, s.rendered_raw_sha256, r.pushed_at, s.modified_at, s.first_seen_at,
+              s.rendered_skill_path, s.current_sha, r.default_branch, r.source_owner, r.source_repo,
+              o.name AS author_name,
               (SELECT COUNT(*) FROM skills repo_skills
                WHERE repo_skills.owner = tagged.owner
                  AND repo_skills.repo = tagged.repo
                  AND repo_skills.source_resolved = 1) AS repo_skill_count
-       FROM (
-       SELECT DISTINCT s.name, s.owner, s.repo, s.display_name, s.slug, r.stars, s.like_count, s.description, s.rendered_raw_sha256, r.pushed_at, s.modified_at, s.first_seen_at,
-              s.rendered_skill_path, s.current_sha, r.default_branch, r.source_owner, r.source_repo,
-              (SELECT o.name FROM owners o WHERE o.owner = s.owner) AS author_name
-       FROM skills s
+       FROM tagged
+       JOIN skills s ON s.owner = tagged.owner AND s.repo = tagged.repo AND s.name = tagged.name
        JOIN repos r ON r.owner = s.owner AND r.repo = s.repo
-       WHERE s.source_resolved = 1 AND ${NOT_BROKEN_SQL} AND (
-         (s.owner, s.repo, s.name) IN (SELECT owner, repo, name FROM skills_fts WHERE skills_fts MATCH ?)
-         OR s.owner = ?
-         OR EXISTS (
-           SELECT 1 FROM skill_generated sg, json_each(sg.payload, '$.tags') je
-           WHERE sg.owner = s.owner AND sg.repo = s.repo AND sg.name = s.name
-             AND sg.kind = 'tags' AND je.value = ?
-         )
-       )
-       ORDER BY r.stars DESC, s.name ASC
-       LIMIT 200
-       ) tagged
+       LEFT JOIN owners o ON o.owner = s.owner
        ORDER BY tagged.stars DESC, tagged.name ASC`,
     )
     .bind(ftsQuery, slug, slug)

layers/registry/server/tasks/drain-skill-dirty.spec.ts

@@ -29,7 +29,6 @@ vi.mock('#shared/schedule-policy', () => ({
 
 vi.mock('../utils/recompute-scores', () => ({
   recomputeIndexabilityForSkill: vi.fn(),
-  recomputeTrustForSkill: vi.fn(),
 }))
 
 vi.stubGlobal('defineScheduledTask', (task: unknown) => task)
@@ -57,4 +56,27 @@ describe('drain-skill-dirty task', () => {
       durationMs: expect.any(Number),
     })
   })
+
+  it('recomputes and removes successful queue entries', async () => {
+    const key = { owner: 'acme', repo: 'tools', name: 'review' }
+    const run = vi.fn().mockResolvedValue({ meta: { changes: 1 } })
+    const all = vi.fn().mockResolvedValue({ results: [key] })
+    const bound = { all, run }
+    const bind = vi.fn(() => bound)
+    const batch = vi.fn().mockResolvedValue([])
+    const db = { prepare: vi.fn(() => ({ bind })), batch }
+    mocks.resolveCloudflareBindings.mockReturnValue({ DB: db })
+
+    const result = await task.run({ context: {} } as never)
+
+    expect(result).toEqual({ result: { drained: 1, failed: 0, scanned: 1 } })
+    expect(run).toHaveBeenCalledOnce()
+    expect(batch).toHaveBeenCalledWith([bound])
+    expect(mocks.reportJobRun).toHaveBeenCalledWith(db, 'drain-skill-dirty', {
+      cron: '*/5 * * * *',
+      status: 'ok',
+      durationMs: expect.any(Number),
+      error: null,
+    })
+  })
 })

layers/registry/server/tasks/drain-skill-dirty.ts

@@ -4,10 +4,7 @@ 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 {
-  recomputeIndexabilityForSkill,
-  recomputeTrustForSkill,
-} from '../utils/recompute-scores'
+import { recomputeIndexabilityForSkill } from '../utils/recompute-scores'
 import { SKILL_COUNTER_RECOMPUTE_SQL } from '../utils/skill-counter-recompute'
 
 const BATCH = 200
@@ -89,10 +86,8 @@ export default defineScheduledTask({
           .then(async () => {
           // Counters just moved; re-score indexability + trust for this
           // skill so seo_index_score / trust_tier track the new inputs.
-          // Indexability writes both sets of columns in one UPDATE; trust
-          // helper is a no-op when indexability already covered the change.
+          // Indexability writes both sets of columns in one UPDATE.
             await recomputeIndexabilityForSkill(db, { owner, repo, name })
-            await recomputeTrustForSkill(db, { owner, repo, name })
             updated++
             successful.push({ owner, repo, name })
           })

layers/registry/server/utils/ai-generation-work.ts

@@ -108,7 +108,10 @@ export type PersistAbstractnessResult
   = | { _tag: 'written' }
     | { _tag: 'source_changed' }
 
-const ELIGIBLE_SKILL_SQL = `
+// Match String.trim(), including Unicode whitespace.
+const NONEMPTY_SOURCE_SQL = `LENGTH(TRIM(s.rendered_raw, char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279))) > 0`
+
+const ELIGIBLE_SKILL_BASE_SQL = `
   SELECT
     s.owner,
     s.repo,
@@ -121,28 +124,30 @@ const ELIGIBLE_SKILL_SQL = `
   WHERE r.broken_since IS NULL
     AND s.current_sha IS NOT NULL
     AND s.rendered_raw IS NOT NULL
-    -- Match String.trim(): empty content cannot produce an embedding or classification.
-    AND LENGTH(TRIM(s.rendered_raw, char(9, 10, 11, 12, 13, 32, 160, 5760, 8192, 8193, 8194, 8195, 8196, 8197, 8198, 8199, 8200, 8201, 8202, 8232, 8233, 8239, 8287, 12288, 65279))) > 0
     AND s.rendered_status = 'ok'
     AND s.seo_indexable = 1
 `
 
+const ELIGIBLE_SKILL_SQL = `${ELIGIBLE_SKILL_BASE_SQL} AND ${NONEMPTY_SOURCE_SQL}`
+
 /**
  * Count the embedding backlog so the limit can respond to it. One extra query
  * against a budget of 850 is a cheap price for not silently falling behind.
  */
 export async function countMissingEmbeddings(db: D1Database): Promise<number> {
   const row = await db.prepare(
-    `SELECT COUNT(*) AS pending FROM (${ELIGIBLE_SKILL_SQL}
-      AND NOT EXISTS (
+    // CASE evaluates the source only when its embedding needs work. Trimming
+    // every completed source cost 2.2 seconds on the live catalog.
+    `SELECT COUNT(*) AS pending FROM (${ELIGIBLE_SKILL_BASE_SQL}
+      AND CASE WHEN NOT EXISTS (
         SELECT 1
         FROM skill_generated generated
         WHERE generated.owner = s.owner
           AND generated.repo = s.repo
           AND generated.name = s.name
           AND generated.kind = 'embedding'
           AND generated.sha = s.current_sha
-      )
+      ) THEN ${NONEMPTY_SOURCE_SQL} ELSE 0 END
     )`,
   ).first<{ pending: number }>()
   return Number(row?.pending ?? 0)

layers/registry/server/utils/recompute-scores.ts

@@ -88,7 +88,10 @@ const BASE_SELECT = `
   s.trust_score,
   s.trust_reasons,
   o.tier AS override_tier,
-  o.reason AS override_reason,
+  o.reason AS override_reason
+`
+
+const LIVE_COUNTER_SELECT = `
   (
     SELECT COUNT(*)
     FROM collection_skills_v2 cs
@@ -135,6 +138,44 @@ const FROM_JOIN = `
   LEFT JOIN repo_trust_overrides o ON o.owner = s.owner AND o.repo = s.repo
 `
 
+// Group once for the full pass. Counting siblings for every skill makes
+// large repositories cost quadratically more reads.
+const ALL_COUNTERS_CTE = `
+  WITH curator_counts AS (
+    SELECT cs.owner, cs.repo, cs.name,
+      COUNT(*) AS curator_count,
+      SUM(CASE WHEN length(trim(COALESCE(cs.reason, ''))) >= 20 THEN 1 ELSE 0 END) AS curator_reason_count
+    FROM collection_skills_v2 cs
+    JOIN collections_v2 c ON c.id = cs.collection_id
+    WHERE c.deleted_at IS NULL
+    GROUP BY cs.owner, cs.repo, cs.name
+  ), social_counts AS (
+    SELECT skill_slug, COUNT(*) AS approved_social_count,
+      SUM(CASE WHEN role = 'author' THEN 1 ELSE 0 END) AS author_social_count
+    FROM skill_social_posts
+    WHERE status = 'approved'
+    GROUP BY skill_slug
+  ), repo_counts AS (
+    SELECT owner, repo, COUNT(*) AS repo_skill_count
+    FROM skills
+    GROUP BY owner, repo
+  )
+`
+
+const ALL_COUNTER_SELECT = `
+  COALESCE(cc.curator_count, 0) AS curator_count,
+  COALESCE(cc.curator_reason_count, 0) AS curator_reason_count,
+  COALESCE(sc.approved_social_count, 0) AS approved_social_count,
+  COALESCE(sc.author_social_count, 0) AS author_social_count,
+  CASE WHEN r.broken_since IS NULL THEN COALESCE(rc.repo_skill_count, 0) ELSE 0 END AS repo_skill_count
+`
+
+const ALL_COUNTER_JOINS = `
+  LEFT JOIN curator_counts cc ON cc.owner = s.owner AND cc.repo = s.repo AND cc.name = s.name
+  LEFT JOIN social_counts sc ON sc.skill_slug = s.slug
+  LEFT JOIN repo_counts rc ON rc.owner = s.owner AND rc.repo = s.repo
+`
+
 function computeFromRow(row: ScoreRow, now: number): {
   isOfficial: boolean
   sourceResolved: boolean
@@ -290,7 +331,7 @@ function updateTrustStmt(
 
 async function fetchOne(db: D1Database, key: SkillKey): Promise<ScoreRow | null> {
   const row = await db
-    .prepare(`SELECT ${BASE_SELECT} ${FROM_JOIN} WHERE s.owner = ?1 AND s.repo = ?2 AND s.name = ?3 LIMIT 1`)
+    .prepare(`SELECT ${BASE_SELECT}, ${LIVE_COUNTER_SELECT} ${FROM_JOIN} WHERE s.owner = ?1 AND s.repo = ?2 AND s.name = ?3 LIMIT 1`)
     .bind(key.owner, key.repo, key.name)
     .first<ScoreRow>()
   return row ?? null
@@ -340,7 +381,8 @@ export async function recomputeAllSkillScores(
 
   const res = await db
     .prepare(
-      `SELECT ${BASE_SELECT} ${FROM_JOIN}
+      `${ALL_COUNTERS_CTE}
+       SELECT ${BASE_SELECT}, ${ALL_COUNTER_SELECT} ${FROM_JOIN} ${ALL_COUNTER_JOINS}
        ORDER BY r.stars DESC, s.owner ASC, s.repo ASC, s.name ASC
        ${limitClause}`,
     )

layers/registry/server/utils/skills-registry.ts

@@ -424,26 +424,28 @@ export async function querySkills(event: H3Event, opts: SkillsQuery): Promise<Sk
     .prepare(`SELECT ${countExpression} as total ${FROM_SKILLS_JOIN_REPOS} ${where}`)
     .bind(...params)
 
-  // The repo Skill count is added after the page cut, so it runs once per row
-  // on the page instead of once per match. The outer query sorts the page
-  // again by the same key, which names result columns and binds nothing.
+  // Rank owner representatives with keys and sort columns only. Read card
+  // metadata after paging, so descriptions and author lookups never enter
+  // the window's sorter. The repo count also runs only for returned rows.
   const pageSelectRow = includeDependencies ? `${SELECT_SKILL_ROW_BASE}, s.rendered_raw` : SELECT_SKILL_ROW_BASE
   const dataStmt = db
     .prepare(uniqueOwners
       ? `WITH ranked_skills AS (
-          SELECT ${pageSelectRow},
+          SELECT s.owner, s.repo, s.name, s.like_count, s.modified_at, r.stars,
             ROW_NUMBER() OVER (PARTITION BY s.owner ORDER BY ${orderBy}) AS owner_rank
           ${FROM_SKILLS_JOIN_REPOS}
           ${where}
         )
-        SELECT paged.*, ${repoSkillCountSql('paged')}
+        SELECT ${selectSkillRow}
         FROM (
           SELECT * FROM ranked_skills
           WHERE owner_rank = 1
           ORDER BY ${rankedOrderBy}
           LIMIT ? OFFSET ?
         ) paged
-        ORDER BY ${rankedOrderBy}`
+        JOIN skills s ON s.owner = paged.owner AND s.repo = paged.repo AND s.name = paged.name
+        ${JOIN_REPOS_SQL}
+        ORDER BY ${orderBy}`
       : `SELECT paged.*, ${repoSkillCountSql('paged')}
         FROM (
           SELECT ${pageSelectRow} ${FROM_SKILLS_JOIN_REPOS} ${where}

test/unit/ai-generation-work.test.ts

@@ -1,11 +1,12 @@
 import type { AbstractnessPayload } from '../../layers/registry/server/utils/ai-generation-work'
 import Database from 'better-sqlite3'
-import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import {
   ABSTRACTNESS_MODEL,
   ABSTRACTNESS_PROMPT_VERSION,
   abstractnessResponseText,
   buildAbstractnessUserPrompt,
+  countMissingEmbeddings,
   parseAbstractnessPayload,
   persistAbstractness,
   runtimeGenerationLimits,
@@ -77,6 +78,24 @@ describe('ai generation work', () => {
     sqlite.prepare('UPDATE skills SET rendered_raw = ?').run(source)
     expect(await selectMissingGeneratedSkills(db, 'abstractness', 50)).toEqual([])
     expect(await selectMissingGeneratedSkills(db, 'embedding', 50)).toEqual([])
+    expect(await countMissingEmbeddings(db)).toBe(0)
+  })
+
+  it('counts only eligible sources without a current embedding', async () => {
+    expect(await countMissingEmbeddings(db)).toBe(1)
+    sqlite.prepare('UPDATE skill_generated SET sha = \'old\' WHERE name = \'paused-only\' AND kind = \'embedding\'').run()
+    expect(await countMissingEmbeddings(db)).toBe(2)
+    sqlite.prepare('UPDATE skills SET rendered_status = \'failed\' WHERE name = \'paused-only\'').run()
+    expect(await countMissingEmbeddings(db)).toBe(1)
+    sqlite.prepare('UPDATE repos SET broken_since = 1').run()
+    expect(await countMissingEmbeddings(db)).toBe(0)
+  })
+
+  it('does not trim sources whose embedding is current', async () => {
+    const trim = vi.fn((source: string) => source.trim())
+    sqlite.function('trim', { varargs: true }, trim)
+    expect(await countMissingEmbeddings(db)).toBe(1)
+    expect(trim.mock.calls.map(([source]) => source)).toEqual(['# Embed'])
   })
 
   it('does not select current runtime kinds because paused batch kinds are missing', async () => {
@@ -267,6 +286,7 @@ function wrapSqlite(sqlite: Database.Database): D1Database {
     all: async <T>() => ({
       results: sqlite.prepare(sql).all(bindings(params)) as T[],
     }),
+    first: async <T>() => (sqlite.prepare(sql).get(bindings(params)) as T | undefined) ?? null,
     run: async () => {
       const result = sqlite.prepare(sql).run(bindings(params))
       return { meta: { changes: result.changes } }