use futures::FutureExt;
use miette::{IntoDiagnostic, Result};
use std::future::Future;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::sync::watch;
use tracing::{debug, error, info, warn};
use url::Url;

use crate::config::{Config, SignatureVerification};
use crate::db::{self, filter as db_filter};
use crate::filter::FilterMode;
use crate::state::AppState;

use super::firehose::{FirehoseHandle, FirehoseShared};
use super::{DbControl, FilterControl, ReposControl, pds::PdsControl};

#[cfg(feature = "indexer")]
use super::{BackfillHandle, crawler};
#[cfg(feature = "indexer")]
use crate::backfill::BackfillWorker;
#[cfg(feature = "indexer")]
use crate::db::load_persisted_crawler_sources;
#[cfg(feature = "indexer")]
use crate::ingest::indexer::FirehoseWorker;

#[cfg(feature = "backlinks")]
use crate::backlinks::BacklinksControl;

use super::seed;

pub(crate) mod run;

/// the top-level handle to a hydrant instance.
///
/// `Hydrant` is cheaply cloneable. all sub-handles share the same underlying state.
/// construct it via [`Hydrant::new`] or [`Hydrant::from_env`], configure the filter
/// and repos as needed, then call [`Hydrant::run`] to start all background components.
#[derive(Clone)]
pub struct Hydrant {
    #[cfg(feature = "indexer")]
    pub crawler: crawler::CrawlerHandle,
    pub firehose: FirehoseHandle,
    #[cfg(feature = "indexer")]
    pub backfill: BackfillHandle,
    pub filter: FilterControl,
    pub pds: PdsControl,
    pub repos: ReposControl,
    pub db: DbControl,
    #[cfg(feature = "backlinks")]
    pub backlinks: crate::backlinks::BacklinksControl,
    pub(crate) state: Arc<AppState>,
    pub(crate) config: Arc<Config>,
    pub(crate) started: Arc<AtomicBool>,
    _priv: (),
}

impl Hydrant {
    /// Returns a reference to the internal DID resolver.
    pub fn resolver(&self) -> &crate::resolver::Resolver {
        &self.state.resolver
    }

    #[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))]
    pub(crate) fn stream_send_timeout(&self) -> std::time::Duration {
        self.config.stream_send_timeout
    }

    /// open the database and configure hydrant from `config`.
    ///
    /// this sets up the database, applies any filter configuration from `config`, and
    /// initializes all sub-handles. no background tasks are started yet: call
    /// [`run`](Self::run) to start all components and drive the instance.
    pub async fn new(config: Config) -> Result<Self> {
        Self::new_inner(config, None).await
    }

    #[cfg(all(test, feature = "indexer_stream"))]
    pub(crate) async fn new_with_resolver(
        config: Config,
        resolver: crate::resolver::Resolver,
    ) -> Result<Self> {
        Self::new_inner(config, Some(resolver)).await
    }

    async fn new_inner(
        config: Config,
        resolver: Option<crate::resolver::Resolver>,
    ) -> Result<Self> {
        info!("{config}");

        #[cfg(feature = "relay")]
        if config.only_index_links {
            miette::bail!("HYDRANT_ONLY_INDEX_LINKS is not supported in relay mode");
        }

        // 1. open database and construct AppState
        let mut state = AppState::new(&config)?;
        if let Some(resolver) = resolver {
            state.resolver = resolver;
        }

        // 2. apply any filter config from env variables
        if config.full_network
            || config.filter_signals.is_some()
            || config.filter_collections.is_some()
            || config.filter_excludes.is_some()
        {
            let mode = config.full_network.then_some(FilterMode::Full);
            let signals = config
                .filter_signals
                .clone()
                .map(crate::patch::SetUpdate::Set);
            let collections = config
                .filter_collections
                .clone()
                .map(crate::patch::SetUpdate::Set);
            let excludes = config
                .filter_excludes
                .clone()
                .map(crate::patch::SetUpdate::Set);

            state
                .db
                .run(move |db| {
                    let mut batch = db.inner.batch();
                    db_filter::apply_patch(
                        &mut batch,
                        &db.filter,
                        mode,
                        signals,
                        collections,
                        excludes,
                    )?;
                    batch.commit().into_diagnostic()
                })
                .await?;

            // 3. reload the live filter into the hot-path arc-swap
            let new_filter = state.db.run(move |db| db_filter::load(&db.filter)).await?;
            state.filter.store(Arc::new(new_filter));
        }

        #[cfg(feature = "indexer")]
        {
            // 4. set crawler enabled state from config, evaluated against the post-patch filter
            let post_patch_crawler = match config.enable_crawler {
                Some(b) => b,
                None => {
                    state.filter.load().mode == FilterMode::Full
                        || !config.crawler_sources.is_empty()
                }
            };
            state.crawler_enabled.send_replace(post_patch_crawler);
        }

        let state = Arc::new(state);

        Ok(Self {
            firehose: FirehoseHandle::new(state.clone()),
            filter: FilterControl(state.clone()),
            pds: PdsControl(state.clone()),
            repos: ReposControl(state.clone()),
            db: DbControl(state.clone()),
            #[cfg(feature = "indexer")]
            crawler: crawler::CrawlerHandle {
                state: state.clone(),
                shared: Arc::new(std::sync::OnceLock::new()),
                tasks: Arc::new(scc::HashMap::new()),
                persisted: Arc::new(scc::HashSet::new()),
            },
            #[cfg(feature = "indexer")]
            backfill: BackfillHandle::new(state.clone()),
            #[cfg(feature = "backlinks")]
            backlinks: crate::backlinks::BacklinksControl(state.clone()),
            state,
            config: Arc::new(config),
            started: Arc::new(AtomicBool::new(false)),
            _priv: (),
        })
    }

