hemx/examples/kanban/src/main.rs

use axum::extract::{Form, Query, Request, State};
use axum::http::{HeaderMap, StatusCode};
use axum::middleware::{self, Next};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Redirect, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use futures_util::{stream, StreamExt};
use hemplate::Hemplate;
use hemx::{Html, IntoEffect};
use hemx_axum::{
    interactions, runtime_js, runtime_js_path, sse, DispatchRegistry, DispatchRejection,
    EffectResponse, InteractionRequest, PageRequest, PageResponse,
};
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 hemx_sync::{
    Channel, PresenceScope, PresenceTracker, PresenceUpdate, SyncEffect as FrameworkSync,
};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::convert::Infallible;
use std::fs::{self, OpenOptions};
use std::future::Future;
use std::io::Write;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;

const COLUMNS: [(&str, &str); 3] = [("backlog", "Backlog"), ("doing", "Doing"), ("done", "Done")];
const ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT: usize = 64;
const ORDINARY_HANDLER_TIMEOUT: Duration = Duration::from_secs(10);
const STARTUP_REPLAY_TIMEOUT: Duration = Duration::from_secs(5);
const MAX_SYNC_STORE_BYTES: usize = 1024 * 1024;
const ACKNOWLEDGEMENT_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
const ACKNOWLEDGEMENT_RECONNECT_BACKOFF: [Duration; 3] = [
    Duration::from_millis(100),
    Duration::from_millis(250),
    Duration::from_millis(500),
];

#[derive(Default)]
struct AppState {
    board: Mutex<BoardState>,
    sync: Mutex<SyncState>,
    sync_store: Option<SyncStore>,
    sync_sessions: SyncSessionTokens,
    acknowledgement_heartbeat_interval: Duration,
    presence: Mutex<PresenceTracker<String>>,
}

#[derive(Default)]
struct SyncSessionTokens {
    alice_alpha_editor: Option<String>,
    bob_alpha_viewer: Option<String>,
    carol_beta_editor: Option<String>,
}

impl SyncSessionTokens {
    fn from_env() -> Self {
        fn token(name: &str) -> Option<String> {
            std::env::var(name).ok().filter(|value| !value.is_empty())
        }
        Self {
            alice_alpha_editor: token("HEMX_KANBAN_SESSION_ALICE_ALPHA_EDITOR"),
            bob_alpha_viewer: token("HEMX_KANBAN_SESSION_BOB_ALPHA_VIEWER"),
            carol_beta_editor: token("HEMX_KANBAN_SESSION_CAROL_BETA_EDITOR"),
        }
    }

    fn is_configured(&self) -> bool {
        self.alice_alpha_editor.is_some()
            || self.bob_alpha_viewer.is_some()
            || self.carol_beta_editor.is_some()
    }

    fn matches(expected: &Option<String>, candidate: &str) -> bool {
        let Some(expected) = expected else {
            return false;
        };
        if expected.len() != candidate.len() {
            return false;
        }
        expected
            .bytes()
            .zip(candidate.bytes())
            .fold(0_u8, |difference, (left, right)| {
                difference | (left ^ right)
            })
            == 0
    }
}

#[derive(Clone)]
struct SyncStore(PathBuf);

#[derive(Default)]
struct SyncState {
    next_sequence: u64,
    acknowledgements: BTreeMap<CommandId, SyncAcknowledgement>,
    retained_after: u64,
    reconnects: BTreeMap<String, u64>,
    transient_failure_limit: u8,
    transient_failures: BTreeMap<CommandId, u8>,
}

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct CommandId(String);

impl CommandId {
    fn parse(value: Option<&String>) -> Result<Self, SyncRejection> {
        Self::parse_str(value.map(String::as_str).unwrap_or_default())
    }

    fn parse_str(value: &str) -> Result<Self, SyncRejection> {
        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, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "lowercase")]
enum CanonicalColumn {
    Todo,
    Doing,
    Done,
}

impl CanonicalColumn {
    fn parse(value: Option<&String>) -> Result<Self, SyncRejection> {
        match value.map(String::as_str).unwrap_or("done") {
            "todo" => Ok(Self::Todo),
            "doing" => Ok(Self::Doing),
            "done" => Ok(Self::Done),
            _ => Err(SyncRejection::BadRequest("invalid column")),
        }
    }

