#!/usr/bin/env python3
"""
TNS Deposit Service
===================
Passive SIP signaling capture service that fires TNS Call Event API token
deposits when outbound INVITEs are observed for registered TNs.

This is the Python reference implementation. Production deployments may
use a Go binary build, but this implementation is fully functional for
single-host deployments and lab/POC environments.

Runs as a systemd service. Reads configuration from
/etc/tns-deposit-service/config.yaml.

Author: TNS Solutions/Sales Engineering
License: Proprietary - TNS Internal Use Only
"""

import argparse
import json
import logging
import os
import re
import signal
import sqlite3
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from http.server import HTTPServer, BaseHTTPRequestHandler
from pathlib import Path
from typing import Optional, Set, List

import requests
import yaml
from scapy.all import sniff, IP, UDP, TCP, Raw, conf as scapy_conf


# ============================================================================
# Version
# ============================================================================

VERSION = "0.1.3"
PRODUCT_NAME = "tns-deposit-service"


def format_uptime(seconds: float) -> str:
    """Render a duration in seconds as e.g. '2 days, 8 hours, 35 minutes, 18 seconds'.
    Omits leading zero units (a fresh start reads '47 seconds', not
    '0 days, 0 hours, 0 minutes, 47 seconds'). Pluralizes correctly."""
    total = int(seconds)
    days, rem = divmod(total, 86400)
    hours, rem = divmod(rem, 3600)
    minutes, secs = divmod(rem, 60)
    parts = []
    for value, label in ((days, "day"), (hours, "hour"),
                         (minutes, "minute"), (secs, "second")):
        # Skip leading zero units, but always include at least the seconds.
        if value == 0 and not parts and label != "second":
            continue
        parts.append(f"{value} {label}" + ("s" if value != 1 else ""))
    return ", ".join(parts)


# ============================================================================
# Configuration
# ============================================================================

@dataclass
class CaptureConfig:
    deployment_mode: str = "native_host"  # "native_host" | "span_tap"
    interface: str = "any"
    sip_ports: List[int] = field(default_factory=lambda: [5060])
    target_ips: List[str] = field(default_factory=list)
    invite_only: bool = True
    promiscuous: bool = False


@dataclass
class TNSConfig:
    api_url: str = ""
    api_timeout_sec: float = 2.0
    api_retries: int = 1


@dataclass
class RegistryConfig:
    mode: str = "registry"  # "registry" (use list) | "all_outbound" (fire on every captured outbound INVITE)
    source: str = "config"  # "config" | "sqlite" | "postgres" | "mysql" - only consulted when mode=registry
    config_tns: List[str] = field(default_factory=list)
    sqlite_path: str = ""
    db_connection: str = ""
    db_query: str = "SELECT tn FROM tns_registered_tns"
    refresh_interval_sec: int = 300


@dataclass
class ServiceConfig:
    log_level: str = "INFO"
    log_format: str = "json"  # "json" | "text"
    max_workers: int = 10
    health_bind: str = "127.0.0.1:8080"
    metrics_enabled: bool = True


@dataclass
class Config:
    capture: CaptureConfig
    tns: TNSConfig
    registry: RegistryConfig
    service: ServiceConfig
    warnings: List[str] = field(default_factory=list)