    /// reads config from environment variables and calls [`Hydrant::new`].
    pub async fn from_env() -> Result<Self> {
        Self::new(Config::from_env()?).await
    }

    /// returns a future that runs the HTTP management API server on the given bind addresses.
    ///
    /// the server exposes all management endpoints (`/filter`, `/repos`, `/ingestion`,
    /// `/stream`, `/stats`, `/db/*`, `/xrpc/*`). it runs indefinitely and resolves
    /// only on error.
    ///
    /// intended for `tokio::spawn` or inclusion in a `select!` / task list. the clone
    /// of `self` is deferred until the future is first polled.
    ///
    /// to disable the HTTP API entirely, simply don't call this method.
    pub fn serve(&self, binds: super::ApiBinds) -> impl Future<Output = Result<()>> {
        let hydrant = self.clone();
        async move { crate::api::serve(hydrant, binds).await }
    }

    /// returns a future that runs the debug HTTP API server on `127.0.0.1:{port}`.
    ///
    /// exposes internal inspection endpoints (`/debug/get`, `/debug/iter`, etc.).
    /// binds only to loopback.
    pub fn serve_debug(&self, port: u16) -> impl Future<Output = Result<()>> {
        let state = self.state.clone();
        async move { crate::api::serve_debug(state, port).await }
    }

    /// iterate sequentially over the current head body of every record in
    /// `collection`, ordered by (repo, rkey).
    ///
    /// values are raw DAG-CBOR. records are keyed repo-major, so this walks
    /// the whole records keyspace and filters; superseded bodies live in the
    /// history keyspace and are not included.
    #[cfg(feature = "indexer")]
    pub fn scan_collection_blocks(
        &self,
        collection: &str,
    ) -> impl Iterator<Item = Result<ScannedBlock>> + 'static {
        let db = self.state.db.indexer.clone();
        let want = collection.to_string();

        let iter = db.record_range::<&[u8], _>(..);
        iter.filter_map(move |item| {
            let (key, value) = match item.into_inner() {
                Ok(kv) => kv,
                Err(e) => return Some(Err(miette::miette!("{e}"))),
            };
            let Some(did_len) = crate::db::keys::indexer::trimmed_did_len(&key) else {
                return None;
            };
            let Ok((col, _)) = crate::db::keys::indexer::split_record_suffix(&key[did_len + 1..])
            else {
                return None;
            };
            if col != want {
                return None;
            }
            // links-only mode stores raw cids, not bodies; there is nothing
            // to scan (a dag-cbor record is a map and never carries the
            // cidv1-dagcbor-sha256 prefix)
            if crate::db::is_cid_record_value(&value) {
                return None;
            }
            // the scanned key keeps the legacy `{collection}|{cid}` shape; the
            // cid is recomputed from the body
            let cid = match jacquard_repo::mst::util::compute_cid(value.as_ref()) {
                Ok(c) => c,
                Err(e) => return Some(Err(miette::miette!("{e}"))),
            };
            let key = crate::db::keys::indexer::block_key(&want, &cid.to_bytes());
            Some(Ok(ScannedBlock {
                key: fjall::Slice::from(key.as_slice()),
                value,
            }))
        })
    }
}

#[cfg(feature = "indexer")]
/// a raw block entry yielded by [`Hydrant::scan_collection_blocks`].
///
/// key and value are kept as raw slices; use the accessor methods for lazy parsing.
#[derive(Clone, Debug)]
pub struct ScannedBlock {
    pub(crate) key: fjall::Slice,
    pub(crate) value: fjall::Slice,
}

#[cfg(feature = "indexer")]
impl ScannedBlock {
    /// the collection name part of the block's key.
    pub fn collection(&self) -> &str {
        let sep_idx = self
            .key
            .iter()
            .position(|&b| b == b'|')
            .expect("malformed block key: missing separator");
        std::str::from_utf8(&self.key[..sep_idx]).expect("collection name must be valid UTF-8")
    }

    /// the raw CID bytes after the separator.
    pub fn cid(&self) -> &[u8] {
        let sep_idx = self
            .key
            .iter()
            .position(|&b| b == b'|')
            .expect("malformed block key: missing separator");
        &self.key[sep_idx + 1..]
    }

