commit ece142938b360219e241e7c2fb0eb938dab8e37c
Author: finwo <finwo@pm.me>
Date: Fri, 18 Sep 2026 23:09:00 +0200
Config, database and storage foundation
Diffstat:
31 files changed, 3513 insertions(+), 0 deletions(-)
diff --git a/.editorconfig b/.editorconfig
@@ -0,0 +1,13 @@
+# 2d0e4a3f-ac19-430c-bf1b-46c68651ce21
+root = true
+
+[*]
+end_of_line = lf
+insert_final_newline = true
+charset = utf-8
+indent_size = 2
+indent_style = space
+trim_trailing_whitespace = true
+
+[Makefile*]
+indent_style = tab
diff --git a/.gitignore b/.gitignore
@@ -0,0 +1,20 @@
+# dependencies
+/node_modules/
+/package-lock.json
+
+# runtime state: sqlite db, mirrors, log spool, artifact storage
+/data/
+
+# local configuration and secrets
+/conductor.yaml
+/.env
+/secrets/
+/keys/
+
+# test scratch
+/test/tmp/
+
+# editor / os noise
+*.swp
+*~
+.DS_Store
diff --git a/LICENSE.md b/LICENSE.md
@@ -0,0 +1,34 @@
+Copyright (c) 2026 finwo
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to use, copy,
+modify, and distribute the Software, subject to the following conditions:
+
+ 1. Redistributions of source code must retain the above copyright notice, this
+ list of conditions, and the following disclaimer.
+
+ 2. Redistributions in binary form, or any public offering of the Software
+ (including hosted or managed services), must reproduce the above copyright
+ notice, this list of conditions, and the following disclaimer in the
+ documentation and/or other materials provided.
+
+ 3. Any redistribution or public offering of the Software must clearly attribute
+ the Software to the original copyright holder, reference this License, and
+ include a link to the official project repository or website.
+
+ 4. The Software may not be renamed, rebranded, or marketed in a manner that
+ implies it is an independent or proprietary product. Derivative works must
+ clearly state that they are based on the Software.
+
+ 5. Modifications to copies of the Software must carry prominent notices stating
+ that changes were made, the nature of the modifications, and the date of the
+ modifications.
+
+Any violation of these conditions terminates the permissions granted herein.
+
+THIS SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE COPYRIGHT
+HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
diff --git a/conductor.example.yaml b/conductor.example.yaml
@@ -0,0 +1,85 @@
+# conductor.example.yaml - server configuration
+#
+# Copy to conductor.yaml and edit. Every value here is a default, so an empty
+# file is a valid configuration: sqlite on disk, local object storage and
+# built-in user accounts.
+#
+# Any value can also be set from the environment, which takes precedence over
+# this file. See ENV_MAP in src/lib/config.js for the full list.
+#
+# Relative paths resolve against the directory holding this file.
+
+server:
+ host: 0.0.0.0
+ port: 8080
+ # Absolute URL this conductor is reachable at. Workers are handed callback
+ # URLs derived from it, so it must be correct behind a reverse proxy.
+ public_url: http://127.0.0.1:8080
+
+read_api:
+ host: 0.0.0.0
+ port: 8081
+
+database:
+ # Leave url unset for sqlite. Otherwise the scheme picks the dialect:
+ # mysql://user:pass@host:3306/conductor
+ # postgres://user:pass@host:5432/conductor
+ # The matching driver package must be installed: mysql2 or pg.
+ url: null
+ path: ./data/conductor.db
+ connection_limit: 10
+
+storage:
+ # Used when no s3 bucket is configured below.
+ path: ./data/storage
+ # Setting a bucket switches artifacts and finished logs to object storage.
+ # endpoint, bucket, access_key_id and secret_access_key are all required
+ # together. Garage and MinIO need force_path_style; AWS S3 does not.
+ s3:
+ endpoint: null
+ region: us-east-1
+ bucket: null
+ access_key_id: null
+ secret_access_key: null
+ force_path_style: true
+
+auth:
+ # Signs built-in session tokens. Generated per boot when unset, which logs
+ # everyone out on restart, so set it in production.
+ session_secret: null
+ session_ttl: 43200
+ # Setting an issuer switches admin authentication to OIDC. Built-in
+ # accounts are used when it is unset.
+ oidc:
+ issuer: null
+ audience: null
+ admin_role: conductor-admin
+ # Created on first boot, only while the users table is empty. With no
+ # password set, one is generated and written to the log once.
+ bootstrap_admin:
+ username: admin
+ password: null
+
+secrets:
+ # 32 bytes as 64 hex characters or base64, for example:
+ # openssl rand -hex 32
+ # Without it, project variables and trigger secrets are stored in the clear
+ # and marked as such, so they can be re-sealed later.
+ encryption_key: null
+
+git:
+ mirror_path: ./data/mirrors
+ fetch_interval: 60
+ timeout: 300
+
+log:
+ spool_path: ./data/logs
+ max_size: 67108864
+
+scheduler:
+ # A claimed job whose worker stops reporting for this long is treated as
+ # lost, then retried or failed. Must exceed reap_interval.
+ heartbeat_timeout: 120
+ reap_interval: 30
+ default_job_timeout: 3600
+ max_attempts: 3
diff --git a/migrations/mysql/001_initial.sql b/migrations/mysql/001_initial.sql
@@ -0,0 +1,144 @@
+-- 001_initial.sql - base schema for conductor (mysql and tidb)
+--
+-- Conventions, kept identical to the sqlite dialect:
+-- timestamps are BIGINT epoch milliseconds
+-- booleans are TINYINT 0 or 1
+-- json documents are TEXT
+-- secrets are TEXT produced by src/lib/secretbox.js
+--
+-- Column widths are chosen so that every indexed column stays within the
+-- InnoDB 3072 byte key limit under utf8mb4.
+
+CREATE TABLE projects (
+ id VARCHAR(64) NOT NULL PRIMARY KEY,
+ name VARCHAR(255) NOT NULL,
+ repo_url TEXT NOT NULL,
+ default_branch VARCHAR(255) NOT NULL DEFAULT 'main',
+ config_path VARCHAR(255) NOT NULL DEFAULT '.conductor.yml',
+ -- How workers obtain the source: 'archive' has them download a tarball
+ -- from the conductor, 'clone' has them git clone repo_url themselves.
+ source_mode VARCHAR(16) NOT NULL DEFAULT 'archive',
+ -- Shared secret for trigger HMAC verification, encrypted at rest.
+ trigger_secret TEXT,
+ enabled TINYINT NOT NULL DEFAULT 1,
+ run_counter BIGINT NOT NULL DEFAULT 0,
+ created_at BIGINT NOT NULL,
+ updated_at BIGINT NOT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE users (
+ id VARCHAR(64) NOT NULL PRIMARY KEY,
+ username VARCHAR(191) NOT NULL UNIQUE,
+ password_hash TEXT NOT NULL,
+ role VARCHAR(32) NOT NULL DEFAULT 'viewer',
+ disabled TINYINT NOT NULL DEFAULT 0,
+ created_at BIGINT NOT NULL,
+ last_login_at BIGINT
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE worker_tokens (
+ id VARCHAR(64) NOT NULL PRIMARY KEY,
+ name VARCHAR(255) NOT NULL,
+ token_hash CHAR(64) NOT NULL UNIQUE,
+ enabled TINYINT NOT NULL DEFAULT 1,
+ created_at BIGINT NOT NULL,
+ last_seen_at BIGINT,
+ last_ip VARCHAR(64)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE runs (
+ id VARCHAR(64) NOT NULL PRIMARY KEY,
+ project_id VARCHAR(64) NOT NULL,
+ number BIGINT NOT NULL,
+ ref VARCHAR(255),
+ base_sha VARCHAR(64),
+ head_sha VARCHAR(64) NOT NULL,
+ -- Named trigger_type rather than trigger, which is reserved in MySQL.
+ trigger_type VARCHAR(32) NOT NULL DEFAULT 'push',
+ actor VARCHAR(255),
+ title TEXT,
+ state VARCHAR(16) NOT NULL DEFAULT 'pending',
+ -- Resolved pipeline as scheduled, retained so a run stays explainable
+ -- after the repository moves on.
+ pipeline MEDIUMTEXT,
+ error TEXT,
+ created_at BIGINT NOT NULL,
+ started_at BIGINT,
+ finished_at BIGINT,
+ CONSTRAINT fk_runs_project FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE,
+ UNIQUE KEY idx_runs_project_number (project_id, number),
+ KEY idx_runs_created (created_at),
+ KEY idx_runs_state (state)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE jobs (
+ id VARCHAR(96) NOT NULL PRIMARY KEY,
+ run_id VARCHAR(64) NOT NULL,
+ -- Expanded name, for example 'package:pkg=musl'.
+ name VARCHAR(191) NOT NULL,
+ -- Template name before matrix and arch expansion.
+ base_name VARCHAR(191) NOT NULL,
+ arch VARCHAR(32),
+ image TEXT NOT NULL,
+ -- json array of worker feature names this job needs.
+ requires TEXT NOT NULL,
+ -- json document: script, env, services, artifacts, cache, matrix values.
+ spec MEDIUMTEXT NOT NULL,
+ state VARCHAR(16) NOT NULL DEFAULT 'queued',
+ allow_failure TINYINT NOT NULL DEFAULT 0,
+ attempt INT NOT NULL DEFAULT 0,
+ max_attempts INT NOT NULL DEFAULT 1,
+ timeout INT NOT NULL,
+ exit_code INT,
+ error TEXT,
+ log_key VARCHAR(255),
+ log_size BIGINT NOT NULL DEFAULT 0,
+ worker_token_id VARCHAR(64),
+ worker_name VARCHAR(255),
+ claimed_at BIGINT,
+ heartbeat_at BIGINT,
+ created_at BIGINT NOT NULL,
+ started_at BIGINT,
+ finished_at BIGINT,
+ CONSTRAINT fk_jobs_run FOREIGN KEY (run_id) REFERENCES runs(id) ON DELETE CASCADE,
+ UNIQUE KEY idx_jobs_run_name (run_id, name),
+ KEY idx_jobs_run (run_id),
+ KEY idx_jobs_state (state),
+ -- Supports the reaper scan for claimed jobs that stopped reporting.
+ KEY idx_jobs_heartbeat (state, heartbeat_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE job_deps (
+ job_id VARCHAR(96) NOT NULL,
+ depends_on_id VARCHAR(96) NOT NULL,
+ PRIMARY KEY (job_id, depends_on_id),
+ KEY idx_job_deps_reverse (depends_on_id),
+ CONSTRAINT fk_job_deps_job FOREIGN KEY (job_id) REFERENCES jobs(id) ON DELETE CASCADE,
+ CONSTRAINT fk_job_deps_target FOREIGN KEY (depends_on_id) REFERENCES jobs(id) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE artifacts (
+ id VARCHAR(64) NOT NULL PRIMARY KEY,
+ job_id VARCHAR(96) NOT NULL,
+ run_id VARCHAR(64) NOT NULL,
+ path VARCHAR(1024) NOT NULL,
+ storage_key VARCHAR(1024) NOT NULL,
+ size BIGINT NOT NULL,
+ sha256 CHAR(64) NOT NULL,
+ created_at BIGINT NOT NULL,
+ expires_at BIGINT,
+ CONSTRAINT fk_artifacts_job FOREIGN KEY (job_id) REFERENCES jobs(id) ON DELETE CASCADE,
+ KEY idx_artifacts_job (job_id),
+ KEY idx_artifacts_run (run_id),
+ KEY idx_artifacts_expires (expires_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+CREATE TABLE project_variables (
+ project_id VARCHAR(64) NOT NULL,
+ name VARCHAR(191) NOT NULL,
+ value TEXT NOT NULL,
+ masked TINYINT NOT NULL DEFAULT 1,
+ created_at BIGINT NOT NULL,
+ PRIMARY KEY (project_id, name),
+ CONSTRAINT fk_project_variables_project FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/migrations/postgres/001_initial.sql b/migrations/postgres/001_initial.sql
@@ -0,0 +1,140 @@
+-- 001_initial.sql - base schema for conductor (postgres)
+--
+-- Conventions, kept identical to the sqlite and mysql dialects:
+-- timestamps are BIGINT epoch milliseconds
+-- booleans are SMALLINT 0 or 1, deliberately not the native BOOLEAN type,
+-- so that a row read back looks the same on every dialect
+-- json documents are TEXT, not JSONB, since nothing queries inside them
+-- secrets are TEXT produced by src/lib/secretbox.js
+
+CREATE TABLE projects (
+ id TEXT NOT NULL PRIMARY KEY,
+ name TEXT NOT NULL,
+ repo_url TEXT NOT NULL,
+ default_branch TEXT NOT NULL DEFAULT 'main',
+ config_path TEXT NOT NULL DEFAULT '.conductor.yml',
+ -- How workers obtain the source: 'archive' has them download a tarball
+ -- from the conductor, 'clone' has them git clone repo_url themselves.
+ source_mode TEXT NOT NULL DEFAULT 'archive',
+ -- Shared secret for trigger HMAC verification, encrypted at rest.
+ trigger_secret TEXT,
+ enabled SMALLINT NOT NULL DEFAULT 1,
+ run_counter BIGINT NOT NULL DEFAULT 0,
+ created_at BIGINT NOT NULL,
+ updated_at BIGINT NOT NULL
+);
+
+CREATE TABLE users (
+ id TEXT NOT NULL PRIMARY KEY,
+ username TEXT NOT NULL UNIQUE,
+ password_hash TEXT NOT NULL,
+ role TEXT NOT NULL DEFAULT 'viewer',
+ disabled SMALLINT NOT NULL DEFAULT 0,
+ created_at BIGINT NOT NULL,
+ last_login_at BIGINT
+);
+
+CREATE TABLE worker_tokens (
+ id TEXT NOT NULL PRIMARY KEY,
+ name TEXT NOT NULL,
+ token_hash CHAR(64) NOT NULL UNIQUE,
+ enabled SMALLINT NOT NULL DEFAULT 1,
+ created_at BIGINT NOT NULL,
+ last_seen_at BIGINT,
+ last_ip TEXT
+);
+
+CREATE TABLE runs (
+ id TEXT NOT NULL PRIMARY KEY,
+ project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+ number BIGINT NOT NULL,
+ ref TEXT,
+ base_sha TEXT,
+ head_sha TEXT NOT NULL,
+ -- Named trigger_type rather than trigger, which is reserved in MySQL.
+ trigger_type TEXT NOT NULL DEFAULT 'push',
+ actor TEXT,
+ title TEXT,
+ state TEXT NOT NULL DEFAULT 'pending',
+ -- Resolved pipeline as scheduled, retained so a run stays explainable
+ -- after the repository moves on.
+ pipeline TEXT,
+ error TEXT,
+ created_at BIGINT NOT NULL,
+ started_at BIGINT,
+ finished_at BIGINT
+);
+
+CREATE UNIQUE INDEX idx_runs_project_number ON runs (project_id, number);
+CREATE INDEX idx_runs_created ON runs (created_at);
+CREATE INDEX idx_runs_state ON runs (state);
+
+CREATE TABLE jobs (
+ id TEXT NOT NULL PRIMARY KEY,
+ run_id TEXT NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
+ -- Expanded name, for example 'package:pkg=musl'.
+ name TEXT NOT NULL,
+ -- Template name before matrix and arch expansion.
+ base_name TEXT NOT NULL,
+ arch TEXT,
+ image TEXT NOT NULL,
+ -- json array of worker feature names this job needs.
+ requires TEXT NOT NULL,
+ -- json document: script, env, services, artifacts, cache, matrix values.
+ spec TEXT NOT NULL,
+ state TEXT NOT NULL DEFAULT 'queued',
+ allow_failure SMALLINT NOT NULL DEFAULT 0,
+ attempt INTEGER NOT NULL DEFAULT 0,
+ max_attempts INTEGER NOT NULL DEFAULT 1,
+ timeout INTEGER NOT NULL,
+ exit_code INTEGER,
+ error TEXT,
+ log_key TEXT,
+ log_size BIGINT NOT NULL DEFAULT 0,
+ worker_token_id TEXT,
+ worker_name TEXT,
+ claimed_at BIGINT,
+ heartbeat_at BIGINT,
+ created_at BIGINT NOT NULL,
+ started_at BIGINT,
+ finished_at BIGINT
+);
+
+CREATE UNIQUE INDEX idx_jobs_run_name ON jobs (run_id, name);
+CREATE INDEX idx_jobs_run ON jobs (run_id);
+CREATE INDEX idx_jobs_state ON jobs (state);
+-- Supports the reaper scan for claimed jobs that stopped reporting.
+CREATE INDEX idx_jobs_heartbeat ON jobs (state, heartbeat_at);
+
+CREATE TABLE job_deps (
+ job_id TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+ depends_on_id TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+ PRIMARY KEY (job_id, depends_on_id)
+);
+
+CREATE INDEX idx_job_deps_reverse ON job_deps (depends_on_id);
+
+CREATE TABLE artifacts (
+ id TEXT NOT NULL PRIMARY KEY,
+ job_id TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+ run_id TEXT NOT NULL,
+ path TEXT NOT NULL,
+ storage_key TEXT NOT NULL,
+ size BIGINT NOT NULL,
+ sha256 CHAR(64) NOT NULL,
+ created_at BIGINT NOT NULL,
+ expires_at BIGINT
+);
+
+CREATE INDEX idx_artifacts_job ON artifacts (job_id);
+CREATE INDEX idx_artifacts_run ON artifacts (run_id);
+CREATE INDEX idx_artifacts_expires ON artifacts (expires_at);
+
+CREATE TABLE project_variables (
+ project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+ name TEXT NOT NULL,
+ value TEXT NOT NULL,
+ masked SMALLINT NOT NULL DEFAULT 1,
+ created_at BIGINT NOT NULL,
+ PRIMARY KEY (project_id, name)
+);
diff --git a/migrations/sqlite/001_initial.sql b/migrations/sqlite/001_initial.sql
@@ -0,0 +1,139 @@
+-- 001_initial.sql - base schema for conductor (sqlite)
+--
+-- Conventions, kept identical to the mysql dialect:
+-- timestamps are INTEGER epoch milliseconds
+-- booleans are INTEGER 0 or 1
+-- json documents are TEXT
+-- secrets are TEXT produced by src/lib/secretbox.js
+
+CREATE TABLE projects (
+ id TEXT NOT NULL PRIMARY KEY,
+ name TEXT NOT NULL,
+ repo_url TEXT NOT NULL,
+ default_branch TEXT NOT NULL DEFAULT 'main',
+ config_path TEXT NOT NULL DEFAULT '.conductor.yml',
+ -- How workers obtain the source: 'archive' has them download a tarball
+ -- from the conductor, 'clone' has them git clone repo_url themselves.
+ source_mode TEXT NOT NULL DEFAULT 'archive',
+ -- Shared secret for trigger HMAC verification, encrypted at rest.
+ trigger_secret TEXT,
+ enabled INTEGER NOT NULL DEFAULT 1,
+ run_counter INTEGER NOT NULL DEFAULT 0,
+ created_at INTEGER NOT NULL,
+ updated_at INTEGER NOT NULL
+);
+
+CREATE TABLE users (
+ id TEXT NOT NULL PRIMARY KEY,
+ username TEXT NOT NULL UNIQUE,
+ password_hash TEXT NOT NULL,
+ role TEXT NOT NULL DEFAULT 'viewer',
+ disabled INTEGER NOT NULL DEFAULT 0,
+ created_at INTEGER NOT NULL,
+ last_login_at INTEGER
+);
+
+CREATE TABLE worker_tokens (
+ id TEXT NOT NULL PRIMARY KEY,
+ name TEXT NOT NULL,
+ token_hash TEXT NOT NULL UNIQUE,
+ enabled INTEGER NOT NULL DEFAULT 1,
+ created_at INTEGER NOT NULL,
+ last_seen_at INTEGER,
+ last_ip TEXT
+);
+
+CREATE TABLE runs (
+ id TEXT NOT NULL PRIMARY KEY,
+ project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+ number INTEGER NOT NULL,
+ ref TEXT,
+ base_sha TEXT,
+ head_sha TEXT NOT NULL,
+ -- Named trigger_type rather than trigger, which is reserved in MySQL.
+ trigger_type TEXT NOT NULL DEFAULT 'push',
+ actor TEXT,
+ title TEXT,
+ state TEXT NOT NULL DEFAULT 'pending',
+ -- Resolved pipeline as scheduled, retained so a run stays explainable
+ -- after the repository moves on.
+ pipeline TEXT,
+ error TEXT,
+ created_at INTEGER NOT NULL,
+ started_at INTEGER,
+ finished_at INTEGER
+);
+
+CREATE UNIQUE INDEX idx_runs_project_number ON runs (project_id, number);
+CREATE INDEX idx_runs_created ON runs (created_at);
+CREATE INDEX idx_runs_state ON runs (state);
+
+CREATE TABLE jobs (
+ id TEXT NOT NULL PRIMARY KEY,
+ run_id TEXT NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
+ -- Expanded name, for example 'package:pkg=musl'.
+ name TEXT NOT NULL,
+ -- Template name before matrix and arch expansion.
+ base_name TEXT NOT NULL,
+ arch TEXT,
+ image TEXT NOT NULL,
+ -- json array of worker feature names this job needs.
+ requires TEXT NOT NULL,
+ -- json document: script, env, services, artifacts, cache, matrix values.
+ spec TEXT NOT NULL,
+ state TEXT NOT NULL DEFAULT 'queued',
+ allow_failure INTEGER NOT NULL DEFAULT 0,
+ attempt INTEGER NOT NULL DEFAULT 0,
+ max_attempts INTEGER NOT NULL DEFAULT 1,
+ timeout INTEGER NOT NULL,
+ exit_code INTEGER,
+ error TEXT,
+ log_key TEXT,
+ log_size INTEGER NOT NULL DEFAULT 0,
+ worker_token_id TEXT,
+ worker_name TEXT,
+ claimed_at INTEGER,
+ heartbeat_at INTEGER,
+ created_at INTEGER NOT NULL,
+ started_at INTEGER,
+ finished_at INTEGER
+);
+
+CREATE UNIQUE INDEX idx_jobs_run_name ON jobs (run_id, name);
+CREATE INDEX idx_jobs_run ON jobs (run_id);
+CREATE INDEX idx_jobs_state ON jobs (state);
+-- Supports the reaper scan for claimed jobs that stopped reporting.
+CREATE INDEX idx_jobs_heartbeat ON jobs (state, heartbeat_at);
+
+CREATE TABLE job_deps (
+ job_id TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+ depends_on_id TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+ PRIMARY KEY (job_id, depends_on_id)
+);
+
+CREATE INDEX idx_job_deps_reverse ON job_deps (depends_on_id);
+
+CREATE TABLE artifacts (
+ id TEXT NOT NULL PRIMARY KEY,
+ job_id TEXT NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+ run_id TEXT NOT NULL,
+ path TEXT NOT NULL,
+ storage_key TEXT NOT NULL,
+ size INTEGER NOT NULL,
+ sha256 TEXT NOT NULL,
+ created_at INTEGER NOT NULL,
+ expires_at INTEGER
+);
+
+CREATE INDEX idx_artifacts_job ON artifacts (job_id);
+CREATE INDEX idx_artifacts_run ON artifacts (run_id);
+CREATE INDEX idx_artifacts_expires ON artifacts (expires_at);
+
+CREATE TABLE project_variables (
+ project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+ name TEXT NOT NULL,
+ value TEXT NOT NULL,
+ masked INTEGER NOT NULL DEFAULT 1,
+ created_at INTEGER NOT NULL,
+ PRIMARY KEY (project_id, name)
+);
diff --git a/package.json b/package.json
@@ -0,0 +1,28 @@
+{
+ "name": "conductor",
+ "version": "0.1.0",
+ "description": "Stand-alone CI system: in-repo pipelines, containerised jobs, pull-based workers",
+ "type": "module",
+ "main": "src/conductor/index.js",
+ "scripts": {
+ "start": "node src/conductor/index.js",
+ "dev": "node --watch src/conductor/index.js",
+ "read-api": "node src/read-api/index.js",
+ "worker": "node src/worker/agent.js",
+ "migrate": "node src/lib/db/migrate-cli.js",
+ "test": "node --test \"test/**/*.test.js\""
+ },
+ "dependencies": {
+ "@fastify/cors": "^11.3.0",
+ "fastify": "^5.12.5",
+ "jose": "^6.2.12",
+ "yaml": "^2.8.1"
+ },
+ "optionalDependencies": {
+ "mysql2": "^3.24.4",
+ "pg": "^8.16.3"
+ },
+ "engines": {
+ "node": ">=24.0.0"
+ }
+}
diff --git a/src/lib/config.js b/src/lib/config.js
@@ -0,0 +1,317 @@
+// src/lib/config.js - configuration loading for all conductor services
+//
+// Resolution order, lowest precedence first:
+// 1. built-in defaults (see DEFAULTS)
+// 2. YAML file, from CONDUCTOR_CONFIG or ./conductor.yaml when present
+// 3. environment variables (see ENV_MAP)
+//
+// Two capabilities switch on automatically when configured, and fall back
+// otherwise:
+// storage.s3.bucket set -> S3 object storage, else local filesystem
+// auth.oidc.issuer set -> OIDC bearer auth, else built-in user accounts
+
+import fs from 'node:fs';
+import path from 'node:path';
+import crypto from 'node:crypto';
+import YAML from 'yaml';
+import { dialectFromUrl } from './db/dialect.js';
+
+const DEFAULTS = {
+ server: {
+ host: '0.0.0.0',
+ port: 8080,
+ // Absolute URL this conductor is reachable at. Used when handing workers
+ // callback URLs, so it must be correct behind a reverse proxy.
+ public_url: 'http://127.0.0.1:8080',
+ },
+ read_api: {
+ host: '0.0.0.0',
+ port: 8081,
+ },
+ database: {
+ // Unset means sqlite at database.path. Otherwise the scheme selects the
+ // dialect: mysql:// or postgres:// (postgresql:// is accepted too).
+ url: null,
+ path: './data/conductor.db',
+ // Pool sizing for the networked dialects, ignored by sqlite.
+ connection_limit: 10,
+ },
+ storage: {
+ path: './data/storage',
+ s3: {
+ endpoint: null,
+ region: 'us-east-1',
+ bucket: null,
+ access_key_id: null,
+ secret_access_key: null,
+ // Garage and MinIO need path style addressing.
+ force_path_style: true,
+ },
+ },
+ auth: {
+ // Signs built-in session tokens. Random per boot when unset, which
+ // invalidates existing sessions on restart.
+ session_secret: null,
+ session_ttl: 43200,
+ oidc: {
+ issuer: null,
+ audience: null,
+ admin_role: 'conductor-admin',
+ },
+ // Created on first boot when the users table is empty.
+ bootstrap_admin: {
+ username: 'admin',
+ password: null,
+ },
+ },
+ secrets: {
+ // 32 bytes, hex or base64, for AES-256-GCM encryption of project
+ // variables at rest. Required only once project variables are used.
+ encryption_key: null,
+ },
+ git: {
+ mirror_path: './data/mirrors',
+ // Seconds before a cached mirror fetch is considered stale.
+ fetch_interval: 60,
+ timeout: 300,
+ },
+ log: {
+ spool_path: './data/logs',
+ // Refuse log appends past this size, to bound a runaway job.
+ max_size: 64 * 1024 * 1024,
+ },
+ scheduler: {
+ // A claimed job whose worker stops sending heartbeats for this long is
+ // considered lost and is requeued or failed.
+ heartbeat_timeout: 120,
+ reap_interval: 30,
+ default_job_timeout: 3600,
+ max_attempts: 3,
+ },
+};
+
+// [environment variable, dotted config path, parser]
+const ENV_MAP = [
+ ['CONDUCTOR_HOST', 'server.host', String],
+ ['CONDUCTOR_PORT', 'server.port', toInt],
+ ['CONDUCTOR_PUBLIC_URL', 'server.public_url', String],
+ ['CONDUCTOR_READ_API_HOST', 'read_api.host', String],
+ ['CONDUCTOR_READ_API_PORT', 'read_api.port', toInt],
+ ['CONDUCTOR_DATABASE_URL', 'database.url', String],
+ ['CONDUCTOR_DATABASE_PATH', 'database.path', String],
+ ['CONDUCTOR_STORAGE_PATH', 'storage.path', String],
+ ['CONDUCTOR_S3_ENDPOINT', 'storage.s3.endpoint', String],
+ ['CONDUCTOR_S3_REGION', 'storage.s3.region', String],
+ ['CONDUCTOR_S3_BUCKET', 'storage.s3.bucket', String],
+ ['CONDUCTOR_S3_ACCESS_KEY_ID', 'storage.s3.access_key_id', String],
+ ['CONDUCTOR_S3_SECRET_ACCESS_KEY', 'storage.s3.secret_access_key', String],
+ ['CONDUCTOR_S3_FORCE_PATH_STYLE', 'storage.s3.force_path_style', toBool],
+ ['CONDUCTOR_SESSION_SECRET', 'auth.session_secret', String],
+ ['CONDUCTOR_OIDC_ISSUER', 'auth.oidc.issuer', String],
+ ['CONDUCTOR_OIDC_AUDIENCE', 'auth.oidc.audience', String],
+ ['CONDUCTOR_OIDC_ADMIN_ROLE', 'auth.oidc.admin_role', String],
+ ['CONDUCTOR_ADMIN_USERNAME', 'auth.bootstrap_admin.username', String],
+ ['CONDUCTOR_ADMIN_PASSWORD', 'auth.bootstrap_admin.password', String],
+ ['CONDUCTOR_SECRET_KEY', 'secrets.encryption_key', String],
+ ['CONDUCTOR_MIRROR_PATH', 'git.mirror_path', String],
+ ['CONDUCTOR_LOG_PATH', 'log.spool_path', String],
+];
+
+// Paths resolved relative to the config file directory, or cwd when there is
+// no config file.
+const PATH_KEYS = ['database.path', 'storage.path', 'git.mirror_path', 'log.spool_path'];
+
+function toInt(v) {
+ const n = parseInt(v, 10);
+ if (!Number.isFinite(n)) throw new Error(`expected an integer, got ${JSON.stringify(v)}`);
+ return n;
+}
+
+function toBool(v) {
+ if (typeof v === 'boolean') return v;
+ const s = String(v).toLowerCase();
+ if (['1', 'true', 'yes', 'on'].includes(s)) return true;
+ if (['0', 'false', 'no', 'off'].includes(s)) return false;
+ throw new Error(`expected a boolean, got ${JSON.stringify(v)}`);
+}
+
+function isPlainObject(v) {
+ return v !== null && typeof v === 'object' && !Array.isArray(v);
+}
+
+function deepMerge(base, overlay) {
+ const out = Array.isArray(base) ? [...base] : { ...base };
+ for (const [k, v] of Object.entries(overlay || {})) {
+ if (v === undefined) continue;
+ out[k] = isPlainObject(v) && isPlainObject(base?.[k]) ? deepMerge(base[k], v) : v;
+ }
+ return out;
+}
+
+function getPath(obj, dotted) {
+ return dotted.split('.').reduce((acc, k) => (acc == null ? acc : acc[k]), obj);
+}
+
+function setPath(obj, dotted, value) {
+ const keys = dotted.split('.');
+ const last = keys.pop();
+ let cur = obj;
+ for (const k of keys) {
+ if (!isPlainObject(cur[k])) cur[k] = {};
+ cur = cur[k];
+ }
+ cur[last] = value;
+}
+
+function deepFreeze(obj) {
+ for (const v of Object.values(obj)) {
+ if (isPlainObject(v) || Array.isArray(v)) deepFreeze(v);
+ }
+ return Object.freeze(obj);
+}
+
+// Reads a 32 byte key given as hex or base64. Returns a Buffer, or null.
+export function parseKey(value, label) {
+ if (!value) return null;
+ let buf = null;
+ if (/^[0-9a-fA-F]{64}$/.test(value)) buf = Buffer.from(value, 'hex');
+ else buf = Buffer.from(value, 'base64');
+ if (buf.length !== 32) {
+ throw new Error(`${label} must decode to 32 bytes, got ${buf.length}; use 64 hex chars or base64`);
+ }
+ return buf;
+}
+
+function validate(cfg) {
+ const errors = [];
+
+ if (!Number.isInteger(cfg.server.port) || cfg.server.port < 1 || cfg.server.port > 65535) {
+ errors.push(`server.port must be between 1 and 65535, got ${cfg.server.port}`);
+ }
+ if (!Number.isInteger(cfg.read_api.port) || cfg.read_api.port < 1 || cfg.read_api.port > 65535) {
+ errors.push(`read_api.port must be between 1 and 65535, got ${cfg.read_api.port}`);
+ }
+ if (cfg.database.url && dialectFromUrl(cfg.database.url) === null) {
+ errors.push(
+ `database.url has an unsupported scheme: ${String(cfg.database.url).split(':')[0]}; ` +
+ 'use mysql://, postgres:// or leave it unset for sqlite'
+ );
+ }
+
+ const s3 = cfg.storage.s3;
+ const s3Given = [s3.endpoint, s3.bucket, s3.access_key_id, s3.secret_access_key].filter(Boolean);
+ if (s3Given.length > 0 && s3Given.length < 4) {
+ errors.push(
+ 'storage.s3 is partially configured; endpoint, bucket, access_key_id and ' +
+ 'secret_access_key are all required, or leave them all unset for local storage'
+ );
+ }
+
+ if (cfg.auth.oidc.issuer) {
+ try {
+ new URL(cfg.auth.oidc.issuer);
+ } catch {
+ errors.push(`auth.oidc.issuer must be an absolute URL, got ${JSON.stringify(cfg.auth.oidc.issuer)}`);
+ }
+ }
+
+ try {
+ new URL(cfg.server.public_url);
+ } catch {
+ errors.push(`server.public_url must be an absolute URL, got ${JSON.stringify(cfg.server.public_url)}`);
+ }
+
+ try {
+ parseKey(cfg.secrets.encryption_key, 'secrets.encryption_key');
+ } catch (e) {
+ errors.push(e.message);
+ }
+
+ if (cfg.scheduler.heartbeat_timeout <= cfg.scheduler.reap_interval) {
+ errors.push(
+ `scheduler.heartbeat_timeout (${cfg.scheduler.heartbeat_timeout}) must be greater than ` +
+ `scheduler.reap_interval (${cfg.scheduler.reap_interval}), or healthy jobs will be reaped`
+ );
+ }
+
+ if (errors.length > 0) {
+ throw new Error(`invalid configuration:\n - ${errors.join('\n - ')}`);
+ }
+}
+
+export function loadConfig(explicitPath) {
+ const file = explicitPath || process.env.CONDUCTOR_CONFIG || 'conductor.yaml';
+ let cfg = structuredClone(DEFAULTS);
+ let baseDir = process.cwd();
+ let source = null;
+
+ if (fs.existsSync(file)) {
+ const text = fs.readFileSync(file, 'utf8');
+ let parsed;
+ try {
+ parsed = YAML.parse(text) || {};
+ } catch (e) {
+ throw new Error(`failed to parse config file ${file}: ${e.message}`);
+ }
+ if (!isPlainObject(parsed)) {
+ throw new Error(`config file ${file} must contain a YAML mapping at the top level`);
+ }
+ cfg = deepMerge(cfg, parsed);
+ baseDir = path.dirname(path.resolve(file));
+ source = path.resolve(file);
+ } else if (explicitPath || process.env.CONDUCTOR_CONFIG) {
+ throw new Error(`config file not found: ${file}`);
+ }
+
+ for (const [env, dotted, parse] of ENV_MAP) {
+ const raw = process.env[env];
+ if (raw === undefined || raw === '') continue;
+ try {
+ setPath(cfg, dotted, parse(raw));
+ } catch (e) {
+ throw new Error(`invalid value for ${env}: ${e.message}`);
+ }
+ }
+
+ // A sqlite: or file: url is just another way of spelling database.path.
+ // Collapse it now so that everything downstream sees one representation.
+ if (cfg.database.url && /^(sqlite|file):/i.test(cfg.database.url)) {
+ const raw = cfg.database.url.replace(/^(sqlite|file):(\/\/)?/i, '');
+ if (!raw) throw new Error(`database.url ${cfg.database.url} does not contain a file path`);
+ cfg.database.path = raw;
+ cfg.database.url = null;
+ }
+
+ for (const key of PATH_KEYS) {
+ const v = getPath(cfg, key);
+ if (typeof v === 'string' && v.length > 0) setPath(cfg, key, path.resolve(baseDir, v));
+ }
+
+ validate(cfg);
+
+ // Derived flags, so callers never re-implement the fallback rules.
+ cfg.source = source;
+ cfg.database.dialect = dialectFromUrl(cfg.database.url);
+ cfg.storage.driver = cfg.storage.s3.bucket ? 's3' : 'local';
+ cfg.auth.mode = cfg.auth.oidc.issuer ? 'oidc' : 'local';
+
+ if (!cfg.auth.session_secret) {
+ cfg.auth.session_secret = crypto.randomBytes(32).toString('hex');
+ cfg.auth.session_secret_ephemeral = true;
+ }
+
+ return deepFreeze(cfg);
+}
+
+// Directories that must exist before a service starts. Which ones matter
+// depends on the selected drivers, so this is derived rather than fixed.
+export function stateDirs(cfg) {
+ const dirs = [cfg.git.mirror_path, cfg.log.spool_path];
+ if (cfg.storage.driver === 'local') dirs.push(cfg.storage.path);
+ if (cfg.database.dialect === 'sqlite') dirs.push(path.dirname(cfg.database.path));
+ return [...new Set(dirs)];
+}
+
+export function ensureStateDirs(cfg) {
+ for (const dir of stateDirs(cfg)) fs.mkdirSync(dir, { recursive: true });
+}
diff --git a/src/lib/db/dialect.js b/src/lib/db/dialect.js
@@ -0,0 +1,17 @@
+// src/lib/db/dialect.js - dialect naming and detection
+//
+// Kept separate from index.js so that configuration loading can resolve a
+// dialect without pulling in any driver module.
+
+export const DIALECTS = ['sqlite', 'mysql', 'postgres'];
+
+// Maps a database.url scheme onto a dialect name. An unset url means sqlite.
+// Returns null for a scheme that is not supported.
+export function dialectFromUrl(url) {
+ if (!url) return 'sqlite';
+ const scheme = String(url).split(':')[0].toLowerCase();
+ if (scheme === 'mysql') return 'mysql';
+ if (scheme === 'postgres' || scheme === 'postgresql') return 'postgres';
+ if (scheme === 'sqlite' || scheme === 'file') return 'sqlite';
+ return null;
+}
diff --git a/src/lib/db/index.js b/src/lib/db/index.js
@@ -0,0 +1,47 @@
+// src/lib/db/index.js - database driver selection
+//
+// Every driver exposes the same surface:
+// all(sql, params) -> array of rows
+// get(sql, params) -> first row, or undefined
+// run(sql, params) -> { changes, lastInsertId }
+// exec(sql) -> no result, for DDL, bind markers are not parsed
+// transaction(fn) -> fn receives a scoped handle, nesting uses savepoints
+// close()
+//
+// Params is an object, and queries use named {name} markers which each driver
+// compiles to its own placeholder syntax. See query.js.
+//
+// db.get('SELECT * FROM jobs WHERE run_id = {run}', { run: runId })
+//
+// Portability rules that apply to every query written against this layer:
+// - named {name} markers only, never ? or $n directly
+// - timestamps are epoch milliseconds, produced by Date.now()
+// - booleans are stored as 0 and 1, never a native boolean type
+// - no dialect specific functions; keep those behind a driver method
+// - no reliance on lastInsertId; identifiers are generated by the caller
+// - no upsert syntax; ON DUPLICATE KEY and ON CONFLICT are not portable,
+// so do the read and the write explicitly inside a transaction
+//
+// One behavioural note on run().changes: mysql counts rows actually changed,
+// while sqlite and postgres count rows matched. An UPDATE that writes a value
+// identical to the current one therefore reports 0 on mysql and 1 elsewhere.
+// The conditional claim pattern used by the scheduler always writes a
+// different state, so it is unaffected; avoid depending on changes for an
+// update that may be a no-op.
+
+import { openSqlite } from './sqlite.js';
+import { openMysql } from './mysql.js';
+import { openPostgres } from './postgres.js';
+
+export { DIALECTS, dialectFromUrl } from './dialect.js';
+
+export async function openDatabase(cfg) {
+ switch (cfg.database.dialect) {
+ case 'mysql': return openMysql(cfg);
+ case 'postgres': return openPostgres(cfg);
+ case 'sqlite': return openSqlite(cfg);
+ default: throw new Error(`unknown database dialect: ${cfg.database.dialect}`);
+ }
+}
+
+export { runMigrations } from './migrate.js';
diff --git a/src/lib/db/migrate.js b/src/lib/db/migrate.js
@@ -0,0 +1,219 @@
+// src/lib/db/migrate.js - forward-only migration runner
+//
+// Migrations live in migrations/<dialect>/NNN_name.sql and are applied in
+// filename order. Each applied file is recorded with a checksum, so editing a
+// migration that has already run is reported as an error rather than silently
+// diverging between environments.
+//
+// Note on atomicity: sqlite and postgres both run DDL inside a transaction,
+// so a failed migration rolls back completely there. MySQL and TiDB implicitly
+// commit on DDL, so a file that fails halfway can leave its earlier statements
+// applied. The surrounding transaction is still worth having on those, since
+// it keeps the bookkeeping insert and any DML atomic. Keep one logical change
+// per file so a partial apply stays easy to reason about.
+
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import crypto from 'node:crypto';
+import { fileURLToPath } from 'node:url';
+import { RULES } from './query.js';
+
+const HERE = path.dirname(fileURLToPath(import.meta.url));
+export const MIGRATIONS_ROOT = path.resolve(HERE, '../../../migrations');
+
+const DDL = {
+ sqlite: `
+ CREATE TABLE IF NOT EXISTS _migrations (
+ filename TEXT PRIMARY KEY,
+ checksum TEXT NOT NULL,
+ executed_at INTEGER NOT NULL
+ )
+ `,
+ mysql: `
+ CREATE TABLE IF NOT EXISTS _migrations (
+ filename VARCHAR(255) NOT NULL PRIMARY KEY,
+ checksum CHAR(64) NOT NULL,
+ executed_at BIGINT NOT NULL
+ ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
+ `,
+ postgres: `
+ CREATE TABLE IF NOT EXISTS _migrations (
+ filename TEXT NOT NULL PRIMARY KEY,
+ checksum CHAR(64) NOT NULL,
+ executed_at BIGINT NOT NULL
+ )
+ `,
+};
+
+// Splits a SQL file into statements on semicolons, while respecting string
+// literals, quoted identifiers and comments. Splitting on a bare semicolon,
+// as the prototype did, corrupts any statement containing one inside a
+// literal. Lexing rules are per dialect, shared with the query compiler.
+export function splitStatements(sql, dialect = 'sqlite') {
+ const rules = RULES[dialect];
+ if (!rules) throw new Error(`unknown dialect: ${dialect}`);
+
+ const out = [];
+ let cur = '';
+ let i = 0;
+
+ while (i < sql.length) {
+ const c = sql[i];
+ const next = sql[i + 1];
+
+ if ((c === '-' && next === '-') || (rules.hash && c === '#')) {
+ const nl = sql.indexOf('\n', i);
+ if (nl === -1) break;
+ cur += '\n';
+ i = nl + 1;
+ continue;
+ }
+
+ if (c === '/' && next === '*') {
+ let depth = 1;
+ let j = i + 2;
+ while (j < sql.length && depth > 0) {
+ if (rules.nestedBlock && sql[j] === '/' && sql[j + 1] === '*') {
+ depth += 1;
+ j += 2;
+ } else if (sql[j] === '*' && sql[j + 1] === '/') {
+ depth -= 1;
+ j += 2;
+ } else {
+ j += 1;
+ }
+ }
+ cur += ' ';
+ i = j;
+ continue;
+ }
+
+ if (c === "'" || c === '"' || (rules.backtick && c === '`')) {
+ const quote = c;
+ cur += c;
+ i += 1;
+ while (i < sql.length) {
+ // Backslash escapes apply to MySQL string literals, not to backtick
+ // quoted identifiers.
+ if (rules.backslash && quote !== '`' && sql[i] === '\\') {
+ cur += sql[i] + (sql[i + 1] ?? '');
+ i += 2;
+ continue;
+ }
+ if (sql[i] === quote) {
+ // A doubled quote is a literal quote, not a terminator.
+ if (sql[i + 1] === quote) {
+ cur += quote + quote;
+ i += 2;
+ continue;
+ }
+ cur += quote;
+ i += 1;
+ break;
+ }
+ cur += sql[i];
+ i += 1;
+ }
+ continue;
+ }
+
+ // Dollar quoted bodies routinely contain semicolons.
+ if (rules.dollar && c === '$') {
+ const m = /^\$([A-Za-z_][A-Za-z0-9_]*)?\$/.exec(sql.slice(i));
+ if (m) {
+ const tag = m[0];
+ const end = sql.indexOf(tag, i + tag.length);
+ const stop = end === -1 ? sql.length : end + tag.length;
+ cur += sql.slice(i, stop);
+ i = stop;
+ continue;
+ }
+ }
+
+ if (c === ';') {
+ if (cur.trim()) out.push(cur.trim());
+ cur = '';
+ i += 1;
+ continue;
+ }
+
+ cur += c;
+ i += 1;
+ }
+
+ if (cur.trim()) out.push(cur.trim());
+ return out;
+}
+
+function checksum(sql) {
+ // Normalize line endings so a CRLF checkout does not invalidate history.
+ return crypto.createHash('sha256').update(sql.replace(/\r\n/g, '\n')).digest('hex');
+}
+
+export async function runMigrations(db, options = {}) {
+ const dir = options.dir || path.join(MIGRATIONS_ROOT, db.dialect);
+ const log = options.logger || ((m) => console.log(`[migrate] ${m}`));
+
+ await db.exec(DDL[db.dialect]);
+
+ const appliedRows = await db.all('SELECT filename, checksum FROM _migrations');
+ const applied = new Map(appliedRows.map((r) => [r.filename, r.checksum]));
+
+ let files;
+ try {
+ files = (await fs.readdir(dir)).filter((f) => f.endsWith('.sql')).sort();
+ } catch (e) {
+ if (e.code === 'ENOENT') throw new Error(`no migrations directory for dialect ${db.dialect}: ${dir}`);
+ throw e;
+ }
+
+ const pending = [];
+ for (const file of files) {
+ const sql = await fs.readFile(path.join(dir, file), 'utf8');
+ const sum = checksum(sql);
+ const prev = applied.get(file);
+ if (prev === undefined) {
+ pending.push({ file, sql, sum });
+ continue;
+ }
+ if (prev !== sum) {
+ throw new Error(
+ `migration ${file} was modified after it was applied\n` +
+ ` recorded: ${prev}\n` +
+ ` on disk: ${sum}\n` +
+ 'Add a new migration instead of editing an applied one.'
+ );
+ }
+ }
+
+ // Detect files removed from disk but still recorded, which usually means a
+ // downgrade or a bad merge.
+ for (const name of applied.keys()) {
+ if (!files.includes(name)) {
+ throw new Error(`migration ${name} is recorded as applied but is missing from ${dir}`);
+ }
+ }
+
+ if (pending.length === 0) {
+ log(`up to date (${files.length} applied)`);
+ return { applied: [], total: files.length };
+ }
+
+ const done = [];
+ for (const { file, sql, sum } of pending) {
+ log(`applying ${file}`);
+ const statements = splitStatements(sql, db.dialect);
+ if (statements.length === 0) throw new Error(`migration ${file} contains no statements`);
+ await db.transaction(async (tx) => {
+ for (const stmt of statements) await tx.exec(stmt);
+ await tx.run(
+ 'INSERT INTO _migrations (filename, checksum, executed_at) VALUES ({file}, {sum}, {at})',
+ { file, sum, at: Date.now() }
+ );
+ });
+ done.push(file);
+ }
+
+ log(`applied ${done.length} migration(s)`);
+ return { applied: done, total: files.length };
+}
diff --git a/src/lib/db/mysql.js b/src/lib/db/mysql.js
@@ -0,0 +1,145 @@
+// src/lib/db/mysql.js - MySQL and TiDB driver built on mysql2
+//
+// mysql2 is an optional dependency: it is only required when database.url
+// uses a mysql scheme, so a default sqlite install does not need it.
+
+import { compileCached, bindParams } from './query.js';
+
+async function loadDriver() {
+ try {
+ return (await import('mysql2/promise')).default;
+ } catch (e) {
+ throw new Error(
+ 'database.url uses a mysql scheme but the mysql2 package is not installed. ' +
+ 'Run `npm install mysql2`, or unset database.url to use sqlite.',
+ { cause: e }
+ );
+ }
+}
+
+function fail(e, sql) {
+ const err = new Error(`mysql: ${e.message}\n sql: ${sql.trim().split('\n')[0]}`);
+ err.cause = e;
+ err.code = e.code;
+ throw err;
+}
+
+// Wraps either the pool or a single transaction connection.
+function wrap(runner) {
+ function compile(sql, params) {
+ const compiled = compileCached(sql, 'mysql');
+ return { text: compiled.sql, values: bindParams(compiled.keys, params, sql) };
+ }
+
+ return {
+ dialect: 'mysql',
+
+ async all(sql, params = {}) {
+ const { text, values } = compile(sql, params);
+ try {
+ const [rows] = await runner.query(text, values);
+ return rows;
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ async get(sql, params = {}) {
+ const { text, values } = compile(sql, params);
+ try {
+ const [rows] = await runner.query(text, values);
+ return rows[0];
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ async run(sql, params = {}) {
+ const { text, values } = compile(sql, params);
+ try {
+ const [res] = await runner.query(text, values);
+ return {
+ changes: res.affectedRows ?? 0,
+ lastInsertId: res.insertId === undefined || res.insertId === 0 ? null : Number(res.insertId),
+ };
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ // Raw DDL, no bind markers are interpreted.
+ async exec(sql) {
+ try {
+ await runner.query(sql);
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+ };
+}
+
+export async function openMysql(cfg) {
+ const mysql = await loadDriver();
+ const pool = mysql.createPool({
+ uri: cfg.database.url,
+ connectionLimit: cfg.database.connection_limit,
+ waitForConnections: true,
+ // Epoch millisecond timestamps sit well inside the safe integer range,
+ // so BIGINT should come back as a number rather than a string.
+ supportBigNumbers: true,
+ bigNumberStrings: false,
+ dateStrings: true,
+ multipleStatements: false,
+ });
+
+ const api = wrap(pool);
+
+ api.transaction = async function transaction(fn) {
+ const conn = await pool.getConnection();
+ const tx = wrap(conn);
+ let depth = 0;
+
+ // Savepoints give the same composability as the other drivers.
+ tx.transaction = async (inner) => {
+ const name = `sp_${depth}`;
+ depth += 1;
+ await conn.query(`SAVEPOINT ${name}`);
+ try {
+ const r = await inner(tx);
+ await conn.query(`RELEASE SAVEPOINT ${name}`);
+ depth -= 1;
+ return r;
+ } catch (e) {
+ try {
+ await conn.query(`ROLLBACK TO SAVEPOINT ${name}`);
+ } catch {
+ // Preserve the original failure.
+ }
+ depth -= 1;
+ throw e;
+ }
+ };
+
+ try {
+ await conn.beginTransaction();
+ const result = await fn(tx);
+ await conn.commit();
+ return result;
+ } catch (e) {
+ try {
+ await conn.rollback();
+ } catch {
+ // Preserve the original failure.
+ }
+ throw e;
+ } finally {
+ conn.release();
+ }
+ };
+
+ api.close = async () => {
+ await pool.end();
+ };
+
+ return api;
+}
diff --git a/src/lib/db/params.js b/src/lib/db/params.js
@@ -0,0 +1,34 @@
+// src/lib/db/params.js - bind value normalization shared by every driver
+//
+// Callers write portable application code and let this file absorb the
+// differences between drivers. node:sqlite rejects booleans, undefined and
+// Date objects outright; mysql2 and pg accept them but would each apply their
+// own timestamp handling. Timestamps are epoch milliseconds everywhere, so
+// Date values are flattened here rather than handed to any driver.
+
+export function normalizeValue(v, key) {
+ if (v === undefined || v === null) return null;
+ if (typeof v === 'boolean') return v ? 1 : 0;
+ if (v instanceof Date) {
+ if (Number.isNaN(v.getTime())) {
+ throw new Error(`bind parameter ${key ? `{${key}} ` : ''}is an invalid Date`);
+ }
+ return v.getTime();
+ }
+ if (typeof v === 'bigint') {
+ if (v > BigInt(Number.MAX_SAFE_INTEGER) || v < BigInt(Number.MIN_SAFE_INTEGER)) {
+ throw new Error(`bind parameter ${key ? `{${key}} ` : ''}exceeds the safe integer range: ${v}`);
+ }
+ return Number(v);
+ }
+ if (typeof v === 'number' || typeof v === 'string') return v;
+ if (Buffer.isBuffer(v) || v instanceof Uint8Array) return v;
+
+ // Objects and arrays are almost always a forgotten JSON.stringify, or a
+ // dotted key that should have addressed a leaf. Both are worth catching.
+ throw new Error(
+ `bind parameter ${key ? `{${key}} ` : ''}has unsupported type ${
+ Array.isArray(v) ? 'array' : typeof v
+ }; serialize it first, or address a nested value with a dotted key`
+ );
+}
diff --git a/src/lib/db/postgres.js b/src/lib/db/postgres.js
@@ -0,0 +1,147 @@
+// src/lib/db/postgres.js - PostgreSQL driver built on pg
+//
+// pg is an optional dependency: it is only required when database.url uses a
+// postgres scheme.
+//
+// Two adaptations are made so that portable queries keep working:
+// - {name} markers compile to $n, see query.js
+// - int8 is parsed to Number, since epoch millisecond timestamps are well
+// inside the safe integer range and pg returns int8 as a string otherwise
+//
+// lastInsertId is always null here. The schema uses application generated
+// string identifiers throughout, so nothing depends on it; retrieving one
+// would need a RETURNING clause the other dialects do not accept.
+
+import { compileCached, bindParams } from './query.js';
+
+async function loadDriver() {
+ try {
+ return (await import('pg')).default;
+ } catch (e) {
+ throw new Error(
+ 'database.url uses a postgres scheme but the pg package is not installed. ' +
+ 'Run `npm install pg`, or unset database.url to use sqlite.',
+ { cause: e }
+ );
+ }
+}
+
+function fail(e, sql) {
+ const err = new Error(`postgres: ${e.message}\n sql: ${sql.trim().split('\n')[0]}`);
+ err.cause = e;
+ err.code = e.code;
+ throw err;
+}
+
+// Wraps either the pool or a single transaction client.
+function wrap(runner) {
+ function compile(sql, params) {
+ const compiled = compileCached(sql, 'postgres');
+ return { text: compiled.sql, values: bindParams(compiled.keys, params, sql) };
+ }
+
+ return {
+ dialect: 'postgres',
+
+ async all(sql, params = {}) {
+ const { text, values } = compile(sql, params);
+ try {
+ const res = await runner.query(text, values);
+ return res.rows;
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ async get(sql, params = {}) {
+ const { text, values } = compile(sql, params);
+ try {
+ const res = await runner.query(text, values);
+ return res.rows[0];
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ async run(sql, params = {}) {
+ const { text, values } = compile(sql, params);
+ try {
+ const res = await runner.query(text, values);
+ return { changes: res.rowCount ?? 0, lastInsertId: null };
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ // Raw DDL, no bind markers are interpreted.
+ async exec(sql) {
+ try {
+ await runner.query(sql);
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+ };
+}
+
+export async function openPostgres(cfg) {
+ const pg = await loadDriver();
+
+ // int8 (oid 20) would otherwise arrive as a string.
+ pg.types.setTypeParser(20, (v) => (v === null ? null : Number(v)));
+
+ const pool = new pg.Pool({
+ connectionString: cfg.database.url,
+ max: cfg.database.connection_limit,
+ });
+
+ const api = wrap(pool);
+
+ api.transaction = async function transaction(fn) {
+ const client = await pool.connect();
+ const tx = wrap(client);
+ let depth = 0;
+
+ tx.transaction = async (inner) => {
+ const name = `sp_${depth}`;
+ depth += 1;
+ await client.query(`SAVEPOINT ${name}`);
+ try {
+ const r = await inner(tx);
+ await client.query(`RELEASE SAVEPOINT ${name}`);
+ depth -= 1;
+ return r;
+ } catch (e) {
+ try {
+ await client.query(`ROLLBACK TO SAVEPOINT ${name}`);
+ } catch {
+ // Preserve the original failure.
+ }
+ depth -= 1;
+ throw e;
+ }
+ };
+
+ try {
+ await client.query('BEGIN');
+ const result = await fn(tx);
+ await client.query('COMMIT');
+ return result;
+ } catch (e) {
+ try {
+ await client.query('ROLLBACK');
+ } catch {
+ // Preserve the original failure.
+ }
+ throw e;
+ } finally {
+ client.release();
+ }
+ };
+
+ api.close = async () => {
+ await pool.end();
+ };
+
+ return api;
+}
diff --git a/src/lib/db/query.js b/src/lib/db/query.js
@@ -0,0 +1,205 @@
+// src/lib/db/query.js - named placeholder compilation
+//
+// Queries are written once, against every dialect, using {name} markers:
+//
+// db.get('SELECT * FROM jobs WHERE run_id = {run} AND state = {state}',
+// { run: runId, state: 'queued' })
+//
+// The marker is compiled to whatever the driver wants (? for sqlite and
+// mysql, $n for postgres) and the values are collected into a positional
+// array in order of appearance. Binding by name rather than by position
+// removes the class of bug where a query gains a clause and every later
+// argument silently shifts.
+//
+// Markers are only recognised in ordinary SQL text. Anything inside a string
+// literal, a quoted identifier, a dollar quoted block or a comment is passed
+// through untouched, so a JSON literal such as '{}' is safe. Outside those,
+// braces are escaped by doubling, so {{name}} emits the literal text {name}.
+//
+// Dotted names address nested values, so {actor.name} reads params.actor.name.
+// A key that exists verbatim on the object wins over path traversal, which
+// lets callers pass an already flattened object if they prefer.
+
+import { normalizeValue } from './params.js';
+
+const NAME = /^\{([A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z0-9_]+)*)\}/;
+
+// Per dialect lexing rules. Everything here is about deciding what to skip.
+export const RULES = {
+ sqlite: { hash: false, backtick: true, backslash: false, dollar: false, nestedBlock: false },
+ mysql: { hash: true, backtick: true, backslash: true, dollar: false, nestedBlock: false },
+ postgres: { hash: false, backtick: false, backslash: false, dollar: true, nestedBlock: true },
+};
+
+function placeholder(dialect, index) {
+ return dialect === 'postgres' ? `$${index}` : '?';
+}
+
+export function compileQuery(sql, dialect) {
+ const rules = RULES[dialect];
+ if (!rules) throw new Error(`unknown dialect: ${dialect}`);
+
+ let out = '';
+ const keys = [];
+ let i = 0;
+
+ while (i < sql.length) {
+ const c = sql[i];
+ const next = sql[i + 1];
+
+ if ((c === '-' && next === '-') || (rules.hash && c === '#')) {
+ const nl = sql.indexOf('\n', i);
+ const end = nl === -1 ? sql.length : nl;
+ out += sql.slice(i, end);
+ i = end;
+ continue;
+ }
+
+ if (c === '/' && next === '*') {
+ let depth = 1;
+ let j = i + 2;
+ while (j < sql.length && depth > 0) {
+ if (rules.nestedBlock && sql[j] === '/' && sql[j + 1] === '*') {
+ depth += 1;
+ j += 2;
+ } else if (sql[j] === '*' && sql[j + 1] === '/') {
+ depth -= 1;
+ j += 2;
+ } else {
+ j += 1;
+ }
+ }
+ out += sql.slice(i, j);
+ i = j;
+ continue;
+ }
+
+ if (c === "'" || c === '"' || (rules.backtick && c === '`')) {
+ const quote = c;
+ let j = i + 1;
+ while (j < sql.length) {
+ if (rules.backslash && quote !== '`' && sql[j] === '\\') {
+ j += 2;
+ continue;
+ }
+ if (sql[j] === quote) {
+ // A doubled quote is an escaped quote, not a terminator.
+ if (sql[j + 1] === quote) {
+ j += 2;
+ continue;
+ }
+ j += 1;
+ break;
+ }
+ j += 1;
+ }
+ out += sql.slice(i, j);
+ i = j;
+ continue;
+ }
+
+ if (rules.dollar && c === '$') {
+ const m = /^\$([A-Za-z_][A-Za-z0-9_]*)?\$/.exec(sql.slice(i));
+ if (m) {
+ const tag = m[0];
+ const end = sql.indexOf(tag, i + tag.length);
+ const stop = end === -1 ? sql.length : end + tag.length;
+ out += sql.slice(i, stop);
+ i = stop;
+ continue;
+ }
+ }
+
+ // Doubling escapes a brace, so {{name}} yields the literal text {name}.
+ if (c === '}' && next === '}') {
+ out += '}';
+ i += 2;
+ continue;
+ }
+
+ if (c === '{') {
+ if (next === '{') {
+ out += '{';
+ i += 2;
+ continue;
+ }
+ const m = NAME.exec(sql.slice(i));
+ if (!m) {
+ throw new Error(
+ `malformed bind marker at offset ${i} in query:\n ${sql.trim()}\n` +
+ 'Expected {name} or {outer.inner}; write {{ for a literal brace.'
+ );
+ }
+ keys.push(m[1]);
+ out += placeholder(dialect, keys.length);
+ i += m[0].length;
+ continue;
+ }
+
+ out += c;
+ i += 1;
+ }
+
+ return { sql: out, keys };
+}
+
+// Compilation is pure and queries come from source code, so the set is finite.
+const cache = new Map();
+
+export function compileCached(sql, dialect) {
+ const cacheKey = `${dialect}\u0000${sql}`;
+ let entry = cache.get(cacheKey);
+ if (!entry) {
+ entry = compileQuery(sql, dialect);
+ cache.set(cacheKey, entry);
+ }
+ return entry;
+}
+
+function resolve(params, key) {
+ // An exact property wins, so a pre-flattened object works unchanged.
+ if (params != null && Object.hasOwn(params, key)) {
+ return { found: true, value: params[key] };
+ }
+ if (!key.includes('.')) return { found: false };
+
+ let cur = params;
+ for (const part of key.split('.')) {
+ if (cur == null || typeof cur !== 'object') return { found: false };
+ if (!Object.hasOwn(cur, part)) return { found: false };
+ cur = cur[part];
+ }
+ return { found: true, value: cur };
+}
+
+export function bindParams(keys, params, sql) {
+ if (keys.length === 0) return [];
+
+ if (params == null || typeof params !== 'object' || Array.isArray(params)) {
+ throw new Error(
+ `bind parameters must be an object, got ${Array.isArray(params) ? 'an array' : typeof params}\n` +
+ ` query: ${sql.trim().split('\n')[0]}\n` +
+ ` expected keys: ${[...new Set(keys)].join(', ')}`
+ );
+ }
+
+ const missing = [];
+ const values = keys.map((key) => {
+ const hit = resolve(params, key);
+ if (!hit.found) {
+ missing.push(key);
+ return null;
+ }
+ return normalizeValue(hit.value, key);
+ });
+
+ if (missing.length > 0) {
+ throw new Error(
+ `missing bind parameter(s): ${[...new Set(missing)].join(', ')}\n` +
+ ` query: ${sql.trim().split('\n')[0]}\n` +
+ ` supplied: ${Object.keys(params).join(', ') || '(none)'}`
+ );
+ }
+
+ return values;
+}
diff --git a/src/lib/db/sqlite.js b/src/lib/db/sqlite.js
@@ -0,0 +1,124 @@
+// src/lib/db/sqlite.js - sqlite driver built on node:sqlite
+//
+// node:sqlite is synchronous. The methods here are async purely so that
+// callers can treat every dialect identically; there is no thread pool behind
+// them. Because each statement runs to completion before the event loop
+// turns, a transaction on this driver cannot interleave with another within
+// the same process.
+
+import { DatabaseSync } from 'node:sqlite';
+import { compileCached, bindParams } from './query.js';
+
+export function openSqlite(cfg) {
+ const db = new DatabaseSync(cfg.database.path);
+
+ // WAL lets the read-api read while the conductor writes.
+ db.exec('PRAGMA journal_mode = WAL');
+ db.exec('PRAGMA foreign_keys = ON');
+ db.exec('PRAGMA busy_timeout = 5000');
+ db.exec('PRAGMA synchronous = NORMAL');
+
+ const statements = new Map();
+
+ function prepare(sql) {
+ const compiled = compileCached(sql, 'sqlite');
+ let stmt = statements.get(compiled.sql);
+ if (!stmt) {
+ stmt = db.prepare(compiled.sql);
+ statements.set(compiled.sql, stmt);
+ }
+ return { stmt, keys: compiled.keys };
+ }
+
+ function fail(e, sql) {
+ const err = new Error(`sqlite: ${e.message}\n sql: ${sql.trim().split('\n')[0]}`);
+ err.cause = e;
+ err.code = e.code;
+ throw err;
+ }
+
+ // node:sqlite hands back null prototype objects. mysql2 and pg both return
+ // ordinary objects, so rows are converted here rather than leaving callers
+ // to discover the difference.
+ const toPlain = (row) => (row === undefined ? undefined : { ...row });
+
+ const api = {
+ dialect: 'sqlite',
+
+ async all(sql, params = {}) {
+ const { stmt, keys } = prepare(sql);
+ const values = bindParams(keys, params, sql);
+ try {
+ return stmt.all(...values).map(toPlain);
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ async get(sql, params = {}) {
+ const { stmt, keys } = prepare(sql);
+ const values = bindParams(keys, params, sql);
+ try {
+ return toPlain(stmt.get(...values));
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ async run(sql, params = {}) {
+ const { stmt, keys } = prepare(sql);
+ const values = bindParams(keys, params, sql);
+ try {
+ const r = stmt.run(...values);
+ return {
+ changes: Number(r.changes),
+ lastInsertId: r.lastInsertRowid === undefined ? null : Number(r.lastInsertRowid),
+ };
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ // Raw DDL, no bind markers are interpreted.
+ async exec(sql) {
+ try {
+ db.exec(sql);
+ } catch (e) {
+ fail(e, sql);
+ }
+ },
+
+ async close() {
+ statements.clear();
+ db.close();
+ },
+ };
+
+ // Nesting uses savepoints so that a helper which opens a transaction can be
+ // composed inside a larger one.
+ let depth = 0;
+ api.transaction = async function transaction(fn) {
+ const name = `sp_${depth}`;
+ if (depth === 0) db.exec('BEGIN IMMEDIATE');
+ else db.exec(`SAVEPOINT ${name}`);
+ depth += 1;
+ try {
+ const result = await fn(api);
+ depth -= 1;
+ if (depth === 0) db.exec('COMMIT');
+ else db.exec(`RELEASE ${name}`);
+ return result;
+ } catch (e) {
+ depth -= 1;
+ try {
+ if (depth === 0) db.exec('ROLLBACK');
+ else db.exec(`ROLLBACK TO ${name}`);
+ } catch {
+ // The original error is more useful than a rollback failure.
+ }
+ throw e;
+ }
+ };
+
+ return api;
+}
diff --git a/src/lib/ids.js b/src/lib/ids.js
@@ -0,0 +1,83 @@
+// src/lib/ids.js - identifier and token generation
+//
+// Identifiers are generated by the application rather than by the database,
+// because lastInsertId is not portable across the three supported dialects
+// and because a conductor needs an id before it has written anything.
+
+import crypto from 'node:crypto';
+
+// Crockford base32 without the ambiguous letters I, L, O and U. Case
+// insensitive and safe to read aloud or paste into a URL.
+const ALPHABET = '0123456789abcdefghjkmnpqrstvwxyz';
+
+export function randomId(length = 24) {
+ // Rejection free: 256 is not a multiple of 32, so mask to 5 bits.
+ const bytes = crypto.randomBytes(length);
+ let out = '';
+ for (let i = 0; i < length; i += 1) out += ALPHABET[bytes[i] & 31];
+ return out;
+}
+
+// Sortable by creation time, which keeps run listings stable without relying
+// on a clock skewed created_at. 8 chars of millisecond timestamp in base32
+// followed by 12 random chars.
+export function timeOrderedId(now = Date.now()) {
+ let stamp = '';
+ let remaining = now;
+ for (let i = 0; i < 8; i += 1) {
+ stamp = ALPHABET[remaining % 32] + stamp;
+ remaining = Math.floor(remaining / 32);
+ }
+ return stamp + randomId(12);
+}
+
+export const newRunId = () => timeOrderedId();
+export const newProjectId = () => randomId(16);
+export const newArtifactId = () => randomId(24);
+export const newUserId = () => randomId(16);
+export const newWorkerTokenId = () => randomId(16);
+
+// Job ids stay human readable, since they appear in logs, storage keys and
+// URLs. The jobs.id column is 96 characters, so an unusually long job name
+// falls back to a truncated form with a hash suffix rather than being
+// rejected or silently colliding.
+export const JOB_ID_MAX = 96;
+
+export function jobId(runId, jobName) {
+ const direct = `${runId}:${jobName}`;
+ if (direct.length <= JOB_ID_MAX) return direct;
+
+ const digest = crypto.createHash('sha256').update(jobName).digest('hex').slice(0, 12);
+ const room = JOB_ID_MAX - runId.length - 1 - 1 - digest.length;
+ if (room < 1) {
+ throw new Error(`run id ${runId} leaves no room for a job id`);
+ }
+ return `${runId}:${jobName.slice(0, room)}~${digest}`;
+}
+
+// Worker and session tokens. The plaintext is shown once and only its hash
+// is stored, so a database leak does not yield usable credentials.
+export function newToken() {
+ return crypto.randomBytes(32).toString('hex');
+}
+
+export function hashToken(token) {
+ return crypto.createHash('sha256').update(String(token)).digest('hex');
+}
+
+// Compares two hex digests without leaking position through timing.
+export function safeEqualHex(a, b) {
+ if (typeof a !== 'string' || typeof b !== 'string' || a.length !== b.length) return false;
+ return crypto.timingSafeEqual(Buffer.from(a, 'hex'), Buffer.from(b, 'hex'));
+}
+
+// A project id derived from a human supplied name, for the common case where
+// an operator wants a readable slug rather than a random string.
+export function slugify(input, fallbackLength = 16) {
+ const slug = String(input)
+ .toLowerCase()
+ .replace(/[^a-z0-9]+/g, '-')
+ .replace(/^-+|-+$/g, '')
+ .slice(0, 64);
+ return slug || randomId(fallbackLength);
+}
diff --git a/src/lib/secretbox.js b/src/lib/secretbox.js
@@ -0,0 +1,87 @@
+// src/lib/secretbox.js - encryption at rest for stored secrets
+//
+// Used for project trigger secrets and project variables. When
+// secrets.encryption_key is configured, values are sealed with AES-256-GCM.
+// When it is not, values are stored base64 encoded and clearly marked as
+// plaintext, so that an operator can tell at a glance which rows are
+// protected and a later key rollout can find what needs upgrading.
+//
+// Stored format is a dotted, self describing string:
+// v1.<iv>.<tag>.<ciphertext> sealed, all parts base64
+// plain.<value> not encrypted, value base64
+//
+// The optional aad argument binds a ciphertext to its location, so a row
+// copied into a different project or variable name fails to open.
+
+import crypto from 'node:crypto';
+
+const IV_BYTES = 12;
+
+export function createSecretBox(key) {
+ const enabled = Boolean(key);
+ if (enabled && (!Buffer.isBuffer(key) || key.length !== 32)) {
+ throw new Error('secret box key must be a 32 byte Buffer');
+ }
+
+ return {
+ enabled,
+
+ seal(plaintext, aad) {
+ if (typeof plaintext !== 'string') throw new Error('secret value must be a string');
+ if (!enabled) return `plain.${Buffer.from(plaintext, 'utf8').toString('base64')}`;
+
+ const iv = crypto.randomBytes(IV_BYTES);
+ const cipher = crypto.createCipheriv('aes-256-gcm', key, iv);
+ if (aad) cipher.setAAD(Buffer.from(aad, 'utf8'));
+ const ct = Buffer.concat([cipher.update(plaintext, 'utf8'), cipher.final()]);
+ const tag = cipher.getAuthTag();
+ return `v1.${iv.toString('base64')}.${tag.toString('base64')}.${ct.toString('base64')}`;
+ },
+
+ open(stored, aad) {
+ if (typeof stored !== 'string' || stored.length === 0) {
+ throw new Error('stored secret is empty');
+ }
+
+ if (stored.startsWith('plain.')) {
+ return Buffer.from(stored.slice(6), 'base64').toString('utf8');
+ }
+
+ if (!stored.startsWith('v1.')) {
+ throw new Error('stored secret has an unrecognised format');
+ }
+ if (!enabled) {
+ throw new Error(
+ 'stored secret is encrypted but secrets.encryption_key is not configured; ' +
+ 'set it to the key this value was sealed with'
+ );
+ }
+
+ const parts = stored.split('.');
+ if (parts.length !== 4) throw new Error('stored secret is malformed');
+ const [, ivB64, tagB64, ctB64] = parts;
+
+ const decipher = crypto.createDecipheriv('aes-256-gcm', key, Buffer.from(ivB64, 'base64'));
+ decipher.setAuthTag(Buffer.from(tagB64, 'base64'));
+ if (aad) decipher.setAAD(Buffer.from(aad, 'utf8'));
+ try {
+ return Buffer.concat([
+ decipher.update(Buffer.from(ctB64, 'base64')),
+ decipher.final(),
+ ]).toString('utf8');
+ } catch (e) {
+ throw new Error(
+ 'stored secret failed to decrypt; the encryption key may have changed, ' +
+ 'or the value was moved between rows',
+ { cause: e }
+ );
+ }
+ },
+
+ // True when a value is stored without encryption, so an admin endpoint
+ // can report what a key rollout would need to re-seal.
+ isPlaintext(stored) {
+ return typeof stored === 'string' && stored.startsWith('plain.');
+ },
+ };
+}
diff --git a/src/lib/storage/index.js b/src/lib/storage/index.js
@@ -0,0 +1,28 @@
+// src/lib/storage/index.js - object storage backend selection
+//
+// S3 is used when storage.s3.bucket is configured, otherwise objects go to
+// the local filesystem. Both backends expose the same surface:
+//
+// put(key, body, opts) -> { key, size, sha256 }
+// get(key, { range }) -> { key, size, stream }, throws StorageNotFound
+// head(key) -> { key, size, modified } or null
+// delete(key) -> resolves even when the key is absent
+// presign(key, opts) -> absolute URL, or null when unsupported
+//
+// Callers must handle a null from presign by streaming get() themselves,
+// since the local backend has no URL to hand out.
+
+import { createLocalStorage } from './local.js';
+import { createS3Storage } from './s3.js';
+
+export { StorageNotFound, assertKey, sanitizeRelativePath } from './key.js';
+
+export function createStorage(cfg) {
+ return cfg.storage.driver === 's3' ? createS3Storage(cfg) : createLocalStorage(cfg);
+}
+
+// Key layout, kept in one place so the read-api and the conductor agree.
+export const keys = {
+ artifact: (runId, jobId, relPath) => `artifacts/${runId}/${jobId}/${relPath}`,
+ log: (runId, jobId) => `logs/${runId}/${jobId}.log`,
+};
diff --git a/src/lib/storage/key.js b/src/lib/storage/key.js
@@ -0,0 +1,52 @@
+// src/lib/storage/key.js - object key validation shared by both backends
+//
+// Keys are slash separated and always relative. Validation is centralised
+// because a key frequently originates from a job artifact path, which is
+// attacker controlled for any worker or repository that is not fully trusted.
+
+export class StorageNotFound extends Error {
+ constructor(key) {
+ super(`object not found: ${key}`);
+ this.name = 'StorageNotFound';
+ this.code = 'ENOENT';
+ this.key = key;
+ }
+}
+
+export function assertKey(key) {
+ if (typeof key !== 'string' || key.length === 0) {
+ throw new Error('storage key must be a non-empty string');
+ }
+ if (key.length > 1024) {
+ throw new Error(`storage key exceeds 1024 characters: ${key.slice(0, 64)}...`);
+ }
+ if (key.startsWith('/')) {
+ throw new Error(`storage key must be relative: ${key}`);
+ }
+ if (key.includes('\\')) {
+ throw new Error(`storage key must use forward slashes: ${key}`);
+ }
+ // A NUL byte truncates the path in some syscalls.
+ if (key.includes('\0')) {
+ throw new Error('storage key contains a NUL byte');
+ }
+ for (const segment of key.split('/')) {
+ if (segment === '' || segment === '.' || segment === '..') {
+ throw new Error(`storage key contains an invalid path segment: ${key}`);
+ }
+ }
+ return key;
+}
+
+// Normalizes a path reported by a worker into a safe key suffix. Returns null
+// when nothing usable remains, so the caller can reject the upload.
+export function sanitizeRelativePath(input) {
+ if (typeof input !== 'string') return null;
+ const parts = input
+ .replace(/\\/g, '/')
+ .split('/')
+ .filter((p) => p !== '' && p !== '.' && p !== '..' && !p.includes('\0'));
+ if (parts.length === 0) return null;
+ const joined = parts.join('/');
+ return joined.length > 512 ? null : joined;
+}
diff --git a/src/lib/storage/local.js b/src/lib/storage/local.js
@@ -0,0 +1,111 @@
+// src/lib/storage/local.js - filesystem object storage
+//
+// The default backend, used whenever no S3 bucket is configured. Objects are
+// plain files under storage.path, so artifacts remain inspectable with
+// ordinary tools and a single node install needs nothing else running.
+
+import fs from 'node:fs';
+import fsp from 'node:fs/promises';
+import path from 'node:path';
+import crypto from 'node:crypto';
+import { pipeline } from 'node:stream/promises';
+import { Readable } from 'node:stream';
+import { StorageNotFound, assertKey } from './key.js';
+
+export function createLocalStorage(cfg) {
+ const root = cfg.storage.path;
+
+ function resolve(key) {
+ assertKey(key);
+ const full = path.resolve(root, key);
+ // Defence in depth: assertKey already rejects traversal, but a symlinked
+ // storage root should not become an escape either.
+ if (full !== root && !full.startsWith(root + path.sep)) {
+ throw new Error(`storage key escapes the storage root: ${key}`);
+ }
+ return full;
+ }
+
+ return {
+ driver: 'local',
+
+ async put(key, body, opts = {}) {
+ const full = resolve(key);
+ await fsp.mkdir(path.dirname(full), { recursive: true });
+
+ // Write to a sibling temp file and rename, so a reader never observes a
+ // partially written object.
+ const tmp = `${full}.${crypto.randomBytes(6).toString('hex')}.part`;
+ const hash = crypto.createHash('sha256');
+ let size = 0;
+
+ try {
+ if (Buffer.isBuffer(body) || typeof body === 'string') {
+ const buf = Buffer.isBuffer(body) ? body : Buffer.from(body, 'utf8');
+ hash.update(buf);
+ size = buf.length;
+ await fsp.writeFile(tmp, buf);
+ } else {
+ const source = body instanceof Readable ? body : Readable.from(body);
+ const out = fs.createWriteStream(tmp);
+ source.on('data', (chunk) => {
+ hash.update(chunk);
+ size += chunk.length;
+ });
+ await pipeline(source, out);
+ }
+ await fsp.rename(tmp, full);
+ } catch (e) {
+ await fsp.rm(tmp, { force: true });
+ throw e;
+ }
+
+ if (opts.size !== undefined && opts.size !== size) {
+ await fsp.rm(full, { force: true });
+ throw new Error(`storage put size mismatch for ${key}: declared ${opts.size}, wrote ${size}`);
+ }
+
+ return { key, size, sha256: hash.digest('hex') };
+ },
+
+ async get(key, opts = {}) {
+ const full = resolve(key);
+ let stat;
+ try {
+ stat = await fsp.stat(full);
+ } catch (e) {
+ if (e.code === 'ENOENT') throw new StorageNotFound(key);
+ throw e;
+ }
+
+ const range = opts.range;
+ const start = range?.start ?? 0;
+ const end = range?.end ?? undefined;
+
+ return {
+ key,
+ size: stat.size,
+ stream: fs.createReadStream(full, end === undefined ? { start } : { start, end }),
+ };
+ },
+
+ async head(key) {
+ try {
+ const stat = await fsp.stat(resolve(key));
+ return { key, size: stat.size, modified: stat.mtimeMs };
+ } catch (e) {
+ if (e.code === 'ENOENT') return null;
+ throw e;
+ }
+ },
+
+ async delete(key) {
+ await fsp.rm(resolve(key), { force: true });
+ },
+
+ // Local files cannot be handed out directly; callers stream them instead.
+ async presign() {
+ return null;
+ },
+ };
+}
diff --git a/src/lib/storage/s3.js b/src/lib/storage/s3.js
@@ -0,0 +1,162 @@
+// src/lib/storage/s3.js - S3 compatible object storage
+//
+// Engaged as soon as storage.s3.bucket is configured. Tested against Garage
+// and MinIO, which both need path style addressing; set force_path_style to
+// false for AWS S3 proper.
+//
+// Requests go out over fetch with SigV4 headers. Uploads of a known Buffer
+// are signed over their content hash; streamed uploads use UNSIGNED-PAYLOAD,
+// which every S3 implementation accepts and which avoids buffering an entire
+// artifact in memory just to hash it.
+
+import crypto from 'node:crypto';
+import { Readable } from 'node:stream';
+import { signRequest, presignUrl, sha256Hex, uriEncode, UNSIGNED_PAYLOAD } from './sigv4.js';
+import { StorageNotFound, assertKey } from './key.js';
+
+export function createS3Storage(cfg) {
+ const s3 = cfg.storage.s3;
+ const endpoint = new URL(s3.endpoint);
+ const creds = {
+ region: s3.region,
+ accessKeyId: s3.access_key_id,
+ secretAccessKey: s3.secret_access_key,
+ };
+
+ // Any path prefix on the endpoint is preserved, so an S3 gateway mounted
+ // under a subpath keeps working.
+ const prefix = endpoint.pathname.replace(/\/+$/, '');
+
+ function objectUrl(key) {
+ assertKey(key);
+ const url = new URL(endpoint.toString());
+ // The signer treats the path as final, so encode it here and exactly
+ // once. Slashes stay literal to keep the key hierarchy intact.
+ const encoded = uriEncode(key, false);
+ if (s3.force_path_style) {
+ url.pathname = `${prefix}/${s3.bucket}/${encoded}`;
+ } else {
+ url.host = `${s3.bucket}.${endpoint.host}`;
+ url.pathname = `${prefix}/${encoded}`;
+ }
+ return url;
+ }
+
+ async function send(method, key, { headers = {}, body, payloadHash } = {}) {
+ const url = objectUrl(key);
+ const signed = signRequest({
+ method,
+ url,
+ headers,
+ payloadHash: payloadHash ?? sha256Hex(''),
+ ...creds,
+ });
+
+ const init = { method, headers: signed };
+ if (body !== undefined) {
+ init.body = body;
+ // Required by undici when streaming a request body.
+ if (body instanceof ReadableStream) init.duplex = 'half';
+ }
+ return fetch(url, init);
+ }
+
+ async function errorFrom(res, action, key) {
+ let detail = '';
+ try {
+ detail = (await res.text()).slice(0, 512);
+ } catch {
+ // The status alone is still useful.
+ }
+ return new Error(`s3 ${action} failed for ${key}: ${res.status} ${res.statusText}\n${detail}`);
+ }
+
+ return {
+ driver: 's3',
+
+ async put(key, body, opts = {}) {
+ const headers = {};
+ if (opts.contentType) headers['content-type'] = opts.contentType;
+
+ let payload;
+ let payloadHash;
+ let size = opts.size;
+ let sha256 = opts.sha256;
+
+ if (Buffer.isBuffer(body) || typeof body === 'string') {
+ const buf = Buffer.isBuffer(body) ? body : Buffer.from(body, 'utf8');
+ payload = buf;
+ sha256 = crypto.createHash('sha256').update(buf).digest('hex');
+ payloadHash = sha256;
+ size = buf.length;
+ headers['content-length'] = String(size);
+ } else {
+ if (size === undefined) {
+ throw new Error(
+ `s3 put of a stream needs an explicit size for ${key}; ` +
+ 'buffer the object first if the length is unknown'
+ );
+ }
+ payloadHash = UNSIGNED_PAYLOAD;
+ headers['content-length'] = String(size);
+ payload = body instanceof Readable ? Readable.toWeb(body) : body;
+ }
+
+ const res = await send('PUT', key, { headers, body: payload, payloadHash });
+ if (!res.ok) throw await errorFrom(res, 'put', key);
+ await res.arrayBuffer();
+
+ return { key, size, sha256: sha256 ?? null };
+ },
+
+ async get(key, opts = {}) {
+ const headers = {};
+ if (opts.range) {
+ const { start = 0, end } = opts.range;
+ headers.range = `bytes=${start}-${end === undefined ? '' : end}`;
+ }
+
+ const res = await send('GET', key, { headers });
+ if (res.status === 404) throw new StorageNotFound(key);
+ if (!res.ok) throw await errorFrom(res, 'get', key);
+
+ const len = res.headers.get('content-length');
+ return {
+ key,
+ size: len === null ? null : Number(len),
+ stream: Readable.fromWeb(res.body),
+ };
+ },
+
+ async head(key) {
+ const res = await send('HEAD', key);
+ if (res.status === 404) return null;
+ if (!res.ok) throw await errorFrom(res, 'head', key);
+ const len = res.headers.get('content-length');
+ const modified = res.headers.get('last-modified');
+ return {
+ key,
+ size: len === null ? null : Number(len),
+ modified: modified ? Date.parse(modified) : null,
+ };
+ },
+
+ async delete(key) {
+ const res = await send('DELETE', key);
+ // A missing object is already in the desired state.
+ if (!res.ok && res.status !== 404) throw await errorFrom(res, 'delete', key);
+ await res.arrayBuffer();
+ },
+
+ // Lets the read-api redirect a client straight at the object store
+ // instead of proxying the bytes.
+ async presign(key, opts = {}) {
+ return presignUrl({
+ method: opts.method || 'GET',
+ url: objectUrl(key),
+ expires: opts.expires ?? 3600,
+ ...creds,
+ });
+ },
+ };
+}
diff --git a/src/lib/storage/sigv4.js b/src/lib/storage/sigv4.js
@@ -0,0 +1,187 @@
+// src/lib/storage/sigv4.js - AWS Signature Version 4 signing
+//
+// Only what an S3 compatible object store needs: PUT, GET, HEAD, DELETE and
+// presigned URLs. Written against node:crypto rather than pulling in the AWS
+// SDK, which would add roughly a hundred transitive packages for four verbs.
+//
+// Verified against the published AWS test vectors, see test/sigv4.test.js.
+
+import crypto from 'node:crypto';
+
+const ALGORITHM = 'AWS4-HMAC-SHA256';
+export const UNSIGNED_PAYLOAD = 'UNSIGNED-PAYLOAD';
+export const EMPTY_SHA256 = crypto.createHash('sha256').update('').digest('hex');
+
+function hmac(key, data) {
+ return crypto.createHmac('sha256', key).update(data, 'utf8').digest();
+}
+
+export function sha256Hex(data) {
+ return crypto.createHash('sha256').update(data).digest('hex');
+}
+
+// RFC 3986 encoding. encodeURIComponent leaves ! ' ( ) * alone, which AWS
+// expects to be percent encoded.
+export function uriEncode(str, encodeSlash = true) {
+ let out = '';
+ for (const ch of String(str)) {
+ if (/[A-Za-z0-9\-._~]/.test(ch)) {
+ out += ch;
+ } else if (ch === '/') {
+ out += encodeSlash ? '%2F' : '/';
+ } else {
+ for (const byte of Buffer.from(ch, 'utf8')) {
+ out += `%${byte.toString(16).toUpperCase().padStart(2, '0')}`;
+ }
+ }
+ }
+ return out;
+}
+
+// 20150830T123600Z and 20150830
+export function amzDate(date = new Date()) {
+ const iso = date.toISOString().replace(/[:-]|\.\d{3}/g, '');
+ return { amz: iso, stamp: iso.slice(0, 8) };
+}
+
+function canonicalQuery(searchParams) {
+ const pairs = [];
+ for (const [k, v] of searchParams) pairs.push([uriEncode(k), uriEncode(v)]);
+ // Sort by encoded key, then encoded value.
+ pairs.sort((a, b) => (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : a[1] < b[1] ? -1 : a[1] > b[1] ? 1 : 0));
+ return pairs.map(([k, v]) => `${k}=${v}`).join('&');
+}
+
+// The canonical URI is the request path exactly as it goes on the wire.
+// Callers must therefore hand in a URL whose path is already RFC 3986
+// encoded, which is what uriEncode(key, false) produces. Encoding here
+// instead would double encode, turning a %20 into %25 20 and yielding
+// SignatureDoesNotMatch for any key containing a space or a plus.
+//
+// Note this is correct for S3 specifically. Other AWS services expect the
+// path to be normalized and encoded a second time.
+function canonicalPath(pathname) {
+ return pathname === '' ? '/' : pathname;
+}
+
+function signingKey(secretAccessKey, stamp, region, service) {
+ const kDate = hmac(`AWS4${secretAccessKey}`, stamp);
+ const kRegion = hmac(kDate, region);
+ const kService = hmac(kRegion, service);
+ return hmac(kService, 'aws4_request');
+}
+
+function buildCanonical({ method, url, headers, payloadHash }) {
+ const lowered = Object.entries(headers)
+ .map(([k, v]) => [k.toLowerCase(), String(v).trim().replace(/\s+/g, ' ')])
+ .sort((a, b) => (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0));
+
+ const canonicalHeaders = lowered.map(([k, v]) => `${k}:${v}\n`).join('');
+ const signedHeaders = lowered.map(([k]) => k).join(';');
+
+ const canonicalRequest = [
+ method.toUpperCase(),
+ canonicalPath(url.pathname),
+ canonicalQuery(url.searchParams),
+ canonicalHeaders,
+ signedHeaders,
+ payloadHash,
+ ].join('\n');
+
+ return { canonicalRequest, signedHeaders };
+}
+
+// Signs a request and returns the headers to send, including Authorization.
+export function signRequest(opts) {
+ const {
+ method,
+ url,
+ headers = {},
+ payloadHash = EMPTY_SHA256,
+ region,
+ service = 's3',
+ accessKeyId,
+ secretAccessKey,
+ date = new Date(),
+ } = opts;
+
+ const target = url instanceof URL ? url : new URL(url);
+ const { amz, stamp } = amzDate(date);
+
+ const signedHeaderSet = {
+ ...headers,
+ host: target.host,
+ 'x-amz-date': amz,
+ };
+ if (service === 's3') signedHeaderSet['x-amz-content-sha256'] = payloadHash;
+
+ const { canonicalRequest, signedHeaders } = buildCanonical({
+ method,
+ url: target,
+ headers: signedHeaderSet,
+ payloadHash,
+ });
+
+ const scope = `${stamp}/${region}/${service}/aws4_request`;
+ const stringToSign = [ALGORITHM, amz, scope, sha256Hex(canonicalRequest)].join('\n');
+ const signature = crypto
+ .createHmac('sha256', signingKey(secretAccessKey, stamp, region, service))
+ .update(stringToSign, 'utf8')
+ .digest('hex');
+
+ return {
+ ...signedHeaderSet,
+ authorization:
+ `${ALGORITHM} Credential=${accessKeyId}/${scope}, ` +
+ `SignedHeaders=${signedHeaders}, Signature=${signature}`,
+ };
+}
+
+// Produces a presigned URL, where the signature travels in the query string
+// and no Authorization header is needed.
+export function presignUrl(opts) {
+ const {
+ method = 'GET',
+ url,
+ headers = {},
+ expires = 3600,
+ region,
+ service = 's3',
+ accessKeyId,
+ secretAccessKey,
+ date = new Date(),
+ } = opts;
+
+ const target = new URL(url instanceof URL ? url.toString() : url);
+ const { amz, stamp } = amzDate(date);
+ const scope = `${stamp}/${region}/${service}/aws4_request`;
+
+ // Only host is signed, so the URL works from any client.
+ const signedHeaderSet = { ...headers, host: target.host };
+ const signedHeaders = Object.keys(signedHeaderSet)
+ .map((k) => k.toLowerCase())
+ .sort()
+ .join(';');
+
+ target.searchParams.set('X-Amz-Algorithm', ALGORITHM);
+ target.searchParams.set('X-Amz-Credential', `${accessKeyId}/${scope}`);
+ target.searchParams.set('X-Amz-Date', amz);
+ target.searchParams.set('X-Amz-Expires', String(expires));
+ target.searchParams.set('X-Amz-SignedHeaders', signedHeaders);
+
+ const { canonicalRequest } = buildCanonical({
+ method,
+ url: target,
+ headers: signedHeaderSet,
+ payloadHash: UNSIGNED_PAYLOAD,
+ });
+
+ const stringToSign = [ALGORITHM, amz, scope, sha256Hex(canonicalRequest)].join('\n');
+ const signature = crypto
+ .createHmac('sha256', signingKey(secretAccessKey, stamp, region, service))
+ .update(stringToSign, 'utf8')
+ .digest('hex');
+
+ target.searchParams.set('X-Amz-Signature', signature);
+ return target.toString();
+}
diff --git a/test/ascii.test.js b/test/ascii.test.js
@@ -0,0 +1,82 @@
+// test/ascii.test.js - enforces the repository wide ASCII rule
+//
+// Every source and documentation file must be plain ASCII. This is a hard
+// project requirement rather than a style preference, so it is tested rather
+// than left to review. Smart quotes, dashes and arrows pasted in from a
+// browser are the usual cause of a failure here.
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
+
+const SKIP_DIRS = new Set(['.git', 'node_modules', 'data', 'tmp']);
+const CHECK_EXT = new Set([
+ '.js', '.mjs', '.cjs', '.json', '.md', '.sql', '.yml', '.yaml',
+ '.html', '.css', '.sh', '.txt',
+]);
+const CHECK_NAMES = new Set(['.editorconfig', '.gitignore', 'Dockerfile', 'post-receive']);
+
+async function* walk(dir) {
+ for (const entry of await fs.readdir(dir, { withFileTypes: true })) {
+ if (entry.isDirectory()) {
+ if (SKIP_DIRS.has(entry.name)) continue;
+ yield* walk(path.join(dir, entry.name));
+ } else if (entry.isFile()) {
+ yield path.join(dir, entry.name);
+ }
+ }
+}
+
+function describeChar(ch) {
+ const code = ch.codePointAt(0);
+ return `U+${code.toString(16).toUpperCase().padStart(4, '0')} ${JSON.stringify(ch)}`;
+}
+
+test('every tracked text file is pure ASCII', async () => {
+ const offences = [];
+
+ for await (const file of walk(ROOT)) {
+ const ext = path.extname(file);
+ const base = path.basename(file);
+ if (!CHECK_EXT.has(ext) && !CHECK_NAMES.has(base)) continue;
+
+ const text = await fs.readFile(file, 'utf8');
+ const lines = text.split('\n');
+ for (let i = 0; i < lines.length; i += 1) {
+ for (const ch of lines[i]) {
+ const code = ch.codePointAt(0);
+ // Tab, and the printable range. Newlines are already stripped.
+ if (code === 9 || (code >= 32 && code <= 126)) continue;
+ offences.push(`${path.relative(ROOT, file)}:${i + 1}: ${describeChar(ch)}`);
+ break;
+ }
+ }
+ }
+
+ assert.deepEqual(offences, [], `non-ASCII characters found:\n${offences.join('\n')}`);
+});
+
+test('no file uses CRLF line endings or trailing whitespace', async () => {
+ const offences = [];
+
+ for await (const file of walk(ROOT)) {
+ const ext = path.extname(file);
+ const base = path.basename(file);
+ if (!CHECK_EXT.has(ext) && !CHECK_NAMES.has(base)) continue;
+
+ const text = await fs.readFile(file, 'utf8');
+ const rel = path.relative(ROOT, file);
+ if (text.includes('\r')) offences.push(`${rel}: contains a carriage return`);
+ const lines = text.split('\n');
+ for (let i = 0; i < lines.length; i += 1) {
+ if (/[ \t]+$/.test(lines[i])) offences.push(`${rel}:${i + 1}: trailing whitespace`);
+ }
+ if (text.length > 0 && !text.endsWith('\n')) offences.push(`${rel}: missing final newline`);
+ }
+
+ assert.deepEqual(offences, [], `formatting problems found:\n${offences.join('\n')}`);
+});
diff --git a/test/config.test.js b/test/config.test.js
@@ -0,0 +1,166 @@
+// test/config.test.js - configuration loading, precedence and validation
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { loadConfig, stateDirs, parseKey } from '../src/lib/config.js';
+
+// loadConfig reads process.env, so every case runs with a clean slate.
+async function withEnv(env, fn) {
+ const saved = {};
+ for (const key of Object.keys(process.env)) {
+ if (key.startsWith('CONDUCTOR_')) {
+ saved[key] = process.env[key];
+ delete process.env[key];
+ }
+ }
+ Object.assign(process.env, env);
+ try {
+ return await fn();
+ } finally {
+ for (const key of Object.keys(process.env)) {
+ if (key.startsWith('CONDUCTOR_')) delete process.env[key];
+ }
+ Object.assign(process.env, saved);
+ }
+}
+
+async function tempDir() {
+ return fs.mkdtemp(path.join(os.tmpdir(), 'conductor-cfg-'));
+}
+
+test('defaults select sqlite, local storage and local auth', async () => {
+ await withEnv({}, () => {
+ const cfg = loadConfig();
+ assert.equal(cfg.database.dialect, 'sqlite');
+ assert.equal(cfg.storage.driver, 'local');
+ assert.equal(cfg.auth.mode, 'local');
+ assert.equal(cfg.server.port, 8080);
+ });
+});
+
+test('configuring an s3 bucket switches the storage driver', async () => {
+ await withEnv({
+ CONDUCTOR_S3_ENDPOINT: 'http://127.0.0.1:3900',
+ CONDUCTOR_S3_BUCKET: 'ci',
+ CONDUCTOR_S3_ACCESS_KEY_ID: 'k',
+ CONDUCTOR_S3_SECRET_ACCESS_KEY: 's',
+ }, () => {
+ assert.equal(loadConfig().storage.driver, 's3');
+ });
+});
+
+test('a partial s3 configuration is rejected rather than half applied', async () => {
+ await withEnv({ CONDUCTOR_S3_BUCKET: 'ci' }, () => {
+ assert.throws(() => loadConfig(), /partially configured/);
+ });
+});
+
+test('configuring an oidc issuer switches the auth mode', async () => {
+ await withEnv({ CONDUCTOR_OIDC_ISSUER: 'https://idp.example.com/realms/ci' }, () => {
+ assert.equal(loadConfig().auth.mode, 'oidc');
+ });
+});
+
+test('a database url selects its dialect', async () => {
+ await withEnv({ CONDUCTOR_DATABASE_URL: 'postgres://u:p@h:5432/ci' }, () => {
+ assert.equal(loadConfig().database.dialect, 'postgres');
+ });
+ await withEnv({ CONDUCTOR_DATABASE_URL: 'mysql://u:p@h:3306/ci' }, () => {
+ assert.equal(loadConfig().database.dialect, 'mysql');
+ });
+});
+
+test('an unsupported database scheme is reported clearly', async () => {
+ await withEnv({ CONDUCTOR_DATABASE_URL: 'mongodb://h/ci' }, () => {
+ assert.throws(() => loadConfig(), /unsupported scheme/);
+ });
+});
+
+test('a sqlite url collapses into database.path', async () => {
+ await withEnv({ CONDUCTOR_DATABASE_URL: 'sqlite:///var/lib/conductor/c.db' }, () => {
+ const cfg = loadConfig();
+ assert.equal(cfg.database.dialect, 'sqlite');
+ assert.equal(cfg.database.url, null);
+ assert.equal(cfg.database.path, '/var/lib/conductor/c.db');
+ });
+});
+
+test('environment variables override the config file', async () => {
+ const dir = await tempDir();
+ const file = path.join(dir, 'conductor.yaml');
+ await fs.writeFile(file, 'server:\n port: 9999\n public_url: https://ci.example.com\n');
+
+ await withEnv({ CONDUCTOR_CONFIG: file, CONDUCTOR_PORT: '7777' }, () => {
+ const cfg = loadConfig();
+ assert.equal(cfg.server.port, 7777);
+ assert.equal(cfg.server.public_url, 'https://ci.example.com');
+ });
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('relative paths resolve against the config file directory', async () => {
+ const dir = await tempDir();
+ const file = path.join(dir, 'conductor.yaml');
+ await fs.writeFile(file, 'database:\n path: ./state/c.db\n');
+
+ await withEnv({ CONDUCTOR_CONFIG: file }, () => {
+ assert.equal(loadConfig().database.path, path.join(dir, 'state/c.db'));
+ });
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('a missing explicit config file is an error, a missing default is not', async () => {
+ await withEnv({ CONDUCTOR_CONFIG: '/nonexistent/conductor.yaml' }, () => {
+ assert.throws(() => loadConfig(), /config file not found/);
+ });
+ await withEnv({}, () => {
+ assert.doesNotThrow(() => loadConfig());
+ });
+});
+
+test('a heartbeat timeout below the reap interval is rejected', async () => {
+ const dir = await tempDir();
+ const file = path.join(dir, 'conductor.yaml');
+ await fs.writeFile(file, 'scheduler:\n heartbeat_timeout: 10\n reap_interval: 30\n');
+
+ await withEnv({ CONDUCTOR_CONFIG: file }, () => {
+ assert.throws(() => loadConfig(), /must be greater than/);
+ });
+ await fs.rm(dir, { recursive: true, force: true });
+});
+
+test('a non-integer port is rejected with the variable named', async () => {
+ await withEnv({ CONDUCTOR_PORT: 'http' }, () => {
+ assert.throws(() => loadConfig(), /CONDUCTOR_PORT/);
+ });
+});
+
+test('parseKey accepts hex and base64, and rejects the wrong length', () => {
+ assert.equal(parseKey('a'.repeat(64), 'k').length, 32);
+ assert.equal(parseKey(Buffer.alloc(32).toString('base64'), 'k').length, 32);
+ assert.throws(() => parseKey('abcd', 'secrets.encryption_key'), /must decode to 32 bytes/);
+});
+
+test('stateDirs omits the storage directory when s3 is in use', async () => {
+ await withEnv({}, () => {
+ assert.ok(stateDirs(loadConfig()).some((d) => d.endsWith('storage')));
+ });
+ await withEnv({
+ CONDUCTOR_S3_ENDPOINT: 'http://127.0.0.1:3900',
+ CONDUCTOR_S3_BUCKET: 'ci',
+ CONDUCTOR_S3_ACCESS_KEY_ID: 'k',
+ CONDUCTOR_S3_SECRET_ACCESS_KEY: 's',
+ }, () => {
+ assert.ok(!stateDirs(loadConfig()).some((d) => d.endsWith('storage')));
+ });
+});
+
+test('the loaded config is frozen', async () => {
+ await withEnv({}, () => {
+ const cfg = loadConfig();
+ assert.throws(() => { cfg.server.port = 1; }, TypeError);
+ });
+});
diff --git a/test/db.test.js b/test/db.test.js
@@ -0,0 +1,188 @@
+// test/db.test.js - sqlite driver and migration runner
+//
+// The mysql and postgres drivers share the query compiler and parameter
+// handling that is exercised here. Their wire behaviour is covered by the
+// integration suite, which needs a live server.
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { openDatabase, runMigrations } from '../src/lib/db/index.js';
+import { splitStatements } from '../src/lib/db/migrate.js';
+import { dialectFromUrl } from '../src/lib/db/dialect.js';
+
+const quiet = () => {};
+
+async function tempDb() {
+ const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'conductor-db-'));
+ const cfg = { database: { dialect: 'sqlite', path: path.join(dir, 'test.db') } };
+ const db = await openDatabase(cfg);
+ return { db, dir, cleanup: async () => { await db.close(); await fs.rm(dir, { recursive: true, force: true }); } };
+}
+
+test('dialect detection covers every supported scheme', () => {
+ assert.equal(dialectFromUrl(null), 'sqlite');
+ assert.equal(dialectFromUrl('mysql://u:p@h/db'), 'mysql');
+ assert.equal(dialectFromUrl('postgres://u:p@h/db'), 'postgres');
+ assert.equal(dialectFromUrl('postgresql://u:p@h/db'), 'postgres');
+ assert.equal(dialectFromUrl('mongodb://h/db'), null);
+});
+
+test('statement splitter respects literals and comments', () => {
+ const sql = [
+ "INSERT INTO t VALUES ('a;b');",
+ '-- a comment with ; in it',
+ 'CREATE TABLE u (id TEXT);',
+ ].join('\n');
+ const parts = splitStatements(sql, 'sqlite');
+ assert.equal(parts.length, 2);
+ assert.ok(parts[0].includes("'a;b'"));
+ assert.ok(parts[1].startsWith('CREATE TABLE u'));
+});
+
+test('statement splitter keeps postgres dollar quoted bodies whole', () => {
+ const sql = "CREATE FUNCTION f() RETURNS void AS $$ BEGIN a; b; END $$ LANGUAGE plpgsql;";
+ assert.equal(splitStatements(sql, 'postgres').length, 1);
+});
+
+test('migrations apply once and are idempotent', async () => {
+ const { db, cleanup } = await tempDb();
+ try {
+ const first = await runMigrations(db, { logger: quiet });
+ assert.equal(first.applied.length, 1);
+
+ const second = await runMigrations(db, { logger: quiet });
+ assert.equal(second.applied.length, 0);
+ assert.equal(second.total, 1);
+ } finally {
+ await cleanup();
+ }
+});
+
+test('an edited migration is refused rather than silently diverging', async () => {
+ const { db, dir, cleanup } = await tempDb();
+ try {
+ const custom = path.join(dir, 'migrations');
+ await fs.mkdir(custom);
+ await fs.writeFile(path.join(custom, '001_a.sql'), 'CREATE TABLE a (id TEXT);');
+ await runMigrations(db, { dir: custom, logger: quiet });
+
+ await fs.writeFile(path.join(custom, '001_a.sql'), 'CREATE TABLE a (id TEXT, extra TEXT);');
+ await assert.rejects(
+ runMigrations(db, { dir: custom, logger: quiet }),
+ /was modified after it was applied/
+ );
+ } finally {
+ await cleanup();
+ }
+});
+
+test('a migration removed from disk is reported', async () => {
+ const { db, dir, cleanup } = await tempDb();
+ try {
+ const custom = path.join(dir, 'migrations');
+ await fs.mkdir(custom);
+ await fs.writeFile(path.join(custom, '001_a.sql'), 'CREATE TABLE a (id TEXT);');
+ await runMigrations(db, { dir: custom, logger: quiet });
+ await fs.rm(path.join(custom, '001_a.sql'));
+
+ await assert.rejects(runMigrations(db, { dir: custom, logger: quiet }), /missing from/);
+ } finally {
+ await cleanup();
+ }
+});
+
+test('rows come back as ordinary objects', async () => {
+ const { db, cleanup } = await tempDb();
+ try {
+ await runMigrations(db, { logger: quiet });
+ const now = Date.now();
+ await db.run(
+ 'INSERT INTO projects (id, name, repo_url, created_at, updated_at) VALUES ({id}, {n}, {u}, {t}, {t})',
+ { id: 'p1', n: 'P', u: 'https://example.invalid/p.git', t: now }
+ );
+ const row = await db.get('SELECT * FROM projects WHERE id = {id}', { id: 'p1' });
+ assert.equal(Object.getPrototypeOf(row), Object.prototype);
+ assert.equal(row.name, 'P');
+ assert.equal(row.enabled, 1);
+ } finally {
+ await cleanup();
+ }
+});
+
+test('a conditional update claims exactly once', async () => {
+ const { db, cleanup } = await tempDb();
+ try {
+ await runMigrations(db, { logger: quiet });
+ const t = Date.now();
+ await db.run('INSERT INTO projects (id,name,repo_url,created_at,updated_at) VALUES ({i},{n},{u},{t},{t})',
+ { i: 'p', n: 'p', u: 'u', t });
+ await db.run('INSERT INTO runs (id,project_id,number,head_sha,created_at) VALUES ({i},{p},{n},{s},{t})',
+ { i: 'r', p: 'p', n: 1, s: 'sha', t });
+ await db.run(
+ 'INSERT INTO jobs (id,run_id,name,base_name,image,requires,spec,timeout,created_at) ' +
+ 'VALUES ({i},{r},{n},{n},{img},{req},{spec},{to},{t})',
+ { i: 'r:b', r: 'r', n: 'b', img: 'alpine', req: '[]', spec: '{}', to: 60, t }
+ );
+
+ const claim = 'UPDATE jobs SET state = {to} WHERE id = {id} AND state = {from}';
+ const first = await db.run(claim, { to: 'running', id: 'r:b', from: 'queued' });
+ const second = await db.run(claim, { to: 'running', id: 'r:b', from: 'queued' });
+ assert.equal(first.changes, 1);
+ assert.equal(second.changes, 0);
+ } finally {
+ await cleanup();
+ }
+});
+
+test('deleting a run cascades to its jobs', async () => {
+ const { db, cleanup } = await tempDb();
+ try {
+ await runMigrations(db, { logger: quiet });
+ const t = Date.now();
+ await db.run('INSERT INTO projects (id,name,repo_url,created_at,updated_at) VALUES ({i},{n},{u},{t},{t})',
+ { i: 'p', n: 'p', u: 'u', t });
+ await db.run('INSERT INTO runs (id,project_id,number,head_sha,created_at) VALUES ({i},{p},{n},{s},{t})',
+ { i: 'r', p: 'p', n: 1, s: 'sha', t });
+ await db.run(
+ 'INSERT INTO jobs (id,run_id,name,base_name,image,requires,spec,timeout,created_at) ' +
+ 'VALUES ({i},{r},{n},{n},{img},{req},{spec},{to},{t})',
+ { i: 'r:b', r: 'r', n: 'b', img: 'alpine', req: '[]', spec: '{}', to: 60, t }
+ );
+
+ await db.run('DELETE FROM runs WHERE id = {id}', { id: 'r' });
+ assert.equal((await db.get('SELECT COUNT(*) AS c FROM jobs', {})).c, 0);
+ } finally {
+ await cleanup();
+ }
+});
+
+test('a failed transaction rolls back, and a failed savepoint keeps the outer work', async () => {
+ const { db, cleanup } = await tempDb();
+ try {
+ await runMigrations(db, { logger: quiet });
+ const t = Date.now();
+ const insert = 'INSERT INTO projects (id,name,repo_url,created_at,updated_at) VALUES ({i},{n},{u},{t},{t})';
+
+ await assert.rejects(db.transaction(async (tx) => {
+ await tx.run(insert, { i: 'rolled', n: 'x', u: 'u', t });
+ throw new Error('boom');
+ }), /boom/);
+ assert.equal((await db.get('SELECT COUNT(*) AS c FROM projects', {})).c, 0);
+
+ await db.transaction(async (tx) => {
+ await tx.run(insert, { i: 'kept', n: 'x', u: 'u', t });
+ await assert.rejects(tx.transaction(async (inner) => {
+ await inner.run(insert, { i: 'inner', n: 'x', u: 'u', t });
+ throw new Error('inner failed');
+ }), /inner failed/);
+ });
+
+ const ids = (await db.all('SELECT id FROM projects ORDER BY id', {})).map((r) => r.id);
+ assert.deepEqual(ids, ['kept']);
+ } finally {
+ await cleanup();
+ }
+});
diff --git a/test/query.test.js b/test/query.test.js
@@ -0,0 +1,92 @@
+// test/query.test.js - named bind marker compilation
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { compileQuery, bindParams } from '../src/lib/db/query.js';
+
+test('compiles markers to the placeholder each dialect expects', () => {
+ const sql = 'SELECT * FROM jobs WHERE run_id = {run} AND state = {state}';
+
+ assert.equal(compileQuery(sql, 'sqlite').sql, 'SELECT * FROM jobs WHERE run_id = ? AND state = ?');
+ assert.equal(compileQuery(sql, 'mysql').sql, 'SELECT * FROM jobs WHERE run_id = ? AND state = ?');
+ assert.equal(compileQuery(sql, 'postgres').sql, 'SELECT * FROM jobs WHERE run_id = $1 AND state = $2');
+
+ assert.deepEqual(compileQuery(sql, 'postgres').keys, ['run', 'state']);
+});
+
+test('numbers postgres placeholders per occurrence, including repeats', () => {
+ const { sql, keys } = compileQuery('SELECT {a}, {b}, {a}', 'postgres');
+ assert.equal(sql, 'SELECT $1, $2, $3');
+ assert.deepEqual(keys, ['a', 'b', 'a']);
+});
+
+test('ignores markers inside string literals and comments', () => {
+ const sql = "SELECT '{notabind}' AS a, {real} -- {alsonot}\n, \"{quoted}\"";
+ const { keys } = compileQuery(sql, 'sqlite');
+ assert.deepEqual(keys, ['real']);
+ assert.ok(compileQuery(sql, 'sqlite').sql.includes("'{notabind}'"));
+});
+
+test('a json literal in default text survives untouched', () => {
+ const { sql, keys } = compileQuery("UPDATE t SET spec = '{}' WHERE id = {id}", 'postgres');
+ assert.equal(sql, "UPDATE t SET spec = '{}' WHERE id = $1");
+ assert.deepEqual(keys, ['id']);
+});
+
+test('doubling a brace escapes it', () => {
+ const { sql, keys } = compileQuery('SELECT {{name}} , {name}', 'sqlite');
+ assert.equal(sql, 'SELECT {name} , ?');
+ assert.deepEqual(keys, ['name']);
+});
+
+test('block comments are skipped, and nest only for postgres', () => {
+ assert.deepEqual(compileQuery('SELECT /* {no} */ {yes}', 'mysql').keys, ['yes']);
+ assert.deepEqual(compileQuery('SELECT /* a /* {no} */ b */ {yes}', 'postgres').keys, ['yes']);
+});
+
+test('postgres dollar quoted bodies are skipped', () => {
+ const { keys } = compileQuery('SELECT $tag$ {no} $tag$, {yes}', 'postgres');
+ assert.deepEqual(keys, ['yes']);
+});
+
+test('mysql backtick identifiers and hash comments are skipped', () => {
+ assert.deepEqual(compileQuery('SELECT `col{no}`, {yes} # {alsono}\n', 'mysql').keys, ['yes']);
+});
+
+test('a malformed marker is rejected with a useful message', () => {
+ assert.throws(() => compileQuery('SELECT {not a name}', 'sqlite'), /malformed bind marker/);
+});
+
+test('binds by name, resolving dotted paths', () => {
+ const { keys, sql } = compileQuery('SELECT {a}, {nested.deep.value}', 'sqlite');
+ assert.deepEqual(bindParams(keys, { a: 1, nested: { deep: { value: 'x' } } }, sql), [1, 'x']);
+});
+
+test('an exact key beats path traversal', () => {
+ const { keys, sql } = compileQuery('SELECT {a.b}', 'sqlite');
+ assert.deepEqual(bindParams(keys, { 'a.b': 'flat', a: { b: 'nested' } }, sql), ['flat']);
+});
+
+test('reports every missing key at once', () => {
+ const { keys, sql } = compileQuery('SELECT {a}, {b}, {c}', 'sqlite');
+ assert.throws(
+ () => bindParams(keys, { b: 1 }, sql),
+ (e) => /missing bind parameter\(s\): a, c/.test(e.message)
+ );
+});
+
+test('rejects an array of parameters, which would silently bind by position', () => {
+ const { keys, sql } = compileQuery('SELECT {a}', 'sqlite');
+ assert.throws(() => bindParams(keys, ['x'], sql), /must be an object/);
+});
+
+test('normalizes booleans, dates and null', () => {
+ const { keys, sql } = compileQuery('SELECT {t}, {f}, {d}, {n}', 'sqlite');
+ const at = new Date('2026-01-02T03:04:05.678Z');
+ assert.deepEqual(bindParams(keys, { t: true, f: false, d: at, n: null }, sql), [1, 0, at.getTime(), null]);
+});
+
+test('rejects an object value, which is nearly always a forgotten stringify', () => {
+ const { keys, sql } = compileQuery('SELECT {spec}', 'sqlite');
+ assert.throws(() => bindParams(keys, { spec: { a: 1 } }, sql), /unsupported type object/);
+});
diff --git a/test/secretbox.test.js b/test/secretbox.test.js
@@ -0,0 +1,102 @@
+// test/secretbox.test.js - encryption at rest for stored secrets
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import crypto from 'node:crypto';
+import { createSecretBox } from '../src/lib/secretbox.js';
+import { jobId, slugify, randomId, timeOrderedId, hashToken, safeEqualHex, JOB_ID_MAX } from '../src/lib/ids.js';
+
+const KEY = crypto.randomBytes(32);
+
+test('seals and opens a value', () => {
+ const box = createSecretBox(KEY);
+ const sealed = box.seal('hunter2');
+ assert.ok(sealed.startsWith('v1.'));
+ assert.ok(!sealed.includes('hunter2'));
+ assert.equal(box.open(sealed), 'hunter2');
+});
+
+test('sealing twice produces different ciphertext', () => {
+ const box = createSecretBox(KEY);
+ assert.notEqual(box.seal('same'), box.seal('same'));
+});
+
+test('without a key, values are stored as marked plaintext', () => {
+ const box = createSecretBox(null);
+ const stored = box.seal('visible');
+ assert.ok(stored.startsWith('plain.'));
+ assert.equal(box.open(stored), 'visible');
+ assert.equal(box.isPlaintext(stored), true);
+});
+
+test('a keyed box can still read values written before a key existed', () => {
+ const plain = createSecretBox(null).seal('legacy');
+ assert.equal(createSecretBox(KEY).open(plain), 'legacy');
+});
+
+test('an unkeyed box refuses encrypted values with an actionable message', () => {
+ const sealed = createSecretBox(KEY).seal('secret');
+ assert.throws(() => createSecretBox(null).open(sealed), /encryption_key is not configured/);
+});
+
+test('the wrong key fails to open', () => {
+ const sealed = createSecretBox(KEY).seal('secret');
+ assert.throws(() => createSecretBox(crypto.randomBytes(32)).open(sealed), /failed to decrypt/);
+});
+
+test('tampering with the ciphertext is detected', () => {
+ const box = createSecretBox(KEY);
+ const parts = box.seal('secret').split('.');
+ const body = Buffer.from(parts[3], 'base64');
+ body[0] ^= 0xff;
+ parts[3] = body.toString('base64');
+ assert.throws(() => box.open(parts.join('.')), /failed to decrypt/);
+});
+
+test('associated data binds a secret to its location', () => {
+ const box = createSecretBox(KEY);
+ const sealed = box.seal('token', 'project:a/var:TOKEN');
+ assert.equal(box.open(sealed, 'project:a/var:TOKEN'), 'token');
+ assert.throws(() => box.open(sealed, 'project:b/var:TOKEN'), /failed to decrypt/);
+});
+
+test('a key of the wrong size is refused at construction', () => {
+ assert.throws(() => createSecretBox(crypto.randomBytes(16)), /32 byte Buffer/);
+});
+
+test('job ids stay readable and within the column width', () => {
+ const runId = timeOrderedId();
+ assert.equal(jobId(runId, 'build'), `${runId}:build`);
+
+ const long = 'a'.repeat(300);
+ const id = jobId(runId, long);
+ assert.ok(id.length <= JOB_ID_MAX, `${id.length} exceeds ${JOB_ID_MAX}`);
+ assert.ok(id.startsWith(`${runId}:`));
+ // Distinct long names must not collide after truncation.
+ assert.notEqual(id, jobId(runId, `${long}b`));
+});
+
+test('time ordered ids sort by creation time', () => {
+ const early = timeOrderedId(1000000000000);
+ const late = timeOrderedId(1000000001000);
+ assert.ok(early < late);
+ assert.equal(early.length, 20);
+});
+
+test('random ids use the expected alphabet', () => {
+ assert.match(randomId(32), /^[0-9abcdefghjkmnpqrstvwxyz]{32}$/);
+});
+
+test('slugify produces readable project ids and falls back when empty', () => {
+ assert.equal(slugify('My Project!'), 'my-project');
+ assert.equal(slugify(' --weird-- '), 'weird');
+ assert.match(slugify('!!!'), /^[0-9a-z]{16}$/);
+});
+
+test('token hashing and constant time comparison', () => {
+ const a = hashToken('token-value');
+ assert.match(a, /^[0-9a-f]{64}$/);
+ assert.equal(safeEqualHex(a, hashToken('token-value')), true);
+ assert.equal(safeEqualHex(a, hashToken('other')), false);
+ assert.equal(safeEqualHex(a, 'short'), false);
+});
diff --git a/test/sigv4.test.js b/test/sigv4.test.js
@@ -0,0 +1,122 @@
+// test/sigv4.test.js - AWS Signature Version 4
+//
+// The vectors come from the published AWS signature test suite. Getting this
+// wrong produces an opaque SignatureDoesNotMatch from the server, so the
+// reference cases are worth pinning.
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { signRequest, presignUrl, uriEncode, amzDate, EMPTY_SHA256 } from '../src/lib/storage/sigv4.js';
+
+const CREDS = {
+ region: 'us-east-1',
+ service: 'service',
+ accessKeyId: 'AKIDEXAMPLE',
+ secretAccessKey: 'wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY',
+ date: new Date(Date.UTC(2015, 7, 30, 12, 36, 0)),
+};
+
+test('get-vanilla matches the reference signature', () => {
+ const headers = signRequest({
+ method: 'GET',
+ url: new URL('https://example.amazonaws.com/'),
+ payloadHash: EMPTY_SHA256,
+ ...CREDS,
+ });
+
+ assert.equal(
+ headers.authorization,
+ 'AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20150830/us-east-1/service/aws4_request, ' +
+ 'SignedHeaders=host;x-amz-date, ' +
+ 'Signature=5fa00fa31553b73ebf1942676e86291e8372ff2a2260956d9b8aae1d763fbf31'
+ );
+});
+
+test('get-vanilla-query-order-key-case matches the reference signature', () => {
+ const headers = signRequest({
+ method: 'GET',
+ url: new URL('https://example.amazonaws.com/?Param2=value2&Param1=value1'),
+ payloadHash: EMPTY_SHA256,
+ ...CREDS,
+ });
+
+ assert.equal(
+ headers.authorization,
+ 'AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20150830/us-east-1/service/aws4_request, ' +
+ 'SignedHeaders=host;x-amz-date, ' +
+ 'Signature=b97d918cfa904a5beff61c982a1b6f458b799221646efd99d3219ec94cdf2500'
+ );
+});
+
+test('uriEncode percent encodes the characters encodeURIComponent leaves alone', () => {
+ assert.equal(uriEncode("a b!'()*"), 'a%20b%21%27%28%29%2A');
+ assert.equal(uriEncode('keep~unreserved-._'), 'keep~unreserved-._');
+ assert.equal(uriEncode('a/b'), 'a%2Fb');
+ assert.equal(uriEncode('a/b', false), 'a/b');
+});
+
+test('amzDate produces both required formats', () => {
+ const { amz, stamp } = amzDate(new Date(Date.UTC(2015, 7, 30, 12, 36, 0)));
+ assert.equal(amz, '20150830T123600Z');
+ assert.equal(stamp, '20150830');
+});
+
+test('s3 requests carry the content hash header', () => {
+ const headers = signRequest({
+ method: 'PUT',
+ url: new URL('https://bucket.example.com/key'),
+ payloadHash: EMPTY_SHA256,
+ region: 'us-east-1',
+ service: 's3',
+ accessKeyId: 'AKIDEXAMPLE',
+ secretAccessKey: 'secret',
+ });
+ assert.equal(headers['x-amz-content-sha256'], EMPTY_SHA256);
+ assert.ok(headers.authorization.includes('x-amz-content-sha256'));
+});
+
+test('presigned urls carry the signature in the query string', () => {
+ const url = new URL(presignUrl({
+ method: 'GET',
+ url: new URL('https://example.com/bucket/key'),
+ expires: 900,
+ region: 'us-east-1',
+ service: 's3',
+ accessKeyId: 'AKIDEXAMPLE',
+ secretAccessKey: 'secret',
+ }));
+
+ assert.equal(url.searchParams.get('X-Amz-Algorithm'), 'AWS4-HMAC-SHA256');
+ assert.equal(url.searchParams.get('X-Amz-Expires'), '900');
+ assert.equal(url.searchParams.get('X-Amz-SignedHeaders'), 'host');
+ assert.match(url.searchParams.get('X-Amz-Signature'), /^[0-9a-f]{64}$/);
+});
+
+test('an already encoded path is not encoded a second time', () => {
+ // A key containing a space reaches the signer as %20. Re-encoding it to
+ // %2520 is the bug this guards against.
+ const url = new URL('https://example.com/bucket/a%20b');
+ const headers = signRequest({
+ method: 'GET',
+ url,
+ payloadHash: EMPTY_SHA256,
+ region: 'us-east-1',
+ service: 's3',
+ accessKeyId: 'AKIDEXAMPLE',
+ secretAccessKey: 'secret',
+ });
+
+ // Signing the same path written differently must differ, proving the path
+ // is taken verbatim rather than normalized.
+ const other = signRequest({
+ method: 'GET',
+ url: new URL('https://example.com/bucket/a%2520b'),
+ payloadHash: EMPTY_SHA256,
+ region: 'us-east-1',
+ service: 's3',
+ accessKeyId: 'AKIDEXAMPLE',
+ secretAccessKey: 'secret',
+ });
+
+ assert.notEqual(headers.authorization, other.authorization);
+});
diff --git a/test/storage.test.js b/test/storage.test.js
@@ -0,0 +1,193 @@
+// test/storage.test.js - object storage backends
+//
+// The local backend is always exercised. The S3 backend runs against a live
+// server only when CONDUCTOR_TEST_S3_ENDPOINT is set, for example:
+//
+// docker run -d -p 9000:9000 -e MINIO_ROOT_USER=testkey \
+// -e MINIO_ROOT_PASSWORD=testsecret123 quay.io/minio/minio server /data
+// CONDUCTOR_TEST_S3_ENDPOINT=http://127.0.0.1:9000 \
+// CONDUCTOR_TEST_S3_KEY=testkey CONDUCTOR_TEST_S3_SECRET=testsecret123 \
+// npm test
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { Readable } from 'node:stream';
+import { createStorage, keys, StorageNotFound, assertKey, sanitizeRelativePath } from '../src/lib/storage/index.js';
+import { signRequest, sha256Hex } from '../src/lib/storage/sigv4.js';
+
+async function drain(stream) {
+ let out = '';
+ for await (const chunk of stream) out += chunk;
+ return out;
+}
+
+async function localStorage() {
+ const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'conductor-store-'));
+ const cfg = { storage: { driver: 'local', path: dir, s3: {} } };
+ return { st: createStorage(cfg), cleanup: () => fs.rm(dir, { recursive: true, force: true }) };
+}
+
+test('key validation rejects traversal and absolute paths', () => {
+ for (const bad of ['../etc/passwd', '/abs', 'a//b', 'a/../b', 'a/./b', '', 'a\\b', 'a\0b']) {
+ assert.throws(() => assertKey(bad), new RegExp('storage key'), `expected ${JSON.stringify(bad)} to be rejected`);
+ }
+ assert.equal(assertKey('artifacts/run/job/dist/app.tar.gz'), 'artifacts/run/job/dist/app.tar.gz');
+});
+
+test('sanitizeRelativePath strips traversal from worker supplied paths', () => {
+ assert.equal(sanitizeRelativePath('../../etc/passwd'), 'etc/passwd');
+ assert.equal(sanitizeRelativePath('./dist/./app.bin'), 'dist/app.bin');
+ assert.equal(sanitizeRelativePath('/abs/path'), 'abs/path');
+ assert.equal(sanitizeRelativePath('..'), null);
+ assert.equal(sanitizeRelativePath(''), null);
+});
+
+test('key layout is stable', () => {
+ assert.equal(keys.artifact('r1', 'r1:build', 'dist/a.bin'), 'artifacts/r1/r1:build/dist/a.bin');
+ assert.equal(keys.log('r1', 'r1:build'), 'logs/r1/r1:build.log');
+});
+
+test('local backend round trips a buffer and reports its digest', async () => {
+ const { st, cleanup } = await localStorage();
+ try {
+ const put = await st.put('a/b/c.txt', Buffer.from('hello conductor'));
+ assert.equal(put.size, 15);
+ assert.equal(put.sha256, sha256Hex(Buffer.from('hello conductor')));
+
+ assert.equal(await drain((await st.get('a/b/c.txt')).stream), 'hello conductor');
+ assert.equal((await st.head('a/b/c.txt')).size, 15);
+ } finally {
+ await cleanup();
+ }
+});
+
+test('local backend serves ranges', async () => {
+ const { st, cleanup } = await localStorage();
+ try {
+ await st.put('r.txt', Buffer.from('hello conductor'));
+ assert.equal(await drain((await st.get('r.txt', { range: { start: 6, end: 14 } })).stream), 'conductor');
+ assert.equal(await drain((await st.get('r.txt', { range: { start: 6 } })).stream), 'conductor');
+ } finally {
+ await cleanup();
+ }
+});
+
+test('local backend accepts a stream', async () => {
+ const { st, cleanup } = await localStorage();
+ try {
+ const put = await st.put('s.log', Readable.from(['line one\n', 'line two\n']));
+ assert.equal(put.size, 18);
+ assert.equal(await drain((await st.get('s.log')).stream), 'line one\nline two\n');
+ } finally {
+ await cleanup();
+ }
+});
+
+test('local backend reports a missing object and deletes idempotently', async () => {
+ const { st, cleanup } = await localStorage();
+ try {
+ await assert.rejects(st.get('nope'), (e) => e instanceof StorageNotFound);
+ assert.equal(await st.head('nope'), null);
+ await st.delete('nope');
+ } finally {
+ await cleanup();
+ }
+});
+
+test('local backend has no presigned urls', async () => {
+ const { st, cleanup } = await localStorage();
+ try {
+ assert.equal(await st.presign('a.txt'), null);
+ } finally {
+ await cleanup();
+ }
+});
+
+test('a partial write leaves no object behind', async () => {
+ const { st, cleanup } = await localStorage();
+ try {
+ const failing = new Readable({
+ read() { this.destroy(new Error('source failed')); },
+ });
+ await assert.rejects(st.put('partial.bin', failing), /source failed/);
+ assert.equal(await st.head('partial.bin'), null);
+ } finally {
+ await cleanup();
+ }
+});
+
+// S3 coverage, opt in.
+const S3_ENDPOINT = process.env.CONDUCTOR_TEST_S3_ENDPOINT;
+const s3Options = { skip: S3_ENDPOINT ? false : 'set CONDUCTOR_TEST_S3_ENDPOINT to run S3 tests' };
+
+test('s3 backend round trips awkward keys against a live server', s3Options, async () => {
+ const bucket = `citest${Date.now().toString(36)}`;
+ const creds = {
+ region: 'us-east-1',
+ accessKeyId: process.env.CONDUCTOR_TEST_S3_KEY,
+ secretAccessKey: process.env.CONDUCTOR_TEST_S3_SECRET,
+ };
+
+ const bucketUrl = new URL(`${S3_ENDPOINT.replace(/\/+$/, '')}/${bucket}`);
+ const created = await fetch(bucketUrl, {
+ method: 'PUT',
+ headers: signRequest({ method: 'PUT', url: bucketUrl, payloadHash: sha256Hex(''), ...creds }),
+ });
+ assert.ok(created.ok, `failed to create bucket: ${created.status}`);
+
+ const st = createStorage({
+ storage: {
+ driver: 's3',
+ s3: {
+ endpoint: S3_ENDPOINT,
+ bucket,
+ region: creds.region,
+ access_key_id: creds.accessKeyId,
+ secret_access_key: creds.secretAccessKey,
+ force_path_style: true,
+ },
+ },
+ });
+
+ // Spaces and plus signs are exactly what a naive signer gets wrong.
+ const key = 'artifacts/r1/r1:build/dist/app bin+v1(final)!.tar.gz';
+ const put = await st.put(key, Buffer.from('hello from s3'), { contentType: 'application/gzip' });
+ assert.equal(put.size, 13);
+
+ assert.equal(await drain((await st.get(key)).stream), 'hello from s3');
+ assert.equal(await drain((await st.get(key, { range: { start: 6, end: 9 } })).stream), 'from');
+ assert.equal((await st.head(key)).size, 13);
+
+ const streamed = await st.put('logs/r1/j.log', Readable.from([Buffer.from('streamed '), Buffer.from('upload')]), { size: 15 });
+ assert.equal(streamed.size, 15);
+
+ const presigned = await fetch(await st.presign(key, { expires: 300 }));
+ assert.equal(presigned.status, 200);
+ assert.equal(await presigned.text(), 'hello from s3');
+
+ await assert.rejects(st.get('missing/object'), (e) => e instanceof StorageNotFound);
+
+ await st.delete(key);
+ assert.equal(await st.head(key), null);
+ await st.delete(key);
+});
+
+test('s3 put of a stream without a size is refused', s3Options, async () => {
+ const st = createStorage({
+ storage: {
+ driver: 's3',
+ s3: {
+ endpoint: S3_ENDPOINT,
+ bucket: 'irrelevant',
+ region: 'us-east-1',
+ access_key_id: 'k',
+ secret_access_key: 's',
+ force_path_style: true,
+ },
+ },
+ });
+ await assert.rejects(st.put('k', Readable.from(['x'])), /needs an explicit size/);
+});
