Realtime

The app holds one Phoenix socket for as long as you are signed in, and joins two kinds of topic on it: waiting:<race_type_slug> while you are in the matchmaking queue, and race:<race_id> while you race. You wrote the server side of both. This page is the client side, in app/lib/data/socket/, and how it lines up with backend/lib/footrace_web/channels/.

The contract both sides keep to is backend/docs/channels.md. Where this page and that file disagree with the Elixir modules, the modules win, and there is one such case below.

Layers

flowchart TB
  subgraph features
    Q[QueueNotifier<br/>features/queue]
    RN[RaceNotifier<br/>features/race]
  end
  subgraph "data/socket (typed)"
    WC[WaitingChannel]
    RC[RaceChannel]
  end
  subgraph "data/socket (transport)"
    ST[[SocketTransport<br/>ChannelTransport<br/>abstract]]
    PT[PhoenixSocketTransport]
    FS[FakeSocket<br/>test/support]
  end
  PS[phoenix_socket package]
  US[FootraceWeb.UserSocket<br/>/socket]
  Q --> WC
  RN --> RC
  WC --> ST
  RC --> ST
  PT -. implements .-> ST
  FS -. implements .-> ST
  PT --> PS
  PS -->|"wss ...?token=jwt&vsn=2.0.0"| US

Three layers, each with one job:

1. Transport interfaces (channel_transport.dart): the smallest slice of Phoenix Channels the app needs, written so a test can fake it.
2. Phoenix implementation (phoenix_transport.dart): that interface over the phoenix_socket package.
3. Typed channel clients (race_channel.dart, waiting_channel.dart): topic names, event names and payload parsing, returning Dart types instead of maps.

Features only see layer 3 plus SocketTransport from a provider.

The transport interfaces

app/lib/data/socket/channel_transport.dart:33-61:

abstract class ChannelTransport {
  String get topic;

  /// Server pushes, excluding replies and Phoenix housekeeping events.
  Stream<ChannelMessage> get messages;

  /// Sends [event] and returns the `ok` reply body. Throws
  /// [ChannelException] on an `error` reply or a timeout.
  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,
  });
}

What maps to what you know:

• A Stream<ChannelMessage> is a Kotlin Flow or an RxSwift Observable: a sequence of values over time you listen to. The nearest Elixir picture is the mailbox of your channel process: handle_info clauses consuming pushes, except here the consumer subscribes and gets a StreamSubscription to cancel later.
• push returns a Future of the reply body. That is the client half of {:reply, {:ok, payload}, socket} in a handle_in. An {:error, %{reason: ...}} reply becomes a thrown ChannelException carrying reason (app/lib/data/socket/channel_transport.dart:18-30).
• join returns a record, (ChannelTransport, Map<String, dynamic>). Dart 3 records are anonymous tuples, like an Elixir {channel, reply}, and callers destructure them the same way: final (channel, reply) = await socket.join(...) (app/lib/data/socket/race_channel.dart:92). The map is what your join/3 returned in {:ok, reply, socket}.

SocketStatus has three values, disconnected, connecting, connected (app/lib/data/socket/channel_transport.dart:4). “Connecting” also covers “reconnecting”.

PhoenixSocketTransport

app/lib/data/socket/phoenix_transport.dart is about 170 lines. The parts that matter:

Connect

@override
Future<void> connect() async {
  if (_socket != null) return;
  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),
          ],
    ),
  );
  // ...

(app/lib/data/socket/phoenix_transport.dart:45-69)

The URL is AppConfig.socketUrl, ws(s)://host/socket/websocket (app/lib/core/config/app_config.dart:33-39). The package appends vsn=2.0.0 itself, and the JWT goes in as the token param, which is what FootraceWeb.UserSocket.connect/3 pattern matches on (backend/lib/footrace_web/channels/user_socket.ex:43).

After connecting, three package streams drive the status (app/lib/data/socket/phoenix_transport.dart:71-78): openStream sets connected, closeStream sets connecting (the package retries on its own with the delays above), and errorStream only logs. A failed first connect() is caught and left alone, again because the package keeps retrying (app/lib/data/socket/phoenix_transport.dart:79-84).

The token is read once per connect, not per reconnect

The class comment says “the token is read at every connect so a refreshed token is picked up on the next reconnect cycle” (app/lib/data/socket/phoenix_transport.dart:12-14). That holds for disconnect() followed by connect(). It does not hold for the package’s own automatic reconnects: PhoenixSocketOptions is built with a fixed params map, and phoenix_socket 0.8.0 also offers dynamicParams for exactly this case, which the app does not use. In practice the app has no token refresh yet, and a dead token ends in a 401 on HTTP, sign-out, and a fresh socket on next sign-in, so the gap is theoretical today.

