//! Anthropic Messages API provider (cloud target).

use async_stream::try_stream;
use eventsource_stream::Eventsource;
use futures::StreamExt;
use serde_json::{json, Value};

use crate::provider::{
    ChatRequest, ChatRole, ContentPart, EventStream, LlmError, LlmEvent, LlmProvider, StopReason,
};

/// The Claude Code identity line Anthropic requires as the first system block
/// when authenticating with a subscription OAuth token.
const OAUTH_SYSTEM_PREAMBLE: &str = "You are Claude Code, Anthropic's official CLI for Claude.";

/// Setup tokens minted by `claude setup-token` carry this prefix. API keys are
/// `sk-ant-api…`, so the shape is enough to pick the auth scheme.
fn is_setup_token(credential: &str) -> bool {
    credential.trim().starts_with("sk-ant-oat")
}

pub struct AnthropicProvider {
    client: reqwest::Client,
    base_url: String,
    api_key: String,
}

impl AnthropicProvider {
    pub fn new(api_key: String) -> AnthropicProvider {
        AnthropicProvider::with_base_url(api_key, "https://api.anthropic.com".into())
    }

    /// Base URL override exists for self-hosted proxies and tests.
    pub fn with_base_url(api_key: String, base_url: String) -> AnthropicProvider {
        AnthropicProvider {
            client: reqwest::Client::new(),
            base_url,
            api_key,
        }
    }

    /// Whether this provider is authenticating with a subscription token
    /// rather than an API key. Callers that report cost attribution care.
    pub fn is_subscription(&self) -> bool {
        is_setup_token(&self.api_key)
    }

    /// The Claude Code identity as its OWN system block, with the caller's
    /// prompt as a second block.
    ///
    /// This used to concatenate the two into one string, and **every call that
    /// set a system prompt failed**. On the OAuth path Anthropic requires the
    /// identity to be the first system BLOCK; a single string that merely
    /// begins with it is rejected — and the rejection arrives as
    /// `429 {"type":"rate_limit_error","message":"Error"}`, which reads as
    /// throttling and is not. Measured on one token, seconds apart:
    ///
    /// ```text
    ///   "PREAMBLE"                     (string)  -> 200
    ///   "PREAMBLE\n\nJudge the …"       (string)  -> 429
    ///   ["PREAMBLE"]                   (blocks)  -> 200
    ///   ["PREAMBLE", "Judge the …"]    (blocks)  -> 200
    /// ```
    ///
    /// while the account itself reported `5h utilization 0.07, status allowed`.
    /// Every judge verdict, preflight probe and refiner call on a subscription
    /// token was failing 100% of the time and being logged as "no capacity",
    /// including a `done_when` verdict that failed closed and burned one of a
    /// phase's three passes.
    ///
    /// Idempotent: a caller whose prompt already opens with the identity does
    /// not get it twice.
    fn oauth_system_blocks(system: &str) -> Value {
        let mut blocks = vec![json!({"type": "text", "text": OAUTH_SYSTEM_PREAMBLE})];
        let rest = system
            .trim_start()
            .strip_prefix(OAUTH_SYSTEM_PREAMBLE)
            .unwrap_or(system)
            .trim();
        if !rest.is_empty() {
            blocks.push(json!({"type": "text", "text": rest}));
        }
        Value::Array(blocks)
    }

    fn wire_messages(request: &ChatRequest) -> Vec<Value> {
        request
            .messages
            .iter()
            .map(|message| {
                let role = match message.role {
                    ChatRole::User => "user",
                    ChatRole::Assistant => "assistant",
                };
                let content: Vec<Value> = message
                    .parts
                    .iter()
                    .map(|part| match part {
                        ContentPart::Text { text } => json!({"type": "text", "text": text}),
                        ContentPart::ToolUse { id, name, input } => {
                            json!({"type": "tool_use", "id": id,
                                   "name": crate::wire_tool_name(name), "input": input})
                        }
                        ContentPart::ToolResult {
                            tool_use_id,
                            content,
                        } => json!({
                            "type": "tool_result",
                            "tool_use_id": tool_use_id,
                            "content": content.to_string(),
                        }),
                    })
                    .collect();
                json!({"role": role, "content": content})
            })
            .collect()
    }
}

fn stop_reason(wire: &str) -> StopReason {
    match wire {
        "tool_use" => StopReason::ToolUse,
        "max_tokens" => StopReason::MaxTokens,
        _ => StopReason::EndTurn,
    }
}

