use fjall::OwnedWriteBatch;

use jacquard_common::IntoStatic;
use jacquard_common::types::did::Did;
use miette::{Context, IntoDiagnostic, Result};
use tracing::debug;

use crate::db::types::{DbAction, DbRkey, DbTid};
use crate::db::{self, Db, Txn, keys};
use crate::filter::FilterConfig;
use crate::ingest::stream::Commit;
use crate::ingest::validation::ValidatedCommit;
use crate::state::AppState;
use crate::types::{GaugeState, RepoState, RepoStatus, ResyncErrorKind, ResyncState};

#[cfg(feature = "indexer_stream")]
use {
    crate::types::{AccountEvt, BroadcastEvent, IdentityEvt, MarshallableEvt},
    jacquard_common::types::string::Handle,
    std::sync::atomic::Ordering,
};
pub(crate) mod backlink_ops;
use record_events::RecordEvents;

pub mod record_events;

pub fn persist_to_resync_buffer(
    txn: &mut Txn<'_>,
    db: &Db,
    did: &Did,
    commit: &Commit,
) -> Result<()> {
    if txn.lock_repo_and_is_excluded(did)? {
        return Ok(());
    }
    let key = keys::resync_buffer_key(did, DbTid::from(&commit.rev));
    let value = rmp_serde::to_vec_named(commit).into_diagnostic()?;
    txn.batch.insert(&db.indexer.resync_buffer, key, value);
    debug!(
        did = %did,
        seq = commit.seq,
        "buffered commit to resync_buffer"
    );
    Ok(())
}

// emitting identity is ephemeral
// we dont replay these, consumers can just fetch identity themselves if they need it
#[cfg(feature = "indexer_stream")]
pub fn make_identity_event(db: &Db, evt: IdentityEvt) -> BroadcastEvent {
    let event_id = db.stream.next_event_id.fetch_add(1, Ordering::SeqCst);
    let marshallable = MarshallableEvt {
        id: event_id,
        kind: crate::types::EventType::Identity,
        record: None,
        identity: Some(evt),
        account: None,
    };
    BroadcastEvent::Ephemeral(Box::new(marshallable))
}

/// announce an identity the caller learned by re-resolving a doc rather than by
/// receiving an event. ephemeral, like every other identity event.
#[cfg(feature = "indexer_stream")]
pub fn announce_identity(db: &Db, did: &Did, handle: Option<Handle>) {
    let evt = IdentityEvt {
        did: did.clone(),
        handle,
    };
    let _ = db.stream.event_tx.send(make_identity_event(db, evt));
}

#[cfg(feature = "indexer_stream")]
pub fn make_account_event(db: &Db, evt: AccountEvt<'static>) -> BroadcastEvent {
    let event_id = db.stream.next_event_id.fetch_add(1, Ordering::SeqCst);
    let marshallable = MarshallableEvt {
        id: event_id,
        kind: crate::types::EventType::Account,
        record: None,
        identity: None,
        account: Some(evt),
    };
    BroadcastEvent::Ephemeral(Box::new(marshallable))
}

pub(crate) fn delete_repo(
    txn: &mut Txn<'_>,
    db: &Db,
    did: &Did,
    repo_state: &RepoState,
) -> Result<()> {
    debug!(did = %did, "deleting repo");

    let repo_key = keys::repo_key(did);
    let metadata_key = keys::repo_metadata_key(did);

    let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?;
    if let Some(metadata_bytes) = metadata_bytes {
        let metadata = db::deser_repo_meta(&metadata_bytes)?;
        txn.batch
            .remove(&db.indexer.pending, keys::pending_key(metadata.index_id));
    }

    // we don't delete from repos, relay uses it as a tombstone
    // todo: we should still delete it after some time
    txn.batch.remove(&db.indexer.resync, &repo_key);
    txn.batch.remove(&db.repo_metadata, &metadata_key);

    // 2. delete from resync buffer
    let resync_prefix = keys::resync_buffer_prefix(did);
    for guard in db.indexer.resync_buffer.prefix(&resync_prefix) {
        let k = guard.key().into_diagnostic()?;
        txn.batch.remove(&db.indexer.resync_buffer, k);
    }

    // 3. move record bodies to history, then delete the heads. history keeps
    // them resolvable for event replay; the synthetic death rev just needs to
    // sort strictly after every real rev we have seen (including future-skewed ones)
    let root_rev = repo_state.root.as_ref().map(|c| &c.rev);
    db::delete_repo_records(txn, db, did, root_rev)?;

    // 5. remove backlinks for all records in this repo
    backlink_ops::delete_repo(&mut txn.batch, db, did)?;

    Ok(())
}

