Bridge and live updates

What this part does

This part answers one question for every open screen: is what I show still true? When one person changes something, every other screen showing it must update quickly. That means a second scorer on the same match, an operator on the schedule, and the console on the medal table.

The user's own words, from the review of Section 7: when two scorers are scoring, the second one's screen should update as soon as the first changes something.

Section 8 answered five questions:

The bridge in one table Built today

The bridge is the one road between an app and omnium. An app runs in an iframe (a page inside the console page). The console is the host. The app and the host talk with postMessage, and only the host talks HTTP to the admin service.

The message types are in packages-ts/omnium-bridge/src/protocol.ts:124-169. The protocol number is PROTOCOL = 1 (protocol.ts:8). The package is published as @fanos/omnium-bridge 1.0.0; the admin panel in the repo uses its source directly.

How it works

The drawing shows today on the left and the agreed design on the right.

Today Built today

1. A command saves its change and one outbox row. The outbox is the domain_event table. The row is written in the same transaction as the change, so it exists exactly when the change does.
2. Each open screen has its own stream. The admin service runs GET /stream/changes for each screen. Once a second it reads new domain_event rows, up to 500, and filters them in Python by event or by fixture.
3. The stream sends "these areas changed". A frame names areas like fixture or medals, not data.
4. The host waits 250 ms, then re-reads. It re-reads, one after another, every watch whose areas were touched. It pushes a new value to the app only if the value changed.
5. The stream ends itself after 10 s. A proxy on AWS cuts every request at 15 s, so the server hangs up first. The browser dials again with its last place.

Agreed design Agreed, to build

1. Told, not asking. Every command transaction ends with pg_notify('omnium_change', outbox_id). Postgres delivers it only when the transaction commits.
2. One listener per server container. Each container keeps one dedicated connection that has run LISTEN omnium_change. A wake-up arrives 0.20 ms after the commit (median, measured).
3. One read per container. On a wake-up, the container reads the new outbox rows once, for all its screens, with the late-commit-safe reader from Section 4 (L9). A safety read runs every 5 s in case a wake-up was lost.
4. Shared watches. A screen tells the server what it watches: a read model and its parameters. The server keeps one copy per distinct watch, not one per screen. Fifty consoles on one schedule day are one watch.
5. Push the value. When a change touches a watch, the server works it out once and pushes the new value to every screen watching it. No screen has to read again.
6. Narrow updates. A read model says which record each row comes from. A change to one match updates that match's row, not the whole day.
7. A version on every value. A screen that holds version 41 and receives 43 knows it missed 42, and re-reads that one record.
8. A stream that never goes quiet. The client reopens the stream after any close, including an error answer. Heartbeats come every 15 s, and they carry the deployed app version and the bridge protocol version.

A worked example

Low-level design

The code below is real unless it is marked "Agreed design, not in the code yet". How commands themselves are written is on Writing a workflow in code.

The outbox row Built today

Every accepted command writes one domain_event row in its own transaction. The scoring pipeline does it here:

# packages/core/src/omnium_core/scoring/pipeline.py:971-979
jobs_outbox.emit(
    session,
    jobs_outbox.COMMAND_ACCEPTED,
    sport_id=fixture.sport_id,
    competition_id=fixture.competition_id,
    fixture_id=fixture.id,
    seq=item.seq,
    payload={"command": code, "touches": ["fixture"]},
)

And the workflow engine (Games operations, Results) does it here:

# packages/core/src/omnium_core/workflows/engine.py:171-188
jobs_outbox.emit(
    session,
    jobs_outbox.COMMAND_ACCEPTED,
    sport_id=fixture.sport_id if fixture is not None else None,
    competition_id=fixture.competition_id if fixture is not None else event.id,
    fixture_id=fixture.id if fixture is not None else None,
    seq=item.seq,
    payload={
        "command": code,
        "workflow": workflow.code,
        "actor": actor.name,
        "event": str(event.id),
        # What a host re-reads for it: only watches that depend on these.
        "touches": touches(report.kinds),
    },
)

In plain words:

• emit (packages/core/src/omnium_core/jobs_outbox.py:44) only calls session.add. The row is saved by the caller's commit, so a rolled-back command leaves no row.
• touches is a list of areas: fixture, medals, entries, integration. A scoring command always says ["fixture"]. The engine works the areas out from the kinds of records the command wrote (workflows/records.py:94). If a kind is unknown, it says None, which means "re-read everything".
• The table is domain_event (model in packages/contract/src/omnium_contract/models/scheduler.py:26). Columns used here: id (big serial), kind, competition_id, fixture_id, seq, payload (JSONB). It is insert-only and pruned after 30 days.

