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.

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