"""async_uploader.py — Non-blocking async upload engine with chunking and circuit breaker.

Key features:
- ThreadPoolExecutor-based background upload (non-blocking)
- Chunked upload with per-chunk SHA256 checksums
- Chunk manifest with sequential acknowledgment
- Circuit breaker (closed → open → half-open)
- Progress tracking via status file (pub/sub compatible)
- Adaptive batch sizing based on network conditions
- Smart retry with Retry-After header support
"""

from __future__ import annotations

import gzip
import hashlib
import json
import os
import random
import time
import uuid
from concurrent.futures import ThreadPoolExecutor, Future
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Any, Callable
import threading

import config_manager

APP_DIR = config_manager.get_app_dir()
DATA_DIR = config_manager.get_data_dir()
LOGS_DIR = config_manager.get_logs_dir()
UPLOAD_DIR = DATA_DIR / "uploads"

# ── Chunk config ─────────────────────────────────────────────────
DEFAULT_CHUNK_SIZE = int(os.environ.get("DATA_CHUNK_SIZE", "100"))  # events per chunk
MAX_CHUNK_SIZE = 500
MIN_CHUNK_SIZE = 20
CHUNK_MANIFEST_FILE = UPLOAD_DIR / ".chunk_manifest_{tenant_id}_{batch_id}.json"

# ── Circuit breaker ──────────────────────────────────────────────
CIRCUIT_FAILURE_THRESHOLD = 5
CIRCUIT_RECOVERY_TIMEOUT_SEC = 60  # seconds before trying half-open
CIRCUIT_HALF_OPEN_SUCCESSES = 2  # successes needed to close circuit


class CircuitState(Enum):
    CLOSED = "closed"   # normal operation
    OPEN = "open"      # failing, reject immediately
    HALF_OPEN = "half_open"  # testing recovery