Join

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));

(app/lib/data/socket/phoenix_transport.dart:109-123)

openStream.first is “the next value of this stream, as a Future”. Waiting up to 15 s for the socket means a screen can call join straight after sign-in without racing the connect. _toException pulls reason out of the error body, so the {:error, %{reason: "not_a_competitor"}} your RaceChannel.join/3 returns arrives as ChannelException('not_a_competitor') (app/lib/data/socket/phoenix_transport.dart:126-131).

Messages

@override
Stream<ChannelMessage> get messages => _channel.messages
    .where((m) => !m.isReply && !m.event.value.startsWith('phx_'))
    .map((m) => ChannelMessage(
          m.event.value,
          PhoenixSocketTransport._asMap(m.payload),
        ));

(app/lib/data/socket/phoenix_transport.dart:150-156)

The package’s channel stream includes replies (phx_reply) and lifecycle events (phx_close, phx_error). This filter leaves only what your channel sent with push/3. where and map on a Stream are the same lazy operators as on a Kotlin Flow.

Rejoining a channel after a reconnect is the package’s job (it has a rejoin timer per channel). The app does not see the rejoin reply; it keeps listening to the same messages stream.

The providers

app/lib/data/socket/socket_service.dart holds three:

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((Object e) => log.w('Socket connect failed: $e'));
      case AuthStatus.signedOut:
        transport.disconnect();
      case AuthStatus.unknown:
        break;
    }
  }, fireImmediately: true);
});

(app/lib/data/socket/socket_service.dart:13-38)

socketLifecycleProvider is a Provider<void>: it produces nothing and exists for its side effect, a listener on auth status. It is kept alive by one ref.watch in the root widget, app/lib/app/app.dart:15. If you know Phoenix better than Riverpod, it plays the part of a supervised process started from Application.start/2 that subscribes to an auth PubSub topic and opens or closes a connection. fireImmediately: true makes it act on the current status straight away, not only on the next change.

Dart switch statements do not fall through, so no break is needed after each case except to make an empty case explicit.

socketStatusProvider (app/lib/data/socket/socket_service.dart:41-45) turns the status stream into a provider for “reconnecting” banners. At d1b602a nothing under lib/ watches it; only app/test/data/socket/socket_service_test.dart does. RaceNotifier listens to statusStream directly instead (app/lib/features/race/race_notifier.dart:141-143).

WaitingChannel: the queue

app/lib/data/socket/waiting_channel.dart wraps waiting:<slug>. The backend design is that joining the topic subscribes you to your own matchmaking notifications without queueing you (backend/lib/footrace_web/channels/waiting_channel.ex:21-31). The client follows it:

static Future<WaitingChannel> join(
  SocketTransport socket,
  String raceTypeSlug,
) async {
  final (channel, reply) = await socket.join(topicFor(raceTypeSlug));
  return WaitingChannel._(
      channel, WaitingJoinReply.fromJson(reply), raceTypeSlug);
}

/// `match` pushes: you were paired, whichever side triggered it.
Stream<MatchInfo> get matches => _channel.messages
    .where((m) => m.event == 'match')
    .map((m) => MatchInfo.fromJson(m.payload));

Future<QueueReply> joinQueue() async =>
    QueueReply.fromJson(await _channel.push('join', const {}));

(app/lib/data/socket/waiting_channel.dart:117-133)

WaitingChannel._(...) is a private named constructor: the only way to get a WaitingChannel is the static join, which guarantees the topic has been joined. Same idea as a Kotlin private constructor with a companion factory.

How QueueNotifier actually uses it is worth knowing, because it does not use joinQueue:

1. It queues over HTTP, POST /api/matchmaking/join (app/lib/features/queue/queue_notifier.dart:98).
2. If not matched at once, it joins waiting:<slug> only to hear the match push (app/lib/features/queue/queue_notifier.dart:122-141).
3. If the join reply already says matched (paired between the HTTP call and the socket join), it handles that as if the push had arrived (line 136).
4. If the channel join throws, it falls back to polling GET /api/competitors/:id/poll every 2 s (app/lib/features/queue/queue_notifier.dart:72, app/lib/features/queue/queue_notifier.dart:143-152).

That works because your backend has one queue, the match_requests table, shared by HTTP and the channel (backend/docs/channels.md:106-108). joinQueue and leaveQueue exist and are tested but no feature calls them at d1b602a. The full flow is in Matchmaking.

RaceChannel: the race

