//! Push local changes to the remote sync server.

use chrono::Utc;
use goingson_core::CoreError;
use sqlx::SqlitePool;
use synckit_client::{ChangeEntry, ChangeOp, DeviceId, Hlc, SyncKitClient};
use tracing::{debug, warn};
use uuid::Uuid;

use super::PUSH_BATCH_LIMIT;
use super::hlc::assign_pending_hlcs;

/// A pending sync_changelog row with its id:
/// (id, table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter).
type PendingRow = (i64, String, String, String, String, Option<String>, Option<i64>, Option<i64>);

#[tracing::instrument(skip_all)]
pub async fn push_changes(
    pool: &SqlitePool,
    client: &SyncKitClient,
    device_id: Uuid,
) -> Result<i64, CoreError> {
    // Stamp any unpushed changes with their HLC before reading them for the wire.
    assign_pending_hlcs(pool, device_id).await?;

    let mut total_pushed: i64 = 0;

    loop {
        let rows: Vec<PendingRow> = sqlx::query_as(
            "SELECT id, table_name, op, row_id, timestamp, data, hlc_wall, hlc_counter \
             FROM sync_changelog WHERE pushed = 0 ORDER BY id ASC LIMIT ?"
        )
        .bind(PUSH_BATCH_LIMIT)
        .fetch_all(pool)
        .await
        .map_err(CoreError::database)?;

        if rows.is_empty() {
            break;
        }

        let row_count = rows.len() as i64;
        let mut pushed_ids: Vec<i64> = Vec::new();
        let changes: Vec<ChangeEntry> = rows
            .into_iter()
            .filter_map(|(id, table, op, row_id, timestamp, data, hlc_wall, hlc_counter)| {
                // Always mark as pushed (even unknown ops) to avoid infinite re-fetch
                pushed_ids.push(id);

                let ts = chrono::DateTime::parse_from_rfc3339(&timestamp)
                    .map(|dt| dt.with_timezone(&Utc))
                    .unwrap_or(chrono::DateTime::UNIX_EPOCH);

                let json_data = data.and_then(|d| serde_json::from_str(&d).ok());

                let change_op = match ChangeOp::from_str_opt(&op) {
                    Some(o) => o,
                    None => {
                        warn!("Skipping changelog entry with unknown op: {}", op);
                        return None;
                    }
                };

                // Stamped by assign_pending_hlcs above; fall back to a
                // timestamp-derived HLC if a row was somehow left unstamped.
                let hlc = match (hlc_wall, hlc_counter) {
                    (Some(wall), Some(counter)) => Hlc { wall_ms: wall, counter: counter as u32, node: DeviceId::new(device_id) },
                    _ => Hlc::from_legacy(ts.timestamp_millis(), DeviceId::new(device_id)),
                };

                Some(ChangeEntry {
                    table,
                    op: change_op,
                    row_id,
                    timestamp: ts,
                    hlc,
                    data: json_data,
                    extra: Default::default(),
                })
            })
            .collect();

        let count = changes.len() as i64;
        let is_last_batch = row_count < PUSH_BATCH_LIMIT;

        client
            .push(DeviceId::new(device_id), changes)
            .await
            .map_err(|e| CoreError::sync(format!("push failed: {}", e)))?;

        // Mark pushed entries in a single transaction
        let mut tx = pool.begin().await.map_err(CoreError::database)?;
        for id in &pushed_ids {
            sqlx::query("UPDATE sync_changelog SET pushed = 1 WHERE id = ?")
                .bind(id)
                .execute(&mut *tx)
                .await
                .map_err(CoreError::database)?;
        }
        tx.commit().await.map_err(CoreError::database)?;

        total_pushed += count;

        if is_last_batch {
            break;
        }
    }

    if total_pushed > 0 {
        debug!("Pushed {} changes", total_pushed);
    }
    Ok(total_pushed)
}