commit 151a4d25eb0556ee8495868e0ff5abd529c7b565
parent f82ffb0051398af639e7cacc8753de2576687242
Author: Silas Brack <silasbrack@gmail.com>
Date: Thu, 11 Jun 2026 14:43:09 +0200
feat: re-implement comment SSE bridge with initial event diagnostic
Pure Datastar approach: data-on:load opens a long-lived SSE stream
to /story/{id}/comments-stream. The bridge subscribes to TrailBase
SSE internally and forwards datastar-patch-elements events.
Key improvements over first attempt:
- Sends current comments immediately on connect (diagnostic)
- data-on:load attribute preserved in SSE response for morph survival
- requestCancellation: 'cleanup' for proper SSE teardown
- Extracted render_comments_event helper to reduce duplication
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Diffstat:
5 files changed, 149 insertions(+), 43 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -2080,6 +2080,7 @@ dependencies = [
"serde_json",
"thiserror 2.0.18",
"tokio",
+ "tokio-stream",
"tracing",
"trailbase-client",
"url",
@@ -2343,6 +2344,18 @@ dependencies = [
]
[[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",
+ "tokio-util",
+]
+
+[[package]]
name = "tokio-util"
version = "0.7.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
@@ -9,6 +9,7 @@ axum-extra = { version = "0.10", features = ["cookie"] }
askama = "0.12"
chrono = { version = "0.4", features = ["serde"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread", "net"] }
+tokio-stream = { version = "0.1", features = ["sync"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
thiserror = "2"
diff --git a/src/handlers/story.rs b/src/handlers/story.rs
@@ -1,16 +1,18 @@
+use std::convert::Infallible;
+
use askama::Template;
use axum::{
- extract::{Path, Query, State},
- response::{Html, IntoResponse, Redirect, Response},
+ extract::{Path, State},
+ response::{Html, Redirect, sse::{Event, Sse}},
Form,
};
use axum_extra::extract::cookie::CookieJar;
use serde::Deserialize;
+use tokio_stream::StreamExt;
use crate::db::{CommentWithMeta, StoryWithMeta};
use crate::error::AppError;
use crate::handlers::auth::get_tokens_from_cookies;
-use crate::sse;
use crate::state::AppState;
use crate::templates::{ApplicationTemplate, CommentsTemplate, EditTemplate, StoryTemplate, SubmitTemplate};
use crate::trailbase::{self, TARGET_TYPE_COMMENT, TARGET_TYPE_STORY};
@@ -69,45 +71,17 @@ async fn fetch_enriched_comments(
Ok(enriched)
}
-#[derive(Debug, Deserialize)]
-#[serde(rename_all = "lowercase")]
-pub enum StoryFragment {
- Comments,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct StoryQuery {
- pub fragment: Option<StoryFragment>,
-}
-
pub async fn show_story(
State(state): State<AppState>,
jar: CookieJar,
Path(id): Path<i64>,
- Query(query): Query<StoryQuery>,
-) -> Result<Response, AppError> {
+) -> Result<Html<String>, AppError> {
let tokens = get_tokens_from_cookies(&jar);
let client = match &tokens {
Some(t) => trailbase::client_with_tokens(&state.trailbase_url, t)?,
None => trailbase::new_client(&state.trailbase_url)?,
};
- // Return comments as a Datastar SSE fragment if requested
- if matches!(query.fragment, Some(StoryFragment::Comments)) {
- let enriched_comments = fetch_enriched_comments(&state, &tokens, &client, id).await?;
- let is_logged_in = tokens.is_some();
- let template = CommentsTemplate {
- comments: enriched_comments,
- is_logged_in,
- };
- let inner_html = template.render()?;
- let html = format!(
- "<section class=\"comments\" id=\"comments-section\">{}</section>",
- inner_html
- );
- return Ok(sse::patch_elements(&html));
- }
-
// Fetch the story
let story = trailbase::get_story(&client, id)
.await?
@@ -161,7 +135,7 @@ pub async fn show_story(
let app = ApplicationTemplate {
content: story_template.render()?,
};
- Ok(Html(app.render()?).into_response())
+ Ok(Html(app.render()?))
}
pub async fn show_submit(
@@ -398,6 +372,132 @@ pub async fn edit_story(
Ok(Redirect::to(&format!("/story/{}", id)))
}
+fn render_comments_event(
+ comments: Vec<CommentWithMeta>,
+ is_logged_in: bool,
+ story_id: i64,
+) -> Option<Event> {
+ let template = CommentsTemplate {
+ comments,
+ is_logged_in,
+ };
+ let inner_html = template.render().ok()?;
+ let html = format!(
+ "<section class=\"comments\" id=\"comments-section\" \
+ data-on:load=\"@get('/story/{story_id}/comments-stream', \
+ {{requestCancellation: 'cleanup'}})\">\
+ {inner_html}</section>"
+ );
+ let data: String = html
+ .lines()
+ .map(|line| format!("elements {line}"))
+ .collect::<Vec<_>>()
+ .join("\n");
+ Some(
+ Event::default()
+ .event("datastar-patch-elements")
+ .data(data),
+ )
+}
+
+/// Long-lived SSE endpoint that bridges TrailBase comment subscriptions
+/// to Datastar patch events. Sends the current comments immediately on
+/// connect, then pushes updates whenever TrailBase notifies of changes.
+pub async fn comments_stream(
+ State(state): State<AppState>,
+ jar: CookieJar,
+ Path(story_id): Path<i64>,
+) -> Sse<impl tokio_stream::Stream<Item = Result<Event, Infallible>>> {
+ let tokens = get_tokens_from_cookies(&jar);
+ let (tx, rx) = tokio::sync::mpsc::channel::<Result<Event, Infallible>>(10);
+
+ tokio::spawn(async move {
+ // Helper to build a TrailBase client
+ let make_client = || -> Option<trailbase_client::Client> {
+ match &tokens {
+ Some(t) => trailbase::client_with_tokens(&state.trailbase_url, t).ok(),
+ None => trailbase::new_client(&state.trailbase_url).ok(),
+ }
+ };
+
+ let is_logged_in = tokens.is_some();
+
+ // Send current comments immediately so we can verify the endpoint works
+ if let Some(client) = make_client() {
+ if let Ok(comments) =
+ fetch_enriched_comments(&state, &tokens, &client, story_id).await
+ {
+ if let Some(event) = render_comments_event(comments, is_logged_in, story_id) {
+ if tx.send(Ok(event)).await.is_err() {
+ return;
+ }
+ }
+ }
+ }
+
+ // Subscribe to TrailBase SSE for change notifications
+ let subscribe_url = format!(
+ "{}/api/records/v1/comment/subscribe/*?filter[story_id]={}",
+ state.trailbase_url.trim_end_matches('/'),
+ story_id
+ );
+
+ let response = match state.http_client.get(&subscribe_url).send().await {
+ Ok(r) => r,
+ Err(e) => {
+ tracing::error!(
+ "Failed to subscribe to comments for story {}: {}",
+ story_id,
+ e
+ );
+ return;
+ }
+ };
+
+ let mut byte_stream = response.bytes_stream();
+ let mut buffer = Vec::new();
+
+ while let Some(chunk) = byte_stream.next().await {
+ let chunk = match chunk {
+ Ok(c) => c,
+ Err(_) => break,
+ };
+ buffer.extend_from_slice(&chunk);
+
+ // Detect SSE event boundary (double newline)
+ if !buffer.windows(2).any(|w| w == b"\n\n") {
+ continue;
+ }
+ buffer.clear();
+
+ // Fetch fresh comments and push to client
+ let client = match make_client() {
+ Some(c) => c,
+ None => continue,
+ };
+
+ match fetch_enriched_comments(&state, &tokens, &client, story_id).await {
+ Ok(comments) => {
+ if let Some(event) = render_comments_event(comments, is_logged_in, story_id) {
+ if tx.send(Ok(event)).await.is_err() {
+ break;
+ }
+ }
+ }
+ Err(e) => {
+ tracing::error!("Failed to fetch comments for SSE: {}", e);
+ }
+ }
+ }
+ });
+
+ let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
+ Sse::new(stream).keep_alive(
+ axum::response::sse::KeepAlive::new()
+ .interval(std::time::Duration::from_secs(30)),
+ )
+}
+
fn render_edit_error(story_id: i64, form: &EditForm, all_tags: &[trailbase::Category], error: &str) -> Html<String> {
let edit = EditTemplate {
error: Some(error.to_string()),
diff --git a/src/routes.rs b/src/routes.rs
@@ -73,6 +73,7 @@ pub fn create_router(state: AppState) -> Router {
.route("/story/{id}/edit", post(handlers::edit_story))
// Comments
+ .route("/story/{id}/comments-stream", get(handlers::comments_stream))
.route("/story/{id}/comment", post(handlers::create_comment))
.route("/comment/{id}/reply", get(handlers::show_reply_form))
.route("/comment/{id}/reply", post(handlers::submit_reply))
diff --git a/templates/story.html b/templates/story.html
@@ -81,16 +81,7 @@
{% endif %}
<section class="comments" id="comments-section"
- data-on:comments-updated="@get('/story/{{ story.id }}?fragment=comments')">
+ data-on:load="@get('/story/{{ story.id }}/comments-stream', {requestCancellation: 'cleanup'})">
{% include "comments.html" %}
</section>
-
- <script>
- (function() {
- var evtSource = new EventSource('/api/records/v1/comment/subscribe/*?filter[story_id]={{ story.id }}');
- evtSource.onmessage = function() {
- document.getElementById('comments-section').dispatchEvent(new CustomEvent('comments-updated'));
- };
- })();
- </script>
</div>
