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

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

use crate::AppState;

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

pub async fn publish_user_event(state: &AppState, uid: i64, event: &Value) {
    let redis = &state.redis;
    // Real-time fan-out is RocketMQ only: the event is written to the outbox and
    // delivered to KenEvent. Redis is used here solely as the OS-push task queue.
    // 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 mut event = event.clone();
        if let Some(notification) = event.get_mut("notification") {
            if let Some(actor_id) = notification.get("actor_id").and_then(Value::as_i64)
                && let Ok(Some(actor)) = crate::users::get_public_user(state, actor_id).await
            {
                notification["actor"] = serde_json::json!({
                    "id": actor.user_id,
                    "username": actor.username,
                    "display_name": actor.displayname,
                    "avatar_url": crate::resolve_user_avatar_url(actor.user_id, &actor.avatar_url),
                });
            }
            if let Some(job) = crate::push::PushJob::from_notification(uid, notification) {
                crate::push::enqueue(redis, &job).await;
            }
        }
        let request_id = crate::middleware::request_id();
        let labels = Labels::new()
            .with("user_id", uid)
            .with_opt("request_id", request_id.clone())
            .with(events::label::ORDERING_KEY, format!("notify:user:{uid}"));
        if let Err(error) = events::outbox::enqueue_now(
            state.database.write(),
            "notify.notification.create",
            labels,
            &event,
            request_id.as_deref().unwrap_or_default(),
            request_id.as_deref(),
        )
        .await
        {
            tracing::warn!(%error, %uid, "notification event enqueue failed");
        }
    }
}

pub async fn publish_notification(state: &AppState, uid: i64, notification: Value) {
    let event = notification_event_json(notification);
    publish_user_event(state, 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, state: &AppState) {
        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(state, uid, &event).await;
        }
    }
}