hotfix: pictique message disappearing

Author: coodos

platforms/pictique-api/src/services/MessageService.ts

@@ -7,7 +7,7 @@ export class MessageService {
     private chatRepository = AppDataSource.getRepository(Chat);
 
     async findById(id: string): Promise<Message | null> {
-        return await this.messageRepository.findOne({ 
+        return await this.messageRepository.findOne({
             where: { id },
             relations: ["chat", "sender"]
         });
@@ -23,7 +23,7 @@ export class MessageService {
         });
 
         const savedMessage = await this.messageRepository.save(message);
-        
+
         // Update the chat's updatedAt timestamp to reflect the latest message
         const chat = await this.chatRepository.findOneBy({ id: chatId });
         if (chat) {
@@ -37,4 +37,4 @@ export class MessageService {
     async createSystemMessage(chatId: string, text: string): Promise<Message> {
         return this.createMessage(null, chatId, text, true);
     }
-} 
+}

platforms/pictique-api/src/web3adapter/watchers/subscriber.ts

@@ -34,6 +34,7 @@ const JUNCTION_TABLE_MAP = {
 export class PostgresSubscriber implements EntitySubscriberInterface {
     private adapter: Web3Adapter;
     private pendingChanges: Map<string, number> = new Map();
+    private recentMessageActivity: Map<string, number> = new Map(); // chatId -> timestamp
 
     constructor() {
         this.adapter = adapter;
@@ -42,6 +43,11 @@ export class PostgresSubscriber implements EntitySubscriberInterface {
         setInterval(() => {
             this.cleanupOldPendingChanges();
         }, 5 * 60 * 1000);
+        
+        // Clean up old message activity every minute
+        setInterval(() => {
+            this.cleanupOldMessageActivity();
+        }, 60 * 1000);
     }
 
     /**
@@ -59,6 +65,20 @@ export class PostgresSubscriber implements EntitySubscriberInterface {
         }
     }
 
+    /**
+     * Clean up old message activity to prevent memory leaks
+     */
+    private cleanupOldMessageActivity(): void {
+        const now = Date.now();
+        const maxAge = 10 * 1000; // 10 seconds
+        
+        for (const [chatId, timestamp] of this.recentMessageActivity.entries()) {
+            if (now - timestamp > maxAge) {
+                this.recentMessageActivity.delete(chatId);
+            }
+        }
+    }
+
     /**
      * Called after entity is loaded.
      */
@@ -203,6 +223,28 @@ export class PostgresSubscriber implements EntitySubscriberInterface {
         const data = this.entityToPlain(entity);
         if (!data.id) return;
 
+        // For messages, track activity per chat to prevent race conditions
+        if (tableName === "messages") {
+            const chatId = data.chat?.id;
+            if (chatId) {
+                console.log(`Recording message activity for chat: ${chatId}`);
+                this.recentMessageActivity.set(chatId, Date.now());
+            }
+        }
+
+        // For chats, check if there was recent message activity
+        if (tableName === "chats") {
+            const chatId = data.id;
+            const lastMessageActivity = this.recentMessageActivity.get(chatId);
+            if (lastMessageActivity) {
+                const timeSinceLastMessage = Date.now() - lastMessageActivity;
+                if (timeSinceLastMessage < 5000) { // 5 seconds
+                    console.log(`Skipping chat ${chatId} change, recent message activity ${timeSinceLastMessage}ms ago`);
+                    return;
+                }
+            }
+        }
+
         // Create a unique key for this entity change to prevent duplicates
         const changeKey = `${tableName}:${entity.id}`;