From aad6f59e96fb3306d54a0c4df7ea80ac310af26a Mon Sep 17 00:00:00 2001
From: Carlos Quintana <74399022+cquintana92@users.noreply.github.com>
Date: Thu, 23 May 2024 15:05:47 +0200
Subject: [PATCH] Improve error handling on event sink (#2117)

* chore: make event_sink return success

* fix: add return to ConsoleEventSink
---
 events/event_sink.py | 12 ++++++++----
 events/runner.py     | 31 +++++++++++++++++--------------
 2 files changed, 25 insertions(+), 18 deletions(-)

diff --git a/events/event_sink.py b/events/event_sink.py
index a51b9e8c..bc00d1f9 100644
--- a/events/event_sink.py
+++ b/events/event_sink.py
@@ -8,15 +8,16 @@ 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):
+    def process(self, event: SyncEvent) -> bool:
         if not EVENT_WEBHOOK:
             LOG.warning("Skipping sending event because there is no webhook configured")
-            return
+            return False
+
         LOG.info(f"Sending event {event.id} to {EVENT_WEBHOOK}")
 
         res = requests.post(
@@ -29,10 +30,13 @@ class HttpEventSink(EventSink):
             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
diff --git a/events/runner.py b/events/runner.py
index 7d0b4b70..e28ff152 100644
--- a/events/runner.py
+++ b/events/runner.py
@@ -22,22 +22,25 @@ class Runner:
             if can_process:
                 event_created_at = event.created_at
                 start_time = arrow.now()
-                self.__sink.process(event)
-                event_id = event.id
-                SyncEvent.delete(event.id, commit=True)
-                LOG.info(f"Marked {event_id} as done")
+                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")
 
-                end_time = arrow.now() - start_time
-                time_between_taken_and_created = start_time - event_created_at
+                    end_time = arrow.now() - start_time
+                    time_between_taken_and_created = start_time - event_created_at
 
-                newrelic.agent.record_custom_metric("Custom/sync_event_processed", 1)
-                newrelic.agent.record_custom_metric(
-                    "Custom/sync_event_process_time", end_time.total_seconds()
-                )
-                newrelic.agent.record_custom_metric(
-                    "Custom/sync_event_elapsed_time",
-                    time_between_taken_and_created.total_seconds(),
-                )
+                    newrelic.agent.record_custom_metric(
+                        "Custom/sync_event_processed", 1
+                    )
+                    newrelic.agent.record_custom_metric(
+                        "Custom/sync_event_process_time", end_time.total_seconds()
+                    )
+                    newrelic.agent.record_custom_metric(
+                        "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: