python-repositories/python_repositories/adapters/file_backed_queue_adapter.py

"""File-backed queue adapter: in-memory hot buffer mirrored to JSONL on disk."""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
import json
import os
from pathlib import Path
import threading
from typing import Any, Self, TextIO

import structlog

from python_repositories.adapters.memory_queue_adapter import (
    MemoryQueueAdapter,
    _parse_age,
)
from python_repositories.config.file_queue_config import FileQueueConfig
from python_repositories.interfaces.queue_repository_interface import (
    QueueRepositoryInterface,
)


def _json_default(value: object) -> str:
    if isinstance(value, datetime):
        return value.isoformat()
    raise TypeError(f"Object of type {type(value)!r} is not JSON serializable")


class FileBackedQueueAdapter(QueueRepositoryInterface):
    """Queue that mirrors an in-memory buffer to an append-friendly JSONL file.

    Does not extend ``ConnectionAwareAdapter``. ``connect()`` replays the JSONL
    file into memory; ``disconnect()`` flushes any open handle.
    """

    def __init__(
        self,
        *,
        config: FileQueueConfig | None = None,
        memory: MemoryQueueAdapter | None = None,
    ) -> None:
        if config is None:
            config = FileQueueConfig.from_env()
        if memory is None:
            memory = MemoryQueueAdapter(
                dedup_keys=config.dedup_keys,
                age_key=config.age_key,
            )
        self._config = config
        self._memory = memory
        self._lock = threading.RLock()
        self._connected = False
        self._append_handle: TextIO | None = None
        self.logger = structlog.get_logger(self.__class__.__name__)

    def __enter__(self) -> Self:
        self.connect()
        return self

    def __exit__(
        self, exc_type: type | None, exc_val: object | None, exc_tb: object | None
    ) -> None:
        del exc_type, exc_val, exc_tb
        self.disconnect()

    def connect(self) -> None:
        """Create parent directories and replay the JSONL file into memory."""
        with self._lock:
            if self._connected:
                return
            path = self._config.path
            path.parent.mkdir(parents=True, exist_ok=True)
            self._memory.clear()
            if path.is_file():
                self._replay_file(path)
            self._append_handle = path.open("a", encoding="utf-8")
            self._connected = True

    def disconnect(self) -> None:
        """Flush and close the append handle if open."""
        with self._lock:
            if self._append_handle is not None:
                self._append_handle.flush()
                try:
                    os.fsync(self._append_handle.fileno())
                except OSError:
                    pass
                self._append_handle.close()
                self._append_handle = None
            self._connected = False

    def is_connected(self) -> bool:
        """Return whether ``connect()`` has completed successfully."""
        return self._connected

    def enqueue(self, items: list[dict[str, Any]]) -> None:
        """Append items to memory and the JSONL file; optionally age-evict."""
        self._require_connected()
        with self._lock:
            added = self._memory.enqueue_and_return_added(items)
            if added:
                self._append_items(added)
            if self._config.age_key is not None:
                cutoff = datetime.now(UTC) - timedelta(hours=self._config.max_age_hours)
                if self._memory.evict_older_than(cutoff):
                    self._compact()

    def dequeue_batch(self, *, max_items: int = 1000) -> list[dict[str, Any]]:
        """Dequeue from memory and compact the JSONL file to match."""
        self._require_connected()
        with self._lock:
            batch = self._memory.dequeue_batch(max_items=max_items)
            if batch:
                self._compact()
            return batch

    def size(self) -> int:
        """Return the number of items currently in the in-memory buffer."""
        self._require_connected()
        return self._memory.size()

    def evict_older_than(self, cutoff: datetime) -> int:
        """Evict aged items from memory and compact the JSONL file."""
        self._require_connected()
        with self._lock:
            removed = self._memory.evict_older_than(cutoff)
            if removed:
                self._compact()
            return removed

    def _require_connected(self) -> None:
        if not self._connected:
            raise RuntimeError("FileBackedQueueAdapter is not connected")

    def _replay_file(self, path: Path) -> None:
        cutoff: datetime | None = None
        if self._config.age_key is not None:
            cutoff = datetime.now(UTC) - timedelta(hours=self._config.max_age_hours)
        with path.open(encoding="utf-8") as handle:
            for line_number, line in enumerate(handle, start=1):
                stripped = line.strip()
                if not stripped:
                    continue
                try:
                    payload = json.loads(stripped)
                except json.JSONDecodeError:
                    self.logger.warning(
                        "Skipping corrupt JSONL line",
                        path=str(path),
                        line_number=line_number,
                    )
                    continue
                if not isinstance(payload, dict):
                    self.logger.warning(
                        "Skipping non-object JSONL line",
                        path=str(path),
                        line_number=line_number,
                    )
                    continue
                item: dict[str, Any] = payload
                if cutoff is not None and self._config.age_key is not None:
                    age_value = item.get(self._config.age_key)
                    if age_value is None:
                        self.logger.warning(
                            "Skipping item missing age key on replay",
                            path=str(path),
                            line_number=line_number,
                            age_key=self._config.age_key,
                        )
                        continue
                    try:
                        if _parse_age(age_value) < cutoff:
                            continue
                    except ValueError:
                        self.logger.warning(
                            "Skipping item with invalid age on replay",
                            path=str(path),
                            line_number=line_number,
                        )
                        continue
                try:
                    self._memory.enqueue_and_return_added([item])
                except ValueError as exc:
                    self.logger.warning(
                        "Skipping invalid item on replay",
                        path=str(path),
                        line_number=line_number,
                        error=str(exc),
                    )

    def _append_items(self, items: list[dict[str, Any]]) -> None:
        if self._append_handle is None:
            raise RuntimeError("append handle is not open")
        for item in items:
            self._append_handle.write(
                json.dumps(item, default=_json_default, separators=(",", ":"))
            )
            self._append_handle.write("\n")
        self._append_handle.flush()

    def _compact(self) -> None:
        """Rewrite the JSONL file from the current in-memory snapshot."""
        path = self._config.path
        tmp_path = path.with_suffix(path.suffix + ".tmp")
        if self._append_handle is not None:
            self._append_handle.flush()
            self._append_handle.close()
            self._append_handle = None
        with tmp_path.open("w", encoding="utf-8") as handle:
            for item in self._memory.snapshot():
                handle.write(
                    json.dumps(item, default=_json_default, separators=(",", ":"))
                )
                handle.write("\n")
            handle.flush()
        tmp_path.replace(path)
        self._append_handle = path.open("a", encoding="utf-8")