use models::chat::room_seq;
use models::chat::room_user_settings::{self, RoomNotificationLevel};
use models::room::{room_members, rooms};
use models::users::user_notifications::NotificationType;
use models::users::user_recent_mentions;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, QueryFilter, Set,
    Statement,
};
use space::MessageNotifications;
use time::OffsetDateTime;
use uuid::Uuid;

use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::realtime::RoomEvent;

/// The recipient set of a DM / group DM message: current room members minus the
/// author. Used for OS pushes, which (unlike mentions) do not write inbox rows.
pub async fn dm_recipients(service: &ChatService, room_id: Uuid, author: i64) -> Vec<i64> {
    room_members::Entity::find()
        .filter(room_members::Column::RoomId.eq(room_id))
        .filter(room_members::Column::LeftAt.is_null())
        .all(service.database.read())
        .await
        .unwrap_or_default()
        .into_iter()
        .map(|member| member.user_id)
        .filter(|uid| *uid != author)
        .collect()
}

pub async fn next_room_seq(db: &impl ConnectionTrait, room_id: Uuid) -> Result<i64, ChatError> {
    let now = OffsetDateTime::now_utc();
    let row = db
        .query_one_raw(Statement::from_sql_and_values(
            DatabaseBackend::Postgres,
            r#"INSERT INTO chat_room_seq (room_id, seq, updated_at)
               VALUES ($1, 1, $2)
               ON CONFLICT (room_id)
               DO UPDATE SET seq = chat_room_seq.seq + 1, updated_at = $2
               RETURNING seq"#,
            [room_id.into(), now.into()],
        ))
        .await?;
    let row = row.ok_or_else(|| ChatError::Internal("seq allocation returned no row".into()))?;
    let seq: i64 = row.try_get("", "seq")?;
    Ok(seq)
}

/// Current room tip without consuming a seq; used for non-message events so
/// edits/pins/deletes never inflate unread counters or reorder listings.
pub async fn current_room_seq(db: &impl ConnectionTrait, room_id: Uuid) -> i64 {
    room_seq::Entity::find_by_id(room_id)
        .one(db)
        .await
        .ok()
        .flatten()
        .map(|s| s.seq)
        .unwrap_or(0)
}

