diff --git a/Cargo.lock b/Cargo.lock index afe0c53..1add51c 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index d0e183d..7c3822e 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/migrations/20260814120000_realtime_cursor.down.sql b/migrations/20260814120000_realtime_cursor.down.sql new file mode 100644 index 0000000..68b751b --- /dev/null +++ b/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; diff --git a/migrations/20260814120000_realtime_cursor.up.sql b/migrations/20260814120000_realtime_cursor.up.sql new file mode 100644 index 0000000..de1337f --- /dev/null +++ b/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(); diff --git a/src/handlers/message.rs b/src/handlers/message.rs index b293d31..e20b31c 100644 --- a/src/handlers/message.rs +++ b/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, + pub limit: Option, +} + +#[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, +} pub async fn list_messages( State(state): State>, + Extension(claims): Extension, Path(conversation_id): Path, + Query(q): Query, ) -> Result { - let messages = sqlx::query_as::<_, Message>( - "SELECT * FROM messages WHERE conversation_id = $1 AND deleted_at IS NULL ORDER BY created_at ASC", + let user = super::sync::user_id(&claims)?; + 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>, Json(body): Json, ) -> Result { - let message = sqlx::query_as::<_, Message>( - "INSERT INTO messages (conversation_id, user_id, content) VALUES ($1, $2, $3) RETURNING *", + let mut tx = state.db.begin().await.map_err(db_err)?; + + 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) - .fetch_one(&state.db) + .bind(seq) + .fetch_optional(&mut *tx) .await .map_err(db_err)?; - Ok((StatusCode::CREATED, Json(message))) + + 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::OK, Json(existing))) + } + } } pub async fn update_message( State(state): State>, Path(id): Path, - Json(body): Json, + Json(body): Json, ) -> Result { - 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 = + 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(|m| Json(m).into_response()) - .ok_or((StatusCode::NOT_FOUND, "Message not found".into())) + .map_err(db_err)?; + + tx.commit().await.map_err(db_err)?; + Ok(Json(message)) } pub async fn delete_message( State(state): State>, Path(id): Path, ) -> Result { - let res = sqlx::query( - "UPDATE messages SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL", - ) - .bind(id) - .execute(&state.db) - .await - .map_err(db_err)?; - if res.rows_affected() > 0 { - Ok(StatusCode::NO_CONTENT.into_response()) - } else { - Err((StatusCode::NOT_FOUND, "Message not found".into())) - } + let mut tx = state.db.begin().await.map_err(db_err)?; + + let conversation_id: Option = + 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_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, } diff --git a/src/handlers/mod.rs b/src/handlers/mod.rs index f92511a..3be8a4a 100644 --- a/src/handlers/mod.rs +++ b/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()) diff --git a/src/handlers/sync.rs b/src/handlers/sync.rs new file mode 100644 index 0000000..aad5c2c --- /dev/null +++ b/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::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 = 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>, + Extension(claims): Extension, +) -> Result { + 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, + }, + Resync { + cursor: i64, + }, +} + +pub async fn changes( + State(state): State>, + Extension(claims): Extension, + Path(conversation_id): Path, + Query(q): Query, +) -> Result { + 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 = 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 })) +} diff --git a/src/handlers/ws.rs b/src/handlers/ws.rs new file mode 100644 index 0000000..bb7c71e --- /dev/null +++ b/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>, + Extension(claims): Extension, +) -> Result { + let user_id = super::sync::user_id(&claims)?; + + let conversations: Vec = + 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, + user_id: Uuid, + token_exp: i64, + conversations: Vec, +) { + 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, + connection: Arc, + state: Arc, + user_id: Uuid, + deadline: Arc, + last_seen: Arc, +) { + 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::(&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 { + let data = jsonwebtoken::decode::( + 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) +} diff --git a/src/hub.rs b/src/hub.rs new file mode 100644 index 0000000..9709135 --- /dev/null +++ b/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>, + 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 { + 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>>, + by_conversation: RwLock>>, + /// 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>>, +} + +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 { + 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 { + 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, 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.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, + } +} diff --git a/src/main.rs b/src/main.rs index d66baa2..4996de5 100644 --- a/src/main.rs +++ b/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, } #[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(); } diff --git a/src/model.rs b/src/model.rs index 45b7016..f62f2ff 100644 --- a/src/model.rs +++ b/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, pub updated_at: chrono::DateTime, pub deleted_at: Option>, diff --git a/src/realtime.rs b/src/realtime.rs new file mode 100644 index 0000000..69ce5c4 --- /dev/null +++ b/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 { + 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 { + 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(); + } +} diff --git a/src/route.rs b/src/route.rs index 9d431dc..c47c84c 100644 --- a/src/route.rs +++ b/src/route.rs @@ -1,6 +1,6 @@ use crate::handlers::{ - attachment, auth, auth_token, conversation, conversation_member, message, reaction, user, - workspace, workspace_invitation, workspace_member, + attachment, auth, auth_token, conversation, conversation_member, message, reaction, sync, user, + workspace, workspace_invitation, workspace_member, ws, }; use crate::middleware::require_auth; use crate::AppState; @@ -81,6 +81,13 @@ pub fn router(state: Arc) -> Router> { "/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)) diff --git a/src/schema.rs b/src/schema.rs index 0fd17d8..66f591b 100644 --- a/src/schema.rs +++ b/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,