//! MC-8 — Fan-out + one shared code; idempotent cross-channel resolve.
//!
//! With BOTH channels enabled, one eligible approval → BOTH sinks (iMessage
//! SpySender + Slack mock) receive a prompt carrying the SAME shared code,
//! minted ONCE. A Slack button-resolve racing an iMessage text-resolve →
//! exactly ONE `approval.resolved`, no double-stamp, no false eviction (the
//! untouched first-writer-wins guard).

mod slack_mock;
use slack_mock::MockSlackTransport;

use car_proto::{CreateHostApprovalRequest, HostApprovalStatus};
use car_server_types::approval_core::ApprovalCore;
use car_messaging::fanout::FanoutCoordinator;
use car_server_types::host::HostState;
use car_messaging::messaging_config::{ChannelId, MessagingConfigStore};
use car_messaging::messaging_orchestrator::{MessageSender, MessagingOrchestrator, SendOutcome};
use car_messaging::slack_adapter::SlackAdapter;
use car_messaging::slack_adapter::SlackTransport; // for `next_event` on the mock
use std::sync::{Arc, Mutex};
use tempfile::TempDir;

const IMSG_HANDLE: &str = "+15551234567";
const SLACK_MEMBER: &str = "U012ABCDEF";
const SLACK_CHANNEL: &str = "D024BE91L";

/// Capturing iMessage send spy (same shape as the #403 SpySender).
#[derive(Default)]
struct SpySender {
    sent: Mutex<Vec<(String, String)>>,
}
impl SpySender {
    fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }
    fn calls(&self) -> Vec<(String, String)> {
        self.sent.lock().unwrap().clone()
    }
}
impl MessageSender for SpySender {
    fn send(&self, handle: &str, body: &str) -> Result<SendOutcome, String> {
        self.sent
            .lock()
            .unwrap()
            .push((handle.to_string(), body.to_string()));
        Ok(SendOutcome::ok())
    }
}

/// Extract the per-approval code the iMessage prompt carries
/// (`Reply \`A0 approve\``) — the token after the first backtick.
fn imsg_code(body: &str) -> Option<String> {
    body.split('`')
        .nth(1)?
        .split_whitespace()
        .next()
        .map(|s| s.to_string())
}

/// Extract the shared code the Slack prompt text carries (`(ref \`A0\`)`) — the
/// token inside the backticks following "ref ".
fn slack_code(text: &str) -> Option<String> {
    let after = text.split("ref `").nth(1)?;
    after.split('`').next().map(|s| s.to_string())
}

struct FanoutHarness {
    host: Arc<HostState>,
    spy: Arc<SpySender>,
    transport: Arc<MockSlackTransport>,
    imsg: Arc<MessagingOrchestrator>,
    slack: Arc<SlackAdapter>,
    coordinator: FanoutCoordinator,
    _dir: TempDir,
}

impl FanoutHarness {
    /// BOTH channels enabled + paired.
    fn both_enabled() -> Self {
        let dir = TempDir::new().expect("tempdir");
        let store = MessagingConfigStore::with_base_dir(dir.path());
        store.set_enabled_for(ChannelId::IMessage, true).unwrap();
        store.add_handle_for(ChannelId::IMessage, IMSG_HANDLE).unwrap();
        store.set_enabled_for(ChannelId::Slack, true).unwrap();
        store.add_handle_for(ChannelId::Slack, SLACK_MEMBER).unwrap();

        let host = Arc::new(HostState::new());
        let spy = SpySender::new();
        let imsg = Arc::new(MessagingOrchestrator::new(
            host.clone(),
            MessagingConfigStore::with_base_dir(dir.path()),
            spy.clone() as Arc<dyn MessageSender>,
            dir.path().to_path_buf(),
        ));
        let transport = MockSlackTransport::new();
        let slack = Arc::new(SlackAdapter::new(
            host.clone(),
            MessagingConfigStore::with_base_dir(dir.path()),
            transport.clone(),
            SLACK_CHANNEL,
        ));
        let coordinator = FanoutCoordinator::new(
            ApprovalCore::new(host.clone()),
            Some(imsg.clone()),
            Some(slack.clone()),
        );
        Self {
            host,
            spy,
            transport,
            imsg,
            slack,
            coordinator,
            _dir: dir,
        }
    }

    async fn seed(&self, action: &str) -> String {
        let req = CreateHostApprovalRequest {
            agent_id: None,
            action: action.to_string(),
            details: serde_json::Value::Null,
            options: vec![],
            system_level: true,
        };
        self.host.create_approval(None, req).await.unwrap().id
    }

    async fn is_resolved(&self, id: &str) -> bool {
        self.host
            .approvals()
            .await
            .into_iter()
            .find(|a| a.id == id)
            .map(|a| a.status == HostApprovalStatus::Resolved)
            .unwrap_or(false)
    }

    async fn resolved_count(&self) -> usize {
        self.host
            .events(500)
            .await
            .into_iter()
            .filter(|e| e.kind == "approval.resolved")
            .count()
    }

    async fn resolution(&self, id: &str) -> Option<String> {
        self.host
            .approvals()
            .await
            .into_iter()
            .find(|a| a.id == id)
            .and_then(|a| a.resolution)
    }
}

