use std::sync::{Arc, Mutex};

use queue::Labels;
use serde_json::Value;

use cache::AppRedis;
use db::AppDatabase;

#[derive(Clone)]
pub struct Notifications {
    database: Arc<AppDatabase>,
    redis: AppRedis,
}

impl Notifications {
    pub fn new(database: Arc<AppDatabase>, redis: AppRedis) -> Self {
        Self { database, redis }
    }
}

impl Notifications {
    #[allow(clippy::too_many_arguments)]
    pub async fn create(
        &self,
        recipient: i64,
        kind: models::users::user_notifications::NotificationType,
        title: String,
        body: Option<String>,
        data: Option<Value>,
        actor: Option<i64>,
    ) -> Result<models::users::user_notifications::Model, sea_orm::DbErr> {
        use sea_orm::TransactionTrait;
        let row = self
            .database
            .write()
            .transaction::<_, models::users::user_notifications::Model, sea_orm::DbErr>(move |tx| {
                Box::pin(async move { insert(tx, recipient, kind, title, body, data, actor).await })
            })
            .await
            .map_err(|error| match error {
                sea_orm::TransactionError::Connection(error)
                | sea_orm::TransactionError::Transaction(error) => error,
            })?;
        self.deliver(recipient, row.to_event_value()).await;
        Ok(row)
    }

    pub async fn deliver(&self, uid: i64, notification: Value) {
        if let Some(job) = crate::protocol::PushJob::from_notification(uid, &notification) {
            crate::protocol::enqueue(&self.redis, &job).await;
        }
    }
}

/// Notifications created inside a DB transaction; flush after commit.
#[derive(Clone, Default)]
pub struct PendingNotifications {
    inner: Arc<Mutex<Vec<(i64, Value)>>>,
}

impl PendingNotifications {
    pub fn push(&self, uid: i64, notification: Value) {
        match self.inner.lock() {
            Ok(mut pending) => pending.push((uid, notification)),
            Err(_) => {
                tracing::warn!("pending notifications lock poisoned");
            }
        }
    }

    pub async fn flush(&self, notifications: &Notifications) {
        let batch: Vec<(i64, Value)> = match self.inner.lock() {
            Ok(mut pending) => std::mem::take(&mut *pending),
            Err(_) => return,
        };
        for (uid, event) in batch {
            notifications.deliver(uid, event).await;
        }
    }
}

#[allow(clippy::too_many_arguments)]
pub async fn insert(
    tx: &impl sea_orm::ConnectionTrait,
    recipient: i64,
    kind: models::users::user_notifications::NotificationType,
    title: String,
    body: Option<String>,
    data: Option<Value>,
    actor: Option<i64>,
) -> Result<models::users::user_notifications::Model, sea_orm::DbErr> {
    use sea_orm::{ActiveModelTrait, Set};
    let row = models::users::user_notifications::ActiveModel {
        id: Set(uuid::Uuid::now_v7()),
        user_id: Set(recipient),
        notification_type: Set(kind),
        title: Set(title),
        body: Set(body),
        data_json: Set(data),
        actor_uid: Set(actor),
        read_at: Set(None),
        created_at: Set(time::OffsetDateTime::now_utc()),
    }
    .insert(tx)
    .await?;
    let labels = Labels::new()
        .with(events::label::RECIPIENT_USER_ID, recipient)
        .with(events::label::USER_ID, recipient);
    events::emit(
        tx,
        events::EventKind::NotifyNotificationCreate,
        labels,
        serde_json::json!({ "kind": "notification", "notification": row.to_event_value() }),
    )
    .await?;
    Ok(row)
}