Day 7: Phoenix channels from the client side

You wrote the server half of this. Today you read the other half. The app keeps one WebSocket open while you are signed in, joins waiting:<slug> while it looks for an opponent, and joins race:<id> for the race itself. Every push, reply and server event in backend/docs/channels.md has a matching line in app/lib/data/socket/, and by the end of the session you should be able to point to each one.

Before you start

Day 5 gave you authStatusProvider and ref.listen. Day 6 gave you AppConfig.socketUrl and the TokenStore that holds the JWT. The realtime structure page has the diagram of how these files depend on each other.

One new Dart idea today is the Stream. A Stream<T> is an asynchronous sequence of values, the same idea as a Kotlin Flow or an RxSwift Observable. Two flavours matter:

A single-subscription stream can be listened to once. Most file and HTTP streams are like this.
A broadcast stream can have many listeners, and a listener only sees events that arrive after it subscribed. It is a hot stream, like a Kotlin SharedFlow with no replay, or a Phoenix.PubSub topic.

You get one from a StreamController, which has add(event) on the producer side and .stream on the consumer side. You consume one with stream.listen(callback), which returns a StreamSubscription. You must cancel() that subscription when you are done. It plays the same role as a Job in a coroutine scope, or a Disposable in Rx. .where, .map, .cast and .first work like their Flow counterparts.

The seam: `SocketTransport` and `ChannelTransport`

Open app/lib/data/socket/channel_transport.dart. The first comment says why it exists: “The slice of Phoenix Channels the app uses, narrow enough to fake in tests.”

abstract class ChannelTransport {
  String get topic;
  Stream<ChannelMessage> get messages;
  Future<Map<String, dynamic>> push(String event, Map<String, dynamic> payload);
  Future<void> leave();
}

abstract class SocketTransport {
  SocketStatus get status;
  Stream<SocketStatus> get statusStream;
  Future<void> connect();
  Future<void> disconnect();
  Future<(ChannelTransport, Map<String, dynamic>)> join(
    String topic, {
    Map<String, dynamic>? params,
  });
}

That is lines 33-44 and 48-61. Line up each member with a Phoenix concept:

Client	Server
`SocketTransport.connect()`	`UserSocket.connect/3`
`join(topic)` returns the transport and the join reply	`join/3` returning `{:ok, reply, socket}`
`join` throws `ChannelException(reason)`	`join/3` returning `{:error, %{reason: ...}}`
`ChannelTransport.push(event, payload)` returns the `ok` body	`handle_in/3` returning `{:reply, {:ok, body}, socket}`
`push` throws `ChannelException`	`{:reply, {:error, %{reason: ...}}, socket}`
`messages`	every `push(socket, event, payload)` from a `handle_info/2`

Look at the return type of join: (ChannelTransport, Map<String, dynamic>). That is a Dart 3 record, an anonymous tuple, like an Elixir {channel, reply}. Callers destructure it with final (channel, reply) = await socket.join(...), which you will see in both channel clients.

ChannelMessage (lines 7-15) is just event plus payload. ChannelException (lines 17-30) carries the server’s reason string, or timeout.

The point of the interface is the fake. app/test/support/fake_socket.dart implements both classes in 89 lines. FakeSocket.join (lines 77-88) returns a scripted reply per topic, or throws a scripted error. An unscripted topic throws ChannelException('not_found'). FakeChannel.emit (line 21) pushes a server event into a broadcast StreamController. Every notifier test that touches the socket uses these two classes instead of a real WebSocket.

The real transport over `phoenix_socket`

Open app/lib/data/socket/phoenix_transport.dart. PhoenixSocketTransport adapts the phoenix_socket package (0.8.0 in the lockfile) to the interface. It is the only file that imports that package.

Connecting

connect() (lines 46-85):

final token = await _readToken();
if (token == null || token.isEmpty) {
  log.w('Socket connect skipped: no token');
  return;
}
_set(SocketStatus.connecting);
final socket = PhoenixSocket(
  _url.toString(),
  socketOptions: PhoenixSocketOptions(
    params: {'token': token},
    reconnectDelays: _reconnectDelays ??
        const [
          Duration.zero,
          Duration(seconds: 1),
          // ... 2, 4, 8, 16
          Duration(seconds: 30),
        ],
  ),
);

