Reconnection and delivery guarantees

The response stream drops periodically even when everything is healthy. That is normal and expected, and a client that treats a disconnect as an error will spend most of its life in the error path.

What makes a drop cheap is that the stream can be resumed. Every frame carries an id:, and a client that keeps the last one it processed loses nothing to a reconnect. What remains is the bookkeeping a resume cannot do for you: a command whose settlement you never saw at all.

What you get, and what you do not

Every frame carries an id:. Reconnect with Last-Event-ID set to the id of the last frame you finished processing, and the stream replays what followed it before it goes live again:

• At most 500 frames per batch, in order. If you were further behind, nothing is dropped: the rest follows on the same connection, and a : catching up comment marks each batch that stopped at the cap.
• As far back as 90 days, which is how long events are kept. An id older than that resumes at the oldest event that remains, and the stream opens with a : replay window exceeded comment so you can tell it from a quiet period.
• Possibly one frame twice. The frame you were reading when the socket went is sent again. Apply frames by id and a repeat is a no-op.

Last-Event-ID catches you up after a drop. To read a thread's past instead — what was said before you were streaming — use GET /v1/conversations/{conversationId}/messages.

Also true, and easy to get wrong:

• A frame is only what Mapier recorded. The stream is served from Mapier's own stored events, so something a transport never reported is not on it — a receipt after iMessage's window, for one.
• An error frame is the end. It says your key or your project stopped working, and reconnecting will not help. A close with no error frame is ordinary.
• Unknown event: names are ignored. The catalog is additive. Closing the socket on a frame you do not recognise is the bug.

The reconnect loop

The outer loop is unconditional: connect with the last id you processed, consume until the socket closes, pause, connect again. The resume is a header on the stream, not a separate request.

const BASE = "https://api.mapier.ai";
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

const MIN_DELAY = 6_000; // 60s ÷ 10 connects
const MAX_DELAY = 30_000;

type Cursor = {
  load(): Promise<string | undefined>;
  save(id: string): Promise<void>;
};

export async function runStream(
  onEvent: (name: string, data: unknown) => Promise<void>,
  cursor: Cursor
) {
  let backoff = MIN_DELAY;

  for (;;) {
    const startedAt = Date.now();
    const lastEventId = await cursor.load();

    try {
      const res = await fetch(`${BASE}/v1/response_stream?events=receipts,delivery_failures`, {
        headers: {
          Authorization: `Bearer ${process.env.MAPIER_API_KEY}`,
          Accept: "text/event-stream",
          ...(lastEventId ? { "Last-Event-ID": lastEventId } : {}),
        },
      });

      if (res.status === 429) {
        await res.body?.cancel();
        const retryAfter = Number(res.headers.get("retry-after")) || 1;
        await sleep(Math.max(retryAfter * 1_000, MIN_DELAY));
        continue;
      }
      if (!res.ok || !res.body) {
        await res.body?.cancel();
        throw new Error(`stream status ${res.status}`);
      }

      await consume(res.body, async (id, name, data) => {
        await onEvent(name, data);
        if (id) await cursor.save(id); // after the frame is handled, never before
      });
    } catch (error) {
      console.warn("response stream dropped", error);
    }

    backoff = Date.now() - startedAt > 30_000 ? MIN_DELAY : Math.min(backoff * 2, MAX_DELAY);
    await sleep(backoff + Math.random() * backoff);
  }
}

Save the id after you have handled its frame. Saved before, a crash between the two skips a frame; saved after, the worst case is handling one twice, which applying by id makes harmless. Keep the cursor somewhere durable: a cursor in memory resumes nothing after a restart.

Three things keep this inside the budget. The floor is the arithmetic: 10 connects a minute is one every six seconds. The doubling handles a stream that fails instantly. The jitter spreads workers that dropped together.

The loop above backs off on every non-429 failure, which is the right default, but the status tells you whether waiting will help. A 401 (invalid_credentials) is your key and will not fix itself. A 409 (project_archived) will not either, and neither will a 422, which names the query parameter it refused. A 503 is transient: upstream_unavailable, or service_busy, which says how long to wait in details.retryAfter.

The budget — 10 connects per minute — is counted per key. Several workers on one key share it; give each worker its own key, or one worker the stream. See Rate limits.

Parsing the stream correctly

The part that bites is buffering. reader.read() gives you whatever bytes arrived, and a chunk boundary can land anywhere: between two events, mid-line, or halfway through a JSON payload. A parser that treats each chunk as a whole message works perfectly in development and corrupts data under load.

