From 16c2922bea1ed5437afc4aaf7dc5f338a9ae5c32 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Fri, 10 Jul 2026 14:57:32 -0400 Subject: [PATCH] feat: events journal read path with filters and backward pagination Co-Authored-By: Claude Fable 5 --- hbd/server/journal.py | 82 +++++++++++++++++++++++++- tests/test_events_journal.py | 110 +++++++++++++++++++++++++++++++++++ 2 files changed, 191 insertions(+), 1 deletion(-) diff --git a/hbd/server/journal.py b/hbd/server/journal.py index a2d0760..3450c65 100644 --- a/hbd/server/journal.py +++ b/hbd/server/journal.py @@ -11,7 +11,7 @@ import os import asyncio from datetime import datetime from pathlib import Path -from typing import Dict, Any, Optional, Iterable +from typing import Dict, Any, Optional, Callable, Iterable, List, Tuple, Union logger = logging.getLogger(__name__) @@ -334,6 +334,86 @@ class MessageJournal: } +def _iter_journal_events(journal_dir: Union[str, Path], journal_file: str) -> Iterable[Dict[str, Any]]: + """Yield event dicts from the journal files, newest event first. + + Reads the current file, then rotated backups newest-to-oldest (backup + names embed rotation timestamps, so filename sort is chronological). + """ + dirp = Path(journal_dir) + files = [dirp / journal_file] + files.extend(sorted(dirp.glob(journal_file + '.*'), reverse=True)) + for f in files: + if not f.is_file(): + continue + try: + lines = f.read_text(encoding='utf-8', errors='replace').splitlines() + except OSError as e: + logger.warning(f"Cannot read journal file {f}: {e}") + continue + for line in reversed(lines): + try: + ev = json.loads(line) + except ValueError: + continue + if isinstance(ev, dict): + yield ev + + +def filter_events( + events: Iterable[Dict[str, Any]], + limit: int = 100, + before: Optional[float] = None, + host: Optional[str] = None, + level: Optional[str] = None, + q: Optional[str] = None, + predicate: Optional[Callable[[Dict[str, Any]], bool]] = None, +) -> Tuple[List[Dict[str, Any]], bool]: + """Filter an iterable of event dicts already ordered newest-first. + + Returns (events, more): up to *limit* matching events, and whether at + least one further matching event exists beyond the limit. + """ + host_l = host.lower() if host else None + level_l = level.lower() if level else None + q_l = q.lower() if q else None + out: List[Dict[str, Any]] = [] + for ev in events: + ts = ev.get('ts') + if before is not None and (not isinstance(ts, (int, float)) or ts >= before): + continue + if host_l and host_l not in str(ev.get('host') or '').lower(): + continue + if level_l and str(ev.get('level') or '').lower() != level_l: + continue + if q_l and q_l not in str(ev.get('message') or '').lower(): + continue + if predicate is not None and not predicate(ev): + continue + if len(out) >= limit: + return out, True + out.append(ev) + return out, False + + +def read_events( + journal_dir: Union[str, Path], + journal_file: str = 'events.journal', + *, + limit: int = 100, + before: Optional[float] = None, + host: Optional[str] = None, + level: Optional[str] = None, + q: Optional[str] = None, + predicate: Optional[Callable[[Dict[str, Any]], bool]] = None, +) -> Tuple[List[Dict[str, Any]], bool]: + """Read filtered events newest-first from the events journal files.""" + return filter_events( + _iter_journal_events(journal_dir, journal_file), + limit=limit, before=before, host=host, level=level, q=q, predicate=predicate, + ) + + # Global journal instance _journal_instance: Optional[MessageJournal] = None diff --git a/tests/test_events_journal.py b/tests/test_events_journal.py index fd8b257..61d9796 100644 --- a/tests/test_events_journal.py +++ b/tests/test_events_journal.py @@ -108,3 +108,113 @@ def test_get_events_journal_defaults(tmp_path): assert ej.max_backups == 10 finally: journal._events_journal_instance = None + + +# ---- read path ------------------------------------------------------------- + + +def _write_journal(path, events): + path.write_text("".join(json.dumps(e) + "\n" for e in events), encoding="utf-8") + + +def _evts(*ts_list): + return [ + {"ts": float(t), "host": f"host{i}", "level": "INFO", "service": None, "message": f"msg {t}"} + for i, t in enumerate(ts_list) + ] + + +def test_read_events_newest_first(tmp_path): + _write_journal(tmp_path / "events.journal", _evts(1, 2, 3)) + events, more = journal.read_events(tmp_path) + assert [e["ts"] for e in events] == [3.0, 2.0, 1.0] + assert more is False + + +def test_read_events_limit_and_more(tmp_path): + _write_journal(tmp_path / "events.journal", _evts(1, 2, 3)) + events, more = journal.read_events(tmp_path, limit=2) + assert [e["ts"] for e in events] == [3.0, 2.0] + assert more is True + + +def test_read_events_before_cursor(tmp_path): + _write_journal(tmp_path / "events.journal", _evts(1, 2, 3)) + events, _ = journal.read_events(tmp_path, before=3.0) + assert [e["ts"] for e in events] == [2.0, 1.0] + + +def test_read_events_spans_rotated_files(tmp_path): + # rotated backup holds the oldest events; current file the newest + _write_journal(tmp_path / "events.journal.20260101-000000", _evts(1, 2)) + _write_journal(tmp_path / "events.journal.20260201-000000", _evts(3, 4)) + _write_journal(tmp_path / "events.journal", _evts(5, 6)) + events, more = journal.read_events(tmp_path, limit=10) + assert [e["ts"] for e in events] == [6.0, 5.0, 4.0, 3.0, 2.0, 1.0] + assert more is False + + +def test_read_events_pagination_across_files(tmp_path): + _write_journal(tmp_path / "events.journal.20260101-000000", _evts(1, 2)) + _write_journal(tmp_path / "events.journal", _evts(3, 4)) + page1, more1 = journal.read_events(tmp_path, limit=3) + assert [e["ts"] for e in page1] == [4.0, 3.0, 2.0] + assert more1 is True + page2, more2 = journal.read_events(tmp_path, limit=3, before=page1[-1]["ts"]) + assert [e["ts"] for e in page2] == [1.0] + assert more2 is False + + +def test_read_events_host_filter_substring_case_insensitive(tmp_path): + evs = [ + {"ts": 1.0, "host": "Wentworth", "level": "INFO", "service": None, "message": "a"}, + {"ts": 2.0, "host": "winter", "level": "INFO", "service": None, "message": "b"}, + ] + _write_journal(tmp_path / "events.journal", evs) + events, _ = journal.read_events(tmp_path, host="went") + assert [e["host"] for e in events] == ["Wentworth"] + + +def test_read_events_level_filter_exact_case_insensitive(tmp_path): + evs = [ + {"ts": 1.0, "host": "h", "level": "CRITICAL", "service": None, "message": "a"}, + {"ts": 2.0, "host": "h", "level": "INFO", "service": None, "message": "b"}, + ] + _write_journal(tmp_path / "events.journal", evs) + events, _ = journal.read_events(tmp_path, level="critical") + assert [e["level"] for e in events] == ["CRITICAL"] + + +def test_read_events_message_filter(tmp_path): + evs = [ + {"ts": 1.0, "host": "h", "level": "INFO", "service": None, "message": "disk almost full"}, + {"ts": 2.0, "host": "h", "level": "INFO", "service": None, "message": "all quiet"}, + ] + _write_journal(tmp_path / "events.journal", evs) + events, _ = journal.read_events(tmp_path, q="Disk") + assert [e["ts"] for e in events] == [1.0] + + +def test_read_events_skips_malformed_lines(tmp_path): + p = tmp_path / "events.journal" + p.write_text('{"ts": 1.0, "host": "h", "level": "INFO", "message": "ok"}\nnot json\n[1,2]\n') + events, _ = journal.read_events(tmp_path) + assert [e["ts"] for e in events] == [1.0] + + +def test_read_events_predicate(tmp_path): + _write_journal(tmp_path / "events.journal", _evts(1, 2)) + events, _ = journal.read_events(tmp_path, predicate=lambda e: e["host"] == "host0") + assert [e["host"] for e in events] == ["host0"] + + +def test_read_events_missing_dir(tmp_path): + events, more = journal.read_events(tmp_path / "nope") + assert events == [] and more is False + + +def test_filter_events_over_in_memory_ring(): + ring = _evts(1, 2, 3) # oldest-first, like data.msgs + events, more = journal.filter_events(reversed(ring), limit=2) + assert [e["ts"] for e in events] == [3.0, 2.0] + assert more is True