def load_config(path: str) -> Config:
    """Load and validate YAML config from disk."""
    if not os.path.exists(path):
        raise FileNotFoundError(f"Config file not found: {path}")

    with open(path) as f:
        raw = yaml.safe_load(f) or {}

    cap = raw.get("capture", {})
    tns = raw.get("tns", {})
    reg = raw.get("registry", {})
    svc = raw.get("service", {})

    cfg = Config(
        capture=CaptureConfig(
            deployment_mode=cap.get("deployment_mode", "native_host"),
            interface=cap.get("interface", "any"),
            sip_ports=cap.get("sip_ports", [5060]),
            target_ips=cap.get("target_ips", []),
            invite_only=cap.get("invite_only", True),
            promiscuous=cap.get("promiscuous", False),
        ),
        tns=TNSConfig(
            api_url=tns.get("api_url", ""),
            api_timeout_sec=tns.get("api_timeout_sec", 2.0),
            api_retries=tns.get("api_retries", 1),
        ),
        registry=RegistryConfig(
            mode=reg.get("mode", "registry"),
            source=reg.get("source", "config"),
            config_tns=reg.get("config_tns", []),
            sqlite_path=reg.get("sqlite_path", ""),
            db_connection=reg.get("db_connection", ""),
            db_query=reg.get("db_query", "SELECT tn FROM tns_registered_tns"),
            refresh_interval_sec=reg.get("refresh_interval_sec", 300),
        ),
        service=ServiceConfig(
            log_level=svc.get("log_level", "INFO"),
            log_format=svc.get("log_format", "json"),
            max_workers=svc.get("max_workers", 10),
            health_bind=svc.get("health_bind", "127.0.0.1:8080"),
            metrics_enabled=svc.get("metrics_enabled", True),
        ),
    )

    # ---- Hard validation (refuse to start) ----
    errors = []
    warnings = []

    if cfg.capture.deployment_mode not in ("native_host", "span_tap"):
        errors.append(
            f"capture.deployment_mode must be 'native_host' or 'span_tap', "
            f"got: {cfg.capture.deployment_mode!r}"
        )

    if not cfg.tns.api_url:
        errors.append("tns.api_url is required")

    if cfg.registry.source == "sqlite" and not cfg.registry.sqlite_path:
        errors.append("registry.sqlite_path required when source=sqlite")
    if cfg.registry.source in ("postgres", "mysql") and not cfg.registry.db_connection:
        errors.append("registry.db_connection required when source=postgres|mysql")

    # ---- registry.mode validation ----
    if cfg.registry.mode not in ("registry", "all_outbound"):
        errors.append(
            f"registry.mode must be 'registry' or 'all_outbound', "
            f"got: {cfg.registry.mode!r}"
        )

    if cfg.registry.mode == "all_outbound":
        # Hard fail: all_outbound without target_ips would fire on every captured
        # INVITE, including internal/inter-extension SIP. The "trust the outbound
        # vantage" safety argument requires that vantage to be explicitly defined.
        if not cfg.capture.target_ips:
            errors.append(
                "registry.mode=all_outbound requires capture.target_ips to be "
                "populated. Without target_ips, the service would fire deposits "
                "on EVERY captured INVITE (including internal signaling). "
                "Populate target_ips with the carrier-facing trunk IP(s) to "
                "define the outbound vantage this mode trusts."
            )
        # Hard fail: span_tap + all_outbound is not supported in 0.1.3. SPAN
        # vantages can see broader traffic than a trunk edge; the "outbound = in
        # scope" argument doesn't necessarily hold. Revisit if a real use case
        # appears.
        if cfg.capture.deployment_mode == "span_tap":
            errors.append(
                "registry.mode=all_outbound is not supported with "
                "deployment_mode=span_tap (a SPAN/mirror vantage can see broader "
                "traffic than a dedicated trunk edge; the all_outbound safety "
                "argument may not hold). Use mode=registry on span_tap deployments."
            )
        # Soft warning: list configured but ignored is operator confusion.
        if cfg.registry.config_tns or cfg.registry.sqlite_path or cfg.registry.db_connection:
            warnings.append(
                "registry.mode=all_outbound: an explicit TN list/source is "
                "configured but will be IGNORED in this mode. Deposits will "
                "fire for every captured outbound INVITE matching target_ips."
            )

    # ---- Mode-specific validation ----
    # SPAN tap requires promiscuous mode to see frames not destined for the local MAC.
    if cfg.capture.deployment_mode == "span_tap" and not cfg.capture.promiscuous:
        errors.append(
            "deployment_mode=span_tap requires capture.promiscuous=true "
            "(SPAN/mirror destinations receive frames addressed to other MACs)"
        )

    # SPAN tap with interface=any is technically valid but usually a misconfiguration.
    if cfg.capture.deployment_mode == "span_tap" and cfg.capture.interface == "any":
        warnings.append(
            "deployment_mode=span_tap with interface=any: consider specifying the "
            "exact mirror destination interface (e.g., 'eth1') for predictable capture."
        )

    # Empty target_ips is a correctness concern in BOTH deployment modes.
    if not cfg.capture.target_ips:
        warnings.append(
            "capture.target_ips is empty: the service will fire deposits on EVERY "
            "captured INVITE, including internal/phone-side signaling that is not "
            "outbound PSTN traffic. Populate target_ips with the IP addresses of the "
            "carrier-facing SBC/trunk interfaces to ensure deposits fire only on the "
            "correct call leg. Acceptable only if the capture point exclusively sees "
            "outbound PSTN signaling (e.g., a dedicated trunk-facing interface)."
        )

    # Native host with promiscuous mode is unusual.
    if cfg.capture.deployment_mode == "native_host" and cfg.capture.promiscuous:
        warnings.append(
            "deployment_mode=native_host with promiscuous=true is unusual. "
            "Native deployments typically capture local traffic and do not need promiscuous mode."
        )

    if errors:
        raise ValueError(
            "Config validation failed:\n  - " + "\n  - ".join(errors)
        )

    cfg.warnings = warnings
    return cfg


# ============================================================================
# BPF filter generation
# ============================================================================

def build_bpf_filter(cap: CaptureConfig) -> str:
    """
    Build a Berkeley Packet Filter expression for kernel-level capture filtering.

    INVITE-only filter uses payload byte matching: bytes 0-3 of UDP payload
    are 'I','N','V','I' (0x49 0x4E 0x56 0x49 = 0x494E5649).
    """
    parts = []

    if len(cap.sip_ports) == 1:
        parts.append(f"udp port {cap.sip_ports[0]}")
    else:
        port_expr = " or ".join(f"udp port {p}" for p in cap.sip_ports)
        parts.append(f"({port_expr})")

    if cap.target_ips:
        ip_expr = " or ".join(f"dst host {ip}" for ip in cap.target_ips)
        parts.append(f"({ip_expr})")

    if cap.invite_only:
        parts.append("udp[8:4] = 0x494e5649")

    return " and ".join(parts)


