//! S4 — `SlackAdapter::run()` cancel / shutdown lifecycle.
//!
//! The mock-injected gates drive `handle_event` directly and never exercise the
//! `run()` loop's cancel + `None`→break close paths. This gate spawns the real
//! `run()` over a `MockSlackTransport`, proves an injected event is handled,
//! then proves BOTH stop paths: (1) flipping the cancel signal exits the loop
//! cleanly, and (2) closing the inbound source (`next_event` → `None`) also
//! exits.

mod slack_mock;
use slack_mock::MockSlackTransport;

use async_trait::async_trait;
use car_proto::{CreateHostApprovalRequest, HostApprovalStatus};
use car_server_types::channel::{ChannelId as ChId, InboundSink};
use car_server_types::host::HostState;
use car_messaging::messaging_config::{ChannelId, MessagingConfigStore};
use car_messaging::slack_adapter::SlackAdapter;
use car_server_types::channel::InboundChannel;
use car_ffi_common::integrations::InboundMessage;
use std::sync::Arc;
use tempfile::TempDir;

const SLACK_MEMBER: &str = "U012ABCDEF";
const SLACK_CHANNEL: &str = "D024BE91L";

/// A do-nothing delivery sink — the Slack adapter delivers internally (button
/// clicks / pairing DMs), so the seam's `sink` is unused, exactly as in boot.
struct NoopSink;

#[async_trait]
impl InboundSink for NoopSink {
    async fn deliver(&self, _channel: ChId, _msg: &InboundMessage) {}
}

#[tokio::test]
async fn run_handles_event_then_exits_on_cancel() {
    let dir = TempDir::new().unwrap();
    let store = MessagingConfigStore::with_base_dir(dir.path());
    store.set_enabled_for(ChannelId::Slack, true).unwrap();
    store.add_handle_for(ChannelId::Slack, SLACK_MEMBER).unwrap();

    let host = Arc::new(HostState::new());
    // Seed an approval the injected Approve button will resolve.
    let id = host
        .create_approval(
            None,
            CreateHostApprovalRequest {
                agent_id: None,
                action: "send wire transfer".to_string(),
                details: serde_json::Value::Null,
                options: vec![],
                system_level: true,
            },
        )
        .await
        .unwrap()
        .id;

    let transport = MockSlackTransport::new();
    let adapter = Arc::new(SlackAdapter::new(
        host.clone(),
        MessagingConfigStore::with_base_dir(dir.path()),
        transport.clone(),
        SLACK_CHANNEL,
    ));

    // Spawn the real run() loop on a cancel watch.
    let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);
    let adapter_run = adapter.clone();
    let run_handle = tokio::spawn(async move {
        let sink = NoopSink;
        adapter_run.run(&sink, cancel_rx).await;
    });

    // Inject an Approve → the loop pulls it off the transport and handles it.
    transport.inject_approve(&id, SLACK_MEMBER);

    // Wait (bounded) for the approval to resolve, proving run() consumed + handled.
    let resolved = wait_until_resolved(&host, &id, std::time::Duration::from_secs(2)).await;
    assert!(resolved, "run() must consume the injected event and resolve the approval");

    // Flip cancel → the loop must exit cleanly (its tokio::select! cancel arm).
    cancel_tx.send(true).unwrap();
    let exited = tokio::time::timeout(std::time::Duration::from_secs(2), run_handle).await;
    assert!(exited.is_ok(), "run() must exit promptly when cancel flips");
    exited.unwrap().expect("run task should not panic");
}

#[tokio::test]
async fn run_exits_when_inbound_source_closes() {
    // The `None`→break close path: when the transport's inbound source is
    // exhausted (sender dropped ⇒ next_event → None), run() exits even without
    // a cancel.
    let dir = TempDir::new().unwrap();
    let store = MessagingConfigStore::with_base_dir(dir.path());
    store.set_enabled_for(ChannelId::Slack, true).unwrap();

    let host = Arc::new(HostState::new());
    let transport = MockSlackTransport::new();
    let adapter = Arc::new(SlackAdapter::new(
        host.clone(),
        MessagingConfigStore::with_base_dir(dir.path()),
        transport.clone(),
        SLACK_CHANNEL,
    ));

    let (_cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);
    let adapter_run = adapter.clone();
    let run_handle = tokio::spawn(async move {
        let sink = NoopSink;
        adapter_run.run(&sink, cancel_rx).await;
    });

    // Drop the only sender so the inbound mpsc closes ⇒ next_event yields None.
    transport.close_inbound();

    let exited = tokio::time::timeout(std::time::Duration::from_secs(2), run_handle).await;
    assert!(
        exited.is_ok(),
        "run() must exit when the inbound source closes (None → break)"
    );
    exited.unwrap().expect("run task should not panic");
}

async fn wait_until_resolved(
    host: &Arc<HostState>,
    id: &str,
    timeout: std::time::Duration,
) -> bool {
    let deadline = std::time::Instant::now() + timeout;
    loop {
        let resolved = host
            .approvals()
            .await
            .into_iter()
            .find(|a| a.id == id)
            .map(|a| a.status == HostApprovalStatus::Resolved)
            .unwrap_or(false);
        if resolved {
            return true;
        }
        if std::time::Instant::now() >= deadline {
            return false;
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
}
