From ec3771f2e75de938960d7bb823cc57c5be8cfb10 Mon Sep 17 00:00:00 2001 From: Bendik Aagaard Lynghaug Date: Wed, 8 Jul 2026 22:14:26 +0200 Subject: [PATCH] Persist chat history to Postgres via durable JetStream consumer DATABASE_URL required at startup; schema self-initializes. Compose gains a postgres service for standalone dev. Co-Authored-By: Claude Fable 5 --- .env.example | 3 + Cargo.lock | 428 ++++++++++++++++++++++++++++++++++++++++++-- Cargo.toml | 7 + README.md | 22 ++- docker-compose.yml | 25 ++- src/app.rs | 27 ++- src/chat.rs | 36 +++- src/main.rs | 15 +- src/server/mod.rs | 2 + src/server/store.rs | 152 ++++++++++++++++ 10 files changed, 696 insertions(+), 21 deletions(-) create mode 100644 src/server/store.rs diff --git a/.env.example b/.env.example index 8e3202a..e740122 100644 --- a/.env.example +++ b/.env.example @@ -3,6 +3,9 @@ # NATS server the chat server publishes/subscribes on. NATS_URL=nats://127.0.0.1:4222 +# Postgres archive for message history (written by the JetStream consumer). +DATABASE_URL=postgres://cnats:cnats@127.0.0.1:5432/cnats + # Base URL of your Kanidm instance (no trailing slash). KANIDM_URL=https://idm.example.com diff --git a/Cargo.lock b/Cargo.lock index 48228c8..b583190 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,6 +11,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -102,6 +108,15 @@ dependencies = [ "syn", ] +[[package]] +name = "atoi" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f28d99ec8bfea296261ca1af174f24225171fea9664ba9003cbebee704810528" +dependencies = [ + "num-traits", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -260,6 +275,12 @@ version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.12.1" @@ -340,6 +361,7 @@ dependencies = [ "openidconnect", "serde", "serde_json", + "sqlx", "tokio", "tower", "tower-http", @@ -521,6 +543,30 @@ dependencies = [ "libc", ] +[[package]] +name = "crc" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d" +dependencies = [ + "crc-catalog", +] + +[[package]] +name = "crc-catalog" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "217698eaf96b4a3f0bc4f3662aaa55bdf913cd54d7204591faa790070c6d0853" + +[[package]] +name = "crossbeam-queue" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "803d13fb3b09d88be9f4dbc29062c66b19bf7170867ceb746d2a8689bf6c7a26" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.22" @@ -732,6 +778,9 @@ name = "either" version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +dependencies = [ + "serde", +] [[package]] name = "either_of" @@ -795,6 +844,17 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "etcetera" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "136d1b5283a1ab77bd9257427ffd09d8667ced0570b6f938942bc7568ed5b943" +dependencies = [ + "cfg-if", + "home", + "windows-sys 0.48.0", +] + [[package]] name = "event-listener" version = "5.4.1" @@ -838,6 +898,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -889,6 +955,17 @@ dependencies = [ "futures-util", ] +[[package]] +name = "futures-intrusive" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f" +dependencies = [ + "futures-core", + "lock_api", + "parking_lot", +] + [[package]] name = "futures-io" version = "0.3.32" @@ -1042,12 +1119,38 @@ version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash", +] + [[package]] name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "hashlink" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1" +dependencies = [ + "hashbrown 0.15.5", +] + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + [[package]] name = "hex" version = "0.4.3" @@ -1072,6 +1175,15 @@ dependencies = [ "digest", ] +[[package]] +name = "home" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "html-escape" version = "0.2.14" @@ -1178,7 +1290,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots", + "webpki-roots 1.0.8", ] [[package]] @@ -1674,6 +1786,18 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "libredox" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c943259e342f1e06ff2da7a83eabdfe7f92ce10262688dbf1895ff0b3e6e4652" +dependencies = [ + "bitflags", + "libc", + "plain", + "redox_syscall 0.9.0", +] + [[package]] name = "litemap" version = "0.8.2" @@ -1740,6 +1864,16 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" +[[package]] +name = "md-5" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf" +dependencies = [ + "cfg-if", + "digest", +] + [[package]] name = "memchr" version = "2.8.3" @@ -2022,7 +2156,7 @@ checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" dependencies = [ "cfg-if", "libc", - "redox_syscall", + "redox_syscall 0.5.18", "smallvec", "windows-link", ] @@ -2101,6 +2235,12 @@ dependencies = [ "spki", ] +[[package]] +name = "plain" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4596b6d070b27117e987119b4dac604f3c58cfb0b191112e24771b2faeac1a6" + [[package]] name = "portable-atomic" version = "1.13.1" @@ -2452,6 +2592,15 @@ dependencies = [ "bitflags", ] +[[package]] +name = "redox_syscall" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c5102a6aaa05aa011a238e178e6bca86d2cb56fc9f586d37cb80f5bca6e07759" +dependencies = [ + "bitflags", +] + [[package]] name = "ref-cast" version = "1.0.25" @@ -2536,7 +2685,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots", + "webpki-roots 1.0.8", ] [[package]] @@ -3129,6 +3278,9 @@ name = "smallvec" version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +dependencies = [ + "serde", +] [[package]] name = "socket2" @@ -3156,12 +3308,143 @@ dependencies = [ "der", ] +[[package]] +name = "sqlx" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fefb893899429669dcdd979aff487bd78f4064e5e7907e4269081e0ef7d97dc" +dependencies = [ + "sqlx-core", + "sqlx-macros", + "sqlx-postgres", +] + +[[package]] +name = "sqlx-core" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6" +dependencies = [ + "base64 0.22.1", + "bytes", + "crc", + "crossbeam-queue", + "either", + "event-listener", + "futures-core", + "futures-intrusive", + "futures-io", + "futures-util", + "hashbrown 0.15.5", + "hashlink", + "indexmap 2.14.0", + "log", + "memchr", + "once_cell", + "percent-encoding", + "rustls", + "serde", + "serde_json", + "sha2", + "smallvec", + "thiserror 2.0.18", + "tokio", + "tokio-stream", + "tracing", + "url", + "webpki-roots 0.26.11", +] + +[[package]] +name = "sqlx-macros" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2d452988ccaacfbf5e0bdbc348fb91d7c8af5bee192173ac3636b5fb6e6715d" +dependencies = [ + "proc-macro2", + "quote", + "sqlx-core", + "sqlx-macros-core", + "syn", +] + +[[package]] +name = "sqlx-macros-core" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19a9c1841124ac5a61741f96e1d9e2ec77424bf323962dd894bdb93f37d5219b" +dependencies = [ + "dotenvy", + "either", + "heck", + "hex", + "once_cell", + "proc-macro2", + "quote", + "serde", + "serde_json", + "sha2", + "sqlx-core", + "sqlx-postgres", + "syn", + "tokio", + "url", +] + +[[package]] +name = "sqlx-postgres" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46" +dependencies = [ + "atoi", + "base64 0.22.1", + "bitflags", + "byteorder", + "crc", + "dotenvy", + "etcetera", + "futures-channel", + "futures-core", + "futures-util", + "hex", + "hkdf", + "hmac", + "home", + "itoa", + "log", + "md-5", + "memchr", + "once_cell", + "rand 0.8.6", + "serde", + "serde_json", + "sha2", + "smallvec", + "sqlx-core", + "stringprep", + "thiserror 2.0.18", + "tracing", + "whoami", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "stringprep" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1" +dependencies = [ + "unicode-bidi", + "unicode-normalization", + "unicode-properties", +] + [[package]] name = "strsim" version = "0.11.1" @@ -3399,6 +3682,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.29.0" @@ -3725,12 +4019,33 @@ version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" +[[package]] +name = "unicode-bidi" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" + [[package]] name = "unicode-ident" version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-normalization" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fd4f6878c9cb28d874b009da9e8d183b5abc80117c40bbd187a1fde336be6e8" +dependencies = [ + "tinyvec", +] + +[[package]] +name = "unicode-properties" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" + [[package]] name = "unicode-segmentation" version = "1.13.3" @@ -3825,6 +4140,12 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "wasite" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b" + [[package]] name = "wasm-bindgen" version = "0.2.126" @@ -3935,6 +4256,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.8", +] + [[package]] name = "webpki-roots" version = "1.0.8" @@ -3944,6 +4274,16 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "whoami" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d4a4db5077702ca3015d3d02d74974948aba2ad9e12ab7df718ee64ccd7e97d" +dependencies = [ + "libredox", + "wasite", +] + [[package]] name = "winapi-util" version = "0.1.11" @@ -4012,13 +4352,22 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" +dependencies = [ + "windows-targets 0.48.5", +] + [[package]] name = "windows-sys" version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets", + "windows-targets 0.52.6", ] [[package]] @@ -4030,34 +4379,67 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c" +dependencies = [ + "windows_aarch64_gnullvm 0.48.5", + "windows_aarch64_msvc 0.48.5", + "windows_i686_gnu 0.48.5", + "windows_i686_msvc 0.48.5", + "windows_x86_64_gnu 0.48.5", + "windows_x86_64_gnullvm 0.48.5", + "windows_x86_64_msvc 0.48.5", +] + [[package]] name = "windows-targets" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", + "windows_aarch64_gnullvm 0.52.6", + "windows_aarch64_msvc 0.52.6", + "windows_i686_gnu 0.52.6", "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", + "windows_i686_msvc 0.52.6", + "windows_x86_64_gnu 0.52.6", + "windows_x86_64_gnullvm 0.52.6", + "windows_x86_64_msvc 0.52.6", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8" + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" +[[package]] +name = "windows_aarch64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" +[[package]] +name = "windows_i686_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e" + [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -4070,24 +4452,48 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" +[[package]] +name = "windows_i686_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406" + [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" +[[package]] +name = "windows_x86_64_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc" + [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" +[[package]] +name = "windows_x86_64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index 50c47eb..c5287ff 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,6 +21,12 @@ tower = { version = "0.5", optional = true } tower-http = { version = "0.6", features = ["fs", "trace"], optional = true } tower-sessions = { version = "0.14", optional = true } async-nats = { version = "0.38", optional = true } +sqlx = { version = "0.8", default-features = false, features = [ + "runtime-tokio", + "tls-rustls", + "postgres", + "macros", +], optional = true } openidconnect = { version = "4", optional = true } futures = { version = "0.3", optional = true } chrono = { version = "0.4", features = ["serde"], optional = true } @@ -59,6 +65,7 @@ ssr = [ "dep:tower-http", "dep:tower-sessions", "dep:async-nats", + "dep:sqlx", "dep:openidconnect", "dep:futures", "dep:chrono", diff --git a/README.md b/README.md index f81903e..feb62cd 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,8 @@ Axum) where every room is a **NATS subject** and sign-in is **Kanidm SSO** browser ──(server fn POST)──▶ axum ──publish──▶ NATS chat.room. browser ◀──(SSE /sse/)── axum ◀─subscribe── NATS chat.room. browser ◀──(302 /auth/*)────── axum ◀──OIDC──▶ Kanidm + axum ◀─JetStream (CHAT / cnats-postgres)─ NATS + └──INSERT──▶ Postgres (history archive) ``` The browser never talks to NATS directly: the server publishes on behalf of the @@ -61,6 +63,7 @@ cp .env.example .env # then edit | Variable | Meaning | | --- | --- | | `NATS_URL` | NATS server, default `nats://127.0.0.1:4222` | +| `DATABASE_URL` | Postgres for message history, default `postgres://cnats:cnats@127.0.0.1:5432/cnats` | | `KANIDM_URL` | Base URL of Kanidm, e.g. `https://idm.example.com` | | `OAUTH2_CLIENT_ID` | The Kanidm oauth2 client name (`cnats` above) | | `OAUTH2_CLIENT_SECRET` | Output of `show-basic-secret` | @@ -70,18 +73,23 @@ cp .env.example .env # then edit ## Run ```bash -docker compose up -d nats +docker compose up -d nats postgres cargo leptos watch # dev, auto-reload on http://127.0.0.1:3000 cargo leptos build --release # production build (binary + target/site) ``` +Or run the whole stack (app image included) with `docker compose up`. + Open two browser windows, sign in, and chat — or join from the CLI: ```bash nats sub 'chat.room.*' -nats pub chat.room.lobby '{"id":"cli-1","room":"lobby","username":"cli","display_name":"CLI","text":"hello from the bus","time":"12:00:00"}' +nats pub chat.room.lobby '{"id":"cli-1","room":"lobby","username":"cli","display_name":"CLI","text":"hello from the bus","time":"12:00:00","ts":0}' ``` +CLI-published messages are archived too — the JetStream consumer sees +everything on `chat.room.*`, not just what this app publishes. + ## Layout ``` @@ -94,6 +102,7 @@ src/ server/ oidc.rs Kanidm OIDC login/callback/logout handlers sse.rs NATS → browser SSE bridge (one subscription per client) + store.rs JetStream → Postgres archive + room history queries style/main.css the console theme ``` @@ -102,9 +111,12 @@ style/main.css the console theme - Sessions are in-memory (`tower-sessions` `MemoryStore`): restart logs everyone out, and it is single-instance. Swap in a Redis/SQL store for multiple replicas. -- Messages are not persisted; you only see traffic while connected. NATS - JetStream (the compose file starts NATS with `-js`) would give history via a - durable stream over `chat.room.*`. +- History: a durable JetStream pull consumer (`CHAT` stream, `cnats-postgres` + consumer) archives `chat.room.*` into Postgres; joining a room backfills the + last 100 messages. Inserts are idempotent on message id (JetStream is + at-least-once). The `CHAT` stream keeps default limits — cap it with + `nats stream edit CHAT --max-age=…` if the bus is chatty, since Postgres + already holds the archive. - The OIDC client verifies ID-token signature, nonce, and CSRF state, and requires PKCE — but there is no token refresh; the app session lives independently of the Kanidm token lifetime. diff --git a/docker-compose.yml b/docker-compose.yml index 2f4c4b2..5630507 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,10 +8,14 @@ services: env_file: .env environment: NATS_URL: nats://nats:4222 + DATABASE_URL: postgres://cnats:cnats@postgres:5432/cnats ports: - "3000:3000" depends_on: - - nats + nats: + condition: service_started + postgres: + condition: service_healthy nats: image: nats:2.10-alpine @@ -19,3 +23,22 @@ services: ports: - "4222:4222" # client connections - "8222:8222" # monitoring UI (http://localhost:8222) + + postgres: + image: postgres:17-alpine + environment: + POSTGRES_USER: cnats + POSTGRES_PASSWORD: cnats + POSTGRES_DB: cnats + ports: + - "5432:5432" + volumes: + - pgdata:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U cnats -d cnats"] + interval: 2s + timeout: 3s + retries: 15 + +volumes: + pgdata: diff --git a/src/app.rs b/src/app.rs index cc52569..d489cd8 100644 --- a/src/app.rs +++ b/src/app.rs @@ -8,6 +8,8 @@ use leptos_router::{ use crate::auth::{current_user, User}; use crate::chat::{is_valid_room, room_subject, ChatMessage, SendMessage, DEFAULT_ROOM, ROOMS}; +#[cfg(feature = "hydrate")] +use crate::chat::room_history; pub fn shell(options: LeptosOptions) -> impl IntoView { view! { @@ -144,7 +146,7 @@ fn ChatShell(room: Memo, user: User) -> impl IntoView { { let es_handle = StoredValue::new_local(None::); Effect::new(move |_| { - let room = room.get(); + let room_name = room.get(); es_handle.update_value(|es| { if let Some(es) = es.take() { es.close(); @@ -152,7 +154,28 @@ fn ChatShell(room: Memo, user: User) -> impl IntoView { }); messages.set(Vec::new()); status.set(BusStatus::Connecting); - es_handle.set_value(open_event_source(&room, messages, status)); + es_handle.set_value(open_event_source(&room_name, messages, status)); + + // Backfill from the Postgres archive; live SSE messages may land + // first, so merge by id (history first, then unseen live ones). + leptos::task::spawn_local(async move { + let Ok(history) = room_history(room_name.clone()).await else { + return; + }; + // The user may have switched rooms while we were fetching. + if room.get_untracked() != room_name { + return; + } + messages.update(|live| { + let mut merged = history; + for m in live.drain(..) { + if !merged.iter().any(|h| h.id == m.id) { + merged.push(m); + } + } + *live = merged; + }); + }); }); on_cleanup(move || { es_handle.update_value(|es| { diff --git a/src/chat.rs b/src/chat.rs index e7820d6..7753dc2 100644 --- a/src/chat.rs +++ b/src/chat.rs @@ -30,8 +30,14 @@ pub struct ChatMessage { pub text: String, /// Pre-formatted UTC wall-clock time, e.g. "14:03:27". pub time: String, + /// Publish instant as UTC epoch milliseconds; orders history. + #[serde(default)] + pub ts: i64, } +/// How many messages `room_history` backfills when joining a room. +pub const HISTORY_LIMIT: i64 = 100; + /// Publishes a message to the room's NATS subject. Requires a signed-in /// session; the sender identity always comes from the session, never from /// the client. @@ -58,13 +64,15 @@ pub async fn send_message(room: String, text: String) -> Result<(), ServerFnErro }; let state = expect_context::(); + let now = chrono::Utc::now(); let msg = ChatMessage { id: uuid::Uuid::new_v4().to_string(), room: room.clone(), username: user.username, display_name: user.display_name, text, - time: chrono::Utc::now().format("%H:%M:%S").to_string(), + time: now.format("%H:%M:%S").to_string(), + ts: now.timestamp_millis(), }; let payload = serde_json::to_vec(&msg).map_err(|e| ServerFnError::new(e.to_string()))?; state @@ -74,3 +82,29 @@ pub async fn send_message(room: String, text: String) -> Result<(), ServerFnErro .map_err(|e| ServerFnError::new(format!("nats publish failed: {e}")))?; Ok(()) } + +/// Returns the most recent messages for a room (oldest first), read from the +/// Postgres archive maintained by the JetStream consumer. +#[server] +pub async fn room_history(room: String) -> Result, ServerFnError> { + use crate::auth::{User, SESSION_USER_KEY}; + use crate::server::AppState; + + if !is_valid_room(&room) { + return Err(ServerFnError::new("unknown room")); + } + let session: tower_sessions::Session = leptos_axum::extract().await?; + if session + .get::(SESSION_USER_KEY) + .await + .map_err(|e| ServerFnError::new(e.to_string()))? + .is_none() + { + return Err(ServerFnError::new("not signed in")); + } + + let state = expect_context::(); + crate::server::store::recent(&state.pool, &room, HISTORY_LIMIT) + .await + .map_err(|e| ServerFnError::new(format!("history unavailable: {e}"))) +} diff --git a/src/main.rs b/src/main.rs index 52ef4dc..8be1c2d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,7 +10,7 @@ async fn main() -> anyhow::Result<()> { Router, }; use cnats::app::{shell, App}; - use cnats::server::{oidc, sse, AppState}; + use cnats::server::{oidc, sse, store, AppState}; use leptos::prelude::*; use leptos_axum::{generate_route_list, LeptosRoutes}; use std::sync::Arc; @@ -34,12 +34,25 @@ async fn main() -> anyhow::Result<()> { tracing::info!(%nats_url, "connecting to NATS"); let nats = async_nats::connect(&nats_url).await?; + let database_url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://cnats:cnats@127.0.0.1:5432/cnats".to_string()); + tracing::info!("connecting to postgres"); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&database_url) + .await?; + store::init_schema(&pool).await?; + + // Archive chat.room.* into Postgres via a durable JetStream consumer. + tokio::spawn(store::run_consumer(nats.clone(), pool.clone())); + let oidc_state = Arc::new(oidc::Oidc::from_env().await?); let state = AppState { leptos_options: leptos_options.clone(), nats, oidc: oidc_state, + pool, }; // Dev-friendly defaults: in-memory sessions, secure cookies only when diff --git a/src/server/mod.rs b/src/server/mod.rs index 55aaed2..2a74a58 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1,5 +1,6 @@ pub mod oidc; pub mod sse; +pub mod store; use axum::extract::FromRef; use leptos::prelude::LeptosOptions; @@ -10,6 +11,7 @@ pub struct AppState { pub leptos_options: LeptosOptions, pub nats: async_nats::Client, pub oidc: Arc, + pub pool: sqlx::PgPool, } impl FromRef for LeptosOptions { diff --git a/src/server/store.rs b/src/server/store.rs new file mode 100644 index 0000000..c226b36 --- /dev/null +++ b/src/server/store.rs @@ -0,0 +1,152 @@ +//! Message archive: a durable JetStream pull consumer drains `chat.room.*` +//! into Postgres, so history survives restarts and includes messages +//! published by any client on the bus (not just this app). + +use std::time::Duration; + +use async_nats::jetstream; +use futures::StreamExt; +use sqlx::PgPool; + +use crate::chat::ChatMessage; + +const STREAM_NAME: &str = "CHAT"; +const CONSUMER_NAME: &str = "cnats-postgres"; + +pub async fn init_schema(pool: &PgPool) -> anyhow::Result<()> { + sqlx::query( + "CREATE TABLE IF NOT EXISTS messages ( + id TEXT PRIMARY KEY, + room TEXT NOT NULL, + username TEXT NOT NULL, + display_name TEXT NOT NULL, + text TEXT NOT NULL, + time TEXT NOT NULL, + ts BIGINT NOT NULL + )", + ) + .execute(pool) + .await?; + sqlx::query( + "CREATE INDEX IF NOT EXISTS messages_room_ts ON messages (room, ts DESC, id)", + ) + .execute(pool) + .await?; + Ok(()) +} + +/// Runs forever; (re)creates the stream/consumer and retries on any failure, +/// so a NATS or Postgres outage never takes the chat server down. +pub async fn run_consumer(nats: async_nats::Client, pool: PgPool) { + loop { + if let Err(err) = consume(&nats, &pool).await { + tracing::error!("archive consumer failed: {err:#}; retrying in 5s"); + } + tokio::time::sleep(Duration::from_secs(5)).await; + } +} + +async fn consume(nats: &async_nats::Client, pool: &PgPool) -> anyhow::Result<()> { + let js = jetstream::new(nats.clone()); + + let stream = js + .get_or_create_stream(jetstream::stream::Config { + name: STREAM_NAME.to_string(), + subjects: vec!["chat.room.*".to_string()], + ..Default::default() + }) + .await + .map_err(|e| anyhow::anyhow!("get_or_create_stream: {e}"))?; + + let consumer = stream + .get_or_create_consumer( + CONSUMER_NAME, + jetstream::consumer::pull::Config { + durable_name: Some(CONSUMER_NAME.to_string()), + ..Default::default() + }, + ) + .await + .map_err(|e| anyhow::anyhow!("get_or_create_consumer: {e}"))?; + + tracing::info!(stream = STREAM_NAME, consumer = CONSUMER_NAME, "archiving to postgres"); + + let mut messages = consumer + .messages() + .await + .map_err(|e| anyhow::anyhow!("consumer messages: {e}"))?; + + while let Some(msg) = messages.next().await { + let msg = msg.map_err(|e| anyhow::anyhow!("pull next: {e}"))?; + match serde_json::from_slice::(&msg.payload) { + Ok(chat) => insert(pool, &chat).await?, + // Malformed payloads (e.g. hand-typed CLI publishes) are logged + // and acked so they don't wedge the consumer. + Err(err) => tracing::warn!("skipping unparseable message: {err}"), + } + msg.ack() + .await + .map_err(|e| anyhow::anyhow!("ack: {e}"))?; + } + Ok(()) +} + +async fn insert(pool: &PgPool, m: &ChatMessage) -> anyhow::Result<()> { + // Idempotent on message id: JetStream is at-least-once, so redeliveries + // after an unacked crash must not duplicate rows. + sqlx::query( + "INSERT INTO messages (id, room, username, display_name, text, time, ts) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (id) DO NOTHING", + ) + .bind(&m.id) + .bind(&m.room) + .bind(&m.username) + .bind(&m.display_name) + .bind(&m.text) + .bind(&m.time) + .bind(m.ts) + .execute(pool) + .await?; + Ok(()) +} + +/// The latest `limit` messages for a room, oldest first. +pub async fn recent(pool: &PgPool, room: &str, limit: i64) -> anyhow::Result> { + #[derive(sqlx::FromRow)] + struct Row { + id: String, + room: String, + username: String, + display_name: String, + text: String, + time: String, + ts: i64, + } + + let mut rows: Vec = sqlx::query_as( + "SELECT id, room, username, display_name, text, time, ts + FROM messages + WHERE room = $1 + ORDER BY ts DESC, id + LIMIT $2", + ) + .bind(room) + .bind(limit) + .fetch_all(pool) + .await?; + + rows.reverse(); + Ok(rows + .into_iter() + .map(|r| ChatMessage { + id: r.id, + room: r.room, + username: r.username, + display_name: r.display_name, + text: r.text, + time: r.time, + ts: r.ts, + }) + .collect()) +}