#[async_trait::async_trait]
impl LlmProvider for AnthropicProvider {
    async fn stream(&self, request: ChatRequest) -> Result<EventStream, LlmError> {
        let mut tools: Vec<Value> = request
            .tools
            .iter()
            .map(|t| {
                json!({
                    "name": crate::wire_tool_name(&t.name),
                    "description": t.description,
                    "input_schema": t.input_schema,
                })
            })
            .collect();
        if request.web_search {
            // Anthropic server-side web search — the model searches the web itself.
            tools.push(json!({"type": "web_search_20250305", "name": "web_search", "max_uses": 5}));
        }
        // A subscription token authenticates as Claude Code: bearer auth, the
        // Claude Code beta set, and a system prompt whose first line is the
        // Claude Code identity. Sending it as `x-api-key` returns 401.
        let oauth = is_setup_token(&self.api_key);
        // Blocks on the OAuth path, plain string on the API-key path — the
        // API-key path never had this constraint and must keep working.
        let system = if oauth {
            Self::oauth_system_blocks(&request.system)
        } else {
            Value::String(request.system.clone())
        };
        let body = json!({
            "model": request.model,
            "max_tokens": request.max_tokens,
            "system": system,
            "messages": AnthropicProvider::wire_messages(&request),
            "tools": tools,
            "stream": true,
        });

        let mut req = self
            .client
            .post(format!("{}/v1/messages", self.base_url))
            .header("anthropic-version", "2023-06-01");
        req = if oauth {
            req.header("authorization", format!("Bearer {}", self.api_key))
                .header(
                    "anthropic-beta",
                    "claude-code-20250219,oauth-2025-04-20,interleaved-thinking-2025-05-14",
                )
        } else {
            req.header("x-api-key", &self.api_key)
        };
        let response = req
            .json(&body)
            .send()
            .await
            .map_err(|e| LlmError::Transport(e.to_string()))?;
        if !response.status().is_success() {
            let status = response.status();
            let detail = response.text().await.unwrap_or_default();
            return Err(LlmError::Api(format!("{status}: {detail}")));
        }

        let mut sse = response.bytes_stream().eventsource();
        let stream = try_stream! {
            // tool_use input arrives as accumulated partial JSON between
            // content_block_start and content_block_stop.
            let mut pending_tool: Option<(String, String, String)> = None; // id, name, json
            let mut input_tokens: u32 = 0;
            while let Some(event) = sse.next().await {
                let event = event.map_err(|e| LlmError::Transport(e.to_string()))?;
                let data: Value = serde_json::from_str(&event.data)
                    .map_err(|e| LlmError::Wire(format!("{e}: {}", event.data)))?;
                match data["type"].as_str().unwrap_or_default() {
                    "content_block_start" => {
                        let block = &data["content_block"];
                        if block["type"] == "tool_use" {
                            pending_tool = Some((
                                block["id"].as_str().unwrap_or_default().to_owned(),
                                crate::internal_tool_name(
                                    block["name"].as_str().unwrap_or_default(),
                                ),
                                String::new(),
                            ));
                        }
                    }
                    "content_block_delta" => {
                        let delta = &data["delta"];
                        match delta["type"].as_str().unwrap_or_default() {
                            "text_delta" => {
                                if let Some(text) = delta["text"].as_str() {
                                    yield LlmEvent::TextDelta(text.to_owned());
                                }
                            }
                            "input_json_delta" => {
                                if let Some((_, _, buf)) = pending_tool.as_mut() {
                                    buf.push_str(delta["partial_json"].as_str().unwrap_or_default());
                                }
                            }
                            _ => {}
                        }
                    }
                    "content_block_stop" => {
                        if let Some((id, name, buf)) = pending_tool.take() {
                            let input: Value = if buf.is_empty() {
                                json!({})
                            } else {
                                serde_json::from_str(&buf)
                                    .map_err(|e| LlmError::Wire(format!("tool input: {e}")))?
                            };
                            yield LlmEvent::ToolUse { id, name, input };
                        }
                    }
                    "message_start" => {
                        input_tokens =
                            data["message"]["usage"]["input_tokens"].as_u64().unwrap_or(0) as u32;
                    }
                    "message_delta" => {
                        // Anthropic reports input_tokens in `message_start`;
                        // z.ai's Anthropic-compatible endpoint sends 0 there and
                        // the real count here, beside output_tokens (measured
                        // 2026-09-14: start `input_tokens: 0`, delta
                        // `input_tokens: 14`). Prefer the delta's figure when
                        // it carries one — the judge's input, which is the
                        // number that empties a plan, read as zero until then.
                        input_tokens = input_tokens_after_delta(input_tokens, &data["usage"]);
                        if let Some(out) = data["usage"]["output_tokens"].as_u64() {
                            yield LlmEvent::Usage {
                                input_tokens,
                                output_tokens: out as u32,
                            };
                        }
                        if let Some(reason) = data["delta"]["stop_reason"].as_str() {
                            yield LlmEvent::Stop(stop_reason(reason));
                        }
                    }
                    "error" => {
                        Err(LlmError::Api(data["error"]["message"]
                            .as_str()
                            .unwrap_or("unknown")
                            .to_owned()))?;
                    }
                    _ => {}
                }
            }
        };
        Ok(Box::pin(stream))
    }
}

