Better launch, cleaner shutdown (wip)

news_fetch/app/utils_slack/message_helpers.py

@@ -14,6 +14,7 @@ LATEST_RECORDED_REACTION = 0
 
 
 def init(client) -> None:
+    """Starts fetching past messages and returns the freshly launched thread"""
     global slack_client
     slack_client = client
 
@@ -26,7 +27,7 @@ def init(client) -> None:
     # fetch all te messages we could have possibly missed
     logger.info("Querying missed messages, threads and reactions. This can take some time.")
     fetch_missed_channel_messages() # not threaded
-    t = Thread(target = fetch_missed_channel_reactions) # threaded, runs in background (usually takes a long time)
+    t = Thread(target = fetch_missed_channel_reactions, daemon=True) # threaded, runs in background (usually takes a long time)
     t.start()
 
     if os.getenv("REDUCEDFETCH", "false") == "true":
@@ -153,16 +154,23 @@ def fetch_missed_channel_reactions():
     logger.info("Starting background fetch of channel reactions...")
     threads = [t for t in models.Thread.select() if not t.is_fully_processed]
     for i,t in enumerate(threads):
+        reactions = []
         try:
             query = slack_client.reactions_get(
                 channel = config["archive_id"],
                 timestamp = t.slack_ts
             )
             reactions = query.get("message", []).get("reactions", []) # default = []
-        except SlackApiError: # probably a rate_limit:
-            logger.error("Hit rate limit while querying reactions. retrying in {}s ({}/{} queries elapsed)".format(config["api_wait_time"], i, len(threads)))
-            time.sleep(int(config["api_wait_time"]))
-            reactions = query.get("message", []).get("reactions", [])
+        except SlackApiError as e:
+            if e.response.get("error", "") == "message_not_found":
+                m = t.initiator_message
+                logger.warning(f"Message (id={m.id}) not found. Skipping and saving...")
+                # this usually means the message is past the 1000 message limit imposed by slack. Mark it as processed in the db
+                m.is_processed_override = True
+                m.save()
+            else: # probably a rate_limit:
+                logger.error("Hit rate limit while querying reactions. retrying in {}s ({}/{} queries elapsed)".format(config["api_wait_time"], i, len(threads)))
+                time.sleep(int(config["api_wait_time"]))
 
         for r in reactions:
             reaction_dict_to_model(r, t)

news_fetch/app/utils_slack/runner.py

@@ -1,5 +1,6 @@
 from slack_bolt import App
 from slack_bolt.adapter.socket_mode import SocketModeHandler
+from slack_sdk.errors import SlackApiError
 
 import logging
 import configuration
@@ -18,7 +19,7 @@ class BotApp(App):
         super().__init__(*args, **kwargs)
         self.callback = callback
 
-    def start(self):
+    def pre_start(self):
         message_helpers.init(self.client)
         missed_messages, missed_reactions = message_helpers.get_unhandled_messages()
 
@@ -124,7 +125,7 @@ class BotApp(App):
         answers = article.slack_info
         for a in answers:
             if a["file_path"]:
-                try: # either, a["file_path"] does not exist, or the upload resulted in an error
+                try: # upload resulted in an error
                     self.client.files_upload(
                         channels = config["archive_id"],
                         initial_comment = f"<@{config['responsible_id']}> \n {a['reply_text']}",
@@ -132,12 +133,13 @@ class BotApp(App):
                         thread_ts = thread.slack_ts
                     )
                     status = True
-                except:
+                except SlackApiError as e:
                     say(
                         "File {} could not be uploaded.".format(a),
                         thread_ts=thread.slack_ts
                     )
                     status = False
+                    self.logger.error(f"File upload failed: {e}")
             else: # anticipated that there is no file!
                 say(
                     f"<@{config['responsible_id']}> \n {a['reply_text']}",
@@ -171,14 +173,17 @@ class BotRunner():
         def handle_incoming_reaction(event, say):
             return self.bot_worker.handle_incoming_reaction(event)
 
-        # target = self.launch
-        # super().__init__(target=target)
+        self.handler = SocketModeHandler(self.bot_worker, config["app_token"])
 
 
     def start(self):
-        self.bot_worker.start()
-        SocketModeHandler(self.bot_worker, config["app_token"]).start()
+        self.bot_worker.pre_start()
+        self.handler.start()
 
 
+    def stop(self):
+        self.handler.close()
+        print("Bye handler!")
+
     # def respond_to_message(self, message):
     #     self.bot_worker.handle_incoming_message(message)