docs — Runtime

/**
 * What the SSE layer provides. Declared structurally rather than imported, so
 * this file has no dependencies at all — the Datastar SDK's
 * ServerSentEventGenerator satisfies it as-is.
 */
export type SsePatch = {
  patchElements(elements: string, options?: Record<string, unknown>): unknown;
  patchSignals(signals: string, options?: Record<string, unknown>): unknown;
  removeElements(
    selector?: string,
    elements?: string,
    options?: Record<string, unknown>,
  ): unknown;
  removeSignals(
    keys: string | string[],
    options?: Record<string, unknown>,
  ): unknown;
  executeScript(script: string, options?: Record<string, unknown>): unknown;
};

/**
 * The two things dispatch needs from an SSE layer. This is exactly the shape of
 * the Datastar SDK's ServerSentEventGenerator, so the suggested setup is to hand
 * the module straight over:
 *
 *   import { ServerSentEventGenerator } from "@starfederation/datastar-sdk/web";
 *   export const { cmd, dcmd } = setupCommands({ sse: ServerSentEventGenerator });
 *
 * Anything else that can open a stream and read signals works too — a different
 * runtime's SDK, a fake in tests, or your own writer.
 */
export type SseAdapter = {
  stream(
    onStart: (patch: SsePatch) => void | Promise<void>,
    options?: Record<string, unknown>,
  ): Response;
  readSignals(
    req: Request,
  ): Promise<
    | { success: true; signals: Record<string, any> }
    | { success: false; error: string }
  >;
};

const emitted: string[] = [];

const fake: SseAdapter = {
  stream(onStart) {
    const patch: SsePatch = {
      patchElements: (e, o) =>
        emitted.push(`elements ${e}${o ? " " + JSON.stringify(o) : ""}`),
      patchSignals: (s) => emitted.push(`signals ${s}`),
      removeElements: (sel) => emitted.push(`remove ${sel}`),
      removeSignals: (k) => emitted.push(`rmsignals ${k}`),
      executeScript: (s) => emitted.push(`script ${s}`),
    };
    open = Promise.resolve(onStart(patch));
    return new Response("fake-stream", { headers: { "x-fake": "1" } });
  },
  async readSignals(req) {
    return { success: true, signals: await req.json() };
  },
};

/**
 * Wrap an SSE layer so the trace can see both directions of a request.
 *
 * The adapter is the right place for this because the adapter opened the
 * stream: it can instrument the patch object on the way in, and it still holds
 * an open stream after the handler resolves, which is the only moment something
 * can patch on the handler's behalf.
 */
export function tracingAdapter(
  sdk: SseAdapter,
  afterHandler: (patch: SsePatch) => void,
): SseAdapter {
  return {
    stream: (onStart, opts) =>
      sdk.stream(async (patch) => {
        const s = als.getStore();
        await onStart(s ? recording(patch, s.patches) : patch);
        flush(200);
        afterHandler(patch);
      }, opts),
    readSignals: async (req) => {
      const r = await sdk.readSignals(req);
      if (r.success) noteSignals(r.signals);
      return r;
    },
  };
}

/**
 * Run one request inside a trace scope.
 *
 * A refused request never reaches a handler, so nothing inside the stream can
 * record it — but the reason is already the status and body of the Response,
 * which is all a trace line needs.
 */
export async function traced(
  req: Request,
  run: () => Promise<Response | null>,
): Promise<Response | null> {
  const u = new URL(req.url);
  const id = u.pathname.startsWith("/command/")
    ? decodeURIComponent(u.pathname.slice("/command/".length))
    : u.pathname;

  return als.run(
    {
      id,
      url: u.pathname + u.search,
      lines: [],
      patches: [],
      signals: {},
      t0: performance.now(),
    },
    async () => {
      const res = await run();
      if (!res || !u.pathname.startsWith("/command/")) return res;

      const streamed = res.headers
        .get("content-type")
        ?.includes("event-stream");
      if (streamed) return res; // already flushed inside the stream
      // Signals-only: a non-stream response from /command/ is always a refusal,
      // because every path that reaches a handler answers over the stream.
      if (!id.startsWith("log.")) {
        store().setLastUrl(u.pathname + u.search);
        store().pushLog({
          level: "refusal",
          command: id,
          msg: await res.clone().text(),
          payload: { status: res.status },
          dur: `${(performance.now() - (als.getStore()?.t0 ?? 0)).toFixed(1)}ms`,
          status: res.status,
        });
      }
      return res;
    },
  );
}

/**
 * Behind a reverse proxy the app sees `http://localhost:3000` while the browser
 * sends `Origin: https://app.example.com`, so a naive `origin === u.origin`
 * check rejects every request. Reconstruct the public origin from the forwarded
 * headers — but ONLY when TRUST_PROXY is set, because those headers are
 * attacker-controlled when nothing is in front of you, and trusting them by
 * default would hand anyone a way to spoof the origin (and the client IP).
 */
export const TRUST_PROXY = process.env.TRUST_PROXY === "1";

export function publicOrigin(req: Request, u: URL): string {
  if (!TRUST_PROXY) return u.origin;
  const host = req.headers.get("x-forwarded-host") ?? req.headers.get("host");
  const proto = req.headers.get("x-forwarded-proto") ?? u.protocol.slice(0, -1);
  return host ? `${proto}://${host}` : u.origin;
}

export const specFor = (fn: Command<any, any>) => specOf.get(fn) ?? {};

export const isSealed = (fn: Command<any, any>) =>
  Boolean(specOf.get(fn)?.sealed);

export const registry = () => new Map(byId);

// ── abuse controls ───────────────────────────────────────────────────────────
// The demo is public and unauthenticated, so the only identity available is the
// connection. A token bucket per IP bounds request rate; the per-session caps in
// store.ts bound what any one visitor can make the server hold.

const RATE = { burst: 40, perSec: 8, sweepMs: 60_000 };

const buckets = new Map<string, { tokens: number; at: number }>();

function allow(ip: string): boolean {
  const now = Date.now();
  const b = buckets.get(ip) ?? { tokens: RATE.burst, at: now };
  b.tokens = Math.min(
    RATE.burst,
    b.tokens + ((now - b.at) / 1000) * RATE.perSec,
  );
  b.at = now;
  if (b.tokens < 1) {
    buckets.set(ip, b);
    return false;
  }
  b.tokens -= 1;
  buckets.set(ip, b);
  return true;
}

const COOKIE = "sa_sid";

const readCookie = (req: Request) =>
  req.headers
    .get("cookie")
    ?.split(";")
    .map((c) => c.trim().split("="))
    .find(([k]) => k === COOKIE)?.[1];

const SECURITY_HEADERS = {
  "x-content-type-options": "nosniff",
  "referrer-policy": "same-origin",
  "x-frame-options": "DENY",
};