use std::sync::Arc;

use chrono::{DateTime, Utc};
use matrix_sdk::Room;
use tokio::sync::{broadcast, mpsc, oneshot, Mutex, RwLock};
use tracing::{debug, error, warn};
use uuid::Uuid;

use dirigent_protocol::accumulator::{AccumulatedMessage, AccumulatedPart, MessageAccumulator, ToolCallData};
use dirigent_protocol::{ContentBlock, Event, MessageRole, SessionUpdate};

/// Command proxy sent from Matrix → Dirigent direction.
///
/// The caller who wires up the share is responsible for translating this
/// into a real `ConnectorCommand::SendMessage` (or equivalent) using their
/// own connector/session handle.
#[derive(Debug, Clone)]
pub struct ConnectorCommandProxy {
    pub session_id: String,
    pub text: String,
}

/// A bidirectional bridge between a Dirigent session and a Matrix room.
///
/// **Dirigent → Matrix**: Subscribes to a broadcast event stream, forwards
/// completed assistant messages and session errors into the Matrix room as
/// `m.notice` messages.
///
/// **Matrix → Dirigent**: The `inject_message` method is called by
/// `MatrixService` when a Matrix message arrives; it sends a
/// `ConnectorCommandProxy` through an mpsc channel that the owner can read.
pub struct MatrixSessionShare {
    /// Connector that owns the session.
    pub connector_id: String,
    /// Session being bridged (native connector session ID).
    pub session_id: String,
    /// Scroll ID for the archived session this share is scoped to.
    ///
    /// Required by `SessionStream::scope()` to select `StreamScope::Session`.
    pub scroll_id: Uuid,
    /// Matrix room ID (as a string, e.g. "!abc:example.com").
    pub room_id: String,
    /// Room handle used by the stream `on_event` path when no legacy
    /// forwarder task is running. Behind an `Option` so `start()` — which
    /// consumes the `Room` when spawning the legacy forwarder — can leave
    /// it empty.
    room_for_stream: Option<Room>,
    /// Shared message accumulator so streaming chunks survive across
    /// multiple `on_event` calls (and the legacy forwarder task).
    accumulator: Arc<Mutex<MessageAccumulator>>,
    /// When this share was activated (for `StreamSummary::active_since`).
    active_since: DateTime<Utc>,

    /// Sender side of the Matrix→Dirigent command channel.
    command_tx: mpsc::Sender<ConnectorCommandProxy>,
    /// Shutdown signal for the event-forwarder task.
    shutdown_tx: Arc<RwLock<Option<oneshot::Sender<()>>>>,
    /// Whether the forwarder task is still running.
    is_active: Arc<RwLock<bool>>,
}

impl MatrixSessionShare {
    /// Construct and start a new `MatrixSessionShare`.
    ///
    /// Spawns a background task that reads from `event_rx`, filters for
    /// events belonging to `(connector_id, session_id)`, and forwards
    /// relevant ones into the Matrix `room`.
    ///
    /// Returns the share and the receiver end of the Matrix→Dirigent channel.
    pub fn start(
        connector_id: String,
        session_id: String,
        room_id: String,
        room: Room,
        event_rx: broadcast::Receiver<Event>,
    ) -> (Self, mpsc::Receiver<ConnectorCommandProxy>) {
        // Legacy start() keeps ownership of the Room inside the forwarder
        // task; `scroll_id` isn't known by the legacy call path so we
        // default it to `Uuid::nil()`. The stream-path consumers use
        // `new_for_stream` instead and supply a real scroll_id.
        Self::start_with_scroll(
            connector_id,
            session_id,
            Uuid::nil(),
            room_id,
            room,
            event_rx,
        )
    }

