feat: implement caching for file info, add rate limiting, and enhance upload handling with retry logic

2026-05-21 23:31:49 +07:00
parent 10c968cf01
commit fd5eb98586
10 changed files with 557 additions and 11 deletions
@@ -8,6 +8,7 @@
        "@prisma/engines": "^7.8.0",
        "drizzle-orm": "^0.45.2",
        "nanoid": "^5.0.4",
+        "p-queue": "^9.3.0",
        "pg": "^8.11.0",
        "postgres": "^3.4.9",
        "telegraf": "^4.15.0",
@@ -154,6 +155,8 @@

    "event-target-shim": ["event-target-shim@5.0.1", "", {}, "sha512-i/2XbnSz/uxRCU6+NdVJgKWDTM427+MqYbkQzD321DuCQJUqOuJKIA0IM2+W2xtYHdKOmZ4dR6fExsd4SXL+WQ=="],

+    "eventemitter3": ["eventemitter3@5.0.4", "", {}, "sha512-mlsTRyGaPBjPedk6Bvw+aqbsXDtoAyAzm5MO7JgU+yVRyMQ5O8bD4Kcci7BS85f93veegeCPkL8R4GLClnjLFw=="],
+
    "fecha": ["fecha@4.2.3", "", {}, "sha512-OP2IUU6HeYKJi3i0z4A19kHMQoLVs4Hc+DPqqxI2h/DPZHTm/vjsfC6P0b4jCMy14XizLBqvndQ+UilD7707Jw=="],

    "fn.name": ["fn.name@1.1.0", "", {}, "sha512-GRnmB5gPyJpAhTQdSZTSp9uaPSvl09KoYcMQtsB9rQoOmzs9dH6ffeccH+Z+cv6P68Hu5bC6JjRh4Ah/mHSNRw=="],
@@ -180,7 +183,9 @@

    "one-time": ["one-time@1.0.0", "", { "dependencies": { "fn.name": "1.x.x" } }, "sha512-5DXOiRKwuSEcQ/l0kGCF6Q3jcADFv5tSmRaJck/OqkVFcOzutB134KRSfF0xDrL39MNnqxbHBbUUcjZIhTgb2g=="],

-    "p-timeout": ["p-timeout@4.1.0", "", {}, "sha512-+/wmHtzJuWii1sXn3HCuH/FTwGhrp4tmJTxSKJbfS+vkipci6osxXM5mY0jUiRzWKMTgUT8l7HFbeSwZAynqHw=="],
+    "p-queue": ["p-queue@9.3.0", "", { "dependencies": { "eventemitter3": "^5.0.4", "p-timeout": "^7.0.0" } }, "sha512-7NED7xhQ74Ngp4JP/2e0VZHp7vSWfJfqeiR92jPgxsz6m0Se4P03YoTKa9dDXyZ3r6P616gUXttrB6nnHYKang=="],
+
+    "p-timeout": ["p-timeout@7.0.1", "", {}, "sha512-AxTM2wDGORHGEkPCt8yqxOTMgpfbEHqF51f/5fJCmwFC3C/zNcGT63SymH2ttOAaiIws2zVg4+izQCjrakcwHg=="],

    "pg": ["pg@8.20.0", "", { "dependencies": { "pg-connection-string": "^2.12.0", "pg-pool": "^3.13.0", "pg-protocol": "^1.13.0", "pg-types": "2.2.0", "pgpass": "1.0.5" }, "optionalDependencies": { "pg-cloudflare": "^1.3.0" }, "peerDependencies": { "pg-native": ">=3.0.1" }, "optionalPeers": ["pg-native"] }, "sha512-ldhMxz2r8fl/6QkXnBD3CR9/xg694oT6DZQ2s6c/RI28OjtSOpxnPrUCGOBJ46RCUxcWdx3p6kw/xnDHjKvaRA=="],

