Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102// spindle websocket log stream// replays from line one on connect and closes when all workflows settle
import { decodeFirst } from "@atcute/cbor";import { spindleUrl } from "#lib/api/spindle.ts";import type * as SubscribeLogs from "#lib/api/lexicons/types/org/tangled/ci/subscribePipelineLogs.ts";
export type LogControl = SubscribeLogs.Control;
export type LogData = SubscribeLogs.Data;
export type LogFrame = | { kind: "control"; control: LogControl } | { kind: "data"; data: LogData } | { kind: "error"; error: string; message: string };
// indigo event header opsconst OP_MESSAGE = 1;const OP_ERROR = -1;
const asRecord = (value: unknown): Record<string, unknown> => typeof value === "object" && value !== null ? (value as Record<string, unknown>) : {};
const str = (value: unknown): string => (typeof value === "string" ? value : "");
// decodes binary websocket frame into dag-cbor header and bodyexport const decodeLogFrame = (bytes: Uint8Array): LogFrame | null => { const [rawHeader, rest] = decodeFirst(bytes); const header = asRecord(rawHeader); if (header.op !== OP_MESSAGE && header.op !== OP_ERROR) return null;
const [rawBody] = decodeFirst(rest); const body = asRecord(rawBody);
if (header.op === OP_ERROR) { return { kind: "error", error: str(body.error), message: str(body.message) }; }
switch (header.t) { case "#control": return { kind: "control", control: body as unknown as LogControl }; case "#data": return { kind: "data", data: body as unknown as LogData }; default: return null; }};
export const pipelineLogsUrl = ( host: string, repo: string, pipeline: string, workflows: string[]): string => { const url = new URL( "/xrpc/org.tangled.ci.subscribePipelineLogs", `${spindleUrl(host).replace(/^http/, "ws")}/` ); url.searchParams.set("repo", repo); url.searchParams.set("pipeline", pipeline); for (const workflow of workflows) url.searchParams.append("workflows", workflow); return url.toString();};
export interface LogStreamHandlers { onFrame: (frame: LogFrame) => void; onClose?: (clean: boolean) => void;}
// opens log subscription, returns teardown functionexport const subscribePipelineLogs = ( host: string, repo: string, pipeline: string, workflows: string[], handlers: LogStreamHandlers): (() => void) => { const socket = new WebSocket(pipelineLogsUrl(host, repo, pipeline, workflows)); socket.binaryType = "arraybuffer"; let live = true;
socket.onmessage = (event: MessageEvent<ArrayBuffer | string>) => { if (!live || typeof event.data === "string") return; const frame = decodeLogFrame(new Uint8Array(event.data)); if (frame) handlers.onFrame(frame); };
socket.onclose = (event: CloseEvent) => { if (!live) return; live = false; handlers.onClose?.(event.wasClean); };
// refused upgrades trigger an error then close, close handler handles it socket.onerror = () => {};
return () => { live = false; socket.close(); };};