Emailzap · Server-Sent Events (SSE)

How realtime events move: Lambda / Celery producers → Valkey pub/sub → Django ASGI workers → chromex & mobile. Click nodes, cards, or table rows for paths & evidence. Snapshot from worktree evidence (docs/memory + code), not a claim of live prod flag state.

Wired (code present + consumer/producer path) Partial / flag-gated / dual-emit Dead / superseded / never shipped Shared package / transport solid = clickable detail · dashed = external / bus / terminal

0 · Prerequisite vocabulary

Terms you need before the rest makes sense.

TermMeaning
HTTP request/responseClient opens TCP (usually TLS) to server; one request → one response. Normally response ends → connection reusable or closed.
Long-lived responseServer deliberately does not finish the response body; keeps writing chunks for minutes/hours. That is the substrate for SSE.
WSGIClassic Python web interface. Sync: one worker thread/process typically busy for whole request lifetime.
ASGIAsync-capable Python web interface. One event loop can juggle many open connections without one OS thread per stream.
GunicornProcess manager: spawns N worker processes, binds port, restarts workers. Does not invent async by itself.
uvicornASGI server. Emailzap runs it as Gunicorn worker class: -k uvicorn.workers.UvicornWorker.
Worker processSeparate OS process with its own memory. Dicts in one worker are invisible to sibling workers.
ALBLoad balancer. Each new HTTP request (including a push POST) can land on any healthy worker.
Pub/SubPublish once → all subscribers get a copy. Valkey/Redis PUBLISH is fire-and-forget (no durable queue for offline consumers).
asyncio.QueueIn-process async mailbox. SSE view awaits queue.get(); another thread/coroutine puts events in.

Django still serves normal sync API views under ASGI — Django 5.2 runs them in a threadpool. Streaming endpoints need the async-capable stack so open SSE connections do not pin sync WSGI workers.

1 · What Server-Sent Events are

Official standard: WHATWG HTML — Server-sent events · tutorial: MDN.

Idea

  • Client opens an HTTP GET to a stream URL.
  • Server responds with Content-Type: text/event-stream and leaves the body open.
  • Server writes discrete events as text; client parses them as they arrive.
  • One direction only: server → client. Client talks back with normal REST/other requests.

Wire format (minimal)

HTTP/1.1 200 OK Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive event: email_categorized data: {"message_id":"abc","_seq_no":12} : this line is a comment (often used as heartbeat) data: {"keepalive":true}
  • Messages end with a blank line.
  • Fields: event: (name), data: (payload; multi-line allowed), id:, retry:.
  • Lines starting with : are comments — browsers ignore them; some mobile libraries drop them before your code sees them (Emailzap heartbeats therefore use a real data: frame for mobile).
  • UTF-8 required.

Browser EventSource vs custom clients

Browser EventSource

  • Built-in reconnect + Last-Event-ID.
  • Cannot set arbitrary headers (no Authorization: Bearer …).
  • Fine for cookie/same-origin pages.

Emailzap clients

  • Chromex: fetch + ReadableStream in offscreen doc (cookies + control).
  • Mobile: react-native-sse shim that can attach Bearer headers.
  • Both implement their own reconnect / watchdog logic.

Proxies must not buffer

If nginx/ALB buffers the response, the client sees nothing until a big chunk fills. Emailzap puts all streams under /api/v1/stream/ with proxy_buffering off. Idle timeouts still kill quiet connections → heartbeats (~30s) keep the path alive under ALB idle limits.

2 · Why SSE (not polling, not WebSockets) here

PollingSSEWebSocket
DirectionClient asks repeatedlyServer pushesFull duplex
ProtocolMany short HTTP callsOne long HTTP responseUpgrade to WS framing
Latency / wasteWorst-case = poll interval; wasted emptiesPush when readyPush; also client→server frames
Fit for Emailzap inboxWas used (e.g. last_synced) — coarseServer events → cache invalidate/patchOverkill: clients already have REST for writes
InfraSimpleNeeds ASGI + non-buffering proxyExtra sticky/upgrade config

Emailzap still uses REST for mutations and some onboarding progress (poll). SSE is the push notification channel for “something changed — refresh / patch cache,” not a replacement for the whole API.

3 · ASGI, Gunicorn, and uvicorn (“unicorn”)

ADR: docs/adrs/sse-framework-asgi-migration.md. Driven by LLM streaming + long-lived event streams.

Problem with sync WSGI workers

  • Sync worker: while an SSE response is open, that worker/thread is occupied.
  • Hundreds of open Gmail/mobile streams ⇒ you run out of workers fast.
  • Async event loop: one process can keep many half-open responses and wake when there is data to write.

What Emailzap actually runs