    /// Same as `start`, but lets callers attach the `scroll_id` of the
    /// archived session — required for `SessionStream::scope()` to be
    /// meaningful when the share is also driven as a stream.
    pub fn start_with_scroll(
        connector_id: String,
        session_id: String,
        scroll_id: Uuid,
        room_id: String,
        room: Room,
        event_rx: broadcast::Receiver<Event>,
    ) -> (Self, mpsc::Receiver<ConnectorCommandProxy>) {
        let (command_tx, command_rx) = mpsc::channel(32);
        let (shutdown_tx, shutdown_rx) = oneshot::channel();
        let is_active = Arc::new(RwLock::new(true));
        let accumulator = Arc::new(Mutex::new(MessageAccumulator::new()));

        let share = MatrixSessionShare {
            connector_id: connector_id.clone(),
            session_id: session_id.clone(),
            scroll_id,
            room_id: room_id.clone(),
            // Legacy path: the forwarder task owns the Room, so we don't
            // hold a second handle on the struct. `on_event` would double-
            // drive delivery if both were active.
            room_for_stream: None,
            accumulator: accumulator.clone(),
            active_since: Utc::now(),
            command_tx,
            shutdown_tx: Arc::new(RwLock::new(Some(shutdown_tx))),
            is_active: is_active.clone(),
        };

        // Spawn the event-forwarder task
        tokio::spawn(run_event_forwarder(
            connector_id,
            session_id,
            room,
            event_rx,
            shutdown_rx,
            is_active,
            accumulator,
        ));

        (share, command_rx)
    }

    /// Construct a share wired for the stream path (`SessionStream`) only.
    ///
    /// No legacy event-forwarder task is spawned — the
    /// `StreamRegistry` worker will drive `on_event` instead. The Room
    /// handle is retained on the struct so `on_event` can deliver to it.
    ///
    /// Returns the share plus the receiver end of the Matrix→Dirigent
    /// command channel (identical semantics to `start`).
    pub fn new_for_stream(
        connector_id: String,
        session_id: String,
        scroll_id: Uuid,
        room_id: String,
        room: Room,
    ) -> (Self, mpsc::Receiver<ConnectorCommandProxy>) {
        let (command_tx, command_rx) = mpsc::channel(32);
        let (shutdown_tx, _shutdown_rx) = oneshot::channel();
        let is_active = Arc::new(RwLock::new(true));
        let accumulator = Arc::new(Mutex::new(MessageAccumulator::new()));

        let share = MatrixSessionShare {
            connector_id,
            session_id,
            scroll_id,
            room_id,
            room_for_stream: Some(room),
            accumulator,
            active_since: Utc::now(),
            command_tx,
            shutdown_tx: Arc::new(RwLock::new(Some(shutdown_tx))),
            is_active,
        };

        (share, command_rx)
    }

    /// Inject a message received from Matrix into the Dirigent session.
    ///
    /// Called by `MatrixService` when a room message arrives (filtered to
    /// skip the bot's own messages). Sends a `ConnectorCommandProxy` through
    /// the internal mpsc channel.
    pub async fn inject_message(&self, text: &str) {
        let proxy = ConnectorCommandProxy {
            session_id: self.session_id.clone(),
            text: text.to_owned(),
        };
        if let Err(e) = self.command_tx.send(proxy).await {
            warn!(
                connector_id = %self.connector_id,
                session_id = %self.session_id,
                "Failed to inject Matrix message into session (channel closed): {}",
                e
            );
        }
    }

    /// Signal the event-forwarder task to stop and wait for it to finish.
    pub async fn shutdown(&self) {
        let tx = self.shutdown_tx.write().await.take();
        if let Some(tx) = tx {
            let _ = tx.send(());
        }
        // Give the task a moment to notice the shutdown signal.
        // The is_active flag is set to false by the task itself when it exits.
    }

    /// Whether the event-forwarder task is still running.
    pub async fn is_active(&self) -> bool {
        *self.is_active.read().await
    }
}

// ---------------------------------------------------------------------------
// Internal event-forwarder task
// ---------------------------------------------------------------------------

