Twitch Ingress

The Twitch Ingress holds the WebSocket shards of a single Twitch EventSub Conduit, keeps them alive across resets, filters incoming payloads, and pushes the survivors onto the NATS bus as normalized events. It is the only service in the system written in Elixir on the BEAM VM; the rest is Go.

The language choice is justified in ADR 0006. The communication substrate (NATS) it sits behind is justified in ADR 0003. The hardware it runs on is described in ADR 0004.

Responsibilities

• Maintain a Twitch EventSub Conduit comprising a small fleet of WebSocket shards. Twitch routes events across the shards on its side; we only have to keep them connected.
• Keep each shard’s WebSocket alive: handle the session_welcome / reconnect dance, reconnect with backoff on close or reset, restart a single shard in isolation when it fails.
• Hold the Twitch app access token used for Helix calls (Conduit management and shard binding), fetching a new one shortly before it expires.
• Filter incoming Twitch payloads and publish the survivors to NATS as normalized events. Chat messages have exactly three outcomes: messages from the special user IDs (secret-store list) always go to the premium lane; messages starting with ! go to the lane matching the broadcaster’s status (premium or standard); plain chat publishes its first line at once and folds identical repeats into one event that lists their senders (squash). A banned broadcaster’s events are dropped.
• Resolve broadcaster status over NATS request-reply from the service that owns broadcaster data (the ingress never reads MySQL directly), behind an in-process read-through cache evicted by invalidation keys on NATS.
• Distribute shard ownership across ingress replicas in the BEAM cluster so that exactly one node owns each shard’s WSS at any time, with re-assignment in seconds when a node leaves.

What this service does not do: process commands of any kind (chat commands or operator commands), persistence, business logic, rate-limit accounting. Anything that needs to react to events lives in Go services downstream and subscribes to the NATS subjects this service publishes.

External shape

flowchart LR
    subgraph Ingress [Twitch Ingress - Elixir / BEAM cluster]
        direction TB
        N1[ingress node A]
        N2[ingress node B]
        N1 <-. Erlang distribution<br/>between pods .-> N2
    end

    NATS[(NATS<br/>core + JetStream)]

    N1 -- publish events --> NATS
    N2 -- publish events --> NATS

Ingress nodes form a single BEAM cluster over Erlang distribution. No public port is exposed. The only thing leaving the cluster is published NATS messages.

Internal shape (OTP supervision tree)

flowchart TB
    App[BagelBot.Ingress.Application<br/>top-level supervisor]

    Cluster[Cluster.Supervisor<br/>libcluster - node discovery]
    Registry[Horde.Registry<br/>shard_id to PID, cluster-wide]
    DynSup[Horde.DynamicSupervisor<br/>spawns and supervises shards]
    NatsConn[Gnat.ConnectionSupervisor<br/>NATS publisher]
    Token[Twitch.AppToken<br/>app access token]

    App --> Cluster
    App --> Registry
    App --> DynSup
    App --> NatsConn
    App --> Token

    DynSup --> Sh1[ShardSession - shard 1<br/>GenServer]
    DynSup --> Sh2[ShardSession - shard 2<br/>GenServer]
    DynSup --> ShN[ShardSession - shard N<br/>GenServer]

    Sh1 -. holds .-> WS1[EventSub WSS shard]
    Sh2 -. holds .-> WS2[EventSub WSS shard]
    ShN -. holds .-> WSN[EventSub WSS shard]

• BagelBot.Ingress.Application is the top-level supervisor. Restart strategy: :one_for_one.
• Cluster.Supervisor (libcluster) discovers peer nodes and triggers Node.connect/1 against them. The strategy follows the environment: Kubernetes DNS when BAGELBOT_K8S_HEADLESS_SERVICE is set, EPMD with BAGELBOT_CLUSTER_HOSTS, otherwise Gossip.
• Horde.Registry is a CRDT-backed, cluster-wide registry. Each shard is registered under a key like {:shard, shard_id}. Cross-node process lookup is a Horde.Registry.lookup/2 call.
• Horde.DynamicSupervisor spawns shards and re-assigns them to surviving nodes on node loss.
• ShardSession is a GenServer per shard. It owns one WebSocket, drives the EventSub protocol state machine, runs the filter on each incoming payload, and publishes survivors to NATS.
• Gnat.ConnectionSupervisor owns the NATS connections (Gnat is the Elixir NATS client). The event bus and the RPC plane use separate connections and identities.
• Twitch.AppToken is a single process holding the app access token. It fetches a new token when the cached one is within a minute of expiry.

The filter itself is a pure function module (no state, no supervisor child). Each ShardSession calls it inline before deciding whether to publish.

Restart strategies