ALB → nginx (sidecar) → Gunicorn └─ workers = N processes each: uvicorn.workers.UvicornWorker loading execfn.asgi:application
  1. Gunicorn owns process lifecycle and the listen socket.
  2. Each child runs uvicorn’s UvicornWorker → ASGI app + asyncio loop (often with uvloop from uvicorn[standard]).
  3. Django ASGI app serves both normal DRF views (threadpooled if sync) and streaming views that yield SSE frames.
  4. First-party helpers live in backend/execfn/common/sse/ (SSEEvent, SSEMixin, SSEStreamingResponse) — not a Django SSE SaaS SDK.

What happens on one open stream (simplified)

  1. Client GET /api/v1/stream/mailbot/events/ (or /api/v1/stream/inbox/events/) hits some worker W.
  2. UserEventStreamView — or InboxEventStreamView for the relational lane — registers an asyncio.Queue for that user on W. Both share one registry: the key is an opaque string, a Mongo id on the legacy lane and the relational family-primary id on the other.
  3. Generator loops: event = await queue.get() → format as SSE → write to response.
  4. Heartbeats yield empty/keepalive data frames so proxies do not idle-timeout.
  5. Client disconnect → cancel → unregister that queue (multi-connection: other queues for same user may remain).

ASGI solves “many open sockets per process.” It does not share memory across Gunicorn workers. That gap is why Valkey exists.

4 · Why Valkey is needed

Valkey = Redis-compatible in-memory store (AWS ElastiCache Valkey in the multi-worker plan). Client library: redis-py speaking the same protocol. Channel name in code: sse:events.

The multi-worker trap (without a bus)

Browser SSE GET ──────────────────────────► Worker 3 _active_streams[user] = {Queue} Lambda/Celery “push to webapp” (old idea: HTTP POST /internal/push-event/) ALB picks Worker 1 at random _active_streams.get(user) → None event dropped ❌ (~1/N delivery)

Open SSE connection lives only in the process that accepted the GET. A later notify that hits another process cannot see that dict. Documented in docs/plans/sse-multi-worker-fix.md.

Pub/Sub fix

Lambda / Celery / any producer │ notify_* → PUBLISH sse:events {user_id, event, data} ▼ Valkey │ fan-out to every web worker’s subscriber thread ├─► Worker 1 (no local queue for user) ignore ├─► Worker 2 (no local queue) ignore └─► Worker 3 (has Queue) push_event → queue → SSE write ✅

Why not “just sticky sessions” or “one worker”?

  • One worker: simple but single point of failure / capacity cliff.
  • Sticky ALB to SSE GET: does not help producers (Celery/Lambda) that are not the browser — they still need a way to address “whoever holds the socket.”
  • Managed SSE SaaS (Pusher/Ably/AppSync): rejected for this fix — would force chromex/mobile client rewrites + store review (sse-multi-worker-fix.md).

Properties that match Emailzap’s model

  • Best-effort: pub/sub does not store history for offline clients. Missed events → client refetch / focus / gap detection — same philosophy as fire-and-forget notify_*.
  • Cross-runtime: Lambda and Celery can PUBLISH without holding HTTP streams.
  • Local fallback: empty VALKEY_ENDPOINT → direct in-process push_event (single-worker / local).

Pub/sub ≠ Redis Streams / SQS. No consumer group replay. If you need durable ordered logs, that is a different tool — Emailzap deliberately chose fire-and-forget fan-out.

5 · How the three ideas fit together

  1. SSE = the browser/mobile-facing contract (long HTTP response, named events).
  2. uvicorn/ASGI workers = how Django can keep many of those responses open efficiently in each process.
  3. Valkey = how an event created in Lambda/Celery (or another worker) reaches the one process that owns that user’s open socket(s).
[ producers: Lambda · Celery · DRF ] │ packages/sse-contracts notify_* ▼ Valkey PUBLISH │ ▼ [ each Gunicorn+uvicorn worker ] subscriber thread → push_event → asyncio.Queue(s) │ ▼ SSE generator → text/event-stream → chromex / mobile

Next tabs show what is wired vs dead in this repo. Theory above is the “why”; those tabs are the “where.”

Official fundamentals (not Emailzap-specific)

Per WHATWG HTML — Server-sent events and MDN:

Ambiguity resolved as SSE: query said “SSC” once — this artifact documents Server-Sent Events. If you meant something else, say so and we’ll regenerate.

1 · Live push path (wired)

Same Valkey channel for backend Celery + Lambda. Multi-connection per user (chromex + mobile) via dict[str, set[Queue]] (EMA-945).

Lambda producers
triggerGmail webhook / onboarding / drafts / classifiers
↓
↓
Backend producers
triggerCelery tasks / DRF mutations (pin, train, …)
↓
↓
Clients open stream
triggerGmail tab open / mobile foreground
↓
↓
producers publish · clients hold long-lived GET
↓
↓
↓
↓