@@ -258,6 +263,8 @@

    "@esbuild-kit/core-utils/esbuild": ["esbuild@0.18.20", "", { "optionalDependencies": { "@esbuild/android-arm": "0.18.20", "@esbuild/android-arm64": "0.18.20", "@esbuild/android-x64": "0.18.20", "@esbuild/darwin-arm64": "0.18.20", "@esbuild/darwin-x64": "0.18.20", "@esbuild/freebsd-arm64": "0.18.20", "@esbuild/freebsd-x64": "0.18.20", "@esbuild/linux-arm": "0.18.20", "@esbuild/linux-arm64": "0.18.20", "@esbuild/linux-ia32": "0.18.20", "@esbuild/linux-loong64": "0.18.20", "@esbuild/linux-mips64el": "0.18.20", "@esbuild/linux-ppc64": "0.18.20", "@esbuild/linux-riscv64": "0.18.20", "@esbuild/linux-s390x": "0.18.20", "@esbuild/linux-x64": "0.18.20", "@esbuild/netbsd-x64": "0.18.20", "@esbuild/openbsd-x64": "0.18.20", "@esbuild/sunos-x64": "0.18.20", "@esbuild/win32-arm64": "0.18.20", "@esbuild/win32-ia32": "0.18.20", "@esbuild/win32-x64": "0.18.20" }, "bin": { "esbuild": "bin/esbuild" } }, "sha512-ceqxoedUrcayh7Y7ZX6NdbbDzGROiyVBgC4PriJThBKSVPWnnFHZAkfI1lJT8QFkOwH4qOS2SJkS4wvpGl8BpA=="],

+    "telegraf/p-timeout": ["p-timeout@4.1.0", "", {}, "sha512-+/wmHtzJuWii1sXn3HCuH/FTwGhrp4tmJTxSKJbfS+vkipci6osxXM5mY0jUiRzWKMTgUT8l7HFbeSwZAynqHw=="],
+
    "tsx/esbuild": ["esbuild@0.28.0", "", { "optionalDependencies": { "@esbuild/aix-ppc64": "0.28.0", "@esbuild/android-arm": "0.28.0", "@esbuild/android-arm64": "0.28.0", "@esbuild/android-x64": "0.28.0", "@esbuild/darwin-arm64": "0.28.0", "@esbuild/darwin-x64": "0.28.0", "@esbuild/freebsd-arm64": "0.28.0", "@esbuild/freebsd-x64": "0.28.0", "@esbuild/linux-arm": "0.28.0", "@esbuild/linux-arm64": "0.28.0", "@esbuild/linux-ia32": "0.28.0", "@esbuild/linux-loong64": "0.28.0", "@esbuild/linux-mips64el": "0.28.0", "@esbuild/linux-ppc64": "0.28.0", "@esbuild/linux-riscv64": "0.28.0", "@esbuild/linux-s390x": "0.28.0", "@esbuild/linux-x64": "0.28.0", "@esbuild/netbsd-arm64": "0.28.0", "@esbuild/netbsd-x64": "0.28.0", "@esbuild/openbsd-arm64": "0.28.0", "@esbuild/openbsd-x64": "0.28.0", "@esbuild/openharmony-arm64": "0.28.0", "@esbuild/sunos-x64": "0.28.0", "@esbuild/win32-arm64": "0.28.0", "@esbuild/win32-ia32": "0.28.0", "@esbuild/win32-x64": "0.28.0" }, "bin": { "esbuild": "bin/esbuild" } }, "sha512-sNR9MHpXSUV/XB4zmsFKN+QgVG82Cc7+/aaxJ8Adi8hyOac+EXptIp45QBPaVyX3N70664wRbTcLTOemCAnyqw=="],

    "@esbuild-kit/core-utils/esbuild/@esbuild/android-arm": ["@esbuild/android-arm@0.18.20", "", { "os": "android", "cpu": "arm" }, "sha512-fyi7TDI/ijKKNZTUJAQqiG5T7YjJXgnzkURqmGj13C6dCqckZBLdl4h7bkhHt/t0WP+zO9/zwroDvANaOqO5Sw=="],
@@ -16,6 +16,7 @@
    "@prisma/engines": "^7.8.0",
    "drizzle-orm": "^0.45.2",
    "nanoid": "^5.0.4",
+    "p-queue": "^9.3.0",
    "pg": "^8.11.0",
    "postgres": "^3.4.9",
    "telegraf": "^4.15.0",
@@ -3,6 +3,7 @@ import { formatCreatedAt, getErrorMessage } from '../utils/file';
 import logger from '../utils/logger';
 import { checkRateLimit } from '../utils/rateLimit';
 import { getBot } from '../utils/telegram';