# ============================================================================
# TN Registry — manages the set of registered TNs to fire deposits for
# ============================================================================

class TNRegistry:
    """Thread-safe registered TN set with periodic refresh from data source."""

    def __init__(self, cfg: RegistryConfig, log: logging.Logger, metrics: Optional["Metrics"] = None):
        self.cfg = cfg
        self.log = log
        self.metrics = metrics
        self._lock = threading.RLock()
        self._tns: Set[str] = set()
        self._last_refresh = 0.0
        self._stop = threading.Event()
        self._refresh_thread: Optional[threading.Thread] = None

    def start(self):
        self.refresh()
        if self.cfg.source != "config":
            self._refresh_thread = threading.Thread(
                target=self._refresh_loop, daemon=True, name="tn-registry-refresh"
            )
            self._refresh_thread.start()

    def stop(self):
        self._stop.set()
        if self._refresh_thread:
            self._refresh_thread.join(timeout=2.0)

    def contains(self, tn: str) -> bool:
        with self._lock:
            return tn in self._tns

    def size(self) -> int:
        with self._lock:
            return len(self._tns)

    def snapshot(self) -> list:
        """Return a sorted copy of the currently-loaded TNs (11-digit NANP)."""
        with self._lock:
            return sorted(self._tns)

    def refresh(self) -> int:
        """Refresh the TN set from the configured source. Returns the new count.

        TNs from any source are normalized to 11-digit NANP format during load,
        so the registry can accept input in 10-digit, 11-digit, or E.164 format
        without the partner having to pre-format their data.

        Timing of the load is recorded in metrics for operational visibility.
        For large data sources (high tens of thousands of rows or more), this
        gives operators an early signal if the refresh is approaching the
        refresh_interval_sec budget.
        """
        load_start = time.time()
        try:
            if self.cfg.source == "config":
                raw_tns = self.cfg.config_tns
            elif self.cfg.source == "sqlite":
                raw_tns = self._load_sqlite()
            elif self.cfg.source in ("postgres", "mysql"):
                raw_tns = self._load_db()
            else:
                raise ValueError(f"Unknown registry source: {self.cfg.source}")

            # Normalize every TN to 11-digit NANP. Skip entries that can't be parsed.
            new_tns: Set[str] = set()
            skipped = 0
            for raw in raw_tns:
                normalized = _normalize_nanp_11(raw)
                if normalized:
                    new_tns.add(normalized)
                else:
                    skipped += 1

            load_duration = time.time() - load_start
            refresh_ts = time.time()

            with self._lock:
                old_count = len(self._tns)
                self._tns = new_tns
                self._last_refresh = refresh_ts
                new_count = len(self._tns)

            if self.metrics is not None:
                self.metrics.set_registry_stats(new_count, load_duration, refresh_ts)

            self.log.info(
                "tn_registry_refreshed",
                extra={
                    "source": self.cfg.source,
                    "previous_count": old_count,
                    "new_count": new_count,
                    "skipped_unparseable": skipped,
                    "load_duration_sec": round(load_duration, 3),
                },
            )
            return new_count
        except Exception as e:
            if self.metrics is not None:
                self.metrics.incr("registry_refresh_failures")
            self.log.error(
                "tn_registry_refresh_failed",
                extra={
                    "source": self.cfg.source,
                    "error": str(e),
                    "load_duration_sec": round(time.time() - load_start, 3),
                },
            )
            return 0

    def _load_sqlite(self) -> List[str]:
        conn = sqlite3.connect(self.cfg.sqlite_path)
        try:
            cur = conn.execute(self.cfg.db_query)
            return [row[0] for row in cur.fetchall() if row[0]]
        finally:
            conn.close()

    def _load_db(self) -> List[str]:
        raise NotImplementedError(
            "Postgres/MySQL connectivity requires production driver. "
            "Use sqlite or config source for now, or implement via Go binary."
        )

    def _refresh_loop(self):
        while not self._stop.wait(self.cfg.refresh_interval_sec):
            self.refresh()


# ============================================================================
# Metrics
# ============================================================================