    /// the key's CID, parsed.
    pub fn parsed_cid(&self) -> Result<cid::Cid, miette::Report> {
        cid::Cid::read_bytes(self.cid()).map_err(|e| miette::miette!("failed to parse CID: {e}"))
    }

    /// the key's CID as a jacquard CID.
    pub fn jacquard_cid(&self) -> Result<jacquard_common::types::cid::Cid, miette::Report> {
        let parsed = cid::Cid::read_bytes(self.cid())
            .map_err(|e| miette::miette!("failed to parse CID: {e}"))?;
        Ok(jacquard_common::types::cid::Cid::ipld(parsed))
    }

    /// the raw DAG-CBOR block bytes.
    pub fn value(&self) -> &[u8] {
        &self.value
    }
}

impl axum::extract::FromRef<Hydrant> for Arc<AppState> {
    fn from_ref(h: &Hydrant) -> Self {
        h.state.clone()
    }
}

#[cfg(all(test, feature = "indexer"))]
mod tests {
    use super::*;
    use crate::config::Config;
    use crate::db::keys;
    use jacquard_common::IntoStatic;
    use tempfile::tempdir;

    fn test_config(path: &std::path::Path) -> Config {
        Config {
            database_path: path.to_path_buf(),
            ..Default::default()
        }
    }

    #[tokio::test]
    async fn test_scan_collection_blocks() -> Result<()> {
        use crate::db::types::DbRkey;
        use jacquard_common::types::string::Did;

        let tmp = tempdir().into_diagnostic()?;
        let hydrant = Hydrant::new(test_config(tmp.path())).await?;
        let state = hydrant.state.clone();

        let did: Did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap();
        let post_body1: &[u8] = b"post_cbor_value";
        let like_body: &[u8] = b"like_cbor_value";
        let post_body2: &[u8] = b"post_cbor_value_2";
        // the scanned cid is recomputed from the body
        let cid1 = jacquard_repo::mst::util::compute_cid(post_body1).unwrap();
        let cid2 = jacquard_repo::mst::util::compute_cid(like_body).unwrap();
        let cid3 = jacquard_repo::mst::util::compute_cid(post_body2).unwrap();

        state
            .db
            .run(move |db| -> Result<()> {
                let mut batch = db.inner.batch();
                db.indexer.stage_record(
                    &mut batch,
                    keys::record_key(&did, "app.bsky.feed.post", &DbRkey::new("a")),
                    post_body1,
                );
                db.indexer.stage_record(
                    &mut batch,
                    keys::record_key(&did, "app.bsky.feed.like", &DbRkey::new("b")),
                    like_body,
                );
                db.indexer.stage_record(
                    &mut batch,
                    keys::record_key(&did, "app.bsky.feed.post", &DbRkey::new("c")),
                    post_body2,
                );
                batch.commit().into_diagnostic()?;
                Ok(())
            })
            .await?;

        // scan posts: ordered by (repo, rkey), so rkey "a" then "c"
        let post_blocks: Vec<ScannedBlock> = hydrant
            .scan_collection_blocks("app.bsky.feed.post")
            .collect::<Result<Vec<_>>>()?;

        assert_eq!(post_blocks.len(), 2);
        assert_eq!(post_blocks[0].collection(), "app.bsky.feed.post");
        assert_eq!(post_blocks[0].cid(), cid1.to_bytes());
        assert_eq!(post_blocks[0].parsed_cid().unwrap(), cid1);
        assert_eq!(
            post_blocks[0].jacquard_cid().unwrap(),
            jacquard_common::types::cid::Cid::ipld(cid1)
        );
        assert_eq!(post_blocks[0].value(), post_body1);

        assert_eq!(post_blocks[1].collection(), "app.bsky.feed.post");
        assert_eq!(post_blocks[1].cid(), cid3.to_bytes());
        assert_eq!(post_blocks[1].parsed_cid().unwrap(), cid3);
        assert_eq!(post_blocks[1].value(), post_body2);

        // scan likes
        let like_blocks: Vec<ScannedBlock> = hydrant
            .scan_collection_blocks("app.bsky.feed.like")
            .collect::<Result<Vec<_>>>()?;

        assert_eq!(like_blocks.len(), 1);
        assert_eq!(like_blocks[0].collection(), "app.bsky.feed.like");
        assert_eq!(like_blocks[0].cid(), cid2.to_bytes());
        assert_eq!(like_blocks[0].parsed_cid().unwrap(), cid2);
        assert_eq!(like_blocks[0].value(), like_body);

        // scan non-existent
        let empty_blocks: Vec<ScannedBlock> = hydrant
            .scan_collection_blocks("app.bsky.feed.repost")
            .collect::<Result<Vec<_>>>()?;
        assert!(empty_blocks.is_empty());

        Ok(())
    }
}