From 8e08f5d88af7fd5c60aa8c00b9402ee4700028cd Mon Sep 17 00:00:00 2001 From: Anton Niklasson Date: Fri, 10 Jul 2026 23:19:53 +0200 Subject: [PATCH] server: integrate SQLite sync engine --- package.json | 2 +- packages/server/package.json | 5 +- packages/server/src/index.ts | 12 +- packages/server/src/routes.test.ts | 135 ++++++++++++++-- packages/server/src/routes.ts | 128 +++------------ packages/server/src/sync.ts | 244 ++++++++++++++--------------- packages/sync/package.json | 2 +- packages/sync/src/cache/store.ts | 47 ++++++ pnpm-lock.yaml | 3 + 9 files changed, 330 insertions(+), 248 deletions(-) diff --git a/package.json b/package.json index b6e5e79..5301d75 100644 --- a/package.json +++ b/package.json @@ -7,7 +7,7 @@ "dev:web": "mkdir -p .logs && concurrently --kill-others -n server,web -c blue,green \"pnpm --filter server dev 2>&1 | tee .logs/server.log\" \"pnpm --filter web dev 2>&1 | tee .logs/web.log\"", "demo": "DEMO=1 pnpm dev", "demo:web": "DEMO=1 pnpm dev:web", - "build": "pnpm --filter server build && pnpm --filter web build", + "build": "pnpm --filter sync build && pnpm --filter server build && pnpm --filter web build", "build:desktop": "pnpm build && pnpm --filter desktop build && pnpm --filter desktop package", "lint": "oxlint", "fmt": "oxfmt .", diff --git a/packages/server/package.json b/packages/server/package.json index dfb9369..9ec25fc 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -3,15 +3,16 @@ "private": true, "type": "module", "scripts": { - "dev": "tsx watch src/index.ts", + "dev": "pnpm --filter sync build && tsx watch src/index.ts", "build": "tsgo", "typecheck": "tsgo --noEmit", - "test": "vitest run" + "test": "pnpm --filter sync build && vitest run" }, "dependencies": { "@hono/node-server": "^1.14.1", "@octokit/rest": "^22.0.1", "hono": "^4.7.6", + "sync": "workspace:*", "yaml": "^2.8.3", "zod": "^4.4.3" }, diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index 83f481f..74ba0ff 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -3,12 +3,9 @@ import { join, normalize, resolve } from "node:path"; import { serve } from "@hono/node-server"; import { Hono } from "hono"; import { cors } from "hono/cors"; -import { loadCache } from "./cache.js"; import { getConfigStatus, getPort } from "./config.js"; import { api } from "./routes.js"; -import { startSync } from "./sync.js"; - -loadCache(); +import { startSync, stopSync } from "./sync.js"; const app = new Hono(); @@ -63,12 +60,15 @@ const server = serve({ fetch: app.fetch, port }); // merge/approve in the packaged desktop app — then exit. The 2s timeout // fallback keeps Ctrl-C feeling instant if a request is wedged. let shuttingDown = false; -const shutdown = () => { +const shutdown = async () => { if (shuttingDown) return; shuttingDown = true; const force = setTimeout(() => process.exit(0), 2000); force.unref(); - server.close(() => process.exit(0)); + server.close(async () => { + await stopSync(); + process.exit(0); + }); }; process.on("SIGINT", shutdown); process.on("SIGTERM", shutdown); diff --git a/packages/server/src/routes.test.ts b/packages/server/src/routes.test.ts index f970fe6..9f1dc9d 100644 --- a/packages/server/src/routes.test.ts +++ b/packages/server/src/routes.test.ts @@ -69,6 +69,123 @@ vi.mock("./cache.js", () => ({ vi.mock("./config.js", () => configStub); vi.mock("./fetchers.js", () => fetchersStub); +vi.mock("./sync.js", () => { + const pending = new Set>(); + const removed: { instanceId: string; repo: string; number: number }[] = []; + const drafts: { + instanceId: string; + repo: string; + number: number; + draft: boolean; + }[] = []; + const keyFor = (kind: string) => + kind === "prs" ? "prs" : kind === "reviews" ? "reviews" : "notifications"; + const fetcherFor = (kind: string) => + kind === "prs" + ? fetchersStub.fetchPrs + : kind === "reviews" + ? fetchersStub.fetchReviews + : fetchersStub.fetchNotifications; + const resyncInstance = async (instanceId: string, kinds: string[]) => { + await Promise.all( + kinds.map(async (kind) => { + let data = (await fetcherFor(kind)(instanceId)) as { + repo?: string; + number?: number; + draft?: boolean; + }[]; + if (kind === "prs" || kind === "reviews") { + data = data + .filter( + (row) => + !removed.some( + (item) => + item.instanceId === instanceId && + item.repo === row.repo && + item.number === row.number, + ), + ) + .map((row) => { + const mutation = drafts.find( + (item) => + item.instanceId === instanceId && + item.repo === row.repo && + item.number === row.number, + ); + return mutation ? { ...row, draft: mutation.draft } : row; + }); + } + cacheStore.set(`${instanceId}:${keyFor(kind)}`, data); + }), + ); + }; + return { + getPrs: (instanceId: string, kind: string) => + cacheStore.get( + `${instanceId}:${kind === "authored" ? "prs" : "reviews"}`, + ) ?? [], + getNotifications: (instanceId: string) => + cacheStore.get(`${instanceId}:notifications`) ?? [], + removeNotification: (instanceId: string, id: string) => { + const rows = (cacheStore.get(`${instanceId}:notifications`) ?? []) as { + id: string; + }[]; + cacheStore.set( + `${instanceId}:notifications`, + rows.filter((row) => row.id !== id), + ); + }, + removePr: (instanceId: string, repo: string, number: number) => { + removed.push({ instanceId, repo, number }); + for (const kind of ["prs", "reviews"]) { + const rows = (cacheStore.get(`${instanceId}:${kind}`) ?? []) as { + repo: string; + number: number; + }[]; + cacheStore.set( + `${instanceId}:${kind}`, + rows.filter((row) => row.repo !== repo || row.number !== number), + ); + } + }, + setPrDraft: ( + instanceId: string, + repo: string, + number: number, + draft: boolean, + ) => { + drafts.push({ instanceId, repo, number, draft }); + for (const kind of ["prs", "reviews"]) { + const rows = (cacheStore.get(`${instanceId}:${kind}`) ?? []) as { + repo: string; + number: number; + }[]; + cacheStore.set( + `${instanceId}:${kind}`, + rows.map((row) => + row.repo === repo && row.number === number + ? { ...row, draft } + : row, + ), + ); + } + }, + resyncInstance, + scheduleResync: (instanceId: string, kinds: string[]) => { + const promise = resyncInstance(instanceId, kinds).finally(() => + pending.delete(promise), + ); + pending.add(promise); + }, + scheduleFullResync: () => {}, + waitForPendingResyncs: async () => { + while (pending.size > 0) await Promise.allSettled(pending); + removed.length = 0; + drafts.length = 0; + }, + }; +}); + vi.mock("./github-client.js", () => ({ getClient: async () => mockOctokit, getInstance: async (id: string) => ({ @@ -207,7 +324,7 @@ describe("POST /config/create", () => { }); describe("POST /config/reload", () => { - it("invalidates cached status, clears data caches when ready, and returns the new status", async () => { + it("invalidates cached status, schedules reconciliation, and returns the new status", async () => { configStub.getConfigStatus.mockResolvedValue({ kind: "ready", instances: [ @@ -226,9 +343,6 @@ describe("POST /config/reload", () => { const res = await call("/config/reload", { method: "POST" }); expect(res.status).toBe(200); expect(configStub.invalidateConfigStatus).toHaveBeenCalled(); - expect(cacheStore.get("github:prs")).toBeNull(); - expect(cacheStore.get("github:reviews")).toBeNull(); - expect(cacheStore.get("github:notifications")).toBeNull(); const body = await res.json(); expect(body.status).toEqual({ kind: "ready", @@ -237,7 +351,7 @@ describe("POST /config/reload", () => { expect(JSON.stringify(body)).not.toContain("SECRET"); }); - it("clears caches for instances no longer in the new payload", async () => { + it("accepts a ready payload after instances were removed", async () => { configStub.getConfigStatus.mockResolvedValue({ kind: "ready", instances: [ @@ -254,9 +368,7 @@ describe("POST /config/reload", () => { cacheStore.set("ghe:reviews", [{ id: 10 }]); cacheStore.set("ghe:notifications", [{ id: 11 }]); await call("/config/reload", { method: "POST" }); - expect(cacheStore.get("ghe:prs")).toBeNull(); - expect(cacheStore.get("ghe:reviews")).toBeNull(); - expect(cacheStore.get("ghe:notifications")).toBeNull(); + expect(configStub.invalidateConfigStatus).toHaveBeenCalled(); }); it("leaves caches alone when reloading into an error state", async () => { @@ -282,13 +394,12 @@ describe("caching behavior on GET /:instanceId/prs", () => { expect(fetchersStub.fetchPrs).not.toHaveBeenCalled(); }); - it("calls fetcher and caches the result when cache is empty", async () => { + it("returns an empty list while the first background sync is pending", async () => { fetchersStub.fetchPrs.mockResolvedValue([{ fresh: true }]); const res = await call("/github/prs"); const body = await res.json(); - expect(body).toEqual([{ fresh: true }]); - expect(fetchersStub.fetchPrs).toHaveBeenCalledWith("github"); - expect(cacheStore.get("github:prs")).toEqual([{ fresh: true }]); + expect(body).toEqual([]); + expect(fetchersStub.fetchPrs).not.toHaveBeenCalled(); }); it("?fresh=1 bypasses cache even when populated", async () => { diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index 05d8130..ef6ec1e 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -1,10 +1,4 @@ import { Hono } from "hono"; -import { - cachedInstanceIds, - getCached, - patchCache, - setCached, -} from "./cache.js"; import { createConfigFromExample, exampleConfig, @@ -14,52 +8,21 @@ import { readConfig, resolveConfigPath, } from "./config.js"; -import { - fetchNotifications, - fetchPrs, - fetchReviews, - latestCheckRunsByName, -} from "./fetchers.js"; +import { latestCheckRunsByName } from "./fetchers.js"; import { clearClients, getClient, getInstance } from "./github-client.js"; -import { recordMutation, scheduleResync } from "./sync.js"; - -interface CachedListItem { - repo: string; - number: number; - draft?: boolean; -} - -function removeFromList(repo: string, number: number) { - return (data: CachedListItem[] | null): CachedListItem[] => - (data ?? []).filter( - (item) => !(item.repo === repo && item.number === number), - ); -} - -function setDraftInList(repo: string, number: number, draft: boolean) { - return (data: CachedListItem[] | null): CachedListItem[] => - (data ?? []).map((item) => - item.repo === repo && item.number === number ? { ...item, draft } : item, - ); -} +import { + getNotifications, + getPrs, + removeNotification, + removePr, + resyncInstance, + scheduleFullResync, + scheduleResync, + setPrDraft, +} from "./sync.js"; const api = new Hono(); -/** Return cached data if available, otherwise fetch live and cache */ -async function cachedOrFetch( - key: string, - fetcher: () => Promise, - fresh = false, -): Promise { - if (!fresh) { - const cached = getCached(key); - if (cached) return cached; - } - const data = await fetcher(); - setCached(key, data); - return data; -} - async function configPayload() { const status = await getConfigStatus(); // Strip tokens before sending — the web only needs id/label/username. @@ -94,23 +57,13 @@ api.get("/config", async (c) => { }); api.post("/config/reload", async (c) => { - // Snapshot cached instance IDs *before* invalidating so we can wipe caches - // for instances the user removed from config — they won't appear in the - // new payload but their stale data would otherwise survive. - const previousIds = cachedInstanceIds(); invalidateConfigStatus(); clearClients(); const payload = await configPayload(); if (payload.status.kind === "ready") { - const ids = new Set([ - ...previousIds, - ...payload.status.instances.map((i) => i.id), - ]); - for (const id of ids) { - setCached(`${id}:prs`, null); - setCached(`${id}:reviews`, null); - setCached(`${id}:notifications`, null); - } + // Reconcile removed instances and populate newly added ones without + // making the config response wait on GitHub. + scheduleFullResync(); } return c.json(payload); }); @@ -127,37 +80,24 @@ api.post("/config/create", (c) => { // Authored PRs with CI + review status api.get("/:instanceId/prs", async (c) => { const { instanceId } = c.req.param(); - const fresh = c.req.query("fresh") === "1"; - const data = await cachedOrFetch( - `${instanceId}:prs`, - () => fetchPrs(instanceId), - fresh, - ); - return c.json(data); + if (c.req.query("fresh") === "1") await resyncInstance(instanceId, ["prs"]); + return c.json(getPrs(instanceId, "authored")); }); // PRs awaiting my review api.get("/:instanceId/reviews", async (c) => { const { instanceId } = c.req.param(); - const fresh = c.req.query("fresh") === "1"; - const data = await cachedOrFetch( - `${instanceId}:reviews`, - () => fetchReviews(instanceId), - fresh, - ); - return c.json(data); + if (c.req.query("fresh") === "1") + await resyncInstance(instanceId, ["reviews"]); + return c.json(getPrs(instanceId, "review_requested")); }); // Notifications (participating) api.get("/:instanceId/notifications", async (c) => { const { instanceId } = c.req.param(); - const fresh = c.req.query("fresh") === "1"; - const data = await cachedOrFetch( - `${instanceId}:notifications`, - () => fetchNotifications(instanceId), - fresh, - ); - return c.json(data); + if (c.req.query("fresh") === "1") + await resyncInstance(instanceId, ["notifications"]); + return c.json(getNotifications(instanceId)); }); // Mark notification as done @@ -168,13 +108,7 @@ api.delete("/:instanceId/notifications/:threadId", async (c) => { await client.activity.markThreadAsDone({ thread_id: Number(threadId) }); // Optimistically remove from cache so it disappears immediately - const cached = getCached<{ id: string }[]>(`${instanceId}:notifications`); - if (cached) { - setCached( - `${instanceId}:notifications`, - cached.filter((n) => n.id !== threadId), - ); - } + removeNotification(instanceId, threadId); scheduleResync(instanceId, ["notifications"]); @@ -268,9 +202,7 @@ api.post("/:instanceId/prs/:owner/:repo/:prNumber/merge", async (c) => { return c.json({ error: "merge_rejected", message }, 422); } - patchCache(`${instanceId}:prs`, removeFromList(fullRepo, num)); - patchCache(`${instanceId}:reviews`, removeFromList(fullRepo, num)); - recordMutation(instanceId, { kind: "removed", repo: fullRepo, number: num }); + removePr(instanceId, fullRepo, num); scheduleResync(instanceId, ["prs", "reviews"]); return c.json({ ok: true }); @@ -290,9 +222,7 @@ api.post("/:instanceId/prs/:owner/:repo/:prNumber/close", async (c) => { state: "closed", }); - patchCache(`${instanceId}:prs`, removeFromList(fullRepo, num)); - patchCache(`${instanceId}:reviews`, removeFromList(fullRepo, num)); - recordMutation(instanceId, { kind: "removed", repo: fullRepo, number: num }); + removePr(instanceId, fullRepo, num); scheduleResync(instanceId, ["prs", "reviews"]); return c.json({ ok: true }); @@ -342,13 +272,7 @@ api.post("/:instanceId/prs/:owner/:repo/:prNumber/toggle-draft", async (c) => { await client.graphql(mutation, { id: pr.node_id }); const newDraft = !pr.draft; - patchCache(`${instanceId}:prs`, setDraftInList(fullRepo, num, newDraft)); - recordMutation(instanceId, { - kind: "draft", - repo: fullRepo, - number: num, - draft: newDraft, - }); + setPrDraft(instanceId, fullRepo, num, newDraft); scheduleResync(instanceId, ["prs"]); return c.json({ ok: true, draft: newDraft }); diff --git a/packages/server/src/sync.ts b/packages/server/src/sync.ts index 10548bd..d1ce7a7 100644 --- a/packages/server/src/sync.ts +++ b/packages/server/src/sync.ts @@ -1,163 +1,159 @@ -import { setCached } from "./cache.js"; -import { getInstances } from "./config.js"; -import { fetchNotifications, fetchPrs, fetchReviews } from "./fetchers.js"; - -const SYNC_INTERVAL = 30_000; // 30s - -export type ResyncKey = "prs" | "reviews" | "notifications"; - -const RESYNC_FETCHERS: Record< - ResyncKey, - (instanceId: string) => Promise -> = { - prs: fetchPrs, - reviews: fetchReviews, - notifications: fetchNotifications, -}; - -const ALL_KEYS: ResyncKey[] = ["prs", "reviews", "notifications"]; - +import { + createSqliteRepository, + createSyncEngine, + openCache, + type PrKind, + type SyncKind, +} from "sync"; + +const { db, path, wiped } = openCache(); +const repo = createSqliteRepository(db); +const engine = createSyncEngine({ repo }); const pending = new Set>(); - -// Tracks recent client-driven mutations so a stale resync (GitHub's search -// index lags by seconds after a merge/close/draft toggle) doesn't re-introduce -// the old state. Entries expire after MUTATION_TTL. -type MutationRecord = - | { kind: "removed"; repo: string; number: number; expiresAt: number } +type Mutation = + | { + kind: "removed"; + instanceId: string; + repo: string; + number: number; + expiresAt: number; + } | { kind: "draft"; + instanceId: string; repo: string; number: number; draft: boolean; expiresAt: number; }; +const mutations: Mutation[] = []; +const MUTATION_TTL_MS = 60_000; -const MUTATION_TTL = 60_000; -const mutations = new Map(); - -function mutationKey(instanceId: string, repo: string, number: number) { - return `${instanceId}:${repo}:${number}`; -} +export type ResyncKey = SyncKind; -export function recordMutation( - instanceId: string, - m: - | { kind: "removed"; repo: string; number: number } - | { kind: "draft"; repo: string; number: number; draft: boolean }, -): void { - mutations.set(mutationKey(instanceId, m.repo, m.number), { - ...m, - expiresAt: Date.now() + MUTATION_TTL, - }); +export function getPrs(instanceId: string, kind: PrKind): unknown[] { + return repo.getPrPayloads(instanceId, kind); } -function activeMutations(instanceId: string): MutationRecord[] { - const now = Date.now(); - const out: MutationRecord[] = []; - for (const [k, v] of mutations) { - if (v.expiresAt <= now) { - mutations.delete(k); - continue; - } - if (k.startsWith(`${instanceId}:`)) out.push(v); - } - return out; +export function getNotifications(instanceId: string) { + return repo.listNotifications(instanceId).map((row) => ({ + id: row.id, + title: row.title, + type: row.type, + reason: row.reason, + repo: row.repo, + url: row.url, + unread: row.unread === 1, + updatedAt: row.updated_at, + })); } -interface ListItem { - repo: string; - number: number; - draft?: boolean; +export function removePr(instanceId: string, repoName: string, number: number) { + repo.removePr(instanceId, repoName, number); + mutations.push({ + kind: "removed", + instanceId, + repo: repoName, + number, + expiresAt: Date.now() + MUTATION_TTL_MS, + }); } -function applyMutations( +export function setPrDraft( instanceId: string, - key: ResyncKey, - data: unknown, -): unknown { - if (key !== "prs" && key !== "reviews") return data; - const muts = activeMutations(instanceId); - if (muts.length === 0) return data; - const items = data as ListItem[]; - - const filtered = items.filter( - (item) => - !muts.some( - (m) => - m.kind === "removed" && - m.repo === item.repo && - m.number === item.number, - ), - ); - - if (key === "reviews") return filtered; - - return filtered.map((item) => { - const draftMut = muts.find( - (m): m is Extract => - m.kind === "draft" && m.repo === item.repo && m.number === item.number, - ); - return draftMut ? { ...item, draft: draftMut.draft } : item; + repoName: string, + number: number, + draft: boolean, +) { + repo.setPrDraft(instanceId, repoName, number, draft); + mutations.push({ + kind: "draft", + instanceId, + repo: repoName, + number, + draft, + expiresAt: Date.now() + MUTATION_TTL_MS, }); } +export function removeNotification(instanceId: string, id: string) { + repo.removeNotification(instanceId, id); +} + export async function resyncInstance( instanceId: string, - keys: ResyncKey[] = ALL_KEYS, + keys: ResyncKey[] = ["prs", "reviews", "notifications"], ): Promise { await Promise.all( - keys.map(async (key) => { - try { - const data = await RESYNC_FETCHERS[key](instanceId); - setCached( - `${instanceId}:${key}`, - applyMutations(instanceId, key, data), - ); - } catch (err) { - console.error( - `Sync failed for ${instanceId}:${key}:`, - err instanceof Error ? err.message : err, - ); + keys.map(async (kind) => { + const summary = await engine.runOnce({ instance: instanceId, kind }); + const now = Date.now(); + for (let i = mutations.length - 1; i >= 0; i--) { + if (mutations[i].expiresAt <= now) mutations.splice(i, 1); + } + if (kind === "prs" || kind === "reviews") { + for (const mutation of mutations) { + if (mutation.instanceId !== instanceId) continue; + if (mutation.kind === "removed") { + repo.removePr(instanceId, mutation.repo, mutation.number); + } else if (kind === "prs") { + repo.setPrDraft( + instanceId, + mutation.repo, + mutation.number, + mutation.draft, + ); + } + } + } + for (const result of summary.results) { + for (const fetch of result.fetches) { + if (fetch.error) { + console.error( + `Sync failed for ${result.instanceId}:${fetch.kind}: ${fetch.error}`, + ); + } + } } }), ); } -/** - * Fire-and-forget resync after a mutation. Lets the route respond fast while - * the cache is refreshed in the background, so the next client poll sees the - * new state without waiting for the 30s sync cycle. - */ export function scheduleResync(instanceId: string, keys: ResyncKey[]): void { - const p = resyncInstance(instanceId, keys).finally(() => { - pending.delete(p); + const promise = resyncInstance(instanceId, keys).finally(() => { + pending.delete(promise); }); - pending.add(p); + pending.add(promise); } -/** Test seam: await all in-flight resyncs. */ -export async function waitForPendingResyncs(): Promise { - while (pending.size > 0) { - await Promise.allSettled(pending); - } +export function scheduleFullResync(): void { + const promise = engine.runOnce().finally(() => { + pending.delete(promise); + }); + pending.add(promise); } -async function syncAll() { - const instances = await getInstances(); - if (instances.length === 0) { - console.log("No instances configured, skipping sync"); - return; - } - console.log(`Syncing ${instances.length} instance(s)...`); - await Promise.all(instances.map((inst) => resyncInstance(inst.id))); - console.log("Sync complete"); +export async function waitForPendingResyncs(): Promise { + while (pending.size > 0) await Promise.allSettled(pending); } export function startSync() { - // Initial sync immediately - syncAll(); - // Then repeat. unref() so the interval alone doesn't keep the event loop - // alive on shutdown — the HTTP listener does that, and gets closed - // explicitly when we exit. - setInterval(syncAll, SYNC_INTERVAL).unref(); + console.log(`Sync cache: ${path}${wiped ? " (schema upgraded)" : ""}`); + engine.start({ + onCycle: (summary) => { + const count = summary.results.reduce( + (total, result) => + total + result.fetches.reduce((n, fetch) => n + fetch.count, 0), + 0, + ); + console.log(`Sync complete: ${count} row(s) in ${summary.durationMs}ms`); + }, + onError: (err) => console.error("Sync failed:", err), + }); +} + +export async function stopSync(): Promise { + await engine.stop(); + await waitForPendingResyncs(); + db.close(); } diff --git a/packages/sync/package.json b/packages/sync/package.json index 3757776..25d12c8 100644 --- a/packages/sync/package.json +++ b/packages/sync/package.json @@ -7,7 +7,7 @@ }, "exports": { ".": { - "types": "./dist/index.d.ts", + "types": "./src/index.ts", "default": "./dist/index.js" } }, diff --git a/packages/sync/src/cache/store.ts b/packages/sync/src/cache/store.ts index 538dbf0..f69518b 100644 --- a/packages/sync/src/cache/store.ts +++ b/packages/sync/src/cache/store.ts @@ -78,11 +78,19 @@ export interface Repository { // prs replacePrs(instanceId: string, kind: PrKind, rows: PrRow[]): void; getPrPayloads(instanceId: string, kind: PrKind): unknown[]; + removePr(instanceId: string, repo: string, number: number): void; + setPrDraft( + instanceId: string, + repo: string, + number: number, + draft: boolean, + ): void; countPrsByKind(instanceId: string): PrKindCount[]; // notifications replaceNotifications(instanceId: string, rows: NotificationRow[]): void; listNotifications(instanceId: string): NotificationRow[]; + removeNotification(instanceId: string, id: string): void; countNotifications(instanceId: string): number; // sync state @@ -130,6 +138,15 @@ export function createSqliteRepository(db: Cache): Repository { countPrsByKind: db.prepare( "SELECT kind, COUNT(*) AS count FROM prs WHERE instance_id = ? GROUP BY kind", ), + removePr: db.prepare( + "DELETE FROM prs WHERE instance_id = ? AND repo = ? AND number = ?", + ), + selectPrsForDraftUpdate: db.prepare( + "SELECT kind, provider_ref, payload FROM prs WHERE instance_id = ? AND repo = ? AND number = ?", + ), + updatePrDraft: db.prepare( + "UPDATE prs SET draft = ?, payload = ? WHERE instance_id = ? AND kind = ? AND provider_ref = ?", + ), deleteNotifications: db.prepare( "DELETE FROM notifications WHERE instance_id = ?", @@ -147,6 +164,9 @@ export function createSqliteRepository(db: Cache): Repository { countNotifications: db.prepare( "SELECT COUNT(*) AS n FROM notifications WHERE instance_id = ?", ), + removeNotification: db.prepare( + "DELETE FROM notifications WHERE instance_id = ? AND id = ?", + ), getSyncState: db.prepare( "SELECT * FROM sync_state WHERE instance_id = ? AND kind = ?", @@ -205,6 +225,30 @@ export function createSqliteRepository(db: Cache): Repository { ( stmts.selectPrPayloads.all(instanceId, kind) as { payload: string }[] ).map((r) => JSON.parse(r.payload)), + removePr: (instanceId, repo, number) => { + stmts.removePr.run(instanceId, repo, number); + }, + setPrDraft: (instanceId, repo, number, draft) => { + const rows = stmts.selectPrsForDraftUpdate.all( + instanceId, + repo, + number, + ) as { + kind: PrKind; + provider_ref: string; + payload: string; + }[]; + for (const row of rows) { + const payload = JSON.parse(row.payload) as Record; + stmts.updatePrDraft.run( + draft ? 1 : 0, + JSON.stringify({ ...payload, draft }), + instanceId, + row.kind, + row.provider_ref, + ); + } + }, countPrsByKind: (instanceId) => stmts.countPrsByKind.all(instanceId) as PrKindCount[], @@ -213,6 +257,9 @@ export function createSqliteRepository(db: Cache): Repository { }, listNotifications: (instanceId) => stmts.listNotifications.all(instanceId) as NotificationRow[], + removeNotification: (instanceId, id) => { + stmts.removeNotification.run(instanceId, id); + }, countNotifications: (instanceId) => (stmts.countNotifications.get(instanceId) as { n: number }).n, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c08222a..bacee6b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -63,6 +63,9 @@ importers: hono: specifier: ^4.7.6 version: 4.12.9 + sync: + specifier: workspace:* + version: link:../sync yaml: specifier: ^2.8.3 version: 2.8.3