use super::*;

pub(super) const MAX_RETAINED_TASKS: usize = 64;
/// 每个任务保留的进度日志行数上限。
pub(super) const MAX_TASK_LOG_LINES: usize = 200;

/// 任务类型：目前覆盖官方同步、校验与 catalog 更新检查。
#[derive(Debug, Clone, Copy)]
pub(super) enum TaskKind {
    Sync,
    Verify,
    Repair,
    /// catalog 更新检查：只做发现 + 拉取计划（dry-run），不下载不审计。
    Refresh,
}

impl TaskKind {
    pub(super) fn method(self) -> &'static str {
        match self {
            Self::Sync => RPC_METHOD_RESOURCE_SYNC,
            Self::Verify => RPC_METHOD_RESOURCE_VERIFY,
            Self::Repair => RPC_METHOD_RESOURCE_REPAIR,
            Self::Refresh => RPC_METHOD_CATALOG_REFRESH,
        }
    }

    /// 由 daemon 基准配置派生该任务的实际同步配置。
    pub(super) fn build_config(
        self,
        base: &OfficialUpdateConfig,
        force: bool,
    ) -> OfficialUpdateConfig {
        let mut config = base.clone();
        match self {
            Self::Sync => {
                config.dry_run = false;
                config.force = config.force || force;
            }
            Self::Verify => {
                config.dry_run = true;
                config.plan = true;
                config.audit_local = true;
                config.repair = false;
                config.force = false;
            }
            Self::Repair => {
                config.dry_run = false;
                config.audit_local = true;
                config.repair = true;
                config.force = false;
            }
            Self::Refresh => {
                config.dry_run = true;
                config.plan = true;
                config.audit_local = false;
                config.repair = false;
                config.force = force;
            }
        }
        config
    }
}

/// 请求取消任务的结果。
pub(super) enum CancelOutcome {
    Requested,
    AlreadyFinished,
    NotFound,
}

/// 单个任务的可轮询记录。
#[derive(Debug, Clone, Serialize)]
pub(super) struct TaskRecord {
    pub(super) id: String,
    pub(super) kind: &'static str,
    /// `queued` | `running` | `succeeded` | `failed` | `cancelled`。
    pub(super) status: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) stage: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) message: Option<String>,
    pub(super) created_at: u64,
    pub(super) updated_at: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) started_at: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) finished_at: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) error: Option<ApiError>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(super) result: Option<serde_json::Value>,
    /// 取消标志，worker 的 should_cancel 检查它；不参与序列化。
    #[serde(skip)]
    pub(super) cancel: Arc<AtomicBool>,
    /// 进度日志（有界），经 task.logs 返回；不参与 task.status 序列化。
    #[serde(skip)]
    pub(super) log: Vec<String>,
}

impl TaskRecord {
    pub(super) fn is_finished(&self) -> bool {
        matches!(self.status, "succeeded" | "failed" | "cancelled")
    }
}

struct TaskStore {
    tasks: HashMap<String, TaskRecord>,
    order: Vec<String>,
    seq: u64,
    /// 任务历史持久化文件路径；`None` 表示纯内存（测试等非 daemon 场景）。
    persist_path: Option<PathBuf>,
}

/// daemon 任务历史持久化文件名（位于 state dir 内，`0600` 原子写）。
pub(super) const TASKS_FILE_NAME: &str = "bat-tasks.json";
/// 任务历史文件结构版本。
pub(super) const TASKS_FILE_VERSION: u32 = 1;

/// 任务历史文件的持久化形态（版本化；daemon 重启后恢复任务历史用）。
#[derive(Debug, Serialize, Deserialize)]
pub(super) struct PersistedTaskFile {
    pub(super) version: u32,
    /// 任务 ID 序号计数器；恢复它避免 pid 复用时新任务与历史任务撞 ID。
    pub(super) seq: u64,
    pub(super) tasks: Vec<PersistedTaskRecord>,
}

