Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

262 lines · 12.0 KBRaw History
1import { mkdtempSync } from "node:fs";
2import { createServer, type IncomingHttpHeaders, type Server, type ServerResponse } from "node:http";
3import type { AddressInfo } from "node:net";
4import { tmpdir } from "node:os";
5import path from "node:path";
6import type { JobCompletion, JobSpec, LogLine, StepReport, StepSpec } from "@diffinery/shared";
7
8export interface FakeApi {
9 url: string;
10 token: string;
11 logs: LogLine[];
12 steps: Array<{ index: number; report: StepReport }>;
13 completion: JobCompletion | null;
14 /** Runtime token the fake results service accepts (run "1", job "job-1"). */
15 runtimeToken: string;
16 /** Cache entries by key (version in `cacheVersions`). */
17 cache: Map<string, Buffer>;
18 cacheVersions: Map<string, string>;
19 /** The Range header of each blob GET ("" for none). */
20 blobGets: string[];
21 /** How many ranged blob GETs past the first range still answer as if the blob had changed (another ETag). */
22 blobChanges: number;
23 /** Finalized artifacts of the run, by name. */
24 artifacts: Map<string, { id: number; body: Buffer; mimeType: string; hash?: string }>;
25 requests: Array<{ method: string; path: string; auth: string | undefined; runner?: string }>;
26 /** Fail the next N log POSTs with 503 (to test retries). */
27 failLogs: number;
28 specs: Map<string, JobSpec>;
29 /** Diffinery's copies of actions, by `owner/repo@ref`, and the actions the runner reported downloading from GitHub. */
30 mirror: Map<string, Buffer>;
31 mirrorReports: string[];
32 close(): Promise<void>;
33 logText(step?: number): string;
34 lastReport(index: number): StepReport | undefined;
35}
36
37export async function startFakeApi(token = "job-token-123"): Promise<FakeApi> {
38 const api: FakeApi = {
39 url: "",
40 token,
41 logs: [],
42 steps: [],
43 completion: null,
44 runtimeToken: fakeRuntimeToken(),
45 cache: new Map(),
46 cacheVersions: new Map(),
47 blobGets: [],
48 blobChanges: 0,
49 artifacts: new Map(),
50 requests: [],
51 failLogs: 0,
52 specs: new Map(),
53 mirror: new Map(),
54 mirrorReports: [],
55 close: async () => {},
56 logText(step?: number) {
57 return this.logs
58 .filter((l) => step === undefined || l.step === step)
59 .map((l) => l.text)
60 .join("\n");
61 },
62 lastReport(index: number) {
63 return [...this.steps].reverse().find((s) => s.index === index)?.report;
64 },
65 };
66 // A minimal Actions results service: the Twirp cache/artifact services and their blob URLs.
67 const pending = new Map<string, { kind: "cache" | "artifact"; name: string; version?: string; mimeType?: string; body?: Buffer }>();
68 let nextId = 1;
69 const runtime = (method: string, u: URL, headers: IncomingHttpHeaders, body: Buffer, res: ServerResponse, json: (s: number, v: unknown) => void) => {
70 const blob = /^\/_runtime\/blob\/([^/]+)$/.exec(u.pathname);
71 if (blob) {
72 const id = decodeURIComponent(blob[1]!);
73 if (method === "PUT") {
74 const p = pending.get(id);
75 if (!p || headers["x-ms-blob-type"] !== "BlockBlob") return json(400, {});
76 p.body = body;
77 res.writeHead(201);
78 return res.end();
79 }
80 const art = [...api.artifacts.values()].find((a) => String(a.id) === id);
81 const b = art?.body ?? api.cache.get(id);
82 if (!b) return json(404, {});
83 api.blobGets.push(String(headers.range ?? ""));
84 // Ranged reads, like Diffinery's blob store.
85 const range = /^bytes=(\d+)-(\d+)$/.exec(String(headers.range ?? ""));
86 const type = { "content-type": art?.mimeType ?? "application/octet-stream", etag: '"e1"', "accept-ranges": "bytes" };
87 if (range && b.length === 0) {
88 res.writeHead(416, { "content-range": "bytes */0" });
89 return res.end();
90 }
91 if (range) {
92 const start = Number(range[1]);
93 const end = Math.min(Number(range[2]), b.length - 1);
94 if (start > 0 && api.blobChanges > 0) {
95 api.blobChanges--;
96 type.etag = '"e2"';
97 }
98 res.writeHead(206, { ...type, "content-range": `bytes ${start}-${end}/${b.length}` });
99 return res.end(b.subarray(start, end + 1));
100 }
101 res.writeHead(200, type);
102 return res.end(b);
103 }
104 if (headers.authorization !== `Bearer ${api.runtimeToken}`) return json(401, { code: "unauthenticated", msg: "bad token" });
105 const req = JSON.parse(body.toString() || "{}") as Record<string, string | string[] | undefined>;
106 const blobUrl = (id: string) => `${api.url}/_runtime/blob/${encodeURIComponent(id)}?sig=x`;
107 switch (u.pathname.replace("/twirp/github.actions.results.api.v1.", "")) {
108 case "CacheService/GetCacheEntryDownloadURL": {
109 const keys = [...api.cache.keys()].filter((k) => api.cacheVersions.get(k) === req.version);
110 const hit = keys.find((k) => k === req.key) ?? ((req.restore_keys ?? []) as string[]).map((p) => keys.find((k) => k.startsWith(p))).find(Boolean);
111 return json(200, hit ? { ok: true, matched_key: hit, signed_download_url: blobUrl(hit) } : { ok: false });
112 }
113 case "CacheService/CreateCacheEntry": {
114 if (api.cache.has(String(req.key))) return json(409, { code: "already_exists", msg: "cache entry already exists" });
115 const id = `upload-${nextId++}`;
116 pending.set(id, { kind: "cache", name: String(req.key), version: String(req.version) });
117 return json(200, { ok: true, signed_upload_url: blobUrl(id) });
118 }
119 case "CacheService/FinalizeCacheEntryUpload": {
120 const [id, p] = [...pending].find(([, p]) => p.kind === "cache" && p.name === req.key) ?? [];
121 if (!id || !p?.body) return json(404, { code: "not_found", msg: "no upload" });
122 pending.delete(id);
123 api.cache.set(p.name, p.body);
124 api.cacheVersions.set(p.name, p.version!);
125 return json(200, { ok: true, entry_id: String(nextId++) });
126 }
127 case "ArtifactService/CreateArtifact": {
128 if (api.artifacts.has(String(req.name))) return json(409, { code: "already_exists", msg: "artifact exists" });
129 const id = `upload-${nextId++}`;
130 pending.set(id, { kind: "artifact", name: String(req.name), mimeType: String(req.mime_type ?? "application/zip") });
131 return json(200, { ok: true, signed_upload_url: blobUrl(id) });
132 }
133 case "ArtifactService/FinalizeArtifact": {
134 const [id, p] = [...pending].find(([, p]) => p.kind === "artifact" && p.name === req.name) ?? [];
135 if (!id || !p?.body) return json(404, { code: "not_found", msg: "no upload" });
136 pending.delete(id);
137 const artifactId = nextId++;
138 api.artifacts.set(p.name, { id: artifactId, body: p.body, mimeType: p.mimeType!, hash: req.hash as string | undefined });
139 return json(200, { ok: true, artifact_id: String(artifactId) });
140 }
141 case "ArtifactService/ListArtifacts":
142 return json(200, {
143 artifacts: [...api.artifacts]
144 .filter(([n]) => !req.name_filter || n === req.name_filter)
145 .map(([name, a]) => ({
146 workflow_run_backend_id: "1",
147 workflow_job_run_backend_id: "job-1",
148 database_id: String(a.id),
149 name,
150 size: String(a.body.length),
151 digest: a.hash?.replace(/^sha256:/, ""),
152 created_at: new Date().toISOString(),
153 })),
154 });
155 case "ArtifactService/GetSignedArtifactURL": {
156 const a = api.artifacts.get(String(req.name));
157 return a ? json(200, { signed_url: blobUrl(String(a.id)) }) : json(404, { code: "not_found", msg: "no artifact" });
158 }
159 case "ArtifactService/DeleteArtifact": {
160 const a = api.artifacts.get(String(req.name));
161 if (!a) return json(404, { code: "not_found", msg: "no artifact" });
162 api.artifacts.delete(String(req.name));
163 return json(200, { ok: true, artifact_id: String(a.id) });
164 }
165 }
166 return json(404, { code: "bad_route", msg: "no such method" });
167 };
168 const server: Server = createServer(async (req, res) => {
169 const chunks: Buffer[] = [];
170 for await (const c of req) chunks.push(c as Buffer);
171 const body = Buffer.concat(chunks);
172 const u = new URL(req.url ?? "/", "http://x");
173 api.requests.push({ method: req.method!, path: u.pathname + u.search, auth: req.headers.authorization, runner: req.headers["x-diffinery-runner"] as string | undefined });
174 const json = (status: number, v: unknown) => {
175 res.writeHead(status, { "content-type": "application/json" });
176 res.end(JSON.stringify(v));
177 };
178 if (u.pathname.startsWith("/twirp/") || u.pathname.startsWith("/_runtime/blob/")) return runtime(req.method!, u, req.headers, body, res, json);
179 if (req.headers.authorization !== `Bearer ${token}`) return json(401, { error: "unauthorized" });
180 const m = /^\/runner\/jobs\/([^/]+)(\/.*)?$/.exec(u.pathname);
181 if (!m) return json(404, {});
182 const sub = m[2] ?? "";
183 if (sub === "" && req.method === "GET") {
184 const s = api.specs.get(decodeURIComponent(m[1]!));
185 return s ? json(200, s) : json(404, {});
186 }
187 if (sub === "/logs" && req.method === "POST") {
188 if (api.failLogs > 0) {
189 api.failLogs--;
190 return json(503, { error: "try later" });
191 }
192 api.logs.push(...(JSON.parse(body.toString()) as { lines: LogLine[] }).lines);
193 return json(200, {});
194 }
195 const sm = /^\/steps\/(\d+)$/.exec(sub);
196 if (sm && req.method === "POST") {
197 api.steps.push({ index: Number(sm[1]), report: JSON.parse(body.toString()) as StepReport });
198 return json(200, {});
199 }
200 if (sub === "/complete" && req.method === "POST") {
201 api.completion = JSON.parse(body.toString()) as JobCompletion;
202 return json(200, {});
203 }
204 if (sub === "/action-mirror" && req.method === "POST") {
205 const a = JSON.parse(body.toString()) as { owner: string; repo: string; ref: string };
206 api.mirrorReports.push(`${a.owner}/${a.repo}@${a.ref}`);
207 return json(200, { result: "stored" });
208 }
209 if (sub === "/action-mirror" && req.method === "GET") {
210 const copy = api.mirror.get(`${u.searchParams.get("owner")}/${u.searchParams.get("repo")}@${u.searchParams.get("ref")}`);
211 if (!copy) return json(404, { message: "no copy of this action" });
212 res.writeHead(200, { "content-type": "application/gzip", "x-diffinery-fetched-at": "2026-10-25T12:00:00.000Z" });
213 return res.end(copy);
214 }
215 if (sub === "/oidc") return json(200, { value: `oidc-for-${u.searchParams.get("audience")}` });
216 return json(404, {});
217 });
218 await new Promise<void>((r) => server.listen(0, "127.0.0.1", () => r()));
219 api.url = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
220 api.close = () => new Promise<void>((r) => server.close(() => r()));
221 return api;
222}
223
224/** A runtime token shaped like the server's: its scope names run "1" and job "job-1". */
225function fakeRuntimeToken(): string {
226 const b64 = (v: unknown) => Buffer.from(JSON.stringify(v)).toString("base64url");
227 return `${b64({ alg: "HS256", typ: "JWT" })}.${b64({ scp: "Actions.GenericRead:1 Actions.Results:1:job-1", exp: Math.floor(Date.now() / 1000) + 3600 })}.sig`;
228}
229
230export function tmp(prefix = "runner-test-"): string {
231 return mkdtempSync(path.join(tmpdir(), prefix));
232}
233
234export function makeSpec(api: FakeApi, steps: StepSpec[], over: Partial<JobSpec> = {}): JobSpec {
235 return {
236 runId: "run-1",
237 jobId: "job-1",
238 jobKey: "build",
239 name: "build",
240 repo: { owner: "acme", name: "widgets", cloneUrl: "http://invalid.test/acme/widgets.git", defaultBranch: "main" },
241 sha: "0123456789abcdef0123456789abcdef01234567",
242 ref: "refs/heads/main",
243 workflowPath: ".github/workflows/ci.yml",
244 github: { event_name: "push", actor: "octo", event: { ref: "refs/heads/main", pusher: { name: "octo" } } },
245 env: {},
246 matrix: {},
247 strategy: { "fail-fast": true, "job-index": 0, "job-total": 1 },
248 needs: {},
249 inputs: {},
250 vars: {},
251 secrets: {},
252 outputs: {},
253 steps,
254 timeoutMinutes: 10,
255 continueOnError: false,
256 idToken: false,
257 apiUrl: api.url,
258 token: api.token,
259 runtime: { token: api.runtimeToken, resultsUrl: `${api.url}/`, cacheUrl: `${api.url}/_runtime/`, runtimeUrl: `${api.url}/_runtime/` },
260 ...over,
261 };
262}