pi_logger 0.8.0

use derivative::Derivative;
use io::ErrorKind;
use log::Record;
use crossbeam_channel as crossbeam;

use log4rs::{
    append::Append,
    config::{Deserialize, Deserializers},
    encode::{
        self, pattern::PatternEncoder, writer::simple::SimpleWriter, Encode, EncoderConfig, Style,
    },
};

use async_httpc::{
    AsyncHttpRequest, AsyncHttpRequestBody, AsyncHttpRequestMethod, AsyncHttpc, AsyncHttpcBuilder,
};

use pi_async_rt::rt::{
    multi_thread::{MultiTaskRuntime, MultiTaskRuntimeBuilder, StealableTaskPool},
    single_thread::{SingleTaskRunner, SingleTaskRuntime},
    spawn_worker_thread, AsyncRuntime,
};

use lazy_static::lazy_static;
use std::sync::RwLock;
use std::{
    fmt::{self, Debug},
    fs::{self, File, OpenOptions},
    io::{self, BufWriter, Error, Write},
    path::{Path, PathBuf},
    sync::Arc,
    thread,
    time::{Duration, Instant},
};

/// An appender which logs to a http.
#[derive(Derivative)]
#[derivative(Debug)]
pub struct SLSAppender {
    write: BatchHttpWrite,
    encoder: Box<dyn Encode>,
}

impl Append for SLSAppender {
    fn append(&self, record: &Record) -> anyhow::Result<()> {
        let mut write = self.write.clone();
        self.encoder.encode(&mut write, record)?;
        write.flush()?;
        Ok(())
    }

    fn flush(&self) {}
}

impl SLSAppender {
    /// Creates a new `SLSAppender` builder.
    pub fn builder() -> SLSAppenderBuilder {
        SLSAppenderBuilder {
            encoder: None,
            append: true,
            batch_config: None,
            source: None,
        }
    }
}

/// A builder for `SLSAppender`s.
pub struct SLSAppenderBuilder {
    encoder: Option<Box<dyn Encode>>,
    append: bool,
    batch_config: Option<SLBatchConfig>,
    source: Option<String>,
}

impl SLSAppenderBuilder {
    /// Sets the output encoder for the `SLSAppender`.
    pub fn encoder(mut self, encoder: Box<dyn Encode>) -> SLSAppenderBuilder {
        self.encoder = Some(encoder);
        self
    }

    /// Determines if the appender will append to or truncate the output file.
    ///
    /// Defaults to `true`.
    pub fn append(mut self, append: bool) -> SLSAppenderBuilder {
        self.append = append;
        self
    }

    /// Sets the batch upload configuration for the `SLSAppender`.
    pub fn batch_config(mut self, config: SLBatchConfig) -> SLSAppenderBuilder {
        self.batch_config = Some(config);
        self
    }

    /// Sets the source identifier for the `SLSAppender`.
    pub fn source(mut self, source: String) -> SLSAppenderBuilder {
        self.source = Some(source);
        self
    }

    /// Consumes the `SLSAppenderBuilder`, producing a `SLSAppender`.
    /// The path argument can contain environment variables of the form $ENV{name_here},
    /// where 'name_here' will be the name of the environment variable that
    /// will be resolved. Note that if the variable fails to resolve,
    /// $ENV{name_here} will NOT be replaced in the path.
    pub fn build(
        self,
        url: String,
        rt: MultiTaskRuntime<()>,
        httpc: AsyncHttpc,
    ) -> io::Result<SLSAppender> {
        let config = self.batch_config.unwrap_or_default();
        let source = self.source.unwrap_or_else(|| "pi_logger".to_string());
        let batch_write = BatchHttpWrite::new(url, rt, httpc, config, source);

        Ok(SLSAppender {
            write: batch_write,
            encoder: self
                .encoder
                .unwrap_or_else(|| Box::new(PatternEncoder::default())),
        })
    }
}

#[derive(Clone)]
struct HttpWrite {
    buf: Vec<u8>,
    rt: MultiTaskRuntime<()>,
    url: String,
    httpc: AsyncHttpc,
}

