diff --git a/lexicons/net/solanaceae/nightshade/feed.json b/lexicons/net/solanaceae/nightshade/feed.json index 165f491..847a827 100644 --- a/lexicons/net/solanaceae/nightshade/feed.json +++ b/lexicons/net/solanaceae/nightshade/feed.json @@ -15,6 +15,11 @@ "format": "uri", "description": "Absolute URL of the feed (RSS/Atom XML endpoint)." }, + "siteUrl": { + "type": "string", + "format": "uri", + "description": "Absolute URL of the feed's homepage/site." + }, "title": { "type": "string", "maxLength": 500, diff --git a/src/server/atproto.ts b/src/server/atproto.ts index 5f8f85e..44c14fd 100644 --- a/src/server/atproto.ts +++ b/src/server/atproto.ts @@ -24,14 +24,15 @@ export class AtprotoRepo { return listAll(this.agent(), this.did, FEED_NSID); } - async createFeed(url: string, title?: string): Promise { + async createFeed(url: string, opts?: { title?: string; siteUrl?: string }): Promise { await this.agent().com.atproto.repo.createRecord({ repo: this.did, collection: FEED_NSID, record: { $type: FEED_NSID, url, - ...(title ? { title } : {}), + ...(opts?.siteUrl ? { siteUrl: opts.siteUrl } : {}), + ...(opts?.title ? { title: opts.title } : {}), createdAt: new Date().toISOString(), } satisfies FeedRecord, }); diff --git a/src/server/db.ts b/src/server/db.ts index be4963d..f9dc57b 100644 --- a/src/server/db.ts +++ b/src/server/db.ts @@ -36,6 +36,33 @@ export function db(): DatabaseSync { last_seen_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_browser_session_did ON browser_session(did); + + CREATE TABLE IF NOT EXISTS cached_feed ( + did TEXT NOT NULL, + rkey TEXT NOT NULL, + url TEXT NOT NULL, + site_url TEXT, + title TEXT, + created_at TEXT NOT NULL, + PRIMARY KEY (did, rkey) + ); + + CREATE TABLE IF NOT EXISTS cached_save ( + did TEXT NOT NULL, + rkey TEXT NOT NULL, + url TEXT NOT NULL, + title TEXT, + created_at TEXT NOT NULL, + read_at TEXT, + PRIMARY KEY (did, rkey) + ); + + CREATE TABLE IF NOT EXISTS cache_meta ( + did TEXT NOT NULL, + collection TEXT NOT NULL, + last_synced_at INTEGER NOT NULL, + PRIMARY KEY (did, collection) + ); `); instance = d; migrateFileStores(d); @@ -84,3 +111,4 @@ function migrateDir( } } } + diff --git a/src/server/index.ts b/src/server/index.ts index 31509e9..80273f2 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -12,6 +12,8 @@ import { DeviceTokenStore } from "./device-tokens.js"; import { buildOAuth } from "./oauth.js"; import { AtprotoRepo } from "./atproto.js"; import { Syncer } from "./sync.js"; +import { RecordCache } from "./record-cache.js"; +import { JetstreamListener } from "./jetstream.js"; import { authRoutes } from "./routes-auth.js"; import { apiRoutes } from "./routes-api.js"; import { deviceRoutes } from "./routes-device.js"; @@ -20,6 +22,7 @@ const mf = new MinifluxClient(config.miniflux.url, config.miniflux.token); const bodies = new BodyCache(); const deviceTokens = new DeviceTokenStore(config.dataDir); const oauth = await buildOAuth(); +const cache = new RecordCache(); const app = new Hono(); app.use(logger()); @@ -35,8 +38,8 @@ if (!isLoopback()) { } app.route("/auth", authRoutes(oauth, deviceTokens)); -app.route("/api", apiRoutes(oauth, mf)); -app.route("/device", deviceRoutes(oauth, mf, bodies, deviceTokens)); +app.route("/api", apiRoutes(oauth, mf, cache)); +app.route("/device", deviceRoutes(oauth, mf, bodies, deviceTokens, cache)); const packageRoot = resolve(import.meta.dirname, "../.."); const publicDir = resolve(packageRoot, "dist/public"); @@ -50,14 +53,38 @@ app.get("*", (c) => { return c.html(indexHtml); }); +// Populate cache on startup for all stored DIDs. +(async () => { + for (const did of oauth.listDids()) { + const session = await oauth.getSessionForDid(did); + if (!session) continue; + try { + await cache.syncAll(new AtprotoRepo(session)); + console.log(`cache: populated for ${did}`); + } catch (e) { + console.error(`cache: startup sync failed for ${did}:`, e); + } + } +})(); + +// Jetstream listener for real-time cache updates. +const jetstream = new JetstreamListener(cache, () => oauth.listDids()); +jetstream.start(); + // Background sync every 5 minutes for every stored DID. setInterval( async () => { for (const did of oauth.listDids()) { - if (!minifluxAllowed(did)) continue; const session = await oauth.getSessionForDid(did); if (!session) continue; - const syncer = new Syncer(config.dataDir, new AtprotoRepo(session), mf); + const repo = new AtprotoRepo(session); + try { + await cache.syncAll(repo); + } catch (e) { + console.error(`cache: periodic sync failed for ${did}:`, e); + } + if (!minifluxAllowed(did)) continue; + const syncer = new Syncer(config.dataDir, repo, mf); try { const res = await syncer.run(); if (res.added || res.removed) { diff --git a/src/server/jetstream.ts b/src/server/jetstream.ts new file mode 100644 index 0000000..5947dcc --- /dev/null +++ b/src/server/jetstream.ts @@ -0,0 +1,122 @@ +import { FEED_NSID, SAVE_NSID, type FeedRecord, type SaveRecord } from "../shared/lexicons.js"; +import type { RecordCache } from "./record-cache.js"; + +const JETSTREAM_URL = "wss://jetstream1.us-east.bsky.network/subscribe"; +const RECONNECT_DELAY_MS = 5_000; +const COLLECTIONS = [FEED_NSID, SAVE_NSID]; + +type CommitEvent = { + did: string; + time_us: number; + kind: "commit"; + commit: { + rev: string; + operation: "create" | "update" | "delete"; + collection: string; + rkey: string; + record?: Record; + cid?: string; + }; +}; + +export class JetstreamListener { + private ws: WebSocket | null = null; + private stopped = false; + private reconnectTimer: ReturnType | null = null; + + constructor( + private cache: RecordCache, + private getDids: () => string[], + ) {} + + start(): void { + this.stopped = false; + this.connect(); + } + + stop(): void { + this.stopped = true; + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + if (this.ws) { + this.ws.close(); + this.ws = null; + } + } + + private connect(): void { + if (this.stopped) return; + + const dids = this.getDids(); + if (dids.length === 0) { + this.scheduleReconnect(); + return; + } + + const url = new URL(JETSTREAM_URL); + for (const c of COLLECTIONS) { + url.searchParams.append("wantedCollections", c); + } + for (const d of dids) { + url.searchParams.append("wantedDids", d); + } + + const ws = new WebSocket(url.toString()); + this.ws = ws; + + ws.addEventListener("open", () => { + console.log("jetstream: connected"); + }); + + ws.addEventListener("message", (ev) => { + try { + const data = JSON.parse(String(ev.data)) as CommitEvent; + if (data.kind === "commit") this.handleCommit(data); + } catch { + // ignore malformed + } + }); + + ws.addEventListener("close", () => { + console.log("jetstream: disconnected"); + this.ws = null; + this.scheduleReconnect(); + }); + + ws.addEventListener("error", (e) => { + console.error("jetstream: error", e); + ws.close(); + }); + } + + private scheduleReconnect(): void { + if (this.stopped) return; + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + this.connect(); + }, RECONNECT_DELAY_MS); + } + + private handleCommit(event: CommitEvent): void { + const { did, commit } = event; + const { collection, rkey, operation, record } = commit; + + if (collection === FEED_NSID) { + if (operation === "delete") { + this.cache.deleteFeed(did, rkey); + } else if (record) { + this.cache.upsertFeed(did, rkey, record as unknown as FeedRecord); + } + } else if (collection === SAVE_NSID) { + if (operation === "delete") { + this.cache.deleteSave(did, rkey); + } else if (record) { + const saveRecord = record as unknown as SaveRecord; + this.cache.upsertSave(did, rkey, saveRecord); + if (!saveRecord.title) this.cache.backfillTitles(did); + } + } + } +} diff --git a/src/server/record-cache.ts b/src/server/record-cache.ts new file mode 100644 index 0000000..9a3004f --- /dev/null +++ b/src/server/record-cache.ts @@ -0,0 +1,182 @@ +import { db } from "./db.js"; +import type { AtprotoRepo } from "./atproto.js"; +import { fetchAndExtract } from "./readability.js"; +import { + FEED_NSID, + SAVE_NSID, + type FeedRecord, + type FeedView, + type SaveRecord, + type SaveView, +} from "../shared/lexicons.js"; + +export class RecordCache { + listFeeds(did: string): FeedView[] { + const rows = db() + .prepare("SELECT rkey, url, site_url, title, created_at FROM cached_feed WHERE did = ?") + .all(did) as Array<{ + rkey: string; url: string; site_url: string | null; + title: string | null; created_at: string; + }>; + return rows.map((r) => ({ + rkey: r.rkey, + record: { + url: r.url, + siteUrl: r.site_url ?? undefined, + title: r.title ?? undefined, + createdAt: r.created_at, + }, + })); + } + + listSaves(did: string): SaveView[] { + const rows = db() + .prepare("SELECT rkey, url, title, created_at, read_at FROM cached_save WHERE did = ?") + .all(did) as Array<{ + rkey: string; + url: string; + title: string | null; + created_at: string; + read_at: string | null; + }>; + return rows.map((r) => ({ + rkey: r.rkey, + record: { + url: r.url, + title: r.title ?? undefined, + createdAt: r.created_at, + readAt: r.read_at ?? undefined, + }, + })); + } + + getSave(did: string, rkey: string): SaveView | null { + const row = db() + .prepare("SELECT rkey, url, title, created_at, read_at FROM cached_save WHERE did = ? AND rkey = ?") + .get(did, rkey) as { + rkey: string; + url: string; + title: string | null; + created_at: string; + read_at: string | null; + } | undefined; + if (!row) return null; + return { + rkey: row.rkey, + record: { + url: row.url, + title: row.title ?? undefined, + createdAt: row.created_at, + readAt: row.read_at ?? undefined, + }, + }; + } + + upsertFeed(did: string, rkey: string, record: FeedRecord): void { + db() + .prepare( + `INSERT OR REPLACE INTO cached_feed (did, rkey, url, site_url, title, created_at) + VALUES (?, ?, ?, ?, ?, ?)`, + ) + .run(did, rkey, record.url, record.siteUrl ?? null, record.title ?? null, record.createdAt); + } + + upsertSave(did: string, rkey: string, record: SaveRecord): void { + db() + .prepare( + `INSERT OR REPLACE INTO cached_save (did, rkey, url, title, created_at, read_at) + VALUES (?, ?, ?, ?, ?, ?)`, + ) + .run(did, rkey, record.url, record.title ?? null, record.createdAt, record.readAt ?? null); + } + + deleteFeed(did: string, rkey: string): void { + db().prepare("DELETE FROM cached_feed WHERE did = ? AND rkey = ?").run(did, rkey); + } + + deleteSave(did: string, rkey: string): void { + db().prepare("DELETE FROM cached_save WHERE did = ? AND rkey = ?").run(did, rkey); + } + + hasSynced(did: string): boolean { + const row = db() + .prepare("SELECT 1 FROM cache_meta WHERE did = ? LIMIT 1") + .get(did) as { 1: number } | undefined; + return !!row; + } + + async syncAll(repo: AtprotoRepo): Promise { + await Promise.all([this.syncFeeds(repo), this.syncSaves(repo)]); + } + + async syncFeeds(repo: AtprotoRepo): Promise { + const feeds = await repo.listFeeds(); + const d = db(); + const txn = d.prepare("BEGIN"); + const commit = d.prepare("COMMIT"); + const rollback = d.prepare("ROLLBACK"); + txn.run(); + try { + d.prepare("DELETE FROM cached_feed WHERE did = ?").run(repo.did); + const ins = d.prepare( + `INSERT INTO cached_feed (did, rkey, url, site_url, title, created_at) VALUES (?, ?, ?, ?, ?, ?)`, + ); + for (const f of feeds) { + ins.run(repo.did, f.rkey, f.record.url, f.record.siteUrl ?? null, f.record.title ?? null, f.record.createdAt); + } + d.prepare( + `INSERT OR REPLACE INTO cache_meta (did, collection, last_synced_at) VALUES (?, ?, ?)`, + ).run(repo.did, FEED_NSID, Date.now()); + commit.run(); + } catch (e) { + rollback.run(); + throw e; + } + } + + async syncSaves(repo: AtprotoRepo): Promise { + const saves = await repo.listSaves(); + const d = db(); + const txn = d.prepare("BEGIN"); + const commit = d.prepare("COMMIT"); + const rollback = d.prepare("ROLLBACK"); + txn.run(); + try { + d.prepare("DELETE FROM cached_save WHERE did = ?").run(repo.did); + const ins = d.prepare( + `INSERT INTO cached_save (did, rkey, url, title, created_at, read_at) VALUES (?, ?, ?, ?, ?, ?)`, + ); + for (const s of saves) { + ins.run( + repo.did, s.rkey, s.record.url, s.record.title ?? null, + s.record.createdAt, s.record.readAt ?? null, + ); + } + d.prepare( + `INSERT OR REPLACE INTO cache_meta (did, collection, last_synced_at) VALUES (?, ?, ?)`, + ).run(repo.did, SAVE_NSID, Date.now()); + commit.run(); + } catch (e) { + rollback.run(); + throw e; + } + this.backfillTitles(repo.did); + } + + backfillTitles(did: string): void { + const rows = db() + .prepare("SELECT rkey, url FROM cached_save WHERE did = ? AND title IS NULL") + .all(did) as Array<{ rkey: string; url: string }>; + if (rows.length === 0) return; + const update = db().prepare( + "UPDATE cached_save SET title = ? WHERE did = ? AND rkey = ?", + ); + for (const row of rows) { + fetchAndExtract(row.url) + .then(({ title }) => { + if (title) update.run(title, did, row.rkey); + }) + .catch(() => {}); + } + } +} diff --git a/src/server/routes-api.ts b/src/server/routes-api.ts index 8e4522b..52d7ead 100644 --- a/src/server/routes-api.ts +++ b/src/server/routes-api.ts @@ -2,6 +2,7 @@ import { Hono } from "hono"; import { getCookie, deleteCookie } from "hono/cookie"; import { AtprotoRepo } from "./atproto.js"; import type { MinifluxClient } from "./miniflux.js"; +import type { RecordCache } from "./record-cache.js"; import { Syncer } from "./sync.js"; import { fetchAndExtract } from "./readability.js"; import type { NightshadeOAuth } from "./oauth.js"; @@ -15,7 +16,7 @@ import { type Env = { Variables: { repo: AtprotoRepo } }; -export function apiRoutes(oauth: NightshadeOAuth, mf: MinifluxClient) { +export function apiRoutes(oauth: NightshadeOAuth, mf: MinifluxClient, cache: RecordCache) { const app = new Hono(); const makeSyncer = (repo: AtprotoRepo) => @@ -40,12 +41,13 @@ export function apiRoutes(oauth: NightshadeOAuth, mf: MinifluxClient) { return next(); }); - // --- Saves --- + // --- Saves (read from cache) --- app.get("/saves", async (c) => { const repo = c.get("repo"); + if (!cache.hasSynced(repo.did)) await cache.syncSaves(repo); const unreadOnly = c.req.query("all") === undefined; const limit = Number(c.req.query("limit") ?? 200); - const views = await repo.listSaves(); + const views = cache.listSaves(repo.did); const filtered = unreadOnly ? views.filter((v) => !v.record.readAt) : views; @@ -67,34 +69,52 @@ export function apiRoutes(oauth: NightshadeOAuth, mf: MinifluxClient) { // filled in on the next successful body fetch from the e-reader path. } const view = await repo.createSave(url, title); + cache.upsertSave(repo.did, view.rkey, view.record); return c.json(view); }); app.delete("/saves/:rkey", async (c) => { const repo = c.get("repo"); - await repo.deleteSave(c.req.param("rkey")); + const rkey = c.req.param("rkey"); + await repo.deleteSave(rkey); + cache.deleteSave(repo.did, rkey); return c.body(null, 204); }); app.post("/saves/:rkey/read", async (c) => { const repo = c.get("repo"); - await repo.markSaveRead(c.req.param("rkey"), true); + const rkey = c.req.param("rkey"); + await repo.markSaveRead(rkey, true); + const existing = cache.getSave(repo.did, rkey); + if (existing) { + cache.upsertSave(repo.did, rkey, { + ...existing.record, + readAt: new Date().toISOString(), + }); + } return c.body(null, 204); }); app.post("/saves/:rkey/unread", async (c) => { const repo = c.get("repo"); - await repo.markSaveRead(c.req.param("rkey"), false); + const rkey = c.req.param("rkey"); + await repo.markSaveRead(rkey, false); + const existing = cache.getSave(repo.did, rkey); + if (existing) { + const { readAt: _, ...rest } = existing.record; + cache.upsertSave(repo.did, rkey, rest); + } return c.body(null, 204); }); - // --- Feeds (atproto-canonical, mirrored to Miniflux) --- + // --- Feeds (read from cache, mirrored to Miniflux) --- app.get("/feeds", async (c) => { const repo = c.get("repo"); if (!minifluxAllowed(repo.did)) return c.json({ error: "miniflux sync not enabled for this account" }, 403); + if (!cache.hasSynced(repo.did)) await cache.syncFeeds(repo); const [feeds, mfFeeds] = await Promise.all([ - repo.listFeeds(), + Promise.resolve(cache.listFeeds(repo.did)), mf.listFeeds().catch(() => []), ]); const siteByFeedUrl = new Map(); @@ -106,7 +126,7 @@ export function apiRoutes(oauth: NightshadeOAuth, mf: MinifluxClient) { const enriched = feeds.map((f) => ({ rkey: f.rkey, record: f.record, - siteUrl: siteByFeedUrl.get(f.record.url) ?? null, + siteUrl: f.record.siteUrl ?? siteByFeedUrl.get(f.record.url) ?? null, minifluxTitle: titleByFeedUrl.get(f.record.url) ?? null, })); enriched.sort((a, b) => { @@ -123,7 +143,8 @@ export function apiRoutes(oauth: NightshadeOAuth, mf: MinifluxClient) { return c.json({ error: "miniflux sync not enabled for this account" }, 403); const { url, title } = await c.req.json<{ url: string; title?: string }>(); if (!url) return c.json({ error: "missing url" }, 400); - await repo.createFeed(url, title); + await repo.createFeed(url, { title }); + await cache.syncFeeds(repo); try { await makeSyncer(repo).run(); } catch (e) { @@ -136,7 +157,9 @@ export function apiRoutes(oauth: NightshadeOAuth, mf: MinifluxClient) { const repo = c.get("repo"); if (!minifluxAllowed(repo.did)) return c.json({ error: "miniflux sync not enabled for this account" }, 403); - await repo.deleteFeed(c.req.param("rkey")); + const rkey = c.req.param("rkey"); + await repo.deleteFeed(rkey); + cache.deleteFeed(repo.did, rkey); try { await makeSyncer(repo).run(); } catch (e) { @@ -157,9 +180,16 @@ export function apiRoutes(oauth: NightshadeOAuth, mf: MinifluxClient) { const repo = c.get("repo"); if (!minifluxAllowed(repo.did)) return c.json({ error: "miniflux sync not enabled for this account" }, 403); + await cache.syncAll(repo); const result = await makeSyncer(repo).run(); return c.json(result); }); + app.post("/cache/sync", async (c) => { + const repo = c.get("repo"); + await cache.syncAll(repo); + return c.body(null, 204); + }); + return app; } diff --git a/src/server/routes-device.ts b/src/server/routes-device.ts index 59d3cc6..d20e45b 100644 --- a/src/server/routes-device.ts +++ b/src/server/routes-device.ts @@ -5,6 +5,7 @@ import type { BodyCache } from "./body-cache.js"; import type { MinifluxClient } from "./miniflux.js"; import type { NightshadeOAuth } from "./oauth.js"; import type { DeviceTokenStore } from "./device-tokens.js"; +import type { RecordCache } from "./record-cache.js"; import { htmlToText } from "./readability.js"; import { renderItem, renderList } from "./reader-format.js"; import { minifluxAllowed } from "./config.js"; @@ -17,6 +18,7 @@ export function deviceRoutes( mf: MinifluxClient, bodies: BodyCache, tokens: DeviceTokenStore, + cache: RecordCache, ) { const app = new Hono(); @@ -47,7 +49,7 @@ export function deviceRoutes( direction: "desc", }) : { entries: [] }, - repo.listSaves(), + Promise.resolve(cache.listSaves(repo.did)), ]); const items: UnifiedItem[] = []; @@ -85,7 +87,7 @@ export function deviceRoutes( try { if (rawId.startsWith("s")) { const rkey = rawId.slice(1); - const save = await repo.getSave(rkey); + const save = cache.getSave(repo.did, rkey) ?? await repo.getSave(rkey); if (!save) return c.text("not found\n", 404); const cached = await bodies.get(save.record.url); return c.text( diff --git a/src/server/sync.ts b/src/server/sync.ts index 6e09edb..a47a610 100644 --- a/src/server/sync.ts +++ b/src/server/sync.ts @@ -73,7 +73,7 @@ export class Syncer { if (!(url in curA)) { const feed = mfList.find((f) => f.feed_url === url); try { - await this.repo.createFeed(url, feed?.title); + await this.repo.createFeed(url, { title: feed?.title, siteUrl: feed?.site_url }); added++; } catch (e) { console.error(`sync: failed to create atproto feed ${url}:`, e); diff --git a/src/shared/lexicons.ts b/src/shared/lexicons.ts index 00f08cb..3dc979c 100644 --- a/src/shared/lexicons.ts +++ b/src/shared/lexicons.ts @@ -4,6 +4,7 @@ export const SAVE_NSID = "net.solanaceae.nightshade.save"; export type FeedRecord = { $type?: typeof FEED_NSID; url: string; + siteUrl?: string; title?: string; createdAt: string; }; diff --git a/src/web/App.tsx b/src/web/App.tsx index 12eb0af..8124a00 100644 --- a/src/web/App.tsx +++ b/src/web/App.tsx @@ -140,11 +140,22 @@ function useAsync(fn: () => Promise, deps: unknown[] = []) { function SavesView() { const [unreadOnly, setUnreadOnly] = useState(true); + const [syncing, setSyncing] = useState(false); const { data, err, loading, reload } = useAsync( () => api.listSaves(!unreadOnly), [unreadOnly], ); + const syncFromPds = async () => { + setSyncing(true); + try { + await api.syncCache(); + } finally { + setSyncing(false); + reload(); + } + }; + return ( <> @@ -161,9 +172,14 @@ function SavesView() { unread only - +
+ + +
{err &&
{err}
} {loading && !data &&
loading…
} @@ -384,7 +400,7 @@ function FeedRow({ }) { const title = view.record.title ?? view.minifluxTitle ?? view.siteUrl ?? view.record.url; - const homepage = view.siteUrl ?? view.record.url; + const homepage = view.record.siteUrl ?? view.siteUrl ?? view.record.url; const remove = async () => { if (!confirm(`Unsubscribe from "${title}"?`)) return; await api.unsubscribe(view.rkey); diff --git a/src/web/api.ts b/src/web/api.ts index e2f4bdc..eca2c25 100644 --- a/src/web/api.ts +++ b/src/web/api.ts @@ -83,6 +83,9 @@ export const api = { syncNow: (): Promise<{ added: number; removed: number }> => req("/api/sync", { method: "POST" }), + syncCache: (): Promise => + req("/api/cache/sync", { method: "POST" }), + listDeviceTokens: (): Promise => req("/auth/device-tokens"), createDeviceToken: (label: string): Promise =>