//! Interface to access the database.

use std::collections::HashMap;
use std::time::Duration;

use crate::activitypub::{self, Object, ObjectOrLink, ObjectsOrLinks};
use crate::http_api::collection_page_id;
use num_traits::ToPrimitive;
use sqlx::postgres::{PgListener, PgPoolOptions};
use sqlx::{Pool, Postgres};
use time::OffsetDateTime;
use tokio::time::sleep;
use tracing::Instrument;

/// The maximum amount of items in a CollectionPage.
const PAGE_SIZE: u32 = 10;

/// A database connection pool holder, with functions for inserting and getting
/// ActivityPub objects to and from.
pub struct Database {
    pool: Pool<Postgres>,
}

impl Database {
    /// Creates a new connection pool and run migrations to bring the database up
    /// to speed.
    pub async fn new(url: &str) -> sqlx::Result<Database> {
        let pool = PgPoolOptions::new()
            .max_connections(10)
            .acquire_timeout(Duration::from_secs(10))
            .connect(url)
            .await?;
        {
            let span = tracing::info_span!("db-migrations");
            sqlx::migrate!().run(&pool).instrument(span).await?;
        }
        Ok(Database { pool })
    }

    // TODO: Make inserts run in a transaction

    /// Inserts the object into the database, or do nothing if it has already
    /// been inserted.
    pub async fn insert(&self, obj: &Object<'_>) -> sqlx::Result<()> {
        let id = obj.id.as_ref().unwrap();
        let type_ = obj.type_.as_ref().unwrap();
        let json = serde_json::to_value(obj).unwrap();
        let query = sqlx::query!(
            "insert into as2_objects (id, obj) values ($1, $2) \
            on conflict do nothing",
            id,
            json,
        );
        let mut conn = self.pool.acquire().await?;
        tracing::debug!("Inserting {type_} into the db: {id}");
        tracing::trace!("{json}");
        query.execute(&mut conn).await?;
        Ok(())
    }

    /// Inserts the item into the collection. Both ids must refer to objects that
    /// have been previously inserted with [Database::insert].
    pub async fn insert_to_collection(&self, collection_id: &str, item_id: &str) -> sqlx::Result<()> {
        let query = sqlx::query!(
            "insert into as2_ordered_collection_items (collection_id, item_id) values ($1, $2)",
            collection_id,
            item_id,
        );
        let mut conn = self.pool.acquire().await?;
        query.execute(&mut conn).await?;
        Ok(())
    }

    /// Returns the object with the id. If the object is an "OrderedCollection",
    /// `page` will be used to provide the specific "CollectionPage", if
    /// provided.
    pub async fn get_object(&self, id: &str, page: Option<u32>) -> sqlx::Result<Option<Object>> {
        let mut conn = self.pool.acquire().await?;
        let query = sqlx::query_scalar!("select obj from as2_objects where id = $1", id);
        match query.fetch_optional(&mut conn).await {
            Ok(None) => Ok(None),
            Ok(Some(row)) => {
                let mut object: Object = serde_json::from_value(row).unwrap();
                if object.has_type("OrderedCollection") {
                    fill_collection_details(&mut conn, &mut object, page).await?;
                } else {
                    let obj_slice = std::slice::from_mut(&mut object);
                    dereference_main_properties(&mut conn, obj_slice).await?;
                }
                Ok(Some(object))
            }
            Err(err) => Err(err),
        }
    }

    /// Adds the activity with the given id to the delivery queue, to be
    /// delivered to the given inbox.
    pub async fn deliver(&self, inbox_url: &str, activity_id: &str) -> sqlx::Result<()> {
        let insert_query = sqlx::query!(
            "insert into inbox_queue
            (inbox_url, activity_id, next_delivery_time_utc, retry_interval) values \
            ($1,        $2,          CURRENT_TIMESTAMP,      interval '1 minute')",
            inbox_url,
            activity_id,
        );
        let notify_query = sqlx::query!("NOTIFY activitypub_inbox_delivery");
        let mut conn = self.pool.acquire().await?;
        insert_query.execute(&mut conn).await?;
        notify_query.execute(&mut conn).await?;
        Ok(())
    }

