"""EVART-01: parent-owned rendering, bounded delivery and media-only HTTP serving."""
import fcntl
import json
import os
from pathlib import Path
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
from collections import Counter

from app.event_artifact_contract import ArtifactConfig, build_body, media_path, STAGE, STAGE_DESCRIPTION, RESERVATION
from app.event_artifact_store import ArtifactStore


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def post_event(endpoint, body, *, authorization=None):
    request = urllib.request.Request(endpoint, json.dumps(body, ensure_ascii=False).encode(),
                                     dict({'Content-Type': 'application/json'}, **({'Authorization': authorization} if authorization else {})), method='POST')
    try:
        with urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect()).open(request, timeout=3) as response:
            status, raw = response.status, response.read(65537)
    except urllib.error.HTTPError as exc:
        raw = exc.read(4096)
        try:
            rejected = json.loads(raw)
        except (ValueError, UnicodeError):
            rejected = {'text': raw.decode('utf-8', errors='replace')[:1024]}
        return False, exc.code in (408, 425, 429) or exc.code >= 500, exc.code, rejected, 'HTTP_REJECTED'
    except (OSError, urllib.error.URLError, TimeoutError):
        return False, True, None, None, 'RESPONSE_UNKNOWN'
    if len(raw) > 65536:
        return False, False, status, None, 'RESPONSE_TOO_LARGE'
    try:
        result = json.loads(raw)
        data = result['data']
        valid = (result['result'] is True and type(data['accepted']) is int and
                 type(data['duplicated']) is int and data['accepted'] >= 0 and
                 data['duplicated'] >= 0 and data['accepted']+data['duplicated'] == 1)
    except (ValueError, TypeError, KeyError):
        valid = False
    safe_response = {'accepted': data['accepted'], 'duplicated': data['duplicated']} if valid else None
    return valid and 200 <= status < 300, False, status, safe_response, None if valid else 'RESPONSE_CONTRACT'


def file_app(root):
    from fastapi import FastAPI, HTTPException
    from starlette.responses import FileResponse
    app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
    @app.api_route('/artifacts/{relative:path}', methods=['GET', 'HEAD'])
    def get_media(relative: str):
        path = media_path(root, relative)
        if path is None:
            raise HTTPException(404)
        return FileResponse(path, media_type='video/mp4' if path.suffix == '.mp4' else 'image/jpeg',
                            headers={'X-Content-Type-Options': 'nosniff', 'Cache-Control': 'private, max-age=60'})
    return app


