root
503 lines
16 KiB
TypeScript
503 lines
16 KiB
TypeScript
// SPEC: PJ2026-010401080313 Workbench实时权威 draft-2026-07-20-p0-process-session-index.
|
|
// Responsibility: accelerate retained hwlab.event.v1 session queries without creating another business authority.
|
|
|
|
import { createHash } from "node:crypto";
|
|
|
|
export function createWorkbenchKafkaSessionIndex(options = {}) {
|
|
const topic = text(options.topic);
|
|
const maxEvents = positiveInteger(options.maxEvents);
|
|
const maxEventsPerSession = positiveInteger(options.maxEventsPerSession);
|
|
const bootstrapTimeoutMs = positiveInteger(options.bootstrapTimeoutMs);
|
|
const rebuildIntervalMs = positiveInteger(options.rebuildIntervalMs);
|
|
const rebuildCooldownMs = positiveInteger(options.rebuildCooldownMs);
|
|
const scanLimit = positiveInteger(options.scanLimit);
|
|
const logger = options.logger ?? console;
|
|
const nowMs = typeof options.nowMs === "function" ? options.nowMs : Date.now;
|
|
if (!topic || !maxEvents || !maxEventsPerSession || !bootstrapTimeoutMs || !rebuildIntervalMs || !rebuildCooldownMs || !scanLimit || typeof options.queryAll !== "function") {
|
|
throw indexError("workbench_kafka_session_index_config_invalid", "Workbench Kafka session index requires complete bounded configuration.");
|
|
}
|
|
|
|
let ready = false;
|
|
let stopped = false;
|
|
let rebuilding = false;
|
|
let rebuildTimer = null;
|
|
let lastRebuildStartedAtMs = null;
|
|
let lastRebuildCompletedAtMs = null;
|
|
let lastFailure = null;
|
|
let bootstrapPromise = null;
|
|
let pendingLiveOverflow = false;
|
|
let state = emptyState();
|
|
const pendingLive = [];
|
|
|
|
function start() {
|
|
if (bootstrapPromise) return bootstrapPromise;
|
|
bootstrapPromise = rebuild("bootstrap");
|
|
rebuildTimer = setInterval(() => {
|
|
void rebuild("interval");
|
|
}, rebuildIntervalMs);
|
|
rebuildTimer.unref?.();
|
|
return bootstrapPromise;
|
|
}
|
|
|
|
function stop() {
|
|
stopped = true;
|
|
if (rebuildTimer) clearInterval(rebuildTimer);
|
|
rebuildTimer = null;
|
|
pendingLive.length = 0;
|
|
}
|
|
|
|
function observeLive(envelope, transport) {
|
|
if (stopped || envelope?.schema !== "hwlab.event.v1") return;
|
|
const record = normalizeRecord({ topic, partition: transport?.partition, offset: transport?.offset, key: sessionId(envelope), timestamp: null, value: envelope });
|
|
if (!record) return;
|
|
if (rebuilding) {
|
|
if (pendingLive.length >= maxEvents) {
|
|
pendingLiveOverflow = true;
|
|
lastFailure = warning("workbench_kafka_session_index_live_buffer_overflow", "Session index rebuild live buffer exceeded its YAML-owned bound.");
|
|
} else {
|
|
pendingLive.push(record);
|
|
}
|
|
}
|
|
if (ready) addRecord(state, record);
|
|
}
|
|
|
|
async function rebuild(reason) {
|
|
if (stopped || rebuilding) return false;
|
|
const startedAtMs = nowMs();
|
|
if (lastRebuildStartedAtMs !== null && reason !== "bootstrap" && startedAtMs - lastRebuildStartedAtMs < rebuildCooldownMs) return false;
|
|
rebuilding = true;
|
|
lastRebuildStartedAtMs = startedAtMs;
|
|
pendingLive.length = 0;
|
|
pendingLiveOverflow = false;
|
|
try {
|
|
const result = await options.queryAll({
|
|
limit: maxEvents,
|
|
scanLimit,
|
|
timeoutMs: bootstrapTimeoutMs
|
|
});
|
|
if (!completeQuery(result)) {
|
|
lastFailure = warning("workbench_kafka_session_index_bootstrap_incomplete", `Session index ${reason} did not reach the Kafka retention barrier.`);
|
|
logWarning(lastFailure, { reason, completionReason: result?.completionReason ?? null });
|
|
return false;
|
|
}
|
|
const candidate = emptyState();
|
|
candidate.endOffsets = normalizeEndOffsets(result.endOffsets);
|
|
for (const raw of Array.isArray(result.events) ? result.events : []) {
|
|
const record = normalizeRecord(raw);
|
|
if (!record) continue;
|
|
addRecord(candidate, record);
|
|
}
|
|
if (pendingLiveOverflow) {
|
|
throw indexError("workbench_kafka_session_index_live_buffer_overflow", "Session index rebuild cannot prove completeness after its live buffer overflowed.");
|
|
}
|
|
for (const record of pendingLive.splice(0)) addRecord(candidate, record);
|
|
state = candidate;
|
|
ready = true;
|
|
lastFailure = null;
|
|
lastRebuildCompletedAtMs = nowMs();
|
|
logger.log?.(JSON.stringify({
|
|
code: "workbench-kafka-session-index-ready",
|
|
reason,
|
|
indexedEventCount: state.records.length,
|
|
indexedSessionCount: state.bySession.size,
|
|
incompleteSessionCount: state.incompleteSessions.size,
|
|
durationMs: lastRebuildCompletedAtMs - startedAtMs,
|
|
valuesPrinted: false
|
|
}));
|
|
return true;
|
|
} catch (error) {
|
|
lastFailure = warning(error?.code ?? "workbench_kafka_session_index_bootstrap_failed", errorMessage(error));
|
|
logWarning(lastFailure, { reason });
|
|
return false;
|
|
} finally {
|
|
rebuilding = false;
|
|
}
|
|
}
|
|
|
|
function query(params = {}) {
|
|
const startedAtMs = nowMs();
|
|
const requestedSessionId = text(params.sessionId);
|
|
const requestedTraceId = text(params.traceId);
|
|
const fallbackReason = !ready
|
|
? "index-not-ready"
|
|
: requestedSessionId && state.incompleteSessions.has(requestedSessionId)
|
|
? "session-capacity-exceeded"
|
|
: !requestedSessionId && !state.globalComplete
|
|
? "global-capacity-exceeded"
|
|
: null;
|
|
if (fallbackReason) return {
|
|
hit: false,
|
|
fallbackReason,
|
|
status: status(),
|
|
warning: warning("workbench_kafka_session_index_fallback", `Session index cannot prove a complete result (${fallbackReason}).`)
|
|
};
|
|
const source = requestedSessionId ? state.bySession.get(requestedSessionId) ?? [] : state.records;
|
|
const events = source.filter((record) => matchesTrace(record.value, requestedTraceId));
|
|
const limit = positiveInteger(params.limit);
|
|
if (limit && events.length > limit) {
|
|
return {
|
|
hit: false,
|
|
fallbackReason: "matched-event-limit",
|
|
status: status(),
|
|
warning: warning("workbench_kafka_session_index_fallback", "Session index result exceeds the request matched-event limit.")
|
|
};
|
|
}
|
|
const elapsedMs = Math.max(0, nowMs() - startedAtMs);
|
|
const firstScannedOffsets = partitionOffsets(events, "first");
|
|
const lastScannedOffsets = partitionOffsets(events, "last");
|
|
return {
|
|
hit: true,
|
|
result: {
|
|
ok: true,
|
|
stream: "hwlab",
|
|
topic,
|
|
groupId: null,
|
|
count: events.length,
|
|
scannedCount: events.length,
|
|
parsedCount: events.length,
|
|
matchedCount: events.length,
|
|
invalidJsonCount: 0,
|
|
filterRejectedCount: 0,
|
|
keyRejectedCount: 0,
|
|
excludedPostBarrierCount: 0,
|
|
completionReason: "session-index",
|
|
completion: {
|
|
reason: "session-index",
|
|
complete: true,
|
|
barrierReached: true,
|
|
retentionStartVerified: true,
|
|
retentionStartMoved: false,
|
|
timeout: false,
|
|
matchedEventLimitReached: false,
|
|
scanLimitReached: false,
|
|
aborted: false,
|
|
valuesRedacted: true
|
|
},
|
|
reachedEndOffsets: true,
|
|
endOffsets: [...state.endOffsets.values()].sort((left, right) => left.partition - right.partition),
|
|
endOffsetsAvailable: state.endOffsets.size > 0,
|
|
endOffsetsError: null,
|
|
firstScannedOffsets,
|
|
lastScannedOffsets,
|
|
limit: limit ?? null,
|
|
scanLimit: positiveInteger(params.scanLimit),
|
|
timeoutMs: positiveInteger(params.timeoutMs),
|
|
filters: compact({ sessionId: requestedSessionId, traceId: requestedTraceId }),
|
|
partitionKeyScoped: Boolean(text(params.partitionKey)),
|
|
targetPartition: events.length > 0 ? events[0].partition : null,
|
|
timing: {
|
|
endOffsetSnapshotMs: 0,
|
|
groupOffsetsMs: 0,
|
|
consumerConnectMs: 0,
|
|
consumerSubscribeMs: 0,
|
|
scanMs: 0,
|
|
cleanupMs: 0,
|
|
indexLookupMs: elapsedMs,
|
|
totalMs: elapsedMs,
|
|
valuesPrinted: false
|
|
},
|
|
index: {
|
|
hit: true,
|
|
source: "process-session-index",
|
|
indexedEventCount: state.records.length,
|
|
indexedSessionCount: state.bySession.size,
|
|
incompleteSessionCount: state.incompleteSessions.size,
|
|
valuesRedacted: true
|
|
},
|
|
events,
|
|
valuesPrinted: false
|
|
}
|
|
};
|
|
}
|
|
|
|
function status() {
|
|
return {
|
|
enabled: true,
|
|
ready,
|
|
rebuilding,
|
|
indexedEventCount: state.records.length,
|
|
indexedSessionCount: state.bySession.size,
|
|
incompleteSessionCount: state.incompleteSessions.size,
|
|
globalComplete: state.globalComplete,
|
|
lastRebuildStartedAt: iso(lastRebuildStartedAtMs),
|
|
lastRebuildCompletedAt: iso(lastRebuildCompletedAtMs),
|
|
warning: lastFailure,
|
|
blocking: false,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function sessionSummaries() {
|
|
if (!ready || !state.globalComplete) {
|
|
const reason = !ready ? "index-not-ready" : "global-capacity-exceeded";
|
|
return {
|
|
ok: false,
|
|
sessions: [],
|
|
warning: warning("workbench_kafka_session_summary_unavailable", `Session summary index is unavailable (${reason}).`),
|
|
status: status(),
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
const sessions = [];
|
|
for (const [id, records] of state.bySession.entries()) {
|
|
if (state.incompleteSessions.has(id)) continue;
|
|
const userEvents = records.map((record) => record.value?.event).filter((event) => firstText(event?.type, event?.eventType) === "user");
|
|
const firstPreview = firstText(userEvents[0]?.text, userEvents[0]?.message);
|
|
if (!firstPreview) continue;
|
|
const lastUserEvent = userEvents.at(-1);
|
|
sessions.push({
|
|
sessionId: id,
|
|
firstUserMessagePreview: firstPreview.slice(0, 240),
|
|
lastUserMessageAt: firstText(lastUserEvent?.createdAt, lastUserEvent?.submittedAt),
|
|
valuesRedacted: true
|
|
});
|
|
}
|
|
return { ok: true, sessions, warning: null, status: status(), valuesRedacted: true };
|
|
}
|
|
|
|
function emptyState() {
|
|
return {
|
|
records: [],
|
|
byTransport: new Map(),
|
|
bySession: new Map(),
|
|
incompleteSessions: new Set(),
|
|
endOffsets: new Map(),
|
|
globalComplete: true
|
|
};
|
|
}
|
|
|
|
function addRecord(target, record) {
|
|
const transportKey = `${record.topic}:${record.partition}:${record.offset}`;
|
|
if (target.byTransport.has(transportKey)) return;
|
|
target.byTransport.set(transportKey, record);
|
|
target.records.push(record);
|
|
updateEndOffset(target.endOffsets, record.partition, record.offset);
|
|
const id = indexedSessionId(record);
|
|
if (id) {
|
|
const bucket = target.bySession.get(id) ?? [];
|
|
bucket.push(record);
|
|
target.bySession.set(id, bucket);
|
|
if (bucket.length > maxEventsPerSession) {
|
|
const removed = bucket.shift();
|
|
if (removed) removeRecord(target, removed);
|
|
target.incompleteSessions.add(id);
|
|
}
|
|
}
|
|
if (target.records.length > maxEvents) {
|
|
const removed = target.records[0];
|
|
removeRecord(target, removed);
|
|
const removedSessionId = indexedSessionId(removed);
|
|
if (removedSessionId) target.incompleteSessions.add(removedSessionId);
|
|
target.globalComplete = false;
|
|
}
|
|
}
|
|
|
|
function removeRecord(target, record) {
|
|
if (!record) return;
|
|
const transportKey = `${record.topic}:${record.partition}:${record.offset}`;
|
|
target.byTransport.delete(transportKey);
|
|
const recordIndex = target.records.indexOf(record);
|
|
if (recordIndex >= 0) target.records.splice(recordIndex, 1);
|
|
const id = indexedSessionId(record);
|
|
const bucket = id ? target.bySession.get(id) : null;
|
|
if (bucket) {
|
|
const bucketIndex = bucket.indexOf(record);
|
|
if (bucketIndex >= 0) bucket.splice(bucketIndex, 1);
|
|
if (bucket.length === 0) target.bySession.delete(id);
|
|
}
|
|
}
|
|
|
|
function logWarning(entry, fields = {}) {
|
|
logger.warn?.(JSON.stringify({
|
|
code: entry.code,
|
|
component: "workbench-kafka-session-index",
|
|
message: entry.message,
|
|
blocking: false,
|
|
...fields,
|
|
valuesPrinted: false
|
|
}));
|
|
}
|
|
|
|
return {
|
|
start,
|
|
stop,
|
|
observeLive,
|
|
query,
|
|
sessionSummaries,
|
|
status,
|
|
rebuild,
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
function normalizeRecord(raw) {
|
|
const recordTopic = text(raw?.topic);
|
|
const partition = integer(raw?.partition);
|
|
const offset = offsetValue(raw?.offset);
|
|
const value = raw?.value;
|
|
if (!recordTopic || partition === null || offset === null || value?.schema !== "hwlab.event.v1") return null;
|
|
return {
|
|
topic: recordTopic,
|
|
partition,
|
|
offset,
|
|
key: text(raw?.key),
|
|
timestamp: text(raw?.timestamp),
|
|
valueSha256: text(raw?.valueSha256) ?? sha256(JSON.stringify(value)),
|
|
value
|
|
};
|
|
}
|
|
|
|
function completeQuery(result) {
|
|
return result?.completion?.complete === true
|
|
&& result?.completion?.barrierReached === true
|
|
&& result?.completion?.retentionStartVerified === true
|
|
&& result?.reachedEndOffsets === true
|
|
&& result?.endOffsetsAvailable === true;
|
|
}
|
|
|
|
function normalizeEndOffsets(value) {
|
|
const result = new Map();
|
|
for (const entry of Array.isArray(value) ? value : []) {
|
|
const partition = integer(entry?.partition);
|
|
const endOffset = offsetValue(entry?.endOffset);
|
|
if (partition === null || endOffset === null) continue;
|
|
result.set(partition, {
|
|
partition,
|
|
startOffset: offsetValue(entry?.startOffset),
|
|
endOffset
|
|
});
|
|
}
|
|
return result;
|
|
}
|
|
|
|
function updateEndOffset(endOffsets, partition, offset) {
|
|
const next = nextOffset(offset);
|
|
if (next === null) return;
|
|
const existing = endOffsets.get(partition);
|
|
if (!existing || compareOffset(next, existing.endOffset) > 0) {
|
|
endOffsets.set(partition, {
|
|
partition,
|
|
startOffset: existing?.startOffset ?? null,
|
|
endOffset: next
|
|
});
|
|
}
|
|
}
|
|
|
|
function sessionId(value) {
|
|
return firstText(
|
|
value?.sessionId,
|
|
value?.hwlabSessionId,
|
|
value?.event?.sessionId,
|
|
value?.event?.hwlabSessionId,
|
|
value?.context?.sessionId,
|
|
value?.context?.hwlabSessionId
|
|
);
|
|
}
|
|
|
|
function indexedSessionId(record) {
|
|
const envelopeSessionId = sessionId(record?.value);
|
|
const key = text(record?.key);
|
|
if (!envelopeSessionId) return null;
|
|
if (key && key !== envelopeSessionId) return null;
|
|
return envelopeSessionId;
|
|
}
|
|
|
|
function matchesTrace(value, traceId) {
|
|
if (!traceId) return true;
|
|
return [
|
|
value?.traceId,
|
|
value?.event?.traceId,
|
|
value?.context?.traceId,
|
|
value?.sourceEvent?.traceId
|
|
].some((candidate) => text(candidate) === traceId);
|
|
}
|
|
|
|
function partitionOffsets(records, mode) {
|
|
const byPartition = new Map();
|
|
for (const record of records) {
|
|
const existing = byPartition.get(record.partition);
|
|
if (existing === undefined || (mode === "first" ? compareOffset(record.offset, existing) < 0 : compareOffset(record.offset, existing) > 0)) {
|
|
byPartition.set(record.partition, record.offset);
|
|
}
|
|
}
|
|
return [...byPartition.entries()]
|
|
.map(([partition, offset]) => ({ partition, offset }))
|
|
.sort((left, right) => left.partition - right.partition);
|
|
}
|
|
|
|
function warning(code, message) {
|
|
return { code, message, blocking: false, valuesRedacted: true };
|
|
}
|
|
|
|
function indexError(code, message) {
|
|
const error = new Error(message);
|
|
error.code = code;
|
|
return error;
|
|
}
|
|
|
|
function errorMessage(error) {
|
|
return error instanceof Error ? error.message : String(error ?? "unknown error");
|
|
}
|
|
|
|
function compact(value) {
|
|
return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== null && item !== undefined && item !== ""));
|
|
}
|
|
|
|
function firstText(...values) {
|
|
for (const value of values) {
|
|
const result = text(value);
|
|
if (result) return result;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function text(value) {
|
|
return typeof value === "string" && value.trim() ? value.trim() : null;
|
|
}
|
|
|
|
function integer(value) {
|
|
const result = Number(value);
|
|
return Number.isInteger(result) && result >= 0 ? result : null;
|
|
}
|
|
|
|
function positiveInteger(value) {
|
|
const result = Number(value);
|
|
return Number.isInteger(result) && result > 0 ? result : null;
|
|
}
|
|
|
|
function offsetValue(value) {
|
|
const result = text(String(value ?? ""));
|
|
if (!result) return null;
|
|
try {
|
|
return BigInt(result) >= 0n ? result : null;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function nextOffset(value) {
|
|
try {
|
|
return String(BigInt(String(value)) + 1n);
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function compareOffset(left, right) {
|
|
try {
|
|
const leftValue = BigInt(String(left));
|
|
const rightValue = BigInt(String(right));
|
|
return leftValue < rightValue ? -1 : leftValue > rightValue ? 1 : 0;
|
|
} catch {
|
|
return String(left).localeCompare(String(right));
|
|
}
|
|
}
|
|
|
|
function iso(value) {
|
|
return Number.isFinite(value) ? new Date(value).toISOString() : null;
|
|
}
|
|
|
|
function sha256(value) {
|
|
return createHash("sha256").update(value).digest("hex");
|
|
}
|