refactor: webhook events (#1661)

* refactored EventBusService to work outside FastAPI

* extended event models

* refactored webhooks to run through event bus

* added basic webhook test route

* changed get_all to page_all

* fixed incorrectly implemented Vue variables

* fixed broken webhook test

* changed factory from staticmethod to classmethod

* made query boundary definitions easier to read

frontend/components/Domain/Group/GroupWebhookEditor.vue

@@ -1,9 +1,9 @@
 <template>
   <div>
     <v-card-text>
-      <v-switch v-model="webhookCopy.enabled" label="$t('general.enabled')"></v-switch>
+      <v-switch v-model="webhookCopy.enabled" :label="$t('general.enabled')"></v-switch>
-      <v-text-field v-model="webhookCopy.name" label="$t('settings.webhooks.webhook-name')"></v-text-field>
+      <v-text-field v-model="webhookCopy.name" :label="$t('settings.webhooks.webhook-name')"></v-text-field>
-      <v-text-field v-model="webhookCopy.url" label="$t('settings.webhooks.webhook-url')"></v-text-field>
+      <v-text-field v-model="webhookCopy.url" :label="$t('settings.webhooks.webhook-url')"></v-text-field>
       <v-time-picker v-model="scheduledTime" class="elevation-2" ampm-in-title format="ampm"></v-time-picker>
     </v-card-text>
     <v-card-actions class="py-0 justify-end">

mealie/routes/_base/base_controllers.py

@@ -122,7 +122,7 @@ class BaseCrudController(BaseUserController):
     Base class for all CRUD controllers to facilitate common CRUD functions.
     """
 
-    event_bus: EventBusService = Depends(EventBusService)
+    event_bus: EventBusService = Depends(EventBusService.create)
 
     def publish_event(self, event_type: EventTypes, document_data: EventDocumentDataBase, message: str = "") -> None:
         self.event_bus.dispatch(

mealie/routes/groups/controller_group_notifications.py

@@ -35,7 +35,7 @@ router = APIRouter(
 
 @controller(router)
 class GroupEventsNotifierController(BaseUserController):
-    event_bus: EventBusService = Depends(EventBusService)
+    event_bus: EventBusService = Depends(EventBusService.create)
 
     @cached_property
     def repo(self):

mealie/routes/groups/controller_webhooks.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from functools import cached_property
 
 from fastapi import APIRouter, Depends
@@ -9,6 +10,7 @@ from mealie.routes._base.mixins import HttpRepo
 from mealie.schema import mapper
 from mealie.schema.group.webhook import CreateWebhook, ReadWebhook, SaveWebhook, WebhookPagination
 from mealie.schema.response.pagination import PaginationQuery
+from mealie.services.scheduler.tasks.post_webhooks import post_group_webhooks
 
 router = APIRouter(prefix="/groups/webhooks", tags=["Groups: Webhooks"])
 
@@ -38,6 +40,14 @@ class ReadWebhookController(BaseUserController):
         save = mapper.cast(data, SaveWebhook, group_id=self.group.id)
         return self.mixins.create_one(save)
 
+    @router.post("/rerun")
+    def rerun_webhooks(self):
+        """Manually re-fires all previously scheduled webhooks for today"""
+
+        start_time = datetime.min.time()
+        start_dt = datetime.combine(datetime.utcnow().date(), start_time)
+        post_group_webhooks(start_dt=start_dt, group_id=self.group.id)
+
     @router.get("/{item_id}", response_model=ReadWebhook)
     def get_one(self, item_id: UUID4):
         return self.mixins.get_one(item_id)

mealie/schema/group/group_events.py

@@ -13,6 +13,9 @@ class GroupEventNotifierOptions(MealieModel):
     If you modify this, make sure to update the EventBusService as well.
     """
 
+    test_message: bool = False
+    webhook_task: bool = False
+
     recipe_created: bool = False
     recipe_updated: bool = False
     recipe_deleted: bool = False

mealie/services/event_bus_service/event_bus_listeners.py

@@ -1,15 +1,21 @@
 import json
+from datetime import datetime, timezone
+from typing import cast
 from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
 
 from fastapi.encoders import jsonable_encoder
 from pydantic import UUID4
 from sqlalchemy.orm.session import Session
 
+from mealie.db.models.group.webhooks import GroupWebhooksModel
+from mealie.repos.all_repositories import get_repositories
 from mealie.repos.repository_factory import AllRepositories
 from mealie.schema.group.group_events import GroupEventNotifierPrivate
+from mealie.schema.group.webhook import ReadWebhook
+from mealie.schema.response.pagination import PaginationQuery
 
-from .event_types import Event
+from .event_types import Event, EventDocumentType, EventTypes, EventWebhookData
-from .publisher import ApprisePublisher, PublisherLike
+from .publisher import ApprisePublisher, PublisherLike, WebhookPublisher
 
 
 class EventListenerBase:
@@ -78,3 +84,48 @@ class AppriseEventListener(EventListenerBase):
     @staticmethod
     def is_custom_url(url: str):
         return url.split(":", 1)[0].lower() in ["form", "forms", "json", "jsons", "xml", "xmls"]
+
+
+class WebhookEventListener(EventListenerBase):
+    def __init__(self, session: Session, group_id: UUID4) -> None:
+        super().__init__(session, group_id, WebhookPublisher())
+        self.repos = get_repositories(session)
+
+    def get_subscribers(self, event: Event) -> list[ReadWebhook]:
+        # we only care about events that contain webhook information
+        if not (event.event_type == EventTypes.webhook_task and isinstance(event.document_data, EventWebhookData)):
+            return []
+
+        scheduled_webhooks = self.get_scheduled_webhooks(
+            event.document_data.webhook_start_dt, event.document_data.webhook_end_dt
+        )
+
+        return scheduled_webhooks
+
+    def publish_to_subscribers(self, event: Event, subscribers: list[ReadWebhook]) -> None:
+        match event.document_data.document_type:
+            case EventDocumentType.mealplan:
+                # TODO: limit mealplan data to a date range instead of returning all mealplans
+                meal_repo = self.repos.meals.by_group(self.group_id)
+                meal_pagination_data = meal_repo.page_all(pagination=PaginationQuery(page=1, per_page=-1))
+                meal_data = meal_pagination_data.items
+                if meal_data:
+                    webhook_data = cast(EventWebhookData, event.document_data)
+                    webhook_data.webhook_body = meal_data
+                    self.publisher.publish(event, [webhook.url for webhook in subscribers])
+
+            case _:
+                # if the document type is not supported, do nothing
+                pass
+
+    def get_scheduled_webhooks(self, start_dt: datetime, end_dt: datetime) -> list[ReadWebhook]:
+        """Fetches all scheduled webhooks from the database"""
+        return (
+            self.session.query(GroupWebhooksModel)
+            .where(
+                GroupWebhooksModel.enabled == True,  # noqa: E712 - required for SQLAlchemy comparison
+                GroupWebhooksModel.scheduled_time > start_dt.astimezone(timezone.utc).time(),
+                GroupWebhooksModel.scheduled_time <= end_dt.astimezone(timezone.utc).time(),
+            )
+            .all()
+        )

mealie/services/event_bus_service/event_bus_service.py

@@ -2,10 +2,15 @@ from typing import Optional
 
 from fastapi import BackgroundTasks, Depends
 from pydantic import UUID4
+from sqlalchemy.orm.session import Session
 
 from mealie.core.config import get_app_settings
 from mealie.db.db_setup import generate_session
-from mealie.services.event_bus_service.event_bus_listeners import AppriseEventListener, EventListenerBase
+from mealie.services.event_bus_service.event_bus_listeners import (
+    AppriseEventListener,
+    EventListenerBase,
+    WebhookEventListener,
+)
 
 from .event_types import Event, EventBusMessage, EventDocumentDataBase, EventTypes
 
@@ -35,12 +40,20 @@ class EventSource:
 
 
 class EventBusService:
-    def __init__(self, bg: BackgroundTasks, session=Depends(generate_session)) -> None:
+    def __init__(
+        self, bg: Optional[BackgroundTasks] = None, session: Optional[Session] = None, group_id: UUID4 | None = None
+    ) -> None:
+        if not session:
+            session = next(generate_session())
+
         self.bg = bg
         self.session = session
-        self.group_id: UUID4 | None = None
+        self.group_id = group_id
 
-        self.listeners: list[EventListenerBase] = [AppriseEventListener(self.session, self.group_id)]
+        self.listeners: list[EventListenerBase] = [
+            AppriseEventListener(self.session, self.group_id),
+            WebhookEventListener(self.session, self.group_id),
+        ]
 
     def dispatch(
         self,
@@ -59,10 +72,18 @@ class EventBusService:
             document_data=document_data,
         )
 
-        self.bg.add_task(self.publish_event, event=event)
+        if self.bg:
+            self.bg.add_task(self.publish_event, event=event)
+
+        else:
+            self.publish_event(event)
 
     def publish_event(self, event: Event) -> None:
         """Publishes the event to all listeners"""
         for listener in self.listeners:
             if subscribers := listener.get_subscribers(event):
                 listener.publish_to_subscribers(event, subscribers)
+
+    @classmethod
+    def create(cls, bg: BackgroundTasks, session=Depends(generate_session), group_id: UUID4 | None = None):
+        return cls(bg, session, group_id)

mealie/services/event_bus_service/event_types.py

@@ -1,11 +1,14 @@
 import uuid
 from datetime import datetime
 from enum import Enum, auto
+from typing import Any
 
 from pydantic import UUID4
 
 from ...schema._mealie.mealie_model import MealieModel
 
+INTERNAL_INTEGRATION_ID = "mealie_generic_user"
+
 
 class EventTypes(Enum):
     """
@@ -18,7 +21,9 @@ class EventTypes(Enum):
     (like shopping list items), modify the event document type instead (which is not tied to a database entry).
     """
 
+    # used internally and cannot be subscribed to
     test_message = auto()
+    webhook_task = auto()
 
     recipe_created = auto()
     recipe_updated = auto()
@@ -54,6 +59,7 @@ class EventDocumentType(Enum):
 
     category = "category"
     cookbook = "cookbook"
+    mealplan = "mealplan"
     shopping_list = "shopping_list"
     shopping_list_item = "shopping_list_item"
     recipe = "recipe"
@@ -122,6 +128,12 @@ class EventTagData(EventDocumentDataBase):
     tag_id: UUID4
 
 
+class EventWebhookData(EventDocumentDataBase):
+    webhook_start_dt: datetime
+    webhook_end_dt: datetime
+    webhook_body: Any
+
+
 class EventBusMessage(MealieModel):
     title: str
     body: str = ""

mealie/services/event_bus_service/publisher.py

@@ -1,6 +1,8 @@
 from typing import Protocol
 
 import apprise
+import requests
+from fastapi.encoders import jsonable_encoder
 
 from mealie.services.event_bus_service.event_types import Event
 
@@ -34,3 +36,15 @@ class ApprisePublisher:
                 raise Exception("Apprise URL Add Failed")
 
         self.apprise.notify(title=event.message.title, body=event.message.body, tag=tags)
+
+
+class WebhookPublisher:
+    def __init__(self, hard_fail=False) -> None:
+        self.hard_fail = hard_fail
+
+    def publish(self, event: Event, notification_urls: list[str]):
+        event_payload = jsonable_encoder(event)
+        for url in notification_urls:
+            r = requests.post(url, json=event_payload, timeout=15)
+            if self.hard_fail:
+                r.raise_for_status()

mealie/services/scheduler/tasks/post_webhooks.py

@@ -1,54 +1,63 @@
 from datetime import datetime, timezone
+from typing import Optional
 
-import requests
-from fastapi.encoders import jsonable_encoder
 from pydantic import UUID4
-from sqlalchemy.orm import Session
 
 from mealie.db.db_setup import create_session
-from mealie.db.models.group.webhooks import GroupWebhooksModel
 from mealie.repos.all_repositories import get_repositories
+from mealie.schema.response.pagination import PaginationQuery
+from mealie.services.event_bus_service.event_bus_service import EventBusService
+from mealie.services.event_bus_service.event_types import (
+    INTERNAL_INTEGRATION_ID,
+    EventDocumentType,
+    EventOperation,
+    EventTypes,
+    EventWebhookData,
+)
 
 last_ran = datetime.now(timezone.utc)
 
 
-def get_scheduled_webhooks(session: Session, bottom: datetime, top: datetime) -> list[GroupWebhooksModel]:
+def post_group_webhooks(start_dt: Optional[datetime] = None, group_id: Optional[UUID4] = None) -> None:
+    """Post webhook events to specified group, or all groups"""
+
+    global last_ran
+
+    # if not specified, start the query at the last time the service ran
+    start_dt = start_dt or last_ran
+
+    # end the query at the current time
+    last_ran = end_dt = datetime.now(timezone.utc)
+
+    if group_id is None:
+        # publish the webhook event to each group's event bus
+        session = create_session()
+        repos = get_repositories(session)
+        groups_data = repos.groups.page_all(PaginationQuery(page=1, per_page=-1))
+        group_ids = [group.id for group in groups_data.items]
+
+    else:
+        group_ids = [group_id]
+
     """
-    get_scheduled_webhooks queries the database for all webhooks scheduled between the bottom and
+    At this time only mealplan webhooks are supported. To add support for more types,
-    top time ranges. It returns a list of GroupWebhooksModel objects.
+    add a dispatch event for that type here (e.g. EventDocumentType.recipe_bulk_report) and
+    handle the webhook data in the webhook event bus listener
     """
 
-    return (
+    event_type = EventTypes.webhook_task
-        session.query(GroupWebhooksModel)
+    event_document_data = EventWebhookData(
-        .where(
+        document_type=EventDocumentType.mealplan,
-            GroupWebhooksModel.enabled == True,  # noqa: E712 - required for SQLAlchemy comparison
+        operation=EventOperation.info,
-            GroupWebhooksModel.scheduled_time > bottom.astimezone(timezone.utc).time(),
+        webhook_start_dt=start_dt,
-            GroupWebhooksModel.scheduled_time <= top.astimezone(timezone.utc).time(),
+        webhook_end_dt=end_dt,
-        )
-        .all()
     )
 
-
+    for group_id in group_ids:
-def post_group_webhooks() -> None:
+        event_bus = EventBusService(group_id=group_id)
-    global last_ran
+        event_bus.dispatch(
-    session = create_session()
+            integration_id=INTERNAL_INTEGRATION_ID,
-    results = get_scheduled_webhooks(session, last_ran, datetime.now())
+            group_id=group_id,
-
+            event_type=event_type,
-    last_ran = datetime.now(timezone.utc)
+            document_data=event_document_data,
-
+        )
-    repos = get_repositories(session)
-
-    memo = {}
-
-    def get_meals(group_id: UUID4):
-        if group_id not in memo:
-            memo[group_id] = repos.meals.get_all(group_id=group_id)
-        return memo[group_id]
-
-    for result in results:
-        meals = get_meals(result.group_id)
-
-        if not meals:
-            continue
-
-        requests.post(result.url, json=jsonable_encoder(meals))

tests/unit_tests/services_tests/scheduler/tasks/test_post_webhook.py

@@ -4,7 +4,7 @@ from pydantic import UUID4
 
 from mealie.repos.repository_factory import AllRepositories
 from mealie.schema.group.webhook import SaveWebhook, WebhookType
-from mealie.services.scheduler.tasks.post_webhooks import get_scheduled_webhooks
+from mealie.services.event_bus_service.event_bus_listeners import WebhookEventListener
 from tests.utils import random_string
 from tests.utils.factories import random_bool
 from tests.utils.fixture_schemas import TestUser
@@ -30,7 +30,7 @@ def webhook_factory(
 
 def test_get_scheduled_webhooks_filter_query(database: AllRepositories, unique_user: TestUser):
     """
-    get_scheduled_webhooks_test tests the get_scheduled_webhooks function.
+    get_scheduled_webhooks_test tests the get_scheduled_webhooks function on the webhook event bus listener.
     """
 
     expected: list[SaveWebhook] = []
@@ -51,7 +51,8 @@ def test_get_scheduled_webhooks_filter_query(database: AllRepositories, unique_u
         if new_item.enabled:
             expected.append(new_item)
 
-    results = get_scheduled_webhooks(database.session, start, datetime.now() + timedelta(minutes=5))
+    event_bus_listener = WebhookEventListener(database.session, unique_user.group_id)  # type: ignore
+    results = event_bus_listener.get_scheduled_webhooks(start, datetime.now() + timedelta(minutes=5))
 
     assert len(results) == len(expected)