"""Runnable demo ingest host: `python -m cl.http`.

Self-contained, honest-label deployment for a demo-trust first live test (e.g. Tali's fork-(a)):
  - in-memory journal (swap for the Postgres store in a durable deploy),
  - self-attested checkpoint anchor (durable JSONL log; label: "self-attested log, no external witness yet"),
  - demo-trust token allowlist (NOT production identity),
  - background auto-anchor on a cadence.

Env:
  CL_INGEST_HOST          bind host (default 127.0.0.1)
  CL_INGEST_PORT          bind port (default 8080)
  CL_CHECKPOINT_LOG       checkpoint JSONL path (default ./checkpoint.log)
  CL_CHECKPOINT_SEED_HEX  32-byte signer seed, hex (default a fixed demo seed — ROTATE before real exposure)
  CL_DEMO_TOKEN           demo-trust bearer token (default "demo-token")
  CL_DEMO_TENANT          demo tenant (default "tenant:demo")
  CL_DEMO_SUB             demo participant/subject (default "urn:participant:demo")
  CL_CONNECTOR_ID         connector id (default "demo-connector")
  CL_CONSEQUENTIAL_TYPES  comma-separated consequential event types (default "payment.execute")
  CL_AUTO_ANCHOR_S        auto-anchor cadence seconds, 0 disables (default 60)
  CL_ANCHOR_MAX_LEAVES    hard cap on leaves fetched+folded per anchor round (default 10000;
                          values < 1 = refuse to start; a tick drains ceil(backlog/cap) rounds)

Anchor backend selection (LIVE_NEO_ON_HOST_FLIP_PLAN §B — the deploy is a config act; the merge changes
nothing by itself):
  CL_ANCHOR_BACKEND       checkpoint | dual | neo | neova | neova2 (default "checkpoint"; unknown = fail-fast)
  CL_NEOVA2_GATEWAY_SEED_HEX / CL_NEOVA2_GATEWAY_ID_HEX   REQUIRED in neova2 (owner-set; NO default)
  CL_NEOVA2_DOOR_HOST / CL_NEOVA2_DOOR_PORT               live NEOva2 door (default 127.0.0.1:7600)
  CL_NEOVA2_GATEWAY_EPOCH / CL_NEOVA2_SEQ_PATH / CL_NEOVA2_TIMEOUT_S   optional knobs
  CL_NEOVA2_PROOF_URL                readback leg (proof server root URL); REQUIRED for TLS anchoring
  CL_NEOVA2_PINNED_QWS_PATH          pinned qualified-witness registry JSON (out-of-band trust
                                     anchor); REQUIRED whenever CL_NEOVA2_PROOF_URL is set
  CL_NEOVA2_READBACK_POLL_S / CL_NEOVA2_READBACK_INTERVAL_S   readback poll knobs (default 120 / 3)
  CL_NEOVA_SUBMIT_URL / CL_NEOVA_SERVE_URL   REQUIRED in neova — the two public seam surfaces
  CL_NEOVA_GATEWAY_SEED_HEX                  REQUIRED in neova (owner-set; NO default)
  CL_NEOVA_GENESIS_PATH                      REQUIRED in neova — pinned genesis the verifier judges against
  CL_NEOVA_NETWORK_ID / CL_NEOVA_NAMESPACE / CL_NEOVA_GATEWAY_ID / CL_NEOVA_SCOPE_KEY_HEX
  CL_NEOVA_ERA / CL_NEOVA_SEQ_PATH / CL_NEOVA_POLL_TIMEOUT_S   optional knobs (safe defaults)
  CL_DUAL_ROUND_LOG       dual-mode round-log JSONL path (default ./dual_anchor_rounds.jsonl)
  CL_NEO_GATEWAY_SEED_HEX REQUIRED in dual/neo (owner-set; NO default — missing = refuse to start)
  CL_NEO_TRANSPORT_KEY    REQUIRED in dual/neo (owner-set; NO default — missing = refuse to start)
  CL_NEO_HOST/PORT        co-located node API (default 127.0.0.1:8080)
  CL_NEO_CHAIN_ID_HEX / CL_NEO_POLICY_SNAPSHOT_ID_HEX / CL_NEO_NAMESPACE_HEX / CL_NEO_ACCOUNT_ID_HEX
                          devnet identifiers (defaults in cl.http.dual_anchor)
  CL_NEO_HEIGHT_MARGIN / CL_NEO_POLL_TIMEOUT_S / CL_NEO_NODE_COUNT   height targeting + polling knobs
"""