impl Debug for HttpWrite {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        Ok(())
    }
}

impl io::Write for HttpWrite {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.buf.write_all(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        let httpc = self.httpc.clone();
        let buf = self.buf.clone();
        let url = self.url.clone();
        let rt = self.rt.clone();
        self.rt.spawn(async move {
            http_request(httpc, url, buf, rt).await;
        });
        self.buf.clear();
        Ok(())
    }
}

impl encode::Write for HttpWrite {
    fn set_style(&mut self, style: &Style) -> io::Result<()> {
        // self.0.set_style(style)
        Ok(())
    }
}

/// 发送日志到阿里云日志服务
pub async fn http_request(
    httpc: AsyncHttpc,
    url: String,
    body: Vec<u8>,
    rt: MultiTaskRuntime<()>,
) -> io::Result<Vec<u8>> {
    let body = AsyncHttpRequestBody::with_binary(body);
    let httpc_copy = httpc.clone();

    let mut resp = httpc_copy
        .build_request(&url, AsyncHttpRequestMethod::Post)
        .add_header("Content-Type", "application/json")
        .add_header("x-log-apiversion", "0.6.0")
        .add_header("x-log-bodyrawsize", "0")
        .set_body(body)
        .send()
        .await?;
    let mut bodyVec: Vec<u8> = Vec::new();
    loop {
        match resp.get_body().await? {
            Some(body) => {
                bodyVec.write_all(&*body);
            }
            None => {
                return Ok(bodyVec);
            }
        }
    }
}

// ============================================================================
// 批量上传相关代码
// ============================================================================

/// SLS 批量上传配置
#[derive(Clone, Debug)]
pub struct SLBatchConfig {
    /// 触发发送的字节数阈值 (默认: 2MB, SLS 限制 10MB)
    pub batch_bytes: usize,

    /// 时间窗口超时(秒) (默认: 1)
    pub timeout_secs: u64,

    /// 最大重试次数 (默认: 3)
    pub max_retries: usize,

    /// 重试延迟(毫秒) (默认: 1000)
    pub retry_delay_ms: u64,
}

impl Default for SLBatchConfig {
    fn default() -> Self {
        Self {
            batch_bytes: 2 * 1024 * 1024,  // 2MB
            timeout_secs: 1,
            max_retries: 3,
            retry_delay_ms: 1000,
        }
    }
}

/// 批量 HTTP 写入器
#[derive(Clone)]
struct BatchHttpWrite {
    /// 发送日志的通道
    sender: crossbeam::Sender<String>,
    /// 配置
    config: SLBatchConfig,
    /// 每次 append 都会 clone，所以每个实例独立拥有自己的 buffer
    buffer: Vec<u8>,
    /// 日志来源标识
    source: String,
}

impl BatchHttpWrite {
    fn new(
        url: String,
        rt: MultiTaskRuntime<()>,
        httpc: AsyncHttpc,
        config: SLBatchConfig,
        source: String,
    ) -> Self {
        // 创建无界通道
        let (sender, receiver) = crossbeam::unbounded();

        // 启动后台批量处理任务
        start_batch_processor(receiver, url, rt, httpc, config.clone(), source.clone());

        Self {
            sender,
            config,
            buffer: Vec::new(),
            source,
        }
    }
}

impl Debug for BatchHttpWrite {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        Ok(())
    }
}

impl io::Write for BatchHttpWrite {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        // encoder.encode() 会分多次写入，先累积到 buffer
        // 在 flush 时一次性发送完整的日志条目
        self.buffer.extend_from_slice(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        // 将 buffer 中的完整日志条目发送到 channel
        if !self.buffer.is_empty() {
            let data = String::from_utf8_lossy(&self.buffer).to_string();
            self.sender.send(data)
                .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
            self.buffer.clear();
        }
        Ok(())
    }
}

impl encode::Write for BatchHttpWrite {
    fn set_style(&mut self, _style: &Style) -> io::Result<()> {
        Ok(())
    }
}

