feat: Implement LRU-Based Account Rotation for Telegram and Twitter with Redis Integration

src/fetchTelegramMessages.ts

@@ -1,16 +1,21 @@
 import { Api, TelegramClient } from "telegram";
 import { trackApiKeyUsage } from './utils/redisUtils';
+import { TelegramAccount } from './services/telegramAccountManager';
 
 export type TelegramMessages = { id: string; content: string; channelId: string };
 
 export async function fetchTelegramMessages(
   client: TelegramClient,
-  channel: string
+  account: TelegramAccount
 ): Promise<TelegramMessages[]> {
+  const channel = account.credentials.TELEGRAM_TG_CHANNEL;
   if (!channel) {
-    throw new Error("TELEGRAM_TG_CHANNEL environment variable is not set.");
+    throw new Error("TELEGRAM_TG_CHANNEL is not set in account credentials.");
+  }
+
+  if (process.env.DEBUG_TELEGRAM === '1') {
+    console.log(`Using Telegram account: ${account.accountId} for channel: ${channel}`);
   }
-  const apiId = process.env.TELEGRAM_API_ID;
 
   // Fetch channel entity to get the actual channel ID
   let entity: Api.Channel;
@@ -52,21 +57,6 @@ export async function fetchTelegramMessages(
     console.log("No messages property found in response:", messages);
   }
 
-  // Track API usage after successful fetch
-  if (apiId) {
-    let accountId: string;
-    try {
-      const me = await client.getMe();
-      if (me && me.id) {
-        accountId = String(me.id);
-      } else {
-        throw new Error('Unable to determine Telegram account ID');
-      }
-    } catch (e) {
-      throw new Error('Unable to determine Telegram account ID');
-    }
-    await trackApiKeyUsage({ accountId, platform: 'telegram' });
-  }
 
   return out;
 }

src/index.ts

@@ -4,21 +4,40 @@ import cron from 'node-cron';
 import { TelegramClient } from "telegram";
 import { StringSession } from "telegram/sessions";
 import { fetchTelegramMessages } from './fetchTelegramMessages';
-import { getApiKeyUsage } from './utils/redisUtils';
+import { telegramAccountManager, TelegramAccount } from './services/telegramAccountManager';
 
-// Replace these with your values
-const apiId = Number(process.env.TELEGRAM_API_ID);
-const apiHash = process.env.TELEGRAM_API_HASH ?? "";
-if (!Number.isFinite(apiId)) {
-  throw new Error("API_ID environment variable is missing or not a valid number.");
-}
-if (!apiHash) {
-  throw new Error("API_HASH environment variable is not set.");
-}
-const stringSession = new StringSession(process.env.TG_SESSION ?? ""); // use existing session if available
+// Create a map to store clients for each account
+const clientMap = new Map<string, TelegramClient>();
 
