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::{db::outbox::StreamOutbox, types::IdentityEvt},
    jacquard_common::types::string::Handle,
    tracing::error,
};
pub(crate) mod backlink_ops;

pub mod record_events;

pub fn persist_to_resync_buffer(
    txn: &mut Txn<'_>,
    db: &Db,
    did: &Did,
    commit: &Commit,
) -> Result<()> {
    if txn.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(())
}

/// drops what `did` buffered at or below `rev`, for a backfill writing a car at `rev`. the car
/// already has those commits, and the drain applying one on top would roll the car back
pub fn stage_buffer_covered_by(
    batch: &mut OwnedWriteBatch,
    db: &Db,
    did: &Did,
    rev: DbTid,
) -> Result<()> {
    let covered = keys::resync_buffer_prefix(did)..=keys::resync_buffer_key(did, rev);
    for guard in db.indexer.resync_buffer.range(covered) {
        batch.remove(&db.indexer.resync_buffer, guard.key().into_diagnostic()?);
    }
    Ok(())
}

/// broadcast identity and account events staged by `stage`. these are
/// ephemeral: they take stream ids but are never persisted or replayed, so
/// consumers that need current state fetch it themselves.
#[cfg(feature = "indexer_stream")]
pub async fn announce(db: &Db, stage: impl FnOnce(&mut StreamOutbox) + Send + 'static) {
    let result = db
        .run(move |db| {
            let mut txn = Txn::new(db);
            stage(&mut txn.outbox.stream);
            txn.commit()
        })
        .await;
    if let Err(e) = result {
        error!(err = %e, "failed to announce stream events");
    }
}

/// announce an identity the caller learned by re-resolving a doc rather than by
/// receiving an event.
#[cfg(feature = "indexer_stream")]
pub async fn announce_identity(db: &Db, did: &Did, handle: Option<Handle>) {
    let identity = IdentityEvt {
        did: did.clone().into_static(),
        handle,
    };
    announce(db, move |events| events.push_identity(identity)).await;
}

/// the first txn of a deleted account's wipe: takes the repo off every queue and forgets what
/// it buffered, then leaves the marker its owner shard moves the heads to history from, a
/// step at a time. a running erase's marker stays, since the erase forgets the heads anyway
pub(crate) fn start_wipe(
    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)?;
        db.indexer
            .stage_pending_remove(&mut txn.batch, did, &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);

    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);
    }

    if matches!(
        db::read_bulk_job(db, did)?,
        Some(crate::types::BulkJobV1::Erase)
    ) {
        return Ok(());
    }
    // the death rev sorts after every rev we have seen, including future-skewed ones, and the
    // marker keeps it so every step files the heads under the same one
    let death_rev = db::death_rev(repo_state.root.as_ref().map(|c| &c.rev));
    db::stage_bulk_job(
        &mut txn.batch,
        db,
        did,
        &crate::types::BulkJobV1::Wipe { death_rev },
    )
}

pub fn transition_repo<'s>(
    txn: &mut Txn<'_>,
    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 db = txn.db;
    let batch = &mut txn.batch;
    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 mut metadata = db::deser_repo_meta(&metadata_bytes)?;
        let pending_key = keys::pending_key(metadata.index_id);
        // the retry worker only requeues these, any other status would strand a deferred refetch
        let retries = repo_state.active
            && matches!(
                new_status,
                RepoStatus::Error(_) | RepoStatus::Desynchronized | RepoStatus::Throttled
            );
        if !retries && std::mem::take(&mut metadata.refetch_deferred) {
            batch.insert(
                &db.repo_metadata,
                &metadata_key,
                db::ser_repo_meta(&metadata)?,
            );
        }

        // manage queues
        match &new_status {
            RepoStatus::Synced => {
                lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced));
                db.indexer.stage_pending_remove(batch, did, &pending_key);
                batch.remove(&db.indexer.resync, &repo_key);
            }
            // the relay writes these for inactive accounts too, and a pds won't hand those over
            // any more than a deactivated one, so a retry would only refetch into the same answer
            RepoStatus::Error(_) | RepoStatus::Desynchronized | RepoStatus::Throttled
                if !repo_state.active =>
            {
                stage_gone(batch, db, lifecycle_transitions, did, &new_status)?;
            }
            RepoStatus::Error(msg) => {
                tracing::warn!("transitioning to error: {msg}");
                lifecycle_transitions.push((
                    did.clone().into_static(),
                    GaugeState::Resync(Some(ResyncErrorKind::Generic)),
                ));
                db.indexer.stage_pending_remove(batch, did, &pending_key);
                // 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 => {
                stage_gone(batch, db, lifecycle_transitions, did, &new_status)?;
            }
            RepoStatus::Deleted => {
                lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced));
                // terminal state: remove from queues, no resync entry needed
                db.indexer.stage_pending_remove(batch, did, &pending_key);
                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
                db.indexer.stage_pending_remove(batch, did, &pending_key);
                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 = match new_status {
        RepoStatus::Synced => true,
        RepoStatus::Error(_) => repo_state.active,
        _ => false,
    };
    repo_state.status = new_status;
    repo_state.touch();
    txn.put_repo(did, &repo_state)?;

    Ok(repo_state)
}