from __future__ import annotations

import logging
import os
import threading

from cl.agent_seam.bridge import ConnectorProfile
from cl.anchor.checkpoint import CheckpointAnchorBackend
from cl.canonical.hash_profile import DEFAULT_PROFILE
from cl.http.app import create_app
from cl.http.identity import make_claims, static_token_verifier
from cl.http.service import IngestConfig, IngestService
from cl.store.memory import InMemoryJournal

_DEMO_SEED = bytes.fromhex("00" * 31 + "01")  # FIXED demo seed — deterministic; ROTATE before real exposure.


def build_service() -> IngestService:
    from cl.http.dual_anchor import build_host_anchor_backend

    seed_hex = os.environ.get("CL_CHECKPOINT_SEED_HEX")
    seed = bytes.fromhex(seed_hex) if seed_hex else _DEMO_SEED
    checkpoint = CheckpointAnchorBackend.from_seed(
        seed, log_path=os.environ.get("CL_CHECKPOINT_LOG", "./checkpoint.log")
    )
    backend = build_host_anchor_backend(
        os.environ.get("CL_ANCHOR_BACKEND", "checkpoint").strip().lower(), checkpoint
    )
    connector_id = os.environ.get("CL_CONNECTOR_ID", "demo-connector")
    consequential = tuple(
        t.strip() for t in os.environ.get("CL_CONSEQUENTIAL_TYPES", "payment.execute").split(",") if t.strip()
    )
    profile = ConnectorProfile(
        connector_id=connector_id,
        consequential_event_types=frozenset(consequential),
    )
    token = os.environ.get("CL_DEMO_TOKEN", "demo-token")
    claims = make_claims(
        sub=os.environ.get("CL_DEMO_SUB", "urn:participant:demo"),
        tenant=os.environ.get("CL_DEMO_TENANT", "tenant:demo"),
    )
    # PR-0: authority-receipt trust anchor. Unset env = None = the legacy
    # caller-supplied path (byte-identical no-op deploy); set = signer resolved
    # server-side + validity window enforced with a real UTC clock.
    from datetime import datetime, timezone

    from cl.agent_seam.signer_registry import registry_from_env

    signer_registry = registry_from_env()
    # Bounded-batch design v0.1: hard cap on leaves per anchor round. FAIL-FAST on < 1 — a host
    # asked to run with an unbounded (or nonsense) round must refuse to start, never silently
    # fall back to the old unbounded behaviour.
    max_leaves_raw = os.environ.get("CL_ANCHOR_MAX_LEAVES", "10000").strip()
    try:
        max_anchor_leaves = int(max_leaves_raw)
    except ValueError:
        raise RuntimeError(
            f"CL_ANCHOR_MAX_LEAVES must be an integer >= 1, got {max_leaves_raw!r}")
    if max_anchor_leaves < 1:
        raise RuntimeError(
            f"CL_ANCHOR_MAX_LEAVES must be >= 1 (unbounded anchor rounds are not a legal "
            f"production state), got {max_anchor_leaves}")
    profiles = {connector_id: profile}
    token_verifier = static_token_verifier({token: claims})

    dsn = os.environ.get("CL_POSTGRES_DSN", "").strip()
    if dsn:
        # Phase 4: durable deploy — Postgres journal + durable anchored-batch
        # store; the service seeds its anchoring watermark from the store at
        # boot (no `-1` reset, no full-journal re-anchor after a crash).
        import psycopg  # optional extra `postgres`; fail-fast if missing

        from cl.connectors import ConnectorRegistry, StoreBackedProfiles, store_token_verifier
        from cl.store.anchor_batches import PostgresAnchorBatchStore
        from cl.store.postgres import PostgresStorage, migrate

        connection = psycopg.connect(dsn)
        migrate(connection)
        # Slice 2 (staza C): membership is data. The registry composes AROUND the
        # static demo pair — demo first, store second — so standing this up cannot
        # change the behaviour of a single existing caller. New members and
        # revocations take effect on the next request, never via a restart.
        #
        # Its OWN connection, deliberately: the registry commits/rolls back on every
        # lookup, on the hot path of every request. Shared with the journal's
        # connection under the threadpool, a registry commit from thread B could
        # prematurely commit thread A's half-written journal entry — an integrity
        # hole. One extra Postgres connection buys full transaction isolation.
        registry = ConnectorRegistry.connect(dsn)
        token_verifier = store_token_verifier(registry, fallback=token_verifier)
        profiles = StoreBackedProfiles(
            profiles, registry, consequential_event_types=frozenset(consequential))

    config = IngestConfig(
        hash_profile_id=DEFAULT_PROFILE.id(),
        profiles=profiles,
        token_verifier=token_verifier,
        signer_registry=signer_registry,
        authority_now=(lambda: datetime.now(timezone.utc).isoformat())
        if signer_registry is not None else None,
        max_anchor_leaves=max_anchor_leaves,
    )
    if dsn:
        return IngestService(
            config, PostgresStorage(connection), backend,
            anchor_batch_store=PostgresAnchorBatchStore(connection))
    return IngestService(config, InMemoryJournal(), backend)