-async function startTelegramCron() {
-  console.log("Starting Telegram client...");
+async function createTelegramClient(account: TelegramAccount): Promise<TelegramClient> {
+  // Check if we already have a client for this account
+  if (clientMap.has(account.accountId)) {
+    return clientMap.get(account.accountId)!;
+  }
+
+  console.log(`Creating Telegram client for account: ${account.accountId}`);
+
+  const apiId = Number(account.credentials.TELEGRAM_API_ID);
+  const apiHash = account.credentials.TELEGRAM_API_HASH;
+
+  if (!Number.isFinite(apiId)) {
+    throw new Error(`Invalid API_ID for account ${account.accountId}`);
+  }
+  if (!apiHash) {
+    throw new Error(`API_HASH not set for account ${account.accountId}`);
+  }
+
+  // Fetch per-account session string from Redis (not env)
+  const redisClient = (telegramAccountManager as any).redisClient;
+  await redisClient.connect?.();
+  const sessionKey = `telegram_session:${account.accountId}`;
+  let sessionStr = await redisClient.get(sessionKey);
+  const isInteractive = Boolean(process.stdin.isTTY);
+  if (!sessionStr && !isInteractive) {
+    throw new Error(`Missing session in Redis for ${account.accountId} (key: ${sessionKey}). Generate and store a session string before running cron.`);
+  }
+  sessionStr = sessionStr || "";
+  const stringSession = new StringSession(sessionStr);
   const client = new TelegramClient(stringSession, apiId, apiHash, {
     connectionRetries: 5,
   });
@@ -30,37 +49,50 @@ async function startTelegramCron() {
     onError: (err) => console.log(err),
   });
 
-  console.log("Logged in successfully!");
-  if (process.env.PRINT_TG_SESSION === "1") {
-    console.log("Your session string:", client.session.save());
-  } else {
-    console.log(
-      "Session created. Set PRINT_TG_SESSION=1 to print it explicitly."
-    );
+  console.log(`Logged in successfully for account: ${account.accountId}`);
+  const saved = client.session.save();
+  if (process.env.PRINT_TG_SESSION === "1" && isInteractive) {
+    // Emit an export-ready line deliberately, instead of dumping secrets in logs
+    console.log(`export ${sessionKey}="${saved}"`);
   }
 
+  // Store the client for reuse
+  clientMap.set(account.accountId, client);
+
+  return client;
+}
+
+async function startTelegramCron() {
+  console.log("Starting Telegram account rotation system...");
+
   // Run once at startup
   try {
-    await fetchTelegramMessages(client, process.env.TELEGRAM_TG_CHANNEL!);
-    // Print Telegram API usage by accountId (not API_ID)
-    const me = await client.getMe();
-    const accountId = String(me.id);
-    const usage = await getApiKeyUsage({ accountId, platform: 'telegram' });
-    console.log('Telegram API usage:', {
-      total_requests: usage.total_requests,
-      last_request: usage.last_request,
-      account_id: usage.account_id
+    const account = await telegramAccountManager.getEarliestUsedAccount();
+    const client = await createTelegramClient(account);
+
+    await fetchTelegramMessages(client, account);
+
+    // Show usage statistics for all accounts
+    const allAccounts = await telegramAccountManager.getAllAccountsUsage();
+    console.log('All Telegram accounts usage:');
+    allAccounts.forEach((acc, index) => {
+      console.log(`  Account ${index + 1} (${acc.accountId}):`);
+      console.log(`    Total requests: ${acc.totalRequests}`);
+      console.log(`    Last used: ${acc.lastUsed || 'Never'}`);
     });
   } catch (err) {
     console.error("Startup Telegram fetch failed:", err);
   }
 
-  // Schedule to run every 5 minutes (no overlap guard)
+  // Schedule to run every 5 minutes with account rotation
   cron.schedule('*/5 * * * *', async () => {
-    console.log('Refetching Telegram messages...');
+    console.log('Refetching Telegram messages with account rotation...');
     try {
-      await fetchTelegramMessages(client, process.env.TELEGRAM_TG_CHANNEL!);
-      // No duplicate print of Telegram API usage
+      const account = await telegramAccountManager.getEarliestUsedAccount();
+      const client = await createTelegramClient(account);
+
+      await fetchTelegramMessages(client, account);
+      console.log(`Fetched messages using account: ${account.accountId}`);
     } catch (err) {
       console.error('Scheduled Telegram fetch failed:', err);
     }

src/services/BaseAccountManager.ts

@@ -0,0 +1,69 @@
+import { createClient, RedisClientType } from 'redis';
+
+export interface BaseAccount {
+    accountId: string;
+    credentials: Record<string, string>;
+    lastUsed?: string;
+    totalRequests?: number;
+}
+
+export abstract class BaseAccountManager<T extends BaseAccount> {
+    protected redisClient: RedisClientType;
+    protected isConnected = false;
+    protected abstract platform: string;
+    protected abstract accountKey: string;
+    protected abstract usageKeyPrefix: string;
+
+    constructor(redisUrl?: string) {
+        this.redisClient = createClient({
+            url: redisUrl || process.env.REDIS_URL,
+        });
+        this.redisClient.on('error', (err) => {
+            console.error(`Redis Client Error in ${this.platform}AccountManager:`, err);
+        });
+    }
+
+    protected async ensureConnected(): Promise<void> {
+        if (!this.isConnected) {
+            await this.redisClient.connect();
+            this.isConnected = true;
+        }
+    }
+
+    protected abstract fetchAllAccounts(): Promise<T[]>;
+
+    async getEarliestUsedAccount(): Promise<T> {
+        await this.ensureConnected();
+        const accounts = await this.fetchAllAccounts();
+        accounts.sort((a, b) => {
+            if (!a.lastUsed && !b.lastUsed) return 0;
+            if (!a.lastUsed) return -1;
+            if (!b.lastUsed) return 1;
+            return new Date(a.lastUsed).getTime() - new Date(b.lastUsed).getTime();
+        });
+        for (const acc of accounts) {
+            const lockKey = `lock:${this.platform}:${acc.accountId}`;
+            const ok = await this.redisClient.set(lockKey, '1', { NX: true, PX: 15000 });
+            if (ok === 'OK') {
+                console.debug(`[${this.platform}AccountManager] Selected account=${acc.accountId} lastUsed=${acc.lastUsed ?? 'Never'} totalRequests=${acc.totalRequests ?? 0}`);
+                return acc;
+            }
+        }
+        throw new Error(`No available ${this.platform} accounts to claim (all locked).`);
+    }
+
+    async markAccountAsUsed(accountId: string): Promise<void> {
+        await this.ensureConnected();
+        await this.trackApiKeyUsageLocal(accountId);
+        try { await this.redisClient.del(`lock:${this.platform}:${accountId}`); } catch { }
+    }
+
+    protected abstract trackApiKeyUsageLocal(accountId: string): Promise<void>;
+
+    async disconnect(): Promise<void> {
+        if (this.isConnected) {
+            await this.redisClient.quit();
+            this.isConnected = false;
+        }
+    }
+}

src/services/telegramAccountManager.ts

@@ -0,0 +1,106 @@
+import { BaseAccountManager, BaseAccount } from './BaseAccountManager';
+import { decrypt } from '../lib/encryption';
+import { getApiKeyUsage, trackApiKeyUsage } from '../utils/redisUtils';
+
+export interface TelegramAccount extends BaseAccount {
+    accountId: string;
+    credentials: {
+        TELEGRAM_API_ID: string;
+        TELEGRAM_API_HASH: string;
+        TELEGRAM_TG_CHANNEL: string;
+    };
+    lastUsed?: string;
+    totalRequests?: number;
+}
+
+export class TelegramAccountManager extends BaseAccountManager<TelegramAccount> {
+    protected platform = 'telegram';
+    protected accountKey = 'telegram-accounts';
+    protected usageKeyPrefix = 'telegram_accounts';
+
+    constructor(redisUrl?: string) {
+        super(redisUrl);
+    }
+
+    /**
+     * Fetch all Telegram accounts from Redis and decrypt their credentials
+     */
+    protected async fetchAllAccounts(): Promise<TelegramAccount[]> {
+        await this.ensureConnected();
+
+        const raw = await this.redisClient.get(this.accountKey);
+        if (!raw) {
+            throw new Error('No Telegram accounts found in Redis');
+        }
+
+        let encryptedAccounts: Record<string, string>[];
+        try {
+            encryptedAccounts = JSON.parse(raw);
+        } catch (e) {
+            throw new Error('Failed to parse Telegram accounts from Redis');
+        }
+
+        const accounts: TelegramAccount[] = [];
+
+        for (let i = 0; i < encryptedAccounts.length; i++) {
+            const encryptedAccount = encryptedAccounts[i];
+
+            try {
+                // Decrypt credentials
+                const credentials = {
+                    TELEGRAM_API_ID: decrypt(encryptedAccount.TELEGRAM_API_ID),
+                    TELEGRAM_API_HASH: decrypt(encryptedAccount.TELEGRAM_API_HASH),
+                    TELEGRAM_TG_CHANNEL: decrypt(encryptedAccount.TELEGRAM_TG_CHANNEL),
+                };
+
+                // Generate account ID from API ID (for uniqueness)
+                const accountId = `telegram_${credentials.TELEGRAM_API_ID}`;
+
+                // Get usage statistics from Redis
+                const usage = await getApiKeyUsage({ accountId, platform: 'telegram' });
+
+                accounts.push({
+                    accountId,
+                    credentials,
+                    lastUsed: usage.last_request || undefined,
+                    totalRequests: usage.total_requests
+                });
+            } catch (e) {
+                console.warn(`Failed to decrypt Telegram account ${i + 1}:`, e);
+                continue;
+            }
+        }
+
+        if (accounts.length === 0) {
+            throw new Error('No valid Telegram accounts could be decrypted');
+        }
+
+        return accounts;
+    }
+
+
+    /**
+     * Local usage tracking for Telegram accounts
+     */
+    protected async trackApiKeyUsageLocal(accountId: string): Promise<void> {
+        await trackApiKeyUsage({ accountId, platform: 'telegram' });
+    }
+
+    /**
+     * Get usage statistics for all accounts (no credentials)
+     */
+    async getAllAccountsUsage(): Promise<Array<{ accountId: string; lastUsed?: string; totalRequests?: number }>> {
+        const accounts = await this.fetchAllAccounts();
+        return accounts.map(({ accountId, lastUsed, totalRequests }) => ({ accountId, lastUsed, totalRequests }));
+    }
+
+    /**
+     * Get all accounts with credentials (full info)
+     */
+    async getAllAccountsWithCredentials(): Promise<TelegramAccount[]> {
+        return await this.fetchAllAccounts();
+    }
+}
+
+// Export singleton instance
+export const telegramAccountManager = new TelegramAccountManager();