feat(kanban): acknowledge idempotent server command

req: sync/001\nreq: sync/005\nreq: sync/006\nreq: sync/007\nreq: sync/008\nreq: sync/012\nreq: sync/013

examples/kanban/Cargo.toml

@@ -6,7 +6,7 @@ publish = false
 
 [features]
 default = ["server"]
-server = ["dep:axum", "dep:futures-util", "dep:hemx-axum", "dep:tokio"]
+server = ["dep:axum", "dep:futures-util", "dep:hemx-axum", "dep:serde", "dep:serde_json", "dep:tokio"]
 client = ["hemx/client"]
 fixture = []
 
@@ -29,6 +29,8 @@ axum = { version = "0.8", optional = true }
 futures-util = { version = "0.3", optional = true }
 hemx = { path = "../../hemx" }
 hemx-axum = { path = "../../hemx-axum", optional = true }
+serde = { version = "1", features = ["derive"], optional = true }
+serde_json = { version = "1", optional = true }
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "time"], optional = true }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]

examples/kanban/src/main.rs

@@ -1,7 +1,9 @@
 use axum::extract::{Query, State};
+use axum::http::StatusCode;
+use axum::response::sse::{Event, KeepAlive, Sse};
 use axum::response::IntoResponse;
-use axum::routing::get;
-use axum::Router;
+use axum::routing::{get, post};
+use axum::{Json, Router};
 use futures_util::{stream, StreamExt};
 use hemplate::Hemplate;
 use hemx::{Html, IntoEffect};
@@ -12,6 +14,7 @@ use hemx_axum::{
 use hemx_kanban_example::ui::board::{self as board};
 use hemx_kanban_example::ui::board_card as card_board;
 use hemx_kanban_example::ui::{self, board as board_ui};
+use serde::Serialize;
 use std::collections::BTreeMap;
 use std::convert::Infallible;
 use std::net::SocketAddr;
@@ -23,6 +26,58 @@ const COLUMNS: [(&str, &str); 3] = [("backlog", "Backlog"), ("doing", "Doing"),
 #[derive(Default)]
 struct AppState {
     board: Mutex<BoardState>,
+    sync: Mutex<SyncState>,
+}
+
+#[derive(Default)]
+struct SyncState {
+    next_sequence: u64,
+    acknowledgements: BTreeMap<CommandId, SyncAcknowledgement>,
+    reconnects: BTreeMap<String, u64>,
+}
+
+#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
+struct CommandId(String);
+
+impl CommandId {
+    fn parse(value: Option<&String>) -> Result<Self, SyncRejection> {
+        let value = value.map(String::as_str).unwrap_or_default();
+        if value.is_empty()
+            || value.len() > 128
+            || !value.bytes().all(|byte| {
+                byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b':' | b'.')
+            })
+        {
+            return Err(SyncRejection::BadRequest("invalid command_id"));
+        }
+        Ok(Self(value.to_owned()))
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct SyncAcknowledgement {
+    command_id: String,
+    server_sequence: u64,
+    card_id: u64,
+    canonical_column: &'static str,
+    status: &'static str,
+}
+
+#[derive(Debug)]
+enum SyncRejection {
+    BadRequest(&'static str),
+    Conflict(&'static str),
+}
+
+impl IntoResponse for SyncRejection {
+    fn into_response(self) -> axum::response::Response {
+        let (status, error) = match self {
+            Self::BadRequest(error) => (StatusCode::BAD_REQUEST, error),
+            Self::Conflict(error) => (StatusCode::CONFLICT, error),
+        };
+        (status, Json(serde_json::json!({ "error": error }))).into_response()
+    }
 }
 
 #[derive(Default, Clone)]
@@ -114,11 +169,17 @@ async fn main() {
                 },
             ],
         }),
+        sync: Mutex::new(SyncState {
+            next_sequence: 1,
+            ..SyncState::default()
+        }),
     });
 
     let app = Router::new()
         .route("/", get(home).post(interact))
         .route("/events", get(events))
+        .route("/sync/commands", post(sync_command))
+        .route("/sync/acknowledgements", get(sync_acknowledgements))
         .route(runtime_js_path(), get(runtime))
         .with_state(state);
 
@@ -172,6 +233,91 @@ async fn events(Query(params): Query<BTreeMap<String, String>>) -> impl IntoResp
     sse(batches)
 }
 