/// parks a repo whose pds won't hand it over until its account comes back. the queue is the
/// caller's to clear
fn stage_gone(
    batch: &mut fjall::OwnedWriteBatch,
    db: &Db,
    lifecycle_transitions: &mut Vec<(Did, GaugeState)>,
    did: &Did,
    status: &RepoStatus,
) -> Result<()> {
    lifecycle_transitions.push((did.clone().into_static(), GaugeState::Resync(None)));
    let resync_state = ResyncState::Gone {
        status: status.clone(),
    };
    batch.insert(
        &db.indexer.resync,
        keys::repo_key(did),
        rmp_serde::to_vec(&resync_state).into_diagnostic()?,
    );
    Ok(())
}

pub struct ApplyCommitResults<'s> {
    pub repo_state: RepoState<'s>,
}

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");

    // the relay puts each commit's root on the row before it gets here, so the row can already
    // be a commit or more ahead, and the relay checks the next commit's since against it
    let root: crate::types::Commit = validated.commit_obj.into();
    if repo_state
        .root
        .as_ref()
        .is_none_or(|current| current.rev < root.rev)
    {
        repo_state.root = Some(root);
    }
    repo_state.touch();

    // the shard commits its batch even when this fails, so everything the commit itself can get
    // wrong is checked before the first write
    let mut writes = Vec::with_capacity(commit.ops.len());
    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())?;

        let block = 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(block) = parsed.blocks.block(&cid_ipld) else {
                    return Err(miette::miette!(
                        "block {cid} not found in CAR for record {did}/{collection}/{rkey}"
                    ));
                };
                let block = block
                    .checked()
                    .map_err(|e| miette::miette!("record {did}/{collection}/{rkey}: {e}"))?;
                Some(block)
            }
            DbAction::Delete => None,
        };
        writes.push((collection, rkey, action, block));
    }

    let mut record_txn = txn.records(state, &DbTid::from(&commit.rev), did)?;
    for (collection, rkey, action, block) in &writes {
        match block {
            Some(block) => record_txn.put_record(collection, rkey, block, *action)?,
            None => record_txn.delete_record(collection, rkey)?,
        }
    }

    record_txn.update_repo_state(&repo_state)?;

    Ok(ApplyCommitResults { repo_state })
}

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 crate::types::{AccountEvt, BroadcastEvent, MarshallableEvt};
    use jacquard_common::{CowStr, types::string::Handle};
    use tokio::sync::broadcast;

    fn next_ephemeral(
        rx: &mut broadcast::Receiver<BroadcastEvent>,
    ) -> Box<MarshallableEvt<'static>> {
        let Ok(BroadcastEvent::Ephemeral(event)) = rx.try_recv() else {
            panic!("expected an ephemeral event");
        };
        event
    }

    #[tokio::test]
    async 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 mut rx = db.stream.event_tx.subscribe();
        let did = Did::new_static("did:web:example.com").unwrap();

        announce_identity(&db, &did, Some(Handle::new_static("alice.test").unwrap())).await;
        let identity = next_ephemeral(&mut rx);
        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 = AccountEvt {
            did,
            active: false,
            status: Some(CowStr::Borrowed("stinkpot")),
        };
        announce(&db, move |events| events.push_account(account)).await;
        let account = next_ephemeral(&mut rx);
        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());

        // ephemeral events take ids but are never persisted
        assert_eq!(db.stream.ids.head(), Some(account.id));
        assert!(db.stream.event_head().unwrap().is_none());
    }
}