@dataclass
class CircuitBreaker:
    failures: int = 0
    state: CircuitState = CircuitState.CLOSED
    last_failure_ts: float = 0.0
    half_open_successes: int = 0

    def record_failure(self) -> None:
        self.failures += 1
        self.last_failure_ts = time.time()
        if self.failures >= CIRCUIT_FAILURE_THRESHOLD:
            self.state = CircuitState.OPEN

    def record_success(self) -> None:
        self.failures = 0
        self.half_open_successes += 1
        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_successes >= CIRCUIT_HALF_OPEN_SUCCESSES:
                self.state = CircuitState.CLOSED
                self.half_open_successes = 0
        elif self.state == CircuitState.CLOSED:
            self.half_open_successes = 0

    def can_attempt(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_ts >= CIRCUIT_RECOVERY_TIMEOUT_SEC:
                self.state = CircuitState.HALF_OPEN
                self.half_open_successes = 0
                return True
            return False
        # HALF_OPEN
        return True

    def to_dict(self) -> dict[str, Any]:
        return {
            "state": self.state.value,
            "failures": self.failures,
            "last_failure_ts": self.last_failure_ts,
            "half_open_successes": self.half_open_successes,
        }


# ── Chunk manifest ───────────────────────────────────────────────

@dataclass
class ChunkInfo:
    chunk_index: int
    events: list[dict[str, Any]]
    sha256: str = ""
    status: str = "pending"  # pending | uploading | uploaded | failed
    attempt: int = 0
    upload_ts: str = ""
    error: str = ""
    compressed_size: int = 0

    def __post_init__(self) -> None:
        if not self.sha256 and self.events:
            raw = json.dumps(self.events, ensure_ascii=False, sort_keys=True).encode("utf-8")
            self.sha256 = hashlib.sha256(raw).hexdigest()

    def compress(self) -> tuple[bytes, int]:
        raw = json.dumps(self.events, ensure_ascii=False, sort_keys=True).encode("utf-8")
        compressed = gzip.compress(raw, compresslevel=6)
        return compressed, len(compressed)


@dataclass
class ChunkManifest:
    batch_id: str
    tenant_id: str
    total_events: int
    total_chunks: int
    chunk_size: int
    created_at: str
    completed_chunks: int = 0
    failed_chunks: int = 0
    status: str = "in_progress"  # in_progress | completed | failed
    server_batch_id: str = ""
    server_compressed_size: int = 0
    total_upload_time_sec: float = 0.0
    retry_count: int = 0
    chunks: list[ChunkInfo] = field(default_factory=list)
    _lock: threading.Lock = field(default_factory=threading.Lock)

    @property
    def all_uploaded(self) -> bool:
        return self.completed_chunks >= self.total_chunks

    def mark_chunk_uploaded(self, chunk_index: int) -> None:
        with self._lock:
            for c in self.chunks:
                if c.chunk_index == chunk_index:
                    c.status = "uploaded"
                    c.upload_ts = datetime.now(timezone.utc).isoformat()
                    break
            self.completed_chunks = sum(1 for c in self.chunks if c.status == "uploaded")
            if self.completed_chunks >= self.total_chunks:
                self.status = "completed"

    def mark_chunk_failed(self, chunk_index: int, error: str) -> None:
        with self._lock:
            for c in self.chunks:
                if c.chunk_index == chunk_index:
                    c.status = "failed"
                    c.error = error
                    c.attempt += 1
                    break
            self.failed_chunks = sum(1 for c in self.chunks if c.status == "failed")
            if self.failed_chunks > self.total_chunks // 2:
                self.status = "failed"

    def progress(self) -> float:
        if self.total_chunks == 0:
            return 1.0
        return self.completed_chunks / self.total_chunks

    def to_dict(self) -> dict[str, Any]:
        return {
            "batch_id": self.batch_id,
            "tenant_id": self.tenant_id,
            "total_events": self.total_events,
            "total_chunks": self.total_chunks,
            "chunk_size": self.chunk_size,
            "created_at": self.created_at,
            "completed_chunks": self.completed_chunks,
            "failed_chunks": self.failed_chunks,
            "status": self.status,
            "server_batch_id": self.server_batch_id,
            "server_compressed_size": self.server_compressed_size,
            "total_upload_time_sec": self.total_upload_time_sec,
            "retry_count": self.retry_count,
            "progress": round(self.progress(), 4),
            "chunks": [
                {
                    "chunk_index": c.chunk_index,
                    "sha256": c.sha256,
                    "status": c.status,
                    "attempt": c.attempt,
                    "upload_ts": c.upload_ts,
                    "compressed_size": c.compressed_size,
                    "error": c.error,
                }
                for c in self.chunks
            ],
        }

    def save(self) -> None:
        UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
        path = UPLOAD_DIR / f".chunk_manifest_{self.tenant_id}_{self.batch_id}.json"
        tmp = path.with_name(f".{os.getpid()}.tmp")
        tmp.write_text(json.dumps(self.to_dict(), ensure_ascii=False), encoding="utf-8")
        tmp.replace(path)

    @classmethod
    def load(cls, tenant_id: str, batch_id: str) -> "ChunkManifest | None":
        path = UPLOAD_DIR / f".chunk_manifest_{tenant_id}_{batch_id}.json"
        if not path.exists():
            return None
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            data.pop("_lock", None)
            return cls(**data)
        except Exception:
            return None


# ── Progress tracker ─────────────────────────────────────────────

class ProgressTracker:
    """Thread-safe progress tracker with file-based pub/sub support."""

    def __init__(self, manifest: ChunkManifest) -> None:
        self._manifest = manifest
        self._lock = threading.Lock()
        self._subscribers: list[Callable[[dict[str, Any]], None]] = []
        self._last_notify_progress = -1.0
        self._last_notify_ts = 0.0
        self._notify_interval_sec = 5.0  # notify at most every 5s
        self._status_file = UPLOAD_DIR / f".upload_progress_{manifest.tenant_id}_{manifest.batch_id}.json"
        self._done_callbacks: list[Callable[[dict[str, Any]], None]] = []

    def subscribe(self, cb: Callable[[dict[str, Any]], None]) -> None:
        with self._lock:
            self._subscribers.append(cb)

    def on_done(self, cb: Callable[[dict[str, Any]], None]) -> None:
        with self._lock:
            self._done_callbacks.append(cb)

    def notify(self, force: bool = False) -> None:
        now = time.time()
        with self._lock:
            progress = self._manifest.progress()
            if not force and progress == self._last_notify_progress and now - self._last_notify_ts < self._notify_interval_sec:
                return
            self._last_notify_progress = progress
            self._last_notify_ts = now

            payload = {
                "batch_id": self._manifest.batch_id,
                "tenant_id": self._manifest.tenant_id,
                "progress": round(progress, 4),
                "completed_chunks": self._manifest.completed_chunks,
                "total_chunks": self._manifest.total_chunks,
                "total_events": self._manifest.total_events,
                "status": self._manifest.status,
                "failed_chunks": self._manifest.failed_chunks,
                "ts": datetime.now(timezone.utc).isoformat(),
            }
            for cb in self._subscribers:
                try:
                    cb(payload)
                except Exception:
                    pass

            self._persist_status(payload)

            if self._manifest.all_uploaded or self._manifest.status == "failed":
                for cb in self._done_callbacks:
                    try:
                        cb(payload)
                    except Exception:
                        pass

    def _persist_status(self, payload: dict[str, Any]) -> None:
        try:
            UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
            path = self._status_file
            tmp = path.with_name(f".{os.getpid()}.tmp")
            tmp.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
            tmp.replace(path)
        except Exception:
            pass

    def get_status(self) -> dict[str, Any]:
        return {
            "batch_id": self._manifest.batch_id,
            "progress": round(self._manifest.progress(), 4),
            "completed_chunks": self._manifest.completed_chunks,
            "total_chunks": self._manifest.total_chunks,
            "status": self._manifest.status,
        }


# ── Adaptive batch sizing ─────────────────────────────────────────

class AdaptiveBatcher:
    """Dynamically adjusts chunk size based on upload latency history."""

    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._recent_latencies: list[float] = []
        self._current_chunk_size = DEFAULT_CHUNK_SIZE
        self._upload_count = 0

    def record_latency(self, chunk_size: int, latency_sec: float) -> None:
        with self._lock:
            self._recent_latencies.append(latency_sec)
            if len(self._recent_latencies) > 20:
                self._recent_latencies.pop(0)
            self._upload_count += 1
            if self._upload_count % 10 == 0:
                self._adjust_chunk_size()

    def _adjust_chunk_size(self) -> None:
        if len(self._recent_latencies) < 5:
            return
        avg_latency = sum(self._recent_latencies) / len(self._recent_latencies)
        # Target: < 2s per chunk upload
        if avg_latency < 1.5:
            self._current_chunk_size = min(MAX_CHUNK_SIZE, int(self._current_chunk_size * 1.25))
        elif avg_latency > 3.0:
            self._current_chunk_size = max(MIN_CHUNK_SIZE, int(self._current_chunk_size * 0.75))
        # Quantize
        self._current_chunk_size = (self._current_chunk_size // 10) * 10

    def get_chunk_size(self) -> int:
        return self._current_chunk_size

    def to_dict(self) -> dict[str, Any]:
        with self._lock:
            return {
                "current_chunk_size": self._current_chunk_size,
                "recent_avg_latency_sec": round(sum(self._recent_latencies) / max(1, len(self._recent_latencies)), 2),
                "sample_count": len(self._recent_latencies),
            }


# ── Network optimizer ─────────────────────────────────────────────

def _estimate_compression_ratio(events: list[dict[str, Any]]) -> float:
    raw = json.dumps(events, ensure_ascii=False, sort_keys=True).encode("utf-8")
    compressed = gzip.compress(raw, compresslevel=6)
    return len(compressed) / max(1, len(raw))


# ── Async upload engine ───────────────────────────────────────────

@dataclass
class UploadTask:
    batch_id: str
    tenant_id: str
    api_key: str | None
    verify_tls: bool
    tenant_hash: str
    chunks: list[ChunkInfo]
    manifest: ChunkManifest
    tracker: ProgressTracker
    circuit: CircuitBreaker
    batcher: AdaptiveBatcher
    result: dict[str, Any] = field(default_factory=lambda: {})
    done: bool = False


class AsyncUploadEngine:
    """Non-blocking upload engine using ThreadPoolExecutor.

    Usage:
        engine = AsyncUploadEngine(max_workers=4)
        future = engine.submit(
            events=[...],
            tenant_id="tenant1",
            tenant_hash="a7f3e2b1",
        )
        # Non-blocking — returns immediately
        result = future.result(timeout=60)  # optional blocking wait
    """

    def __init__(
        self,
        max_workers: int = 4,
        server_url: str | None = None,
    ) -> None:
        self._executor = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="uploader")
        self._server_url = server_url or os.environ.get(
            "DATA_SERVER_URL", "http://43.167.210.121:8000"
        )
        self._active_tasks: dict[str, Future[dict[str, Any]]] = {}
        self._lock = threading.Lock()
        self._circuit = CircuitBreaker()
        self._batcher = AdaptiveBatcher()

    def submit(
        self,
        events: list[dict[str, Any]],
        tenant_id: str,
        tenant_hash: str,
        api_key: str | None = None,
        verify_tls: bool = True,
        batch_id: str | None = None,
        on_progress: Callable[[dict[str, Any]], None] | None = None,
        on_done: Callable[[dict[str, Any]], None] | None = None,
    ) -> Future[dict[str, Any]]:
        """Submit an upload job. Returns immediately (non-blocking)."""
        if not batch_id:
            batch_id = str(uuid.uuid4())

        # Build chunk manifest
        chunk_size = self._batcher.get_chunk_size()
        chunks = self._chunk_events(events, chunk_size)

        manifest = ChunkManifest(
            batch_id=batch_id,
            tenant_id=tenant_id,
            total_events=len(events),
            total_chunks=len(chunks),
            chunk_size=chunk_size,
            created_at=datetime.now(timezone.utc).isoformat(),
            chunks=chunks,
        )

        tracker = ProgressTracker(manifest)
        if on_progress:
            tracker.subscribe(on_progress)
        if on_done:
            tracker.on_done(on_done)

        task = UploadTask(
            batch_id=batch_id,
            tenant_id=tenant_id,
            api_key=api_key,
            verify_tls=verify_tls,
            tenant_hash=tenant_hash,
            chunks=chunks,
            manifest=manifest,
            tracker=tracker,
            circuit=self._circuit,
            batcher=self._batcher,
        )

        def _run() -> dict[str, Any]:
            return self._execute_upload(task)

        future: Future[dict[str, Any]] = self._executor.submit(_run)

        with self._lock:
            self._active_tasks[batch_id] = future

        future.add_done_callback(lambda f: self._cleanup(batch_id))
        return future

    def _chunk_events(self, events: list[dict[str, Any]], chunk_size: int) -> list[ChunkInfo]:
        chunks: list[ChunkInfo] = []
        for i in range(0, len(events), chunk_size):
            chunk_events = events[i:i + chunk_size]
            chunks.append(ChunkInfo(chunk_index=i // chunk_size, events=chunk_events))
        return chunks

    def _execute_upload(self, task: UploadTask) -> dict[str, Any]:
        """Execute the full chunked upload with circuit breaker and retries."""
        start_time = time.time()
        result: dict[str, Any] = {
            "batch_id": task.batch_id,
            "tenant_id": task.tenant_id,
            "uploaded": 0,
            "failed": 0,
            "skipped": 0,
            "chunks_total": len(task.chunks),
            "chunks_uploaded": 0,
            "chunks_failed": 0,
            "errors": [],
            "server_batch_id": "",
            "total_time_sec": 0.0,
        }

        if not task.circuit.can_attempt():
            result["errors"].append(f"circuit_open: server unavailable, try after {CIRCUIT_RECOVERY_TIMEOUT_SEC}s")
            result["status"] = "circuit_open"
            return result

        # Upload chunks sequentially (server expects ordered append)
        for chunk in task.chunks:
            chunk_start = time.time()
            if task.manifest.status == "failed":
                result["errors"].append("manifest_failed: stopping further uploads")
                break

            try:
                upload_res = self._upload_single_chunk(chunk, task)
                chunk_latency = time.time() - chunk_start
                task.batcher.record_latency(len(chunk.events), chunk_latency)

                if upload_res.get("success"):
                    task.manifest.mark_chunk_uploaded(chunk.chunk_index)
                    result["uploaded"] += len(chunk.events)
                    result["chunks_uploaded"] += 1
                    task.circuit.record_success()
                    if not result["server_batch_id"]:
                        result["server_batch_id"] = upload_res.get("server_batch_id", "")
                else:
                    error = upload_res.get("error", "unknown")
                    task.manifest.mark_chunk_failed(chunk.chunk_index, error)
                    result["failed"] += len(chunk.events)
                    result["chunks_failed"] += 1
                    result["errors"].append(f"chunk_{chunk.chunk_index}: {error}")
                    task.circuit.record_failure()

            except Exception as e:
                error = str(e)
                task.manifest.mark_chunk_failed(chunk.chunk_index, error)
                result["failed"] += len(chunk.events)
                result["chunks_failed"] += 1
                result["errors"].append(f"chunk_{chunk.chunk_index}: {error}")
                task.circuit.record_failure()

            task.manifest.save()
            task.tracker.notify()
            manifest_payload = task.tracker.get_status()
            for cb in task.tracker._subscribers:
                try:
                    cb(manifest_payload)
                except Exception:
                    pass

        result["total_time_sec"] = round(time.time() - start_time, 2)
        result["status"] = "completed" if task.manifest.all_uploaded else "partial"
        if result["chunks_failed"] > result["chunks_total"] // 2:
            result["status"] = "failed"

        task.manifest.save()
        task.tracker.notify(force=True)
        return result

    def _upload_single_chunk(
        self,
        chunk: ChunkInfo,
        task: UploadTask,
    ) -> dict[str, Any]:
        """Upload one chunk to the server with retry and Retry-After support."""
        compressed, _ = chunk.compress()
        import base64
        encoded = base64.b64encode(compressed).decode("ascii")

        payload = {
            "batch_id": task.batch_id,
            "chunk_index": chunk.chunk_index,
            "total_chunks": task.manifest.total_chunks,
            "chunk_sha256": chunk.sha256,
            "event_count": len(chunk.events),
            "compression": "gzip+base64",
            "events_data": encoded,
        }

        headers: dict[str, str] = {
            "Content-Type": "application/json",
            "Content-Encoding": "identity",
            "X-Api-Key": task.api_key or os.environ.get("DATA_API_KEY", ""),
            "X-Tenant-Hash": task.tenant_hash,
            "X-Batch-Id": task.batch_id,
            "X-Chunk-Index": str(chunk.chunk_index),
        }

        max_retries = 5
        retry_delay = 1.0

        for attempt in range(1, max_retries + 1):
            try:
                import urllib.request
                import urllib.error

                url = f"{self._server_url}/api/data/chunk_ingest"
                data = json.dumps(payload, ensure_ascii=False).encode("utf-8")

                req = urllib.request.Request(
                    url,
                    data=data,
                    headers=headers,
                    method="POST",
                )

                if not task.verify_tls:
                    import ssl
                    ctx = ssl.create_default_context()
                    ctx.check_hostname = False
                    ctx.verify_mode = ssl.CERT_NONE
                    response = urllib.request.urlopen(req, timeout=30, context=ctx)
                else:
                    response = urllib.request.urlopen(req, timeout=30)

                body = response.read()
                res_data = json.loads(body.decode("utf-8"))

                if response.status == 200:
                    return {
                        "success": True,
                        "server_batch_id": res_data.get("batch_id", task.batch_id),
                    }
                elif response.status == 409:
                    # Idempotent — chunk already uploaded
                    return {"success": True, "server_batch_id": res_data.get("batch_id", task.batch_id)}
                else:
                    return {"success": False, "error": f"http_{response.status}"}

            except urllib.error.HTTPError as e:
                body = e.read() if e.fp else b""
                try:
                    err_data = json.loads(body.decode("utf-8"))
                    error_msg = err_data.get("error", str(e))
                except Exception:
                    error_msg = str(e)

                # Retry on 5xx or specific recoverable errors
                if e.code >= 500:
                    if attempt < max_retries:
                        # Check Retry-After header
                        retry_after = e.headers.get("Retry-After")
                        if retry_after:
                            try:
                                delay = int(retry_after)
                            except ValueError:
                                delay = int(float(retry_after))
                        else:
                            delay = retry_delay
                        time.sleep(delay)
                        retry_delay = min(60.0, retry_delay * 2)
                        continue
                    return {"success": False, "error": f"http_{e.code}_exhausted: {error_msg}"}
                else:
                    return {"success": False, "error": f"http_{e.code}: {error_msg}"}

            except Exception as e:
                if attempt < max_retries:
                    jitter = random.uniform(0.5, 1.5)
                    wait = retry_delay * jitter
                    time.sleep(wait)
                    retry_delay = min(60.0, retry_delay * 2)
                    continue
                return {"success": False, "error": str(e)}

        return {"success": False, "error": "max_retries_exceeded"}

    def _cleanup(self, batch_id: str) -> None:
        with self._lock:
            self._active_tasks.pop(batch_id, None)

    def get_active_count(self) -> int:
        with self._lock:
            return len(self._active_tasks)

    def get_circuit_status(self) -> dict[str, Any]:
        return self._circuit.to_dict()

    def get_batcher_status(self) -> dict[str, Any]:
        return self._batcher.to_dict()

    def shutdown(self, wait: bool = True) -> None:
        self._executor.shutdown(wait=wait)


# ── Singleton instance ────────────────────────────────────────────

_upload_engine: AsyncUploadEngine | None = None
_engine_lock = threading.Lock()


def get_upload_engine(
    max_workers: int = 4,
    server_url: str | None = None,
) -> AsyncUploadEngine:
    """Get or create the singleton upload engine."""
    global _upload_engine
    with _engine_lock:
        if _upload_engine is None:
            _upload_engine = AsyncUploadEngine(max_workers=max_workers, server_url=server_url)
        return _upload_engine


# ── High-level API ───────────────────────────────────────────────

def async_upload_batch(
    events: list[dict[str, Any]],
    tenant_id: str,
    api_key: str | None = None,
    verify_tls: bool = True,
    tenant_hash: str | None = None,
    batch_id: str | None = None,
    on_progress: Callable[[dict[str, Any]], None] | None = None,
    on_done: Callable[[dict[str, Any]], None] | None = None,
) -> Future[dict[str, Any]]:
    """Submit a batch for async upload. Non-blocking.

    Returns a Future[dict] with keys:
        uploaded, failed, skipped, chunks_total, chunks_uploaded,
        chunks_failed, server_batch_id, total_time_sec, status, errors
    """
    if not tenant_hash:
        from src.data_pipeline.anonymizer import hash_tenant_id
        tenant_hash = hash_tenant_id(tenant_id)

    engine = get_upload_engine()
    return engine.submit(
        events=events,
        tenant_id=tenant_id,
        tenant_hash=tenant_hash,
        api_key=api_key,
        verify_tls=verify_tls,
        batch_id=batch_id,
        on_progress=on_progress,
        on_done=on_done,
    )


def get_upload_progress(
    tenant_id: str,
    batch_id: str,
) -> dict[str, Any] | None:
    """Read upload progress from status file (for external polling)."""
    path = UPLOAD_DIR / f".upload_progress_{tenant_id}_{batch_id}.json"
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return None


def check_upload_circuit() -> dict[str, Any]:
    """Check circuit breaker status."""
    engine = get_upload_engine()
    return {
        "circuit": engine.get_circuit_status(),
        "batcher": engine.get_batcher_status(),
    }