//! Background task completion sink.
//!
//! The sink is the durable boundary between a runner completion and later
//! mailbox/wake work: it writes the terminal task state to the
//! [`BackgroundTaskStore`](domain::ports::BackgroundTaskStore) first, then emits
//! a lightweight "ready to deliver" signal. It never wakes an agent and never
//! enqueues mailbox items; B4/B5 consume the ready signal.

use std::collections::HashSet;
use std::sync::Arc;

use thiserror::Error;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::sync::Mutex;
use tokio::task::JoinHandle;

use domain::ports::{BackgroundTaskCompletion, BackgroundTaskRunner, BackgroundTaskStore};
use domain::{
    AgentInbox, BackgroundTaskPortError, InboxError, InboxItem, InboxItemKind, InboxReceiptStatus,
    InboxSource, ProjectId, TaskId,
};

/// Signal emitted only after a completion has been persisted.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BackgroundTaskReadyToDeliver {
    /// Persisted task id.
    pub task_id: TaskId,
    /// Owning project, copied from the persisted task.
    pub project_id: ProjectId,
    /// Agent that owns the completion delivery.
    pub owner_agent_id: domain::AgentId,
}

/// Outcome of processing one completion event.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BackgroundCompletionSinkOutcome {
    /// Completion was persisted and signalled as ready to deliver.
    PersistedAndSignalled(BackgroundTaskReadyToDeliver),
    /// Task was already terminal, so the duplicate completion was ignored.
    IgnoredAlreadyTerminal {
        /// Ignored task id.
        task_id: TaskId,
    },
    /// Another completion for this task was already processed by this sink.
    IgnoredDuplicate {
        /// Ignored task id.
        task_id: TaskId,
    },
    /// The task no longer exists in the store.
    IgnoredMissingTask {
        /// Ignored task id.
        task_id: TaskId,
    },
}

/// Errors raised by the completion sink.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum BackgroundCompletionSinkError {
    /// Store operation failed.
    #[error("background completion store failed: {0}")]
    Store(#[from] BackgroundTaskPortError),
    /// The ready-to-deliver channel is closed.
    #[error("background completion ready signal channel is closed")]
    ReadySignalClosed,
}

/// Consumes runner completions, persists terminal state, then signals delivery.
pub struct BackgroundCompletionSink {
    store: Arc<dyn BackgroundTaskStore>,
    ready: UnboundedSender<BackgroundTaskReadyToDeliver>,
    processed: Mutex<HashSet<TaskId>>,
}

/// Handle for the ready-to-inbox bridge task.
pub type BackgroundReadyInboxBridgeHandle = JoinHandle<()>;

impl BackgroundCompletionSink {
    /// Builds a sink from a task store and ready-to-deliver channel.
    #[must_use]
    pub fn new(
        store: Arc<dyn BackgroundTaskStore>,
        ready: UnboundedSender<BackgroundTaskReadyToDeliver>,
    ) -> Self {
        Self {
            store,
            ready,
            processed: Mutex::new(HashSet::new()),
        }
    }

    /// Starts consuming [`BackgroundTaskRunner::subscribe_completions`].
    ///
    /// The domain completion stream is a blocking iterator, so consumption runs on
    /// a blocking task and re-enters the current Tokio runtime for persistence.
    ///
    /// # Panics
    /// Panics if called outside a Tokio runtime.
    #[must_use]
    pub fn start_from_runner(
        self: Arc<Self>,
        runner: Arc<dyn BackgroundTaskRunner>,
    ) -> JoinHandle<()> {
        let mut stream = runner.subscribe_completions();
        let runtime = tokio::runtime::Handle::current();
        tokio::task::spawn_blocking(move || {
            while let Some(completion) = stream.next() {
                let _ = runtime.block_on(self.process_completion(completion));
            }
        })
    }

