CCP restore/tunnel/upgrade + upgrade.sh release-mode fixes + volunteer dashboard polish
- Add instance restore model, routes, and agent backup/restore endpoints - Add Pangolin tunnel service (subdomain prefix, teardown action, CCP client) - Add slug mutex for concurrent operation safety in agent - Expand upgrade service with remote driver orchestration - Fix upgrade.sh to properly handle release-mode installs (no git operations) - Add CCP registration flags to config.sh (--ccp-url, --ccp-invite-code, --ccp-agent-url) - Auto-detect JVB advertise IP in non-interactive mode - Polish volunteer dashboard ActionStepsList with highlighted step component - Add ticketed event description field + volunteer dashboard query refinements Bunker Admin
This commit is contained in:
@@ -26,6 +26,7 @@ const envSchema = z.object({
|
||||
INSTANCE_SLUG: z.string().default(''),
|
||||
INSTANCE_DOMAIN: z.string().default(''),
|
||||
INSTANCE_BASE_PATH: z.string().default(''),
|
||||
COMPOSE_PROJECT: z.string().default(''),
|
||||
});
|
||||
|
||||
function validateEnv() {
|
||||
|
||||
@@ -1,105 +1,623 @@
|
||||
import { Router, Request, Response } from 'express';
|
||||
import { param } from '../utils/params';
|
||||
import fs from 'fs/promises';
|
||||
import path from 'path';
|
||||
import { exec as execCb } from 'child_process';
|
||||
import { createReadStream, createWriteStream } from 'fs';
|
||||
import { pipeline as pipelineCb, Transform } from 'stream';
|
||||
import { promisify } from 'util';
|
||||
import * as docker from '../services/docker.service';
|
||||
import path from 'path';
|
||||
import crypto from 'crypto';
|
||||
import { spawn } from 'child_process';
|
||||
import { getSlugEntry } from '../services/registry.service';
|
||||
import { env } from '../config/env';
|
||||
import { logger } from '../utils/logger';
|
||||
import { withSlugLock, SlugBusyError, isSlugLocked } from '../services/slug-mutex';
|
||||
import { AgentError } from '../middleware/error-handler';
|
||||
|
||||
const pipeline = promisify(pipelineCb);
|
||||
|
||||
const exec = promisify(execCb);
|
||||
const router = Router();
|
||||
|
||||
// POST /instance/:slug/backup — Run pg_dump + tar uploads → return backup info
|
||||
router.post('/instance/:slug/backup', async (req: Request, res: Response) => {
|
||||
const entry = await getSlugEntry(param(req, 'slug'));
|
||||
const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
|
||||
const backupDir = path.join(env.AGENT_DATA_DIR, 'backups', param(req, 'slug'), timestamp);
|
||||
await fs.mkdir(backupDir, { recursive: true });
|
||||
// ─── Helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
const { pgPassword } = req.body;
|
||||
const ID_REGEX = /^[a-zA-Z0-9_-]+$/;
|
||||
const ARCHIVE_PREFIX = 'changemaker-v2-backup-';
|
||||
const ARCHIVE_SUFFIX = '.tar.gz';
|
||||
|
||||
function backupsDirFor(slug: string): string {
|
||||
return path.join(env.AGENT_DATA_DIR, 'backups', slug);
|
||||
}
|
||||
|
||||
function archivePathFor(slug: string, id: string): string {
|
||||
return path.join(backupsDirFor(slug), `${ARCHIVE_PREFIX}${id}${ARCHIVE_SUFFIX}`);
|
||||
}
|
||||
|
||||
async function sha256File(filePath: string): Promise<string> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const hash = crypto.createHash('sha256');
|
||||
const stream = createReadStream(filePath);
|
||||
stream.on('data', (chunk) => hash.update(chunk));
|
||||
stream.on('end', () => resolve(hash.digest('hex')));
|
||||
stream.on('error', reject);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the manifest.json out of a backup archive without extracting it.
|
||||
* backup.sh stores it at <archive>/changemaker-v2-backup-<ts>/manifest.json
|
||||
*/
|
||||
async function readManifestFromArchive(archivePath: string): Promise<unknown | null> {
|
||||
return new Promise((resolve) => {
|
||||
const proc = spawn('tar', ['-xzOf', archivePath, '--wildcards', '*/manifest.json'], {
|
||||
stdio: ['ignore', 'pipe', 'ignore'],
|
||||
});
|
||||
let buf = '';
|
||||
proc.stdout.on('data', (chunk) => (buf += chunk.toString('utf-8')));
|
||||
proc.on('error', () => resolve(null));
|
||||
proc.on('close', (code) => {
|
||||
if (code !== 0 || !buf.trim()) return resolve(null);
|
||||
try {
|
||||
resolve(JSON.parse(buf));
|
||||
} catch {
|
||||
resolve(null);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract the timestamp ID from a filename like "changemaker-v2-backup-20260409_143000.tar.gz".
|
||||
*/
|
||||
function idFromFilename(filename: string): string | null {
|
||||
if (!filename.startsWith(ARCHIVE_PREFIX) || !filename.endsWith(ARCHIVE_SUFFIX)) return null;
|
||||
return filename.slice(ARCHIVE_PREFIX.length, filename.length - ARCHIVE_SUFFIX.length);
|
||||
}
|
||||
|
||||
// ─── Routes ───────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* POST /instance/:slug/backup
|
||||
* Shells out to the remote CML's scripts/backup.sh. Returns archive metadata
|
||||
* so the CCP can immediately stream it down via the /download endpoint.
|
||||
*/
|
||||
router.post('/instance/:slug/backup', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
const entry = await getSlugEntry(slug);
|
||||
|
||||
try {
|
||||
// 1. pg_dump
|
||||
const dumpFile = path.join(backupDir, 'database.sql');
|
||||
const dump = await docker.composeExec(
|
||||
entry.basePath, entry.composeProject,
|
||||
'v2-postgres',
|
||||
'pg_dump -U changemaker -d changemaker',
|
||||
300_000,
|
||||
pgPassword ? { PGPASSWORD: pgPassword } : undefined
|
||||
);
|
||||
await fs.writeFile(dumpFile, dump, 'utf-8');
|
||||
const result = await withSlugLock(slug, 'backup', async () => {
|
||||
const backupsDir = backupsDirFor(slug);
|
||||
await fs.mkdir(backupsDir, { recursive: true });
|
||||
|
||||
// Gzip the dump
|
||||
await exec(`gzip '${dumpFile}'`, { timeout: 120_000 });
|
||||
// Verify scripts/backup.sh exists
|
||||
const scriptPath = path.join(entry.basePath, 'scripts', 'backup.sh');
|
||||
try {
|
||||
await fs.access(scriptPath);
|
||||
} catch {
|
||||
throw new AgentError(500, `scripts/backup.sh not found at ${scriptPath}`, 'BACKUP_SCRIPT_MISSING');
|
||||
}
|
||||
|
||||
// 2. Tar uploads if exists
|
||||
const uploadsDir = path.join(entry.basePath, 'uploads');
|
||||
let hasUploads = false;
|
||||
try {
|
||||
await fs.access(uploadsDir);
|
||||
hasUploads = true;
|
||||
} catch { /* no uploads dir */ }
|
||||
|
||||
if (hasUploads) {
|
||||
await exec(
|
||||
`tar -czf '${path.join(backupDir, 'uploads.tar.gz')}' -C '${entry.basePath}' uploads`,
|
||||
{ timeout: 300_000 }
|
||||
// Snapshot existing archive filenames so we can identify the new one
|
||||
const existingFiles = new Set(
|
||||
(await fs.readdir(backupsDir)).filter((f) => f.startsWith(ARCHIVE_PREFIX) && f.endsWith(ARCHIVE_SUFFIX))
|
||||
);
|
||||
}
|
||||
|
||||
// 3. Create final archive
|
||||
const archiveName = `backup-${param(req, 'slug')}-${timestamp}.tar.gz`;
|
||||
const archivePath = path.join(env.AGENT_DATA_DIR, 'backups', archiveName);
|
||||
await exec(
|
||||
`tar -czf '${archivePath}' -C '${path.dirname(backupDir)}' '${timestamp}'`,
|
||||
{ timeout: 300_000 }
|
||||
);
|
||||
const logPath = path.join(backupsDir, `backup-${Date.now()}.log`);
|
||||
const logFd = await fs.open(logPath, 'w');
|
||||
|
||||
// Clean up temp dir
|
||||
await fs.rm(backupDir, { recursive: true, force: true });
|
||||
// Spawn backup.sh with cwd=basePath so its .env detection works.
|
||||
// Retention is effectively disabled here — CCP manages retention of
|
||||
// the streamed-down archives, not the agent's transient copies.
|
||||
//
|
||||
// Container names: backup.sh defaults to `changemaker-v2-postgres` and
|
||||
// `listmonk-db`, which match the main CML's `container_name:` overrides.
|
||||
// If a deployment has custom naming, the operator can set PG_CONTAINER /
|
||||
// LISTMONK_PG_CONTAINER in the instance's own .env (backup.sh loads it).
|
||||
const spawnEnv: NodeJS.ProcessEnv = {
|
||||
...process.env,
|
||||
BACKUP_DIR: backupsDir,
|
||||
RETENTION_DAYS: '36500', // ~100 years; CCP controls retention
|
||||
};
|
||||
|
||||
const stats = await fs.stat(archivePath);
|
||||
const backupId = timestamp;
|
||||
logger.info(`[backup] Running scripts/backup.sh for ${slug} (basePath=${entry.basePath})`);
|
||||
|
||||
logger.info(`[backup] Created backup for ${param(req, 'slug')}: ${archivePath} (${stats.size} bytes)`);
|
||||
const exitCode: number = await new Promise((resolve, reject) => {
|
||||
const proc = spawn('bash', ['scripts/backup.sh'], {
|
||||
cwd: entry.basePath,
|
||||
env: spawnEnv,
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
});
|
||||
proc.stdout.on('data', (chunk) => logFd.write(chunk).catch(() => {}));
|
||||
proc.stderr.on('data', (chunk) => logFd.write(chunk).catch(() => {}));
|
||||
proc.on('error', reject);
|
||||
proc.on('close', (code) => resolve(code ?? 1));
|
||||
});
|
||||
|
||||
res.json({
|
||||
backupId,
|
||||
archivePath,
|
||||
sizeBytes: stats.size,
|
||||
timestamp,
|
||||
await logFd.close();
|
||||
|
||||
if (exitCode !== 0) {
|
||||
// Return the tail of the log so the CCP can display it
|
||||
let logTail = '';
|
||||
try {
|
||||
const fullLog = await fs.readFile(logPath, 'utf-8');
|
||||
logTail = fullLog.split('\n').slice(-40).join('\n');
|
||||
} catch { /* ignore */ }
|
||||
throw new AgentError(500, `backup.sh exited with code ${exitCode}\n${logTail}`, 'BACKUP_FAILED');
|
||||
}
|
||||
|
||||
// Find the new archive
|
||||
const afterFiles = (await fs.readdir(backupsDir)).filter(
|
||||
(f) => f.startsWith(ARCHIVE_PREFIX) && f.endsWith(ARCHIVE_SUFFIX)
|
||||
);
|
||||
const newFiles = afterFiles.filter((f) => !existingFiles.has(f));
|
||||
if (newFiles.length === 0) {
|
||||
throw new AgentError(500, 'backup.sh succeeded but no new archive was created', 'BACKUP_NO_OUTPUT');
|
||||
}
|
||||
// Pick the most recently modified (in case of oddities)
|
||||
newFiles.sort();
|
||||
const newest = newFiles[newFiles.length - 1] as string;
|
||||
const archivePath = path.join(backupsDir, newest);
|
||||
const backupId = idFromFilename(newest);
|
||||
if (!backupId || !ID_REGEX.test(backupId)) {
|
||||
throw new AgentError(500, `Unexpected archive filename: ${newest}`, 'BACKUP_NAME_INVALID');
|
||||
}
|
||||
|
||||
const stats = await fs.stat(archivePath);
|
||||
const sha256 = await sha256File(archivePath);
|
||||
const manifest = await readManifestFromArchive(archivePath);
|
||||
|
||||
// Delete the log file once we know the backup succeeded
|
||||
try { await fs.unlink(logPath); } catch { /* ignore */ }
|
||||
|
||||
logger.info(`[backup] ${slug}: created ${newest} (${stats.size} bytes, sha256=${sha256.substring(0, 16)}...)`);
|
||||
|
||||
return {
|
||||
backupId,
|
||||
filename: newest,
|
||||
sizeBytes: stats.size,
|
||||
sha256,
|
||||
manifest,
|
||||
createdAt: stats.mtime.toISOString(),
|
||||
};
|
||||
});
|
||||
|
||||
res.json(result);
|
||||
} catch (err) {
|
||||
// Clean up on failure
|
||||
try { await fs.rm(backupDir, { recursive: true, force: true }); } catch { /* ignore */ }
|
||||
if (err instanceof SlugBusyError) {
|
||||
res.status(409).json({ error: 'SLUG_BUSY', message: err.message });
|
||||
return;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
// GET /instance/:slug/backup/:id/download — Stream backup archive
|
||||
router.get('/instance/:slug/backup/:id/download', async (req: Request, res: Response) => {
|
||||
const archiveName = `backup-${param(req, 'slug')}-${param(req, 'id')}.tar.gz`;
|
||||
const archivePath = path.join(env.AGENT_DATA_DIR, 'backups', archiveName);
|
||||
/**
|
||||
* GET /instance/:slug/backups
|
||||
* Lists backup archives currently held on the agent for this slug.
|
||||
*/
|
||||
router.get('/instance/:slug/backups', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
await getSlugEntry(slug); // validate slug is registered
|
||||
|
||||
const backupsDir = backupsDirFor(slug);
|
||||
let entries: string[] = [];
|
||||
try {
|
||||
await fs.access(archivePath);
|
||||
entries = await fs.readdir(backupsDir);
|
||||
} catch {
|
||||
res.json({ data: [] });
|
||||
return;
|
||||
}
|
||||
|
||||
const results = [];
|
||||
for (const filename of entries) {
|
||||
const id = idFromFilename(filename);
|
||||
if (!id) continue;
|
||||
try {
|
||||
const stats = await fs.stat(path.join(backupsDir, filename));
|
||||
results.push({
|
||||
backupId: id,
|
||||
filename,
|
||||
sizeBytes: stats.size,
|
||||
createdAt: stats.mtime.toISOString(),
|
||||
});
|
||||
} catch { /* skip */ }
|
||||
}
|
||||
results.sort((a, b) => (a.createdAt < b.createdAt ? 1 : -1));
|
||||
res.json({ data: results });
|
||||
});
|
||||
|
||||
/**
|
||||
* GET /instance/:slug/backup/:id/download
|
||||
* Streams the backup archive (supports Content-Length so the CCP can verify size).
|
||||
*/
|
||||
router.get('/instance/:slug/backup/:id/download', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
const id = param(req, 'id');
|
||||
if (!ID_REGEX.test(id)) {
|
||||
res.status(400).json({ error: 'INVALID_ID', message: 'Invalid backup id' });
|
||||
return;
|
||||
}
|
||||
await getSlugEntry(slug);
|
||||
|
||||
const archivePath = archivePathFor(slug, id);
|
||||
try {
|
||||
const stats = await fs.stat(archivePath);
|
||||
res.setHeader('Content-Type', 'application/gzip');
|
||||
res.setHeader('Content-Length', String(stats.size));
|
||||
res.setHeader('Content-Disposition', `attachment; filename="${path.basename(archivePath)}"`);
|
||||
const stream = createReadStream(archivePath);
|
||||
stream.on('error', (err) => {
|
||||
logger.error(`[backup] stream error for ${archivePath}: ${err.message}`);
|
||||
if (!res.headersSent) res.status(500).end();
|
||||
else res.destroy(err);
|
||||
});
|
||||
stream.pipe(res);
|
||||
} catch {
|
||||
res.status(404).json({ error: 'NOT_FOUND', message: 'Backup archive not found' });
|
||||
return;
|
||||
}
|
||||
});
|
||||
|
||||
const stats = await fs.stat(archivePath);
|
||||
res.setHeader('Content-Type', 'application/gzip');
|
||||
res.setHeader('Content-Length', stats.size);
|
||||
res.setHeader('Content-Disposition', `attachment; filename="${archiveName}"`);
|
||||
/**
|
||||
* DELETE /instance/:slug/backup/:id
|
||||
* Deletes the archive from the agent's disk. The CCP calls this after it has
|
||||
* successfully streamed the archive to its own storage.
|
||||
*/
|
||||
router.delete('/instance/:slug/backup/:id', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
const id = param(req, 'id');
|
||||
if (!ID_REGEX.test(id)) {
|
||||
res.status(400).json({ error: 'INVALID_ID', message: 'Invalid backup id' });
|
||||
return;
|
||||
}
|
||||
await getSlugEntry(slug);
|
||||
|
||||
const { createReadStream } = await import('fs');
|
||||
const stream = createReadStream(archivePath);
|
||||
stream.pipe(res);
|
||||
const archivePath = archivePathFor(slug, id);
|
||||
// Path traversal defense: ensure the resolved path is still inside the slug's backups dir
|
||||
const resolved = path.resolve(archivePath);
|
||||
const boundary = path.resolve(backupsDirFor(slug));
|
||||
if (!resolved.startsWith(boundary + path.sep)) {
|
||||
res.status(400).json({ error: 'INVALID_ID', message: 'Invalid backup id' });
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await fs.unlink(archivePath);
|
||||
logger.info(`[backup] ${slug}: deleted ${path.basename(archivePath)}`);
|
||||
res.json({ deleted: true });
|
||||
} catch (err) {
|
||||
const code = (err as NodeJS.ErrnoException).code;
|
||||
if (code === 'ENOENT') {
|
||||
res.status(404).json({ error: 'NOT_FOUND', message: 'Backup archive not found' });
|
||||
return;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
// ─── Restore ──────────────────────────────────────────────────────────
|
||||
|
||||
// Hard cap on a single restore upload. The CCP is trusted, but a buggy or
|
||||
// compromised CCP shouldn't be able to fill the agent's disk in one request.
|
||||
// 20 GB is well above any realistic Changemaker Lite backup size.
|
||||
const MAX_RESTORE_UPLOAD_BYTES = 20 * 1024 * 1024 * 1024;
|
||||
|
||||
function restoresDirFor(slug: string): string {
|
||||
return path.join(env.AGENT_DATA_DIR, 'restores', slug);
|
||||
}
|
||||
|
||||
function restoreUploadDir(slug: string, uploadId: string): string {
|
||||
return path.join(restoresDirFor(slug), uploadId);
|
||||
}
|
||||
|
||||
interface RestoreState {
|
||||
status: 'UPLOADED' | 'RUNNING' | 'COMPLETED' | 'FAILED';
|
||||
uploadId: string;
|
||||
startedAt: string;
|
||||
completedAt?: string;
|
||||
exitCode?: number;
|
||||
logTail?: string;
|
||||
errorMessage?: string;
|
||||
options?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
async function readRestoreState(slug: string, uploadId: string): Promise<RestoreState | null> {
|
||||
const statePath = path.join(restoreUploadDir(slug, uploadId), 'restore-state.json');
|
||||
try {
|
||||
const content = await fs.readFile(statePath, 'utf-8');
|
||||
return JSON.parse(content) as RestoreState;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function writeRestoreState(slug: string, uploadId: string, state: RestoreState): Promise<void> {
|
||||
const statePath = path.join(restoreUploadDir(slug, uploadId), 'restore-state.json');
|
||||
await fs.writeFile(statePath, JSON.stringify(state, null, 2), 'utf-8');
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /instance/:slug/restore/upload?sha256=<hex>
|
||||
* Accepts an application/octet-stream upload of a backup archive and writes
|
||||
* it to the agent's restores directory. Verifies SHA256 as it streams — if
|
||||
* the hash doesn't match, the partial file is deleted and we return 400.
|
||||
*
|
||||
* Returns `{ uploadId, sizeBytes, sha256 }`.
|
||||
*/
|
||||
router.post('/instance/:slug/restore/upload', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
await getSlugEntry(slug);
|
||||
|
||||
if (isSlugLocked(slug, 'restore')) {
|
||||
res.status(409).json({ error: 'SLUG_BUSY', message: 'A restore is already in progress for this slug' });
|
||||
return;
|
||||
}
|
||||
if (isSlugLocked(slug, 'backup')) {
|
||||
res.status(409).json({ error: 'SLUG_BUSY', message: 'A backup is in progress for this slug' });
|
||||
return;
|
||||
}
|
||||
|
||||
const expectedSha256 = typeof req.query.sha256 === 'string' ? req.query.sha256.toLowerCase() : undefined;
|
||||
if (!expectedSha256 || !/^[a-f0-9]{64}$/.test(expectedSha256)) {
|
||||
res.status(400).json({ error: 'VALIDATION', message: 'sha256 query parameter required (64 hex chars)' });
|
||||
return;
|
||||
}
|
||||
|
||||
const uploadId = crypto.randomBytes(16).toString('hex');
|
||||
const uploadDir = restoreUploadDir(slug, uploadId);
|
||||
await fs.mkdir(uploadDir, { recursive: true });
|
||||
const archivePath = path.join(uploadDir, 'archive.tar.gz');
|
||||
|
||||
const hash = crypto.createHash('sha256');
|
||||
let bytesWritten = 0;
|
||||
const hashTransform = new Transform({
|
||||
transform(chunk: Buffer, _enc, cb) {
|
||||
bytesWritten += chunk.length;
|
||||
if (bytesWritten > MAX_RESTORE_UPLOAD_BYTES) {
|
||||
// Abort the stream — pipeline() will reject and the catch block below
|
||||
// will remove the partial upload directory.
|
||||
cb(new AgentError(
|
||||
413,
|
||||
`Upload exceeds maximum allowed size of ${MAX_RESTORE_UPLOAD_BYTES} bytes`,
|
||||
'UPLOAD_TOO_LARGE'
|
||||
));
|
||||
return;
|
||||
}
|
||||
hash.update(chunk);
|
||||
cb(null, chunk);
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
const writeStream = createWriteStream(archivePath);
|
||||
await pipeline(req, hashTransform, writeStream);
|
||||
const sha256 = hash.digest('hex');
|
||||
|
||||
if (sha256 !== expectedSha256) {
|
||||
// Integrity failure — nuke the upload
|
||||
await fs.rm(uploadDir, { recursive: true, force: true });
|
||||
res.status(400).json({
|
||||
error: 'SHA256_MISMATCH',
|
||||
message: `Expected sha256 ${expectedSha256}, got ${sha256}`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const stats = await fs.stat(archivePath);
|
||||
|
||||
// Persist initial state so the progress endpoint works even before apply
|
||||
await writeRestoreState(slug, uploadId, {
|
||||
status: 'UPLOADED',
|
||||
uploadId,
|
||||
startedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
logger.info(`[restore] ${slug}: uploaded ${bytesWritten} bytes (sha256=${sha256.substring(0, 16)}...) upload_id=${uploadId}`);
|
||||
|
||||
res.json({
|
||||
uploadId,
|
||||
sizeBytes: stats.size,
|
||||
sha256,
|
||||
});
|
||||
} catch (err) {
|
||||
// Stream error or write error — clean up
|
||||
try { await fs.rm(uploadDir, { recursive: true, force: true }); } catch { /* ignore */ }
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* POST /instance/:slug/restore/:uploadId/apply
|
||||
* Body: { confirm: true, skipDb?, skipUploads?, skipListmonk?, dryRun? }
|
||||
*
|
||||
* Fires off `scripts/restore.sh --archive <path> --force` in the background
|
||||
* and writes progress to restore-state.json. The CCP polls the progress
|
||||
* endpoint for updates. Mutex prevents concurrent restores/backups.
|
||||
*/
|
||||
router.post('/instance/:slug/restore/:uploadId/apply', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
const uploadId = param(req, 'uploadId');
|
||||
if (!ID_REGEX.test(uploadId)) {
|
||||
res.status(400).json({ error: 'INVALID_ID', message: 'Invalid upload id' });
|
||||
return;
|
||||
}
|
||||
const entry = await getSlugEntry(slug);
|
||||
|
||||
const { confirm, skipDb, skipUploads, skipListmonk, dryRun } = req.body ?? {};
|
||||
if (confirm !== true) {
|
||||
res.status(400).json({ error: 'CONFIRMATION_REQUIRED', message: 'Body must include { confirm: true }' });
|
||||
return;
|
||||
}
|
||||
|
||||
const uploadDir = restoreUploadDir(slug, uploadId);
|
||||
// Path traversal defense
|
||||
const resolvedDir = path.resolve(uploadDir);
|
||||
const boundary = path.resolve(restoresDirFor(slug));
|
||||
if (!resolvedDir.startsWith(boundary + path.sep)) {
|
||||
res.status(400).json({ error: 'INVALID_ID', message: 'Invalid upload id' });
|
||||
return;
|
||||
}
|
||||
|
||||
const archivePath = path.join(uploadDir, 'archive.tar.gz');
|
||||
try {
|
||||
await fs.access(archivePath);
|
||||
} catch {
|
||||
res.status(404).json({ error: 'NOT_FOUND', message: 'Upload not found or already applied' });
|
||||
return;
|
||||
}
|
||||
|
||||
// Verify scripts/restore.sh exists
|
||||
const scriptPath = path.join(entry.basePath, 'scripts', 'restore.sh');
|
||||
try {
|
||||
await fs.access(scriptPath);
|
||||
} catch {
|
||||
res.status(500).json({ error: 'RESTORE_SCRIPT_MISSING', message: `scripts/restore.sh not found at ${scriptPath}` });
|
||||
return;
|
||||
}
|
||||
|
||||
// Check mutex state (don't block — tell caller it's busy)
|
||||
if (isSlugLocked(slug, 'restore') || isSlugLocked(slug, 'backup')) {
|
||||
res.status(409).json({ error: 'SLUG_BUSY', message: 'Slug is busy with backup or restore' });
|
||||
return;
|
||||
}
|
||||
|
||||
// Fire-and-forget: acquire lock and run in background. Return immediately
|
||||
// so CCP can start polling /progress.
|
||||
const options = {
|
||||
skipDb: !!skipDb,
|
||||
skipUploads: !!skipUploads,
|
||||
skipListmonk: !!skipListmonk,
|
||||
dryRun: !!dryRun,
|
||||
};
|
||||
|
||||
await writeRestoreState(slug, uploadId, {
|
||||
status: 'RUNNING',
|
||||
uploadId,
|
||||
startedAt: new Date().toISOString(),
|
||||
options,
|
||||
});
|
||||
|
||||
// Build restore.sh args (all flags, no user input interpolated into a shell string)
|
||||
const args = ['scripts/restore.sh', '--archive', archivePath, '--force'];
|
||||
if (options.skipDb) args.push('--skip-db');
|
||||
if (options.skipUploads) args.push('--skip-uploads');
|
||||
if (options.skipListmonk) args.push('--skip-listmonk');
|
||||
if (options.dryRun) args.push('--dry-run');
|
||||
|
||||
const logPath = path.join(uploadDir, 'restore.log');
|
||||
|
||||
// Schedule the background task — don't await inside the handler
|
||||
void withSlugLock(slug, 'restore', async () => {
|
||||
const logFd = await fs.open(logPath, 'w');
|
||||
logger.info(`[restore] ${slug}: running ${args.join(' ')} (cwd=${entry.basePath})`);
|
||||
|
||||
const exitCode: number = await new Promise((resolve, reject) => {
|
||||
const proc = spawn('bash', args, {
|
||||
cwd: entry.basePath,
|
||||
env: { ...process.env },
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
});
|
||||
proc.stdout.on('data', (chunk) => logFd.write(chunk).catch(() => {}));
|
||||
proc.stderr.on('data', (chunk) => logFd.write(chunk).catch(() => {}));
|
||||
proc.on('error', reject);
|
||||
proc.on('close', (code) => resolve(code ?? 1));
|
||||
});
|
||||
|
||||
await logFd.close();
|
||||
|
||||
// Read the tail of the log for the state file
|
||||
let logTail = '';
|
||||
try {
|
||||
const fullLog = await fs.readFile(logPath, 'utf-8');
|
||||
logTail = fullLog.split('\n').slice(-80).join('\n');
|
||||
} catch { /* ignore */ }
|
||||
|
||||
const state: RestoreState = {
|
||||
status: exitCode === 0 ? 'COMPLETED' : 'FAILED',
|
||||
uploadId,
|
||||
startedAt: (await readRestoreState(slug, uploadId))?.startedAt || new Date().toISOString(),
|
||||
completedAt: new Date().toISOString(),
|
||||
exitCode,
|
||||
logTail,
|
||||
options,
|
||||
...(exitCode !== 0 ? { errorMessage: `restore.sh exited with code ${exitCode}` } : {}),
|
||||
};
|
||||
await writeRestoreState(slug, uploadId, state);
|
||||
|
||||
logger.info(`[restore] ${slug}: restore.sh finished with exit ${exitCode}`);
|
||||
}).catch(async (err) => {
|
||||
logger.error(`[restore] ${slug}: background restore failed: ${(err as Error).message}`);
|
||||
// If the mutex was the issue, state is already written. Otherwise, mark failed.
|
||||
if (!(err instanceof SlugBusyError)) {
|
||||
try {
|
||||
await writeRestoreState(slug, uploadId, {
|
||||
status: 'FAILED',
|
||||
uploadId,
|
||||
startedAt: new Date().toISOString(),
|
||||
completedAt: new Date().toISOString(),
|
||||
errorMessage: (err as Error).message,
|
||||
options,
|
||||
});
|
||||
} catch { /* ignore */ }
|
||||
}
|
||||
});
|
||||
|
||||
res.status(202).json({ applied: true, uploadId, options });
|
||||
});
|
||||
|
||||
/**
|
||||
* GET /instance/:slug/restore/:uploadId/progress
|
||||
* Returns the current state of a running or completed restore.
|
||||
*/
|
||||
router.get('/instance/:slug/restore/:uploadId/progress', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
const uploadId = param(req, 'uploadId');
|
||||
if (!ID_REGEX.test(uploadId)) {
|
||||
res.status(400).json({ error: 'INVALID_ID', message: 'Invalid upload id' });
|
||||
return;
|
||||
}
|
||||
await getSlugEntry(slug);
|
||||
|
||||
const state = await readRestoreState(slug, uploadId);
|
||||
if (!state) {
|
||||
res.status(404).json({ error: 'NOT_FOUND', message: 'Restore not found' });
|
||||
return;
|
||||
}
|
||||
res.json(state);
|
||||
});
|
||||
|
||||
/**
|
||||
* DELETE /instance/:slug/restore/:uploadId
|
||||
* Removes a restore upload directory. Refuses if a restore is currently running.
|
||||
*/
|
||||
router.delete('/instance/:slug/restore/:uploadId', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
const uploadId = param(req, 'uploadId');
|
||||
if (!ID_REGEX.test(uploadId)) {
|
||||
res.status(400).json({ error: 'INVALID_ID', message: 'Invalid upload id' });
|
||||
return;
|
||||
}
|
||||
await getSlugEntry(slug);
|
||||
|
||||
const uploadDir = restoreUploadDir(slug, uploadId);
|
||||
const resolvedDir = path.resolve(uploadDir);
|
||||
const boundary = path.resolve(restoresDirFor(slug));
|
||||
if (!resolvedDir.startsWith(boundary + path.sep)) {
|
||||
res.status(400).json({ error: 'INVALID_ID', message: 'Invalid upload id' });
|
||||
return;
|
||||
}
|
||||
|
||||
const state = await readRestoreState(slug, uploadId);
|
||||
if (state?.status === 'RUNNING') {
|
||||
res.status(409).json({ error: 'RESTORE_RUNNING', message: 'Cannot delete a running restore' });
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await fs.rm(uploadDir, { recursive: true, force: true });
|
||||
res.json({ deleted: true });
|
||||
} catch (err) {
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
export default router;
|
||||
|
||||
@@ -4,6 +4,13 @@ import { registerSlug, unregisterSlug, listSlugs } from '../services/registry.se
|
||||
|
||||
const router = Router();
|
||||
|
||||
// SECURITY: defense-in-depth slug validation. The CCP enforces ^[a-z0-9-]+$
|
||||
// upstream via Zod, but the registry slug is later interpolated into
|
||||
// filesystem paths (backupsDirFor, etc.), so we validate independently here.
|
||||
// A poisoned registry entry could otherwise let a compromised or buggy CCP
|
||||
// escape AGENT_DATA_DIR.
|
||||
const SLUG_RE = /^[a-z0-9-]{2,50}$/;
|
||||
|
||||
// POST /instances/register — Register a slug→basePath mapping
|
||||
router.post('/instances/register', async (req: Request, res: Response) => {
|
||||
const { slug, basePath, composeProject } = req.body;
|
||||
@@ -11,14 +18,23 @@ router.post('/instances/register', async (req: Request, res: Response) => {
|
||||
res.status(400).json({ error: 'VALIDATION', message: 'slug, basePath, and composeProject required' });
|
||||
return;
|
||||
}
|
||||
if (typeof slug !== 'string' || !SLUG_RE.test(slug)) {
|
||||
res.status(400).json({ error: 'VALIDATION', message: 'Invalid slug format (expected ^[a-z0-9-]{2,50}$)' });
|
||||
return;
|
||||
}
|
||||
await registerSlug(slug, basePath, composeProject);
|
||||
res.json({ registered: slug });
|
||||
});
|
||||
|
||||
// DELETE /instances/:slug — Unregister slug
|
||||
router.delete('/instances/:slug', async (req: Request, res: Response) => {
|
||||
await unregisterSlug(param(req, 'slug'));
|
||||
res.json({ unregistered: param(req, 'slug') });
|
||||
const slug = param(req, 'slug');
|
||||
if (!SLUG_RE.test(slug)) {
|
||||
res.status(400).json({ error: 'VALIDATION', message: 'Invalid slug format' });
|
||||
return;
|
||||
}
|
||||
await unregisterSlug(slug);
|
||||
res.json({ unregistered: slug });
|
||||
});
|
||||
|
||||
// GET /instances — List all managed slugs
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
import { Router, Request, Response } from 'express';
|
||||
import { param } from '../utils/params';
|
||||
import { execFile } from 'child_process';
|
||||
import { execFile, spawn } from 'child_process';
|
||||
import { promisify } from 'util';
|
||||
import fs from 'fs/promises';
|
||||
import path from 'path';
|
||||
import { getSlugEntry } from '../services/registry.service';
|
||||
import { logger } from '../utils/logger';
|
||||
import { withSlugLock, SlugBusyError, isSlugLocked } from '../services/slug-mutex';
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
const router = Router();
|
||||
@@ -13,9 +14,108 @@ const router = Router();
|
||||
/** Validate a git branch name — prevent shell injection. */
|
||||
const SAFE_BRANCH = /^[a-zA-Z0-9][a-zA-Z0-9_.\/-]{0,99}$/;
|
||||
|
||||
// POST /instance/:slug/upgrade/start — Run upgrade.sh
|
||||
/**
|
||||
* Max age of an in-progress upgrade (by progress.json mtime) before we
|
||||
* consider a previous attempt dead and allow a new one through.
|
||||
*
|
||||
* SECURITY NOTE: this must be LONGER than the CCP's REMOTE_UPGRADE_TIMEOUT
|
||||
* AND longer than any realistic legitimate upgrade duration. The concern is
|
||||
* a concurrent-upgrade scenario:
|
||||
* - upgrade.sh is running and legitimately slow (large image pull + DB
|
||||
* migration)
|
||||
* - at 15 min the CCP side times out and marks the row FAILED
|
||||
* - admin clicks "Upgrade" again → CCP's DB check sees no active row
|
||||
* - if this staleness window is <= realistic upgrade time, the second
|
||||
* /upgrade/start call would ALSO pass this check, spawning a second
|
||||
* upgrade.sh process racing against the still-running first one
|
||||
*
|
||||
* 45 min gives headroom over the 15-min CCP timeout and covers realistic
|
||||
* upgrade durations. For a truly bulletproof guard, switch to a PID lock
|
||||
* file that verifies the process is still alive.
|
||||
*/
|
||||
const STALE_UPGRADE_MTIME_MS = 45 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* Returns true if there's an in-progress upgrade visible on disk.
|
||||
*
|
||||
* Used as a second-line guard in case the in-memory mutex was lost to an
|
||||
* agent restart mid-upgrade. The check looks at progress.json mtime and
|
||||
* the absence of a result.json — together they indicate "started but not
|
||||
* finished within the staleness window".
|
||||
*/
|
||||
async function isUpgradeRunningOnDisk(basePath: string): Promise<boolean> {
|
||||
const progressPath = path.join(basePath, 'data', 'upgrade', 'progress.json');
|
||||
const resultPath = path.join(basePath, 'data', 'upgrade', 'result.json');
|
||||
|
||||
let progressStat: import('fs').Stats;
|
||||
try {
|
||||
progressStat = await fs.stat(progressPath);
|
||||
} catch {
|
||||
return false; // no progress file → no in-progress upgrade
|
||||
}
|
||||
|
||||
// If a result file exists with mtime >= progress mtime, the run is finished
|
||||
try {
|
||||
const resultStat = await fs.stat(resultPath);
|
||||
if (resultStat.mtimeMs >= progressStat.mtimeMs) return false;
|
||||
} catch { /* no result file yet */ }
|
||||
|
||||
// Stale: progress file is old and no result was written → assume the
|
||||
// previous attempt died and let a new one through
|
||||
if (Date.now() - progressStat.mtimeMs > STALE_UPGRADE_MTIME_MS) return false;
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
// POST /instance/:slug/upgrade/check — Run upgrade-check.sh and return status.json
|
||||
router.post('/instance/:slug/upgrade/check', async (req: Request, res: Response) => {
|
||||
const slug = param(req, 'slug');
|
||||
const entry = await getSlugEntry(slug);
|
||||
|
||||
// Refuse during a running upgrade — check writes status.json which could
|
||||
// race with upgrade.sh writing other files in data/upgrade/
|
||||
if (isSlugLocked(slug, 'upgrade') || await isUpgradeRunningOnDisk(entry.basePath)) {
|
||||
res.status(409).json({ error: 'SLUG_BUSY', message: 'An upgrade is currently running' });
|
||||
return;
|
||||
}
|
||||
|
||||
const scriptPath = path.join(entry.basePath, 'scripts', 'upgrade-check.sh');
|
||||
try {
|
||||
await fs.access(scriptPath);
|
||||
} catch {
|
||||
res.status(404).json({ error: 'SCRIPT_NOT_FOUND', message: `upgrade-check.sh not found at ${scriptPath}` });
|
||||
return;
|
||||
}
|
||||
|
||||
// Run upgrade-check.sh — it writes data/upgrade/status.json. Use execFile
|
||||
// (no shell) and a 60s timeout. Failures are non-fatal: the script may
|
||||
// still have written status.json before erroring out, so we always try
|
||||
// to read it afterwards.
|
||||
try {
|
||||
await execFileAsync('bash', [scriptPath], {
|
||||
cwd: entry.basePath,
|
||||
timeout: 60_000,
|
||||
maxBuffer: 4 * 1024 * 1024,
|
||||
env: { ...process.env, COMPOSE_ANSI: 'never' },
|
||||
});
|
||||
} catch (err) {
|
||||
logger.warn(`[upgrade] ${slug}: upgrade-check.sh failed: ${(err as Error).message}`);
|
||||
// continue — try to read status.json anyway
|
||||
}
|
||||
|
||||
const statusPath = path.join(entry.basePath, 'data', 'upgrade', 'status.json');
|
||||
try {
|
||||
const content = await fs.readFile(statusPath, 'utf-8');
|
||||
res.json(JSON.parse(content));
|
||||
} catch {
|
||||
res.status(500).json({ error: 'STATUS_NOT_AVAILABLE', message: 'upgrade-check.sh did not produce status.json' });
|
||||
}
|
||||
});
|
||||
|
||||
// POST /instance/:slug/upgrade/start — Run upgrade.sh in the background
|
||||
router.post('/instance/:slug/upgrade/start', async (req: Request, res: Response) => {
|
||||
const entry = await getSlugEntry(param(req, 'slug'));
|
||||
const slug = param(req, 'slug');
|
||||
const entry = await getSlugEntry(slug);
|
||||
const { skipBackup, useRegistry, branch } = req.body || {};
|
||||
|
||||
// SECURITY: Validate branch name to prevent injection
|
||||
@@ -28,26 +128,64 @@ router.post('/instance/:slug/upgrade/start', async (req: Request, res: Response)
|
||||
try {
|
||||
await fs.access(scriptPath);
|
||||
} catch {
|
||||
res.status(400).json({ error: 'NOT_FOUND', message: 'upgrade.sh not found' });
|
||||
res.status(404).json({ error: 'NOT_FOUND', message: 'upgrade.sh not found' });
|
||||
return;
|
||||
}
|
||||
|
||||
// SECURITY: Use execFile with args array — no shell interpolation
|
||||
const args = ['--api-mode', '--force'];
|
||||
// Refuse if an upgrade is already running (in-memory or on-disk indicators)
|
||||
if (isSlugLocked(slug, 'upgrade') || await isUpgradeRunningOnDisk(entry.basePath)) {
|
||||
res.status(409).json({ error: 'SLUG_BUSY', message: 'An upgrade is already in progress' });
|
||||
return;
|
||||
}
|
||||
// Backup or restore concurrency: refuse to start an upgrade while either is running
|
||||
if (isSlugLocked(slug, 'backup') || isSlugLocked(slug, 'restore')) {
|
||||
res.status(409).json({ error: 'SLUG_BUSY', message: 'A backup or restore is currently running' });
|
||||
return;
|
||||
}
|
||||
|
||||
// Clear stale progress/result files before starting so the on-disk staleness
|
||||
// check doesn't think a brand-new upgrade is still finishing.
|
||||
const progressPath = path.join(entry.basePath, 'data', 'upgrade', 'progress.json');
|
||||
const resultPath = path.join(entry.basePath, 'data', 'upgrade', 'result.json');
|
||||
await fs.mkdir(path.dirname(progressPath), { recursive: true });
|
||||
await fs.rm(progressPath, { force: true });
|
||||
await fs.rm(resultPath, { force: true });
|
||||
|
||||
// SECURITY: Use spawn with args array — no shell interpolation
|
||||
const args: string[] = [scriptPath, '--api-mode', '--force'];
|
||||
if (skipBackup) args.push('--skip-backup');
|
||||
if (useRegistry) args.push('--use-registry');
|
||||
if (branch) args.push('--branch', branch);
|
||||
|
||||
// Fire-and-forget — CCP polls progress
|
||||
execFileAsync('bash', [scriptPath, ...args], {
|
||||
cwd: entry.basePath,
|
||||
timeout: 600_000,
|
||||
maxBuffer: 10 * 1024 * 1024,
|
||||
// Schedule the background task under the slug lock. Use void so the
|
||||
// promise doesn't block the response. Errors are caught and logged; the
|
||||
// CCP detects them via the absence of a result file or via the timeout.
|
||||
void withSlugLock(slug, 'upgrade', async () => {
|
||||
logger.info(`[upgrade] ${slug}: spawning ${args.join(' ')} (cwd=${entry.basePath})`);
|
||||
try {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const proc = spawn('bash', args, {
|
||||
cwd: entry.basePath,
|
||||
env: { ...process.env, COMPOSE_ANSI: 'never' },
|
||||
stdio: ['ignore', 'ignore', 'ignore'], // upgrade.sh writes its own logs
|
||||
});
|
||||
proc.on('error', reject);
|
||||
proc.on('close', (code) => {
|
||||
if (code === 0) resolve();
|
||||
else reject(new Error(`upgrade.sh exited with code ${code}`));
|
||||
});
|
||||
});
|
||||
logger.info(`[upgrade] ${slug}: upgrade.sh completed`);
|
||||
} catch (err) {
|
||||
logger.error(`[upgrade] ${slug}: ${(err as Error).message}`);
|
||||
}
|
||||
}).catch((err) => {
|
||||
logger.error(`[upgrade] ${param(req, 'slug')} failed: ${(err as Error).message}`);
|
||||
if (!(err instanceof SlugBusyError)) {
|
||||
logger.error(`[upgrade] ${slug}: lock or background error: ${(err as Error).message}`);
|
||||
}
|
||||
});
|
||||
|
||||
res.json({ started: true });
|
||||
res.status(202).json({ started: true });
|
||||
});
|
||||
|
||||
// GET /instance/:slug/upgrade/progress — Read progress.json
|
||||
|
||||
@@ -53,8 +53,24 @@ if (hasCerts()) {
|
||||
app.use(errorHandler);
|
||||
|
||||
const server = https.createServer(tlsOptions, app);
|
||||
server.listen(env.AGENT_PORT, () => {
|
||||
server.listen(env.AGENT_PORT, async () => {
|
||||
logger.info(`CCP Agent (mTLS) listening on port ${env.AGENT_PORT}`);
|
||||
|
||||
// Auto-register this instance's slug if configured
|
||||
if (env.INSTANCE_SLUG && env.INSTANCE_BASE_PATH) {
|
||||
const { registerSlug, getSlugEntry } = await import('./services/registry.service');
|
||||
try {
|
||||
await getSlugEntry(env.INSTANCE_SLUG);
|
||||
logger.debug(`[registry] Slug ${env.INSTANCE_SLUG} already registered`);
|
||||
} catch {
|
||||
// Detect compose project name: use env override, or derive from basePath directory name
|
||||
// (Docker Compose default: directory name with special chars stripped)
|
||||
const pathMod = await import('path');
|
||||
const composeProject = env.COMPOSE_PROJECT
|
||||
|| pathMod.basename(env.INSTANCE_BASE_PATH).replace(/[^a-zA-Z0-9]/g, '').toLowerCase();
|
||||
await registerSlug(env.INSTANCE_SLUG, env.INSTANCE_BASE_PATH, composeProject);
|
||||
}
|
||||
}
|
||||
});
|
||||
} else {
|
||||
// Pre-approval mode — start HTTP, only health + phone-home polling
|
||||
|
||||
65
changemaker-control-panel/agent/src/services/slug-mutex.ts
Normal file
65
changemaker-control-panel/agent/src/services/slug-mutex.ts
Normal file
@@ -0,0 +1,65 @@
|
||||
/**
|
||||
* Per-slug single-flight mutex.
|
||||
*
|
||||
* Guards long-running, mutating operations (backup, restore, upgrade) so that
|
||||
* two concurrent CCP calls for the same slug can't trample each other.
|
||||
*
|
||||
* Usage:
|
||||
* await withSlugLock(slug, 'backup', async () => { ... });
|
||||
*
|
||||
* If a lock is already held for (slug, op), throws SlugBusyError which the
|
||||
* route handler should convert to HTTP 409.
|
||||
*/
|
||||
|
||||
export class SlugBusyError extends Error {
|
||||
constructor(public slug: string, public op: string) {
|
||||
super(`Slug ${slug} is busy: ${op} already in progress`);
|
||||
this.name = 'SlugBusyError';
|
||||
}
|
||||
}
|
||||
|
||||
type LockKey = string;
|
||||
const locks = new Map<LockKey, { op: string; startedAt: number }>();
|
||||
|
||||
function key(slug: string, op: string): LockKey {
|
||||
return `${slug}::${op}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run `fn` while holding a single-flight lock on (slug, op).
|
||||
* Throws SlugBusyError immediately if another call is already running.
|
||||
*/
|
||||
export async function withSlugLock<T>(
|
||||
slug: string,
|
||||
op: string,
|
||||
fn: () => Promise<T>
|
||||
): Promise<T> {
|
||||
const k = key(slug, op);
|
||||
if (locks.has(k)) {
|
||||
throw new SlugBusyError(slug, op);
|
||||
}
|
||||
locks.set(k, { op, startedAt: Date.now() });
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
locks.delete(k);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns true if a lock is currently held for (slug, op).
|
||||
*/
|
||||
export function isSlugLocked(slug: string, op: string): boolean {
|
||||
return locks.has(key(slug, op));
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns debug info about all active locks.
|
||||
*/
|
||||
export function listActiveLocks(): Array<{ slug: string; op: string; ageMs: number }> {
|
||||
const now = Date.now();
|
||||
return Array.from(locks.entries()).map(([k, v]) => {
|
||||
const [slug] = k.split('::');
|
||||
return { slug: slug ?? '', op: v.op, ageMs: now - v.startedAt };
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user