class Metrics:
    """Thread-safe in-memory metrics counters."""

    def __init__(self):
        self._lock = threading.Lock()
        self.invites_seen = 0
        self.invites_parsed = 0
        self.parse_errors = 0
        self.tn_not_registered = 0
        self.deposits_fired = 0
        self.deposits_succeeded = 0
        self.deposits_failed = 0
        self.deposits_422_non_wireless = 0
        self.api_total_latency_ms_sum = 0.0
        self.api_total_latency_ms_count = 0
        self.registry_size = 0
        self.registry_last_load_duration_sec = 0.0
        self.registry_last_refresh_ts = 0.0
        self.registry_refresh_failures = 0
        self.started_at = time.time()

    def set_registry_stats(self, size: int, duration_sec: float, refresh_ts: float):
        with self._lock:
            self.registry_size = size
            self.registry_last_load_duration_sec = round(duration_sec, 3)
            self.registry_last_refresh_ts = round(refresh_ts, 1)

    def incr(self, field: str, by: int = 1):
        with self._lock:
            setattr(self, field, getattr(self, field) + by)

    def observe_latency(self, latency_ms: float):
        with self._lock:
            self.api_total_latency_ms_sum += latency_ms
            self.api_total_latency_ms_count += 1

    def snapshot(self) -> dict:
        with self._lock:
            avg_latency = (
                self.api_total_latency_ms_sum / self.api_total_latency_ms_count
                if self.api_total_latency_ms_count > 0
                else 0.0
            )
            return {
                "invites_seen": self.invites_seen,
                "invites_parsed": self.invites_parsed,
                "parse_errors": self.parse_errors,
                "tn_not_registered": self.tn_not_registered,
                "deposits_fired": self.deposits_fired,
                "deposits_succeeded": self.deposits_succeeded,
                "deposits_failed": self.deposits_failed,
                "deposits_422_non_wireless": self.deposits_422_non_wireless,
                "avg_api_latency_ms": round(avg_latency, 2),
                "registry_size": self.registry_size,
                "registry_last_load_duration_sec": self.registry_last_load_duration_sec,
                "registry_last_refresh_ts": self.registry_last_refresh_ts,
                "registry_refresh_failures": self.registry_refresh_failures,
                "uptime_sec": round(time.time() - self.started_at, 1),
                "uptime_human": format_uptime(time.time() - self.started_at),
            }

    def prometheus_text(self) -> str:
        s = self.snapshot()
        prefix = "tns_deposit_service"
        lines = [
            f"# HELP {prefix}_invites_seen Total SIP INVITEs observed by capture.",
            f"# TYPE {prefix}_invites_seen counter",
            f"{prefix}_invites_seen {s['invites_seen']}",
            f"# HELP {prefix}_invites_parsed INVITEs successfully parsed for orig/term TNs.",
            f"# TYPE {prefix}_invites_parsed counter",
            f"{prefix}_invites_parsed {s['invites_parsed']}",
            f"# HELP {prefix}_parse_errors INVITEs that could not be parsed.",
            f"# TYPE {prefix}_parse_errors counter",
            f"{prefix}_parse_errors {s['parse_errors']}",
            f"# HELP {prefix}_deposits_fired Deposit API calls initiated.",
            f"# TYPE {prefix}_deposits_fired counter",
            f"{prefix}_deposits_fired {s['deposits_fired']}",
            f"# HELP {prefix}_deposits_succeeded Deposit API calls that succeeded.",
            f"# TYPE {prefix}_deposits_succeeded counter",
            f"{prefix}_deposits_succeeded {s['deposits_succeeded']}",
            f"# HELP {prefix}_deposits_failed Deposit API calls that failed.",
            f"# TYPE {prefix}_deposits_failed counter",
            f"{prefix}_deposits_failed {s['deposits_failed']}",
            f"# HELP {prefix}_deposits_422 Deposits returning 422 (TN not on wireless).",
            f"# TYPE {prefix}_deposits_422 counter",
            f"{prefix}_deposits_422 {s['deposits_422_non_wireless']}",
            f"# HELP {prefix}_avg_api_latency_ms Average end-to-end deposit latency in ms.",
            f"# TYPE {prefix}_avg_api_latency_ms gauge",
            f"{prefix}_avg_api_latency_ms {s['avg_api_latency_ms']}",
            f"# HELP {prefix}_registry_size Current number of registered TNs loaded in memory.",
            f"# TYPE {prefix}_registry_size gauge",
            f"{prefix}_registry_size {s['registry_size']}",
            f"# HELP {prefix}_registry_last_load_duration_seconds Duration of the most recent registry load/refresh.",
            f"# TYPE {prefix}_registry_last_load_duration_seconds gauge",
            f"{prefix}_registry_last_load_duration_seconds {s['registry_last_load_duration_sec']}",
            f"# HELP {prefix}_registry_last_refresh_timestamp Unix timestamp of the most recent successful registry refresh.",
            f"# TYPE {prefix}_registry_last_refresh_timestamp gauge",
            f"{prefix}_registry_last_refresh_timestamp {s['registry_last_refresh_ts']}",
            f"# HELP {prefix}_registry_refresh_failures Count of refresh attempts that failed.",
            f"# TYPE {prefix}_registry_refresh_failures counter",
            f"{prefix}_registry_refresh_failures {s['registry_refresh_failures']}",
            f"# HELP {prefix}_uptime_seconds Uptime of the service in seconds.",
            f"# TYPE {prefix}_uptime_seconds gauge",
            f"{prefix}_uptime_seconds {s['uptime_sec']}",
        ]
        return "\n".join(lines) + "\n"