async fn run_event_forwarder(
    connector_id: String,
    session_id: String,
    room: Room,
    mut event_rx: broadcast::Receiver<Event>,
    shutdown_rx: oneshot::Receiver<()>,
    is_active: Arc<RwLock<bool>>,
    accumulator: Arc<Mutex<MessageAccumulator>>,
) {
    // Fuse the shutdown signal so we can use it inside tokio::select!
    let mut shutdown_rx = shutdown_rx;

    loop {
        tokio::select! {
            _ = &mut shutdown_rx => {
                debug!(
                    connector_id = %connector_id,
                    session_id = %session_id,
                    "MatrixSessionShare forwarder received shutdown signal"
                );
                break;
            }
            result = event_rx.recv() => {
                match result {
                    Err(broadcast::error::RecvError::Closed) => {
                        debug!(
                            connector_id = %connector_id,
                            session_id = %session_id,
                            "Event broadcast channel closed, stopping forwarder"
                        );
                        break;
                    }
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        warn!(
                            connector_id = %connector_id,
                            session_id = %session_id,
                            "Event forwarder lagged by {} messages",
                            n
                        );
                        continue;
                    }
                    Ok(event) => {
                        let mut acc = accumulator.lock().await;
                        handle_event(
                            &event,
                            &connector_id,
                            &session_id,
                            &room,
                            &mut *acc,
                        ).await;
                    }
                }
            }
        }
    }

    *is_active.write().await = false;
}