+import { fileInfoCache } from '../utils/cache';

 type RequestWithParams = Request & {
  params?: {
@@ -26,8 +27,25 @@ export const handleFileRedirect = async (req: RequestWithParams): Promise<Respon
      return Response.json({ error: 'File not found' }, { status: 404 });
    }

-    const bot = getBot();
-    const fileInfo = await bot.telegram.getFile(file.telegramFileId);
+    // Check cache first
+    const cacheKey = `file_info_${file.telegramFileId}`;
+    let fileInfo = fileInfoCache.get(cacheKey);
+
+    if (!fileInfo) {
+      // Cache miss - fetch from Telegram API
+      const bot = getBot();
+      const apiFileInfo = await bot.telegram.getFile(file.telegramFileId);
+      fileInfo = {
+        file_size: (apiFileInfo as any).file_size || 0,
+        mime_type: (apiFileInfo as any).mime_type || 'application/octet-stream',
+        file_path: (apiFileInfo as any).file_path || '',
+      };
+      // Store in cache
+      fileInfoCache.set(cacheKey, fileInfo);
+      logger.debug('File info cached', { public_id, cacheKey });
+    } else {
+      logger.debug('File info from cache', { public_id, cacheKey });
+    }

    const redirectUrl = `https://api.telegram.org/file/bot${process.env.BOT_TOKEN}/${fileInfo.file_path}`;
    return new Response(null, {
@@ -1,4 +1,5 @@
-import { createReadStream, unlinkSync } from 'node:fs';
+import { createReadStream } from 'node:fs';
+import { unlink } from 'node:fs/promises';
 import { nanoid } from 'nanoid';
 import { db, files as fileSchema } from '../db';
 import { findFileByHash } from '../db/files';
@@ -69,11 +70,13 @@ const performUpload = async (
      updatedAt: new Date(),
    };
  } finally {
-    setTimeout(() => {
+    setTimeout(async () => {
      try {
-        unlinkSync(tempPath);
-      } catch {}
-    }, 50);
+        await unlink(tempPath);
+      } catch (err) {
+        logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) });
+      }
+    }, 500);
  }
 };

