feat: events journal write primitives (log_event, backfill, get_events_journal)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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
+from typing import Dict, Any, Optional, Callable, Iterable, List, Tuple, Union
 
 logger = logging.getLogger(__name__)
 
@@ -149,10 +149,35 @@ class MessageJournal:
                     self._current_size += len(json_bytes)
                     
                     logger.debug(f"Logged message from {addr[0]}: {msg.get('ID', 'UNKNOWN')}")
-                    
+
             except Exception as e:
                 logger.error(f"Error writing to journal: {e}")
-    
+
+    async def log_event(self, event: Dict[str, Any]):
+        """Write a caller-provided dict verbatim as one JSONL line (with rotation)."""
+        if not self.enabled or not self._initialized:
+            return
+        async with self._lock:
+            try:
+                line = json.dumps(event, separators=(',', ':')) + '\n'
+                nbytes = len(line.encode('utf-8'))
+                if self._current_size + nbytes > self.max_size:
+                    await self._rotate()
+                if self._file_handle:
+                    self._file_handle.write(line)
+                    self._file_handle.flush()
+                    self._current_size += nbytes
+            except Exception as e:
+                logger.error(f"Error writing event to journal: {e}")
+
+    async def backfill(self, events: Iterable[Dict[str, Any]]):
+        """One-time seed: write *events* only if the journal file is currently empty."""
+        if not self.enabled or not self._initialized or self._current_size > 0:
+            return
+        for ev in events:
+            if isinstance(ev, dict):
+                await self.log_event(ev)
+
     async def _rotate(self):
         """
         Rotate the journal file.
@@ -332,7 +357,7 @@ def get_journal(config: Optional[Dict[str, Any]] = None) -> MessageJournal:
 async def log_message(msg: Dict[str, Any], addr: tuple, timestamp: Optional[float] = None):
     """
     Convenience function to log a message using the global journal.
-    
+
     Args:
         msg: Parsed message dictionary
         addr: Source address (ip, port) tuple
@@ -340,3 +365,28 @@ async def log_message(msg: Dict[str, Any], addr: tuple, timestamp: Optional[floa
     """
     journal = get_journal()
     await journal.log_message(msg, addr, timestamp)
+
+
+# Global events journal instance (human-readable event log, written by notify.eventlog)
+_events_journal_instance: Optional[MessageJournal] = None
+
+
+def get_events_journal(config: Optional[Dict[str, Any]] = None) -> MessageJournal:
+    """Get or create the global events journal instance.
+
+    Uses the events_journal_* config keys; shares journal_dir and
+    journal_enabled with the raw-datagram journal.
+    """
+    global _events_journal_instance
+    if _events_journal_instance is None:
+        cfg = config or {}
+        _events_journal_instance = MessageJournal(
+            {
+                'journal_dir': cfg.get('journal_dir', '/var/log/heartbeat'),
+                'journal_file': cfg.get('events_journal_file', 'events.journal'),
+                'journal_max_size': cfg.get('events_journal_max_size', 10 * 1024 * 1024),
+                'journal_max_backups': cfg.get('events_journal_max_backups', 10),
+                'journal_enabled': cfg.get('journal_enabled', True),
+            }
+        )
+    return _events_journal_instance

tests/test_events_journal.py

@@ -0,0 +1,110 @@
+"""Tests for the dedicated events journal (write path and read path)."""
+import asyncio
+import json
+
+from hbd.server import journal
+
+
+def _make_journal(tmp_path, **overrides):
+    cfg = {"journal_dir": str(tmp_path), "journal_file": "events.journal"}
+    cfg.update(overrides)
+    j = journal.MessageJournal(cfg)
+    assert asyncio.run(j.initialize())
+    return j
+
+
+def _read_lines(tmp_path, name="events.journal"):
+    return (tmp_path / name).read_text(encoding="utf-8").splitlines()
+
+
+EV1 = {"ts": 1000.0, "host": "h1", "level": "INFO", "service": None, "message": "host up"}
+EV2 = {"ts": 2000.0, "host": "h2", "level": "CRITICAL", "service": "cpu", "message": "cpu high"}
+
+
+def test_log_event_writes_one_json_line(tmp_path):
+    j = _make_journal(tmp_path)
+    asyncio.run(j.log_event(EV1))
+    asyncio.run(j.close())
+    lines = _read_lines(tmp_path)
+    assert len(lines) == 1
+    assert json.loads(lines[0]) == EV1
+
+
+def test_log_event_appends_in_order(tmp_path):
+    j = _make_journal(tmp_path)
+    asyncio.run(j.log_event(EV1))
+    asyncio.run(j.log_event(EV2))
+    asyncio.run(j.close())
+    lines = _read_lines(tmp_path)
+    assert [json.loads(ln)["ts"] for ln in lines] == [1000.0, 2000.0]
+
+
+def test_log_event_rotates_at_max_size(tmp_path):
+    # max_size fits one serialized event line (~77 bytes) but not two, so the
+    # second write triggers exactly one rotation
+    j = _make_journal(tmp_path, journal_max_size=120)
+    asyncio.run(j.log_event(EV1))
+    asyncio.run(j.log_event(EV2))
+    asyncio.run(j.close())
+    backups = list(tmp_path.glob("events.journal.*"))
+    assert len(backups) == 1
+    assert json.loads(backups[0].read_text().splitlines()[0]) == EV1
+    assert json.loads(_read_lines(tmp_path)[0]) == EV2
+
+
+def test_log_event_noop_when_disabled(tmp_path):
+    j = journal.MessageJournal(
+        {"journal_dir": str(tmp_path), "journal_file": "events.journal", "journal_enabled": False}
+    )
+    asyncio.run(j.initialize())
+    asyncio.run(j.log_event(EV1))
+    assert not (tmp_path / "events.journal").exists()
+
+
+def test_backfill_seeds_empty_journal(tmp_path):
+    j = _make_journal(tmp_path)
+    asyncio.run(j.backfill([EV1, EV2]))
+    asyncio.run(j.close())
+    lines = _read_lines(tmp_path)
+    assert len(lines) == 2
+    assert json.loads(lines[0]) == EV1
+
+
+def test_backfill_skipped_when_journal_nonempty(tmp_path):
+    (tmp_path / "events.journal").write_text(json.dumps(EV1) + "\n")
+    j = _make_journal(tmp_path)  # initialize() picks up the existing size
+    asyncio.run(j.backfill([EV2]))
+    asyncio.run(j.close())
+    assert len(_read_lines(tmp_path)) == 1
+
+
+def test_get_events_journal_uses_events_config_keys(tmp_path):
+    journal._events_journal_instance = None
+    try:
+        ej = journal.get_events_journal(
+            {
+                "journal_dir": str(tmp_path),
+                "events_journal_file": "ev.jsonl",
+                "events_journal_max_size": 12345,
+                "events_journal_max_backups": 3,
+            }
+        )
+        assert ej.journal_file == "ev.jsonl"
+        assert ej.max_size == 12345
+        assert ej.max_backups == 3
+        assert ej.journal_dir == tmp_path
+        # singleton: second call returns the same instance
+        assert journal.get_events_journal() is ej
+    finally:
+        journal._events_journal_instance = None
+
+
+def test_get_events_journal_defaults(tmp_path):
+    journal._events_journal_instance = None
+    try:
+        ej = journal.get_events_journal({"journal_dir": str(tmp_path)})
+        assert ej.journal_file == "events.journal"
+        assert ej.max_size == 10 * 1024 * 1024
+        assert ej.max_backups == 10
+    finally:
+        journal._events_journal_instance = None