review fixes

docker/Dockerfile

@@ -29,17 +29,16 @@ RUN pacman -S --noconfirm --asdeps \
         python-filelock \
         python-inflection \
         python-pyelftools \
-        python-requests \
+        python-requests
-        && \
+RUN pacman -S --noconfirm --asdeps \
-    pacman -S --noconfirm --asdeps \
         base-devel \
         python-build \
         python-flit \
         python-installer \
+        python-setuptools \
         python-tox \
-        python-wheel \
+        python-wheel
-        && \
+RUN pacman -S --noconfirm --asdeps \
-    pacman -S --noconfirm --asdeps \
         git \
         python-aiohttp \
         python-aiohttp-openmetrics \
@@ -48,9 +47,8 @@ RUN pacman -S --noconfirm --asdeps \
         python-cryptography \
         python-jinja \
         python-systemd \
-        rsync \
+        rsync
-        && \
+RUN runuser -u build -- install-aur-package \
-    runuser -u build -- install-aur-package \
         python-aioauth-client \
         python-sphinx-typlog-theme \
         python-webargs \
@@ -59,6 +57,7 @@ RUN pacman -S --noconfirm --asdeps \
         python-aiohttp-jinja2 \
         python-aiohttp-session \
         python-aiohttp-security \
+        python-aiohttp-sse-git \
         python-requests-unixsocket2
 
 # install ahriman

docs/ahriman.core.status.rst

@@ -12,6 +12,14 @@ ahriman.core.status.client module
    :no-undoc-members:
    :show-inheritance:
 
+ahriman.core.status.event\_bus module
+-------------------------------------
+
+.. automodule:: ahriman.core.status.event_bus
+   :members:
+   :no-undoc-members:
+   :show-inheritance:
+
 ahriman.core.status.local\_client module
 ----------------------------------------
 

docs/ahriman.web.schemas.rst

@@ -92,6 +92,14 @@ ahriman.web.schemas.error\_schema module
    :no-undoc-members:
    :show-inheritance:
 
+ahriman.web.schemas.event\_bus\_filter\_schema module
+-----------------------------------------------------
+
+.. automodule:: ahriman.web.schemas.event_bus_filter_schema
+   :members:
+   :no-undoc-members:
+   :show-inheritance:
+
 ahriman.web.schemas.event\_schema module
 ----------------------------------------
 
@@ -356,6 +364,14 @@ ahriman.web.schemas.search\_schema module
    :no-undoc-members:
    :show-inheritance:
 
+ahriman.web.schemas.sse\_schema module
+--------------------------------------
+
+.. automodule:: ahriman.web.schemas.sse_schema
+   :members:
+   :no-undoc-members:
+   :show-inheritance:
+
 ahriman.web.schemas.status\_schema module
 -----------------------------------------
 

docs/ahriman.web.views.v1.auditlog.rst

@@ -4,6 +4,14 @@ ahriman.web.views.v1.auditlog package
 Submodules
 ----------
 
+ahriman.web.views.v1.auditlog.event\_bus module
+-----------------------------------------------
+
+.. automodule:: ahriman.web.views.v1.auditlog.event_bus
+   :members:
+   :no-undoc-members:
+   :show-inheritance:
+
 ahriman.web.views.v1.auditlog.events module
 -------------------------------------------
 

docs/requirements.txt

@@ -7,10 +7,13 @@ aiohttp==3.11.18
     #   ahriman (pyproject.toml)
     #   aiohttp-cors
     #   aiohttp-jinja2
+    #   aiohttp-sse
 aiohttp-cors==0.8.1
     # via ahriman (pyproject.toml)
 aiohttp-jinja2==1.6
     # via ahriman (pyproject.toml)
+aiohttp-sse==2.2.0
+    # via ahriman (pyproject.toml)
 aiosignal==1.3.2
     # via aiohttp
 alabaster==1.0.0

package/archlinux/PKGBUILD