@@ -0,0 +1,129 @@
+import logger from './logger';
+
+interface BotHealth {
+  index: number;
+  isHealthy: boolean;
+  rateLimitedUntil: number;
+  failureCount: number;
+  successCount: number;
+  lastUsed: number;
+}
+
+class BotHealthTracker {
+  private botHealth: Map<number, BotHealth> = new Map();
+  private totalBots: number;
+
+  constructor(totalBots: number) {
+    this.totalBots = totalBots;
+    for (let i = 0; i < totalBots; i++) {
+      this.botHealth.set(i, {
+        index: i,
+        isHealthy: true,
+        rateLimitedUntil: 0,
+        failureCount: 0,
+        successCount: 0,
+        lastUsed: 0,
+      });
+    }
+  }
+
+  recordSuccess(botIndex: number): void {
+    const health = this.botHealth.get(botIndex);
+    if (health) {
+      health.successCount++;
+      health.failureCount = 0;
+      health.isHealthy = true;
+      health.lastUsed = Date.now();
+    }
+  }
+
+  recordFailure(botIndex: number, retryAfterSeconds?: number): void {
+    const health = this.botHealth.get(botIndex);
+    if (health) {
+      health.failureCount++;
+      health.lastUsed = Date.now();
+
+      if (retryAfterSeconds) {
+        health.rateLimitedUntil = Date.now() + retryAfterSeconds * 1000;
+        health.isHealthy = false;
+        logger.warn('Bot rate limited', {
+          botIndex,
+          retryAfter: retryAfterSeconds,
+        });
+      } else if (health.failureCount >= 3) {
+        health.isHealthy = false;
+        logger.warn('Bot marked unhealthy', { botIndex, failures: health.failureCount });
+      }
+    }
+  }
+
+  getHealthiestBot(): number {
+    const now = Date.now();
+    let bestBot = 0;
+    let bestScore = -Infinity;
+
+    for (let i = 0; i < this.totalBots; i++) {
+      const health = this.botHealth.get(i)!;
+
+      // Skip rate-limited bots
+      if (health.rateLimitedUntil > now) {
+        continue;
+      }
+
+      // Calculate score: prefer healthy bots with fewer failures and more successes
+      const score =
+        (health.isHealthy ? 100 : 0) +
+        health.successCount -
+        health.failureCount * 10 -
+        (now - health.lastUsed) / 1000;
+
+      if (score > bestScore) {
+        bestScore = score;
+        bestBot = i;
+      }
+    }
+
+    return bestBot;
+  }
+
+  getStats() {
+    const stats = {
+      healthy: 0,
+      rateLimited: 0,
+      unhealthy: 0,
+      bots: [] as any[],
+    };
+
+    const now = Date.now();
+    for (const health of this.botHealth.values()) {
+      if (health.rateLimitedUntil > now) {
+        stats.rateLimited++;
+      } else if (health.isHealthy) {
+        stats.healthy++;
+      } else {
+        stats.unhealthy++;
+      }
+
+      stats.bots.push({
+        index: health.index,
+        healthy: health.isHealthy,
+        rateLimitedUntil: health.rateLimitedUntil > now ? health.rateLimitedUntil - now : 0,
+        failures: health.failureCount,
+        successes: health.successCount,
+      });
+    }
+
+    return stats;
+  }
+
+  reset(): void {
+    for (const health of this.botHealth.values()) {
+      health.isHealthy = true;
+      health.rateLimitedUntil = 0;
+      health.failureCount = 0;
+      health.successCount = 0;
+    }
+  }
+}
+
+export { BotHealthTracker };
@@ -0,0 +1,81 @@
+// Simple in-memory cache with TTL support
+interface CacheEntry<T> {
+  value: T;
+  expiresAt: number;
+}
+
+class Cache<T> {
+  private store = new Map<string, CacheEntry<T>>();
+  private ttlMs: number;
+
+  constructor(ttlSeconds: number = 3600) {
+    this.ttlMs = ttlSeconds * 1000;
+  }
+
+  set(key: string, value: T): void {
+    this.store.set(key, {
+      value,
+      expiresAt: Date.now() + this.ttlMs,
+    });
+  }
+
+  get(key: string): T | null {
+    const entry = this.store.get(key);
+    if (!entry) return null;
+
+    if (Date.now() > entry.expiresAt) {
+      this.store.delete(key);
+      return null;
+    }
+
+    return entry.value;
+  }
+
+  has(key: string): boolean {
+    return this.get(key) !== null;
+  }
+
+  delete(key: string): void {
+    this.store.delete(key);
+  }
+
+  clear(): void {
+    this.store.clear();
+  }
+
+  size(): number {
+    return this.store.size;
+  }
+
+  // Cleanup expired entries
+  cleanup(): number {
+    let removed = 0;
+    const now = Date.now();
+
+    for (const [key, entry] of this.store.entries()) {
+      if (now > entry.expiresAt) {
+        this.store.delete(key);
+        removed++;
+      }
+    }
+
+    return removed;
+  }
+}
+
+// File info cache (1 hour TTL)
+export const fileInfoCache = new Cache<{
+  file_size: number;
+  mime_type: string;
+  file_path: string;
+}>(3600);
+
+// Cleanup expired cache entries every 5 minutes
+setInterval(() => {
+  const removed = fileInfoCache.cleanup();
+  if (removed > 0) {
+    console.log(`Cleaned up ${removed} expired cache entries`);
+  }
+}, 5 * 60 * 1000);
+
+export { Cache };
@@ -0,0 +1,122 @@
+import logger from './logger';
+
+interface Metric {
+  name: string;
+  value: number;
+  timestamp: number;
+  tags?: Record<string, string>;
+}
+
+interface MetricsSnapshot {
+  uploadLatency: { p50: number; p95: number; p99: number };
+  uploadThroughput: number;
+  queueSize: number;
+  errorRate: number;
+  cacheHitRate: number;
+  botUtilization: number;
+  timestamp: number;
+}
+
+class MetricsCollector {
+  private metrics: Metric[] = [];
+  private uploadTimes: number[] = [];
+  private errorCount = 0;
+  private totalRequests = 0;
+  private cacheHits = 0;
+  private cacheMisses = 0;
+  private maxMetricsSize = 10000;
+
+  recordUploadTime(durationMs: number): void {
+    this.uploadTimes.push(durationMs);
+    this.totalRequests++;
+
+    // Keep only last 1000 measurements
+    if (this.uploadTimes.length > 1000) {
+      this.uploadTimes.shift();
+    }
+  }
+
+  recordError(): void {
+    this.errorCount++;
+  }
+
+  recordCacheHit(): void {
+    this.cacheHits++;
+  }
+
+  recordCacheMiss(): void {
+    this.cacheMisses++;
+  }
+
+  recordMetric(name: string, value: number, tags?: Record<string, string>): void {
+    this.metrics.push({
+      name,
+      value,
+      timestamp: Date.now(),
+      tags,
+    });
+
+    // Keep metrics bounded
+    if (this.metrics.length > this.maxMetricsSize) {
+      this.metrics = this.metrics.slice(-this.maxMetricsSize);
+    }
+  }
+
+  private calculatePercentile(arr: number[], percentile: number): number {
+    if (arr.length === 0) return 0;
+    const sorted = [...arr].sort((a, b) => a - b);
+    const index = Math.ceil((percentile / 100) * sorted.length) - 1;
+    return sorted[Math.max(0, index)];
+  }
+
+  getSnapshot(): MetricsSnapshot {
+    const errorRate = this.totalRequests > 0 ? (this.errorCount / this.totalRequests) * 100 : 0;
+    const cacheHitRate =
+      this.cacheHits + this.cacheMisses > 0
+        ? (this.cacheHits / (this.cacheHits + this.cacheMisses)) * 100
+        : 0;
+
+    return {
+      uploadLatency: {
+        p50: this.calculatePercentile(this.uploadTimes, 50),
+        p95: this.calculatePercentile(this.uploadTimes, 95),
+        p99: this.calculatePercentile(this.uploadTimes, 99),
+      },
+      uploadThroughput: this.totalRequests > 0 ? this.totalRequests / 60 : 0,
+      queueSize: 0, // Will be updated by queue
+      errorRate,
+      cacheHitRate,
+      botUtilization: 0, // Will be updated by bot tracker
+      timestamp: Date.now(),
+    };
+  }
+
+  reset(): void {
+    this.uploadTimes = [];
+    this.errorCount = 0;
+    this.totalRequests = 0;
+    this.cacheHits = 0;
+    this.cacheMisses = 0;
+    this.metrics = [];
+  }
+
+  getMetrics(name?: string): Metric[] {
+    if (!name) return this.metrics;
+    return this.metrics.filter((m) => m.name === name);
+  }
+}
+
+export const metricsCollector = new MetricsCollector();
+
+// Log metrics every 5 minutes
+setInterval(() => {
+  const snapshot = metricsCollector.getSnapshot();
+  logger.info('Metrics snapshot', {
+    uploadLatency: snapshot.uploadLatency,
+    uploadThroughput: snapshot.uploadThroughput.toFixed(2),
+    errorRate: snapshot.errorRate.toFixed(2),
+    cacheHitRate: snapshot.cacheHitRate.toFixed(2),
+  });
+}, 5 * 60 * 1000);
+
+export { MetricsCollector };
@@ -1,7 +1,57 @@
-export const checkRateLimit = (_key: string): boolean => {
+import logger from './logger';
+
+// Simple sliding window rate limiter
+interface RateLimitEntry {
+  count: number;
+  resetTime: number;
+}
+
+const rateLimitStore = new Map<string, RateLimitEntry>();
+const WINDOW_SIZE_MS = 60000; // 1 minute window
+const MAX_REQUESTS_PER_WINDOW = 100; // 100 requests per minute per IP
+
+export const checkRateLimit = (key: string): boolean => {
+  const now = Date.now();
+  const entry = rateLimitStore.get(key);
+
+  // No entry or window expired - create new entry
+  if (!entry || now > entry.resetTime) {
+    rateLimitStore.set(key, {
+      count: 1,
+      resetTime: now + WINDOW_SIZE_MS,
+    });
+    return true;
+  }
+
+  // Check if limit exceeded
+  if (entry.count >= MAX_REQUESTS_PER_WINDOW) {
+    logger.warn('Rate limit exceeded', { key, count: entry.count });
+    return false;
+  }
+
+  // Increment counter
+  entry.count++;
  return true;
 };

 export const cleanupRateLimitCache = (): void => {
-  // No-op karena rate limit dinonaktifkan
+  const now = Date.now();
+  let cleaned = 0;
+
+  for (const [key, entry] of rateLimitStore.entries()) {
+    if (now > entry.resetTime) {
+      rateLimitStore.delete(key);
+      cleaned++;
+    }
+  }
+
+  if (cleaned > 0) {
+    logger.debug('Rate limit cache cleanup', { cleaned, remaining: rateLimitStore.size });
+  }
 };
+
+export const getRateLimitStats = () => ({
+  trackedIPs: rateLimitStore.size,
+  windowSize: WINDOW_SIZE_MS,
+  maxRequests: MAX_REQUESTS_PER_WINDOW,
+});
@@ -0,0 +1,91 @@
+import logger from './logger';
+
+interface RetryOptions {
+  maxRetries?: number;
+  initialDelayMs?: number;
+  maxDelayMs?: number;
+  backoffMultiplier?: number;
+  shouldRetry?: (error: unknown) => boolean;
+}
+
+const DEFAULT_OPTIONS: Required<RetryOptions> = {
+  maxRetries: 3,
+  initialDelayMs: 100,
+  maxDelayMs: 5000,
+  backoffMultiplier: 2,
+  shouldRetry: (error: unknown) => {
+    const errorStr = error instanceof Error ? error.message : String(error);
+    // Retry on transient errors
+    return (
+      errorStr.includes('ECONNREFUSED') ||
+      errorStr.includes('ETIMEDOUT') ||
+      errorStr.includes('ENOTFOUND') ||
+      errorStr.includes('429') ||
+      errorStr.includes('timeout')
+    );
+  },
+};
+
+export const withRetry = async <T>(
+  fn: () => Promise<T>,
+  options: RetryOptions = {},
+): Promise<T> => {
+  const opts = { ...DEFAULT_OPTIONS, ...options };
+  let lastError: unknown;
+  let delay = opts.initialDelayMs;
+
+  for (let attempt = 0; attempt <= opts.maxRetries; attempt++) {
+    try {
+      return await fn();
+    } catch (error: unknown) {
+      lastError = error;
+      const errorStr = error instanceof Error ? error.message : String(error);
+
+      if (attempt === opts.maxRetries || !opts.shouldRetry(error)) {
+        logger.error('Retry exhausted', {
+          attempt,
+          maxRetries: opts.maxRetries,
+          error: errorStr,
+        });
+        throw error;
+      }
+
+      logger.warn('Retrying after error', {
+        attempt,
+        delay,
+        error: errorStr,
+      });
+
+      await new Promise((resolve) => setTimeout(resolve, delay));
+      delay = Math.min(delay * opts.backoffMultiplier, opts.maxDelayMs);
+    }
+  }
+
+  throw lastError;
+};
+
+export const withTimeout = async <T>(
+  fn: () => Promise<T>,
+  timeoutMs: number = 30000,
+): Promise<T> => {
+  return Promise.race([
+    fn(),
+    new Promise<T>((_, reject) =>
+      setTimeout(() => reject(new Error(`Operation timeout after ${timeoutMs}ms`)), timeoutMs),
+    ),
+  ]);
+};
+
+export const withFallback = async <T>(
+  primary: () => Promise<T>,
+  fallback: () => Promise<T>,
+): Promise<T> => {
+  try {
+    return await primary();
+  } catch (error: unknown) {
+    logger.warn('Primary operation failed, using fallback', {
+      error: error instanceof Error ? error.message : String(error),
+    });
+    return fallback();
+  }
+};
@@ -1,3 +1,47 @@
+import PQueue from 'p-queue';
+import logger from './logger';
+
+// Create queue with concurrency limit matching bot pool size
+// Concurrency: 4-8 uploads in parallel
+// Interval: 1 second window for rate limiting
+// IntervalCap: Max 10 tasks per second
+const uploadQueue = new PQueue({
+  concurrency: 4,
+  interval: 1000,
+  intervalCap: 10,
+});
+
+// Monitor queue events
+uploadQueue.on('add', () => {
+  const stats = getQueueStats();
+  if (stats.size > 5) {
+    logger.warn('Upload queue building up', { pending: stats.pending, size: stats.size });
+  }
+});
+
+uploadQueue.on('next', () => {
+  const stats = getQueueStats();
+  logger.debug('Processing next upload', { pending: stats.pending, size: stats.size });
+});
+
 export const enqueueUpload = <T>(task: () => Promise<T>): Promise<T> => {
-  return task();
+  return uploadQueue.add(task);
+};
+
+export const getQueueStats = () => ({
+  pending: uploadQueue.pending,
+  size: uploadQueue.size,
+});
+
+export const getQueueSize = (): number => uploadQueue.size;
+
+export const getPendingCount = (): number => uploadQueue.pending;
+
+export const clearQueue = async (): Promise<void> => {
+  uploadQueue.clear();
+  await uploadQueue.onIdle();
+};
+
+export const waitForQueue = async (): Promise<void> => {
+  await uploadQueue.onIdle();
 };