feat(realtime): add monotonic cursor, change log and WebSocket fan-out

Messages were only reachable by polling. This adds the server half of the
realtime path: a cursor that cannot skip rows, a change log, and a socket that
pushes "conversation X is at seq N".

The cursor is a counter on conversations, drawn with UPDATE ... RETURNING
inside the mutating transaction. That takes the conversation row lock, so a
second writer blocks until the first commits and commit order equals seq order.
A BIGSERIAL on messages does not have this property: it hands out numbers at
INSERT time while rows become visible at COMMIT time, and those two orders
diverge under concurrent senders — a reader following seq > n then loses a
message permanently. Two tests cover this; the second one asserts that the
naive design *does* lose a row, so the first is proven to measure something.

created_at loses its ordering role. It is NOW(), i.e. transaction start, so it
is neither unique nor commit-ordered — the existing ORDER BY created_at was
already wrong, independently of realtime.

Every mutation appends to conversation_events, and that row is the event. A
three-line trigger on that table fires pg_notify, so no future write path can
forget to signal, and NOTIFY being transactional means it fires exactly when
the row becomes visible.

Fan-out always goes through NOTIFY, even with a single instance. One path
instead of two means the tested path is the production path, and running more
than one instance needs no second implementation. Per connection the pending
work is a map rather than a queue: only the newest seq per conversation
matters, so a newer value overwrites an older one, there is no overflow and no
overflow error path. Entries are merged with max(), because NOTIFY only orders
per listener. When the listener reconnects, every client is told to resync —
notifications sent during the gap are gone, as NOTIFY keeps no history.