2 · Dead / superseded paths (do not design as if live)

Early plan docs still mention these. Code search in this worktree found no live implementations.

Planned inline onboarding SSE
Internal HTTP push
SW-resident EventStreamService

Status = presence of wiring in this worktree’s code + project memory. Runtime PostHog flag % and prod Valkey health are not asserted here.

SurfacePathStatusNotes

Backend owns the HTTP SSE endpoint, the in-process queue registry, and Celery/DRF emit sites. Streaming requires ASGI (Gunicorn + uvicorn.workers.UvicornWorker) so one worker can hold many open streams without tying a sync thread each.

ASGI + nginx (from ADR)

Docstring on UserEventStreamView still says events are pushed “via the internal push endpoint”. That endpoint is not present in this worktree; live path is Valkey → subscriber → queue. Treat the docstring as stale.

Lambda never holds browser sockets. It configures the shared notifier at import time and PUBLISHes. No local_push_hook (no in-process streams).

Emit-active Lambdas (this worktree)

Producer families (high level)

Chromex holds one long-lived stream in an offscreen document (MV3). Sidebar is the single useRealtimeEvents subscriber. Legacy handlers are live, and hints are consumed through inbox-core's coalescing invalidator. Lane selection is a one-time probe per open(): it tries the relational path first and on 409 or 404 flips a module global to the legacy path and falls through to the existing retry. 404 tolerance makes deploy ordering safe. Only a 401 re-probes — a 403 must not, because an account the legacy stream 403s can also be 409'd by the relational one, and the two rejections would alternate forever.

Mobile SSE is foreground-only (background → push path). Transport: react-native-sse (third-party EventSource shim) with named-event allowlist. Must register every dispatcher event name or frames are silently dropped. Lane selection reads the migration atom rather than probing: sseUrl(migrated) picks the endpoint, and the atom is a dependency of the auth effect so a mid-session flip tears the connection down and reconnects to the other lane.

packages/sse-contracts is the shared producer protocol (Python, zero non-stdlib deps). packages/inbox-core/src/shared/invalidate.ts is the shared TS consumer: it coalesces hints for 200 ms per QueryClient, dedupes by account:thread, OR-merges facts across the window, then invalidates. Both clients are wired.

Retained legacy events vs invalidation hints (wire)

Retained legacy eventsHints (contract v1.1)
Namesemail_categorized, email_enriched, draft.ready, briefing.pin_changed, sender_category_updated, email.state_changed, email.soft_deleted, email.category_changedemail.created, email.updated
LaneLegacy stream (Mongo producers)Relational stream only — the URL, not the event name, is the version boundary, which is why the names could be reused
PayloadRows / patches a consumer merges{account_id, thread_ids[]} + optional facts; the client refetches
Seq_seq_no attached in the shared notifier path (EMA-985)None (skip_seq). A hint goes stale, never wrong, so there is nothing to sequence, gap-detect, or replay
Batching50-item truncation on the legacy mutation echoesChunked at HINT_MAX_BATCH_SIZE = 50 into consecutive events — nothing is dropped
Missing field—Fail-open: an absent fact means "refresh anyway", so a producer may omit a fact it cannot cheaply determine and new facts need no contract rev

Hint facts

EventFactEffect when present
email.createdbriefing: falseClient skips its briefing refresh — the threads are too new to be there
email.creatednew_sender: trueClient also refreshes contacts
email.updatedcorrelation_idThe client that caused the change can skip its own echo (lands on the wire; suppression deferred)

The full-row v2 protocol is retired, not dormant: EmailRowV2, the *_batch hint names, gap detection, and replay are gone, and ff-sse-v2-emit no longer exists in code. Rows on the wire coupled every producer to a client serializer, and an out-of-order merge leaves a cache that is wrong rather than merely stale — which is precisely what forced the sequencing machinery. email.created fires once per thread lifetime; every later change, including a new message on that thread, is an email.updated.

Third-party vs first-party

PieceLibrary?Evidence
SSE wire framework (Django)first-partybackend/execfn/common/sse/
ASGI serveruvicornpyproject.toml + ADR
Pub/sub brokerValkey / redis-pystream_registry.py, multi-worker plan
Producer schemasfirst-party packagepackages/sse-contracts (stdlib only)
Chromex clientfirst-party fetch + ReadableStreamoffscreen.ts (not browser EventSource — needs cookies/headers control)
Mobile clientreact-native-sse ^1.2.1mobile/package.json + sse-transport-rns.ts
Managed push SaaS (Pusher/Ably/AppSync)rejecteddocs/plans/sse-multi-worker-fix.md
Web frontend/ app SSEno wiring foundgrep under frontend/src

Why ASGI workers matter