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 <noreply@anthropic.com>

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<String>, user: User) -> impl IntoView {
     {
         let es_handle = StoredValue::new_local(None::<web_sys::EventSource>);
         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<String>, 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| {

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::<AppState>();
+    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<Vec<ChatMessage>, 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::<User>(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::<AppState>();
+    crate::server::store::recent(&state.pool, &room, HISTORY_LIMIT)
+        .await
+        .map_err(|e| ServerFnError::new(format!("history unavailable: {e}")))
+}

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

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<oidc::Oidc>,
+    pub pool: sqlx::PgPool,
 }
 
 impl FromRef<AppState> for LeptosOptions {

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::<ChatMessage>(&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<Vec<ChatMessage>> {
+    #[derive(sqlx::FromRow)]
+    struct Row {
+        id: String,
+        room: String,
+        username: String,
+        display_name: String,
+        text: String,
+        time: String,
+        ts: i64,
+    }
+
+    let mut rows: Vec<Row> = 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())
+}