feat: capture Fuseki error bodies into events + server-side filtering for the admin event viewer (#1050)

* feat(triple-store): capture Fuseki error body into TripleStoreError events

HTTP-backed adapters discarded the server response body on failure, so TripleStoreError events only carried a generic "500 Server Error for url ..." string — useless for telling whether a Fuseki 500 is OOM, disk-full or a lock abort.

- Add Exceptions.RequestError (TripleStorePorts) carrying operation, status_code, response_body (truncated to 4KB), endpoint, attempts. Plain-data domain exception so the service layer needs no HTTP-library import.
- ApacheJenaTDB2: read response.text on failure and raise RequestError; the distributed-lock acquire timeout now raises RequestError(operation=acquire_write_lock) instead of a bare RuntimeError.
- TripleStoreService: enrich TripleStoreError.message via exc.detail(); query()/query_view() now emit TripleStoreError on read failures (previously silent).

* feat(nexus): server-side filtering, search and paging for the admin event viewer

The admin event viewer fetched the last 100 events of ALL types then filtered/searched client-side, so rare events (e.g. TripleStoreError) were invisible and the type dropdown only listed whatever was in the loaded sample.

- EventService/EventSQLiteAdapter + ports: add newest_first (ORDER BY seq DESC -> last N matching, via the existing (event_type, seq) index) and search (case-insensitive LIKE over the payload, wildcards escaped). Additive and backward-compatible — cursor readers keep ASC ordering.
- /api/admin/events/recent: add event_class (type URI), q (whole-log search) and before_seq (cursor); limit is applied AFTER the filters so you get the last N of that kind. New /api/admin/events/types lists the full LogProcess registry for the dropdown.
- Web viewer: server-driven type dropdown, debounced server-side search over the whole log, 'Load older' cursor paging; the live poll respects active filters.

Author: Dr0p42

libs/naas-abi-core/naas_abi_core/services/event/EventPort.py

@@ -52,11 +52,18 @@ def query(
         until_timestamp: str | None = None,
         json_filter: dict | None = None,
         limit: int | None = None,
+        newest_first: bool = False,
+        search: str | None = None,
     ) -> list[StoredEvent]:
-        """Read events matching the filters, ordered by `seq` ascending.
+        """Read events matching the filters.
+
+        Ordered by `seq` ascending by default; pass ``newest_first=True`` to order
+        descending so ``limit`` returns the most recent N matching rows.
 
         ``json_filter`` is an EventBridge-style dict translated by the adapter
         into backend-native pushdown (SQLite JSON1 for the SQLite adapter).
+        ``search`` is a case-insensitive substring matched against the raw payload
+        text (keys and values).
         """
 
     @abstractmethod
@@ -118,8 +125,15 @@ def query(
         until_timestamp: str | None = None,
         filter: dict | None = None,
         limit: int | None = None,
+        newest_first: bool = False,
+        search: str | None = None,
     ) -> list[Any]:
-        """Return reconstructed event instances matching the filters."""
+        """Return reconstructed event instances matching the filters.
+
+        ``newest_first=True`` orders by ``seq`` descending so ``limit`` yields the
+        most recent N matches. ``search`` is a case-insensitive substring match
+        over the raw payload text.
+        """
 
     @abstractmethod
     def iter_query(

libs/naas-abi-core/naas_abi_core/services/event/EventService.py

@@ -131,12 +131,19 @@ def query(
         until_timestamp: str | None = None,
         filter: dict | None = None,
         limit: int | None = None,
+        newest_first: bool = False,
+        search: str | None = None,
     ) -> list[Any]:
         """Read events.
 
         ``filter`` is an EventBridge-style dict; see :mod:`EventFilter` for
         the supported syntax. Translated to SQLite JSON1 SQL — pushdown
         happens at the adapter, not in Python.
+
+        ``newest_first=True`` orders results by ``seq`` descending so ``limit``
+        returns the most recent N matches (e.g. "last 100 of this event type").
+        ``search`` is a case-insensitive substring matched against the raw event
+        payload (keys and values).
         """
         event_type = str(event_class._class_uri) if event_class is not None else None
         rows = self._adapter.query(
@@ -147,6 +154,8 @@ def query(
             until_timestamp=until_timestamp,
             json_filter=filter,
             limit=limit,
+            newest_first=newest_first,
+            search=search,
         )
         return [self._reconstruct(row, event_class) for row in rows]
 

libs/naas-abi-core/naas_abi_core/services/event/adapters/secondary/EventSQLiteAdapter.py

@@ -96,7 +96,22 @@ def query(
         until_timestamp: str | None = None,
         json_filter: dict | None = None,
         limit: int | None = None,
+        newest_first: bool = False,
+        search: str | None = None,
     ) -> list[StoredEvent]:
+        """Read events matching the given filters.
+
+        ``newest_first`` controls ordering: by default rows come back oldest-first
+        (``seq ASC``) — the order cursor-based readers rely on. Pass
+        ``newest_first=True`` to order ``seq DESC`` so ``limit`` yields the *most
+        recent* N matching rows (e.g. "last 100 of this type") rather than the
+        oldest N. The composite ``(event_type, seq)`` index serves both directions.
+
+        ``search`` is a case-insensitive substring matched against the raw JSON
+        payload text (keys and values) — a server-side equivalent of the admin
+        viewer's previous client-side ``JSON.stringify(...).includes()``. LIKE
+        wildcards in the term are escaped so they match literally.
+        """
         clauses: list[str] = []
         params: list[object] = []
         if event_type is not None:
@@ -119,11 +134,18 @@ def query(
             if where_sql:
                 clauses.append(where_sql)
                 params.extend(where_params)
+        if search:
+            # Escape LIKE metacharacters so the term matches literally.
+            like = (
+                search.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+            )
+            clauses.append("CAST(payload AS TEXT) LIKE ? ESCAPE '\\'")
+            params.append(f"%{like}%")
 
         sql = "SELECT id, event_type, seq, timestamp, payload FROM events"
         if clauses:
             sql += " WHERE " + " AND ".join(clauses)
-        sql += " ORDER BY seq ASC"
+        sql += " ORDER BY seq DESC" if newest_first else " ORDER BY seq ASC"
         if limit is not None:
             sql += " LIMIT ?"
             params.append(limit)

libs/naas-abi-core/naas_abi_core/services/event/adapters/secondary/EventSQLiteAdapter_test.py

@@ -121,6 +121,79 @@ def test_max_seq(adapter):
     assert adapter.max_seq(event_type="urn:Type:DoesNotExist") == 0
 
 
+# ---------------------------------------------------------------------------
+# newest_first ordering ("last N of a type")
+# ---------------------------------------------------------------------------
+
+
+def test_query_newest_first_orders_seq_desc(adapter):
+    for i in range(5):
+        adapter.append(f"urn:e{i}", "urn:Type:A", _ts(i), b"p")
+    rows = adapter.query(newest_first=True)
+    assert [r.seq for r in rows] == [5, 4, 3, 2, 1]
+
+
+def test_query_newest_first_with_limit_returns_most_recent(adapter):
+    for i in range(5):
+        adapter.append(f"urn:e{i}", "urn:Type:A", _ts(i), b"p")
+    rows = adapter.query(newest_first=True, limit=2)
+    # The two MOST RECENT, not the two oldest.
+    assert [r.seq for r in rows] == [5, 4]
+
+
+def test_query_newest_first_with_type_returns_last_n_of_that_type(adapter):
+    # Interleave types; the rare type B is sparse and old relative to seq.
+    adapter.append("urn:b1", "urn:Type:B", _ts(0), b"p")
+    for i in range(10):
+        adapter.append(f"urn:a{i}", "urn:Type:A", _ts(i + 1), b"p")
+    adapter.append("urn:b2", "urn:Type:B", _ts(20), b"p")
+
+    rows = adapter.query(event_type="urn:Type:B", newest_first=True, limit=100)
+    # Both B events are returned even though A dominates the recent seq window.
+    assert [r.id for r in rows] == ["urn:b2", "urn:b1"]
+
+
+def test_query_default_ordering_is_unchanged(adapter):
+    for i in range(3):
+        adapter.append(f"urn:e{i}", "urn:Type:A", _ts(i), b"p")
+    # Default (no newest_first) still oldest-first — cursor readers depend on it.
+    rows = adapter.query()
+    assert [r.seq for r in rows] == [1, 2, 3]
+
+
+# ---------------------------------------------------------------------------
+# search (substring over payload)
+# ---------------------------------------------------------------------------
+
+
+def test_query_search_matches_payload_substring(adapter):
+    adapter.append("urn:e1", "urn:Type:A", _ts(0), b'{"message":"disk full"}')
+    adapter.append("urn:e2", "urn:Type:A", _ts(1), b'{"message":"all good"}')
+    adapter.append("urn:e3", "urn:Type:A", _ts(2), b'{"message":"DISK pressure"}')
+
+    rows = adapter.query(search="disk")
+    # Case-insensitive over the raw JSON text.
+    assert {r.id for r in rows} == {"urn:e1", "urn:e3"}
+
+
+def test_query_search_combines_with_type_and_newest_first(adapter):
+    adapter.append("urn:e1", "urn:Type:A", _ts(0), b'{"status":500}')
+    adapter.append("urn:e2", "urn:Type:B", _ts(1), b'{"status":500}')
+    adapter.append("urn:e3", "urn:Type:A", _ts(2), b'{"status":500}')
+
+    rows = adapter.query(event_type="urn:Type:A", search="500", newest_first=True)
+    assert [r.id for r in rows] == ["urn:e3", "urn:e1"]
+
+
+def test_query_search_escapes_like_wildcards(adapter):
+    adapter.append("urn:e1", "urn:Type:A", _ts(0), b'{"path":"a/b"}')
+    adapter.append("urn:e2", "urn:Type:A", _ts(1), b'{"path":"axb"}')
+
+    # "a%b" must match literally, not as the LIKE wildcard "a<anything>b".
+    rows = adapter.query(search="a%b")
+    assert rows == []
+
+
 # ---------------------------------------------------------------------------
 # cursor / query_for_consumer
 # ---------------------------------------------------------------------------

libs/naas-abi-core/naas_abi_core/services/triple_store/TripleStorePorts.py

@@ -22,6 +22,57 @@ class GraphNotFoundError(Exception):
     class GraphAlreadyExistsError(Exception):
         pass
 
+    class RequestError(Exception):
+        """Raised by HTTP-backed adapters when the triple-store server rejects a request.
+
+        Carries the server's own diagnostics (status code + response body) so the
+        domain layer can record a meaningful ``TripleStoreError`` event instead of a
+        generic transport-error string. This is a plain-data domain exception: the
+        ``TripleStoreService`` catches it without importing any HTTP library, keeping
+        business logic technology-agnostic.
+
+        Attributes:
+            operation: The triple-store operation that failed (e.g. ``"update"``,
+                ``"query"``, ``"acquire_write_lock"``).
+            status_code: HTTP status returned by the server, when applicable.
+            response_body: The server's response body (already truncated by the
+                adapter), which holds the real cause — OOM, disk-full, lock abort,
+                a TDB2 stack trace, etc.
+            endpoint: The endpoint the request targeted.
+            attempts: How many attempts were made before giving up.
+        """
+
+        def __init__(
+            self,
+            *,
+            operation: str,
+            message: str,
+            status_code: int | None = None,
+            response_body: str | None = None,
+            endpoint: str | None = None,
+            attempts: int | None = None,
+        ) -> None:
+            self.operation = operation
+            self.status_code = status_code
+            self.response_body = response_body
+            self.endpoint = endpoint
+            self.attempts = attempts
+            super().__init__(message)
+
+        def detail(self) -> str:
+            """Single multi-line summary suitable for logs and event messages."""
+            fields = [f"operation={self.operation}"]
+            if self.status_code is not None:
+                fields.append(f"status={self.status_code}")
+            if self.endpoint:
+                fields.append(f"endpoint={self.endpoint}")
+            if self.attempts is not None:
+                fields.append(f"attempts={self.attempts}")
+            summary = f"{' '.join(fields)}: {super().__str__()}"
+            if self.response_body:
+                summary += f"\nServer response:\n{self.response_body}"
+            return summary
+
 
 class OntologyEvent(Enum):
     INSERT = "INSERT"

libs/naas-abi-core/naas_abi_core/services/triple_store/TripleStoreService.py

@@ -11,6 +11,7 @@
 from naas_abi_core import logger
 from naas_abi_core.services.ServiceBase import ServiceBase
 from naas_abi_core.services.triple_store.TripleStorePorts import (
+    Exceptions,
     ITripleStorePort,
     ITripleStoreService,
     OntologyEvent,
@@ -117,6 +118,21 @@ def __publish_event(self, event: object) -> None:
             # Triple store mutations are the source of truth; event logging must not break them.
             logger.warning(f"TripleStoreService: failed to publish event: {exc}")
 
+    @staticmethod
+    def _error_message(exc: Exception) -> str:
+        """Build a rich, loggable message from an adapter exception.
+
+        HTTP-backed adapters (e.g. Apache Jena Fuseki) raise
+        :class:`Exceptions.RequestError`, which carries the server's own status
+        code and response body — the actual cause of a 5xx (OOM, disk-full, lock
+        abort, TDB2 stack trace). Surface that detail so the persisted
+        ``TripleStoreError`` event is genuinely diagnosable instead of a generic
+        "500 Server Error" string. Any other exception falls back to ``str``.
+        """
+        if isinstance(exc, Exceptions.RequestError):
+            return exc.detail()
+        return str(exc)
+
     def _hash_value(self, value: object) -> str:
         return hashlib.sha256(str(value).encode("utf-8")).hexdigest()[:32]
 
@@ -140,7 +156,7 @@ def insert(
                 TripleStoreError(
                     operation="insert",
                     graph_name=str(graph_name),
-                    message=str(exc),
+                    message=self._error_message(exc),
                 )
             )
             raise
@@ -185,7 +201,7 @@ def remove(
                 TripleStoreError(
                     operation="remove",
                     graph_name=str(graph_name),
-                    message=str(exc),
+                    message=self._error_message(exc),
                 )
             )
             raise
@@ -219,10 +235,28 @@ def get(self) -> Graph:
         return self.__triple_store_adapter.get()
 
     def query(self, query: str) -> rdflib.query.Result:
-        return self.__triple_store_adapter.query(query)
+        try:
+            return self.__triple_store_adapter.query(query)
+        except Exception as exc:
+            self.__publish_event(
+                TripleStoreError(
+                    operation="query",
+                    message=self._error_message(exc),
+                )
+            )
+            raise
 
     def query_view(self, view: str, query: str) -> rdflib.query.Result:
-        return self.__triple_store_adapter.query_view(view, query)
+        try:
+            return self.__triple_store_adapter.query_view(view, query)
+        except Exception as exc:
+            self.__publish_event(
+                TripleStoreError(
+                    operation="query_view",
+                    message=self._error_message(exc),
+                )
+            )
+            raise
 
     def create_graph(self, graph_name: URIRef) -> None:
         assert graph_name is not None
@@ -235,7 +269,7 @@ def create_graph(self, graph_name: URIRef) -> None:
                 TripleStoreError(
                     operation="create_graph",
                     graph_name=str(graph_name),
-                    message=str(exc),
+                    message=self._error_message(exc),
                 )
             )
             raise
@@ -253,7 +287,7 @@ def clear_graph(self, graph_name: URIRef) -> None:
                 TripleStoreError(
                     operation="clear_graph",
                     graph_name=str(graph_name),
-                    message=str(exc),
+                    message=self._error_message(exc),
                 )
             )
             raise
@@ -268,7 +302,7 @@ def drop_graph(self, graph_name: URIRef) -> None:
                 TripleStoreError(
                     operation="drop_graph",
                     graph_name=str(graph_name),
-                    message=str(exc),
+                    message=self._error_message(exc),
                 )
             )
             raise
@@ -435,7 +469,7 @@ def _apply_schema_for_file_with_event(
                 TripleStoreError(
                     operation="load_schema",
                     filepath=filepath,
-                    message=str(exc),
+                    message=self._error_message(exc),
                 )
             )
             raise
@@ -579,7 +613,7 @@ def remove_schema(self, filepath: str):
                 TripleStoreError(
                     operation="remove_schema",
                     filepath=filepath,
-                    message=str(exc),
+                    message=self._error_message(exc),
                 )
             )
             raise