#[tokio::test]
async fn both_channels_receive_one_shared_code() {
    let h = FanoutHarness::both_enabled();
    let id = h.seed("send wire transfer").await;

    // One fan-out tick: mint ONE shared code, deliver to BOTH channels.
    h.coordinator.observe_and_fanout().await;

    // iMessage sink received exactly one prompt.
    let imsg_calls = h.spy.calls();
    assert_eq!(imsg_calls.len(), 1, "iMessage got one prompt: {imsg_calls:?}");
    assert_eq!(imsg_calls[0].0, IMSG_HANDLE, "to the paired iMessage handle");
    let code_a = imsg_code(&imsg_calls[0].1).expect("iMessage carries a code");

    // Slack sink received exactly one Block Kit post, button value = approval_id.
    let posts = h.transport.captured();
    assert_eq!(posts.len(), 1, "Slack got one post: {posts:?}");
    assert!(
        posts[0].has_approve_deny_buttons(),
        "Slack post carries Approve/Deny buttons"
    );
    assert_eq!(
        posts[0].button_value().as_deref(),
        Some(id.as_str()),
        "Slack button value carries the approval_id"
    );
    let code_b = slack_code(&posts[0].text).expect("Slack carries the shared code");

    // The SAME shared code reached both channels (minted once).
    assert_eq!(
        code_a, code_b,
        "iMessage and Slack carry the SAME shared code (minted once)"
    );

    // MC-1 byte-for-byte: the fan-out iMessage body is byte-identical to the
    // poll-path grammar `outbound_body(approval, code)` — the EXACT function the
    // iMessage-only poll path renders. This pins the MC-1 guarantee against a
    // future drift in `send_shared_prompt` (e.g. a reworded prompt that no
    // longer matches the poll path, or a stray code-format change).
    let approval = h
        .host
        .approvals()
        .await
        .into_iter()
        .find(|a| a.id == id)
        .expect("the seeded approval row");
    assert_eq!(
        imsg_calls[0].1,
        car_messaging::messaging_orchestrator::outbound_body(&approval, &code_a),
        "fan-out iMessage body is byte-for-byte the poll-path outbound_body grammar (MC-1)"
    );

    // Idempotency: a second tick does NOT re-prompt either channel.
    h.coordinator.observe_and_fanout().await;
    assert_eq!(h.spy.calls().len(), 1, "no duplicate iMessage prompt");
    assert_eq!(h.transport.post_count(), 1, "no duplicate Slack post");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cross_channel_race_resolves_exactly_once() {
    let h = FanoutHarness::both_enabled();
    let id = h.seed("delete prod database").await;
    h.coordinator.observe_and_fanout().await;

    // Pull the shared code so the iMessage text-resolve can name it.
    let code = imsg_code(&h.spy.calls()[0].1).expect("shared code");

    // The RACE: a Slack button-resolve and an iMessage text-resolve for the
    // SAME approval. The first to hit the untouched first-writer-wins guard
    // wins; the second no-ops.
    //
    // Drive both: Slack Approve button, then iMessage `<code> deny`. Exactly one
    // approval.resolved must fire; the row must NOT double-stamp.
    h.transport.inject_approve(&id, SLACK_MEMBER);
    let ev = h.transport.next_event().await.unwrap();

    // Issue both resolves under a REAL race: spawn each onto its own worker
    // thread (multi_thread runtime, 2 workers) so they execute on separate
    // threads, then join. A bare current-thread `tokio::join!` only cooperatively
    // interleaves at await points — this exercises genuine concurrent entry into
    // the first-writer-wins guard.
    let slack = h.slack.clone();
    let imsg = h.imsg.clone();
    let imsg_msg = car_ffi_common::integrations::InboundMessage {
        rowid: 1,
        handle_id: IMSG_HANDLE.to_string(),
        body: format!("{code} deny"),
        date: 0,
    };
    let slack_task = tokio::spawn(async move {
        slack.handle_event(&ev).await;
    });
    let imsg_task = tokio::spawn(async move {
        imsg.handle_inbound(&imsg_msg).await;
    });
    let (sr, ir) = tokio::join!(slack_task, imsg_task);
    sr.expect("slack resolve task");
    ir.expect("imsg resolve task");

    assert!(h.is_resolved(&id).await, "the approval is Resolved");
    assert_eq!(
        h.resolved_count().await,
        1,
        "exactly ONE approval.resolved across both channels (no double-resolve)"
    );
    // The first writer's resolution stands; the loser did not overwrite it.
    let res = h.resolution(&id).await;
    assert!(
        res.as_deref() == Some("approve") || res.as_deref() == Some("deny"),
        "exactly one resolution stamped (the first writer's), got {res:?}"
    );

    // After the race, a second fan-out tick must NOT re-prompt (the resolved
    // approval is evicted from the shared map; no false re-surfacing).
    let imsg_before = h.spy.calls().len();
    let slack_before = h.transport.post_count();
    h.coordinator.observe_and_fanout().await;
    assert_eq!(
        h.spy.calls().len(),
        imsg_before,
        "resolved approval not re-prompted on iMessage"
    );
    assert_eq!(
        h.transport.post_count(),
        slack_before,
        "resolved approval not re-prompted on Slack"
    );
}

#[tokio::test]
async fn fanout_with_only_slack_enabled_still_one_code() {
    // Single-channel fan-out (only Slack) still flows through the one shared-code
    // path — proving fan-out reduces cleanly to single-channel delivery.
    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());
    let transport = MockSlackTransport::new();
    let slack = Arc::new(SlackAdapter::new(
        host.clone(),
        MessagingConfigStore::with_base_dir(dir.path()),
        transport.clone(),
        SLACK_CHANNEL,
    ));
    let coordinator =
        FanoutCoordinator::new(ApprovalCore::new(host.clone()), None, Some(slack));

    let req = CreateHostApprovalRequest {
        agent_id: None,
        action: "ship it".to_string(),
        details: serde_json::Value::Null,
        options: vec![],
        system_level: true,
    };
    let id = host.create_approval(None, req).await.unwrap().id;

    coordinator.observe_and_fanout().await;
    let posts = transport.captured();
    assert_eq!(posts.len(), 1, "Slack-only fan-out posts once");
    assert_eq!(posts[0].button_value().as_deref(), Some(id.as_str()));
}
