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.
Refresh per-tenant OAuth tokens before expiry and re-authenticate when tokens roll over.
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); everything else is 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/>over Tailscale .-> 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. The distribution port is reachable only inside the tailnet from ADR 0004; 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[TokenRefresher<br/>per-tenant OAuth]

    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. EPMD strategy in production, with the peer list passed via config.
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 connection pool (Gnat is the Elixir NATS client). The ingress only publishes; it does not subscribe.
TokenRefresher is one process per tenant. It holds the current OAuth refresh token, schedules a refresh ahead of expiry, and notifies shards when a new access token is available.

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

Process	Strategy	Notes
`Application`	`:one_for_one`	A subsystem crash does not take down siblings.
`Horde.DynamicSupervisor`	`:one_for_one`	One shard crashing only restarts that shard.
`ShardSession`	`:transient`	Normal shutdown (shard removed) does not restart; crash does, with backoff.
`TokenRefresher`	`:permanent`	Always restart; without tokens the tenant cannot work.
`Gnat.ConnectionSupervisor`	`:permanent`	Always restart; ingress is useless without NATS.

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:

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}.
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.
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 one subject: cache invalidation keys.

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.broadcaster: 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 tool. A queue group ensures exactly one replica answers; the reply covers every shard via the Horde registry.

Outbound: events

Subject	Payload (JSON)	Notes
`twitch.ingress.event.premium`	chat: `{type, lane, broadcaster_user_id, chatter_user_id, chatter_user_login, text, ts, msg_id}`; non-chat: EventSub payload as delivered by Twitch, plus `{type, lane, received_at, shard_id, msg_id}`	Premium broadcasters + special IDs.
`twitch.ingress.event.standard`	same shapes as premium	Standard broadcasters, and events with no extractable broadcaster.
`twitch.ingress.event.stream`	EventSub payload plus `{type, lane, received_at, shard_id, msg_id}`	Only `stream.online` / `stream.offline`, regardless of broadcaster status.

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

Subject	Payload
`twitch.ingress.status.shard.up`	`{tenant, shard_id, node, since}`
`twitch.ingress.status.shard.bound`	`{shard_id, node, session_id, kind, at}` — `kind` is `fresh` (full connect + Helix bind) or `moved` (session_reconnect handshake completed)
`twitch.ingress.status.shard.down`	`{tenant, shard_id, node, reason}` — `reason` includes `duplicate_resolved` when a netsplit-heal duplicate stands down
`twitch.ingress.status.node.joined`	`{node, version, ts}`
`twitch.ingress.status.node.left`	`{node, reason, ts}`

Status is a JetStream stream with short retention (10 minutes is enough for any observer to catch up).

Configuration

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

Variable	Purpose	Example
`BAGELBOT_NODE_NAME`	Erlang long-name of this node.	`[email protected]`
`BAGELBOT_ERLANG_COOKIE`	Shared cookie for distribution. Provided by the secret store.	(opaque)
`BAGELBOT_CLUSTER_HOSTS`	Comma-separated peer long-names for libcluster EPMD strategy.	`[email protected],[email protected]`
`NATS_URL`	NATS connection URL inside the VCN.	`nats://nats.internal:4222`
`NATS_CREDS_FILE`	Path to the NATS credentials file (JWT + nkey).	`/run/secrets/nats.creds`
`TWITCH_CLIENT_ID`	App client ID for Twitch API calls.	(opaque)
`TWITCH_CLIENT_SECRET`	App client secret.	(opaque)
`TWITCH_CONDUIT_ID`	The Conduit this ingress owns.	`conduit_abc123`
`TWITCH_CONDUIT_SHARD_COUNT`	Desired number of shards in the Conduit.	`4`
`TWITCH_EVENTSUB_WSS_URL`	Twitch EventSub WebSocket endpoint. Pinned via config.	`wss://eventsub.wss.twitch.tv/ws`
`TWITCH_SPECIAL_USER_IDS`	Chatter IDs that always route to the premium lane. From the secret store.	`1001,1002`
`NATS_SUBJECT_LANE_PREMIUM`	Premium lane subject (all event types).	`twitch.ingress.event.premium`
`NATS_SUBJECT_LANE_STANDARD`	Standard lane subject (all event types).	`twitch.ingress.event.standard`
`NATS_SUBJECT_LANE_STREAM`	Dedicated lane for stream.online / stream.offline events only.	`twitch.ingress.event.stream`
`NEW_RELIC_LICENSE_KEY`	Enables New Relic monitoring; absent, the agent is a no-op.	(opaque)
`NEW_RELIC_APP_NAME`	New Relic application name.	`itsbagelbot-twitch-ingress`
`NATS_BROADCASTER_STATUS_SUBJECT`	Request-reply subject for broadcaster status lookups.	`bagel.rpc.broadcaster.status.get`
`NATS_CACHE_INVALIDATION_SUBJECT`	Subject carrying broadcaster cache invalidation keys.	`bagel.cache.invalidate.broadcaster`
`NATS_ADMIN_SUBJECT`	Request-reply subject answered with the shard state snapshot.	`twitch.ingress.admin.shards.get`
`BROADCASTER_CACHE_TTL_SECONDS`	TTL of the in-process broadcaster status cache.	`300`
`LOG_LEVEL`	`debug` / `info` / `warn` / `error`.	`info`
`OTEL_EXPORTER_OTLP_ENDPOINT`	Optional OTLP target for traces and metrics.	`http://otel-collector:4318`

