Stream a mandate's events

Events are a running record of a mandate: every status change, action, approval, and finding, as it happens. This guide streams one mandate's events, keeps working through your own restarts, and handles duplicates and gaps.

Open the stream

stream.ts

import { Shadway, isKnownEvent } from "shadway";

const shadway = new Shadway();
const controller = new AbortController();

for await (const event of shadway
  .mandate(mandateId)
  .events.stream({ signal: controller.signal })) {
  if (!isKnownEvent(event)) continue;

  console.log(event.sequence, event.type);

  if (
    event.type === "mandate.status_changed" &&
    ["completed", "failed", "canceled", "expired"].includes(event.data.mandate.status)
  ) {
    break;
  }
}

stream.go

client, err := shadway.New()
if err != nil {
	log.Fatal(err)
}

for event, err := range client.Mandate(mandateID).Stream(ctx, shadway.EventStreamOptions{}) {
	if err != nil {
		log.Fatal(err)
	}
	if !event.IsKnown() {
		continue // A future event type this SDK version does not model yet.
	}

	fmt.Println(event.Sequence, event.Type)

	if data, ok := event.Data.(*shadway.MandateStatusChangedData); ok {
		switch data.Mandate.Status {
		case shadway.MandateStatusCompleted, shadway.MandateStatusFailed,
			shadway.MandateStatusCanceled, shadway.MandateStatusExpired:
			return
		}
	}
}

The stream reconnects and deduplicates by event ID on its own. Ending the stream stops observation only. It never affects the mandate. New event types appear over time, so always skip what you don't recognize instead of failing on it.

Handle specific event types

Each event type carries a typed payload. Narrow before you touch data.

narrow.ts

for await (const event of shadway.mandate(mandateId).events.stream({ signal })) {
  if (!isKnownEvent(event)) continue;

  switch (event.type) {
    case "mandate.status_changed":
      console.log(event.data.previousStatus, "->", event.data.mandate.status);
      break;
    case "mandate.approval_requested":
      console.log("needs a decision:", event.data.approval.question);
      break;
    case "capability.execution_failed":
      console.log("failed:", event.data.execution.error?.message);
      break;
  }
}

narrow.go

for event, err := range client.Mandate(mandateID).Stream(ctx, shadway.EventStreamOptions{}) {
	if err != nil {
		log.Fatal(err)
	}
	switch data := event.Data.(type) {
	case *shadway.MandateStatusChangedData:
		fmt.Println(data.PreviousStatus, "->", data.Mandate.Status)
	case *shadway.ApprovalEventData:
		fmt.Println("needs a decision:", data.Approval.Question)
	case *shadway.CapabilityExecutionEventData:
		if event.Type == shadway.EventTypeCapabilityExecutionFailed && data.Execution.Error != nil {
			fmt.Println("failed:", data.Execution.Error.Message)
		}
	}
}

Resume after a restart

Persist the last event ID you processed. A database row per mandate is enough. Pass it when you reconnect, and the stream replays everything after that point, so nothing that happened while you were down is skipped.

resume.ts

let lastEventId = await loadCheckpoint(mandateId); // Your storage.

for await (const event of shadway
  .mandate(mandateId)
  .events.stream({ signal, ...(lastEventId ? { lastEventId } : {}) })) {
  if (!isKnownEvent(event)) continue;

  await handle(event); // Process before checkpointing.
  await saveCheckpoint(mandateId, event.id);
  lastEventId = event.id;
}

resume.go

lastEventID := loadCheckpoint(mandateID) // Your storage.

for event, err := range client.Mandate(mandateID).Stream(ctx, shadway.EventStreamOptions{
	LastEventID: lastEventID,
}) {
	if err != nil {
		log.Fatal(err)
	}
	if !event.IsKnown() {
		continue
	}

	handle(event) // Process before checkpointing.
	saveCheckpoint(mandateID, event.ID)
}

Delivery is at least once: after a reconnect you may see an event you already processed. Make handling idempotent. Keying your side effects by event.id is the simplest way.

Event order and gaps

sequence goes up within a mandate and never repeats, so use it to put events in order and to notice one missing. It is not gap-free, so do not treat a skipped number as data loss by itself. When you need a complete picture, list the mandate's events, which return in ascending sequence order:

backfill.ts

for await (const event of shadway.mandate(mandateId).events.list()) {
  console.log(event.sequence, event.type);
}

There is also a workspace-wide stream (shadway.events / client.Events) that carries every mandate's events in occurrence order. Use it when one consumer feeds your own system. The per-mandate patterns apply there too, except that ordering across mandates has no sequence. Deduplicate by event ID and order events within each mandate by its sequence.

For delivery to your servers without holding a connection open, use webhooks. They carry the same event objects.

Continue building

Wait for a result

When you only need the ending, not the story.

Verify webhooks

The same events, pushed to your servers.