//! Background sync worker — the bridge that makes the CQRS split eventually
//! consistent.
//!
//! It polls the SQLite outbox, migrates each pending write to Postgres, and on
//! success broadcasts a WebSocket alert for every changed field whose custom
//! data type (`Live<T>`) requested one.

use crate::state::AppState;
use serde_json::{json, Value};
use std::time::Duration;

/// Spawn the worker on the current tokio runtime. Runs until the process exits.
pub fn spawn(state: AppState) {
    tokio::spawn(async move {
        loop {
            if let Err(e) = tick(&state).await {
                eprintln!("[roundabout_orm] sync worker error: {e}");
            }
            tokio::time::sleep(Duration::from_millis(500)).await;
        }
    });
}

/// Drain one batch of pending writes.
async fn tick(state: &AppState) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let pending = crate::db::sqlite::fetch_pending(&state.sqlite, 50).await?;
    for row in pending {
        let Some(ent) = state.manifest.iter().find(|e| e.name == row.entity) else {
            crate::db::sqlite::set_status(&state.sqlite, row.id, "error").await?;
            continue;
        };

        let changes: serde_json::Map<String, Value> = serde_json::from_str(&row.payload_json)
            .ok()
            .and_then(|v: Value| v.as_object().cloned())
            .unwrap_or_default();

        match crate::db::postgres::apply_update(&state.pg, ent, &row.row_id, &changes).await {
            Ok(()) => {
                crate::db::sqlite::set_status(&state.sqlite, row.id, "done").await?;
                broadcast_alerts(state, ent, &row.row_id, &changes);
            }
            Err(e) => {
                eprintln!("[roundabout_orm] migrate outbox#{} failed: {e}", row.id);
                crate::db::sqlite::set_status(&state.sqlite, row.id, "error").await?;
            }
        }
    }
    Ok(())
}

/// Emit a WS alert for each changed field flagged `ws` in the manifest.
fn broadcast_alerts(
    state: &AppState,
    ent: &roundabout_orm_schema::EntityDef,
    row_id: &str,
    changes: &serde_json::Map<String, Value>,
) {
    let Some(tx) = state.ws_hub.get(&ent.table) else {
        return;
    };
    for (field, value) in changes {
        let is_live = ent.field(field).map(|f| f.ws).unwrap_or(false);
        if !is_live {
            continue;
        }
        let alert = json!({
            "entity": ent.name,
            "table": ent.table,
            "row_id": row_id,
            "field": field,
            "value": value,
        });
        // Ignore send errors: a closed channel just means no subscribers.
        let _ = tx.send(alert.to_string());
    }
}