params: {'token': token} becomes the ?token=... query string, which arrives as the first argument of connect(%{"token" => token}, socket, _connect_info) at backend/lib/footrace_web/channels/user_socket.ex:43. The token is read on every connect(), so a token refreshed since the last connect gets used.

The three listen calls at lines 71-78 turn the package’s open, close and error streams into the app’s SocketStatus. Each StreamSubscription goes into _subs so disconnect() can cancel them all (lines 91-94). That is the cleanup discipline from the Streams section. A forgotten subscription keeps a closure, and whatever it captured, alive.

_status (line 30) is a broadcast controller, and _set (lines 39-43) only emits on a real change. Reconnecting is the package’s job. Your app only reports connecting when the socket closes (lines 73-76).

The docs and the code disagree here. backend/docs/channels.md:14-15 says “A missing or invalid token still connects, but every channel join then fails with {"reason": "unauthorized"}.” The socket module itself refuses a bad token outright. user_socket.ex:39-41 says “A socket with a bad or expired token is refused outright”, and connect/3 returns :error at line 50. Only a socket with no token param at all connects anonymously (lines 54-56). The code wins. For the app this means an expired token gives repeated failed connects with backoff, not a connected socket with failing joins. The client never asks for a socket without a token (lines 48-52 skip the connect), so the anonymous path is never used.

Joining

join() (lines 100-124):

if (!socket.isConnected) {
  // Joining while disconnected leaves the channel errored in the
  // package; wait for the (re)connect first.
  await socket.openStream.first.timeout(const Duration(seconds: 15),
      onTimeout: () {
    throw const ChannelException.timeout();
  });
}
final channel = socket.addChannel(topic: topic, parameters: params);
final reply = await channel.join().future;
if (!reply.isOk) {
  channel.close();
  throw _toException(reply);
}
return (_PhoenixChannelTransport(channel), _asMap(reply.response));

openStream.first is “a Future that completes with the next event on this stream”. .timeout(...) puts a deadline on any Future, the same as withTimeout in Kotlin. _toException (lines 126-131) pulls reason out of the error body. For race:<id> that is one of unauthorized, not_a_competitor, not_found, race_over, token_generation_failed (backend/docs/channels.md:50-56).

The channel adapter

_PhoenixChannelTransport (lines 142-176) has one line worth reading twice:

Stream<ChannelMessage> get messages => _channel.messages
    .where((m) => !m.isReply && !m.event.value.startsWith('phx_'))
    .map((m) => ChannelMessage(/* ... */));

The package’s message stream includes replies to your own pushes and Phoenix housekeeping (phx_reply, phx_close, phx_error). The filter leaves only what your handle_info/2 clauses pushed. push (lines 158-166) awaits the reply and returns its body or throws.

The typed clients, next to their Elixir modules

The transport speaks maps. The two channel classes turn those maps into types, the same job fromJson does for HTTP.

`WaitingChannel` and `FootraceWeb.WaitingChannel`

Open app/lib/data/socket/waiting_channel.dart and backend/lib/footrace_web/channels/waiting_channel.ex side by side.

Client	Server	Contract
`WaitingChannel.join(socket, slug)`, topic `'waiting:$slug'` (lines 114-124)	`join("waiting:" <> race_type_slug, ...)`, line 21. Subscribes the channel process to `matchmaking:user:<id>` and replies with `status_payload`	`channels.md:110-123`
`WaitingJoinReply.fromJson`: `idle`, `queued`, or `matched` with `raceId` (lines 80-88)	`status_payload/1`, lines 95-101	`channels.md:113-117`
`joinQueue()` pushes `'join'`, decodes `QueueReply` (lines 132-133)	`handle_in("join", ...)`, lines 38-58	`channels.md:127-144`
`leaveQueue()` pushes `'leave'`, reads `cancelled` (lines 136-139)	`handle_in("leave", ...)`, lines 61-66	`channels.md:146-147`
`matches`: `messages.where(event == 'match').map(MatchInfo.fromJson)` (lines 127-129)	`handle_info({:matched, race, competitor}, ...)` pushes `"match"`, lines 73-76	`channels.md:152-156`