Today's stream: one polling loop per screen Built today

The constants first:

# packages/admin/src/omnium_admin/routers/bridge.py:766-771
_POLL_SECONDS = 1.0
_KEEPALIVE_POLLS = 15
_TREE_POLLS = 60
#: How soon a browser dials again after a stream ends. Short, because a stream
#: now ends on purpose every few seconds and the gap is time with no updates.
_RETRY_MS = 1000

The loop that runs for each open screen (trimmed):

# packages/admin/src/omnium_admin/routers/bridge.py:901-958 (trimmed)
async def body() -> AsyncIterator[bytes]:
    cursor = start
    polls = 0
    ends_at = clock.time() + limit if limit > 0 else None
    yield f"retry: {_RETRY_MS}\n: connected\n\n".encode() + _place(cursor)
    while True:
        async with factory() as session:
            if event_row is not None and polls % _TREE_POLLS == 0:
                ids = (await session.execute(subjects.event_tree(event_row.id))).scalars()
                tree = {*ids, event_row.id}
            rows = (
                await session.execute(
                    sa.select(DomainEvent.id, DomainEvent.kind, DomainEvent.fixture_id,
                              DomainEvent.competition_id, DomainEvent.payload)
                    .where(DomainEvent.id > cursor)
                    .order_by(DomainEvent.id)
                    .limit(500)
                )
            ).all()
        polls += 1
        if rows:
            cursor = int(rows[-1].id)
            changes = [c for c in map(_as_change, rows) if wanted(c, ...)]
            if changes:
                yield _frame(cursor, changes)
        ...
        if ends_at is not None and clock.time() >= ends_at:
            yield _place(cursor)      # hang up before the proxy does
            return
        if not rows:
            await asyncio.sleep(_POLL_SECONDS)

What each part does, and what it costs:

• One query a second per screen. DomainEvent.id > cursor with LIMIT 500 runs every second, even when nothing changed.
• Every screen reads every row. The query has no filter for the screen's event or fixture. wanted() (bridge.py:795) drops the rows the screen does not need, in Python, after they were read. During a big import every screen reads every row.
• The event tree. For an event screen, the loop reads the list of competitions under the event when polls % 60 == 0. Because polls starts at 0 in each new stream, and a stream lasts 10 s, this read runs at the start of every stream, so about every 10 s.
• The 10 s hang-up. ends_at comes from stream_max_seconds. Each new stream also checks the login and looks up the event by code.
• Late commits are skipped. id > cursor assumes ids commit in order. They do not: a transaction that took id 100 can commit after one that took id 101. Once the cursor passed 101, row 100 is never sent. Section 4 (L9) agreed a late-commit-safe reader.
• Keepalive. A : keepalive comment is sent every 15 polls. With 10 s streams, this line is not reached (from the code, not tested).

Counting from the code, an event screen that sees no changes runs, per 10 s stream: about 10 outbox reads, 1 tree read, 1 login check and 1 event lookup. That is about 13 queries per 10 s, or about 78 a minute. The design doc rounds this to 70 to 80 a minute. A fixture screen skips the tree and event reads.

The frame it sends carries areas and places, not data:

# packages/admin/src/omnium_admin/routers/bridge.py:826-856 (trimmed)
def _frame(seq: int, changes: list[dict[str, Any]]) -> bytes:
    body = {
        "seq": seq,
        "changes": [
            {"seq": change["seq"], "kind": change["kind"],
             "fixtureId": ..., "competitionId": ...,
             "workflow": change["workflow"], "command": change["command"],
             "touches": change["touches"]}
            for change in changes
        ],
    }
    return f"event: changes\nid: {seq}\ndata: {json.dumps(body)}\n\n".encode()

seq here is the stream's place in the outbox, not a version of any record. A screen cannot tell from it that it missed a value.

Why the stream ends after 10 s Built today

# packages/core/src/omnium_core/settings.py:259-267
# --- The admin panel's live stream (/stream/changes) ---------------------
# How long one stream stays open before the server ends it and the browser
# dials again, carrying on from the last change it saw. 0 keeps it open.
#
# Not 0 by default because of AWS: admin-api sits behind ECS Service Connect,
# whose proxy ends EVERY request after 15 seconds unless the service sets a
# timeout. A stream cut there shows "Reconnecting" and can drop a change.
# Ending first, cleanly, avoids both. Set 0 once that timeout is lifted.
stream_max_seconds: float = Field(default=10.0, ge=0)

The agreed design raises the per-request timeout on the admin service and sets stream_max_seconds = 0 (G5). The infra change is tracked in Section 14 (Deploy).

