//! Push worker：从 Redis 队列取任务，过闸门，逐设备投递。
//!
//! 闸门顺序是刻意的——先看偏好位（纯内存），再看在线状态（一次 Redis
//! EXISTS），最后才回查数据库组装文案。用户关掉某类推送或已经在线时，
//! worker 不做任何数据库往返。

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use export::AppState;
use export::push::{PushJob, online_key};
use models::users::user_preferences;
use models::users::user_push_subscriptions::{self, PushSubscriptionPlatform};
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
use tracing::Instrument as _;

use crate::client::{DeliveryOutcome, NotificationPayload, SubscriptionKeys, VapidKeys, deliver};

use crate::error::PushError;

/// 队列空转的阻塞时长；同时决定进程响应 shutdown 的粒度。
const CONSUME_TIMEOUT_SECS: u64 = 5;

pub struct PushWorker {
    state: AppState,
    keys: Arc<VapidKeys>,
    client: reqwest::Client,
    cancel: tokio_util::sync::CancellationToken,
    /// 重试计数按任务指纹累计。
    ///
    /// 计数留在进程内是刻意的：持久化它需要再开一张表或改载荷字段，而代价
    /// 只是进程重启后多投一次——对"尽力而为"的通知完全可接受。
    attempts: Arc<Mutex<HashMap<String, u32>>>,
}

impl PushWorker {
    pub fn new(
        state: AppState,
        cancel: tokio_util::sync::CancellationToken,
    ) -> Result<Self, PushError> {
        let keys = VapidKeys::from_config(&state.config.push)?;
        let timeout = Duration::from_secs(state.config.push.request_timeout_secs.max(1));
        let client = reqwest::Client::builder()
            .timeout(timeout)
            .build()
            .map_err(|error| PushError::Internal(format!("building http client: {error}")))?;
        Ok(Self {
            state,
            keys: Arc::new(keys),
            client,
            cancel,
            attempts: Arc::new(Mutex::new(HashMap::new())),
        })
    }

    pub async fn run(self) {
        let topic = export::push::QUEUE_TOPIC.to_string();
        loop {
            let payload = tokio::select! {
                _ = self.cancel.cancelled() => break,
                result = self.state.redis.brpop_bytes(&topic, CONSUME_TIMEOUT_SECS) => {
                    match result {
                        Ok(Some(payload)) => payload,
                        Ok(None) => continue,
                        Err(error) => {
                            tracing::warn!(%error, "push queue consume failed");
                            tokio::time::sleep(Duration::from_secs(2)).await;
                            continue;
                        }
                    }
                }
            };

            let job = match PushJob::decode(&payload) {
                Ok(job) => job,
                Err(error) => {
                    tracing::warn!(%error, "dropping malformed push job");
                    export::push::enqueue_dlq(&self.state.redis, &payload).await;
                    continue;
                }
            };

            let span = tracing::info_span!(
                "push.delivery",
                user = %job.user_id,
                category = job.category.as_str(),
                request_id = tracing::field::Empty
            );
            if let Some(request_id) = job.request_id.as_deref() {
                span.record("request_id", request_id);
            }
            oltp::propagation::set_parent_from_traceparent(&span, job.trace.as_deref());
            if let Err(error) = export::middleware::with_request_id(
                job.request_id.as_deref(),
                self.handle(&job).instrument(span),
            )
            .await
            {
                tracing::warn!(
                    %error,
                    user = %job.user_id,
                    category = job.category.as_str(),
                    "push job failed"
                );
            }
        }
        tracing::debug!("push worker stopped");
    }