The keys line up because of to_client/1 at waiting_channel.ex:103, which camelCases the snake_case map from Racing.Matchmaking.match_payload/2. So race_id leaves the server as raceId, and MatchInfo.fromJson reads json['raceId'] (line 53).

Two notes. MatchCompetitor (lines 21-34) reads id, userId and name, and ignores bot and botSlug, which channels.md:139-140 documents. And the app never calls joinQueue() or leaveQueue(). A grep of lib/ finds no caller. Matchmaking queues over HTTP and only listens on the channel. Day 8 shows why. Both methods are tested (app/test/data/socket/waiting_channel_test.dart:65-105) and ready for the day the app switches.

`RaceChannel` and `FootraceWeb.RaceChannel`

Open app/lib/data/socket/race_channel.dart and backend/lib/footrace_web/channels/race_channel.ex.

The events are a sealed class hierarchy (lines 5-42):

sealed class RaceEvent {
  const RaceEvent();
}

class RaceCountdown extends RaceEvent { /* startsAt */ }
class RaceStarted extends RaceEvent { /* race */ }
class RaceUpdated extends RaceEvent { /* snapshot */ }
class CompetitorWon extends RaceEvent { /* competitorId */ }
// CompetitorFinished, CompetitorAborted, RaceFinished

sealed means every subclass must be in this file, so a switch over a RaceEvent is checked for exhaustiveness at compile time. This is Kotlin’s sealed class exactly. In Elixir terms it is the set of message shapes your handle_info/2 clauses match, except the compiler tells you when you forgot one.

parseRaceEvent (lines 46-64) is the decoder, one switch arm per server push:

Event string (app)	`handle_info` clause (server)
`'countdown'`	`{:countdown_started, starts_at}`, `race_channel.ex:99-102`
`'raceStarted'`	`:race_started`, lines 105-109
`'raceUpdate'`	`:race_hydrated` (lines 112-116) and `{:race_update, state}` (lines 119-122)
`'competitorWon'`	`{:competitor_won, id}`, lines 131-134
`'competitorFinished'`	`{:competitor_finished, id}`, lines 125-128
`'competitorAborted'`	`{:competitor_aborted, id}`, lines 137-140
`'raceFinished'`	`:race_finished`, lines 145-148

Two defensive choices. An unknown event returns null (line 58), “so a newer backend never breaks an older app”. A malformed payload is caught by on Object (line 60) and also returns null. events (lines 97-100) drops the nulls with .where((e) => e != null).cast<RaceEvent>(). Your server’s own catch-all, handle_info(message, socket) at line 153, does the same for messages it does not know.

The client-to-server side:

RaceChannel.join (lines 91-94) decodes the join reply into RaceJoinReply: the competitor token plus a RaceSnapshot. The server builds that reply at race_channel.ex:40 as Map.put(state_payload(state, race), :token, token).
pushPositions (lines 105-108) sends {'positions': fixes}. That is the body handle_in("position", ...) parses at race_channel.ex:54-79, and the same body as POST /api/races/:id/positions (channels.md:80-91).
abort (line 110) pushes 'abort' with an empty body. That matches handle_in("abort", ...) at lines 82-92.

RaceNotifier is the one caller of RaceChannel.join (app/lib/features/race/race_notifier.dart:137). If the join throws, it falls back to GET /api/races/:id (lines 144-157). Day 11 covers that.

Keeping the socket alive: `socketLifecycleProvider`

Open app/lib/data/socket/socket_service.dart.

final socketTransportProvider = Provider<SocketTransport>((ref) {
  final config = ref.watch(appConfigProvider);
  final tokens = ref.watch(tokenStoreProvider);
  final transport =
      PhoenixSocketTransport(url: config.socketUrl, readToken: tokens.read);
  ref.onDispose(transport.dispose);
  return transport;
});

final socketLifecycleProvider = Provider<void>((ref) {
  final transport = ref.watch(socketTransportProvider);
  ref.listen<AuthStatus>(authStatusProvider, (_, status) {
    switch (status) {
      case AuthStatus.signedIn:
        transport.connect().catchError(/* log */);
      case AuthStatus.signedOut:
        transport.disconnect();
      case AuthStatus.unknown:
        break;
    }
  }, fireImmediately: true);
});