/// Handle a single protocol event: forward relevant ones to the Matrix room.
async fn handle_event(
    event: &Event,
    connector_id: &str,
    session_id: &str,
    room: &Room,
    accumulator: &mut MessageAccumulator,
) {
    use matrix_sdk::ruma::events::room::message::RoomMessageEventContent;

    match event {
        // -- Streaming accumulation: gather chunks into the accumulator --
        Event::SessionUpdate {
            connector_id: cid,
            session_id: sid,
            update,
        } if cid == connector_id && sid == session_id => {
            // Send typing indicator on first chunk of a new message
            let is_agent_chunk = matches!(
                update,
                SessionUpdate::AgentMessageChunk { .. }
                    | SessionUpdate::AgentThoughtChunk { .. }
                    | SessionUpdate::ToolCall { .. }
            );
            let msg_id = match update {
                SessionUpdate::AgentMessageChunk { message_id, .. }
                | SessionUpdate::AgentThoughtChunk { message_id, .. }
                | SessionUpdate::ToolCall { message_id, .. }
                | SessionUpdate::ToolCallUpdate { message_id, .. } => Some(message_id.as_str()),
                _ => None,
            };
            if is_agent_chunk {
                if let Some(mid) = msg_id {
                    if !accumulator.has_buffer(mid) {
                        // First chunk for this message — start typing indicator
                        let _ = room.typing_notice(true).await;
                    }
                }
            }

            match update {
                SessionUpdate::AgentMessageChunk {
                    message_id,
                    content,
                    ..
                } => {
                    accumulator.add_chunk(message_id, session_id, &connector_id, "assistant", content.clone());
                }
                SessionUpdate::AgentThoughtChunk {
                    message_id,
                    content,
                    ..
                } => {
                    if let ContentBlock::Text { text } = content {
                        accumulator.add_thinking(message_id, session_id, &connector_id, text);
                    }
                }
                SessionUpdate::ToolCall {
                    message_id,
                    tool_call,
                    ..
                } => {
                    accumulator.add_or_update_tool_call(
                        message_id,
                        ToolCallData {
                            id: tool_call.id.clone(),
                            tool_name: tool_call.tool_name.clone(),
                            input: tool_call.raw_input.clone().unwrap_or_default(),
                            output: tool_call.raw_output.clone(),
                        },
                    );
                }
                SessionUpdate::ToolCallUpdate {
                    message_id,
                    tool_call,
                    ..
                } => {
                    accumulator.add_or_update_tool_call(
                        message_id,
                        ToolCallData {
                            id: tool_call.id.clone(),
                            tool_name: tool_call.tool_name.clone(),
                            input: tool_call.raw_input.clone().unwrap_or_default(),
                            output: tool_call.raw_output.clone(),
                        },
                    );
                }
                _ => {} // UserMessageChunk, Unknown -- not forwarded
            }
        }

        // -- Streaming finalization: send accumulated content on TurnComplete --
        Event::TurnComplete {
            connector_id: cid,
            session_id: sid,
            message_id,
            ..
        } if cid == connector_id && sid == session_id => {
            if let Some(accumulated) = accumulator.finalize(message_id) {
                if accumulated.role == "assistant" && !accumulated.is_empty() {
                    send_accumulated_to_matrix(&accumulated, room).await;
                }
            }
            // Stop typing indicator
            let _ = room.typing_notice(false).await;
            debug!(
                connector_id = %connector_id,
                session_id = %session_id,
                message_id = %message_id,
                "TurnComplete received for bridged session"
            );
        }

        // -- Non-streaming fallback: send content from MessageCompleted --
        Event::MessageCompleted {
            connector_id: cid,
            message,
        } if cid == connector_id && message.session_id == session_id => {
            if message.role != MessageRole::Assistant {
                debug!(
                    connector_id = %connector_id,
                    session_id = %session_id,
                    role = ?message.role,
                    "Skipping non-assistant MessageCompleted"
                );
                return;
            }

            // Non-streaming path: content populated directly in MessageCompleted.
            // Skip if the accumulator already has data for this message (streaming
            // path handles delivery via TurnComplete instead).
            if !message.content.is_empty() && !accumulator.has_buffer(&message.id) {
                let accumulated = AccumulatedMessage::from_message_parts(
                    message.id.clone(),
                    message.session_id.clone(),
                    connector_id.to_string(),
                    "assistant".to_string(),
                    &message.content,
                );
                send_accumulated_to_matrix(&accumulated, room).await;
            }
        }

        Event::SessionError {
            connector_id: cid,
            session_id: sid,
            error_message,
            is_recoverable,
            ..
        } if cid == connector_id && sid == session_id => {
            let notice = if *is_recoverable {
                format!("\u{26a0}\u{fe0f} Session warning: {}", error_message)
            } else {
                format!("\u{274c} Session error (unrecoverable): {}", error_message)
            };
            let content = RoomMessageEventContent::notice_plain(notice);
            if let Err(e) = room.send(content).await {
                error!(
                    connector_id = %connector_id,
                    session_id = %session_id,
                    "Failed to send session error to Matrix room: {}",
                    e
                );
            }
        }

        // Events for *this* session but not handled above (expected, low noise)
        Event::SessionIdle { session_id: sid, .. }
        | Event::SessionMetadataUpdated {
            session_id: sid, ..
        }
            if sid == session_id =>
        {
            debug!(
                connector_id = %connector_id,
                session_id = %session_id,
                event = event_name(event),
                "Ignoring non-forwarded event for this session"
            );
        }

        // Events for a *different* session/connector -- expected, just noise
        Event::MessageCompleted {
            connector_id: cid,
            message,
        } if cid != connector_id || message.session_id != session_id => {
            // Different session's message -- expected on a shared broadcast channel
        }
        Event::TurnComplete {
            connector_id: cid,
            session_id: sid,
            ..
        } if cid != connector_id || sid != session_id => {
            // Different session -- expected
        }
        Event::SessionError {
            connector_id: cid,
            session_id: sid,
            ..
        } if cid != connector_id || sid != session_id => {
            // Different session -- expected
        }
        Event::SessionUpdate {
            connector_id: cid,
            session_id: sid,
            ..
        } if cid != connector_id || sid != session_id => {
            // Different session -- expected
        }

        // Connector lifecycle, inspector, system events -- expected on broadcast
        Event::ConnectorCreated { .. }
        | Event::ConnectorRemoved { .. }
        | Event::ConnectorStateChanged { .. }
        | Event::Connected
        | Event::Disconnected
        | Event::InspectorSnapshot { .. }
        | Event::InspectorNodeRegistered { .. }
        | Event::InspectorNodeRemoved { .. }
        | Event::InspectorStateChanged { .. }
        | Event::InspectorPropertiesUpdated { .. }
        | Event::SystemTaskStatusChanged { .. }
        | Event::SessionsListed { .. }
        | Event::SessionCreated { .. }
        | Event::SessionUpdated { .. }
        | Event::SessionDeleted { .. }
        | Event::SessionClosed { .. }
        | Event::MessagesListed { .. }
        | Event::SessionSystemMessageSet { .. }
        | Event::SessionMetadataReceived { .. }
        | Event::SessionTransferred { .. }
        | Event::ForwardingPanic { .. }
        | Event::AgentRequest { .. }
        | Event::AcpClientConnected { .. }
        | Event::AcpClientDisconnected { .. }
        | Event::AcpClientSessionOpened { .. }
        | Event::AcpClientSessionRouted { .. }
        | Event::MessageStarted { .. }
        | Event::MessageFailed { .. }
        | Event::Error { .. }
        | Event::SessionRegistered { .. } => {
            // Expected broadcast traffic, not relevant to this share
        }

        other => {
            warn!(
                connector_id = %connector_id,
                session_id = %session_id,
                event = event_name(other),
                "Unhandled event type in Matrix forwarder"
            );
        }
    }
}

