Publishing events

A channel or connecting consumer handler publishes with the shortcut on its context:

import type { UpdatesResponseHandler } from "../.embabel/realm.d.ts";

export const updates: UpdatesResponseHandler = async (body, ctx) => {
  if (typeof body !== "string") return {};
  const batch = JSON.parse(body);
  let offset: number | undefined;
  for (const update of batch.result) {
    await ctx.publish(
      "messages",
      { text: update.message.text, chatId: update.message.chat.id },
      String(update.update_id),
      { streamId: String(update.message.chat.id) },
    );
    offset = update.update_id + 1;
  }
  return offset === undefined ? {} : { cursor: String(offset), cursorParam: "offset" };
};

ctx.publish(source, event, key, options) files event, a JSON object, under one of the realm’s own sources and resolves to a Receipt, {id, offset, replayed}, where id is the journal receipt’s id. key is the event id, and options may carry streamId and occurredAt. The journal matches a retry on source, key, stream and payload, so a frame the host redelivers gets the first receipt back (replayed: true) and keeps the first occurredAt. The same event id with a different payload is refused. Only the names in the realm’s sources block compile as source.

Any handler in a captured realm can publish through ctx.gateway.channel, which the generated types type with the receipts the host returns:

import type { RecordHandler } from "../.embabel/realm.d.ts";

export const record: RecordHandler = async (_input, ctx) => {
  const { receiptId, offset, replayed } = await ctx.gateway.channel.publish({
    source: "changes",
    eventId: "note-42-v3",
    streamId: "note-42",
    occurredAt: new Date().toISOString(),
    payload: { noteId: "note-42" },
  });
  ctx.log(`filed ${receiptId} at ${offset}${replayed ? " (replayed)" : ""}`);
  return {};
};

Call	Resolves to
publish({source, eventId, streamId, occurredAt, payload})	ChannelPublishReceipt: {receiptId, offset, replayed}
publishBatch({source, batchId, expectedPosition, nextPosition, events})	ChannelBatchReceipt: {receiptId, offsets, replayed}, one offset per event in order
position({source})	{position}, a string, or null before the first batch

A poller commits a page and its cursor together with position and publishBatch; each event in events is {eventId, streamId, occurredAt, payload}. The receipt from ctx.gateway.channel names its id receiptId, and the ctx.publish shortcut names it id. The limits and retry windows are under publication and polling positions.