use axum::http::StatusCode;

use super::super::AppState;
use super::super::core;
use super::super::guards::DedupGuard;
use super::config::PipelineConfig;

/// Maximum allowed message content length (bytes).
pub(in super::super) const MAX_MESSAGE_BYTES: usize =
    roboticus_core::limits::MAX_USER_MESSAGE_BYTES;

// ── Pipeline request / response types ─────────────────────────────────────

/// Input from a connector to the unified pipeline.
///
/// Connectors do exactly three things: parse input, call `run_pipeline()`,
/// format output. This struct is how they pass raw materials to the factory.
///
/// All business logic (injection defense, session resolution, decomposition,
/// delegation, authority resolution, shortcuts, inference, guards, cost
/// recording, cache, ingestion) is handled by the pipeline, not the caller.
pub(in super::super) struct PipelineRequest<'a> {
    pub state: &'a AppState,
    pub config: PipelineConfig,
    /// Original user message content (pre-sanitization).
    pub raw_content: &'a str,
    /// Caller-provided session ID (for resuming existing sessions).
    pub session_id_hint: Option<&'a str>,
    /// Scope hint for session resolution (e.g., web scope or channel scope).
    /// Computed by the connector from channel-specific context.
    pub scope_hint: Option<roboticus_db::sessions::SessionScope>,
    /// Whether the user is correcting/contradicting the previous reply.
    pub is_correction_turn: bool,
    /// Channel-specific context for authority resolution (ChannelClaim mode only).
    pub channel_context: Option<ChannelClaimContext>,
    /// Multimodal content parts (images, audio transcriptions) created by
    /// `enrich_multimodal()`. Merged into the user `UnifiedMessage.parts`
    /// at context assembly time. `None` for text-only paths.
    pub content_parts: Option<Vec<roboticus_llm::format::ContentPart>>,
}

/// Channel-specific data needed exclusively for authority resolution.
/// Populated by channel connectors; `None` for API/cron paths.
pub(in super::super) struct ChannelClaimContext {
    pub sender_id: String,
    pub chat_id: String,
    pub platform: String,
    pub sender_in_allowlist: bool,
    pub allowlist_configured: bool,
    pub trusted_sender_ids: Vec<String>,
}

/// Result of `run_pipeline()`.
///
/// Each variant maps to a different connector response format:
/// - `Complete` → JSON response (API), channel reply (Telegram), string (cron)
/// - `SpecialistProposal` → interactive prompt for specialist creation
/// - `StreamReady` → SSE stream setup (web streaming)
pub(in super::super) enum PipelineOutcome {
    /// Standard completed inference.
    Complete {
        session_id: String,
        user_message_id: String,
        result: Box<core::PipelineResult>,
    },
    /// Specialist creation proposal — connector must present to user.
    SpecialistProposal { session_id: String, prompt: String },
    /// Streaming: pre-inference work is done, ready for SSE delivery.
    StreamReady(Box<StreamContext>),
}

/// Everything the streaming connector needs to run the SSE loop.
///
/// The pipeline has already performed: injection defense, dedup, session
/// resolution, message storage, turn creation, decomposition, delegation,
/// authority resolution, and shortcut dispatch. The connector only needs to
/// stream inference output and handle post-stream bookkeeping when done.
pub(in super::super) struct StreamContext {
    pub session_id: String,
    pub turn_id: String,
    pub user_content: String,
    pub prepared: core::PreparedInference,
    /// If a shortcut matched, the result is pre-computed — emit as synthetic SSE.
    pub shortcut_result: Option<core::InferenceOutput>,
    /// Dedup guard — ownership transfers to the SSE stream for RAII release.
    pub dedup_guard: DedupGuard,
    pub agent_id: String,
    /// Pre-resolved streaming provider. `None` when a shortcut matched (no provider needed).
    pub resolved_stream: Option<super::super::routing::StreamResolvedContext>,
    /// Guard set to apply during buffer and post-stream phases. Propagated from
    /// `PipelineConfig::guard_set` so the streaming connector honors the
    /// configured preset instead of hardcoding `full()`.
    pub guard_set: super::config::GuardSetPreset,
}

