pikasTech-HWLAB/internal/db/runtime-store-postgres-notify.ts

/* PostgreSQL projection notifications wake SSE readers; the outbox remains the payload authority. */

const WORKBENCH_PROJECTION_CHANNEL = "hwlab_workbench_projection";
const RECONNECT_DELAY_MS = 500;
const hubs = new WeakMap();

export async function subscribeWorkbenchProjectionCommits(store, listener) {
  if (typeof listener !== "function") throw codedError("workbench_projection_listener_invalid", "Projection notification listener must be a function.");
  let hub = hubs.get(store);
  if (!hub) {
    hub = { store, listeners: new Set(), client: null, connectPromise: null, reconnectTimer: null, connectedOnce: false, closing: false };
    hubs.set(store, hub);
  }
  hub.listeners.add(listener);
  hub.closing = false;
  await ensureConnected(hub);
  let subscribed = true;
  return async () => {
    if (!subscribed) return;
    subscribed = false;
    hub.listeners.delete(listener);
    if (hub.listeners.size === 0) await closeHub(hub);
  };
}

async function ensureConnected(hub) {
  if (hub.client || hub.connectPromise || hub.closing || hub.listeners.size === 0) return hub.connectPromise;
  hub.connectPromise = (async () => {
    const queryClient = await hub.store.getQueryClient();
    const client = typeof queryClient?.connect === "function" ? await queryClient.connect() : queryClient;
    if (!client || typeof client.query !== "function" || typeof client.on !== "function") {
      client?.release?.();
      throw codedError("workbench_projection_notify_unsupported", "Postgres runtime adapter requires a dedicated LISTEN connection.");
    }
    const onNotification = (message) => {
      if (message?.channel !== WORKBENCH_PROJECTION_CHANNEL) return;
      const signal = parseSignal(message.payload);
      for (const notify of [...hub.listeners]) {
        try { notify(signal); } catch {}
      }
    };
    const onDisconnect = (error) => disconnectHubClient(hub, client, error);
    client.on("notification", onNotification);
    client.on("error", onDisconnect);
    client.on("end", onDisconnect);
    client.__hwlabProjectionNotifyHandlers = { onNotification, onDisconnect };
    try {
      await client.query(`LISTEN ${WORKBENCH_PROJECTION_CHANNEL}`);
    } catch (error) {
      detachClient(client, error);
      throw error;
    }
    if (hub.closing || hub.listeners.size === 0) {
      detachClient(client);
      return;
    }
    hub.client = client;
    if (hub.connectedOnce) emitRecovery(hub);
    hub.connectedOnce = true;
  })().finally(() => {
    hub.connectPromise = null;
  });
  return hub.connectPromise;
}

function disconnectHubClient(hub, client, error) {
  if (client && hub.client !== client) return;
  if (client) {
    hub.client = null;
    detachClient(client, error);
  }
  scheduleReconnect(hub, error);
}

function scheduleReconnect(hub, error) {
  if (hub.closing || hub.listeners.size === 0 || hub.reconnectTimer) return;
  hub.store.logger?.warn?.({ event: "workbench_projection_notify_disconnected", errorCode: error?.code ?? "UNKNOWN", valuesRedacted: true });
  hub.reconnectTimer = setTimeout(() => {
    hub.reconnectTimer = null;
    void ensureConnected(hub).catch((connectError) => scheduleReconnect(hub, connectError));
  }, RECONNECT_DELAY_MS);
  hub.reconnectTimer.unref?.();
}

async function closeHub(hub) {
  hub.closing = true;
  if (hub.reconnectTimer) clearTimeout(hub.reconnectTimer);
  hub.reconnectTimer = null;
  const client = hub.client;
  hub.client = null;
  if (client) {
    try { await client.query(`UNLISTEN ${WORKBENCH_PROJECTION_CHANNEL}`); } catch {}
    detachClient(client);
  }
  hubs.delete(hub.store);
}

function detachClient(client, error = null) {
  const handlers = client?.__hwlabProjectionNotifyHandlers;
  if (handlers) {
    client.off?.("notification", handlers.onNotification);
    client.off?.("error", handlers.onDisconnect);
    client.off?.("end", handlers.onDisconnect);
    delete client.__hwlabProjectionNotifyHandlers;
  }
  client?.release?.(error || undefined);
}

function emitRecovery(hub) {
  for (const notify of [...hub.listeners]) {
    try { notify({ recovery: true, valuesRedacted: true }); } catch {}
  }
}

function parseSignal(payload) {
  try {
    const value = JSON.parse(String(payload ?? "{}"));
    return {
      sessionId: text(value.sessionId),
      traceId: text(value.traceId),
      outboxSeq: nonNegativeInteger(value.outboxSeq),
      valuesRedacted: true
    };
  } catch {
    return { recovery: true, valuesRedacted: true };
  }
}

function text(value) {
  const normalized = String(value ?? "").trim();
  return normalized || null;
}

function nonNegativeInteger(value) {
  const number = Number(value);
  return Number.isInteger(number) && number >= 0 ? number : 0;
}

function codedError(code, message) {
  const error = new Error(message);
  error.code = code;
  return error;
}

export { WORKBENCH_PROJECTION_CHANNEL };