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 matters more is what the stream does not promise. Every guarantee listed below as missing is one a client can quietly assume it has, and the resulting bug — a settlement that never arrives, an outcome silently treated as success — surfaces in production rather than in testing.

What you do not get

This list is deliberate negative space. None of these are planned omissions being hidden; they are the current shape of the system, and building against them is the difference between a client that works and one that mostly works.

• No id: field, so no Last-Event-ID. SSE has a built-in resume mechanism. It is not implemented here, and the browser EventSource reconnect behaviour that depends on it gains you nothing.
• No replay, no backfill, no cursor. There is no sequence number, no timestamp parameter and no endpoint that returns events you missed. Anything that happened while you were disconnected is gone.
• At-most-once delivery. An event can be lost and is never redelivered. A settlement you never receive is indistinguishable from a command still in flight.
• No ordering guarantee. Nothing enforces the order events arrive in, and nothing in a payload lets you detect that one is missing.
• No delivery receipts. A succeeded settlement means iMessage accepted the send, not that it reached a device or was read.
• No typing or read events. Neither is exposed as an event or available as a command.

Two of these compound. Because delivery is at-most-once and there is no backfill, the stream cannot be your source of truth for whether work completed. It is a fast path to the answer, with your own bookkeeping underneath.

A reconnect loop

The outer loop is unconditional: connect, consume until the socket closes, pause, connect again. The only conditional part is how long you pause, and the 429 case.

const BASE = process.env.MAPIER_BASE_URL!;
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

export async function runStream(onEvent: (name: string, data: unknown) => void) {
  let backoff = 1_000;

  for (;;) {
    const startedAt = Date.now();

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

      if (res.status === 429) {
        // Connections are what the budget counts. Respect the header.
        // Cancel the body: an unconsumed response holds its socket out of the
        // pool until GC, so a tight 429 loop would leak one per attempt.
        await res.body?.cancel();
        const retryAfter = Number(res.headers.get("retry-after")) || 1;
        await sleep(retryAfter * 1_000);
        continue;
      }
      if (!res.ok || !res.body) {
        await res.body?.cancel();
        throw new Error(`stream status ${res.status}`);
      }

      await consume(res.body, onEvent); // returns when the socket closes
    } catch (error) {
      console.warn("response stream dropped", error);
    }

    // A stream that lived a while was healthy; one that died at once was not.
    backoff = Date.now() - startedAt > 30_000 ? 1_000 : Math.min(backoff * 2, 30_000);
    await sleep(backoff);
  }
}

The backoff distinction is what keeps this well-behaved. A long-lived stream that finally drops should reconnect straight away. A stream that fails instantly — a connector that has gone away, a bad credential — should not reconnect sixty times a minute.

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 (unauthenticated) is your credential and will not fix itself, so it is worth surfacing rather than retrying silently. A 404 (no_connector) means the account has no Mac connector to stream from — retryable, but only once someone brings the connector back. A 503 (stream_unavailable) is transient and is exactly what backoff is for. A 405 means the request was not a GET.

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 event: and data: fields, and dispatch when a blank line arrives.

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

  let buffer = "";
  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);

      // A blank line terminates an event.
      if (line === "") {
        if (dataLines.length > 0) dispatch(eventName, dataLines.join("\n"), onEvent);
        eventName = "message";
        dataLines = [];
        continue;
      }

      // ": connected" and ": ping" are comments, not events.
      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 === "event") eventName = fieldValue;
      else if (field === "data") dataLines.push(fieldValue);
      // Any other field, including a future "id", is ignored.
    }
  }
}

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

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

await runStream((name, data) => {
  if (name === "command") onCommandEvent(data as CommandEvent);
  else if (name === "message") autoReply(data as InboundMessage);
  // Unrecognised names are ignored rather than thrown on.
});

Designing for lost events

Because a settlement can never arrive at all, your client needs a record of what it queued that does not depend on the stream.

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. Under at-most-once delivery that state is permanent, and the work silently stalls.

Forward compatibility

The errorCode list is not closed, and a command frame already carries a field that is present only sometimes — conversationId, on a resolved group.create. 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":
  case "no_effect":
    markDone(event.commandId);
    break;

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

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

  default:
    // A status this code does not recognise. Not an error, and not a success —
    // the only honest handling is the same as ambiguous.
    markOutcomeUnknown(event.commandId, event.errorCode);
}

The same applies to errorCode. Match the codes you handle specially and let everything else fall through to a generic failure — an unrecognised code is never a reason to assume the command worked.

Multiple streams are a broadcast

You may hold several connections open at once, and every connection receives every event for the account. This is a fan-out, not a work queue: opening a second stream does not split the load, it duplicates it.

Two workers on one account both see the same settlement and the same inbound message. If that matters — and for anything that sends a reply, it does — deduplicate on commandId for settlements and on externalId for inbound messages, and derive your Idempotency-Key from the event that triggered the work so a duplicate is adopted rather than executed twice.

Related