/// Result of post-stream bookkeeping — data the SSE connector needs for
/// the final `stream_end` event.
pub(in super::super) struct StreamFinalization {
    pub assistant_content: String,
    pub content_blocked: bool,
    pub tokens_in: u32,
    pub tokens_out: u32,
    pub model: String,
}

/// Post-stream bookkeeping: sanitize output, persist message/turn, record
/// cost, update breakers/capacity, store cache, ingest into memory.
///
/// Returns data needed for the final SSE events. Called by the streaming
/// connector after the chunk stream ends.
#[allow(clippy::too_many_arguments)]
pub(in super::super) async fn finalize_stream(
    state: &AppState,
    session_id: &str,
    turn_id: &str,
    user_content: &str,
    model: &str,
    cache_hash: &str,
    unified_resp: &roboticus_llm::format::UnifiedResponse,
    provider_prefix: &str,
    cost_in: f64,
    cost_out: f64,
    stream_start: std::time::Instant,
    system_prompt_hash: &str,
) -> StreamFinalization {
    let assistant_content =
        core::sanitize_model_output(unified_resp.content.clone(), state.hmac_secret.as_ref());

    let content_blocked = roboticus_agent::injection::scan_output(&assistant_content);
    let assistant_content = if content_blocked {
        tracing::warn!("L4 output scan flagged streaming response");
        "[Response blocked by output safety filter]".to_string()
    } else {
        assistant_content
    };

    let stream_latency_ms = (stream_start.elapsed().as_millis() as u64).min(i64::MAX as u64) as i64;
    if !assistant_content.trim().is_empty() {
        if let Err(e) = roboticus_db::sessions::append_message(
            &state.db,
            session_id,
            "assistant",
            &assistant_content,
        ) {
            tracing::error!(error = %e, session_id = %session_id, "failed to persist assistant response after streaming");
        }

        let cost =
            unified_resp.tokens_in as f64 * cost_in + unified_resp.tokens_out as f64 * cost_out;
        if let Err(e) = roboticus_db::sessions::create_turn_with_id(
            &state.db,
            turn_id,
            session_id,
            Some(model),
            Some(unified_resp.tokens_in as i64),
            Some(unified_resp.tokens_out as i64),
            Some(cost),
        ) {
            tracing::warn!(error = %e, turn_id = %turn_id, "failed to persist streaming turn");
        }

        core::record_cost(
            state,
            model,
            provider_prefix,
            unified_resp.tokens_in as i64,
            unified_resp.tokens_out as i64,
            cost,
            None,
            false,
            Some(stream_latency_ms),
            None,
            false,
            Some(turn_id),
        );

        {
            let mut llm = state.llm.write().await;
            llm.breakers.record_success(provider_prefix);
            let total_tokens = unified_resp.tokens_in + unified_resp.tokens_out;
            llm.capacity.record(provider_prefix, total_tokens as u64);
            let pressured = llm.capacity.is_sustained_hot(provider_prefix);
            llm.breakers
                .set_capacity_pressure(provider_prefix, pressured);
        }

        core::store_in_cache(
            state,
            cache_hash,
            user_content,
            &assistant_content,
            model,
            unified_resp.tokens_out as i64,
            &[], // streaming path: intents not threaded; acknowledgement exemption handled by guard chain
        )
        .await;

        core::post_turn_ingest(
            state,
            session_id,
            user_content,
            &assistant_content,
            &[],
            system_prompt_hash,
        );
    } else {
        tracing::warn!(session_id = %session_id, "skipping persistence of empty streaming response");
    }

    StreamFinalization {
        assistant_content,
        content_blocked,
        tokens_in: unified_resp.tokens_in,
        tokens_out: unified_resp.tokens_out,
        model: unified_resp.model.clone(),
    }
}