The rules this implements are the SSE ones: accumulate into a buffer, split off only complete lines, ignore comment lines beginning with :, collect id:, event: and data: fields, and dispatch when a blank line arrives. Unknown event: names are handed to you so you can ignore them, and each frame's id is handed over with it so the loop above can save it once the frame is handled.

export async function consume(
  body: ReadableStream<Uint8Array>,
  onEvent: (id: string | undefined, name: string, data: unknown) => Promise<void>
) {
  const reader = body.getReader();
  const decoder = new TextDecoder();

  let buffer = "";
  let eventId: string | undefined;
  let eventName = "message"; // the SSE default, overridden by every Mapier frame
  let dataLines: string[] = [];

  for (;;) {
    const { value, done } = await reader.read();
    if (done) return; // a partial event at the end is incomplete: discard it

    buffer += decoder.decode(value, { stream: true });

    let newline: number;
    while ((newline = buffer.indexOf("\n")) !== -1) {
      const line = buffer.slice(0, newline).replace(/\r$/, "");
      buffer = buffer.slice(newline + 1);

      if (line === "") {
        if (dataLines.length > 0) await dispatch(eventId, eventName, dataLines.join("\n"), onEvent);
        eventId = undefined;
        eventName = "message";
        dataLines = [];
        continue;
      }

      if (line.startsWith(":")) continue;

      const colon = line.indexOf(":");
      const field = colon === -1 ? line : line.slice(0, colon);
      let fieldValue = colon === -1 ? "" : line.slice(colon + 1);
      if (fieldValue.startsWith(" ")) fieldValue = fieldValue.slice(1);

      if (field === "id") eventId = fieldValue;
      else if (field === "event") eventName = fieldValue;
      else if (field === "data") dataLines.push(fieldValue);
      // "retry" is the reconnect hint; the loop above keeps its own delay.
    }
  }
}

async function dispatch(
  id: string | undefined,
  name: string,
  raw: string,
  onEvent: (id: string | undefined, name: string, data: unknown) => Promise<void>
) {
  let data: unknown;
  try {
    data = JSON.parse(raw);
  } catch {
    return; // one malformed frame must not kill the connection
  }
  await onEvent(id, name, data);
}

Wiring the two together, with a handler that routes by event name:

await runStream(async (name, data) => {
  if (name === "command") await onCommandFrame(data as CommandFrame);
  else if (name === "message") await autoReply(data as InboundMessage);
  else if (name === "receipt") await onReceipt(data as ReceiptFrame);
  else if (name === "delivery_failed") await onDeliveryFailed(data as DeliveryFailedFrame);
  else if (name === "error") await stopAndAlert(data as ErrorFrame);
  // Unrecognised names are ignored rather than thrown on.
}, cursor);

Designing for the settlement you never see

A resume replays every frame Mapier recorded after your cursor, so a drop on its own loses nothing. What it cannot give you is a settlement that has not happened yet — or one that never will, because the command is still waiting on a transport that has gone quiet.

const rows = await db.unsettledCommandsOlderThan(6 * 60_000);
for (const row of rows) await markOutcomeUnknown(row.commandId);

The shape to avoid is the one that reads naturally: send, wait for the settlement, and assume anything still waiting is still working. A process that never sweeps is one whose work silently stalls.

Forward compatibility

The errorCode list is not closed, and a frame can carry keys the reference does not list. Two habits keep a client working across changes.

Parse tolerantly. Read the fields you need by name. Avoid exhaustive destructuring, strict schema validation that rejects unknown keys, or anything that treats an extra field as an error.

Give every branch a default. A value you did not anticipate should land somewhere useful rather than falling off the end of a switch and being treated as success:

switch (event.status) {
  case "succeeded":
    markSettled(event.commandId); // settled, not delivered
    break;

  case "failed":
  case "expired":
  case "cancelled":
    markNotDone(event.commandId, event.failureClass, event.errorCode);
    break;

  case "ambiguous":
  case "no_effect":
    markOutcomeUnknown(event.commandId, event.errorCode);
    break;

  default:
    markOutcomeUnknown(event.commandId, event.errorCode);
}

The same applies to failureClass — branch on it, and give it a default too — to errorCode and to event: names. Match the ones you handle and let everything else fall through.

Multiple streams are a broadcast

You may hold several connections open at once — four per key — and every connection receives every frame its filters admit. This is a fan-out, not a work queue: opening a second stream does not split the load, it duplicates it.

Two workers reading two streams both see the same settlement and the same inbound message. If that matters — and for anything that sends a reply, it does — deduplicate on the frame's id, which is the same on every stream that carries it. Derive your Idempotency-Key from the frame that triggered the work so a duplicate is adopted rather than executed twice.

Related