Provider recipes

import { createHash } from "node:crypto";
import { webhook } from "@use-crux/core/signal/transport";
import { z } from "zod";

const MAX_BYTES = 64 * 1024;

const ProviderBodySchema = z.object({
  accountId: z.string().min(1),
  eventId: z.string().min(1),
  topic: z.string().min(1),
});

/** Reject oversized trusted Content-Length before any body read. */
function assertContentLengthBudget(request: Request, maxBytes: number): void {
  const header = request.headers.get("content-length");
  if (header === null) {
    return;
  }

  const length = Number(header);
  if (!Number.isFinite(length) || length < 0 || length > maxBytes) {
    throw new Error("payload too large");
  }
}

/**
 * Single bounded accumulation from the request stream.
 * Aborts as soon as total bytes would exceed `maxBytes`.
 */
async function readBoundedBody(
  request: Request,
  maxBytes: number,
): Promise<Uint8Array> {
  if (!request.body) {
    return new Uint8Array(0);
  }

  const reader = request.body.getReader();
  const chunks: Uint8Array[] = [];
  let total = 0;

  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) {
        break;
      }
      if (!value || value.byteLength === 0) {
        continue;
      }

      total += value.byteLength;
      if (total > maxBytes) {
        throw new Error("payload too large");
      }

      chunks.push(value);
    }
  } finally {
    reader.releaseLock();
  }

  const raw = new Uint8Array(total);
  let offset = 0;
  for (const chunk of chunks) {
    raw.set(chunk, offset);
    offset += chunk.byteLength;
  }
  return raw;
}

function sha256Hex(bytes: Uint8Array): string {
  return createHash("sha256").update(bytes).digest("hex");
}

export const ingress = webhook({
  async handle(request) {
    assertContentLengthBudget(request, MAX_BYTES);

    const raw = await readBoundedBody(request, MAX_BYTES);
    const signature = request.headers.get("x-signature");
    if (!verify(signature, raw)) {
      throw new Error("unauthenticated");
    }

    const body = ProviderBodySchema.parse(
      JSON.parse(new TextDecoder().decode(raw)),
    );

    return {
      accountId: body.accountId,
      eventId: body.eventId,
      authenticatedRouting: { source: "webhook", topic: body.topic },
      payload: {
        kind: "inline-base64url",
        value: Buffer.from(raw).toString("base64url"),
        byteLength: raw.byteLength,
        sha256: sha256Hex(raw),
      },
    };
  },
});

import {
  acceptTransportEnvelope,
  TransportEnvelopeConflictError,
  type RuntimeAcceptedTransportEnvelope,
} from "@use-crux/core/runtime";

export async function postWebhook(request: Request, store: RuntimeStore) {
  const handled = await ingress.handle(request);
  const envelope: RuntimeAcceptedTransportEnvelope = {
    _tag: "RuntimeAcceptedTransportEnvelope",
    schemaVersion: 1,
    bindingId: "binding.orders",
    adapterId: "orders.webhook",
    provider: "orders",
    accountId: handled.accountId,
    eventId: handled.eventId,
    receivedAt: new Date().toISOString(),
    authenticatedRouting: handled.authenticatedRouting,
    payload: handled.payload,
    configRef: { id: "config.orders", revision: "1" },
    target: { kind: "signal", signalId: "order.submitted" },
  };

  try {
    // Accepted and duplicate both resolve with acknowledge: true.
    await acceptTransportEnvelope({
      store,
      namespace: "prod",
      envelope,
    });
    return new Response("ok", { status: 200 });
  } catch (error) {
    if (error instanceof TransportEnvelopeConflictError) {
      // Conflict throws; do not acknowledge as success.
      return new Response("conflict", { status: 409 });
    }

    throw error;
  }
}

import {
  createTransportNormalizationRunner,
  replayTransportEnvelope,
  transportBindingHealth,
  transportStatistics,
  projectTransportEnvelope,
} from "@use-crux/core/runtime";

const runner = createTransportNormalizationRunner({
  store,
  namespace: "prod",
  providers: [orders],
});