#[derive(Debug, Serialize, Deserialize)]
pub(super) struct PersistedTaskRecord {
    id: String,
    kind: String,
    pub(super) status: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    stage: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    message: Option<String>,
    created_at: u64,
    updated_at: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    started_at: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    finished_at: Option<u64>,
    /// `ApiError` 的序列化形态（code/kind/domain/location/message/retryable）。
    #[serde(default, skip_serializing_if = "Option::is_none")]
    error: Option<serde_json::Value>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    result: Option<serde_json::Value>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    log: Vec<String>,
}

/// 把持久化的任务类型映射回静态字符串；未识别（如未来版本新增）返回 `None`。
fn task_kind_static(kind: &str) -> Option<&'static str> {
    match kind {
        RPC_METHOD_RESOURCE_SYNC => Some(RPC_METHOD_RESOURCE_SYNC),
        RPC_METHOD_RESOURCE_VERIFY => Some(RPC_METHOD_RESOURCE_VERIFY),
        RPC_METHOD_RESOURCE_REPAIR => Some(RPC_METHOD_RESOURCE_REPAIR),
        RPC_METHOD_CATALOG_REFRESH => Some(RPC_METHOD_CATALOG_REFRESH),
        _ => None,
    }
}

/// 把持久化的任务状态映射回静态字符串；未识别返回 `None`。
fn task_status_static(status: &str) -> Option<&'static str> {
    match status {
        "queued" => Some("queued"),
        "running" => Some("running"),
        "succeeded" => Some("succeeded"),
        "failed" => Some("failed"),
        "cancelled" => Some("cancelled"),
        _ => None,
    }
}

impl PersistedTaskRecord {
    fn from_record(record: &TaskRecord) -> Self {
        Self {
            id: record.id.clone(),
            kind: record.kind.to_string(),
            status: record.status.to_string(),
            stage: record.stage.clone(),
            message: record.message.clone(),
            created_at: record.created_at,
            updated_at: record.updated_at,
            started_at: record.started_at,
            finished_at: record.finished_at,
            error: record
                .error
                .as_ref()
                .and_then(|error| serde_json::to_value(error).ok()),
            result: record.result.clone(),
            log: record.log.clone(),
        }
    }

    /// 还原为内存任务记录；kind/status 未识别时返回 `None`（调用方计数跳过）。
    fn into_record(self) -> Option<TaskRecord> {
        let kind = task_kind_static(&self.kind)?;
        let status = task_status_static(&self.status)?;
        // 错误从序列化形态还原：code 经码表反查（未登记回退 internal），
        // location 固定为任务执行器（当前全部任务错误的唯一来源）。
        let error = self.error.as_ref().map(|value| {
            let code = value
                .get("code")
                .and_then(serde_json::Value::as_str)
                .and_then(ErrorCode::from_id)
                .unwrap_or(ErrorCode::INTERNAL);
            let message = value
                .get("message")
                .and_then(serde_json::Value::as_str)
                .unwrap_or("<持久化错误信息缺失>")
                .to_string();
            ApiError::new(code, "task.executor", message)
        });
        Some(TaskRecord {
            id: self.id,
            kind,
            status,
            stage: self.stage,
            message: self.message,
            created_at: self.created_at,
            updated_at: self.updated_at,
            started_at: self.started_at,
            finished_at: self.finished_at,
            error,
            result: self.result,
            cancel: Arc::new(AtomicBool::new(false)),
            log: self.log,
        })
    }
}

/// 读取任务历史文件。文件缺失返回 `Ok(None)`；symlink、解析失败或版本不支持返回 `Err`。
fn load_persisted_tasks(path: &Path) -> Result<Option<PersistedTaskFile>, String> {
    let Some(bytes) = read_file_no_symlink(path, "任务历史")? else {
        return Ok(None);
    };
    let file: PersistedTaskFile = serde_json::from_slice(&bytes)
        .map_err(|error| format!("解析任务历史失败 {}：{error}", path.display()))?;
    if file.version != TASKS_FILE_VERSION {
        return Err(format!(
            "不支持的任务历史版本 {}，文件 {}",
            file.version,
            path.display()
        ));
    }
    Ok(Some(file))
}