@@ -77,7 +77,7 @@ package_ahriman-triggers() {
 package_ahriman-web() {
     pkgname='ahriman-web'
     pkgdesc="ArcH linux ReposItory MANager, web server"
-    depends=("$pkgbase-core=$pkgver" 'python-aiohttp-cors' 'python-aiohttp-jinja2')
+    depends=("$pkgbase-core=$pkgver" 'python-aiohttp-cors' 'python-aiohttp-jinja2' 'python-aiohttp-sse-git')
     optdepends=('python-aioauth-client: OAuth2 authorization support'
                 'python-aiohttp-apispec>=3.0.0: autogenerated API documentation'
                 'python-aiohttp-openmetrics: HTTP metrics support'

pyproject.toml

@@ -58,6 +58,7 @@ web = [
     "aiohttp",
     "aiohttp_cors",
     "aiohttp_jinja2",
+    "aiohttp_sse",
 ]
 web-auth = [
     "ahriman[web]",

src/ahriman/core/configuration/configuration.py

@@ -80,7 +80,7 @@ class Configuration(configparser.RawConfigParser):
         """
         configparser.RawConfigParser.__init__(
             self,
-            dict_type=ConfigurationMultiDict if allow_multi_key else dict,
+            dict_type=ConfigurationMultiDict if allow_multi_key else dict,  # type: ignore[arg-type]
             allow_no_value=allow_no_value,
             strict=False,
             empty_lines_in_values=not allow_multi_key,

src/ahriman/core/configuration/shell_template.py

@@ -150,6 +150,6 @@ class ShellTemplate(Template):
                         break
 
         kwargs.update(mapping)
-        kwargs.update(dict(generator(kwargs)))
+        substituted = dict(generator(kwargs))
 
-        return self.safe_substitute(kwargs)
+        return self.safe_substitute(kwargs | substituted)

src/ahriman/core/status/event_bus.py

@@ -0,0 +1,113 @@
+#
+# Copyright (c) 2021-2026 ahriman team.
+#
+# This file is part of ahriman
+# (see https://github.com/arcan1s/ahriman).
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+import uuid
+
+from asyncio import Lock, Queue, QueueFull
+from typing import Any
+
+from ahriman.core.log import LazyLogging
+from ahriman.models.event import EventType
+
+
+SSEvent = tuple[str, dict[str, Any]]
+
+
+class EventBus(LazyLogging):
+    """
+    event bus implementation
+
+    Attributes:
+        max_size(int): maximum size of queue
+    """
+
+    def __init__(self, max_size: int) -> None:
+        """
+        Args:
+            max_size(int): maximum size of queue
+        """
+        self.max_size = max_size
+
+        self._lock = Lock()
+        self._subscribers: dict[str, tuple[list[EventType] | None, Queue[SSEvent | None]]] = {}
+
+    async def broadcast(self, event_type: EventType, object_id: str | None, **kwargs: Any) -> None:
+        """
+        broadcast event to all subscribers
+
+        Args:
+            event_type(EventType): event type
+            object_id(str | None): object identifier (e.g. package base)
+            **kwargs(Any): additional event data
+        """
+        event: dict[str, Any] = {"object_id": object_id}
+        event.update(kwargs)
+
+        async with self._lock:
+            for subscriber_id, (topics, queue) in self._subscribers.items():
+                if topics is not None and event_type not in topics:
+                    continue
+                try:
+                    queue.put_nowait((event_type, event))
+                except QueueFull:
+                    self.logger.warning("discard message to slow subscriber %s", subscriber_id)
+
+    async def shutdown(self) -> None:
+        """
+        gracefully shutdown all subscribers
+        """
+        async with self._lock:
+            for _, queue in self._subscribers.values():
+                try:
+                    queue.put_nowait(None)
+                except QueueFull:
+                    pass
+                queue.shutdown()
+
+    async def subscribe(self, topics: list[EventType] | None = None) -> tuple[str, Queue[SSEvent | None]]:
+        """
+        register new subscriber
+
+        Args:
+            topics(list[EventType] | None, optional): list of event types to filter by. If ``None`` is set,
+                all events will be delivered (Default value = None)
+
+        Returns:
+            tuple[str, Queue[SSEvent | None]]: subscriber identifier and associated queue
+        """
+        subscriber_id = str(uuid.uuid4())
+        queue: Queue[SSEvent | None] = Queue(self.max_size)
+
+        async with self._lock:
+            self._subscribers[subscriber_id] = (topics, queue)
+
+        return subscriber_id, queue
+
+    async def unsubscribe(self, subscriber_id: str) -> None:
+        """
+        unsubscribe from events
+
+        Args:
+            subscriber_id(str): subscriber unique identifier
+        """
+        async with self._lock:
+            result = self._subscribers.pop(subscriber_id, None)
+        if result is not None:
+            _, queue = result
+            queue.shutdown()

src/ahriman/core/status/watcher.py

@@ -17,15 +17,16 @@
 # You should have received a copy of the GNU General Public License
 # along with this program. If not, see <http://www.gnu.org/licenses/>.
 #
-from collections.abc import Callable
+# pylint: disable=too-many-public-methods
+from asyncio import Lock
 from dataclasses import replace
-from threading import Lock
+from typing import Self
-from typing import Any, Self
 
 from ahriman.core.exceptions import UnknownPackageError
 from ahriman.core.log import LazyLogging
 from ahriman.core.repository.package_info import PackageInfo
 from ahriman.core.status import Client
+from ahriman.core.status.event_bus import EventBus
 from ahriman.models.build_status import BuildStatus, BuildStatusEnum
 from ahriman.models.changes import Changes
 from ahriman.models.dependencies import Dependencies
@@ -41,51 +42,74 @@ class Watcher(LazyLogging):
 
     Attributes:
         client(Client): reporter instance
+        event_bus(EventBus): event bus instance
         package_info(PackageInfo): package info instance
         status(BuildStatus): daemon status
     """
 
-    def __init__(self, client: Client, package_info: PackageInfo) -> None:
+    def __init__(self, client: Client, package_info: PackageInfo, event_bus: EventBus) -> None:
         """
         Args:
             client(Client): reporter instance
             package_info(PackageInfo): package info instance
+            event_bus(EventBus): event bus instance
         """
         self.client = client
         self.package_info = package_info
+        self.event_bus = event_bus
 
         self._lock = Lock()
         self._known: dict[str, tuple[Package, BuildStatus]] = {}
         self.status = BuildStatus()
 
-    @property
+    async def event_add(self, event: Event) -> None:
-    def packages(self) -> list[tuple[Package, BuildStatus]]:
         """
-        get current known packages list
+        create new event
+
+        Args:
+            event(Event): audit log event
+        """
+        self.client.event_add(event)
+
+    async def event_get(self, event: str | EventType | None, object_id: str | None,
+                        from_date: int | float | None = None, to_date: int | float | None = None,
+                        limit: int = -1, offset: int = 0) -> list[Event]:
+        """
+        retrieve list of events
+
+        Args:
+            event(str | EventType | None): filter by event type
+            object_id(str | None): filter by event object
+            from_date(int | float | None, optional): minimal creation date, inclusive (Default value = None)
+            to_date(int | float | None, optional): maximal creation date, exclusive (Default value = None)
+            limit(int, optional): limit records to the specified count, -1 means unlimited (Default value = -1)
+            offset(int, optional): records offset (Default value = 0)
 
         Returns:
-            list[tuple[Package, BuildStatus]]: list of packages together with their statuses
+            list[Event]: list of audit log events
         """
-        with self._lock:
+        return self.client.event_get(event, object_id, from_date, to_date, limit, offset)
-            return list(self._known.values())
 
-    event_add: Callable[[Event], None]
+    async def load(self) -> None:
-
-    event_get: Callable[[str | EventType | None, str | None, int | None, int | None, int, int], list[Event]]
-
-    def load(self) -> None:
         """
         load packages from local database
         """
-        with self._lock:
+        async with self._lock:
             self._known = {
                 package.base: (package, status)
                 for package, status in self.client.package_get(None)
             }
 
-    logs_rotate: Callable[[int], None]
+    async def logs_rotate(self, keep_last_records: int) -> None:
+        """
+        remove older logs from storage
 
-    def package_archives(self, package_base: str) -> list[Package]:
+        Args:
+            keep_last_records(int): number of last records to keep
+        """
+        self.client.logs_rotate(keep_last_records)
+
+    async def package_archives(self, package_base: str) -> list[Package]:
         """
         get known package archives
 
@@ -97,15 +121,51 @@ class Watcher(LazyLogging):
         """
         return self.package_info.package_archives(package_base)
 
-    package_changes_get: Callable[[str], Changes]
+    async def package_changes_get(self, package_base: str) -> Changes:
+        """
+        get package changes
 
-    package_changes_update: Callable[[str, Changes], None]
+        Args:
+            package_base(str): package base to retrieve
 
-    package_dependencies_get: Callable[[str], Dependencies]
+        Returns:
+            Changes: package changes if available and empty object otherwise
+        """
+        return self.client.package_changes_get(package_base)
 
-    package_dependencies_update: Callable[[str, Dependencies], None]
+    async def package_changes_update(self, package_base: str, changes: Changes) -> None:
+        """
+        update package changes
 
-    def package_get(self, package_base: str) -> tuple[Package, BuildStatus]:
+        Args:
+            package_base(str): package base to update
+            changes(Changes): changes descriptor
+        """
+        self.client.package_changes_update(package_base, changes)
+
+    async def package_dependencies_get(self, package_base: str) -> Dependencies:
+        """
+        get package dependencies
+
+        Args:
+            package_base(str): package base to retrieve
+
+        Returns:
+            list[Dependencies]: package implicit dependencies if available
+        """
+        return self.client.package_dependencies_get(package_base)
+
+    async def package_dependencies_update(self, package_base: str, dependencies: Dependencies) -> None:
+        """
+        update package dependencies
+
+        Args:
+            package_base(str): package base to update
+            dependencies(Dependencies): dependencies descriptor
+        """
+        self.client.package_dependencies_update(package_base, dependencies)
+
+    async def package_get(self, package_base: str) -> tuple[Package, BuildStatus]:
         """
         get current package base build status
 
@@ -119,18 +179,12 @@ class Watcher(LazyLogging):
             UnknownPackageError: if no package found
         """
         try:
-            with self._lock:
+            async with self._lock:
                 return self._known[package_base]
         except KeyError:
             raise UnknownPackageError(package_base) from None
 
-    package_logs_add: Callable[[LogRecord], None]
+    async def package_hold_update(self, package_base: str, *, enabled: bool) -> None:
-
-    package_logs_get: Callable[[str, str | None, str | None, int, int], list[LogRecord]]
-
-    package_logs_remove: Callable[[str, str | None], None]
-
-    def package_hold_update(self, package_base: str, *, enabled: bool) -> None:
         """
         update package hold status
 
@@ -138,29 +192,98 @@ class Watcher(LazyLogging):
             package_base(str): package base name
             enabled(bool): new hold status
         """
-        package, status = self.package_get(package_base)
+        package, status = await self.package_get(package_base)
-        with self._lock:
+        async with self._lock:
             self._known[package_base] = (package, replace(status, is_held=enabled))
         self.client.package_hold_update(package_base, enabled=enabled)
 
-    package_patches_get: Callable[[str, str | None], list[PkgbuildPatch]]
+        await self.event_bus.broadcast(EventType.PackageHeld, package_base, is_held=enabled)
 
-    package_patches_remove: Callable[[str, str], None]
+    async def package_logs_add(self, log_record: LogRecord) -> None:
+        """
+        post log record
 
-    package_patches_update: Callable[[str, PkgbuildPatch], None]
+        Args:
+            log_record(LogRecord): log record
+        """
+        self.client.package_logs_add(log_record)
 
-    def package_remove(self, package_base: str) -> None:
+        await self.event_bus.broadcast(EventType.BuildLog, log_record.log_record_id.package_base, **log_record.view())
+
+    async def package_logs_get(self, package_base: str, version: str | None = None, process_id: str | None = None,
+                               limit: int = -1, offset: int = 0) -> list[LogRecord]:
+        """
+        get package logs
+
+        Args:
+            package_base(str): package base
+            version(str | None, optional): package version to search (Default value = None)
+            process_id(str | None, optional): process identifier to search (Default value = None)
+            limit(int, optional): limit records to the specified count, -1 means unlimited (Default value = -1)
+            offset(int, optional): records offset (Default value = 0)
+
+        Returns:
+            list[LogRecord]: package logs
+        """
+        return self.client.package_logs_get(package_base, version, process_id, limit, offset)
+
+    async def package_logs_remove(self, package_base: str, version: str | None) -> None:
+        """
+        remove package logs
+
+        Args:
+            package_base(str): package base
+            version(str | None): package version to remove logs. If ``None`` is set, all logs will be removed
+        """
+        self.client.package_logs_remove(package_base, version)
+
+    async def package_patches_get(self, package_base: str, variable: str | None) -> list[PkgbuildPatch]:
+        """
+        get package patches
+
+        Args:
+            package_base(str): package base to retrieve
+            variable(str | None): optional filter by patch variable
+
+        Returns:
+            list[PkgbuildPatch]: list of patches for the specified package
+        """
+        return self.client.package_patches_get(package_base, variable)
+
+    async def package_patches_remove(self, package_base: str, variable: str | None) -> None:
+        """
+        remove package patch
+
+        Args:
+            package_base(str): package base to update
+            variable(str | None): patch name. If ``None`` is set, all patches will be removed
+        """
+        self.client.package_patches_remove(package_base, variable)
+
+    async def package_patches_update(self, package_base: str, patch: PkgbuildPatch) -> None:
+        """
+        create or update package patch
+
+        Args:
+            package_base(str): package base to update
+            patch(PkgbuildPatch): package patch
+        """
+        self.client.package_patches_update(package_base, patch)
+
+    async def package_remove(self, package_base: str) -> None:
         """
         remove package base from known list if any
 
         Args:
             package_base(str): package base
         """
-        with self._lock:
+        async with self._lock:
             self._known.pop(package_base, None)
         self.client.package_remove(package_base)
 
-    def package_status_update(self, package_base: str, status: BuildStatusEnum) -> None:
+        await self.event_bus.broadcast(EventType.PackageRemoved, package_base)
+
+    async def package_status_update(self, package_base: str, status: BuildStatusEnum) -> None:
         """
         update package status
 
@@ -168,12 +291,14 @@ class Watcher(LazyLogging):
             package_base(str): package base to update
             status(BuildStatusEnum): new build status
         """
-        package, current_status = self.package_get(package_base)
+        package, current_status = await self.package_get(package_base)
-        with self._lock:
+        async with self._lock:
             self._known[package_base] = (package, BuildStatus(status, is_held=current_status.is_held))
         self.client.package_status_update(package_base, status)
 
-    def package_update(self, package: Package, status: BuildStatusEnum) -> None:
+        await self.event_bus.broadcast(EventType.PackageStatusChanged, package_base, status=status.value)
+
+    async def package_update(self, package: Package, status: BuildStatusEnum) -> None:
         """
         update package
 
@@ -181,12 +306,32 @@ class Watcher(LazyLogging):
             package(Package): package description
             status(BuildStatusEnum): new build status
         """
-        with self._lock:
+        async with self._lock:
             _, current_status = self._known.get(package.base, (package, BuildStatus()))
             self._known[package.base] = (package, BuildStatus(status, is_held=current_status.is_held))
         self.client.package_update(package, status)
 
-    def status_update(self, status: BuildStatusEnum) -> None:
+        await self.event_bus.broadcast(
+            EventType.PackageUpdated, package.base, status=status.value, version=package.version,
+        )
+
+    async def packages(self) -> list[tuple[Package, BuildStatus]]:
+        """
+        get current known packages list
+
+        Returns:
+            list[tuple[Package, BuildStatus]]: list of packages together with their statuses
+        """
+        async with self._lock:
+            return list(self._known.values())
+
+    async def shutdown(self) -> None:
+        """
+        gracefully shutdown watcher
+        """
+        await self.event_bus.shutdown()
+
+    async def status_update(self, status: BuildStatusEnum) -> None:
         """
         update service status
 
@@ -195,6 +340,8 @@ class Watcher(LazyLogging):
         """
         self.status = BuildStatus(status)
 
+        await self.event_bus.broadcast(EventType.ServiceStatusChanged, None, status=status.value)
+
     def __call__(self, package_base: str | None) -> Self:
         """
         extract client for future calls
@@ -204,24 +351,11 @@ class Watcher(LazyLogging):
 
         Returns:
             Self: instance of self to pass calls to the client
-        """
-        if package_base is not None:
-            _ = self.package_get(package_base)
-        return self
-
-    def __getattr__(self, item: str) -> Any:
-        """
-        proxy methods for reporter client
-
-        Args:
-            item(str): property name
-
-        Returns:
-            Any: attribute by its name
 
         Raises:
-            AttributeError: in case if no such attribute found
+            UnknownPackageError: if no package found
         """
-        if (method := getattr(self.client, item, None)) is not None:
+        # keep check here instead of calling package_get to keep this method synchronized
-            return method
+        if package_base is not None and package_base not in self._known:
-        raise AttributeError(f"'{self.__class__.__qualname__}' object has no attribute '{item}'")
+            raise UnknownPackageError(package_base)
+        return self

src/ahriman/models/event.py

@@ -28,16 +28,24 @@ class EventType(StrEnum):
     predefined event types
 
     Attributes:
+        BuildLog(EventType): new build log line
+        PackageHeld(EventType): package hold status has been changed
         PackageOutdated(EventType): package has been marked as out-of-date
         PackageRemoved(EventType): package has been removed
+        PackageStatusChanged(EventType): package build status has been changed
         PackageUpdateFailed(EventType): package update has been failed
         PackageUpdated(EventType): package has been updated
+        ServiceStatusChanged(EventType): service status has been changed
     """
 
+    BuildLog = "build-log"
+    PackageHeld = "package-held"
     PackageOutdated = "package-outdated"
     PackageRemoved = "package-removed"
+    PackageStatusChanged = "package-status-changed"
     PackageUpdateFailed = "package-update-failed"
     PackageUpdated = "package-updated"
+    ServiceStatusChanged = "service-status-changed"
 
 
 class Event:

src/ahriman/web/schemas/__init__.py

@@ -28,6 +28,7 @@ from ahriman.web.schemas.configuration_schema import ConfigurationSchema
 from ahriman.web.schemas.counters_schema import CountersSchema
 from ahriman.web.schemas.dependencies_schema import DependenciesSchema
 from ahriman.web.schemas.error_schema import ErrorSchema
+from ahriman.web.schemas.event_bus_filter_schema import EventBusFilterSchema
 from ahriman.web.schemas.event_schema import EventSchema
 from ahriman.web.schemas.event_search_schema import EventSearchSchema
 from ahriman.web.schemas.file_schema import FileSchema
@@ -61,6 +62,7 @@ from ahriman.web.schemas.repository_id_schema import RepositoryIdSchema
 from ahriman.web.schemas.repository_stats_schema import RepositoryStatsSchema
 from ahriman.web.schemas.rollback_schema import RollbackSchema
 from ahriman.web.schemas.search_schema import SearchSchema
+from ahriman.web.schemas.sse_schema import SSESchema
 from ahriman.web.schemas.status_schema import StatusSchema
 from ahriman.web.schemas.update_flags_schema import UpdateFlagsSchema
 from ahriman.web.schemas.worker_schema import WorkerSchema

src/ahriman/web/schemas/event_bus_filter_schema.py

@@ -0,0 +1,33 @@
+#
+# Copyright (c) 2021-2026 ahriman team.
+#
+# This file is part of ahriman
+# (see https://github.com/arcan1s/ahriman).
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+from ahriman.models.event import EventType
+from ahriman.web.apispec import fields
+from ahriman.web.schemas.repository_id_schema import RepositoryIdSchema
+
+
+class EventBusFilterSchema(RepositoryIdSchema):
+    """
+    request event bus filter schema
+    """
+
+    event = fields.List(fields.String(), metadata={
+        "description": "Event type filter",
+        "example": [EventType.PackageUpdated],
+    })

src/ahriman/web/schemas/sse_schema.py

@@ -0,0 +1,35 @@
+#
+# Copyright (c) 2021-2026 ahriman team.
+#
+# This file is part of ahriman
+# (see https://github.com/arcan1s/ahriman).
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+from ahriman.models.event import EventType
+from ahriman.web.apispec import Schema, fields
+
+
+class SSESchema(Schema):
+    """
+    response SSE schema
+    """
+
+    event = fields.String(required=True, metadata={
+        "description": "Event type",
+        "example": EventType.PackageUpdated,
+    })
+    data = fields.Dict(keys=fields.String(), values=fields.Raw(), metadata={
+        "description": "Event data",
+    })

src/ahriman/web/views/base.py

@@ -209,8 +209,8 @@ class BaseView(View, CorsViewMixin):
             HTTPBadRequest: if supplied parameters are invalid
         """
         try:
-            limit = int(self.request.query.get("limit", -1))
+            limit = int(self.request.query.get("limit", default=-1))
-            offset = int(self.request.query.get("offset", 0))
+            offset = int(self.request.query.get("offset", default=0))
         except ValueError as ex:
             raise HTTPBadRequest(reason=str(ex))
 

src/ahriman/web/views/v1/auditlog/event_bus.py

@@ -0,0 +1,103 @@
+#
+# Copyright (c) 2021-2026 ahriman team.
+#
+# This file is part of ahriman
+# (see https://github.com/arcan1s/ahriman).
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+import json
+
+from aiohttp.web import HTTPBadRequest, StreamResponse
+from aiohttp_sse import EventSourceResponse, sse_response
+from asyncio import Queue, QueueShutDown, wait_for
+from typing import ClassVar
+
+from ahriman.core.status.event_bus import SSEvent
+from ahriman.models.event import EventType
+from ahriman.models.user_access import UserAccess
+from ahriman.web.apispec.decorators import apidocs
+from ahriman.web.schemas import EventBusFilterSchema, SSESchema
+from ahriman.web.views.base import BaseView
+
+
+class EventBusView(BaseView):
+    """
+    event bus SSE view
+
+    Attributes:
+        GET_PERMISSION(UserAccess): (class attribute) get permissions of self
+    """
+
+    GET_PERMISSION: ClassVar[UserAccess] = UserAccess.Full
+    ROUTES = ["/api/v1/events/stream"]
+
+    @staticmethod
+    async def _run(response: EventSourceResponse, queue: Queue[SSEvent | None]) -> None:
+        """
+        read events from queue and send them to the client
+
+        Args:
+            response(EventSourceResponse): SSE response instance
+            queue(Queue[SSEvent | None]): subscriber queue
+        """
+        while response.is_connected():
+            try:
+                message = await wait_for(queue.get(), timeout=response.ping_interval)
+            except TimeoutError:
+                continue
+
+            if message is None:
+                break  # terminate queue on sentinel event
+            event_type, data = message
+
+            await response.send(json.dumps(data), event=event_type)
+
+    @apidocs(
+        tags=["Audit log"],
+        summary="Live updates",
+        description="Stream live updates via SSE",
+        permission=GET_PERMISSION,
+        error_400_enabled=True,
+        error_404_description="Repository is unknown",
+        schema=SSESchema(many=True),
+        query_schema=EventBusFilterSchema,
+    )
+    async def get(self) -> StreamResponse:
+        """
+        subscribe on updates
+
+        Returns:
+            StreamResponse: 200 with streaming updates
+
+        Raises:
+            HTTPBadRequest: if invalid event type is supplied
+        """
+        try:
+            topics = [EventType(event) for event in self.request.query.getall("event", [])] or None
+        except ValueError as ex:
+            raise HTTPBadRequest(reason=str(ex))
+        event_bus = self.service().event_bus
+
+        async with sse_response(self.request) as response:
+            subscription_id, queue = await event_bus.subscribe(topics)
+
+            try:
+                await self._run(response, queue)
+            except (ConnectionResetError, QueueShutDown):
+                pass
+            finally:
+                await event_bus.unsubscribe(subscription_id)
+
+        return response

src/ahriman/web/views/v1/auditlog/events.py

@@ -67,7 +67,7 @@ class EventsView(BaseView):
         except ValueError as ex:
             raise HTTPBadRequest(reason=str(ex))
 
-        events = self.service().event_get(event, object_id, from_date, to_date, limit, offset)
+        events = await self.service().event_get(event, object_id, from_date, to_date, limit, offset)
         response = [event.view() for event in events]
 
         return self.json_response(response)
@@ -94,6 +94,6 @@ class EventsView(BaseView):
         except Exception as ex:
             raise HTTPBadRequest(reason=str(ex))
 
-        self.service().event_add(event)
+        await self.service().event_add(event)
 
         raise HTTPNoContent

src/ahriman/web/views/v1/packages/archives.py

@@ -60,6 +60,6 @@ class Archives(StatusViewGuard, BaseView):
         """
         package_base = self.request.match_info["package"]
 
-        archives = self.service(package_base=package_base).package_archives(package_base)
+        archives = await self.service(package_base=package_base).package_archives(package_base)
 
         return self.json_response([archive.view() for archive in archives])

src/ahriman/web/views/v1/packages/changes.py

@@ -63,7 +63,7 @@ class ChangesView(StatusViewGuard, BaseView):
         """
         package_base = self.request.match_info["package"]
 
-        changes = self.service(package_base=package_base).package_changes_get(package_base)
+        changes = await self.service(package_base=package_base).package_changes_get(package_base)
 
         return self.json_response(changes.view())
 
@@ -97,6 +97,6 @@ class ChangesView(StatusViewGuard, BaseView):
             raise HTTPBadRequest(reason=str(ex))
 
         changes = Changes(last_commit_sha, change, pkgbuild)
-        self.service().package_changes_update(package_base, changes)
+        await self.service().package_changes_update(package_base, changes)
 
         raise HTTPNoContent

src/ahriman/web/views/v1/packages/dependencies.py

@@ -63,7 +63,7 @@ class DependenciesView(StatusViewGuard, BaseView):
         """
         package_base = self.request.match_info["package"]
 
-        dependencies = self.service(package_base=package_base).package_dependencies_get(package_base)
+        dependencies = await self.service(package_base=package_base).package_dependencies_get(package_base)
 
         return self.json_response(dependencies.view())
 
@@ -95,6 +95,6 @@ class DependenciesView(StatusViewGuard, BaseView):
         except Exception as ex:
             raise HTTPBadRequest(reason=str(ex))
 
-        self.service(package_base=package_base).package_dependencies_update(package_base, dependencies)
+        await self.service(package_base=package_base).package_dependencies_update(package_base, dependencies)
 
         raise HTTPNoContent

src/ahriman/web/views/v1/packages/hold.py

@@ -68,7 +68,7 @@ class HoldView(StatusViewGuard, BaseView):
             raise HTTPBadRequest(reason=str(ex))
 
         try:
-            self.service().package_hold_update(package_base, enabled=is_held)
+            await self.service().package_hold_update(package_base, enabled=is_held)
         except UnknownPackageError:
             raise HTTPNotFound(reason=f"Package {package_base} is unknown")
 

src/ahriman/web/views/v1/packages/logs.py

@@ -62,7 +62,7 @@ class LogsView(StatusViewGuard, BaseView):
         """
         package_base = self.request.match_info["package"]
         version = self.request.query.get("version")
-        self.service().package_logs_remove(package_base, version)
+        await self.service().package_logs_remove(package_base, version)
 
         raise HTTPNoContent
 
@@ -89,8 +89,8 @@ class LogsView(StatusViewGuard, BaseView):
         package_base = self.request.match_info["package"]
 
         try:
-            _, status = self.service().package_get(package_base)
+            _, status = await self.service().package_get(package_base)
-            logs = self.service(package_base=package_base).package_logs_get(package_base, None, None, -1, 0)
+            logs = await self.service(package_base=package_base).package_logs_get(package_base, None, None, -1, 0)
         except UnknownPackageError:
             raise HTTPNotFound(reason=f"Package {package_base} is unknown")
 
@@ -127,6 +127,6 @@ class LogsView(StatusViewGuard, BaseView):
         except Exception as ex:
             raise HTTPBadRequest(reason=str(ex))
 
-        self.service().package_logs_add(log_record)
+        await self.service().package_logs_add(log_record)
 
         raise HTTPNoContent

src/ahriman/web/views/v1/packages/package.py

@@ -66,7 +66,7 @@ class PackageView(StatusViewGuard, BaseView):
             HTTPNoContent: on success response
         """
         package_base = self.request.match_info["package"]
-        self.service().package_remove(package_base)
+        await self.service().package_remove(package_base)
 
         raise HTTPNoContent
 
@@ -94,7 +94,7 @@ class PackageView(StatusViewGuard, BaseView):
         repository_id = self.repository_id()
 
         try:
-            package, status = self.service(repository_id).package_get(package_base)
+            package, status = await self.service(repository_id).package_get(package_base)
         except UnknownPackageError:
             raise HTTPNotFound(reason=f"Package {package_base} is unknown")
 
@@ -137,9 +137,9 @@ class PackageView(StatusViewGuard, BaseView):
 
         try:
             if package is None:
-                self.service().package_status_update(package_base, status)
+                await self.service().package_status_update(package_base, status)
             else:
-                self.service().package_update(package, status)
+                await self.service().package_update(package, status)
         except UnknownPackageError:
             raise HTTPBadRequest(reason=f"Package {package_base} is unknown, but no package body set")
 

src/ahriman/web/views/v1/packages/packages.py

@@ -67,7 +67,7 @@ class PackagesView(StatusViewGuard, BaseView):
         stop = offset + limit if limit >= 0 else None
 
         repository_id = self.repository_id()
-        packages = self.service(repository_id).packages
+        packages = await self.service(repository_id).packages()
 
         comparator: Callable[[tuple[Package, BuildStatus]], Comparable] = lambda items: items[0].base
         response = [
@@ -95,6 +95,6 @@ class PackagesView(StatusViewGuard, BaseView):
         Raises:
             HTTPNoContent: on success response
         """
-        self.service().load()
+        await self.service().load()
 
         raise HTTPNoContent

src/ahriman/web/views/v1/packages/patch.py

@@ -57,7 +57,7 @@ class PatchView(StatusViewGuard, BaseView):
         package_base = self.request.match_info["package"]
         variable = self.request.match_info["patch"]
 
-        self.service().package_patches_remove(package_base, variable)
+        await self.service().package_patches_remove(package_base, variable)
 
         raise HTTPNoContent
 
@@ -83,7 +83,7 @@ class PatchView(StatusViewGuard, BaseView):
         package_base = self.request.match_info["package"]
         variable = self.request.match_info["patch"]
 
-        patches = self.service().package_patches_get(package_base, variable)
+        patches = await self.service().package_patches_get(package_base, variable)
 
         selected = next((patch for patch in patches if patch.key == variable), None)
         if selected is None:

src/ahriman/web/views/v1/packages/patches.py

@@ -57,7 +57,7 @@ class PatchesView(StatusViewGuard, BaseView):
             Response: 200 with package patches on success
         """
         package_base = self.request.match_info["package"]
-        patches = self.service().package_patches_get(package_base, None)
+        patches = await self.service().package_patches_get(package_base, None)
 
         response = [patch.view() for patch in patches]
         return self.json_response(response)
@@ -88,6 +88,6 @@ class PatchesView(StatusViewGuard, BaseView):
         except Exception as ex:
             raise HTTPBadRequest(reason=str(ex))
 
-        self.service().package_patches_update(package_base, PkgbuildPatch.parse(key, value))
+        await self.service().package_patches_update(package_base, PkgbuildPatch.parse(key, value))
 
         raise HTTPNoContent

src/ahriman/web/views/v1/service/logs.py

@@ -59,6 +59,6 @@ class LogsView(BaseView):
         except Exception as ex:
             raise HTTPBadRequest(reason=str(ex))
 
-        self.service().logs_rotate(keep_last_records)
+        await self.service().logs_rotate(keep_last_records)
 
         raise HTTPNoContent

src/ahriman/web/views/v1/status/status.py

@@ -62,7 +62,7 @@ class StatusView(StatusViewGuard, BaseView):
             Response: 200 with service status object
         """
         repository_id = self.repository_id()
-        packages = self.service(repository_id).packages
+        packages = await self.service(repository_id).packages()
         counters = Counters.from_packages(packages)
         stats = RepositoryStats.from_packages([package for package, _ in packages])
 
@@ -101,6 +101,6 @@ class StatusView(StatusViewGuard, BaseView):
         except Exception as ex:
             raise HTTPBadRequest(reason=str(ex))
 
-        self.service().status_update(status)
+        await self.service().status_update(status)
 
         raise HTTPNoContent

src/ahriman/web/views/v2/packages/logs.py

@@ -67,7 +67,7 @@ class LogsView(StatusViewGuard, BaseView):
         version = self.request.query.get("version", None)
         process = self.request.query.get("process_id", None)
 
-        logs = self.service(package_base=package_base).package_logs_get(package_base, version, process, limit, offset)
+        logs = await self.service(package_base=package_base).package_logs_get(package_base, version, process, limit, offset)
 
         head = self.request.query.get("head", "false")
         # pylint: disable=protected-access

src/ahriman/web/web.py

@@ -33,6 +33,7 @@ from ahriman.core.exceptions import InitializeError
 from ahriman.core.repository.package_info import PackageInfo
 from ahriman.core.spawn import Spawn
 from ahriman.core.status import Client
+from ahriman.core.status.event_bus import EventBus
 from ahriman.core.status.watcher import Watcher
 from ahriman.models.repository_id import RepositoryId
 from ahriman.web.apispec.info import setup_apispec
@@ -108,7 +109,9 @@ def _create_watcher(path: Path, repository_id: RepositoryId) -> Watcher:
     package_info.reporter = client
     package_info.repository_id = repository_id
 
-    return Watcher(client, package_info)
+    event_bus = EventBus(configuration.getint("web", "max_queue_size", fallback=0))
+
+    return Watcher(client, package_info, event_bus)
 
 
 async def _on_shutdown(application: Application) -> None:
@@ -118,6 +121,8 @@ async def _on_shutdown(application: Application) -> None:
     Args:
         application(Application): web application instance
     """
+    for watcher in application[WatcherKey].values():
+        await watcher.shutdown()
     application.logger.warning("server terminated")
 
 
@@ -135,7 +140,7 @@ async def _on_startup(application: Application) -> None:
 
     try:
         for watcher in application[WatcherKey].values():
-            watcher.load()
+            await watcher.load()
     except Exception:
         message = "could not load packages"
         application.logger.exception(message)

tests/ahriman/conftest.py

@@ -19,6 +19,7 @@ from ahriman.core.repository import Repository
 from ahriman.core.repository.package_info import PackageInfo
 from ahriman.core.spawn import Spawn
 from ahriman.core.status import Client
+from ahriman.core.status.event_bus import EventBus
 from ahriman.core.status.watcher import Watcher
 from ahriman.models.aur_package import AURPackage
 from ahriman.models.build_status import BuildStatus, BuildStatusEnum
@@ -690,4 +691,5 @@ def watcher(local_client: Client) -> Watcher:
         Watcher: package status watcher test instance
     """
     package_info = PackageInfo()
-    return Watcher(local_client, package_info)
+    event_bus = EventBus(0)
+    return Watcher(local_client, package_info, event_bus)

tests/ahriman/core/status/conftest.py

@@ -2,6 +2,7 @@ import pytest
 
 from ahriman.core.configuration import Configuration
 from ahriman.core.status import Client
+from ahriman.core.status.event_bus import EventBus
 from ahriman.core.status.web_client import WebClient
 
 
@@ -16,6 +17,17 @@ def client() -> Client:
     return Client()
 
 
+@pytest.fixture
+def event_bus() -> EventBus:
+    """
+    fixture for event bus
+
+    Returns:
+        EventBus: even bus test instance
+    """
+    return EventBus(0)
+
+
 @pytest.fixture
 def web_client(configuration: Configuration) -> WebClient:
     """

tests/ahriman/core/status/test_event_bus.py

@@ -0,0 +1,107 @@
+import pytest
+
+from ahriman.core.status.event_bus import EventBus
+from ahriman.models.event import EventType
+from ahriman.models.package import Package
+
+
+async def test_broadcast(event_bus: EventBus, package_ahriman: Package) -> None:
+    """
+    must broadcast event to all subscribers
+    """
+    _, queue = await event_bus.subscribe()
+    await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base, version=package_ahriman.version)
+
+    message = queue.get_nowait()
+    assert message == (
+        EventType.PackageUpdated,
+        {"object_id": package_ahriman.base, "version": package_ahriman.version},
+    )
+
+
+async def test_broadcast_with_topics(event_bus: EventBus, package_ahriman: Package) -> None:
+    """
+    must broadcast event to subscribers with matching topics
+    """
+    _, queue = await event_bus.subscribe([EventType.PackageUpdated])
+    await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base)
+    assert not queue.empty()
+
+
+async def test_broadcast_topic_isolation(event_bus: EventBus, package_ahriman: Package) -> None:
+    """
+    must not broadcast event to subscribers with non-matching topics
+    """
+    _, queue = await event_bus.subscribe([EventType.BuildLog])
+    await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base)
+    assert queue.empty()
+
+
+async def test_broadcast_queue_full(event_bus: EventBus, package_ahriman: Package) -> None:
+    """
+    must discard message to slow subscriber
+    """
+    event_bus.max_size = 1
+    _, queue = await event_bus.subscribe()
+
+    await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base)
+    await event_bus.broadcast(EventType.PackageRemoved, package_ahriman.base)
+    assert queue.qsize() == 1
+
+
+async def test_shutdown(event_bus: EventBus) -> None:
+    """
+    must send sentinel to all subscribers on shutdown
+    """
+    _, queue = await event_bus.subscribe()
+
+    await event_bus.shutdown()
+    message = queue.get_nowait()
+    assert message is None
+
+
+async def test_shutdown_queue_full(event_bus: EventBus, package_ahriman: Package) -> None:
+    """
+    must handle shutdown when queue is full
+    """
+    event_bus.max_size = 1
+    _, queue = await event_bus.subscribe()
+
+    await event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base)
+    await event_bus.shutdown()
+
+
+async def test_subscribe(event_bus: EventBus) -> None:
+    """
+    must register new subscriber
+    """
+    subscriber_id, queue = await event_bus.subscribe()
+
+    assert subscriber_id
+    assert queue.empty()
+    assert subscriber_id in event_bus._subscribers
+
+
+async def test_subscribe_with_topics(event_bus: EventBus) -> None:
+    """
+    must register subscriber with topic filter
+    """
+    subscriber_id, _ = await event_bus.subscribe([EventType.BuildLog])
+    topics, _ = event_bus._subscribers[subscriber_id]
+    assert topics == [EventType.BuildLog]
+
+
+async def test_unsubscribe(event_bus: EventBus) -> None:
+    """
+    must remove subscriber
+    """
+    subscriber_id, _ = await event_bus.subscribe()
+    await event_bus.unsubscribe(subscriber_id)
+    assert subscriber_id not in event_bus._subscribers
+
+
+async def test_unsubscribe_unknown(event_bus: EventBus) -> None:
+    """
+    must not fail on unknown subscriber removal
+    """
+    await event_bus.unsubscribe("unknown")

tests/ahriman/core/status/test_watcher.py

@@ -5,34 +5,53 @@ from pytest_mock import MockerFixture
 from ahriman.core.exceptions import UnknownPackageError
 from ahriman.core.status.watcher import Watcher
 from ahriman.models.build_status import BuildStatus, BuildStatusEnum
+from ahriman.models.changes import Changes
+from ahriman.models.dependencies import Dependencies
+from ahriman.models.event import Event, EventType
+from ahriman.models.log_record import LogRecord
+from ahriman.models.log_record_id import LogRecordId
 from ahriman.models.package import Package
+from ahriman.models.pkgbuild_patch import PkgbuildPatch
 
 
-def test_packages(watcher: Watcher, package_ahriman: Package) -> None:
+async def test_event_add(watcher: Watcher, mocker: MockerFixture) -> None:
     """
-    must return list of available packages
+    must create new event
     """
-    assert not watcher.packages
+    event = Event("event", "object")
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.event_add")
 
-    watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
+    await watcher.event_add(event)
-    assert watcher.packages
+    cache_mock.assert_called_once_with(event)
 
 
-def test_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_event_get(watcher: Watcher, mocker: MockerFixture) -> None:
+    """
+    must retrieve events
+    """
+    event = Event("event", "object")
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.event_get", return_value=[event])
+
+    result = await watcher.event_get(None, None)
+    assert result == [event]
+    cache_mock.assert_called_once_with(None, None, None, None, -1, 0)
+
+
+async def test_load(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must correctly load packages
     """
     cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_get",
                               return_value=[(package_ahriman, BuildStatus())])
 
-    watcher.load()
+    await watcher.load()
     cache_mock.assert_called_once_with(None)
     package, status = watcher._known[package_ahriman.base]
     assert package == package_ahriman
     assert status.status == BuildStatusEnum.Unknown
 
 
-def test_load_known(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_load_known(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must correctly load packages with known statuses
     """
@@ -40,147 +59,307 @@ def test_load_known(watcher: Watcher, package_ahriman: Package, mocker: MockerFi
     mocker.patch("ahriman.core.status.local_client.LocalClient.package_get", return_value=[(package_ahriman, status)])
     watcher._known = {package_ahriman.base: (package_ahriman, status)}
 
-    watcher.load()
+    await watcher.load()
     _, status = watcher._known[package_ahriman.base]
     assert status.status == BuildStatusEnum.Success
 
 
-def test_package_archives(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_logs_rotate(watcher: Watcher, mocker: MockerFixture) -> None:
+    """
+    must rotate logs
+    """
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.logs_rotate")
+    await watcher.logs_rotate(42)
+    cache_mock.assert_called_once_with(42)
+
+
+async def test_package_archives(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must return package archives from package info
     """
     archives_mock = mocker.patch("ahriman.core.repository.package_info.PackageInfo.package_archives",
                                  return_value=[package_ahriman])
 
-    result = watcher.package_archives(package_ahriman.base)
+    result = await watcher.package_archives(package_ahriman.base)
     assert result == [package_ahriman]
     archives_mock.assert_called_once_with(package_ahriman.base)
 
 
-def test_package_get(watcher: Watcher, package_ahriman: Package) -> None:
+async def test_package_get(watcher: Watcher, package_ahriman: Package) -> None:
     """
     must return package status
     """
     watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
-    package, status = watcher.package_get(package_ahriman.base)
+    package, status = await watcher.package_get(package_ahriman.base)
     assert package == package_ahriman
     assert status.status == BuildStatusEnum.Unknown
 
 
-def test_package_get_failed(watcher: Watcher, package_ahriman: Package) -> None:
+async def test_package_get_failed(watcher: Watcher, package_ahriman: Package) -> None:
     """
     must fail on unknown package
     """
     with pytest.raises(UnknownPackageError):
-        watcher.package_get(package_ahriman.base)
+        await watcher.package_get(package_ahriman.base)
 
 
-def test_package_hold_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_package_changes_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must return package changes
+    """
+    changes = Changes("sha")
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_changes_get",
+                              return_value=changes)
+
+    assert await watcher.package_changes_get(package_ahriman.base) == changes
+    cache_mock.assert_called_once_with(package_ahriman.base)
+
+
+async def test_package_changes_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must update package changes
+    """
+    changes = Changes("sha")
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_changes_update")
+
+    await watcher.package_changes_update(package_ahriman.base, changes)
+    cache_mock.assert_called_once_with(package_ahriman.base, changes)
+
+
+async def test_package_dependencies_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must return package dependencies
+    """
+    dependencies = Dependencies({"path": [package_ahriman.base]})
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_dependencies_get",
+                              return_value=dependencies)
+
+    assert await watcher.package_dependencies_get(package_ahriman.base) == dependencies
+    cache_mock.assert_called_once_with(package_ahriman.base)
+
+
+async def test_package_dependencies_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must update package dependencies
+    """
+    dependencies = Dependencies({"path": [package_ahriman.base]})
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_dependencies_update")
+
+    await watcher.package_dependencies_update(package_ahriman.base, dependencies)
+    cache_mock.assert_called_once_with(package_ahriman.base, dependencies)
+
+
+async def test_package_hold_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must update package hold status
     """
     cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_hold_update")
+    broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
     watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
 
-    watcher.package_hold_update(package_ahriman.base, enabled=True)
+    await watcher.package_hold_update(package_ahriman.base, enabled=True)
     cache_mock.assert_called_once_with(package_ahriman.base, enabled=True)
     _, status = watcher._known[package_ahriman.base]
     assert status.is_held is True
+    broadcast_mock.assert_called_once_with(EventType.PackageHeld, package_ahriman.base, is_held=True)
 
 
-def test_package_hold_update_unknown(watcher: Watcher, package_ahriman: Package) -> None:
+async def test_package_hold_update_unknown(watcher: Watcher, package_ahriman: Package) -> None:
     """
     must fail on unknown package hold update
     """
     with pytest.raises(UnknownPackageError):
-        watcher.package_hold_update(package_ahriman.base, enabled=True)
+        await watcher.package_hold_update(package_ahriman.base, enabled=True)
 
 
-def test_package_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_package_logs_add(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must post log record
+    """
+    log_record = LogRecord(LogRecordId(package_ahriman.base, "1.0.0"), 42.0, "message")
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_logs_add")
+    broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
+
+    await watcher.package_logs_add(log_record)
+    cache_mock.assert_called_once_with(log_record)
+    broadcast_mock.assert_called_once_with(EventType.BuildLog, package_ahriman.base, **log_record.view())
+
+
+async def test_package_logs_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must return package logs
+    """
+    log_record = LogRecord(LogRecordId(package_ahriman.base, "1.0.0"), 42.0, "message")
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_logs_get",
+                              return_value=[log_record])
+
+    assert await watcher.package_logs_get(package_ahriman.base) == [log_record]
+    cache_mock.assert_called_once_with(package_ahriman.base, None, None, -1, 0)
+
+
+async def test_package_logs_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must remove package logs
+    """
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_logs_remove")
+    await watcher.package_logs_remove(package_ahriman.base, None)
+    cache_mock.assert_called_once_with(package_ahriman.base, None)
+
+
+async def test_package_patches_get(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must return package patches
+    """
+    patch = PkgbuildPatch("key", "value")
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_patches_get", return_value=[patch])
+
+    assert await watcher.package_patches_get(package_ahriman.base, None) == [patch]
+    cache_mock.assert_called_once_with(package_ahriman.base, None)
+
+
+async def test_package_patches_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must remove package patches
+    """
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_patches_remove")
+    await watcher.package_patches_remove(package_ahriman.base, None)
+    cache_mock.assert_called_once_with(package_ahriman.base, None)
+
+
+async def test_package_patches_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+    """
+    must update package patches
+    """
+    patch = PkgbuildPatch("key", "value")
+    cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_patches_update")
+
+    await watcher.package_patches_update(package_ahriman.base, patch)
+    cache_mock.assert_called_once_with(package_ahriman.base, patch)
+
+
+async def test_package_remove(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must remove package base
     """
     cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_remove")
+    broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
     watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
 
-    watcher.package_remove(package_ahriman.base)
+    await watcher.package_remove(package_ahriman.base)
     assert not watcher._known
     cache_mock.assert_called_once_with(package_ahriman.base)
+    broadcast_mock.assert_called_once_with(EventType.PackageRemoved, package_ahriman.base)
 
 
-def test_package_remove_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_package_remove_unknown(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must not fail on unknown base removal
     """
     cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_remove")
-    watcher.package_remove(package_ahriman.base)
+    broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
+
+    await watcher.package_remove(package_ahriman.base)
     cache_mock.assert_called_once_with(package_ahriman.base)
+    broadcast_mock.assert_called_once_with(EventType.PackageRemoved, package_ahriman.base)
 
 
-def test_package_status_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_package_status_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must update package status only for known package
     """
     cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_status_update")
+    broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
     watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
 
-    watcher.package_status_update(package_ahriman.base, BuildStatusEnum.Success)
+    await watcher.package_status_update(package_ahriman.base, BuildStatusEnum.Success)
     cache_mock.assert_called_once_with(package_ahriman.base, pytest.helpers.anyvar(int))
     package, status = watcher._known[package_ahriman.base]
     assert package == package_ahriman
     assert status.status == BuildStatusEnum.Success
+    broadcast_mock.assert_called_once_with(
+        EventType.PackageStatusChanged, package_ahriman.base, status=BuildStatusEnum.Success.value,
+    )
 
 
-def test_package_status_update_preserves_hold(watcher: Watcher, package_ahriman: Package,
+async def test_package_status_update_preserves_hold(watcher: Watcher, package_ahriman: Package,
-                                              mocker: MockerFixture) -> None:
+                                                    mocker: MockerFixture) -> None:
     """
     must preserve hold status on package status update
     """
     mocker.patch("ahriman.core.status.local_client.LocalClient.package_status_update")
+    mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
     watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus(is_held=True))}
 
-    watcher.package_status_update(package_ahriman.base, BuildStatusEnum.Success)
+    await watcher.package_status_update(package_ahriman.base, BuildStatusEnum.Success)
     _, status = watcher._known[package_ahriman.base]
     assert status.is_held is True
 
 
-def test_package_status_update_unknown(watcher: Watcher, package_ahriman: Package) -> None:
+async def test_package_status_update_unknown(watcher: Watcher, package_ahriman: Package) -> None:
     """
     must fail on unknown package status update only
     """
     with pytest.raises(UnknownPackageError):
-        watcher.package_status_update(package_ahriman.base, BuildStatusEnum.Unknown)
+        await watcher.package_status_update(package_ahriman.base, BuildStatusEnum.Unknown)
 
 
-def test_package_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_package_update(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
     """
     must add package to cache
     """
     cache_mock = mocker.patch("ahriman.core.status.local_client.LocalClient.package_update")
+    broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
 
-    watcher.package_update(package_ahriman, BuildStatusEnum.Unknown)
+    await watcher.package_update(package_ahriman, BuildStatusEnum.Unknown)
-    assert watcher.packages
+    assert await watcher.packages()
     cache_mock.assert_called_once_with(package_ahriman, pytest.helpers.anyvar(int))
+    broadcast_mock.assert_called_once_with(
+        EventType.PackageUpdated, package_ahriman.base,
+        status=BuildStatusEnum.Unknown.value, version=package_ahriman.version,
+    )
 
 
-def test_package_update_preserves_hold(watcher: Watcher, package_ahriman: Package, mocker: MockerFixture) -> None:
+async def test_package_update_preserves_hold(watcher: Watcher, package_ahriman: Package,
+                                             mocker: MockerFixture) -> None:
     """
     must preserve hold status on package update
     """
     mocker.patch("ahriman.core.status.local_client.LocalClient.package_update")
     watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus(is_held=True))}
 
-    watcher.package_update(package_ahriman, BuildStatusEnum.Success)
+    await watcher.package_update(package_ahriman, BuildStatusEnum.Success)
     _, status = watcher._known[package_ahriman.base]
     assert status.is_held is True
 
 
-def test_status_update(watcher: Watcher) -> None:
+async def test_packages(watcher: Watcher, package_ahriman: Package) -> None:
+    """
+    must return list of available packages
+    """
+    assert not await watcher.packages()
+
+    watcher._known = {package_ahriman.base: (package_ahriman, BuildStatus())}
+    assert await watcher.packages()
+
+
+async def test_shutdown(watcher: Watcher, mocker: MockerFixture) -> None:
+    """
+    must gracefully shutdown watcher
+    """
+    shutdown_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.shutdown")
+    await watcher.shutdown()
+    shutdown_mock.assert_called_once_with()
+
+
+async def test_status_update(watcher: Watcher, mocker: MockerFixture) -> None:
     """
     must update service status
     """
-    watcher.status_update(BuildStatusEnum.Success)
+    broadcast_mock = mocker.patch("ahriman.core.status.event_bus.EventBus.broadcast")
+
+    await watcher.status_update(BuildStatusEnum.Success)
     assert watcher.status.status == BuildStatusEnum.Success
+    broadcast_mock.assert_called_once_with(EventType.ServiceStatusChanged, None, status=BuildStatusEnum.Success.value)
 
 
 def test_call(watcher: Watcher, package_ahriman: Package) -> None:
@@ -204,18 +383,3 @@ def test_call_failed(watcher: Watcher, package_ahriman: Package) -> None:
     """
     with pytest.raises(UnknownPackageError):
         assert watcher(package_ahriman.base)
-
-
-def test_getattr(watcher: Watcher) -> None:
-    """
-    must return client method call
-    """
-    assert watcher.package_logs_remove
-
-
-def test_getattr_unknown_method(watcher: Watcher) -> None:
-    """
-    must raise AttributeError in case if no reporter attribute found
-    """
-    with pytest.raises(AttributeError):
-        assert watcher.random_method

tests/ahriman/web/schemas/test_event_bus_filter_schema.py

@@ -0,0 +1 @@
+# schema testing goes in view class tests

tests/ahriman/web/schemas/test_sse_schema.py

@@ -0,0 +1 @@
+# schema testing goes in view class tests

tests/ahriman/web/views/v1/auditlog/test_view_v1_auditlog_event_bus.py

@@ -0,0 +1,130 @@
+import asyncio
+import pytest
+
+from aiohttp.test_utils import TestClient
+from asyncio import Queue
+from pytest_mock import MockerFixture
+from unittest.mock import AsyncMock
+
+from ahriman.core.status.watcher import Watcher
+from ahriman.models.event import EventType
+from ahriman.models.package import Package
+from ahriman.models.user_access import UserAccess
+from ahriman.web.keys import WatcherKey
+from ahriman.web.views.v1.auditlog.event_bus import EventBusView
+
+
+async def _producer(watcher: Watcher, package_ahriman: Package) -> None:
+    """
+    create producer
+
+    Args:
+        watcher(Watcher): watcher test instance
+        package_ahriman(Package): package test instance
+    """
+    await asyncio.sleep(0.1)
+    await watcher.event_bus.broadcast(EventType.PackageRemoved, package_ahriman.base)
+    await watcher.event_bus.broadcast(EventType.PackageUpdated, package_ahriman.base, status="success")
+    await asyncio.sleep(0.1)
+    await watcher.event_bus.shutdown()
+
+
+async def test_get_permission() -> None:
+    """
+    must return correct permission for the request
+    """
+    for method in ("GET",):
+        request = pytest.helpers.request("", "", method)
+        assert await EventBusView.get_permission(request) == UserAccess.Full
+
+
+def test_routes() -> None:
+    """
+    must return correct routes
+    """
+    assert EventBusView.ROUTES == ["/api/v1/events/stream"]
+
+
+async def test_run_timeout() -> None:
+    """
+    must handle timeout and continue loop
+    """
+    queue = Queue()
+
+    async def _shutdown() -> None:
+        await asyncio.sleep(0.05)
+        await queue.put(None)
+
+    response = AsyncMock()
+    response.is_connected = lambda: True
+    response.ping_interval = 0.01
+
+    asyncio.create_task(_shutdown())
+    await EventBusView._run(response, queue)
+
+
+async def test_get(client: TestClient, package_ahriman: Package) -> None:
+    """
+    must stream events via SSE
+    """
+    watcher = next(iter(client.app[WatcherKey].values()))
+    asyncio.create_task(_producer(watcher, package_ahriman))
+    request_schema = pytest.helpers.schema_request(EventBusView.get, location="querystring")
+    # no content validation here because it is a streaming response
+
+    assert not request_schema.validate({})
+    response = await client.get("/api/v1/events/stream")
+    assert response.status == 200
+
+    body = await response.text()
+    assert EventType.PackageUpdated in body
+    assert "ahriman" in body
+
+
+async def test_get_with_topic_filter(client: TestClient, package_ahriman: Package) -> None:
+    """
+    must filter events by topic
+    """
+    watcher = next(iter(client.app[WatcherKey].values()))
+    asyncio.create_task(_producer(watcher, package_ahriman))
+    request_schema = pytest.helpers.schema_request(EventBusView.get, location="querystring")
+
+    payload = {"event": [EventType.PackageUpdated]}
+    assert not request_schema.validate(payload)
+    response = await client.get("/api/v1/events/stream", params=payload)
+    assert response.status == 200
+
+    body = await response.text()
+    assert EventType.PackageUpdated in body
+    assert EventType.PackageRemoved not in body
+
+
+async def test_get_bad_request(client: TestClient) -> None:
+    """
+    must return bad request for invalid event type
+    """
+    response_schema = pytest.helpers.schema_response(EventBusView.get, code=400)
+
+    response = await client.get("/api/v1/events/stream", params={"event": "invalid"})
+    assert response.status == 400
+    assert not response_schema.validate(await response.json())
+
+
+async def test_get_not_found(client: TestClient) -> None:
+    """
+    must return not found for unknown repository
+    """
+    response_schema = pytest.helpers.schema_response(EventBusView.get, code=404)
+
+    response = await client.get("/api/v1/events/stream", params={"architecture": "unknown", "repository": "unknown"})
+    assert response.status == 404
+    assert not response_schema.validate(await response.json())
+
+
+async def test_get_connection_reset(client: TestClient, mocker: MockerFixture) -> None:
+    """
+    must handle connection reset
+    """
+    mocker.patch.object(EventBusView, "_run", side_effect=ConnectionResetError)
+    response = await client.get("/api/v1/events/stream")
+    assert response.status == 200