use crate::backfill::admission::BackfillAdmission;
use crate::backfill::client::ThrottledHttpClient;
use crate::backfill::error::BackfillError;
use crate::config::BackfillStrategy;
use crate::db::{self, types::TrimmedDid};
use crate::ingest::indexer::IndexerTx;
use crate::net::public_http_with_timeout;
use crate::state::AppState;
use crate::util::WatchEnabledExt;
use jacquard_common::IntoStatic;
use jacquard_common::types::did::Did;
use miette::Result;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tracing::{Instrument, debug, error, info};

pub mod process;
pub mod task;

pub struct BackfillWorker {
    state: Arc<AppState>,
    buffer_tx: IndexerTx,
    http: ThrottledHttpClient,
    semaphore: Arc<Semaphore>,
    admission: BackfillAdmission,
    verify_signatures: bool,
    strategy: BackfillStrategy,
    in_flight: Arc<scc::HashSet<Did>>,
    deferred: Arc<scc::HashSet<Did>>,
    enabled: tokio::sync::watch::Receiver<bool>,
}

impl BackfillWorker {
    pub fn new(
        state: Arc<AppState>,
        buffer_tx: IndexerTx,
        timeout: Duration,
        concurrency_limit: usize,
        verify_signatures: bool,
        strategy: BackfillStrategy,
        proxies: Vec<url::Url>,
        enabled: tokio::sync::watch::Receiver<bool>,
    ) -> Result<Self> {
        // A CONNECT proxy resolves the origin itself, so it cannot carry the
        // socket-coupled public-PDS guarantee. Reject it at startup instead of
        // silently changing the address policy or pacing budget.
        if !proxies.is_empty() {
            return Err(miette::miette!(
                "backfill proxies are unsupported for public-PDS egress; remove HYDRANT_BACKFILL_PROXIES"
            ));
        }
        let client = public_http_with_timeout(timeout);
        info!("direct-only backfill client ready");

        let admission =
            BackfillAdmission::new(concurrency_limit, state.throttler.per_pds_concurrency());

        Ok(Self {
            state: state.clone(),
            buffer_tx,
            http: ThrottledHttpClient::new_with_body_limit(
                client,
                state.throttler.clone(),
                state.max_car_body_bytes,
            ),
            semaphore: Arc::new(Semaphore::new(concurrency_limit)),
            admission,
            verify_signatures,
            strategy,
            in_flight: Arc::new(scc::HashSet::new()),
            deferred: Arc::new(scc::HashSet::new()),
            enabled,
        })
    }
}

struct InFlightGuard {
    did: Did,
    set: Arc<scc::HashSet<Did>>,
}

impl Drop for InFlightGuard {
    fn drop(&mut self) {
        let _ = self.set.remove_sync(&self.did);
    }
}

impl BackfillWorker {
    pub async fn run(mut self) {
        info!("backfill worker started");

        loop {
            self.enabled.wait_enabled("backfill").await;
            let mut spawned = 0;

            for guard in self.state.db.indexer.pending.iter() {
                let (key, value) = match guard.into_inner() {
                    Ok(kv) => kv,
                    Err(e) => {
                        error!(err = %e, "failed to read pending entry");
                        db::check_poisoned(&e);
                        continue;
                    }
                };

                let did = match TrimmedDid::try_from(value.as_ref()) {
                    Ok(d) => d.to_did(),
                    Err(e) => {
                        error!(err = %e, "invalid did in pending value");
                        continue;
                    }
                };

                // check before trying to acquire a permit so we dont acquire a permit
                // for no reason, the read will be cheap anyhow
                if self.in_flight.contains_sync(&did) || self.deferred.contains_sync(&did) {
                    continue;
                }

                let permit = match self.semaphore.clone().try_acquire_owned() {
                    Ok(p) => p,
                    Err(_) => break,
                };

                // only mark as in flight if we can acquire a permit
                if self
                    .in_flight
                    .insert_sync(did.clone().into_static())
                    .is_err()
                {
                    // a task is already running, weh
                    // so we don't need this one anymore...
                    break;
                }

                let guard = InFlightGuard {
                    did: did.clone().into_static(),
                    set: self.in_flight.clone(),
                };

                let state = self.state.clone();
                let http = self.http.clone();
                let did = did.clone();
                let buffer_tx = self.buffer_tx.clone();
                let verify = self.verify_signatures;
                let strategy = self.strategy;
                let admission = self.admission.clone();
                let deferred = self.deferred.clone();

                let span = tracing::info_span!("backfill", did = %did);
                tokio::spawn(
                    async move {
                        let _guard = guard;
                        let res = task::did_task(
                            &state, http, buffer_tx, &did, key, admission, permit, verify, strategy,
                        )
                        .await;

                        match &res {
                            Ok(task::TaskDisposition::Deferred) => {
                                let _ = deferred.insert_sync(did.clone().into_static());
                            }
                            Ok(task::TaskDisposition::Finished) | Err(_) => {
                                deferred.clear_sync();
                            }
                        }

                        if let Err(e) = res {
                            match &e {
                                BackfillError::Ratelimited
                                | BackfillError::PreemptivelyThrottled => {
                                    debug!(err = %e, "process failed");
                                }
                                _ => {
                                    error!(err = %e, "process failed");
                                }
                            }
                            if let BackfillError::Generic(report) = &e {
                                db::check_poisoned_report(report);
                            }
                        }

                        // wake worker to pick up more (in case we were sleeping at limit)
                        state.backfill_notify.notify_one();
                    }
                    .instrument(span),
                );

                spawned += 1;
            }

            if spawned == 0 {
                // wait for new tasks
                self.state.backfill_notify.notified().await;
            }
            // loop immediately since we might have more tasks
        }
    }
}