# ============================================================================
# SIP parsing
# ============================================================================

@dataclass
class InviteInfo:
    originating: str
    terminating: str
    call_id: str
    capture_ts: float


_RE_INVITE_LINE = re.compile(rb"^INVITE\s+sip:([^@\s;>]+)", re.IGNORECASE)
_RE_FROM = re.compile(rb"^From:.*?sip:([^@>\s;]+)", re.IGNORECASE | re.MULTILINE)
_RE_TO = re.compile(rb"^To:.*?sip:([^@>\s;]+)", re.IGNORECASE | re.MULTILINE)
_RE_PAI = re.compile(rb"^P-Asserted-Identity:.*?sip:([^@>\s;]+)", re.IGNORECASE | re.MULTILINE)
_RE_CALLID = re.compile(rb"^Call-ID:\s*([^\r\n]+)", re.IGNORECASE | re.MULTILINE)


def _normalize_nanp_11(raw) -> Optional[str]:
    """
    Normalize a phone number to 11-digit NANP format (1NPANXXXXXX).

    The TNS Call Event API expects 11-digit NANP format. This function
    accepts inputs in 10-digit (NPANXXXXXX), 11-digit (1NPANXXXXXX), or
    E.164 (+1NPANXXXXXX or other) format and returns 11-digit NANP.

    For non-NANP international numbers, returns the digit string as-is
    without a leading '+'. These numbers will likely receive a 422 from
    the TNS API (non-wireless or non-NANP), which is handled as expected.

    Accepts either bytes (from SIP parsing) or str (from registry loads).
    Returns None if the input has no recognizable digits, or if the
    input appears to be an IP address / hostname rather than a phone
    number (contains '.' or ':' — as happens with some SBCs/PBXes that
    put only host:port in the Request-URI instead of user@host).
    """
    if isinstance(raw, bytes):
        s = raw.decode("ascii", errors="ignore").strip()
    else:
        s = str(raw).strip()

    # Reject anything that looks like an IP address, hostname, or host:port.
    # A '.' or ':' in the raw string is a strong signal this isn't a phone
    # number - some MBGs/SBCs put "sip:HOST:PORT" in the Request-URI with
    # no user part, which would otherwise silently normalize to garbage
    # digits (e.g. "64.27.210.15:5060" -> "64272101055060").
    if "." in s or ":" in s:
        return None

    digits = re.sub(r"\D", "", s)
    if not digits:
        return None
    # Sanity bound: real phone numbers are at most ~15 digits (E.164 max).
    # Anything longer is a parse error, not a number.
    if len(digits) > 15:
        return None
    if len(digits) == 10:
        # Bare 10-digit NANP: prepend country code 1
        return f"1{digits}"
    if len(digits) == 11 and digits.startswith("1"):
        # Already 11-digit NANP
        return digits
    if digits.startswith("011"):
        # International dialed format (011 + country code + number) — strip the 011 prefix
        return digits[3:]
    # Anything else (other international E.164 stripped of '+', short codes, etc.):
    # return digits as-is. Will likely 422 from TNS but isn't our job to filter here.
    return digits


def parse_invite(payload: bytes, capture_ts: float) -> Optional[InviteInfo]:
    if not payload.startswith(b"INVITE"):
        return None

    m = _RE_INVITE_LINE.match(payload)
    if not m:
        return None

    # Terminating number extraction:
    # Standard SIP INVITEs put the called party as user@host in the
    # Request-URI (e.g. "INVITE sip:12145551234@carrier.com SIP/2.0").
    # However, some SBCs and MBGs (notably certain Mitel Border Gateway
    # and Broadworks configurations) put ONLY the host:port in the
    # Request-URI with no user part (e.g. "INVITE sip:64.27.210.15:5060
    # SIP/2.0"), placing the actual called party number in the To
    # header instead. In that case _normalize_nanp_11 will return None
    # (because the captured value contains '.' or ':'), and we fall
    # back to the To header for the called number.
    terminating = _normalize_nanp_11(m.group(1))
    if terminating is None:
        m_to = _RE_TO.search(payload)
        if m_to:
            terminating = _normalize_nanp_11(m_to.group(1))

    m_pai = _RE_PAI.search(payload)
    m_from = _RE_FROM.search(payload)
    if m_pai:
        originating = _normalize_nanp_11(m_pai.group(1))
    elif m_from:
        originating = _normalize_nanp_11(m_from.group(1))
    else:
        return None

    m_cid = _RE_CALLID.search(payload)
    call_id = m_cid.group(1).decode("ascii", errors="ignore").strip() if m_cid else "unknown"

    if not originating or not terminating:
        return None

    return InviteInfo(originating, terminating, call_id, capture_ts)


