Real-Time (WebSocket)

The realtime library provides WebSocket-based real-time communication with JWT authentication, message routing, and topic-based pub/sub.

Connection lifecycle

Client connects with JWT: ws://host/ws?token=<jwt>
Server verifies JWT via the IJWTVerifier the application supplies
Connection record created and registered
Client sends and receives messages
On disconnect: cleanup registry, unsubscribe from all topics

Message routing types

Type Target Use case

Type	Target	Use case
`:broadcast`	All connections	System announcements
`:user`	Specific user-id	Direct messages
`:role`	Users with a specific role	Admin notifications
`:connection`	Specific connection-id	Job progress updates

:broadcast

All connections

System announcements

:user

Specific user-id

Direct messages

:role

Users with a specific role

Admin notifications

:connection

Specific connection-id

Job progress updates

Sending messages

(require '[wagoe.realtime.ports :as ports])

;; Broadcast to all
(ports/broadcast service {:type :announcement :text "System maintenance in 5 minutes"})

;; Send to specific user
(ports/send-to-user service user-id {:type :notification :data {...}})

;; Send to all users with a role
(ports/send-to-role service :admin {:type :alert :data {...}})

;; Send to specific connection
(ports/send-to-connection service conn-id {:type :progress :percent 75})

Topic-based pub/sub

;; Subscribe connection to a topic
(ports/subscribe-to-topic pubsub-mgr conn-id "order:123")

;; Publish to all topic subscribers
(ports/publish-to-topic service "order:123" {:type :order-updated :payload {...}})

;; Unsubscribe
(ports/unsubscribe-from-topic pubsub-mgr conn-id "order:123")

Live htmx fragments (server-sent events)

Push rendered Hiccup to a page without writing JavaScript. The stream rides the web session, so mount it as an ordinary web route, not :public. A client names its topics in the URL; :authorize-topic decides which it may have, and without it every topic is refused.

(require '[wagoe.realtime.shell.handlers.sse :as sse])

;; route, in the module's :web routes
["/orders/events"
 {:get {:handler (sse/sse-handler svc
                                  :authorize-topic (fn [request topic]
                                                     (may-see-order? (:user request) topic)))}}]

;; page
[:div {:hx-ext "sse" :sse-connect "/web/orders/events?topic=order:123"}
 [:table [:tbody#order-rows {:sse-swap "message" :hx-swap "beforeend"}]]]

;; from a handler, a job, any replica
(sse/push-fragment! svc "order:123" [:tr [:td "Shipped"]])
(sse/push-fragment! svc "order:123" [:span#badge [:t :orders/count {:n 4}]] {:event "badge"})

[:t …] markers resolve in each subscriber’s locale. An element with :hx-swap-oob swaps by id, so one push can update several places. Under the :redis provider fragments cross the wire: keep them plain data.

Each open stream holds one server thread for as long as the page is open, and writes a heartbeat every 15 seconds; a closed tab is cleaned up at the next one. The :wagoe/realtime key caps streams per node with :max-streams (default 32, under Jetty’s 50 threads); a stream over the cap is told to retry in 10 seconds. A handler accepts at most :max-topics (16) topics per stream.

An open stream does not see a logout by itself. Pass :revalidate, asked every heartbeat, and the stream closes once it answers false; the browser reconnects and gets a 401:

(require '[wagoe.user.ports :as user-ports]
         '[wagoe.user.shell.middleware :as user-mw])

(sse/sse-handler svc
                 :authorize-topic may-see-topic?
                 :revalidate (fn [request]
                               (when-let [session (some->> (user-mw/extract-session-token request)
                                                           (user-ports/validate-session user-service))]
                                 (:active (user-ports/get-user-by-id user-service (:user-id session))))))

validate-session also refreshes the session’s last access, so an open page keeps a sliding session alive.

Integrant configuration

:wagoe/realtime
{:provider     :memory              ; or :redis, to share topics across replicas
 :max-streams  32                   ; SSE streams per node
 :jwt-verifier #ig/ref :my-app/jwt-verifier} ; your IJWTVerifier; realtime ships none

;; :redis also takes :host, :port, :password, :database, :channel and :key-prefix

See realtime library for the full reference.