Per-tenant credentials (OAuth refresh tokens, EventSub subscription configuration) come from the relational database described in ADR 0005.

Deployment

Architecture: ARM and Intel. Scheduled across both node1 (ARM) and node2 (Intel/x86_64) for High Availability (HA). Images are built per node: ARM natively on the operator Mac, Intel natively on node2 — no cross-arch emulation in the pipeline.
Runtime: OTP 27+, Elixir 1.17+. Released as a Mix release (mix release), running on a distroless Erlang 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 binds to the Tailscale interface only. EPMD bound to the same address. No public port. NATS is reached over the internal VCN.
Health checks:
- Liveness: a small HTTP endpoint on localhost that confirms the application supervisor is alive.
- Readiness: Gnat.ping/1 to NATS plus Node.list/0 containing at least one peer when scaled beyond one node.
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: JSON to stdout, structured via Logger with metadata. Fields: tenant, shard_id, node, event.
Metrics: Telemetry events emitted by Gnat, Horde, and our own :telemetry.execute/3 calls. Exported via the OpenTelemetry SDK to the collector at OTEL_EXPORTER_OTLP_ENDPOINT.
Traces: Each shard’s payload-handling path opens a span; outbound NATS publishes carry the trace context as a header so downstream Go services see continuity.
Key metrics:
- ingress.shard.count{state} (gauge): live shards by state (connected, reconnecting).
- ingress.shard.reconnects_total (counter): reconnect attempts; sustained spikes flag a Twitch outage.
- ingress.filter.dropped_total{reason} (counter): events dropped by the filter, by reason.
- ingress.nats.publish_latency_ms{subject} (histogram).
- ingress.token.refreshes_total{tenant, outcome} (counter).
- ingress.cluster.members (gauge): nodes in Node.list/0.

Failure modes and how the service responds

Failure	What happens
A shard’s WebSocket drops	`ShardSession` traps the close, restarts via supervisor, reconnects with backoff. Sibling shards untouched.
Twitch returns 401 on a shard	Shard asks `TokenRefresher` for a fresh token; on success it re-authenticates, on failure it crashes and is restarted by the supervisor.
One ingress node dies	`Horde` re-assigns its shards to surviving nodes. New owners re-open WebSockets and re-register them with Twitch. Status events emitted.
NATS unreachable	`Gnat.ConnectionSupervisor` reconnects. Outbound events are buffered (bounded) and dropped on overflow with a warn-level log; we prefer drop over unbounded memory growth.
Erlang distribution partitions	Each side becomes its own quorum-less cluster and may run duplicate shards. On heal, the registry merge flags the duplicate, and resolution is health-based, not arbitrary: the copy that is bound to the Conduit keeps the shard (reclaiming the registration if the registry picked the unbound one), the redundant copy stands down gracefully and publishes `shard.down` with reason `duplicate_resolved`.
Single bad message format from Twitch	Shard crashes, supervisor restarts, Twitch will resend through the Conduit. No retry storm on bad payloads.

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, including the tailnet that BEAM distribution rides over.
ADR 0005: the database that holds per-tenant Twitch credentials.
ADR 0006: the language and runtime choice for this service.