use crate::types::{PlcEntry, PlcOperationType};
use deadpool_postgres::{Object, Pool};
use reqwest::Client;
use slog::{Logger, debug, error, info};
use std::env::var;
use std::time::{Duration, Instant};
use tokio::sync::OnceCell;
use tokio_postgres::Statement;
use tokio_postgres::types::Type;

// PLC rate limit is 100req/min - we aim a little lower
const SLEEP_TIME_F: f32 = (60. / 90.) * 1000.;
const SLEEP_TIME: Duration = Duration::from_millis(SLEEP_TIME_F as u64);

static INSERT_DID_STMT: OnceCell<Statement> = OnceCell::const_new();
static INSERT_OP_STMT: OnceCell<Statement> = OnceCell::const_new();

pub async fn importer(log: Logger, pool: Pool, start_after: Option<String>) {
    let plc_main = var("PLC_UPSTREAM").unwrap_or("https://plc.directory".to_string());
    let user_agent = var("PLC_USER_AGENT").unwrap_or("parakeet-plc".to_string());

    let mut after = start_after;

    if let Some(after) = &after {
        info!(log, "starting import at {after}");
    }

    let client = Client::builder().user_agent(user_agent).build().unwrap();
    let conn = pool.get().await.unwrap();

    loop {
        let start = Instant::now();

        match upstream_export(&client, &plc_main, after.clone()).await {
            Ok((entries, errors)) => {
                debug!(
                    log,
                    "exported {} entries, {} failed",
                    entries.len(),
                    errors.len()
                );

                if !errors.is_empty() {
                    error!(log, "{} entries failed to decode", errors.len());
                }

                for entry in &entries {
                    if let Err(e) = insert_did_and_op(&conn, entry).await {
                        error!(log, "failed to insert operation for did {}: {e}", entry.did);
                    }
                }

                if let Some(new_after) = entries.last() {
                    let new_after = new_after.created_at;
                    after = Some(new_after.to_rfc3339_opts(chrono::SecondsFormat::Millis, true));
                }
            }
            Err(err) => error!(log, "failed to export PLC directory: {err}"),
        }

        let end = start.elapsed();
        tokio::time::sleep(SLEEP_TIME.saturating_sub(end)).await;
    }
}

async fn upstream_export(
    client: &Client,
    plc: &str,
    after: Option<String>,
) -> eyre::Result<(Vec<PlcEntry>, Vec<serde_json::Error>)> {
    let mut uri = format!("{plc}/export?count=1000");
    if let Some(after) = after {
        uri += format!("&after={}", after).as_str();
    }

    let export_res = client.get(uri).send().await?;

    let export = export_res.text().await?;
    let (exports, errors) = export
        .lines()
        .map(serde_json::from_str)
        .fold((Vec::new(), Vec::new()), fold_exports);

    Ok((exports, errors))
}

fn fold_exports<A, B>(
    (mut oks, mut errors): (Vec<A>, Vec<B>),
    current: Result<A, B>,
) -> (Vec<A>, Vec<B>) {
    match current {
        Ok(data) => oks.push(data),
        Err(err) => errors.push(err),
    };

    (oks, errors)
}

async fn insert_did_and_op(conn: &Object, entry: &PlcEntry) -> eyre::Result<()> {
    let insert_did_stmt = INSERT_DID_STMT
        .get_or_init(|| init_insert_did_stmt(conn))
        .await;
    let insert_op_stmt = INSERT_OP_STMT
        .get_or_init(|| init_insert_op_stmt(conn))
        .await;

    let operation = serde_json::to_value(&entry.operation)?;

    let (prev, sig) = match &entry.operation {
        PlcOperationType::Create(op) => (op.prev.map(|v| v.to_string()), &op.sig),
        PlcOperationType::Tombstone(op) => (op.prev.map(|v| v.to_string()), &op.sig),
        PlcOperationType::Operation(op) => (op.prev.map(|v| v.to_string()), &op.sig),
    };

    conn.execute(insert_did_stmt, &[&entry.did]).await?;
    conn.execute(
        insert_op_stmt,
        &[
            &entry.did,
            &entry.cid.to_string(),
            &prev,
            &sig,
            &entry.nullified,
            &operation,
            &entry.created_at,
        ],
    )
    .await?;

    Ok(())
}

async fn init_insert_did_stmt(conn: &Object) -> Statement {
    conn.prepare_typed(
        "INSERT INTO dids (did) VALUES ($1) ON CONFLICT DO NOTHING",
        &[Type::TEXT],
    )
    .await
    .unwrap()
}

async fn init_insert_op_stmt(conn: &Object) -> Statement {
    conn.prepare_typed(
        include_str!("sql/insert_op.sql"),
        &[
            Type::TEXT,
            Type::TEXT,
            Type::TEXT,
            Type::TEXT,
            Type::BOOL,
            Type::JSONB,
            Type::TIMESTAMPTZ,
        ],
    )
    .await
    .unwrap()
}