pub mod links;
pub mod store;

pub mod api;

use std::ops::Bound;
use std::sync::Arc;

use miette::{IntoDiagnostic, Result};
use smol_str::SmolStr;
use tracing::warn;

use crate::state::AppState;

/// handle for querying the backlinks index.
#[derive(Clone)]
pub struct BacklinksControl(pub(crate) Arc<AppState>);

pub struct BacklinkEntry {
    pub uri: SmolStr,
    pub cid: SmolStr,
}

pub struct BacklinksPage {
    pub backlinks: Vec<BacklinkEntry>,
    /// raw key bytes of the last returned entry, used as cursor for next page
    pub next_cursor: Option<Vec<u8>>,
}

impl BacklinksControl {
    /// begin building a paginated fetch of backlinks pointing to `subject`.
    pub fn fetch(&self, subject: impl Into<String>) -> BacklinksFetch {
        BacklinksFetch {
            state: self.0.clone(),
            subject: subject.into(),
            collection: None,
            path: None,
            dids: Vec::new(),
            limit: 50,
            reverse: false,
            cursor: None,
        }
    }

    /// begin building a backlinks count query for `subject`.
    pub fn count(&self, subject: impl Into<String>) -> BacklinksCount {
        BacklinksCount {
            state: self.0.clone(),
            subject: subject.into(),
            collection: None,
            path: None,
            dids: Vec::new(),
        }
    }
}

/// a paginated fetch against the backlinks reverse index.
///
/// obtain via [`BacklinksControl::fetch`] and execute with [`BacklinksFetch::run`].
pub struct BacklinksFetch {
    state: Arc<AppState>,
    subject: String,
    collection: Option<String>,
    path: Option<String>,
    dids: Vec<String>,
    limit: usize,
    reverse: bool,
    cursor: Option<Vec<u8>>,
}

impl BacklinksFetch {
    /// filter results to the given source collection NSID.
    pub fn collection(mut self, collection: impl Into<String>) -> Self {
        self.collection = Some(collection.into());
        self
    }

    /// filter results to specific source author DIDs.
    pub fn dids(mut self, dids: Vec<String>) -> Self {
        self.dids = dids;
        self
    }

    /// filter results to a specific dotted field path within the record (e.g. `.subject.uri`).
    /// has no effect unless a collection is also set.
    pub fn path(mut self, path: impl Into<String>) -> Self {
        self.path = Some(path.into());
        self
    }

    /// convenience: parse a `source` parameter of the form `collection` or `collection:path`
    /// and apply both filters. the path component (if present) has `.` prepended automatically.
    pub fn source(self, source: &str) -> Self {
        match source.split_once(':') {
            Some((col, p)) => self.collection(col).path(format!(".{p}")),
            None => self.collection(source),
        }
    }

    /// maximum number of results to return (default 50).
    pub fn limit(mut self, limit: usize) -> Self {
        self.limit = limit;
        self
    }

    /// if `true`, return results in reverse key order (default `false`).
    pub fn reverse(mut self, reverse: bool) -> Self {
        self.reverse = reverse;
        self
    }

    /// resume from the raw key bytes returned as `next_cursor` from a previous page.
    pub fn cursor(mut self, cursor: Vec<u8>) -> Self {
        self.cursor = Some(cursor);
        self
    }