    fn index(self) -> usize {
        match self {
            Self::Todo => 0,
            Self::Doing => 1,
            Self::Done => 2,
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
struct SyncAcknowledgement {
    command_id: String,
    server_sequence: u64,
    card_id: u64,
    canonical_column: CanonicalColumn,
    status: &'static str,
    #[serde(skip_serializing)]
    tenant: String,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct SyncSnapshot {
    schema_version: u8,
    server_sequence: u64,
    cards: Vec<SnapshotCard>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct SnapshotCard {
    id: u64,
    column: CanonicalColumn,
}

#[derive(Debug)]
enum SyncRejection {
    BadRequest(&'static str),
    Unauthorized(&'static str),
    Forbidden(&'static str),
    Conflict(&'static str),
    Transient,
    Storage,
}

impl IntoResponse for SyncRejection {
    fn into_response(self) -> axum::response::Response {
        let (status, kind, error) = match self {
            Self::BadRequest(error) => (StatusCode::BAD_REQUEST, "invalid-command", error),
            Self::Unauthorized(error) => (StatusCode::UNAUTHORIZED, "authorization-denial", error),
            Self::Forbidden(error) => (StatusCode::FORBIDDEN, "authorization-denial", error),
            Self::Conflict(error) => (StatusCode::CONFLICT, "command-conflict", error),
            Self::Transient => (
                StatusCode::SERVICE_UNAVAILABLE,
                "transport-failure",
                "transient sync failure",
            ),
            Self::Storage => (
                StatusCode::INTERNAL_SERVER_ERROR,
                "storage-failure",
                "sync storage failed",
            ),
        };
        (
            status,
            Json(serde_json::json!({ "kind": kind, "error": error })),
        )
            .into_response()
    }
}

#[derive(Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct PersistedSync {
    schema_version: u8,
    next_sequence: u64,
    acknowledgements: Vec<PersistedAcknowledgement>,
}

#[derive(Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct PersistedAcknowledgement {
    command_id: String,
    server_sequence: u64,
    card_id: u64,
    canonical_column: CanonicalColumn,
    #[serde(default)]
    tenant: Option<String>,
}

impl SyncStore {
    async fn load(&self) -> Result<SyncState, String> {
        if !self.0.exists() {
            return Ok(SyncState {
                next_sequence: 1,
                ..SyncState::default()
            });
        }
        let bytes = run_with_timeout(
            STARTUP_REPLAY_TIMEOUT,
            "startup sync-store read/replay",
            tokio::fs::read(&self.0),
        )
        .await?
        .map_err(|error| format!("read {}: {error}", self.0.display()))?;
        if bytes.len() > MAX_SYNC_STORE_BYTES {
            return Err(format!(
                "sync store {} exceeds {MAX_SYNC_STORE_BYTES} bytes",
                self.0.display()
            ));
        }
        let persisted: PersistedSync = serde_json::from_slice(&bytes)
            .map_err(|error| format!("decode {}: {error}", self.0.display()))?;
        if !matches!(persisted.schema_version, 1 | 2) || persisted.next_sequence == 0 {
            return Err(format!("unsupported sync store {}", self.0.display()));
        }
        let mut acknowledgements = BTreeMap::new();
        for stored in persisted.acknowledgements {
            let command_id = CommandId::parse_str(&stored.command_id)
                .map_err(|_| format!("invalid command id in {}", self.0.display()))?;
            if stored.server_sequence == 0 || stored.card_id == 0 {
                return Err(format!("invalid acknowledgement in {}", self.0.display()));
            }
            let tenant = if persisted.schema_version == 1 {
                "demo".to_owned()
            } else {
                stored
                    .tenant
                    .filter(|tenant| matches!(tenant.as_str(), "demo" | "alpha" | "beta"))
                    .ok_or_else(|| {
                        format!("invalid acknowledgement tenant in {}", self.0.display())
                    })?
            };
            let acknowledgement = SyncAcknowledgement {
                command_id: stored.command_id,
                server_sequence: stored.server_sequence,
                card_id: stored.card_id,
                canonical_column: stored.canonical_column,
                status: "accepted",
                tenant,
            };
            if acknowledgements
                .insert(command_id, acknowledgement)
                .is_some()
            {
                return Err(format!("duplicate command id in {}", self.0.display()));
            }
        }
        Ok(SyncState {
            next_sequence: persisted.next_sequence,
            acknowledgements,
            ..SyncState::default()
        })
    }

    fn persist(
        &self,
        sync: &SyncState,
        acknowledgement: &SyncAcknowledgement,
    ) -> Result<(), String> {
        let mut acknowledgements = sync
            .acknowledgements
            .values()
            .map(PersistedAcknowledgement::from)
            .collect::<Vec<_>>();
        acknowledgements.push(PersistedAcknowledgement::from(acknowledgement));
        acknowledgements.sort_by_key(|item| item.server_sequence);
        let persisted = PersistedSync {
            schema_version: 2,
            next_sequence: sync.next_sequence + 1,
            acknowledgements,
        };
        let bytes = serde_json::to_vec_pretty(&persisted)
            .map_err(|error| format!("encode {}: {error}", self.0.display()))?;
        if let Some(parent) = self.0.parent() {
            fs::create_dir_all(parent)
                .map_err(|error| format!("create {}: {error}", parent.display()))?;
        }
        let temporary = self.0.with_extension("tmp");
        let mut file = OpenOptions::new()
            .create(true)
            .truncate(true)
            .write(true)
            .open(&temporary)
            .map_err(|error| format!("open {}: {error}", temporary.display()))?;
        file.write_all(&bytes)
            .and_then(|()| file.sync_all())
            .map_err(|error| format!("write {}: {error}", temporary.display()))?;
        fs::rename(&temporary, &self.0)
            .map_err(|error| format!("replace {}: {error}", self.0.display()))?;
        if let Some(parent) = self.0.parent() {
            fs::File::open(parent)
                .and_then(|directory| directory.sync_all())
                .map_err(|error| format!("sync {}: {error}", parent.display()))?;
        }
        Ok(())
    }
}

impl From<&SyncAcknowledgement> for PersistedAcknowledgement {
    fn from(value: &SyncAcknowledgement) -> Self {
        Self {
            command_id: value.command_id.clone(),
            server_sequence: value.server_sequence,
            card_id: value.card_id,
            canonical_column: value.canonical_column,
            tenant: Some(value.tenant.clone()),
        }
    }
}

#[derive(Default, Clone)]
struct BoardState {
    next_id: u64,
    cards: Vec<Card>,
}

#[derive(Clone)]
struct Card {
    id: u64,
    title: String,
    column: usize,
}

#[derive(Hemplate)]
struct AppShell {
    runtime_src: &'static str,
    body: Html,
}

#[derive(Hemplate)]
struct LegacySyncFixture {
    runtime_src: &'static str,
}

#[derive(Hemplate)]
#[hemplate = "partials"]
struct BoardColumns {
    columns: Vec<BoardColumn>,
}

#[derive(Hemplate)]
#[hemplate = "partials"]
struct BoardColumn {
    title: &'static str,
    cards: Vec<BoardCard>,
}

#[derive(Hemplate)]
#[hemplate = "partials"]
struct BoardCard {
    id: u64,
    title: String,
    left_disabled: bool,
    right_disabled: bool,
}

#[derive(Hemplate)]
struct Board {
    options: Html,
    board: Html,
}

#[derive(Hemplate)]
#[hemplate = "partials"]
struct ColumnOptions {
    options: Vec<ColumnOption>,
}

#[derive(Hemplate)]
#[hemplate = "partials"]
struct ColumnOption {
    id: &'static str,
    title: &'static str,
}

#[derive(Hemplate)]
#[hemplate = "partials"]
struct Presence {
    count: u64,
}

async fn run_with_timeout<F, T>(
    duration: Duration,
    operation: &'static str,
    future: F,
) -> Result<T, String>
where
    F: Future<Output = T>,
{
    tokio::time::timeout(duration, future)
        .await
        .map_err(|_| format!("{operation} timed out after {} ms", duration.as_millis()))
}

async fn bounded_handler(duration: Duration, request: Request, next: Next) -> Response {
    match run_with_timeout(duration, "ordinary request", next.run(request)).await {
        Ok(response) => response,
        Err(message) => (StatusCode::GATEWAY_TIMEOUT, message).into_response(),
    }
}

async fn ordinary_handler_timeout(request: Request, next: Next) -> Response {
    bounded_handler(ORDINARY_HANDLER_TIMEOUT, request, next).await
}

#[tokio::main]
async fn main() {
    let sync_store = std::env::var_os("HEMX_KANBAN_SYNC_STORE")
        .map(PathBuf::from)
        .map(SyncStore);
    let mut sync = if let Some(store) = sync_store.as_ref() {
        store
            .load()
            .await
            .unwrap_or_else(|error| panic!("cannot start with sync store: {error}"))
    } else {
        SyncState {
            next_sequence: 1,
            ..SyncState::default()
        }
    };
    sync.retained_after = std::env::var("HEMX_KANBAN_RETAINED_AFTER")
        .ok()
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or_default();
    sync.transient_failure_limit = std::env::var("HEMX_KANBAN_SYNC_FAILURES")
        .ok()
        .and_then(|value| value.parse::<u8>().ok())
        .unwrap_or_else(|| u8::from(std::env::var_os("HEMX_KANBAN_FAIL_FIRST_SYNC").is_some()));
    let mut board = initial_board();
    for acknowledgement in sync.acknowledgements.values() {
        if let Some(card) = board
            .cards
            .iter_mut()
            .find(|card| card.id == acknowledgement.card_id)
        {
            card.column = acknowledgement.canonical_column.index();
        }
    }
    let state = Arc::new(AppState {
        board: Mutex::new(board),
        sync: Mutex::new(sync),
        sync_store,
        sync_sessions: SyncSessionTokens::from_env(),
        acknowledgement_heartbeat_interval: std::env::var("HEMX_KANBAN_ACK_HEARTBEAT_MS")
            .ok()
            .and_then(|value| value.parse::<u64>().ok())
            .filter(|milliseconds| *milliseconds > 0)
            .map(Duration::from_millis)
            .unwrap_or(ACKNOWLEDGEMENT_HEARTBEAT_INTERVAL),
        presence: Mutex::new(PresenceTracker::default()),
    });

    let ordinary_routes = Router::new()
        .route("/", get(home).post(interact))
        .route("/move", post(move_card_without_script))
        .route("/events", get(events))
        .route("/sync/broadcast", get(sync_broadcast))
        .route("/sync/ack", get(sync_ack))
        .route("/sync/context", get(sync_context))
        .route("/sync/commands", post(sync_command))
        .route("/sync/snapshot", get(sync_snapshot))
        .route(runtime_js_path(), get(runtime))
        .layer(middleware::from_fn(ordinary_handler_timeout));
    let ordinary_routes = if std::env::var_os("HEMX_KANBAN_LEGACY_SYNC_FIXTURE").as_deref()
        == Some(std::ffi::OsStr::new("1"))
    {
        ordinary_routes
            .route("/sync-demo", get(legacy_sync_fixture))
            .route("/sync.js", get(legacy_sync_fixture_js))
    } else {
        ordinary_routes
    };
    let app = ordinary_routes
        .merge(Router::new().route("/sync/acknowledgements", get(sync_acknowledgements)))
        .with_state(state);

    let addr = std::env::var("HEMX_KANBAN_ADDR")
        .map(|value| value.parse::<SocketAddr>().expect("valid HEMX_KANBAN_ADDR"))
        .unwrap_or_else(|_| SocketAddr::from(([127, 0, 0, 1], 3001)));
    let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
    println!("hemx Kanban example: http://{addr}");
    axum::serve(listener, app).await.unwrap();
}

fn initial_board() -> BoardState {
    BoardState {
        next_id: 4,
        cards: vec![
            Card {
                id: 1,
                title: "Write requirements".into(),
                column: 0,
            },
            Card {
                id: 2,
                title: "Build browser example".into(),
                column: 1,
            },
            Card {
                id: 3,
                title: "Verify with HTTP".into(),
                column: 2,
            },
        ],
    }
}

// req: examples/001 req: component/003
async fn home(State(state): State<Arc<AppState>>, request: PageRequest) -> impl IntoResponse {
    let board = state.board.lock().unwrap().clone();
    request
        .page_html(page_html(&board), shell)
        .title("hemx Kanban")
        .fingerprint(ui::BUILD_FINGERPRINT)
}

#[derive(Deserialize)]
#[serde(rename_all = "lowercase")]
enum MoveDirection {
    Left,
    Right,
}

#[derive(Deserialize)]
struct MoveCardForm {
    card_id: u64,
    direction: MoveDirection,
}

// req: accessibility/001 req: ms/001
async fn move_card_without_script(
    State(state): State<Arc<AppState>>,
    Form(command): Form<MoveCardForm>,
) -> Result<Redirect, StatusCode> {
    let mut board = state.board.lock().unwrap();
    let moved = update_card(&mut board, Some(command.card_id), |card| {
        card.column = match command.direction {
            MoveDirection::Left => card.column.saturating_sub(1),
            MoveDirection::Right => (card.column + 1).min(COLUMNS.len() - 1),
        };
    });
    moved
        .then(|| Redirect::to("/"))
        .ok_or(StatusCode::BAD_REQUEST)
}

async fn runtime() -> impl IntoResponse {
    runtime_js()
}

async fn legacy_sync_fixture() -> impl IntoResponse {
    PageResponse::full(
        ui::page(&LegacySyncFixture {
            runtime_src: runtime_js_path(),
        })
        .into_string(),
    )
    .title("hemx Kanban sync")
    .fingerprint(ui::BUILD_FINGERPRINT)
}

async fn legacy_sync_fixture_js() -> impl IntoResponse {
    (
        [("content-type", "text/javascript; charset=utf-8")],
        include_str!("../tests/fixtures/legacy-sync.js"),
    )
}

async fn interact(
    State(state): State<Arc<AppState>>,
    request: InteractionRequest,
) -> Result<EffectResponse, DispatchRejection> {
    request.dispatch(registry(state))
}

// req: push/001 req: push/003 req: examples/001
struct PresenceSignal {
    channel: Channel,
    count: usize,
}

impl PresenceScope for PresenceSignal {
    fn presence_channel(&self) -> Channel {
        self.channel.clone()
    }
}

#[hemx_sync::presence]
fn presence_changed(signal: PresenceSignal) -> impl hemx::IntoEffect {
    board::presence.put(&Presence {
        count: u64::try_from(signal.count).expect("presence count fits u64"),
    })
}

async fn sync_ack(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Query(params): Query<BTreeMap<String, String>>,
) -> Result<Response, SyncRejection> {
    let principal = current_sync_principal(&headers, &state.sync_sessions)?;
    let command_id = CommandId::parse(params.get("command_id"))?;
    let acknowledgement = {
        let sync = state.sync.lock().unwrap();
        let acknowledgement =
            sync.acknowledgements
                .get(&command_id)
                .ok_or(SyncRejection::Conflict(
                    "command has no canonical acknowledgement",
                ))?;
        if !visible_acknowledgement(principal, acknowledgement) {
            return Err(SyncRejection::Forbidden(
                "current tenant cannot access command acknowledgement",
            ));
        }
        acknowledgement.clone()
    };
    let batch = (
        FrameworkSync::ack(board::atoms::sync_ack),
        board::sync_status.text(format!(
            "Canonical acknowledgement {} at server sequence {}",
            acknowledgement.command_id, acknowledgement.server_sequence
        )),
    )
        .into_batch(ui::BUILD_FINGERPRINT);
    Ok(sse(stream::iter([Ok::<_, Infallible>(batch)]).boxed()).into_response())
}

async fn sync_broadcast(
    State(state): State<Arc<AppState>>,
    Query(params): Query<BTreeMap<String, String>>,
) -> Response {
    let Some(channel) = params
        .get("channel")
        .and_then(|channel| Channel::new(channel).ok())
    else {
        return (StatusCode::BAD_REQUEST, "missing or invalid sync channel").into_response();
    };
    if channel.as_str() != "board" {
        return StatusCode::NOT_FOUND.into_response();
    }
    let member = params.get("member").cloned();
    let count = {
        let mut presence = state.presence.lock().unwrap();
        match (params.get("action").map(String::as_str), member.as_ref()) {
            (Some("join"), Some(member)) => presence.join(channel.clone(), member.clone()).count,
            (Some("leave"), Some(member)) => presence.leave(&channel, member).count,
            (None | Some("snapshot"), None) => presence.count(&channel),
            _ => {
                return (StatusCode::BAD_REQUEST, "invalid presence action or member")
                    .into_response();
            }
        }
    };
    let broadcast =
        presence_changed(PresenceSignal { channel, count }).into_broadcast(ui::BUILD_FINGERPRINT);
    let (_channel, effect_batch) = broadcast.into_parts();
    sse(stream::iter([Ok::<_, Infallible>(effect_batch)]).boxed()).into_response()
}

async fn events(Query(params): Query<BTreeMap<String, String>>) -> impl IntoResponse {
    if params.contains_key("once") {
        let effect = board::presence.put(&Presence { count: 1 });
        return sse(stream::iter([Ok::<_, Infallible>(
            effect.into_batch(ui::BUILD_FINGERPRINT),
        )])
        .boxed());
    }

    let batches = stream::unfold(1_u64, |count| async move {
        tokio::time::sleep(Duration::from_secs(4)).await;
        let effect = board::presence.put(&Presence { count });
        Some((
            Ok::<_, Infallible>(effect.into_batch(ui::BUILD_FINGERPRINT)),
            count + 1,
        ))
    })
    .boxed();
    sse(batches)
}

#[derive(Clone, Copy)]
struct CurrentSyncPrincipal {
    principal: &'static str,
    tenant: &'static str,
    can_replay: bool,
}

fn current_sync_principal(
    headers: &HeaderMap,
    sessions: &SyncSessionTokens,
) -> Result<CurrentSyncPrincipal, SyncRejection> {
    let session = headers
        .get(axum::http::header::COOKIE)
        .and_then(|value| value.to_str().ok())
        .and_then(|cookies| {
            cookies.split(';').find_map(|cookie| {
                cookie
                    .trim()
                    .strip_prefix("hemx_kanban_session=")
                    .map(str::trim)
            })
        });
    let Some(session) = session else {
        if sessions.is_configured() {
            return Err(SyncRejection::Unauthorized(
                "current sync session is required",
            ));
        }
        return Ok(CurrentSyncPrincipal {
            principal: "demo",
            tenant: "demo",
            can_replay: true,
        });
    };
    if SyncSessionTokens::matches(&sessions.alice_alpha_editor, session) {
        return Ok(CurrentSyncPrincipal {
            principal: "alice",
            tenant: "alpha",
            can_replay: true,
        });
    }
    if SyncSessionTokens::matches(&sessions.bob_alpha_viewer, session) {
        return Ok(CurrentSyncPrincipal {
            principal: "bob",
            tenant: "alpha",
            can_replay: false,
        });
    }
    if SyncSessionTokens::matches(&sessions.carol_beta_editor, session) {
        return Ok(CurrentSyncPrincipal {
            principal: "carol",
            tenant: "beta",
            can_replay: true,
        });
    }
    Err(SyncRejection::Unauthorized(
        "current sync session is invalid",
    ))
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct SyncContext {
    account_partition: String,
}

// req: sync/020 req: auth/005
async fn sync_context(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
) -> Result<Json<SyncContext>, SyncRejection> {
    let principal = current_sync_principal(&headers, &state.sync_sessions)?;
    Ok(Json(SyncContext {
        account_partition: format!("{}:{}", principal.tenant, principal.principal),
    }))
}

fn visible_card(principal: CurrentSyncPrincipal, card_id: u64) -> bool {
    principal.tenant == "demo"
        || (principal.tenant == "beta" && card_id == 2)
        || (principal.tenant == "alpha" && card_id != 2)
}

fn visible_acknowledgement(
    principal: CurrentSyncPrincipal,
    acknowledgement: &SyncAcknowledgement,
) -> bool {
    principal.tenant == "demo" || acknowledgement.tenant == principal.tenant
}

fn authorize_sync_replay(
    principal: CurrentSyncPrincipal,
    card_id: u64,
) -> Result<(), SyncRejection> {
    if principal.principal == "demo" && principal.tenant == "demo" {
        return Ok(());
    }
    if !principal.can_replay {
        return Err(SyncRejection::Forbidden(
            "current principal cannot replay commands",
        ));
    }
    let card_tenant = if card_id == 2 { "beta" } else { "alpha" };
    if principal.tenant != card_tenant {
        return Err(SyncRejection::Forbidden(
            "current tenant cannot access command target",
        ));
    }
    Ok(())
}

// req: sync/001 req: sync/008 req: sync/012 req: sync/019 req: security/004
async fn sync_command(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    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 canonical_column = CanonicalColumn::parse(params.get("column"))?;
    let principal = current_sync_principal(&headers, &state.sync_sessions)?;
    authorize_sync_replay(principal, card_id)?;

    let mut sync = state.sync.lock().unwrap();
    if let Some(existing) = sync.acknowledgements.get(&command_id) {
        if !visible_acknowledgement(principal, existing) {
            return Err(SyncRejection::Forbidden(
                "current tenant cannot access command acknowledgement",
            ));
        }
        if existing.card_id != card_id || existing.canonical_column != canonical_column {
            return Err(SyncRejection::Conflict(
                "command_id was already used for a different payload",
            ));
        }
        return Ok(Json(existing.clone()));
    }
    let transient_failure_limit = sync.transient_failure_limit;
    if transient_failure_limit > 0 {
        let failures = sync
            .transient_failures
            .entry(command_id.clone())
            .or_default();
        if *failures < transient_failure_limit {
            *failures += 1;
            return Err(SyncRejection::Transient);
        }
    }

    let mut board = state.board.lock().unwrap();
    let card_index = board
        .cards
        .iter()
        .position(|card| card.id == card_id)
        .ok_or(SyncRejection::BadRequest("unknown card_id"))?;
    let server_sequence = sync
        .acknowledgements
        .values()
        .filter(|acknowledgement| visible_acknowledgement(principal, acknowledgement))
        .map(|acknowledgement| acknowledgement.server_sequence)
        .max()
        .unwrap_or_default()
        + 1;
    let acknowledgement = SyncAcknowledgement {
        command_id: command_id.0.clone(),
        server_sequence,
        card_id,
        canonical_column,
        status: "accepted",
        tenant: principal.tenant.to_owned(),
    };
    if let Some(store) = &state.sync_store {
        store.persist(&sync, &acknowledgement).map_err(|error| {
            eprintln!("sync persistence failed: {error}");
            SyncRejection::Storage
        })?;
    }
    board.cards[card_index].column = canonical_column.index();
    sync.next_sequence += 1;
    sync.acknowledgements
        .insert(command_id, acknowledgement.clone());
    Ok(Json(acknowledgement))
}

// req: sync/007 req: sync/020 req: auth/005 req: security/004
async fn sync_snapshot(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
) -> Result<Json<SyncSnapshot>, SyncRejection> {
    let principal = current_sync_principal(&headers, &state.sync_sessions)?;
    let board = state.board.lock().unwrap();
    let sync = state.sync.lock().unwrap();
    let cards = board
        .cards
        .iter()
        .filter(|card| visible_card(principal, card.id))
        .map(|card| SnapshotCard {
            id: card.id,
            column: canonical_column(card.column),
        })
        .collect();
    let server_sequence = sync
        .acknowledgements
        .values()
        .filter(|acknowledgement| visible_acknowledgement(principal, acknowledgement))
        .map(|acknowledgement| acknowledgement.server_sequence)
        .max()
        .unwrap_or_default();
    Ok(Json(SyncSnapshot {
        schema_version: 1,
        server_sequence,
        cards,
    }))
}

fn canonical_column(column: usize) -> CanonicalColumn {
    match column {
        1 => CanonicalColumn::Doing,
        2 => CanonicalColumn::Done,
        _ => CanonicalColumn::Todo,
    }
}

// req: sync/005 req: sync/006 req: sync/007 req: sync/013 req: auth/005 req: security/004
async fn sync_acknowledgements(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Query(params): Query<BTreeMap<String, String>>,
) -> Result<Sse<impl futures_util::Stream<Item = Result<Event, Infallible>>>, SyncRejection> {
    let principal = current_sync_principal(&headers, &state.sync_sessions)?;
    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 persistent_stream = reconnect_key.is_some();

    let mut sync = state.sync.lock().unwrap();
    if let Some(key) = reconnect_key {
        let attempts = sync.reconnects.entry(key).or_default();
        *attempts += 1;
        if let Some(backoff) = usize::try_from(*attempts - 1)
            .ok()
            .and_then(|index| ACKNOWLEDGEMENT_RECONNECT_BACKOFF.get(index))
        {
            return Ok(Sse::new(
                stream::iter([Ok(Event::default()
                    .comment(format!("reconnect-attempt-{attempts}"))
                    .retry(*backoff))])
                .boxed(),
            )
            .keep_alive(KeepAlive::new().interval(state.acknowledgement_heartbeat_interval)));
        }
    }
    let first_available = sync
        .acknowledgements
        .values()
        .filter(|acknowledgement| {
            visible_acknowledgement(principal, acknowledgement)
                && acknowledgement.server_sequence > sync.retained_after
        })
        .map(|acknowledgement| acknowledgement.server_sequence)
        .min();
    let latest = sync
        .acknowledgements
        .values()
        .filter(|acknowledgement| visible_acknowledgement(principal, acknowledgement))
        .map(|acknowledgement| acknowledgement.server_sequence)
        .max()
        .unwrap_or_default();
    let history_missing = after < latest
        && first_available.is_none_or(|first_sequence| after.saturating_add(1) < first_sequence);
    let pending_count = sync
        .acknowledgements
        .values()
        .filter(|acknowledgement| {
            visible_acknowledgement(principal, acknowledgement)
                && acknowledgement.server_sequence > after
                && acknowledgement.server_sequence > sync.retained_after
        })
        .count();
    let slow_consumer = pending_count > ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT;
    let events = if history_missing || slow_consumer {
        vec![Ok(Event::default()
            .id(latest.to_string())
            .event("snapshot-required")
            .json_data(serde_json::json!({
                "after": after,
                "firstAvailable": first_available,
                "latest": latest,
                "snapshotUrl": "/sync/snapshot",
                "reason": if slow_consumer { "slow-consumer" } else { "missing-history" },
                "pendingCount": pending_count,
                "bufferLimit": ACKNOWLEDGEMENT_STREAM_BUFFER_LIMIT,
            }))
            .expect("serializable missing history event"))]
    } else {
        sync.acknowledgements
            .values()
            .filter(|acknowledgement| {
                visible_acknowledgement(principal, acknowledgement)
                    && acknowledgement.server_sequence > after
                    && acknowledgement.server_sequence > sync.retained_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>>>()
    };
    drop(sync);
    let heartbeat_interval = state.acknowledgement_heartbeat_interval;
    let event_stream = stream::iter(events).boxed();
    let response_stream = if persistent_stream {
        let heartbeat = stream::unfold(heartbeat_interval, |interval| async move {
            tokio::time::sleep(interval).await;
            Some((
                Ok(Event::default()
                    .event("heartbeat")
                    .data("{\"status\":\"ok\"}")),
                interval,
            ))
        });
        event_stream.chain(heartbeat).boxed()
    } else {
        event_stream
    };
    Ok(Sse::new(response_stream).keep_alive(
        KeepAlive::new()
            .interval(heartbeat_interval)
            .text("heartbeat"),
    ))
}

fn registry(state: Arc<AppState>) -> impl DispatchRegistry {
    interactions(ui::BUILD_FINGERPRINT)
        .on(board::create_card, {
            let state = state.clone();
            move |form| {
                // req: examples/001 req: form/002
                let title = form.value("title").unwrap_or("").trim();
                let column = parse_column(form.value("column"));
                let mut board = state.board.lock().unwrap();
                if !title.is_empty() {
                    let id = board.next_id;
                    board.next_id += 1;
                    board.cards.push(Card {
                        id,
                        title: title.into(),
                        column,
                    });
                }
                board_effects(&board, "Card added")
            }
        })
        .on(card_board::move_left, {
            let state = state.clone();
            move |form| {
                // req: examples/001 req: list/003
                let mut board = state.board.lock().unwrap();
                let moved = update_card(&mut board, form.parse("card_id"), |card| {
                    card.column = card.column.saturating_sub(1);
                });
                board_effects(
                    &board,
                    if moved {
                        "Card moved left"
                    } else {
                        "Card not found"
                    },
                )
            }
        })
        .on(card_board::move_right, {
            let state = state.clone();
            move |form| {
                // req: examples/001 req: list/003
                let mut board = state.board.lock().unwrap();
                let moved = update_card(&mut board, form.parse("card_id"), |card| {
                    card.column = (card.column + 1).min(COLUMNS.len() - 1);
                });
                board_effects(
                    &board,
                    if moved {
                        "Card moved right"
                    } else {
                        "Card not found"
                    },
                )
            }
        })
        .on(card_board::delete_card, {
            let state = state.clone();
            move |form| {
                // req: examples/001 req: list/003
                let mut board = state.board.lock().unwrap();
                let before = board.cards.len();
                if let Some(id) = form.parse::<u64>("card_id") {
                    board.cards.retain(|card| card.id != id);
                }
                board_effects(
                    &board,
                    if board.cards.len() < before {
                        "Card deleted"
                    } else {
                        "Card not found"
                    },
                )
            }
        })
}

fn board_effects(board: &BoardState, notice: &'static str) -> impl IntoEffect {
    (
        board::board.put(&board_view(board)),
        board::notice.text(notice),
        board::create_card_form.clear(),
    )
}

fn update_card(
    board: &mut BoardState,
    card_id: Option<u64>,
    update: impl FnOnce(&mut Card),
) -> bool {
    let Some(id) = card_id else {
        return false;
    };
    let Some(card) = board.cards.iter_mut().find(|card| card.id == id) else {
        return false;
    };
    update(card);
    true
}

fn parse_column(value: Option<&str>) -> usize {
    let id = value.unwrap_or(COLUMNS[0].0);
    COLUMNS
        .iter()
        .position(|(column_id, _)| *column_id == id)
        .unwrap_or(0)
}

fn page_html(board: &BoardState) -> Html {
    // req: html_safety/002 req: view/001
    board_ui::page(&Board {
        options: render_options(),
        board: ui::page(&board_view(board)),
    })
}

fn shell(body: Html) -> Html {
    // req: html_safety/001 req: html_safety/002 req: axum_integration/001
    ui::page(&AppShell {
        runtime_src: runtime_js_path(),
        body,
    })
}

fn render_options() -> Html {
    // req: html_safety/002 req: view/001
    ui::page(&ColumnOptions {
        options: COLUMNS
            .iter()
            .map(|(id, title)| ColumnOption { id, title })
            .collect(),
    })
}

fn board_view(board: &BoardState) -> BoardColumns {
    // req: html_safety/002 req: view/001
    BoardColumns {
        columns: COLUMNS
            .iter()
            .enumerate()
            .map(|(idx, (_, title))| BoardColumn {
                title,
                cards: board
                    .cards
                    .iter()
                    .filter(|card| card.column == idx)
                    .map(render_card)
                    .collect(),
            })
            .collect(),
    }
}

fn render_card(card: &Card) -> BoardCard {
    BoardCard {
        id: card.id,
        title: card.title.clone(),
        left_disabled: card.column == 0,
        right_disabled: card.column + 1 == COLUMNS.len(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use hemx_test::{
        class_child_selector, disabled_button_selector, element_class_selector,
        escaped_markup_selector, form_selector, keyed_selector, root_element_selector,
        select_options_selector, small_text_selector, strong_text_selector,
    };
    use scraper::{Html, Selector};
    use std::sync::atomic::{AtomicBool, Ordering};
    use tower::ServiceExt;

    struct CancelProof(Arc<AtomicBool>);

    impl Drop for CancelProof {
        fn drop(&mut self) {
            self.0.store(true, Ordering::SeqCst);
        }
    }

    // req: operations/003
    #[tokio::test]
    async fn ordinary_handlers_timeout_and_cancel_inflight_work() {
        let cancelled = Arc::new(AtomicBool::new(false));
        let proof = Arc::clone(&cancelled);
        let app = Router::new()
            .route(
                "/slow",
                get(move || {
                    let proof = Arc::clone(&proof);
                    async move {
                        let _cancel_proof = CancelProof(proof);
                        std::future::pending::<Response>().await
                    }
                }),
            )
            .layer(middleware::from_fn(|request, next| async move {
                bounded_handler(Duration::from_millis(20), request, next).await
            }));
        let response = app
            .oneshot(
                Request::get("/slow")
                    .body(axum::body::Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT);
        assert!(cancelled.load(Ordering::SeqCst));
    }

    // req: operations/003
    #[tokio::test]
    async fn startup_replay_timeout_cancels_inflight_work_with_a_named_error() {
        let cancelled = Arc::new(AtomicBool::new(false));
        let proof = Arc::clone(&cancelled);
        let error = run_with_timeout(Duration::from_millis(20), "startup replay", async move {
            let _cancel_proof = CancelProof(proof);
            std::future::pending::<()>().await;
        })
        .await
        .expect_err("startup replay must time out");
        assert_eq!(error, "startup replay timed out after 20 ms");
        assert!(cancelled.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn startup_replay_rejects_oversized_store_before_decoding() {
        let path = std::env::temp_dir().join(format!(
            "hemx-kanban-oversized-store-{}.json",
            std::process::id()
        ));
        fs::write(&path, vec![b' '; MAX_SYNC_STORE_BYTES + 1]).unwrap();
        let error = match SyncStore(path.clone()).load().await {
            Ok(_) => panic!("oversized store must be rejected"),
            Err(error) => error,
        };
        let _ = fs::remove_file(path);
        assert!(error.contains("exceeds 1048576 bytes"), "{error}");
    }

    fn selector(value: &str) -> Selector {
        Selector::parse(value).expect("test selector parses")
    }

    // req: html_safety/002 req: view/001 req: test/005
    #[test]
    fn kanban_page_is_composed_by_a_hemplate_view() {
        let html = page_html(&BoardState::default());
        assert!(!html.as_str().contains("__OPTIONS__"));
        assert!(!html.as_str().contains("__BOARD__"));

        let document = Html::parse_fragment(html.as_str());
        assert_eq!(
            document
                .select(&selector(&root_element_selector("section", "kanban")))
                .count(),
            1
        );
        assert_eq!(
            document
                .select(&selector(&select_options_selector("column")))
                .count(),
            3
        );
        assert_eq!(
            document.select(&selector(&form_selector("header"))).count(),
            1
        );
    }

    // req: html_safety/002 req: view/001 req: test/005
    #[test]
    fn board_payload_is_rendered_by_a_hemplate_view() {
        let board = BoardState {
            next_id: 2,
            cards: vec![Card {
                id: 1,
                title: "<b>Compile checked</b>".to_owned(),
                column: 0,
            }],
        };

        let html = ui::page(&board_view(&board));
        let document = Html::parse_fragment(html.as_str());
        assert_eq!(
            document
                .select(&selector(&class_child_selector(
                    "columns", "section", "column"
                )))
                .count(),
            3
        );
        let card = document
            .select(&selector(&keyed_selector("article.card", 1)))
            .next()
            .expect("card renders");
        let title = card
            .select(&selector(strong_text_selector()))
            .next()
            .expect("card title renders");
        assert_eq!(title.text().collect::<String>(), "<b>Compile checked</b>");
        assert!(card
            .select(&selector(&escaped_markup_selector("b")))
            .next()
            .is_none());
        assert_eq!(
            card.select(&selector(disabled_button_selector())).count(),
            1
        );
    }

    // req: html_safety/002 req: view/001 req: test/005
    #[test]
    fn presence_payload_is_rendered_by_a_hemplate_view() {
        let html = ui::page(&Presence { count: 7 });
        let document = Html::parse_fragment(html.as_str());
        assert_eq!(
            document
                .select(&selector(&element_class_selector("span", "presence")))
                .count(),
            2
        );
        assert_eq!(
            document
                .select(&selector(small_text_selector()))
                .next()
                .map(|small| small.text().collect::<String>()),
            Some("tick #7".to_owned())
        );
    }
}