def _debug_extract_raw(payload: bytes) -> dict:
    """
    Extract the raw regex captures from an INVITE payload for DEBUG logging.

    Returns a dict of the raw bytes (decoded to str) that the parser regexes
    captured before _normalize_nanp_11 was applied. Used only when the module
    logger is at DEBUG level to show exactly what the parser saw on the wire
    vs. what it produced after normalization.

    Returns empty strings for any header the parser didn't find, so the
    log line has a consistent shape.
    """
    def _decode_group(m):
        if not m:
            return ""
        try:
            return m.group(1).decode("ascii", errors="replace").strip()
        except Exception:
            return "<decode error>"

    return {
        "request_uri_raw": _decode_group(_RE_INVITE_LINE.match(payload)),
        "from_raw": _decode_group(_RE_FROM.search(payload)),
        "to_raw": _decode_group(_RE_TO.search(payload)),
        "pai_raw": _decode_group(_RE_PAI.search(payload)),
    }


# ============================================================================
# Dispatcher — fires the TNS Call Event API call asynchronously
# ============================================================================

class Dispatcher:
    def __init__(self, cfg: Config, registry: TNRegistry, metrics: Metrics, log: logging.Logger):
        self.cfg = cfg
        self.registry = registry
        self.metrics = metrics
        self.log = log
        # No authentication on v0.1.x — the current TNS Call Event API does not
        # require credentials. If a future API version introduces auth, a new
        # service version will be released to handle it.
        #
        # Headers match the validated Kamailio wire format (Content-Type +
        # Accept). The User-Agent is added for server-side identification of
        # this service version — TNS may use it for fleet observability.
        self.session = requests.Session()
        self.session.headers.update({
            "Content-Type": "application/json",
            "Accept": "*/*",
            "User-Agent": f"{PRODUCT_NAME}/{VERSION}",
        })

        # Match the HTTP connection pool size to max_workers so workers
        # never block waiting for a connection. urllib3's default pool size
        # is 10, which would silently bottleneck deployments with larger
        # max_workers values. We size the pool to max_workers and keep a
        # floor of 10 to avoid pathological small-pool behavior.
        from requests.adapters import HTTPAdapter
        pool_size = max(10, cfg.service.max_workers)
        adapter = HTTPAdapter(
            pool_connections=pool_size,
            pool_maxsize=pool_size,
            pool_block=False,
        )
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)

        self.executor = ThreadPoolExecutor(
            max_workers=cfg.service.max_workers,
            thread_name_prefix="dispatcher",
        )

    def submit(self, info: InviteInfo):
        self.executor.submit(self._fire, info)

    def _fire(self, info: InviteInfo):
        # In all_outbound mode, the vantage (target_ips on a trunk-edge interface)
        # is what guarantees these are legitimate outbound calls; the registry
        # is intentionally not consulted. In registry mode (default), the TN
        # must be on the loaded list or we don't fire.
        if self.cfg.registry.mode != "all_outbound":
            if not self.registry.contains(info.originating):
                self.metrics.incr("tn_not_registered")
                return

        self.metrics.incr("deposits_fired")
        api_start = time.time()

        # Build the request body in the exact shape the TNS Call Event API
        # expects, matching the validated Kamailio wire format:
        #   {"calling": <int>, "called": <int>, "type": "deposit"}
        #
        # Numbers are sent as JSON integers (no quotes around them) because
        # that is what the Kamailio integration sends and what TNS has
        # validated. The internal registry/parser representation is a string
        # ('12145551212') for ease of comparison/normalization; we convert to
        # int at serialization time only.
        #
        # call_id is intentionally NOT sent to TNS — we retain it locally for
        # log correlation only.
        try:
            calling_int = int(info.originating)
            called_int = int(info.terminating)
        except (TypeError, ValueError):
            # Should never happen if the parser/normalizer did their jobs,
            # but guard against an unexpected non-numeric value rather than
            # crashing the worker thread.
            self.metrics.incr("deposits_failed")
            self.log.error("deposit_payload_invalid", extra={
                "call_id": info.call_id,
                "originating": info.originating,
                "terminating": info.terminating,
                "error": "could not coerce normalized TN to integer",
            })
            return

        body = {
            "calling": calling_int,
            "called": called_int,
            "type": "deposit",
        }

        if self.log.isEnabledFor(logging.DEBUG):
            self.log.debug("deposit_request", extra={
                "call_id": info.call_id,
                "api_url": self.cfg.tns.api_url,
                "request_body": body,
            })

        try:
            resp = self.session.post(
                self.cfg.tns.api_url,
                json=body,
                timeout=self.cfg.tns.api_timeout_sec,
            )
            api_end = time.time()
            total_latency_ms = (api_end - info.capture_ts) * 1000
            self.metrics.observe_latency(total_latency_ms)

            if self.log.isEnabledFor(logging.DEBUG):
                self.log.debug("deposit_response", extra={
                    "call_id": info.call_id,
                    "status": resp.status_code,
                    "response_body": resp.text,
                    "latency_ms": round(total_latency_ms, 1),
                })

            if resp.status_code in (200, 201, 202):
                self.metrics.incr("deposits_succeeded")
                self.log.info("deposit_succeeded", extra={
                    "calling": calling_int,
                    "called": called_int,
                    "call_id": info.call_id,
                    "total_latency_ms": round(total_latency_ms, 1),
                    "status": resp.status_code,
                    "registry_mode": self.cfg.registry.mode,
                })
            elif resp.status_code == 422:
                self.metrics.incr("deposits_422_non_wireless")
                self.log.debug("deposit_422_non_wireless", extra={
                    "called": called_int,
                    "call_id": info.call_id,
                    "registry_mode": self.cfg.registry.mode,
                })
            else:
                self.metrics.incr("deposits_failed")
                self.log.warning("deposit_failed", extra={
                    "call_id": info.call_id,
                    "status": resp.status_code,
                    "body": resp.text[:200],
                    "registry_mode": self.cfg.registry.mode,
                })
        except requests.exceptions.RequestException as e:
            self.metrics.incr("deposits_failed")
            self.log.error("deposit_exception", extra={
                "call_id": info.call_id,
                "error": str(e),
                "registry_mode": self.cfg.registry.mode,
            })

    def shutdown(self):
        self.log.info("dispatcher_shutdown_start")
        self.executor.shutdown(wait=True)
        self.log.info("dispatcher_shutdown_complete")