// Runtime maintenance / worker cadence normally calls this.
await runner.runOnce();

const stats = await transportStatistics({ store, namespace: "prod" });
console.log(stats.total.retried, stats.total.deadLettered);

// Restart-safe binding health: identity, status, last owner, cursor age, counts.
// `program` is the immutable RuntimeProgram that declares managed bindings.
const health = await transportBindingHealth({
  store,
  namespace: "prod",
  program, // createRuntimeProgram({ providers, transports })
});
console.log(health.bindings[0]?.status, health.bindings[0]?.cursor.ageMs);

// Operators requeue dead letters explicitly — workers never invent replay.
const revived = await replayTransportEnvelope({
  store,
  namespace: "prod",
  provider: "orders",
  accountId: "acct_1",
  eventId: "evt_poison",
});

const view = projectTransportEnvelope(revived);
// view.lineage is occurrence identities only; no payload bytes.

import { stream } from "@use-crux/core/signal/transport";
import { ManagedStreamTerminalError } from "@use-crux/core/runtime";
import {
  managedTransportBinding,
  signalProvider,
} from "@use-crux/core/signal/provider";

export const marketplaceStream = signalProvider({
  id: "marketplace.stream",
  transport: stream({
    async *open({ cursor, signal, configRef }) {
      const session = await openMarketplaceSession({
        resumeFrom: cursor,
        revision: configRef.revision,
        signal,
      });

      try {
        for await (const frame of session.frames) {
          if (signal.aborted) {
            break;
          }

          if (frame.kind === "heartbeat") {
            // Cursor-only progress: Runtime may checkpoint immediately.
            yield { kind: "cursor", cursor: frame.cursor };
            continue;
          }

          if (frame.kind === "auth_revoked") {
            // Terminal: durable faulted status, no automatic reconnect.
            throw new ManagedStreamTerminalError(
              "MARKETPLACE_AUTH_REVOKED",
              "provider revoked credentials",
            );
          }

          yield {
            kind: "envelope",
            accountId: frame.accountId,
            eventId: frame.eventId,
            authenticatedRouting: { source: "stream", topic: frame.topic },
            payload: encodeInlinePayload(frame.body),
            // Checkpoint only after this envelope is durably accepted.
            cursor: frame.cursor,
          };
        }
      } finally {
        await session.close();
      }
    },
  }),
  signals: { orderSubmitted },
  async onEvent(envelope, { signals }) {
    await signals.orderSubmitted.publish({
      orderId: decodeOrderId(envelope.payload),
    });
  },
});

export const marketplaceStreamBinding = managedTransportBinding(
  marketplaceStream,
  {
    id: "binding.marketplace.stream",
    configRef: { id: "config.marketplace.stream", revision: "1" },
    signalId: "order.submitted",
  },
);

import {
  sse,
  classifySseHttpStatus,
  sseHttpStatusErrorCode,
} from "@use-crux/core/signal/transport";
import { ManagedStreamTerminalError } from "@use-crux/core/runtime";
import {
  managedTransportBinding,
  signalProvider,
} from "@use-crux/core/signal/provider";

export const marketplaceSse = signalProvider({
  id: "marketplace.sse",
  transport: sse({
    async *open({ cursor, signal, configRef }) {
      // Map the durable Runtime cursor to the HTTP Last-Event-ID header.
      const response = await connectMarketplaceSse({
        lastEventId: cursor,
        revision: configRef.revision,
        signal,
      });

      if (!response.ok) {
        // Release an unused body when present; null bodies are a no-op.
        await response.body?.cancel?.();
        if (classifySseHttpStatus(response.status) === "terminal") {
          throw new ManagedStreamTerminalError(
            sseHttpStatusErrorCode(response.status),
            `SSE connect rejected with HTTP ${response.status}`,
          );
        }
        throw new Error(`SSE connect transient HTTP ${response.status}`);
      }

      try {
        for await (const frame of parseSse(response.body, signal)) {
          if (signal.aborted) {
            break;
          }

          if (frame.kind === "comment" && frame.id) {
            // Cursor-only: genuine new Last-Event-ID without an envelope.
            yield { kind: "cursor", lastEventId: frame.id };
            continue;
          }

          if (frame.kind === "event") {
            yield {
              kind: "envelope",
              accountId: frame.accountId,
              eventId: frame.eventId,
              authenticatedRouting: {
                source: "sse",
                eventType: frame.eventType ?? "message",
              },
              payload: encodeInlinePayload(frame.body),
              // Progress through this event inclusive (wire id:).
              lastEventId: frame.id ?? undefined,
            };
          }
        }
      } finally {
        await response.body?.cancel?.();
      }
    },
  }),
  signals: { orderSubmitted },
  async onEvent(envelope, { signals }) {
    await signals.orderSubmitted.publish({
      orderId: decodeOrderId(envelope.payload),
    });
  },
});

