feat: events journal read path with filters and backward pagination

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-10 14:57:32 -04:00
parent 0343eb4a24
commit 16c2922bea
2 changed files with 191 additions and 1 deletions
@@ -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

@@ -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