diff --git a/src/servers/sidecar/headscale/companion.ts b/src/servers/sidecar/headscale/companion.ts new file mode 100644 index 00000000..190f1201 --- /dev/null +++ b/src/servers/sidecar/headscale/companion.ts @@ -0,0 +1,199 @@ +import { getActiveHeadscaleCredentials, type HeadscaleServerCredentials } from 'officerdb'; +import { badRequest, methodNotAllowed, notFound, type OfficerContext } from './routes'; + +// The Officer Companion API — a small service deployed NEXT TO each Headscale server that answers what the +// admin API structurally cannot: is the container up, what do its logs say, and start/stop/restart it. +// Contract: COMMS/HEADSCALE_COMPANION_API.md. +// +// Three facts shape everything here. +// +// 1. It lives at `${server.url}/officer-api` and authenticates with the SAME admin API key we already +// store, validated locally against Headscale's own key store — so auth keeps working while Headscale +// is down, which is exactly when `/restart` matters. Nothing new to register, and the key still never +// leaves this sidecar. +// +// 2. It is OPTIONAL and per-server. Of the four servers registered here today, one has it deployed. So +// "no companion" is a normal state, not an error: every route below answers 200 with +// `{available: false, reason}` rather than failing, and the UI degrades to what the admin API can do. +// Distinguishing the two 502s is the whole trick — nginx returns HTML when the companion is down, +// the companion returns JSON when a docker op fails. Branch on whether the body parses. +// +// 3. `GET /health` is ALWAYS 200, at every verdict. Never key anything off its HTTP status; read +// `verdict`. That inversion is deliberate on their side and is preserved on ours. + +/** + * Every route answers `{available: true, ...}` or `{available: false, reason}` at HTTP 200. Not having a + * companion is a state to render, not a request that failed — the admin API on the same domain is + * independent and may still be working, so this must not surface as an error the UI swallows. + */ +const unavailable = (reason: string) => ({ available: false as const, reason }); + +const DEFAULT_TIMEOUT_MS = 20_000; + +type CompanionCall = { path: string; method?: string; timeoutMs?: number; signal?: AbortSignal }; + +/** + * One request to a server's companion. Returns the raw Response, or a reason string when the companion + * itself could not be reached — the caller decides how to present that, because for this feature + * "unreachable" is information rather than a failure. + */ +async function callCompanion( + creds: HeadscaleServerCredentials, + { path, method = 'GET', timeoutMs = DEFAULT_TIMEOUT_MS, signal }: CompanionCall, +): Promise { + let res: Response; + try { + res = await fetch(`${creds.url}/officer-api${path}`, { + method, + headers: { authorization: `Bearer ${creds.apiKey}`, accept: 'application/json' }, + signal: signal ?? AbortSignal.timeout(timeoutMs), + }); + } catch (err) { + if (err instanceof Error && err.name === 'TimeoutError') return 'the companion timed out'; + // A TLS failure or DNS miss on the server's own domain: the whole host is unreachable, not just this. + return 'could not reach the server'; + } + + if (res.status === 401) return 'the companion rejected the stored API key'; + + // Both 404 and 502 are ambiguous, and the same test settles both: a JSON body means the companion + // answered (no such container / the docker op failed) and that answer belongs to the caller; a + // non-JSON body means we never reached it — nginx's own 502 page, or a route that isn't there at all. + const isJson = (res.headers.get('content-type') ?? '').includes('json'); + if (res.status === 404 && !isJson) return 'this server has no companion at /officer-api'; + if (res.status === 502 && !isJson) return 'the companion is not deployed on this server'; + if (res.status >= 500 && !isJson) return `the companion returned ${res.status}`; + return res; +} + +/** Parse a companion JSON body, or a reason when it isn't JSON after all. */ +async function readBody(res: Response): Promise | string> { + const text = await res.text().catch(() => ''); + if (!text) return 'the companion returned an empty body'; + try { + const parsed = JSON.parse(text) as unknown; + if (!parsed || typeof parsed !== 'object') return 'the companion returned an unexpected body'; + return parsed as Record; + } catch { + return 'the companion returned a non-JSON body'; + } +} + +/** The active server's credentials, or a 409 the UI already knows how to render. */ +async function activeCreds(userId: number): Promise { + const creds = await getActiveHeadscaleCredentials(userId); + if (!creds) { + return Response.json({ error: 'no active Headscale server', code: 'no_active_server' }, { status: 409 }); + } + return creds; +} + +/** `GET /_officer/companion/health` — verdict, container state and, when unhappy, its own diagnosis. */ +async function health(creds: HeadscaleServerCredentials): Promise { + const res = await callCompanion(creds, { path: '/health' }); + if (typeof res === 'string') return Response.json(unavailable(res)); + + const body = await readBody(res); + if (typeof body === 'string') return Response.json(unavailable(body)); + // Passed through as-is. The companion owns this vocabulary and versions it; re-shaping it here would mean + // a new verdict or a new likely-cause silently disappearing on the way to the screen. + return Response.json({ available: true, health: body }); +} + +/** `GET /_officer/companion/logs?tail=N` — a snapshot of the last N lines. */ +async function logs(creds: HeadscaleServerCredentials, url: URL): Promise { + const tail = Number(url.searchParams.get('tail') ?? 200); + if (!Number.isInteger(tail) || tail < 1 || tail > 2000) return badRequest('tail must be 1–2000'); + + const res = await callCompanion(creds, { path: `/logs?tail=${tail}` }); + if (typeof res === 'string') return Response.json(unavailable(res)); + + const body = await readBody(res); + if (typeof body === 'string') return Response.json(unavailable(body)); + const lines = Array.isArray(body.lines) ? body.lines.filter((l): l is string => typeof l === 'string') : []; + return Response.json({ available: true, lines }); +} + +/** + * `GET /_officer/companion/logs/stream?tail=N` — the live tail, relayed frame for frame. + * + * The browser cannot open this itself: EventSource sends no Authorization header, and the key it would need + * is one this sidecar exists to keep. So the stream is proxied, and the body is returned UNTOUCHED — a + * ReadableStream that Bun cancels when the client disconnects, which aborts the upstream fetch in turn. + * Buffering it into frames here would break that, and would also mean a log line waiting on our own flush. + */ +async function logStream(creds: HeadscaleServerCredentials, ctx: OfficerContext): Promise { + const tail = Number(ctx.url.searchParams.get('tail') ?? 200); + if (!Number.isInteger(tail) || tail < 1 || tail > 2000) return badRequest('tail must be 1–2000'); + + // No timeout: a quiet log is the normal case and must not look like a dropped connection. The request's + // own signal is the lifetime — when the panel closes, this closes. + const res = await callCompanion(creds, { + path: `/logs?tail=${tail}&follow=1`, + signal: ctx.req.signal, + }); + + // An unavailable companion still answers in the stream's own vocabulary, so the client has one parser and + // one place to show a problem rather than a second, JSON-shaped failure mode. + if (typeof res === 'string') { + return new Response(`event: error\ndata: ${res}\n\n`, { + headers: { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' }, + }); + } + + return new Response(res.body, { + status: 200, + headers: { + 'content-type': 'text/event-stream', + 'cache-control': 'no-cache', + // Belt and braces through our own proxy chain, matching what the companion already sets. + 'x-accel-buffering': 'no', + }, + }); +} + +const ACTIONS = new Set(['restart', 'stop', 'start']); + +/** + * `POST /_officer/companion/:action` — restart / stop / start the Headscale container. + * + * Every one of these drops every node's control-plane connection for the duration. That is the intended + * "kill it" behaviour and the reason the UI asks twice; it is not something to retry automatically. + */ +async function action(creds: HeadscaleServerCredentials, name: string): Promise { + // 60s: docker restart on a busy container is not fast, and a timeout here reads as "did it work?" — the + // one question this feature exists to answer. + const res = await callCompanion(creds, { path: `/${name}`, method: 'POST', timeoutMs: 60_000 }); + if (typeof res === 'string') return Response.json(unavailable(res)); + + const body = await readBody(res); + if (typeof body === 'string') return Response.json(unavailable(body)); + return Response.json({ available: true, ...body }); +} + +/** Dispatch `/_officer/companion/...`. Always acts on the ACTIVE server, like every other domain route. */ +export async function handleCompanionRoute(ctx: OfficerContext, rest: string[]): Promise { + const creds = await activeCreds(ctx.userId); + if (creds instanceof Response) return creds; + + const [head, tail] = rest; + + if (head === 'health' && !tail) { + if (ctx.req.method !== 'GET') return methodNotAllowed(); + return health(creds); + } + + if (head === 'logs') { + if (ctx.req.method !== 'GET') return methodNotAllowed(); + if (!tail) return logs(creds, ctx.url); + if (tail === 'stream') return logStream(creds, ctx); + return notFound(); + } + + if (head && ACTIONS.has(head) && !tail) { + if (ctx.req.method !== 'POST') return methodNotAllowed(); + return action(creds, head); + } + + return notFound(); +} diff --git a/src/servers/sidecar/headscale/routes.ts b/src/servers/sidecar/headscale/routes.ts index ca899a85..bc9f55d7 100644 --- a/src/servers/sidecar/headscale/routes.ts +++ b/src/servers/sidecar/headscale/routes.ts @@ -5,6 +5,7 @@ import { handleUsersRoute } from './users'; import { handleKeysRoute } from './keys'; import { handleEnrollRoute } from './enroll'; import { handleSshTestRoute } from './ssh'; +import { handleCompanionRoute } from './companion'; // Officer-owned routes for the headscale sidecar — the entire feature surface lives under /_officer/. // @@ -62,6 +63,9 @@ export async function handleOfficerRoute(req: Request, url: URL): Promise = { + ok: { tone: 'ok', label: 'Healthy', blurb: 'The container is running and Headscale is answering.' }, + degraded: { + tone: 'warn', + label: 'Degraded', + blurb: 'The container is running, but Headscale is not answering properly.', + }, + down: { tone: 'bad', label: 'Down', blurb: 'The container is not running.' }, + unknown: { tone: 'idle', label: 'Unknown', blurb: 'Docker does not know this container.' }, +}; + +const ACTIONS: { id: CompanionAction; label: string; icon: typeof RotateCw; confirm: string }[] = [ + { + id: 'restart', + label: 'Restart', + icon: RotateCw, + confirm: 'Restart the Headscale container? Every node loses its control-plane connection until it is back.', + }, + { + id: 'stop', + label: 'Stop', + icon: Square, + confirm: 'Stop the Headscale container? Every node stays disconnected until you start it again.', + }, + { id: 'start', label: 'Start', icon: Play, confirm: 'Start the Headscale container?' }, +]; + +/** The RFC3339 zero date the companion passes through from docker for a container that never finished. */ +const isZeroDate = (iso: string) => iso.startsWith('0001-'); + +const Row = ({ label, value }: { label: string; value: React.ReactNode }) => ( +
+ {label} + {value} +
+); + +const ContainerFacts = ({ container }: { container: CompanionContainer }) => ( +
+ + + + {!container.running && !isZeroDate(container.finishedAt) && ( + + )} + +
+); + +/** Evidence, shown only when the verdict is not ok — likely causes first, then the raw tail behind them. */ +const Evidence = ({ health }: { health: CompanionHealthBody }) => { + const causes = health.likelyCauses ?? []; + const recent = health.recentLogs ?? []; + if (causes.length === 0 && recent.length === 0 && !health.healthcheckOutput) return null; + + return ( +
+ {causes.length > 0 && ( +
+
+ + Likely causes +
+
    + {causes.map((cause) => ( +
  • + {cause} +
    • + ))} +
+ {/* The companion reads these out of the logs heuristically. Saying so is the difference between a + hint the owner checks and a diagnosis they trust and then chase down the wrong hole. */} +

Guessed from the logs — treat them as leads, not answers.

+
+ )} + + {health.healthcheckOutput && ( +
+
Healthcheck output
+ 
+            {health.healthcheckOutput}
+
+
+ )} + + {recent.length > 0 && ( +
+
Last lines before now
+ 
+            {recent.join('\n')}
+
+
+ )} +
+ ); +}; + +const TAILS = [100, 500, 2000]; + +const LogViewer = () => { + const [tail, setTail] = useState(200); + const [follow, setFollow] = useState(false); + const snapshot = useCompanionLogs(tail, !follow); + const stream = useCompanionLogStream(follow, tail); + const boxRef = useRef(null); + + const lines = follow ? stream.lines : snapshot.data?.available ? snapshot.data.lines : []; + + // Pin to the bottom while following. Only while following: scrolling a snapshot back to the top and having + // it yanked down again would be the viewer fighting the reader. + useEffect(() => { + if (follow && boxRef.current) boxRef.current.scrollTop = boxRef.current.scrollHeight; + }, [follow, lines.length]); + + const unavailable = !follow && snapshot.data && !snapshot.data.available ? snapshot.data.reason : null; + + return ( + +
+
+ + Logs + {follow && ( + + + {stream.live ? 'live' : 'stopped'} + + )} +
+ +
+ {TAILS.map((n) => ( + + ))} +
+ + {follow ? ( + + ) : ( + + )} +
+ + {stream.error && ( +
{stream.error}
+ )} + {unavailable &&
{unavailable}
} + + 
+        {lines.length > 0
+          ? lines.join('\n')
+          : snapshot.isLoading
+            ? 'Loading…'
+            : follow
+              ? 'Waiting for output…'
+              : 'No log lines.'}
+
+ + ); +}; + +const Lifecycle = ({ running }: { running: boolean | null }) => { + const action = useCompanionAction(); + const [pending, setPending] = useState(null); + + const run = async (id: CompanionAction, confirm: string) => { + if (!window.confirm(confirm)) return; + setPending(id); + try { + await action.mutateAsync(id); + } catch { + /* surfaced from action.error below */ + } finally { + setPending(null); + } + }; + + const result = action.data; + + return ( +
+
+ {ACTIONS.map(({ id, label, icon: Icon, confirm }) => ( + + ))} + Acts on the container, not on Officer. +
+ + {action.error != null && The action could not be sent.} + {result && !result.available && {result.reason}} + {result && result.available && !result.ok && ( + {result.error ?? 'Docker refused the action.'} + )} + {result && result.available && result.ok && ( +
+ {result.action}: {result.result} +
+ )} +
+ ); +}; + +export const DiagnosticsView = () => { + const { active } = useHeadscaleServers(); + const query = useCompanionHealth(); + const result = query.data; + + return ( + +
+ : undefined} + /> + + {result && !result.available ? ( + +
+
+ + No companion on this server +
+

+ {result.reason}. The Officer Companion is a small service deployed next to Headscale that can see its + container — it is what makes health, logs and restart possible from here. +

+

+ This says nothing about Headscale itself: it is served by the same domain but a different process, so + the other sections may be working normally. When the companion is missing, the Console section is the + way in. +

+
+ + ) : result ? ( + <> + +
+
+ +
+
+
{VERDICTS[result.health.verdict].label}
+

+ {result.health.reason ?? VERDICTS[result.health.verdict].blurb} +

+
+
+ +
+ + {result.health.probe && } + {result.health.container && } +
+ + + + + + + + ) : null} +
+ + ); +}; diff --git a/src/workspaces/officerdev/src/apps/Headscale/HeadscaleNav.tsx b/src/workspaces/officerdev/src/apps/Headscale/HeadscaleNav.tsx index a47c447c..7007a281 100644 --- a/src/workspaces/officerdev/src/apps/Headscale/HeadscaleNav.tsx +++ b/src/workspaces/officerdev/src/apps/Headscale/HeadscaleNav.tsx @@ -1,6 +1,6 @@ import type { LucideIcon } from 'lucide-react'; import { NavLink } from 'react-router'; -import { Network, Server, Laptop, Users, KeyRound, TerminalSquare, Check } from 'lucide-react'; +import { Network, Server, Laptop, Users, KeyRound, Activity, TerminalSquare, Check } from 'lucide-react'; import { HEADSCALE_SECTIONS, headscaleSectionPath, type HeadscaleSectionId } from './shared'; import { useHeadscaleServers } from './useHeadscaleServers'; @@ -18,6 +18,7 @@ const ICONS: Record = { nodes: Laptop, users: Users, keys: KeyRound, + diagnostics: Activity, console: TerminalSquare, }; diff --git a/src/workspaces/officerdev/src/apps/Headscale/HeadscaleView.tsx b/src/workspaces/officerdev/src/apps/Headscale/HeadscaleView.tsx index d1907245..6d9bdb87 100644 --- a/src/workspaces/officerdev/src/apps/Headscale/HeadscaleView.tsx +++ b/src/workspaces/officerdev/src/apps/Headscale/HeadscaleView.tsx @@ -3,6 +3,7 @@ import { ServersView } from './ServersView'; import { NodesView } from './NodesView'; import { UsersView } from './UsersView'; import { KeysView } from './KeysView'; +import { DiagnosticsView } from './DiagnosticsView'; import { ConsoleView } from './ConsoleView'; // Right panel of the /headscale workspace — renders the section named by the URL. @@ -20,6 +21,8 @@ export const HeadscaleView = () => { return ; case 'keys': return ; + case 'diagnostics': + return ; case 'console': return ; default: diff --git a/src/workspaces/officerdev/src/apps/Headscale/shared.ts b/src/workspaces/officerdev/src/apps/Headscale/shared.ts index 398be1bf..4458686b 100644 --- a/src/workspaces/officerdev/src/apps/Headscale/shared.ts +++ b/src/workspaces/officerdev/src/apps/Headscale/shared.ts @@ -9,6 +9,7 @@ export const HEADSCALE_SECTIONS = [ { id: 'nodes', label: 'Nodes' }, { id: 'users', label: 'Users' }, { id: 'keys', label: 'Pre-auth keys' }, + { id: 'diagnostics', label: 'Diagnostics' }, { id: 'console', label: 'Console' }, ] as const; @@ -56,6 +57,54 @@ export type HeadscaleSshTest = { ok: boolean; error?: string; ms: number }; /** Officer's supported floor, restated for UI copy. The sidecar is the enforcer; this is only a label. */ export const MIN_HEADSCALE_VERSION = '0.29'; +// ── Companion API ───────────────────────────────────────────────────────────────────────────────── +// The Officer Companion is a service deployed next to a Headscale server that can see the container the +// admin API is served from: whether it is running, what it logged, and start/stop/restart. It is optional +// and per-server, so `available: false` is a first-class state rather than an error — the admin API on the +// same domain is independent and may still work. Contract: COMMS/HEADSCALE_COMPANION_API.md. + +/** Never available for a companion that is missing — the reason says which flavour of missing. */ +type Unavailable = { available: false; reason: string }; + +export type CompanionVerdict = 'ok' | 'degraded' | 'down' | 'unknown'; + +export type CompanionContainer = { + status: string; + running: boolean; + exitCode: number; + restartCount: number; + startedAt: string; + /** The RFC3339 zero date (`0001-…`) while the container is running. */ + finishedAt: string; + /** Null when the image defines no healthcheck. */ + healthcheck: string | null; +}; + +export type CompanionHealthBody = { + verdict: CompanionVerdict; + /** Whether Headscale's own HTTP is answering — the "is the control plane serving?" signal. */ + connected: boolean; + container?: CompanionContainer; + /** Human string for the probe outcome, e.g. `GET /health -> 200`, `unreachable`, `container not running`. */ + probe?: string; + /** Only on `unknown`: docker has no container by that name. */ + reason?: string; + /** Only when not ok — best-effort guesses read out of the logs. May be empty. */ + likelyCauses?: string[]; + healthcheckOutput?: string | null; + recentLogs?: string[]; +}; + +export type CompanionHealthResult = ({ available: true } & { health: CompanionHealthBody }) | Unavailable; +export type CompanionLogsResult = { available: true; lines: string[] } | Unavailable; + +/** The three lifecycle verbs. `stop`/`start` are what the companion calls `disconnect`/`reconnect`. */ +export type CompanionAction = 'restart' | 'stop' | 'start'; + +export type CompanionActionResult = + | { available: true; ok: boolean; action?: string; result?: string; error?: string } + | Unavailable; + // ── Domain objects ──────────────────────────────────────────────────────────────────────────────── // Ids are strings because Headscale's are uint64 — never parse them to numbers. diff --git a/src/workspaces/officerdev/src/apps/Headscale/useHeadscaleCompanion.ts b/src/workspaces/officerdev/src/apps/Headscale/useHeadscaleCompanion.ts new file mode 100644 index 00000000..ed5f16ee --- /dev/null +++ b/src/workspaces/officerdev/src/apps/Headscale/useHeadscaleCompanion.ts @@ -0,0 +1,165 @@ +import { useEffect, useRef, useState } from 'react'; +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; +import { useClient, getHeaders } from 'hooks/useClient'; +import type { CompanionAction, CompanionActionResult, CompanionHealthResult, CompanionLogsResult } from './shared'; + +// Client for the active server's Officer Companion. Everything here goes through the headscale sidecar, +// because the companion authenticates with the Headscale admin key — which is encrypted in Postgres and +// decryptable only there. The browser never sees it and never talks to the companion directly. + +const BASE = '/headscale/_officer/companion'; +const HEALTH_KEY = ['headscale', 'companion', 'health'] as const; + +/** + * The container's health, polled. + * + * Polling is the point rather than a convenience: this section is what you have open while waiting for a + * restart to take, so it has to move on its own. 10s is fast enough to watch a container come back and slow + * enough that a degraded server isn't being hammered while it struggles. + */ +export function useCompanionHealth() { + const { get } = useClient(); + return useQuery({ + queryKey: HEALTH_KEY, + queryFn: () => get(`${BASE}/health`), + refetchInterval: 10_000, + // A verdict from ten seconds ago is stale by definition; never show one on remount without refetching. + staleTime: 0, + }); +} + +/** A snapshot of the last N lines. The live tail is a separate thing — see useCompanionLogStream. */ +export function useCompanionLogs(tail: number, enabled: boolean) { + const { get } = useClient(); + return useQuery({ + queryKey: ['headscale', 'companion', 'logs', tail], + queryFn: () => get(`${BASE}/logs?tail=${tail}`), + enabled, + staleTime: 0, + }); +} + +/** + * Restart / stop / start the Headscale container. + * + * Every one of these drops every node's control-plane connection, so nothing here retries and nothing here + * fires without the owner having confirmed. The health query is invalidated on settle — including on + * failure, where "did it happen anyway?" is exactly the question. + */ +export function useCompanionAction() { + const { post } = useClient(); + const qc = useQueryClient(); + return useMutation({ + mutationFn: (action: CompanionAction) => post(`${BASE}/${action}`), + onSettled: () => qc.invalidateQueries({ queryKey: HEALTH_KEY }), + }); +} + +/** How many lines the viewer keeps. Beyond this the browser, not the server, becomes the bottleneck. */ +const MAX_LINES = 5000; + +export type LogStream = { + lines: string[]; + /** Set when the stream ended badly — the companion's own `event: error` frame, or a dropped connection. */ + error: string | null; + /** True between opening the request and the stream ending, however it ends. */ + live: boolean; + clear: () => void; +}; + +/** + * The live log tail, over `fetch()` rather than `EventSource`. + * + * `EventSource` cannot send an `Authorization` header and every hop of this chain needs one — Officer's own + * bearer token to reach the platform, and the Headscale admin key from there on. So the SSE framing is + * parsed by hand: split on blank lines, read `data:` and `event:`. It is a small parser and it only has to + * handle what the companion emits (one line per frame, an `event: error` frame before a fatal close). + * + * `follow` changing off aborts mid-stream; the abort is deliberately not reported as an error, since it is + * the owner switching the toggle rather than anything going wrong. + */ +export function useCompanionLogStream(follow: boolean, tail: number): LogStream { + const [lines, setLines] = useState([]); + const [error, setError] = useState(null); + const [live, setLive] = useState(false); + // Batched through a ref: a busy container emits faster than React can render, and one setState per line + // would spend the whole frame budget on log output. + const pending = useRef([]); + + useEffect(() => { + if (!follow) return; + + const controller = new AbortController(); + let flushTimer: ReturnType | null = null; + setError(null); + setLive(true); + + const flush = () => { + if (pending.current.length === 0) return; + const batch = pending.current; + pending.current = []; + setLines((prev) => { + const next = prev.concat(batch); + return next.length > MAX_LINES ? next.slice(next.length - MAX_LINES) : next; + }); + }; + + const run = async () => { + try { + const res = await fetch(`/api${BASE}/logs/stream?tail=${tail}`, { + headers: getHeaders(), + signal: controller.signal, + }); + if (!res.ok || !res.body) { + setError(`the log stream returned ${res.status}`); + return; + } + + flushTimer = setInterval(flush, 200); + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + + // Frames are separated by a blank line; anything after the last one is a partial frame and waits. + const frames = buffer.split('\n\n'); + buffer = frames.pop() ?? ''; + + for (const frame of frames) { + let event = 'message'; + const data: string[] = []; + for (const rawLine of frame.split('\n')) { + if (rawLine.startsWith('event:')) event = rawLine.slice(6).trim(); + else if (rawLine.startsWith('data:')) data.push(rawLine.slice(5).replace(/^ /, '')); + } + if (data.length === 0) continue; + const text = data.join('\n'); + if (event === 'error') setError(text); + else pending.current.push(text); + } + } + } catch (err) { + if ((err as Error | null)?.name !== 'AbortError') setError('the log stream disconnected'); + } finally { + if (flushTimer) clearInterval(flushTimer); + flush(); + // An aborted stream is already being torn down by the effect that replaced this one; letting it set + // state here would flash "not live" onto a stream that is about to reopen. + if (!controller.signal.aborted) setLive(false); + } + }; + + void run(); + return () => { + controller.abort(); + if (flushTimer) clearInterval(flushTimer); + setLive(false); + }; + }, [follow, tail]); + + return { lines, error, live, clear: () => setLines([]) }; +}