"""Local educational inbox/reconciler; no Amazon, EventBridge or SQS client."""

from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime
import json
import math
from pathlib import Path
import re
import sqlite3
from typing import Callable


MAX_JSON_BYTES = 65_536
MAX_DEPTH = 20
KEY_FIELDS = ("seller_id", "sku", "marketplace_id")
EVENT_FIELDS = {
    "schema_version", "source", "subscription", "notification_id",
    *KEY_FIELDS, "notification_type", "event_time", "payload",
}
KINDS = {"listing_status_changed", "listing_issues_changed"}


class InvalidMessage(ValueError):
    """Not the exact normalized educational input contract."""


class IdentityConflict(ValueError):
    """One notification identity was reused with different normalized content."""


class WrongKey(ValueError):
    """A read result belongs to a different seller/SKU/marketplace."""


def _pairs(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise InvalidMessage("duplicate JSON object member")
        result[key] = value
    return result


def _tree(value, depth=0):
    if depth > MAX_DEPTH:
        raise InvalidMessage("JSON nesting exceeds the lab bound")
    if value is None or type(value) in (bool, int):
        return
    if type(value) is float:
        if not math.isfinite(value):
            raise InvalidMessage("non-finite JSON number")
    elif type(value) is str:
        try:
            value.encode("utf-8")
        except UnicodeError:
            raise InvalidMessage("invalid Unicode string") from None
    elif type(value) is list:
        for item in value:
            _tree(item, depth + 1)
    elif type(value) is dict:
        for key, item in value.items():
            if type(key) is not str:
                raise InvalidMessage("JSON object keys must be strings")
            _tree(key, depth + 1)
            _tree(item, depth + 1)
    else:
        raise InvalidMessage("value is not JSON")


def _canonical(value):
    _tree(value)
    try:
        encoded = json.dumps(value, sort_keys=True, separators=(",", ":"),
                             ensure_ascii=True, allow_nan=False)
    except (ValueError, RecursionError):
        raise InvalidMessage("invalid JSON value") from None
    if len(encoded.encode("utf-8")) > MAX_JSON_BYTES:
        raise InvalidMessage("canonical JSON exceeds the lab byte bound")
    return encoded


def parse_event(raw: str) -> tuple[dict, str]:
    """Validate a normalized JSON event, preserving key strings exactly."""
    try:
        if type(raw) is not str or len(raw.encode("utf-8")) > MAX_JSON_BYTES:
            raise InvalidMessage("input must be bounded UTF-8 JSON text")
        event = json.loads(raw, object_pairs_hook=_pairs)
    except (UnicodeError, ValueError, RecursionError):
        raise InvalidMessage("invalid, duplicate-member or oversized JSON") from None
    if type(event) is not dict or set(event) != EVENT_FIELDS:
        raise InvalidMessage("normalized event fields are missing or unknown")
    if type(event["schema_version"]) is not int or event["schema_version"] != 1:
        raise InvalidMessage("unsupported educational schema_version")
    for name in ("source", "subscription", "notification_id", *KEY_FIELDS):
        value = event[name]
        if (type(value) is not str or not 1 <= len(value) <= 256
                or value != value.strip() or any(ord(c) < 32 for c in value)):
            raise InvalidMessage(f"{name} must be an explicit nonempty bounded key")
    if (type(event["notification_type"]) is not str
            or event["notification_type"] not in KINDS):
        raise InvalidMessage("unknown normalized notification_type")
    stamp = event["event_time"]
    if type(stamp) is not str or not re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ", stamp):
        raise InvalidMessage("event_time must use YYYY-MM-DDTHH:MM:SSZ")
    try:
        datetime.strptime(stamp, "%Y-%m-%dT%H:%M:%SZ")
    except ValueError:
        raise InvalidMessage("invalid event_time") from None
    if type(event["payload"]) is not dict:
        raise InvalidMessage("payload must be a JSON object")
    return event, _canonical(event)


@dataclass(frozen=True, order=True)
class ListingKey:
    seller_id: str
    sku: str
    marketplace_id: str

    @classmethod
    def from_event(cls, event):
        return cls(*(event[name] for name in KEY_FIELDS))

    def values(self):
        return self.seller_id, self.sku, self.marketplace_id


@dataclass(frozen=True)
class Ticket:
    key: ListingKey
    generation: int


@dataclass(frozen=True)
class ReadResult:
    key: ListingKey
    document: dict


@dataclass(frozen=True)
class IngestResult:
    disposition: str
    may_ack: bool  # Logical queue decision only; this module never acknowledges.


SCHEMA = """
CREATE TABLE IF NOT EXISTS inbox (
    source TEXT NOT NULL,
    subscription TEXT NOT NULL,
    notification_id TEXT NOT NULL,
    canonical_event TEXT NOT NULL,
    PRIMARY KEY (source, subscription, notification_id)
);
CREATE TABLE IF NOT EXISTS work (
    seller_id TEXT NOT NULL,
    sku TEXT NOT NULL,
    marketplace_id TEXT NOT NULL,
    generation INTEGER NOT NULL CHECK (typeof(generation) = 'integer' AND generation > 0),
    applied_generation INTEGER NOT NULL DEFAULT 0
        CHECK (applied_generation >= 0 AND applied_generation <= generation),
    snapshot_json TEXT,
    PRIMARY KEY (seller_id, sku, marketplace_id)
);
"""


class Store:
    """One reconciliation worker; ingestion may happen during its read.

    Tickets are generation comparisons, not claims/leases or security tokens.
    Use a local SQLite database and a separate connection for each ingester.
    """

    def __init__(self, path: str | Path):
        self._db = sqlite3.connect(path, isolation_level=None, timeout=5)
        self._db.row_factory = sqlite3.Row
        self._db.execute("PRAGMA journal_mode=WAL")
        self._db.execute("PRAGMA synchronous=FULL")
        self._db.executescript(SCHEMA)

    def __enter__(self):
        return self

    def __exit__(self, *_):
        self.close()

    def close(self):
        self._db.close()

    @contextmanager
    def _transaction(self):
        self._db.execute("BEGIN IMMEDIATE")
        try:
            yield
            self._db.execute("COMMIT")
        except BaseException:
            if self._db.in_transaction:
                self._db.execute("ROLLBACK")
            raise

    def ingest(self, raw: str) -> IngestResult:
        event, canonical = parse_event(raw)
        identity = tuple(event[k] for k in ("source", "subscription", "notification_id"))
        key = ListingKey.from_event(event)
        with self._transaction():
            prior = self._db.execute(
                "SELECT canonical_event FROM inbox WHERE source=? AND subscription=? AND notification_id=?",
                identity,
            ).fetchone()
            if prior is not None:
                if prior["canonical_event"] != canonical:
                    raise IdentityConflict("same identity has different normalized content; do not ack")
                disposition = "duplicate"
            else:
                self._db.execute("INSERT INTO inbox VALUES (?, ?, ?, ?)", (*identity, canonical))
                self._db.execute("""
                    INSERT INTO work (seller_id, sku, marketplace_id, generation)
                    VALUES (?, ?, ?, 1)
                    ON CONFLICT (seller_id, sku, marketplace_id)
                    DO UPDATE SET generation = generation + 1
                """, key.values())
                disposition = "stored"
        # The transaction has committed before this logical decision is returned.
        return IngestResult(disposition, may_ack=True)

    def state(self, key: ListingKey) -> dict | None:
        row = self._db.execute("""
            SELECT generation, applied_generation, snapshot_json FROM work
            WHERE seller_id=? AND sku=? AND marketplace_id=?
        """, key.values()).fetchone()
        if row is None:
            return None
        return {
            "generation": row["generation"],
            "applied_generation": row["applied_generation"],
            "dirty": row["generation"] > row["applied_generation"],
            "snapshot": json.loads(row["snapshot_json"]) if row["snapshot_json"] is not None else None,
        }

    def dirty_keys(self) -> list[ListingKey]:
        return [ListingKey(*row) for row in self._db.execute("""
            SELECT seller_id, sku, marketplace_id FROM work
            WHERE generation > applied_generation ORDER BY seller_id, sku, marketplace_id
        """)]

    def inbox_count(self) -> int:
        return self._db.execute("SELECT count(*) FROM inbox").fetchone()[0]

    def begin_refresh(self, key: ListingKey) -> Ticket | None:
        state = self.state(key)
        return Ticket(key, state["generation"]) if state and state["dirty"] else None

    def complete(self, ticket: Ticket, result: ReadResult) -> bool:
        """Atomically save this local read and clear only the observed generation."""
        if ticket.key != result.key:
            raise WrongKey("read key does not match ticket key")
        if type(result.document) is not dict:
            raise InvalidMessage("read document must be a JSON object")
        document = _canonical(result.document)
        with self._transaction():
            updated = self._db.execute("""
                UPDATE work SET snapshot_json=?, applied_generation=?
                WHERE seller_id=? AND sku=? AND marketplace_id=?
                  AND generation=? AND applied_generation < ?
            """, (document, ticket.generation, *ticket.key.values(),
                  ticket.generation, ticket.generation)).rowcount
        return updated == 1


def refresh_once(store: Store, key: ListingKey, reader: Callable[[ListingKey], ReadResult]) -> str:
    """Read outside the transaction. Read failures propagate and leave work dirty."""
    ticket = store.begin_refresh(key)
    if ticket is None:
        return "clean"
    result = reader(key)
    return "applied" if store.complete(ticket, result) else "superseded"