Try again

firehose/data_stream.py

@@ -1,5 +1,5 @@
 from collections import defaultdict
-from datetime import datetime, timedelta
+from datetime import datetime, timezone
 
 from atproto import (
     AtUri,
@@ -19,13 +19,6 @@
     models.AppBskyFeedPost: models.ids.AppBskyFeedPost,
 }
 
-def is_healthy():
-    with db.connection_context():
-        state = SubscriptionState.get_or_none()
-        if not state or not state.last_indexed_at:
-            return False
-        return (datetime.now() - state.last_indexed_at) < timedelta(minutes=5)
-
 def _get_ops_by_type(commit: models.ComAtprotoSyncSubscribeRepos.Commit) -> defaultdict:
     """
     Processes a commit message and extracts operations of interest, grouping them by record type.
@@ -174,7 +167,7 @@ def on_message_handler(message: firehose_models.MessageFrame) -> None:
             with db.atomic():
                 SubscriptionState.update(
                     cursor=commit.seq,
-                    last_indexed_at=datetime.now()
+                    last_indexed_at=datetime.now(timezone.utc),
                 ).where(SubscriptionState.service == name).execute()
 
 

firehose/health_check.py

@@ -1,6 +1,15 @@
 # health_check.py
 from data_stream import is_healthy
 from utils.logger import logger
+from datetime import datetime, timedelta, timezone
+from database import db, SubscriptionState
+
+def is_healthy():
+    with db.connection_context():
+        state = SubscriptionState.get_or_none()
+        if not state or not state.last_indexed_at:
+            return False
+        return (datetime.now(timezone.utc) - state.last_indexed_at) < timedelta(minutes=5)
 
 if __name__ == "__main__":
     if is_healthy():