updates-worker.js

require('dotenv').config({ path: './.env' })

const fs = require('fs')
const { Telegraf } = require('telegraf')
const { createRedisClient } = require('./utils/redis')
const { getMaxWorkers, getWorkerConcurrentLimit, getWorkerHandlerTimeout, getWorkerIndex } = require('./utils/worker-config')
const { normalizeQuoteConfig } = require('./utils/quote-config')
const { db } = require('./database')
const { stats } = require('./middlewares')

// Cache config at startup
let cachedConfig = normalizeQuoteConfig()
try {
  if (fs.existsSync('./config.json')) {
    cachedConfig = normalizeQuoteConfig(JSON.parse(fs.readFileSync('./config.json', 'utf8')))
  }
} catch (error) {
  console.error('Error loading config:', error.message)
}

const logWithTimestamp = (message) => {
  const workerId = process.env.pm_id || process.pid
  console.log(`[${new Date().toISOString()}] [WORKER-${workerId}] ${message}`)
}

const errorWithTimestamp = (message, ...args) => {
  const workerId = process.env.pm_id || process.pid
  console.error(`[${new Date().toISOString()}] [WORKER-${workerId}] ${message}`, ...args)
}

function formatDuration (milliseconds) {
  const seconds = Math.max(0, Math.floor(milliseconds / 1000))
  const minutes = Math.floor(seconds / 60)
  const hours = Math.floor(minutes / 60)

  if (hours > 0) return `${hours}h ${minutes % 60}m`
  if (minutes > 0) return `${minutes}m ${seconds % 60}s`
  return `${seconds}s`
}

class TelegramProcessor {
  constructor () {
    this.maxWorkers = getMaxWorkers()
    this.bot = new Telegraf(process.env.BOT_TOKEN, {
      handlerTimeout: getWorkerHandlerTimeout()
    })

    // Main Redis connection for queue and stats
    this.redis = createRedisClient({
      retryDelayOnFailover: 100,
      maxRetriesPerRequest: 3
    })

    // Separate connection for TDLib pub/sub
    this.tdlibRedis = createRedisClient({
      retryDelayOnFailover: 100,
      maxRetriesPerRequest: 3
    })

    this.isProcessing = false
    this.processedCount = 0
    this.errorCount = 0
    this.lastProcessedCount = 0
    this.lastErrorCount = 0
    this.lastStatsAt = Date.now()
    this.lastProcessedAt = null
    this.startedAt = Date.now()
    this.concurrentLimit = getWorkerConcurrentLimit() // Process updates concurrently per worker
    this.activePromises = new Set() // Track active processing promises
    this.workerId = process.env.WORKER_INDEX || process.env.pm_id || process.pid
    this.workerIndex = getWorkerIndex(process.env.WORKER_INDEX, this.workerId, this.maxWorkers)
    this.queueName = `telegram:updates:worker:${this.workerIndex}`
    this.pendingTDLibRequests = new Map() // Track pending TDLib requests

    this.setupRedisEvents()
    this.setupBot()
  }

  setupRedisEvents () {
    this.redis.on('connect', () => {
      logWithTimestamp('Connected to Redis')
    })

    this.redis.on('error', (error) => {
      errorWithTimestamp('Redis error:', error.message)
    })

    // Setup TDLib response handler once
    this.tdlibRedis.on('message', (channel, message) => {
      if (channel === 'tdlib:responses') {
        try {
          const response = JSON.parse(message)
          const pending = this.pendingTDLibRequests.get(response.id)
          if (pending) {
            this.pendingTDLibRequests.delete(response.id)
            clearTimeout(pending.timeout)
            if (response.error) {
              pending.reject(new Error(response.error))
            } else {
              pending.resolve(response.result)
            }
          }
        } catch (_error) {
          // Ignore parse errors
        }
      }
    })
  }

  createTDLibProxy () {
    // Create proxy that sends TDLib requests through Redis
    return new Proxy({}, {
      get: (target, prop) => {
        return (...args) => {
          return new Promise((resolve, reject) => {
            const requestId = `${this.workerId}-${Date.now()}-${Math.random()}`

            // Send request through Redis
            const request = {
              id: requestId,
              method: prop,
              args: args
            }

            // Set timeout
            const timeout = setTimeout(() => {
              this.pendingTDLibRequests.delete(requestId)
              reject(new Error(`TDLib ${prop} timeout`))
            }, 5000)

            // Store pending request
            this.pendingTDLibRequests.set(requestId, { resolve, reject, timeout })

            // Publish request
            this.redis.publish('tdlib:requests', JSON.stringify(request))
          })
        }
      }
    })
  }

  setupBot () {
    // Idempotent boot-time seed of the quote counter. $setOnInsert is a no-op
    // once the document exists, so concurrent workers racing here is safe.
    db.ready.then(() =>
      db.Counter.findOneAndUpdate(
        { _id: 'quote' },
        { $setOnInsert: { seq: 0 } },
        { upsert: true }
      )
    ).catch(err => {
      errorWithTimestamp('[boot] failed to seed Counter{_id:quote}', err.message || err)
    })

    // Set up database and config context
    this.bot.use(async (ctx, next) => {
      ctx.db = db
      ctx.config = cachedConfig
      return next()
    })

    // TDLib proxy through Redis IPC (like old architecture)
    this.bot.use(async (ctx, next) => {
      ctx.tdlib = this.createTDLibProxy()
      return next()
    })

    // Add stats middleware
    this.bot.use(stats.middleware())

    // Add main handler
    const handler = require('./handler')
    this.bot.use(handler)

    // Error handling
    this.bot.catch((err) => {
      this.errorCount++
      errorWithTimestamp('Bot error:', err.message)
    })
  }