# ============================================================================
# Health/Metrics HTTP server
# ============================================================================

class HealthHandler(BaseHTTPRequestHandler):
    metrics: Metrics = None
    registry: TNRegistry = None
    registry_mode: str = "registry"

    def do_GET(self):
        if self.path == "/healthz":
            self._respond(200, "text/plain", "ok\n")
        elif self.path == "/metrics" and self.__class__.metrics:
            self._respond(200, "text/plain; version=0.0.4",
                          self.__class__.metrics.prometheus_text())
        elif self.path == "/status":
            payload = {
                "product": PRODUCT_NAME,
                "version": VERSION,
                "registry_mode": self.__class__.registry_mode,
                "metrics": self.__class__.metrics.snapshot() if self.__class__.metrics else {},
                "registry_size": self.__class__.registry.size() if self.__class__.registry else 0,
            }
            self._respond(200, "application/json", json.dumps(payload, indent=2) + "\n")
        elif self.path == "/registry":
            mode = self.__class__.registry_mode
            if mode == "all_outbound":
                payload = {
                    "product": PRODUCT_NAME,
                    "version": VERSION,
                    "registry_mode": mode,
                    "registry_size": 0,
                    "registered_tns": [],
                    "note": "Service is in all_outbound mode; the TN registry is intentionally "
                            "not consulted. Deposits fire for every captured outbound INVITE "
                            "matching capture.target_ips.",
                }
            else:
                tns = self.__class__.registry.snapshot() if self.__class__.registry else []
                payload = {
                    "product": PRODUCT_NAME,
                    "version": VERSION,
                    "registry_mode": mode,
                    "registry_size": len(tns),
                    "registered_tns": tns,
                }
            self._respond(200, "application/json", json.dumps(payload, indent=2) + "\n")
        else:
            self._respond(404, "text/plain", "not found\n")

    def _respond(self, code: int, ctype: str, body: str):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body.encode("utf-8"))

    def log_message(self, fmt, *args):
        pass


def start_health_server(bind: str, metrics: Metrics, registry: TNRegistry,
                         registry_mode: str, log: logging.Logger) -> HTTPServer:
    host, port = bind.rsplit(":", 1)
    HealthHandler.metrics = metrics
    HealthHandler.registry = registry
    HealthHandler.registry_mode = registry_mode
    server = HTTPServer((host, int(port)), HealthHandler)
    t = threading.Thread(target=server.serve_forever, daemon=True, name="health-server")
    t.start()
    log.info("health_server_started", extra={"bind": bind})
    return server


# ============================================================================
# Logging
# ============================================================================

