"""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")