all

src/bots/services/BalanceUpdateService.ts

@@ -2,6 +2,7 @@ import { supabase } from '../../db/supabase';
 import { createLogger } from '../../utils/logger';
 import { BotContext } from '../types/botTypes';
 import { messageManager } from '../utils/messageManager';
+import { initializeBot, stopBot } from '../botManager';
 
 const logger = createLogger('BalanceUpdateService');
 
@@ -9,10 +10,14 @@ export class BalanceUpdateService {
   private static instance: BalanceUpdateService;
   private bot: any;
   private subscription: any;
+  private botsSubscription: any;
 
   private constructor() {
     // Delay the initial subscription setup to ensure bot is ready
-    setTimeout(() => this.setupRealtimeSubscription(), 1000);
+    setTimeout(() => {
+      this.setupRealtimeSubscription();
+      this.setupBotsSubscription();
+    }, 1000);
   }
 
   public static getInstance(): BalanceUpdateService {
@@ -100,6 +105,94 @@ export class BalanceUpdateService {
     }
   }
 
+  private async setupBotsSubscription() {
+    try {
+      logger.info('Starting bots subscription setup...');
+
+      // Unsubscribe from any existing subscription
+      if (this.botsSubscription) {
+        await this.botsSubscription.unsubscribe();
+        logger.info('Unsubscribed from previous bots subscription');
+      }
+
+      // Create a new channel for bots
+      const channel = supabase.channel('bots_changes', {
+        config: {
+          broadcast: { self: true }
+        }
+      });
+
+      // Subscribe to changes in bots table
+      channel
+        .on(
+          'postgres_changes',
+          {
+            event: '*', // Listen to all events (INSERT, UPDATE, DELETE)
+            schema: 'public',
+            table: 'bots'
+          },
+          async (payload: any) => {
+            logger.info('Received bots table change:', JSON.stringify(payload, null, 2));
+            
+            try {
+              const botId = payload.new?.id || payload.old?.id;
+              if (!botId) {
+                logger.error('No bot ID found in payload');
+                return;
+              }
+
+              // Stop the bot first
+              await stopBot(botId);
+              logger.info(`Stopped bot ${botId}`);
+
+              // If it's an UPDATE or INSERT, restart the bot with new data
+              if (payload.eventType === 'UPDATE' || payload.eventType === 'INSERT') {
+                const { data: botData, error } = await supabase
+                  .from('bots')
+                  .select('*')
+                  .eq('id', botId)
+                  .single();
+
+                if (error) {
+                  logger.error(`Error fetching bot data: ${error.message}`);
+                  return;
+                }
+
+                if (botData) {
+                  const result = await initializeBot(botData.bot_token, botData);
+                  if (result.success) {
+                    logger.info(`Bot ${botId} restarted successfully`);
+                  } else {
+                    logger.error(`Failed to restart bot ${botId}: ${result.message}`);
+                  }
+                }
+              }
+            } catch (error) {
+              logger.error('Error handling bot update:', error);
+            }
+          }
+        )
+        .subscribe((status: string) => {
+          logger.info('Bots subscription status changed:', status);
+          if (status === 'SUBSCRIBED') {
+            logger.info('Successfully subscribed to bots changes');
+            this.botsSubscription = channel;
+          } else if (status === 'CLOSED') {
+            logger.error('Bots subscription closed unexpectedly');
+            setTimeout(() => this.setupBotsSubscription(), 5000);
+          } else if (status === 'CHANNEL_ERROR') {
+            logger.error('Bots channel error occurred');
+            setTimeout(() => this.setupBotsSubscription(), 5000);
+          }
+        });
+
+      logger.info('Bots subscription setup completed');
+    } catch (error) {
+      logger.error('Error setting up bots subscription:', error);
+      setTimeout(() => this.setupBotsSubscription(), 5000);
+    }
+  }
+
   private async handleBalanceUpdate(payload: any) {
     logger.info('Processing balance update payload:', JSON.stringify(payload, null, 2));