Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
3.8 kB · 99 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100import { randomBytes } from "node:crypto";import fs from "node:fs/promises";import path from "node:path";
interface LockRecord { pid: number; token: string; createdAt: string;}
export class SingleProcessDirectoryLock { private released = false;
private constructor( private readonly filePath: string, private readonly token: string, private readonly handle: fs.FileHandle, ) {}
static async acquire(directory: string, name = "oauth-store"): Promise<SingleProcessDirectoryLock> { if (!path.isAbsolute(directory)) throw new Error("Single-process lock directory must be absolute"); if (!/^[a-z][a-z0-9-]*$/.test(name)) throw new Error("Single-process lock name is invalid"); await fs.mkdir(directory, { recursive: true, mode: 0o700 }); const directoryStat = await fs.lstat(directory); if (!directoryStat.isDirectory() || directoryStat.isSymbolicLink()) { throw new Error("Single-process lock directory must be a real directory"); } await fs.chmod(directory, 0o700); const filePath = path.join(directory, `${name}.lock`); for (let attempt = 0; attempt < 2; attempt += 1) { const token = randomBytes(24).toString("base64url"); let handle: fs.FileHandle; try { handle = await fs.open(filePath, "wx", 0o600); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error; const existing = await readLockRecord(filePath); if (processIsAlive(existing.pid)) { throw new Error(`OAuth store is already owned by process ${existing.pid}`); } if (attempt > 0) throw new Error("OAuth store lock could not be reclaimed safely"); await fs.unlink(filePath); continue; } const record: LockRecord = { pid: process.pid, token, createdAt: new Date().toISOString() }; try { await handle.writeFile(`${JSON.stringify(record)}\n`, "utf8"); await handle.sync(); await fs.chmod(filePath, 0o600); return new SingleProcessDirectoryLock(filePath, token, handle); } catch (error) { await handle.close().catch(() => undefined); await fs.unlink(filePath).catch(() => undefined); throw error; } } throw new Error("OAuth store lock could not be acquired"); }
async release(): Promise<void> { if (this.released) return; this.released = true; await this.handle.close(); const existing = await readLockRecord(this.filePath).catch(() => undefined); if (existing?.token === this.token && existing.pid === process.pid) { await fs.unlink(this.filePath).catch((error: NodeJS.ErrnoException) => { if (error.code !== "ENOENT") throw error; }); } }}
async function readLockRecord(filePath: string): Promise<LockRecord> { const stat = await fs.lstat(filePath); if (!stat.isFile() || stat.isSymbolicLink() || stat.size > 4_096) { throw new Error("OAuth store lock file is invalid; operator review is required"); } try { const parsed = JSON.parse(await fs.readFile(filePath, "utf8")) as Partial<LockRecord>; if (!Number.isSafeInteger(parsed.pid) || Number(parsed.pid) < 1 || typeof parsed.token !== "string" || parsed.token.length < 16) { throw new Error("shape"); } return { pid: Number(parsed.pid), token: parsed.token, createdAt: String(parsed.createdAt ?? "") }; } catch { throw new Error("OAuth store lock file is invalid; operator review is required"); }}
function processIsAlive(pid: number): boolean { try { process.kill(pid, 0); return true; } catch (error) { const code = (error as NodeJS.ErrnoException).code; if (code === "ESRCH") return false; if (code === "EPERM") return true; throw error; }}