/// 任务注册表句柄：包住内存存储，供 RPC handler 与 worker 共享。
///
/// 通过方法访问（而非直接摸内部 map），便于将来换成 Redis 等持久化后端。
#[derive(Clone)]
pub(super) struct TaskRegistry {
    inner: Arc<Mutex<TaskStore>>,
}

impl TaskRegistry {
    /// 纯内存注册表（无持久化）；生产 daemon 走 [`Self::with_persistence`]。
    #[cfg(test)]
    pub(super) fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(TaskStore {
                tasks: HashMap::new(),
                order: Vec::new(),
                seq: 0,
                persist_path: None,
            })),
        }
    }

    /// 从 state dir 恢复任务历史并启用持久化。
    ///
    /// 中断时仍处于 queued/running 的任务标记为 `failed`（`TASK_INTERRUPTED`）；
    /// 文件缺失按空历史处理；文件损坏或版本不支持时改名 `.corrupt` 留证并从
    /// 空历史开始。返回注册表与恢复摘要（供 daemon 日志记录）。
    pub(super) fn with_persistence(state_dir: &Path) -> (Self, String) {
        let path = state_dir.join(TASKS_FILE_NAME);
        let now = unix_seconds_now();
        let mut seq = 0;
        let mut tasks = HashMap::new();
        let mut order = Vec::new();
        let summary = match load_persisted_tasks(&path) {
            Ok(None) => "无历史任务文件，从空任务历史开始".to_string(),
            Ok(Some(file)) => {
                seq = file.seq;
                let total = file.tasks.len();
                let mut interrupted = 0usize;
                let mut skipped = 0usize;
                for persisted in file.tasks {
                    let Some(mut record) = persisted.into_record() else {
                        skipped += 1;
                        continue;
                    };
                    if !record.is_finished() {
                        interrupted += 1;
                        record.status = "failed";
                        record.finished_at = Some(now);
                        record.updated_at = now;
                        record.error = Some(ApiError::new(
                            ErrorCode::TASK_INTERRUPTED,
                            "task.executor",
                            "daemon 停止/重启导致任务中断",
                        ));
                        record
                            .log
                            .push("[daemon] 任务因 daemon 停止/重启而中断".to_string());
                    }
                    if tasks.insert(record.id.clone(), record.clone()).is_none() {
                        order.push(record.id);
                    } else {
                        skipped += 1;
                    }
                }
                format!("恢复任务历史 {total} 条（标记中断 {interrupted} 条，跳过无法识别 {skipped} 条）")
            }
            Err(error) => {
                // 保留损坏文件供诊断（改名而非覆盖），从空历史开始。
                let corrupt = path.with_extension("json.corrupt");
                if fs::rename(&path, &corrupt).is_ok() {
                    format!(
                        "任务历史不可用（{error}）；原文件已改名保留为 {}",
                        corrupt.display()
                    )
                } else {
                    format!("任务历史不可用（{error}）；且无法改名保留原文件")
                }
            }
        };
        let registry = Self {
            inner: Arc::new(Mutex::new(TaskStore {
                tasks,
                order,
                seq,
                persist_path: Some(path),
            })),
        };
        // 把中断标记（或空历史）立即写回，保证文件与内存视图一致。
        registry.lock().persist();
        (registry, summary)
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, TaskStore> {
        self.inner
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    /// 创建 queued 任务并返回 task_id。
    pub(super) fn create(&self, kind: TaskKind) -> String {
        let now = unix_seconds_now();
        let mut store = self.lock();
        store.seq += 1;
        let id = format!("task-{}-{}", std::process::id(), store.seq);
        let record = TaskRecord {
            id: id.clone(),
            kind: kind.method(),
            status: "queued",
            stage: None,
            message: None,
            created_at: now,
            updated_at: now,
            started_at: None,
            finished_at: None,
            error: None,
            result: None,
            cancel: Arc::new(AtomicBool::new(false)),
            log: Vec::new(),
        };
        store.tasks.insert(id.clone(), record);
        store.order.push(id.clone());
        store.prune();
        store.persist();
        id
    }

    pub(super) fn update<F: FnOnce(&mut TaskRecord)>(&self, id: &str, update: F) {
        let mut store = self.lock();
        let mut status_changed = false;
        if let Some(record) = store.tasks.get_mut(id) {
            let previous_status = record.status;
            update(record);
            record.updated_at = unix_seconds_now();
            status_changed = record.status != previous_status;
        }
        // 只在生命周期转换时落盘；stage/message/log 的高频进度更新以内存为准，
        // 随下一次转换一起写入（避免每个进度事件一次磁盘写）。
        if status_changed {
            store.persist();
        }
    }

    pub(super) fn get(&self, id: &str) -> Option<TaskRecord> {
        self.lock().tasks.get(id).cloned()
    }

    /// 返回任务的取消标志（与 worker 共享同一 Arc）。
    pub(super) fn cancel_flag(&self, id: &str) -> Option<Arc<AtomicBool>> {
        self.lock()
            .tasks
            .get(id)
            .map(|record| Arc::clone(&record.cancel))
    }

    /// 追加一行进度日志，超出上限时丢弃最旧的。
    pub(super) fn append_log(&self, id: &str, line: String) {
        let mut store = self.lock();
        if let Some(record) = store.tasks.get_mut(id) {
            record.log.push(line);
            if record.log.len() > MAX_TASK_LOG_LINES {
                let overflow = record.log.len() - MAX_TASK_LOG_LINES;
                record.log.drain(0..overflow);
            }
        }
    }

    /// 返回任务的进度日志。
    pub(super) fn logs(&self, id: &str) -> Option<Vec<String>> {
        self.lock().tasks.get(id).map(|record| record.log.clone())
    }

    /// 请求取消任务：未结束的置取消标志，已结束的原样返回，不存在返回 NotFound。
    pub(super) fn request_cancel(&self, id: &str) -> CancelOutcome {
        let store = self.lock();
        match store.tasks.get(id) {
            None => CancelOutcome::NotFound,
            Some(record) if record.is_finished() => CancelOutcome::AlreadyFinished,
            Some(record) => {
                record.cancel.store(true, Ordering::Relaxed);
                CancelOutcome::Requested
            }
        }
    }

    /// 返回全部任务，最新创建的在前。
    pub(super) fn list(&self) -> Vec<TaskRecord> {
        let store = self.lock();
        store
            .order
            .iter()
            .rev()
            .filter_map(|id| store.tasks.get(id).cloned())
            .collect()
    }
}

impl TaskStore {
    /// 把当前任务历史落盘（`0600` 原子写、不跟随 symlink）。
    ///
    /// 持久化未启用时为 no-op；写失败只记 stderr（进 daemon 日志），
    /// 不让持久化故障拖垮任务执行本身。
    fn persist(&self) {
        let Some(path) = &self.persist_path else {
            return;
        };
        let file = PersistedTaskFile {
            version: TASKS_FILE_VERSION,
            seq: self.seq,
            tasks: self
                .order
                .iter()
                .filter_map(|id| self.tasks.get(id))
                .map(PersistedTaskRecord::from_record)
                .collect(),
        };
        match serde_json::to_vec_pretty(&file) {
            Ok(bytes) => {
                if let Err(error) = write_file_atomic(path, &bytes, PRIVATE_FILE_MODE, "任务历史")
                {
                    eprintln!("[daemon] 任务历史落盘失败：{error}");
                }
            }
            Err(error) => eprintln!("[daemon] 任务历史序列化失败：{error}"),
        }
    }

    /// 裁剪最旧的已结束任务，把内存占用控制在上限内；运行中/排队中的任务不裁剪。
    fn prune(&mut self) {
        while self.order.len() > MAX_RETAINED_TASKS {
            let Some(position) = self.order.iter().position(|id| {
                self.tasks
                    .get(id)
                    .map(TaskRecord::is_finished)
                    .unwrap_or(true)
            }) else {
                break;
            };
            let id = self.order.remove(position);
            self.tasks.remove(&id);
        }
    }
}

/// 提交给任务 worker 的作业（配置已按任务类型派生完毕）。
pub(super) struct TaskJob {
    pub(super) id: String,
    pub(super) config: OfficialUpdateConfig,
    /// 与任务记录共享的取消标志。
    pub(super) cancel: Arc<AtomicBool>,
}

/// daemon 任务上下文：RPC handler 借它创建任务、入队和读取。
#[derive(Clone)]
pub(super) struct DaemonTaskContext {
    pub(super) registry: TaskRegistry,
    pub(super) queue: mpsc::Sender<TaskJob>,
    pub(super) base_config: OfficialUpdateConfig,
    pub(super) restart_controller: DaemonRestartController,
}

pub(super) type DaemonRestartController = fn(&Path) -> anyhow::Result<u32>;

/// 任务 worker：单线程 FIFO 消费任务队列，串行执行官方同步/校验。
///
/// 每个任务执行前获取进程内 `sync_lock`，与 watch 循环互斥（等待而非撞文件锁失败）；
/// 进度写入任务记录；`should_cancel` 接 daemon 停止标志，停机时中止在途任务。
pub(super) fn run_task_worker(
    receiver: mpsc::Receiver<TaskJob>,
    registry: TaskRegistry,
    sync_lock: Arc<Mutex<()>>,
    control: DaemonControl,
) {
    let service = OfficialUpdateService::new();
    for job in receiver {
        registry.update(&job.id, |record| {
            record.status = "running";
            record.started_at = Some(unix_seconds_now());
        });

        let cancel = Arc::clone(&job.cancel);
        let run_result = {
            let _sync_guard = sync_lock
                .lock()
                .unwrap_or_else(|poison| poison.into_inner());
            let progress_registry = registry.clone();
            let progress_id = job.id.clone();
            let cancel_check = Arc::clone(&cancel);
            let stop_control = Arc::clone(&control);
            service.run_with_progress_and_cancellation(
                &job.config,
                |event| {
                    progress_registry
                        .append_log(&progress_id, format!("[{}] {}", event.stage, event.message));
                    progress_registry.update(&progress_id, |record| {
                        record.stage = Some(event.stage.to_string());
                        record.message = Some(event.message.clone());
                    });
                },
                || {
                    cancel_check.load(Ordering::Relaxed)
                        || daemon_control_stop_requested(Some(&stop_control))
                },
            )
        };

        match run_result {
            Ok(report) => registry.update(&job.id, |record| {
                record.status = "succeeded";
                record.finished_at = Some(unix_seconds_now());
                record.result = serde_json::to_value(&report).ok();
            }),
            Err(error) => {
                let cancelled = cancel.load(Ordering::Relaxed);
                // 下载失败携带类型化 DownloadError（含准确网络域码）；其余归 internal。
                let code = error
                    .downcast_ref::<bat_infrastructure::DownloadError>()
                    .map(bat_infrastructure::DownloadError::code)
                    .unwrap_or(ErrorCode::INTERNAL);
                registry.update(&job.id, |record| {
                    record.finished_at = Some(unix_seconds_now());
                    if cancelled {
                        record.status = "cancelled";
                        record.error = Some(ApiError::new(
                            ErrorCode::INTERNAL,
                            "task.executor",
                            "任务已取消",
                        ));
                    } else {
                        record.status = "failed";
                        record.error =
                            Some(ApiError::new(code, "task.executor", error.to_string()));
                    }
                });
            }
        }
    }
}