From 6410143a351c05990871894eb354252ef7fa27fc Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 2 Oct 2026 08:17:33 -0500 Subject: [PATCH] Reduce D1 reads with incremental summaries and bounded API caching (#14) --- README.md | 63 ++++ docs/d1-read-optimization-measurements.json | 225 ++++++++++++ scripts/measure-read-cost.ts | 342 +++++++++++++++++ src/db/types.ts | 2 +- src/github/sync-refresh.ts | 52 +++ src/github/sync.ts | 386 ++++++++++++-------- src/github/users.ts | 56 ++- src/web/read-cache.ts | 158 ++++++++ src/web/routes.ts | 187 ++++------ src/web/worker.ts | 19 +- tests/github-users-sync.test.ts | 99 +++++ tests/read-cache.test.ts | 146 ++++++++ tests/sync-refresh.test.ts | 103 ++++++ tests/worker.test.ts | 15 + 14 files changed, 1556 insertions(+), 297 deletions(-) create mode 100644 docs/d1-read-optimization-measurements.json create mode 100644 scripts/measure-read-cost.ts create mode 100644 src/github/sync-refresh.ts create mode 100644 src/web/read-cache.ts create mode 100644 tests/read-cache.test.ts create mode 100644 tests/sync-refresh.test.ts diff --git a/README.md b/README.md index 9cc1ce4..58cf22b 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,69 @@ The dashboard runs as a Cloudflare Worker with a D1 database. Merges to `master` are automatically deployed via CI — manual deploys are only needed for initial setup or debugging. +## D1 read usage and freshness + +Scheduled sync refreshes contributor summaries only for affected PR/comment authors. +It records pending work in the existing `sync_state` table before changing source +rows, because D1's REST client executes batches sequentially. Interrupted PR saves +are replayed even if the PR has since closed. Failed summary refreshes remain +queued, and `last_sync_at` advances only after the entire sync succeeds. The +existing workflow serializes sync jobs; other callers must also use a single sync +writer per database. + +The first sync with this implementation performs one full contributor repair and +stores `incremental_users_version=1`. Later unchanged syncs read only queue/marker +rows for this stage. This adds no tables, columns, indexes, or schema-version +migration; it uses the normalized handles and indexes already created by schema +version 2. Existing data, scoring rules, and response shapes are preserved. + +The Worker caches successful public API GET responses per database binding and +Worker isolate. It checks `last_sync_at` at most once per minute and clears cached +responses when the marker changes. Every response also expires five minutes after +its fetch started, bounding staleness after partial sync failures, out-of-band +writes, and time-based score changes. Revision-check failures discard the cache. +Authorization/cookie requests, errors, private responses, static routes, and sync +requests bypass it. The local Node server remains uncached. + +The cache retains at most 64 responses / 16 MiB, accepts responses up to 4 MiB, and +coalesces at most four distinct fills. Oversized responses pass through without +being retained. Deployments/new isolates start cold. This reduces repeated reads +but does not guarantee that an account stays within the free quota. + +On a local fixture with 11,000 PRs, 25,000 comments, and 165,000 files, compared with +commit `3dec121`, SQLite reported: + +| Operation | Previous row visits | New row visits | +| --- | ---: | ---: | +| Default PR list, uncached | 246,875 | 96,051 | +| Author-filtered PR list | 209,090 | 221 | +| PR detail | 11,024 | 35 | +| Contributor profile | 36,118 | 146 | +| Similar PRs | 224,763 | 11,326 | +| Contributor refresh: full rebuild → one affected handle | 236,501 | 132 | + +An unchanged contributor-refresh stage uses two marker queries and one indexed +row visit. A warm cached list uses zero database queries; its periodic unchanged +revision check uses one query/row. These stage measurements exclude the rest of +the scheduled sync. Cold stats queries are unchanged: 6,803 scan visits plus +36,000 rows counted through SQLite's fast `COUNT(*)` path, which scanstatus does +not report. Similar-PR ranking still compares all PR bodies; its lower read cost +does not establish that cold requests fit the Worker's CPU limit. + +To reproduce locally, install dependencies with Node 22.18 or newer and run: + +```bash +node --experimental-strip-types scripts/measure-read-cost.ts 3dec121 +``` + +The script requires `clang`, compiles a SQLite scanstatus harness from the +repository's installed dependency, and writes synthetic fixtures, exact SQL, +query plans, response comparisons, and results under the system temporary +directory. It makes no GitHub or Cloudflare requests. It checks exact response +parity, including mixed-case authors, temporarily missing summaries, similarity +ranking, and cached responses. Measurements are local SQLite row visits, not +Cloudflare billed reads or Worker CPU measurements. + ## How scoring works Every PR receives a **composite score from 0 to 180**, built from ten signals. The goal is to surface PRs that are most likely to be worth reviewing right now — small, well-tested PRs from reliable contributors with passing CI will naturally float to the top. diff --git a/docs/d1-read-optimization-measurements.json b/docs/d1-read-optimization-measurements.json new file mode 100644 index 0000000..c00bb41 --- /dev/null +++ b/docs/d1-read-optimization-measurements.json @@ -0,0 +1,225 @@ +{ + "baselineCommit": "3dec1213066c61d9535c45bae6c0b49c2ab8a246", + "sourceAtStart": { + "src/web/routes.ts": "d11864d72d565bf4be47e04308f4b79382f5a4cd19bf1ac9c518664dc31de54f", + "src/web/read-cache.ts": "4912dae21c0c9f80a49bedd077d1296611e6fcd10eca78f057640ff96990e45f", + "src/github/users.ts": "a5a00f71bddefae3ccc29c42d755f98e129a12608d38f6174b5ed86d5d2c06f1", + "src/github/sync-refresh.ts": "a89ae6964c546e5d6b4b680f7e633511882068f8de677f1ca0e81fac8d1f3055" + }, + "sourceAtEnd": { + "src/web/routes.ts": "d11864d72d565bf4be47e04308f4b79382f5a4cd19bf1ac9c518664dc31de54f", + "src/web/read-cache.ts": "4912dae21c0c9f80a49bedd077d1296611e6fcd10eca78f057640ff96990e45f", + "src/github/users.ts": "a5a00f71bddefae3ccc29c42d755f98e129a12608d38f6174b5ed86d5d2c06f1", + "src/github/sync-refresh.ts": "a89ae6964c546e5d6b4b680f7e633511882068f8de677f1ca0e81fac8d1f3055" + }, + "sourcesStableDuringRun": true, + "fixture": { + "prs": 11000, + "comments": 25000, + "files": 165000, + "contributors": 1499, + "openPRs": 3500, + "edgeCases": [ + "mixed-case author", + "one temporarily missing contributor summary" + ] + }, + "fixedNow": "2026-10-02T12:00:00.000Z", + "caveats": [ + "Local SQLite row visits, not Cloudflare billed rows.", + "readWorkProxy adds fast COUNT(*) table cardinalities, which scanstatus does not report.", + "Timings are single local runs, not Worker CPU measurements.", + "Warm application-cache behavior is tested separately; these are uncached route costs." + ], + "results": { + "list": { + "before": { + "queries": 5, + "scanRows": 246875, + "fastCountCardinality": 0, + "readWorkProxy": 246875, + "elapsedMs": 123.58 + }, + "after": { + "queries": 14, + "scanRows": 96051, + "fastCountCardinality": 0, + "readWorkProxy": 96051, + "elapsedMs": 58.02 + }, + "identicalJson": true + }, + "authorList": { + "before": { + "queries": 5, + "scanRows": 209090, + "fastCountCardinality": 0, + "readWorkProxy": 209090, + "elapsedMs": 71.24 + }, + "after": { + "queries": 2, + "scanRows": 221, + "fastCountCardinality": 0, + "readWorkProxy": 221, + "elapsedMs": 0.87 + }, + "identicalJson": true + }, + "mixedCaseAuthor": { + "before": { + "queries": 5, + "scanRows": 209085, + "fastCountCardinality": 0, + "readWorkProxy": 209085, + "elapsedMs": 69.62 + }, + "after": { + "queries": 2, + "scanRows": 230, + "fastCountCardinality": 0, + "readWorkProxy": 230, + "elapsedMs": 1.47 + }, + "identicalJson": true + }, + "missingSummary": { + "before": { + "queries": 5, + "scanRows": 209085, + "fastCountCardinality": 0, + "readWorkProxy": 209085, + "elapsedMs": 71.37 + }, + "after": { + "queries": 2, + "scanRows": 216, + "fastCountCardinality": 0, + "readWorkProxy": 216, + "elapsedMs": 0.7 + }, + "identicalJson": true + }, + "detail": { + "before": { + "queries": 7, + "scanRows": 11024, + "fastCountCardinality": 0, + "readWorkProxy": 11024, + "elapsedMs": 2.86 + }, + "after": { + "queries": 6, + "scanRows": 35, + "fastCountCardinality": 0, + "readWorkProxy": 35, + "elapsedMs": 0.85 + }, + "identicalJson": true + }, + "profile": { + "before": { + "queries": 3, + "scanRows": 36118, + "fastCountCardinality": 0, + "readWorkProxy": 36118, + "elapsedMs": 3.08 + }, + "after": { + "queries": 3, + "scanRows": 146, + "fastCountCardinality": 0, + "readWorkProxy": 146, + "elapsedMs": 0.86 + }, + "identicalJson": true + }, + "stats": { + "before": { + "queries": 8, + "scanRows": 6803, + "fastCountCardinality": 36000, + "readWorkProxy": 42803, + "elapsedMs": 0.62 + }, + "after": { + "queries": 8, + "scanRows": 6803, + "fastCountCardinality": 36000, + "readWorkProxy": 42803, + "elapsedMs": 0.62 + }, + "identicalJson": true + }, + "similar": { + "before": { + "queries": 4, + "scanRows": 224763, + "fastCountCardinality": 0, + "readWorkProxy": 224763, + "elapsedMs": 166.9 + }, + "after": { + "queries": 4, + "scanRows": 11326, + "fastCountCardinality": 0, + "readWorkProxy": 11326, + "elapsedMs": 55.43 + }, + "identicalJson": true, + "identicalRanking": true + }, + "fullRefresh": { + "queries": 4, + "scanRows": 236501, + "fastCountCardinality": 0, + "readWorkProxy": 236501, + "elapsedMs": 31.75 + }, + "targetedRefresh": { + "queries": 2, + "scanRows": 132, + "fastCountCardinality": 0, + "readWorkProxy": 132, + "elapsedMs": 0.39 + }, + "unchangedRefresh": { + "queries": 0, + "scanRows": 0, + "fastCountCardinality": 0, + "readWorkProxy": 0, + "elapsedMs": 0.06 + }, + "unchangedSyncRefresh": { + "queries": 2, + "scanRows": 1, + "fastCountCardinality": 0, + "readWorkProxy": 1 + }, + "changedContributor": { + "identicalSummaries": true + }, + "cache-cold": { + "queries": 15, + "scanRows": 96052, + "fastCountCardinality": 0, + "readWorkProxy": 96052, + "state": "miss" + }, + "cache-warm": { + "queries": 0, + "scanRows": 0, + "fastCountCardinality": 0, + "readWorkProxy": 0, + "state": "hit" + }, + "cache-revisionCheck": { + "queries": 1, + "scanRows": 1, + "fastCountCardinality": 0, + "readWorkProxy": 1, + "state": "hit" + }, + "cacheParity": true + } +} diff --git a/scripts/measure-read-cost.ts b/scripts/measure-read-cost.ts new file mode 100644 index 0000000..487bca0 --- /dev/null +++ b/scripts/measure-read-cost.ts @@ -0,0 +1,342 @@ +/** + * Local-only D1 read-cost proxy. Run with Node >=22.18: + * node --experimental-strip-types scripts/measure-read-cost.ts [baseline-ref] + * + * Creates synthetic data under /tmp, imports the baseline and working-tree + * implementations, captures their actual SQL, and measures it using SQLite's + * sqlite3_stmt_scanstatus API. Does not load configuration, call GitHub/Cloudflare, + * or change data outside its temporary directory. Requires a C compiler; builds + * a local scanstatus harness from better-sqlite3's bundled SQLite amalgamation. + * + * SQLite visits are not D1 billing: notably SQLite's fast COUNT(*) opcode reports + * no scan rows. We report those table cardinalities separately, not as zero cost. + */ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { execFileSync } from 'node:child_process'; +import { createRequire } from 'node:module'; +import { fileURLToPath } from 'node:url'; +import { performance } from 'node:perf_hooks'; +import { createHash } from 'node:crypto'; + +const require = createRequire(import.meta.url); +const ts = require('typescript'); +// Import the source directly without overwriting the repository's build output. +require.extensions['.ts'] = (module: any, filename: string) => { + const source = fs.readFileSync(filename, 'utf8'); + const output = ts.transpileModule(source, { + compilerOptions: { target: ts.ScriptTarget.ES2022, module: ts.ModuleKind.CommonJS, + esModuleInterop: true }, fileName: filename, + }).outputText; + module._compile(output, filename); +}; + +const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..'); +const fingerprints = () => Object.fromEntries(['src/web/routes.ts', 'src/web/read-cache.ts', + 'src/github/users.ts', 'src/github/sync-refresh.ts'].filter(file => fs.existsSync(path.join(root, file))) + .map(file => [file, createHash('sha256').update(fs.readFileSync(path.join(root, file))).digest('hex')])); +const sourceAtStart = fingerprints(); +const baselineRef = process.argv[2] ?? '3dec121'; +const temp = fs.mkdtempSync(path.join(os.tmpdir(), 'pr-triage-read-cost-')); +// macOS sqlite3 3.51.0's .scanstats crashes on COUNT(DISTINCT). Use the SQLite +// version bundled with this repository and the scanstatus C API directly. +const harness = path.join(temp, 'scanstatus'); +const harnessSource = path.join(temp, 'scanstatus.c'); +fs.writeFileSync(harnessSource, ` +#include +#include +#include "sqlite3.h" +int main(int argc, char **argv) { + if (argc != 3) return 2; + sqlite3 *db = 0; + if (sqlite3_open(argv[1], &db) != SQLITE_OK) return 3; + FILE *input = fopen(argv[2], "rb"); + if (!input) return 4; + fseek(input, 0, SEEK_END); long size = ftell(input); rewind(input); + char *text = calloc((size_t)size + 1, 1); + if (fread(text, 1, (size_t)size, input) != (size_t)size) return 5; + fclose(input); + const char *next = text, *tail = 0; int query = 0; + while (*next) { + sqlite3_stmt *statement = 0; + if (sqlite3_prepare_v2(db, next, -1, &statement, &tail) != SQLITE_OK) { + fprintf(stderr, "prepare: %s\\n", sqlite3_errmsg(db)); return 6; + } + next = tail; + if (!statement) continue; + int result; + while ((result = sqlite3_step(statement)) == SQLITE_ROW) {} + if (result != SQLITE_DONE) { + fprintf(stderr, "step: %s\\n", sqlite3_errmsg(db)); return 7; + } + printf("QUERY_%d\\n", query++); + for (int index = 0;; index++) { + sqlite3_int64 visits = 0, loops = 0; const char *explain = 0; + if (sqlite3_stmt_scanstatus(statement, index, SQLITE_SCANSTAT_NVISIT, &visits)) break; + sqlite3_stmt_scanstatus(statement, index, SQLITE_SCANSTAT_NLOOP, &loops); + sqlite3_stmt_scanstatus(statement, index, SQLITE_SCANSTAT_EXPLAIN, &explain); + printf("%s (loops=%lld rows=%lld)\\n", explain, (long long)loops, (long long)visits); + } + sqlite3_finalize(statement); + } + free(text); sqlite3_close(db); return 0; +} +`); +const sqliteSource = path.join(root, 'node_modules/better-sqlite3/deps/sqlite3'); +execFileSync('clang', ['-O2', '-DSQLITE_ENABLE_STMT_SCANSTATUS', '-DSQLITE_ENABLE_FTS5', + '-I', sqliteSource, harnessSource, path.join(sqliteSource, 'sqlite3.c'), '-lm', '-o', harness], { + maxBuffer: 4 * 1024 * 1024, +}); +const baselineRoot = path.join(temp, 'baseline'); +fs.mkdirSync(baselineRoot); +const baselineCommit = execFileSync('git', ['rev-parse', baselineRef], { cwd: root, encoding: 'utf8' }).trim(); +for (const filename of execFileSync('git', ['ls-tree', '-r', '--name-only', baselineCommit, 'src'], { + cwd: root, encoding: 'utf8', +}).trim().split('\n')) { + const destination = path.join(baselineRoot, filename); + fs.mkdirSync(path.dirname(destination), { recursive: true }); + fs.writeFileSync(destination, execFileSync('git', ['show', `${baselineCommit}:${filename}`], { cwd: root })); +} +fs.symlinkSync(path.join(root, 'node_modules'), path.join(baselineRoot, 'node_modules'), 'dir'); +const load = (directory: string, name: string) => require(path.join(directory, 'src', name)); +const { SqliteClient } = load(root, 'db/sqlite.ts'); +const fixturePath = path.join(temp, 'fixture.sqlite'); +const fixture = new SqliteClient(fixturePath); +await load(baselineRoot, 'db/bootstrap.ts').initializeDb(fixture); +const raw = fixture.db; +const now = Date.UTC(2026, 9, 2, 12); +const originalNow = Date.now; +Date.now = () => now; +const date = (index: number) => new Date(now - (index % 30) * 86400000).toISOString(); + +raw.transaction(() => { + const pr = raw.prepare(`INSERT INTO pull_requests + (number,title,body,author,author_handle,head_sha,mergeable,mergeable_state,state,labels_json, + additions,deletions,changed_files,created_at,updated_at) + VALUES (?,?,?,?,?,?,1,'clean',?,'[]',?,?,15,?,?)`); + const comment = raw.prepare(`INSERT INTO pr_comments + (comment_id,pr_number,author,author_handle,body,created_at,updated_at) VALUES (?,?,?,?,?,?,?)`); + const file = raw.prepare('INSERT INTO pr_files(pr_number,filename,status) VALUES (?,?,?)'); + const check = raw.prepare(`INSERT INTO check_runs(pr_number,name,status,conclusion,updated_at) + VALUES (?,?,'completed',?,?)`); + const score = raw.prepare(`INSERT INTO greptile_scores + (pr_number,comment_id,confidence_score,comment_body,created_at) VALUES (?,?,?,'Review',?)`); + const review = raw.prepare(`INSERT INTO llm_reviews(pr_number,review_json,model,prompt_version) + VALUES (?,'{}','fixture','1')`); + for (let i = 1; i <= 11000; i++) { + const owner = i % 1000; + const body = `Topic ${i % 79}: change the implementation for module ${i % 131}. ` + + `Add validation and support for workflow ${i % 41}; preserve compatible behavior. ` + + (i % 5 === 0 ? 'Thinking path: consider existing usage. ' : '') + + (i % 7 === 0 ? `Fixes #${i + 50}.` : ''); + // One account deliberately has mixed display casing across its PRs. + const author = owner === 1 ? (i % 3 === 0 ? 'USER1' : 'user1') : `User${owner}`; + pr.run(i, `Improve topic ${i % 79} for module ${i % 131}`, body, + author, `user${owner}`, `sha-${i}`, + i <= 3500 ? 'open' : i % 3 === 0 ? 'closed' : 'merged', i % 600, i % 80, date(i), date(i + 1)); + for (let j = 0; j < 15; j++) { + const name = j === 0 && i % 3 === 0 ? `src/topic${i % 100}/file.test.ts` : + `src/topic${i % 100}/file${(i + j) % 300}.ts`; + file.run(i, name, 'modified'); + } + if (i <= 3500) { + check.run(i, 'tests', i % 9 === 0 ? 'failure' : 'success', date(i)); + check.run(i, 'lint', 'success', date(i)); + } + if (i % 4 === 0) score.run(i, 100000 + i, i % 5 + 1, date(i)); + if (i % 20 === 0) review.run(i); + } + for (let i = 1; i <= 25000; i++) { + const owner = i % 1500; + comment.run(i, i % 11000 + 1, `User${owner}`, `user${owner}`, + `Fixture comment ${i}: topic ${i % 79}`, date(i), date(i)); + } + raw.prepare('INSERT INTO sync_state(key,value) VALUES (?,?)').run('last_sync_at', '2026-10-02 11:00:00'); + raw.prepare('INSERT INTO sync_state(key,value) VALUES (?,?)').run('merged_count', '5000'); + raw.prepare('INSERT INTO sync_state(key,value) VALUES (?,?)').run('closed_count', '2500'); + raw.prepare('INSERT INTO sync_state(key,value) VALUES (?,?)').run('incremental_users_version', '1'); +})(); +await load(baselineRoot, 'github/users.ts').rebuildGitHubUsers(fixture); +// Simulate the source/summary gap during an in-progress contributor refresh. +await fixture.run("DELETE FROM github_users WHERE handle = 'user2'"); +await fixture.close(); + +type Statement = { sql: string; params: any[] }; +function recordingDb(db: any, statements: Statement[]) { + return Object.fromEntries(['get', 'all', 'run', 'exec', 'runBatch', 'close'].map(method => [method, + async (sql: any, params: any[] = []) => { + if (method === 'runBatch') statements.push(...sql); + else if (method !== 'close') statements.push({ sql, params }); + return db[method](sql, params); + }])); +} + +function bind(statement: Statement): string { + let index = 0; + const sql = statement.sql.replace(/'(?:''|[^'])*'|"(?:""|[^"])*"|\?/g, (token) => { + if (token !== '?') return token; + if (index >= statement.params.length) throw new Error('Too few SQL parameters'); + const value = statement.params[index++]; + if (value === null || value === undefined) return 'NULL'; + if (typeof value === 'number') return String(value); + return `'${String(value).replaceAll("'", "''")}'`; + }); + if (index !== statement.params.length) throw new Error('Too many SQL parameters'); + return sql.trim().replace(/;$/, '') + ';'; +} + +function measure(name: string, statements: Statement[]) { + const replay = path.join(temp, `${name}.replay.sqlite`); + fs.copyFileSync(fixturePath, replay); + const script = statements.map(bind).join('\n'); + const scriptPath = path.join(temp, `${name}.sql`); + fs.writeFileSync(scriptPath, script); + const output = execFileSync(harness, [replay, scriptPath], { encoding: 'utf8', maxBuffer: 32 * 1024 * 1024 }); + fs.writeFileSync(path.join(temp, `${name}.scanstats.txt`), output); + const scanRows = [...output.matchAll(/\brows=(\d+)/g)].reduce((sum, match) => sum + Number(match[1]), 0); + // Op_Count bypasses scanstatus; its rows are meaningful to D1 even though + // native SQLite retrieves B-tree cardinality without visiting every row. + let fastCountCardinality = 0; + const cardinalities: Record = { pull_requests: 11000, pr_comments: 25000, + pr_files: 165000, github_users: 1499, check_runs: 7000, greptile_scores: 2750, llm_reviews: 550 }; + for (const statement of statements) { + const match = statement.sql.trim().match(/^SELECT COUNT\(\*\)\s+(?:AS\s+)?\w+\s+FROM\s+(\w+)\s*;?$/i); + if (match) fastCountCardinality += cardinalities[match[1]] ?? 0; + } + return { queries: statements.length, scanRows, fastCountCardinality, + readWorkProxy: scanRows + fastCountCardinality }; +} + +function canonical(value: any): string { + if (value === undefined) return 'undefined'; + if (Array.isArray(value)) return '[' + value.map(canonical).join(',') + ']'; + if (value && typeof value === 'object') return '{' + Object.keys(value).sort().map(key => JSON.stringify(key) + ':' + canonical(value[key])).join(',') + '}'; + return JSON.stringify(value); +} + +const routeCases = { + list: '/prs', + authorList: '/prs?state=all&author=USER704', + mixedCaseAuthor: '/prs?state=all&author=USER1', + missingSummary: '/prs?state=all&author=USER2', + detail: '/prs/9704', + profile: '/github-users/user704', + stats: '/stats', + similar: '/prs/9704/similar', +}; +const results: Record = {}; +for (const [name, url] of Object.entries(routeCases)) { + const payloads: any[] = []; + const pair: Record = {}; + for (const [label, directory] of [['before', baselineRoot], ['after', root]]) { + const db = new SqliteClient(fixturePath); + const statements: Statement[] = []; + const routes = load(directory, 'web/routes.ts').createRoutes(async () => recordingDb(db, statements)); + const started = performance.now(); + const response = await routes.request('http://fixture.test' + url); + const json = await response.json(); + const elapsedMs = Math.round((performance.now() - started) * 100) / 100; + if (response.status !== 200) throw new Error(`${label} ${name}: ${response.status} ${JSON.stringify(json)}`); + payloads.push(json); + await db.close(); + fs.writeFileSync(path.join(temp, `${name}-${label}.json`), JSON.stringify(json, null, 2)); + pair[label] = { ...measure(`${name}-${label}`, statements), elapsedMs }; + } + pair.identicalJson = canonical(payloads[0]) === canonical(payloads[1]); + if (name === 'similar') pair.identicalRanking = canonical(payloads[0].similar.map((pr: any) => pr.number)) === canonical(payloads[1].similar.map((pr: any) => pr.number)); + results[name] = pair; + console.log(name, JSON.stringify(pair)); +} + +for (const [name, directory, handles] of [ + ['fullRefresh', baselineRoot, undefined], + ['targetedRefresh', root, ['user704']], + ['unchangedRefresh', root, []], +] as const) { + const dbPath = path.join(temp, `${name}.sqlite`); + fs.copyFileSync(fixturePath, dbPath); + const db = new SqliteClient(dbPath); + const statements: Statement[] = []; + const started = performance.now(); + await load(directory, 'github/users.ts').rebuildGitHubUsers(recordingDb(db, statements), handles); + const elapsedMs = Math.round((performance.now() - started) * 100) / 100; + await db.close(); + results[name] = { ...measure(name, statements), elapsedMs }; + console.log(name, JSON.stringify(results[name])); +} + +if (fs.existsSync(path.join(root, 'src/github/sync-refresh.ts'))) { + const db = new SqliteClient(fixturePath); + const statements: Statement[] = []; + await load(root, 'github/sync-refresh.ts').refreshPendingUsers(recordingDb(db, statements)); + results.unchangedSyncRefresh = measure('unchangedSyncRefresh', statements); + console.log('unchangedSyncRefresh', JSON.stringify(results.unchangedSyncRefresh)); + await db.close(); +} + +// Verify that a real source change has the same aggregate result in both modes. +const contributorPayloads: any[] = []; +for (const [label, directory, handles] of [ + ['before', baselineRoot, undefined], ['after', root, ['user704']], +] as const) { + const dbPath = path.join(temp, `changed-contributor-${label}.sqlite`); + fs.copyFileSync(fixturePath, dbPath); + const db = new SqliteClient(dbPath); + await db.run("UPDATE pull_requests SET state = 'closed', updated_at = '2026-10-02T12:00:00.000Z' WHERE number = 9704"); + await db.run("UPDATE pr_comments SET updated_at = '2026-10-02T12:00:00.000Z' WHERE comment_id = 704"); + await load(directory, 'github/users.ts').rebuildGitHubUsers(db, handles); + // Exclude the deliberately absent *unrelated* summary: only the full repair + // restores it. Timestamps record refresh time, not contributor contents. + const rows = await db.all(`SELECT handle,display_handle,pr_count,open_pr_count, + merged_pr_count,closed_unmerged_pr_count,comment_count,latest_pr_number, + latest_pr_at,latest_comment_id,latest_comment_at FROM github_users + WHERE handle != 'user2' ORDER BY handle`); + contributorPayloads.push(rows); + await db.close(); +} +results.changedContributor = { identicalSummaries: canonical(contributorPayloads[0]) === canonical(contributorPayloads[1]) }; +console.log('changedContributor', JSON.stringify(results.changedContributor)); + +// Exercise the real cache with a real SQLite-backed route. This includes the +// indexed revision read and proves warm hits avoid the expensive query group. +if (fs.existsSync(path.join(root, 'src/web/read-cache.ts'))) { + const db = new SqliteClient(fixturePath); + const statements: Statement[] = []; + const tracked = recordingDb(db, statements); + const routes = load(root, 'web/routes.ts').createRoutes(async () => tracked); + let cacheNow = 0; + const { ReadCache } = load(root, 'web/read-cache.ts'); + const cache = new ReadCache(async () => { + const row = await tracked.get("SELECT value FROM sync_state WHERE key = 'last_sync_at'"); + return row?.value ?? null; + }, { now: () => cacheNow }); + const request = new Request('http://fixture.test/api/prs'); + const payloads: any[] = []; + for (const [name, elapsed] of [['cold', 0], ['warm', 1], ['revisionCheck', 60_001]] as const) { + cacheNow = elapsed; + statements.length = 0; + const response = await cache.fetch(request, async () => routes.request('http://fixture.test/prs')); + payloads.push(await response.json()); + results[`cache-${name}`] = { ...measure(`cache-${name}`, statements), state: response.headers.get('X-PR-Cache') }; + console.log(`cache-${name}`, JSON.stringify(results[`cache-${name}`])); + } + results.cacheParity = payloads.every(value => canonical(value) === canonical(payloads[0])); + await db.close(); +} + +Date.now = originalNow; +const sourceAtEnd = fingerprints(); +const report = { baselineCommit, sourceAtStart, sourceAtEnd, + sourcesStableDuringRun: canonical(sourceAtStart) === canonical(sourceAtEnd), + fixture: { prs: 11000, comments: 25000, files: 165000, + contributors: 1499, openPRs: 3500, edgeCases: ['mixed-case author', 'one temporarily missing contributor summary'] }, fixedNow: new Date(now).toISOString(), + caveats: ['Local SQLite row visits, not Cloudflare billed rows.', + 'readWorkProxy adds fast COUNT(*) table cardinalities, which scanstatus does not report.', + 'Timings are single local runs, not Worker CPU measurements.', + 'Warm application-cache behavior is tested separately; these are uncached route costs.'], results }; +fs.writeFileSync(path.join(temp, 'results.json'), JSON.stringify(report, null, 2)); +console.log('Artifacts:', temp); +if (Object.values(results).some(result => result && typeof result === 'object' && + (result.identicalJson === false || result.identicalSummaries === false || result.identicalRanking === false)) + || results.cacheParity === false) process.exitCode = 1; diff --git a/src/db/types.ts b/src/db/types.ts index f75a2ee..c5c2486 100644 --- a/src/db/types.ts +++ b/src/db/types.ts @@ -9,7 +9,7 @@ export interface DbClient { run(sql: string, params?: any[]): Promise; get(sql: string, params?: any[]): Promise; all(sql: string, params?: any[]): Promise; - /** Execute multiple statements in a single round-trip (D1) or transaction (SQLite) */ + /** Execute statements in order. REST D1 is sequential; callers must not assume atomicity. */ runBatch(statements: BatchStatement[]): Promise; exec(sql: string): Promise; close(): Promise; diff --git a/src/github/sync-refresh.ts b/src/github/sync-refresh.ts new file mode 100644 index 0000000..f8fe334 --- /dev/null +++ b/src/github/sync-refresh.ts @@ -0,0 +1,52 @@ +import type { DbClient } from '../db/types'; +import { normalizeGitHubHandle, rebuildGitHubUsers } from './users'; + +const PR_PREFIX = 'pending_pr:'; +const USER_PREFIX = 'pending_user:'; + +export async function pendingPRs(db: DbClient): Promise> { + const rows = await db.all<{ key: string }>( + "SELECT key FROM sync_state WHERE key >= 'pending_pr:' AND key < 'pending_pr;'", + ); + return new Set(rows.map(row => Number(row.key.slice(PR_PREFIX.length)))); +} + +// The REST DbClient is sequential, NOT transactional. Persist retry information +// before any source writes; a failed prefix must prevent the source mutation. +export async function beginPRRefresh(db: DbClient, number: number, authors: string[], token: string): Promise { + await db.run('INSERT OR REPLACE INTO sync_state(key, value) VALUES (?, ?)', [PR_PREFIX + number, token]); + const previous = await db.all<{ handle: string }>(` + SELECT author_handle AS handle FROM pull_requests WHERE number = ? + UNION SELECT author_handle FROM pr_comments WHERE pr_number = ? + `, [number, number]); + const handles = new Set([...previous.map(row => row.handle), ...authors].map(normalizeGitHubHandle).filter(Boolean)); + for (const handle of handles) { + await db.run(`INSERT INTO sync_state(key, value) VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value + WHERE sync_state.value IS NOT excluded.value`, [USER_PREFIX + handle, token]); + } +} + +export async function finishPRRefresh(db: DbClient, number: number, token: string): Promise { + await db.run('DELETE FROM sync_state WHERE key = ? AND value = ?', [PR_PREFIX + number, token]); +} + +export async function refreshPendingUsers(db: DbClient): Promise { + const initialized = await db.get<{ value: string }>("SELECT value FROM sync_state WHERE key = 'incremental_users_version'"); + // One initial repair also covers old syncs that wrote source rows and then + // failed their aggregate rebuild. No schema/index migration is needed. + const pending = await db.all<{ key: string; value: string }>( + "SELECT key, value FROM sync_state WHERE key >= 'pending_user:' AND key < 'pending_user;'", + ); + if (initialized?.value !== '1') { + await rebuildGitHubUsers(db); + await db.run("INSERT INTO sync_state(key, value) VALUES ('incremental_users_version', '1') ON CONFLICT(key) DO UPDATE SET value = excluded.value"); + } else { + await rebuildGitHubUsers(db, pending.map(row => row.key.slice(USER_PREFIX.length))); + } + // Compare tokens so a newer enqueue cannot be erased by an older refresh. + // Never clear the queue before the aggregate write succeeds. + for (const row of pending) { + await db.run('DELETE FROM sync_state WHERE key = ? AND value = ?', [row.key, row.value]); + } +} diff --git a/src/github/sync.ts b/src/github/sync.ts index 52829e7..8a1dfdb 100644 --- a/src/github/sync.ts +++ b/src/github/sync.ts @@ -3,16 +3,43 @@ import { getDb } from '../db/client'; import { BatchStatement } from '../db/types'; import { parseGreptileScores } from './comments'; import { CheckRun } from './checks'; -import { normalizeGitHubHandle, rebuildGitHubUsers } from './users'; +import { normalizeGitHubHandle } from './users'; +import { beginPRRefresh, finishPRRefresh, pendingPRs, refreshPendingUsers } from './sync-refresh'; +import { randomUUID } from 'node:crypto'; import chalk from 'chalk'; export interface SyncOptions { full?: boolean; } +// Both GitHub's list and detail responses provide these fields. Their unused +// label metadata has different nullability, so accept only the fields we read. +interface SyncPRSource { + number: number; + title: string; + body?: string | null; + user: { login: string } | null; + head: { sha: string }; + labels: Array; + created_at: string; + updated_at: string; +} + +interface SyncPRDetail { + mergeable: boolean | null; + mergeable_state: string; + additions: number; + deletions: number; + changed_files: number; +} + export async function syncPullRequests(opts: SyncOptions = {}): Promise { const octokit = getOctokit(); const db = await getDb(); + const token = randomUUID(); + const retryPRs = await pendingPRs(db); + const incompletePRs = new Set(); + let failures = 0; console.log(chalk.blue('Fetching open pull requests...')); @@ -41,187 +68,204 @@ export async function syncPullRequests(opts: SyncOptions = {}): Promise { let completed = 0; let skipped = 0; - const tasks = prs.map(pr => limit(async () => { - try { - const existing = cached.get(pr.number); + const syncPR = async (pr: SyncPRSource, state = 'open', prefetchedDetail?: SyncPRDetail) => { + const existing = cached.get(pr.number); - // Skip detail fetch if the PR hasn't changed since last sync - if (!opts.full && existing && existing.updated_at === pr.updated_at) { - skipped++; - completed++; - process.stdout.write(`\r ${chalk.green(`${completed}/${prs.length}`)} synced (${chalk.yellow(`${skipped} skipped`)})`); - return; - } + // Skip detail fetch if the PR hasn't changed since last sync + if (state === 'open' && !opts.full && !retryPRs.has(pr.number) && existing && existing.updated_at === pr.updated_at) { + skipped++; + completed++; + process.stdout.write(`\r ${chalk.green(`${completed}/${prs.length}`)} synced (${chalk.yellow(`${skipped} skipped`)})`); + return; + } - let mergeable: boolean | null = null; - let mergeableState: string | null = null; - let additions = 0; - let deletions = 0; - let changedFiles = 0; + let mergeable: boolean | null = null; + let mergeableState: string | null = null; + let additions = 0; + let deletions = 0; + let changedFiles = 0; - // Only call pulls.get (with retry) when head_sha changed; otherwise reuse cached mergeable - const headShaChanged = !existing || existing.head_sha !== pr.head.sha; + // Only call pulls.get (with retry) when head_sha changed; otherwise reuse cached mergeable + const headShaChanged = !existing || existing.head_sha !== pr.head.sha; - if (headShaChanged) { - for (let attempt = 0; attempt < 3; attempt++) { - const { data: detail } = await octokit.rest.pulls.get({ - owner: REPO_OWNER, - repo: REPO_NAME, - pull_number: pr.number, - }); - mergeable = detail.mergeable; - mergeableState = detail.mergeable_state; - additions = detail.additions ?? 0; - deletions = detail.deletions ?? 0; - changedFiles = detail.changed_files ?? 0; - if (mergeable !== null) break; - await new Promise(resolve => setTimeout(resolve, 1000)); - } - } else { - // head_sha unchanged — reuse cached mergeable, still fetch detail once for LOC stats + if (prefetchedDetail) { + mergeable = prefetchedDetail.mergeable ?? null; + mergeableState = prefetchedDetail.mergeable_state ?? null; + additions = prefetchedDetail.additions ?? 0; + deletions = prefetchedDetail.deletions ?? 0; + changedFiles = prefetchedDetail.changed_files ?? 0; + } else if (headShaChanged) { + for (let attempt = 0; attempt < 3; attempt++) { const { data: detail } = await octokit.rest.pulls.get({ owner: REPO_OWNER, repo: REPO_NAME, pull_number: pr.number, }); - mergeable = existing.mergeable === null ? null : existing.mergeable === 1; - mergeableState = existing.mergeable_state; + mergeable = detail.mergeable; + mergeableState = detail.mergeable_state; additions = detail.additions ?? 0; deletions = detail.deletions ?? 0; changedFiles = detail.changed_files ?? 0; + if (mergeable !== null) break; + await new Promise(resolve => setTimeout(resolve, 1000)); } - - const comments = await octokit.paginate(octokit.rest.issues.listComments, { - owner: REPO_OWNER, - repo: REPO_NAME, - issue_number: pr.number, - per_page: 100, - }); - - const scores = parseGreptileScores(comments); - - const { data: checksData } = await octokit.rest.checks.listForRef({ - owner: REPO_OWNER, - repo: REPO_NAME, - ref: pr.head.sha, - per_page: 100, - }); - - const checks: CheckRun[] = checksData.check_runs.map(cr => ({ - name: cr.name, - status: cr.status, - conclusion: cr.conclusion ?? null, - updatedAt: cr.completed_at ?? cr.started_at ?? new Date().toISOString(), - })); - - // Extract labels (name + color) - const labels = (pr.labels || []).map((l: any) => ({ - name: typeof l === 'string' ? l : l.name, - color: typeof l === 'string' ? null : l.color, - })); - - // --- Batch all DB writes for this PR --- - const batch: BatchStatement[] = []; - const authorHandle = normalizeGitHubHandle(pr.user?.login ?? 'unknown'); - - // Upsert PR - batch.push({ - sql: `INSERT INTO pull_requests (number, title, body, author, author_handle, head_sha, mergeable, mergeable_state, state, labels_json, additions, deletions, changed_files, created_at, updated_at, fetched_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, ?, ?, ?, ?, datetime('now')) - ON CONFLICT(number) DO UPDATE SET - title=excluded.title, body=excluded.body, author=excluded.author, author_handle=excluded.author_handle, - head_sha=excluded.head_sha, mergeable=excluded.mergeable, - mergeable_state=excluded.mergeable_state, state='open', labels_json=excluded.labels_json, - additions=excluded.additions, deletions=excluded.deletions, changed_files=excluded.changed_files, - updated_at=excluded.updated_at, fetched_at=datetime('now')`, - params: [ - pr.number, - pr.title, - pr.body ?? null, - pr.user?.login ?? 'unknown', - authorHandle, - pr.head.sha, - mergeable === null ? null : mergeable ? 1 : 0, - mergeableState, - JSON.stringify(labels), - additions, - deletions, - changedFiles, - pr.created_at, - pr.updated_at, - ], - }); - - // Upsert comments + FTS - for (const comment of comments) { - if (!comment.body) continue; - batch.push({ - sql: `INSERT INTO pr_comments (comment_id, pr_number, author, author_handle, body, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(comment_id) DO UPDATE SET - author=excluded.author, author_handle=excluded.author_handle, body=excluded.body, updated_at=excluded.updated_at`, - params: [ - comment.id, - pr.number, - comment.user?.login ?? 'unknown', - normalizeGitHubHandle(comment.user?.login ?? 'unknown'), - comment.body, - comment.created_at, - comment.updated_at, - ], - }); - batch.push({ - sql: `INSERT OR REPLACE INTO pr_comments_fts(rowid, body) VALUES (?, ?)`, - params: [comment.id, comment.body], - }); - } - - // Upsert greptile scores - for (const score of scores) { - batch.push({ - sql: `INSERT INTO greptile_scores (pr_number, comment_id, confidence_score, comment_body, created_at) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(comment_id) DO UPDATE SET - confidence_score=excluded.confidence_score, comment_body=excluded.comment_body`, - params: [pr.number, score.commentId, score.confidenceScore, score.commentBody, score.createdAt], - }); - } - - // Sync changed files - const files = await octokit.paginate(octokit.rest.pulls.listFiles, { + } else { + // head_sha unchanged — reuse cached mergeable, still fetch detail once for LOC stats + const { data: detail } = await octokit.rest.pulls.get({ owner: REPO_OWNER, repo: REPO_NAME, pull_number: pr.number, - per_page: 100, }); + mergeable = existing.mergeable === null ? null : existing.mergeable === 1; + mergeableState = existing.mergeable_state; + additions = detail.additions ?? 0; + deletions = detail.deletions ?? 0; + changedFiles = detail.changed_files ?? 0; + } + const comments = await octokit.paginate(octokit.rest.issues.listComments, { + owner: REPO_OWNER, + repo: REPO_NAME, + issue_number: pr.number, + per_page: 100, + }); + + const scores = parseGreptileScores(comments); + + const { data: checksData } = await octokit.rest.checks.listForRef({ + owner: REPO_OWNER, + repo: REPO_NAME, + ref: pr.head.sha, + per_page: 100, + }); + + const checks: CheckRun[] = checksData.check_runs.map(cr => ({ + name: cr.name, + status: cr.status, + conclusion: cr.conclusion ?? null, + updatedAt: cr.completed_at ?? cr.started_at ?? new Date().toISOString(), + })); + + // Extract labels (name + color) + const labels = (pr.labels || []).map((l: any) => ({ + name: typeof l === 'string' ? l : l.name, + color: typeof l === 'string' ? null : l.color, + })); + + // --- Batch all DB writes for this PR --- + const batch: BatchStatement[] = []; + const authorHandle = normalizeGitHubHandle(pr.user?.login ?? 'unknown'); + + // Upsert PR + batch.push({ + sql: `INSERT INTO pull_requests (number, title, body, author, author_handle, head_sha, mergeable, mergeable_state, state, labels_json, additions, deletions, changed_files, created_at, updated_at, fetched_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) + ON CONFLICT(number) DO UPDATE SET + title=excluded.title, body=excluded.body, author=excluded.author, author_handle=excluded.author_handle, + head_sha=excluded.head_sha, mergeable=excluded.mergeable, + mergeable_state=excluded.mergeable_state, state=excluded.state, labels_json=excluded.labels_json, + additions=excluded.additions, deletions=excluded.deletions, changed_files=excluded.changed_files, + updated_at=excluded.updated_at, fetched_at=datetime('now')`, + params: [ + pr.number, + pr.title, + pr.body ?? null, + pr.user?.login ?? 'unknown', + authorHandle, + pr.head.sha, + mergeable === null ? null : mergeable ? 1 : 0, + mergeableState, + state, + JSON.stringify(labels), + additions, + deletions, + changedFiles, + pr.created_at, + pr.updated_at, + ], + }); + + // Upsert comments + FTS + for (const comment of comments) { + if (!comment.body) continue; batch.push({ - sql: 'DELETE FROM pr_files WHERE pr_number = ?', - params: [pr.number], + sql: `INSERT INTO pr_comments (comment_id, pr_number, author, author_handle, body, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(comment_id) DO UPDATE SET + author=excluded.author, author_handle=excluded.author_handle, body=excluded.body, updated_at=excluded.updated_at`, + params: [ + comment.id, + pr.number, + comment.user?.login ?? 'unknown', + normalizeGitHubHandle(comment.user?.login ?? 'unknown'), + comment.body, + comment.created_at, + comment.updated_at, + ], }); - for (const file of files) { - batch.push({ - sql: `INSERT INTO pr_files (pr_number, filename, status) VALUES (?, ?, ?)`, - params: [pr.number, file.filename, file.status], - }); - } + batch.push({ + sql: `INSERT OR REPLACE INTO pr_comments_fts(rowid, body) VALUES (?, ?)`, + params: [comment.id, comment.body], + }); + } - // Upsert check runs - for (const check of checks) { - batch.push({ - sql: `INSERT INTO check_runs (pr_number, name, status, conclusion, updated_at) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(pr_number, name) DO UPDATE SET - status=excluded.status, conclusion=excluded.conclusion, updated_at=excluded.updated_at`, - params: [pr.number, check.name, check.status, check.conclusion, check.updatedAt], - }); - } + // Upsert greptile scores + for (const score of scores) { + batch.push({ + sql: `INSERT INTO greptile_scores (pr_number, comment_id, confidence_score, comment_body, created_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(comment_id) DO UPDATE SET + confidence_score=excluded.confidence_score, comment_body=excluded.comment_body`, + params: [pr.number, score.commentId, score.confidenceScore, score.commentBody, score.createdAt], + }); + } - await db.runBatch(batch); + // Sync changed files + const files = await octokit.paginate(octokit.rest.pulls.listFiles, { + owner: REPO_OWNER, + repo: REPO_NAME, + pull_number: pr.number, + per_page: 100, + }); + batch.push({ + sql: 'DELETE FROM pr_files WHERE pr_number = ?', + params: [pr.number], + }); + for (const file of files) { + batch.push({ + sql: `INSERT INTO pr_files (pr_number, filename, status) VALUES (?, ?, ?)`, + params: [pr.number, file.filename, file.status], + }); + } + + // Upsert check runs + for (const check of checks) { + batch.push({ + sql: `INSERT INTO check_runs (pr_number, name, status, conclusion, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(pr_number, name) DO UPDATE SET + status=excluded.status, conclusion=excluded.conclusion, updated_at=excluded.updated_at`, + params: [pr.number, check.name, check.status, check.conclusion, check.updatedAt], + }); + } + + await beginPRRefresh(db, pr.number, [authorHandle, ...comments.filter(c => c.body).map(c => c.user?.login ?? 'unknown')], token); + await db.runBatch(batch); + await finishPRRefresh(db, pr.number, token); + + if (state === 'open') { completed++; process.stdout.write(`\r ${chalk.green(`${completed}/${prs.length}`)} synced (${chalk.yellow(`${skipped} skipped`)})`); + } + }; + + const tasks = prs.map(pr => limit(async () => { + try { + await syncPR(pr); } catch (err: any) { + failures++; + incompletePRs.add(pr.number); completed++; console.error(chalk.red(`\nError syncing PR #${pr.number}: ${err.message}`)); } @@ -231,8 +275,7 @@ export async function syncPullRequests(opts: SyncOptions = {}): Promise { // Detect PRs in DB that are no longer open const openNumbers = new Set(prs.map(p => p.number)); - const stalePRs = await db.all<{ number: number }>(`SELECT number FROM pull_requests WHERE state = 'open'`); - const toCheck = stalePRs.filter(p => !openNumbers.has(p.number)); + const toCheck = [...new Set([...cached.keys(), ...retryPRs])].filter(number => !openNumbers.has(number)).map(number => ({ number })); if (toCheck.length > 0) { console.log(chalk.blue(`\nChecking ${toCheck.length} PRs no longer open...`)); @@ -243,13 +286,24 @@ export async function syncPullRequests(opts: SyncOptions = {}): Promise { const { data } = await octokit.rest.pulls.get({ owner: REPO_OWNER, repo: REPO_NAME, pull_number: p.number, }); + if (retryPRs.has(p.number)) { + // A previous REST batch may have stopped after saving updated_at or + // deleting files. Replay every source write even if GitHub has since + // closed the PR; a state-only update must not acknowledge that retry. + await syncPR(data, data.state === 'open' ? 'open' : data.merged ? 'merged' : 'closed', data); + return; + } if (data.state === 'open') { // PR is still open — pagination returned partial results, leave it alone return; } const newState = data.merged ? 'merged' : 'closed'; + await beginPRRefresh(db, p.number, [], token); await db.run(`UPDATE pull_requests SET state = ? WHERE number = ?`, [newState, p.number]); + await finishPRRefresh(db, p.number, token); } catch { + failures++; + incompletePRs.add(p.number); // API error — don't assume closed, leave state unchanged console.error(chalk.yellow(`\nCould not verify PR #${p.number}, leaving state unchanged`)); } @@ -278,7 +332,17 @@ export async function syncPullRequests(opts: SyncOptions = {}): Promise { let synced = 0; for (const cpr of closedPRs.slice(0, 500)) { + // A scalar history update cannot acknowledge a partial full-source + // replay. The open/stale paths own these durable retry markers. + if (retryPRs.has(cpr.number) || incompletePRs.has(cpr.number)) continue; const state = cpr.merged_at ? 'merged' : 'closed'; + const author = cpr.user?.login ?? 'unknown'; + const authorHandle = normalizeGitHubHandle(author); + const old = await db.get('SELECT title, body, author, author_handle, head_sha, state, updated_at FROM pull_requests WHERE number = ?', [cpr.number]); + if (old && old.title === cpr.title && old.body === (cpr.body ?? null) + && old.author === author && old.author_handle === authorHandle && old.head_sha === cpr.head.sha + && old.state === state && old.updated_at === cpr.updated_at) continue; + await beginPRRefresh(db, cpr.number, [authorHandle], token); await db.run(` INSERT INTO pull_requests (number, title, body, author, author_handle, head_sha, state, labels_json, created_at, updated_at, fetched_at) VALUES (?, ?, ?, ?, ?, ?, ?, '[]', ?, ?, datetime('now')) @@ -303,10 +367,12 @@ export async function syncPullRequests(opts: SyncOptions = {}): Promise { cpr.user?.login ?? 'unknown', normalizeGitHubHandle(cpr.user?.login ?? 'unknown'), cpr.head.sha, state, cpr.created_at, cpr.updated_at, ]); + await finishPRRefresh(db, cpr.number, token); synced++; } console.log(chalk.green(` ${synced} merged/closed PRs synced`)); } catch (err: any) { + failures++; console.error(chalk.yellow(`Could not fetch closed PRs: ${err.message}`)); } } @@ -327,14 +393,16 @@ export async function syncPullRequests(opts: SyncOptions = {}): Promise { await db.run(`INSERT INTO sync_state (key, value) VALUES ('merged_count', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value WHERE sync_state.value IS NOT excluded.value`, [String(mergedRes.data.total_count)]); await db.run(`INSERT INTO sync_state (key, value) VALUES ('closed_count', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value WHERE sync_state.value IS NOT excluded.value`, [String(closedRes.data.total_count)]); } catch (err: any) { + failures++; console.error(chalk.yellow(`Could not fetch closed/merged counts: ${err.message}`)); } + await refreshPendingUsers(db); + if (failures) throw new Error(`Sync incomplete: ${failures} operation(s) failed; pending PRs will be retried.`); await db.run(` INSERT INTO sync_state (key, value) VALUES ('last_sync_at', datetime('now')) ON CONFLICT(key) DO UPDATE SET value=datetime('now') `); - await rebuildGitHubUsers(db); const synced = completed - skipped; console.log(chalk.green(`\nSync complete. ${synced} PRs synced, ${skipped} unchanged (skipped).`)); diff --git a/src/github/users.ts b/src/github/users.ts index 0a6fbaa..cf15f4e 100644 --- a/src/github/users.ts +++ b/src/github/users.ts @@ -160,19 +160,34 @@ export function parseGitHubLeaderboardSort(input: string | null | undefined): Gi } } -export async function rebuildGitHubUsers(db: DbClient): Promise { - await db.run(` - UPDATE pull_requests - SET author_handle = LOWER(TRIM(COALESCE(NULLIF(author_handle, ''), author))) - WHERE COALESCE(author, '') != '' - AND author_handle IS NOT LOWER(TRIM(COALESCE(NULLIF(author_handle, ''), author))) - `); - await db.run(` - UPDATE pr_comments - SET author_handle = LOWER(TRIM(COALESCE(NULLIF(author_handle, ''), author))) - WHERE COALESCE(author, '') != '' - AND author_handle IS NOT LOWER(TRIM(COALESCE(NULLIF(author_handle, ''), author))) - `); +// Passing handles uses the existing author_handle indexes. The full path is +// reserved for initialization/repair; normal syncs refresh only queued handles. +export async function rebuildGitHubUsers(db: DbClient, handles?: string[]): Promise { + if (handles) { + handles = [...new Set(handles.map(normalizeGitHubHandle).filter(Boolean))]; + if (!handles.length) return; + // D1 limits bound parameters per statement. Keep each refresh below 100. + if (handles.length > 80) { + for (let i = 0; i < handles.length; i += 80) { + await rebuildGitHubUsers(db, handles.slice(i, i + 80)); + } + return; + } + } + if (!handles) { + await db.run(` + UPDATE pull_requests + SET author_handle = LOWER(TRIM(COALESCE(NULLIF(author_handle, ''), author))) + WHERE COALESCE(author, '') != '' + AND author_handle IS NOT LOWER(TRIM(COALESCE(NULLIF(author_handle, ''), author))) + `); + await db.run(` + UPDATE pr_comments + SET author_handle = LOWER(TRIM(COALESCE(NULLIF(author_handle, ''), author))) + WHERE COALESCE(author, '') != '' + AND author_handle IS NOT LOWER(TRIM(COALESCE(NULLIF(author_handle, ''), author))) + `); + } // Keep existing summaries (and their creation dates). An unchanged sync // should not rewrite every contributor or temporarily empty the leaderboard. @@ -192,14 +207,14 @@ export async function rebuildGitHubUsers(db: DbClient): Promise { created_at, updated_at ) - WITH author_handles AS ( - SELECT author_handle AS handle + WITH author_handles(handle) AS ( + ${handles ? handles.map(() => 'SELECT ?').join(' UNION ') : `SELECT author_handle AS handle FROM pull_requests WHERE COALESCE(author_handle, '') != '' UNION SELECT author_handle AS handle FROM pr_comments - WHERE COALESCE(author_handle, '') != '' + WHERE COALESCE(author_handle, '') != ''`} ) SELECT ah.handle, @@ -256,7 +271,8 @@ export async function rebuildGitHubUsers(db: DbClient): Promise { datetime('now'), datetime('now') FROM author_handles ah - WHERE true + WHERE EXISTS (SELECT 1 FROM pull_requests WHERE author_handle = ah.handle) + OR EXISTS (SELECT 1 FROM pr_comments WHERE author_handle = ah.handle) ORDER BY ah.handle ON CONFLICT(handle) DO UPDATE SET display_handle = excluded.display_handle, @@ -280,10 +296,10 @@ export async function rebuildGitHubUsers(db: DbClient): Promise { OR github_users.latest_pr_at IS NOT excluded.latest_pr_at OR github_users.latest_comment_id IS NOT excluded.latest_comment_id OR github_users.latest_comment_at IS NOT excluded.latest_comment_at - `); + `, handles ?? []); await db.run(` DELETE FROM github_users - WHERE NOT EXISTS (SELECT 1 FROM pull_requests WHERE author_handle = github_users.handle) + WHERE ${handles ? `handle IN (${handles.map(() => '?').join(',')}) AND` : ''} NOT EXISTS (SELECT 1 FROM pull_requests WHERE author_handle = github_users.handle) AND NOT EXISTS (SELECT 1 FROM pr_comments WHERE author_handle = github_users.handle) - `); + `, handles ?? []); } diff --git a/src/web/read-cache.ts b/src/web/read-cache.ts new file mode 100644 index 0000000..e9134b1 --- /dev/null +++ b/src/web/read-cache.ts @@ -0,0 +1,158 @@ +type Snapshot = { body: ArrayBuffer; headers: [string, string][]; status: number }; +type Entry = Snapshot & { expires: number }; +type Result = Snapshot | Response; + +interface CacheOptions { + now?: () => number; + ttlMs?: number; + revisionIntervalMs?: number; + maxEntries?: number; + maxBytes?: number; + maxEntryBytes?: number; + maxPending?: number; +} + +// These endpoints expose public data. Mutations, capability probes, static +// assets and future routes must explicitly opt in rather than being cached. +const PUBLIC_READ = /^\/api\/(?:prs(?:\/\d+(?:\/(?:comments|similar))?)?|stats|labels|authors|leaderboard|github-users\/[^/]+|search)$/; + +/** Per-database, per-isolate read cache; never writes to D1 or external storage. */ +export class ReadCache { + private entries = new Map(); + private pending = new Map>(); + private bytes = 0; + private revision: string | null | undefined; + private checkedAt = -Infinity; + private checking?: Promise; + private generation = 0; + private now: () => number; + private ttl: number; + private interval: number; + private maxEntries: number; + private maxBytes: number; + private maxEntryBytes: number; + private maxPending: number; + + constructor(private readRevision: () => Promise, options: CacheOptions = {}) { + this.now = options.now ?? Date.now; + this.ttl = options.ttlMs ?? 5 * 60_000; + this.interval = options.revisionIntervalMs ?? 60_000; + this.maxEntries = options.maxEntries ?? 64; + this.maxBytes = options.maxBytes ?? 16 * 1024 * 1024; + this.maxEntryBytes = Math.min(options.maxEntryBytes ?? 4 * 1024 * 1024, this.maxBytes); + this.maxPending = options.maxPending ?? 4; + } + + private clear(): void { + this.entries.clear(); + this.bytes = 0; + this.generation++; + } + + private async checkRevision(): Promise { + if (this.now() - this.checkedAt < this.interval) return true; + if (this.checking) return this.checking; + this.checking = (async () => { + try { + const revision = await this.readRevision(); + if (revision !== this.revision) this.clear(); + this.revision = revision; + this.checkedAt = this.now(); + return true; + } catch { + // A failed freshness check cannot extend the life of a cached result. + this.clear(); + return false; + } + })(); + try { return await this.checking; } finally { this.checking = undefined; } + } + + private remove(key: string, entry: Entry): void { + this.entries.delete(key); + this.bytes -= entry.body.byteLength; + } + + private response(snapshot: Result, state: string): Response { + if (snapshot instanceof Response) return snapshot; + const headers = new Headers(snapshot.headers); + headers.set('X-PR-Cache', state); + // Keep the freshness contract in this cache, not independent browser caches. + headers.set('Cache-Control', 'no-store'); + return new Response(snapshot.body.slice(0), { status: snapshot.status, headers }); + } + + private async snapshot(response: Response): Promise { + if (response.status !== 200 || !response.headers.get('content-type')?.includes('application/json') + || response.headers.has('set-cookie') || /private|no-store/i.test(response.headers.get('cache-control') ?? '') + || Number(response.headers.get('content-length')) > this.maxEntryBytes || !response.body) return response; + // Read a clone up to a fixed limit. If it is too large, return the original + // stream instead of buffering the rest of a large search/comment response. + const reader = response.clone().body!.getReader(); + const chunks: Uint8Array[] = []; + let length = 0; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + length += value.byteLength; + if (length > this.maxEntryBytes) { + void reader.cancel(); + return response; + } + chunks.push(value); + } + void response.body.cancel(); + const body = new Uint8Array(length); + let offset = 0; + for (const chunk of chunks) { body.set(chunk, offset); offset += chunk.byteLength; } + return { body: body.buffer, headers: [...response.headers.entries()], status: response.status }; + } + + async fetch(request: Request, next: () => Promise): Promise { + const url = new URL(request.url); + if (request.method !== 'GET' || !PUBLIC_READ.test(url.pathname) + || request.headers.has('authorization') || request.headers.has('cookie')) return next(); + if (!await this.checkRevision()) return next(); + + const key = url.href; // Includes origin, full path and every query parameter. + const cached = this.entries.get(key); + if (cached && cached.expires > this.now()) { + this.entries.delete(key); + this.entries.set(key, cached); + return this.response(cached, 'hit'); + } + if (cached) this.remove(key, cached); + + const generation = this.generation; + const pendingKey = `${generation}:${key}`; + const existing = this.pending.get(pendingKey); + if (existing) { + const result = await existing; + // Uncacheable streams belong to the original caller. Cloning them would + // leave a tee branch buffering arbitrarily large bodies for followers. + return result instanceof Response ? next() : this.response(result, 'coalesced'); + } + // Bound in-flight retained responses as well as completed entries. + if (this.pending.size >= this.maxPending) return next(); + const expires = this.now() + this.ttl; + const promise = (async () => { + const response = await next(); + const snapshot = await this.snapshot(response); + if (!(snapshot instanceof Response) && generation === this.generation && expires > this.now()) { + for (const [entryKey, entry] of this.entries) { + if (entry.expires <= this.now()) this.remove(entryKey, entry); + } + while (this.entries.size >= this.maxEntries || this.bytes + snapshot.body.byteLength > this.maxBytes) { + const [entryKey, entry] = this.entries.entries().next().value!; + this.remove(entryKey, entry); + } + this.entries.set(key, { ...snapshot, expires }); + this.bytes += snapshot.body.byteLength; + } + return snapshot; + })(); + this.pending.set(pendingKey, promise); + try { return this.response(await promise, 'miss'); } + finally { this.pending.delete(pendingKey); } + } +} diff --git a/src/web/routes.ts b/src/web/routes.ts index 0ce4937..50aa28a 100644 --- a/src/web/routes.ts +++ b/src/web/routes.ts @@ -84,6 +84,9 @@ const PR_SELECT = ` pr.mergeable, pr.mergeable_state, pr.state, pr.labels_json, pr.additions, pr.deletions, pr.changed_files, pr.created_at, pr.updated_at, + EXISTS(SELECT 1 FROM pr_files WHERE pr_number = pr.number AND (${TEST_FILE_SQL})) AS has_tests, + (${THINKING_PATH_SQL}) AS has_thinking_path, + (${ISSUE_LINK_SQL}) AS has_issue_link, (SELECT MAX(gs.confidence_score) FROM greptile_scores gs WHERE gs.pr_number = pr.number) as greptile_score, (SELECT COUNT(*) FROM check_runs cr WHERE cr.pr_number = pr.number) as total_checks, (SELECT COUNT(*) FROM check_runs cr WHERE cr.pr_number = pr.number AND cr.status = 'completed' AND cr.conclusion NOT IN ('success', 'skipped', 'neutral')) as failed_checks, @@ -117,35 +120,30 @@ function jaccard(a: Set, b: Set): number { return union === 0 ? 0 : intersection / union; } -function extractDirs(filename: string): string[] { - const parts = filename.split('/'); - const dirs: string[] = []; - for (let i = 1; i < parts.length; i++) { - dirs.push(parts.slice(0, i).join('/')); - } - return dirs; -} - -// --- Batch author stats helper --- - -function buildAuthorMap(rows: Array<{ author: string; state: string; cnt: number }>): Map { - const authorMap = new Map(); - for (const r of rows) { - if (!authorMap.has(r.author)) { - authorMap.set(r.author, { openCount: 0, mergedCount: 0, closedCount: 0, totalCount: 0, mergeRate: 0, isFirstContribution: false }); +async function loadAuthorStats(db: DbClient, handles: string[]): Promise> { + const authors = new Map(); + const unique = [...new Set(handles)]; + for (let i = 0; i < unique.length; i += 80) { + const batch = unique.slice(i, i + 80); + const rows = await db.all<{ author: string; state: string; cnt: number }>(` + SELECT author, state, COUNT(*) AS cnt FROM pull_requests + WHERE author_handle IN (${batch.map(() => '?').join(',')}) GROUP BY author, state + `, batch); + for (const row of rows) { + const stats = authors.get(row.author) ?? { openCount: 0, mergedCount: 0, closedCount: 0, totalCount: 0, mergeRate: 0, isFirstContribution: false }; + if (row.state === 'open') stats.openCount = row.cnt; + else if (row.state === 'merged') stats.mergedCount = row.cnt; + else if (row.state === 'closed') stats.closedCount = row.cnt; + authors.set(row.author, stats); } - const s = authorMap.get(r.author)!; - if (r.state === 'open') s.openCount = r.cnt; - else if (r.state === 'merged') s.mergedCount = r.cnt; - else if (r.state === 'closed') s.closedCount = r.cnt; } - for (const [, s] of authorMap) { - s.totalCount = s.openCount + s.mergedCount + s.closedCount; - const decided = s.mergedCount + s.closedCount; - s.mergeRate = decided > 0 ? s.mergedCount / decided : 0; - s.isFirstContribution = s.totalCount === 1; + for (const stats of authors.values()) { + stats.totalCount = stats.openCount + stats.mergedCount + stats.closedCount; + const decided = stats.mergedCount + stats.closedCount; + stats.mergeRate = decided > 0 ? stats.mergedCount / decided : 0; + stats.isFirstContribution = stats.totalCount === 1; } - return authorMap; + return authors; } /** Create API routes with an injected DB client — no Node.js imports */ @@ -173,7 +171,7 @@ export function createRoutes(getDb: () => Promise): Hono { params.push(state); } if (author) { - conditions.push('COALESCE(pr.author_handle, LOWER(pr.author)) = ?'); + conditions.push('pr.author_handle = ?'); params.push(normalizeGitHubHandle(author)); } @@ -183,28 +181,22 @@ export function createRoutes(getDb: () => Promise): Hono { let candidates = rows.map(buildCandidate); - // Batch queries for bonus signals - const [authorStatRows, testFileRows, tpRows, issueRows] = await Promise.all([ - db.all<{ author: string; state: string; cnt: number }>('SELECT author, state, COUNT(*) as cnt FROM pull_requests GROUP BY author, state'), - db.all<{ pr_number: number }>(`SELECT DISTINCT pr_number FROM pr_files WHERE ${TEST_FILE_SQL}`), - db.all<{ number: number }>(`SELECT number FROM pull_requests WHERE ${THINKING_PATH_SQL}`), - db.all<{ number: number }>(`SELECT number FROM pull_requests WHERE ${ISSUE_LINK_SQL}`), - ]); - - const authorMap = buildAuthorMap(authorStatRows); - const prsWithTests = new Set(testFileRows.map(r => r.pr_number)); - const prsWithThinkingPath = new Set(tpRows.map(r => r.number)); - const prsWithIssueLink = new Set(issueRows.map(r => r.number)); + // Preserve the existing case-sensitive scoring groups and live source + // counts, even while contributor summaries are being refreshed by sync. + const authorStats = await loadAuthorStats(db, candidates.map(candidate => candidate.authorHandle)); const now = Date.now(); - // Enrich candidates with all scoring signals - for (const c of candidates as any[]) { - const stats = authorMap.get(c.author) || { openCount: 0, mergedCount: 0, closedCount: 0, totalCount: 0, mergeRate: 0, isFirstContribution: true }; + // Signals are evaluated only for the selected PRs. Author lookups and test + // files use existing indexes; body checks reuse the selected row. + for (let i = 0; i < candidates.length; i++) { + const c: any = candidates[i]; + const row = rows[i]; + const stats = authorStats.get(c.author) ?? { openCount: 0, mergedCount: 0, closedCount: 0, totalCount: 0, mergeRate: 0, isFirstContribution: true }; const contrib = computeContributorScore(stats); const cPts = contributorPts(contrib.score); - const tPts = testPts(prsWithTests.has(c.number)); - const tpPts = thinkingPathPts(prsWithThinkingPath.has(c.number)); - const ilPts = issueLinkPts(prsWithIssueLink.has(c.number)); + const tPts = testPts(Boolean(row.has_tests)); + const tpPts = thinkingPathPts(Boolean(row.has_thinking_path)); + const ilPts = issueLinkPts(Boolean(row.has_issue_link)); const fresh = freshnessPts(c.createdAt, now); c.contributorPts = cPts; @@ -281,27 +273,13 @@ export function createRoutes(getDb: () => Promise): Hono { review: JSON.parse(r.review_json), })); - // Contributor stats - const authorRows = await db.all<{ state: string }>( - 'SELECT state FROM pull_requests WHERE author = ?', [candidate.author] - ); - const openCount = authorRows.filter(r => r.state === 'open').length; - const mergedCount = authorRows.filter(r => r.state === 'merged').length; - const closedCount = authorRows.filter(r => r.state === 'closed').length; - const totalCount = authorRows.length; - const decided = mergedCount + closedCount; - const mergeRate = decided > 0 ? mergedCount / decided : 0; - const isFirstContribution = totalCount === 1; - - const authorStats: AuthorStats = { openCount, mergedCount, closedCount, totalCount, mergeRate, isFirstContribution }; + const authorMap = await loadAuthorStats(db, [candidate.authorHandle]); + const authorStats = authorMap.get(candidate.author) ?? { openCount: 0, mergedCount: 0, closedCount: 0, totalCount: 0, mergeRate: 0, isFirstContribution: false }; const contributor = computeContributorScore(authorStats); const cPts = contributorPts(contributor.score); // Bonus signals - const testFiles = await db.get<{ cnt: number }>( - `SELECT COUNT(*) as cnt FROM pr_files WHERE pr_number = ? AND (${TEST_FILE_SQL})`, [prNumber] - ); - const hasTests = (testFiles?.cnt ?? 0) > 0; + const hasTests = Boolean(row.has_tests); const body = (pr as any)?.body ?? ''; const hasThinkingPath = detectThinkingPath(body); const hasIssueLink = detectIssueLink(body); @@ -428,39 +406,12 @@ export function createRoutes(getDb: () => Promise): Hono { ); if (!pr) return c.json({ error: 'PR not found' }, 404); - const others = await db.all<{ number: number; title: string; body: string | null; author: string; author_handle: string | null; created_at: string; greptile_score: number | null; total_checks: number; failed_checks: number; pending_checks: number; human_comments: number; additions: number | null; deletions: number | null; mergeable: number | null; mergeable_state: string | null }>( - `SELECT pr.number, pr.title, pr.body, pr.author, COALESCE(pr.author_handle, LOWER(pr.author)) as author_handle, pr.created_at, - pr.additions, pr.deletions, pr.mergeable, pr.mergeable_state, - (SELECT MAX(gs.confidence_score) FROM greptile_scores gs WHERE gs.pr_number = pr.number) as greptile_score, - (SELECT COUNT(*) FROM check_runs cr WHERE cr.pr_number = pr.number) as total_checks, - (SELECT COUNT(*) FROM check_runs cr WHERE cr.pr_number = pr.number AND cr.status = 'completed' AND cr.conclusion NOT IN ('success', 'skipped', 'neutral')) as failed_checks, - (SELECT COUNT(*) FROM check_runs cr WHERE cr.pr_number = pr.number AND cr.status != 'completed') as pending_checks, - (SELECT COUNT(*) FROM pr_comments pc WHERE pc.pr_number = pr.number AND pc.author NOT LIKE '%[bot]') as human_comments - FROM pull_requests pr WHERE pr.number != ?`, [prNumber] + // Rank using text first. Scoring metadata and file overlap cannot change + // the ranking, so load those indexed rows only for the eight results. + const others = await db.all<{ number: number; title: string; body: string | null; author: string; author_handle: string; created_at: string }>( + 'SELECT number, title, body, author, author_handle, created_at FROM pull_requests WHERE number != ? ORDER BY number', [prNumber] ); - const srcFiles = await db.all<{ filename: string }>( - 'SELECT filename FROM pr_files WHERE pr_number = ?', [prNumber] - ); - const srcFileSet = new Set(srcFiles.map(f => f.filename)); - const srcDirSet = new Set(srcFiles.flatMap(f => extractDirs(f.filename))); - - const allFiles = await db.all<{ pr_number: number; filename: string }>( - 'SELECT pr_number, filename FROM pr_files WHERE pr_number != ?', [prNumber] - ); - const filesByPR = new Map>(); - const dirsByPR = new Map>(); - for (const f of allFiles) { - if (!filesByPR.has(f.pr_number)) { - filesByPR.set(f.pr_number, new Set()); - dirsByPR.set(f.pr_number, new Set()); - } - filesByPR.get(f.pr_number)!.add(f.filename); - for (const dir of extractDirs(f.filename)) { - dirsByPR.get(f.pr_number)!.add(dir); - } - } - const srcTitleWords = new Set(tokenize(pr.title)); const srcBodyWords = tokenize(pr.body || ''); const srcBodyShingles = wordShingles(srcBodyWords, 3); @@ -488,14 +439,6 @@ export function createRoutes(getDb: () => Promise): Hono { bodySim = jaccard(srcBodyBigrams, otherBodyBigrams); } - const otherFileSet = filesByPR.get(other.number) || new Set(); - const fileSim = jaccard(srcFileSet, otherFileSet); - - let sharedFiles = 0; - for (const f of srcFileSet) { - if (otherFileSet.has(f)) sharedFiles++; - } - // Score by text only — file overlap is displayed but doesn't affect ranking const overall = titleSim * 0.4 + bodySim * 0.6; @@ -513,33 +456,47 @@ export function createRoutes(getDb: () => Promise): Hono { relationship = 'similar topic'; } - const otherCiStatus = deriveCIStatus(other.total_checks, other.failed_checks, other.pending_checks); - const otherHasConflicts = other.mergeable === 0 || other.mergeable_state === 'dirty'; - const prScore = computeBaseScore(other.greptile_score, otherCiStatus, otherHasConflicts, other.human_comments, other.additions ?? 0, other.deletions ?? 0); - simResults.push({ number: other.number, title: other.title, author: other.author, authorHandle: normalizeGitHubHandle(other.author_handle ?? other.author), created_at: other.created_at, - score: prScore, + score: 0, titleSimilarity: Math.round(titleSim * 100) / 100, bodySimilarity: Math.round(bodySim * 100) / 100, - fileSimilarity: Math.round(fileSim * 100) / 100, + fileSimilarity: 0, overallScore: Math.round(overall * 100) / 100, - sharedFiles, + sharedFiles: 0, potentialCopy, relationship, }); } simResults.sort((a, b) => b.overallScore - a.overallScore); - - return c.json({ - pr: prNumber, - similar: simResults.slice(0, 8), - }); + const similar = simResults.slice(0, 8); + if (similar.length) { + const numbers = similar.map(item => item.number); + const placeholders = numbers.map(() => '?').join(','); + const metadata = await db.all(`${PR_SELECT} WHERE pr.number IN (${placeholders})`, numbers); + const scores = new Map(metadata.map(row => [row.number, buildCandidate(row).compositeScore])); + const files = await db.all<{ pr_number: number; filename: string }>( + `SELECT pr_number, filename FROM pr_files WHERE pr_number IN (?, ${placeholders})`, [prNumber, ...numbers], + ); + const filesByPR = new Map>(); + for (const file of files) { + if (!filesByPR.has(file.pr_number)) filesByPR.set(file.pr_number, new Set()); + filesByPR.get(file.pr_number)!.add(file.filename); + } + const source = filesByPR.get(prNumber) ?? new Set(); + for (const item of similar) { + const other = filesByPR.get(item.number) ?? new Set(); + item.score = scores.get(item.number) ?? 0; + item.fileSimilarity = Math.round(jaccard(source, other) * 100) / 100; + item.sharedFiles = [...source].filter(file => other.has(file)).length; + } + } + return c.json({ pr: prNumber, similar }); }); // List unique labels across all PRs @@ -600,7 +557,7 @@ export function createRoutes(getDb: () => Promise): Hono { (SELECT COALESCE(pc.author_handle, LOWER(pc.author)) FROM pr_comments pc WHERE pc.pr_number = pr.number ORDER BY pc.created_at DESC, pc.comment_id DESC LIMIT 1) as latest_comment_author_handle, (SELECT pc.body FROM pr_comments pc WHERE pc.pr_number = pr.number ORDER BY pc.created_at DESC, pc.comment_id DESC LIMIT 1) as latest_comment_body FROM pull_requests pr - WHERE COALESCE(pr.author_handle, LOWER(pr.author)) = ? + WHERE pr.author_handle = ? ORDER BY pr.updated_at DESC, pr.number DESC LIMIT 40 `, [handle]); @@ -617,7 +574,7 @@ export function createRoutes(getDb: () => Promise): Hono { pr.title as pr_title FROM pr_comments pc JOIN pull_requests pr ON pr.number = pc.pr_number - WHERE COALESCE(pc.author_handle, LOWER(pc.author)) = ? + WHERE pc.author_handle = ? ORDER BY pc.created_at DESC, pc.comment_id DESC LIMIT 8 `, [handle]); diff --git a/src/web/worker.ts b/src/web/worker.ts index ab630b2..07c342f 100644 --- a/src/web/worker.ts +++ b/src/web/worker.ts @@ -1,5 +1,6 @@ import { createApp } from './app'; import { D1BindingClient } from '../db/d1-binding'; +import { ReadCache } from './read-cache'; import DASHBOARD_HTML from './index.html'; import DASHBOARD_FAVICON from './favicon.svg'; @@ -7,11 +8,25 @@ export interface Env { DB: D1Database; } +// Reuse the application and cache across requests, isolated by database binding. +const instances = new WeakMap; cache: ReadCache }>(); + export default { async fetch(request: Request, env: Env): Promise { // The sync CLI owns schema initialization and migrations. Serving the // dashboard must never spend the write quota or fail before returning HTML. - const app = createApp(async () => new D1BindingClient(env.DB), DASHBOARD_HTML, DASHBOARD_FAVICON); - return app.fetch(request); + let instance = instances.get(env.DB); + if (!instance) { + const db = new D1BindingClient(env.DB); + const app = createApp(async () => db, DASHBOARD_HTML, DASHBOARD_FAVICON); + const cache = new ReadCache(async () => { + const row = await db.get<{ value: string }>("SELECT value FROM sync_state WHERE key = 'last_sync_at'"); + return row?.value ?? null; + }); + instance = { app, cache }; + instances.set(env.DB, instance); + } + const { app, cache } = instance; + return cache.fetch(request, async () => app.fetch(request)); }, }; diff --git a/tests/github-users-sync.test.ts b/tests/github-users-sync.test.ts index e1b7ebc..e58ae36 100644 --- a/tests/github-users-sync.test.ts +++ b/tests/github-users-sync.test.ts @@ -233,4 +233,103 @@ describe('syncPullRequests github users', () => { expect(await db.all('SELECT * FROM github_users ORDER BY handle')).toEqual(users); expect(await db.get('SELECT state FROM pull_requests WHERE number = 4')).toEqual({ state: 'merged' }); }); + + test.each(['open', 'closed', 'closed-in-history'])('replays a partial REST batch when the PR is now %s', async state => { + mockState.octokit.rest.search.issuesAndPullRequests.mockReset(); + mockState.octokit.rest.search.issuesAndPullRequests.mockResolvedValue({ data: { total_count: 1 } }); + await db.run("INSERT INTO sync_state(key,value) VALUES ('last_sync_at','previous-success')"); + // Emulate REST D1, where runBatch applies statements sequentially. Fail + // after the PR upsert but before its comments/files have been stored. + let failed = false; + const spy = vi.spyOn(db, 'runBatch').mockImplementation(async statements => { + for (const statement of statements) { + const failComment = state === 'open' && statement.sql.includes('INSERT INTO pr_comments') && statement.params[0] === 201; + const failFile = state !== 'open' && statement.sql.includes('INSERT INTO pr_files') && statement.params[0] === 1; + if (!failed && (failComment || failFile)) { + failed = true; + throw new Error('interrupted REST batch'); + } + await db.run(statement.sql, statement.params); + } + }); + await expect(syncPullRequests()).rejects.toThrow('Sync incomplete'); + expect(await db.get('SELECT updated_at FROM pull_requests WHERE number=1')) + .toEqual({ updated_at: '2026-04-01T10:00:00Z' }); + if (state === 'open') expect(await db.get('SELECT comment_id FROM pr_comments WHERE comment_id=201')).toBeNull(); + else expect(await db.all('SELECT filename FROM pr_files WHERE pr_number=1')).toEqual([]); + expect(await db.get("SELECT value FROM sync_state WHERE key='last_sync_at'")) + .toEqual({ value: 'previous-success' }); + expect(await db.get("SELECT key FROM sync_state WHERE key='pending_pr:1'")) + .toEqual({ key: 'pending_pr:1' }); + spy.mockRestore(); + if (state !== 'open') { + const paginate = mockState.octokit.paginate.getMockImplementation()!; + const prs = await paginate(mockState.pullsList, { state: 'open' }); + mockState.octokit.paginate.mockImplementation(async (endpoint: unknown, params: any) => { + if (endpoint === mockState.pullsList && params.state === 'open') return prs.filter((pr: any) => pr.number !== 1); + if (endpoint === mockState.pullsList && params.state === 'closed' && state === 'closed-in-history') { + return [{ ...prs[0], merged_at: '2026-04-01T15:00:00Z' }]; + } + return paginate(endpoint, params); + }); + const get = mockState.octokit.rest.pulls.get.getMockImplementation()!; + mockState.octokit.rest.pulls.get.mockImplementation(async (params: any) => { + if (params.pull_number === 1) return { data: { ...prs[0], state: 'closed', merged: true, mergeable: true, mergeable_state: 'clean' } }; + return get(params); + }); + } + if (state === 'closed-in-history') { + // Fail full replay once more. The lightweight closed-history pass must + // not erase the durable marker for still-missing files. + const retry = vi.spyOn(db, 'runBatch').mockImplementation(async statements => { + for (const statement of statements) { + if (statement.sql.includes('INSERT INTO pr_files') && statement.params[0] === 1) throw new Error('retry interrupted'); + await db.run(statement.sql, statement.params); + } + }); + await expect(syncPullRequests()).rejects.toThrow('Sync incomplete'); + expect(await db.get("SELECT key FROM sync_state WHERE key='pending_pr:1'")) + .toEqual({ key: 'pending_pr:1' }); + expect(await db.all('SELECT filename FROM pr_files WHERE pr_number=1')).toEqual([]); + expect(await db.get("SELECT value FROM sync_state WHERE key='last_sync_at'")) + .toEqual({ value: 'previous-success' }); + retry.mockRestore(); + } + await syncPullRequests(); + expect(await db.get('SELECT comment_id FROM pr_comments WHERE comment_id=201')) + .toEqual({ comment_id: 201 }); + expect(await db.get("SELECT key FROM sync_state WHERE key='pending_pr:1'")).toBeNull(); + expect(await db.all('SELECT filename FROM pr_files WHERE pr_number=1')) + .toEqual([{ filename: 'src/web/routes.ts' }]); + expect(await db.get('SELECT state FROM pull_requests WHERE number=1')) + .toEqual({ state: state === 'open' ? 'open' : 'merged' }); + expect(await db.get("SELECT comment_count FROM github_users WHERE handle='alice'")) + .toEqual({ comment_count: 2 }); + expect(await db.get("SELECT value FROM sync_state WHERE key='last_sync_at'")) + .not.toEqual({ value: 'previous-success' }); + }); + + test('retries aggregate failures even when every PR is unchanged, without advancing success early', async () => { + mockState.octokit.rest.search.issuesAndPullRequests.mockReset(); + mockState.octokit.rest.search.issuesAndPullRequests.mockResolvedValue({ data: { total_count: 1 } }); + await db.run("INSERT INTO sync_state(key,value) VALUES ('last_sync_at','previous-success')"); + const run = db.run.bind(db); + const spy = vi.spyOn(db, 'run').mockImplementation(async (sql, params) => { + if (sql.includes('INSERT INTO github_users')) throw new Error('aggregate interrupted'); + return run(sql, params); + }); + await expect(syncPullRequests()).rejects.toThrow('aggregate interrupted'); + expect(await db.get("SELECT value FROM sync_state WHERE key='last_sync_at'")) + .toEqual({ value: 'previous-success' }); + spy.mockRestore(); + mockState.octokit.rest.pulls.get.mockClear(); + await syncPullRequests(); + expect(mockState.octokit.rest.pulls.get).not.toHaveBeenCalled(); + expect(await db.get("SELECT comment_count FROM github_users WHERE handle='alice'")) + .toEqual({ comment_count: 2 }); + expect(await db.all("SELECT key FROM sync_state WHERE key >= 'pending_user:' AND key < 'pending_user;'")) + .toEqual([]); + expect(await db.get("SELECT value FROM sync_state WHERE key='last_sync_at'")) + .not.toEqual({ value: 'previous-success' }); + }); }); diff --git a/tests/read-cache.test.ts b/tests/read-cache.test.ts new file mode 100644 index 0000000..2cde574 --- /dev/null +++ b/tests/read-cache.test.ts @@ -0,0 +1,146 @@ +import { describe, expect, test, vi } from 'vitest'; +import { ReadCache } from '../src/web/read-cache'; + +const request = (path = '/api/stats', init?: RequestInit) => new Request(`https://example.test${path}`, init); +const json = (value: unknown, status = 200) => Response.json(value, { status }); + +describe('public read cache', () => { + test('reuses responses, checks revision once a minute, and invalidates on successful sync', async () => { + let now = 0; + let revision = 'first'; + const readRevision = vi.fn(async () => revision); + const cache = new ReadCache(readRevision, { now: () => now }); + const next = vi.fn(async () => json({ revision })); + expect(await (await cache.fetch(request(), next)).json()).toEqual({ revision: 'first' }); + expect((await cache.fetch(request(), next)).headers.get('X-PR-Cache')).toBe('hit'); + expect(readRevision).toHaveBeenCalledTimes(1); + expect(next).toHaveBeenCalledTimes(1); + now = 60_001; + await cache.fetch(request(), next); + expect(readRevision).toHaveBeenCalledTimes(2); + expect(next).toHaveBeenCalledTimes(1); + revision = 'second'; + now = 120_002; + expect(await (await cache.fetch(request(), next)).json()).toEqual({ revision: 'second' }); + expect(next).toHaveBeenCalledTimes(2); + }); + + test('expires even without a successful sync, bounding stale partial writes and time-based scores', async () => { + let now = 0; + const cache = new ReadCache(async () => null, { now: () => now }); + const next = vi.fn(async () => json({ now })); + await cache.fetch(request(), next); + now = 299_999; + expect(await (await cache.fetch(request(), next)).json()).toEqual({ now: 0 }); + now = 300_000; + expect(await (await cache.fetch(request(), next)).json()).toEqual({ now }); + expect(next).toHaveBeenCalledTimes(2); + }); + + test('keeps query keys separate and coalesces concurrent misses', async () => { + const cache = new ReadCache(async () => 'one'); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const next = vi.fn(async () => { await gate; return json(['open']); }); + const first = cache.fetch(request('/api/prs?state=open'), next); + const second = cache.fetch(request('/api/prs?state=open'), next); + release(); + const responses = await Promise.all([first, second]); + expect(await Promise.all(responses.map(response => response.json()))).toEqual([['open'], ['open']]); + expect(next).toHaveBeenCalledTimes(1); + await cache.fetch(request('/api/prs?state=closed'), next); + expect(next).toHaveBeenCalledTimes(2); + }); + + test('never caches errors, private responses or bypassed routes', async () => { + const revision = vi.fn(async () => 'one'); + const cache = new ReadCache(revision); + const failed = vi.fn(async () => json({ error: 'quota' }, 500)); + await cache.fetch(request(), failed); + await cache.fetch(request(), failed); + expect(failed).toHaveBeenCalledTimes(2); + const privateResponse = vi.fn(async () => Response.json({}, { headers: { 'Cache-Control': 'private' } })); + await cache.fetch(request(), privateResponse); + await cache.fetch(request(), privateResponse); + expect(privateResponse).toHaveBeenCalledTimes(2); + const bypass = vi.fn(async () => json({})); + for (const req of [request('/'), request('/api/sync'), request('/api/scoring'), + request('/api/stats', { method: 'POST' }), request('/api/stats', { headers: { Authorization: 'test' } }), + request('/api/stats', { headers: { Cookie: 'session=test' } })]) { + await cache.fetch(req, bypass); + await cache.fetch(req, bypass); + } + expect(bypass).toHaveBeenCalledTimes(12); + expect(revision).toHaveBeenCalledTimes(1); + }); + + test('discards stale cache if revision read fails and never stores thrown failures', async () => { + let now = 0; + const revision = vi.fn(async () => 'one'); + const cache = new ReadCache(revision, { now: () => now }); + const next = vi.fn(async () => json({ count: 1 })); + await cache.fetch(request(), next); + now = 60_000; + revision.mockRejectedValueOnce(new Error('quota')); + const response = await cache.fetch(request(), async () => json({ error: 'quota' }, 500)); + expect(response.status).toBe(500); + await expect(cache.fetch(request(), async () => { throw new Error('network'); })).rejects.toThrow('network'); + await cache.fetch(request(), next); + expect(next).toHaveBeenCalledTimes(2); + }); + + test('bounds retained responses by bytes and entry count', async () => { + const cache = new ReadCache(async () => 'one', { maxEntries: 1, maxBytes: 20 }); + const next = vi.fn(async () => json({ n: 1 })); + await cache.fetch(request('/api/prs'), next); + await cache.fetch(request('/api/stats'), next); + await cache.fetch(request('/api/prs'), next); + expect(next).toHaveBeenCalledTimes(3); + const large = vi.fn(async () => json('x'.repeat(100))); + await cache.fetch(request('/api/authors'), large); + await cache.fetch(request('/api/authors'), large); + expect(large).toHaveBeenCalledTimes(2); + }); + + test('passes oversized streams through without buffering the rest or sharing the stream with followers', async () => { + const cache = new ReadCache(async () => 'one', { maxEntryBytes: 8 }); + const pulls: number[] = []; + const next = vi.fn(async () => { + const index = pulls.length; + pulls.push(0); + return new Response(new ReadableStream({ + pull(controller) { + const count = pulls[index]++; + controller.enqueue(new TextEncoder().encode(count === 0 || count === 101 ? '"' : 'a')); + if (count === 101) controller.close(); + }, + }), { headers: { 'Content-Type': 'application/json' } }); + }); + const [first, second] = await Promise.all([cache.fetch(request(), next), cache.fetch(request(), next)]); + expect(next).toHaveBeenCalledTimes(2); + expect(pulls[0]).toBeLessThan(102); + expect(pulls[1]).toBeLessThan(102); + expect(await first.json()).toBe('a'.repeat(100)); + expect(await second.json()).toBe('a'.repeat(100)); + const third = await cache.fetch(request(), next); + expect(next).toHaveBeenCalledTimes(3); + expect(await third.json()).toBe('a'.repeat(100)); + }); + + test('does not let an older in-flight result repopulate the cache after invalidation', async () => { + let now = 0; + let revision = 'old'; + const cache = new ReadCache(async () => revision, { now: () => now }); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const old = cache.fetch(request(), async () => { await gate; return json('old'); }); + // Allow the old revision read and request to start. + await new Promise(resolve => setTimeout(resolve, 0)); + revision = 'new'; + now = 60_001; + expect(await (await cache.fetch(request(), async () => json('new'))).json()).toBe('new'); + release(); + expect(await (await old).json()).toBe('old'); + expect(await (await cache.fetch(request(), async () => json('wrong'))).json()).toBe('new'); + }); +}); diff --git a/tests/sync-refresh.test.ts b/tests/sync-refresh.test.ts new file mode 100644 index 0000000..45c60a5 --- /dev/null +++ b/tests/sync-refresh.test.ts @@ -0,0 +1,103 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { initializeDb } from '../src/db/bootstrap'; +import { SqliteClient } from '../src/db/sqlite'; +import { beginPRRefresh, finishPRRefresh, pendingPRs, refreshPendingUsers } from '../src/github/sync-refresh'; +import { rebuildGitHubUsers } from '../src/github/users'; + +describe('durable incremental contributor refresh', () => { + let db: SqliteClient; + beforeEach(async () => { + db = await initializeDb(new SqliteClient(':memory:')); + await db.run(`INSERT INTO pull_requests (number,title,author,author_handle,head_sha,created_at,updated_at) + VALUES (1,'One','Alice','alice','a','2026-01-01','2026-01-01'), + (2,'Two','Bob','bob','b','2026-01-01','2026-01-01')`); + await db.run(`INSERT INTO pr_comments(comment_id,pr_number,author,author_handle,body,created_at,updated_at) + VALUES (11,1,'Reviewer','reviewer','Review','2026-01-01','2026-01-01')`); + await refreshPendingUsers(db); + }); + afterEach(async () => { await db.close(); }); + const summaries = () => db.all(`SELECT handle,display_handle,pr_count,open_pr_count,merged_pr_count, + closed_unmerged_pr_count,comment_count,latest_pr_number,latest_pr_at,latest_comment_id,latest_comment_at + FROM github_users ORDER BY handle`); + + test('refreshes old/new PR and comment authors with the same result as a full rebuild', async () => { + const untouched = await db.get("SELECT * FROM github_users WHERE handle='bob'"); + await db.run("UPDATE github_users SET created_at='2020-01-01' WHERE handle='alice'"); + await beginPRRefresh(db, 1, ['NewAlice', 'NewReviewer'], 'run-1'); + await db.run("UPDATE pull_requests SET author='NewAlice', author_handle='newalice',state='merged',updated_at='2026-02-01' WHERE number=1"); + await db.run("UPDATE pr_comments SET author='NewReviewer',author_handle='newreviewer',updated_at='2026-02-02' WHERE comment_id=11"); + await finishPRRefresh(db, 1, 'run-1'); + await refreshPendingUsers(db); + const incremental = await summaries(); + expect(incremental.map((row: any) => row.handle)).toEqual(['bob', 'newalice', 'newreviewer']); + expect(await db.get("SELECT * FROM github_users WHERE handle='bob'")).toEqual(untouched); + expect(await pendingPRs(db)).toEqual(new Set()); + await rebuildGitHubUsers(db); + expect(await summaries()).toEqual(incremental); + }); + + test('keeps pending work through a failed aggregate refresh and retries idempotently', async () => { + await beginPRRefresh(db, 1, ['Alice'], 'run-1'); + await db.run("UPDATE pull_requests SET state='closed' WHERE number=1"); + await finishPRRefresh(db, 1, 'run-1'); + const run = db.run.bind(db); + const spy = vi.spyOn(db, 'run').mockImplementation(async (sql, params) => { + if (sql.includes('INSERT INTO github_users')) throw new Error('temporary failure'); + return run(sql, params); + }); + await expect(refreshPendingUsers(db)).rejects.toThrow('temporary failure'); + expect(await db.all("SELECT key FROM sync_state WHERE key >= 'pending_user:' AND key < 'pending_user;'")) + .toEqual([{ key: 'pending_user:alice' }, { key: 'pending_user:reviewer' }]); + spy.mockRestore(); + await refreshPendingUsers(db); + expect(await db.get("SELECT closed_unmerged_pr_count FROM github_users WHERE handle='alice'")) + .toEqual({ closed_unmerged_pr_count: 1 }); + expect(await db.all("SELECT key FROM sync_state WHERE key >= 'pending_user:' AND key < 'pending_user;'")) + .toEqual([]); + const before = await summaries(); + await refreshPendingUsers(db); + expect(await summaries()).toEqual(before); + }); + + test('an unchanged refresh reads only markers and never scans or updates source tables', async () => { + const get = vi.spyOn(db, 'get'); + const all = vi.spyOn(db, 'all'); + const run = vi.spyOn(db, 'run'); + await refreshPendingUsers(db); + expect(get).toHaveBeenCalledTimes(1); + expect(all).toHaveBeenCalledTimes(1); + expect([...get.mock.calls, ...all.mock.calls].every(([sql]) => sql.includes('sync_state'))).toBe(true); + expect(run).not.toHaveBeenCalled(); + }); + + test('retains a newer enqueue when finishing an older PR or aggregate refresh', async () => { + await beginPRRefresh(db, 1, ['Alice'], 'old'); + await beginPRRefresh(db, 1, ['Alice'], 'new'); + await finishPRRefresh(db, 1, 'old'); + expect(await pendingPRs(db)).toEqual(new Set([1])); + const run = db.run.bind(db); + let advanced = false; + const spy = vi.spyOn(db, 'run').mockImplementation(async (sql, params) => { + if (sql.includes('INSERT INTO github_users') && !advanced) { + advanced = true; + await run("UPDATE sync_state SET value='newer' WHERE key='pending_user:alice'"); + } + return run(sql, params); + }); + await refreshPendingUsers(db); + spy.mockRestore(); + expect(await db.get("SELECT value FROM sync_state WHERE key='pending_user:alice'")) + .toEqual({ value: 'newer' }); + }); + + test('chunks a large pending set below the D1 bound-parameter limit', async () => { + for (let i = 0; i < 105; i++) { + await db.run('INSERT INTO sync_state(key,value) VALUES (?,?)', [`pending_user:missing${i}`, 'run']); + } + const run = vi.spyOn(db, 'run'); + await refreshPendingUsers(db); + expect(run.mock.calls.every(([, params]) => (params?.length ?? 0) <= 80)).toBe(true); + expect(await db.all("SELECT key FROM sync_state WHERE key >= 'pending_user:' AND key < 'pending_user;'")) + .toEqual([]); + }); +}); diff --git a/tests/worker.test.ts b/tests/worker.test.ts index b3e5021..af55775 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -41,4 +41,19 @@ describe('Worker read-only requests', () => { expect(prepare.mock.calls.every(([sql]) => sql.startsWith('SELECT'))).toBe(true); expect(write).not.toHaveBeenCalled(); }); + + test('keeps warm API reads across requests and isolates different database bindings', async () => { + const prepare = vi.fn((sql: string) => ({ + bind: () => ({ first: async () => sql.includes('COUNT') ? { cnt: 3 } : null }), + })); + const env = { DB: { prepare } } as any; + const request = new Request('https://example.com/api/stats'); + expect((await worker.fetch(request, env)).status).toBe(200); + const reads = prepare.mock.calls.length; + const warm = await worker.fetch(request, env); + expect(warm.headers.get('X-PR-Cache')).toBe('hit'); + expect(prepare).toHaveBeenCalledTimes(reads); + expect((await worker.fetch(request, { DB: { prepare } } as any)).headers.get('X-PR-Cache')).toBe('miss'); + expect(prepare).toHaveBeenCalledTimes(reads * 2); + }); });