pub fn transition_repo<'s>(
    batch: &mut OwnedWriteBatch,
    db: &Db,
    lifecycle_transitions: &mut Vec<(Did, GaugeState)>,
    did: &Did,
    mut repo_state: RepoState<'s>,
    new_status: RepoStatus,
) -> Result<RepoState<'s>> {
    debug!(did = %did, status = ?new_status, "updating repo status");

    let repo_key = keys::repo_key(did);
    let metadata_key = keys::repo_metadata_key(did);

    let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?;
    if let Some(metadata_bytes) = metadata_bytes {
        let metadata = db::deser_repo_meta(&metadata_bytes)?;
        let pending_key = keys::pending_key(metadata.index_id);

        // manage queues
        match &new_status {
            RepoStatus::Synced => {
                lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced));
                batch.remove(&db.indexer.pending, pending_key.as_slice());
                batch.remove(&db.indexer.resync, &repo_key);
            }
            RepoStatus::Error(msg) => {
                tracing::warn!("transitioning to error: {msg}");
                lifecycle_transitions.push((
                    did.clone().into_static(),
                    GaugeState::Resync(Some(ResyncErrorKind::Generic)),
                ));
                batch.remove(&db.indexer.pending, pending_key.as_slice());
                // TODO: we need to make errors have kind instead of "message" in repo status
                // and then pass it to resync error kind
                let resync_state = crate::types::ResyncState::Error {
                    kind: crate::types::ResyncErrorKind::Generic,
                    retry_count: 0,
                    next_retry: chrono::Utc::now().timestamp(),
                };
                batch.insert(
                    &db.indexer.resync,
                    &repo_key,
                    rmp_serde::to_vec(&resync_state).into_diagnostic()?,
                );
            }
            RepoStatus::Deactivated | RepoStatus::Takendown | RepoStatus::Suspended => {
                lifecycle_transitions.push((did.clone().into_static(), GaugeState::Resync(None)));
                // this shouldnt be needed since a repo wont be in a pending state when it gets to any of these states
                // batch.remove(&db.pending, &pending_key);
                let resync_state = ResyncState::Gone {
                    status: new_status.clone(),
                };
                batch.insert(
                    &db.indexer.resync,
                    &repo_key,
                    rmp_serde::to_vec(&resync_state).into_diagnostic()?,
                );
            }
            RepoStatus::Deleted => {
                lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced));
                // terminal state: remove from queues, no resync entry needed
                batch.remove(&db.indexer.pending, pending_key.as_slice());
                batch.remove(&db.indexer.resync, &repo_key);
            }
            RepoStatus::Desynchronized | RepoStatus::Throttled => {
                lifecycle_transitions.push((
                    did.clone().into_static(),
                    GaugeState::Resync(Some(ResyncErrorKind::Generic)),
                ));
                // like an error: remove from pending and schedule a resync attempt
                batch.remove(&db.indexer.pending, pending_key.as_slice());
                let resync_state = crate::types::ResyncState::Error {
                    kind: crate::types::ResyncErrorKind::Generic,
                    retry_count: 0,
                    next_retry: chrono::Utc::now().timestamp(),
                };
                batch.insert(
                    &db.indexer.resync,
                    &repo_key,
                    rmp_serde::to_vec(&resync_state).into_diagnostic()?,
                );
            }
        }
    }

    repo_state.active = matches!(new_status, RepoStatus::Synced | RepoStatus::Error(_));
    repo_state.status = new_status;
    repo_state.touch();
    batch.insert(&db.repos, repo_key, db::ser_repo_state(&repo_state)?);

    Ok(repo_state)
}

pub struct ApplyCommitResults<'s> {
    pub repo_state: RepoState<'s>,
    #[cfg_attr(not(feature = "indexer_stream"), allow(dead_code))]
    pub events: RecordEvents,
}

