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, load_persisted_firehose_sources};
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> {
        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 state = AppState::new(&config)?;

        // 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 every block stored for `collection`, ordered by CID.
    ///
    /// values are raw DAG-CBOR. deleted or updated records may leave orphan blocks
    /// behind since the CAS is content-addressed, not record-referenced.
    #[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 collection = collection.to_string();

        let mut start = Vec::with_capacity(collection.len() + 1);
        start.extend_from_slice(collection.as_bytes());
        start.push(b'|');

        let mut end = Vec::with_capacity(collection.len() + 1);
        end.extend_from_slice(collection.as_bytes());
        end.push(b'|' + 1);

        let iter = db.block_range((
            std::ops::Bound::Included(start),
            std::ops::Bound::Excluded(end),
        ));

        iter.map(|item| {
            let (key, value) = item.into_inner().into_diagnostic()?;
            Ok(ScannedBlock { key, 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<'static>, 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 cid::Cid;
    use cid::multihash::Multihash;
    use tempfile::tempdir;

    const SHA2_256: u64 = 0x12;
    const DAG_CBOR: u64 = 0x71;

    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<()> {
        let tmp = tempdir().into_diagnostic()?;
        let hydrant = Hydrant::new(test_config(tmp.path())).await?;
        let state = hydrant.state.clone();

        let hash1 = [1u8; 32];
        let mh1 = Multihash::<64>::wrap(SHA2_256, &hash1).unwrap();
        let valid_cid1 = Cid::new_v1(DAG_CBOR, mh1);
        let cid_bytes1 = valid_cid1.to_bytes();

        let hash2 = [2u8; 32];
        let mh2 = Multihash::<64>::wrap(SHA2_256, &hash2).unwrap();
        let valid_cid2 = Cid::new_v1(DAG_CBOR, mh2);
        let cid_bytes2 = valid_cid2.to_bytes();

        let hash3 = [3u8; 32];
        let mh3 = Multihash::<64>::wrap(SHA2_256, &hash3).unwrap();
        let valid_cid3 = Cid::new_v1(DAG_CBOR, mh3);
        let cid_bytes3 = valid_cid3.to_bytes();

        let cid_bytes1_val = cid_bytes1.clone();
        let cid_bytes2_val = cid_bytes2.clone();
        let cid_bytes3_val = cid_bytes3.clone();

        // Insert a couple of blocks for two collections: "app.bsky.feed.post" and "app.bsky.feed.like"
        state
            .db
            .run(move |db| -> Result<()> {
                let mut batch = db.inner.batch();

                let post_key = keys::indexer::block_key("app.bsky.feed.post", &cid_bytes1_val);
                let post_val = b"post_cbor_value";
                db.indexer
                    .stage_block(&mut batch, post_key, post_val.to_vec());

                let like_key = keys::indexer::block_key("app.bsky.feed.like", &cid_bytes2_val);
                let like_val = b"like_cbor_value";
                db.indexer
                    .stage_block(&mut batch, like_key, like_val.to_vec());

                // Let's add one more post to test multiple blocks in same collection
                let post_key2 = keys::indexer::block_key("app.bsky.feed.post", &cid_bytes3_val);
                let post_val2 = b"post_cbor_value_2";
                db.indexer
                    .stage_block(&mut batch, post_key2, post_val2.to_vec());

                batch.commit().into_diagnostic()?;
                Ok(())
            })
            .await?;

        // Scan posts
        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(), cid_bytes1);
        assert_eq!(post_blocks[0].parsed_cid().unwrap(), valid_cid1);
        assert_eq!(
            post_blocks[0].jacquard_cid().unwrap(),
            jacquard_common::types::cid::Cid::ipld(valid_cid1)
        );
        assert_eq!(post_blocks[0].value(), b"post_cbor_value");

        assert_eq!(post_blocks[1].collection(), "app.bsky.feed.post");
        assert_eq!(post_blocks[1].cid(), cid_bytes3);
        assert_eq!(post_blocks[1].parsed_cid().unwrap(), valid_cid3);
        assert_eq!(post_blocks[1].value(), b"post_cbor_value_2");

        // 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(), cid_bytes2);
        assert_eq!(like_blocks[0].parsed_cid().unwrap(), valid_cid2);
        assert_eq!(like_blocks[0].value(), b"like_cbor_value");

        // 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(())
    }
}