"""Optional independent decoder for local Oak SQLite v1 committed trees.

No Oak code or executable is imported or invoked. Optional blake3/zstandard

packages belong to a pinned reader environment, not the base harness. Unknown

formats and missing instruments are unavailable evidence; corrupt objects fail.

"""

from __future__ import annotations

import hashlib

import importlib.metadata

import json

from pathlib import Path

import re

import sqlite3

import sys

import time

import tempfile

READER_VERSION = 'oak-sqlite-v1-2026-09-18.1'

MAX_OBJECT = 32 * 1024 * 1024

MAX_TOTAL = 128 * 1024 * 1024

MAX_ENTRIES = 8192

MAX_DATABASE = 512 * 1024 * 1024

class Unavailable(Exception):

    pass

class Invalid(Exception):

    pass

def read_committed_tree(repo: Path, destination: Path) -> dict:

    try:

        import blake3

        import zstandard

    except ImportError as exc:

        raise Unavailable('reader dependencies unavailable: blake3 and zstandard required') from exc

    db_path = repo / '.oak' / 'oak.db'

    if (repo.is_symlink() or (repo / '.oak').is_symlink() or db_path.is_symlink()

            or not db_path.is_file()):

        raise Unavailable('local Oak SQLite database missing or linked')

    if destination.exists() or destination.is_symlink():

        raise Invalid('reader destination must not exist')

    deadline = time.monotonic() + 60

    destination.mkdir(parents=True)

    def hashed(raw, expected):

        if not isinstance(expected, str) or not re.fullmatch('[0-9a-f]{64}', expected):

            raise Invalid('invalid object hash')

        if blake3.blake3(raw).hexdigest() != expected:

            raise Invalid('object hash mismatch')

        return raw

    def decode(content, codec):

        if not isinstance(content, bytes) or len(content) > MAX_OBJECT:

            raise Invalid('stored object exceeds reader bound')

        if codec == 0:

            return content

        if codec != 1:

            raise Unavailable('unsupported object codec')

        try:

            size = zstandard.frame_content_size(content)

            if zstandard.get_frame_parameters(content).window_size > MAX_OBJECT:

                raise Invalid('decompression window exceeds reader bound')

            if size not in (-1, zstandard.CONTENTSIZE_UNKNOWN) and not 0 <= size <= MAX_OBJECT:

                raise Invalid('decompressed object exceeds reader bound')

            raw = zstandard.ZstdDecompressor(max_window_size=MAX_OBJECT).decompress(

                content, max_output_size=MAX_OBJECT, allow_extra_data=False)

        except zstandard.ZstdError as exc:

            raise Invalid('invalid compressed object') from exc

        if len(raw) > MAX_OBJECT:

            raise Invalid('decompressed object exceeds reader bound')

        return raw

    def database_snapshot():

        if any(Path(str(db_path) + suffix).exists() for suffix in ('-wal', '-journal')):

            raise Unavailable('active SQLite journal; quiescent database required')

        with db_path.open('rb') as stream:

            content = stream.read(MAX_DATABASE + 1)

        if len(content) > MAX_DATABASE:

            raise Unavailable('database exceeds independent reader bound')

        return content

    # Snapshot only quiescent databases. immutable=1 is safe on this private copy,

    # never on a live WAL database; the original bytes are checked again below.

    snapshot = database_snapshot()

    database_sha256 = hashlib.sha256(snapshot).hexdigest()

    with tempfile.TemporaryDirectory(prefix='oakbench-reader-db-') as temp:

        snapshot_path = Path(temp) / 'snapshot.db'

        snapshot_path.write_bytes(snapshot)

        del snapshot

        return _read_snapshot(snapshot_path, database_sha256, database_snapshot,

                              destination, deadline, hashed, decode, blake3)