/// 启动后台批量处理任务
fn start_batch_processor(
    receiver: crossbeam::Receiver<String>,
    url: String,
    rt: MultiTaskRuntime<()>,
    httpc: AsyncHttpc,
    config: SLBatchConfig,
    source: String,
) {
    rt.clone().spawn(async move {
        let mut buffer = Vec::with_capacity(128);
        let mut last_flush = Instant::now();
        let mut current_bytes = 0;

        loop {
            // 检查超时条件
            if !buffer.is_empty() &&
               last_flush.elapsed() >= Duration::from_secs(config.timeout_secs) {
                send_batch_with_retry(&buffer, &httpc, &url, &rt, &config, &source).await;
                buffer.clear();
                current_bytes = 0;
                last_flush = Instant::now();
            }

            // 使用 try_iter 非阻塞接收所有待处理日志
            let mut received_any = false;
            for log_entry in receiver.try_iter() {
                received_any = true;
                let entry_bytes = log_entry.len();
                current_bytes += entry_bytes;
                buffer.push(log_entry);

                // 检查字节数条件
                if current_bytes >= config.batch_bytes {
                    send_batch_with_retry(&buffer, &httpc, &url, &rt, &config, &source).await;
                    buffer.clear();
                    current_bytes = 0;
                    last_flush = Instant::now();
                }
            }

            // 如果没有收到任何日志且缓冲区为空，短暂休眠避免忙等待
            if !received_any && buffer.is_empty() {
                rt.timeout(100).await;
            } else if !received_any {
                // 有数据但未达到发送条件，短暂休眠后继续检查
                rt.timeout(10).await;
            }
        }
    });
}

/// 发送批量日志（带重试）
async fn send_batch_with_retry(
    logs: &[String],
    httpc: &AsyncHttpc,
    url: &str,
    rt: &MultiTaskRuntime<()>,
    config: &SLBatchConfig,
    source: &str,
) {
    let payload = build_batch_payload(logs, source);

    for retry in 0..=config.max_retries {
        match http_send_batch(httpc, url, &payload).await {
            Ok(resp_body) => {
                // 检查响应内容，SLS 可能返回 200 但内容包含错误
                if let Ok(resp_str) = String::from_utf8(resp_body) {
                    if resp_str.contains("\"errorCode\"") || resp_str.contains("\"error\"") {
                        eprintln!("[SLS Appender] Warning: SLS returned error response: {}", resp_str);
                    }
                }
                return; // 成功，直接返回
            }
            Err(e) if retry < config.max_retries => {
                eprintln!("[SLS Appender] Warning: Send failed (attempt {}/{}), error: {:?}. Retrying in {}ms...",
                    retry + 1, config.max_retries + 1, e, config.retry_delay_ms);
                rt.timeout(config.retry_delay_ms as usize).await;
            }
            Err(e) => {
                eprintln!("[SLS Appender] Error: Failed to send logs after {} retries. Final error: {:?}",
                    config.max_retries + 1, e);
                eprintln!("[SLS Appender] Failed to send {} log entries to: {}", logs.len(), url);
                return;
            }
        }
    }
}

