use std::sync::Arc;

use serde_json::Value;

use crate::AppState;
use crate::db::{adapt_sql, now_rfc3339};
use crate::event_log::{EventLog, Severity, log_event};
use crate::lexicon::{LexiconType, ParsedLexicon, ProcedureAction};

/// The static collection we always include for lexicon schema updates.
pub const LEXICON_SCHEMA_COLLECTION: &str = "com.atproto.lexicon.schema";

/// A generic record event that can originate from any source (Jetstream, backfill, etc.).
pub struct RecordEvent {
    pub did: String,
    pub collection: String,
    pub rkey: String,
    pub action: String,
    pub record: Option<Value>,
    pub cid: Option<String>,
}

/// Process a record event: upsert/delete the record in the database, run index
/// hooks, and handle lexicon schema events.
pub async fn handle_record_event(state: &AppState, record: &RecordEvent) {
    let db = &state.db;
    let lexicons = &state.lexicons;

    let uri = format!("at://{}/{}/{}", record.did, record.collection, record.rkey);

    // Handle lexicon schema events for tracked network lexicons.
    if record.collection == LEXICON_SCHEMA_COLLECTION {
        handle_lexicon_schema_event(state, &record.did, record).await;
        return;
    }

    // Skip records whose collection is not tracked by a registered record-type lexicon.
    let lex = lexicons.get(&record.collection).await;
    let is_tracked = lex
        .as_ref()
        .is_some_and(|l| l.lexicon_type == LexiconType::Record);

    if !is_tracked {
        tracing::debug!(
            collection = %record.collection,
            "skipping record for untracked collection"
        );
        return;
    }

    // Skip records for local-only lexicons. These should never arrive from the
    // firehose or backfill in the first place, but defense in depth: if a remote
    // record happens to share a URI with a local-only collection (or someone
    // republishes one), we refuse to overwrite the local row.
    if lex.as_ref().is_some_and(|l| l.is_local) {
        tracing::debug!(
            collection = %record.collection,
            uri = %uri,
            "skipping network-sourced record for local-only collection"
        );
        return;
    }

    match record.action.as_str() {
        "create" | "update" => {
            let rec = match &record.record {
                Some(r) => r,
                None => return,
            };
            let cid = record.cid.as_deref().unwrap_or_default();

            // Run any service_hooks registered for this action+collection
            // through the shared dispatcher. Fail-open: a dead-lettered
            // hook falls through with the original record so a buggy
            // script can't permanently block the firehose.
            let kind = match record.action.as_str() {
                "create" => crate::lua::ServiceHookKind::RecordCreated(record.collection.clone()),
                "update" => crate::lua::ServiceHookKind::RecordUpdated(record.collection.clone()),
                _ => unreachable!("outer match guards action to create|update"),
            };
            let outcome = crate::lua::dispatch_record_event(
                state,
                kind,
                &record.action,
                &uri,
                &record.did,
                &record.collection,
                &record.rkey,
                Some(rec),
            )
            .await;

            let rec_to_store = match outcome {
                crate::lua::RecordHookOutcome::Continue { record: Some(v) } => v,
                crate::lua::RecordHookOutcome::Continue { record: None } => {
                    // Hook cleared the body for a create/update — treat
                    // as skip rather than indexing an empty record.
                    log_event(
                        db,
                        EventLog {
                            event_type: "record.skipped".to_string(),
                            severity: Severity::Info,
                            actor_did: None,
                            subject: Some(uri.clone()),
                            detail: serde_json::json!({
                                "collection": record.collection,
                                "did": record.did,
                                "rkey": record.rkey,
                                "reason": "hook cleared record body",
                            }),
                        },
                        state.db_backend,
                    )
                    .await;
                    return;
                }
                crate::lua::RecordHookOutcome::Skip => {
                    log_event(
                        db,
                        EventLog {
                            event_type: "record.skipped".to_string(),
                            severity: Severity::Info,
                            actor_did: None,
                            subject: Some(uri.clone()),
                            detail: serde_json::json!({
                                "collection": record.collection,
                                "did": record.did,
                                "rkey": record.rkey,
                                "reason": "hook returned nil",
                            }),
                        },
                        state.db_backend,
                    )
                    .await;
                    return;
                }
            };

            let now = now_rfc3339();
            let backend = state.db_backend;
            let insert_sql = adapt_sql(
                r#"
                INSERT INTO records (uri, did, collection, rkey, record, cid, indexed_at, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT (uri) DO UPDATE
                    SET record = EXCLUDED.record,
                        cid = EXCLUDED.cid,
                        indexed_at = ?
                "#,
                backend,
            );
            match sqlx::query(&insert_sql)
                .bind(&uri)
                .bind(&record.did)
                .bind(&record.collection)
                .bind(&record.rkey)
                .bind(serde_json::to_string(&rec_to_store).unwrap_or_default())
                .bind(cid)
                .bind(&now)
                .bind(&now)
                .bind(&now)
                .execute(db)
                .await
            {
                Ok(_) => {
                    let _ = crate::record_refs::sync_refs(
                        db,
                        &uri,
                        &record.collection,
                        &rec_to_store,
                        backend,
                    )
                    .await;

                    log_event(
                        db,
                        EventLog {
                            event_type: "record.created".to_string(),
                            severity: Severity::Info,
                            actor_did: None,
                            subject: Some(uri.clone()),
                            detail: serde_json::json!({
                                "collection": record.collection,
                                "did": record.did,
                                "rkey": record.rkey,
                            }),
                        },
                        backend,
                    )
                    .await;

                    crate::labeler::backfill_labels_for_uri(Arc::new(state.clone()), uri.clone());
                }
                Err(e) => {
                    tracing::warn!(uri = %uri, "failed to upsert record: {e}");
                    log_event(
                        db,
                        EventLog {
                            event_type: "record.created".to_string(),
                            severity: Severity::Error,
                            actor_did: None,
                            subject: Some(uri.clone()),
                            detail: serde_json::json!({
                                "collection": record.collection,
                                "did": record.did,
                                "rkey": record.rkey,
                                "error": e.to_string(),
                            }),
                        },
                        backend,
                    )
                    .await;
                }
            }
        }
        "delete" => {
            let backend = state.db_backend;

            // Run any service_hooks for record.deleted:<nsid>. The
            // record body is None throughout — hooks can only inspect
            // the URI/DID/collection/rkey and decide whether to skip.
            let outcome = crate::lua::dispatch_record_event(
                state,
                crate::lua::ServiceHookKind::RecordDeleted(record.collection.clone()),
                "delete",
                &uri,
                &record.did,
                &record.collection,
                &record.rkey,
                None,
            )
            .await;

            if matches!(outcome, crate::lua::RecordHookOutcome::Skip) {
                log_event(
                    db,
                    EventLog {
                        event_type: "record.skipped".to_string(),
                        severity: Severity::Info,
                        actor_did: None,
                        subject: Some(uri.clone()),
                        detail: serde_json::json!({
                            "collection": record.collection,
                            "did": record.did,
                            "rkey": record.rkey,
                            "reason": "hook returned nil",
                        }),
                    },
                    backend,
                )
                .await;
                return;
            }

            let delete_sql = adapt_sql("DELETE FROM records WHERE uri = ?", backend);
            match sqlx::query(&delete_sql).bind(&uri).execute(db).await {
                Ok(_) => {
                    log_event(
                        db,
                        EventLog {
                            event_type: "record.deleted".to_string(),
                            severity: Severity::Info,
                            actor_did: None,
                            subject: Some(uri.clone()),
                            detail: serde_json::json!({
                                "collection": record.collection,
                                "did": record.did,
                                "rkey": record.rkey,
                            }),
                        },
                        backend,
                    )
                    .await;
                }
                Err(e) => {
                    tracing::warn!(uri = %uri, "failed to delete record: {e}");
                    log_event(
                        db,
                        EventLog {
                            event_type: "record.deleted".to_string(),
                            severity: Severity::Error,
                            actor_did: None,
                            subject: Some(uri.clone()),
                            detail: serde_json::json!({
                                "collection": record.collection,
                                "did": record.did,
                                "rkey": record.rkey,
                                "error": e.to_string(),
                            }),
                        },
                        backend,
                    )
                    .await;
                }
            }
        }
        _ => {}
    }
}

/// Handle a `com.atproto.lexicon.schema` record event for tracked network lexicons.
pub async fn handle_lexicon_schema_event(state: &AppState, did: &str, record: &RecordEvent) {
    let db = &state.db;
    let lexicons = &state.lexicons;
    let collections_tx = &state.collections_tx;
    let nsid = &record.rkey;

    let backend = state.db_backend;

    // Check if this NSID is one we're tracking and the DID matches the authority.
    let select_sql = adapt_sql(
        "SELECT target_collection FROM lexicons WHERE id = ? AND source = 'network' AND authority_did = ?",
        backend,
    );
    let tracked: Option<(Option<String>,)> = sqlx::query_as(&select_sql)
        .bind(nsid)
        .bind(did)
        .fetch_optional(db)
        .await
        .unwrap_or(None);

    let target_collection = match tracked {
        Some((tc,)) => tc,
        None => return, // Not a tracked network lexicon.
    };

    match record.action.as_str() {
        "create" | "update" => {
            let rec = match &record.record {
                Some(r) => r,
                None => return,
            };

            let parsed = match ParsedLexicon::parse(
                rec.clone(),
                1,
                target_collection.clone(),
                ProcedureAction::Upsert,
                None,
                None,
                None,
                false,
            ) {
                Ok(p) => p,
                Err(e) => {
                    tracing::warn!(nsid, "failed to parse lexicon schema event: {e}");
                    return;
                }
            };

            let is_record = parsed.lexicon_type == crate::lexicon::LexiconType::Record;

            // Upsert into lexicons table with last_fetched_at.
            let now = now_rfc3339();
            let upsert_sql = adapt_sql(
                r#"
                INSERT INTO lexicons (id, lexicon_json, backfill, target_collection, source, authority_did, last_fetched_at, created_at)
                VALUES (?, ?, 0, ?, 'network', ?, ?, ?)
                ON CONFLICT (id) DO UPDATE SET
                    lexicon_json = EXCLUDED.lexicon_json,
                    target_collection = EXCLUDED.target_collection,
                    last_fetched_at = ?,
                    revision = lexicons.revision + 1,
                    updated_at = ?
                "#,
                backend,
            );
            if let Err(e) = sqlx::query(&upsert_sql)
                .bind(nsid)
                .bind(serde_json::to_string(rec).unwrap_or_default())
                .bind(&target_collection)
                .bind(did)
                .bind(&now)
                .bind(&now)
                .bind(&now)
                .bind(&now)
                .execute(db)
                .await
            {
                tracing::warn!(nsid, "failed to upsert lexicon from event: {e}");
                return;
            }

            lexicons.upsert(parsed).await;
            tracing::info!(nsid, "updated network lexicon from network event");

            if is_record {
                let collections = lexicons.get_record_collections().await;
                let _ = collections_tx.send(collections);
            }
        }
        "delete" => {
            // Remove from lexicons table and registry.
            let delete_sql = adapt_sql("DELETE FROM lexicons WHERE id = ?", backend);
            let _ = sqlx::query(&delete_sql).bind(nsid).execute(db).await;

            let was_present = lexicons.remove(nsid).await;
            if was_present {
                tracing::info!(nsid, "removed network lexicon from network delete event");
                let collections = lexicons.get_record_collections().await;
                let _ = collections_tx.send(collections);
            }
        }
        _ => {}
    }
}