    /// Waits until the next time there's deliveries to be delivered.
    ///
    /// ### Implementation details
    ///
    /// Returns either when the "activitypub_inbox_delivery" channel is notified
    /// (i.e. when a new delivery is added to the db), or when the earliest
    /// "next_delivery_time_utc" is reached, whichever is first.
    pub async fn wait_for_inbox_delivery(&self) -> sqlx::Result<()> {
        let wait_duration = {
            let query = sqlx::query!(
                // "extract (epoch from <interval>)" returns the total amount of seconds in the interval
                "select greatest(0, extract(epoch from next_delivery_time_utc - CURRENT_TIMESTAMP)) \
                  as wait_time from inbox_queue order by next_delivery_time_utc asc limit 1",
            );
            let mut conn = self.pool.acquire().await?;
            let result = query.fetch_optional(&mut conn).await?;
            result
                .and_then(|row| row.wait_time)
                .map(|wait_time| Duration::from_secs(wait_time.to_u64().unwrap() + 1))
        };
        let mut listener = PgListener::connect_with(&self.pool).await?;
        listener.listen("activitypub_inbox_delivery").await?;
        let notify_fut = listener.recv();
        if let Some(wait_duration) = wait_duration {
            tokio::select! {
                _ = notify_fut => Ok(()),
                _ = sleep(wait_duration) => Ok(())
            }
        } else {
            notify_fut.await?;
            Ok(())
        }
    }

    /// Returns a list of `(inbox, activity to send to inbox)` tuples that are
    /// due to be sent now, and marks the rows as having attempted a delivery at
    /// the time of calling this function.
    pub async fn get_deliveries(&self) -> sqlx::Result<Vec<(String, Object)>> {
        let mut conn = self.pool.acquire().await?;
        let now = OffsetDateTime::now_utc();
        let query = sqlx::query!(
            "update inbox_queue \
            set next_delivery_time_utc = next_delivery_time_utc + retry_interval, \
              retry_interval = retry_interval * 2 \
            from as2_objects \
            where activity_id = id and \
              ((next_delivery_time_utc <= $1 and latest_delivery_attempt_time_utc is null) or \
               COALESCE(latest_delivery_attempt_time_utc + interval '1 minute' <= $1, false)) \
            returning inbox_url, obj",
            now
        );
        let rows = query.fetch_all(&mut conn).await?;
        let (inbox_urls, mut objects): (Vec<String>, Vec<Object>) = rows
            .into_iter()
            .map(|row| (row.inbox_url, serde_json::from_value(row.obj).unwrap()))
            .unzip();
        dereference_main_properties(&mut conn, &mut objects).await?;
        Ok(inbox_urls.into_iter().zip(objects.into_iter()).collect())
    }

    /// Marks deliveries as failed, to be retried. The slice should contain
    /// tuples containing the (inbox_url, activity_id).
    pub async fn reschedule_failed_deliveries(&self, deliveries: &[(&str, &str)]) -> sqlx::Result<()> {
        let deliveries = deliveries
            .iter()
            .map(|(inbox_url, activity_id)| format!("{activity_id}{inbox_url}"))
            .collect::<Vec<String>>();
        let query = sqlx::query!(
            "update inbox_queue set latest_delivery_attempt_time_utc = null \
            where (activity_id || inbox_url) = ANY($1)",
            &deliveries,
        );
        let mut conn = self.pool.acquire().await?;
        query.execute(&mut conn).await?;
        Ok(())
    }

    /// Marks deliveries as successful, clearing them out of the delivery queue.
    /// The slice should contain tuples containing the (inbox_url, activity_id).
    pub async fn clear_successful_deliveries(&self, deliveries: &[(&str, &str)]) -> sqlx::Result<()> {
        let deliveries = deliveries
            .iter()
            .map(|(inbox_url, activity_id)| format!("{activity_id}{inbox_url}"))
            .collect::<Vec<String>>();
        let query = sqlx::query!(
            "delete from inbox_queue where (activity_id || inbox_url) = ANY($1)",
            &deliveries,
        );
        let mut conn = self.pool.acquire().await?;
        query.execute(&mut conn).await?;
        Ok(())
    }
}

