4.44.3

app/events/event_sink.py

@@ -1,19 +1,42 @@
+import requests
+
 from abc import ABC, abstractmethod
+from app.config import EVENT_WEBHOOK, EVENT_WEBHOOK_SKIP_VERIFY_SSL
 from app.log import LOG
 from app.models import SyncEvent
 
 
 class EventSink(ABC):
     @abstractmethod
-    def process(self, event: SyncEvent):
+    def process(self, event: SyncEvent) -> bool:
         pass
 
 
 class HttpEventSink(EventSink):
-    def process(self, event: SyncEvent):
-        pass
+    def process(self, event: SyncEvent) -> bool:
+        if not EVENT_WEBHOOK:
+            LOG.warning("Skipping sending event because there is no webhook configured")
+            return False
+
+        LOG.info(f"Sending event {event.id} to {EVENT_WEBHOOK}")
+
+        res = requests.post(
+            url=EVENT_WEBHOOK,
+            data=event.content,
+            headers={"Content-Type": "application/x-protobuf"},
+            verify=not EVENT_WEBHOOK_SKIP_VERIFY_SSL,
+        )
+        if res.status_code != 200:
+            LOG.warning(
+                f"Failed to send event to webhook: {res.status_code} {res.text}"
+            )
+            return False
+        else:
+            LOG.info(f"Event {event.id} sent successfully to webhook")
+            return True
 
 
 class ConsoleEventSink(EventSink):
-    def process(self, event: SyncEvent):
+    def process(self, event: SyncEvent) -> bool:
         LOG.info(f"Handling event {event.id}")
+        return True

app/events/event_source.py

@@ -13,6 +13,8 @@ from typing import Callable, NoReturn
 _DEAD_LETTER_THRESHOLD_MINUTES = 10
 _DEAD_LETTER_INTERVAL_SECONDS = 30
 
+_POSTGRES_RECONNECT_INTERVAL_SECONDS = 5
+
 
 class EventSource(ABC):
     @abstractmethod
@@ -22,9 +24,19 @@ class EventSource(ABC):
 
 class PostgresEventSource(EventSource):
     def __init__(self, connection_string: str):
-        self.__connection = psycopg2.connect(connection_string)
+        self.__connection_string = connection_string
+        self.__connect()
 
     def run(self, on_event: Callable[[SyncEvent], NoReturn]):
+        while True:
+            try:
+                self.__listen(on_event)
+            except Exception as e:
+                LOG.warn(f"Error listening to events: {e}")
+                sleep(_POSTGRES_RECONNECT_INTERVAL_SECONDS)
+                self.__connect()
+
+    def __listen(self, on_event: Callable[[SyncEvent], NoReturn]):
         self.__connection.set_isolation_level(
             psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT
         )
@@ -44,12 +56,24 @@ class PostgresEventSource(EventSource):
                         webhook_id = int(notify.payload)
                         event = SyncEvent.get_by(id=webhook_id)
                         if event is not None:
-                            on_event(event)
+                            if event.mark_as_taken():
+                                on_event(event)
+                            else:
+                                LOG.info(
+                                    f"Event {event.id} was handled by another runner"
+                                )
                         else:
                             LOG.info(f"Could not find event with id={notify.payload}")
                     except Exception as e:
                         LOG.warn(f"Error getting event: {e}")
 
+    def __connect(self):
+        self.__connection = psycopg2.connect(self.__connection_string)
+
+        from app.db import Session
+
+        Session.close()
+
 
 class DeadLetterEventSource(EventSource):
     @newrelic.agent.background_task()
@@ -73,3 +97,4 @@ class DeadLetterEventSource(EventSource):
                     sleep(_DEAD_LETTER_INTERVAL_SECONDS)
             except Exception as e:
                 LOG.warn(f"Error getting dead letter event: {e}")
+                sleep(_DEAD_LETTER_INTERVAL_SECONDS)

app/events/runner.py

@@ -18,11 +18,10 @@ class Runner:
     @newrelic.agent.background_task()
     def __on_event(self, event: SyncEvent):
         try:
-            can_process = event.mark_as_taken()
-            if can_process:
-                event_created_at = event.created_at
-                start_time = arrow.now()
-                self.__sink.process(event)
+            event_created_at = event.created_at
+            start_time = arrow.now()
+            success = self.__sink.process(event)
+            if success:
                 event_id = event.id
                 SyncEvent.delete(event.id, commit=True)
                 LOG.info(f"Marked {event_id} as done")
@@ -38,8 +37,6 @@ class Runner:
                     "Custom/sync_event_elapsed_time",
                     time_between_taken_and_created.total_seconds(),
                 )
-            else:
-                LOG.info(f"{event.id} was handled by another runner")
         except Exception as e:
             LOG.warn(f"Exception processing event [id={event.id}]: {e}")
             newrelic.agent.record_custom_metric("Custom/sync_event_failed", 1)