stream.py

"""Real-time stream event writer — persists SSE events to a named Modal Dict."""

import time
import uuid
from typing import Any

import modal

_STREAM_PREFIX = "osint-stream-"


def write_stream_event(scan_id: str, event_type: str, payload: dict[str, Any]) -> None:
    """
    Write one SSE event to the persistent stream dict for scan_id.

    Keys use UUIDs so concurrent resolver calls never collide.
    next_seq is a best-effort monotonic counter (small gaps OK under concurrency).
    Events are best-effort: failures are silently swallowed.
    """
    if not scan_id:
        return
    try:
        sd = modal.Dict.from_name(f"{_STREAM_PREFIX}{scan_id}", create_if_missing=True)
        # next_seq has a benign read-then-write race under concurrency: two
        # callers may read the same seq and both increment to the same value.
        # This is acceptable because event keys are UUID-based so no data is
        # lost — only the seq counter may have duplicates or small gaps.
        seq: int = sd.get("next_seq", 0)
        sd["next_seq"] = seq + 1
        sd[f"evt_{uuid.uuid4().hex}"] = {
            "seq": seq,
            "type": event_type,
            "payload": payload,
            "ts": time.time(),
        }
    except Exception:
        pass  # stream events are best-effort