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

use serde_json::Value;
use uuid::Uuid;

use crate::redis::AppRedis;

/// The chat WS per-user channel; every crate that writes a notification
/// publishes to it so connected clients receive the row in real time.
pub fn user_channel(uid: i64) -> String {
    format!("chat:user:{uid}")
}

pub fn notification_event_json(notification: Value) -> Value {
    serde_json::json!({
        "kind": "notification",
        "notification": notification,
    })
}

pub async fn publish_user_event(redis: &AppRedis, uid: i64, event: &Value) {
    let payload = match serde_json::to_string(event) {
        Ok(payload) => payload,
        Err(error) => {
            tracing::warn!(%error, %uid, "user event serialize failed");
            return;
        }
    };
    let channel = user_channel(uid);
    if let Err(error) = redis.publish(&channel, &payload).await {
        tracing::warn!(%error, %channel, "user event publish failed");
    }
    // Every `notification` user event is also a candidate for an OS push.
    // Enqueueing here — rather than at each of the nine writers — is what makes
    // "websocket fan-out and push fan-out never drift" true by construction.
    if event.get("kind").and_then(Value::as_str) == Some("notification")
        && let Some(notification) = event.get("notification")
        && let Some(job) = crate::push::PushJob::from_notification(uid, notification)
    {
        crate::push::enqueue(redis, &job).await;
    }
}

pub async fn publish_notification(redis: &AppRedis, uid: i64, notification: Value) {
    let event = notification_event_json(notification);
    publish_user_event(redis, uid, &event).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_event_json(notification))),
            Err(_) => {
                tracing::warn!("pending notifications lock poisoned");
            }
        }
    }

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