pub async fn record_mentions(
    db: &impl ConnectionTrait,
    room_id: Uuid,
    message_id: Uuid,
    uids: &[i64],
) -> Result<(), ChatError> {
    let now = OffsetDateTime::now_utc();
    for &uid in uids {
        user_recent_mentions::ActiveModel {
            id: Set(Uuid::now_v7()),
            user_id: Set(uid),
            channel_id: Set(room_id),
            message_id: Set(message_id),
            created_at: Set(now),
        }
        .insert(db)
        .await?;
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
pub async fn notify_user(
    db: &impl ConnectionTrait,
    pending: &push::notify::PendingNotifications,
    recipient: i64,
    ntype: NotificationType,
    title: String,
    body: Option<String>,
    data: Option<serde_json::Value>,
    actor_uid: Option<i64>,
) -> Result<(), ChatError> {
    // The row stores already-rendered text: there is no schema column for the
    // language, so callers resolve it per recipient and pass finished copy in.
    let row = push::notify::insert(db, recipient, ntype, title, body, data, actor_uid).await?;
    pending.push(recipient, row.to_event_value());
    Ok(())
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NotifyDecision {
    Notify,
    Skip,
}

pub(crate) fn mute_active(muted: bool, mute_until: Option<OffsetDateTime>) -> bool {
    muted && mute_until.is_none_or(|until| until > OffsetDateTime::now_utc())
}

pub async fn notification_decision(
    service: &ChatService,
    room_id: Uuid,
    space_id: Option<Uuid>,
    uid: i64,
    mentioned: bool,
    everyone: bool,
) -> Result<NotifyDecision, ChatError> {
    if let Some(setting) = room_user_settings::Entity::find_by_id((uid, room_id))
        .one(service.database.read())
        .await?
    {
        if mute_active(setting.muted, setting.mute_until) {
            return Ok(if mentioned && !everyone {
                NotifyDecision::Notify
            } else {
                NotifyDecision::Skip
            });
        }
        match setting.notification_level {
            RoomNotificationLevel::All => return Ok(NotifyDecision::Notify),
            RoomNotificationLevel::Mentions | RoomNotificationLevel::Nothing => {
                return Ok(if mentioned && !everyone {
                    NotifyDecision::Notify
                } else {
                    NotifyDecision::Skip
                });
            }
            RoomNotificationLevel::Default => {}
        }
    }

    if let Some(space_id) = space_id
        && let Some(member) = service.participants.access.member(space_id, uid).await?
    {
        if mute_active(member.muted, member.mute_until) && !(mentioned && !everyone) {
            return Ok(NotifyDecision::Skip);
        }
        return Ok(match member.message_notifications {
            MessageNotifications::All => NotifyDecision::Notify,
            MessageNotifications::Mentions | MessageNotifications::Nothing => {
                if mentioned && (!everyone || !member.suppress_everyone) {
                    NotifyDecision::Notify
                } else {
                    NotifyDecision::Skip
                }
            }
        });
    }

    Ok(NotifyDecision::Notify)
}

/// One @everyone fan-out. Grouped into a struct because the room scope, the
/// message and the actor are all `Uuid`-shaped and would be easy to transpose.
#[derive(Clone)]
pub struct EveryoneScope {
    pub room_id: Uuid,
    pub space_id: Option<Uuid>,
    pub is_private: bool,
    pub message_id: Uuid,
    pub body: String,
    pub actor: i64,
    pub explicit: Vec<i64>,
    pub request_id: Option<String>,
    pub trace: Option<String>,
}

/// The recipients of an @everyone mention: room/space members minus the actor
/// and the explicitly mentioned users.
pub async fn everyone_candidates(service: &ChatService, scope: &EveryoneScope) -> Vec<i64> {
    let room_id = scope.room_id;
    let actor = scope.actor;
    let candidates: Vec<i64> = match scope.space_id {
        None => room_members::Entity::find()
            .filter(room_members::Column::RoomId.eq(room_id))
            .filter(room_members::Column::LeftAt.is_null())
            .all(service.database.read())
            .await
            .unwrap_or_default()
            .into_iter()
            .map(|m| m.user_id)
            .collect(),
        Some(space_id) => {
            let Ok(space) = service.participants.access.by_id(space_id).await else {
                return Vec::new();
            };
            let mut ids: Vec<i64> = space
                .members(actor)
                .await
                .unwrap_or_default()
                .into_iter()
                .filter(|m| !m.suppress_everyone)
                .map(|m| m.user_id)
                .collect();
            if scope.is_private {
                let mut allowed: Vec<i64> = models::room::room_members::Entity::find()
                    .filter(models::room::room_members::Column::RoomId.eq(room_id))
                    .filter(models::room::room_members::Column::LeftAt.is_null())
                    .all(service.database.read())
                    .await
                    .unwrap_or_default()
                    .into_iter()
                    .map(|row| row.user_id)
                    .collect();
                if let Ok(Some(room)) = rooms::Entity::find_by_id(room_id)
                    .one(service.database.read())
                    .await
                {
                    allowed.push(room.owner_id);
                }
                ids.retain(|uid| allowed.contains(uid));
            }
            ids
        }
    };

    candidates
        .into_iter()
        .filter(|uid| *uid != actor && !scope.explicit.contains(uid))
        .collect()
}

/// Fan out an @everyone mention to one bounded batch of recipients: writes
/// notifications and recent mentions, then pushes a WS user event.
pub async fn fanout_everyone_chunk(
    service: &ChatService,
    scope: &EveryoneScope,
    recipients: &[i64],
) {
    let data = serde_json::json!({
        "room_id": scope.room_id,
        "message_id": scope.message_id,
        "everyone": true,
    });
    for &uid in recipients {
        match notification_decision(service, scope.room_id, scope.space_id, uid, true, true).await {
            Ok(NotifyDecision::Notify) => {}
            _ => continue,
        }
        let now = OffsetDateTime::now_utc();
        let _ = user_recent_mentions::ActiveModel {
            id: Set(Uuid::now_v7()),
            user_id: Set(uid),
            channel_id: Set(scope.room_id),
            message_id: Set(scope.message_id),
            created_at: Set(now),
        }
        .insert(service.database.write())
        .await;
        let _ = service
            .audience
            .notifications
            .create(
                uid,
                NotificationType::Mention,
                i18n::catalog::notification_mention_message_title(
                    service.audience.recipient_locale(uid).await,
                )
                .to_string(),
                Some(scope.body.clone()),
                Some(data.clone()),
                Some(scope.actor),
            )
            .await;
        service.audience.realtime
            .publish_user(
                uid,
                &serde_json::json!({"kind": "mention", "room_id": scope.room_id, "message_id": scope.message_id, "everyone": true}),
            )
            .await;
    }
}

/// Mention e-mails honour per-room / per-space mute and notification level:
/// Nothing, muted (even mid-window) and blocked settings opt out.
pub async fn mention_email_allowed(
    service: &ChatService,
    room_id: Uuid,
    space_id: Option<Uuid>,
    uid: i64,
) -> Result<bool, ChatError> {
    if let Some(setting) = room_user_settings::Entity::find_by_id((uid, room_id))
        .one(service.database.read())
        .await?
    {
        if mute_active(setting.muted, setting.mute_until) {
            return Ok(false);
        }
        match setting.notification_level {
            RoomNotificationLevel::Nothing => return Ok(false),
            RoomNotificationLevel::All | RoomNotificationLevel::Mentions => return Ok(true),
            RoomNotificationLevel::Default => {}
        }
    }
    if let Some(space_id) = space_id
        && let Some(member) = service.participants.access.member(space_id, uid).await?
    {
        if mute_active(member.muted, member.mute_until) {
            return Ok(false);
        }
        return Ok(!matches!(
            member.message_notifications,
            MessageNotifications::Nothing
        ));
    }
    Ok(true)
}

pub async fn publish_message_event(
    service: &ChatService,
    room_id: Uuid,
    seq: i64,
    message: serde_json::Value,
) {
    let event = RoomEvent::MessagePublished {
        room_id,
        seq,
        message,
    };
    if let Ok(payload) = serde_json::to_value(&event) {
        crate::services::outbound::enqueue_outbound_event(
            service,
            room_id,
            "message_published",
            &payload,
        )
        .await;
    }
    service.audience.realtime.publish_room(&event).await;
}

pub async fn advance_read_state(
    db: &impl ConnectionTrait,
    uid: i64,
    room_id: Uuid,
    message_id: Option<Uuid>,
    seq: i64,
) -> Result<bool, ChatError> {
    let now = OffsetDateTime::now_utc();
    let row = db
        .query_one_raw(Statement::from_sql_and_values(
            DatabaseBackend::Postgres,
            r#"INSERT INTO chat_read_state
                   (user_id, room_id, last_read_message_id, last_read_seq, last_viewed_at, updated_at)
               VALUES ($1, $2, $3, $4, $5, $5)
               ON CONFLICT (user_id, room_id) DO UPDATE SET
                   last_read_message_id = EXCLUDED.last_read_message_id,
                   last_read_seq = EXCLUDED.last_read_seq,
                   last_viewed_at = EXCLUDED.last_viewed_at,
                   updated_at = EXCLUDED.updated_at
               WHERE chat_read_state.last_read_seq < EXCLUDED.last_read_seq
               RETURNING room_id"#,
            [uid.into(), room_id.into(), message_id.into(), seq.into(), now.into()],
        ))
        .await?;
    Ok(row.is_some())
}