    /// Processes a single completion event.
    ///
    /// Order is strict: `store.save(terminal_task)` must succeed before the ready
    /// signal is sent. If persistence fails, no ready signal is emitted and the
    /// task id is released so a later retry can persist it.
    ///
    /// # Errors
    /// [`BackgroundCompletionSinkError`] if the store fails, the task transition
    /// is invalid, or the ready channel is closed after persistence.
    pub async fn process_completion(
        &self,
        completion: BackgroundTaskCompletion,
    ) -> Result<BackgroundCompletionSinkOutcome, BackgroundCompletionSinkError> {
        {
            let mut processed = self.processed.lock().await;
            if processed.contains(&completion.task_id) {
                return Ok(BackgroundCompletionSinkOutcome::IgnoredDuplicate {
                    task_id: completion.task_id,
                });
            }
            processed.insert(completion.task_id);
        }

        let outcome = self.persist_and_signal(completion.clone()).await;
        if outcome.is_err() {
            self.processed.lock().await.remove(&completion.task_id);
        }
        outcome
    }

    async fn persist_and_signal(
        &self,
        completion: BackgroundTaskCompletion,
    ) -> Result<BackgroundCompletionSinkOutcome, BackgroundCompletionSinkError> {
        let Some(task) = self.store.get(completion.task_id).await? else {
            return Ok(BackgroundCompletionSinkOutcome::IgnoredMissingTask {
                task_id: completion.task_id,
            });
        };
        if task.is_terminal() {
            return Ok(BackgroundCompletionSinkOutcome::IgnoredAlreadyTerminal {
                task_id: completion.task_id,
            });
        }

        let terminal = task.complete(completion.result).map_err(|e| {
            BackgroundCompletionSinkError::Store(BackgroundTaskPortError::Invalid(e.to_string()))
        })?;
        self.store.save(&terminal).await?;

        let ready = BackgroundTaskReadyToDeliver {
            task_id: terminal.id,
            project_id: terminal.project_id,
            owner_agent_id: terminal.owner_agent_id,
        };
        self.ready
            .send(ready.clone())
            .map_err(|_| BackgroundCompletionSinkError::ReadySignalClosed)?;
        Ok(BackgroundCompletionSinkOutcome::PersistedAndSignalled(
            ready,
        ))
    }
}

/// Starts the B3→B4 bridge: persisted completions become inbox items.
///
/// Overflow of completion/system items is intentionally non-fatal: the completion
/// is already durable and remains delivery-pending for boot reconcile.
#[must_use]
pub fn start_background_ready_inbox_bridge(
    mut ready: UnboundedReceiver<BackgroundTaskReadyToDeliver>,
    inbox: Arc<dyn AgentInbox>,
) -> BackgroundReadyInboxBridgeHandle {
    tokio::spawn(async move {
        while let Some(ready) = ready.recv().await {
            let item = InboxItem {
                id: domain::TicketId::new_random(),
                agent_id: ready.owner_agent_id,
                source: InboxSource::BackgroundTask {
                    task_id: ready.task_id,
                },
                kind: InboxItemKind::BackgroundCompletion,
                body: format!("Background task {} completed.", ready.task_id),
                created_at_ms: now_ms(),
                correlation_id: Some(ready.task_id.to_string()),
            };
            match inbox.enqueue_message(
                domain::RuntimeAgentKey::new(ready.project_id, ready.owner_agent_id),
                item,
            ) {
                Ok(receipt) if receipt.status == InboxReceiptStatus::Deferred => {
                    application::diag!(
                        "[background-task] completion deferred: task={} owner={} queue_depth={}",
                        ready.task_id,
                        ready.owner_agent_id,
                        receipt.depth
                    );
                }
                Ok(_) => {}
                Err(InboxError::InboxFull { .. }) => {
                    application::diag!(
                        "[background-task] completion inbox full but durable: task={} owner={}",
                        ready.task_id,
                        ready.owner_agent_id
                    );
                }
                Err(err) => {
                    application::diag!(
                        "[background-task] completion inbox enqueue failed: task={} owner={} err={err}",
                        ready.task_id,
                        ready.owner_agent_id
                    );
                }
            }
        }
    })
}

fn now_ms() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
        .unwrap_or(0)
}