"""Worker entry point (Layer 4, spec 10).

One chunk per worker process. The worker reads a byte range of the source
(read-only), parses it to canonical records, writes a staging parquet, fsyncs,
computes the content id and physical SHA-256, renames into the final ticks
directory and returns a stats payload. Workers never share memory; they
communicate with the parent only through files and their return value.
"""

import json
import os
import time

from . import versions
from .canonical import (
    tick_chunk_content_id, preservation_chunk_content_id, serialize_preservation,
)
from .config import validate_config
from .parse import parse_chunk
from .storage import ticks_chunk_path, ticks_staging_path, malformed_chunk_path, preservation_chunk_path, write_ticks_chunk
from .util import (
    read_binary_range, sha256_file, sha256_bytes,
)

UTF8_BOM = b"\xef\xbb\xbf"


def _serialize_keys(records):
    """(ts_ms, bid_u, ask_u, vol) per canonical serialization rule (spec 8.5)."""
    return [(r[0], r[1], r[2], r[4]) for r in records]


def _now_iso():
    import datetime
    return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z")


def worker_main(config_path, source_path, out_root, source_id,
                chunk_index, byte_start, byte_end, global_line_start,
                expect_header, certify_time_ms):
    """Spawn-safe worker body. Returns a stats dict (picklable)."""
    with open(config_path, "r", encoding="utf-8") as fh:
        cfg = json.load(fh)
    validate_config(cfg)

    started = time.time()
    data = read_binary_range(source_path, byte_start, byte_end)
    if chunk_index == 0 and data.startswith(UTF8_BOM):
        data = data[len(UTF8_BOM):]

    records, malformed, counters, flags = parse_chunk(
        data, chunk_index=chunk_index, byte_start=byte_start,
        global_line_start=global_line_start, cfg=cfg,
        expect_header=expect_header, certify_time_ms=certify_time_ms)

    # Persist malformed sidecar (jsonl) before the canonical file rename so a
    # crash mid-write leaves only an orphan .tmp (spec 10.4).
    os.makedirs(os.path.dirname(ticks_staging_path(out_root, chunk_index)),
                exist_ok=True)
    if malformed:
        mp = malformed_chunk_path(out_root, chunk_index)
        os.makedirs(os.path.dirname(mp), exist_ok=True)
        with open(mp, "wb") as fh:
            for rec in malformed:
                fh.write((json.dumps(rec, sort_keys=True,
                                     separators=(",", ":")) + "\n").encode("utf-8"))
            fh.flush()
            os.fsync(fh.fileno())

    # Source-preservation layer (spec 8.6): the G_TICKSTORY_MT5 ``last``
    # column is persisted per canonical row before the canonical rename, so a
    # crash mid-write leaves only an orphan .tmp. Dotted runs write nothing.
    preservation_content_id = None
    preservation_sha256 = None
    if cfg["source_grammar"] == versions.GRAMMAR_TICKSTORY_MT5 and records:
        last_u_list = flags.get("last_u") or []
        if len(last_u_list) == len(records):
            pp = preservation_chunk_path(out_root, chunk_index)
            os.makedirs(os.path.dirname(pp), exist_ok=True)
            payload = b"".join(
                serialize_preservation(r[5], lu)
                for r, lu in zip(records, last_u_list))
            with open(pp, "wb") as fh:
                fh.write(payload)
                fh.flush()
                os.fsync(fh.fileno())
            preservation_content_id = preservation_chunk_content_id(
                [(r[5], lu) for r, lu in zip(records, last_u_list)])
            preservation_sha256 = sha256_file(pp)
    staging = ticks_staging_path(out_root, chunk_index)
    write_ticks_chunk(staging, records, cfg, source_id, chunk_index)
    content_id = tick_chunk_content_id(_serialize_keys(records))
    physical_sha = sha256_file(staging)
    os.replace(staging, ticks_chunk_path(out_root, chunk_index))

    stats = {
        "chunk_index": chunk_index,
        "rows_parsed": flags["rows_parsed"],
        "is_header": flags["has_header"],
        "canonical_rows": len(records),
        "malformed_count": len(malformed),
        "preservation_content_id": preservation_content_id,
        "preservation_sha256": preservation_sha256,
        "malformed": counters,
        "content_id": content_id,
        "file_sha256": physical_sha,
        "elapsed_sec": time.time() - started,
        "ts_utc": _now_iso(),
    }
    return stats