pickle -> sqlite store: D6 async writes, D9 reload race, one-shot pickle migration

fjerkroa_bot/ai_responder.py

@@ -12,6 +12,8 @@ from pathlib import Path
 from pprint import pformat
 from typing import Any, Dict, List, Optional, Tuple, Union
 
+from .persistence import PersistentStore
+
 
 def pp(*args, **kw):
     if "width" not in kw:
@@ -138,17 +140,16 @@ class AIResponder(AIResponderBase):
         self.history: List[Dict[str, Any]] = []
         self.memory: str = "I am an assistant."
         self.rate_limit_backoff = exponential_backoff()
-        self.history_file: Optional[Path] = None
-        self.memory_file: Optional[Path] = None
+        self.store: Optional[PersistentStore] = None
         if "history-directory" in self.config:
-            self.history_file = Path(self.config["history-directory"]).expanduser() / f"{self.channel}.dat"
-            if self.history_file.exists():
-                with open(self.history_file, "rb") as fd:
-                    self.history = pickle.load(fd)
-            self.memory_file = Path(self.config["history-directory"]).expanduser() / f"{self.channel}.memory"
-            if self.memory_file.exists():
-                with open(self.memory_file, "rb") as fd:
-                    self.memory = pickle.load(fd)
+            directory = Path(self.config["history-directory"]).expanduser()
+            self.store = PersistentStore(directory / "bot.db")
+            # Legacy pickles import once, then live on as *.migrated (PER-03)
+            self.store.migrate_pickles(self.channel, directory / f"{self.channel}.dat", directory / f"{self.channel}.memory")
+            self.history = self.store.load_history(self.channel)
+            stored_memory = self.store.load_memory(self.channel)
+            if stored_memory is not None:
+                self.memory = stored_memory
         logging.info(f"memmory:\n{self.memory}")
 
     def message(self, message: AIMessage, limit: Optional[int] = None) -> List[Dict[str, Any]]:
@@ -225,9 +226,6 @@ class AIResponder(AIResponderBase):
                 self.history.append({"role": "user", "content": str(message)})
                 while len(self.history) > limit:
                     self.shrink_history_by_one()
-                if self.history_file is not None:
-                    with open(self.history_file, "wb") as fd:
-                        pickle.dump(self.history, fd)
                 return True
         return False
 
@@ -269,15 +267,17 @@ class AIResponder(AIResponderBase):
         self.history.append(answer)
         while len(self.history) > limit:
             self.shrink_history_by_one()
-        if self.history_file is not None:
-            with open(self.history_file, "wb") as fd:
-                pickle.dump(self.history, fd)
 
     def update_memory(self, memory) -> None:
         self.memory = memory
-        if self.memory_file is not None:
-            with open(self.memory_file, "wb") as fd:
-                pickle.dump(self.memory, fd)
+
+    async def _persist_history(self) -> None:
+        if self.store is not None:
+            await asyncio.to_thread(self.store.save_history, self.channel, list(self.history))
+
+    async def _persist_memory(self) -> None:
+        if self.store is not None:
+            await asyncio.to_thread(self.store.save_memory, self.channel, self.memory)
 
     async def handle_picture(self, response: Dict) -> bool:
         if not isinstance(response.get("picture"), (type(None), str)):
@@ -299,6 +299,7 @@ class AIResponder(AIResponderBase):
     async def memoize(self, message_user: str, answer_user: str, message: str, answer: str) -> None:
         self.memory = await self.memory_rewrite(self.memory, message_user, answer_user, message, answer)
         self.update_memory(self.memory)
+        await self._persist_memory()
 
     async def memoize_reaction(self, message_user: str, reaction_user: str, operation: str, reaction: str, message: str) -> None:
         quoted_message = message.replace("\n", "\n> ")
@@ -312,6 +313,7 @@ class AIResponder(AIResponderBase):
 
         # Check if a short path applies, return an empty AIResponse if it does
         if self.short_path(message, limit):
+            await self._persist_history()
             return AIResponse(None, False, None, None, None, False, False)
 
         # Number of retries for sending the message; failed attempts are
@@ -347,8 +349,9 @@ class AIResponder(AIResponderBase):
             answer_message = await self.post_process(message, response)
             answer["content"] = str(answer_message)
 
-            # Update message history
+            # Update message history; persistence runs off the loop (PER-05)
             self.update_history(messages[-1], answer, limit, message.historise_question)
+            await self._persist_history()
             logging.info(f"got this answer:\n{str(answer_message)}")
 
             # Update memory