| 1 | import { ZERO_OID, diffTrees, fromUtf8, isTreeMode, readCommit, readTree, resolvePath } from "@diffinery/git-core"; |
| 2 | import type { Actor, Conclusion, Json, JobCompletion, JobSpec, LogLine, NeedsResult, StepReport } from "@diffinery/shared"; |
| 3 | import { |
| 4 | evaluateIf, |
| 5 | evaluateValue, |
| 6 | expandMatrix, |
| 7 | interpolate, |
| 8 | isWarning, |
| 9 | jobOrder, |
| 10 | matchesTrigger, |
| 11 | parseWorkflow, |
| 12 | type ExprContext, |
| 13 | type Job, |
| 14 | type StatusFns, |
| 15 | type TriggerEvent, |
| 16 | type Workflow, |
| 17 | } from "@diffinery/workflow-yaml"; |
| 18 | import { decryptSecret, encryptSecret } from "../lib/crypto.js"; |
| 19 | import { HttpError, newId, now, randomToken, sha256 } from "../lib/util.js"; |
| 20 | import type { Env } from "../env.js"; |
| 21 | import type { RepoDO } from "./RepoDO.js"; |
| 22 | import { atLeast, type Caller } from "./types.js"; |
| 23 | |
| 24 | const WORKFLOW_DIRS = [".github/workflows", ".diffinery/workflows"]; |
| 25 | const SUPPORTED_LABEL = /^(ubuntu-.*|linux|diffinery(-.*)?|self-hosted|x64|amd64)$/i; |
| 26 | |
| 27 | export type RunRow = { |
| 28 | id: string; |
| 29 | run_number: number; |
| 30 | workflow_path: string; |
| 31 | workflow_name: string; |
| 32 | event: string; |
| 33 | ref: string; |
| 34 | sha: string; |
| 35 | actor: string; |
| 36 | status: "queued" | "in_progress" | "completed" | "waiting"; |
| 37 | conclusion: Conclusion | "timed_out" | null; |
| 38 | created_at: string; |
| 39 | updated_at: string; |
| 40 | pr_number: number | null; |
| 41 | attempt: number; |
| 42 | event_payload: string; |
| 43 | inputs: string; |
| 44 | concurrency_group: string | null; |
| 45 | display_title: string; |
| 46 | }; |
| 47 | |
| 48 | export type JobRow = { |
| 49 | id: string; |
| 50 | run_id: string; |
| 51 | job_key: string; |
| 52 | name: string; |
| 53 | status: "queued" | "waiting" | "in_progress" | "completed"; |
| 54 | conclusion: Conclusion | "timed_out" | null; |
| 55 | matrix: string; |
| 56 | strategy: string; |
| 57 | started_at: string | null; |
| 58 | completed_at: string | null; |
| 59 | outputs: string; |
| 60 | steps: string; |
| 61 | failure: string | null; |
| 62 | token_hash: string | null; |
| 63 | summary: string | null; |
| 64 | container: string | null; |
| 65 | timeout_minutes: number; |
| 66 | /** runs-on labels for self-hosted jobs (JSON array); null for Diffinery-hosted containers. */ |
| 67 | labels: string | null; |
| 68 | }; |
| 69 | |
| 70 | export type RunInfo = { |
| 71 | id: string; |
| 72 | number: number; |
| 73 | attempt: number; |
| 74 | workflowPath: string; |
| 75 | workflowName: string; |
| 76 | title: string; |
| 77 | event: string; |
| 78 | ref: string; |
| 79 | sha: string; |
| 80 | actor: Actor; |
| 81 | status: RunRow["status"]; |
| 82 | conclusion: RunRow["conclusion"]; |
| 83 | createdAt: string; |
| 84 | updatedAt: string; |
| 85 | prNumber: number | null; |
| 86 | }; |
| 87 | |
| 88 | export type JobInfo = { |
| 89 | id: string; |
| 90 | runId: string; |
| 91 | key: string; |
| 92 | name: string; |
| 93 | status: JobRow["status"]; |
| 94 | conclusion: JobRow["conclusion"]; |
| 95 | matrix: Record<string, Json>; |
| 96 | startedAt: string | null; |
| 97 | completedAt: string | null; |
| 98 | outputs: Record<string, string>; |
| 99 | steps: Array<StepReport & { index: number }>; |
| 100 | failure: JobCompletion["failure"] | null; |
| 101 | summary: string | null; |
| 102 | }; |
| 103 | |
| 104 | const toRun = (r: RunRow): RunInfo => ({ |
| 105 | id: r.id, |
| 106 | number: r.run_number, |
| 107 | attempt: r.attempt, |
| 108 | workflowPath: r.workflow_path, |
| 109 | workflowName: r.workflow_name, |
| 110 | title: r.display_title, |
| 111 | event: r.event, |
| 112 | ref: r.ref, |
| 113 | sha: r.sha, |
| 114 | actor: JSON.parse(r.actor), |
| 115 | status: r.status, |
| 116 | conclusion: r.conclusion, |
| 117 | createdAt: r.created_at, |
| 118 | updatedAt: r.updated_at, |
| 119 | prNumber: r.pr_number, |
| 120 | }); |
| 121 | |
| 122 | const toJob = (j: JobRow): JobInfo => ({ |
| 123 | id: j.id, |
| 124 | runId: j.run_id, |
| 125 | key: j.job_key, |
| 126 | name: j.name, |
| 127 | status: j.status, |
| 128 | conclusion: j.conclusion, |
| 129 | matrix: JSON.parse(j.matrix), |
| 130 | startedAt: j.started_at, |
| 131 | completedAt: j.completed_at, |
| 132 | outputs: JSON.parse(j.outputs), |
| 133 | steps: JSON.parse(j.steps), |
| 134 | failure: j.failure ? JSON.parse(j.failure) : null, |
| 135 | summary: j.summary, |
| 136 | }); |
| 137 | |
| 138 | const statusFor = (needs: NeedsResult[], cancelled: boolean): StatusFns => ({ |
| 139 | success: () => !cancelled && needs.every((n) => n.result === "success" || n.result === "skipped"), |
| 140 | failure: () => needs.some((n) => n.result === "failure"), |
| 141 | cancelled: () => cancelled, |
| 142 | always: () => true, |
| 143 | }); |
| 144 | |
| 145 | /** |
| 146 | * GitHub Actions compatible run orchestration state. The Cloudflare Workflow |
| 147 | * (`RunWorkflow`) drives a run through these methods; the runner agent inside |
| 148 | * each job's container reports back through the runner* methods. |
| 149 | */ |
| 150 | export class Actions { |
| 151 | constructor( |
| 152 | private repo: RepoDO, |
| 153 | private env: Env, |
| 154 | ) {} |
| 155 | |
| 156 | private get sql() { |
| 157 | return this.repo.sql; |
| 158 | } |
| 159 | |
| 160 | // ---------------------------------------------------------------- workflow discovery |
| 161 | |
| 162 | async loadWorkflows(sha: string): Promise<Array<{ path: string; workflow?: Workflow; errors: string[] }>> { |
| 163 | const commit = await readCommit(this.repo.store, sha); |
| 164 | const out: Array<{ path: string; workflow?: Workflow; errors: string[] }> = []; |
| 165 | for (const dir of WORKFLOW_DIRS) { |
| 166 | const entry = await resolvePath(this.repo.store, commit.tree, dir); |
| 167 | if (!entry || !isTreeMode(entry.mode)) continue; |
| 168 | for (const f of await readTree(this.repo.store, entry.oid)) { |
| 169 | if (!/\.ya?ml$/.test(f.name) || isTreeMode(f.mode)) continue; |
| 170 | const path = `${dir}/${f.name}`; |
| 171 | const o = await this.repo.store.read(f.oid); |
| 172 | if (!o) continue; |
| 173 | const { workflow, errors } = parseWorkflow(fromUtf8(o.content), path); |
| 174 | out.push({ path, workflow, errors: errors.filter((e) => !isWarning(e)).map((e) => `${e.path}${e.line ? `:${e.line}` : ""}: ${e.message}`) }); |
| 175 | } |
| 176 | } |
| 177 | return out; |
| 178 | } |
| 179 | |
| 180 | async listWorkflows(): Promise<Array<{ path: string; name: string; errors: string[]; triggers: string[] }>> { |
| 181 | const sha = await this.repo.resolve("HEAD"); |
| 182 | if (!sha) return []; |
| 183 | return (await this.loadWorkflows(sha)).map((w) => ({ path: w.path, name: w.workflow?.name ?? w.path.split("/").pop()!, errors: w.errors, triggers: w.workflow ? Object.keys(w.workflow.on) : [] })); |
| 184 | } |
| 185 | |
| 186 | private repositoryPayload() { |
| 187 | const m = this.repo.meta; |
| 188 | return { id: m.repoId, name: m.name, full_name: `${m.owner}/${m.name}`, owner: { login: m.owner }, default_branch: m.defaultBranch, private: true, html_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}`, clone_url: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git` }; |
| 189 | } |
| 190 | |
| 191 | // ---------------------------------------------------------------- triggers |
| 192 | |
| 193 | async onPush(ev: { ref: string; before: string; after: string; forced: boolean; created: boolean; deleted: boolean }, caller: Caller): Promise<void> { |
| 194 | if (ev.deleted) return; |
| 195 | const workflows = await this.loadWorkflows(ev.after); |
| 196 | let changedFiles: string[] | undefined; |
| 197 | if (ev.before !== ZERO_OID && this.repo.store.has(ev.before)) { |
| 198 | const [a, b] = await Promise.all([readCommit(this.repo.store, ev.before), readCommit(this.repo.store, ev.after)]); |
| 199 | changedFiles = (await diffTrees(this.repo.store, a.tree, b.tree)).map((c) => c.path); |
| 200 | } |
| 201 | const head = await readCommit(this.repo.store, ev.after); |
| 202 | const payload = { |
| 203 | ref: ev.ref, |
| 204 | before: ev.before, |
| 205 | after: ev.after, |
| 206 | created: ev.created, |
| 207 | deleted: ev.deleted, |
| 208 | forced: ev.forced, |
| 209 | repository: this.repositoryPayload(), |
| 210 | pusher: { name: caller.login }, |
| 211 | sender: { login: caller.login, type: caller.kind === "agent" ? "Bot" : "User" }, |
| 212 | head_commit: { id: ev.after, message: head.message, timestamp: new Date(head.committer.timestamp * 1000).toISOString(), author: { name: head.author.name, email: head.author.email } }, |
| 213 | commits: [{ id: ev.after, message: head.message }], |
| 214 | }; |
| 215 | if (ev.ref === `refs/heads/${this.repo.meta.defaultBranch}`) await this.syncSchedules(workflows); |
| 216 | for (const w of workflows) { |
| 217 | if (!w.workflow) { |
| 218 | if (w.errors.length) await this.createFailedRun(w.path, "push", ev.ref, ev.after, caller, w.errors); |
| 219 | continue; |
| 220 | } |
| 221 | if (!matchesTrigger(w.workflow, { name: "push", ref: ev.ref, changedFiles })) continue; |
| 222 | await this.createRun(w.workflow, { event: "push", ref: ev.ref, sha: ev.after, payload, caller, title: head.message.split("\n")[0] ?? "" }); |
| 223 | } |
| 224 | } |
| 225 | |
| 226 | async onPullRequest(action: string, number: number, caller: Caller): Promise<void> { |
| 227 | const { issue } = this.repo.collab.get(number); |
| 228 | const pull = issue.pull!; |
| 229 | const files = await this.repo.collab.pullFiles(number); |
| 230 | const workflows = await this.loadWorkflows(pull.headSha); |
| 231 | const baseSha = this.repo.getRef(`refs/heads/${pull.base}`); |
| 232 | const payload = { |
| 233 | action, |
| 234 | number, |
| 235 | pull_request: { |
| 236 | number, |
| 237 | title: issue.title, |
| 238 | body: issue.body, |
| 239 | draft: pull.draft, |
| 240 | user: { login: issue.author.login }, |
| 241 | head: { ref: pull.head, sha: pull.headSha, repo: this.repositoryPayload() }, |
| 242 | base: { ref: pull.base, sha: baseSha, repo: this.repositoryPayload() }, |
| 243 | html_url: `${this.env.PUBLIC_URL}/${this.repo.meta.owner}/${this.repo.meta.name}/pull/${number}`, |
| 244 | }, |
| 245 | repository: this.repositoryPayload(), |
| 246 | sender: { login: caller.login }, |
| 247 | }; |
| 248 | for (const w of workflows) { |
| 249 | if (!w.workflow) continue; |
| 250 | for (const name of ["pull_request", "pull_request_target"] as const) { |
| 251 | const ev: TriggerEvent = { name, action, baseRef: pull.base, changedFiles: files.files.map((f) => f.path) }; |
| 252 | if (!matchesTrigger(w.workflow, ev)) continue; |
| 253 | await this.createRun(w.workflow, { |
| 254 | event: name, |
| 255 | ref: `refs/pull/${number}/head`, |
| 256 | sha: pull.headSha, |
| 257 | payload, |
| 258 | caller, |
| 259 | prNumber: number, |
| 260 | headRef: pull.head, |
| 261 | baseRef: pull.base, |
| 262 | title: issue.title, |
| 263 | }); |
| 264 | } |
| 265 | } |
| 266 | } |
| 267 | |
| 268 | async dispatch(workflowFile: string, ref: string, inputs: Record<string, unknown>, caller: Caller): Promise<RunInfo> { |
| 269 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to run workflows"); |
| 270 | const sha = await this.repo.resolve(ref); |
| 271 | if (!sha) throw new HttpError(422, `unknown ref ${ref}`); |
| 272 | const all = await this.loadWorkflows(sha); |
| 273 | const w = all.find((x) => x.path === workflowFile || x.path.endsWith(`/${workflowFile}`)); |
| 274 | if (!w?.workflow) throw new HttpError(404, `workflow ${workflowFile} not found at ${ref}${w?.errors.length ? `: ${w.errors.join("; ")}` : ""}`); |
| 275 | if (!w.workflow.on.workflow_dispatch) throw new HttpError(422, "workflow does not have a workflow_dispatch trigger"); |
| 276 | const defs = w.workflow.on.workflow_dispatch.inputs ?? {}; |
| 277 | const resolved: Record<string, unknown> = {}; |
| 278 | for (const [k, d] of Object.entries(defs)) { |
| 279 | const v = inputs[k] ?? d.default; |
| 280 | if (d.required && (v === undefined || v === "")) throw new HttpError(422, `input '${k}' is required`); |
| 281 | if (d.type === "choice" && d.options && v !== undefined && !d.options.includes(String(v))) throw new HttpError(422, `input '${k}' must be one of ${d.options.join(", ")}`); |
| 282 | resolved[k] = d.type === "boolean" ? v === true || v === "true" : d.type === "number" ? Number(v) : v; |
| 283 | } |
| 284 | const fullRef = ref.startsWith("refs/") ? ref : this.repo.getRef(`refs/heads/${ref}`) ? `refs/heads/${ref}` : `refs/tags/${ref}`; |
| 285 | const payload = { inputs: resolved, ref: fullRef, repository: this.repositoryPayload(), sender: { login: caller.login }, workflow: w.path }; |
| 286 | return (await this.createRun(w.workflow, { event: "workflow_dispatch", ref: fullRef, sha, payload, caller, inputs: resolved, title: w.workflow.name ?? w.path }))!; |
| 287 | } |
| 288 | |
| 289 | async schedule(workflowPath: string, cron: string): Promise<void> { |
| 290 | const sha = await this.repo.resolve("HEAD"); |
| 291 | if (!sha) return; |
| 292 | const w = (await this.loadWorkflows(sha)).find((x) => x.path === workflowPath); |
| 293 | if (!w?.workflow || !matchesTrigger(w.workflow, { name: "schedule", cron })) return; |
| 294 | const system: Caller = { id: "system", login: "diffinery-scheduler", kind: "system", role: "write" }; |
| 295 | await this.createRun(w.workflow, { event: "schedule", ref: this.repo.headRef, sha, payload: { schedule: cron, repository: this.repositoryPayload() }, caller: system, title: w.workflow.name ?? w.path }); |
| 296 | } |
| 297 | |
| 298 | private async syncSchedules(workflows: Array<{ path: string; workflow?: Workflow }>): Promise<void> { |
| 299 | const repoId = this.repo.meta.repoId; |
| 300 | const stmts = [this.env.DB.prepare("DELETE FROM schedules WHERE repo_id = ?").bind(repoId)]; |
| 301 | for (const w of workflows) for (const s of w.workflow?.on.schedule ?? []) stmts.push(this.env.DB.prepare("INSERT OR IGNORE INTO schedules (repo_id, workflow_path, cron) VALUES (?, ?, ?)").bind(repoId, w.path, s.cron)); |
| 302 | await this.env.DB.batch(stmts); |
| 303 | } |
| 304 | |
| 305 | // ---------------------------------------------------------------- runs |
| 306 | |
| 307 | private nextRunNumber(): number { |
| 308 | this.sql.exec("INSERT INTO counters (name, value) VALUES ('run', 1) ON CONFLICT(name) DO UPDATE SET value = value + 1"); |
| 309 | return this.sql.exec<{ value: number }>("SELECT value FROM counters WHERE name = 'run'").one().value; |
| 310 | } |
| 311 | |
| 312 | private githubContext(run: RunRow, jobKey: string, token: string): Record<string, unknown> { |
| 313 | const m = this.repo.meta; |
| 314 | const payload = JSON.parse(run.event_payload) as Record<string, unknown>; |
| 315 | const actor = JSON.parse(run.actor) as Actor; |
| 316 | const refName = run.ref.replace(/^refs\/(heads|tags)\//, "").replace(/^refs\/pull\/(\d+)\/head$/, "$1/merge"); |
| 317 | const pr = payload.pull_request as { head?: { ref?: string }; base?: { ref?: string } } | undefined; |
| 318 | return { |
| 319 | event_name: run.event, |
| 320 | event: payload, |
| 321 | sha: run.sha, |
| 322 | ref: run.ref, |
| 323 | ref_name: refName, |
| 324 | ref_type: run.ref.startsWith("refs/tags/") ? "tag" : "branch", |
| 325 | ref_protected: false, |
| 326 | head_ref: pr?.head?.ref ?? "", |
| 327 | base_ref: pr?.base?.ref ?? "", |
| 328 | repository: `${m.owner}/${m.name}`, |
| 329 | repository_id: m.repoId, |
| 330 | repository_owner: m.owner, |
| 331 | repositoryUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, |
| 332 | actor: actor.login, |
| 333 | triggering_actor: actor.login, |
| 334 | run_id: run.id, |
| 335 | run_number: String(run.run_number), |
| 336 | run_attempt: String(run.attempt), |
| 337 | workflow: run.workflow_name, |
| 338 | workflow_ref: `${m.owner}/${m.name}/${run.workflow_path}@${run.ref}`, |
| 339 | job: jobKey, |
| 340 | server_url: this.env.PUBLIC_URL, |
| 341 | api_url: `${this.env.PUBLIC_URL}/api/v3`, |
| 342 | graphql_url: `${this.env.PUBLIC_URL}/api/graphql`, |
| 343 | token, |
| 344 | retention_days: "90", |
| 345 | secret_source: "Actions", |
| 346 | }; |
| 347 | } |
| 348 | |
| 349 | private async createFailedRun(path: string, event: string, ref: string, sha: string, caller: Caller, errors: string[]): Promise<void> { |
| 350 | const id = newId(); |
| 351 | const t = now(); |
| 352 | this.sql.exec( |
| 353 | "INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, conclusion, created_at, updated_at, event_payload, display_title) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'completed', 'failure', ?, ?, ?, ?)", |
| 354 | id, |
| 355 | this.nextRunNumber(), |
| 356 | path, |
| 357 | path, |
| 358 | event, |
| 359 | ref, |
| 360 | sha, |
| 361 | JSON.stringify({ id: caller.id, login: caller.login, kind: caller.kind }), |
| 362 | t, |
| 363 | t, |
| 364 | "{}", |
| 365 | `Invalid workflow file: ${errors[0] ?? path}`, |
| 366 | ); |
| 367 | const jobId = newId(); |
| 368 | this.sql.exec( |
| 369 | "INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, 'invalid', 'Invalid workflow file', 'completed', 'failure', ?, ?, ?)", |
| 370 | jobId, |
| 371 | id, |
| 372 | t, |
| 373 | t, |
| 374 | JSON.stringify({ message: errors.join("\n") }), |
| 375 | ); |
| 376 | } |
| 377 | |
| 378 | async createRun( |
| 379 | wf: Workflow, |
| 380 | o: { event: string; ref: string; sha: string; payload: Record<string, unknown>; caller: Caller | { id: string; login: string; kind: Actor["kind"] }; prNumber?: number; inputs?: Record<string, unknown>; headRef?: string; baseRef?: string; title: string }, |
| 381 | ): Promise<RunInfo | null> { |
| 382 | const id = newId(); |
| 383 | const t = now(); |
| 384 | const run: RunRow = { |
| 385 | id, |
| 386 | run_number: this.nextRunNumber(), |
| 387 | workflow_path: wf.path, |
| 388 | workflow_name: wf.name ?? wf.path, |
| 389 | event: o.event, |
| 390 | ref: o.ref, |
| 391 | sha: o.sha, |
| 392 | actor: JSON.stringify({ id: o.caller.id, login: o.caller.login, kind: o.caller.kind }), |
| 393 | status: "queued", |
| 394 | conclusion: null, |
| 395 | created_at: t, |
| 396 | updated_at: t, |
| 397 | pr_number: o.prNumber ?? null, |
| 398 | attempt: 1, |
| 399 | event_payload: JSON.stringify(o.payload), |
| 400 | inputs: JSON.stringify(o.inputs ?? {}), |
| 401 | concurrency_group: null, |
| 402 | display_title: o.title, |
| 403 | }; |
| 404 | const ctx: ExprContext = { github: this.githubContext(run, "", ""), inputs: o.inputs ?? {}, vars: this.variables() }; |
| 405 | try { |
| 406 | if (wf.runName) run.display_title = interpolate(wf.runName, ctx); |
| 407 | if (wf.concurrency) run.concurrency_group = interpolate(wf.concurrency.group, ctx); |
| 408 | } catch (e) { |
| 409 | console.warn("run-name/concurrency evaluation failed", e); |
| 410 | } |
| 411 | this.sql.exec( |
| 412 | "INSERT INTO runs (id, run_number, workflow_path, workflow_name, event, ref, sha, actor, status, created_at, updated_at, pr_number, attempt, event_payload, inputs, concurrency_group, display_title) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 413 | run.id, |
| 414 | run.run_number, |
| 415 | run.workflow_path, |
| 416 | run.workflow_name, |
| 417 | run.event, |
| 418 | run.ref, |
| 419 | run.sha, |
| 420 | run.actor, |
| 421 | run.status, |
| 422 | run.created_at, |
| 423 | run.updated_at, |
| 424 | run.pr_number, |
| 425 | run.attempt, |
| 426 | run.event_payload, |
| 427 | run.inputs, |
| 428 | run.concurrency_group, |
| 429 | run.display_title, |
| 430 | ); |
| 431 | if (run.concurrency_group && wf.concurrency) { |
| 432 | const cancelInProgress = evaluateValue(wf.concurrency.cancelInProgress, ctx); |
| 433 | const others = this.sql.exec<{ id: string }>("SELECT id FROM runs WHERE concurrency_group = ? AND status != 'completed' AND id != ?", run.concurrency_group, id).toArray(); |
| 434 | for (const other of others) if (cancelInProgress === true || cancelInProgress === "true") await this.cancelInternal(other.id, "cancelled by a newer run in the same concurrency group"); |
| 435 | } |
| 436 | await this.env.RUN_WORKFLOW.create({ id, params: { repoId: this.repo.meta.repoId, runId: id } }); |
| 437 | const m = this.repo.meta; |
| 438 | this.repo.emit({ type: "workflow_run", action: "requested", repoId: m.repoId, owner: m.owner, repo: m.name, runId: id, actor: JSON.parse(run.actor), at: t }); |
| 439 | this.repo.broadcast("runs", { action: "created", runId: id }); |
| 440 | return toRun(run); |
| 441 | } |
| 442 | |
| 443 | private runRow(id: string): RunRow { |
| 444 | const r = this.sql.exec<RunRow>("SELECT * FROM runs WHERE id = ?", id).toArray()[0]; |
| 445 | if (!r) throw new HttpError(404, "run not found"); |
| 446 | return r; |
| 447 | } |
| 448 | |
| 449 | private jobRow(id: string): JobRow { |
| 450 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", id).toArray()[0]; |
| 451 | if (!j) throw new HttpError(404, "job not found"); |
| 452 | return j; |
| 453 | } |
| 454 | |
| 455 | listRuns(opts: { workflow?: string; branch?: string; status?: string; event?: string; sha?: string; prNumber?: number; limit?: number; offset?: number } = {}): { runs: RunInfo[]; total: number } { |
| 456 | const where: string[] = []; |
| 457 | const args: unknown[] = []; |
| 458 | if (opts.workflow) { |
| 459 | where.push("(workflow_path = ? OR workflow_path LIKE ?)"); |
| 460 | args.push(opts.workflow, `%/${opts.workflow}`); |
| 461 | } |
| 462 | if (opts.branch) { |
| 463 | where.push("ref = ?"); |
| 464 | args.push(`refs/heads/${opts.branch}`); |
| 465 | } |
| 466 | if (opts.status) { |
| 467 | where.push(opts.status === "completed" || opts.status === "queued" || opts.status === "in_progress" ? "status = ?" : "conclusion = ?"); |
| 468 | args.push(opts.status); |
| 469 | } |
| 470 | if (opts.event) { |
| 471 | where.push("event = ?"); |
| 472 | args.push(opts.event); |
| 473 | } |
| 474 | if (opts.sha) { |
| 475 | where.push("sha = ?"); |
| 476 | args.push(opts.sha); |
| 477 | } |
| 478 | if (opts.prNumber) { |
| 479 | where.push("pr_number = ?"); |
| 480 | args.push(opts.prNumber); |
| 481 | } |
| 482 | const w = where.length ? `WHERE ${where.join(" AND ")}` : ""; |
| 483 | const total = this.sql.exec<{ n: number }>(`SELECT count(*) AS n FROM runs ${w}`, ...args).one().n; |
| 484 | const rows = this.sql.exec<RunRow>(`SELECT * FROM runs ${w} ORDER BY created_at DESC LIMIT ? OFFSET ?`, ...args, Math.min(opts.limit ?? 25, 100), opts.offset ?? 0).toArray(); |
| 485 | return { runs: rows.map(toRun), total }; |
| 486 | } |
| 487 | |
| 488 | getRun(id: string): RunInfo & { jobs: JobInfo[] } { |
| 489 | return { ...toRun(this.runRow(id)), jobs: this.jobs(id) }; |
| 490 | } |
| 491 | |
| 492 | jobs(runId: string): JobInfo[] { |
| 493 | return this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? ORDER BY rowid", runId).toArray().map(toJob); |
| 494 | } |
| 495 | |
| 496 | getJob(id: string): JobInfo { |
| 497 | return toJob(this.jobRow(id)); |
| 498 | } |
| 499 | |
| 500 | async logs(jobId: string, afterSeq = -1): Promise<{ lines: Array<LogLine & { seq: number }>; complete: boolean }> { |
| 501 | const j = this.jobRow(jobId); |
| 502 | const rows = this.sql.exec<LogLine & { seq: number }>("SELECT seq, t, step, text FROM logs WHERE job_id = ? AND seq > ? ORDER BY seq LIMIT 5000", jobId, afterSeq).toArray(); |
| 503 | if (rows.length === 0 && j.status === "completed") { |
| 504 | const archived = await this.env.BUCKET.get(this.logKey(jobId)); |
| 505 | if (archived && afterSeq < 0) return { lines: (await archived.json()) as Array<LogLine & { seq: number }>, complete: true }; |
| 506 | } |
| 507 | return { lines: rows, complete: j.status === "completed" }; |
| 508 | } |
| 509 | |
| 510 | private logKey(jobId: string) { |
| 511 | return `repos/${this.repo.meta.repoId}/logs/${jobId}.json`; |
| 512 | } |
| 513 | |
| 514 | // ---------------------------------------------------------------- orchestration (called by RunWorkflow) |
| 515 | |
| 516 | async plan(runId: string): Promise<{ levels: string[][]; error?: string; concurrencyBlocked: boolean }> { |
| 517 | const run = this.runRow(runId); |
| 518 | if (run.status === "completed") return { levels: [], error: "run already completed", concurrencyBlocked: false }; |
| 519 | const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path); |
| 520 | if (!wf?.workflow) return { levels: [], error: `workflow file not found or invalid: ${wf?.errors.join("; ") ?? run.workflow_path}`, concurrencyBlocked: false }; |
| 521 | const { order, errors } = jobOrder(wf.workflow); |
| 522 | if (errors.length) return { levels: [], error: errors.map((e) => e.message).join("; "), concurrencyBlocked: false }; |
| 523 | const blocked = |
| 524 | !!run.concurrency_group && |
| 525 | this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM runs WHERE concurrency_group = ? AND status = 'in_progress' AND id != ?", run.concurrency_group, runId).one().n > 0; |
| 526 | if (!blocked) this.sql.exec("UPDATE runs SET status = 'in_progress', updated_at = ? WHERE id = ? AND status = 'queued'", now(), runId); |
| 527 | return { levels: order, concurrencyBlocked: blocked }; |
| 528 | } |
| 529 | |
| 530 | private needsResults(runId: string, keys: string[]): Record<string, NeedsResult> { |
| 531 | const out: Record<string, NeedsResult> = {}; |
| 532 | for (const key of keys) { |
| 533 | const rows = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ?", runId, key).toArray(); |
| 534 | let result: Conclusion = "success"; |
| 535 | const outputs: Record<string, string> = {}; |
| 536 | if (rows.length === 0) result = "skipped"; |
| 537 | for (const r of rows) { |
| 538 | Object.assign(outputs, JSON.parse(r.outputs)); |
| 539 | const c = r.conclusion === "timed_out" ? "failure" : (r.conclusion ?? "failure"); |
| 540 | if (c === "failure") result = "failure"; |
| 541 | else if (c === "cancelled" && result !== "failure") result = "cancelled"; |
| 542 | else if (c === "skipped" && result === "success" && rows.length === 1) result = "skipped"; |
| 543 | } |
| 544 | out[key] = { result, outputs }; |
| 545 | } |
| 546 | return out; |
| 547 | } |
| 548 | |
| 549 | /** |
| 550 | * Start every matrix instance of a job. Returns the ids the workflow must wait |
| 551 | * for (instances beyond max-parallel are queued and started as others finish). |
| 552 | */ |
| 553 | async startJobKey(runId: string, jobKey: string): Promise<{ jobIds: string[] }> { |
| 554 | const run = this.runRow(runId); |
| 555 | const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path)!.workflow!; |
| 556 | const job = wf.jobs[jobKey]!; |
| 557 | const cancelled = run.conclusion === "cancelled"; |
| 558 | const needs = this.needsResults(runId, job.needs); |
| 559 | const status = statusFor(Object.values(needs), cancelled); |
| 560 | const baseCtx: ExprContext = { github: this.githubContext(run, jobKey, ""), needs, inputs: JSON.parse(run.inputs), vars: this.variables(), env: wf.env }; |
| 561 | const fail = (name: string, message: string, conclusion: Conclusion = "failure") => { |
| 562 | const id = newId(); |
| 563 | const t = now(); |
| 564 | this.sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, ?, ?, 'completed', ?, ?, ?, ?)", id, runId, jobKey, name, conclusion, t, t, conclusion === "failure" ? JSON.stringify({ message }) : null); |
| 565 | this.repo.broadcast(`run:${runId}`, { action: "job", jobId: id }); |
| 566 | return { jobIds: [] as string[] }; |
| 567 | }; |
| 568 | let shouldRun: boolean; |
| 569 | try { |
| 570 | shouldRun = evaluateIf(job.if, baseCtx, status); |
| 571 | } catch (e) { |
| 572 | return fail(job.name ?? jobKey, `Error evaluating 'if': ${(e as Error).message}`); |
| 573 | } |
| 574 | if (!shouldRun) return fail(job.name ?? jobKey, "", "skipped"); |
| 575 | if (job.uses) return fail(job.name ?? jobKey, `Reusable workflows (jobs.${jobKey}.uses) are not supported yet on Diffinery.`); |
| 576 | const labels = job.runsOn.map((l) => String(evaluateValue(l, baseCtx))); |
| 577 | const unsupported = labels.filter((l) => !SUPPORTED_LABEL.test(l)); |
| 578 | const selfHosted = labels.some((l) => l.toLowerCase() === "self-hosted"); |
| 579 | if (unsupported.length && !selfHosted) return fail(job.name ?? jobKey, `No Diffinery runner matches runs-on: ${unsupported.join(", ")}. Linux (ubuntu-*) runners are hosted; anything else needs runs-on: [self-hosted, ...] and a registered runner.`); |
| 580 | if (job.container) return fail(job.name ?? jobKey, "Job containers (jobs.<id>.container) are not supported yet; steps run directly on the Ubuntu runner image."); |
| 581 | if (job.services && Object.keys(job.services).length) return fail(job.name ?? jobKey, "Service containers (jobs.<id>.services) are not supported yet on Diffinery runners."); |
| 582 | |
| 583 | let combos: Array<Record<string, unknown>>; |
| 584 | try { |
| 585 | combos = expandMatrix(job.strategy?.matrix, baseCtx); |
| 586 | } catch (e) { |
| 587 | return fail(job.name ?? jobKey, `Error expanding matrix: ${(e as Error).message}`); |
| 588 | } |
| 589 | if (combos.length > 256) return fail(job.name ?? jobKey, "A matrix may produce at most 256 jobs."); |
| 590 | const failFast = evaluateValue(job.strategy?.failFast ?? true, baseCtx) !== false; |
| 591 | const maxParallel = Number(evaluateValue(job.strategy?.maxParallel ?? combos.length, baseCtx)) || combos.length; |
| 592 | const ids: string[] = []; |
| 593 | combos.forEach((matrix, i) => { |
| 594 | const ctx = { ...baseCtx, matrix, strategy: { "fail-fast": failFast, "job-index": i, "job-total": combos.length, "max-parallel": maxParallel } }; |
| 595 | let name = job.name ? interpolate(job.name, ctx) : jobKey; |
| 596 | if (!job.name && Object.keys(matrix).length) name = `${jobKey} (${Object.values(matrix).map((v) => (typeof v === "object" ? JSON.stringify(v) : String(v))).join(", ")})`; |
| 597 | const timeout = Number(evaluateValue(job.timeoutMinutes, ctx)) || 360; |
| 598 | const id = newId(); |
| 599 | this.sql.exec( |
| 600 | "INSERT INTO jobs (id, run_id, job_key, name, status, matrix, strategy, timeout_minutes, labels) VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?)", |
| 601 | id, |
| 602 | runId, |
| 603 | jobKey, |
| 604 | name, |
| 605 | JSON.stringify(matrix), |
| 606 | JSON.stringify(ctx.strategy), |
| 607 | Math.min(timeout, 360), |
| 608 | selfHosted ? JSON.stringify(labels.map((l) => l.toLowerCase())) : null, |
| 609 | ); |
| 610 | ids.push(id); |
| 611 | }); |
| 612 | for (const id of ids.slice(0, maxParallel)) await this.launch(id); |
| 613 | this.repo.broadcast(`run:${runId}`, { action: "jobs" }); |
| 614 | return { jobIds: ids }; |
| 615 | } |
| 616 | |
| 617 | private checkName(run: RunRow, job: JobRow): string { |
| 618 | return `${run.workflow_name} / ${job.name}`; |
| 619 | } |
| 620 | |
| 621 | private async launch(jobId: string): Promise<void> { |
| 622 | const job = this.jobRow(jobId); |
| 623 | const run = this.runRow(job.run_id); |
| 624 | const m = this.repo.meta; |
| 625 | if (job.labels) { |
| 626 | // Self-hosted: park the job until a registered runner with matching labels claims it. |
| 627 | this.sql.exec("UPDATE jobs SET status = 'waiting', container = NULL WHERE id = ?", jobId); |
| 628 | this.repo.upsertCheck({ sha: run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` }); |
| 629 | return; |
| 630 | } |
| 631 | const token = `djt_${m.repoId}_${jobId}_${randomToken(24)}`; |
| 632 | const container = `job-${jobId}`; |
| 633 | this.sql.exec("UPDATE jobs SET status = 'waiting', token_hash = ?, container = ? WHERE id = ?", await sha256(token), container, jobId); |
| 634 | this.repo.upsertCheck({ sha: run.sha, name: this.checkName(run, job), status: "queued", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` }); |
| 635 | try { |
| 636 | const stub = this.env.RUNNER.get(this.env.RUNNER.idFromName(container)); |
| 637 | await stub.startJob({ repoId: m.repoId, jobId, token, apiUrl: this.env.PUBLIC_URL }); |
| 638 | } catch (e) { |
| 639 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `Could not start a runner: ${(e as Error).message}` } }); |
| 640 | } |
| 641 | } |
| 642 | |
| 643 | async jobTimedOut(jobId: string): Promise<void> { |
| 644 | const j = this.jobRow(jobId); |
| 645 | if (j.status === "completed") return; |
| 646 | await this.stopContainer(j); |
| 647 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The job exceeded its timeout of ${j.timeout_minutes} minutes.` } }, "timed_out"); |
| 648 | } |
| 649 | |
| 650 | async containerStopped(jobId: string, exitCode: number | null): Promise<void> { |
| 651 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", jobId).toArray()[0]; |
| 652 | if (!j || j.status === "completed") return; |
| 653 | await this.completeJob(jobId, { conclusion: "failure", outputs: {}, failure: { message: `The runner stopped unexpectedly (exit code ${exitCode ?? "unknown"}).` } }); |
| 654 | } |
| 655 | |
| 656 | /** |
| 657 | * A self-hosted runner asks for work. Returns the oldest waiting job whose |
| 658 | * runs-on labels are all offered by the runner, with a fresh job token. |
| 659 | */ |
| 660 | async claimJob(runnerName: string, runnerLabels: string[], caller: Caller): Promise<{ jobId: string; token: string; runId: string; name: string } | null> { |
| 661 | if (!atLeast(caller.role, "admin")) throw new HttpError(403, "admin access is required to run self-hosted jobs"); |
| 662 | const offered = new Set(["self-hosted", ...runnerLabels.map((l) => l.toLowerCase())]); |
| 663 | const waiting = this.sql |
| 664 | .exec<JobRow>("SELECT j.* FROM jobs j JOIN runs r ON r.id = j.run_id WHERE j.status = 'waiting' AND j.labels IS NOT NULL AND j.container IS NULL AND r.status != 'completed' ORDER BY r.created_at, j.rowid") |
| 665 | .toArray(); |
| 666 | const job = waiting.find((j) => (JSON.parse(j.labels!) as string[]).every((l) => offered.has(l))); |
| 667 | if (!job) return null; |
| 668 | const m = this.repo.meta; |
| 669 | const token = `djt_${m.repoId}_${job.id}_${randomToken(24)}`; |
| 670 | const container = `self-hosted:${runnerName.slice(0, 64)}`; |
| 671 | this.sql.exec("UPDATE jobs SET token_hash = ?, container = ? WHERE id = ? AND container IS NULL", await sha256(token), container, job.id); |
| 672 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId: job.id }); |
| 673 | return { jobId: job.id, token, runId: job.run_id, name: job.name }; |
| 674 | } |
| 675 | |
| 676 | private async stopContainer(j: JobRow): Promise<void> { |
| 677 | if (!j.container || j.container.startsWith("self-hosted:")) return; |
| 678 | try { |
| 679 | await this.env.RUNNER.get(this.env.RUNNER.idFromName(j.container)).stopJob(); |
| 680 | } catch (e) { |
| 681 | console.warn("stop container failed", e); |
| 682 | } |
| 683 | } |
| 684 | |
| 685 | async finishRun(runId: string): Promise<RunInfo> { |
| 686 | const run = this.runRow(runId); |
| 687 | const jobs = this.jobs(runId); |
| 688 | let conclusion: RunRow["conclusion"] = run.conclusion === "cancelled" ? "cancelled" : "success"; |
| 689 | if (conclusion !== "cancelled") { |
| 690 | if (jobs.some((j) => j.conclusion === "failure" || j.conclusion === "timed_out")) conclusion = "failure"; |
| 691 | else if (jobs.some((j) => j.conclusion === "cancelled")) conclusion = "cancelled"; |
| 692 | } |
| 693 | const t = now(); |
| 694 | this.sql.exec("UPDATE runs SET status = 'completed', conclusion = ?, updated_at = ? WHERE id = ?", conclusion, t, runId); |
| 695 | const m = this.repo.meta; |
| 696 | this.repo.emit({ type: "workflow_run", action: "completed", repoId: m.repoId, owner: m.owner, repo: m.name, runId, conclusion: conclusion ?? undefined, actor: JSON.parse(run.actor), at: t }); |
| 697 | this.repo.broadcast(`run:${runId}`, { action: "completed", conclusion }); |
| 698 | this.repo.broadcast("runs", { action: "completed", runId }); |
| 699 | return toRun(this.runRow(runId)); |
| 700 | } |
| 701 | |
| 702 | async plannedError(runId: string, message: string): Promise<void> { |
| 703 | const t = now(); |
| 704 | this.sql.exec("INSERT INTO jobs (id, run_id, job_key, name, status, conclusion, started_at, completed_at, failure) VALUES (?, ?, 'setup', 'Set up workflow', 'completed', 'failure', ?, ?, ?)", newId(), runId, t, t, JSON.stringify({ message })); |
| 705 | } |
| 706 | |
| 707 | // ---------------------------------------------------------------- cancellation and re-runs |
| 708 | |
| 709 | private async cancelInternal(runId: string, reason: string): Promise<void> { |
| 710 | const run = this.runRow(runId); |
| 711 | if (run.status === "completed") return; |
| 712 | this.sql.exec("UPDATE runs SET conclusion = 'cancelled', updated_at = ? WHERE id = ?", now(), runId); |
| 713 | for (const j of this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND status != 'completed'", runId).toArray()) { |
| 714 | await this.stopContainer(j); |
| 715 | await this.completeJob(j.id, { conclusion: "cancelled", outputs: {}, failure: { message: reason } }); |
| 716 | } |
| 717 | } |
| 718 | |
| 719 | async cancel(runId: string, caller: Caller): Promise<void> { |
| 720 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied"); |
| 721 | await this.cancelInternal(runId, `Cancelled by ${caller.login}`); |
| 722 | } |
| 723 | |
| 724 | async rerun(runId: string, caller: Caller): Promise<RunInfo> { |
| 725 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "permission denied"); |
| 726 | const old = this.runRow(runId); |
| 727 | const wf = (await this.loadWorkflows(old.sha)).find((w) => w.path === old.workflow_path)?.workflow; |
| 728 | if (!wf) throw new HttpError(422, "workflow file no longer valid at this commit"); |
| 729 | const run = await this.createRun(wf, { |
| 730 | event: old.event, |
| 731 | ref: old.ref, |
| 732 | sha: old.sha, |
| 733 | payload: JSON.parse(old.event_payload), |
| 734 | caller, |
| 735 | prNumber: old.pr_number ?? undefined, |
| 736 | inputs: JSON.parse(old.inputs), |
| 737 | title: old.display_title, |
| 738 | }); |
| 739 | this.sql.exec("UPDATE runs SET attempt = ?, run_number = ? WHERE id = ?", old.attempt + 1, old.run_number, run!.id); |
| 740 | return toRun(this.runRow(run!.id)); |
| 741 | } |
| 742 | |
| 743 | // ---------------------------------------------------------------- runner protocol |
| 744 | |
| 745 | /** Validate a job or sandbox token. Returns what it grants. */ |
| 746 | async runnerAuth(token: string): Promise<{ kind: "job"; jobId: string; runId: string } | { kind: "sandbox"; sandboxId: string; caller: Caller } | null> { |
| 747 | const parts = token.split("_"); |
| 748 | if (parts.length < 4) return null; |
| 749 | const hash = await sha256(token); |
| 750 | if (parts[0] === "djt") { |
| 751 | const j = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE id = ?", parts[2]!).toArray()[0]; |
| 752 | if (!j || j.token_hash !== hash || j.status === "completed") return null; |
| 753 | return { kind: "job", jobId: j.id, runId: j.run_id }; |
| 754 | } |
| 755 | if (parts[0] === "dst") { |
| 756 | const s = this.sql.exec<{ id: string; secret: string; created_by: string; status: string }>("SELECT * FROM sandboxes WHERE id = ?", parts[2]!).toArray()[0]; |
| 757 | if (!s || s.secret !== hash || s.status !== "running") return null; |
| 758 | return { kind: "sandbox", sandboxId: s.id, caller: JSON.parse(s.created_by) as Caller }; |
| 759 | } |
| 760 | return null; |
| 761 | } |
| 762 | |
| 763 | private async authJob(jobId: string, token: string): Promise<JobRow> { |
| 764 | const auth = await this.runnerAuth(token); |
| 765 | if (!auth || auth.kind !== "job" || auth.jobId !== jobId) throw new HttpError(401, "invalid job token"); |
| 766 | return this.jobRow(jobId); |
| 767 | } |
| 768 | |
| 769 | /** Claims for an OIDC id token; side-effect free (unlike runnerSpec). */ |
| 770 | async oidcClaims(jobId: string, token: string): Promise<{ allowed: boolean; github: Record<string, string> }> { |
| 771 | const job = await this.authJob(jobId, token); |
| 772 | if (job.status !== "in_progress") throw new HttpError(409, "job is not running"); |
| 773 | const run = this.runRow(job.run_id); |
| 774 | const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path)!.workflow!; |
| 775 | const def: Job = wf.jobs[job.job_key]!; |
| 776 | const allowed = wf.permissions === "write-all" || (typeof wf.permissions === "object" && wf.permissions["id-token"] === "write") || (typeof def.permissions === "object" && def.permissions["id-token"] === "write") || def.permissions === "write-all"; |
| 777 | const ctx = this.githubContext(run, job.job_key, token) as Record<string, Json>; |
| 778 | const github: Record<string, string> = {}; |
| 779 | for (const [k, v] of Object.entries(ctx)) if (typeof v === "string" || typeof v === "number") github[k] = String(v); |
| 780 | return { allowed, github }; |
| 781 | } |
| 782 | |
| 783 | async runnerSpec(jobId: string, token: string): Promise<JobSpec> { |
| 784 | const job = await this.authJob(jobId, token); |
| 785 | const run = this.runRow(job.run_id); |
| 786 | const wf = (await this.loadWorkflows(run.sha)).find((w) => w.path === run.workflow_path)!.workflow!; |
| 787 | const def: Job = wf.jobs[job.job_key]!; |
| 788 | const m = this.repo.meta; |
| 789 | const t = now(); |
| 790 | this.sql.exec("UPDATE jobs SET status = 'in_progress', started_at = ? WHERE id = ?", t, jobId); |
| 791 | this.repo.upsertCheck({ sha: run.sha, name: this.checkName(run, job), status: "in_progress", runId: run.id, jobId, detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}` }); |
| 792 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId }); |
| 793 | const secrets: Record<string, string> = { ...(await this.orgSecrets()), ...(await this.repoSecrets()), GITHUB_TOKEN: token, DIFFINERY_TOKEN: token }; |
| 794 | const idToken = wf.permissions === "write-all" || (typeof wf.permissions === "object" && wf.permissions["id-token"] === "write") || (typeof def.permissions === "object" && def.permissions["id-token"] === "write") || def.permissions === "write-all"; |
| 795 | return { |
| 796 | runId: run.id, |
| 797 | jobId, |
| 798 | jobKey: job.job_key, |
| 799 | name: job.name, |
| 800 | repo: { owner: m.owner, name: m.name, cloneUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, defaultBranch: m.defaultBranch }, |
| 801 | sha: run.sha, |
| 802 | ref: run.ref, |
| 803 | workflowPath: run.workflow_path, |
| 804 | github: this.githubContext(run, job.job_key, token), |
| 805 | env: { ...wf.env, ...def.env }, |
| 806 | defaults: { run: { ...wf.defaults.run, ...def.defaults.run } }, |
| 807 | matrix: JSON.parse(job.matrix), |
| 808 | strategy: JSON.parse(job.strategy), |
| 809 | needs: this.needsResults(run.id, def.needs), |
| 810 | inputs: JSON.parse(run.inputs), |
| 811 | vars: this.variables(), |
| 812 | secrets, |
| 813 | outputs: def.outputs, |
| 814 | steps: def.steps.map((s) => ({ ...s, "timeout-minutes": s["timeout-minutes"] === undefined ? undefined : Number(s["timeout-minutes"]) })), |
| 815 | timeoutMinutes: job.timeout_minutes, |
| 816 | continueOnError: def.continueOnError === true || def.continueOnError === "true", |
| 817 | idToken: !!idToken, |
| 818 | apiUrl: this.env.PUBLIC_URL, |
| 819 | token, |
| 820 | }; |
| 821 | } |
| 822 | |
| 823 | async runnerLogs(jobId: string, token: string, lines: LogLine[]): Promise<void> { |
| 824 | await this.authJob(jobId, token); |
| 825 | let seq = this.sql.exec<{ n: number | null }>("SELECT max(seq) AS n FROM logs WHERE job_id = ?", jobId).one().n ?? -1; |
| 826 | const out: Array<LogLine & { seq: number }> = []; |
| 827 | for (const l of lines.slice(0, 5000)) { |
| 828 | seq++; |
| 829 | const text = l.text.length > 16384 ? `${l.text.slice(0, 16384)}…` : l.text; |
| 830 | this.sql.exec("INSERT INTO logs (job_id, seq, t, step, text) VALUES (?, ?, ?, ?, ?)", jobId, seq, l.t, l.step, text); |
| 831 | out.push({ ...l, text, seq }); |
| 832 | } |
| 833 | this.repo.broadcast(`job:${jobId}`, { action: "logs", lines: out }); |
| 834 | } |
| 835 | |
| 836 | async runnerStep(jobId: string, token: string, index: number, report: StepReport): Promise<void> { |
| 837 | const job = await this.authJob(jobId, token); |
| 838 | const steps = JSON.parse(job.steps) as Array<StepReport & { index: number }>; |
| 839 | const existing = steps.findIndex((s) => s.index === index); |
| 840 | const next = { ...(existing >= 0 ? steps[existing] : {}), ...report, index }; |
| 841 | if (existing >= 0) steps[existing] = next; |
| 842 | else steps.push(next); |
| 843 | steps.sort((a, b) => a.index - b.index); |
| 844 | this.sql.exec("UPDATE jobs SET steps = ? WHERE id = ?", JSON.stringify(steps), jobId); |
| 845 | this.repo.broadcast(`job:${jobId}`, { action: "step", step: next }); |
| 846 | this.repo.broadcast(`run:${job.run_id}`, { action: "job", jobId }); |
| 847 | } |
| 848 | |
| 849 | async runnerComplete(jobId: string, token: string, completion: JobCompletion): Promise<void> { |
| 850 | await this.authJob(jobId, token); |
| 851 | await this.completeJob(jobId, completion); |
| 852 | } |
| 853 | |
| 854 | private async completeJob(jobId: string, completion: JobCompletion, override?: "timed_out"): Promise<void> { |
| 855 | const job = this.jobRow(jobId); |
| 856 | if (job.status === "completed") return; |
| 857 | const run = this.runRow(job.run_id); |
| 858 | const t = now(); |
| 859 | const conclusion = override ?? completion.conclusion; |
| 860 | this.sql.exec( |
| 861 | "UPDATE jobs SET status = 'completed', conclusion = ?, completed_at = ?, outputs = ?, failure = ?, summary = ?, token_hash = NULL, started_at = COALESCE(started_at, ?) WHERE id = ?", |
| 862 | conclusion, |
| 863 | t, |
| 864 | JSON.stringify(completion.outputs ?? {}), |
| 865 | completion.failure ? JSON.stringify(completion.failure) : null, |
| 866 | completion.summary ?? null, |
| 867 | t, |
| 868 | jobId, |
| 869 | ); |
| 870 | const m = this.repo.meta; |
| 871 | this.repo.upsertCheck({ |
| 872 | sha: run.sha, |
| 873 | name: this.checkName(run, job), |
| 874 | status: "completed", |
| 875 | conclusion: conclusion === "timed_out" ? "timed_out" : conclusion, |
| 876 | runId: run.id, |
| 877 | jobId, |
| 878 | detailsUrl: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}/actions/runs/${run.id}/job/${jobId}`, |
| 879 | output: completion.failure ? { title: "Failed", summary: completion.failure.message } : { title: conclusion }, |
| 880 | }); |
| 881 | // Archive logs to R2 and free the Durable Object's storage. |
| 882 | const lines = this.sql.exec<LogLine & { seq: number }>("SELECT seq, t, step, text FROM logs WHERE job_id = ? ORDER BY seq", jobId).toArray(); |
| 883 | if (lines.length) { |
| 884 | await this.env.BUCKET.put(this.logKey(jobId), JSON.stringify(lines), { httpMetadata: { contentType: "application/json" } }); |
| 885 | this.sql.exec("DELETE FROM logs WHERE job_id = ?", jobId); |
| 886 | } |
| 887 | this.repo.broadcast(`job:${jobId}`, { action: "completed", conclusion }); |
| 888 | this.repo.broadcast(`run:${run.id}`, { action: "job", jobId }); |
| 889 | |
| 890 | // fail-fast: cancel unfinished siblings of a failed matrix job. |
| 891 | const siblings = this.sql.exec<JobRow>("SELECT * FROM jobs WHERE run_id = ? AND job_key = ? AND id != ? AND status != 'completed'", run.id, job.job_key, jobId).toArray(); |
| 892 | const strategy = JSON.parse(job.strategy || "{}") as { "fail-fast"?: boolean }; |
| 893 | if (conclusion === "failure" && strategy["fail-fast"] !== false && siblings.length) { |
| 894 | for (const s of siblings) { |
| 895 | await this.stopContainer(s); |
| 896 | await this.completeJob(s.id, { conclusion: "cancelled", outputs: {}, failure: { message: `Cancelled because ${job.name} failed (fail-fast).` } }); |
| 897 | } |
| 898 | } else { |
| 899 | const queued = siblings.find((s) => s.status === "queued"); |
| 900 | if (queued) await this.launch(queued.id); |
| 901 | } |
| 902 | try { |
| 903 | const instance = await this.env.RUN_WORKFLOW.get(run.id); |
| 904 | await instance.sendEvent({ type: `job-${jobId}`, payload: { conclusion } }); |
| 905 | } catch (e) { |
| 906 | console.error("failed to notify workflow", e); |
| 907 | } |
| 908 | } |
| 909 | |
| 910 | // ---------------------------------------------------------------- secrets and variables |
| 911 | |
| 912 | private async repoSecrets(): Promise<Record<string, string>> { |
| 913 | const out: Record<string, string> = {}; |
| 914 | for (const r of this.sql.exec<{ name: string; ciphertext: string }>("SELECT name, ciphertext FROM secrets").toArray()) { |
| 915 | out[r.name] = await decryptSecret(this.env.MASTER_KEY, `repo:${this.repo.meta.repoId}`, r.ciphertext); |
| 916 | } |
| 917 | return out; |
| 918 | } |
| 919 | |
| 920 | private async orgSecrets(): Promise<Record<string, string>> { |
| 921 | const m = this.repo.meta; |
| 922 | if (m.ownerKind !== "org") return {}; |
| 923 | const rows = await this.env.DB.prepare("SELECT name, ciphertext FROM org_secrets WHERE org_id = ?").bind(m.ownerId).all<{ name: string; ciphertext: string }>(); |
| 924 | const out: Record<string, string> = {}; |
| 925 | for (const r of rows.results) out[r.name] = await decryptSecret(this.env.MASTER_KEY, `org:${m.ownerId}`, r.ciphertext); |
| 926 | return out; |
| 927 | } |
| 928 | |
| 929 | async setSecret(name: string, value: string): Promise<void> { |
| 930 | if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(name) || /^GITHUB_/i.test(name)) throw new HttpError(422, "invalid secret name"); |
| 931 | this.sql.exec("INSERT OR REPLACE INTO secrets (name, ciphertext, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), await encryptSecret(this.env.MASTER_KEY, `repo:${this.repo.meta.repoId}`, value), now()); |
| 932 | } |
| 933 | |
| 934 | deleteSecret(name: string): void { |
| 935 | this.sql.exec("DELETE FROM secrets WHERE name = ?", name.toUpperCase()); |
| 936 | } |
| 937 | |
| 938 | secretNames(): Array<{ name: string; updatedAt: string }> { |
| 939 | return this.sql.exec<{ name: string; updated_at: string }>("SELECT name, updated_at FROM secrets ORDER BY name").toArray().map((r) => ({ name: r.name, updatedAt: r.updated_at })); |
| 940 | } |
| 941 | |
| 942 | setVariable(name: string, value: string | null): void { |
| 943 | if (value === null) this.sql.exec("DELETE FROM variables WHERE name = ?", name); |
| 944 | else this.sql.exec("INSERT OR REPLACE INTO variables (name, value, updated_at) VALUES (?, ?, ?)", name.toUpperCase(), value, now()); |
| 945 | } |
| 946 | |
| 947 | variables(): Record<string, string> { |
| 948 | return Object.fromEntries(this.sql.exec<{ name: string; value: string }>("SELECT name, value FROM variables").toArray().map((r) => [r.name, r.value])); |
| 949 | } |
| 950 | |
| 951 | // ---------------------------------------------------------------- cache and artifacts |
| 952 | |
| 953 | async cacheLookup(jobId: string, token: string, key: string, restoreKeys: string[]): Promise<{ key: string; r2Key: string } | null> { |
| 954 | await this.authJob(jobId, token); |
| 955 | const exact = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE key = ?", key).toArray()[0]; |
| 956 | let hit = exact?.key; |
| 957 | if (!hit) { |
| 958 | for (const prefix of restoreKeys) { |
| 959 | const r = this.sql.exec<{ key: string }>("SELECT key FROM caches WHERE key >= ? AND key < ? ORDER BY created_at DESC LIMIT 1", prefix, `${prefix}`).toArray()[0]; |
| 960 | if (r) { |
| 961 | hit = r.key; |
| 962 | break; |
| 963 | } |
| 964 | } |
| 965 | } |
| 966 | if (!hit) return null; |
| 967 | this.sql.exec("UPDATE caches SET last_used = ? WHERE key = ?", now(), hit); |
| 968 | return { key: hit, r2Key: this.cacheR2Key(hit) }; |
| 969 | } |
| 970 | |
| 971 | cacheR2Key(key: string): string { |
| 972 | return `repos/${this.repo.meta.repoId}/cache/${encodeURIComponent(key)}`; |
| 973 | } |
| 974 | |
| 975 | async cacheSaved(jobId: string, token: string, key: string, size: number): Promise<string[]> { |
| 976 | await this.authJob(jobId, token); |
| 977 | this.sql.exec("INSERT OR REPLACE INTO caches (key, size, created_at, last_used) VALUES (?, ?, ?, ?)", key, size, now(), now()); |
| 978 | // Evict least recently used entries above 10 GB per repository. |
| 979 | const evicted: string[] = []; |
| 980 | let total = this.sql.exec<{ n: number | null }>("SELECT sum(size) AS n FROM caches").one().n ?? 0; |
| 981 | while (total > 10 * 1024 ** 3) { |
| 982 | const old = this.sql.exec<{ key: string; size: number }>("SELECT key, size FROM caches ORDER BY last_used LIMIT 1").one(); |
| 983 | this.sql.exec("DELETE FROM caches WHERE key = ?", old.key); |
| 984 | evicted.push(this.cacheR2Key(old.key)); |
| 985 | total -= old.size; |
| 986 | } |
| 987 | return evicted; |
| 988 | } |
| 989 | |
| 990 | async artifactSaved(jobId: string, token: string, name: string, size: number): Promise<{ r2Key: string }> { |
| 991 | const job = await this.authJob(jobId, token); |
| 992 | this.sql.exec("INSERT OR REPLACE INTO artifacts (id, run_id, name, size, created_at) VALUES (?, ?, ?, ?, ?)", newId(), job.run_id, name, size, now()); |
| 993 | return { r2Key: this.artifactR2Key(job.run_id, name) }; |
| 994 | } |
| 995 | |
| 996 | artifactR2Key(runId: string, name: string): string { |
| 997 | return `repos/${this.repo.meta.repoId}/artifacts/${runId}/${encodeURIComponent(name)}`; |
| 998 | } |
| 999 | |
| 1000 | artifacts(runId: string): Array<{ name: string; size: number; createdAt: string; r2Key: string }> { |
| 1001 | return this.sql.exec<{ name: string; size: number; created_at: string }>("SELECT name, size, created_at FROM artifacts WHERE run_id = ? ORDER BY name", runId).toArray().map((a) => ({ name: a.name, size: a.size, createdAt: a.created_at, r2Key: this.artifactR2Key(runId, a.name) })); |
| 1002 | } |
| 1003 | |
| 1004 | /** Runs of the same commit; lets a job download artifacts uploaded by earlier jobs. */ |
| 1005 | runIdForJob(jobId: string): string { |
| 1006 | return this.jobRow(jobId).run_id; |
| 1007 | } |
| 1008 | |
| 1009 | // ---------------------------------------------------------------- sandboxes for agents |
| 1010 | |
| 1011 | async createSandbox(caller: Caller, ref?: string): Promise<{ id: string; ref: string }> { |
| 1012 | if (!atLeast(caller.role, "write")) throw new HttpError(403, "write access is required to create a sandbox"); |
| 1013 | const sha = await this.repo.resolve(ref ?? "HEAD"); |
| 1014 | if (!sha) throw new HttpError(422, "unknown ref"); |
| 1015 | const id = newId(); |
| 1016 | const m = this.repo.meta; |
| 1017 | const token = `dst_${m.repoId}_${id}_${randomToken(24)}`; |
| 1018 | const secret = randomToken(32); |
| 1019 | this.sql.exec("INSERT INTO sandboxes (id, secret, created_by, ref, created_at, status) VALUES (?, ?, ?, ?, ?, 'running')", id, await sha256(token), JSON.stringify(caller), ref ?? m.defaultBranch, now()); |
| 1020 | const stub = this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)); |
| 1021 | await stub.boot({ |
| 1022 | DIFFINERY_CLONE_URL: `${this.env.PUBLIC_URL}/${m.owner}/${m.name}.git`, |
| 1023 | DIFFINERY_REF: ref ?? m.defaultBranch, |
| 1024 | DIFFINERY_JOB_TOKEN: token, |
| 1025 | DIFFINERY_SANDBOX_SECRET: secret, |
| 1026 | DIFFINERY_API_URL: this.env.PUBLIC_URL, |
| 1027 | GIT_AUTHOR_NAME: caller.login, |
| 1028 | GIT_AUTHOR_EMAIL: `${caller.login}@users.noreply.diffinery.com`, |
| 1029 | GIT_COMMITTER_NAME: caller.login, |
| 1030 | GIT_COMMITTER_EMAIL: `${caller.login}@users.noreply.diffinery.com`, |
| 1031 | }, secret); |
| 1032 | return { id, ref: ref ?? m.defaultBranch }; |
| 1033 | } |
| 1034 | |
| 1035 | getSandbox(id: string, caller: Caller): { id: string; ref: string; status: string; createdAt: string } { |
| 1036 | const s = this.sql.exec<{ id: string; ref: string; status: string; created_at: string; created_by: string }>("SELECT * FROM sandboxes WHERE id = ?", id).toArray()[0]; |
| 1037 | if (!s) throw new HttpError(404, "sandbox not found"); |
| 1038 | const owner = JSON.parse(s.created_by) as Caller; |
| 1039 | if (owner.id !== caller.id && !atLeast(caller.role, "admin")) throw new HttpError(403, "not your sandbox"); |
| 1040 | return { id: s.id, ref: s.ref, status: s.status, createdAt: s.created_at }; |
| 1041 | } |
| 1042 | |
| 1043 | async closeSandbox(id: string, caller: Caller): Promise<void> { |
| 1044 | this.getSandbox(id, caller); |
| 1045 | this.sql.exec("UPDATE sandboxes SET status = 'closed' WHERE id = ?", id); |
| 1046 | await this.env.SANDBOX.get(this.env.SANDBOX.idFromName(id)).shutdown(); |
| 1047 | } |
| 1048 | } |