//! Table column whitelists and row-level apply logic (upsert, delete).
//!
//! # SQL Safety: `format!()` for table and column names
//!
//! Several functions in this module use `format!()` to interpolate table and column
//! names into SQL strings (e.g., `apply_upsert`, `apply_delete`).
//! This is safe because:
//!
//! 1. **Table names come from hardcoded constants** (`UPSERT_ORDER`, `DELETE_ORDER`) -- never
//!    from user input or remote data.
//! 2. **Column names come from `table_columns()`**, a compile-time whitelist of `&'static str`
//!    literals -- also never from user input.
//! 3. **All user-supplied values** (row IDs, field data) are passed through `sqlx::query().bind()`,
//!    which parameterizes them safely.
//! 4. **Unknown table names are rejected** before any SQL is constructed: both `apply_upsert` and
//!    `apply_delete` return an error if `table_columns()` returns `None`.

use goingson_core::CoreError;
use sqlx::{Row, SqliteConnection};

use super::EMAIL_ACCOUNT_SYNC_COLS;

/// The single source of truth for which columns each table syncs.
///
/// CHRONIC-C invariant: every column a sync changelog trigger emits into
/// `sync_changelog.data` (via `json_object(...)`) must appear here, and every column
/// here must be emitted by the trigger. `apply_upsert` and `create_initial_snapshot`
/// both read their column list from this table, so the apply side has exactly one list.
/// The trigger side lives in raw migration SQL (`migrations/sqlite/*.sql`); the
/// `trigger_columns_match_whitelist` round-trip test reads the live trigger DDL and
/// asserts the emitted `json_object` keys equal this list for every table, so the two
/// can never silently diverge again. This is the bug behind UF-1: the trigger emitted
/// `recurrence_rule` but it was missing here, so it arrived NULL on every pull.
pub(crate) const SYNCED_COLUMNS: &[(&str, &[&str])] = &[
    ("projects", &[
        "id", "name", "description", "project_type", "status", "created_at", "user_id",
    ]),
    ("tasks", &[
        "id", "project_id", "description", "status", "priority", "due", "tags", "urgency",
        "recurrence", "recurrence_rule", "created_at", "user_id", "recurrence_parent_id",
        "source_email_id", "snoozed_until", "waiting_for_response", "waiting_since",
        "expected_response_date", "scheduled_start", "scheduled_duration", "is_focus",
        "focus_set_at", "contact_id", "milestone_id", "completed_at", "estimated_minutes",
        "actual_minutes",
    ]),
    ("events", &[
        "id", "project_id", "title", "description", "start_time", "end_time", "location",
        "user_id", "linked_task_id", "recurrence", "recurrence_parent_id", "recurrence_rule",
        "contact_id", "block_type", "external_source", "external_id", "is_read_only",
        "snoozed_until", "reminder_offsets_seconds",
    ]),
    ("contacts", &[
        "id", "user_id", "display_name", "nickname", "company", "title", "notes", "tags",
        "birthday", "timezone", "external_source", "external_id", "created_at", "updated_at",
    ]),
    ("contact_emails", &["id", "contact_id", "address", "label", "is_primary"]),
    ("contact_phones", &["id", "contact_id", "number", "label", "is_primary"]),
    ("contact_social_handles", &["id", "contact_id", "platform", "handle", "url"]),
    ("contact_custom_fields", &["id", "contact_id", "label", "value", "url"]),
    ("annotations", &["id", "task_id", "timestamp", "note"]),
    ("subtasks", &[
        "id", "task_id", "text", "is_completed", "position", "created_at", "linked_task_id",
    ]),
    ("task_status_tokens", &[
        "id", "task_id", "kind", "reference", "state", "is_primary", "position", "created_at",
    ]),
    ("milestones", &[
        "id", "user_id", "project_id", "name", "description", "position", "target_date",
        "status", "created_at",
    ]),
    ("time_sessions", &[
        "id", "task_id", "user_id", "started_at", "ended_at", "duration_minutes", "created_at",
    ]),
    ("attachments", &[
        "id", "user_id", "task_id", "project_id", "filename", "file_size", "mime_type",
        "blob_hash", "source_email_id", "created_at",
    ]),
    ("sync_accounts", &[
        "id", "user_id", "provider", "account_name", "email", "sync_calendars", "sync_contacts",
        "calendar_ids", "sync_interval_minutes", "enabled", "created_at",
    ]),
    ("email_accounts", EMAIL_ACCOUNT_SYNC_COLS),
    ("daily_notes", &[
        "id", "user_id", "note_date", "went_well", "could_improve", "is_reviewed", "reviewed_at",
        "created_at", "updated_at",
    ]),
    ("saved_views", &[
        "id", "user_id", "name", "view_type", "filters", "sort_by", "sort_order", "is_pinned",
        "position", "created_at", "updated_at",
    ]),
    ("weekly_reviews", &[
        "id", "user_id", "week_start_date", "completed_at", "notes", "vacation_days",
    ]),
    ("monthly_goals", &[
        "id", "user_id", "month", "text", "status", "position", "created_at", "updated_at",
    ]),
    ("monthly_reflections", &[
        "id", "user_id", "month", "highlight_text", "change_text", "completed_at",
    ]),
];

