| 1 | import { createReadStream } from "node:fs"; |
| 2 | import { stat } from "node:fs/promises"; |
| 3 | import { Readable } from "node:stream"; |
| 4 | import type { JobCompletion, JobSpec, LogLine, StepReport } from "@diffinery/shared"; |
| 5 | |
| 6 | export class HttpError extends Error { |
| 7 | constructor( |
| 8 | public status: number, |
| 9 | message: string, |
| 10 | public body?: string, |
| 11 | ) { |
| 12 | super(message); |
| 13 | } |
| 14 | } |
| 15 | |
| 16 | const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); |
| 17 | |
| 18 | export interface RequestOptions { |
| 19 | method?: string; |
| 20 | /** JSON body, or a factory producing a raw body (re-invoked on retry). */ |
| 21 | json?: unknown; |
| 22 | body?: () => Promise<{ body: BodyInit; length?: number; contentType?: string }>; |
| 23 | retries?: number; |
| 24 | /** Return the response for these statuses instead of throwing. */ |
| 25 | allow?: number[]; |
| 26 | headers?: Record<string, string>; |
| 27 | } |
| 28 | |
| 29 | /** |
| 30 | * Client for `${apiUrl}/runner/jobs/${jobId}` endpoints. When `base` is null |
| 31 | * (offline `--spec` mode) all calls are no-ops. |
| 32 | */ |
| 33 | export class RunnerApi { |
| 34 | constructor( |
| 35 | public readonly base: string | null, |
| 36 | private readonly token: string, |
| 37 | private readonly opts: { minBackoffMs?: number; maxBackoffMs?: number } = {}, |
| 38 | ) {} |
| 39 | |
| 40 | get online(): boolean { |
| 41 | return this.base !== null; |
| 42 | } |
| 43 | |
| 44 | url(path: string): string { |
| 45 | return `${this.base}${path}`; |
| 46 | } |
| 47 | |
| 48 | authHeaders(): Record<string, string> { |
| 49 | return { authorization: `Bearer ${this.token}` }; |
| 50 | } |
| 51 | |
| 52 | async request(path: string, o: RequestOptions = {}): Promise<Response> { |
| 53 | if (!this.base) throw new Error("runner API not configured"); |
| 54 | return this.requestUrl(this.url(path), o); |
| 55 | } |
| 56 | |
| 57 | async requestUrl(url: string, o: RequestOptions = {}, withAuth = true): Promise<Response> { |
| 58 | const retries = o.retries ?? 6; |
| 59 | const min = this.opts.minBackoffMs ?? 500; |
| 60 | const max = this.opts.maxBackoffMs ?? 15000; |
| 61 | let lastErr: unknown; |
| 62 | for (let attempt = 0; attempt <= retries; attempt++) { |
| 63 | if (attempt > 0) await sleep(Math.min(max, min * 2 ** (attempt - 1))); |
| 64 | try { |
| 65 | const headers: Record<string, string> = { ...(withAuth ? this.authHeaders() : {}), ...(o.headers ?? {}) }; |
| 66 | const init: RequestInit & { duplex?: string } = { method: o.method ?? "GET", headers }; |
| 67 | if (o.json !== undefined) { |
| 68 | headers["content-type"] = "application/json"; |
| 69 | init.body = JSON.stringify(o.json); |
| 70 | } else if (o.body) { |
| 71 | const b = await o.body(); |
| 72 | init.body = b.body; |
| 73 | if (b.contentType) headers["content-type"] = b.contentType; |
| 74 | if (b.length !== undefined) headers["content-length"] = String(b.length); |
| 75 | init.duplex = "half"; |
| 76 | } |
| 77 | const res = await fetch(url, init); |
| 78 | if (res.ok || o.allow?.includes(res.status)) return res; |
| 79 | const text = await res.text().catch(() => ""); |
| 80 | const err = new HttpError(res.status, `${init.method} ${redactUrl(url)} -> ${res.status} ${text.slice(0, 300)}`, text); |
| 81 | if (res.status === 408 || res.status === 429 || res.status >= 500) { |
| 82 | lastErr = err; |
| 83 | continue; |
| 84 | } |
| 85 | throw err; |
| 86 | } catch (e) { |
| 87 | if (e instanceof HttpError && !(e.status === 408 || e.status === 429 || e.status >= 500)) throw e; |
| 88 | lastErr = e; |
| 89 | } |
| 90 | } |
| 91 | throw lastErr instanceof Error ? lastErr : new Error(String(lastErr)); |
| 92 | } |
| 93 | |
| 94 | async getSpec(): Promise<JobSpec> { |
| 95 | const res = await this.request("", { retries: 8 }); |
| 96 | return (await res.json()) as JobSpec; |
| 97 | } |
| 98 | |
| 99 | async postLogs(lines: LogLine[]): Promise<void> { |
| 100 | if (!this.base) return; |
| 101 | await this.request("/logs", { method: "POST", json: { lines }, retries: 0 }); |
| 102 | } |
| 103 | |
| 104 | async postStep(index: number, report: StepReport): Promise<void> { |
| 105 | if (!this.base) return; |
| 106 | await this.request(`/steps/${index}`, { method: "POST", json: report }); |
| 107 | } |
| 108 | |
| 109 | async complete(c: JobCompletion): Promise<void> { |
| 110 | if (!this.base) return; |
| 111 | await this.request("/complete", { method: "POST", json: c, retries: 10 }); |
| 112 | } |
| 113 | |
| 114 | /** Looks up a cache entry. Returns null on miss. */ |
| 115 | async cacheLookup(key: string, restoreKeys: string[]): Promise<{ key: string; url: string } | null> { |
| 116 | const q = new URLSearchParams({ key }); |
| 117 | if (restoreKeys.length) q.set("restore", restoreKeys.join(",")); |
| 118 | const res = await this.request(`/cache?${q}`, { allow: [404, 204] }); |
| 119 | if (res.status === 404 || res.status === 204) return null; |
| 120 | const j = (await res.json()) as { key?: string; url?: string } | null; |
| 121 | if (!j || !j.key) return null; |
| 122 | return { key: j.key, url: j.url ?? `${this.base}/cache/${encodeURIComponent(j.key)}` }; |
| 123 | } |
| 124 | |
| 125 | /** Downloads a URL (relative URLs resolve against the API; auth sent only to the API origin). */ |
| 126 | async download(url: string): Promise<Response> { |
| 127 | const abs = new URL(url, this.base ?? undefined).toString(); |
| 128 | const sameOrigin = this.base ? new URL(abs).origin === new URL(this.base).origin : false; |
| 129 | return this.requestUrl(abs, {}, sameOrigin); |
| 130 | } |
| 131 | |
| 132 | async putFile(path: string, file: string, contentType = "application/gzip"): Promise<Response> { |
| 133 | return this.request(path, { |
| 134 | method: "PUT", |
| 135 | body: async () => { |
| 136 | const s = await stat(file); |
| 137 | return { body: Readable.toWeb(createReadStream(file)) as unknown as BodyInit, length: s.size, contentType }; |
| 138 | }, |
| 139 | allow: [409], |
| 140 | }); |
| 141 | } |
| 142 | } |
| 143 | |
| 144 | function redactUrl(u: string): string { |
| 145 | return u.replace(/([?&](?:token|sig|signature)=)[^&]+/gi, "$1***"); |
| 146 | } |
| 147 | |
| 148 | /** |
| 149 | * Batches log lines and ships them every `intervalMs` or when `maxLines` |
| 150 | * accumulate. Lines are never dropped: failed batches are retried with |
| 151 | * exponential backoff and kept in order. |
| 152 | */ |
| 153 | export class LogBatcher { |
| 154 | private queue: LogLine[] = []; |
| 155 | private timer: NodeJS.Timeout | null = null; |
| 156 | private inflight: Promise<void> | null = null; |
| 157 | private failures = 0; |
| 158 | private nextAttemptAt = 0; |
| 159 | private closed = false; |
| 160 | |
| 161 | constructor( |
| 162 | private readonly sink: (lines: LogLine[]) => Promise<void>, |
| 163 | private readonly opts: { intervalMs?: number; maxLines?: number; maxBatch?: number; maxBackoffMs?: number; onError?: (e: unknown) => void } = {}, |
| 164 | ) { |
| 165 | this.timer = setInterval(() => void this.tick(), this.opts.intervalMs ?? 500); |
| 166 | this.timer.unref?.(); |
| 167 | } |
| 168 | |
| 169 | push(line: LogLine): void { |
| 170 | this.queue.push(line); |
| 171 | if (this.queue.length >= (this.opts.maxLines ?? 200)) void this.tick(); |
| 172 | } |
| 173 | |
| 174 | get pending(): number { |
| 175 | return this.queue.length; |
| 176 | } |
| 177 | |
| 178 | private tick(): Promise<void> { |
| 179 | if (this.inflight) return this.inflight; |
| 180 | if (this.queue.length === 0) return Promise.resolve(); |
| 181 | if (Date.now() < this.nextAttemptAt) return Promise.resolve(); |
| 182 | const batch = this.queue.slice(0, this.opts.maxBatch ?? 1000); |
| 183 | this.inflight = this.sink(batch) |
| 184 | .then(() => { |
| 185 | this.queue.splice(0, batch.length); |
| 186 | this.failures = 0; |
| 187 | this.nextAttemptAt = 0; |
| 188 | }) |
| 189 | .catch((e) => { |
| 190 | this.failures++; |
| 191 | this.nextAttemptAt = Date.now() + Math.min(this.opts.maxBackoffMs ?? 30000, 250 * 2 ** this.failures); |
| 192 | this.opts.onError?.(e); |
| 193 | }) |
| 194 | .finally(() => { |
| 195 | this.inflight = null; |
| 196 | }); |
| 197 | return this.inflight; |
| 198 | } |
| 199 | |
| 200 | /** Flushes everything, retrying until `deadlineMs` elapses. */ |
| 201 | async flush(deadlineMs = 120000): Promise<boolean> { |
| 202 | const until = Date.now() + deadlineMs; |
| 203 | while (this.queue.length > 0 || this.inflight) { |
| 204 | if (Date.now() > until) return false; |
| 205 | if (this.inflight) { |
| 206 | await this.inflight; |
| 207 | continue; |
| 208 | } |
| 209 | const wait = this.nextAttemptAt - Date.now(); |
| 210 | if (wait > 0) await sleep(Math.min(wait, Math.max(0, until - Date.now()))); |
| 211 | await this.tick(); |
| 212 | } |
| 213 | return true; |
| 214 | } |
| 215 | |
| 216 | async close(deadlineMs?: number): Promise<boolean> { |
| 217 | if (this.closed) return true; |
| 218 | this.closed = true; |
| 219 | if (this.timer) clearInterval(this.timer); |
| 220 | return this.flush(deadlineMs); |
| 221 | } |
| 222 | } |