def run_drain_tick(service: IngestService) -> int:
    """One auto-anchor tick: drain the backlog FOUND AT TICK START as a sequence of bounded rounds.

    The round count is computed from a snapshot of the backlog, never re-read mid-tick (owner
    correction to bounded-batch design v0.1): an open `while anchor_pending()` loop could run
    forever whenever new events arrive at least as fast as anchoring — a tick must always
    terminate, and events that arrive during the tick simply belong to the next one. There is no
    N-per-cadence throughput ceiling: a backlog of B drains in ceil(B / max_anchor_leaves)
    back-to-back bounded rounds within the same tick. Returns the number of batches anchored."""
    backlog = service.anchor_backlog()
    rounds = (backlog + service.max_anchor_leaves - 1) // service.max_anchor_leaves
    anchored = 0
    for _ in range(rounds):
        if service.anchor_pending() is None:
            break
        anchored += 1
    return anchored


def _start_auto_anchor(service: IngestService, period_s: float) -> None:
    if period_s <= 0:
        return

    def loop() -> None:
        timer = threading.Event()
        while not timer.wait(period_s):
            try:
                run_drain_tick(service)
            except Exception as error:  # noqa: BLE001
                # Phase 4: the host must not die on a transient anchor
                # error — but a drop is NEVER silent. The durable watermark
                # guarantees the same span is retried next cadence.
                service.anchor_failures += 1
                logging.getLogger("cl.host").warning(
                    "anchor_pending failed (retry next cadence, failure "
                    "#%d): %s", service.anchor_failures, error)

    threading.Thread(target=loop, daemon=True).start()


def main() -> None:
    import uvicorn

    service = build_service()
    _start_auto_anchor(service, float(os.environ.get("CL_AUTO_ANCHOR_S", "60")))
    app = create_app(service)
    uvicorn.run(
        app,
        host=os.environ.get("CL_INGEST_HOST", "127.0.0.1"),
        port=int(os.environ.get("CL_INGEST_PORT", "8080")),
    )


if __name__ == "__main__":
    main()