/// Render markdown text to HTML for Matrix consumption.
fn markdown_to_html(markdown: &str) -> String {
    use pulldown_cmark::{Options, Parser};

    let mut options = Options::empty();
    options.insert(Options::ENABLE_STRIKETHROUGH);
    options.insert(Options::ENABLE_TABLES);

    let parser = Parser::new_ext(markdown, options);
    let mut html = String::new();
    pulldown_cmark::html::push_html(&mut html, parser);
    html
}

/// Send an accumulated message to a Matrix room, one message per content part.
async fn send_accumulated_to_matrix(msg: &AccumulatedMessage, room: &Room) {
    use matrix_sdk::ruma::events::room::message::RoomMessageEventContent;

    for part in &msg.parts {
        let content = match part {
            AccumulatedPart::Text { text } if !text.is_empty() => {
                let html = markdown_to_html(text);
                Some(RoomMessageEventContent::text_html(text.clone(), html))
            }
            AccumulatedPart::Thinking { text } if !text.is_empty() => {
                Some(RoomMessageEventContent::notice_plain(format!(
                    "\u{1f4ad} {text}"
                )))
            }
            AccumulatedPart::Tool { data } => {
                let mut notice = format!("\u{1f527} Tool: {}", data.tool_name);
                if let Some(out) = &data.output {
                    let out_str = if let Some(s) = out.as_str() {
                        s.to_string()
                    } else {
                        serde_json::to_string_pretty(out).unwrap_or_default()
                    };
                    if !out_str.is_empty() {
                        let truncated = if out_str.len() > 500 {
                            format!("{}... (truncated)", &out_str[..500])
                        } else {
                            out_str
                        };
                        notice.push_str(&format!("\nOutput: {truncated}"));
                    }
                }
                Some(RoomMessageEventContent::notice_plain(notice))
            }
            _ => None,
        };
        if let Some(content) = content {
            if let Err(e) = room.send(content).await {
                error!("Failed to send message part to Matrix room: {}", e);
            }
        }
    }
}

/// Return a human-readable name for an event variant (for logging).
fn event_name(event: &Event) -> &'static str {
    match event {
        Event::SessionsListed { .. } => "SessionsListed",
        Event::SessionCreated { .. } => "SessionCreated",
        Event::SessionUpdated { .. } => "SessionUpdated",
        Event::SessionMetadataUpdated { .. } => "SessionMetadataUpdated",
        Event::SessionDeleted { .. } => "SessionDeleted",
        Event::SessionClosed { .. } => "SessionClosed",
        Event::SessionSystemMessageSet { .. } => "SessionSystemMessageSet",
        Event::SessionIdle { .. } => "SessionIdle",
        Event::SessionMetadataReceived { .. } => "SessionMetadataReceived",
        Event::TurnComplete { .. } => "TurnComplete",
        Event::SessionError { .. } => "SessionError",
        Event::SessionTransferred { .. } => "SessionTransferred",
        Event::ForwardingPanic { .. } => "ForwardingPanic",
        Event::SessionUpdate { .. } => "SessionUpdate",
        Event::AgentRequest { .. } => "AgentRequest",
        Event::AcpClientConnected { .. } => "AcpClientConnected",
        Event::AcpClientDisconnected { .. } => "AcpClientDisconnected",
        Event::AcpClientSessionOpened { .. } => "AcpClientSessionOpened",
        Event::AcpClientSessionRouted { .. } => "AcpClientSessionRouted",
        Event::MessagesListed { .. } => "MessagesListed",
        Event::MessageStarted { .. } => "MessageStarted",
        Event::MessageCompleted { .. } => "MessageCompleted",
        Event::MessageFailed { .. } => "MessageFailed",
        Event::ConnectorCreated { .. } => "ConnectorCreated",
        Event::ConnectorRemoved { .. } => "ConnectorRemoved",
        Event::ConnectorStateChanged { .. } => "ConnectorStateChanged",
        Event::Connected => "Connected",
        Event::Disconnected => "Disconnected",
        Event::Error { .. } => "Error",
        Event::InspectorSnapshot { .. } => "InspectorSnapshot",
        Event::InspectorNodeRegistered { .. } => "InspectorNodeRegistered",
        Event::InspectorNodeRemoved { .. } => "InspectorNodeRemoved",
        Event::InspectorStateChanged { .. } => "InspectorStateChanged",
        Event::InspectorPropertiesUpdated { .. } => "InspectorPropertiesUpdated",
        Event::SessionRegistered { .. } => "SessionRegistered",
        Event::SystemTaskStatusChanged { .. } => "SystemTaskStatusChanged",
    }
}

