refactor: remove global upload queue

Per-bot queues now handle concurrency internally. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-29 13:28:38 +07:00
parent a986ce1e08
commit adbf9b5efa
3 changed files with 0 additions and 150 deletions
@@ -1,7 +1,6 @@
 import { serve } from 'bun';
 import { config } from './config/index';
 import { fileInfoCache } from './infrastructure/cache/index';
 import { clearQueue, getQueueStats, waitForQueue } from './infrastructure/telegram/upload-queue';
 import { startBot } from './interfaces/bot/handler';
 import { handleS3Request } from './interfaces/http/controllers/s3-controller';
 import { cleanupRateLimitCache } from './interfaces/http/middleware/rate-limit';
@@ -67,19 +66,6 @@ const gracefulShutdown = async (signal: string): Promise<void> => {
  logger.info('Closing HTTP server — no new requests accepted');
  server.stop();
  // Drain pending upload queue with a timeout
  const { pending, size } = getQueueStats();
  if (pending > 0 || size > 0) {
    logger.info('Draining upload queue', { pending, size });
    const drainTimeout = setTimeout(() => {
      logger.warn('Upload queue drain timeout — clearing remaining tasks');
      clearQueue();
    }, 30_000);
    await waitForQueue();
    clearTimeout(drainTimeout);
    logger.info('Upload queue drained');
  }
  logger.info('Stopping Telegram bot');
  bot.stop(signal);
@@ -1,100 +0,0 @@
 import PQueue from 'p-queue';
 import { config } from '../../env';
 import logger from '../../shared/logger/index';
 /**
 * Maximum number of pending (queued + in-flight) upload tasks before
 * new submissions are rejected. Prevents unbounded memory growth when
 * Telegram is slow or unavailable.
 */
 const MAX_QUEUE_PENDING = 1000;
 /**
 * P-queue instance for serialising Telegram upload tasks.
 *
 * Concurrency is governed by {@link config.uploadConcurrency}.
 * Built-in logging emits warnings when the queue grows beyond 5 pending items.
 */
 const uploadQueue = new PQueue({
  concurrency: config.uploadConcurrency,
 });
 /* Monitor queue growth and emit warnings for large backlogs */
 uploadQueue.on('add', () => {
  const stats = getQueueStats();
  if (stats.size > 5) {
    logger.warn('Upload queue building up', {
      pending: stats.pending,
      size: stats.size,
      max: MAX_QUEUE_PENDING,
    });
  }
 });
 uploadQueue.on('next', () => {
  const stats = getQueueStats();
  logger.debug('Processing next upload', { pending: stats.pending, size: stats.size });
 });
 /**
 * Enqueue an upload task to be executed by the queue.
 *
 * Tasks are executed in FIFO order, subject to the concurrency limit.
 *
 * @param task - An async function representing the upload operation.
 * @returns A promise that resolves with the task's result.
 */
 export const enqueueUpload = <T>(task: () => Promise<T>): Promise<T> => {
  const stats = getQueueStats();
  if (stats.pending + stats.size > MAX_QUEUE_PENDING) {
    return Promise.reject(
      new Error(
        `Upload queue full (${stats.pending + stats.size} pending, max ${MAX_QUEUE_PENDING})`,
      ),
    );
  }
  return uploadQueue.add(task);
 };
 /**
 * Get current queue statistics.
 *
 * @returns An object with `pending` (actively executing) and `size` (waiting) counts.
 */
 export const getQueueStats = (): { pending: number; size: number } => ({
  pending: uploadQueue.pending,
  size: uploadQueue.size,
 });
 /**
 * Get the number of items waiting in the queue (not yet started).
 *
 * @returns The number of queued items.
 */
 export const getQueueSize = (): number => uploadQueue.size;
 /**
 * Get the number of items currently being processed.
 *
 * @returns The number of pending (in-flight) items.
 */
 export const getPendingCount = (): number => uploadQueue.pending;
 /**
 * Clear all pending items and wait for in-flight ones to finish.
 *
 * @returns A promise that resolves when the queue is idle after clearing.
 */
 export const clearQueue = async (): Promise<void> => {
  uploadQueue.clear();
  await uploadQueue.onIdle();
 };
 /**
 * Wait for the queue to become idle (all tasks finished).
 *
 * @returns A promise that resolves when no tasks are pending or in-flight.
 */
 export const waitForQueue = async (): Promise<void> => {
  await uploadQueue.onIdle();
 };
@@ -1,36 +0,0 @@
 import { describe, expect, it } from 'bun:test';
 import { enqueueUpload } from '../src/infrastructure/telegram/upload-queue';
 describe('Telegram Queue', () => {
  it('should process tasks in parallel without limit', async () => {
    let activeTasks = 0;
    let maxActiveTasks = 0;
    const createTask = (id: number, delayMs: number) => {
      return async () => {
        activeTasks++;
        if (activeTasks > maxActiveTasks) {
          maxActiveTasks = activeTasks;
        }
        await new Promise((resolve) => setTimeout(resolve, delayMs));
        activeTasks--;
        return id;
      };
    };
    const promises = [
      enqueueUpload(createTask(1, 50)),
      enqueueUpload(createTask(2, 50)),
      enqueueUpload(createTask(3, 50)),
      enqueueUpload(createTask(4, 50)),
    ];
    const results = await Promise.all(promises);
    expect(results).toEqual([1, 2, 3, 4]);
    // Concurrency limit is removed, so active tasks should be able to reach 4 (fully parallel)
    expect(maxActiveTasks).toBe(4);
  });
 });