admin / Syanpse-Vanguard

"""Normalizer worker.

Consumes raw NDJSON batches from the broker consumer group, parses each line
into a normalized EventModel, and bulk-writes to storage. At-least-once: a
message is only acked after its batch is durably written. Idempotent replay is
provided by the agent's monotonic seq (dedup can be layered later)."""
from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import signal

from sv.bootstrap import shutdown, startup
from sv.config import settings
from sv.models.event import EventModel
from sv.pipeline.parsers import normalize

log = logging.getLogger("sv.normalizer")


def default_consumer() -> str:
    return f"{os.uname().nodename if hasattr(os, 'uname') else 'win'}-{os.getpid()}"


async def run_loop(engine, broker, stop: asyncio.Event, consumer: str) -> None:
    """Consume-parse-write-ack loop over a SHARED engine/broker. Used both by the
    standalone worker and by the consolidated single-process entrypoint (sv.main),
    where it runs as a background task alongside the API on one DuckDB connection."""
    log.info("normalizer loop up (consumer=%s)", consumer)
    while not stop.is_set():
        msgs = await broker.read_group(
            consumer, count=settings.normalizer_batch,
            block_ms=settings.normalizer_block_ms,
        )
        if not msgs:
            continue

        events: list[EventModel] = []
        for m in msgs:
            events.extend(_parse_lines(m.payload))

        if events:
            # Apply keyword alert rules BEFORE storing so matched events are
            # persisted with their Alert severity + name. Returns (event,
            # webhook_id) pairs for rules that should fire a webhook on match.
            alert_targets = []
            try:
                from sv import alerts
                alert_targets = await alerts.apply_rules(engine, events)
            except Exception as exc:  # noqa: BLE001 — alerts must never break ingest
                log.warning("alert-rule matching error: %s", exc)

            written = await engine.write_batch(events)
            log.debug("wrote %d events from %d messages", written, len(msgs))
            # Fire outbound webhooks for qualifying events (fire-and-forget so
            # slow/failing receivers never stall the pipeline).
            asyncio.create_task(_safe_dispatch(engine, events, alert_targets))

        for m in msgs:                           # ack only after durable write
            await broker.ack(m.msg_id)
    log.info("normalizer loop stopped")


async def _safe_dispatch(engine, events, alert_targets=None) -> None:
    try:
        from sv import webhooks
        await webhooks.dispatch(engine, events, direct=alert_targets)
    except Exception as exc:  # noqa: BLE001 — webhooks must never break ingest
        log.warning("webhook dispatch error: %s", exc)


async def _run(concurrency: int) -> None:
    """Standalone worker entrypoint (owns its own engine/broker). Only safe as a
    separate process when the storage backend is a shared server (ClickHouse) —
    NOT with embedded DuckDB, which is single-writer. See sv.main for the
    consolidated DuckDB deployment."""
    engine, broker = await startup(with_broker=True)
    stop = asyncio.Event()

    def _handle(*_):
        stop.set()

    try:
        signal.signal(signal.SIGINT, _handle)
        signal.signal(signal.SIGTERM, _handle)
    except (ValueError, AttributeError):
        pass  # not on main thread / unsupported platform

    try:
        await run_loop(engine, broker, stop, default_consumer())
    finally:
        await shutdown(engine, broker)


def _parse_lines(payload: bytes) -> list[EventModel]:
    out: list[EventModel] = []
    for line in payload.split(b"\n"):
        line = line.strip()
        if not line:
            continue
        try:
            raw = json.loads(line)
            out.append(normalize(raw))
        except Exception as exc:                 # tolerant: drop the bad line, keep the batch
            log.warning("dropping malformed event: %s", exc)
    return out


def main() -> None:
    ap = argparse.ArgumentParser(description="Synapse-Vanguard normalizer worker")
    ap.add_argument("--concurrency", type=int, default=4)
    args = ap.parse_args()
    logging.basicConfig(level=settings.log_level)
    asyncio.run(_run(args.concurrency))


if __name__ == "__main__":
    main()