"""Thread-safe in-process tracking queue for the Convert Python SDK (Story 2.3).

The queue batches conversion events per visitor before delivery, mirroring the

JS SDK ``ApiManager`` request queue (``api-manager.ts`` ``enqueue`` /

``releaseQueue``). It is the lightweight, synchronous enqueue side of tracking:

* :meth:`TrackingQueue.enqueue` only appends a typed snake_case

:class:`~convert_sdk.domain.results.ConversionEvent` (plus the visitor's

active segments) under a :class:`threading.Lock`. It performs **no network

I/O and no wire serialization** so it stays under the NFR5 10 ms budget. Wire

mapping happens later, at flush time, in ``tracking/payloads.py`` (the queue

holds snake_case domain items only — Critical Warning #8).

* Items are grouped per ``visitor_id`` so one visitor accumulates multiple

events in a single ``visitors[]`` entry, matching the JS per-visitor queue

shape.

* Reaching the configured ``batch_size`` signals a size-triggered release;

:meth:`enqueue` returns ``True`` so the caller funnels the release through the

ONE shared release path (Critical Warning #4) — the queue itself never calls a

transport.

* :class:`ReleaseReason` is a typed enum (``size`` / ``explicit`` / ``timeout``

/ ``atexit``) rather than the JS free-form release strings — a deliberate

Pythonic improvement (F-031).

The queue is thread-safe (``threading.Lock``) so it does not need replacing when

async transport is added later (architecture Async-Readiness guardrail).

"""

from __future__ import annotations

import enum

import threading

from dataclasses import dataclass, field

from typing import Any, List, Mapping, Optional, Union

from convert_sdk.domain.results import BucketingEvent, ConversionEvent

# Union type for all events the queue can hold (Story 2.5). Both ConversionEvent

# and BucketingEvent carry a ``visitor_id`` field so per-visitor grouping works

# unchanged. The queue holds snake_case domain objects only — wire serialization

# is dispatched per-type at flush time in ``tracking/tracker.py``.

TrackedEvent = Union[ConversionEvent, BucketingEvent]

class ReleaseReason(str, enum.Enum):

"""Why the queue was released (typed; never a free-form string — F-031).

Mirrors the JS ``releaseQueue(reason)`` call sites with typed values:

* ``SIZE`` — the queue reached the configured ``batch_size``.

* ``EXPLICIT`` — a caller invoked ``Core.flush()``.

* ``TIMEOUT`` — the opt-in periodic ``threading.Timer`` fired.

* ``ATEXIT`` — the best-effort interpreter-shutdown hook fired.

"""

SIZE = "size"

EXPLICIT = "explicit"

TIMEOUT = "timeout"

ATEXIT = "atexit"

@dataclass

class VisitorQueueItem:

"""One per-visitor queue entry: ``(visitor_id, events, segments)``.

Mirrors the JS ``enqueue(visitorId, event, segments)`` item shape. ``events``

accumulates every conversion event tracked for the visitor in this batch;

``segments`` carries the visitor's active segments (latest write wins) for

the eventual wire ``visitors[].segments``. All values stay snake_case

domain objects — no wire mapping happens here.

"""

visitor_id: str

events: List[TrackedEvent] = field(default_factory=list)

segments: Optional[Mapping[str, Any]] = None

class TrackingQueue:

"""A thread-safe, per-visitor batching queue of conversion events.

Args:

batch_size: The number of enqueued events that triggers a size-based

release. Mirrors the JS ``DEFAULT_BATCH_SIZE`` (10) when defaulted

via ``SDKConfig``.

"""

def __init__(self, batch_size: int = 10) -> None:

self._batch_size = batch_size

self._lock = threading.Lock()

# Insertion-ordered per-visitor grouping (dict preserves order).

self._items: "dict[str, VisitorQueueItem]" = {}

self._event_count = 0

# Story 5.2: the queue tracks the account/project identity it is bound to

# so a background config refresh can re-point delivery attribution to the

# new project (JS parity: ApiManager.setData()). ``None`` until set by the

# tracker; the wire envelope is still built from the tracker's snapshot,

# this metadata records the latest authoritative identity after a swap.

self._account_id: Optional[str] = None

self._project_id: Optional[str] = None

@property

def length(self) -> int:

"""The total number of queued events across all visitors."""

with self._lock:

return self._event_count

@property

def account_id(self) -> Optional[str]:

"""The account id this queue is currently bound to (post-swap identity)."""

with self._lock:

return self._account_id

@property

def project_id(self) -> Optional[str]:

"""The project id this queue is currently bound to (post-swap identity)."""

with self._lock:

return self._project_id

def update_snapshot_metadata(

self, *, account_id: Optional[str], project_id: Optional[str]

) -> None:

"""Re-point the queue's account/project identity after a config refresh.

Mirrors the JS SDK ``ApiManager.setData()`` — when a refresh swaps in a

snapshot with a different project, conversions queued afterwards must be

attributed to the new project, not the project that was current when the

SDK first initialized. The latest write wins under the queue lock.

"""

with self._lock:

self._account_id = account_id

self._project_id = project_id

def enqueue(

self,

event: TrackedEvent,

*,

segments: Optional[Mapping[str, Any]] = None,

) -> bool:

"""Append a tracked event (conversion or bucketing) for its visitor; lightweight and sync.

Groups the event under its ``visitor_id`` and records the visitor's

active ``segments`` (latest write wins). Performs no network I/O and no

wire serialization (NFR5). Returns ``True`` when this enqueue brings the

total event count to the configured ``batch_size`` — the signal that the

caller should release the queue via the shared release path with

:attr:`ReleaseReason.SIZE`. Returns ``False`` otherwise.

Both :class:`~convert_sdk.domain.results.ConversionEvent` and

:class:`~convert_sdk.domain.results.BucketingEvent` are accepted; they

coexist in the same per-visitor batch and are dispatched per-type at flush

time in :meth:`~convert_sdk.tracking.tracker.Tracker._build_batch_payload`.

"""

with self._lock:

item = self._items.get(event.visitor_id)

if item is None:

item = VisitorQueueItem(visitor_id=event.visitor_id)

self._items[event.visitor_id] = item

item.events.append(event)

if segments is not None:

item.segments = segments

self._event_count += 1

return self._event_count >= self._batch_size

def items(self) -> List[VisitorQueueItem]:

"""A snapshot list of the current per-visitor items (does not drain)."""

with self._lock:

return list(self._items.values())

def drain(self) -> List[VisitorQueueItem]:

"""Atomically remove and return all queued per-visitor items.

Used by the single shared release path: the caller drains, serializes,

and delivers. On a successful delivery the queue is already empty; on a

failed delivery the caller re-enqueues or restores the drained items

(this story leaves the queue intact by draining only inside the release

path that owns delivery — see the tracker).

"""

with self._lock:

drained = list(self._items.values())

self._items = {}

self._event_count = 0

return drained

def restore(self, items: List[VisitorQueueItem]) -> None:

"""Put drained items back (used when delivery fails, to avoid loss).

Merges restored items ahead of any events enqueued since the drain so

ordering stays stable per visitor.

"""

if not items:

return

with self._lock:

merged: "dict[str, VisitorQueueItem]" = {}

for item in items:

merged[item.visitor_id] = item

# Append any events enqueued after the drain.

for visitor_id, current in self._items.items():

if visitor_id in merged:

merged[visitor_id].events.extend(current.events)

if current.segments is not None:

merged[visitor_id].segments = current.segments

else:

merged[visitor_id] = current

self._items = merged

self._event_count = sum(len(i.events) for i in merged.values())