// ---------------------------------------------------------------------------
// SessionShare trait implementation
// ---------------------------------------------------------------------------

#[async_trait::async_trait]
impl dirigent_protocol::sharing::SessionShare for MatrixSessionShare {
    fn summary(&self) -> dirigent_protocol::sharing::ShareSummary {
        dirigent_protocol::sharing::ShareSummary {
            id: format!("matrix:{}:{}", self.connector_id, self.session_id),
            connector_id: self.connector_id.clone(),
            session_id: self.session_id.clone(),
            backend: "matrix".to_string(),
            destination: self.room_id.clone(),
            active: self.is_active.try_read().map(|g| *g).unwrap_or(false),
        }
    }

    fn is_active(&self) -> bool {
        self.is_active.try_read().map(|g| *g).unwrap_or(false)
    }

    async fn shutdown(&self) {
        // Delegate to the existing shutdown method (same implementation)
        let tx = self.shutdown_tx.write().await.take();
        if let Some(tx) = tx {
            let _ = tx.send(());
        }
    }
}

// ---------------------------------------------------------------------------
// SessionStream trait implementation (Phase 4 migration, Task 18)
// ---------------------------------------------------------------------------
//
// Dual-impl: `MatrixSessionShare` keeps its bi-directional `SessionShare`
// impl (room management, `inject_message`) while also gaining a
// uni-directional `SessionStream` impl so the `StreamRegistry` can drive it
// via the central `SharingBus`.
//
// When driven as a stream, the caller (factory) is expected to construct
// the share via `new_for_stream(..)` so a Room handle is stored on the
// struct. `on_event` then translates bus events back to the legacy
// `handle_event` dispatcher, preserving the accumulator state across
// calls via the shared `Arc<Mutex<MessageAccumulator>>`.

#[async_trait::async_trait]
impl dirigent_protocol::streaming::SessionStream for MatrixSessionShare {
    fn summary(&self) -> dirigent_protocol::streaming::StreamSummary {
        dirigent_protocol::streaming::StreamSummary {
            name: format!("{}:{}", self.connector_id, self.session_id),
            kind: dirigent_protocol::streaming::StreamKind::Matrix,
            target: format!("matrix:{}", self.room_id),
            active_since: self.active_since,
        }
    }

    fn scope(&self) -> dirigent_protocol::streaming::StreamScope {
        dirigent_protocol::streaming::StreamScope::Session {
            scroll_id: self.scroll_id,
        }
    }

    async fn on_event(
        &self,
        event: &dirigent_protocol::streaming::BusEvent,
    ) -> dirigent_protocol::streaming::StreamOutcome {
        // If we were started via the legacy `start()` path, the Room
        // handle lives inside the forwarder task and this method has no
        // way to deliver. Treat that as a deliberate skip rather than a
        // transport failure.
        let room = match &self.room_for_stream {
            Some(r) => r,
            None => return dirigent_protocol::streaming::StreamOutcome::Skipped,
        };

        let mut acc = self.accumulator.lock().await;
        handle_event(
            &*event.event,
            &self.connector_id,
            &self.session_id,
            room,
            &mut *acc,
        )
        .await;
        dirigent_protocol::streaming::StreamOutcome::Ok
    }

    async fn shutdown(&self) {
        // Delegate to the bi-directional shutdown; both impls share the
        // same underlying oneshot + is_active signal.
        let tx = self.shutdown_tx.write().await.take();
        if let Some(tx) = tx {
            let _ = tx.send(());
        }
        *self.is_active.write().await = false;
    }
}
