root
100 lines
4.7 KiB
TypeScript
100 lines
4.7 KiB
TypeScript
/*
|
|
* Immutable Workbench projection outbox rows are the common replay and realtime SSE authority.
|
|
*/
|
|
|
|
export const WORKBENCH_REALTIME_AUTHORITY_VERSION = "workbench-realtime-authority-v2";
|
|
|
|
export function projectionOutboxRealtimeEvents(snapshot = {}, { includeSnapshot = false, includeEvents = true } = {}) {
|
|
const facts = objectValue(snapshot.facts);
|
|
const rows = includeEvents ? factArray(snapshot.events) : [];
|
|
const emitted = new Set();
|
|
const output = [];
|
|
for (const row of rows) {
|
|
const family = textValue(row.entityFamily) || textValue(row.payload?.family) || projectionOutboxFamily(row);
|
|
const fact = projectionOutboxFact(row);
|
|
if (!fact || !SUPPORTED_FAMILIES.has(family)) continue;
|
|
const item = projectionRealtimeItem(row, fact, family);
|
|
if (!item) continue;
|
|
emitted.add(`${family}:${item.payload.entity.id}`);
|
|
output.push(item);
|
|
}
|
|
if (includeSnapshot) {
|
|
const cutoff = nonNegativeInteger(snapshot.cutoffOutboxSeq);
|
|
for (const [family, name] of [["messages", "workbench.message.snapshot"], ["turns", "workbench.turn.snapshot"]]) {
|
|
for (const fact of factArray(facts[family])) {
|
|
const id = projectionFactId(fact, family);
|
|
if (!id || emitted.has(`${family}:${id}`)) continue;
|
|
const item = projectionRealtimeItem({ outboxSeq: cutoff, entityFamily: family, entityId: id, projectedSeq: fact.projectedSeq, projectionRevision: fact.projectedSeq, createdAt: fact.updatedAt }, fact, family, name);
|
|
if (item) output.push(item);
|
|
}
|
|
}
|
|
}
|
|
return output.sort((left, right) => nonNegativeInteger(left.payload.cursor?.outboxSeq) - nonNegativeInteger(right.payload.cursor?.outboxSeq));
|
|
}
|
|
|
|
function projectionRealtimeItem(row, fact, family, forcedName = null) {
|
|
const id = textValue(row.entityId) || projectionFactId(fact, family);
|
|
if (!id) return null;
|
|
const outboxSeq = nonNegativeInteger(row.outboxSeq);
|
|
const version = nonNegativeInteger(row.projectionRevision ?? row.projectedSeq ?? fact.projectedSeq ?? outboxSeq);
|
|
const projectionRevision = String(version);
|
|
const base = {
|
|
contractVersion: "workbench-sync-v1",
|
|
realtimeAuthority: WORKBENCH_REALTIME_AUTHORITY_VERSION,
|
|
realtimeSource: "projection-outbox",
|
|
sessionId: textValue(fact.sessionId ?? row.sessionId),
|
|
traceId: textValue(fact.traceId ?? row.traceId),
|
|
cursor: { outboxSeq, traceSeq: nonNegativeInteger(row.projectedSeq ?? fact.projectedSeq) },
|
|
entity: { family, id, version, entityVersion: version, outboxSeq, traceSeq: nonNegativeInteger(row.projectedSeq ?? fact.projectedSeq), projectionRevision, authority: WORKBENCH_REALTIME_AUTHORITY_VERSION },
|
|
family,
|
|
id,
|
|
commitType: textValue(row.commitType),
|
|
terminal: row.terminal === true || fact.terminal === true,
|
|
sealed: row.sealed === true || fact.sealed === true,
|
|
projectionRevision,
|
|
outboxEventId: textValue(row.outboxEventId),
|
|
eventCreatedAt: textValue(row.createdAt ?? fact.updatedAt ?? fact.createdAt),
|
|
valuesRedacted: true
|
|
};
|
|
if (family === "messages") return { name: forcedName ?? "workbench.message.snapshot", payload: { ...base, type: "message.snapshot", reason: "projection-outbox", message: fact } };
|
|
if (family === "turns") return { name: forcedName ?? "workbench.turn.snapshot", payload: { ...base, type: "turn.snapshot", reason: "projection-outbox", turn: fact } };
|
|
return { name: forcedName ?? "workbench.trace.event", payload: { ...base, type: "trace.event", event: fact, snapshot: { traceId: base.traceId, sessionId: base.sessionId, status: fact.status ?? null, events: [fact], eventCount: 1 } } };
|
|
}
|
|
|
|
function projectionOutboxFact(row) {
|
|
const fact = row?.payload?.fact;
|
|
return fact && typeof fact === "object" && !Array.isArray(fact) ? fact : null;
|
|
}
|
|
|
|
function projectionOutboxFamily(row) {
|
|
if (row.commitType === "message") return "messages";
|
|
if (row.commitType === "terminal" && row.entityFamily !== "traceEvents") return "turns";
|
|
return "traceEvents";
|
|
}
|
|
|
|
function projectionFactId(fact, family) {
|
|
if (family === "messages") return textValue(fact.messageId ?? fact.id);
|
|
if (family === "turns") return textValue(fact.turnId ?? fact.traceId);
|
|
return textValue(fact.id ?? fact.sourceEventId);
|
|
}
|
|
|
|
function factArray(value) {
|
|
return Array.isArray(value) ? value : [];
|
|
}
|
|
|
|
function objectValue(value) {
|
|
return value && typeof value === "object" && !Array.isArray(value) ? value : {};
|
|
}
|
|
|
|
function textValue(value) {
|
|
const text = typeof value === "string" ? value.trim() : value === null || value === undefined ? "" : String(value).trim();
|
|
return text || "";
|
|
}
|
|
|
|
function nonNegativeInteger(value) {
|
|
const parsed = Number.parseInt(String(value ?? ""), 10);
|
|
return Number.isInteger(parsed) && parsed >= 0 ? parsed : 0;
|
|
}
|
|
|
|
const SUPPORTED_FAMILIES = new Set(["messages", "turns", "traceEvents"]);
|