class ArtifactService:
    def __init__(self, config):
        self.config = config
        self.store = ArtifactStore(config)
        self.stop = threading.Event()
        self.workers = []
        self.http = None
        self.http_thread = None
        self.processes = {}
        self.metrics_lock = threading.Lock()
        self.activity = {}
        self.counts = Counter()
        self.peaks = Counter()
        self.cached = {}
        self.monitor = None
        self.lock = None
        self.error = None
        self.relay = None

    @classmethod
    def from_environment(cls):
        config = ArtifactConfig.from_environment()
        return cls(config) if config else None

    def start(self):
        import uvicorn
        self.config.root.mkdir(parents=True, exist_ok=True)
        self.lock = (self.config.root / '.worker.lock').open('a+')
        fcntl.flock(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        if self.config.enabled:
            from app.secure_event_transport import RelayTransport
            self.relay = RelayTransport.from_environment(self.config.endpoint)
            self.store.initialize()
            self.store.recover()
        self.http = uvicorn.Server(uvicorn.Config(file_app(self.config.root), host='0.0.0.0', port=self.config.port,
                                                 log_level='warning', access_log=False))
        self.http_thread = threading.Thread(target=self.http.run, name='artifact-http', daemon=True)
        self.http_thread.start()
        deadline = time.monotonic()+5
        while not self.http.started and self.http_thread.is_alive() and time.monotonic() < deadline:
            time.sleep(.05)
        if not self.http.started:
            raise RuntimeError('Artifact HTTP server did not start.')
        if not self.config.enabled:
            return
        for kind in ('render', 'send'):
            for index in range(4):
                name = f'artifact-{kind}-{index}'
                self.activity[name] = 'idle'
                worker = threading.Thread(target=self._run, args=(kind,name), name=name, daemon=True)
                self.workers.append(worker)
                worker.start()
        self.monitor = threading.Thread(target=self._monitor, name='artifact-status', daemon=True)
        self.monitor.start()

    def status(self):
        with self.metrics_lock:
            current = Counter(self.activity.values())
            storage = self.cached.get('storage',{})
            accounted = storage.get('observed_bytes',0)+storage.get('reserved_bytes',0)
            return dict(enabled=self.config.enabled, http_enabled=True, workers=len(self.workers), render_limit=4, active=sum(v!='idle' for v in self.activity.values()),
                        activity=dict(current), rendering=len(self.processes), peaks=dict(self.peaks),
                        counts=dict(self.counts), error=self.error, ledger=self.cached,
                        storage_warning=accounted>=self.config.warning,
                        storage_warning_bytes=self.config.warning,storage_cap_bytes=self.config.cap)

    def _monitor(self):
        last_scan = time.monotonic()
        while not self.stop.is_set():
            try:
                self.store.expire()
                snapshot = self.store.summary()
                snapshot['sampled_at'] = time.time()
                with self.metrics_lock:
                    self.cached = snapshot
                if time.monotonic()-last_scan >= 60:
                    self.store.scan_storage()
                    last_scan = time.monotonic()
            except Exception as exc:
                self.error = 'ARTIFACT_STATUS_' + type(exc).__name__.upper()
            self.stop.wait(2)

    def _render(self, row):
        stage = self.config.root / '.staging' / row['event_id']
        final = self.config.root / row['relative_path']
        media, error = {}, row['error']
        if row['artifact_state']=='EXPIRED' or row['confirmed_at']+self.config.lifetime<=time.time():
            self.store.cleanup_expired(row)
            with self.metrics_lock:
                self.counts['expired_cleaned'] += 1
            return
        process = None
        if row['artifact_state'] == 'CAPTURE_READY':
            try:
                env = dict(os.environ, CONTEXTRA_ARTIFACT_PARENT_PID=str(os.getpid()),
                           OMP_NUM_THREADS='1', OPENBLAS_NUM_THREADS='1')
                process = subprocess.Popen([sys.executable, '-m', 'app.artifact_media', str(stage)],
                                                env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                                                start_new_session=True)
                with self.metrics_lock:
                    self.processes[row['event_id']] = process
                    self.peaks['rendering'] = max(self.peaks['rendering'],len(self.processes))
                process.communicate(timeout=60)
                if process.returncode:
                    raise RuntimeError('Renderer exited unsuccessfully.')
                ready = stage / 'ready'
                size = sum(p.stat().st_size for p in stage.rglob('*') if p.is_file())
                if size > RESERVATION:
                    raise RuntimeError('Artifact reservation exceeded.')
                manifest = json.loads((ready/'event.json').read_text())
                media = manifest['media']
                manifest.update(aiRecordedStage=STAGE, aiRecordedStageDescription=STAGE_DESCRIPTION)
                (ready/'event.json').write_text(json.dumps(manifest, ensure_ascii=False, indent=2))
                final.parent.mkdir(parents=True, exist_ok=True)
                if final.exists():
                    raise RuntimeError('Event directory already published.')
                os.rename(ready, final)
                # Accepted Event and overlay are now durable; discard only this job's intermediate cache.
                (stage/'capture.bin').unlink()
                (stage/'job.json').unlink()
                stage.rmdir()
            except Exception as exc:
                error = 'ARTIFACT_' + type(exc).__name__.upper()
                if process is not None and process.poll() is None:
                    process.kill()
                    process.wait(timeout=3)
            finally:
                with self.metrics_lock:
                    self.processes.pop(row['event_id'],None)
        if error and not final.exists():
            stage.mkdir(parents=True, exist_ok=True)
            (stage/'failure.json').write_text(json.dumps(dict(event=row['payload'], error=error,
                aiRecordedStage=STAGE, aiRecordedStageDescription=STAGE_DESCRIPTION), ensure_ascii=False, indent=2))
        size = sum(p.stat().st_size for root in (stage, final) if root.exists() for p in root.rglob('*') if p.is_file())
        self.store.rendered(row, media, error, size)
        with self.metrics_lock:
            self.counts['render_failed' if error else 'render_completed'] += 1

    def _deliver(self, row):
        body = row['body'] or build_body(row['payload'], row['media'], row['error'], self.config.public_base, row['relative_path'])
        body = self.store.prepare_delivery(row, body)
        if body is None:
            return
        if self.stop.is_set():
            self.store.finish_attempt(row,'PENDING',None,None,'STOP_BEFORE_SEND')
            return
        attempt = self.store.begin_attempt(row)
        if attempt is None:
            return
        if self.relay is None:
            ok, retry, status, response, error = post_event(self.config.endpoint, body)
        else:
            ok, retry, status, response, error = self.relay.post(body, post_event)
        state = 'SENT' if ok else 'RETRY' if retry and attempt < 2 else 'FAILED'
        self.store.finish_attempt(row,state,status,response,error)
        with self.metrics_lock:
            self.counts[state.lower()] += 1

    def _run(self, kind, name):
        while not self.stop.is_set():
            row = None
            try:
                row = self.store.claim(kind)
                if row is None:
                    self.stop.wait(.5)
                    continue
                with self.metrics_lock:
                    self.activity[name] = kind
                    self.peaks['active'] = max(self.peaks['active'],sum(v!='idle' for v in self.activity.values()))
                (self._render if kind=='render' else self._deliver)(row)
            except Exception as exc:
                self.error = 'EVENT_JOB_' + type(exc).__name__.upper()
                with self.metrics_lock:
                    self.counts['job_errors'] += 1
                if row is not None:
                    try:
                        if kind=='send':
                            self.store.finish_attempt(row,'FAILED',None,None,self.error)
                        else:
                            stage = self.config.root/'.staging'/row['event_id']
                            size = sum(p.stat().st_size for p in stage.rglob('*') if p.is_file())
                            self.store.rendered(row,{},self.error,size)
                    except Exception:
                        # Preserve claim/reservation on cleanup failure; never fake a release.
                        row = None
                self.stop.wait(.5)
            finally:
                if row is not None:
                    try:
                        self.store.release_claim(row,kind)
                    except Exception as exc:
                        self.error = 'EVENT_RELEASE_' + type(exc).__name__.upper()
                with self.metrics_lock:
                    self.activity[name] = 'idle'

    def close(self, timeout=65):
        self.stop.set()
        deadline = time.monotonic()+timeout
        for worker in self.workers:
            worker.join(max(0,deadline-time.monotonic()))
        with self.metrics_lock:
            remaining = list(self.processes.values())
        for process in remaining:
            if process.poll() is None:
                process.kill()
                process.wait(timeout=3)
        for worker in self.workers:
            worker.join(1)
        if self.monitor is not None:
            self.monitor.join(4)
        if self.http is not None:
            self.http.should_exit = True
        if self.http_thread is not None:
            self.http_thread.join(3)
        okay = (not any(w.is_alive() for w in self.workers) and not remaining
                and (self.monitor is None or not self.monitor.is_alive())
                and (self.http_thread is None or not self.http_thread.is_alive()))
        if okay and self.lock is not None:
            self.store.close()
            self.lock.close()
        return okay