Sockets are closed by hand on shutdown: axum's graceful shutdown neither waits
for nor closes upgraded connections (axum#3003, fix PR #3831 still open).

Also adds membership checks to the message history endpoint, which had none,
and paginates it — it previously returned entire conversations unbounded.

Not yet done, marked as TODO in the code: reactions and attachments do not emit
events, and create/update/delete still trust user_id from the request body.

Cargo.lock

@@ -57,6 +57,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
 dependencies = [
  "axum-core",
+ "base64",
  "bytes",
  "form_urlencoded",
  "futures-util",
@@ -75,8 +76,10 @@ dependencies = [
  "serde_json",
  "serde_path_to_error",
  "serde_urlencoded",
+ "sha1 0.10.7",
  "sync_wrapper",
  "tokio",
+ "tokio-tungstenite",
  "tower",
  "tower-layer",
  "tower-service",
@@ -297,6 +300,12 @@ dependencies = [
  "cmov",
 ]
 
+[[package]]
+name = "data-encoding"
+version = "2.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06"
+
 [[package]]
 name = "deranged"
 version = "0.5.8"
@@ -469,6 +478,17 @@ version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed"
 
+[[package]]
+name = "futures-macro"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.34"
@@ -489,6 +509,7 @@ checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc"
 dependencies = [
  "futures-core",
  "futures-io",
+ "futures-macro",
  "futures-sink",
  "futures-task",
  "memchr",
@@ -519,6 +540,18 @@ dependencies = [
  "wasm-bindgen",
 ]
 
+[[package]]
+name = "getrandom"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi 5.3.0",
+ "wasip2",
+]
+
 [[package]]
 name = "getrandom"
 version = "0.4.3"
@@ -527,7 +560,7 @@ checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099"
 dependencies = [
  "cfg-if",
  "libc",
- "r-efi",
+ "r-efi 6.0.0",
  "rand_core 0.10.1",
 ]
 
@@ -1044,6 +1077,15 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
 
+[[package]]
+name = "ppv-lite86"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
+dependencies = [
+ "zerocopy",
+]
+
 [[package]]
 name = "proc-macro2"
 version = "1.0.107"
@@ -1062,12 +1104,28 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
+
 [[package]]
 name = "r-efi"
 version = "6.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
 
+[[package]]
+name = "rand"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41"
+dependencies = [
+ "rand_chacha",
+ "rand_core 0.9.5",
+]
+
 [[package]]
 name = "rand"
 version = "0.10.2"
@@ -1079,6 +1137,16 @@ dependencies = [
  "rand_core 0.10.1",
 ]
 
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.9.5",
+]
+
 [[package]]
 name = "rand_core"
 version = "0.6.4"
@@ -1088,6 +1156,15 @@ dependencies = [
  "getrandom 0.2.17",
 ]
 
+[[package]]
+name = "rand_core"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
+dependencies = [
+ "getrandom 0.3.4",
+]
+
 [[package]]
 name = "rand_core"
 version = "0.10.1"
@@ -1125,12 +1202,14 @@ dependencies = [
  "axum",
  "chrono",
  "dotenv",
+ "futures-util",
  "jsonwebtoken",
  "rand_core 0.6.4",
  "serde",
  "serde_json",
  "sqlx",
  "tokio",
+ "tokio-util",
  "uuid",
 ]
 
@@ -1252,6 +1331,17 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "sha1"
+version = "0.10.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.2.17",
+ "digest 0.10.7",
+]
+
 [[package]]
 name = "sha1"
 version = "0.11.0"
@@ -1457,7 +1547,7 @@ dependencies = [
  "log",
  "percent-encoding",
  "serde",
- "sha1",
+ "sha1 0.11.0",
  "sha2 0.11.0",
  "sqlx-core",
  "thiserror",
@@ -1489,7 +1579,7 @@ dependencies = [
  "log",
  "md-5",
  "memchr",
- "rand",
+ "rand 0.10.2",
  "serde",
  "serde_json",
  "sha2 0.11.0",
@@ -1704,6 +1794,32 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-tungstenite"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite",
+]
+
+[[package]]
+name = "tokio-util"
+version = "0.7.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "futures-util",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tower"
 version = "0.5.3"
@@ -1764,6 +1880,22 @@ dependencies = [
  "once_cell",
 ]
 
+[[package]]
+name = "tungstenite"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8"
+dependencies = [
+ "bytes",
+ "data-encoding",
+ "http",
+ "httparse",
+ "log",
+ "rand 0.9.5",
+ "sha1 0.10.7",
+ "thiserror",
+]
+
 [[package]]
 name = "typenum"
 version = "1.20.1"
@@ -1851,6 +1983,15 @@ version = "0.11.1+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
 
+[[package]]
+name = "wasip2"
+version = "1.0.4+wasi-0.2.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487"
+dependencies = [
+ "wit-bindgen",
+]
+
 [[package]]
 name = "wasm-bindgen"
 version = "0.2.127"
@@ -2052,6 +2193,12 @@ version = "0.52.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
 
+[[package]]
+name = "wit-bindgen"
+version = "0.57.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
+
 [[package]]
 name = "writeable"
 version = "0.6.3"
@@ -2081,6 +2228,26 @@ dependencies = [
  "synstructure",
 ]
 
+[[package]]
+name = "zerocopy"
+version = "0.8.56"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb"
+dependencies = [
+ "zerocopy-derive",
+]
+
+[[package]]
+name = "zerocopy-derive"
+version = "0.8.56"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
 [[package]]
 name = "zerofrom"
 version = "0.1.8"

Cargo.toml

@@ -4,7 +4,9 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-axum = "0.8.9"
+axum = { version = "0.8.9", features = ["ws"] }
+tokio-util = { version = "0.7", features = ["rt"] }
+futures-util = "0.3"
 dotenv = "0.15.0"
 serde_json = "1.0.151"
 tokio = { version = "1", features = ["full"] }

migrations/20260814120000_realtime_cursor.down.sql

@@ -0,0 +1,7 @@
+DROP TRIGGER IF EXISTS conversation_events_notify ON conversation_events;
+DROP FUNCTION IF EXISTS notify_conversation_event();
+DROP TABLE IF EXISTS conversation_events;
+DROP INDEX IF EXISTS messages_conversation_seq_idx;
+ALTER TABLE messages ALTER COLUMN seq DROP NOT NULL;
+ALTER TABLE messages DROP COLUMN IF EXISTS seq;
+ALTER TABLE conversations DROP COLUMN IF EXISTS seq;

migrations/20260814120000_realtime_cursor.up.sql

@@ -0,0 +1,69 @@
+-- Realtime-Grundlage: monotoner Cursor je Conversation + Änderungs-Log.
+--
+-- Warum ein Zähler auf `conversations` und nicht BIGSERIAL auf `messages`:
+-- Eine Sequenz vergibt die Nummer beim INSERT, sichtbar wird die Zeile aber erst
+-- beim COMMIT. Diese beiden Reihenfolgen können auseinanderlaufen — Transaktion A
+-- holt 100, B holt 101 und committet zuerst; ein Leser bei `since = 101` verliert
+-- Nachricht 100 für immer. Der Zähler hier wird per UPDATE gezogen und nimmt damit
+-- den Zeilen-Lock der Conversation: der zweite Schreiber blockiert bis zum COMMIT
+-- des ersten, also ist Commit-Reihenfolge == seq-Reihenfolge.
+
+ALTER TABLE conversations ADD COLUMN seq BIGINT NOT NULL DEFAULT 0;
+
+-- Das Änderungs-Log. Eine Zeile je Mutation — diese Zeile *ist* der Event, der
+-- über den WebSocket geht. Deckt auch ab, was keine eigene Nachrichtenzeile hat
+-- (Reactions, Attachments, später Presence und Read-Receipts).
+CREATE TABLE conversation_events (
+    conversation_id UUID        NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
+    seq             BIGINT      NOT NULL,
+    kind            TEXT        NOT NULL,
+    target_id       UUID        NOT NULL,
+    at              TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (conversation_id, seq)
+);
+-- Der Primärschlüssel bedient bereits `WHERE conversation_id = $1 AND seq > $2
+-- ORDER BY seq` — der Backfill braucht keinen weiteren Index.
+
+-- Unveränderlicher Sortierschlüssel einer Nachricht: die seq ihres
+-- Erzeugungs-Events. Damit verliert `created_at` seine Sortierrolle — es ist NOW(),
+-- also der Transaktionsstart, weder eindeutig noch commit-geordnet.
+ALTER TABLE messages ADD COLUMN seq BIGINT;
+
+-- Bestand nachziehen: nach created_at durchnummerieren, dann den Conversation-
+-- Zähler auf das Maximum setzen.
+WITH nummeriert AS (
+    SELECT id, row_number() OVER (PARTITION BY conversation_id ORDER BY created_at, id) AS n
+    FROM messages
+)
+UPDATE messages m
+SET    seq = nummeriert.n
+FROM   nummeriert
+WHERE  m.id = nummeriert.id;
+
+UPDATE conversations c
+SET    seq = COALESCE((SELECT max(m.seq) FROM messages m WHERE m.conversation_id = c.id), 0);
+
+ALTER TABLE messages ALTER COLUMN seq SET NOT NULL;
+CREATE UNIQUE INDEX messages_conversation_seq_idx ON messages(conversation_id, seq);
+
+-- Für Bestandsnachrichten werden bewusst KEINE Log-Zeilen nachgetragen: ein Client
+-- ohne Cursor bekommt `resync` und lädt über den Historien-Endpunkt, nie über das
+-- Log. Das Log braucht also keine Vergangenheit, nur Zukunft.
+
+-- Der Signalpfad. Der Trigger enthält keine Fachlogik — er feuert nur. Damit kann
+-- kein künftiger Schreibweg das Benachrichtigen vergessen, und weil NOTIFY
+-- transaktional ist, feuert er erst beim COMMIT, also genau dann, wenn die Zeile
+-- auch sichtbar wird.
+CREATE FUNCTION notify_conversation_event() RETURNS TRIGGER AS $$
+BEGIN
+    PERFORM pg_notify(
+        'conversation_changed',
+        json_build_object('conversation', NEW.conversation_id, 'seq', NEW.seq)::text
+    );
+    RETURN NULL;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE TRIGGER conversation_events_notify
+    AFTER INSERT ON conversation_events
+    FOR EACH ROW EXECUTE FUNCTION notify_conversation_event();

src/handlers/message.rs

@@ -1,26 +1,74 @@
 use axum::{
-    extract::{Path, State},
+    extract::{Path, Query, State},
     http::StatusCode,
     response::IntoResponse,
-    Json,
+    Extension, Json,
 };
 use std::sync::Arc;
 use uuid::Uuid;
-use crate::{model::Message, schema::MessageSchema, AppState};
+
 use super::db_err;
+use crate::realtime::{self, MESSAGE_CREATED, MESSAGE_DELETED, MESSAGE_EDITED};
+use crate::{model::Claims, model::Message, schema::MessageSchema, AppState};
+
+// TODO: create/update/delete still trust `user_id` and `conversation_id` from
+// the request body and check no membership. Pre-existing, but the realtime
+// design leans on these endpoints being the security boundary — close this
+// before any instance is exposed.
+
+// Lock ordering: every path that touches both a conversation and one of its
+// messages takes the conversation row lock FIRST (that is what `record_event`
+// does). Doing it the other way round in one handler and this way round in
+// another is how you get deadlocks under concurrent edits.
+
+#[derive(serde::Deserialize)]
+pub struct HistoryQuery {
+    /// Exclusive upper bound, for scrolling back. Omit for the newest page.
+    pub before: Option<i64>,
+    pub limit: Option<i64>,
+}
+
+#[derive(serde::Serialize)]
+pub struct History {
+    /// The conversation's current seq — what a client adopts as its cursor
+    /// after a resync.
+    pub cursor: i64,
+    pub messages: Vec<Message>,
+}
 
 pub async fn list_messages(
     State(state): State<Arc<AppState>>,
+    Extension(claims): Extension<Claims>,
     Path(conversation_id): Path<Uuid>,
+    Query(q): Query<HistoryQuery>,
 ) -> Result<impl IntoResponse, (StatusCode, String)> {
-    let messages = sqlx::query_as::<_, Message>(
+    let user = super::sync::user_id(&claims)?;
-        "SELECT * FROM messages WHERE conversation_id = $1 AND deleted_at IS NULL ORDER BY created_at ASC",
+    super::sync::ensure_member(&state.db, conversation_id, user).await?;
+
+    let limit = q.limit.unwrap_or(50).clamp(1, 200);
+    let cursor: i64 = sqlx::query_scalar("SELECT seq FROM conversations WHERE id = $1")
+        .bind(conversation_id)
+        .fetch_one(&state.db)
+        .await
+        .map_err(db_err)?;
+
+    // Ordered by seq, not created_at: created_at is NOW(), i.e. transaction
+    // start, so under concurrent senders its order can contradict commit order.
+    // Newest page first, then reversed, so `before` walks backwards.
+    let mut messages = sqlx::query_as::<_, Message>(
+        "SELECT * FROM messages
+         WHERE conversation_id = $1 AND deleted_at IS NULL AND ($2::BIGINT IS NULL OR seq < $2)
+         ORDER BY seq DESC LIMIT $3",
     )
     .bind(conversation_id)
+    .bind(q.before)
+    .bind(limit)
     .fetch_all(&state.db)
     .await
     .map_err(db_err)?;
-    Ok(Json(messages))
+    messages.reverse();
+
+    Ok(Json(History { cursor, messages }))
 }
 
 pub async fn get_message(
@@ -40,50 +88,121 @@ pub async fn create_message(
     State(state): State<Arc<AppState>>,
     Json(body): Json<MessageSchema>,
 ) -> Result<impl IntoResponse, (StatusCode, String)> {
-    let message = sqlx::query_as::<_, Message>(
+    let mut tx = state.db.begin().await.map_err(db_err)?;
-        "INSERT INTO messages (conversation_id, user_id, content) VALUES ($1, $2, $3) RETURNING *",
+
+    let seq = realtime::record_event(&mut tx, body.conversation_id, MESSAGE_CREATED, body.id)
+        .await
+        .map_err(db_err)?;
+
+    // The id comes from the client, so a retry after an unclear network error
+    // arrives as the same insert. DO NOTHING turns that into a no-op rather than
+    // a duplicate or a 500.
+    let inserted = sqlx::query_as::<_, Message>(
+        "INSERT INTO messages (id, conversation_id, user_id, content, seq)
+         VALUES ($1, $2, $3, $4, $5)
+         ON CONFLICT (id) DO NOTHING
+         RETURNING *",
     )
+    .bind(body.id)
     .bind(body.conversation_id)
     .bind(body.user_id)
     .bind(&body.content)
+    .bind(seq)
+    .fetch_optional(&mut *tx)
+    .await
+    .map_err(db_err)?;
+
+    match inserted {
+        Some(message) => {
+            tx.commit().await.map_err(db_err)?;
+            Ok((StatusCode::CREATED, Json(message)))
+        }
+        None => {
+            // Roll back so the sequence bump and the log row disappear with it —
+            // a duplicate send must not advance anybody's cursor.
+            tx.rollback().await.map_err(db_err)?;
+            let existing = sqlx::query_as::<_, Message>("SELECT * FROM messages WHERE id = $1")
+                .bind(body.id)
                 .fetch_one(&state.db)
                 .await
                 .map_err(db_err)?;
-    Ok((StatusCode::CREATED, Json(message)))
+            Ok((StatusCode::OK, Json(existing)))
+        }
+    }
 }
 
 pub async fn update_message(
     State(state): State<Arc<AppState>>,
     Path(id): Path<Uuid>,
-    Json(body): Json<MessageSchema>,
+    Json(body): Json<UpdateMessageSchema>,
 ) -> Result<impl IntoResponse, (StatusCode, String)> {
-    sqlx::query_as::<_, Message>(
+    let mut tx = state.db.begin().await.map_err(db_err)?;
+
+    // Read the parent first so the conversation lock is taken before the message
+    // row — see the lock-ordering note at the top.
+    let conversation_id: Option<Uuid> =
+        sqlx::query_scalar("SELECT conversation_id FROM messages WHERE id = $1 AND deleted_at IS NULL")
+            .bind(id)
+            .fetch_optional(&mut *tx)
+            .await
+            .map_err(db_err)?;
+
+    let Some(conversation_id) = conversation_id else {
+        return Err((StatusCode::NOT_FOUND, "Message not found".into()));
+    };
+
+    realtime::record_event(&mut tx, conversation_id, MESSAGE_EDITED, id)
+        .await
+        .map_err(db_err)?;
+
+    // `seq` deliberately stays as it was: an edit moves the message in cursor
+    // space, not in the conversation.
+    let message = sqlx::query_as::<_, Message>(
         "UPDATE messages SET content = $1, updated_at = NOW()
          WHERE id = $2 AND deleted_at IS NULL RETURNING *",
     )
     .bind(&body.content)
     .bind(id)
-    .fetch_optional(&state.db)
+    .fetch_one(&mut *tx)
     .await
-    .map_err(db_err)?
+    .map_err(db_err)?;
-    .map(|m| Json(m).into_response())
+
-    .ok_or((StatusCode::NOT_FOUND, "Message not found".into()))
+    tx.commit().await.map_err(db_err)?;
+    Ok(Json(message))
 }
 
 pub async fn delete_message(
     State(state): State<Arc<AppState>>,
     Path(id): Path<Uuid>,
 ) -> Result<impl IntoResponse, (StatusCode, String)> {
-    let res = sqlx::query(
+    let mut tx = state.db.begin().await.map_err(db_err)?;
-        "UPDATE messages SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL",
+
-    )
+    let conversation_id: Option<Uuid> =
+        sqlx::query_scalar("SELECT conversation_id FROM messages WHERE id = $1 AND deleted_at IS NULL")
             .bind(id)
-    .execute(&state.db)
+            .fetch_optional(&mut *tx)
             .await
             .map_err(db_err)?;
-    if res.rows_affected() > 0 {
+
-        Ok(StatusCode::NO_CONTENT.into_response())
+    let Some(conversation_id) = conversation_id else {
-    } else {
+        return Err((StatusCode::NOT_FOUND, "Message not found".into()));
-        Err((StatusCode::NOT_FOUND, "Message not found".into()))
+    };
-    }
+
+    realtime::record_event(&mut tx, conversation_id, MESSAGE_DELETED, id)
+        .await
+        .map_err(db_err)?;
+
+    sqlx::query("UPDATE messages SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL")
+        .bind(id)
+        .execute(&mut *tx)
+        .await
+        .map_err(db_err)?;
+
+    tx.commit().await.map_err(db_err)?;
+    Ok(StatusCode::NO_CONTENT)
+}
+
+#[derive(serde::Deserialize)]
+pub struct UpdateMessageSchema {
+    pub content: String,
 }

src/handlers/mod.rs

@@ -7,10 +7,12 @@ pub mod conversation;
 pub mod conversation_member;
 pub mod message;
 pub mod reaction;
+pub mod sync;
 pub mod user;
 pub mod workspace;
 pub mod workspace_invitation;
 pub mod workspace_member;
+pub mod ws;
 
 pub(crate) fn db_err(e: sqlx::Error) -> (StatusCode, String) {
     (StatusCode::INTERNAL_SERVER_ERROR, e.to_string())

src/handlers/sync.rs

@@ -0,0 +1,150 @@
+//! Manifest and backfill — the two endpoints a client uses to catch up.
+//!
+//! The WebSocket only ever says "conversation X is now at seq N". Everything
+//! with content is pulled through here, and authorization lives here too: the
+//! event carries no content, so the realtime registry is a delivery index, not a
+//! security boundary. These handlers are the boundary.
+
+use axum::{
+    extract::{Path, Query, State},
+    http::StatusCode,
+    response::IntoResponse,
+    Extension, Json,
+};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use uuid::Uuid;
+
+use super::db_err;
+use crate::{model::Claims, model::Message, AppState};
+
+/// Ticket 06: above this many entities a delta is not worth building — the
+/// client is told to resync instead. Tunable without a protocol change.
+const DELTA_LIMIT: i64 = 500;
+
+pub(crate) fn user_id(claims: &Claims) -> Result<Uuid, (StatusCode, String)> {
+    Uuid::parse_str(&claims.sub)
+        .map_err(|_| (StatusCode::UNAUTHORIZED, "Malformed subject in token".into()))
+}
+
+pub(crate) async fn ensure_member(
+    db: &sqlx::PgPool,
+    conversation_id: Uuid,
+    user: Uuid,
+) -> Result<(), (StatusCode, String)> {
+    let member: Option<Uuid> = sqlx::query_scalar(
+        "SELECT user_id FROM conversation_members WHERE conversation_id = $1 AND user_id = $2",
+    )
+    .bind(conversation_id)
+    .bind(user)
+    .fetch_optional(db)
+    .await
+    .map_err(db_err)?;
+
+    // 404 rather than 403: whether a conversation exists is itself information.
+    member
+        .map(|_| ())
+        .ok_or((StatusCode::NOT_FOUND, "Conversation not found".into()))
+}
+
+#[derive(Serialize, sqlx::FromRow)]
+pub struct ManifestEntry {
+    pub conversation_id: Uuid,
+    pub seq: i64,
+}
+
+/// One cheap query that answers "what has moved since I was last here" for every
+/// conversation at once. Without it a client with forty conversations would fire
+/// forty backfills after every reconnect — and reconnects are routine on mobile.
+pub async fn sync(
+    State(state): State<Arc<AppState>>,
+    Extension(claims): Extension<Claims>,
+) -> Result<impl IntoResponse, (StatusCode, String)> {
+    let user = user_id(&claims)?;
+
+    let entries = sqlx::query_as::<_, ManifestEntry>(
+        "SELECT c.id AS conversation_id, c.seq
+         FROM conversations c
+         JOIN conversation_members m ON m.conversation_id = c.id
+         WHERE m.user_id = $1 AND c.deleted_at IS NULL
+         ORDER BY c.id",
+    )
+    .bind(user)
+    .fetch_all(&state.db)
+    .await
+    .map_err(db_err)?;
+
+    Ok(Json(entries))
+}
+
+#[derive(Deserialize)]
+pub struct ChangesQuery {
+    pub since: i64,
+}
+
+/// `resync` is a normal outcome, not an error — it happens on every first
+/// connection. Sending it as 4xx would run it through `error_for_status()`
+/// reflexes, fill error logs, and let reconnect logic mistake it for a network
+/// failure. Hence one 200 with a discriminated body.
+#[derive(Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum Changes {
+    Delta {
+        cursor: i64,
+        /// Includes tombstones — a client that never sees the soft-deleted row
+        /// never learns about the deletion.
+        messages: Vec<Message>,
+    },
+    Resync {
+        cursor: i64,
+    },
+}
+
+pub async fn changes(
+    State(state): State<Arc<AppState>>,
+    Extension(claims): Extension<Claims>,
+    Path(conversation_id): Path<Uuid>,
+    Query(q): Query<ChangesQuery>,
+) -> Result<impl IntoResponse, (StatusCode, String)> {
+    let user = user_id(&claims)?;
+    ensure_member(&state.db, conversation_id, user).await?;
+
+    let cursor: i64 = sqlx::query_scalar("SELECT seq FROM conversations WHERE id = $1")
+        .bind(conversation_id)
+        .fetch_one(&state.db)
+        .await
+        .map_err(db_err)?;
+
+    // One over the limit, so a full page is distinguishable from an overflow.
+    let targets: Vec<Uuid> = sqlx::query_scalar(
+        "SELECT DISTINCT target_id FROM conversation_events
+         WHERE conversation_id = $1 AND seq > $2
+         ORDER BY target_id
+         LIMIT $3",
+    )
+    .bind(conversation_id)
+    .bind(q.since)
+    .bind(DELTA_LIMIT + 1)
+    .fetch_all(&state.db)
+    .await
+    .map_err(db_err)?;
+
+    if targets.len() as i64 > DELTA_LIMIT {
+        return Ok(Json(Changes::Resync { cursor }));
+    }
+
+    // Resolved current state, not the events themselves: idempotent, and ten
+    // edits of one message collapse into one row instead of ten.
+    // TODO: reactions and attachments once they emit events — today only
+    // message_* kinds exist, so resolving messages covers every target.
+    let messages = sqlx::query_as::<_, Message>(
+        "SELECT * FROM messages WHERE conversation_id = $1 AND id = ANY($2) ORDER BY seq",
+    )
+    .bind(conversation_id)
+    .bind(&targets)
+    .fetch_all(&state.db)
+    .await
+    .map_err(db_err)?;
+
+    Ok(Json(Changes::Delta { cursor, messages }))
+}

src/handlers/ws.rs

@@ -0,0 +1,204 @@
+//! The realtime socket.
+//!
+//! Carries only `{conversation, seq}` downstream — no content ever crosses it,
+//! which is what lets the REST handlers stay the single authorization boundary.
+//!
+//! Auth rides in on the normal `Authorization: Bearer` header, because the
+//! client owns this connection from Rust rather than from a WebView. That also
+//! means real HTTP status codes reach the client on a rejected handshake; a
+//! browser would only ever see close code 1006.
+
+use std::sync::atomic::{AtomicI64, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::{
+    extract::ws::{CloseFrame, Message, WebSocket, WebSocketUpgrade},
+    extract::State,
+    http::StatusCode,
+    response::IntoResponse,
+    Extension,
+};
+use futures_util::{SinkExt, StreamExt};
+use serde::Deserialize;
+use uuid::Uuid;
+
+use super::db_err;
+use crate::hub::{close, Connection, MAX_CONNECTIONS_PER_USER};
+use crate::{model::Claims, AppState};
+
+/// The client pings; we only answer and remember. One direction serves both
+/// sides: the client notices a dead link from the missing pong, we notice it
+/// from the missing ping. 15s keeps us under the strictest documented proxy idle
+/// timeout (Azure Application Gateway, 20s).
+const HEARTBEAT: Duration = Duration::from_secs(15);
+const HEARTBEAT_GRACE: i64 = 2;
+/// How often we check the two deadlines. Cheap, so it can be coarse.
+const TICK: Duration = Duration::from_secs(5);
+
+#[derive(Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+enum Inbound {
+    /// The client refreshed its access token over REST and hands it in here, so
+    /// the session survives without a reconnect. At 15 minutes of token life and
+    /// several connected servers, forced reconnects would otherwise be constant.
+    Reauth { token: String },
+}
+
+pub async fn handler(
+    ws: WebSocketUpgrade,
+    State(state): State<Arc<AppState>>,
+    Extension(claims): Extension<Claims>,
+) -> Result<impl IntoResponse, (StatusCode, String)> {
+    let user_id = super::sync::user_id(&claims)?;
+
+    let conversations: Vec<Uuid> =
+        sqlx::query_scalar("SELECT conversation_id FROM conversation_members WHERE user_id = $1")
+            .bind(user_id)
+            .fetch_all(&state.db)
+            .await
+            .map_err(db_err)?;
+
+    // Displace the oldest rather than refuse the newest: on mobile the OS kills
+    // apps without closing sockets, so refusing would let a dead connection lock
+    // the user out of their own account.
+    if state.hub.connection_count(user_id) >= MAX_CONNECTIONS_PER_USER {
+        if let Some(oldest) = state.hub.oldest_session(user_id) {
+            state.hub.displace(oldest);
+        }
+    }
+
+    Ok(ws.on_upgrade(move |socket| {
+        run(socket, state, user_id, claims.exp as i64, conversations)
+    }))
+}
+
+async fn run(
+    socket: WebSocket,
+    state: Arc<AppState>,
+    user_id: Uuid,
+    token_exp: i64,
+    conversations: Vec<Uuid>,
+) {
+    let session = Uuid::new_v4();
+    let connection = state.hub.register(session, user_id, &conversations);
+
+    let (mut sink, stream) = socket.split();
+    let deadline = Arc::new(AtomicI64::new(token_exp));
+    let last_seen = Arc::new(AtomicI64::new(now()));
+
+    let reader = tokio::spawn(read_loop(
+        stream,
+        connection.clone(),
+        state.clone(),
+        user_id,
+        deadline.clone(),
+        last_seen.clone(),
+    ));
+
+    let mut ticker = tokio::time::interval(TICK);
+    let close_code = loop {
+        tokio::select! {
+            outbound = connection.take() => {
+                let mut failed = false;
+                for message in outbound {
+                    let encoded = match serde_json::to_string(&message) {
+                        Ok(encoded) => encoded,
+                        Err(e) => { eprintln!("ws: could not encode {message:?}: {e}"); continue }
+                    };
+                    if sink.send(Message::Text(encoded.into())).await.is_err() {
+                        failed = true;
+                        break;
+                    }
+                }
+                if failed { break close::RESTART }
+            }
+            _ = ticker.tick() => {
+                if now() >= deadline.load(Ordering::SeqCst) {
+                    break close::TOKEN_EXPIRED;
+                }
+                if now() - last_seen.load(Ordering::SeqCst)
+                    > HEARTBEAT.as_secs() as i64 * HEARTBEAT_GRACE
+                {
+                    break close::RESTART;
+                }
+            }
+            _ = connection.cancel.cancelled() => break connection.close_code(),
+        }
+    };
+
+    // axum 0.8 dropped `WebSocket::close`, so the frame goes out by hand.
+    let _ = sink
+        .send(Message::Close(Some(CloseFrame {
+            code: close_code,
+            reason: reason_for(close_code).into(),
+        })))
+        .await;
+
+    connection.cancel.cancel();
+    reader.abort();
+    state.hub.unregister(session);
+}
+
+async fn read_loop(
+    mut stream: futures_util::stream::SplitStream<WebSocket>,
+    connection: Arc<Connection>,
+    state: Arc<AppState>,
+    user_id: Uuid,
+    deadline: Arc<AtomicI64>,
+    last_seen: Arc<AtomicI64>,
+) {
+    while let Some(received) = stream.next().await {
+        let Ok(message) = received else { break };
+        last_seen.store(now(), Ordering::SeqCst);
+
+        match message {
+            // axum answers Ping itself; seeing it here is what keeps last_seen
+            // fresh.
+            Message::Ping(_) | Message::Pong(_) => {}
+            Message::Close(_) => break,
+            Message::Text(text) => match serde_json::from_str::<Inbound>(&text) {
+                Ok(Inbound::Reauth { token }) => {
+                    match verify(&state, &token, user_id) {
+                        Some(exp) => deadline.store(exp, Ordering::SeqCst),
+                        None => {
+                            connection.close_with(close::UNAUTHORIZED);
+                            return;
+                        }
+                    }
+                }
+                Err(e) => eprintln!("ws: undecodable frame: {e}"),
+            },
+            Message::Binary(_) => {}
+        }
+    }
+    connection.close_with(close::RESTART);
+}
+
+/// A refreshed token must belong to the same user — otherwise a session could be
+/// walked over to another account without reconnecting.
+fn verify(state: &AppState, token: &str, expected: Uuid) -> Option<i64> {
+    let data = jsonwebtoken::decode::<Claims>(
+        token,
+        &jsonwebtoken::DecodingKey::from_secret(state.jwt_secret.as_bytes()),
+        &jsonwebtoken::Validation::default(),
+    )
+    .ok()?;
+    (Uuid::parse_str(&data.claims.sub).ok()? == expected).then_some(data.claims.exp as i64)
+}
+
+fn reason_for(code: u16) -> &'static str {
+    match code {
+        close::TOKEN_EXPIRED => "token expired",
+        close::DISPLACED => "displaced by a newer connection",
+        close::UNAUTHORIZED => "not authorized",
+        _ => "server going away",
+    }
+}
+
+fn now() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs() as i64)
+        .unwrap_or(0)
+}

src/hub.rs

@@ -0,0 +1,325 @@
+//! Fan-out: getting "conversation X moved to seq N" to the right sockets.
+//!
+//! Two things make this simpler than it looks.
+//!
+//! First, only the *newest* seq per conversation matters. Delivering 5, then 6,
+//! then 7 is equivalent to delivering only 7, because the client fetches current
+//! state by cursor either way. So the per-connection buffer is a map, not a
+//! queue: a newer value overwrites an older one, there is no overflow, and
+//! therefore no overflow error path.
+//!
+//! Second, the event carries no content — only an id and a number. Delivery is
+//! an index, not a security boundary; the REST handlers do the authorization.
+//!
+//! Delivery goes through Postgres `NOTIFY` even with a single instance. One path
+//! instead of two means the tested path is the production path, and running more
+//! than one instance needs no second implementation.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, RwLock};
+
+use serde::Serialize;
+use sqlx::postgres::PgListener;
+use sqlx::PgPool;
+use tokio::sync::Notify;
+use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
+
+pub const CHANNEL: &str = "conversation_changed";
+
+/// Ticket 07: generous against the legitimate case — desktop, phone, tablet,
+/// second machine — and still a hard ceiling.
+pub const MAX_CONNECTIONS_PER_USER: usize = 8;
+
+pub type SessionId = Uuid;
+
+/// What the server sends down a socket. Deliberately tiny: about 60 bytes, well
+/// under the 8 kB `NOTIFY` payload limit that shaped this protocol.
+#[derive(Debug, Serialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum Outbound {
+    Changed { conversation: Uuid, seq: i64 },
+    /// "Re-pull the manifest." Sent when the listener reconnected and events in
+    /// the gap are simply gone — NOTIFY has no history.
+    Resync,
+}
+
+/// Close codes. The client maps these onto three behaviours: reconnect now,
+/// reconnect with backoff, or do not reconnect at all. `DISPLACED` is why the
+/// table exists — without an explicit "stay down", a displaced device reconnects
+/// and displaces the next one, and two devices push each other around in circles.
+pub mod close {
+    /// Server going down. Reconnect immediately, with jitter.
+    pub const RESTART: u16 = 1012;
+    /// No valid token any more. Refresh, then reconnect; if the refresh fails,
+    /// send the user to sign-in.
+    pub const TOKEN_EXPIRED: u16 = 4001;
+    /// A newer connection took this slot. Do not reconnect.
+    pub const DISPLACED: u16 = 4002;
+    /// Account gone or not permitted. Do not reconnect.
+    pub const UNAUTHORIZED: u16 = 4003;
+}
+
+pub struct Connection {
+    pub user_id: Uuid,
+    pub cancel: CancellationToken,
+    /// Monotonic registration order, so "displace the oldest" means the oldest
+    /// and not merely some deterministic pick.
+    registered: u64,
+    close_code: AtomicU16,
+    /// conversation -> highest seq not yet written to the socket.
+    dirty: Mutex<HashMap<Uuid, i64>>,
+    resync: AtomicBool,
+    signal: Notify,
+}
+
+impl Connection {
+    pub fn close_code(&self) -> u16 {
+        self.close_code.load(Ordering::SeqCst)
+    }
+
+    pub fn close_with(&self, code: u16) {
+        self.close_code.store(code, Ordering::SeqCst);
+        self.cancel.cancel();
+    }
+
+    /// Blocks until there is something to send, then hands over everything at
+    /// once — coalescing falls out of the map for free.
+    pub async fn take(&self) -> Vec<Outbound> {
+        loop {
+            {
+                let mut dirty = self.dirty.lock().unwrap();
+                let resync = self.resync.swap(false, Ordering::SeqCst);
+                if resync || !dirty.is_empty() {
+                    let mut out = Vec::with_capacity(dirty.len() + 1);
+                    if resync {
+                        // Ordered first: the client throws its cache away, so
+                        // anything queued behind it would be redundant anyway.
+                        out.push(Outbound::Resync);
+                    }
+                    for (conversation, seq) in dirty.drain() {
+                        out.push(Outbound::Changed { conversation, seq });
+                    }
+                    return out;
+                }
+            }
+            self.signal.notified().await;
+        }
+    }
+
+    fn mark(&self, conversation: Uuid, seq: i64) {
+        let mut dirty = self.dirty.lock().unwrap();
+        // max, not insert: NOTIFY guarantees ordering only per listener, so a
+        // blind overwrite could replace a newer seq with an older one.
+        dirty
+            .entry(conversation)
+            .and_modify(|held| *held = (*held).max(seq))
+            .or_insert(seq);
+        drop(dirty);
+        self.signal.notify_one();
+    }
+
+    fn mark_resync(&self) {
+        self.resync.store(true, Ordering::SeqCst);
+        self.signal.notify_one();
+    }
+}
+
+#[derive(Default)]
+pub struct Hub {
+    next_registration: AtomicU64,
+    connections: RwLock<HashMap<SessionId, Arc<Connection>>>,
+    by_conversation: RwLock<HashMap<Uuid, HashSet<SessionId>>>,
+    /// Needed because a *join* cannot be routed by conversation — the joining
+    /// user is not in that set yet, which is the whole point of the join.
+    by_user: RwLock<HashMap<Uuid, HashSet<SessionId>>>,
+}
+
+impl Hub {
+    pub fn connection_count(&self, user_id: Uuid) -> usize {
+        self.by_user
+            .read()
+            .unwrap()
+            .get(&user_id)
+            .map_or(0, |s| s.len())
+    }
+
+    /// The oldest session of this user, for the connection limit. Displacing the
+    /// oldest rather than refusing the newest matters on mobile, where the OS
+    /// kills apps without closing sockets — otherwise a stale connection locks
+    /// the user out of their own account.
+    pub fn oldest_session(&self, user_id: Uuid) -> Option<SessionId> {
+        let by_user = self.by_user.read().unwrap();
+        let sessions = by_user.get(&user_id)?;
+        let connections = self.connections.read().unwrap();
+        sessions
+            .iter()
+            .filter_map(|s| connections.get(s).map(|c| (c.registered, *s)))
+            .min()
+            .map(|(_, session)| session)
+    }
+
+    pub fn register(
+        &self,
+        session: SessionId,
+        user_id: Uuid,
+        conversations: &[Uuid],
+    ) -> Arc<Connection> {
+        let connection = Arc::new(Connection {
+            user_id,
+            cancel: CancellationToken::new(),
+            registered: self.next_registration.fetch_add(1, Ordering::SeqCst),
+            close_code: AtomicU16::new(close::RESTART),
+            dirty: Mutex::new(HashMap::new()),
+            resync: AtomicBool::new(false),
+            signal: Notify::new(),
+        });
+
+        self.connections
+            .write()
+            .unwrap()
+            .insert(session, connection.clone());
+        self.by_user
+            .write()
+            .unwrap()
+            .entry(user_id)
+            .or_default()
+            .insert(session);
+
+        let mut by_conversation = self.by_conversation.write().unwrap();
+        for conversation in conversations {
+            by_conversation
+                .entry(*conversation)
+                .or_default()
+                .insert(session);
+        }
+
+        connection
+    }
+
+    /// Dropped connections are the normal case, not the exception — leaking
+    /// registry entries is the classic bug here.
+    pub fn unregister(&self, session: SessionId) {
+        let removed = self.connections.write().unwrap().remove(&session);
+        let Some(connection) = removed else { return };
+
+        let mut by_user = self.by_user.write().unwrap();
+        if let Some(sessions) = by_user.get_mut(&connection.user_id) {
+            sessions.remove(&session);
+            if sessions.is_empty() {
+                by_user.remove(&connection.user_id);
+            }
+        }
+        drop(by_user);
+
+        let mut by_conversation = self.by_conversation.write().unwrap();
+        by_conversation.retain(|_, sessions| {
+            sessions.remove(&session);
+            !sessions.is_empty()
+        });
+    }
+
+    pub fn displace(&self, session: SessionId) {
+        if let Some(connection) = self.connections.read().unwrap().get(&session) {
+            connection.close_with(close::DISPLACED);
+        }
+    }
+
+    /// Close every socket with 1012 so clients reconnect promptly instead of
+    /// waiting for a timeout. Needed because axum's graceful shutdown does not
+    /// wait for — or close — upgraded connections (axum#3003, PR #3831 open).
+    pub fn close_all(&self) {
+        for connection in self.connections.read().unwrap().values() {
+            connection.close_with(close::RESTART);
+        }
+    }
+
+    fn publish(&self, conversation: Uuid, seq: i64) {
+        let by_conversation = self.by_conversation.read().unwrap();
+        let Some(sessions) = by_conversation.get(&conversation) else {
+            return;
+        };
+        let connections = self.connections.read().unwrap();
+        for session in sessions {
+            if let Some(connection) = connections.get(session) {
+                connection.mark(conversation, seq);
+            }
+        }
+    }
+
+    fn resync_everyone(&self) {
+        for connection in self.connections.read().unwrap().values() {
+            connection.mark_resync();
+        }
+    }
+}
+
+#[derive(serde::Deserialize)]
+struct Notification {
+    conversation: Uuid,
+    seq: i64,
+}
+
+/// Listens to Postgres and feeds the hub. Runs for the life of the process.
+///
+/// `try_recv` returning `Ok(None)` means sqlx transparently re-established the
+/// connection — and notifications sent during that gap are gone for good, since
+/// NOTIFY keeps no history. Every client is therefore told to resync, which is
+/// cheap and rare, and closes the one hole this design would otherwise have.
+pub async fn run_listener(pool: PgPool, hub: Arc<Hub>, cancel: CancellationToken) {
+    loop {
+        let mut listener = match PgListener::connect_with(&pool).await {
+            Ok(listener) => listener,
+            Err(e) => {
+                eprintln!("listener: could not connect: {e}");
+                if wait_or_cancel(&cancel).await {
+                    return;
+                }
+                continue;
+            }
+        };
+
+        if let Err(e) = listener.listen(CHANNEL).await {
+            eprintln!("listener: LISTEN failed: {e}");
+            if wait_or_cancel(&cancel).await {
+                return;
+            }
+            continue;
+        }
+
+        loop {
+            tokio::select! {
+                _ = cancel.cancelled() => return,
+                received = listener.try_recv() => match received {
+                    Ok(Some(notification)) => {
+                        match serde_json::from_str::<Notification>(notification.payload()) {
+                            Ok(n) => hub.publish(n.conversation, n.seq),
+                            Err(e) => eprintln!("listener: undecodable payload: {e}"),
+                        }
+                    }
+                    Ok(None) => {
+                        eprintln!("listener: reconnected — asking every client to resync");
+                        hub.resync_everyone();
+                    }
+                    Err(e) => {
+                        eprintln!("listener: {e}");
+                        break;
+                    }
+                },
+            }
+        }
+
+        if wait_or_cancel(&cancel).await {
+            return;
+        }
+    }
+}
+
+/// Returns true if shutdown was requested while waiting.
+async fn wait_or_cancel(cancel: &CancellationToken) -> bool {
+    tokio::select! {
+        _ = cancel.cancelled() => true,
+        _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => false,
+    }
+}

src/main.rs

@@ -3,8 +3,10 @@ use std::sync::Arc;
 use axum::Router;
 
 mod handlers;
+mod hub;
 mod middleware;
 mod model;
+mod realtime;
 mod route;
 mod schema;
 
@@ -14,6 +16,7 @@ use sqlx::{PgPool, postgres::PgPoolOptions};
 pub struct AppState {
     pub db: PgPool,
     pub jwt_secret: String,
+    pub hub: Arc<hub::Hub>,
 }
 
 #[tokio::main]
@@ -42,12 +45,41 @@ async fn main() {
     let port: String = std::env::var("API_PORT").expect("API_PORT must be set!");
     let addr: String = format!("{}:{}", ip_addr, port);
 
-    let state = Arc::new(AppState { db: pool, jwt_secret });
+    let state = Arc::new(AppState {
+        db: pool.clone(),
+        jwt_secret,
+        hub: Arc::new(hub::Hub::default()),
+    });
+
+    // Fan-out always goes through Postgres NOTIFY, even with a single instance:
+    // one path means the tested path is the production path, and running more
+    // than one instance later needs no second implementation.
+    let shutdown = tokio_util::sync::CancellationToken::new();
+    tokio::spawn(hub::run_listener(
+        pool,
+        state.hub.clone(),
+        shutdown.clone(),
+    ));
+
     let app: Router = Router::new()
         .nest("/api/v1", route::router(state.clone()))
-        .with_state(state);
+        .with_state(state.clone());
 
     let listener = tokio::net::TcpListener::bind(&addr).await.unwrap();
     println!("Server started successfully at {}", addr);
-    axum::serve(listener, app).await.unwrap();
+
+    // axum's graceful shutdown neither closes nor waits for upgraded
+    // connections (axum#3003; fix PR #3831 still open), so the sockets are
+    // closed here by hand before the runtime goes away.
+    let closing = state.clone();
+    axum::serve(listener, app)
+        .with_graceful_shutdown(async move {
+            let _ = tokio::signal::ctrl_c().await;
+            println!("Shutting down — closing realtime connections");
+            closing.hub.close_all();
+            shutdown.cancel();
+            tokio::time::sleep(std::time::Duration::from_millis(250)).await;
+        })
+        .await
+        .unwrap();
 }

src/model.rs

@@ -147,6 +147,10 @@ pub struct Message {
     pub conversation_id: Uuid,
     pub user_id: Uuid,
     pub content: String,
+    /// Immutable sort key within the conversation. Ordering follows this, never
+    /// `created_at` — that is NOW(), i.e. transaction start, so it is neither
+    /// unique nor commit-ordered.
+    pub seq: i64,
     pub created_at: chrono::DateTime<chrono::Utc>,
     pub updated_at: chrono::DateTime<chrono::Utc>,
     pub deleted_at: Option<chrono::DateTime<chrono::Utc>>,

src/realtime.rs

@@ -0,0 +1,235 @@
+//! Cursor allocation and the change log.
+//!
+//! Everything that mutates a conversation goes through `record_event`. It bumps
+//! the conversation's counter and writes the log row in one step, so you cannot
+//! get a sequence number without also emitting the event that goes with it.
+//!
+//! The `UPDATE ... RETURNING` is what makes the cursor safe: it takes the row
+//! lock on the conversation, so a second writer blocks until the first commits.
+//! Commit order therefore equals seq order, and a reader polling `seq > n` can
+//! never skip a row. A plain BIGSERIAL does not have this property — it hands out
+//! numbers at INSERT time, but rows become visible at COMMIT time, and those two
+//! orders can diverge.
+//!
+//! Must be called inside the same transaction as the mutation it describes.
+
+use sqlx::PgConnection;
+use uuid::Uuid;
+
+pub const MESSAGE_CREATED: &str = "message_created";
+pub const MESSAGE_EDITED: &str = "message_edited";
+pub const MESSAGE_DELETED: &str = "message_deleted";
+
+pub async fn record_event(
+    tx: &mut PgConnection,
+    conversation_id: Uuid,
+    kind: &str,
+    target_id: Uuid,
+) -> Result<i64, sqlx::Error> {
+    let seq: i64 =
+        sqlx::query_scalar("UPDATE conversations SET seq = seq + 1 WHERE id = $1 RETURNING seq")
+            .bind(conversation_id)
+            .fetch_one(&mut *tx)
+            .await?;
+
+    // The INSERT fires the pg_notify trigger — but only on COMMIT, since NOTIFY
+    // is transactional. Listeners therefore never see an event before the row.
+    sqlx::query(
+        "INSERT INTO conversation_events (conversation_id, seq, kind, target_id)
+         VALUES ($1, $2, $3, $4)",
+    )
+    .bind(conversation_id)
+    .bind(seq)
+    .bind(kind)
+    .bind(target_id)
+    .execute(&mut *tx)
+    .await?;
+
+    Ok(seq)
+}
+
+#[cfg(test)]
+mod tests {
+    //! Guards the cursor against the gap described above. Needs a running
+    //! Postgres with the migrations applied:
+    //!
+    //!     docker compose up -d db && cargo test realtime
+    //!
+    //! These tests deliberately panic instead of skipping when DATABASE_URL is
+    //! missing — a correctness test that quietly passes without running is worse
+    //! than no test at all.
+
+    use super::*;
+    use sqlx::{postgres::PgPoolOptions, PgPool};
+    use std::time::Duration;
+
+    async fn pool() -> PgPool {
+        dotenv::dotenv().ok();
+        let url = std::env::var("DATABASE_URL")
+            .expect("DATABASE_URL must be set — these tests need a real Postgres");
+        PgPoolOptions::new()
+            .max_connections(5)
+            .connect(&url)
+            .await
+            .expect("could not connect to Postgres")
+    }
+
+    /// A fresh conversation with an owner, so the tests do not interfere.
+    async fn fixture(pool: &PgPool) -> Uuid {
+        let user: Uuid = sqlx::query_scalar(
+            "INSERT INTO users (email, password_hash, username, display_name)
+             VALUES ($1, 'x', $2, 'Probe') RETURNING id",
+        )
+        .bind(format!("probe-{}@example.invalid", Uuid::new_v4()))
+        .bind(format!("probe-{}", Uuid::new_v4()))
+        .fetch_one(pool)
+        .await
+        .unwrap();
+
+        sqlx::query_scalar(
+            "INSERT INTO conversations (kind, name, created_by)
+             VALUES ('channel', 'cursor probe', $1) RETURNING id",
+        )
+        .bind(user)
+        .fetch_one(pool)
+        .await
+        .unwrap()
+    }
+
+    /// Everything visible after `since`, the way the backfill endpoint reads it.
+    async fn read_since(pool: &PgPool, conversation: Uuid, since: i64) -> Vec<i64> {
+        sqlx::query_scalar(
+            "SELECT seq FROM conversation_events
+             WHERE conversation_id = $1 AND seq > $2 ORDER BY seq",
+        )
+        .bind(conversation)
+        .bind(since)
+        .fetch_all(pool)
+        .await
+        .unwrap()
+    }
+
+    /// Two senders overlap: A starts first and commits last. A reader polling
+    /// with a cursor in between must still end up seeing both.
+    #[tokio::test]
+    async fn concurrent_senders_leave_no_gap() {
+        let pool = pool().await;
+        let conversation = fixture(&pool).await;
+
+        let mut a = pool.begin().await.unwrap();
+        let a_seq = record_event(&mut a, conversation, MESSAGE_CREATED, Uuid::new_v4())
+            .await
+            .unwrap();
+
+        // B runs concurrently and will block on the conversation row lock until
+        // A commits. That blocking IS the mechanism under test.
+        let pool_b = pool.clone();
+        let b = tokio::spawn(async move {
+            let mut tx = pool_b.begin().await.unwrap();
+            let seq = record_event(&mut tx, conversation, MESSAGE_CREATED, Uuid::new_v4())
+                .await
+                .unwrap();
+            tx.commit().await.unwrap();
+            seq
+        });
+
+        // Give B time to actually reach the lock.
+        tokio::time::sleep(Duration::from_millis(300)).await;
+
+        // Nothing is committed yet, so a reader sees nothing — in particular it
+        // cannot advance its cursor past A.
+        assert!(
+            read_since(&pool, conversation, 0).await.is_empty(),
+            "reader saw an uncommitted event"
+        );
+
+        a.commit().await.unwrap();
+        let b_seq = b.await.unwrap();
+
+        assert_eq!(a_seq, 1, "A committed first, so it must hold the lower seq");
+        assert_eq!(b_seq, 2);
+
+        // Read incrementally, exactly like a client following its cursor.
+        let mut cursor = 0i64;
+        let mut seen = Vec::new();
+        loop {
+            let batch = read_since(&pool, conversation, cursor).await;
+            if batch.is_empty() {
+                break;
+            }
+            cursor = *batch.last().unwrap();
+            seen.extend(batch);
+        }
+        assert_eq!(seen, vec![1, 2], "a message was skipped by the cursor");
+    }
+
+    /// The same interleaving against a plain BIGSERIAL — the obvious design.
+    /// This one MUST show the gap; if it ever stops failing to deliver a row,
+    /// the test above is no longer proving anything.
+    #[tokio::test]
+    async fn naive_sequence_does_lose_a_row() {
+        let pool = pool().await;
+        let probe = Uuid::new_v4();
+
+        sqlx::query(
+            "CREATE TABLE IF NOT EXISTS naive_cursor_probe (
+                 seq BIGSERIAL PRIMARY KEY, probe UUID NOT NULL)",
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+
+        let mut a = pool.begin().await.unwrap();
+        let a_seq: i64 =
+            sqlx::query_scalar("INSERT INTO naive_cursor_probe (probe) VALUES ($1) RETURNING seq")
+                .bind(probe)
+                .fetch_one(&mut *a)
+                .await
+                .unwrap();
+
+        // No lock is taken, so B sails past A and gets the higher number.
+        let mut b = pool.begin().await.unwrap();
+        let b_seq: i64 =
+            sqlx::query_scalar("INSERT INTO naive_cursor_probe (probe) VALUES ($1) RETURNING seq")
+                .bind(probe)
+                .fetch_one(&mut *b)
+                .await
+                .unwrap();
+        b.commit().await.unwrap();
+
+        assert!(a_seq < b_seq);
+
+        // The reader polls while A is still open, sees only B, and advances its
+        // cursor past A's number.
+        let read = |since: i64| {
+            let pool = pool.clone();
+            async move {
+                sqlx::query_scalar::<_, i64>(
+                    "SELECT seq FROM naive_cursor_probe WHERE probe = $1 AND seq > $2 ORDER BY seq",
+                )
+                .bind(probe)
+                .bind(since)
+                .fetch_all(&pool)
+                .await
+                .unwrap()
+            }
+        };
+
+        let first = read(0).await;
+        assert_eq!(first, vec![b_seq]);
+        let cursor = *first.last().unwrap();
+
+        a.commit().await.unwrap();
+
+        assert!(
+            read(cursor).await.is_empty(),
+            "expected the naive design to lose A — it did not, so this test no longer \
+             demonstrates the failure it exists to demonstrate"
+        );
+
+        sqlx::query("DROP TABLE naive_cursor_probe")
+            .execute(&pool)
+            .await
+            .ok();
+    }
+}

src/route.rs

@@ -1,6 +1,6 @@
 use crate::handlers::{
-    attachment, auth, auth_token, conversation, conversation_member, message, reaction, user,
+    attachment, auth, auth_token, conversation, conversation_member, message, reaction, sync, user,
-    workspace, workspace_invitation, workspace_member,
+    workspace, workspace_invitation, workspace_member, ws,
 };
 use crate::middleware::require_auth;
 use crate::AppState;
@@ -81,6 +81,13 @@ pub fn router(state: Arc<AppState>) -> Router<Arc<AppState>> {
             "/conversations/{conversation_id}/members/{user_id}",
             delete(conversation_member::remove_member),
         )
+        // Realtime. The socket carries only {conversation, seq}; everything with
+        // content is pulled through the two endpoints below.
+        .route("/ws", get(ws::handler))
+        // Catch-up: one cheap manifest for every conversation, then targeted
+        // backfills for the few that actually moved.
+        .route("/sync", get(sync::sync))
+        .route("/conversations/{id}/changes", get(sync::changes))
         // Messages
         .route("/conversations/{id}/messages", get(message::list_messages))
         .route("/messages", post(message::create_message))

src/schema.rs

@@ -81,6 +81,12 @@ pub struct ConversationMemberSchema {
 
 #[derive(Deserialize, Serialize, Debug)]
 pub struct MessageSchema {
+    /// Supplied by the client, not the server. Without it the client cannot
+    /// reconcile its optimistically rendered message with the one that comes
+    /// back over the realtime channel — it would show up twice. Bonus: sending
+    /// becomes idempotent, so a retry after an unclear network error is a no-op
+    /// instead of a duplicate.
+    pub id: Uuid,
     pub conversation_id: Uuid,
     pub user_id: Uuid,
     pub content: String,