    /// 一条任务的完整流程：闸门 → 组装 → 逐订阅投递。
    async fn handle(&self, job: &PushJob) -> Result<(), PushError> {
        if self.skip_by_preferences(job).await? {
            self.forget(job);
            return Ok(());
        }
        if self.skip_online(job.user_id).await {
            self.forget(job);
            return Ok(());
        }

        let subscriptions = user_push_subscriptions::Entity::find()
            .filter(user_push_subscriptions::Column::UserId.eq(job.user_id))
            .filter(user_push_subscriptions::Column::Platform.eq(PushSubscriptionPlatform::WebPush))
            .all(self.state.database.read())
            .await
            .map_err(|error| PushError::Internal(format!("subscription lookup failed: {error}")))?;
        if subscriptions.is_empty() {
            self.forget(job);
            return Ok(());
        }

        let payload = self.build_payload(job).await?;
        let ttl_secs = self.state.config.push.ttl_secs;
        let mut retry = false;
        for subscription in subscriptions {
            let keys = SubscriptionKeys {
                endpoint: subscription.endpoint.clone(),
                p256dh_key: subscription.p256dh_key.clone(),
                auth_key: subscription.auth_key.clone(),
            };
            match deliver(&self.client, &self.keys, &keys, &payload, ttl_secs).await {
                Ok(DeliveryOutcome::Delivered) => {
                    oltp::metrics::push_delivery("delivered");
                    tracing::debug!(user = %job.user_id, "push delivered");
                }
                Ok(DeliveryOutcome::Gone) => {
                    oltp::metrics::push_delivery("gone");
                    // 用户清了浏览器数据或卸载了站点：一次性清理，不是错误。
                    tracing::debug!(user = %job.user_id, "push subscription expired; removing");
                    if let Err(error) =
                        user_push_subscriptions::Entity::delete_by_id(subscription.id)
                            .exec(self.state.database.write())
                            .await
                    {
                        tracing::warn!(%error, "failed to remove expired push subscription");
                    }
                }
                Ok(DeliveryOutcome::Rejected) => {
                    oltp::metrics::push_delivery("rejected");
                    tracing::warn!(
                        user = %job.user_id,
                        endpoint = %subscription.endpoint,
                        "push rejected by service; sending to dlq"
                    );
                    self.to_dlq(job).await;
                    self.forget(job);
                    return Ok(());
                }
                Ok(DeliveryOutcome::Retry) => {
                    oltp::metrics::push_delivery("retry");
                    retry = true;
                }
                Err(PushError::Subscription(message)) => {
                    oltp::metrics::push_delivery("subscription_error");
                    // 订阅本身不可修（endpoint 或密钥解析失败），重投没有意义。
                    tracing::warn!(user = %job.user_id, %message, "dropping broken subscription");
                    if let Err(error) =
                        user_push_subscriptions::Entity::delete_by_id(subscription.id)
                            .exec(self.state.database.write())
                            .await
                    {
                        tracing::warn!(%error, "failed to remove broken push subscription");
                    }
                }
                Err(error) => {
                    oltp::metrics::push_delivery("error");
                    tracing::warn!(%error, user = %job.user_id, "push delivery error");
                    retry = true;
                }
            }
        }

        if retry {
            self.retry_or_drop(job).await;
        } else {
            self.forget(job);
        }
        Ok(())
    }

    /// 用户偏好里的禁用位；`Other` 类不受位控制，只受"是否有订阅"约束。
    async fn skip_by_preferences(&self, job: &PushJob) -> Result<bool, PushError> {
        let Some(flags) = user_preferences::Entity::find_by_id(job.user_id)
            .one(self.state.database.read())
            .await
            .map_err(|error| PushError::Internal(format!("preferences lookup failed: {error}")))?
        else {
            // 没有偏好行 = 默认全开。
            return Ok(false);
        };
        Ok(job.category.is_disabled_by(flags.push_flags))
    }

    /// "用户在线"= 任意标签页持有聊天 WS 连接（集合里还有未过期的连接成员）；
    /// 此时站内 toast 已经送达，再弹系统通知就是重复提醒。
    async fn skip_online(&self, uid: i64) -> bool {
        export::redis::zset_live_count(&self.state.redis, &online_key(uid))
            .await
            .unwrap_or(0)
            > 0
    }

    /// 文案与深链的最终组装：作者名、头像、点击 URL 都需要数据库或存储。
    async fn build_payload(&self, job: &PushJob) -> Result<NotificationPayload, PushError> {
        let (title, icon) = match job.actor_id {
            Some(actor_id) => {
                let actor = auth::users::get_public_user(&auth::Accounts::new(self.state.database.connection().clone(), self.state.cache.clone()), actor_id)
                    .await
                    .map_err(|error| {
                        PushError::Internal(format!("actor lookup failed: {error}"))
                    })?;
                match actor {
                    Some(actor) => {
                        let name = if actor.displayname.trim().is_empty() {
                            actor.username.clone()
                        } else {
                            actor.displayname.clone()
                        };
                        (
                            format!("{name} · {}", job.title),
                            self.avatar_url(actor_id, &actor.avatar_url).await,
                        )
                    }
                    None => (job.title.clone(), None),
                }
            }
            None => (job.title.clone(), None),
        };

        Ok(NotificationPayload {
            title,
            body: job.body.clone(),
            icon,
            tag: job
                .data
                .as_ref()
                .and_then(|data| data.get("room_id"))
                .and_then(|value| value.as_str())
                .map(|room_id| format!("chat:{room_id}")),
            url: self.target_url(job),
        })
    }