That is lines 13-38. socketLifecycleProvider is a Provider<void>. It has no value. It exists only for the side effect of its ref.listen. fireImmediately: true runs the callback once with the current status, so an app restored straight into signedIn connects without waiting for a change. Dart switch statements do not fall through, so there is no break after each case. The one break is for the empty unknown case.

Who keeps it alive? FootraceApp.build does ref.watch(socketLifecycleProvider) at app/lib/app/app.dart:15. A provider with no watchers would never be built, so this line is what turns the socket on. In OTP terms it is like starting a child under the application supervisor that only exists to subscribe to a topic and react.

socketStatusProvider (lines 41-45) is a StreamProvider, written as an async* generator:

final socketStatusProvider = StreamProvider<SocketStatus>((ref) async* {
  final transport = ref.watch(socketTransportProvider);
  yield transport.status;
  yield* transport.statusStream;
});

async* makes a function that returns a Stream. yield emits one value. yield* forwards everything from another stream. It is Kotlin’s flow { emit(current); emitAll(other) }. Emitting the current value first matters because statusStream is broadcast and has no replay.

Try it

1. Run the socket tests.

flutter test test/data/socket/

Open app/test/data/socket/socket_service_test.dart:10-31. It builds a container with testOverrides(token: 'jwt') and then one more override, socketTransportProvider.overrideWithValue(socket). testOverrides already overrides that provider, twice in fact (app/test/support/fakes.dart:554 and 558). It works because Riverpod 2.6.1 keeps the last override in the list for a provider. The one at line 554 is dead code, since line 558 always replaces it. That is worth a one-line cleanup PR.

2. Add an event and watch the decoder ignore it. In app/test/data/socket/race_channel_test.dart, after the test at line 112 (“every documented event decodes to its type”), add:

test('an unknown event is dropped, the next one still arrives', () async {
  final channel = await RaceChannel.join(socket, 'race-1');
  final fake = socket.channels['race:race-1']!;
  final next = channel.events.first;
  fake.emit('lapCompleted', {'lap': 2});
  fake.emit('competitorWon', {'competitorId': 'c-bot'});
  expect(await next, isA<CompetitorWon>());
});

The setUp at lines 45-49 already scripts the race:race-1 join reply. Run flutter test test/data/socket/race_channel_test.dart. Then change 'competitorWon' to 'competitor_won' and run it again. The test hangs until the default 30-second timeout. That is the cost of the “unknown returns null” rule: a misspelled event fails silently rather than loudly.

Check yourself

What does the server receive when PhoenixSocketTransport.connect() runs, and which Elixir function sees it?
Why does the messages getter filter out events starting with phx_?
A signed-in user’s token expires while the app is open. According to the backend code, what happens on the next socket reconnect? According to channels.md?
Why is socketLifecycleProvider watched from FootraceApp.build even though nothing reads its value?
What makes RaceEvent a better fit for sealed than RaceStatus is?

Answers

A WebSocket upgrade to /socket/websocket?token=<jwt>&vsn=.... FootraceWeb.UserSocket.connect/3 matches %{"token" => token} at backend/lib/footrace_web/channels/user_socket.ex:43.
The package’s stream includes replies and Phoenix control messages. The filter at app/lib/data/socket/phoenix_transport.dart:151-152 leaves only events a handle_info pushed.
The code refuses the connection (connect/3 returns :error, user_socket.ex:50), so the package keeps retrying with backoff. The doc says it would connect and then fail joins with unauthorized. The code wins. Separately, the next HTTP call gets a 401 and sessionExpired signs the user out, which disconnects the socket.
Providers are lazy and only built when something reads or watches them. The watch at app/lib/app/app.dart:15 builds it once and keeps its ref.listen alive for the app’s lifetime.
Its cases carry different data (a timestamp, a snapshot, a competitor id). RaceStatus cases carry nothing, so an enum is enough.

Day 8: Matchmaking queue and the lobby puts WaitingChannel to work inside QueueNotifier, with timers and a polling fallback.

Day 7: Phoenix channels from the client side

Before you start

The seam: SocketTransport and ChannelTransport

The real transport over phoenix_socket