  async processUpdate (updateData) {
    try {
      const update = JSON.parse(updateData)

      // Add processing timestamp
      update._processor_started = Date.now()

      // Process through bot
      await this.bot.handleUpdate(update)

      this.processedCount++
      this.lastProcessedAt = Date.now()

      // Update global processed counter
      await this.redis
        .multi()
        .incr('telegram:processed_count')
        .set('telegram:last_processed_at', this.lastProcessedAt)
        .exec()

      // Don't log each update - only batch stats
    } catch (error) {
      this.errorCount++
      errorWithTimestamp('Error processing update:', error.message)

      // Track error globally
      await this.redis.incr('telegram:error_count')
    }
  }

  async startProcessing () {
    if (this.isProcessing) {
      return
    }

    this.isProcessing = true
    logWithTimestamp(`Starting concurrent processing (limit: ${this.concurrentLimit})...`)

    while (this.isProcessing) {
      try {
        // Only get new update if we have capacity
        if (this.activePromises.size < this.concurrentLimit) {
          const result = await this.redis.brpop(this.queueName, 1)

          if (result && result[1]) {
            // Start processing concurrently (don't await)
            const processPromise = this.processUpdate(result[1])
              .then(() => {
                this.activePromises.delete(processPromise)
              })
              .catch((error) => {
                this.activePromises.delete(processPromise)
                errorWithTimestamp('Concurrent processing error:', error.message)
              })

            this.activePromises.add(processPromise)
          }
        } else {
          // Wait for at least one promise to complete
          await Promise.race(this.activePromises)
        }
      } catch (error) {
        errorWithTimestamp('Processing loop error:', error.message)
        await new Promise(resolve => setTimeout(resolve, 100))
      }
    }

    // Wait for all active promises to complete before stopping
    if (this.activePromises.size > 0) {
      logWithTimestamp(`Waiting for ${this.activePromises.size} active processes to complete...`)
      await Promise.allSettled(this.activePromises)
    }
  }

  async start () {
    try {
      // Wait for database to be ready before processing
      logWithTimestamp('Waiting for MongoDB connection...')
      await db.ready
      logWithTimestamp('MongoDB connected')

      await this.redis.connect()
      await this.tdlibRedis.connect()

      // Subscribe to TDLib responses once after connection
      await this.tdlibRedis.subscribe('tdlib:responses')

      logWithTimestamp(`Starting Telegram processor (Worker ${this.workerId}[${this.workerIndex}] -> ${this.queueName})...`)

      // Start processing loop
      this.startProcessing()

      // Worker stats every 10 seconds
      setInterval(async () => {
        try {
          const now = Date.now()
          const queueSize = await this.redis.llen(this.queueName)
          const totalProcessed = await this.redis.get('telegram:processed_count') || 0
          const totalErrors = await this.redis.get('telegram:error_count') || 0
          const activeCount = this.activePromises.size
          const secondsSinceLastStats = Math.max(1, Math.round((now - this.lastStatsAt) / 1000))
          const processedDelta = Math.max(0, this.processedCount - this.lastProcessedCount)
          const errorDelta = Math.max(0, this.errorCount - this.lastErrorCount)
          const localRate = (processedDelta / secondsSinceLastStats).toFixed(2)
          const lastProcessed = this.lastProcessedAt ? `${formatDuration(now - this.lastProcessedAt)} ago` : 'none yet'

          logWithTimestamp(`Live: ${this.isProcessing ? 'active' : 'stopped'} [${this.workerIndex}] | Queue: ${queueSize} | Local: ${this.processedCount} (+${processedDelta}/${secondsSinceLastStats}s, ${localRate}/s) | Global: ${totalProcessed} | Errors: ${totalErrors} (+${errorDelta}) | Active jobs: ${activeCount}/${this.concurrentLimit} | Last job: ${lastProcessed} | Uptime: ${formatDuration(now - this.startedAt)}`)
          this.lastProcessedCount = this.processedCount
          this.lastErrorCount = this.errorCount
          this.lastStatsAt = now
        } catch (error) {
          errorWithTimestamp('Stats error:', error.message)
        }
      }, 10000) // Every 10 seconds

      logWithTimestamp('Processor started successfully')
    } catch (error) {
      errorWithTimestamp('Failed to start processor:', error.message)
      process.exit(1)
    }
  }

  async stop () {
    logWithTimestamp('Stopping processor...')
    this.isProcessing = false
    await this.redis.quit()
    await this.tdlibRedis.quit()
    logWithTimestamp('Processor stopped')
  }
}

// Create and start processor
const processor = new TelegramProcessor()

// Graceful shutdown
process.once('SIGINT', () => processor.stop())
process.once('SIGTERM', () => processor.stop())

// Start the processor
processor.start()

// Handle uncaught exceptions
process.on('uncaughtException', (error) => {
  errorWithTimestamp('Uncaught Exception:', error.message)
  processor.stop()
  process.exit(1)
})

process.on('unhandledRejection', (reason, promise) => {
  errorWithTimestamp('Unhandled Rejection at:', promise, 'reason:', reason)
})