/// Errors from `run_pipeline()`.
///
/// Each connector converts these into channel-appropriate error responses.
pub(crate) enum PipelineError {
    /// Input rejected (empty, invalid). Maps to HTTP 400.
    BadRequest(String),
    /// Input exceeds maximum length. Maps to HTTP 413.
    PayloadTooLarge(String),
    /// Injection blocked. Maps to HTTP 403.
    InjectionBlocked { threat_score: f64 },
    /// Duplicate in-flight request. Maps to HTTP 429.
    DuplicateRequest,
    /// Session not found, wrong agent, or cross-channel access denied.
    SessionError(String, StatusCode),
    /// Internal failure. Maps to HTTP 500.
    Internal(String),
}

impl std::fmt::Display for PipelineError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::BadRequest(msg) => write!(f, "bad request: {msg}"),
            Self::PayloadTooLarge(msg) => write!(f, "payload too large: {msg}"),
            Self::InjectionBlocked { threat_score } => {
                write!(f, "injection blocked (score={threat_score})")
            }
            Self::DuplicateRequest => write!(f, "duplicate request already in flight"),
            Self::SessionError(msg, _) => write!(f, "session error: {msg}"),
            Self::Internal(msg) => write!(f, "internal error: {msg}"),
        }
    }
}

/// Canonical HTTP error mapping for `PipelineError`.
///
/// Serialized as RFC 9457 Problem Details (`application/problem+json`).
/// Used by API-facing connectors (handlers, streaming) via `?` operator,
/// eliminating duplicated match blocks in each connector.
impl axum::response::IntoResponse for PipelineError {
    fn into_response(self) -> axum::response::Response {
        use serde_json::json;
        let (status, detail, extensions) = match self {
            Self::BadRequest(msg) => (StatusCode::BAD_REQUEST, msg, json!({})),
            Self::PayloadTooLarge(msg) => (StatusCode::PAYLOAD_TOO_LARGE, msg, json!({})),
            Self::InjectionBlocked { threat_score } => (
                StatusCode::FORBIDDEN,
                "message_blocked".into(),
                json!({
                    "reason": "prompt injection detected",
                    "threat_score": threat_score,
                }),
            ),
            Self::DuplicateRequest => (
                StatusCode::TOO_MANY_REQUESTS,
                "duplicate_request".into(),
                json!({"reason": "identical request already in flight"}),
            ),
            Self::SessionError(msg, status) => (status, msg, json!({})),
            Self::Internal(msg) => (StatusCode::INTERNAL_SERVER_ERROR, msg, json!({})),
        };
        let mut body = json!({
            "type": "about:blank",
            "title": status.canonical_reason().unwrap_or("Error"),
            "status": status.as_u16(),
            "detail": detail,
        });
        // Merge RFC 9457 extension members (threat_score, reason, etc.)
        if let (Some(base), Some(ext)) = (body.as_object_mut(), extensions.as_object()) {
            for (k, v) in ext {
                base.insert(k.clone(), v.clone());
            }
        }
        (
            status,
            [(axum::http::header::CONTENT_TYPE, "application/problem+json")],
            axum::Json(body),
        )
            .into_response()
    }
}

/// Resolve the web scope hint for API-facing connectors.
///
/// Encapsulates the common pattern: when no `session_id` is provided,
/// resolve a `SessionScope` from the request body + config. Returns `None`
/// when `session_id` is already present (session will be looked up directly).
pub(in super::super) async fn resolve_web_scope_hint(
    state: &AppState,
    body: &super::super::AgentMessageRequest,
) -> Result<Option<roboticus_db::sessions::SessionScope>, PipelineError> {
    if body.session_id.is_some() {
        return Ok(None);
    }
    let config = state.config.read().await;
    match super::super::guards::resolve_web_scope(&config, body) {
        Ok(scope) => Ok(Some(scope)),
        Err(msg) => Err(PipelineError::BadRequest(msg.to_string())),
    }
}
