From fd5eb98586471e609a95402cf692bfd9f60c9235 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Thu, 21 May 2026 23:31:49 +0700 Subject: [PATCH] feat: implement caching for file info, add rate limiting, and enhance upload handling with retry logic --- bun.lock | 9 ++- package.json | 1 + src/routes/files.ts | 22 ++++++- src/routes/upload.ts | 13 ++-- src/utils/botHealth.ts | 129 +++++++++++++++++++++++++++++++++++++ src/utils/cache.ts | 81 +++++++++++++++++++++++ src/utils/metrics.ts | 122 +++++++++++++++++++++++++++++++++++ src/utils/rateLimit.ts | 54 +++++++++++++++- src/utils/retry.ts | 91 ++++++++++++++++++++++++++ src/utils/telegramQueue.ts | 46 ++++++++++++- 10 files changed, 557 insertions(+), 11 deletions(-) create mode 100644 src/utils/botHealth.ts create mode 100644 src/utils/cache.ts create mode 100644 src/utils/metrics.ts create mode 100644 src/utils/retry.ts diff --git a/bun.lock b/bun.lock index 3944b1e..97d63f5 100644 --- a/bun.lock +++ b/bun.lock @@ -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=="], diff --git a/package.json b/package.json index c2de5da..abb2844 100644 --- a/package.json +++ b/package.json @@ -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", diff --git a/src/routes/files.ts b/src/routes/files.ts index f7143ed..a76d96b 100644 --- a/src/routes/files.ts +++ b/src/routes/files.ts @@ -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 { + setTimeout(async () => { try { - unlinkSync(tempPath); - } catch {} - }, 50); + await unlink(tempPath); + } catch (err) { + logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) }); + } + }, 500); } }; diff --git a/src/utils/botHealth.ts b/src/utils/botHealth.ts new file mode 100644 index 0000000..f7c0cd5 --- /dev/null +++ b/src/utils/botHealth.ts @@ -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 = 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 }; diff --git a/src/utils/cache.ts b/src/utils/cache.ts new file mode 100644 index 0000000..df272dd --- /dev/null +++ b/src/utils/cache.ts @@ -0,0 +1,81 @@ +// Simple in-memory cache with TTL support +interface CacheEntry { + value: T; + expiresAt: number; +} + +class Cache { + private store = new Map>(); + 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 }; diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts new file mode 100644 index 0000000..9a88153 --- /dev/null +++ b/src/utils/metrics.ts @@ -0,0 +1,122 @@ +import logger from './logger'; + +interface Metric { + name: string; + value: number; + timestamp: number; + tags?: Record; +} + +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): 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 }; diff --git a/src/utils/rateLimit.ts b/src/utils/rateLimit.ts index 1db174f..4774b9b 100644 --- a/src/utils/rateLimit.ts +++ b/src/utils/rateLimit.ts @@ -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(); +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, +}); diff --git a/src/utils/retry.ts b/src/utils/retry.ts new file mode 100644 index 0000000..d3af93e --- /dev/null +++ b/src/utils/retry.ts @@ -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 = { + 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 ( + fn: () => Promise, + options: RetryOptions = {}, +): Promise => { + 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 ( + fn: () => Promise, + timeoutMs: number = 30000, +): Promise => { + return Promise.race([ + fn(), + new Promise((_, reject) => + setTimeout(() => reject(new Error(`Operation timeout after ${timeoutMs}ms`)), timeoutMs), + ), + ]); +}; + +export const withFallback = async ( + primary: () => Promise, + fallback: () => Promise, +): Promise => { + try { + return await primary(); + } catch (error: unknown) { + logger.warn('Primary operation failed, using fallback', { + error: error instanceof Error ? error.message : String(error), + }); + return fallback(); + } +}; diff --git a/src/utils/telegramQueue.ts b/src/utils/telegramQueue.ts index 2e51a54..18f9882 100644 --- a/src/utils/telegramQueue.ts +++ b/src/utils/telegramQueue.ts @@ -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 = (task: () => Promise): Promise => { - 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 => { + uploadQueue.clear(); + await uploadQueue.onIdle(); +}; + +export const waitForQueue = async (): Promise => { + await uploadQueue.onIdle(); };