class JSONFormatter(logging.Formatter):
    RESERVED = {
        "name", "msg", "args", "levelname", "levelno", "pathname", "filename",
        "module", "exc_info", "exc_text", "stack_info", "lineno", "funcName",
        "created", "msecs", "relativeCreated", "thread", "threadName",
        "processName", "process", "message",
    }

    def format(self, record):
        obj = {
            "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(record.created)),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        for k, v in record.__dict__.items():
            if k not in self.RESERVED:
                obj[k] = v
        return json.dumps(obj)


def setup_logging(svc: ServiceConfig) -> logging.Logger:
    log = logging.getLogger(PRODUCT_NAME)
    log.setLevel(getattr(logging, svc.log_level.upper(), logging.INFO))
    log.handlers.clear()

    handler = logging.StreamHandler(sys.stdout)
    if svc.log_format == "json":
        handler.setFormatter(JSONFormatter())
    else:
        handler.setFormatter(logging.Formatter(
            "%(asctime)s %(levelname)s %(name)s: %(message)s"
        ))
    log.addHandler(handler)
    log.propagate = False
    return log


# ============================================================================
# Packet handler
# ============================================================================

def make_packet_handler(metrics: Metrics, dispatcher: Dispatcher, log: logging.Logger):
    def handle(pkt):
        try:
            capture_ts = time.time()
            if not pkt.haslayer(Raw):
                return
            payload = bytes(pkt[Raw].load)
            if not payload.startswith(b"INVITE"):
                return
            metrics.incr("invites_seen")
            info = parse_invite(payload, capture_ts)
            if not info:
                metrics.incr("parse_errors")
                if log.isEnabledFor(logging.DEBUG):
                    log.debug("parse_failed", extra={
                        "reason": "parse_invite returned None",
                        **_debug_extract_raw(payload),
                    })
                return
            metrics.incr("invites_parsed")
            if log.isEnabledFor(logging.DEBUG):
                log.debug("invite_parsed", extra={
                    "call_id": info.call_id,
                    "originating_normalized": info.originating,
                    "terminating_normalized": info.terminating,
                    **_debug_extract_raw(payload),
                })
            dispatcher.submit(info)
        except Exception as e:
            log.exception("packet_handler_error", extra={"error": str(e)})
    return handle


# ============================================================================
# Main
# ============================================================================

def main():
    parser = argparse.ArgumentParser(
        prog=PRODUCT_NAME,
        description="TNS Deposit Service — Passive SIP signaling capture for Branded Calling token deposit.",
    )
    parser.add_argument("--config", default=f"/etc/{PRODUCT_NAME}/config.yaml",
                        help=f"Path to YAML config file (default: /etc/{PRODUCT_NAME}/config.yaml)")
    parser.add_argument("--version", action="version", version=f"{PRODUCT_NAME} {VERSION}")
    parser.add_argument("--check-config", action="store_true",
                        help="Validate the config file and exit")
    parser.add_argument("--print-bpf", action="store_true",
                        help="Print the BPF filter that would be applied and exit")
    args = parser.parse_args()

    try:
        cfg = load_config(args.config)
    except Exception as e:
        print(f"FATAL: Config error: {e}", file=sys.stderr)
        sys.exit(2)

    if args.check_config:
        if cfg.warnings:
            print("Config valid, but with warnings:")
            for w in cfg.warnings:
                print(f"  WARN: {w}")
        else:
            print("Config valid.")
        sys.exit(0)

    bpf = build_bpf_filter(cfg.capture)

    if args.print_bpf:
        print(bpf)
        sys.exit(0)

    log = setup_logging(cfg.service)
    log.info("service_starting", extra={
        "product": PRODUCT_NAME,
        "version": VERSION,
        "config_path": args.config,
        "deployment_mode": cfg.capture.deployment_mode,
        "interface": cfg.capture.interface,
        "bpf_filter": bpf,
        "registry_mode": cfg.registry.mode,
        "registry_source": cfg.registry.source,
        "invite_only": cfg.capture.invite_only,
        "promiscuous": cfg.capture.promiscuous,
        "auth_mode": "none",
    })

    for w in cfg.warnings:
        log.warning("config_warning", extra={"detail": w})

    metrics = Metrics()
    registry = TNRegistry(cfg.registry, log, metrics)
    registry.start()

    if cfg.registry.mode == "all_outbound":
        log.info("all_outbound_mode_active", extra={
            "detail": "registry.mode=all_outbound: TN registry is intentionally not consulted. "
                      "Deposits will fire for every captured outbound INVITE matching target_ips.",
            "target_ips": cfg.capture.target_ips,
        })
    elif registry.size() == 0:
        log.warning("registry_empty_on_startup",
                    extra={"source": cfg.registry.source,
                           "detail": "No TNs loaded. Service will run but will not fire deposits."})

    dispatcher = Dispatcher(cfg, registry, metrics, log)
    health_server = start_health_server(
        cfg.service.health_bind, metrics, registry, cfg.registry.mode, log
    )

    shutdown_event = threading.Event()

    def handle_signal(signum, _frame):
        if shutdown_event.is_set():
            return
        shutdown_event.set()
        log.info("shutdown_signal_received", extra={"signal": signum})
        try:
            health_server.shutdown()
        except Exception:
            pass
        dispatcher.shutdown()
        registry.stop()
        log.info("metrics_final", extra=metrics.snapshot())
        log.info("service_stopped")
        sys.exit(0)

    signal.signal(signal.SIGTERM, handle_signal)
    signal.signal(signal.SIGINT, handle_signal)

    handler = make_packet_handler(metrics, dispatcher, log)

    scapy_conf.sniff_promisc = 1 if cfg.capture.promiscuous else 0

    log.info("capture_starting", extra={"interface": cfg.capture.interface, "filter": bpf})
    try:
        sniff(
            iface=cfg.capture.interface if cfg.capture.interface != "any" else None,
            filter=bpf,
            prn=handler,
            store=False,
        )
    except Exception as e:
        log.exception("capture_failed", extra={"error": str(e)})
        sys.exit(1)


if __name__ == "__main__":
    main()
