L7 Offline & PWA: Outbox-Kern, IndexedDB-Store, Sync-Engine und Bundle-Logik
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -0,0 +1,350 @@
|
||||
import type { SyncOperationInput, SyncOpResult, SyncOpType } from "@/lib/sync/envelope";
|
||||
import { CONFLICTING_OPS } from "@/lib/sync/envelope";
|
||||
import { ctxKeyOf, type BlobEntry, type OfflineContext, type OutboxEntry, type OutboxError } from "./types";
|
||||
|
||||
/**
|
||||
* Pure outbox logic (lane L7, Spec §23.4/§23.5, ARCHITEKTUR §4.6) — no IndexedDB, no fetch.
|
||||
* Tested in scripts/test-offline-core.ts with the in-memory store.
|
||||
*
|
||||
* Rules:
|
||||
* - ops of one work order are sent strictly in creation order; an op that is still pending
|
||||
* (backoff, waiting for its upload) blocks the later ops of the SAME order only
|
||||
* - terminal outcomes (applied / conflict / rejected) never block other ops
|
||||
* - blob uploads happen before the op that references them (`documentId: "blob:<clientId>"`)
|
||||
* - server idMaps (client id → server id) are applied to the payloads of pending ops
|
||||
* - transient failures (network, internal) → exponential backoff; deterministic ones are final
|
||||
*/
|
||||
|
||||
export const MAX_BATCH = 50;
|
||||
export const BLOB_REF_PREFIX = "blob:";
|
||||
export const BACKOFF_BASE_MS = 2_000;
|
||||
export const BACKOFF_MAX_MS = 5 * 60_000;
|
||||
|
||||
/** Ops that change WorkOrder.version on the server (status changes). */
|
||||
export const VERSION_CHANGING_OPS: readonly SyncOpType[] = ["session.start", "session.pause", "session.resume", "session.end", "work_order.transition", "report.submit"];
|
||||
|
||||
const TRANSIENT_CODES = new Set<OutboxError["code"]>(["internal", "network"]);
|
||||
|
||||
export type QueuedOpInput = {
|
||||
opType: SyncOpType;
|
||||
payload: Record<string, unknown>;
|
||||
entityType?: string;
|
||||
entityId?: string;
|
||||
baseVersion?: number;
|
||||
};
|
||||
|
||||
export const blobRef = (clientId: string) => `${BLOB_REF_PREFIX}${clientId}`;
|
||||
|
||||
/** Client ids of all blob references in a payload (deep). */
|
||||
export function blobRefsIn(value: unknown): string[] {
|
||||
const out = new Set<string>();
|
||||
const walk = (v: unknown) => {
|
||||
if (typeof v === "string") {
|
||||
if (v.startsWith(BLOB_REF_PREFIX)) out.add(v.slice(BLOB_REF_PREFIX.length));
|
||||
} else if (Array.isArray(v)) v.forEach(walk);
|
||||
else if (v && typeof v === "object") Object.values(v).forEach(walk);
|
||||
};
|
||||
walk(value);
|
||||
return [...out];
|
||||
}
|
||||
|
||||
export function workOrderIdOf(op: Pick<QueuedOpInput, "payload" | "entityType" | "entityId">): string | null {
|
||||
const fromPayload = op.payload.workOrderId;
|
||||
if (typeof fromPayload === "string" && fromPayload) return fromPayload;
|
||||
return op.entityType === "work_order" && op.entityId ? op.entityId : null;
|
||||
}
|
||||
|
||||
export const isPending = (e: Pick<OutboxEntry, "status">) => e.status === "queued" || e.status === "sending";
|
||||
|
||||
/** Exponential backoff with jitter (0.5–1.0 × of the step); `random` injectable for tests. */
|
||||
export function backoffMs(attempts: number, random: () => number = Math.random): number {
|
||||
const step = Math.min(BACKOFF_MAX_MS, BACKOFF_BASE_MS * 2 ** Math.max(0, attempts - 1));
|
||||
return Math.round(step * (0.5 + random() * 0.5));
|
||||
}
|
||||
|
||||
let seqCounter = 0;
|
||||
/** Monotonic per device (ms timestamp × 1000 + counter). */
|
||||
export function nextSeq(now: Date): number {
|
||||
seqCounter = (seqCounter + 1) % 1000;
|
||||
return now.getTime() * 1000 + seqCounter;
|
||||
}
|
||||
|
||||
export function createEntry(ctx: OfflineContext, op: QueuedOpInput, opts: { clientOpId: string; now: Date; existing: OutboxEntry[]; seq?: number }): OutboxEntry {
|
||||
const workOrderId = workOrderIdOf(op);
|
||||
const iso = opts.now.toISOString();
|
||||
// A conflict-checked op queued behind our own pending status change of the same order takes its
|
||||
// baseVersion from that op's server result — otherwise our own chain would always conflict.
|
||||
const chainedBase =
|
||||
CONFLICTING_OPS.includes(op.opType) &&
|
||||
!!workOrderId &&
|
||||
opts.existing.some((e) => e.workOrderId === workOrderId && isPending(e) && VERSION_CHANGING_OPS.includes(e.opType));
|
||||
return {
|
||||
clientOpId: opts.clientOpId,
|
||||
ctxKey: ctxKeyOf(ctx),
|
||||
tenantId: ctx.tenantId,
|
||||
userId: ctx.userId,
|
||||
opType: op.opType,
|
||||
payload: op.payload,
|
||||
entityType: op.entityType,
|
||||
entityId: op.entityId,
|
||||
baseVersion: op.baseVersion,
|
||||
chainedBase: chainedBase || undefined,
|
||||
workOrderId,
|
||||
seq: opts.seq ?? nextSeq(opts.now),
|
||||
status: "queued",
|
||||
attempts: 0,
|
||||
lastError: null,
|
||||
clientCreatedAt: iso,
|
||||
updatedAt: iso,
|
||||
nextAttemptAt: null,
|
||||
appliedAt: null,
|
||||
blobRefs: blobRefsIn(op.payload),
|
||||
result: null,
|
||||
};
|
||||
}
|
||||
|
||||
const orderKey = (e: OutboxEntry) => e.workOrderId ?? "_global";
|
||||
const due = (at: string | null, now: Date) => !at || new Date(at).getTime() <= now.getTime();
|
||||
|
||||
/** Entries left in `sending` by an interrupted pass (tab closed, crash) go back to the queue. */
|
||||
export function recoverSending(ops: OutboxEntry[], now: Date): OutboxEntry[] {
|
||||
return ops.filter((o) => o.status === "sending").map((o) => ({ ...o, status: "queued" as const, updatedAt: now.toISOString() }));
|
||||
}
|
||||
|
||||
/** Blobs that must be uploaded before the next batch (referenced by pending ops, not in backoff). */
|
||||
export function selectUploads(ops: OutboxEntry[], blobs: BlobEntry[], now: Date): BlobEntry[] {
|
||||
const byId = new Map(blobs.map((b) => [b.clientId, b]));
|
||||
const out: BlobEntry[] = [];
|
||||
const seen = new Set<string>();
|
||||
for (const op of [...ops].sort((a, b) => a.seq - b.seq)) {
|
||||
if (op.status !== "queued") continue;
|
||||
for (const ref of op.blobRefs) {
|
||||
const b = byId.get(ref);
|
||||
if (!b || seen.has(ref) || b.status === "uploaded") continue;
|
||||
if (b.status === "failed" && b.nextAttemptAt === null) continue; // final failure
|
||||
if (!due(b.nextAttemptAt, now)) continue;
|
||||
seen.add(ref);
|
||||
out.push(b);
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Payload with blob references replaced by their uploaded documentIds; null if an upload is missing. */
|
||||
export function resolvePayload(op: OutboxEntry, blobs: BlobEntry[]): Record<string, unknown> | null {
|
||||
if (op.blobRefs.length === 0) return op.payload;
|
||||
const docs = new Map(blobs.filter((b) => b.status === "uploaded" && b.documentId).map((b) => [b.clientId, b.documentId as string]));
|
||||
if (!op.blobRefs.every((r) => docs.has(r))) return null;
|
||||
const walk = (v: unknown): unknown => {
|
||||
if (typeof v === "string" && v.startsWith(BLOB_REF_PREFIX)) return docs.get(v.slice(BLOB_REF_PREFIX.length)) ?? v;
|
||||
if (Array.isArray(v)) return v.map(walk);
|
||||
if (v && typeof v === "object") return Object.fromEntries(Object.entries(v).map(([k, x]) => [k, walk(x)]));
|
||||
return v;
|
||||
};
|
||||
return walk(op.payload) as Record<string, unknown>;
|
||||
}
|
||||
|
||||
export type Batch = { entries: OutboxEntry[]; operations: SyncOperationInput[] };
|
||||
|
||||
/**
|
||||
* Next batch (max `max` ops): in seq order; per work order stop at the first op that cannot be
|
||||
* sent yet (backoff, missing upload, failed upload) — later ops of other orders still go.
|
||||
*/
|
||||
export function selectBatch(ops: OutboxEntry[], blobs: BlobEntry[], now: Date, max = MAX_BATCH): Batch {
|
||||
const blocked = new Set<string>();
|
||||
const entries: OutboxEntry[] = [];
|
||||
const operations: SyncOperationInput[] = [];
|
||||
for (const op of [...ops].sort((a, b) => a.seq - b.seq)) {
|
||||
if (entries.length >= max) break;
|
||||
const key = orderKey(op);
|
||||
if (op.status === "sending") {
|
||||
blocked.add(key);
|
||||
continue;
|
||||
}
|
||||
if (op.status !== "queued") continue;
|
||||
if (blocked.has(key)) continue;
|
||||
if (!due(op.nextAttemptAt, now)) {
|
||||
blocked.add(key);
|
||||
continue;
|
||||
}
|
||||
const payload = resolvePayload(op, blobs);
|
||||
if (!payload) {
|
||||
blocked.add(key);
|
||||
continue;
|
||||
}
|
||||
entries.push(op);
|
||||
operations.push({
|
||||
clientOpId: op.clientOpId,
|
||||
opType: op.opType,
|
||||
...(op.entityType ? { entityType: op.entityType } : {}),
|
||||
...(op.entityId ? { entityId: op.entityId } : {}),
|
||||
...(op.baseVersion !== undefined ? { baseVersion: op.baseVersion } : {}),
|
||||
payload,
|
||||
clientCreatedAt: op.clientCreatedAt,
|
||||
});
|
||||
}
|
||||
return { entries, operations };
|
||||
}
|
||||
|
||||
/** Replaces client ids by server ids in a payload (deep). The own `clientId` field stays (idempotency). */
|
||||
export function applyIdMapToPayload(payload: Record<string, unknown>, idMap: Record<string, string>): Record<string, unknown> {
|
||||
const walk = (v: unknown, key?: string): unknown => {
|
||||
if (typeof v === "string") return key !== "clientId" && Object.hasOwn(idMap, v) ? idMap[v] : v;
|
||||
if (Array.isArray(v)) return v.map((x) => walk(x));
|
||||
if (v && typeof v === "object") return Object.fromEntries(Object.entries(v).map(([k, x]) => [k, walk(x, k)]));
|
||||
return v;
|
||||
};
|
||||
return walk(payload) as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function originalStatus(result: SyncOpResult): "applied" | "conflict" | "rejected" {
|
||||
if (!result.errorCode) return "applied";
|
||||
if (result.errorCode === "conflict" && /original status: conflict/.test(result.message ?? "")) return "conflict";
|
||||
return "rejected";
|
||||
}
|
||||
|
||||
/**
|
||||
* Applies the server results of a batch. Returns all changed entries (batch entries and pending
|
||||
* entries rewritten by idMap / chained baseVersion) plus the merged idMap.
|
||||
*/
|
||||
export function applyResults(
|
||||
ops: OutboxEntry[],
|
||||
batch: OutboxEntry[],
|
||||
results: SyncOpResult[],
|
||||
now: Date,
|
||||
random: () => number = Math.random,
|
||||
): { changed: OutboxEntry[]; idMap: Record<string, string> } {
|
||||
const iso = now.toISOString();
|
||||
const byOp = new Map(results.map((r) => [r.clientOpId, r]));
|
||||
const changed = new Map<string, OutboxEntry>();
|
||||
const idMap: Record<string, string> = {};
|
||||
const versions = new Map<string, number>();
|
||||
|
||||
for (const entry of batch) {
|
||||
const r = byOp.get(entry.clientOpId);
|
||||
let next: OutboxEntry;
|
||||
if (!r) {
|
||||
next = transient(entry, { code: "internal", message: "missing result" }, now, random);
|
||||
} else {
|
||||
const status = r.status === "duplicate" ? originalStatus(r) : r.status;
|
||||
if (status === "applied") {
|
||||
next = { ...entry, status: "applied", appliedAt: iso, updatedAt: iso, lastError: null, nextAttemptAt: null, result: r };
|
||||
Object.assign(idMap, r.idMap ?? {});
|
||||
if (entry.workOrderId && r.entityVersion !== undefined) versions.set(entry.workOrderId, r.entityVersion);
|
||||
} else if (status === "conflict") {
|
||||
next = { ...entry, status: "conflict", updatedAt: iso, lastError: { code: "conflict", message: r.message }, nextAttemptAt: null, result: r };
|
||||
} else {
|
||||
const code = r.errorCode ?? "internal";
|
||||
next = TRANSIENT_CODES.has(code)
|
||||
? transient(entry, { code, message: r.message }, now, random)
|
||||
: { ...entry, status: "rejected", updatedAt: iso, lastError: { code, message: r.message }, nextAttemptAt: null, result: r };
|
||||
}
|
||||
}
|
||||
changed.set(next.clientOpId, next);
|
||||
}
|
||||
|
||||
const inBatch = new Set(batch.map((b) => b.clientOpId));
|
||||
const hasIds = Object.keys(idMap).length > 0;
|
||||
for (const op of ops) {
|
||||
if (inBatch.has(op.clientOpId) || !isPending(op)) continue;
|
||||
let next = op;
|
||||
if (hasIds) {
|
||||
const payload = applyIdMapToPayload(op.payload, idMap);
|
||||
if (JSON.stringify(payload) !== JSON.stringify(op.payload)) next = { ...next, payload, updatedAt: iso };
|
||||
}
|
||||
if (next.chainedBase && next.workOrderId && versions.has(next.workOrderId)) {
|
||||
next = { ...next, baseVersion: versions.get(next.workOrderId), updatedAt: iso };
|
||||
}
|
||||
if (next !== op) changed.set(next.clientOpId, next);
|
||||
}
|
||||
// a chained op in the same batch already went out with the old baseVersion — nothing to do there
|
||||
return { changed: [...changed.values()], idMap };
|
||||
}
|
||||
|
||||
function transient(entry: OutboxEntry, error: OutboxError, now: Date, random: () => number): OutboxEntry {
|
||||
const attempts = entry.attempts + 1;
|
||||
return {
|
||||
...entry,
|
||||
status: "queued",
|
||||
attempts,
|
||||
lastError: error,
|
||||
updatedAt: now.toISOString(),
|
||||
nextAttemptAt: new Date(now.getTime() + backoffMs(attempts, random)).toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
/** Whole batch failed on transport level (offline, 5xx, 401). */
|
||||
export function applyTransportFailure(batch: OutboxEntry[], error: OutboxError, now: Date, random: () => number = Math.random): OutboxEntry[] {
|
||||
return batch.map((e) => transient(e, error, now, random));
|
||||
}
|
||||
|
||||
/** Manual "erneut versuchen": transient entries retry now; rejected ones get a fresh clientOpId (the server stored the old outcome). */
|
||||
export function retryEntry(entry: OutboxEntry, now: Date, newClientOpId: string): OutboxEntry | null {
|
||||
if (entry.status === "applied" || entry.status === "conflict" || entry.status === "sending") return null;
|
||||
const base = { ...entry, status: "queued" as const, nextAttemptAt: null, updatedAt: now.toISOString(), acknowledged: undefined };
|
||||
return entry.status === "rejected" ? { ...base, clientOpId: newClientOpId, attempts: 0, lastError: null, result: null } : base;
|
||||
}
|
||||
|
||||
/** Only rejected entries can be discarded (conflicts stay listed; they are reviewed in the office). */
|
||||
export const canDiscard = (e: Pick<OutboxEntry, "status">) => e.status === "rejected";
|
||||
|
||||
export type OutboxSummary = {
|
||||
pendingOps: number;
|
||||
pendingUploads: number;
|
||||
uploadBytes: number;
|
||||
failedUploads: number;
|
||||
conflicts: number;
|
||||
rejected: number;
|
||||
/** oldest pending op (ISO) */
|
||||
oldestPending: string | null;
|
||||
waitingForAuth: boolean;
|
||||
};
|
||||
|
||||
export function summarize(ops: OutboxEntry[], blobs: BlobEntry[]): OutboxSummary {
|
||||
const pending = ops.filter(isPending);
|
||||
const openBlobs = blobs.filter((b) => b.status !== "uploaded");
|
||||
return {
|
||||
pendingOps: pending.length,
|
||||
pendingUploads: openBlobs.filter((b) => b.status !== "failed" || b.nextAttemptAt !== null).length,
|
||||
uploadBytes: openBlobs.reduce((s, b) => s + b.size, 0),
|
||||
failedUploads: openBlobs.filter((b) => b.status === "failed" && b.nextAttemptAt === null).length,
|
||||
conflicts: ops.filter((o) => o.status === "conflict" && !o.acknowledged).length,
|
||||
rejected: ops.filter((o) => o.status === "rejected" && !o.acknowledged).length,
|
||||
oldestPending: pending.length ? pending.reduce((m, o) => (o.clientCreatedAt < m ? o.clientCreatedAt : m), pending[0].clientCreatedAt) : null,
|
||||
waitingForAuth: pending.some((o) => o.lastError?.code === "unauthorized"),
|
||||
};
|
||||
}
|
||||
|
||||
/** Entries that can be removed: applied before the last bundle pull, acknowledged rejections. */
|
||||
export function prunable(ops: OutboxEntry[], bundleSyncedAt: string | null): string[] {
|
||||
return ops
|
||||
.filter((o) => (o.status === "applied" && !!o.appliedAt && !!bundleSyncedAt && o.appliedAt <= bundleSyncedAt) || (o.status === "rejected" && o.acknowledged))
|
||||
.map((o) => o.clientOpId);
|
||||
}
|
||||
|
||||
/** i18n key (messages offline.problem.*) explaining a failed entry in plain language. */
|
||||
export function problemKey(e: Pick<OutboxEntry, "status" | "opType" | "lastError">): string {
|
||||
if (e.status === "conflict") {
|
||||
if (e.opType === "work_order.transition") return "conflictTransition";
|
||||
if (e.opType === "report.submit") return "conflictReport";
|
||||
return "conflict";
|
||||
}
|
||||
switch (e.lastError?.code) {
|
||||
case "not_found":
|
||||
return "notFound";
|
||||
case "forbidden":
|
||||
return "forbidden";
|
||||
case "blocked":
|
||||
return "blocked";
|
||||
case "invalid":
|
||||
return "invalid";
|
||||
case "upload":
|
||||
return "upload";
|
||||
case "unauthorized":
|
||||
return "unauthorized";
|
||||
case "network":
|
||||
return "network";
|
||||
default:
|
||||
return "internal";
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user