    /// execute and return a page of backlink entries.
    pub async fn run(self) -> Result<BacklinksPage> {
        self.state
            .db
            .run(move |db| {
                let scan_prefix = store::reverse_scan_prefix(
                    &self.subject,
                    self.collection.as_deref(),
                    self.path.as_deref(),
                );

                let iter: Box<dyn Iterator<Item = _>> = if !self.reverse {
                    if let Some(cursor_bytes) = &self.cursor {
                        Box::new(db.backlinks.range::<&[u8], _>((
                            Bound::Excluded(cursor_bytes.as_slice()),
                            Bound::Unbounded,
                        )))
                    } else {
                        Box::new(db.backlinks.range(scan_prefix.as_slice()..))
                    }
                } else {
                    // for reverse scans, bound the end at the cursor (exclusive) or the
                    // prefix end (increment last byte to get an exclusive upper bound)
                    let end: Vec<u8> = if let Some(cursor_bytes) = &self.cursor {
                        cursor_bytes.clone()
                    } else {
                        let mut end = scan_prefix.clone();
                        if let Some(last) = end.last_mut() {
                            *last = last.saturating_add(1);
                        }
                        end
                    };
                    Box::new(
                        db.backlinks
                            .range(scan_prefix.as_slice()..end.as_slice())
                            .rev(),
                    )
                };

                let mut backlinks = Vec::with_capacity(self.limit);
                let mut next_cursor = None;
                let mut last_returned_key = None;

                for item in iter {
                    let key = item.key().into_diagnostic()?;
                    if !key.starts_with(scan_prefix.as_slice()) {
                        break;
                    }

                    let Some((col, _path, did, rkey)) = store::source_from_reverse_key(&key) else {
                        warn!("backlinks: could not extract source from reverse key");
                        continue;
                    };

                    if !self.dids.is_empty() && !self.dids.iter().any(|d| d == did) {
                        continue;
                    }

                    let Some(cid) = store::lookup_cid(&db.indexer, did, col, rkey) else {
                        // record deleted after backlink was written
                        continue;
                    };
                    if backlinks.len() >= self.limit {
                        next_cursor = last_returned_key;
                        break;
                    }
                    let at_uri = format!("at://{did}/{col}/{rkey}");
                    backlinks.push(BacklinkEntry {
                        uri: SmolStr::new(at_uri),
                        cid: SmolStr::new(cid),
                    });
                    last_returned_key = Some(key.to_vec());
                }

                Ok(BacklinksPage {
                    backlinks,
                    next_cursor,
                })
            })
            .await
    }
}

/// a count query against the backlinks reverse index.
///
/// obtain via [`BacklinksControl::count`] and execute with [`BacklinksCount::run`].
pub struct BacklinksCount {
    state: Arc<AppState>,
    subject: String,
    collection: Option<String>,
    path: Option<String>,
    dids: Vec<String>,
}

impl BacklinksCount {
    /// filter to the given source collection NSID.
    pub fn collection(mut self, collection: impl Into<String>) -> Self {
        self.collection = Some(collection.into());
        self
    }

    /// filter results to specific source author DIDs.
    pub fn dids(mut self, dids: Vec<String>) -> Self {
        self.dids = dids;
        self
    }

    /// filter to a specific dotted field path within the record (e.g. `.subject.uri`).
    /// has no effect unless a collection is also set.
    pub fn path(mut self, path: impl Into<String>) -> Self {
        self.path = Some(path.into());
        self
    }

    /// parse a `source` parameter of the form `collection` or `collection:path`
    /// and apply both filters. the path component (if present) has `.` prepended automatically.
    pub fn source(self, source: &str) -> Self {
        match source.split_once(':') {
            Some((col, p)) => self.collection(col).path(format!(".{p}")),
            None => self.collection(source),
        }
    }

    /// execute and return the total count of matching entries.
    pub async fn run(self) -> Result<u64> {
        self.state
            .db
            .run(move |db| {
                let scan_prefix = store::reverse_scan_prefix(
                    &self.subject,
                    self.collection.as_deref(),
                    self.path.as_deref(),
                );

                if !self.dids.is_empty() {
                    let mut count = 0;
                    for item in db.backlinks.prefix(&scan_prefix) {
                        let key = item.key().into_diagnostic()?;
                        let Some((_col, _path, did, _rkey)) = store::source_from_reverse_key(&key)
                        else {
                            continue;
                        };
                        if self.dids.iter().any(|d| d == did) {
                            count += 1;
                        }
                    }
                    Ok(count)
                } else {
                    Ok(db.backlinks.prefix(&scan_prefix).count() as u64)
                }
            })
            .await
    }
}