/// 将 JSON Value 中所有值转换为字符串（SLS PutWebTracking 要求）
/// 注意：SLS PutWebTracking 只支持一层结构，所有值必须是字符串
/// 嵌套对象会被递归处理，保持扁平结构
fn convert_values_to_strings(value: &serde_json::Value) -> serde_json::Value {
    match value {
        serde_json::Value::String(s) => serde_json::Value::String(s.clone()),
        serde_json::Value::Number(n) => serde_json::Value::String(n.to_string()),
        serde_json::Value::Bool(b) => serde_json::Value::String(b.to_string()),
        serde_json::Value::Null => serde_json::Value::String("null".to_string()),
        // 数组：递归处理每个元素，如果所有元素都是基本类型则保持数组，否则转为字符串
        serde_json::Value::Array(arr) => {
            let converted: Vec<serde_json::Value> = arr.iter().map(convert_values_to_strings).collect();
            // 检查是否所有元素都是字符串
            let all_strings = converted.iter().all(|v| matches!(v, serde_json::Value::String(_)));
            if all_strings {
                serde_json::Value::Array(converted)
            } else {
                // 还有嵌套结构，整个数组转为 JSON 字符串
                serde_json::Value::String(serde_json::to_string(&converted).unwrap_or_default())
            }
        }
        // 对象：递归处理每个值
        serde_json::Value::Object(obj) => {
            let mut new_obj = serde_json::Map::new();
            for (k, v) in obj.iter() {
                let converted = convert_values_to_strings(v);
                // 如果转换后的值不是字符串，再将其序列化为 JSON 字符串
                let final_value = match &converted {
                    serde_json::Value::String(_) => converted,
                    serde_json::Value::Array(arr) => {
                        // 数组中如果都是字符串，保持数组；否则转为字符串
                        let all_strings = arr.iter().all(|v| matches!(v, serde_json::Value::String(_)));
                        if all_strings {
                            converted
                        } else {
                            serde_json::Value::String(serde_json::to_string(&converted).unwrap_or_default())
                        }
                    }
                    serde_json::Value::Object(_) => {
                        // 嵌套对象转为 JSON 字符串
                        serde_json::Value::String(serde_json::to_string(&converted).unwrap_or_default())
                    }
                    _ => serde_json::Value::String(serde_json::to_string(&converted).unwrap_or_default()),
                };
                new_obj.insert(k.clone(), final_value);
            }
            serde_json::Value::Object(new_obj)
        }
    }
}

/// 构建符合 SLS PutWebTracking 格式的批量 payload
fn build_batch_payload(logs: &[String], source: &str) -> String {
    let mut log_entries = Vec::new();
    let mut skipped = 0;

    for entry in logs.iter() {
        // 尝试直接解析为单个 JSON 对象
        if let Ok(v) = serde_json::from_str::<serde_json::Value>(entry) {
            // 将所有值转换为字符串
            let converted = convert_values_to_strings(&v);
            log_entries.push(converted);
            continue;
        }

        // 如果失败，可能是多条 JSON 对象用换行符分隔的情况（来自 __logs__ 数组）
        // 尝试按行分割，分别解析每行
        let mut has_valid_entry = false;
        for line in entry.lines() {
            let trimmed = line.trim();
            if trimmed.is_empty() {
                continue;
            }
            if let Ok(v) = serde_json::from_str::<serde_json::Value>(trimmed) {
                // 将所有值转换为字符串
                let converted = convert_values_to_strings(&v);
                log_entries.push(converted);
                has_valid_entry = true;
            } else {
                skipped += 1;
                eprintln!("[SLS Appender] Warning: Failed to parse log line as JSON: {}", trimmed);
            }
        }

        // 如果整条 entry 都没有有效记录，才计入跳过
        if !has_valid_entry {
            skipped += 1;
        }
    }

    if skipped > 0 {
        eprintln!("[SLS Appender] Warning: Skipped {} invalid log entries", skipped);
    }

    serde_json::json!({
        "__source__": source,
        "__logs__": log_entries
    }).to_string()
}

/// HTTP 发送批量日志
async fn http_send_batch(
    httpc: &AsyncHttpc,
    url: &str,
    body: &str,
) -> io::Result<Vec<u8>> {
    let body_bytes = body.as_bytes();
    let body_req = AsyncHttpRequestBody::with_binary(body_bytes.to_vec());

    let mut resp = httpc
        .build_request(url, AsyncHttpRequestMethod::Post)
        .add_header("Content-Type", "application/json")
        .add_header("x-log-apiversion", "0.6.0")
        .add_header("x-log-bodyrawsize", &body_bytes.len().to_string())
        .set_body(body_req)
        .send()
        .await?;

    let mut body_vec = Vec::new();
    loop {
        match resp.get_body().await? {
            Some(b) => {
                body_vec.write_all(&*b)?;
            }
            None => {
                return Ok(body_vec);
            }
        }
    }
}