/// Fill in the [Collection] properties, or [CollectionPage] properties of the
/// given `page`, if it's a [`Some`]. The `object` is expected to have the
/// "OrderedCollection" type.
///
/// [Collection]:
///     https://www.w3.org/TR/activitystreams-vocabulary/#dfn-collection
/// [CollectionPage]:
///     https://www.w3.org/TR/activitystreams-vocabulary/#dfn-collectionpage
async fn fill_collection_details(
    conn: &mut sqlx::postgres::PgConnection,
    object: &mut Object<'_>,
    page: Option<u32>,
) -> Result<(), sqlx::Error> {
    let id = object.id.as_ref().unwrap();
    let query = sqlx::query_scalar!(
        "select count(item_id) from as2_ordered_collection_items where collection_id = $1",
        id,
    );
    let count = query.fetch_one(&mut *conn).await?.unwrap() as u32;
    object.total_items = Some(count);
    let first_page_id = collection_page_id(id, 0);
    let last_page = count / PAGE_SIZE;
    let last_page_id = collection_page_id(id, last_page);

    if let Some(page) = page {
        let offset = (page * PAGE_SIZE) as i64;
        let limit = PAGE_SIZE as i64;
        let query = sqlx::query_scalar!(
            "select obj from as2_objects \
            join as2_ordered_collection_items on (item_id = id) \
            where collection_id = $1 \
            order by item_order_number desc \
            limit $2 offset $3",
            id,
            limit,
            offset,
        );
        let mut objs = query
            .fetch_all(&mut *conn)
            .await?
            .into_iter()
            .map(|v| serde_json::from_value(v).unwrap())
            .collect::<Vec<Object>>();
        dereference_main_properties(&mut *conn, &mut objs).await?;
        let objs = objs
            .into_iter()
            .map(|obj| ObjectOrLink::Object(Box::new(obj)))
            .collect::<Vec<ObjectOrLink>>();

        // Page was given, make necessary changes to become a CollectionPage
        object.type_ = activitypub::type_collection_page();
        object.ordered_items = Some(ObjectsOrLinks::Multiple(objs));
        object.part_of = Some(ObjectOrLink::Link(id.clone()));
        if page > 0 {
            object.prev = Some(ObjectOrLink::Link(collection_page_id(id, page - 1).into()))
        }
        if page < last_page {
            object.next = Some(ObjectOrLink::Link(collection_page_id(id, page + 1).into()))
        }
        let current_id = collection_page_id(id, page);
        object.id = Some(current_id.clone().into());
        object.current = Some(ObjectOrLink::Link(current_id.into()));
    } else {
        object.current = Some(ObjectOrLink::Link(first_page_id.clone().into()));
    }

    object.first = Some(ObjectOrLink::Link(first_page_id.into()));
    object.last = Some(ObjectOrLink::Link(last_page_id.into()));
    Ok(())
}

/// Dereference some important properties (actor, object, target) for each
/// object.
async fn dereference_main_properties<'a>(
    conn: &mut sqlx::postgres::PgConnection,
    objects: &mut [Object<'a>],
) -> Result<(), sqlx::Error> {
    let mut links = Vec::new();
    for obj in &mut *objects {
        obj.actor.iter().for_each(|o| o.for_links(|l| links.push(l.clone())));
        obj.object.iter().for_each(|o| o.for_link(|l| links.push(l.clone())));
        obj.target.iter().for_each(|o| o.for_link(|l| links.push(l.clone())));
    }
    links.sort();
    links.dedup();
    let links = links.into_iter().map(|s| s.to_string()).collect::<Vec<String>>();
    let query = sqlx::query!("select id, obj from as2_objects where id = ANY($1)", &links[..]);
    let objects_by_id = query
        .fetch_all(conn)
        .await?
        .into_iter()
        .map(|record| (record.id, serde_json::from_value(record.obj).unwrap()))
        .collect::<HashMap<String, Object>>();
    let deref = |obj_or_link: &mut ObjectOrLink| {
        if let ObjectOrLink::Link(link) = obj_or_link {
            if let Some(obj) = objects_by_id.get(link.as_ref()) {
                *obj_or_link = ObjectOrLink::Object(Box::new(obj.clone()));
            }
        }
    };
    for obj in objects {
        obj.actor.iter_mut().for_each(|o| o.for_objects_or_links_mut(deref));
        obj.object.iter_mut().for_each(|obj_or_link| deref(obj_or_link));
        obj.target.iter_mut().for_each(|obj_or_link| deref(obj_or_link));
    }
    Ok(())
}