Today's client: the browser side Built today

The bridge's HTTP client opens the stream with the browser's EventSource:

// packages-ts/omnium-bridge/src/http.ts:244-257
changes(onChange) {
  const live = { event: options.live?.event, fixture: options.live?.fixture };
  const Source =
    options.EventSource === undefined
      ? (globalThis as { EventSource?: EventSourceConstructor }).EventSource
      : options.EventSource;
  if (Source) {
    // A browser's EventSource cannot send a header, so the token rides in
    // the address — for this one route only.
    const stream = new Source(
      `${base}/stream/changes` + query({ ...live, access_token: options.token?.() }),
    );
    stream.addEventListener("changes", (event) => onChange(touchesOf(event?.data)));
    return () => stream.close();
  }
  // ... otherwise poll with ?once=true every 20 s

The host turns "areas changed" into re-reads:

// packages-ts/omnium-bridge/src/host.ts:188-228 (trimmed)
changed(touches: string[] | null = null): void {
  if (touches === null) this.everything = true;
  else if (touches.length === 0) return;
  else for (const area of touches) this.touched.add(area);
  if (this.settleTimer) clearTimeout(this.settleTimer);
  this.settleTimer = setTimeout(() => {
    this.settleTimer = null;
    void this.refresh();
  }, this.options.settleMs ?? 250);
}

refresh(): Promise<void> {
  ...
  for (const [watchId, watch] of [...this.watches]) {
    if (!everything && watch.depends && !watch.depends.some((area) => touched.has(area))) {
      continue;
    }
    const { value, depends } = await this.readWith(watch, watch.name, watch.params);
    const text = JSON.stringify(value);
    if (!this.watches.has(watchId) || text === watch.last) continue;
    watch.last = text;
    this.post({ type: "push", watchId, value });
  }
  ...
}

The console wraps EventSource to show the live dot:

// frontend-admin/src/lib/bridge/frame.ts:194-206
constructor(url: string) {
  this.source = new Source(url);
  this.source.addEventListener("open", () => {
    this.settle();
    onState("open");
  });
  this.source.addEventListener("error", () => {
    if (this.pending !== null) return;
    this.pending = setTimeout(() => {
      this.pending = null;
      onState("retrying");
    }, graceMs);
  });
}

In plain words:

• The URL is fixed at the start. The login token is put in the address once. The browser's own reconnects reuse that address, with the same token.
• The re-reads are one after another. refresh() awaits each watch in turn. A console with four schedule watches does four reads in a row for one ball.
• An error only changes the label. After 5 s of error (RECONNECT_GRACE_MS = 5_000, frame.ts:170), the dot says "Reconnecting". Nothing opens a new stream. By the browser's rules, an answer that is not 200 (an expired login gives 401, a proxy can give 502) closes an EventSource for good. So the screen says "Reconnecting" forever and never updates again.
• Old code is checked once. The host checks the protocol number only when the app says ready (host.ts:261-264). The app also sends its own version in ready, but nothing compares it, and nothing tells an open app that a new version was deployed.

The wake-up Agreed, to build

Agreed design, not in the code yet. Searching the repo for pg_notify, LISTEN and add_listener in packages/*/src finds nothing today.

-- Agreed design, not in the code yet (from the Section 8 doc).
-- In every command transaction, right after the outbox insert (Section 4):
INSERT INTO domain_event (...) VALUES (...) RETURNING id;
SELECT pg_notify('omnium_change', $outbox_id::text);   -- delivered only if this transaction commits
COMMIT;

The listener, one per admin container:

# Agreed design, not in the code yet: admin/src/omnium_admin/live/listener.py (new)
async def run(dsn: str, on_change: Callable[[], Awaitable[None]]) -> None:
    while True:
        try:
            conn = await asyncpg.connect(dsn)                 # outside the pool, never in a transaction
            wake = asyncio.Event()
            await conn.add_listener("omnium_change", lambda *_: wake.set())
            await on_change()                                  # catch up from the last place after (re)connecting
            while True:
                try:
                    await asyncio.wait_for(wake.wait(), timeout=5.0)   # the safety read every 5 s
                except asyncio.TimeoutError:
                    pass
                wake.clear()
                await on_change()                              # one read for any number of notifications
        except (OSError, asyncpg.PostgresError):
            await asyncio.sleep(backoff())                     # reconnect with growing waits

What matters in it:

• Its own connection, outside the pool. A listener inside a transaction receives no notifications, and a long transaction that listens can hold up Postgres's notification queue.
• The wake-up is only a doorbell. The outbox is the truth. A lost notification costs at most 5 s, because the safety read finds the row anyway.
• Many notifications, one read. wake is a flag. Ten commits in one moment cause one read.
• Reconnect reads from the last place. After a dropped connection, on_change() runs before waiting again, so nothing is missed.

Shared watches Agreed, to build

Agreed design, not in the code yet. Today each browser's host keeps its own watches (packages-ts/omnium-bridge/src/host.ts:289-302). In the design, the server keeps them, once per distinct view:

# Agreed design, not in the code yet: admin/src/omnium_admin/live/watches.py (new)
# per container, in memory
watches: dict[WatchKey, Watch]            # WatchKey = (read name, canonical params)
class Watch:
    value: bytes                          # canonical JSON of the last value
    version: int                          # the highest record version inside it
    depends: set[str]                     # areas, as read models declare today
    records: set[uuid.UUID]               # which records it shows, for narrow updates
    screens: set[StreamId]

The rules (G2, G3):

Narrow updates need one change to read models (G6). Today the schedule reads say only an area:

# packages/core/src/omnium_core/workflows/reads_schedule.py:295, 338, 392
@read_model("games.schedule_days", depends=("fixture",))
@read_model("games.schedule_day", depends=("fixture",))
@read_model("games.schedule_choices", depends=("fixture",))

So any change to any match re-reads all of them. In the design, each read model also declares which records each row comes from. depends stays. The registry is ReadModel in packages/core/src/omnium_core/workflows/reads.py:52-57.

The new stream Agreed, to build

Agreed design, not in the code yet. /stream/changes (bridge.py:859) is replaced by three routes:

GET  /stream?subject=event:asian-games-2026            -> text/event-stream, first frame: "hello" with stream id, app version, protocol version
POST /stream/{id}/watch   {name, params}               -> {watchId, version, value}   (the first value at once)
POST /stream/{id}/unwatch {watchId}

event: value      id: <outbox place>   data: {"watchId": "w7", "version": 205, "value": {...}}
event: resync     data: {"watchId": "w7"}              -> the screen reads that watch again
event: heartbeat  data: {"app": "2026.10.06-3", "protocol": 2}     every 15 s

How the parts fit:

• Watches are registered over HTTP on an open stream (G4). The stream stays one-way. No WebSockets are needed.
• id: is the outbox place. The browser sends it back as Last-Event-ID when it reconnects after a network error.
• version is per record. It is what lets a screen notice a gap (B5).
• The client wrapper reopens after any close (B6, G5). For an error answer, which the browser never retries, the wrapper opens a new stream itself, with a fresh login from the panel and its last place. It then sends its watches again. Waits between tries grow.
• 30 s of silence counts as broken. Heartbeats come every 15 s; two missed heartbeats trigger a reopen.
• Heartbeats carry versions (B7). A newer app shows a quiet "A new version is ready" banner with Reload. A protocol the old code cannot read shows "Please reload", and the panel stops sending that app's commands.

Changes on the client side, from the design doc's file list:

New server files

All new, from the design doc's file list:

What a screen shows

Today the dot has four labels: "Live", "Reconnecting", "Connecting" and "Checking every 20 s" when there is no EventSource (frame.ts:151-158).

The public API stream is a different thing

GET /v1/stream in the public API (packages/api/src/omnium_api/routers/stream.py:115-151) is for fans, not staff screens. It subscribes to a Redis pub/sub channel and sends {entity, ids, seq}, with a 15 s heartbeat. Redis pub/sub keeps no replay, so a client that missed a message must re-fetch. It answers 503 when no Redis is set. Section 8 does not change it; fan delivery is on Stats, feeds and delivery.

When things go wrong

Every case ends with each screen showing the latest data, or saying clearly that it cannot.

Decisions

All 14 decisions are Agreed (closed 6 Oct 2026). None was dropped.

Built today, or still to build

The design doc's build order puts the cheapest big win first:

1. notify() in the engine after every outbox insert. Harmless while nobody listens.
2. The listener and the late-commit-safe reader in each admin container, feeding today's /stream/changes. The per-screen polling stops here.
3. The client wrapper that reopens after any close (fixes silent screens), and the proxy timeout.
4. Shared watches and the new stream routes; the host forwards values instead of re-reading.
5. Record keys on the schedule read models (narrow row updates).
6. Versions on values, gap detection, the Catching up state.
7. App and protocol versions in heartbeats; the "new version" banner.

Numbers

The tests that will prove the design (Section 12):

Read next

• Console and scorer apps: the screens that sit on top of this stream, and how they stay fast.
• Commands and the workflow engine: where the outbox row is written, and the late-commit-safe reader.
• Stats, feeds and delivery: the same wake-up drives the stats queue and the feed builder.