| 3cbdca6 | | | 1 | import type { Dirent } from "node:fs"; |
| 3cbdca6 | | | 2 | import { appendFile, mkdir, readFile, readdir } from "node:fs/promises"; |
| 3cbdca6 | | | 3 | import { dirname, resolve } from "node:path"; |
| 3cbdca6 | | | 4 | import type { FastifyInstance } from "fastify"; |
| 3cbdca6 | | | 5 | import { z } from "zod"; |
| 3cbdca6 | | | 6 | |
| 3cbdca6 | | | 7 | interface RingLogEntry { |
| 3cbdca6 | | | 8 | ts: string; |
| 3cbdca6 | | | 9 | source: string; |
| 3cbdca6 | | | 10 | level: string; |
| 3cbdca6 | | | 11 | message: string; |
| 3cbdca6 | | | 12 | payload: unknown; |
| 3cbdca6 | | | 13 | owner?: string; |
| 3cbdca6 | | | 14 | repo?: string; |
| 3cbdca6 | | | 15 | } |
| 3cbdca6 | | | 16 | |
| 3cbdca6 | | | 17 | interface RingInstanceSummary { |
| 3cbdca6 | | | 18 | owner: string; |
| 3cbdca6 | | | 19 | repo: string; |
| 3cbdca6 | | | 20 | total: number; |
| 3cbdca6 | | | 21 | last_ts: string | null; |
| 3cbdca6 | | | 22 | last_level: string | null; |
| 3cbdca6 | | | 23 | last_message: string | null; |
| 3cbdca6 | | | 24 | } |
| 3cbdca6 | | | 25 | |
| 3cbdca6 | | | 26 | const querySchema = z.object({ |
| 3cbdca6 | | | 27 | limit: z.coerce.number().int().min(1).max(1000).default(200), |
| 3cbdca6 | | | 28 | }); |
| 3cbdca6 | | | 29 | |
| 3cbdca6 | | | 30 | const repoParamsSchema = z.object({ |
| 3cbdca6 | | | 31 | owner: z.string().min(1).max(120).regex(/^[A-Za-z0-9._-]+$/), |
| 3cbdca6 | | | 32 | repo: z.string().min(1).max(120).regex(/^[A-Za-z0-9._-]+$/), |
| 3cbdca6 | | | 33 | }); |
| 3cbdca6 | | | 34 | |
| 3cbdca6 | | | 35 | function toRecord(value: unknown): value is Record<string, unknown> { |
| 3cbdca6 | | | 36 | return typeof value === "object" && value !== null && !Array.isArray(value); |
| 3cbdca6 | | | 37 | } |
| 3cbdca6 | | | 38 | |
| 3cbdca6 | | | 39 | function coerceHeaderString(value: string | string[] | undefined): string | null { |
| 3cbdca6 | | | 40 | if (typeof value === "string") return value.trim() || null; |
| 3cbdca6 | | | 41 | if (Array.isArray(value) && typeof value[0] === "string") { |
| 3cbdca6 | | | 42 | const first = value[0].trim(); |
| 3cbdca6 | | | 43 | return first || null; |
| 3cbdca6 | | | 44 | } |
| 3cbdca6 | | | 45 | return null; |
| 3cbdca6 | | | 46 | } |
| 3cbdca6 | | | 47 | |
| 3cbdca6 | | | 48 | function safeStringify(value: unknown): string { |
| 3cbdca6 | | | 49 | try { |
| 3cbdca6 | | | 50 | const serialized = JSON.stringify(value); |
| 3cbdca6 | | | 51 | return typeof serialized === "string" ? serialized : String(value); |
| 3cbdca6 | | | 52 | } catch { |
| 3cbdca6 | | | 53 | return String(value); |
| 3cbdca6 | | | 54 | } |
| 3cbdca6 | | | 55 | } |
| 3cbdca6 | | | 56 | |
| 3cbdca6 | | | 57 | function normalizePayload( |
| 3cbdca6 | | | 58 | body: unknown, |
| 3cbdca6 | | | 59 | fallbackSource: string |
| 3cbdca6 | | | 60 | ): RingLogEntry { |
| 3cbdca6 | | | 61 | const now = new Date().toISOString(); |
| 3cbdca6 | | | 62 | |
| 3cbdca6 | | | 63 | if (typeof body === "string") { |
| 3cbdca6 | | | 64 | return { |
| 3cbdca6 | | | 65 | ts: now, |
| 3cbdca6 | | | 66 | source: fallbackSource, |
| 3cbdca6 | | | 67 | level: "info", |
| 3cbdca6 | | | 68 | message: body, |
| 3cbdca6 | | | 69 | payload: body, |
| 3cbdca6 | | | 70 | }; |
| 3cbdca6 | | | 71 | } |
| 3cbdca6 | | | 72 | |
| 3cbdca6 | | | 73 | if (typeof body === "number" || typeof body === "boolean" || body === null) { |
| 3cbdca6 | | | 74 | return { |
| 3cbdca6 | | | 75 | ts: now, |
| 3cbdca6 | | | 76 | source: fallbackSource, |
| 3cbdca6 | | | 77 | level: "info", |
| 3cbdca6 | | | 78 | message: String(body), |
| 3cbdca6 | | | 79 | payload: body, |
| 3cbdca6 | | | 80 | }; |
| 3cbdca6 | | | 81 | } |
| 3cbdca6 | | | 82 | |
| 3cbdca6 | | | 83 | if (Array.isArray(body)) { |
| 3cbdca6 | | | 84 | return { |
| 3cbdca6 | | | 85 | ts: now, |
| 3cbdca6 | | | 86 | source: fallbackSource, |
| 3cbdca6 | | | 87 | level: "info", |
| 3cbdca6 | | | 88 | message: `Array(${body.length})`, |
| 3cbdca6 | | | 89 | payload: body, |
| 3cbdca6 | | | 90 | }; |
| 3cbdca6 | | | 91 | } |
| 3cbdca6 | | | 92 | |
| 3cbdca6 | | | 93 | if (toRecord(body)) { |
| 3cbdca6 | | | 94 | const source = |
| 3cbdca6 | | | 95 | typeof body.source === "string" && body.source.trim() |
| 3cbdca6 | | | 96 | ? body.source.trim() |
| 3cbdca6 | | | 97 | : fallbackSource; |
| 3cbdca6 | | | 98 | const level = |
| 3cbdca6 | | | 99 | typeof body.level === "string" && body.level.trim() |
| 3cbdca6 | | | 100 | ? body.level.trim() |
| 3cbdca6 | | | 101 | : "info"; |
| 3cbdca6 | | | 102 | const message = |
| 3cbdca6 | | | 103 | typeof body.message === "string" && body.message.trim() |
| 3cbdca6 | | | 104 | ? body.message |
| 3cbdca6 | | | 105 | : safeStringify(body); |
| 3cbdca6 | | | 106 | return { |
| 3cbdca6 | | | 107 | ts: now, |
| 3cbdca6 | | | 108 | source, |
| 3cbdca6 | | | 109 | level, |
| 3cbdca6 | | | 110 | message, |
| 3cbdca6 | | | 111 | payload: body, |
| 3cbdca6 | | | 112 | }; |
| 3cbdca6 | | | 113 | } |
| 3cbdca6 | | | 114 | |
| 3cbdca6 | | | 115 | return { |
| 3cbdca6 | | | 116 | ts: now, |
| 3cbdca6 | | | 117 | source: fallbackSource, |
| 3cbdca6 | | | 118 | level: "info", |
| 3cbdca6 | | | 119 | message: "Unsupported payload", |
| 3cbdca6 | | | 120 | payload: body, |
| 3cbdca6 | | | 121 | }; |
| 3cbdca6 | | | 122 | } |
| 3cbdca6 | | | 123 | |
| 3cbdca6 | | | 124 | async function readLogFile(path: string): Promise<string> { |
| 3cbdca6 | | | 125 | try { |
| 3cbdca6 | | | 126 | return await readFile(path, "utf8"); |
| 3cbdca6 | | | 127 | } catch { |
| 3cbdca6 | | | 128 | return ""; |
| 3cbdca6 | | | 129 | } |
| 3cbdca6 | | | 130 | } |
| 3cbdca6 | | | 131 | |
| 3cbdca6 | | | 132 | function parseLogLine(line: string): RingLogEntry { |
| 3cbdca6 | | | 133 | try { |
| 3cbdca6 | | | 134 | return JSON.parse(line) as RingLogEntry; |
| 3cbdca6 | | | 135 | } catch { |
| 3cbdca6 | | | 136 | return { |
| 3cbdca6 | | | 137 | ts: new Date().toISOString(), |
| 3cbdca6 | | | 138 | source: "ring", |
| 3cbdca6 | | | 139 | level: "info", |
| 3cbdca6 | | | 140 | message: line, |
| 3cbdca6 | | | 141 | payload: line, |
| 3cbdca6 | | | 142 | }; |
| 3cbdca6 | | | 143 | } |
| 3cbdca6 | | | 144 | } |
| 3cbdca6 | | | 145 | |
| 3cbdca6 | | | 146 | async function readEntries(path: string, limit: number): Promise<{ entries: RingLogEntry[]; total: number }> { |
| 3cbdca6 | | | 147 | const content = await readLogFile(path); |
| 3cbdca6 | | | 148 | const lines = content.split("\n").filter(Boolean); |
| 3cbdca6 | | | 149 | const total = lines.length; |
| 3cbdca6 | | | 150 | const entries = lines |
| 3cbdca6 | | | 151 | .slice(Math.max(0, total - limit)) |
| 3cbdca6 | | | 152 | .map((line) => parseLogLine(line)) |
| 3cbdca6 | | | 153 | .reverse(); |
| 3cbdca6 | | | 154 | return { entries, total }; |
| 3cbdca6 | | | 155 | } |
| 3cbdca6 | | | 156 | |
| 3cbdca6 | | | 157 | function sortInstancesDescByNewest(a: RingInstanceSummary, b: RingInstanceSummary): number { |
| 3cbdca6 | | | 158 | const aTs = a.last_ts ? new Date(a.last_ts).getTime() : 0; |
| 3cbdca6 | | | 159 | const bTs = b.last_ts ? new Date(b.last_ts).getTime() : 0; |
| 3cbdca6 | | | 160 | if (aTs !== bTs) return bTs - aTs; |
| 3cbdca6 | | | 161 | return b.total - a.total; |
| 3cbdca6 | | | 162 | } |
| 3cbdca6 | | | 163 | |
| 3cbdca6 | | | 164 | export async function ringRoutes(app: FastifyInstance) { |
| 3cbdca6 | | | 165 | // Allow plain text ingestion as well as JSON. |
| 3cbdca6 | | | 166 | app.addContentTypeParser( |
| 3cbdca6 | | | 167 | "text/plain", |
| 3cbdca6 | | | 168 | { parseAs: "string" }, |
| 3cbdca6 | | | 169 | (_request, body, done) => done(null, body) |
| 3cbdca6 | | | 170 | ); |
| 3cbdca6 | | | 171 | |
| 3cbdca6 | | | 172 | const ringDataDir = resolve(process.env.RING_DATA_DIR ?? "./data/ring"); |
| 3cbdca6 | | | 173 | const globalLogPath = resolve( |
| 3cbdca6 | | | 174 | process.env.RING_LOG_PATH ?? `${ringDataDir}/logs.ndjson` |
| 3cbdca6 | | | 175 | ); |
| 3cbdca6 | | | 176 | const repoLogsRoot = resolve( |
| 3cbdca6 | | | 177 | process.env.RING_REPO_LOG_ROOT ?? `${ringDataDir}/repos` |
| 3cbdca6 | | | 178 | ); |
| 3cbdca6 | | | 179 | |
| 3cbdca6 | | | 180 | function repoLogPath(owner: string, repo: string): string { |
| 3cbdca6 | | | 181 | return resolve(repoLogsRoot, owner, `${repo}.ndjson`); |
| 3cbdca6 | | | 182 | } |
| 3cbdca6 | | | 183 | |
| 3cbdca6 | | | 184 | // Backward-compatible global ingest. |
| 3cbdca6 | | | 185 | app.post("/ring/logs", async (request, reply) => { |
| 3cbdca6 | | | 186 | const source = |
| 3cbdca6 | | | 187 | coerceHeaderString(request.headers["x-ring-source"]) ?? "ingest"; |
| 3cbdca6 | | | 188 | const entry = normalizePayload(request.body, source); |
| 3cbdca6 | | | 189 | const serialized = `${JSON.stringify(entry)}\n`; |
| 3cbdca6 | | | 190 | |
| 3cbdca6 | | | 191 | try { |
| 3cbdca6 | | | 192 | await mkdir(dirname(globalLogPath), { recursive: true }); |
| 3cbdca6 | | | 193 | await appendFile(globalLogPath, serialized, "utf8"); |
| 3cbdca6 | | | 194 | return { ok: true, entry }; |
| 3cbdca6 | | | 195 | } catch (error) { |
| 3cbdca6 | | | 196 | request.log.error({ error }, "Failed to append Ring log"); |
| 3cbdca6 | | | 197 | return reply.code(500).send({ error: "Failed to write log entry" }); |
| 3cbdca6 | | | 198 | } |
| 3cbdca6 | | | 199 | }); |
| 3cbdca6 | | | 200 | |
| 3cbdca6 | | | 201 | // Backward-compatible global log view. |
| 3cbdca6 | | | 202 | app.get<{ Querystring: { limit?: string } }>("/ring/logs", async (request) => { |
| 3cbdca6 | | | 203 | const parsed = querySchema.safeParse(request.query); |
| 3cbdca6 | | | 204 | const limit = parsed.success ? parsed.data.limit : 200; |
| 3cbdca6 | | | 205 | return await readEntries(globalLogPath, limit); |
| 3cbdca6 | | | 206 | }); |
| 3cbdca6 | | | 207 | |
| 3cbdca6 | | | 208 | // Repo-scoped ingest. |
| 3cbdca6 | | | 209 | app.post<{ Params: { owner: string; repo: string } }>( |
| 3cbdca6 | | | 210 | "/repos/:owner/:repo/ring/logs", |
| 3cbdca6 | | | 211 | async (request, reply) => { |
| 3cbdca6 | | | 212 | const paramsParsed = repoParamsSchema.safeParse(request.params); |
| 3cbdca6 | | | 213 | if (!paramsParsed.success) { |
| 3cbdca6 | | | 214 | return reply.code(400).send({ error: "Invalid owner/repo" }); |
| 3cbdca6 | | | 215 | } |
| 3cbdca6 | | | 216 | const { owner, repo } = paramsParsed.data; |
| 3cbdca6 | | | 217 | const source = |
| 3cbdca6 | | | 218 | coerceHeaderString(request.headers["x-ring-source"]) ?? `${owner}/${repo}`; |
| 3cbdca6 | | | 219 | const baseEntry = normalizePayload(request.body, source); |
| 3cbdca6 | | | 220 | const entry: RingLogEntry = { ...baseEntry, owner, repo }; |
| 3cbdca6 | | | 221 | const serialized = `${JSON.stringify(entry)}\n`; |
| 3cbdca6 | | | 222 | |
| 3cbdca6 | | | 223 | try { |
| 3cbdca6 | | | 224 | const path = repoLogPath(owner, repo); |
| 3cbdca6 | | | 225 | await mkdir(dirname(path), { recursive: true }); |
| 3cbdca6 | | | 226 | await appendFile(path, serialized, "utf8"); |
| 3cbdca6 | | | 227 | return { ok: true, entry }; |
| 3cbdca6 | | | 228 | } catch (error) { |
| 3cbdca6 | | | 229 | request.log.error({ error }, "Failed to append repo Ring log"); |
| 3cbdca6 | | | 230 | return reply.code(500).send({ error: "Failed to write log entry" }); |
| 3cbdca6 | | | 231 | } |
| 3cbdca6 | | | 232 | } |
| 3cbdca6 | | | 233 | ); |
| 3cbdca6 | | | 234 | |
| 3cbdca6 | | | 235 | // Repo-scoped log view. |
| 3cbdca6 | | | 236 | app.get<{ |
| 3cbdca6 | | | 237 | Params: { owner: string; repo: string }; |
| 3cbdca6 | | | 238 | Querystring: { limit?: string }; |
| 3cbdca6 | | | 239 | }>("/repos/:owner/:repo/ring/logs", async (request, reply) => { |
| 3cbdca6 | | | 240 | const paramsParsed = repoParamsSchema.safeParse(request.params); |
| 3cbdca6 | | | 241 | if (!paramsParsed.success) { |
| 3cbdca6 | | | 242 | return reply.code(400).send({ error: "Invalid owner/repo" }); |
| 3cbdca6 | | | 243 | } |
| 3cbdca6 | | | 244 | const { owner, repo } = paramsParsed.data; |
| 3cbdca6 | | | 245 | const queryParsed = querySchema.safeParse(request.query); |
| 3cbdca6 | | | 246 | const limit = queryParsed.success ? queryParsed.data.limit : 200; |
| 3cbdca6 | | | 247 | |
| 3cbdca6 | | | 248 | return await readEntries(repoLogPath(owner, repo), limit); |
| 3cbdca6 | | | 249 | }); |
| 3cbdca6 | | | 250 | |
| 3cbdca6 | | | 251 | // List all repo Ring instances that have been set up (log file exists). |
| 3cbdca6 | | | 252 | app.get("/ring/instances", async () => { |
| 3cbdca6 | | | 253 | const instances: RingInstanceSummary[] = []; |
| 3cbdca6 | | | 254 | let ownerDirs: Dirent[] = []; |
| 3cbdca6 | | | 255 | try { |
| 3cbdca6 | | | 256 | ownerDirs = await readdir(repoLogsRoot, { withFileTypes: true }); |
| 3cbdca6 | | | 257 | } catch { |
| 3cbdca6 | | | 258 | return { instances }; |
| 3cbdca6 | | | 259 | } |
| 3cbdca6 | | | 260 | |
| 3cbdca6 | | | 261 | for (const ownerDir of ownerDirs) { |
| 3cbdca6 | | | 262 | if (!ownerDir.isDirectory()) continue; |
| 3cbdca6 | | | 263 | const owner = ownerDir.name; |
| 3cbdca6 | | | 264 | const ownerPath = resolve(repoLogsRoot, owner); |
| 3cbdca6 | | | 265 | let files: Dirent[] = []; |
| 3cbdca6 | | | 266 | try { |
| 3cbdca6 | | | 267 | files = await readdir(ownerPath, { withFileTypes: true }); |
| 3cbdca6 | | | 268 | } catch { |
| 3cbdca6 | | | 269 | continue; |
| 3cbdca6 | | | 270 | } |
| 3cbdca6 | | | 271 | for (const file of files) { |
| 3cbdca6 | | | 272 | if (!file.isFile() || !file.name.endsWith(".ndjson")) continue; |
| 3cbdca6 | | | 273 | const repo = file.name.slice(0, -7); |
| 3cbdca6 | | | 274 | const path = resolve(ownerPath, file.name); |
| 3cbdca6 | | | 275 | const content = await readLogFile(path); |
| 3cbdca6 | | | 276 | const lines = content.split("\n").filter(Boolean); |
| 3cbdca6 | | | 277 | const total = lines.length; |
| 3cbdca6 | | | 278 | const last = total > 0 ? parseLogLine(lines[total - 1]) : null; |
| 3cbdca6 | | | 279 | instances.push({ |
| 3cbdca6 | | | 280 | owner, |
| 3cbdca6 | | | 281 | repo, |
| 3cbdca6 | | | 282 | total, |
| 3cbdca6 | | | 283 | last_ts: last?.ts ?? null, |
| 3cbdca6 | | | 284 | last_level: last?.level ?? null, |
| 3cbdca6 | | | 285 | last_message: last?.message ?? null, |
| 3cbdca6 | | | 286 | }); |
| 3cbdca6 | | | 287 | } |
| 3cbdca6 | | | 288 | } |
| 3cbdca6 | | | 289 | |
| 3cbdca6 | | | 290 | instances.sort(sortInstancesDescByNewest); |
| 3cbdca6 | | | 291 | return { instances }; |
| 3cbdca6 | | | 292 | }); |
| 3cbdca6 | | | 293 | } |