pub fn apply_commit<'s>(
    txn: &mut crate::db::Txn<'_>,
    state: &AppState,
    mut repo_state: RepoState<'s>,
    validated: ValidatedCommit<'_>,
    filter: &FilterConfig,
) -> Result<ApplyCommitResults<'s>> {
    let commit = validated.commit;
    let parsed = validated.parsed_blocks;
    let did = &commit.repo;
    debug!(did = %did, commit = %commit.commit, "applying commit");

    repo_state.root = Some(validated.commit_obj.into());
    repo_state.touch();

    let mut record_txn = txn.records(state, &DbTid::from(&commit.rev), did)?;

    for op in &commit.ops {
        let (collection, rkey) = parse_path(&op.path)?;

        if !filter.matches_collection(collection) {
            continue;
        }

        let rkey = DbRkey::new(rkey);
        let action = DbAction::try_from(op.action.as_str())?;

        match action {
            DbAction::Create | DbAction::Update => {
                let Some(cid) = &op.cid else {
                    continue;
                };
                let cid_ipld = cid
                    .to_ipld()
                    .into_diagnostic()
                    .wrap_err("expected valid cid from relay")?;
                let Some(bytes) = parsed.blocks.get(&cid_ipld) else {
                    return Err(miette::miette!(
                        "block {cid} not found in CAR for record {did}/{collection}/{rkey}"
                    ));
                };
                record_txn.put_record(collection, &rkey, &cid_ipld, bytes, action)?;
            }
            DbAction::Delete => {
                record_txn.delete_record(collection, &rkey)?;
            }
        }
    }

    record_txn.update_repo_state(&repo_state)?;
    let events = record_txn.finish()?;

    Ok(ApplyCommitResults { repo_state, events })
}

pub fn parse_path(path: &str) -> Result<(&str, &str)> {
    let mut parts = path.splitn(2, '/');
    let collection = parts.next().wrap_err("missing collection")?;
    let rkey = parts.next().wrap_err("missing rkey")?;

    if collection.is_empty() || jacquard_common::types::nsid::Nsid::new(collection).is_err() {
        miette::bail!("invalid collection NSID: {collection}");
    }
    if rkey.is_empty() || jacquard_common::types::string::Rkey::new(rkey).is_err() {
        miette::bail!("invalid record key (rkey): {rkey}");
    }

    Ok((collection, rkey))
}

#[cfg(all(test, feature = "indexer_stream"))]
mod tests {
    use super::*;
    use crate::config::Config;
    use jacquard_common::{CowStr, types::string::Handle};

    fn ephemeral(event: BroadcastEvent) -> Box<MarshallableEvt<'static>> {
        let BroadcastEvent::Ephemeral(event) = event else {
            panic!("expected ephemeral event");
        };
        event
    }

    #[test]
    fn ephemeral_identity_and_account_json_contracts() {
        let tmp = tempfile::tempdir().unwrap();
        let db = Db::open(&Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        })
        .unwrap();
        let did = Did::new_static("did:web:example.com").unwrap();

        let identity = ephemeral(make_identity_event(
            &db,
            IdentityEvt {
                did: did.clone(),
                handle: Some(Handle::new_static("alice.test").unwrap()),
            },
        ));
        let identity_json = serde_json::to_value(identity.as_ref()).unwrap();
        assert_eq!(identity_json["type"], "identity");
        assert_eq!(identity_json["identity"]["did"], "did:web:example.com");
        assert_eq!(identity_json["identity"]["handle"], "alice.test");
        assert!(identity_json.get("record").is_none());
        assert!(identity_json.get("account").is_none());

        let account = ephemeral(make_account_event(
            &db,
            AccountEvt {
                did,
                active: false,
                status: Some(CowStr::Borrowed("stinkpot")),
            },
        ));
        let account_json = serde_json::to_value(account.as_ref()).unwrap();
        assert_eq!(account_json["id"].as_u64(), Some(identity.id + 1));
        assert_eq!(account_json["type"], "account");
        assert_eq!(account_json["account"]["did"], "did:web:example.com");
        assert_eq!(account_json["account"]["active"], false);
        assert_eq!(account_json["account"]["status"], "stinkpot");
        assert!(account_json.get("record").is_none());
        assert!(account_json.get("identity").is_none());
    }
}