Codex
1407 lines
57 KiB
JavaScript
1407 lines
57 KiB
JavaScript
import { spawn } from "node:child_process";
|
|
import { createHash, createHmac, randomUUID, timingSafeEqual } from "node:crypto";
|
|
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
|
|
import { mkdir, open, readFile, readdir, rename, stat, unlink } from "node:fs/promises";
|
|
import { createServer } from "node:http";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
|
|
export function createGithubSyncHandler(config) {
|
|
const controller = config.controller ?? createDurableSyncController(config);
|
|
if (!Number.isInteger(config.responseBudgetMs) || config.responseBudgetMs < 1) throw new Error("responseBudgetMs must be a positive integer");
|
|
return async (req, res) => {
|
|
const responseDeadlineMs = Date.now() + config.responseBudgetMs;
|
|
const requestContext = { deliveryId: null, repository: null, ref: null, requestedCommit: null };
|
|
try {
|
|
const url = new URL(req.url || "/", "http://gitea-github-sync.local");
|
|
if (req.method === "GET" && url.pathname === "/healthz") {
|
|
const readiness = controller.readiness();
|
|
writeJson(res, readiness.ready ? 200 : 503, { ok: readiness.ready, repos: config.repos.length, durableInbox: readiness, valuesPrinted: false });
|
|
return;
|
|
}
|
|
if (req.method === "GET" && url.pathname === "/status") {
|
|
const status = await controller.status();
|
|
writeJson(res, status.ready ? 200 : 503, { ok: status.ready, ...status, valuesPrinted: false });
|
|
return;
|
|
}
|
|
if (req.method !== "POST" || url.pathname !== config.path) {
|
|
writeJson(res, 404, { ok: false, error: "not-found", valuesPrinted: false });
|
|
return;
|
|
}
|
|
const deliveryId = singleHeader(req.headers["x-github-delivery"]);
|
|
requestContext.deliveryId = deliveryId || null;
|
|
const body = await waitWithinResponseBudget(
|
|
readBody(req, config.maxBodyBytes),
|
|
responseDeadlineMs,
|
|
"request-body-response-deadline-exhausted",
|
|
);
|
|
verifySignature(req.headers["x-hub-signature-256"], body, config.webhookSecret);
|
|
const event = String(req.headers["x-github-event"] || "");
|
|
if (!deliveryId) throw new HttpError(400, "missing-github-delivery-id");
|
|
if (event === "ping") {
|
|
writeJson(res, 200, { ok: true, event, deliveryId, disposition: "pong", valuesPrinted: false });
|
|
return;
|
|
}
|
|
if (event !== "push") {
|
|
writeJson(res, 202, { ok: true, event, deliveryId, disposition: "ignored-event", valuesPrinted: false });
|
|
return;
|
|
}
|
|
const payload = parsePayload(body);
|
|
const repository = String(payload?.repository?.full_name || "");
|
|
const ref = String(payload?.ref || "");
|
|
const requestedCommit = String(payload?.after || "");
|
|
Object.assign(requestContext, { repository, ref, requestedCommit });
|
|
const repo = config.repos.find((item) => item?.upstream?.repository === repository && ref === `refs/heads/${item?.upstream?.branch}`);
|
|
if (!repo || /^0+$/u.test(requestedCommit)) {
|
|
writeJson(res, 202, { ok: true, event, deliveryId, repository, ref, disposition: "ignored-repo-ref-or-delete", valuesPrinted: false });
|
|
return;
|
|
}
|
|
if (!/^[0-9a-f]{40,64}$/u.test(requestedCommit)) throw new HttpError(400, "invalid-push-after");
|
|
const delivery = {
|
|
deliveryId,
|
|
repository,
|
|
ref,
|
|
requestedCommit,
|
|
payloadSha: createHash("sha256").update(body).digest("hex"),
|
|
receivedAt: new Date().toISOString(),
|
|
};
|
|
const acceptPromise = Promise.resolve().then(() => controller.accept(repo, delivery));
|
|
let result;
|
|
try {
|
|
result = await waitWithinResponseBudget(
|
|
acceptPromise,
|
|
responseDeadlineMs,
|
|
"durable-acceptance-response-deadline-exhausted",
|
|
);
|
|
} catch (error) {
|
|
if (error instanceof HttpError && error.message === "durable-acceptance-response-deadline-exhausted") {
|
|
void acceptPromise.then(
|
|
(lateResult) => safeLog(config.log, deliveryLog("github-to-gitea-delivery-accepted-after-response-deadline", repo, delivery, {
|
|
ok: true,
|
|
state: lateResult.state,
|
|
duplicate: lateResult.duplicate,
|
|
})),
|
|
(lateError) => safeLog(config.log, deliveryLog("github-to-gitea-delivery-persist-failed-after-response-deadline", repo, delivery, {
|
|
ok: false,
|
|
errorType: "durable-acceptance-late-failure",
|
|
error: redact(String(lateError?.message || lateError)),
|
|
})),
|
|
);
|
|
}
|
|
throw error;
|
|
}
|
|
const refsCommitted = result.state === "committed";
|
|
writeJson(res, 202, {
|
|
ok: true,
|
|
event,
|
|
deliveryId,
|
|
repository,
|
|
ref,
|
|
repo: repo.key,
|
|
requestedCommit,
|
|
state: result.state,
|
|
disposition: result.duplicate ? "duplicate-durable-inbox-record" : "durable-inbox-accepted",
|
|
refsCommitted,
|
|
...(refsCommitted && result.result ? { result: result.result } : {}),
|
|
valuesPrinted: false,
|
|
});
|
|
} catch (error) {
|
|
const status = error instanceof HttpError ? error.status : 503;
|
|
safeLog(config.log, {
|
|
event: "github-to-gitea-request-rejected",
|
|
...requestContext,
|
|
ok: false,
|
|
status,
|
|
error: redact(String(error?.message || error)),
|
|
valuesPrinted: false,
|
|
});
|
|
writeJson(res, status, { ok: false, error: error instanceof HttpError ? error.message : "internal-sync-error", valuesPrinted: false });
|
|
}
|
|
};
|
|
}
|
|
|
|
export function createDurableInboxStore(options) {
|
|
const inboxDirectory = options.path;
|
|
const maxBytes = options.maxBytes;
|
|
const now = options.now ?? Date.now;
|
|
let totalBytes = 0;
|
|
let nextAcceptedSequence = 1;
|
|
let initialized = false;
|
|
let metadata = null;
|
|
let mutationTail = Promise.resolve();
|
|
|
|
function withMutationLock(task) {
|
|
const result = mutationTail.then(task);
|
|
mutationTail = result.catch(() => undefined);
|
|
return result;
|
|
}
|
|
|
|
async function initialize() {
|
|
await mkdir(inboxDirectory, { recursive: true, mode: 0o700 });
|
|
const probePath = join(inboxDirectory, `.readiness-${randomUUID()}`);
|
|
await durableWrite(probePath, '{"ok":true}\n');
|
|
await unlink(probePath);
|
|
await syncDirectory(inboxDirectory);
|
|
const metadataPath = join(inboxDirectory, ".journal-metadata");
|
|
try {
|
|
metadata = validateJournalMetadata(JSON.parse(await readFile(metadataPath, "utf8")));
|
|
} catch (error) {
|
|
if (error?.code !== "ENOENT") throw new DurableInboxError(503, "durable-inbox-metadata-invalid");
|
|
metadata = validateJournalMetadata({
|
|
version: 1,
|
|
kind: "GiteaGithubSyncDurableInbox",
|
|
createdAt: new Date(now()).toISOString(),
|
|
});
|
|
await durableWrite(metadataPath, serializeRecord(metadata));
|
|
}
|
|
totalBytes = 0;
|
|
nextAcceptedSequence = 1;
|
|
for (const entry of await readdir(inboxDirectory, { withFileTypes: true })) {
|
|
if (!entry.isFile() || !entry.name.endsWith(".json")) continue;
|
|
const path = join(inboxDirectory, entry.name);
|
|
const record = await readRecordPath(path);
|
|
nextAcceptedSequence = Math.max(nextAcceptedSequence, record.acceptedSequence + 1);
|
|
totalBytes += (await stat(path)).size;
|
|
}
|
|
initialized = true;
|
|
return { totalBytes, maxBytes };
|
|
}
|
|
|
|
async function accept(repo, delivery) {
|
|
return await withMutationLock(async () => {
|
|
assertInitialized();
|
|
const path = recordPath(inboxDirectory, delivery.deliveryId);
|
|
const existing = await readRecordPath(path, true);
|
|
const fingerprint = deliveryFingerprint(repo, delivery);
|
|
if (existing) {
|
|
if (existing.fingerprint !== fingerprint) throw new DurableInboxError(409, "github-delivery-id-payload-conflict");
|
|
return { record: existing, duplicate: true };
|
|
}
|
|
const now = delivery.receivedAt || new Date().toISOString();
|
|
const record = {
|
|
version: 1,
|
|
deliveryId: delivery.deliveryId,
|
|
fingerprint,
|
|
repoKey: repo.key,
|
|
repository: delivery.repository,
|
|
ref: delivery.ref,
|
|
requestedCommit: delivery.requestedCommit,
|
|
payloadSha: delivery.payloadSha,
|
|
receivedAt: now,
|
|
acceptedAt: now,
|
|
acceptedSequence: nextAcceptedSequence,
|
|
updatedAt: now,
|
|
state: "accepted",
|
|
totalAttempts: 0,
|
|
cycleAttempt: 0,
|
|
nextAttemptAt: now,
|
|
result: null,
|
|
lastError: null,
|
|
};
|
|
const serialized = serializeRecord(validateInboxRecord(record));
|
|
if (totalBytes + Buffer.byteLength(serialized) > maxBytes) {
|
|
throw new DurableInboxError(503, "durable-inbox-capacity-exhausted");
|
|
}
|
|
await durableWrite(path, serialized);
|
|
totalBytes += Buffer.byteLength(serialized);
|
|
nextAcceptedSequence += 1;
|
|
return { record, duplicate: false };
|
|
});
|
|
}
|
|
|
|
async function get(deliveryId) {
|
|
assertInitialized();
|
|
return await readRecordPath(recordPath(inboxDirectory, deliveryId), true);
|
|
}
|
|
|
|
async function list() {
|
|
assertInitialized();
|
|
const records = [];
|
|
for (const entry of await readdir(inboxDirectory, { withFileTypes: true })) {
|
|
if (!entry.isFile() || !entry.name.endsWith(".json")) continue;
|
|
records.push(await readRecordPath(join(inboxDirectory, entry.name)));
|
|
}
|
|
return sortInboxRecordsByAcceptedSequence(records);
|
|
}
|
|
|
|
async function update(deliveryId, updater) {
|
|
return await withMutationLock(async () => {
|
|
assertInitialized();
|
|
const path = recordPath(inboxDirectory, deliveryId);
|
|
const previous = await readRecordPath(path);
|
|
const previousSize = (await stat(path)).size;
|
|
const next = updater(structuredClone(previous));
|
|
const identityFields = ["version", "deliveryId", "fingerprint", "repoKey", "repository", "ref", "requestedCommit", "payloadSha", "receivedAt", "acceptedAt", "acceptedSequence"];
|
|
if (!next || identityFields.some((field) => next[field] !== previous[field])) {
|
|
throw new DurableInboxError(503, "durable-inbox-invalid-record-update");
|
|
}
|
|
const serialized = serializeRecord(validateInboxRecord(next));
|
|
await durableWrite(path, serialized);
|
|
totalBytes += Buffer.byteLength(serialized) - previousSize;
|
|
return next;
|
|
});
|
|
}
|
|
|
|
async function recoverProcessing(now) {
|
|
const records = await list();
|
|
for (const record of records) {
|
|
if (record.state !== "processing") continue;
|
|
await update(record.deliveryId, (current) => ({
|
|
...current,
|
|
state: "accepted",
|
|
updatedAt: now,
|
|
nextAttemptAt: now,
|
|
recoveredAfterRestart: true,
|
|
lastError: { errorType: "worker-restarted-during-processing", retryable: true },
|
|
}));
|
|
}
|
|
}
|
|
|
|
async function cleanupCommitted(cutoffMs) {
|
|
return await withMutationLock(async () => {
|
|
assertInitialized();
|
|
let removed = 0;
|
|
const entries = (await readdir(inboxDirectory, { withFileTypes: true }))
|
|
.filter((entry) => entry.isFile() && entry.name.endsWith(".json"));
|
|
const records = [];
|
|
for (const entry of entries) {
|
|
const path = join(inboxDirectory, entry.name);
|
|
records.push({ entry, path, record: await readRecordPath(path) });
|
|
}
|
|
const latestCommittedByRepo = new Map();
|
|
for (const item of records) {
|
|
if (item.record.state !== "committed") continue;
|
|
const current = latestCommittedByRepo.get(item.record.repoKey);
|
|
if (!current || item.record.acceptedSequence > current.record.acceptedSequence) {
|
|
latestCommittedByRepo.set(item.record.repoKey, item);
|
|
}
|
|
}
|
|
for (const item of records) {
|
|
const { path, record } = item;
|
|
if (record.state !== "committed" || latestCommittedByRepo.get(record.repoKey) === item || Date.parse(record.committedAt) > cutoffMs) continue;
|
|
const size = (await stat(path)).size;
|
|
await unlink(path);
|
|
totalBytes = Math.max(0, totalBytes - size);
|
|
removed += 1;
|
|
}
|
|
if (removed > 0) await syncDirectory(inboxDirectory);
|
|
return removed;
|
|
});
|
|
}
|
|
|
|
function capacityAvailable() {
|
|
return initialized && totalBytes < maxBytes;
|
|
}
|
|
|
|
function usage() {
|
|
return { totalBytes, maxBytes, capacityAvailable: capacityAvailable() };
|
|
}
|
|
|
|
function journalMetadata() {
|
|
assertInitialized();
|
|
return structuredClone(metadata);
|
|
}
|
|
|
|
function assertInitialized() {
|
|
if (!initialized) throw new DurableInboxError(503, "durable-inbox-unavailable");
|
|
}
|
|
|
|
return { initialize, accept, get, list, update, recoverProcessing, cleanupCommitted, capacityAvailable, usage, journalMetadata };
|
|
}
|
|
|
|
export function createDurableSyncController(options) {
|
|
const now = options.now ?? Date.now;
|
|
const store = options.store ?? createDurableInboxStore(options.inbox);
|
|
const retry = options.workerRetry;
|
|
const workerAbort = new AbortController();
|
|
const executor = options.executor ?? createSyncCoordinator({
|
|
...options,
|
|
signal: workerAbort.signal,
|
|
responseBudgetMs: retry.attemptTimeoutMs,
|
|
retryMaxAttempts: 1,
|
|
retryInitialDelayMs: retry.initialDelayMs,
|
|
retryMaxDelayMs: retry.maxDelayMs,
|
|
});
|
|
const repoByKey = new Map(options.repos.map((repo) => [repo.key, repo]));
|
|
const repoTails = new Map();
|
|
const activeDeliveries = new Set();
|
|
const inFlight = new Set();
|
|
let storageReady = false;
|
|
let storageError = "durable-inbox-starting";
|
|
let capacityExhausted = false;
|
|
let stopped = true;
|
|
let scanTimer = null;
|
|
let scanRunning = false;
|
|
let lastCleanupAt = 0;
|
|
|
|
async function start() {
|
|
if (!stopped) return;
|
|
stopped = false;
|
|
try {
|
|
await store.initialize();
|
|
await store.recoverProcessing(new Date(now()).toISOString());
|
|
storageReady = true;
|
|
storageError = null;
|
|
capacityExhausted = false;
|
|
safeLog(options.log, { event: "github-to-gitea-inbox-ready", ...store.usage(), valuesPrinted: false });
|
|
wake(0);
|
|
} catch (error) {
|
|
storageReady = false;
|
|
storageError = typedStorageError(error);
|
|
safeLog(options.log, { event: "github-to-gitea-inbox-unavailable", ok: false, errorType: storageError, error: redact(String(error?.message || error)), valuesPrinted: false });
|
|
wake(retry.scanIntervalMs);
|
|
}
|
|
}
|
|
|
|
async function accept(repo, delivery) {
|
|
if (!storageReady) throw new HttpError(503, storageError || "durable-inbox-unavailable");
|
|
try {
|
|
const accepted = await store.accept(repo, delivery);
|
|
capacityExhausted = false;
|
|
safeLog(options.log, deliveryLog("github-to-gitea-delivery-accepted", repo, delivery, {
|
|
ok: true,
|
|
state: accepted.record.state,
|
|
duplicate: accepted.duplicate,
|
|
}));
|
|
wake(0);
|
|
return {
|
|
state: accepted.record.state,
|
|
duplicate: accepted.duplicate,
|
|
result: accepted.record.state === "committed" && accepted.record.result
|
|
? publicSyncResult(accepted.record.result)
|
|
: null,
|
|
};
|
|
} catch (error) {
|
|
if (error instanceof DurableInboxError && error.status === 409) throw new HttpError(409, error.message);
|
|
if (!(error instanceof DurableInboxError && error.message === "durable-inbox-capacity-exhausted")) {
|
|
storageReady = false;
|
|
storageError = typedStorageError(error);
|
|
} else {
|
|
capacityExhausted = true;
|
|
}
|
|
safeLog(options.log, deliveryLog("github-to-gitea-delivery-persist-failed", repo, delivery, {
|
|
ok: false,
|
|
state: "rejected",
|
|
errorType: error instanceof DurableInboxError ? error.message : "durable-inbox-write-failed",
|
|
error: redact(String(error?.message || error)),
|
|
}));
|
|
throw new HttpError(503, error instanceof DurableInboxError ? error.message : "durable-inbox-write-failed");
|
|
}
|
|
}
|
|
|
|
function readiness() {
|
|
const usage = store.usage();
|
|
return {
|
|
ready: storageReady && usage.capacityAvailable && !capacityExhausted,
|
|
storageReady,
|
|
capacityAvailable: usage.capacityAvailable && !capacityExhausted,
|
|
totalBytes: usage.totalBytes,
|
|
maxBytes: usage.maxBytes,
|
|
errorType: storageReady ? (usage.capacityAvailable && !capacityExhausted ? null : "durable-inbox-capacity-exhausted") : storageError,
|
|
};
|
|
}
|
|
|
|
async function status() {
|
|
let records = [];
|
|
if (storageReady) {
|
|
try {
|
|
records = await store.list();
|
|
} catch (error) {
|
|
storageReady = false;
|
|
storageError = typedStorageError(error);
|
|
}
|
|
}
|
|
const counts = { accepted: 0, processing: 0, committed: 0, failed: 0 };
|
|
for (const record of records) {
|
|
if (Object.hasOwn(counts, record.state)) counts[record.state] += 1;
|
|
}
|
|
return {
|
|
...readiness(),
|
|
journal: storageReady ? store.journalMetadata() : null,
|
|
counts,
|
|
deliveries: statusRecords(records).map(publicInboxRecord),
|
|
failedDeliveryRecovery: await readFailedDeliveryRecoveryStatus(options.failedDeliveryRecoveryStatePath),
|
|
};
|
|
}
|
|
|
|
function wake(delayMs) {
|
|
if (stopped) return;
|
|
if (scanTimer !== null) clearTimeout(scanTimer);
|
|
scanTimer = setTimeout(() => void scan(), Math.max(0, delayMs));
|
|
}
|
|
|
|
async function scan() {
|
|
if (stopped || scanRunning) return;
|
|
scanRunning = true;
|
|
try {
|
|
if (!storageReady) {
|
|
await store.initialize();
|
|
await store.recoverProcessing(new Date(now()).toISOString());
|
|
storageReady = true;
|
|
storageError = null;
|
|
}
|
|
if (now() - lastCleanupAt >= options.inbox.cleanupIntervalMs) {
|
|
const removed = await store.cleanupCommitted(now() - options.inbox.committedRetentionMs);
|
|
lastCleanupAt = now();
|
|
if (removed > 0) {
|
|
capacityExhausted = false;
|
|
safeLog(options.log, { event: "github-to-gitea-inbox-retention", removed, ...store.usage(), valuesPrinted: false });
|
|
}
|
|
}
|
|
const currentTime = now();
|
|
for (const record of sortInboxRecordsByAcceptedSequence(await store.list())) {
|
|
if (!["accepted", "failed"].includes(record.state)) continue;
|
|
if (Date.parse(record.nextAttemptAt || record.receivedAt) > currentTime) continue;
|
|
schedule(record);
|
|
}
|
|
} catch (error) {
|
|
storageReady = false;
|
|
storageError = typedStorageError(error);
|
|
safeLog(options.log, { event: "github-to-gitea-inbox-scan-failed", ok: false, errorType: storageError, error: redact(String(error?.message || error)), valuesPrinted: false });
|
|
} finally {
|
|
scanRunning = false;
|
|
wake(retry.scanIntervalMs);
|
|
}
|
|
}
|
|
|
|
function schedule(record) {
|
|
if (activeDeliveries.has(record.deliveryId) || stopped) return;
|
|
activeDeliveries.add(record.deliveryId);
|
|
const previous = repoTails.get(record.repoKey) ?? Promise.resolve();
|
|
const task = previous.catch(() => undefined).then(() => processRecord(record.deliveryId))
|
|
.catch((error) => {
|
|
if (error instanceof DurableInboxError) {
|
|
storageReady = false;
|
|
storageError = typedStorageError(error);
|
|
}
|
|
safeLog(options.log, {
|
|
event: "github-to-gitea-worker-exception",
|
|
deliveryId: record.deliveryId,
|
|
repo: record.repoKey,
|
|
ok: false,
|
|
errorType: "durable-worker-exception",
|
|
error: redact(String(error?.message || error)),
|
|
valuesPrinted: false,
|
|
});
|
|
})
|
|
.finally(() => {
|
|
activeDeliveries.delete(record.deliveryId);
|
|
inFlight.delete(task);
|
|
wake(0);
|
|
});
|
|
repoTails.set(record.repoKey, task.catch(() => undefined));
|
|
inFlight.add(task);
|
|
}
|
|
|
|
async function processRecord(deliveryId) {
|
|
let record = await store.get(deliveryId);
|
|
if (!record || record.state === "committed") return;
|
|
const repo = repoByKey.get(record.repoKey);
|
|
if (!repo) {
|
|
await markFailed(record, { ok: false, retryable: false, errorType: "repository-config-missing" });
|
|
return;
|
|
}
|
|
const attemptStartedAt = new Date(now()).toISOString();
|
|
record = await store.update(deliveryId, (current) => ({
|
|
...current,
|
|
state: "processing",
|
|
updatedAt: attemptStartedAt,
|
|
processingAt: attemptStartedAt,
|
|
totalAttempts: current.totalAttempts + 1,
|
|
cycleAttempt: current.cycleAttempt + 1,
|
|
}));
|
|
const delivery = recordDelivery(record);
|
|
safeLog(options.log, deliveryLog("github-to-gitea-delivery-processing", repo, delivery, {
|
|
ok: true,
|
|
state: "processing",
|
|
totalAttempts: record.totalAttempts,
|
|
cycleAttempt: record.cycleAttempt,
|
|
}));
|
|
let result;
|
|
try {
|
|
result = validateExecutorTerminalResult(await executor.execute(repo, delivery));
|
|
} catch (error) {
|
|
await markFailed(record, {
|
|
ok: false,
|
|
retryable: true,
|
|
errorType: workerAbort.signal.aborted || stopped
|
|
? "source-sync-executor-aborted"
|
|
: error instanceof ExecutorTerminalResultError
|
|
? "source-sync-executor-invalid-result"
|
|
: "source-sync-executor-threw",
|
|
error: redact(String(error?.message || error)),
|
|
});
|
|
return;
|
|
}
|
|
if (result.ok) {
|
|
const committedAt = new Date(now()).toISOString();
|
|
await store.update(deliveryId, (current) => ({
|
|
...current,
|
|
state: "committed",
|
|
updatedAt: committedAt,
|
|
committedAt,
|
|
nextAttemptAt: null,
|
|
result: publicSyncResult(result),
|
|
lastError: null,
|
|
}));
|
|
safeLog(options.log, deliveryLog("github-to-gitea-delivery-committed", repo, delivery, {
|
|
ok: true,
|
|
state: "committed",
|
|
sourceCommit: result.sourceCommit,
|
|
authorityCommit: result.authorityCommit,
|
|
snapshotRef: result.snapshotRef,
|
|
disposition: result.disposition,
|
|
totalAttempts: record.totalAttempts,
|
|
}));
|
|
return;
|
|
}
|
|
await markFailed(record, result);
|
|
}
|
|
|
|
async function markFailed(record, result) {
|
|
const repo = repoByKey.get(record.repoKey) ?? { key: record.repoKey };
|
|
const terminal = result.retryable === false || record.cycleAttempt >= retry.maxAttempts;
|
|
const delayMs = terminal
|
|
? retry.terminalRetryDelayMs
|
|
: retryDelayMs(record.cycleAttempt, retry.initialDelayMs, retry.maxDelayMs);
|
|
const failedAt = new Date(now()).toISOString();
|
|
const nextAttemptAt = new Date(now() + delayMs).toISOString();
|
|
await store.update(record.deliveryId, (current) => ({
|
|
...current,
|
|
state: "failed",
|
|
updatedAt: failedAt,
|
|
failedAt,
|
|
nextAttemptAt,
|
|
cycleAttempt: terminal ? 0 : current.cycleAttempt,
|
|
result: null,
|
|
lastError: {
|
|
errorType: result.errorType || "source-sync-failed",
|
|
retryable: result.retryable !== false,
|
|
terminal,
|
|
},
|
|
}));
|
|
safeLog(options.log, deliveryLog("github-to-gitea-delivery-failed", repo, recordDelivery(record), {
|
|
ok: false,
|
|
state: "failed",
|
|
errorType: result.errorType || "source-sync-failed",
|
|
error: result.error ? redact(result.error) : null,
|
|
terminal,
|
|
automaticRetryScheduled: true,
|
|
nextAttemptAt,
|
|
totalAttempts: record.totalAttempts,
|
|
cycleAttempt: record.cycleAttempt,
|
|
}));
|
|
}
|
|
|
|
async function stop() {
|
|
if (stopped && storageError === "durable-inbox-stopping") return;
|
|
stopped = true;
|
|
storageReady = false;
|
|
storageError = "durable-inbox-stopping";
|
|
if (scanTimer !== null) clearTimeout(scanTimer);
|
|
workerAbort.abort();
|
|
const settling = Promise.allSettled([...inFlight]);
|
|
await Promise.race([settling, sleep(options.shutdownGraceMs)]);
|
|
}
|
|
|
|
return { start, stop, accept, readiness, status, scan };
|
|
}
|
|
|
|
export function createSyncCoordinator(options) {
|
|
const states = new Map();
|
|
const deliveryOwners = new Map();
|
|
const sync = options.sync ?? ((repo, delivery, deadline) => syncRepository(repo, delivery, { ...options, ...deadline }));
|
|
const wait = options.sleep ?? sleep;
|
|
const now = options.now ?? Date.now;
|
|
|
|
function stateFor(key) {
|
|
let state = states.get(key);
|
|
if (!state) {
|
|
state = { tail: Promise.resolve(), active: null, lastSuccess: null };
|
|
states.set(key, state);
|
|
}
|
|
return state;
|
|
}
|
|
|
|
function execute(repo, delivery) {
|
|
const requestStartedAt = now();
|
|
const deadlineAt = requestStartedAt + options.responseBudgetMs;
|
|
const fingerprint = deliveryFingerprint(repo, delivery);
|
|
const owner = deliveryOwners.get(delivery.deliveryId);
|
|
if (owner && owner.fingerprint !== fingerprint) {
|
|
throw new HttpError(409, "github-delivery-id-payload-conflict");
|
|
}
|
|
if (owner?.promise) {
|
|
safeLog(options.log, deliveryLog("github-to-gitea-delivery-duplicate", repo, delivery, { disposition: "join-in-flight" }));
|
|
return owner.promise.then((result) => ({ ...result, duplicate: true }));
|
|
}
|
|
if (owner && owner.fingerprint === fingerprint && owner.result?.ok === true) {
|
|
safeLog(options.log, deliveryLog("github-to-gitea-delivery-duplicate", repo, delivery, { disposition: "durable-refs-already-committed" }));
|
|
return Promise.resolve({ ...owner.result, duplicate: true });
|
|
}
|
|
const state = stateFor(repo.key);
|
|
if (state.lastSuccess?.fingerprint === fingerprint) {
|
|
safeLog(options.log, deliveryLog("github-to-gitea-delivery-duplicate", repo, delivery, { disposition: "durable-refs-already-committed" }));
|
|
return Promise.resolve({ ...state.lastSuccess.result, duplicate: true });
|
|
}
|
|
const promise = state.tail.then(async () => {
|
|
state.active = { delivery, fingerprint };
|
|
const startedAt = requestStartedAt;
|
|
let terminal = null;
|
|
for (let syncAttempt = 1; syncAttempt <= options.retryMaxAttempts; syncAttempt += 1) {
|
|
if (now() >= deadlineAt) {
|
|
terminal = { ok: false, retryable: true, errorType: "response-deadline-exhausted" };
|
|
safeLog(options.log, deliveryLog("github-to-gitea-sync", repo, delivery, {
|
|
ok: false,
|
|
errorType: terminal.errorType,
|
|
syncAttempt,
|
|
maxSyncAttempts: options.retryMaxAttempts,
|
|
willRetry: false,
|
|
terminal: true,
|
|
elapsedMs: now() - startedAt,
|
|
}));
|
|
break;
|
|
}
|
|
let result;
|
|
try {
|
|
result = await sync(repo, delivery, { deadlineAt, now, signal: options.signal });
|
|
} catch (error) {
|
|
result = { ok: false, retryable: true, errorType: "sync-exception", error: String(error?.message || error) };
|
|
}
|
|
if (now() >= deadlineAt) {
|
|
result = {
|
|
ok: false,
|
|
retryable: true,
|
|
errorType: "response-deadline-exhausted",
|
|
sourceCommit: result.sourceCommit ?? null,
|
|
authorityCommit: result.authorityCommit ?? null,
|
|
snapshotRef: result.snapshotRef ?? null,
|
|
committedBeforeDeadlineUnknown: result.ok === true,
|
|
};
|
|
}
|
|
const retryable = result.retryable !== false;
|
|
const willRetry = !result.ok && retryable && syncAttempt < options.retryMaxAttempts;
|
|
terminal = result;
|
|
safeLog(options.log, deliveryLog("github-to-gitea-sync", repo, delivery, {
|
|
ok: result.ok,
|
|
sourceCommit: result.sourceCommit ?? null,
|
|
authorityCommit: result.authorityCommit ?? null,
|
|
snapshotRef: result.snapshotRef ?? null,
|
|
disposition: result.disposition ?? null,
|
|
errorType: result.errorType ?? null,
|
|
error: result.error ? redact(result.error) : null,
|
|
syncAttempt,
|
|
maxSyncAttempts: options.retryMaxAttempts,
|
|
willRetry,
|
|
terminal: !willRetry,
|
|
elapsedMs: now() - startedAt,
|
|
}));
|
|
if (!willRetry) break;
|
|
try {
|
|
const remainingMs = Math.max(0, deadlineAt - now());
|
|
await wait(Math.min(remainingMs, retryDelayMs(syncAttempt, options.retryInitialDelayMs, options.retryMaxDelayMs)));
|
|
} catch (error) {
|
|
safeLog(options.log, deliveryLog("github-to-gitea-retry-delay-error", repo, delivery, { error: redact(String(error?.message || error)) }));
|
|
}
|
|
}
|
|
if (terminal?.ok) state.lastSuccess = { fingerprint, result: terminal };
|
|
const previousDeliveryId = state.lastDeliveryId;
|
|
state.lastDeliveryId = delivery.deliveryId;
|
|
deliveryOwners.set(delivery.deliveryId, { fingerprint, promise: null, result: terminal });
|
|
if (previousDeliveryId && previousDeliveryId !== delivery.deliveryId) deliveryOwners.delete(previousDeliveryId);
|
|
return terminal ?? { ok: false, errorType: "sync-produced-no-result", retryable: true };
|
|
});
|
|
const settled = promise.finally(() => {
|
|
state.active = null;
|
|
const current = deliveryOwners.get(delivery.deliveryId);
|
|
if (current?.promise === settled) deliveryOwners.set(delivery.deliveryId, { fingerprint, promise: null, result: null });
|
|
});
|
|
state.tail = settled.catch(() => undefined);
|
|
deliveryOwners.set(delivery.deliveryId, { fingerprint, promise: settled });
|
|
return settled;
|
|
}
|
|
|
|
return { execute, stateFor };
|
|
}
|
|
|
|
export async function syncRepository(repo, delivery, runtime) {
|
|
const work = mkdtempSync(join(tmpdir(), `gitea-gh-sync-${repo.key}-`));
|
|
let sourceCommit = null;
|
|
let snapshotRef = null;
|
|
let authorityCommit = null;
|
|
const execute = (args) => runWithinDeadline(runtime, args);
|
|
const executeResult = (args) => runResultWithinDeadline(runtime, args);
|
|
try {
|
|
await execute(["git", "init", "--bare", work]);
|
|
const snapshotRefForRequest = `${repo.snapshot.prefix.replace(/\/+$/u, "")}/${delivery.requestedCommit}`;
|
|
const giteaUrl = runtime.giteaBaseUrl + new URL(repo.gitea.readUrl).pathname;
|
|
const branchRef = `refs/heads/${repo.upstream.branch}`;
|
|
const initialRemoteRefs = parseLsRemote((await execute([
|
|
"git", "-C", work, "-c", `http.extraHeader=${runtime.giteaAuthHeader}`,
|
|
"ls-remote", giteaUrl, branchRef, snapshotRefForRequest,
|
|
])).stdout);
|
|
const currentCommit = initialRemoteRefs.get(branchRef) ?? null;
|
|
const snapshotCommit = initialRemoteRefs.get(snapshotRefForRequest) ?? null;
|
|
if (currentCommit) {
|
|
await execute(["git", "-C", work, "-c", `http.extraHeader=${runtime.giteaAuthHeader}`, "fetch", "--no-tags", giteaUrl, `+${branchRef}:refs/remotes/gitea/${repo.upstream.branch}`]);
|
|
}
|
|
if (snapshotCommit && snapshotCommit !== currentCommit) {
|
|
await execute(["git", "-C", work, "-c", `http.extraHeader=${runtime.giteaAuthHeader}`, "fetch", "--no-tags", giteaUrl, `+${snapshotRefForRequest}:refs/remotes/gitea/requested-snapshot`]);
|
|
}
|
|
let requestedObject = await executeResult(["git", "-C", work, "cat-file", "-e", `${delivery.requestedCommit}^{commit}`]);
|
|
if (requestedObject.status !== 0) {
|
|
const branchFetch = await executeResult([
|
|
"git", "-C", work, "-c", `http.extraHeader=${runtime.githubAuthHeader}`,
|
|
"fetch", "--no-tags", repo.upstream.cloneUrl,
|
|
`+refs/heads/${repo.upstream.branch}:refs/remotes/github/${repo.upstream.branch}`,
|
|
]);
|
|
requestedObject = await executeResult(["git", "-C", work, "cat-file", "-e", `${delivery.requestedCommit}^{commit}`]);
|
|
if (requestedObject.status !== 0) {
|
|
const exactFetch = await executeResult(["git", "-C", work, "-c", `http.extraHeader=${runtime.githubAuthHeader}`, "fetch", "--no-tags", repo.upstream.cloneUrl, `${delivery.requestedCommit}:refs/remotes/github/delivery`]);
|
|
if (exactFetch.status !== 0) {
|
|
return {
|
|
ok: false,
|
|
retryable: true,
|
|
errorType: "requested-commit-temporarily-unavailable",
|
|
error: `${branchFetch.stderr}\n${exactFetch.stderr}`,
|
|
};
|
|
}
|
|
requestedObject = await executeResult(["git", "-C", work, "cat-file", "-e", `${delivery.requestedCommit}^{commit}`]);
|
|
}
|
|
}
|
|
if (requestedObject.status !== 0) return { ok: false, retryable: false, errorType: "requested-commit-not-a-commit" };
|
|
sourceCommit = (await execute(["git", "-C", work, "rev-parse", `${delivery.requestedCommit}^{commit}`])).stdout.trim();
|
|
if (sourceCommit !== delivery.requestedCommit) {
|
|
return {
|
|
ok: false,
|
|
retryable: false,
|
|
errorType: "source-commit-mismatch",
|
|
sourceCommit,
|
|
error: `fetched source commit ${sourceCommit} does not equal webhook after ${delivery.requestedCommit}`,
|
|
};
|
|
}
|
|
snapshotRef = snapshotRefForRequest;
|
|
if (snapshotCommit && snapshotCommit !== sourceCommit) {
|
|
return { ok: false, retryable: false, errorType: "immutable-snapshot-conflict", sourceCommit, authorityCommit: currentCommit, snapshotRef };
|
|
}
|
|
let branchDisposition = "created";
|
|
let updateBranch = true;
|
|
if (currentCommit) {
|
|
const currentIsAncestor = await ancestry(executeResult, work, currentCommit, sourceCommit);
|
|
const sourceIsAncestor = await ancestry(executeResult, work, sourceCommit, currentCommit);
|
|
if (currentCommit === sourceCommit) {
|
|
branchDisposition = "current";
|
|
updateBranch = false;
|
|
} else if (currentIsAncestor === true) {
|
|
branchDisposition = "fast-forwarded";
|
|
} else if (sourceIsAncestor === true) {
|
|
branchDisposition = "superseded";
|
|
updateBranch = false;
|
|
} else if (currentIsAncestor === null || sourceIsAncestor === null) {
|
|
return { ok: false, retryable: true, errorType: "ancestry-proof-failed", sourceCommit, authorityCommit: currentCommit, snapshotRef };
|
|
} else {
|
|
return { ok: false, retryable: false, errorType: "source-history-diverged", sourceCommit, authorityCommit: currentCommit, snapshotRef };
|
|
}
|
|
}
|
|
const refspecs = [];
|
|
const pushArgs = ["git", "-C", work, "-c", `http.extraHeader=${runtime.giteaAuthHeader}`, "push", "--atomic"];
|
|
if (updateBranch) {
|
|
pushArgs.push(`--force-with-lease=${branchRef}:${currentCommit ?? ""}`);
|
|
refspecs.push(`${sourceCommit}:${branchRef}`);
|
|
}
|
|
if (!snapshotCommit) refspecs.push(`${sourceCommit}:${snapshotRef}`);
|
|
if (refspecs.length > 0) {
|
|
const pushed = await executeResult([...pushArgs, giteaUrl, ...refspecs]);
|
|
if (pushed.status !== 0) {
|
|
return { ok: false, retryable: true, errorType: "atomic-push-race-or-failure", sourceCommit, authorityCommit: currentCommit, snapshotRef, error: pushed.stderr };
|
|
}
|
|
}
|
|
authorityCommit = branchDisposition === "superseded" ? currentCommit : sourceCommit;
|
|
const proofRefs = parseLsRemote((await execute([
|
|
"git", "-C", work, "-c", `http.extraHeader=${runtime.giteaAuthHeader}`,
|
|
"ls-remote", giteaUrl, branchRef, snapshotRef,
|
|
])).stdout);
|
|
const provenAuthorityCommit = proofRefs.get(branchRef) ?? null;
|
|
const provenSnapshotCommit = proofRefs.get(snapshotRef) ?? null;
|
|
if (provenAuthorityCommit !== authorityCommit || provenSnapshotCommit !== sourceCommit) {
|
|
return {
|
|
ok: false,
|
|
retryable: true,
|
|
errorType: "post-push-authority-proof-failed",
|
|
sourceCommit,
|
|
authorityCommit: provenAuthorityCommit,
|
|
snapshotRef,
|
|
};
|
|
}
|
|
return {
|
|
ok: true,
|
|
sourceCommit,
|
|
authorityCommit,
|
|
snapshotRef,
|
|
disposition: branchDisposition === "superseded" ? "superseded-with-immutable-snapshot" : "atomic-source-authority-committed",
|
|
};
|
|
} catch (error) {
|
|
return { ok: false, retryable: true, errorType: "git-source-sync-failed", sourceCommit, authorityCommit, snapshotRef, error: String(error?.message || error) };
|
|
} finally {
|
|
rmSync(work, { recursive: true, force: true });
|
|
}
|
|
}
|
|
|
|
export async function run(args, timeoutMs = 120000) {
|
|
const result = await runResult(args, timeoutMs);
|
|
if (result.status !== 0) {
|
|
throw new Error(`${args[0]} failed status=${result.status ?? "error"} stderr=${result.stderr}`);
|
|
}
|
|
return { stdout: result.stdout, stderr: result.stderr };
|
|
}
|
|
|
|
export function runResult(args, timeoutMs = 120000, signal = null) {
|
|
const boundedTimeoutMs = Math.max(1, Math.floor(timeoutMs));
|
|
if (signal?.aborted) return Promise.resolve({ status: null, stdout: "", stderr: "command aborted before spawn" });
|
|
return new Promise((resolve) => {
|
|
let stdout = "";
|
|
let stderr = "";
|
|
let timedOut = false;
|
|
let aborted = false;
|
|
let settled = false;
|
|
let deadlineTimer = null;
|
|
let forceKillTimer = null;
|
|
let abortCommand = () => {};
|
|
const child = spawn(args[0], args.slice(1), {
|
|
stdio: ["ignore", "pipe", "pipe"],
|
|
detached: process.platform !== "win32",
|
|
env: {
|
|
...process.env,
|
|
GIT_TERMINAL_PROMPT: "0",
|
|
GIT_CONFIG_NOSYSTEM: "1",
|
|
GIT_CONFIG_GLOBAL: "/dev/null",
|
|
},
|
|
});
|
|
const finish = (status, error = "") => {
|
|
if (settled) return;
|
|
settled = true;
|
|
if (deadlineTimer !== null) clearTimeout(deadlineTimer);
|
|
if (forceKillTimer !== null) clearTimeout(forceKillTimer);
|
|
signal?.removeEventListener("abort", abortCommand);
|
|
resolve({
|
|
status,
|
|
stdout,
|
|
stderr: appendTail(stderr, error, 1600),
|
|
});
|
|
};
|
|
child.stdout.on("data", (chunk) => {
|
|
stdout = appendTail(stdout, String(chunk), 65536);
|
|
});
|
|
child.stderr.on("data", (chunk) => {
|
|
stderr = appendTail(stderr, String(chunk), 1600);
|
|
});
|
|
child.once("error", (error) => finish(null, String(error?.message || error)));
|
|
child.once("close", (status, signal) => {
|
|
const terminalError = aborted
|
|
? `command aborted; signal=${signal || "none"}`
|
|
: timedOut
|
|
? `command deadline exceeded after ${boundedTimeoutMs}ms; signal=${signal || "none"}`
|
|
: "";
|
|
finish(aborted || timedOut ? null : status, terminalError);
|
|
});
|
|
abortCommand = () => {
|
|
aborted = true;
|
|
signalProcessTree(child, "SIGTERM");
|
|
forceKillTimer = setTimeout(() => signalProcessTree(child, "SIGKILL"), 50);
|
|
};
|
|
signal?.addEventListener("abort", abortCommand, { once: true });
|
|
if (signal?.aborted) abortCommand();
|
|
deadlineTimer = setTimeout(() => {
|
|
timedOut = true;
|
|
signalProcessTree(child, "SIGTERM");
|
|
forceKillTimer = setTimeout(() => signalProcessTree(child, "SIGKILL"), 50);
|
|
}, boundedTimeoutMs);
|
|
});
|
|
}
|
|
|
|
export function verifySignature(signatureHeader, body, webhookSecret) {
|
|
const signature = singleHeader(signatureHeader);
|
|
if (!signature || !signature.startsWith("sha256=")) throw new HttpError(401, "missing-github-signature");
|
|
const expected = `sha256=${createHmac("sha256", webhookSecret).update(body).digest("hex")}`;
|
|
if (!timingSafeEqualString(signature, expected)) throw new HttpError(401, "invalid-github-signature");
|
|
}
|
|
|
|
export function retryDelayMs(syncAttempt, initialDelayMs, maxDelayMs) {
|
|
return Math.min(maxDelayMs, initialDelayMs * (2 ** Math.max(0, syncAttempt - 1)));
|
|
}
|
|
|
|
export function loadRuntimeConfig() {
|
|
const githubToken = requiredEnv("GITHUB_TOKEN");
|
|
const giteaUsername = requiredEnv("GITEA_USERNAME");
|
|
const giteaPassword = requiredEnv("GITEA_PASSWORD");
|
|
return {
|
|
port: Number.parseInt(process.env.UNIDESK_GITEA_WEBHOOK_PORT || "8080", 10),
|
|
path: process.env.UNIDESK_GITEA_WEBHOOK_PATH || "/_unidesk/github-to-gitea",
|
|
responseBudgetMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_RESPONSE_BUDGET_MS"),
|
|
repos: JSON.parse(readFileSync(process.env.UNIDESK_GITEA_REPOS_PATH || "/etc/gitea-github-sync/repos.json", "utf8")),
|
|
webhookSecret: requiredEnv("GITHUB_WEBHOOK_SECRET"),
|
|
giteaBaseUrl: requiredEnv("UNIDESK_GITEA_SERVICE_BASE_URL").replace(/\/+$/u, ""),
|
|
inbox: {
|
|
path: requiredEnv("UNIDESK_GITEA_WEBHOOK_INBOX_PATH"),
|
|
maxBytes: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_INBOX_MAX_BYTES"),
|
|
committedRetentionMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_COMMITTED_RETENTION_SECONDS") * 1000,
|
|
cleanupIntervalMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_CLEANUP_INTERVAL_MS"),
|
|
},
|
|
failedDeliveryRecoveryStatePath: process.env.UNIDESK_GITEA_HOOK_RECOVERY_STATE_PATH || null,
|
|
workerRetry: {
|
|
maxAttempts: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_RETRY_MAX_ATTEMPTS"),
|
|
initialDelayMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_RETRY_INITIAL_DELAY_MS"),
|
|
maxDelayMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_RETRY_MAX_DELAY_MS"),
|
|
terminalRetryDelayMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_RETRY_TERMINAL_DELAY_MS"),
|
|
attemptTimeoutMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_WORKER_ATTEMPT_TIMEOUT_MS"),
|
|
scanIntervalMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_WORKER_SCAN_INTERVAL_MS"),
|
|
},
|
|
shutdownGraceMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_SHUTDOWN_GRACE_MS"),
|
|
maxBodyBytes: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_MAX_BODY_BYTES"),
|
|
githubAuthHeader: `Authorization: Basic ${Buffer.from(`x-access-token:${githubToken}`).toString("base64")}`,
|
|
giteaAuthHeader: `Authorization: Basic ${Buffer.from(`${giteaUsername}:${giteaPassword}`).toString("base64")}`,
|
|
log,
|
|
run,
|
|
runResult,
|
|
};
|
|
}
|
|
|
|
export async function readFailedDeliveryRecoveryStatus(statePath) {
|
|
if (!statePath) return { enabled: false, ready: true, state: "disabled", counts: {}, entries: [] };
|
|
let ledger;
|
|
try {
|
|
ledger = JSON.parse(await readFile(statePath, "utf8"));
|
|
} catch (error) {
|
|
if (error?.code === "ENOENT") return { enabled: true, ready: false, state: "not-initialized", counts: {}, entries: [] };
|
|
return { enabled: true, ready: false, state: "invalid-or-unreadable", errorType: "hook-recovery-ledger-unreadable", counts: {}, entries: [] };
|
|
}
|
|
if (ledger?.version !== 1 || ledger?.kind !== "GiteaGithubHookRecoveryLedger" || !ledger.records || typeof ledger.records !== "object" || Array.isArray(ledger.records)) {
|
|
return { enabled: true, ready: false, state: "invalid-or-unreadable", errorType: "hook-recovery-ledger-invalid", counts: {}, entries: [] };
|
|
}
|
|
const allowedStates = new Set(["pending", "attempted", "delivery-succeeded-awaiting-source", "source-committed", "terminal-exhausted"]);
|
|
const records = Object.values(ledger.records);
|
|
const validRecord = (item) => item && typeof item === "object"
|
|
&& allowedStates.has(item.state)
|
|
&& typeof item.repository === "string" && item.repository.length > 0
|
|
&& (typeof item.hookId === "number" || typeof item.hookId === "string")
|
|
&& typeof item.deliveryId === "string" && item.deliveryId.length > 0
|
|
&& Number.isInteger(item.attempts) && item.attempts >= 0
|
|
&& Number.isInteger(item.cooldownMs) && item.cooldownMs >= 0
|
|
&& typeof item.updatedAt === "string" && item.updatedAt.length > 0;
|
|
if (records.some((item) => !validRecord(item))) {
|
|
return { enabled: true, ready: false, state: "invalid-or-unreadable", errorType: "hook-recovery-ledger-record-invalid", counts: {}, entries: [] };
|
|
}
|
|
const counts = { pending: 0, attempted: 0, "delivery-succeeded-awaiting-source": 0, "source-committed": 0, "terminal-exhausted": 0 };
|
|
for (const item of records) counts[item.state] += 1;
|
|
const entries = records
|
|
.sort((left, right) => String(right.updatedAt || "").localeCompare(String(left.updatedAt || "")))
|
|
.slice(0, 50)
|
|
.map((item) => ({
|
|
repository: item.repository,
|
|
hookId: item.hookId,
|
|
deliveryId: item.deliveryId,
|
|
state: item.state,
|
|
attempts: item.attempts,
|
|
cooldownMs: item.cooldownMs,
|
|
nextAt: item.nextAt ?? null,
|
|
updatedAt: item.updatedAt,
|
|
lastStatusCode: item.lastStatusCode ?? null,
|
|
completionReason: item.completionReason ?? null,
|
|
}));
|
|
return { enabled: true, ready: true, state: "ready", counts, entries };
|
|
}
|
|
|
|
export function startServer(config = loadRuntimeConfig()) {
|
|
const controller = config.controller ?? createDurableSyncController(config);
|
|
const server = createServer(createGithubSyncHandler({ ...config, controller }));
|
|
void controller.start();
|
|
server.listen(config.port, "0.0.0.0", () => {
|
|
safeLog(config.log, { event: "github-to-gitea-sync-listening", port: config.port, path: config.path, repos: config.repos.map((repo) => repo.key), valuesPrinted: false });
|
|
});
|
|
const shutdown = async () => {
|
|
server.close();
|
|
await controller.stop();
|
|
};
|
|
server.once("close", () => void controller.stop());
|
|
process.once("SIGTERM", shutdown);
|
|
process.once("SIGINT", shutdown);
|
|
return server;
|
|
}
|
|
|
|
class HttpError extends Error {
|
|
constructor(status, message) {
|
|
super(message);
|
|
this.status = status;
|
|
}
|
|
}
|
|
|
|
class DurableInboxError extends Error {
|
|
constructor(status, message) {
|
|
super(message);
|
|
this.status = status;
|
|
}
|
|
}
|
|
|
|
class ExecutorTerminalResultError extends Error {}
|
|
|
|
function validateExecutorTerminalResult(result) {
|
|
if (!result || typeof result !== "object" || typeof result.ok !== "boolean") {
|
|
throw new ExecutorTerminalResultError("executor terminal result must be an object with boolean ok");
|
|
}
|
|
if (result.ok === false) {
|
|
if (typeof result.retryable !== "boolean" || !/^[a-z0-9][a-z0-9-]{0,127}$/u.test(result.errorType || "")) {
|
|
throw new ExecutorTerminalResultError("failed executor terminal result must include retryable and typed errorType");
|
|
}
|
|
return result;
|
|
}
|
|
if (!/^[0-9a-f]{40,64}$/u.test(result.sourceCommit || "")
|
|
|| !/^[0-9a-f]{40,64}$/u.test(result.authorityCommit || "")
|
|
|| !/^refs\/[A-Za-z0-9._/-]+$/u.test(result.snapshotRef || "")
|
|
|| !["atomic-source-authority-committed", "superseded-with-immutable-snapshot"].includes(result.disposition)) {
|
|
throw new ExecutorTerminalResultError("successful executor terminal result must include source, authority, snapshot and disposition proof");
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function deliveryFingerprint(repo, delivery) {
|
|
return createHash("sha256")
|
|
.update([repo.key, delivery.deliveryId, delivery.repository, delivery.ref, delivery.requestedCommit, delivery.payloadSha].join("\0"))
|
|
.digest("hex");
|
|
}
|
|
|
|
function nonRetryableStatus(errorType) {
|
|
return [
|
|
"github-delivery-id-payload-conflict",
|
|
"source-commit-mismatch",
|
|
"immutable-snapshot-conflict",
|
|
"source-history-diverged",
|
|
].includes(errorType) ? 409 : 422;
|
|
}
|
|
|
|
function safeLog(logger, payload) {
|
|
try {
|
|
logger(payload);
|
|
} catch {
|
|
// Logging must never change delivery success or retry semantics.
|
|
}
|
|
}
|
|
|
|
function parseLsRemote(output) {
|
|
const refs = new Map();
|
|
for (const line of String(output || "").split("\n")) {
|
|
const [commit, ref] = line.trim().split(/\s+/u);
|
|
if (commit && ref) refs.set(ref, commit);
|
|
}
|
|
return refs;
|
|
}
|
|
|
|
async function ancestry(executeResult, work, ancestor, descendant) {
|
|
const result = await executeResult(["git", "-C", work, "merge-base", "--is-ancestor", ancestor, descendant]);
|
|
if (result.status === 0) return true;
|
|
if (result.status === 1) return false;
|
|
return null;
|
|
}
|
|
|
|
async function runWithinDeadline(runtime, args) {
|
|
const result = await runResultWithinDeadline(runtime, args);
|
|
if (result.status !== 0) throw new Error(`${args[0]} failed status=${result.status ?? "deadline"} stderr=${result.stderr}`);
|
|
return { stdout: result.stdout, stderr: result.stderr };
|
|
}
|
|
|
|
async function runResultWithinDeadline(runtime, args) {
|
|
const remainingMs = Math.floor(runtime.deadlineAt - runtime.now());
|
|
if (remainingMs < 1) return { status: null, stdout: "", stderr: "response deadline exhausted before command" };
|
|
if (runtime.signal?.aborted) return { status: null, stdout: "", stderr: "command aborted before spawn" };
|
|
return await runtime.runResult(args, remainingMs, runtime.signal);
|
|
}
|
|
|
|
function appendTail(previous, next, limit) {
|
|
return `${previous}${next}`.slice(-limit);
|
|
}
|
|
|
|
function signalProcessTree(child, signal) {
|
|
if (process.platform !== "win32" && child.pid) {
|
|
try {
|
|
process.kill(-child.pid, signal);
|
|
return;
|
|
} catch {
|
|
// Fall back to the direct child when the process group already exited.
|
|
}
|
|
}
|
|
child.kill(signal);
|
|
}
|
|
|
|
function recordPath(directory, deliveryId) {
|
|
return join(directory, `${createHash("sha256").update(String(deliveryId)).digest("hex")}.json`);
|
|
}
|
|
|
|
function serializeRecord(record) {
|
|
return `${JSON.stringify(record)}\n`;
|
|
}
|
|
|
|
async function readRecordPath(path, missingIsNull = false) {
|
|
try {
|
|
const parsed = JSON.parse(await readFile(path, "utf8"));
|
|
return validateInboxRecord(parsed);
|
|
} catch (error) {
|
|
if (missingIsNull && error?.code === "ENOENT") return null;
|
|
if (error instanceof DurableInboxError) throw error;
|
|
throw new DurableInboxError(503, `durable-inbox-read-failed:${error?.code || "unknown"}`);
|
|
}
|
|
}
|
|
|
|
async function durableWrite(path, content) {
|
|
const temporaryPath = `${path}.tmp-${randomUUID()}`;
|
|
let handle = null;
|
|
try {
|
|
handle = await open(temporaryPath, "wx", 0o600);
|
|
await handle.writeFile(content, "utf8");
|
|
await handle.sync();
|
|
await handle.close();
|
|
handle = null;
|
|
await rename(temporaryPath, path);
|
|
await syncDirectory(join(path, ".."));
|
|
} catch (error) {
|
|
if (handle) await handle.close().catch(() => undefined);
|
|
await unlink(temporaryPath).catch(() => undefined);
|
|
if (error instanceof DurableInboxError) throw error;
|
|
throw new DurableInboxError(503, `durable-inbox-write-failed:${error?.code || "unknown"}`);
|
|
}
|
|
}
|
|
|
|
async function syncDirectory(directory) {
|
|
const handle = await open(directory, "r");
|
|
try {
|
|
await handle.sync();
|
|
} finally {
|
|
await handle.close();
|
|
}
|
|
}
|
|
|
|
function typedStorageError(error) {
|
|
if (error instanceof DurableInboxError) return error.message.split(":", 1)[0];
|
|
return "durable-inbox-unavailable";
|
|
}
|
|
|
|
function recordDelivery(record) {
|
|
return {
|
|
deliveryId: record.deliveryId,
|
|
repository: record.repository,
|
|
ref: record.ref,
|
|
requestedCommit: record.requestedCommit,
|
|
payloadSha: record.payloadSha,
|
|
receivedAt: record.receivedAt,
|
|
};
|
|
}
|
|
|
|
function publicSyncResult(result) {
|
|
return {
|
|
sourceCommit: result.sourceCommit ?? null,
|
|
authorityCommit: result.authorityCommit ?? null,
|
|
snapshotRef: result.snapshotRef ?? null,
|
|
disposition: result.disposition ?? null,
|
|
};
|
|
}
|
|
|
|
function publicInboxRecord(record) {
|
|
return {
|
|
deliveryId: record.deliveryId,
|
|
repo: record.repoKey,
|
|
repository: record.repository,
|
|
ref: record.ref,
|
|
requestedCommit: record.requestedCommit,
|
|
payloadSha: record.payloadSha,
|
|
receivedAt: record.receivedAt,
|
|
state: record.state,
|
|
acceptedSequence: record.acceptedSequence,
|
|
totalAttempts: record.totalAttempts,
|
|
cycleAttempt: record.cycleAttempt,
|
|
nextAttemptAt: record.nextAttemptAt,
|
|
updatedAt: record.updatedAt,
|
|
result: record.result,
|
|
lastError: record.lastError,
|
|
};
|
|
}
|
|
|
|
function statusRecords(records) {
|
|
const selected = new Map();
|
|
for (const record of records.slice(-100)) selected.set(record.deliveryId, record);
|
|
const latestByRepo = new Map();
|
|
const latestCommittedByRepo = new Map();
|
|
for (const record of records) {
|
|
const current = latestByRepo.get(record.repoKey);
|
|
if (!current || record.acceptedSequence > current.acceptedSequence) latestByRepo.set(record.repoKey, record);
|
|
if (record.state === "committed") {
|
|
const committed = latestCommittedByRepo.get(record.repoKey);
|
|
if (!committed || record.acceptedSequence > committed.acceptedSequence) latestCommittedByRepo.set(record.repoKey, record);
|
|
}
|
|
}
|
|
for (const record of latestByRepo.values()) selected.set(record.deliveryId, record);
|
|
for (const record of latestCommittedByRepo.values()) selected.set(record.deliveryId, record);
|
|
return sortInboxRecordsByAcceptedSequence([...selected.values()]);
|
|
}
|
|
|
|
export function sortInboxRecordsByAcceptedSequence(records) {
|
|
return [...records].sort((left, right) => left.acceptedSequence - right.acceptedSequence);
|
|
}
|
|
|
|
function validateInboxRecord(record) {
|
|
const invalid = () => { throw new DurableInboxError(503, "durable-inbox-record-invalid"); };
|
|
if (!record || record.version !== 1) invalid();
|
|
if (typeof record.deliveryId !== "string" || record.deliveryId.length < 1 || record.deliveryId.length > 160) invalid();
|
|
if (!/^[a-zA-Z0-9._-]+$/u.test(record.repoKey || "")) invalid();
|
|
if (!/^[^/\s]+\/[^/\s]+$/u.test(record.repository || "")) invalid();
|
|
if (!/^refs\/heads\/[^\s]+$/u.test(record.ref || "")) invalid();
|
|
if (!/^[0-9a-f]{40,64}$/u.test(record.requestedCommit || "")) invalid();
|
|
if (!/^[0-9a-f]{64}$/u.test(record.payloadSha || "")) invalid();
|
|
if (!/^[0-9a-f]{64}$/u.test(record.fingerprint || "")) invalid();
|
|
if (!["accepted", "processing", "committed", "failed"].includes(record.state)) invalid();
|
|
if (!isTimestamp(record.receivedAt) || !isTimestamp(record.acceptedAt) || !isTimestamp(record.updatedAt)) invalid();
|
|
if (!Number.isInteger(record.acceptedSequence) || record.acceptedSequence < 1) invalid();
|
|
if (!Number.isInteger(record.totalAttempts) || record.totalAttempts < 0 || !Number.isInteger(record.cycleAttempt) || record.cycleAttempt < 0) invalid();
|
|
if (record.nextAttemptAt !== null && !isTimestamp(record.nextAttemptAt)) invalid();
|
|
if (record.lastError !== null) {
|
|
if (typeof record.lastError !== "object" || !/^[a-z0-9][a-z0-9-]{0,127}$/u.test(record.lastError.errorType || "")) invalid();
|
|
if (typeof record.lastError.retryable !== "boolean") invalid();
|
|
if (record.lastError.terminal !== undefined && typeof record.lastError.terminal !== "boolean") invalid();
|
|
}
|
|
const expectedFingerprint = createHash("sha256")
|
|
.update([record.repoKey, record.deliveryId, record.repository, record.ref, record.requestedCommit, record.payloadSha].join("\0"))
|
|
.digest("hex");
|
|
if (!timingSafeEqualString(record.fingerprint, expectedFingerprint)) invalid();
|
|
if (record.state === "accepted" && (!isTimestamp(record.nextAttemptAt) || record.result !== null)) invalid();
|
|
if (record.state === "processing" && (!isTimestamp(record.processingAt) || record.result !== null)) invalid();
|
|
if (record.state === "failed" && (!isTimestamp(record.failedAt) || !isTimestamp(record.nextAttemptAt) || record.lastError === null || record.result !== null)) invalid();
|
|
if (record.state === "committed") {
|
|
if (!isTimestamp(record.committedAt) || record.nextAttemptAt !== null || record.lastError !== null || !record.result || typeof record.result !== "object") invalid();
|
|
if (!/^[0-9a-f]{40,64}$/u.test(record.result.sourceCommit || "") || !/^[0-9a-f]{40,64}$/u.test(record.result.authorityCommit || "")) invalid();
|
|
if (record.result.sourceCommit !== record.requestedCommit) invalid();
|
|
if (!/^refs\/[a-zA-Z0-9._/-]+$/u.test(record.result.snapshotRef || "") || !record.result.snapshotRef.endsWith(`/${record.requestedCommit}`)) invalid();
|
|
if (!["atomic-source-authority-committed", "superseded-with-immutable-snapshot"].includes(record.result.disposition)) invalid();
|
|
}
|
|
return record;
|
|
}
|
|
|
|
function validateJournalMetadata(metadata) {
|
|
if (!metadata || metadata.version !== 1 || metadata.kind !== "GiteaGithubSyncDurableInbox" || !isTimestamp(metadata.createdAt)) {
|
|
throw new DurableInboxError(503, "durable-inbox-metadata-invalid");
|
|
}
|
|
return metadata;
|
|
}
|
|
|
|
function isTimestamp(value) {
|
|
return typeof value === "string" && Number.isFinite(Date.parse(value));
|
|
}
|
|
|
|
function deliveryLog(event, repo, delivery, extra = {}) {
|
|
return {
|
|
event,
|
|
repo: repo.key,
|
|
deliveryId: delivery.deliveryId,
|
|
repository: delivery.repository,
|
|
ref: delivery.ref,
|
|
requestedCommit: delivery.requestedCommit,
|
|
...extra,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function parsePayload(body) {
|
|
try {
|
|
return JSON.parse(body.toString("utf8") || "{}");
|
|
} catch {
|
|
throw new HttpError(400, "invalid-json-payload");
|
|
}
|
|
}
|
|
|
|
function singleHeader(value) {
|
|
return String(Array.isArray(value) ? value[0] : value || "").trim();
|
|
}
|
|
|
|
function timingSafeEqualString(a, b) {
|
|
const left = Buffer.from(a);
|
|
const right = Buffer.from(b);
|
|
return left.length === right.length && timingSafeEqual(left, right);
|
|
}
|
|
|
|
function readBody(req, maxBodyBytes) {
|
|
return new Promise((resolve, reject) => {
|
|
const chunks = [];
|
|
let bytes = 0;
|
|
let rejected = false;
|
|
req.on("data", (chunk) => {
|
|
bytes += chunk.length;
|
|
if (!rejected && bytes > maxBodyBytes) {
|
|
rejected = true;
|
|
reject(new HttpError(413, "request-body-too-large"));
|
|
return;
|
|
}
|
|
if (!rejected) chunks.push(chunk);
|
|
});
|
|
req.on("end", () => {
|
|
if (!rejected) resolve(Buffer.concat(chunks));
|
|
});
|
|
req.on("error", reject);
|
|
});
|
|
}
|
|
|
|
export function waitWithinResponseBudget(promise, deadlineMs, errorType) {
|
|
const remainingMs = deadlineMs - Date.now();
|
|
if (remainingMs <= 0) return Promise.reject(new HttpError(503, errorType));
|
|
return new Promise((resolve, reject) => {
|
|
let settled = false;
|
|
const timer = setTimeout(() => {
|
|
if (settled) return;
|
|
settled = true;
|
|
reject(new HttpError(503, errorType));
|
|
}, remainingMs);
|
|
Promise.resolve(promise).then(
|
|
(value) => {
|
|
if (settled) return;
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
resolve(value);
|
|
},
|
|
(error) => {
|
|
if (settled) return;
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
reject(error);
|
|
},
|
|
);
|
|
});
|
|
}
|
|
|
|
function writeJson(res, status, payload) {
|
|
if (res.headersSent || res.destroyed) return;
|
|
res.writeHead(status, { "Content-Type": "application/json" });
|
|
res.end(`${JSON.stringify(payload)}\n`);
|
|
}
|
|
|
|
function sleep(ms) {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
}
|
|
|
|
function log(payload) {
|
|
console.log(JSON.stringify(payload));
|
|
}
|
|
|
|
function redact(value) {
|
|
return String(value || "")
|
|
.replace(/Authorization: Basic [A-Za-z0-9+/=]+/gu, "Authorization: Basic <redacted>")
|
|
.replace(/x-access-token:[^@\s]+/gu, "x-access-token:<redacted>")
|
|
.slice(-1600);
|
|
}
|
|
|
|
function requiredEnv(name) {
|
|
const value = process.env[name];
|
|
if (!value) throw new Error(`missing env ${name}`);
|
|
return value;
|
|
}
|
|
|
|
function positiveIntegerEnv(name) {
|
|
const value = Number.parseInt(requiredEnv(name), 10);
|
|
if (!Number.isFinite(value) || value < 1) throw new Error(`invalid positive integer env ${name}`);
|
|
return value;
|
|
}
|