1// Live pages: server-sent events for /events/<owner>/<repo>/ (one repo) and2// /events/ (everything). Tails $FORGE_STATE/events.jsonl, which the hooks and3// builds append to, and streams each new event to the pages watching it.4//5// Each event goes out twice: an unnamed message with a line of HTML for the6// page's live status (ssexi swaps it in), and a named event (push, built) the7// page's script acts on, e.g. refreshing its content when a build finishes.8//9// A private repo's events stay off /events/, and its /events/<owner>/<repo>/10// streams only to a reader signed in with a passkey: the session cookie is11// checked against identity.db (read-only), the user against .forge/keys and12// the repo's readers (lib/access.ts). Anyone else gets the 404 a repo that13// doesn't exist gets. Whether a repo is private is read from its14// git-daemon-export-ok at each event, not only from the event's flag, and15// the session and the right to read are checked again too, so making a16// repo private, removing a reader or signing out takes effect at once.17//18// Streams stay open, so there are only so many: at most ANON_MAX for19// people not signed in to a private repo's stream, and PER_IP_MAX from any20// one address (lib/limits.ts clientIp). Past either, a 503 with Retry-After.21import http from 'node:http';22import fs from 'node:fs';23import path from 'node:path';24import { fileURLToPath } from 'node:url';25import { canRead, isPublic as flagged, repoDir, users } from './access.ts';26import { Identity, cookieValue } from './identity.ts';27import { clientIp, timeouts } from './limits.ts';2829export const ANON_MAX = Number(process.env.FORGE_EVENTS_MAX) || 2000;30export const PER_IP_MAX = Number(process.env.FORGE_EVENTS_PER_IP) || 16;3132export interface ForgeEvent {33 t: number;34 repo: string;35 type: 'push' | 'built' | 'comment' | 'approve';36 user?: string;37 refs?: string;38 review?: number;39 ms?: number | null;40 rendered?: number;41 // Set by the hooks and builds for a private repo's events.42 private?: boolean;43}4445interface Client {46 repo: string | null;47 // Watching a private repo: the session that may.48 session: string | null;49 ip: string;50 res: http.ServerResponse;51 ping: NodeJS.Timeout;52}5354// A repo is public while its bare repo has git-daemon-export-ok.55export function isPublic(root: string, repo: string): boolean {56 const dir = repoDir(root, repo);57 return !!dir && flagged(dir);58}5960const ENTITIES: Record<string, string> = { '&': '&', '<': '<', '>': '>', '"': '"' };61const esc = (s: unknown): string => String(s ?? '').replace(/[&<>"]/g, (c) => ENTITIES[c]);6263function status(ev: ForgeEvent): string {64 if (ev.type === 'push') {65 const what = (ev.refs ?? '').split(' ').filter(Boolean).map((r) => r.replace(/^refs\/(heads|tags)\//, '')).join(', ');66 return `<span class="live-busy">${esc(ev.user || 'someone')} pushed ${esc(what || ev.repo)}, updating</span>`;67 }68 if (ev.type === 'built') return `<span class="live-on">updated ${esc(ev.repo)}</span>`;69 if (ev.type === 'comment' || ev.type === 'approve') {70 return `<span class="live-busy">${esc(ev.user || 'someone')} ${ev.type === 'comment' ? 'commented on' : 'approved'} #${esc(ev.review)}, updating</span>`;71 }72 return '';73}7475function send(res: http.ServerResponse, ev: ForgeEvent): void {76 const html = status(ev);77 if (html) res.write(`data: ${html}\n\n`);78 res.write(`event: ${ev.type}\ndata: ${JSON.stringify(ev)}\n\n`);79}8081// The repos live in the directory above the state directory (FORGE_ROOT/.forge).82export function serveEvents(state: string, port: number, root = path.dirname(state)): http.Server {83 const file = path.join(state, 'events.jsonl');84 const clients = new Set<Client>();85 // identity.db, read-only, once it exists.86 let id: Identity | null = null;87 // May the holder of this session cookie read the private repo?88 const reader = (secret: string | null, repo: string): boolean => {89 if (!secret) return false;90 id ??= Identity.open(state, { readOnly: true });91 const s = id?.session(secret);92 const dir = repoDir(root, repo);93 return !!s && !!dir && !flagged(dir) && users(state).has(s.user) && canRead(dir, s.user);94 };95 // Open streams by address, and those without a session.96 const perIp = new Map<string, number>();97 let anon = 0;98 const drop = (c: Client): void => {99 if (!clients.delete(c)) return;100 clearInterval(c.ping);101 const n = (perIp.get(c.ip) ?? 1) - 1;102 if (n > 0) perIp.set(c.ip, n);103 else perIp.delete(c.ip);104 if (!c.session) anon--;105 };106 const broadcast = (ev: ForgeEvent): void => {107 if (ev.private || !isPublic(root, ev.repo)) {108 // Only its signed-in readers hear of it; pages of a repo made private109 // since they connected, or whose reader may not read any more, hear110 // no more of it.111 for (const c of clients) {112 if (c.repo !== ev.repo) continue;113 if (reader(c.session, ev.repo)) send(c.res, ev);114 else {115 drop(c);116 c.res.end();117 }118 }119 return;120 }121 for (const c of clients) if (!c.repo || c.repo === ev.repo) send(c.res, ev);122 };123124 // Follow the file: read whatever was appended since last time.125 let offset = fs.existsSync(file) ? fs.statSync(file).size : 0;126 let partial = '';127 const poll = (): void => {128 let size: number;129 try {130 size = fs.statSync(file).size;131 } catch {132 return;133 }134 if (size < offset) offset = 0; // rotated135 if (size === offset) return;136 const fd = fs.openSync(file, 'r');137 const buf = Buffer.alloc(size - offset);138 fs.readSync(fd, buf, 0, buf.length, offset);139 fs.closeSync(fd);140 offset = size;141 const lines = (partial + buf.toString('utf8')).split('\n');142 partial = lines.pop() ?? '';143 for (const line of lines) {144 try {145 broadcast(JSON.parse(line) as ForgeEvent);146 } catch {}147 }148 };149 fs.watchFile(file, { interval: 300 }, poll);150151 return timeouts(http.createServer((req, res) => {152 const m = /^\/events\/(?:([A-Za-z0-9_.-]+)\/([A-Za-z0-9_.-]+)\/?)?$/.exec(new URL(req.url ?? '/', 'http://x').pathname);153 const repo = m?.[1] ? `${m[1]}/${m[2]}` : null;154 const secret = cookieValue(req.headers.cookie);155 let session: string | null = null;156 if (repo && !isPublic(root, repo) && reader(secret, repo)) session = secret;157 if (!m || req.method !== 'GET' || (repo && !isPublic(root, repo) && !session)) {158 res.writeHead(404).end();159 return;160 }161 const ip = clientIp(req);162 if ((perIp.get(ip) ?? 0) >= PER_IP_MAX || (!session && anon >= ANON_MAX)) {163 res.writeHead(503, { 'Retry-After': '60', 'Cache-Control': 'no-store' }).end();164 return;165 }166 res.writeHead(200, { 'Content-Type': 'text/event-stream; charset=utf-8', 'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no' });167 res.write('retry: 5000\n');168 // Connected and idle: nothing to show until someone pushes.169 res.write('data: \n\n');170 const client: Client = { repo, session, ip, res, ping: setInterval(() => res.write(': ping\n\n'), 20000) };171 clients.add(client);172 perIp.set(ip, (perIp.get(ip) ?? 0) + 1);173 if (!session) anon++;174 req.on('close', () => drop(client));175 })).listen(port, '127.0.0.1', () => console.log(`events on 127.0.0.1:${port}, following ${file}`));176}177178// Run directly: node lib/events-server.ts179if (process.argv[1] && fs.realpathSync(process.argv[1]) === fileURLToPath(import.meta.url)) {180 serveEvents(process.env.FORGE_STATE || path.resolve('demo/root/.forge'), Number(process.env.FORGE_EVENTS_PORT || 8091));181}