/// Return the syncable column whitelist for a given table, or `None` if unknown.
pub(crate) fn table_columns(table: &str) -> Option<&'static [&'static str]> {
    SYNCED_COLUMNS
        .iter()
        .find(|(t, _)| *t == table)
        .map(|(_, cols)| *cols)
}

/// Apply an upsert for a remote change.
///
/// Uses `INSERT ... ON CONFLICT(id) DO UPDATE` rather than `INSERT OR REPLACE`.
/// `INSERT OR REPLACE` deletes the conflicting row before reinserting it, which
/// fires `ON DELETE CASCADE` on children (annotations, subtasks, contact_*) and
/// would wipe them on any parent re-upsert. `ON CONFLICT DO UPDATE` mutates the
/// row in place, so children survive and referential integrity holds regardless
/// of FK enforcement. Every syncable table has `id` as its primary key.
#[tracing::instrument(skip_all)]
pub(crate) async fn apply_upsert(
    conn: &mut SqliteConnection,
    table: &str,
    _row_id: &str,
    data: &serde_json::Value,
) -> Result<(), CoreError> {
    // Email accounts use a bespoke ON CONFLICT that preserves local credentials
    if table == "email_accounts" {
        return apply_email_account_upsert(conn, data).await;
    }

    let columns = table_columns(table)
        .ok_or_else(|| CoreError::bad_request(format!("unknown syncable table: {}", table)))?;

    // A remote null for a NOT NULL column is invalid input the changelog never
    // emits. INSERT OR REPLACE silently coerced it to the column default; to
    // keep that tolerance under ON CONFLICT, omit such columns: a new row then
    // takes the schema default and an existing row keeps its current value.
    // Nullable columns keep null (so legitimate clears — e.g. un-snoozing —
    // still propagate). NOT NULL columns are derived from the live schema, not
    // a hand-maintained list, so they can't drift.
    let not_null = not_null_columns(conn, table).await?;
    let included: Vec<&str> = columns
        .iter()
        .copied()
        .filter(|c| !(data[*c].is_null() && not_null.contains(*c)))
        .collect();

    let col_list = included.join(", ");
    let placeholders = included.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
    let update_set = included
        .iter()
        .filter(|c| **c != "id")
        .map(|c| format!("{} = excluded.{}", c, c))
        .collect::<Vec<_>>()
        .join(", ");
    // If only `id` survived, there is nothing to update on conflict.
    let conflict = if update_set.is_empty() {
        "DO NOTHING".to_string()
    } else {
        format!("DO UPDATE SET {}", update_set)
    };
    let sql = format!(
        "INSERT INTO {} ({}) VALUES ({}) ON CONFLICT(id) {}",
        table, col_list, placeholders, conflict
    );

    let mut query = sqlx::query(&sql);

    for col in &included {
        query = bind_json_value(query, &data[*col]);
    }

    query
        .execute(&mut *conn)
        .await
        .map_err(CoreError::database)?;

    Ok(())
}

