tmk241/hemx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(kanban): partition local queues by account

Browse Source 
req: sync/020

req: auth/005

req: security/004

req: operations/002
This commit is contained in:
slhx agent
2026-07-13 19:00:04 +02:00
parent 6de4dcb73b
commit 3ef84e7331
7 changed files with 239 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/kanban/static/command-log.js
+66 -22
View File
@@ -1,10 +1,12 @@
const DATABASE = "hemx-kanban-v1";
const DATABASE_VERSION = 2;
const DATABASE_VERSION = 3;
const COMMANDS = "commands";
const META = "meta";
const ACCOUNT_INDEX = "byAccountPartition";
const COMMAND_SCHEMA = 2;
const LEGACY_COMMAND_SCHEMA = 1;
const MIGRATION_KEY = "commandSchemaMigration";
const ACCOUNT_PARTITION_SESSION = "hemx-kanban-account-partition-v1";
const EXPORT_SCHEMA = 1;
const MAX_REPLAY_COMMANDS = 64;
const REPLAY_BUDGET_MS = 100;
@@ -28,23 +30,30 @@ function completed(transaction) {
function migrateCommandLog(request, oldVersion) {
const database = request.result;
if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" });
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 commands = transaction.objectStore(COMMANDS);
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)) {
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: "done" });
commands.put({
...command,
schemaVersion: COMMAND_SCHEMA,
targetColumn: command.targetColumn || "done",
accountPartition: command.accountPartition || "demo:demo",
});
}
meta.put({ from: LEGACY_COMMAND_SCHEMA, to: COMMAND_SCHEMA, migrated: legacy.length }, MIGRATION_KEY);
meta.put({ from: oldVersion, to: DATABASE_VERSION, migrated: legacy.length }, MIGRATION_KEY);
}, { once: true });
}
@@ -54,6 +63,28 @@ function openCommandLog() {
return result(request);
}
async function currentAccountPartition() {
let response;
try {
response = await fetch("/sync/context", { credentials: "same-origin", cache: "no-store" });
} catch (error) {
const cached = sessionStorage.getItem(ACCOUNT_PARTITION_SESSION);
if (cached) return cached;
throw error;
}
if (!response.ok) {
const cached = sessionStorage.getItem(ACCOUNT_PARTITION_SESSION);
if (response.status === 404 && cached) return cached;
throw new Error(`account context failed with ${response.status}`);
}
const context = await response.json();
if (!context || typeof context.accountPartition !== "string" || !context.accountPartition) {
throw new Error("account context omitted accountPartition");
}
sessionStorage.setItem(ACCOUNT_PARTITION_SESSION, context.accountPartition);
return context.accountPartition;
}
function clientReady(root) {
if (root.hasAttribute("data-hemx-client-ready")) return Promise.resolve();
return new Promise((resolve) => {
@@ -85,7 +116,7 @@ function stableSession() {
return session;
}
async function appendReorder(database, wire) {
async function appendReorder(database, accountPartition, wire) {
const transaction = database.transaction([COMMANDS, META], "readwrite");
const done = completed(transaction);
const completion = done.then(
@@ -94,14 +125,17 @@ async function appendReorder(database, wire) {
);
const meta = transaction.objectStore(META);
const commands = transaction.objectStore(COMMANDS);
const actorRequest = result(meta.get("actor"));
const causalRequest = result(meta.get("causal"));
const actorKey = `actor:${accountPartition}`;
const causalKey = `causal:${accountPartition}`;
const actorRequest = result(meta.get(actorKey));
const causalRequest = result(meta.get(causalKey));
const [storedActor, storedCausal] = await Promise.all([actorRequest, causalRequest]);
const actor = storedActor || crypto.randomUUID();
const causal = (storedCausal || 0) + 1;
const command = {
id: `${actor}:${causal}`,
schemaVersion: COMMAND_SCHEMA,
accountPartition,
actor,
session: stableSession(),
causal,
@@ -114,10 +148,10 @@ async function appendReorder(database, wire) {
let append;
let counted;
try {
meta.put(actor, "actor");
meta.put(causal, "causal");
meta.put(actor, actorKey);
meta.put(causal, causalKey);
append = result(commands.add(command));
counted = result(commands.count());
counted = result(commands.index(ACCOUNT_INDEX).count(accountPartition));
} catch (error) {
transaction.abort();
await completion;
@@ -134,10 +168,10 @@ async function appendReorder(database, wire) {
}
}
async function storedCommands(database) {
async function storedCommands(database, accountPartition) {
const transaction = database.transaction(COMMANDS, "readonly");
const done = completed(transaction);
const commands = await result(transaction.objectStore(COMMANDS).getAll());
const commands = await result(transaction.objectStore(COMMANDS).index(ACCOUNT_INDEX).getAll(accountPartition));
await done;
return commands.sort((left, right) => left.causal - right.causal);
}
@@ -161,6 +195,7 @@ function validate(command) {
throw new Error(`unsupported durable command ${command.id || "record"}`);
}
if (typeof command.id !== "string" || !command.id) invalidCommand(command, "id");
if (typeof command.accountPartition !== "string" || !command.accountPartition) invalidCommand(command, "accountPartition");
if (typeof command.actor !== "string" || !command.actor) invalidCommand(command, "actor");
if (typeof command.session !== "string" || !command.session) invalidCommand(command, "session");
if (!Number.isSafeInteger(command.causal) || command.causal < 1) invalidCommand(command, "causal");
@@ -221,10 +256,16 @@ function exportCommands(root, commands) {
root.dispatchEvent(new CustomEvent("kanban:commands-exported", { detail: payload }));
}
async function clearCommands(database) {
async function clearCommands(database, accountPartition) {
const transaction = database.transaction(COMMANDS, "readwrite");
const done = completed(transaction);
transaction.objectStore(COMMANDS).clear();
const commands = transaction.objectStore(COMMANDS);
const cursor = commands.index(ACCOUNT_INDEX).openKeyCursor(IDBKeyRange.only(accountPartition));
cursor.addEventListener("success", () => {
if (!cursor.result) return;
commands.delete(cursor.result.primaryKey);
cursor.result.continue();
});
await done;
}
@@ -244,7 +285,7 @@ function disarmRecoveryControls(controls) {
}
}
function installRecoveryControls(root, database) {
function installRecoveryControls(root, database, accountPartition) {
const controls = [...root.querySelectorAll("[data-kanban-command-action]")];
for (const control of controls) {
control.addEventListener("click", async () => {
@@ -260,12 +301,12 @@ function installRecoveryControls(root, database) {
controls.forEach((item) => { item.disabled = true; });
try {
if (action === "export") {
exportCommands(root, await storedCommands(database));
exportCommands(root, await storedCommands(database, accountPartition));
controls.forEach((item) => { item.disabled = false; });
return;
}
if (action === "delete") {
await clearCommands(database);
await clearCommands(database, accountPartition);
root.dispatchEvent(new CustomEvent("kanban:commands-deleted"));
} else if (action === "reset") {
await resetLocalData(database);
@@ -286,6 +327,9 @@ function installRecoveryControls(root, database) {
async function start() {
const root = document.querySelector(ROOT);
if (!root) return;
root.setAttribute("data-kanban-load-id", crypto.randomUUID());
const accountPartition = await currentAccountPartition();
root.setAttribute("data-kanban-account-partition", accountPartition);
const databasePromise = openCommandLog();
const offlineReady = prepareOfflineShell(root).catch((error) => report(root, "offline", error));
await clientReady(root);
@@ -299,7 +343,7 @@ async function start() {
let command;
let count;
try {
({ command, count } = await appendReorder(await databasePromise, wire));
({ command, count } = await appendReorder(await databasePromise, accountPartition, wire));
} catch (error) {
report(root, "persist", error);
throw error;
@@ -328,10 +372,10 @@ async function start() {
wasmHandler = window.hemx.registerClientHandler("reorder_card", durableHandler);
if (typeof wasmHandler !== "function") throw new Error("reorder_card WASM handler is not registered");
const database = await databasePromise;
installRecoveryControls(root, database);
installRecoveryControls(root, database, accountPartition);
root.setAttribute("data-kanban-replay-limit", String(MAX_REPLAY_COMMANDS));
try {
const commands = await storedCommands(database);
const commands = await storedCommands(database, accountPartition);
if (commands.length > MAX_REPLAY_COMMANDS) throw new ReplayLimitError(commands.length);
commands.forEach(validate);
const replayStarted = performance.now();
examples/kanban/static/sync.js
+57 -14
View File
@@ -1,6 +1,7 @@
const DATABASE = "hemx-kanban-v1";
const DATABASE_VERSION = 2;
const DATABASE_VERSION = 3;
const COMMANDS = "commands";
const ACCOUNT_INDEX = "byAccountPartition";
const COMMAND_SCHEMA = 2;
const LEGACY_COMMAND_SCHEMA = 1;
const MIGRATION_KEY = "commandSchemaMigration";
@@ -10,8 +11,8 @@ 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;
const LEASE_KEY = "uploaderLease";
let database;
let accountPartition;
let uploadLimit;
let retryTimer;
let leaseTimer;
@@ -50,23 +51,30 @@ function transactionDone(transaction) {
function migrateCommandLog(request, oldVersion) {
const database = request.result;
if (!database.objectStoreNames.contains(COMMANDS)) database.createObjectStore(COMMANDS, { keyPath: "id" });
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 commands = transaction.objectStore(COMMANDS);
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)) {
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: "done" });
commands.put({
...command,
schemaVersion: COMMAND_SCHEMA,
targetColumn: command.targetColumn || "done",
accountPartition: command.accountPartition || "demo:demo",
});
}
meta.put({ from: LEGACY_COMMAND_SCHEMA, to: COMMAND_SCHEMA, migrated: legacy.length }, MIGRATION_KEY);
meta.put({ from: oldVersion, to: DATABASE_VERSION, migrated: legacy.length }, MIGRATION_KEY);
}, { once: true });
}
@@ -79,7 +87,7 @@ async function openLog() {
async function pendingCommands(database) {
const transaction = database.transaction(COMMANDS, "readonly");
const done = transactionDone(transaction);
const commands = await requestResult(transaction.objectStore(COMMANDS).getAll());
const commands = await requestResult(transaction.objectStore(COMMANDS).index(ACCOUNT_INDEX).getAll(accountPartition));
await done;
return commands.sort((left, right) => left.causal - right.causal);
}
@@ -105,13 +113,14 @@ async function claimUploaderLease(database) {
const done = transactionDone(transaction);
const meta = transaction.objectStore("meta");
const now = Date.now();
const current = await requestResult(meta.get(LEASE_KEY));
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, LEASE_KEY);
meta.put(lease, leaseKey);
await done;
return { leader: true, ...lease };
}
@@ -120,8 +129,9 @@ async function releaseUploaderLease(database) {
const transaction = database.transaction("meta", "readwrite");
const done = transactionDone(transaction);
const meta = transaction.objectStore("meta");
const current = await requestResult(meta.get(LEASE_KEY));
if (current?.owner === TAB_ID) meta.delete(LEASE_KEY);
const leaseKey = `uploaderLease:${accountPartition}`;
const current = await requestResult(meta.get(leaseKey));
if (current?.owner === TAB_ID) meta.delete(leaseKey);
await done;
}
@@ -147,7 +157,7 @@ function setPhase(phase, message) {
}
function validatePending(command) {
if (!command || command.schemaVersion !== COMMAND_SCHEMA || command.kind !== "reorder_card" || typeof command.id !== "string" || !command.id || typeof command.cardId !== "string" || !command.cardId || command.targetColumn !== "done") {
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;
@@ -164,6 +174,23 @@ function setManualRetryAvailable(available) {
else root.removeAttribute("data-sync-manual-retry");
}
function setExportAvailable(available) {
root.querySelector("[data-sync-export]").disabled = !available;
}
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);
root.setAttribute("data-sync-error", error instanceof Error ? error.message : String(error));
@@ -225,6 +252,7 @@ function finishUpload() {
async function continuePendingWork() {
const commands = await pendingCommands(database);
root.setAttribute("data-sync-pending-count", String(commands.length));
setExportAvailable(commands.length > 0);
if (commands.length === 0) return;
if (uploadsThisRun >= uploadLimit) {
setManualRetryAvailable(true);
@@ -282,7 +310,9 @@ async function synchronize(command) {
uploadedTotal += 1;
root.setAttribute("data-sync-uploaded-this-run", String(uploadsThisRun));
root.setAttribute("data-sync-uploaded-total", String(uploadedTotal));
root.setAttribute("data-sync-pending-count", String((await pendingCommands(database)).length));
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-ack-command-id", canonical.commandId);
root.setAttribute("data-sync-canonical-column", canonical.canonicalColumn);
@@ -373,6 +403,14 @@ async function runLeaseLoop(command) {
async function start() {
if (!root) return;
const contextResponse = await fetch("/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();
@@ -390,6 +428,7 @@ async function start() {
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));
setExportAvailable(commands.length > 0);
setManualRetryAvailable(false);
if (commands.length === 0) {
setPhase("idle", "No pending commands.");
@@ -397,6 +436,10 @@ async function start() {
}
const command = validatePending(commands[0]);
root.addEventListener("click", async (event) => {
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");