root
419 lines
28 KiB
TypeScript
419 lines
28 KiB
TypeScript
/*
|
|
* Durable AgentRun Kafka projector storage operations.
|
|
* Kafka I/O is intentionally outside every transaction in this module.
|
|
*/
|
|
import { createHash } from "node:crypto";
|
|
|
|
import {
|
|
indexWorkbenchAggregateEvents,
|
|
inputFactWithAggregateSeq,
|
|
nonNegativeInteger,
|
|
normalizeWorkbenchAggregateEventsForFacts,
|
|
normalizeWorkbenchFacts,
|
|
parseJsonColumn,
|
|
stableJson,
|
|
textOr
|
|
} from "./runtime-store-core.ts";
|
|
|
|
export async function commitAgentRunKafkaProjection(store, params = {}) {
|
|
const transport = normalizeTransport(params.transport);
|
|
const canonicalEvent = objectValue(params.canonicalEvent);
|
|
const projectedEvent = objectValue(params.projectedEvent);
|
|
const sourceEventId = requiredText(canonicalEvent.eventId, "canonical AgentRun eventId");
|
|
const traceId = requiredText(projectedEvent.traceId ?? canonicalEvent.traceId, "canonical AgentRun traceId");
|
|
const sessionId = text(projectedEvent.sessionId ?? canonicalEvent.hwlabSessionId);
|
|
const sourceSeq = nonNegativeInteger(canonicalEvent.sourceSeq ?? canonicalEvent.event?.seq);
|
|
const inputSha256 = requiredText(transport.inputSha256, "Kafka input sha256");
|
|
const now = store.now();
|
|
const result = await store.withDurableTransaction("workbench.kafka-projection.commit", async (client) => {
|
|
await lockKafkaInboxIdentity(client, transport, sourceEventId);
|
|
await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-projection:${traceId}`]);
|
|
const existing = await client.query(
|
|
"SELECT source_topic, source_partition, source_offset, source_event_id, input_sha256, status, projected_seq FROM workbench_kafka_inbox WHERE (source_topic = $1 AND source_partition = $2 AND source_offset = $3) OR (source_topic = $1 AND source_event_id = $4) ORDER BY created_at ASC LIMIT 1",
|
|
[transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, sourceEventId]
|
|
);
|
|
const prior = existing.rows?.[0] ?? null;
|
|
if (prior) {
|
|
if (prior.input_sha256 !== inputSha256) {
|
|
const diagnostic = { code: "workbench_kafka_inbox_hash_conflict", message: "Kafka transport or source event identity was reused with a different payload hash.", valuesRedacted: true };
|
|
if (prior.status !== "projected") {
|
|
await client.query(
|
|
"UPDATE workbench_kafka_inbox SET status = 'failed', error_json = $1, updated_at = $2 WHERE source_topic = $3 AND source_partition = $4 AND source_offset = $5",
|
|
[stableJson(diagnostic), now, prior.source_topic, prior.source_partition, prior.source_offset]
|
|
);
|
|
}
|
|
await insertDlq(client, { ...transport, sourceEventId, inputSha256, error: diagnostic, envelope: redactedEnvelope(canonicalEvent), createdAt: now });
|
|
return { conflict: true, diagnostic, duplicate: false, projectedSeq: nonNegativeInteger(prior.projected_seq) };
|
|
}
|
|
return { duplicate: true, conflict: false, status: prior.status, projectedSeq: nonNegativeInteger(prior.projected_seq) };
|
|
}
|
|
await client.query(
|
|
"INSERT INTO workbench_kafka_inbox (source_topic, source_partition, source_offset, source_key, source_event_id, input_sha256, trace_id, session_id, run_id, command_id, source_seq, status, projected_seq, envelope_json, error_json, created_at, updated_at, projected_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,'processing',NULL,$12,'{}',$13,$13,NULL)",
|
|
[transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, transport.sourceKey, sourceEventId, inputSha256, traceId, sessionId, text(canonicalEvent.runId), text(canonicalEvent.commandId), sourceSeq, stableJson(redactedEnvelope(canonicalEvent)), now]
|
|
);
|
|
const checkpointResult = await client.query("SELECT checkpoint_json FROM workbench_projection_checkpoints WHERE trace_id = $1 FOR UPDATE", [traceId]);
|
|
const previousCheckpoint = parseJsonColumn(checkpointResult.rows?.[0]?.checkpoint_json, null);
|
|
const sessionResult = sessionId
|
|
? await client.query("SELECT owner_user_id, project_id, conversation_id, thread_id, session_json FROM workbench_sessions WHERE session_id = $1 FOR UPDATE", [sessionId])
|
|
: { rows: [] };
|
|
const existingSession = sessionResult.rows?.[0] ?? null;
|
|
const maxResult = await client.query(
|
|
"SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events WHERE trace_id = $1), 0), COALESCE((SELECT projected_seq FROM workbench_projection_checkpoints WHERE trace_id = $1), 0))::int AS max_projected_seq",
|
|
[traceId]
|
|
);
|
|
const projectedSeq = nonNegativeInteger(maxResult.rows?.[0]?.max_projected_seq) + 1;
|
|
if (typeof params.factsFactory !== "function") throw codedError("workbench_kafka_facts_factory_required", "Kafka projection requires a pure facts factory.");
|
|
const built = await params.factsFactory({ projectedSeq, previousCheckpoint, projectedAt: now });
|
|
let facts = null;
|
|
let persistedEvents = [];
|
|
if (built?.written !== false) {
|
|
facts = normalizeWorkbenchFacts(mergeKafkaSessionIdentity(built?.facts ?? {}, existingSession), params.requestMeta ?? {}, now);
|
|
const persisted = await persistFacts(store, client, facts, params.requestMeta ?? {});
|
|
persistedEvents = persisted.events;
|
|
}
|
|
const committedProjectedSeq = built?.written === false
|
|
? nonNegativeInteger(built?.projectedSeq) || nonNegativeInteger(previousCheckpoint?.projectedSeq) || projectedSeq
|
|
: projectedSeq;
|
|
const projectionWritten = built?.written !== false;
|
|
if (projectionWritten) {
|
|
const hwlabEvent = objectValue(params.hwlabEvent);
|
|
const hwlabEventId = requiredText(hwlabEvent.eventId, "HWLAB Kafka eventId");
|
|
await client.query(
|
|
"INSERT INTO hwlab_kafka_outbox (event_id, topic, partition_key, payload_json, headers_json, attempt_count, next_attempt_at, lease_owner, lease_expires_at, published_at, last_error, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,0,$6,NULL,NULL,NULL,NULL,$6,$6) ON CONFLICT (event_id) DO NOTHING",
|
|
[hwlabEventId, requiredText(params.hwlabTopic, "HWLAB Kafka topic"), requiredText(params.partitionKey, "HWLAB Kafka partition key"), stableJson(hwlabEvent), stableJson(objectValue(params.headers)), now]
|
|
);
|
|
}
|
|
await client.query(
|
|
"UPDATE workbench_kafka_inbox SET status = 'projected', projected_seq = $1, updated_at = $2, projected_at = $2 WHERE source_topic = $3 AND source_partition = $4 AND source_offset = $5",
|
|
[committedProjectedSeq, now, transport.sourceTopic, transport.sourcePartition, transport.sourceOffset]
|
|
);
|
|
return { duplicate: false, conflict: false, projectedSeq: committedProjectedSeq, facts, events: persistedEvents, projectionWritten, suppressedAfterSeal: built?.suppressedAfterSeal === true };
|
|
});
|
|
if (result.facts) store.memory.writeWorkbenchFacts({ facts: result.facts }, params.requestMeta ?? {});
|
|
return { ...result, transport, sourceEventId, traceId, sessionId, sourceSeq, valuesPrinted: false };
|
|
}
|
|
|
|
export async function recordFailedAgentRunKafkaMessage(store, params = {}) {
|
|
const transport = normalizeTransport(params.transport);
|
|
const now = store.now();
|
|
const error = redactedError(params.error);
|
|
const sourceEventId = text(params.sourceEventId);
|
|
const inputSha256 = requiredText(transport.inputSha256, "Kafka input sha256");
|
|
const result = await store.withDurableTransaction("workbench.kafka-projection.dlq", async (client) => {
|
|
await lockKafkaInboxIdentity(client, transport, sourceEventId);
|
|
const existing = await client.query("SELECT source_topic, source_partition, source_offset, input_sha256, status, projected_seq FROM workbench_kafka_inbox WHERE (source_topic = $1 AND source_partition = $2 AND source_offset = $3) OR ($4::text IS NOT NULL AND source_topic = $1 AND source_event_id = $4) ORDER BY created_at ASC LIMIT 1 FOR UPDATE", [transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, sourceEventId]);
|
|
const prior = existing.rows?.[0] ?? null;
|
|
if (prior) {
|
|
const sameTransport = prior.source_partition === transport.sourcePartition && String(prior.source_offset) === transport.sourceOffset;
|
|
const diagnostic = prior.input_sha256 !== inputSha256
|
|
? { ...error, code: "workbench_kafka_inbox_hash_conflict" }
|
|
: sameTransport ? error : { ...error, code: "workbench_kafka_inbox_source_event_duplicate" };
|
|
if (prior.status !== "projected") {
|
|
await client.query("UPDATE workbench_kafka_inbox SET status = 'failed', error_json = $1, updated_at = $2 WHERE source_topic = $3 AND source_partition = $4 AND source_offset = $5", [stableJson(diagnostic), now, prior.source_topic, prior.source_partition, prior.source_offset]);
|
|
}
|
|
await insertDlq(client, { ...transport, sourceEventId, inputSha256, error: diagnostic, envelope: redactedEnvelope(params.envelope), createdAt: now });
|
|
return { duplicate: sameTransport, conflict: !sameTransport || prior.input_sha256 !== inputSha256, priorStatus: prior.status, projectedSeq: nonNegativeInteger(prior.projected_seq), diagnostic };
|
|
}
|
|
await client.query(
|
|
"INSERT INTO workbench_kafka_inbox (source_topic, source_partition, source_offset, source_key, source_event_id, input_sha256, trace_id, session_id, run_id, command_id, source_seq, status, projected_seq, envelope_json, error_json, created_at, updated_at, projected_at) VALUES ($1,$2,$3,$4,$5,$6,NULL,NULL,NULL,NULL,0,'failed',NULL,$7,$8,$9,$9,NULL) ON CONFLICT (source_topic, source_partition, source_offset) DO UPDATE SET status = 'failed', error_json = EXCLUDED.error_json, updated_at = EXCLUDED.updated_at",
|
|
[transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, transport.sourceKey, sourceEventId, inputSha256, stableJson(redactedEnvelope(params.envelope)), stableJson(error), now]
|
|
);
|
|
await insertDlq(client, { ...transport, sourceEventId, inputSha256, error, envelope: redactedEnvelope(params.envelope), createdAt: now });
|
|
return { duplicate: false, conflict: false, priorStatus: null, diagnostic: error };
|
|
});
|
|
return { recorded: true, transport, sourceEventId, error: result.diagnostic, ...result, valuesPrinted: false };
|
|
}
|
|
|
|
export async function recordIgnoredAgentRunKafkaMessage(store, params = {}) {
|
|
const transport = normalizeTransport(params.transport);
|
|
const envelope = objectValue(params.envelope);
|
|
const sourceEventId = requiredText(envelope.eventId, "canonical AgentRun eventId");
|
|
const inputSha256 = requiredText(transport.inputSha256, "Kafka input sha256");
|
|
const now = store.now();
|
|
return store.withDurableTransaction("workbench.kafka-projection.ignore", async (client) => {
|
|
await lockKafkaInboxIdentity(client, transport, sourceEventId);
|
|
const existing = await client.query("SELECT input_sha256, status FROM workbench_kafka_inbox WHERE (source_topic = $1 AND source_partition = $2 AND source_offset = $3) OR (source_topic = $1 AND source_event_id = $4) ORDER BY created_at ASC LIMIT 1", [transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, sourceEventId]);
|
|
const prior = existing.rows?.[0] ?? null;
|
|
if (prior) {
|
|
if (prior.input_sha256 !== inputSha256) {
|
|
const diagnostic = { code: "workbench_kafka_inbox_hash_conflict", message: "Ignored Kafka identity was reused with a different payload hash.", valuesRedacted: true };
|
|
await insertDlq(client, { ...transport, sourceEventId, inputSha256, error: diagnostic, envelope: redactedEnvelope(envelope), createdAt: now });
|
|
return { ignored: true, duplicate: false, conflict: true, status: prior.status, diagnostic, valuesPrinted: false };
|
|
}
|
|
return { ignored: true, duplicate: true, status: prior.status, valuesPrinted: false };
|
|
}
|
|
await client.query(
|
|
"INSERT INTO workbench_kafka_inbox (source_topic, source_partition, source_offset, source_key, source_event_id, input_sha256, trace_id, session_id, run_id, command_id, source_seq, status, projected_seq, envelope_json, error_json, created_at, updated_at, projected_at) VALUES ($1,$2,$3,$4,$5,$6,NULL,NULL,$7,$8,$9,'ignored',NULL,$10,'{}',$11,$11,NULL)",
|
|
[transport.sourceTopic, transport.sourcePartition, transport.sourceOffset, transport.sourceKey, sourceEventId, inputSha256, text(envelope.runId), text(envelope.commandId), nonNegativeInteger(envelope.sourceSeq), stableJson(redactedEnvelope(envelope)), now]
|
|
);
|
|
return { ignored: true, duplicate: false, status: "ignored", valuesPrinted: false };
|
|
});
|
|
}
|
|
|
|
export async function claimHwlabKafkaOutbox(store, { owner, leaseMs, limit } = {}) {
|
|
const leaseOwner = requiredText(owner, "HWLAB Kafka relay owner");
|
|
const boundedLimit = boundedInteger(limit, 100, 1, 500);
|
|
const now = store.now();
|
|
const leaseExpiresAt = new Date(Date.parse(now) + boundedInteger(leaseMs, 60_000, 1_000, 3_600_000)).toISOString();
|
|
const result = await store.withDurableTransaction("hwlab.kafka-outbox.claim", (client) => client.query(
|
|
`WITH candidates AS (
|
|
SELECT candidate.outbox_seq FROM hwlab_kafka_outbox candidate
|
|
WHERE candidate.published_at IS NULL
|
|
AND candidate.next_attempt_at <= $1
|
|
AND (candidate.lease_expires_at IS NULL OR candidate.lease_expires_at <= $1)
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM hwlab_kafka_outbox earlier
|
|
WHERE earlier.published_at IS NULL
|
|
AND earlier.partition_key = candidate.partition_key
|
|
AND earlier.outbox_seq < candidate.outbox_seq
|
|
)
|
|
ORDER BY candidate.outbox_seq ASC
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT $2
|
|
)
|
|
UPDATE hwlab_kafka_outbox o SET attempt_count = o.attempt_count + 1, lease_owner = $3, lease_expires_at = $4, updated_at = $1
|
|
FROM candidates WHERE o.outbox_seq = candidates.outbox_seq
|
|
RETURNING o.*`,
|
|
[now, boundedLimit, leaseOwner, leaseExpiresAt]
|
|
));
|
|
return (result.rows ?? []).map(outboxRow);
|
|
}
|
|
|
|
export async function completeHwlabKafkaOutbox(store, item = {}) {
|
|
const result = await store.query(
|
|
"UPDATE hwlab_kafka_outbox SET published_at = $1, lease_owner = NULL, lease_expires_at = NULL, last_error = NULL, updated_at = $1 WHERE outbox_seq = $2 AND lease_owner = $3 AND attempt_count = $4 AND lease_expires_at = $5 AND lease_expires_at > $1 AND published_at IS NULL RETURNING outbox_seq",
|
|
[store.now(), item.outboxSeq, item.leaseOwner, item.attemptCount, item.leaseExpiresAt]
|
|
);
|
|
if (!result.rows?.[0]) throw codedError("hwlab_kafka_outbox_lease_conflict", "HWLAB Kafka outbox claim is stale.");
|
|
return { completed: true, outboxSeq: Number(result.rows[0].outbox_seq), valuesPrinted: false };
|
|
}
|
|
|
|
export async function retryHwlabKafkaOutbox(store, item = {}, error = null, retryAt = null) {
|
|
const now = store.now();
|
|
const nextAttemptAt = text(retryAt) ?? now;
|
|
const result = await store.query(
|
|
"UPDATE hwlab_kafka_outbox SET next_attempt_at = $1, lease_owner = NULL, lease_expires_at = NULL, last_error = $2, updated_at = $3 WHERE outbox_seq = $4 AND lease_owner = $5 AND attempt_count = $6 AND lease_expires_at = $7 AND lease_expires_at > $3 AND published_at IS NULL RETURNING outbox_seq",
|
|
[nextAttemptAt, stableJson(redactedError(error)), now, item.outboxSeq, item.leaseOwner, item.attemptCount, item.leaseExpiresAt]
|
|
);
|
|
if (!result.rows?.[0]) throw codedError("hwlab_kafka_outbox_lease_conflict", "HWLAB Kafka outbox claim is stale.");
|
|
return { retried: true, outboxSeq: Number(result.rows[0].outbox_seq), nextAttemptAt, valuesPrinted: false };
|
|
}
|
|
|
|
export async function hwlabKafkaProjectorStatus(store) {
|
|
const result = await store.query(
|
|
`SELECT
|
|
COUNT(*) FILTER (WHERE published_at IS NULL)::int AS backlog_count,
|
|
COUNT(*) FILTER (WHERE published_at IS NULL AND attempt_count > 0)::int AS retry_count,
|
|
MIN(created_at) FILTER (WHERE published_at IS NULL) AS oldest_pending_at,
|
|
(SELECT COUNT(*)::int FROM workbench_kafka_inbox WHERE status = 'failed') AS failed_inbox_count,
|
|
(SELECT COUNT(*)::int FROM workbench_kafka_dlq) AS dlq_count
|
|
FROM hwlab_kafka_outbox`, []
|
|
);
|
|
const row = result.rows?.[0] ?? {};
|
|
return { backlogCount: Number(row.backlog_count ?? 0), retryCount: Number(row.retry_count ?? 0), oldestPendingAt: row.oldest_pending_at ?? null, failedInboxCount: Number(row.failed_inbox_count ?? 0), dlqCount: Number(row.dlq_count ?? 0), valuesPrinted: false };
|
|
}
|
|
|
|
export async function readAtomicWorkbenchProjectionSync(store, params = {}) {
|
|
const since = nonNegativeInteger(params.afterOutboxSeq ?? params.afterSeq ?? params.since);
|
|
const limit = boundedInteger(params.limit, 100, 1, 500);
|
|
const traceId = text(params.traceId);
|
|
const sessionId = text(params.sessionId);
|
|
const snapshotOnly = params.snapshotOnly === true;
|
|
const deltaOnly = params.deltaOnly === true;
|
|
if (!traceId && !sessionId) throw codedError("workbench_sync_scope_required", "Atomic Workbench sync requires sessionId or traceId.");
|
|
if (snapshotOnly && deltaOnly) throw codedError("workbench_sync_mode_invalid", "snapshotOnly and deltaOnly are mutually exclusive.");
|
|
return store.withDurableTransaction("workbench.projection-sync.read", async (client) => {
|
|
await client.query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY");
|
|
const filter = traceId ? { column: "trace_id", value: traceId } : { column: "session_id", value: sessionId };
|
|
const cutoffResult = await client.query(`SELECT COALESCE(MAX(outbox_seq), 0)::bigint AS cutoff FROM workbench_projection_outbox WHERE ${filter.column} = $1`, [filter.value]);
|
|
const cutoff = Number(cutoffResult.rows?.[0]?.cutoff ?? 0);
|
|
const outboxResult = snapshotOnly ? { rows: [] } : await client.query(
|
|
`SELECT outbox_seq, outbox_event_id, entity_family, entity_id, event_seq, aggregate_id, aggregate_seq, projection_revision, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE outbox_seq > $1 AND outbox_seq <= $2 AND ${filter.column} = $3 ORDER BY outbox_seq ASC LIMIT $4`,
|
|
[since, cutoff, filter.value, limit + 1]
|
|
);
|
|
const allRows = (outboxResult.rows ?? []).map(projectionOutboxRow);
|
|
const events = allRows.slice(0, limit);
|
|
const facts = deltaOnly ? {} : await readSyncFacts(client, { traceId, sessionId, limit: 500, families: snapshotOnly ? ["sessions", "messages", "turns"] : null });
|
|
const hasMore = allRows.length > limit;
|
|
return { facts, events, cutoffOutboxSeq: cutoff, cursorOutboxSeq: snapshotOnly ? cutoff : hasMore ? (events.at(-1)?.outboxSeq ?? since) : cutoff, hasMore, valuesPrinted: false };
|
|
});
|
|
}
|
|
|
|
async function persistFacts(store, client, facts, requestMeta) {
|
|
const aggregateEvents = normalizeWorkbenchAggregateEventsForFacts(facts, requestMeta, store.now());
|
|
const events = [];
|
|
for (const event of aggregateEvents) events.push(await store.persistWorkbenchAggregateEvent(event, client));
|
|
const eventIndex = indexWorkbenchAggregateEvents(events);
|
|
const inputFacts = facts.inputs.map((fact) => inputFactWithAggregateSeq(fact, eventIndex));
|
|
for (const fact of inputFacts) await store.persistWorkbenchSessionInputFact(fact, client);
|
|
for (const fact of facts.sessions) await store.persistWorkbenchSessionFact(fact, client);
|
|
for (const fact of facts.messages) await store.persistWorkbenchMessageFact(fact, client);
|
|
for (const fact of facts.parts) await store.persistWorkbenchPartFact(fact, client);
|
|
for (const fact of facts.turns) await store.persistWorkbenchTurnFact(fact, client);
|
|
for (const fact of facts.traceEvents) await store.persistWorkbenchTraceEventFact(fact, client);
|
|
for (const fact of facts.checkpoints) await store.persistWorkbenchProjectionCheckpoint(fact, client);
|
|
for (const fact of facts.messages) {
|
|
const event = eventIndex.get(`message:${fact.messageId}`) ?? eventIndex.get(`messages:${fact.messageId}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
|
|
await store.persistWorkbenchProjectionOutbox(outboxRecord(fact, event, "messages", "message", { role: fact.role, status: fact.status }), client);
|
|
}
|
|
for (const fact of facts.traceEvents) {
|
|
const event = eventIndex.get(`traceEvent:${fact.id}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
|
|
await store.persistWorkbenchProjectionOutbox(outboxRecord(fact, event, "traceEvents", fact.terminal ? "terminal" : "event", { type: fact.eventType }), client);
|
|
}
|
|
for (const fact of facts.turns.filter((item) => item.terminal || item.sealed)) {
|
|
const event = eventIndex.get(`turn:${fact.turnId}`) ?? eventIndex.get(`sourceEvent:${fact.sourceEventId}`) ?? null;
|
|
await store.persistWorkbenchProjectionOutbox(outboxRecord(fact, event, "turns", "terminal", { status: fact.status, failureKind: fact.failureKind }), client);
|
|
}
|
|
return { events, inputFacts };
|
|
}
|
|
|
|
function outboxRecord(fact, event, family, commitType, metadata) {
|
|
const entityId = text(family === "messages" ? fact.messageId : family === "turns" ? fact.turnId : fact.id) ?? text(fact.traceId) ?? "projection";
|
|
return {
|
|
outboxEventId: projectionOutboxEventId({ fact, event, family, entityId, commitType }),
|
|
entityFamily: family,
|
|
entityId,
|
|
eventSeq: event?.eventSeq ?? null,
|
|
aggregateId: event?.aggregateId ?? null,
|
|
aggregateSeq: event?.aggregateSeq ?? 0,
|
|
projectionRevision: event?.projectionRevision ?? fact.projectedSeq,
|
|
traceId: fact.traceId,
|
|
sessionId: fact.sessionId,
|
|
turnId: fact.turnId,
|
|
messageId: fact.messageId,
|
|
projectedSeq: fact.projectedSeq,
|
|
sourceSeq: fact.sourceSeq,
|
|
sourceEventId: fact.sourceEventId,
|
|
commitType,
|
|
terminal: Boolean(fact.terminal),
|
|
sealed: Boolean(fact.sealed),
|
|
payload: {
|
|
family,
|
|
fact,
|
|
metadata: { ...metadata, eventSeq: event?.eventSeq ?? null, aggregateSeq: event?.aggregateSeq ?? null, projectedSeq: fact.projectedSeq, valuesRedacted: true },
|
|
valuesRedacted: true
|
|
},
|
|
createdAt: fact.occurredAt ?? fact.updatedAt ?? fact.createdAt
|
|
};
|
|
}
|
|
|
|
export function projectionOutboxEventId({ fact = {}, event = null, family, entityId, commitType }) {
|
|
const sourceEventId = text(fact.sourceEventId);
|
|
if (sourceEventId) return `${sourceEventId}:${family}:${entityId}:${commitType}`;
|
|
const identity = stableJson({
|
|
family,
|
|
entityId,
|
|
commitType,
|
|
eventSeq: event?.eventSeq ?? null,
|
|
aggregateId: event?.aggregateId ?? null,
|
|
aggregateSeq: event?.aggregateSeq ?? 0,
|
|
projectionRevision: event?.projectionRevision ?? fact.projectedSeq ?? 0,
|
|
projectedSeq: fact.projectedSeq ?? 0,
|
|
sourceSeq: fact.sourceSeq ?? 0
|
|
});
|
|
return `projection:${createHash("sha256").update(identity).digest("hex")}`;
|
|
}
|
|
|
|
function mergeKafkaSessionIdentity(facts, row) {
|
|
if (!row) return facts;
|
|
const existing = parseJsonColumn(row.session_json, {});
|
|
const sessions = Array.isArray(facts?.sessions) ? facts.sessions.map((fact) => ({
|
|
...existing,
|
|
...fact,
|
|
ownerUserId: fact.ownerUserId ?? row.owner_user_id ?? existing.ownerUserId ?? null,
|
|
projectId: fact.projectId ?? row.project_id ?? existing.projectId ?? null,
|
|
conversationId: fact.conversationId ?? row.conversation_id ?? existing.conversationId ?? null,
|
|
threadId: fact.threadId ?? row.thread_id ?? existing.threadId ?? null,
|
|
ownerRole: fact.ownerRole ?? existing.ownerRole ?? null
|
|
})) : [];
|
|
return { ...facts, sessions };
|
|
}
|
|
|
|
async function readSyncFacts(client, { traceId, sessionId, limit, families = null }) {
|
|
const familySet = Array.isArray(families) ? new Set(families) : null;
|
|
const specs = [
|
|
["sessions", "workbench_sessions", "session_json", traceId ? "last_trace_id" : "session_id"],
|
|
["messages", "workbench_messages", "message_json", traceId ? "trace_id" : "session_id"],
|
|
["parts", "workbench_parts", "part_json", traceId ? "trace_id" : "session_id"],
|
|
["turns", "workbench_turns", "turn_json", traceId ? "trace_id" : "session_id"],
|
|
["traceEvents", "workbench_trace_events", "event_json", traceId ? "trace_id" : "session_id"],
|
|
["checkpoints", "workbench_projection_checkpoints", "checkpoint_json", traceId ? "trace_id" : "session_id"]
|
|
].filter(([family]) => !familySet || familySet.has(family));
|
|
const value = traceId ?? sessionId;
|
|
const facts = {};
|
|
for (const [family, table, jsonColumn, column] of specs) {
|
|
const result = await client.query(`SELECT ${jsonColumn} FROM ${table} WHERE ${column} = $1 ORDER BY updated_at ASC LIMIT $2`, [value, limit]);
|
|
facts[family] = (result.rows ?? []).map((row) => parseJsonColumn(row[jsonColumn], null)).filter(Boolean);
|
|
}
|
|
return facts;
|
|
}
|
|
|
|
async function insertDlq(client, record) {
|
|
await client.query(
|
|
"INSERT INTO workbench_kafka_dlq (source_topic, source_partition, source_offset, source_event_id, input_sha256, error_code, error_message, envelope_json, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) ON CONFLICT (source_topic, source_partition, source_offset) DO UPDATE SET error_code = EXCLUDED.error_code, error_message = EXCLUDED.error_message, envelope_json = EXCLUDED.envelope_json",
|
|
[record.sourceTopic, record.sourcePartition, record.sourceOffset, record.sourceEventId, record.inputSha256, record.error.code, record.error.message, stableJson(record.envelope), record.createdAt]
|
|
);
|
|
}
|
|
|
|
async function lockKafkaInboxIdentity(client, transport, sourceEventId) {
|
|
const transportIdentity = `${transport.sourceTopic}:transport:${transport.sourcePartition}:${transport.sourceOffset}`;
|
|
await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-kafka-inbox:${transportIdentity}`]);
|
|
if (sourceEventId) {
|
|
const eventIdentity = `${transport.sourceTopic}:event:${sourceEventId}`;
|
|
await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-kafka-inbox:${eventIdentity}`]);
|
|
}
|
|
}
|
|
|
|
function normalizeTransport(value = {}) {
|
|
const input = objectValue(value);
|
|
const sourceTopic = requiredText(input.sourceTopic ?? input.topic, "Kafka source topic");
|
|
const sourcePartition = Number(input.sourcePartition ?? input.partition);
|
|
const sourceOffset = text(input.sourceOffset ?? input.offset);
|
|
if (!Number.isInteger(sourcePartition) || sourcePartition < 0) throw codedError("workbench_kafka_partition_invalid", "Kafka source partition must be a non-negative integer.");
|
|
if (!/^\d+$/u.test(sourceOffset ?? "")) throw codedError("workbench_kafka_offset_invalid", "Kafka source offset must be a non-negative integer string.");
|
|
return { sourceTopic, sourcePartition, sourceOffset, inputSha256: text(input.inputSha256), sourceKey: text(input.sourceKey), valuesPrinted: false };
|
|
}
|
|
|
|
function projectionOutboxRow(row) {
|
|
return { outboxSeq: Number(row.outbox_seq), outboxEventId: row.outbox_event_id, entityFamily: row.entity_family, entityId: row.entity_id, eventSeq: row.event_seq == null ? null : Number(row.event_seq), aggregateId: row.aggregate_id, aggregateSeq: Number(row.aggregate_seq ?? 0), projectionRevision: Number(row.projection_revision ?? 0), traceId: row.trace_id, sessionId: row.session_id, turnId: row.turn_id, messageId: row.message_id, projectedSeq: Number(row.projected_seq ?? 0), sourceSeq: Number(row.source_seq ?? 0), sourceEventId: row.source_event_id, commitType: row.commit_type, terminal: Boolean(row.terminal), sealed: Boolean(row.sealed), payload: parseJsonColumn(row.payload_json, {}), createdAt: row.created_at, valuesPrinted: false };
|
|
}
|
|
|
|
function outboxRow(row) {
|
|
return { outboxSeq: Number(row.outbox_seq), eventId: row.event_id, topic: row.topic, partitionKey: row.partition_key, payload: parseJsonColumn(row.payload_json, {}), headers: parseJsonColumn(row.headers_json, {}), attemptCount: Number(row.attempt_count ?? 0), nextAttemptAt: row.next_attempt_at, leaseOwner: row.lease_owner, leaseExpiresAt: row.lease_expires_at, createdAt: row.created_at, valuesPrinted: false };
|
|
}
|
|
|
|
function redactedEnvelope(value) {
|
|
const input = objectValue(value);
|
|
return { schema: text(input.schema), eventType: text(input.eventType), eventId: text(input.eventId), outboxSeq: nonNegativeInteger(input.outboxSeq), sourceSeq: nonNegativeInteger(input.sourceSeq), traceId: text(input.traceId), sessionId: text(input.sessionId), hwlabSessionId: text(input.hwlabSessionId), runId: text(input.runId), commandId: text(input.commandId), valuesRedacted: true };
|
|
}
|
|
|
|
function redactedError(error) {
|
|
return { code: text(error?.code) ?? "workbench_kafka_message_invalid", message: String(error?.message ?? error ?? "Kafka message is invalid.").slice(0, 500), valuesRedacted: true };
|
|
}
|
|
|
|
function objectValue(value) {
|
|
return value && typeof value === "object" && !Array.isArray(value) ? value : {};
|
|
}
|
|
|
|
function text(value) {
|
|
const normalized = textOr(value, "");
|
|
return normalized || null;
|
|
}
|
|
|
|
function requiredText(value, label) {
|
|
const normalized = text(value);
|
|
if (!normalized) throw codedError("workbench_kafka_contract_invalid", `${label} is required.`);
|
|
return normalized;
|
|
}
|
|
|
|
function boundedInteger(value, fallback, min, max) {
|
|
const parsed = Number(value);
|
|
return Number.isInteger(parsed) && parsed >= min && parsed <= max ? parsed : fallback;
|
|
}
|
|
|
|
function codedError(code, message) {
|
|
const error = new Error(message);
|
|
error.code = code;
|
|
return error;
|
|
}
|