app/lib/data/socket/race_channel.dart wraps race:<race_id>. Its centrepiece is a sealed event hierarchy:

sealed class RaceEvent {
  const RaceEvent();
}

class RaceCountdown extends RaceEvent {
  const RaceCountdown(this.startsAt);
  final DateTime startsAt;
}
// RaceStarted, RaceUpdated, CompetitorWon, CompetitorFinished,
// CompetitorAborted, RaceFinished

(app/lib/data/socket/race_channel.dart:5-42)

and one parser for all server pushes:

RaceEvent? parseRaceEvent(ChannelMessage message) {
  final p = message.payload;
  try {
    return switch (message.event) {
      'countdown' => RaceCountdown(DateTime.parse(p['startsAt'] as String)),
      'raceStarted' =>
        RaceStarted(RaceInfo.fromJson(Map<String, dynamic>.from(p['race']))),
      'raceUpdate' => RaceUpdated(RaceSnapshot.fromJson(p)),
      'competitorWon' => CompetitorWon(p['competitorId'].toString()),
      'competitorFinished' => CompetitorFinished(p['competitorId'].toString()),
      'competitorAborted' => CompetitorAborted(p['competitorId'].toString()),
      'raceFinished' => RaceFinished(RaceSnapshot.fromJson(p)),
      _ => null,
    };
  } on Object {
    // A malformed payload is the server's bug; do not kill the stream.
    return null;
  }
}

(app/lib/data/socket/race_channel.dart:46-64)

sealed works like a Kotlin sealed class: all subclasses live in this file, so a switch over RaceEvent is checked for exhaustiveness. RaceNotifier._onEvent is that switch, and reads very much like a set of handle_info clauses (app/lib/features/race/race_notifier.dart:178-199):

switch (event) {
  case RaceCountdown(:final startsAt):
    _update((st) => st.copyWith(startsAt: startsAt));
    _schedule(startsAt);
  case RaceStarted():
    _startRacing();
  // ...
}

RaceCountdown(:final startsAt) is an object pattern that matches the type and binds the field, the Dart spelling of %RaceCountdown{starts_at: starts_at}. Unknown events return null and are filtered out (app/lib/data/socket/race_channel.dart:97-100), so a newer backend that adds an event never breaks an older app.

The join reply is the full engine state plus a competitor token (app/lib/data/socket/race_channel.dart:67-77), matching {:ok, Map.put(state_payload(state, race), :token, token), socket} at backend/lib/footrace_web/channels/race_channel.ex:40. The app keeps the token field but does not use it.

What goes up the race channel

RaceChannel.abort() sends abort and is used, with HTTP as the fallback when there is no channel (app/lib/features/race/race_notifier.dart:347-363).

RaceChannel.pushPositions() exists (app/lib/data/socket/race_channel.dart:105-108) and is not called anywhere at d1b602a. GPS fixes go to POST /api/races/:id/positions from the foreground-service isolate instead, because that isolate keeps running with the screen off and has no access to the socket, which lives in the UI isolate. The server reply to each upload carries the same {race, competitors, competitor} shape as raceUpdate, and RaceNotifier applies it the same way. See Tracking.

When the race channel is not there

If RaceChannel.join throws, RaceNotifier.build fetches GET /api/races/:id for the state, marks socketDown, and carries on. Upload replies then keep the screen current every two seconds. not_a_competitor and not_found are fatal; anything else falls back (app/lib/features/race/race_notifier.dart:144-158).

Topics and events, both sides

