use std::time::Duration;

use export::AppState;
use models::users::user_recent_mentions;
use models::users::user_scheduled_messages::{self, ScheduledMessageStatus};
use sea_orm::sea_query::Expr;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};
use time::OffsetDateTime;

use crate::errors::ChatError;
use crate::services::messages::SendMessageCmd;
use crate::services::service::ChatService;

pub(crate) const STUCK_CLAIM_MINUTES: i64 = 10;
const MENTION_RETENTION_DAYS: i64 = 90;

impl ChatService {
    pub async fn deliver_due_scheduled_messages(&self) -> Result<u64, ChatError> {
        let now = OffsetDateTime::now_utc();

        // Recover claims abandoned by a crashed instance.
        let _ = user_scheduled_messages::Entity::update_many()
            .col_expr(
                user_scheduled_messages::Column::Status,
                Expr::value(ScheduledMessageStatus::Pending),
            )
            .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Sending))
            .filter(
                user_scheduled_messages::Column::ScheduledAt
                    .lt(now - time::Duration::minutes(STUCK_CLAIM_MINUTES)),
            )
            .exec(self.db().write())
            .await?;

        let due = user_scheduled_messages::Entity::find()
            .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Pending))
            .filter(user_scheduled_messages::Column::ScheduledAt.lte(now))
            .limit(100)
            .all(self.db().write())
            .await?;
        let count = due.len() as u64;
        for scheduled in due {
            // Atomic claim so concurrent replicas deliver each row at most once.
            let claimed = user_scheduled_messages::Entity::update_many()
                .col_expr(
                    user_scheduled_messages::Column::Status,
                    Expr::value(ScheduledMessageStatus::Sending),
                )
                .filter(user_scheduled_messages::Column::Id.eq(scheduled.id))
                .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Pending))
                .exec(self.db().write())
                .await?;
            if claimed.rows_affected == 0 {
                continue;
            }

            let id = scheduled.id;
            let user_id = scheduled.user_id;
            let channel_id = scheduled.target_channel_id;
            let cmd: SendMessageCmd = match serde_json::from_value(scheduled.payload.clone()) {
                Ok(cmd) => cmd,
                Err(error) => {
                    tracing::warn!(%error, %id, "invalid scheduled message payload");
                    let mut am: user_scheduled_messages::ActiveModel = scheduled.into();
                    am.status = Set(ScheduledMessageStatus::Cancelled);
                    am.update(self.db().write()).await?;
                    continue;
                }
            };
            let next_status = match self.send_message(user_id, channel_id, cmd).await {
                Ok(_) => ScheduledMessageStatus::Sent,
                Err(error) => {
                    tracing::warn!(%error, %id, "scheduled message delivery failed");
                    // Authz/validation failures are permanent; transient slow
                    // mode retries later.
                    if error.is_rate_limited() {
                        ScheduledMessageStatus::Pending
                    } else {
                        ScheduledMessageStatus::Failed
                    }
                }
            };
            user_scheduled_messages::Entity::update_many()
                .col_expr(
                    user_scheduled_messages::Column::Status,
                    Expr::value(next_status),
                )
                .filter(user_scheduled_messages::Column::Id.eq(id))
                .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Sending))
                .exec(self.db().write())
                .await?;
        }
        Ok(count)
    }

    pub async fn prune_old_mentions(&self) -> Result<(), ChatError> {
        let cutoff = OffsetDateTime::now_utc() - time::Duration::days(MENTION_RETENTION_DAYS);
        user_recent_mentions::Entity::delete_many()
            .filter(user_recent_mentions::Column::CreatedAt.lt(cutoff))
            .exec(self.db().write())
            .await?;
        Ok(())
    }
}

pub async fn run_background_tasks(state: AppState) {
    let service = ChatService::new(state);
    let mut interval = tokio::time::interval(Duration::from_secs(30));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let mut tick: u32 = 0;
    loop {
        interval.tick().await;
        if let Err(error) = service.deliver_due_reminders().await {
            tracing::warn!(%error, "chat reminder delivery failed");
        }
        if let Err(error) = service.deliver_due_scheduled_messages().await {
            tracing::warn!(%error, "chat scheduled message delivery failed");
        }
        tick = tick.wrapping_add(1);
        if tick % 120 == 0
            && let Err(error) = service.prune_old_mentions().await
        {
            tracing::warn!(%error, "chat mention prune failed");
        }
    }
}