From c035fbfb9f495f4d135ef49997a982acbcff2fe0 Mon Sep 17 00:00:00 2001 From: Christian Johannesen Date: Sun, 18 Oct 2026 16:05:50 +0200 Subject: [PATCH] feat(api): add editorial API and source monitoring worker --- apps/api/package.json | 24 +++++++ apps/api/src/app.test.ts | 28 ++++++++ apps/api/src/app.ts | 94 ++++++++++++++++++++++++ apps/api/src/hash-password.ts | 4 ++ apps/api/src/repository.ts | 128 +++++++++++++++++++++++++++++++++ apps/api/src/server.ts | 12 ++++ apps/api/tsconfig.json | 5 ++ apps/worker/package.json | 8 +++ apps/worker/src/index.ts | 31 ++++++++ apps/worker/src/poller.test.ts | 21 ++++++ apps/worker/src/poller.ts | 55 ++++++++++++++ apps/worker/tsconfig.json | 5 ++ 12 files changed, 415 insertions(+) create mode 100644 apps/api/package.json create mode 100644 apps/api/src/app.test.ts create mode 100644 apps/api/src/app.ts create mode 100644 apps/api/src/hash-password.ts create mode 100644 apps/api/src/repository.ts create mode 100644 apps/api/src/server.ts create mode 100644 apps/api/tsconfig.json create mode 100644 apps/worker/package.json create mode 100644 apps/worker/src/index.ts create mode 100644 apps/worker/src/poller.test.ts create mode 100644 apps/worker/src/poller.ts create mode 100644 apps/worker/tsconfig.json diff --git a/apps/api/package.json b/apps/api/package.json new file mode 100644 index 0000000..58ce007 --- /dev/null +++ b/apps/api/package.json @@ -0,0 +1,24 @@ +{ + "name": "@tech-library/api", + "version": "0.1.0", + "type": "module", + "scripts": { + "dev": "tsx watch src/server.ts", + "build": "tsc -p tsconfig.json", + "start": "node dist/server.js", + "typecheck": "tsc -p tsconfig.json --noEmit", + "test": "vitest run src", + "password:hash": "tsx src/hash-password.ts" + }, + "dependencies": { + "@fastify/cookie": "^11.0.2", + "@fastify/cors": "^11.1.0", + "@fastify/helmet": "^13.0.2", + "@tech-library/contracts": "*", + "@tech-library/database": "*", + "argon2": "^0.44.0", + "fastify": "^5.6.0", + "zod": "^4.1.5" + }, + "devDependencies": { "tsx": "^4.20.5", "typescript": "^5.9.2", "vitest": "^5.0.5" } +} diff --git a/apps/api/src/app.test.ts b/apps/api/src/app.test.ts new file mode 100644 index 0000000..d7df47a --- /dev/null +++ b/apps/api/src/app.test.ts @@ -0,0 +1,28 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import argon2 from 'argon2'; +import { openDatabase, seedDatabase } from '@tech-library/database'; +import { buildApp } from './app.js'; + +describe('API', () => { + let context: ReturnType; + beforeEach(() => { context = openDatabase(':memory:'); seedDatabase(context.sqlite); }); + afterEach(() => context.sqlite.close()); + + it('lists and searches published topics', async () => { + const app = await buildApp({ sqlite: context.sqlite }); + const response = await app.inject('/api/v1/topics?q=TypeScript'); + expect(response.statusCode).toBe(200); + expect(response.json().items[0].slug).toBe('typescript'); + await app.close(); + }); + + it('protects admin routes and authenticates with a CSRF-bound session', async () => { + const hash = await argon2.hash('correct horse battery staple'); + const app = await buildApp({ sqlite: context.sqlite, adminPasswordHash: hash, adminUsername: 'admin', sessionSecret: 'a-secret-with-at-least-thirty-two-characters' }); + expect((await app.inject({ method: 'GET', url: '/api/v1/admin/findings' })).statusCode).toBe(401); + const login = await app.inject({ method: 'POST', url: '/api/v1/admin/login', payload: { username: 'admin', password: 'correct horse battery staple' } }); + expect(login.statusCode).toBe(200); + expect(login.json().csrfToken).toBeTruthy(); + await app.close(); + }); +}); diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts new file mode 100644 index 0000000..ea7bd41 --- /dev/null +++ b/apps/api/src/app.ts @@ -0,0 +1,94 @@ +import Fastify, { type FastifyReply, type FastifyRequest } from 'fastify'; +import cookie from '@fastify/cookie'; +import cors from '@fastify/cors'; +import helmet from '@fastify/helmet'; +import argon2 from 'argon2'; +import { randomBytes, randomUUID } from 'node:crypto'; +import type Database from 'better-sqlite3'; +import { LibraryRepository } from './repository.js'; + +declare module 'fastify' { interface FastifyRequest { adminSession?: { id: string; csrfToken: string } } } + +export interface AppOptions { sqlite: Database.Database; adminUsername?: string; adminPasswordHash?: string; sessionSecret?: string; webOrigin?: string } + +export async function buildApp(options: AppOptions) { + const app = Fastify({ logger: process.env.NODE_ENV !== 'test' }); + const repo = new LibraryRepository(options.sqlite); + const secret = options.sessionSecret ?? process.env.SESSION_SECRET ?? 'development-secret-must-be-replaced'; + await app.register(cookie, { secret }); + await app.register(cors, { origin: options.webOrigin ?? process.env.WEB_ORIGIN ?? 'http://localhost:5173', credentials: true }); + await app.register(helmet, { contentSecurityPolicy: false }); + + app.get('/health', async () => ({ status: 'ok', time: new Date().toISOString() })); + app.get('/api/v1/topics', async (request) => { + const query = request.query as Record; + return repo.listTopics({ q: query.q, domain: query.domain, type: query.type, page: Math.max(1, Number(query.page) || 1), pageSize: Math.min(100, Math.max(1, Number(query.pageSize) || 24)) }); + }); + app.get('/api/v1/topics/:slug', async (request, reply) => { + const result = repo.getTopic((request.params as { slug: string }).slug); + return result ?? reply.code(404).send({ error: 'Topic not found' }); + }); + app.get('/api/v1/topics/:slug/revisions', async (request) => repo.revisions((request.params as { slug: string }).slug)); + app.get('/api/v1/facets', async () => repo.facets()); + + app.post('/api/v1/admin/login', async (request, reply) => { + const body = request.body as { username?: string; password?: string }; + const username = options.adminUsername ?? process.env.ADMIN_USERNAME ?? 'admin'; + const hash = options.adminPasswordHash ?? process.env.ADMIN_PASSWORD_HASH; + if (!hash || body.username !== username || !(await argon2.verify(hash, body.password ?? ''))) return reply.code(401).send({ error: 'Invalid credentials' }); + const id = randomUUID(); + const csrfToken = randomBytes(24).toString('base64url'); + const now = new Date(); + const expires = new Date(now.getTime() + 8 * 60 * 60 * 1000); + options.sqlite.prepare('DELETE FROM sessions WHERE expires_at < ?').run(now.toISOString()); + options.sqlite.prepare('INSERT INTO sessions(id,csrf_token,expires_at,created_at) VALUES(?,?,?,?)').run(id, csrfToken, expires.toISOString(), now.toISOString()); + reply.setCookie('tech_library_session', id, { path: '/', httpOnly: true, sameSite: 'strict', secure: process.env.NODE_ENV === 'production', signed: true, expires }); + return { authenticated: true, csrfToken, expiresAt: expires.toISOString() }; + }); + + app.post('/api/v1/admin/logout', { preHandler: requireAdmin(options.sqlite, false) }, async (request, reply) => { + options.sqlite.prepare('DELETE FROM sessions WHERE id=?').run(request.adminSession!.id); + reply.clearCookie('tech_library_session', { path: '/' }); + return { authenticated: false }; + }); + app.get('/api/v1/admin/session', { preHandler: requireAdmin(options.sqlite, false) }, async (request) => ({ authenticated: true, csrfToken: request.adminSession!.csrfToken })); + + app.post('/api/v1/admin/revisions', { preHandler: requireAdmin(options.sqlite) }, async (request, reply) => handle(reply, () => repo.createDraft(request.body))); + app.post('/api/v1/admin/revisions/:id/approve', { preHandler: requireAdmin(options.sqlite) }, async (request, reply) => handle(reply, () => repo.approveRevision((request.params as { id: string }).id))); + app.post('/api/v1/admin/revisions/:id/reject', { preHandler: requireAdmin(options.sqlite) }, async (request, reply) => handle(reply, () => repo.rejectRevision((request.params as { id: string }).id))); + app.post('/api/v1/admin/revisions/:id/publish', { preHandler: requireAdmin(options.sqlite) }, async (request, reply) => handle(reply, () => repo.publishRevision((request.params as { id: string }).id))); + app.post('/api/v1/admin/topics/:slug/rollback', { preHandler: requireAdmin(options.sqlite) }, async (request, reply) => handle(reply, () => repo.rollback((request.params as { slug: string }).slug, (request.body as { revisionId: string }).revisionId))); + app.get('/api/v1/admin/findings', { preHandler: requireAdmin(options.sqlite, false) }, async (request) => repo.listFindings((request.query as { status?: string }).status)); + app.post('/api/v1/admin/source-poll', { preHandler: requireAdmin(options.sqlite) }, async () => { + const id = randomUUID(); + options.sqlite.prepare("INSERT INTO worker_jobs(id,type,status,created_at) VALUES(?,'source-poll','pending',?)").run(id, new Date().toISOString()); + return { id, status: 'pending' }; + }); + app.post('/api/v1/admin/findings/:id/:action', { preHandler: requireAdmin(options.sqlite) }, async (request, reply) => handle(reply, () => { + const { id, action } = request.params as { id: string; action: string }; + if (action !== 'approve' && action !== 'reject') throw new Error('Invalid action'); + return repo.reviewFinding(id, action === 'approve' ? 'approved' : 'rejected'); + })); + + app.setErrorHandler((error, _request, reply) => { + app.log.error(error); + const status = typeof error === 'object' && error !== null && 'issues' in error ? 400 : 500; + reply.code(status).send({ error: status === 400 && error instanceof Error ? error.message : 'Internal server error' }); + }); + return app; +} + +function requireAdmin(sqlite: Database.Database, csrf = true) { + return async (request: FastifyRequest, reply: FastifyReply) => { + const signed = request.unsignCookie(request.cookies.tech_library_session ?? ''); + if (!signed.valid || !signed.value) return reply.code(401).send({ error: 'Authentication required' }); + const session = sqlite.prepare('SELECT id,csrf_token csrfToken FROM sessions WHERE id=? AND expires_at>?').get(signed.value, new Date().toISOString()) as { id: string; csrfToken: string } | undefined; + if (!session) return reply.code(401).send({ error: 'Session expired' }); + if (csrf && request.headers['x-csrf-token'] !== session.csrfToken) return reply.code(403).send({ error: 'Invalid CSRF token' }); + request.adminSession = session; + }; +} + +async function handle(reply: FastifyReply, operation: () => unknown) { + try { return operation(); } catch (error) { return reply.code(400).send({ error: error instanceof Error ? error.message : 'Invalid request' }); } +} diff --git a/apps/api/src/hash-password.ts b/apps/api/src/hash-password.ts new file mode 100644 index 0000000..1de8191 --- /dev/null +++ b/apps/api/src/hash-password.ts @@ -0,0 +1,4 @@ +import argon2 from 'argon2'; +const password = process.argv[2]; +if (!password) { console.error('Usage: npm run password:hash -w @tech-library/api -- "password"'); process.exit(1); } +console.log(await argon2.hash(password, { type: argon2.argon2id })); diff --git a/apps/api/src/repository.ts b/apps/api/src/repository.ts new file mode 100644 index 0000000..61516a7 --- /dev/null +++ b/apps/api/src/repository.ts @@ -0,0 +1,128 @@ +import { randomUUID } from 'node:crypto'; +import type Database from 'better-sqlite3'; +import { topicRevisionInputSchema, type TopicDetail, type TopicListItem, type TopicRevisionInput } from '@tech-library/contracts'; + +const parse = (value: string): T => JSON.parse(value) as T; + +export class LibraryRepository { + constructor(private readonly db: Database.Database) {} + + listTopics({ q = '', domain, type, page = 1, pageSize = 24 }: { q?: string; domain?: string; type?: string; page?: number; pageSize?: number }) { + const params: unknown[] = []; + let where = 'WHERE t.active_revision_id IS NOT NULL'; + if (q) { where += ' AND t.id IN (SELECT topic_id FROM topic_search WHERE topic_search MATCH ?)'; params.push(q.replace(/["']/g, '')); } + if (domain) { where += " AND json_extract(r.payload, '$.domains') LIKE ?"; params.push(`%${domain}%`); } + if (type) { where += " AND json_extract(r.payload, '$.type') = ?"; params.push(type); } + const base = `FROM topics t JOIN revisions r ON r.id = t.active_revision_id ${where}`; + const total = Number((this.db.prepare(`SELECT count(*) AS count ${base}`).get(...params) as { count: number }).count); + const rows = this.db.prepare(`SELECT t.id, t.slug, r.payload, COALESCE(r.published_at, r.reviewed_at, r.created_at) verified_at ${base} ORDER BY json_extract(r.payload, '$.title') LIMIT ? OFFSET ?`).all(...params, pageSize, (page - 1) * pageSize) as Array>; + return { items: rows.map(toListItem), total, page, pageSize }; + } + + getTopic(slug: string): TopicDetail | null { + const row = this.db.prepare(`SELECT t.id, t.slug, r.id revision_id, r.number, r.payload, COALESCE(r.published_at, r.reviewed_at, r.created_at) verified_at FROM topics t JOIN revisions r ON r.id = t.active_revision_id WHERE t.slug = ?`).get(slug) as Record | undefined; + if (!row) return null; + const item = toListItem(row as Record); + const data = parse(String(row.payload)); + const sourceRows = this.db.prepare(`SELECT s.* FROM sources s JOIN revision_sources rs ON rs.source_id=s.id WHERE rs.revision_id=? ORDER BY s.verified_at DESC`).all(row.revision_id) as Array>; + const relationRows = this.db.prepare(`SELECT rel.type, t.id, t.slug, r.payload, COALESCE(r.published_at,r.created_at) verified_at FROM relations rel JOIN topics t ON t.id=rel.target_topic_id JOIN revisions r ON r.id=t.active_revision_id WHERE rel.source_topic_id=?`).all(row.id) as Array>; + return { + ...item, aliases: data.aliases, audience: data.audience, supportedVersions: data.supportedVersions, + content: data.content, revisionId: String(row.revision_id), revisionNumber: Number(row.number), + relations: relationRows.map((relation) => ({ type: relation.type as TopicDetail['relations'][number]['type'], topic: toListItem(relation) })), + sources: sourceRows.map((source) => ({ id: source.id!, title: source.title!, url: source.url!, publisher: source.publisher!, type: source.source_type as TopicDetail['sources'][number]['type'], verifiedAt: source.verified_at!, claims: parse(source.claims!) })) + }; + } + + facets() { + const rows = this.db.prepare(`SELECT r.payload FROM topics t JOIN revisions r ON r.id=t.active_revision_id`).all() as Array<{ payload: string }>; + const payloads = rows.map((row) => parse(row.payload)); + return { + domains: countValues(payloads.flatMap((value) => value.domains)), + types: countValues(payloads.map((value) => value.type)), + tags: countValues(payloads.flatMap((value) => value.tags)) + }; + } + + revisions(slug: string) { + return this.db.prepare(`SELECT r.id, r.number, r.status, r.change_note changeNote, r.created_at createdAt, r.reviewed_at reviewedAt, r.published_at publishedAt, r.id=t.active_revision_id active FROM revisions r JOIN topics t ON t.id=r.topic_id WHERE t.slug=? ORDER BY r.number DESC`).all(slug); + } + + createDraft(input: unknown) { + const data = topicRevisionInputSchema.parse(input); + const now = new Date().toISOString(); + let topic = this.db.prepare('SELECT id FROM topics WHERE slug=?').get(data.slug) as { id: string } | undefined; + if (!topic) { topic = { id: randomUUID() }; this.db.prepare('INSERT INTO topics(id,slug,created_at,updated_at) VALUES(?,?,?,?)').run(topic.id, data.slug, now, now); } + const number = Number((this.db.prepare('SELECT COALESCE(max(number),0)+1 next FROM revisions WHERE topic_id=?').get(topic.id) as { next: number }).next); + const id = randomUUID(); + this.db.prepare('INSERT INTO revisions(id,topic_id,number,status,payload,change_note,created_at) VALUES(?,?,?,?,?,?,?)').run(id, topic.id, number, 'draft', JSON.stringify(data), data.changeNote, now); + this.audit('draft.created', 'revision', id, { topicId: topic.id, number }); + return { id, topicId: topic.id, number, status: 'draft' }; + } + + approveRevision(id: string) { return this.setRevisionStatus(id, 'approved'); } + rejectRevision(id: string) { return this.setRevisionStatus(id, 'rejected'); } + + publishRevision(id: string) { + const revision = this.db.prepare('SELECT * FROM revisions WHERE id=?').get(id) as Record | undefined; + if (!revision || revision.status !== 'approved') throw new Error('Only approved revisions can be published'); + this.db.transaction(() => { + this.db.prepare('UPDATE revisions SET published_at=? WHERE id=?').run(new Date().toISOString(), id); + this.db.prepare('UPDATE topics SET active_revision_id=?, updated_at=? WHERE id=?').run(id, new Date().toISOString(), revision.topic_id); + this.reindex(revision.topic_id!, id); + this.audit('revision.published', 'revision', id, { topicId: revision.topic_id }); + })(); + return { id, published: true }; + } + + rollback(slug: string, revisionId: string) { + const row = this.db.prepare(`SELECT r.id, r.topic_id FROM revisions r JOIN topics t ON t.id=r.topic_id WHERE t.slug=? AND r.id=? AND r.status='approved'`).get(slug, revisionId) as { id: string; topic_id: string } | undefined; + if (!row) throw new Error('Approved revision not found'); + this.db.transaction(() => { + this.db.prepare('UPDATE topics SET active_revision_id=?,updated_at=? WHERE id=?').run(row.id, new Date().toISOString(), row.topic_id); + this.reindex(row.topic_id, row.id); + this.audit('topic.rolled_back', 'topic', row.topic_id, { revisionId }); + })(); + return { revisionId, active: true }; + } + + listFindings(status?: string) { + return status ? this.db.prepare('SELECT * FROM findings WHERE status=? ORDER BY discovered_at DESC').all(status) : this.db.prepare('SELECT * FROM findings ORDER BY discovered_at DESC').all(); + } + + reviewFinding(id: string, status: 'approved' | 'rejected') { + const result = this.db.prepare("UPDATE findings SET status=?,reviewed_at=? WHERE id=? AND status='pending'").run(status, new Date().toISOString(), id); + if (!result.changes) throw new Error('Pending finding not found'); + this.audit(`finding.${status}`, 'finding', id, {}); + return { id, status }; + } + + private setRevisionStatus(id: string, status: 'approved' | 'rejected') { + const result = this.db.prepare("UPDATE revisions SET status=?,reviewed_at=? WHERE id=? AND status='draft'").run(status, new Date().toISOString(), id); + if (!result.changes) throw new Error('Draft revision not found'); + this.audit(`revision.${status}`, 'revision', id, {}); + return { id, status }; + } + + private reindex(topicId: string, revisionId: string) { + const revision = this.db.prepare('SELECT payload FROM revisions WHERE id=?').get(revisionId) as { payload: string }; + const data = parse(revision.payload); + this.db.prepare('DELETE FROM topic_search WHERE topic_id=?').run(topicId); + this.db.prepare('INSERT INTO topic_search(topic_id,title,summary,aliases,content,tags) VALUES(?,?,?,?,?,?)').run(topicId, data.title, data.summary, data.aliases.join(' '), JSON.stringify(data.content), data.tags.join(' ')); + } + + private audit(action: string, entityType: string, entityId: string, metadata: object) { + this.db.prepare('INSERT INTO audit_log(id,action,entity_type,entity_id,metadata,created_at) VALUES(?,?,?,?,?,?)').run(randomUUID(), action, entityType, entityId, JSON.stringify(metadata), new Date().toISOString()); + } +} + +function toListItem(row: Record): TopicListItem { + const data = parse(row.payload!); + return { id: row.id!, slug: row.slug!, title: data.title, summary: data.summary, type: data.type, domains: data.domains, tags: data.tags, currentVersion: data.currentVersion, verifiedAt: row.verified_at! }; +} + +function countValues(values: string[]) { + const counts = new Map(); + values.forEach((value) => counts.set(value, (counts.get(value) ?? 0) + 1)); + return [...counts].map(([value, count]) => ({ value, count })).sort((a, b) => a.value.localeCompare(b.value)); +} diff --git a/apps/api/src/server.ts b/apps/api/src/server.ts new file mode 100644 index 0000000..b54adb3 --- /dev/null +++ b/apps/api/src/server.ts @@ -0,0 +1,12 @@ +import { openDatabase, seedDatabase } from '@tech-library/database'; +import { buildApp } from './app.js'; + +const { sqlite } = openDatabase(); +seedDatabase(sqlite); +const app = await buildApp({ sqlite }); +const port = Number(process.env.PORT ?? 3001); + +const close = async () => { await app.close(); sqlite.close(); process.exit(0); }; +process.on('SIGINT', close); +process.on('SIGTERM', close); +await app.listen({ port, host: '0.0.0.0' }); diff --git a/apps/api/tsconfig.json b/apps/api/tsconfig.json new file mode 100644 index 0000000..44f4ad8 --- /dev/null +++ b/apps/api/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "outDir": "dist", "rootDir": "src" }, + "include": ["src/**/*.ts"] +} diff --git a/apps/worker/package.json b/apps/worker/package.json new file mode 100644 index 0000000..d9609f1 --- /dev/null +++ b/apps/worker/package.json @@ -0,0 +1,8 @@ +{ + "name": "@tech-library/worker", + "version": "0.1.0", + "type": "module", + "scripts": { "dev": "tsx watch src/index.ts", "build": "tsc -p tsconfig.json", "start": "node dist/index.js", "typecheck": "tsc -p tsconfig.json --noEmit", "test": "vitest run src" }, + "dependencies": { "@tech-library/database": "*" }, + "devDependencies": { "tsx": "^4.20.5", "typescript": "^5.9.2", "vitest": "^5.0.5", "@types/better-sqlite3": "^7.6.13" } +} diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts new file mode 100644 index 0000000..721f89d --- /dev/null +++ b/apps/worker/src/index.ts @@ -0,0 +1,31 @@ +import { openDatabase } from '@tech-library/database'; +import { pollSources } from './poller.js'; + +const { sqlite } = openDatabase(); +const intervalHours = Math.max(1, Number(process.env.SOURCE_POLL_INTERVAL_HOURS ?? 24)); +let running = false; + +async function run(jobId?: string) { + if (running) return; + running = true; + if (jobId) sqlite.prepare("UPDATE worker_jobs SET status='running',started_at=? WHERE id=?").run(new Date().toISOString(), jobId); + try { + const result = await pollSources(sqlite); + if (jobId) sqlite.prepare("UPDATE worker_jobs SET status='completed',completed_at=?,result=? WHERE id=?").run(new Date().toISOString(), JSON.stringify(result), jobId); + console.log(JSON.stringify({ level: 'info', event: 'source_poll.completed', ...result, time: new Date().toISOString() })); + } catch (error) { + if (jobId) sqlite.prepare("UPDATE worker_jobs SET status='failed',completed_at=?,result=? WHERE id=?").run(new Date().toISOString(), JSON.stringify({ error: error instanceof Error ? error.message : 'Unknown error' }), jobId); + console.error(JSON.stringify({ level: 'error', event: 'source_poll.failed', error: error instanceof Error ? error.message : 'Unknown error', time: new Date().toISOString() })); + } finally { running = false; } +} + +setInterval(() => { + const job = sqlite.prepare("SELECT id FROM worker_jobs WHERE type='source-poll' AND status='pending' ORDER BY created_at LIMIT 1").get() as { id: string } | undefined; + if (job) void run(job.id); +}, 5_000); +setInterval(() => void run(), intervalHours * 60 * 60 * 1000); +void run(); + +const close = () => { sqlite.close(); process.exit(0); }; +process.on('SIGINT', close); +process.on('SIGTERM', close); diff --git a/apps/worker/src/poller.test.ts b/apps/worker/src/poller.test.ts new file mode 100644 index 0000000..fe35a6f --- /dev/null +++ b/apps/worker/src/poller.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from 'vitest'; +import { openDatabase, seedDatabase } from '@tech-library/database'; +import { normalize, pollSources } from './poller.js'; + +describe('source poller', () => { + it('normalizes unstable page whitespace and scripts', () => { + expect(normalize('

Hi

world')).toBe('

Hi

world'); + }); + + it('is idempotent and creates one finding for a changed fingerprint', async () => { + const { sqlite } = openDatabase(':memory:'); + seedDatabase(sqlite); + const response = (body: string) => async () => ({ ok: true, status: 200, text: async () => body, headers: { get: () => null } }); + await pollSources(sqlite, response('Initial') as never); + await pollSources(sqlite, response('Changed') as never); + await pollSources(sqlite, response('Changed') as never); + const count = (sqlite.prepare('SELECT count(*) count FROM findings').get() as { count: number }).count; + expect(count).toBe(32); + sqlite.close(); + }); +}); diff --git a/apps/worker/src/poller.ts b/apps/worker/src/poller.ts new file mode 100644 index 0000000..0e7d481 --- /dev/null +++ b/apps/worker/src/poller.ts @@ -0,0 +1,55 @@ +import { createHash, randomUUID } from 'node:crypto'; +import type Database from 'better-sqlite3'; + +export interface FetchResult { ok: boolean; status: number; text(): Promise; headers: { get(name: string): string | null } } +export type Fetcher = (url: string, init?: { headers?: Record; signal?: AbortSignal }) => Promise; + +export async function pollSources(sqlite: Database.Database, fetcher: Fetcher = fetch as Fetcher) { + const subscriptions = sqlite.prepare('SELECT * FROM source_subscriptions WHERE enabled=1').all() as Array>; + let discovered = 0; + let unchanged = 0; + let failed = 0; + for (const source of subscriptions) { + const checkedAt = new Date().toISOString(); + try { + const response = await fetcher(source.url!, { headers: { 'user-agent': 'TechLibrary/0.1 (+source-monitor)' }, signal: AbortSignal.timeout(20_000) }); + if (!response.ok) throw new Error(`HTTP ${response.status}`); + const body = normalize(await response.text()); + const fingerprint = createHash('sha256').update(body).digest('hex'); + if (!source.last_fingerprint) { + sqlite.prepare('UPDATE source_subscriptions SET last_fingerprint=?,last_checked_at=?,last_error=NULL WHERE id=?').run(fingerprint, checkedAt, source.id); + unchanged++; + continue; + } + if (fingerprint === source.last_fingerprint) { + sqlite.prepare('UPDATE source_subscriptions SET last_checked_at=?,last_error=NULL WHERE id=?').run(checkedAt, source.id); + unchanged++; + continue; + } + const title = extractTitle(body) ?? 'Official source changed'; + const findingFingerprint = createHash('sha256').update(`${source.url}:${fingerprint}`).digest('hex'); + sqlite.transaction(() => { + sqlite.prepare(`INSERT OR IGNORE INTO findings(id,adapter,topic_id,source_url,fingerprint,status,title,summary,raw_metadata,discovered_at) VALUES(?,?,?,?,?,'pending',?,?,?,?)`).run( + randomUUID(), source.adapter, source.topic_id, source.url, findingFingerprint, title, + 'The normalized content of this official source changed. Review the source and create a revision if the change affects the library entry.', + JSON.stringify({ contentFingerprint: fingerprint, contentLength: body.length, etag: response.headers.get('etag'), lastModified: response.headers.get('last-modified') }), checkedAt + ); + sqlite.prepare('UPDATE source_subscriptions SET last_fingerprint=?,last_checked_at=?,last_error=NULL WHERE id=?').run(fingerprint, checkedAt, source.id); + })(); + discovered++; + } catch (error) { + failed++; + sqlite.prepare('UPDATE source_subscriptions SET last_checked_at=?,last_error=? WHERE id=?').run(checkedAt, error instanceof Error ? error.message : 'Unknown fetch error', source.id); + } + } + return { checked: subscriptions.length, discovered, unchanged, failed }; +} + +export function normalize(input: string) { + return input.replace(//gi, '').replace(//gi, '').replace(/\s+/g, ' ').trim(); +} + +function extractTitle(input: string) { + const match = input.match(/]*>(.*?)<\/title>/i); + return match?.[1]?.replace(/<[^>]+>/g, '').trim(); +} diff --git a/apps/worker/tsconfig.json b/apps/worker/tsconfig.json new file mode 100644 index 0000000..44f4ad8 --- /dev/null +++ b/apps/worker/tsconfig.json @@ -0,0 +1,5 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { "outDir": "dist", "rootDir": "src" }, + "include": ["src/**/*.ts"] +}