/// The input-token figure to report once a `message_delta` has arrived.
///
/// `start` is what `message_start` said. Anthropic puts the real count there
/// and nothing in the delta; z.ai's Anthropic-compatible endpoint puts 0 there
/// and the real count in the delta's `usage` (measured 2026-09-14). A nonzero
/// delta figure wins; anything else keeps what `message_start` said, so the
/// Anthropic path is unchanged.
fn input_tokens_after_delta(start: u32, delta_usage: &serde_json::Value) -> u32 {
    match delta_usage["input_tokens"].as_u64() {
        Some(n) if n > 0 => n as u32,
        _ => start,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// z.ai reports input in the delta; Anthropic reports it at the start.
    /// Before this the judge's input tokens — the number that empties a
    /// plan — read as zero on every GLM verdict.
    #[test]
    fn input_tokens_come_from_whichever_frame_carries_them() {
        use serde_json::json;
        // z.ai shape: start says 0, delta says 14.
        assert_eq!(input_tokens_after_delta(0, &json!({"input_tokens": 14, "output_tokens": 16})), 14);
        // Anthropic shape: start said 812, delta has no input figure.
        assert_eq!(input_tokens_after_delta(812, &json!({"output_tokens": 40})), 812);
        // A delta that explicitly says 0 must not erase the start's figure.
        assert_eq!(input_tokens_after_delta(812, &json!({"input_tokens": 0, "output_tokens": 40})), 812);
    }

    #[test]
    fn setup_tokens_are_distinguished_from_api_keys() {
        assert!(is_setup_token("sk-ant-oat01-abc"));
        assert!(is_setup_token("  sk-ant-oat01-abc  "), "trims first");
        assert!(!is_setup_token("sk-ant-api03-abc"));
        assert!(!is_setup_token(""));
    }

    /// The identity must be its own FIRST block, and the caller's prompt a
    /// SECOND one. Concatenating them into a single string is what made every
    /// system-prompt-setting call return `429 rate_limit_error` on a token
    /// whose account was at 7% utilization.
    #[test]
    fn the_oauth_system_is_blocks_with_the_identity_first() {
        let v = AnthropicProvider::oauth_system_blocks("Judge the condition.");
        let blocks = v.as_array().expect("system must be an ARRAY, not a string");
        assert_eq!(blocks.len(), 2);
        assert_eq!(blocks[0]["type"], "text");
        assert_eq!(blocks[0]["text"], OAUTH_SYSTEM_PREAMBLE);
        assert_eq!(blocks[1]["text"], "Judge the condition.");
    }

    /// Idempotent: a caller that already opens with the identity must not send
    /// it twice — duplicate identity is token waste on every single call.
    #[test]
    fn the_identity_is_never_duplicated() {
        let already = format!("{OAUTH_SYSTEM_PREAMBLE}\n\nJudge the condition.");
        let v = AnthropicProvider::oauth_system_blocks(&already);
        let blocks = v.as_array().unwrap();
        assert_eq!(blocks.len(), 2, "{blocks:?}");
        assert_eq!(blocks[0]["text"], OAUTH_SYSTEM_PREAMBLE);
        assert_eq!(blocks[1]["text"], "Judge the condition.");
        let text = serde_json::to_string(&v).unwrap();
        assert_eq!(text.matches(OAUTH_SYSTEM_PREAMBLE).count(), 1);
    }

    /// An empty caller prompt yields the identity ALONE — never a trailing
    /// empty block, which the API rejects.
    #[test]
    fn an_empty_system_prompt_yields_the_identity_alone() {
        for empty in ["", "   ", OAUTH_SYSTEM_PREAMBLE] {
            let v = AnthropicProvider::oauth_system_blocks(empty);
            let blocks = v.as_array().unwrap();
            assert_eq!(blocks.len(), 1, "{empty:?} -> {blocks:?}");
            assert_eq!(blocks[0]["text"], OAUTH_SYSTEM_PREAMBLE);
        }
    }

    #[test]
    fn is_subscription_reports_the_credential_kind() {
        assert!(AnthropicProvider::new("sk-ant-oat01-x".into()).is_subscription());
        assert!(!AnthropicProvider::new("sk-ant-api03-x".into()).is_subscription());
    }
}