"""Independent Tor lifecycle observer; never expands the FD6 HostWatch budget."""
import array
import json
import os
import socket
import struct
from c01_host_watch import now


def exact_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError('duplicate Tor progress key')
        result[key] = value
    return result


class Watch:
    def __init__(self, control, application, pid):
        self.control, self.application, self.pid = control, application, pid
        control.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
        control.setblocking(False)
        self.state, self.sequence, self.last_mutation = 'starting', 0, 0
        self.deadline = now() + 2_000_000_000
        self.reason = None

    def revoke(self, reason):
        if self.state != 'failed':
            self.state, self.reason = 'failed', reason
            try:
                self.application.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
            self.control.close()

    def release_reaped(self):
        self.revoke('reaped')
        self.application.close()

    def receive(self):
        if self.state == 'failed':
            return False
        stamp = now()
        if self.deadline is not None and stamp >= self.deadline:
            self.revoke('deadline')
            return False
        try:
            raw, anc, flags, _ = self.control.recvmsg(
                4097, socket.CMSG_SPACE(12) + socket.CMSG_SPACE(128),
                socket.MSG_DONTWAIT | socket.MSG_CMSG_CLOEXEC)
        except BlockingIOError:
            return True
        except OSError:
            self.revoke('closed')
            return False
        credentials, unexpected = [], False
        for level, kind, data in anc:
            if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS):
                rights = array.array('i')
                rights.frombytes(data[:len(data) - len(data) % rights.itemsize])
                for fd in rights:
                    os.close(fd)
                unexpected = True
            elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS) and len(data) == 12:
                credentials.append(struct.unpack('3i', data))
            else:
                unexpected = True
        try:
            if unexpected or credentials != [(self.pid, 0, 0)] or flags & ~socket.MSG_CMSG_CLOEXEC or not 0 < len(raw) <= 4096:
                raise ValueError('Tor progress peer/framing')
            value = json.loads(raw, object_pairs_hook=exact_object)
            if type(value) is not dict or set(value) != {'version', 'sequence', 'deadline_boottime_ns', 'phase'}:
                raise ValueError('Tor progress fields')
            version, sequence, deadline, phase = (value[k] for k in
                ('version', 'sequence', 'deadline_boottime_ns', 'phase'))
            if type(version) is not int or version != 1 or type(sequence) is not int or not 0 <= sequence < 2**64 or type(deadline) is not int or not 0 <= deadline < 2**64:
                raise ValueError('Tor progress numeric bounds')
            if phase == 'Ready':
                if self.state != 'starting' or sequence != 0 or deadline != 0:
                    raise ValueError('Tor readiness order')
                self.state, self.deadline = 'idle', None
            elif phase == 'Begin':
                if self.state != 'idle' or not stamp < deadline <= stamp + 60_000_000_000:
                    raise ValueError('Tor admission deadline/order')
                if sequence != 0 and sequence < self.last_mutation:
                    raise ValueError('Tor mutation sequence regression')
                self.state, self.sequence, self.deadline = 'busy', sequence, deadline
            elif phase == 'Complete':
                if self.state != 'busy' or (sequence, deadline) != (self.sequence, self.deadline):
                    raise ValueError('Tor completion mismatch')
                if sequence:
                    self.last_mutation = sequence
                self.state, self.deadline = 'idle', None
            else:
                raise ValueError('Tor uncertain or unknown phase')
        except (ValueError, TypeError, UnicodeError):
            self.revoke('protocol or uncertain outcome')
        return self.state != 'failed'