pikasTech-HWLAB/internal/cloud/workbench-kafka-rail-stream.ts

// SPEC: PJ2026-010401080313 Workbench realtime authority draft-2026-07-22-p0-rail-kafka-projection.
// Responsibility: deliver one scoped Kafka-derived session rail snapshot and semantic live deltas.

export function createWorkbenchKafkaRailProjectionStream(options = {}) {
  const bridge = options.bridge;
  const sessionIds = normalizeSessionIds(options.sessionIds);
  const sessionIdSet = new Set(sessionIds);
  const fingerprints = new Map();
  const pending = new Map();
  let unsubscribe = null;
  let live = false;
  let stopped = false;
  let failed = false;
  let snapshotCount = 0;
  let liveCount = 0;
  let deliveryChain = Promise.resolve(true);

  const fail = (error) => {
    if (failed || stopped) return;
    failed = true;
    pending.clear();
    unsubscribe?.();
    unsubscribe = null;
    options.onFailure?.(error);
  };

  const deliver = (summary, phase) => {
    if (stopped || failed || !summary) return Promise.resolve(false);
    const fingerprint = railSummaryFingerprint(summary);
    if (fingerprints.get(summary.sessionId) === fingerprint) return Promise.resolve(true);
    fingerprints.set(summary.sessionId, fingerprint);
    if (phase === "snapshot") snapshotCount += 1;
    else liveCount += 1;
    deliveryChain = deliveryChain.then(() => options.deliver?.({
      type: "session.rail",
      phase,
      authority: "hwlab.kafka.session-index",
      sessionId: summary.sessionId,
      session: summary,
      valuesRedacted: true
    }) ?? true);
    return deliveryChain;
  };

  const projectLive = (envelope) => {
    if (stopped || failed) return;
    const sessionId = envelopeSessionId(envelope);
    if (!sessionIdSet.has(sessionId)) return;
    const result = bridge?.workbenchSessionSummaries?.({ sessionIds: [sessionId] });
    if (result?.ok !== true) {
      fail(railProjectionError(result));
      return;
    }
    const summary = result.sessions?.find((item) => item?.sessionId === sessionId) ?? null;
    if (!summary) {
      if (fingerprints.has(sessionId)) {
        fail(railProjectionError({ warning: { code: "workbench_kafka_session_rail_missing", message: "Kafka rail projection lost a previously visible scoped session." } }));
      }
      return;
    }
    if (!live) pending.set(sessionId, summary);
    else void deliver(summary, "live").catch(fail);
  };

  async function start() {
    if (sessionIds.length === 0) return { ok: true, sessionIds, snapshotCount: 0, liveCount: 0, valuesRedacted: true };
    if (typeof bridge?.waitWorkbenchSessionRailReady !== "function" || typeof bridge?.workbenchSessionSummaries !== "function" || typeof bridge?.subscribeLiveHwlabEvents !== "function") {
      throw railProjectionError({ warning: { code: "workbench_kafka_session_rail_unconfigured", message: "Kafka rail projection runtime is not configured." } });
    }
    await bridge.waitWorkbenchSessionRailReady();
    if (stopped) return { ok: false, sessionIds, snapshotCount, liveCount, valuesRedacted: true };
    unsubscribe = bridge.subscribeLiveHwlabEvents(projectLive);
    const result = bridge.workbenchSessionSummaries({ sessionIds });
    if (result?.ok !== true) throw railProjectionError(result);
    const bySessionId = new Map((Array.isArray(result.sessions) ? result.sessions : []).map((summary) => [summary?.sessionId, summary]));
    for (const sessionId of sessionIds) {
      const summary = bySessionId.get(sessionId);
      if (!summary) continue;
      await deliver(summary, "snapshot");
    }
    live = true;
    for (const summary of pending.values()) await deliver(summary, "live");
    pending.clear();
    return { ok: true, sessionIds, snapshotCount, liveCount, valuesRedacted: true };
  }

  function stop() {
    if (stopped) return;
    stopped = true;
    pending.clear();
    unsubscribe?.();
    unsubscribe = null;
  }

  function status() {
    return { ok: !failed, live, stopped, sessionIds, snapshotCount, liveCount, valuesRedacted: true };
  }

  return { start, stop, status, valuesRedacted: true };
}

function railSummaryFingerprint(summary) {
  return JSON.stringify([
    summary?.sessionId ?? null,
    summary?.firstUserMessagePreview ?? null,
    summary?.lastUserMessageAt ?? null,
    summary?.lastTraceId ?? null,
    summary?.status ?? null
  ]);
}

function envelopeSessionId(envelope) {
  return text(envelope?.sessionId ?? envelope?.hwlabSessionId ?? envelope?.event?.sessionId);
}

function normalizeSessionIds(values) {
  return [...new Set((Array.isArray(values) ? values : [values]).map(text).filter(Boolean))].sort();
}

function railProjectionError(result) {
  const code = text(result?.warning?.code) || "workbench_kafka_session_rail_unavailable";
  const message = text(result?.warning?.message) || "Kafka session rail projection is unavailable.";
  const error = new Error(message);
  error.code = code;
  return error;
}

function text(value) {
  return typeof value === "string" && value.trim() ? value.trim() : null;
}