export const marketplaceSseBinding = managedTransportBinding(marketplaceSse, {
  id: "binding.marketplace.sse",
  configRef: { id: "config.marketplace.sse", revision: "1" },
  signalId: "order.submitted",
});

import {
  websocket,
  createBoundedPushBuffer,
  classifyWebSocketCloseCode,
  webSocketCloseErrorCode,
} from "@use-crux/core/signal/transport";
import { ManagedStreamTerminalError } from "@use-crux/core/runtime";
import {
  managedTransportBinding,
  signalProvider,
} from "@use-crux/core/signal/provider";

export const marketplaceWs = signalProvider({
  id: "marketplace.ws",
  transport: websocket({
    async *open({ cursor, signal, configRef }) {
      const socket = await connectMarketplaceWs({
        resume: cursor,
        revision: configRef.revision,
        signal,
      });
      const buffer = createBoundedPushBuffer({ capacity: 64, signal });

      let finalized = false;
      const finalize = () => {
        if (finalized) {
          return;
        }
        finalized = true;
        signal.removeEventListener("abort", onAbort);
        socket.close();
      };
      const onAbort = () => {
        finalize();
      };

      socket.onmessage = (event) => {
        try {
          const message = mapMarketplaceFrame(event);
          if (message.kind === "heartbeat") {
            buffer.push({ kind: "cursor", cursor: message.cursor });
            return;
          }

          buffer.push({
            kind: "envelope",
            accountId: message.accountId,
            eventId: message.eventId,
            authenticatedRouting: { source: "websocket" },
            payload: encodeInlinePayload(message.body),
            cursor: message.cursor,
            // Optional: Runtime calls this only after durable accept + checkpoint.
            acknowledge: () => socket.sendAck(message.wireId),
          });
        } catch (error) {
          // Fail first so a later clean onclose cannot mask parse/overflow faults.
          buffer.fail(error);
          finalize();
        }
      };

      socket.onclose = (closeEvent) => {
        const kind = classifyWebSocketCloseCode(closeEvent.code);
        if (kind === "normal") {
          // close() is a no-op after fail(), so prior failure stays authoritative.
          buffer.close();
          return;
        }
        if (kind === "terminal") {
          buffer.fail(
            new ManagedStreamTerminalError(
              webSocketCloseErrorCode(closeEvent.code),
              `WebSocket closed with code ${closeEvent.code}`,
            ),
          );
          return;
        }
        buffer.fail(new Error(`WebSocket closed with code ${closeEvent.code}`));
      };

      signal.addEventListener("abort", onAbort);
      try {
        yield* buffer.items;
      } finally {
        // Early return, clean EOF, and throw all close the socket and detach abort.
        finalize();
      }
    },
  }),
  signals: { orderSubmitted },
  async onEvent(envelope, { signals }) {
    await signals.orderSubmitted.publish({
      orderId: decodeOrderId(envelope.payload),
    });
  },
});

export const marketplaceWsBinding = managedTransportBinding(marketplaceWs, {
  id: "binding.marketplace.ws",
  configRef: { id: "config.marketplace.ws", revision: "1" },
  signalId: "order.submitted",
});