def _read_snapshot(snapshot_path, database_sha256, database_snapshot, destination,

                   deadline, hashed, decode, blake3):

    total = 0

    entries_seen = 0

    with sqlite3.connect(snapshot_path.as_uri() + '?mode=ro&immutable=1', uri=True, timeout=5) as db:

        db.execute('PRAGMA query_only=ON')

        db.execute('PRAGMA trusted_schema=OFF')

        db.set_progress_handler(lambda: int(time.monotonic() > deadline), 1000)

        db.execute('BEGIN')

        def metadata(key):

            row = db.execute('SELECT value FROM metadata WHERE key=? AND length(value)<=4096', (key,)).fetchone()

            return row[0] if row else None

        if metadata('hash_format') not in (None, 'v1'):

            raise Unavailable('unsupported Oak hash format')

        head = metadata('head')

        branch = metadata('current_branch')

        if not head or not branch:

            raise Unavailable('committed HEAD unavailable')

        commit = db.execute('SELECT branch_name,parent_hash,merge_parent_hash,manifest_hash,author,message,timestamp FROM commits WHERE hash=? AND length(CAST(branch_name AS BLOB))<=4096 AND coalesce(length(CAST(parent_hash AS BLOB)),0)<=64 AND coalesce(length(CAST(merge_parent_hash AS BLOB)),0)<=64 AND length(CAST(manifest_hash AS BLOB))=64 AND length(CAST(author AS BLOB))<=4096 AND coalesce(length(CAST(message AS BLOB)),0)<=1048576 AND length(CAST(timestamp AS BLOB))<=128', (head,)).fetchone()

        if not commit or any(v is not None and (not isinstance(v, str) or len(v) > MAX_OBJECT) for v in commit):

            raise Invalid('committed HEAD object missing or invalid')

        # Oak v1 stores the same RFC3339 timestamp text used in its hash preimage.

        hashed('\n'.join(v or '' for v in commit).encode(), head)

        root_hash = commit[3]

        def blob(object_hash):

            row = db.execute('SELECT content,size,codec FROM blobs WHERE hash=? AND length(content)<=?', (object_hash, MAX_OBJECT)).fetchone()

            if row is None:

                if db.execute('SELECT 1 FROM blobs WHERE hash=?', (object_hash,)).fetchone():

                    raise Invalid('stored blob exceeds reader bound')

                # Chunk-only hydration is not yet qualified by this reader.

                raise Unavailable('complete blob unavailable; chunk-only storage unsupported')

            content, size, codec = row

            if type(size) is not int or not 0 <= size <= MAX_OBJECT:

                raise Invalid('blob size exceeds reader bound')

            raw = decode(content, codec)

            if len(raw) != size:

                raise Invalid('blob size mismatch')

            return hashed(raw, object_hash)

        def visit(tree_hash, relative, ancestors):

            nonlocal total, entries_seen

            if len(ancestors) >= 32 or tree_hash in ancestors or time.monotonic() > deadline:

                raise Invalid('tree depth, cycle or time bound exceeded')

            if tree_hash == blake3.blake3(b'').hexdigest():

                return

            row = db.execute('SELECT content FROM trees WHERE hash=? AND (content IS NULL OR length(content)<=?)', (tree_hash, MAX_OBJECT)).fetchone()

            if row is None:

                raise Invalid('committed tree missing or oversized')

            if row[0] is None:

                if db.execute('SELECT 1 FROM tree_entries WHERE tree_hash=? AND (length(CAST(name AS BLOB))>4096 OR length(CAST(hash AS BLOB))!=64 OR length(CAST(kind AS BLOB))>4 OR length(CAST(mode AS BLOB))>10) LIMIT 1', (tree_hash,)).fetchone():

                    raise Invalid('stored tree entry exceeds reader bound')

                stored = db.execute('SELECT kind,name,hash,mode FROM tree_entries WHERE tree_hash=? ORDER BY name LIMIT ?', (tree_hash, MAX_ENTRIES + 1)).fetchall()

                raw = '\n'.join('\t'.join((kind, name, obj, 'tree' if kind == 'tree' else mode)) for kind, name, obj, mode in stored).encode()

            else:

                raw = decode(row[0], 1)

            hashed(raw, tree_hash)

            try:

                lines = raw.decode('utf-8').split('\n') if raw else []

            except UnicodeError as exc:

                raise Invalid('invalid tree encoding') from exc

            previous = None

            for line in lines:

                parts = line.split('\t')

                if len(parts) != 4:

                    raise Invalid('invalid tree entry')

                kind, name, object_hash, mode = parts

                if (not name or name in ('.', '..', '.oak', '.git') or any(c in name for c in '/\\\x00\n\r')

                        or (previous is not None and name <= previous)):

                    raise Invalid('unsafe or unordered tree path')

                previous = name

                entries_seen += 1

                if entries_seen > MAX_ENTRIES:

                    raise Invalid('tree entry bound exceeded')

                path = relative / name

                if (destination / path).exists() or (destination / path).is_symlink():

                    raise Unavailable('filesystem path collision on reader host')

                if kind == 'tree' and mode == 'tree':

                    (destination / path).mkdir()

                    visit(object_hash, path, ancestors | {tree_hash})

                elif kind == 'blob' and mode in ('regular', 'executable'):

                    content = blob(object_hash)

                    total += len(content)

                    if total > MAX_TOTAL:

                        raise Invalid('total payload bound exceeded')

                    target = destination / path

                    target.write_bytes(content)

                    target.chmod(0o755 if mode == 'executable' else 0o644)

                else:

                    raise Unavailable('unsupported tree entry kind or mode')

        visit(root_hash, Path(), set())

        db.commit()

        if hashlib.sha256(database_snapshot()).hexdigest() != database_sha256:

            raise Unavailable('database changed during independent read')

    return {'head': head, 'database_sha256': database_sha256, 'reader_version': READER_VERSION,

            'reader_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),

            'blake3_version': importlib.metadata.version('blake3'),

            'zstandard_version': importlib.metadata.version('zstandard')}

def main():

    try:

        if len(sys.argv) != 3:

            raise Invalid('usage: oak_sqlite_reader.py REPOSITORY NEW_DESTINATION')

        result = read_committed_tree(Path(sys.argv[1]), Path(sys.argv[2]))

        code = 0

    except Unavailable as exc:

        result, code = {'unavailable': str(exc)}, 77

    except Invalid as exc:

        result, code = {'corrupt': True, 'reason': str(exc)}, 1

    except (OSError, ValueError, TypeError, sqlite3.Error) as exc:

        result, code = {'unavailable': 'reader instrument error: ' + str(exc)}, 77

    print(json.dumps(result, sort_keys=True))

    return code

if __name__ == '__main__':

    raise SystemExit(main())