"""METAPUB-04: active-channel writers and fenced stopped-heartbeat handoff."""

import math
import threading
import time

from app.metadata_publisher import MetadataPublisher, RedisSnapshotSink, camera_key


class ChannelMetadataPublisher(MetadataPublisher):
    def __init__(self, settings, *, sink_factory=None, clock=time.monotonic):
        if settings.worker_mode != 'channel':
            raise ValueError('Channel publisher requires channel mode.')
        self._factory = sink_factory or (lambda key: RedisSnapshotSink(settings))
        super().__init__(settings, clock=clock, sink_factory=lambda _: self._factory('control'))
        self._workers = {}
        self._control_busy = None
        self._worker_errors = {}
        self._terminal_errors = set()
        self._inflight_bytes = {}

    def start(self):
        with self._cv:
            if self._closed:
                raise RuntimeError('Publisher is closed.')
            if self._thread is None:
                self._thread = threading.Thread(target=self._control, name='metadata-control', daemon=True)
                self._thread.start()
                for thread in self._workers.values():
                    thread.start()

    def _reap(self):
        # A finally notification is not proof of thread termination.
        for key, thread in tuple(self._workers.items()):
            if thread.ident is not None and not thread.is_alive():
                del self._workers[key]
                self._metrics['workers_reclaimed'] += 1

    def _same_state(self, key, state):
        if state in ('STARTING', 'RUNNING'):
            thread = self._workers.get(key)
            return bool(thread is not None and (thread.ident is None or thread.is_alive())
                        and key not in self._worker_errors)
        return True

    def _admit(self, key, old, run_id, generation, state):
        self._reap()
        active = state in ('STARTING', 'RUNNING')
        if active:
            thread = self._workers.get(key)
            if thread is not None:
                return bool(old and old['runId'] == run_id and old['runGeneration'] == generation
                            and old['state'] in ('STARTING', 'RUNNING'))
            if key in self._worker_errors or key in self._terminal_errors:
                self._metrics['worker_unavailable'] += 1
                return False
            if len(self._workers) >= self.settings.active_worker_limit:
                self._metrics['worker_limit'] += 1
                return False
            # No old STOPPED write may cross activation and delete the new Frame stream.
            if not self._cv.wait_for(lambda: self._control_busy != key or self._closed,
                                     self.settings.io_timeout * 2 + .1):
                self._metrics['control_handoff_timeout'] += 1
                return False
            if self._closed or self._channels.get(key) is not old:
                return False
            if key in self._worker_errors or key in self._terminal_errors:
                return False
            self._reap()
            if len(self._workers) >= self.settings.active_worker_limit or key in self._workers:
                self._metrics['worker_limit'] += 1
                return False
            thread = threading.Thread(target=self._channel_run, args=(key,),
                                      name='metadata-channel-' + key[:12], daemon=True)
            self._workers[key] = thread
            if self._thread is not None:
                try:
                    thread.start()
                except RuntimeError:
                    del self._workers[key]
                    self._metrics['worker_start_error'] += 1
                    return False
            self._metrics['workers_created'] += 1
        self._lanes[key] = key
        return True

    def _take(self, key, lane):
        at, payload, size = self._pending.pop(key)
        self._bytes -= size
        status = dict(self._channels[key], publisher=dict(self._metrics))
        if payload is not None and self.clock() - at > self.settings.max_age:
            self._metrics['expired'] += 1
            self._profile.count(key, 'expired')
            payload, size = None, 0
        self._inflight_bytes[lane] = size
        token = self._profile.token()
        kind = 'frame' if payload is not None else 'status'
        self._profile.duration(token, key, kind + '_queue', self.clock() - at)
        trace = self._profile.dequeue(key, kind, lane)
        return status, payload, token, kind, trace

    def _write(self, sink, key, lane, item):
        status, payload, token, kind, trace = item
        if token is not None:
            started, cpu_start = time.monotonic_ns(), time.thread_time_ns()
        succeeded = False
        try:
            sink.write(key, status, payload)
            succeeded = True
        except Exception:
            pass
        if token is not None:
            finished, cpu = time.monotonic_ns(), time.thread_time_ns() - cpu_start
        with self._cv:
            accounted = time.monotonic_ns() if token is not None else None
            self._inflight_bytes.pop(lane, None)
            if succeeded:
                self._metrics['sent'] += 1
                self._metrics['sent_' + ('frames' if payload is not None else 'status')] += 1
            else:
                self._metrics['send_error'] += 1
            if token is not None:
                if token == self._profile.token():
                    self._profile.count(key, 'sent_' + kind if succeeded else 'send_error')
                self._profile.duration(token, key, kind + '_write_wall', (finished - started) / 1e9)
                self._profile.duration(token, key, kind + '_write_cpu', cpu / 1e9)
                self._profile.duration(token, key, kind + '_completion_lock', (accounted - finished) / 1e9)
                self._profile.complete(token, trace, started, finished, cpu, accounted)
            if status['state'] in ('STOPPED', 'ERROR'):
                if succeeded:
                    self._terminal_errors.discard(key)
                else:
                    self._terminal_errors.add(key)
            self._cv.notify_all()
        return succeeded

    def _close_sink(self, sink, key):
        try:
            sink.close()
        except Exception:
            with self._cv:
                self._metrics['close_error'] += 1
                self._worker_errors[key] = 'sink_close_failed'

    def _channel_run(self, key):
        sink = None
        try:
            sink = self._factory(key)
            while True:
                with self._cv:
                    self._cv.wait_for(lambda: self._closed or key in self._pending)
                    if self._closed:
                        return
                    item = self._take(key, key)
                self._write(sink, key, key, item)
                if item[0]['state'] in ('STOPPED', 'ERROR'):
                    return
        except Exception:
            with self._cv:
                self._worker_errors[key] = 'worker_failed'
                self._metrics['worker_error'] += 1
        finally:
            if sink is not None:
                self._close_sink(sink, key)
            with self._cv:
                self._cv.notify_all()

    def _eligible_control(self):
        return next((key for key in self._pending if key not in self._workers
                     and self._channels[key]['state'] in ('STOPPED', 'ERROR')), None)

    def _control(self):
        try:
            while True:
                with self._cv:
                    self._reap()
                    self._cv.wait_for(lambda: self._closed or self._eligible_control() is not None,
                                      self.settings.stream_ttl / 3)
                    if self._closed:
                        return
                    self._reap()
                    if self.clock() - self._heartbeat_at >= self.settings.stream_ttl / 3:
                        for key in self._channels:
                            if key not in self._pending:
                                self._queue(key, None, 'heartbeat')
                        self._heartbeat_at = self.clock()
                    key = self._eligible_control()
                    if key is None:
                        continue
                    self._control_busy = key
                    item = self._take(key, 'control')
                try:
                    self._write(self.sink, key, 'control', item)
                finally:
                    with self._cv:
                        self._control_busy = None
                        self._cv.notify_all()
        finally:
            self._close_sink(self.sink, 'control')

    def wait_channel(self, camera_id, run_id, generation, timeout):
        if not math.isfinite(timeout) or timeout < 0:
            raise ValueError('Invalid channel shutdown timeout.')
        key = camera_key(camera_id)
        with self._cv:
            status = self._channels.get(key)
            if not status or (status['runId'], status['runGeneration']) != (run_id, generation):
                return False
            if status['state'] not in ('STOPPED', 'ERROR'):
                return False
            thread = self._workers.get(key)
        if thread is not None and thread.ident is not None:
            thread.join(timeout)
        with self._cv:
            self._reap()
            return key not in self._workers and key not in self._worker_errors and key not in self._terminal_errors

    def metrics(self, include_timeline=False):
        with self._cv:
            self._reap()
            result = super().metrics(include_timeline)
            result.update(workerMode='channel', workers=len(self._workers),
                          workerLimit=self.settings.active_worker_limit,
                          controlWorkers=int(self._thread is not None and self._thread.is_alive()),
                          retiringWorkers=sum(self._channels[k]['state'] in ('STOPPED', 'ERROR')
                                              for k in self._workers),
                          workerErrors=dict(self._worker_errors), terminalErrors=len(self._terminal_errors),
                          inflightBytes=sum(self._inflight_bytes.values()),
                          inflightByteLimit=self.settings.active_worker_limit * self.settings.max_message_bytes)
            return result

    def close(self, timeout=1.0):
        if not math.isfinite(timeout) or timeout < 0:
            raise ValueError('Invalid shutdown timeout.')
        with self._cv:
            self._closed = True
            self._metrics['shutdown_drop'] += len(self._pending)
            self._pending.clear()
            self._bytes = 0
            self._cv.notify_all()
            threads = list(self._workers.values()) + ([self._thread] if self._thread else [])
            for key, thread in tuple(self._workers.items()):
                if thread.ident is None:
                    del self._workers[key]
        deadline = time.monotonic() + timeout
        for thread in threads:
            if thread.ident is not None:
                thread.join(max(0., deadline - time.monotonic()))
        with self._cv:
            self._reap()
            return not any(t.is_alive() for t in threads) and not self._worker_errors