/// Names of the NOT NULL columns of a table, read from the live schema.
///
/// Derived via `PRAGMA table_info` rather than a static list so it can never
/// drift from the migrations. `table` is already validated against the column
/// whitelist by the caller, so interpolating it is safe.
async fn not_null_columns(
    conn: &mut SqliteConnection,
    table: &str,
) -> Result<std::collections::HashSet<String>, CoreError> {
    let rows = sqlx::query(&format!("PRAGMA table_info({})", table))
        .fetch_all(&mut *conn)
        .await
        .map_err(CoreError::database)?;

    let mut set = std::collections::HashSet::new();
    for row in rows {
        let notnull: i64 = row.try_get("notnull").map_err(CoreError::database)?;
        if notnull != 0 {
            let name: String = row.try_get("name").map_err(CoreError::database)?;
            set.insert(name);
        }
    }
    Ok(set)
}

/// Apply an upsert for email_accounts that preserves local credentials.
///
/// Uses INSERT ... ON CONFLICT(id) DO UPDATE to only touch the 16 config columns,
/// leaving `password`, `oauth2_access_token`, `oauth2_refresh_token`, and
/// `oauth2_token_expires_at` untouched on existing rows. New rows get `password = ''`
/// to satisfy the NOT NULL constraint.
#[tracing::instrument(skip_all)]
async fn apply_email_account_upsert(
    conn: &mut SqliteConnection,
    data: &serde_json::Value,
) -> Result<(), CoreError> {
    let cols = EMAIL_ACCOUNT_SYNC_COLS;

    // INSERT columns: 16 sync cols + password (hardcoded to '')
    let mut insert_cols: Vec<&str> = cols.to_vec();
    insert_cols.push("password");

    let col_list = insert_cols.join(", ");
    let placeholders = insert_cols.iter().map(|_| "?").collect::<Vec<_>>().join(", ");

    // ON CONFLICT: only update the 16 sync columns
    let update_set = cols
        .iter()
        .filter(|c| **c != "id")
        .map(|c| format!("{} = excluded.{}", c, c))
        .collect::<Vec<_>>()
        .join(", ");

    let sql = format!(
        "INSERT INTO email_accounts ({}) VALUES ({}) ON CONFLICT(id) DO UPDATE SET {}",
        col_list, placeholders, update_set
    );

    let mut query = sqlx::query(&sql);

    // Bind the 16 sync columns from data
    for col in cols {
        query = bind_json_value(query, &data[*col]);
    }

    // Bind password = '' for the INSERT
    query = query.bind("");

    query
        .execute(&mut *conn)
        .await
        .map_err(CoreError::database)?;

    Ok(())
}

/// Bind a JSON value to a sqlx query.
pub(crate) fn bind_json_value<'q>(
    query: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>,
    val: &'q serde_json::Value,
) -> sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>> {
    match val {
        serde_json::Value::String(s) => query.bind(s.as_str()),
        serde_json::Value::Number(n) => {
            if let Some(i) = n.as_i64() {
                query.bind(i)
            } else if let Some(f) = n.as_f64() {
                query.bind(f)
            } else {
                query.bind(None::<String>)
            }
        }
        serde_json::Value::Bool(b) => query.bind(if *b { 1i32 } else { 0i32 }),
        serde_json::Value::Null => query.bind(None::<String>),
        _ => {
            // Arrays/objects: serialize as JSON string -- need owned String
            query.bind(val.to_string())
        }
    }
}

/// Apply a DELETE for a remote change.
#[tracing::instrument(skip_all)]
pub(crate) async fn apply_delete(conn: &mut SqliteConnection, table: &str, row_id: &str) -> Result<(), CoreError> {
    // Validate table name is in our whitelist
    if table_columns(table).is_none() {
        return Err(CoreError::bad_request(format!("unknown syncable table: {}", table)));
    }

    let sql = format!("DELETE FROM {} WHERE id = ?", table);
    sqlx::query(&sql)
        .bind(row_id)
        .execute(&mut *conn)
        .await
        .map_err(CoreError::database)?;

    Ok(())
}