Reconnect backoff inside a shard uses jittered exponential backoff capped at 60 seconds.

Sharding and ownership

Twitch handles the routing of events across the Conduit’s shards on its side; we do not pick which channel goes where. What we own is keeping each shard’s WebSocket connected and re-assigning that ownership across BEAM nodes when the cluster changes.

The flow is:

1. At boot, the cluster reads the Conduit’s shard count and ensures one ShardSession per shard exists in the cluster, started via Horde.DynamicSupervisor.start_child and registered in Horde.Registry under {:shard, shard_id}.
2. When a node leaves the cluster, Horde re-assigns its shards to surviving nodes. The new owner re-opens the WebSocket and re-registers it with Twitch via the EventSub API.
3. When a node joins, Horde migrates shards to it per Ingress.ShardDistribution: round-robin by shard id across the alive nodes (sorted by name), so five shards on two nodes always split 3/2. Placement is deterministic, so active redistribution moves only the shards whose target actually changed.

We deliberately do not use NATS KV for this. The BEAM cluster’s registry is authoritative, in-memory, and updates synchronously across nodes. See ADR 0006 for the reasoning.

NATS contracts

The ingress publishes events and status, issues request-reply calls for broadcaster status, and subscribes to cache invalidation keys, data.users.changed (trial eviction), and its own admin RPC subjects.

Subject prefixes:

• twitch.ingress.event.<lane>: outbound, normalized events the ingress publishes after filtering. Exactly three subjects: premium / standard (laned by broadcaster status) and stream (only stream online/offline events); the EventSub type travels in the payload, not the subject.
• twitch.ingress.status.*: outbound, lifecycle and health signals.
• bagel.rpc.broadcaster.status.get: request-reply, broadcaster status lookups against the owning Go service.
• bagel.cache.invalidate.status: inbound, evicts entries from the in-process broadcaster status cache.
• twitch.ingress.admin.shards.get: inbound request-reply, answered by Ingress.AdminRpc with a cluster-wide shard state snapshot for the Admin console. A queue group ensures exactly one replica answers; the reply covers every shard via the Horde registry.
• twitch.ingress.admin.shards.scale, .shards.autoscale and twitch.ingress.admin.trials.* (list, add, remove, set_enabled): inbound request-reply for shard scaling and trial channels.
• bagel.rpc.ingress.conduit.get: inbound request-reply returning the live Conduit ID (used by Twitch Outgress). Health is answered on bagel.rpc.health.ingress.

Outbound: events

Chat notices and every other subscribed EventSub type travel on the premium/standard subjects; consumers dispatch on the payload’s type field. Payloads carry msg_id so consumers can de-duplicate Twitch redeliveries.

Outbound: status

Status subjects share the TWITCH_INGRESS JetStream stream with the premium and stream lanes, which keeps messages for 10 seconds.

Configuration

Environment-driven. All values arrive as env vars and are read once at boot.

Deployment

• Replicas: 3, spread one per host (topologySpreadConstraints on kubernetes.io/hostname) and kept off the control-plane and worker-role nodes (deploy/k8s/twitch-ingress.yaml).
• Runtime: Elixir 1.17+ (mix.exs), built with Elixir 1.20 on OTP 28. Released as a Mix release (mix release) and run on a debian:trixie-slim base image.
• Process model: one container per node. The BEAM is the concurrency runtime; we do not run multiple BEAM VMs per pod.
• Networking: Erlang distribution is cluster-internal. No public port. NATS is reached over the cluster network.
• Health checks: the Kubernetes liveness and readiness probes run /app/bin/ingress pid. A separate HTTP listener (STATUS_PORT, default 8080) serves /healthz, /readyz and /status.
• Rolling updates: one node at a time. SIGTERM triggers :init.stop/0, which runs Application stop callbacks and lets Horde drain shards to surviving nodes before exit.

Observability

• Logs: Logger to stdout with shard_id and node metadata; LOG_LEVEL sets the level.
• Metrics: in-process counters are flushed every second to New Relic as Custom/Ingress/* metrics, and notable shard events are reported as IngressEvent custom events. Without NEW_RELIC_LICENSE_KEY the agent is a no-op.
• Traces: a sampled share of notifications (INGRESS_TRACE_SAMPLE_RATE) runs as a New Relic Ingress transaction, with the dispatcher queue wait recorded as an attribute.

Failure modes and how the service responds

References

• ADR 0001: the rewrite to microservices, originally surfaced the Twitch zombie-socket problem this service exists to solve.
• ADR 0003: NATS as the communication substrate; defines the subject space this service publishes into.
• ADR 0004: the fleet this service runs on.
• ADR 0006: the language and runtime choice for this service.