slhx agent
756 lines
34 KiB
JavaScript
756 lines
34 KiB
JavaScript
const DATABASE = "hemx-kanban-v1";
|
|
const DATABASE_VERSION = 3;
|
|
const COMMANDS = "commands";
|
|
const ACCOUNT_INDEX = "byAccountPartition";
|
|
const COMMAND_SCHEMA = 2;
|
|
const LEGACY_COMMAND_SCHEMA = 1;
|
|
const MIGRATION_KEY = "commandSchemaMigration";
|
|
const MAX_ATTEMPTS = 3;
|
|
const BACKOFF_MS = [25, 50];
|
|
const REQUEST_TIMEOUT_MS = 1_000;
|
|
const ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT = 64;
|
|
const root = document.querySelector("[data-kanban-sync]");
|
|
const TAB_ID = sessionStorage.getItem("hemx-kanban-sync-tab-id") || crypto.randomUUID();
|
|
const LEASE_MS = 5000;
|
|
const LEASE_POLL_MS = 100;
|
|
let database;
|
|
let accountPartition;
|
|
let uploadLimit;
|
|
let retryTimer;
|
|
let leaseTimer;
|
|
let acknowledgementSource;
|
|
const activeRequests = new Set();
|
|
let synchronizing = false;
|
|
let uploadsThisRun = 0;
|
|
let uploadedTotal = 0;
|
|
let inFlightUploads = 0;
|
|
let maxObservedInFlight = 0;
|
|
let acknowledgementStartedAt;
|
|
let conflictCount = 0;
|
|
let rejectionCount = 0;
|
|
let activeConflict;
|
|
let manualRetryCommand;
|
|
let stopped = false;
|
|
|
|
class UploadError extends Error {
|
|
constructor(status, retryable, kind, reason) {
|
|
super(`sync upload failed with ${status}`);
|
|
this.name = "UploadError";
|
|
this.status = status;
|
|
this.retryable = retryable;
|
|
this.kind = kind;
|
|
this.reason = reason;
|
|
}
|
|
}
|
|
|
|
// req: operations/003
|
|
export async function fetchWithTimeout(
|
|
input,
|
|
init = {},
|
|
fetchImplementation = fetch,
|
|
timeoutMs = REQUEST_TIMEOUT_MS,
|
|
) {
|
|
if (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1) {
|
|
throw new TypeError("sync request timeout must be a positive integer");
|
|
}
|
|
const controller = new AbortController();
|
|
const timeout = setTimeout(
|
|
() => controller.abort(new DOMException(`sync request timed out after ${timeoutMs} ms`, "TimeoutError")),
|
|
timeoutMs,
|
|
);
|
|
activeRequests.add(controller);
|
|
try {
|
|
return await fetchImplementation(input, { ...init, signal: controller.signal });
|
|
} finally {
|
|
clearTimeout(timeout);
|
|
activeRequests.delete(controller);
|
|
}
|
|
}
|
|
|
|
function requestResult(request) {
|
|
return new Promise((resolve, reject) => {
|
|
request.addEventListener("success", () => resolve(request.result), { once: true });
|
|
request.addEventListener("error", () => reject(request.error || new Error("IndexedDB request failed")), { once: true });
|
|
});
|
|
}
|
|
|
|
function transactionDone(transaction) {
|
|
return new Promise((resolve, reject) => {
|
|
transaction.addEventListener("complete", resolve, { once: true });
|
|
transaction.addEventListener("abort", () => reject(transaction.error || new Error("IndexedDB transaction aborted")), { once: true });
|
|
transaction.addEventListener("error", () => reject(transaction.error || new Error("IndexedDB transaction failed")), { once: true });
|
|
});
|
|
}
|
|
|
|
function migrateCommandLog(request, oldVersion) {
|
|
const database = request.result;
|
|
const commands = database.objectStoreNames.contains(COMMANDS)
|
|
? request.transaction.objectStore(COMMANDS)
|
|
: database.createObjectStore(COMMANDS, { keyPath: "id" });
|
|
if (!commands.indexNames.contains(ACCOUNT_INDEX)) commands.createIndex(ACCOUNT_INDEX, "accountPartition");
|
|
if (!database.objectStoreNames.contains("meta")) database.createObjectStore("meta");
|
|
if (oldVersion === 0 || oldVersion >= DATABASE_VERSION) return;
|
|
const transaction = request.transaction;
|
|
const meta = transaction.objectStore("meta");
|
|
const all = commands.getAll();
|
|
all.addEventListener("success", () => {
|
|
const legacy = all.result;
|
|
if (legacy.some((command) => command.schemaVersion !== LEGACY_COMMAND_SCHEMA && command.schemaVersion !== COMMAND_SCHEMA)) {
|
|
transaction.abort();
|
|
return;
|
|
}
|
|
for (const command of legacy) {
|
|
commands.put({
|
|
...command,
|
|
schemaVersion: COMMAND_SCHEMA,
|
|
targetColumn: command.targetColumn || "done",
|
|
accountPartition: command.accountPartition || "demo:demo",
|
|
queuedAt: Number.isSafeInteger(command.queuedAt) ? command.queuedAt : Date.now(),
|
|
});
|
|
}
|
|
meta.put({ from: oldVersion, to: DATABASE_VERSION, migrated: legacy.length }, MIGRATION_KEY);
|
|
}, { once: true });
|
|
}
|
|
|
|
async function openLog() {
|
|
const request = indexedDB.open(DATABASE, DATABASE_VERSION);
|
|
request.addEventListener("upgradeneeded", (event) => migrateCommandLog(request, event.oldVersion));
|
|
return requestResult(request);
|
|
}
|
|
|
|
export function validateQueuedCommand(command) {
|
|
if (!command || Object.getPrototypeOf(command) !== Object.prototype) {
|
|
throw new TypeError("queued command must be an object");
|
|
}
|
|
if (command.schemaVersion !== COMMAND_SCHEMA) {
|
|
throw new RangeError(`unsupported queued command schema version ${command.schemaVersion}`);
|
|
}
|
|
const boundedString = (field, maximum) => {
|
|
const value = command[field];
|
|
if (typeof value !== "string" || value.length === 0 || value.length > maximum) {
|
|
throw new TypeError(`queued command ${field} is invalid`);
|
|
}
|
|
};
|
|
boundedString("id", 256);
|
|
boundedString("accountPartition", 128);
|
|
boundedString("actor", 128);
|
|
boundedString("session", 128);
|
|
boundedString("cardId", 128);
|
|
if (!Number.isSafeInteger(command.causal) || command.causal < 1) {
|
|
throw new TypeError("queued command causal is invalid");
|
|
}
|
|
const queuedAt = command.queuedAt === undefined ? 0 : command.queuedAt;
|
|
if (!Number.isSafeInteger(queuedAt) || queuedAt < 0) {
|
|
throw new TypeError("queued command queuedAt is invalid");
|
|
}
|
|
if (command.kind !== "reorder_card") throw new TypeError(`unknown queued command kind ${command.kind}`);
|
|
if (command.targetColumn !== "done") throw new TypeError(`unknown queued command target ${command.targetColumn}`);
|
|
if (!["click", "drop", "keydown"].includes(command.eventKind)) {
|
|
throw new TypeError(`unknown queued command event kind ${command.eventKind}`);
|
|
}
|
|
if (command.key !== null && (typeof command.key !== "string" || command.key.length > 64)) {
|
|
throw new TypeError("queued command key is invalid");
|
|
}
|
|
return command.queuedAt === undefined ? { ...command, queuedAt } : command;
|
|
}
|
|
|
|
async function pendingCommands(database) {
|
|
const transaction = database.transaction(COMMANDS, "readonly");
|
|
const done = transactionDone(transaction);
|
|
const commands = await requestResult(transaction.objectStore(COMMANDS).index(ACCOUNT_INDEX).getAll(accountPartition));
|
|
await done;
|
|
return commands.map(validateQueuedCommand).sort((left, right) => left.causal - right.causal);
|
|
}
|
|
|
|
async function removePendingCommand(database, commandId) {
|
|
const transaction = database.transaction(COMMANDS, "readwrite");
|
|
const done = transactionDone(transaction);
|
|
transaction.objectStore(COMMANDS).delete(commandId);
|
|
await done;
|
|
}
|
|
|
|
function decideRebase(snapshot, command) {
|
|
const canonical = snapshot.cards.find((card) => String(card.id) === command.cardId);
|
|
if (!canonical) return { kind: "conflicted", reason: "card-missing", canonicalColumn: "missing" };
|
|
if (command.kind === "reorder_card" && canonical.column === "done") {
|
|
return { kind: "converged", reason: "intent-already-canonical", canonicalColumn: canonical.column };
|
|
}
|
|
return { kind: "conflicted", reason: "canonical-state-diverged", canonicalColumn: canonical.column };
|
|
}
|
|
|
|
// The built-in policy is deliberately a named module export: applications that
|
|
// need custom merge or CRDT semantics must import and wire a different policy.
|
|
export function reconcileServerAuthoritative(snapshot, commandSequence, serverResults) {
|
|
if (!snapshot || !Array.isArray(snapshot.cards) || !Number.isSafeInteger(snapshot.serverSequence)) {
|
|
throw new TypeError("reconciliation snapshot is invalid");
|
|
}
|
|
if (!Array.isArray(commandSequence) || !Array.isArray(serverResults)) {
|
|
throw new TypeError("reconciliation commands and server results must be arrays");
|
|
}
|
|
const resultCursor = serverResults.reduce((cursor, result) => {
|
|
if (!result || !Number.isSafeInteger(result.serverSequence)) {
|
|
throw new TypeError("reconciliation server result is invalid");
|
|
}
|
|
return Math.max(cursor, result.serverSequence);
|
|
}, 0);
|
|
if (resultCursor > snapshot.serverSequence) {
|
|
throw new RangeError("reconciliation server result is newer than the canonical snapshot");
|
|
}
|
|
const command = commandSequence[0];
|
|
const decision = command
|
|
? decideRebase(snapshot, command)
|
|
: { kind: "idle", reason: "no-pending-command", canonicalColumn: "unchanged" };
|
|
return {
|
|
model: "server-authoritative-v1",
|
|
snapshotSequence: snapshot.serverSequence,
|
|
serverResultCursor: resultCursor,
|
|
serverResultCount: serverResults.length,
|
|
commandCount: commandSequence.length,
|
|
retainedCommandCount: decision.kind === "converged"
|
|
? Math.max(0, commandSequence.length - 1)
|
|
: commandSequence.length,
|
|
decision,
|
|
};
|
|
}
|
|
|
|
async function claimUploaderLease(database) {
|
|
const transaction = database.transaction("meta", "readwrite");
|
|
const done = transactionDone(transaction);
|
|
const meta = transaction.objectStore("meta");
|
|
const now = Date.now();
|
|
const leaseKey = `uploaderLease:${accountPartition}`;
|
|
const current = await requestResult(meta.get(leaseKey));
|
|
if (current && current.owner !== TAB_ID && current.expiresAt > now) {
|
|
await done;
|
|
return { leader: false, owner: current.owner, expiresAt: current.expiresAt };
|
|
}
|
|
const lease = { owner: TAB_ID, expiresAt: now + LEASE_MS };
|
|
meta.put(lease, leaseKey);
|
|
await done;
|
|
return { leader: true, ...lease };
|
|
}
|
|
|
|
async function releaseUploaderLease(database) {
|
|
const transaction = database.transaction("meta", "readwrite");
|
|
const done = transactionDone(transaction);
|
|
const meta = transaction.objectStore("meta");
|
|
const leaseKey = `uploaderLease:${accountPartition}`;
|
|
const current = await requestResult(meta.get(leaseKey));
|
|
if (current?.owner === TAB_ID) meta.delete(leaseKey);
|
|
await done;
|
|
}
|
|
|
|
function publishLease(lease) {
|
|
root.setAttribute("data-sync-tab-id", TAB_ID);
|
|
root.setAttribute("data-sync-leader", String(lease.leader));
|
|
root.setAttribute("data-sync-lease-owner", lease.owner || TAB_ID);
|
|
root.setAttribute("data-sync-lease-expires", String(lease.expiresAt));
|
|
}
|
|
|
|
async function commitConvergedRebase(database, snapshot, command) {
|
|
const transaction = database.transaction([COMMANDS, "meta"], "readwrite");
|
|
const done = transactionDone(transaction);
|
|
transaction.objectStore("meta").put(snapshot, "canonicalSnapshot");
|
|
transaction.objectStore("meta").put(snapshot.serverSequence, "acknowledgementCursor");
|
|
transaction.objectStore(COMMANDS).delete(command.queueCommandId || command.id);
|
|
await done;
|
|
}
|
|
|
|
function setPhase(phase, message) {
|
|
root.setAttribute("data-sync-phase", phase);
|
|
root.querySelector('[role="status"]').textContent = message;
|
|
}
|
|
|
|
function ageBucket(milliseconds) {
|
|
if (milliseconds < 1000) return "lt-1s";
|
|
if (milliseconds < 10000) return "1s-10s";
|
|
if (milliseconds < 60000) return "10s-1m";
|
|
return "gte-1m";
|
|
}
|
|
|
|
function latencyBucket(milliseconds) {
|
|
if (milliseconds < 50) return "lt-50ms";
|
|
if (milliseconds < 250) return "50ms-250ms";
|
|
if (milliseconds < 1000) return "250ms-1s";
|
|
return "gte-1s";
|
|
}
|
|
|
|
function publishDiagnostics(commands) {
|
|
const queued = Array.isArray(commands) ? commands : [];
|
|
const oldest = queued.reduce((value, command) => {
|
|
return Number.isSafeInteger(command.queuedAt) ? Math.min(value, command.queuedAt) : value;
|
|
}, Date.now());
|
|
root.setAttribute("data-sync-diag-queue-count", String(queued.length));
|
|
root.setAttribute("data-sync-diag-oldest-age-bucket", queued.length === 0 ? "empty" : ageBucket(Date.now() - oldest));
|
|
root.setAttribute("data-sync-diag-cursor", root.getAttribute("data-sync-ack-sequence") || "0");
|
|
root.setAttribute("data-sync-diag-conflicts", String(conflictCount));
|
|
root.setAttribute("data-sync-diag-rejections", String(rejectionCount));
|
|
const diagnostics = root.querySelector("[data-sync-diagnostics]");
|
|
diagnostics.textContent = `Queue ${queued.length}; oldest ${root.getAttribute("data-sync-diag-oldest-age-bucket")}; cursor ${root.getAttribute("data-sync-diag-cursor")}; acknowledgement ${root.getAttribute("data-sync-diag-ack-latency-bucket") || "none"}; conflicts ${conflictCount}; rejections ${rejectionCount}.`;
|
|
}
|
|
|
|
function validatePending(command) {
|
|
if (!command || command.schemaVersion !== COMMAND_SCHEMA || command.accountPartition !== accountPartition || command.kind !== "reorder_card" || typeof command.id !== "string" || !command.id || typeof command.cardId !== "string" || !command.cardId || command.targetColumn !== "done") {
|
|
throw new Error("invalid pending command");
|
|
}
|
|
return command;
|
|
}
|
|
|
|
function setOnline(online) {
|
|
root.setAttribute("data-sync-connection", online ? "online" : "offline");
|
|
}
|
|
|
|
function setManualRetryAvailable(available) {
|
|
const retry = root.querySelector("[data-sync-retry]");
|
|
retry.disabled = !available;
|
|
if (available) root.setAttribute("data-sync-manual-retry", "available");
|
|
else root.removeAttribute("data-sync-manual-retry");
|
|
}
|
|
|
|
function setExportAvailable(available) {
|
|
root.querySelector("[data-sync-export]").disabled = !available;
|
|
}
|
|
|
|
function setConflictResolutionAvailable(available) {
|
|
root.querySelector("[data-sync-use-canonical]").disabled = !available;
|
|
root.querySelector("[data-sync-keep-local]").disabled = !available;
|
|
}
|
|
|
|
async function keepLocalChange() {
|
|
if (!activeConflict) return;
|
|
const { command, snapshot } = activeConflict;
|
|
const retryCommand = {
|
|
...command,
|
|
id: `${command.id}:keep:${snapshot.serverSequence}`,
|
|
queueCommandId: command.id,
|
|
conflictResolution: "keep-local-change",
|
|
basedOnServerSequence: snapshot.serverSequence,
|
|
};
|
|
setConflictResolutionAvailable(false);
|
|
root.setAttribute("data-sync-conflict-resolution", "keep-local-pending");
|
|
root.setAttribute("data-sync-resolution-command-id", retryCommand.id);
|
|
root.setAttribute("data-sync-resolved-command-id", command.id);
|
|
synchronizing = false;
|
|
clearTimeout(leaseTimer);
|
|
await releaseUploaderLease(database);
|
|
root.setAttribute("data-sync-leader", "false");
|
|
await synchronize(retryCommand);
|
|
}
|
|
|
|
async function useCanonicalState() {
|
|
if (!activeConflict) return;
|
|
const { command, snapshot } = activeConflict;
|
|
setConflictResolutionAvailable(false);
|
|
await removePendingCommand(database, command.id);
|
|
const remaining = await pendingCommands(database);
|
|
root.setAttribute("data-sync-conflict-resolution", "used-canonical-state");
|
|
root.setAttribute("data-sync-resolved-command-id", command.id);
|
|
root.setAttribute("data-sync-pending-count", String(remaining.length));
|
|
setExportAvailable(remaining.length > 0);
|
|
setPhase("conflict-resolved", `Used canonical snapshot ${snapshot.serverSequence}; removed ${command.id} and retained ${remaining.length} queued command${remaining.length === 1 ? "" : "s"}.`);
|
|
activeConflict = undefined;
|
|
uploadsThisRun = 0;
|
|
synchronizing = false;
|
|
clearTimeout(leaseTimer);
|
|
await releaseUploaderLease(database);
|
|
root.setAttribute("data-sync-leader", "false");
|
|
await continuePendingWork();
|
|
}
|
|
|
|
async function exportPendingWork() {
|
|
const commands = await pendingCommands(database);
|
|
if (commands.length === 0) return;
|
|
const payload = JSON.stringify({ accountPartition, commands }, null, 2);
|
|
const url = URL.createObjectURL(new Blob([payload], { type: "application/json" }));
|
|
const link = document.createElement("a");
|
|
link.href = url;
|
|
link.download = "hemx-kanban-queue.json";
|
|
link.click();
|
|
URL.revokeObjectURL(url);
|
|
root.setAttribute("data-sync-exported-count", String(commands.length));
|
|
}
|
|
|
|
function scheduleManualRetry(command, error) {
|
|
clearTimeout(retryTimer);
|
|
manualRetryCommand = command;
|
|
root.setAttribute("data-sync-error", error instanceof Error ? error.message : String(error));
|
|
setManualRetryAvailable(true);
|
|
setPhase("offline", "Sync is offline after bounded retries; the durable command remains queued. Retry now when ready.");
|
|
root.dispatchEvent(new CustomEvent("kanban:sync-exhausted", { detail: { commandId: command.id, attempts: MAX_ATTEMPTS } }));
|
|
}
|
|
|
|
async function upload(command) {
|
|
root.setAttribute("data-sync-max-attempts", String(MAX_ATTEMPTS));
|
|
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt += 1) {
|
|
root.setAttribute("data-sync-attempts", String(attempt));
|
|
setPhase(attempt === 1 ? "uploading" : "retrying", `Uploading ${command.id} (attempt ${attempt} of ${MAX_ATTEMPTS}).`);
|
|
try {
|
|
const query = new URLSearchParams({ command_id: command.id, card_id: command.cardId, column: command.targetColumn });
|
|
const response = await fetchWithTimeout(`/sync/commands?${query}`, { method: "POST" });
|
|
if (response.status === 503 && attempt < MAX_ATTEMPTS) {
|
|
const base = BACKOFF_MS[attempt - 1];
|
|
const delay = base + Math.floor(Math.random() * base);
|
|
root.setAttribute("data-sync-last-backoff-base-ms", String(base));
|
|
root.setAttribute("data-sync-last-backoff-ms", String(delay));
|
|
root.dispatchEvent(new CustomEvent("kanban:sync-retry", { detail: { attempt, base, delay } }));
|
|
await new Promise((resolve) => setTimeout(resolve, delay));
|
|
continue;
|
|
}
|
|
if (!response.ok) {
|
|
const problem = await response.json().catch(() => ({}));
|
|
const kind = typeof problem.kind === "string" ? problem.kind : "unclassified-rejection";
|
|
const reason = typeof problem.error === "string" ? problem.error : "unclassified rejection";
|
|
throw new UploadError(response.status, response.status >= 500, kind, reason);
|
|
}
|
|
return response.json();
|
|
} catch (error) {
|
|
if (error instanceof UploadError && !error.retryable) throw error;
|
|
if (attempt === MAX_ATTEMPTS) throw error;
|
|
const base = BACKOFF_MS[attempt - 1];
|
|
const delay = base + Math.floor(Math.random() * base);
|
|
root.setAttribute("data-sync-last-backoff-base-ms", String(base));
|
|
root.setAttribute("data-sync-last-backoff-ms", String(delay));
|
|
root.dispatchEvent(new CustomEvent("kanban:sync-retry", { detail: { attempt, base, delay } }));
|
|
await new Promise((resolve) => setTimeout(resolve, delay));
|
|
}
|
|
}
|
|
throw new Error("sync retry limit exhausted");
|
|
}
|
|
|
|
function beginUpload() {
|
|
inFlightUploads += 1;
|
|
maxObservedInFlight = Math.max(maxObservedInFlight, inFlightUploads);
|
|
root.setAttribute("data-sync-in-flight", String(inFlightUploads));
|
|
root.setAttribute("data-sync-max-observed-in-flight", String(maxObservedInFlight));
|
|
}
|
|
|
|
function finishUpload() {
|
|
inFlightUploads -= 1;
|
|
root.setAttribute("data-sync-in-flight", String(inFlightUploads));
|
|
}
|
|
|
|
async function continuePendingWork() {
|
|
const commands = await pendingCommands(database);
|
|
root.setAttribute("data-sync-pending-count", String(commands.length));
|
|
publishDiagnostics(commands);
|
|
setExportAvailable(commands.length > 0);
|
|
if (commands.length === 0) return;
|
|
if (uploadsThisRun >= uploadLimit) {
|
|
setManualRetryAvailable(true);
|
|
setPhase("backpressured", `Upload limit ${uploadLimit} reached; ${commands.length} durable command${commands.length === 1 ? " remains" : "s remain"} queued. Retry now to continue.`);
|
|
return;
|
|
}
|
|
setTimeout(() => synchronize(validatePending(commands[0])).catch(failPermanently), 0);
|
|
}
|
|
|
|
async function renewOfflineLease(command) {
|
|
if (stopped || root.getAttribute("data-sync-phase") !== "offline") return;
|
|
const lease = await claimUploaderLease(database);
|
|
publishLease(lease);
|
|
if (!lease.leader) {
|
|
setPhase("standby", "Another tab owns sync; waiting for lease takeover.");
|
|
leaseTimer = setTimeout(() => runLeaseLoop(command).catch(failPermanently), LEASE_POLL_MS);
|
|
return;
|
|
}
|
|
leaseTimer = setTimeout(
|
|
() => renewOfflineLease(command).catch(failPermanently),
|
|
LEASE_MS / 2,
|
|
);
|
|
}
|
|
|
|
async function synchronize(command) {
|
|
if (synchronizing) return;
|
|
synchronizing = true;
|
|
const lease = await claimUploaderLease(database);
|
|
publishLease(lease);
|
|
if (!lease.leader) {
|
|
synchronizing = false;
|
|
setPhase("standby", "Another tab owns sync; waiting for lease takeover.");
|
|
return;
|
|
}
|
|
clearTimeout(leaseTimer);
|
|
leaseTimer = setTimeout(() => {
|
|
if (stopped || root.getAttribute("data-sync-phase") === "acknowledged") return;
|
|
if (root.getAttribute("data-sync-phase") === "offline") {
|
|
renewOfflineLease(command).catch(failPermanently);
|
|
} else {
|
|
synchronize(command).catch(failPermanently);
|
|
}
|
|
}, LEASE_MS / 2);
|
|
root.removeAttribute("data-sync-error");
|
|
root.removeAttribute("data-sync-manual-retry");
|
|
setOnline(navigator.onLine);
|
|
try {
|
|
beginUpload();
|
|
let acknowledgement;
|
|
try {
|
|
acknowledgement = await upload(command);
|
|
} finally {
|
|
finishUpload();
|
|
}
|
|
setOnline(true);
|
|
root.setAttribute("data-sync-upload-sequence", String(acknowledgement.serverSequence));
|
|
acknowledgementStartedAt = performance.now();
|
|
setPhase("awaiting-ack", `Command ${command.id} uploaded; awaiting canonical acknowledgement.`);
|
|
|
|
const reconnect = command.session || command.actor || "kanban";
|
|
const source = new EventSource(`/sync/acknowledgements?after=0&reconnect=${encodeURIComponent(reconnect)}`);
|
|
acknowledgementSource = source;
|
|
let opens = 0;
|
|
source.addEventListener("open", () => {
|
|
opens += 1;
|
|
root.setAttribute("data-sync-transport-opens", String(opens));
|
|
root.setAttribute("data-sync-stream-state", "open");
|
|
});
|
|
source.addEventListener("heartbeat", () => {
|
|
const heartbeats = Number(root.getAttribute("data-sync-heartbeats") || "0") + 1;
|
|
root.setAttribute("data-sync-heartbeats", String(heartbeats));
|
|
root.setAttribute("data-sync-stream-state", "healthy");
|
|
});
|
|
source.addEventListener("error", () => {
|
|
const reconnects = Number(root.getAttribute("data-sync-reconnects") || "0") + 1;
|
|
root.setAttribute("data-sync-reconnects", String(reconnects));
|
|
root.setAttribute("data-sync-stream-state", "reconnecting");
|
|
});
|
|
source.addEventListener("acknowledgement", async (event) => {
|
|
const canonical = JSON.parse(event.data);
|
|
if (canonical.commandId !== command.id) return;
|
|
source.close();
|
|
if (acknowledgementSource === source) acknowledgementSource = undefined;
|
|
root.setAttribute("data-sync-pending-before-ack", String((await pendingCommands(database)).length));
|
|
const queueCommandId = command.queueCommandId || command.id;
|
|
await removePendingCommand(database, queueCommandId);
|
|
manualRetryCommand = undefined;
|
|
if (command.conflictResolution === "keep-local-change") {
|
|
activeConflict = undefined;
|
|
setConflictResolutionAvailable(false);
|
|
root.setAttribute("data-sync-conflict-resolution", "kept-local-change");
|
|
root.setAttribute("data-sync-resolved-command-id", queueCommandId);
|
|
}
|
|
uploadsThisRun += 1;
|
|
uploadedTotal += 1;
|
|
root.setAttribute("data-sync-uploaded-this-run", String(uploadsThisRun));
|
|
root.setAttribute("data-sync-uploaded-total", String(uploadedTotal));
|
|
const pendingAfterAck = (await pendingCommands(database)).length;
|
|
root.setAttribute("data-sync-pending-count", String(pendingAfterAck));
|
|
setExportAvailable(pendingAfterAck > 0);
|
|
root.setAttribute("data-sync-ack-sequence", String(canonical.serverSequence));
|
|
root.setAttribute("data-sync-diag-cursor", String(canonical.serverSequence));
|
|
root.setAttribute("data-sync-diag-ack-latency-bucket", latencyBucket(performance.now() - acknowledgementStartedAt));
|
|
root.setAttribute("data-sync-canonical-column", canonical.canonicalColumn);
|
|
publishDiagnostics(await pendingCommands(database));
|
|
setPhase("acknowledged", `Queued change acknowledged in ${canonical.canonicalColumn}.`);
|
|
root.dispatchEvent(new CustomEvent("kanban:sync-acknowledged", { detail: canonical }));
|
|
synchronizing = false;
|
|
clearTimeout(leaseTimer);
|
|
await releaseUploaderLease(database);
|
|
root.setAttribute("data-sync-leader", "false");
|
|
await continuePendingWork();
|
|
});
|
|
source.addEventListener("snapshot-required", async (event) => {
|
|
const missing = JSON.parse(event.data);
|
|
const response = await fetchWithTimeout(missing.snapshotUrl);
|
|
if (!response.ok) throw new Error(`snapshot failed with ${response.status}`);
|
|
const snapshot = await response.json();
|
|
const queued = await pendingCommands(database);
|
|
const reconciliation = reconcileServerAuthoritative(snapshot, queued, [{
|
|
status: "snapshot-required",
|
|
serverSequence: missing.latest,
|
|
}]);
|
|
const decision = reconciliation.decision;
|
|
const converged = decision.kind === "converged";
|
|
root.setAttribute("data-sync-reconciliation-model", reconciliation.model);
|
|
root.setAttribute("data-sync-reconciliation-result-cursor", String(reconciliation.serverResultCursor));
|
|
root.setAttribute("data-sync-reconciliation-retained-count", String(reconciliation.retainedCommandCount));
|
|
root.setAttribute("data-sync-snapshot-sequence", String(snapshot.serverSequence));
|
|
root.setAttribute("data-sync-snapshot-schema", String(snapshot.schemaVersion));
|
|
root.setAttribute("data-sync-snapshot-card-count", String(snapshot.cards.length));
|
|
root.setAttribute("data-sync-rebase-pending-count", String(queued.length));
|
|
root.setAttribute("data-sync-rebase-decision", decision.kind);
|
|
root.setAttribute("data-sync-rebase-reason", decision.reason);
|
|
root.setAttribute("data-sync-canonical-column", decision.canonicalColumn);
|
|
if (converged) {
|
|
activeConflict = undefined;
|
|
manualRetryCommand = undefined;
|
|
setConflictResolutionAvailable(false);
|
|
await commitConvergedRebase(database, snapshot, command);
|
|
if (command.conflictResolution === "keep-local-change") {
|
|
root.setAttribute("data-sync-conflict-resolution", "kept-local-change");
|
|
root.setAttribute("data-sync-resolved-command-id", command.queueCommandId);
|
|
}
|
|
root.setAttribute("data-sync-pending-count", String((await pendingCommands(database)).length));
|
|
root.setAttribute("data-sync-ack-sequence", String(snapshot.serverSequence));
|
|
setPhase("rebased", `Canonical snapshot ${snapshot.serverSequence} already satisfies ${command.id}; committed and removed the pending command.`);
|
|
root.dispatchEvent(new CustomEvent("kanban:sync-rebased", { detail: { snapshot, command, decision } }));
|
|
} else {
|
|
conflictCount += 1;
|
|
activeConflict = { command, snapshot, decision };
|
|
publishDiagnostics(await pendingCommands(database));
|
|
setConflictResolutionAvailable(true);
|
|
setPhase("conflicted", `Canonical snapshot ${snapshot.serverSequence} conflicts with ${command.id} (${decision.reason}); the pending command remains queued.`);
|
|
root.dispatchEvent(new CustomEvent("kanban:sync-conflicted", { detail: { snapshot, command, decision } }));
|
|
}
|
|
synchronizing = false;
|
|
source.close();
|
|
if (acknowledgementSource === source) acknowledgementSource = undefined;
|
|
clearTimeout(leaseTimer);
|
|
await releaseUploaderLease(database);
|
|
root.setAttribute("data-sync-leader", "false");
|
|
if (converged) await continuePendingWork();
|
|
});
|
|
} catch (error) {
|
|
synchronizing = false;
|
|
if (error instanceof UploadError && !error.retryable) {
|
|
setOnline(true);
|
|
clearTimeout(leaseTimer);
|
|
root.setAttribute("data-sync-error", error.message);
|
|
root.setAttribute("data-sync-error-status", String(error.status));
|
|
const remaining = await pendingCommands(database);
|
|
setManualRetryAvailable(false);
|
|
rejectionCount += 1;
|
|
publishDiagnostics(remaining);
|
|
if (command.conflictResolution === "keep-local-change") {
|
|
manualRetryCommand = undefined;
|
|
setConflictResolutionAvailable(true);
|
|
root.setAttribute("data-sync-error-kind", error.kind);
|
|
root.setAttribute("data-sync-error-reason", error.reason);
|
|
root.setAttribute("data-sync-pending-count", String(remaining.length));
|
|
root.setAttribute("data-sync-conflict-resolution", "keep-local-rejected");
|
|
setPhase("resolution-rejected", `Keep-local command ${command.id} was rejected (${error.status}: ${error.reason}); the conflicted command and ${remaining.length - 1} queued suffix command${remaining.length === 2 ? "" : "s"} remain in order.`);
|
|
} else if (error.kind === "authorization-denial") {
|
|
root.setAttribute("data-sync-error-kind", "authorization-denial");
|
|
root.setAttribute("data-sync-pending-count", "redacted");
|
|
root.setAttribute("data-sync-redacted-pending", "true");
|
|
root.removeAttribute("data-sync-error-reason");
|
|
root.removeAttribute("data-sync-rejected-command-id");
|
|
setPhase("authorization-denied", "Current session cannot access local queued work. Sign back into the owning account to continue.");
|
|
} else {
|
|
root.setAttribute("data-sync-error-kind", "permanent-rejection");
|
|
root.setAttribute("data-sync-error-reason", error.reason);
|
|
root.setAttribute("data-sync-rejected-command-id", command.id);
|
|
root.setAttribute("data-sync-pending-count", String(remaining.length));
|
|
setPhase("rejected", `Command ${command.id} was permanently rejected (${error.status}: ${error.reason}); ${remaining.length} durable command${remaining.length === 1 ? " remains" : "s remain"} queued for review.`);
|
|
}
|
|
await releaseUploaderLease(database);
|
|
root.setAttribute("data-sync-leader", "false");
|
|
return;
|
|
}
|
|
setOnline(false);
|
|
scheduleManualRetry(command, error);
|
|
}
|
|
}
|
|
|
|
async function runLeaseLoop(command) {
|
|
if (stopped) return;
|
|
const phase = root.getAttribute("data-sync-phase");
|
|
if (phase === "acknowledged" || phase === "rebased" || phase === "conflicted" || phase === "failed") return;
|
|
if (root.getAttribute("data-sync-leader") === "true") {
|
|
await synchronize(command);
|
|
return;
|
|
}
|
|
const lease = await claimUploaderLease(database);
|
|
publishLease(lease);
|
|
if (lease.leader) {
|
|
await synchronize(command);
|
|
return;
|
|
}
|
|
setPhase("standby", "Another tab owns sync; waiting for lease takeover.");
|
|
leaseTimer = setTimeout(() => runLeaseLoop(command).catch(failPermanently), LEASE_POLL_MS);
|
|
}
|
|
|
|
async function start() {
|
|
if (!root) return;
|
|
root.setAttribute("data-sync-request-timeout-ms", String(REQUEST_TIMEOUT_MS));
|
|
root.setAttribute("data-sync-stream-buffer-limit", String(ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT));
|
|
const contextResponse = await fetchWithTimeout("/sync/context", { credentials: "same-origin", cache: "no-store" });
|
|
if (!contextResponse.ok) throw new Error(`account context failed with ${contextResponse.status}`);
|
|
const context = await contextResponse.json();
|
|
if (!context || typeof context.accountPartition !== "string" || !context.accountPartition) {
|
|
throw new Error("account context omitted accountPartition");
|
|
}
|
|
accountPartition = context.accountPartition;
|
|
root.setAttribute("data-sync-account-partition", accountPartition);
|
|
uploadLimit = Number.parseInt(root.getAttribute("data-sync-upload-limit"), 10);
|
|
if (!Number.isSafeInteger(uploadLimit) || uploadLimit < 1) throw new Error("data-sync-upload-limit must be a positive integer");
|
|
database = await openLog();
|
|
const migration = await requestResult(database.transaction("meta", "readonly").objectStore("meta").get(MIGRATION_KEY));
|
|
root.setAttribute("data-sync-database-version", String(database.version));
|
|
root.setAttribute("data-sync-command-schema", String(COMMAND_SCHEMA));
|
|
if (migration) {
|
|
root.setAttribute("data-sync-migration-from", String(migration.from));
|
|
root.setAttribute("data-sync-migration-to", String(migration.to));
|
|
root.setAttribute("data-sync-migrated-count", String(migration.migrated));
|
|
}
|
|
const commands = await pendingCommands(database);
|
|
root.setAttribute("data-sync-uploaded-this-run", "0");
|
|
root.setAttribute("data-sync-uploaded-total", "0");
|
|
root.setAttribute("data-sync-in-flight", "0");
|
|
root.setAttribute("data-sync-max-observed-in-flight", "0");
|
|
root.setAttribute("data-sync-pending-count", String(commands.length));
|
|
publishDiagnostics(commands);
|
|
root.setAttribute("data-sync-diag-ack-latency-bucket", "none");
|
|
setExportAvailable(commands.length > 0);
|
|
setConflictResolutionAvailable(false);
|
|
setManualRetryAvailable(false);
|
|
if (commands.length === 0) {
|
|
setPhase("idle", "No pending commands.");
|
|
return;
|
|
}
|
|
const command = validatePending(commands[0]);
|
|
root.addEventListener("click", async (event) => {
|
|
if (event.target.closest("[data-sync-keep-local]")) {
|
|
await keepLocalChange();
|
|
return;
|
|
}
|
|
if (event.target.closest("[data-sync-use-canonical]")) {
|
|
await useCanonicalState();
|
|
return;
|
|
}
|
|
if (event.target.closest("[data-sync-export]")) {
|
|
await exportPendingWork();
|
|
return;
|
|
}
|
|
if (!event.target.closest("[data-sync-retry]")) return;
|
|
uploadsThisRun = 0;
|
|
root.setAttribute("data-sync-uploaded-this-run", "0");
|
|
setManualRetryAvailable(false);
|
|
const [next] = await pendingCommands(database);
|
|
const retry = manualRetryCommand || (next && validatePending(next));
|
|
if (retry) synchronize(retry).catch(failPermanently);
|
|
});
|
|
window.addEventListener("online", async () => {
|
|
if (root.getAttribute("data-sync-phase") !== "offline") return;
|
|
setManualRetryAvailable(false);
|
|
const [next] = await pendingCommands(database);
|
|
const retry = manualRetryCommand || (next && validatePending(next));
|
|
if (retry) synchronize(retry).catch(failPermanently);
|
|
});
|
|
await runLeaseLoop(command);
|
|
}
|
|
|
|
window.addEventListener("pagehide", () => {
|
|
stopped = true;
|
|
clearTimeout(leaseTimer);
|
|
clearTimeout(retryTimer);
|
|
if (acknowledgementSource) {
|
|
acknowledgementSource.close();
|
|
root.setAttribute("data-sync-stream-state", "cancelled");
|
|
}
|
|
acknowledgementSource = undefined;
|
|
for (const controller of activeRequests) {
|
|
controller.abort(new DOMException("sync cancelled because page is hidden", "AbortError"));
|
|
}
|
|
if (database) releaseUploaderLease(database).catch(() => {});
|
|
});
|
|
|
|
function failPermanently(error) {
|
|
synchronizing = false;
|
|
root.setAttribute("data-sync-error", error instanceof Error ? error.message : String(error));
|
|
setPhase("failed", "Sync failed; the durable command remains queued.");
|
|
}
|
|
|
|
start().catch((error) => {
|
|
if (!root) return;
|
|
failPermanently(error);
|
|
});
|