# -*- coding: utf-8 -*-
"""本地 spool 缓存 + 断点续传 —— 丢失率 ≤ 0.02% 的实现保障（PRD 9 章）。

机制：
1. 每次采集样本先写入本地 spool 文件（JSON Lines，按小时分片）；
2. Kafka 上行确认（ack）后才删除对应记录；
3. 网关重启时扫描 spool 目录，未确认记录全部重发 —— 断点续传；
4. 上行通道抖动不丢数据，仅增加本地缓存占用（受 cache_limit_bytes 约束）。

文件命名：{shard_time:%Y%m%d%H}.spool.jsonl
"""
from __future__ import annotations

import json
import os
import threading
import time
from typing import List

from drivers.base import SampleValue


class SpoolStore:
    """本地 spool：追加写 + 按 ack 删除（断点续传）。"""

    def __init__(self, spool_dir: str, cache_limit_bytes: int = 512 * 1024 * 1024):
        self.spool_dir = spool_dir
        self.cache_limit_bytes = cache_limit_bytes
        os.makedirs(spool_dir, exist_ok=True)
        self._lock = threading.Lock()

    # ------------------------------------------------------------------
    def _shard_path(self, ts: float) -> str:
        return os.path.join(
            self.spool_dir,
            time.strftime("%Y%m%d%H", time.localtime(ts)) + ".spool.jsonl",
        )

    def append(self, device_id: str, point_id: str, value: SampleValue, ts: float) -> None:
        """写入一条待上行的样本记录。"""
        record = {
            "device_id": device_id,
            "point_id": point_id,
            "value": value,
            "ts": ts,
        }
        with self._lock:
            with open(self._shard_path(ts), "a", encoding="utf-8") as fh:
                fh.write(json.dumps(record, ensure_ascii=False) + "\n")

    def pending_records(self, limit: int = 10000) -> List[dict]:
        """读取所有未确认记录（断点续传：重启后调用，全部重发）。"""
        records: List[dict] = []
        with self._lock:
            for name in sorted(os.listdir(self.spool_dir)):
                if not name.endswith(".spool.jsonl"):
                    continue
                path = os.path.join(self.spool_dir, name)
                with open(path, "r", encoding="utf-8") as fh:
                    for line in fh:
                        line = line.strip()
                        if line:
                            records.append(json.loads(line))
                        if len(records) >= limit:
                            return records
        return records

    def ack(self, record: dict) -> None:
        """上行确认后删除对应记录。

        record 中 value/ts 为 None 的字段不参与匹配（Kafka 投递回调
        仅有 point_id 时，按 point_id 删除最早一条未确认记录）。
        """
        with self._lock:
            for name in sorted(os.listdir(self.spool_dir)):
                if not name.endswith(".spool.jsonl"):
                    continue
                path = os.path.join(self.spool_dir, name)
                try:
                    with open(path, "r", encoding="utf-8") as fh:
                        lines = fh.readlines()
                except OSError:
                    continue
                kept, removed = [], False
                for line in lines:
                    line = line.strip()
                    if not line:
                        continue
                    parsed = json.loads(line)
                    matched = all(
                        record.get(k) is None or parsed.get(k) == v
                        for k, v in record.items()
                    )
                    if not removed and matched:
                        removed = True  # 删除第一条匹配记录
                    else:
                        kept.append(line)
                if removed:
                    with open(path, "w", encoding="utf-8") as fh:
                        fh.write("\n".join(kept) + ("\n" if kept else ""))
                    break

    def total_pending(self) -> int:
        """当前未确认记录总数（健康度上报用）。"""
        return len(self.pending_records(limit=10 ** 9))