Direction	Topic	Event	Client	Server
connect	/socket	token param	app/lib/data/socket/phoenix_transport.dart:57	backend/lib/footrace_web/channels/user_socket.ex:43-52
join	waiting:<slug>	reply {status, raceId?, ...}	app/lib/data/socket/waiting_channel.dart:80-88	backend/lib/footrace_web/channels/waiting_channel.ex:21-31
client →	waiting:<slug>	join (queue)	app/lib/data/socket/waiting_channel.dart:132-133 (unused)	backend/lib/footrace_web/channels/waiting_channel.ex:38-58
client →	waiting:<slug>	leave	app/lib/data/socket/waiting_channel.dart:136-139 (unused)	backend/lib/footrace_web/channels/waiting_channel.ex:61-66
server →	waiting:<slug>	match	app/lib/data/socket/waiting_channel.dart:127-129	backend/lib/footrace_web/channels/waiting_channel.ex:73-76
join	race:<id>	reply {token, race, competitors}	app/lib/data/socket/race_channel.dart:91-94	backend/lib/footrace_web/channels/race_channel.ex:20-46
client →	race:<id>	position	app/lib/data/socket/race_channel.dart:105-108 (unused)	backend/lib/footrace_web/channels/race_channel.ex:54-79
client →	race:<id>	abort	app/lib/data/socket/race_channel.dart:110	backend/lib/footrace_web/channels/race_channel.ex:82-92
server →	race:<id>	countdown	app/lib/data/socket/race_channel.dart:50	backend/lib/footrace_web/channels/race_channel.ex:99-100
server →	race:<id>	raceStarted	app/lib/data/socket/race_channel.dart:51-52	backend/lib/footrace_web/channels/race_channel.ex:105-107
server →	race:<id>	raceUpdate	app/lib/data/socket/race_channel.dart:53	backend/lib/footrace_web/channels/race_channel.ex:112-120
server →	race:<id>	competitorWon	app/lib/data/socket/race_channel.dart:54	backend/lib/footrace_web/channels/race_channel.ex:131-132
server →	race:<id>	competitorFinished	app/lib/data/socket/race_channel.dart:55	backend/lib/footrace_web/channels/race_channel.ex:125-126
server →	race:<id>	competitorAborted	app/lib/data/socket/race_channel.dart:56	backend/lib/footrace_web/channels/race_channel.ex:137-138
server →	race:<id>	raceFinished	app/lib/data/socket/race_channel.dart:57	backend/lib/footrace_web/channels/race_channel.ex:145-146

Fields the client drops: botSlug on competitors (neither CompetitorState nor MatchCompetitor reads it) and bot on MatchCompetitor. The fixture contract test lists the deliberate ignores; see Testing.

A race, end to end over the socket

sequenceDiagram
  autonumber
  participant UI as QueueNotifier / RaceNotifier
  participant T as PhoenixSocketTransport
  participant US as UserSocket
  participant WC as WaitingChannel (server)
  participant RC as RaceChannel (server)
  participant RP as RaceProcess
  Note over UI,T: auth becomes signedIn
  UI->>T: connect() via socketLifecycleProvider
  T->>US: ws /socket/websocket?token=jwt
  US-->>T: connected (current_user assigned)
  UI->>UI: POST /api/matchmaking/join (HTTP) → queued
  UI->>T: join("waiting:one_km")
  T->>WC: phx_join
  WC-->>T: ok {status: "queued"}
  Note over WC: another runner queues
  WC-->>T: push "match" {raceId, startsAt, me, opponent}
  T-->>UI: MatchInfo → QueueMatched
  UI->>T: race screen opens, join("race:<id>")
  T->>RC: phx_join
  RC->>RP: ensure_process, get_state
  RC-->>T: ok {token, race, competitors}
  RP-->>RC: {:countdown_started, t}
  RC-->>T: push "countdown" {startsAt}
  RP-->>RC: :race_started
  RC-->>T: push "raceStarted"
  loop every applied position, either runner
    RP-->>RC: {:race_update, state}
    RC-->>T: push "raceUpdate" {race, competitors}
  end
  RP-->>RC: {:competitor_won, id}
  RC-->>T: push "competitorWon"
  RP-->>RC: :race_finished
  RC-->>T: push "raceFinished" (final state)

Positions do not appear in this diagram on purpose: they travel over HTTP from the service isolate, and the raceUpdate pushes in the loop are the server’s reaction to them.

Where the contract doc and the code disagree

backend/docs/channels.md:14-15 says a missing or invalid token “still connects, but every channel join then fails with unauthorized”. UserSocket.connect/3 does that only for a missing token. For a token that fails Users.get_user_from_client_token/1, it returns :error and the socket is refused at connect (backend/lib/footrace_web/channels/user_socket.ex:39-52; the comment above it says so too).

On the client that shows up as the socket never reaching connected: phoenix_socket keeps retrying with the same stale token, and any join waits 15 s for openStream and then throws ChannelException.timeout(). Both features handle that by falling back to HTTP (polling for the queue, GET /api/races/:id for the race), and the next authenticated HTTP call gets a 401 and signs you out. So the app copes; the doc is the thing that is out of date.

Testing the socket

Because features depend on SocketTransport, not on phoenix_socket, tests replace the whole socket with FakeSocket (app/test/support/fake_socket.dart:43), which can script join replies, refuse joins with a reason, and push server events. testOverrides installs it for every widget test (app/test/support/fakes.dart:554). PhoenixSocketTransport itself is the one piece that is only exercised against a real backend, by the integration test; see Testing.

Where to go next

• Day 7: Phoenix channels from the client side
• Day 8: Matchmaking queue and the lobby
• Race flow puts the channel next to the tracker and the uploads.
• Failure modes covers network loss mid-race.