"""Crash-readable, fsynced debugger records. No game or transport dependencies."""
import hashlib
import json
import os
from pathlib import Path
import threading
import time


def encode(value):
    return json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()


def durable_json(path, value):
    path = Path(path)
    temp = path.with_suffix(path.suffix + '.tmp')
    with temp.open('wb') as f:
        os.chmod(temp, 0o600)
        f.write(encode(value)); f.flush(); os.fsync(f.fileno())
    temp.replace(path)
    fd = os.open(path.parent, os.O_DIRECTORY)
    try: os.fsync(fd)
    finally: os.close(fd)


class Journal:
    def __init__(self, folder, quota=256*1024*1024, reserve=2*1024*1024):
        self.folder = Path(folder)
        self.folder.mkdir(parents=True, exist_ok=True, mode=0o700)
        self.path = self.folder/'journal.jsonl'
        self.quota, self.reserve = quota, reserve
        self.sequence = 0
        self.operations = {}
        self._lock = threading.RLock()
        for row in self.rows():
            self.sequence = row['sequence']
            if row['kind'] in ('intent', 'result'):
                self.operations[row['payload']['operation_id']] = row['payload']

    def rows(self):
        if not self.path.exists(): return []
        rows = []
        with self.path.open('rb') as f:
            for line in f:
                # A torn tail is evidence, not a valid completed operation. Refuse
                # appending until an explicit recovery preserves and repairs it.
                if not line.endswith(b'\n'):
                    raise ValueError('journal_torn_tail: preserve journal and recover before appending')
                row = json.loads(line)
                if row['sequence'] != len(rows)+1:
                    raise ValueError('journal_sequence_gap')
                rows.append(row)
        return rows

    def append(self, kind, payload, cleanup=False):
        # Cancellation/control requests and the native worker share this writer.
        # Sequence allocation, quota accounting, write and publication are one
        # critical section, not merely individually atomic Python statements.
        with self._lock:
            return self._append(kind, payload, cleanup)

    def _append(self, kind, payload, cleanup=False):
        row = dict(schema_version=1, sequence=self.sequence+1, kind=kind,
                   host_monotonic_ns=time.monotonic_ns(), payload=payload)
        data = encode(row)+b'\n'
        size = self.path.stat().st_size if self.path.exists() else 0
        if size+len(data) > self.quota-(0 if cleanup else self.reserve):
            raise OSError('journal_quota: new work refused; cleanup reserve retained')
        with self.path.open('ab') as f:
            os.chmod(self.path, 0o600)
            f.write(data); f.flush(); os.fsync(f.fileno())
        self.sequence += 1
        if kind in ('intent', 'result'):
            self.operations[payload['operation_id']] = payload
        return row

    def manifest(self):
        with self._lock:
            return self._manifest()

    def _manifest(self):
        data = self.path.read_bytes() if self.path.exists() else b''
        value = dict(schema_version=1, kind='debugger_journal_manifest',
                     journal='journal.jsonl', bytes=len(data), rows=self.sequence,
                     sha256=hashlib.sha256(data).hexdigest())
        durable_json(self.folder/'manifest.json', value)
        return value


def verify(path):
    path = Path(path)
    m = json.loads(path.read_text())
    if m['journal'] != 'journal.jsonl': raise ValueError('manifest journal must be a sibling journal.jsonl')
    if set(m)!={'schema_version','kind','journal','bytes','rows','sha256'} or m['schema_version']!=1 or m['kind']!='debugger_journal_manifest':
        raise ValueError('unsupported journal manifest schema')
    raw = (path.parent/m['journal']).read_bytes()
    if len(raw) != m['bytes'] or hashlib.sha256(raw).hexdigest() != m['sha256']:
        raise ValueError('journal digest/length mismatch')
    rows = Journal(path.parent).rows()
    if len(rows) != m['rows']: raise ValueError('journal row count mismatch')
    pending = set()
    owner=next((r['payload'] for r in rows if r['kind']=='owner'),None)
    for row in rows:
        p = row['payload']
        if row['kind'] in ('intent','result') and owner and p.get('session_id')!=owner['session_id']:
            raise ValueError('operation session identity mismatch')
        if row['kind']=='native_reply' and owner and owner.get('native_session') and p.get('session')!=owner['native_session']:
            raise ValueError('native session identity mismatch')
        if row['kind'] == 'intent':
            if p['operation_id'] in pending: raise ValueError('duplicate operation intent')
            pending.add(p['operation_id'])
        elif row['kind'] == 'result':
            if p['operation_id'] not in pending: raise ValueError('result without intent')
            pending.remove(p['operation_id'])
    return dict(ok=True, rows=len(rows), pending_operations=sorted(pending), manifest=m)


def recover_torn_tail(folder):
    """Call only while holding the owner lock. Preserve every byte before repair."""
    folder=Path(folder); path=folder/'journal.jsonl'
    raw=path.read_bytes()
    if not raw or raw.endswith(b'\n'): return Journal(folder)
    digest=hashlib.sha256(raw).hexdigest()
    backup=folder/('journal-torn-'+digest+'.jsonl')
    with backup.open('xb') as f:
        os.chmod(backup,0o600); f.write(raw); f.flush(); os.fsync(f.fileno())
    prefix=raw[:raw.rfind(b'\n')+1]
    temp=folder/'journal-recovery.tmp'
    with temp.open('wb') as f:
        os.chmod(temp,0o600); f.write(prefix); f.flush(); os.fsync(f.fileno())
    temp.replace(path)
    fd=os.open(folder,os.O_DIRECTORY)
    try: os.fsync(fd)
    finally: os.close(fd)
    journal=Journal(folder)
    journal.append('journal_tail_recovery',dict(preserved_file=backup.name,sha256=digest,
                   original_bytes=len(raw),retained_complete_bytes=len(prefix),discarded_from_active_stream_bytes=len(raw)-len(prefix)),cleanup=True)
    return journal