    /// 头像在库里存的是 `storage://` locator，预签名后才是浏览器可取到的
    /// 地址；外链头像（导入数据）原样透传，空值表示前端渲染首字母。
    async fn avatar_url(&self, uid: i64, stored: &str) -> Option<String> {
        if stored.is_empty() {
            return None;
        }
        if !stored.starts_with("storage://") {
            return Some(stored.to_string());
        }
        let key = stored.strip_prefix("storage://")?;
        if key.is_empty() {
            return None;
        }
        match self.state.storage.presign_get(key, 7 * 24 * 60 * 60) {
            Ok(url) => Some(url),
            Err(error) => {
                tracing::warn!(%error, %uid, "avatar presign failed");
                None
            }
        }
    }

    /// 点击行为：与前端 `notificationTarget` 的规则保持一致，避免通知落到
    /// 一个前端不认识的地址上；`data` 缺字段时退到站内通知中心。
    fn target_url(&self, job: &PushJob) -> String {
        let base = self.state.config.web.web_base_url.trim_end_matches('/');
        let fallback = format!("{base}/app/@me/notify");
        let Some(data) = job.data.as_ref() else {
            return fallback;
        };
        let field = |key: &str| data.get(key).and_then(|value| value.as_str());
        let scope = field("space_name").unwrap_or("@me");
        if let Some(room_id) = field("room_id") {
            let jump = field("message_id")
                .map(|message_id| format!("?jump={message_id}"))
                .unwrap_or_default();
            return format!("{base}/app/{scope}/chat/{room_id}{jump}");
        }
        if let (Some(board_id), Some(issue_number)) = (field("board_id"), field("autoincrement_id"))
        {
            return format!("{base}/app/{scope}/board/{board_id}?issue={issue_number}");
        }
        if scope != "@me" {
            return format!("{base}/app/{scope}");
        }
        fallback
    }

    /// 暂时性失败：退避后重投，超过上限进 DLQ。
    ///
    /// 退避在独立任务里等待而不是就地 `sleep`：队列是单个 FIFO，就地等待会让
    /// 一条坏任务把整个消费者按分钟级卡住，后面所有用户的推送跟着排长队。
    async fn retry_or_drop(&self, job: &PushJob) {
        let config = self.state.config.push.clone();
        let attempt = {
            let mut attempts = match self.attempts.lock() {
                Ok(guard) => guard,
                Err(_) => return,
            };
            let entry = attempts.entry(job_key(job)).or_insert(0);
            *entry += 1;
            *entry
        };
        if attempt >= config.max_attempts.max(1) {
            tracing::warn!(user = %job.user_id, attempt, "push permanently failed; sending to dlq");
            self.to_dlq(job).await;
            self.forget(job);
            return;
        }
        let backoff = config.retry_backoff_base_secs.max(1) * u64::from(attempt);
        let redis = self.state.redis.clone();
        let cancel = self.cancel.clone();
        let Ok(payload) = job.encode() else {
            return;
        };
        let user_id = job.user_id;
        tokio::spawn(async move {
            tokio::select! {
                _ = cancel.cancelled() => return,
                _ = tokio::time::sleep(Duration::from_secs(backoff)) => {}
            }
            if let Err(error) = redis.lpush_bytes(export::push::QUEUE_TOPIC, &payload).await {
                tracing::warn!(%error, %user_id, "push requeue failed");
            }
        });
    }

    async fn to_dlq(&self, job: &PushJob) {
        if let Ok(payload) = job.encode() {
            export::push::enqueue_dlq(&self.state.redis, &payload).await;
        }
    }

    fn forget(&self, job: &PushJob) {
        if let Ok(mut attempts) = self.attempts.lock() {
            attempts.remove(&job_key(job));
        }
    }
}

/// 同一任务的稳定指纹：一条站内通知的重投必须落回同一个计数桶。私信推送
/// 没有通知行，退化成"收件人 + 分类 + 标题"，对同一房间的消息足够唯一。
fn job_key(job: &PushJob) -> String {
    match job.notification_id {
        Some(id) => id.to_string(),
        None => format!("{}:{}:{}", job.user_id, job.category.as_str(), job.title),
    }
}