mod filter;
pub(crate) mod parsers;
mod resource;

use self::filter::{FilePatternMatcher, Filter};
use self::parsers::{Parser, ParserEnum};
use self::resource::Resource;
use super::EventSourcePipe;
use crate::errors::{IntoDiagnostic, Result};
use futures::TryStreamExt;
use opendal::Operator;
use serde::Deserialize;
use serde::Serialize;
use std::collections::HashMap;
use std::time::Duration;
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
use tracing::instrument;

#[derive(Clone, Debug, Deserialize, Serialize)]
pub(crate) struct Config {
    #[serde(with = "humantime_serde")]
    pub(crate) polling_interval: Duration,
    pub(crate) kind: String,
    pub(crate) parameters: HashMap<String, String>,
    pub(crate) recursive: bool,
    pub(crate) path_patterns: Vec<String>,
    pub(crate) parser: parsers::Config,
    #[serde(default)]
    pub(crate) try_read_headers_json: bool,
    /// Base metadata to include in all `EventSource` instances created by this extractor.
    /// The `context.source` field will be automatically populated if not set.
    #[serde(default)]
    pub(crate) metadata: serde_json::Value,
}

pub(crate) struct OpendalExtractor {
    op: Operator,
    polling_interval: Duration,
    recursive: bool,
    filter: Filter,
    parser: ParserEnum,
    try_read_headers_json: bool,
    state_op: Option<Operator>,
    source_name: String,
}

impl OpendalExtractor {
    pub(crate) fn try_from(
        config: &Config,
        next: EventSourcePipe,
        state_op: Option<Operator>,
        source_name: String,
    ) -> Result<Self> {
        let op: Operator =
            Operator::via_iter(&config.kind, config.parameters.clone()).into_diagnostic()?;
        let filter = Filter::from_patterns(FilePatternMatcher::from(&config.path_patterns)?);
        let parser = config.parser.make_parser(config.metadata.clone(), next)?;
        let try_read_headers_json = config.try_read_headers_json;
        Ok(Self {
            op,
            polling_interval: config.polling_interval,
            recursive: config.recursive,
            filter,
            parser,
            try_read_headers_json,
            state_op,
            source_name,
        })
    }

    #[instrument(skip(self))]
    pub(crate) async fn run_once(&mut self) -> Result<usize> {
        let op = &self.op;
        let filter = &self.filter;
        let recursive = self.recursive;
        let try_read_headers_json = self.try_read_headers_json;
        let parser = &mut self.parser;
        tracing::debug!(filter=? filter, scheme =? op.info().scheme(), root =? op.info().root(), "scanning");
        let mut lister = op
        .lister_with("")
        .recursive(recursive)
        // Make sure content-length and last-modified been fetched.
        .await.into_diagnostic()?;
        let mut count = 0;
        while let Some(entry) = lister.try_next().await.into_diagnostic()? {
            let resource = Resource::from_entry(op, entry, try_read_headers_json).await;
            if filter.accept(&resource) {
                count += 1;
                if let Err(err) = parser.parse(op, &resource).await {
                    tracing::warn!(?err, path = resource.path(), "fail to process, skip");
                }
            }
        }
        Ok(count)
    }

    pub(crate) async fn run(&mut self, cancel_token: CancellationToken) -> Result<()> {
        if let Some(state_op) = &self.state_op
            && let Some(ts) = crate::state::load_ts_after(state_op, &self.source_name).await
        {
            self.filter.set_ts_after(ts);
        }
        while !cancel_token.is_cancelled() {
            if let Err(err) = self.run_once().await {
                tracing::warn!(?err, scheme =? self.op.info().scheme(), root =? self.op.info().root(), "fail during scanning");
            }
            tokio::select! {
                () = sleep(self.polling_interval) => {},
                () = cancel_token.cancelled() => {},
            }
            self.filter.jump_to_next_ts_window();
            if let Some(state_op) = &self.state_op
                && let Err(err) =
                    crate::state::save_ts_after(state_op, &self.source_name, self.filter.ts_after())
                        .await
            {
                tracing::warn!(?err, source = %self.source_name, "failed to save checkpoint");
            }
        }
        Ok(())
    }
}