+// req: sync/001 req: sync/008 req: sync/012
+async fn sync_command(
+    State(state): State<Arc<AppState>>,
+    Query(params): Query<BTreeMap<String, String>>,
+) -> Result<Json<SyncAcknowledgement>, SyncRejection> {
+    let command_id = CommandId::parse(params.get("command_id"))?;
+    let card_id = params
+        .get("card_id")
+        .and_then(|value| value.parse::<u64>().ok())
+        .filter(|value| *value > 0)
+        .ok_or(SyncRejection::BadRequest("invalid card_id"))?;
+
+    let mut sync = state.sync.lock().unwrap();
+    if let Some(existing) = sync.acknowledgements.get(&command_id) {
+        if existing.card_id != card_id {
+            return Err(SyncRejection::Conflict(
+                "command_id was already used for a different payload",
+            ));
+        }
+        return Ok(Json(existing.clone()));
+    }
+
+    let mut board = state.board.lock().unwrap();
+    let card = board
+        .cards
+        .iter_mut()
+        .find(|card| card.id == card_id)
+        .ok_or(SyncRejection::BadRequest("unknown card_id"))?;
+    card.column = 2;
+    let acknowledgement = SyncAcknowledgement {
+        command_id: command_id.0.clone(),
+        server_sequence: sync.next_sequence,
+        card_id,
+        canonical_column: "done",
+        status: "accepted",
+    };
+    sync.next_sequence += 1;
+    sync.acknowledgements
+        .insert(command_id, acknowledgement.clone());
+    Ok(Json(acknowledgement))
+}
+
+// req: sync/005 req: sync/006 req: sync/007 req: sync/013
+async fn sync_acknowledgements(
+    State(state): State<Arc<AppState>>,
+    Query(params): Query<BTreeMap<String, String>>,
+) -> Sse<impl futures_util::Stream<Item = Result<Event, Infallible>>> {
+    let after = params
+        .get("after")
+        .and_then(|value| value.parse::<u64>().ok())
+        .unwrap_or_default();
+    let reconnect_key = params
+        .get("reconnect")
+        .filter(|value| !value.is_empty())
+        .cloned();
+
+    let mut sync = state.sync.lock().unwrap();
+    if let Some(key) = reconnect_key {
+        let attempts = sync.reconnects.entry(key).or_default();
+        *attempts += 1;
+        if *attempts == 1 {
+            return Sse::new(
+                stream::iter([Ok(Event::default()
+                    .comment("reconnect")
+                    .retry(Duration::from_millis(25)))])
+                .boxed(),
+            )
+            .keep_alive(KeepAlive::default());
+        }
+    }
+    let events = sync
+        .acknowledgements
+        .values()
+        .filter(|acknowledgement| acknowledgement.server_sequence > after)
+        .map(|acknowledgement| {
+            Ok(Event::default()
+                .id(acknowledgement.server_sequence.to_string())
+                .event("acknowledgement")
+                .json_data(acknowledgement)
+                .expect("serializable acknowledgement"))
+        })
+        .collect::<Vec<Result<Event, Infallible>>>();
+    Sse::new(stream::iter(events).boxed()).keep_alive(KeepAlive::default())
+}
+
 fn registry(state: Arc<AppState>) -> impl DispatchRegistry {
     interactions(ui::BUILD_FINGERPRINT)
         .on(board::create_card, {

examples/kanban/tests/browser_e2e.rs

@@ -88,6 +88,131 @@ async fn server_first_route_does_not_load_optional_client_assets() -> WebDriverR
     result.and(quit)
 }
 
+#[tokio::test]
+async fn idempotent_server_command_is_acknowledged_after_reconnect() -> WebDriverResult<()> {
+    // test req: sync/001 req: sync/005 req: sync/006 req: sync/007
+    // test req: sync/008 req: sync/012 req: sync/013
+    let app_port = available_port();
+    let app_addr = format!("127.0.0.1:{app_port}");
+    let mut app = Command::new(env!("CARGO_BIN_EXE_hemx-kanban-example"));
+    app.env("HEMX_KANBAN_ADDR", &app_addr);
+    let _app = TestProcess::start(app, "hemx-kanban", &app_addr, STARTUP_TIMEOUT)
+        .expect("start ready hemx-kanban");
+
+    let webdriver_port = available_port();
+    let webdriver_addr = format!("127.0.0.1:{webdriver_port}");
+    let mut webdriver = Command::new("geckodriver");
+    webdriver.arg("--port").arg(webdriver_port.to_string());
+    let _webdriver = TestProcess::start(webdriver, "geckodriver", &webdriver_addr, STARTUP_TIMEOUT)
+        .expect("start ready geckodriver");
+
+    let mut caps = DesiredCapabilities::firefox();
+    caps.set_headless()?;
+    let driver = WebDriver::new(&format!("http://{webdriver_addr}"), caps).await?;
+    let result = async {
+        driver.goto(&format!("http://{app_addr}/")).await?;
+        driver.find(By::Css("section[data-hemx-root='kanban']")).await?;
+        driver
+            .execute(
+                r#"
+                window.__syncProof = { ready: false, opens: 0, events: [], error: null };
+                (async () => {
+                  const endpoint = '/sync/commands?command_id=actor-1%3A1&card_id=1';
+                  const firstResponse = await fetch(endpoint, { method: 'POST' });
+                  const first = await firstResponse.json();
+                  const duplicateResponse = await fetch(endpoint, { method: 'POST' });
+                  const duplicate = await duplicateResponse.json();
+                  const conflictResponse = await fetch('/sync/commands?command_id=actor-1%3A1&card_id=2', { method: 'POST' });
+                  const conflict = await conflictResponse.json();
+                  window.__syncProof.command = {
+                    firstStatus: firstResponse.status,
+                    duplicateStatus: duplicateResponse.status,
+                    first,
+                    duplicate,
+                    conflictStatus: conflictResponse.status,
+                    conflict,
+                  };
+                  const source = new EventSource('/sync/acknowledgements?after=0&reconnect=browser-proof');
+                  source.onopen = () => { window.__syncProof.opens += 1; };
+                  source.addEventListener('acknowledgement', (event) => {
+                    window.__syncProof.events.push({ id: event.lastEventId, acknowledgement: JSON.parse(event.data) });
+                    window.__syncProof.ready = true;
+                    source.close();
+                  });
+                  source.onerror = () => {
+                    if (source.readyState === EventSource.CLOSED && !window.__syncProof.ready) {
+                      window.__syncProof.error = 'acknowledgement stream closed';
+                    }
+                  };
+                })().catch((error) => { window.__syncProof.error = String(error); });
+                return true;
+                "#,
+                Vec::new(),
+            )
+            .await?;
+        wait_until(
+            &driver,
+            "return window.__syncProof.ready === true || window.__syncProof.error !== null",
+        )
+        .await?;
+        let proof = driver
+            .execute("return window.__syncProof", Vec::new())
+            .await?
+            .json()
+            .clone();
+        assert!(proof["error"].is_null(), "sync failed: {proof}");
+        assert!(
+            proof["opens"].as_u64().is_some_and(|opens| opens >= 2),
+            "transport did not reconnect: {proof}"
+        );
+        assert_eq!(proof["command"]["firstStatus"], 200);
+        assert_eq!(proof["command"]["duplicateStatus"], 200);
+        assert_eq!(proof["command"]["first"], proof["command"]["duplicate"]);
+        assert_eq!(proof["command"]["first"]["commandId"], "actor-1:1");
+        assert_eq!(proof["command"]["first"]["serverSequence"], 1);
+        assert_eq!(proof["command"]["first"]["cardId"], 1);
+        assert_eq!(proof["command"]["first"]["canonicalColumn"], "done");
+        assert_eq!(proof["command"]["first"]["status"], "accepted");
+        assert_eq!(proof["command"]["conflictStatus"], 409);
+        assert_eq!(
+            proof["command"]["conflict"]["error"],
+            "command_id was already used for a different payload"
+        );
+        assert_eq!(proof["events"].as_array().map(Vec::len), Some(1));
+        assert_eq!(proof["events"][0]["id"], "1");
+        assert_eq!(
+            proof["events"][0]["acknowledgement"],
+            proof["command"]["first"]
+        );
+
+        driver.refresh().await?;
+        let canonical = driver
+            .execute(
+                "return [...document.querySelectorAll('section.column')].map((column) => ({ title: column.querySelector('h2').textContent, cards: [...column.querySelectorAll('[data-key]')].map((card) => card.dataset.key) }))",
+                Vec::new(),
+            )
+            .await?
+            .json()
+            .clone();
+        assert_eq!(canonical[2]["title"], "Done");
+        assert_eq!(canonical[2]["cards"], serde_json::json!(["1", "3"]));
+        Ok(())
+    }
+    .await;
+    let quit = driver.quit().await;
+    result.and(quit)
+}
+
+async fn wait_until(driver: &WebDriver, script: &str) -> WebDriverResult<()> {
+    for _ in 0..200 {
+        if driver.execute(script, Vec::new()).await?.json().as_bool() == Some(true) {
+            return Ok(());
+        }
+        tokio::time::sleep(Duration::from_millis(25)).await;
+    }
+    panic!("browser condition timed out: {script}");
+}
+
 fn available_port() -> u16 {
     TcpListener::bind("127.0.0.1:0")
         .expect("reserve browser test port")