diff --git a/.env.example b/.env.example index 7719a69..525dba9 100644 --- a/.env.example +++ b/.env.example @@ -7,4 +7,5 @@ PORT=3000 NODE_ENV=production LOG_LEVEL=info RATE_LIMIT_WINDOW_MS=60000 -RATE_LIMIT_MAX_REQUESTS=30 \ No newline at end of file +RATE_LIMIT_MAX_REQUESTS=30 +# TRUST_PROXY=true # Uncomment when behind reverse proxy (Traefik, Nginx) for correct client IP detection \ No newline at end of file diff --git a/schema.sql b/schema.sql index bcaf1c5..22941be 100644 --- a/schema.sql +++ b/schema.sql @@ -21,14 +21,6 @@ CREATE TABLE IF NOT EXISTS files ( updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -ALTER TABLE files ADD COLUMN IF NOT EXISTS file_hash VARCHAR; -ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_telegram_file_id VARCHAR; -ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_storage_message_id BIGINT; -ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_file_name VARCHAR; -ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_entry_name VARCHAR; -ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_mime_type VARCHAR; -ALTER TABLE files ADD COLUMN IF NOT EXISTS archive_size_bytes BIGINT; - CREATE INDEX IF NOT EXISTS idx_files_public_id ON files(public_id); CREATE INDEX IF NOT EXISTS idx_files_telegram_file_id ON files(telegram_file_id); CREATE INDEX IF NOT EXISTS idx_files_file_hash ON files(file_hash); diff --git a/src/bot.ts b/src/bot.ts index 5f1af65..ac1116a 100644 --- a/src/bot.ts +++ b/src/bot.ts @@ -42,6 +42,15 @@ export const startBot = async (): Promise> => { ); }); + // Logging middleware must be registered BEFORE the media handler so all events are captured + bot.use((ctx, next) => { + logger.info('Telegram event received', { + type: 'type' in ctx.update ? ctx.update.type : undefined, + chat_id: ctx.chat?.id, + }); + return next(); + }); + const mediaBot = bot as unknown as MediaEventRegistrar; mediaBot.on( ['document', 'photo', 'video', 'audio', 'voice', 'animation', 'sticker', 'video_note'], @@ -116,14 +125,6 @@ export const startBot = async (): Promise> => { }, ); - bot.use((ctx, next) => { - logger.info('Telegram event received', { - type: 'type' in ctx.update ? ctx.update.type : undefined, - chat_id: ctx.chat?.id, - }); - return next(); - }); - await bot.launch(); logger.info('Telegram bot started', { botToken: `${config.botToken?.substring(0, 10)}...` }); diff --git a/src/index.ts b/src/index.ts index 5049597..9376553 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,7 +5,9 @@ import { handleFileInfo, handleFileRedirect } from './routes/files'; import { handleHealth } from './routes/health'; import { handleSwaggerHtml, handleSwaggerJson } from './routes/swagger'; import { handleUpload } from './routes/upload'; +import { fileInfoCache } from './utils/cache'; import logger from './utils/logger'; +import { metricsCollector } from './utils/metrics'; import { cleanupRateLimitCache, withRateLimit } from './utils/rateLimit'; const server = serve({ @@ -52,6 +54,28 @@ const gracefulShutdown = async (signal: string): Promise => { process.on('SIGTERM', () => gracefulShutdown('SIGTERM')); process.on('SIGINT', () => gracefulShutdown('SIGINT')); +// Periodic maintenance intervals setInterval(cleanupRateLimitCache, 60000); +setInterval( + () => { + const removed = fileInfoCache.cleanup(); + if (removed > 0) { + logger.info(`Cleaned up ${removed} expired cache entries`); + } + }, + 5 * 60 * 1000, +); +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, +); logger.info('Application running successfully'); diff --git a/src/routes/files.ts b/src/routes/files.ts index 71e0556..5ffbf31 100644 --- a/src/routes/files.ts +++ b/src/routes/files.ts @@ -1,11 +1,11 @@ import { createReadStream } from 'node:fs'; -import { unlink } from 'node:fs/promises'; import { nanoid } from 'nanoid'; import { findFileByPublicId } from '../db/files'; import { fileInfoCache } from '../utils/cache'; -import { formatCreatedAt, getErrorMessage } from '../utils/file'; +import { cleanupTempFile, formatCreatedAt, getErrorMessage } from '../utils/file'; import logger from '../utils/logger'; -import { getFileInfo } from '../utils/telegram'; +import { metricsCollector } from '../utils/metrics'; +import { getFileInfo, type TelegramFileInfo } from '../utils/telegram'; import { locateZipEntry } from '../utils/zip'; type RequestWithParams = Request & { @@ -16,30 +16,24 @@ type RequestWithParams = Request & { const getTelegramFileInfo = async (telegramFileId: string, public_id: string) => { const cacheKey = `file_info_${telegramFileId}`; - let fileInfo = fileInfoCache.get(cacheKey) as any; + let fileInfo = fileInfoCache.get(cacheKey) as TelegramFileInfo | null; if (!fileInfo) { + metricsCollector.recordCacheMiss(); fileInfo = await getFileInfo(telegramFileId); fileInfoCache.set(cacheKey, fileInfo); logger.debug('File info cached', { public_id, cacheKey }); } else { + metricsCollector.recordCacheHit(); logger.debug('File info from cache', { public_id, cacheKey }); } - return fileInfo as { file_size: number; mime_type: string; file_path: string; bot_token: string }; + return fileInfo; }; const buildTelegramFileUrl = (filePath: string, botToken: string): string => `https://api.telegram.org/file/bot${botToken}/${filePath}`; -const cleanupTempFile = async (tempPath: string): Promise => { - try { - await unlink(tempPath); - } catch (err) { - logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) }); - } -}; - const sanitizeFilenameHeader = (fileName: string): string => fileName.replace(/[\\"]/g, '').replace(/[\n\r]/g, ''); @@ -93,7 +87,7 @@ export const handleFileRedirect = async (req: RequestWithParams): Promise { return null; }; -const cleanupTempFile = async (tempPath: string): Promise => { - try { - await unlink(tempPath); - } catch (err) { - logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) }); - } -}; - const streamFileToTemp = async (file: File, maxSizeBytes: number): Promise => { const tempPath = `/tmp/teleuploader-${nanoid()}`; const writer = createWriteStream(tempPath); @@ -145,6 +138,7 @@ const writeBufferToTemp = async (fileBuffer: Buffer, fileHash: string): Promise< }; export const handleUpload = async (req: Request): Promise => { + const startTime = performance.now(); try { const contentType = req.headers.get('content-type') || ''; const oversizedResponse = rejectOversizedRequest(req); @@ -161,9 +155,12 @@ export const handleUpload = async (req: Request): Promise => { { status: 400 }, ); } catch (error: unknown) { + metricsCollector.recordError(); const message = getErrorMessage(error); logger.error('Upload error', { error: message }); return Response.json({ error: message }, { status: 500 }); + } finally { + metricsCollector.recordUploadTime(performance.now() - startTime); } }; diff --git a/src/utils/botHealth.ts b/src/utils/botHealth.ts deleted file mode 100644 index f7c0cd5..0000000 --- a/src/utils/botHealth.ts +++ /dev/null @@ -1,129 +0,0 @@ -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 index bbdc398..4dd28ad 100644 --- a/src/utils/cache.ts +++ b/src/utils/cache.ts @@ -71,15 +71,4 @@ export const fileInfoCache = new Cache<{ bot_token: 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/file.ts b/src/utils/file.ts index c581383..b99c4cf 100644 --- a/src/utils/file.ts +++ b/src/utils/file.ts @@ -1,7 +1,18 @@ +import { unlink } from 'node:fs/promises'; +import logger from './logger'; + export const getErrorMessage = (error: unknown): string => { return error instanceof Error ? error.message : String(error); }; +export const cleanupTempFile = async (tempPath: string): Promise => { + try { + await unlink(tempPath); + } catch (err) { + logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) }); + } +}; + interface FileMetadata { publicId: string; telegramFileId: string; diff --git a/src/utils/logger.ts b/src/utils/logger.ts index 910d666..4b99051 100644 --- a/src/utils/logger.ts +++ b/src/utils/logger.ts @@ -12,16 +12,14 @@ const logger = winston.createLogger({ // Write all logs including error logs to file new winston.transports.File({ filename: 'logs/error.log', level: 'error' }), new winston.transports.File({ filename: 'logs/combined.log' }), + // Console transport for docker logs / CLI visibility + new winston.transports.Console({ + format: + process.env.NODE_ENV !== 'production' + ? winston.format.combine(winston.format.colorize(), winston.format.simple()) + : winston.format.json(), + }), ], }); -// If not production, also log to console -if (process.env.NODE_ENV !== 'production') { - logger.add( - new winston.transports.Console({ - format: winston.format.combine(winston.format.colorize(), winston.format.simple()), - }), - ); -} - export default logger; diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts index 1358675..fc49103 100644 --- a/src/utils/metrics.ts +++ b/src/utils/metrics.ts @@ -1,4 +1,4 @@ -import logger from './logger'; +// No imports needed — logger used only by setInterval which moved to index.ts interface Metric { name: string; @@ -108,18 +108,4 @@ class MetricsCollector { 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/telegram.ts b/src/utils/telegram.ts index e0e508f..bfbed4f 100644 --- a/src/utils/telegram.ts +++ b/src/utils/telegram.ts @@ -129,27 +129,6 @@ const buildSendPayload = (fileType: string, fileName: string): SendPayload => { return basePayload; }; -const getMediaGroupType = (fileType: string): string => { - if (fileType === 'photo') return 'photo'; - if (fileType === 'video') return 'video'; - if (fileType === 'audio') return 'audio'; - return 'document'; -}; - -interface MediaGroupPayloadItem { - type: string; - media: string; - caption: string; -} - -const buildMediaGroup = (items: MediaGroupItem[]): MediaGroupPayloadItem[] => { - return items.map((item) => ({ - type: getMediaGroupType(item.fileType), - media: item.fileId, - caption: item.fileName, - })); -}; - export const forwardToStorage = async ( fileChunk: unknown, fileName: string, @@ -184,57 +163,6 @@ export const forwardToStorage = async ( } }; -export interface MediaGroupItem { - fileId: string; - fileName: string; - fileType: string; -} - -export const forwardMediaGroupToStorage = async ( - items: MediaGroupItem[], -): Promise<{ - storageMessageId: number; - telegramFileIds: string[]; - telegramFileUniqueIds: string[]; -}> => { - try { - const result = await enqueueUpload(async (): Promise => { - const mediaGroup = buildMediaGroup(items); - - return executeWithBotRetry((activeBot) => { - const sendMediaGroup = activeBot.telegram.sendMediaGroup as unknown as ( - chatId: number, - media: MediaGroupPayloadItem[], - ) => Promise; - return sendMediaGroup(config.storageChatId, mediaGroup); - }); - }); - - const messages = Array.isArray(result) ? result : [result]; - const storageMessageId = messages[0]?.message_id || 0; - - const telegramFileIds: string[] = []; - const telegramFileUniqueIds: string[] = []; - - for (let i = 0; i < messages.length; i++) { - const uploadedFile = extractUploadedFile(messages[i], items[i]?.fileType || 'document'); - telegramFileIds.push(uploadedFile?.file_id || ''); - telegramFileUniqueIds.push(uploadedFile?.file_unique_id || ''); - } - - return { - storageMessageId, - telegramFileIds, - telegramFileUniqueIds, - }; - } catch (error: unknown) { - logger.error('Failed to forward media group to storage', { - error: error instanceof Error ? error.message : String(error), - }); - throw error; - } -}; - export const getFileInfo = async (telegramFileId: string): Promise => { let lastError: unknown; for (const activeBot of bots) { diff --git a/src/utils/uploadBatcher.ts b/src/utils/uploadBatcher.ts index b5aa18e..799afd4 100644 --- a/src/utils/uploadBatcher.ts +++ b/src/utils/uploadBatcher.ts @@ -1,11 +1,9 @@ import { createReadStream } from 'node:fs'; -import { unlink } from 'node:fs/promises'; import { nanoid } from 'nanoid'; import { db, files as fileSchema } from '../db'; import type { NewFile } from '../db/schema'; import { config } from '../env'; -import { getErrorMessage } from './file'; -import logger from './logger'; +import { cleanupTempFile } from './file'; import { forwardToStorage } from './telegram'; import { createZip, type ZipEntry } from './zip'; @@ -38,14 +36,6 @@ const BATCH_WINDOW_MS = 2000; let pendingUploads: PendingUpload[] = []; let flushTimer: ReturnType | null = null; -const cleanupTempFile = async (tempPath: string): Promise => { - try { - await unlink(tempPath); - } catch (error) { - logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(error) }); - } -}; - const buildUploadedFile = ( item: BatchUploadItem, entry: ZipEntry, @@ -124,6 +114,12 @@ const flushUploads = async (): Promise => { } finally { await Promise.all(batch.map((item) => cleanupTempFile(item.prepared.tempPath))); if (zipTempPath) await cleanupTempFile(zipTempPath); + // Reschedule timer if new items arrived during async processing + if (pendingUploads.length > 0 && !flushTimer) { + flushTimer = setTimeout(() => { + void flushUploads(); + }, BATCH_WINDOW_MS); + } } }; diff --git a/test/bootstrap.test.ts b/test/bootstrap.test.ts index af2a329..0f3d9fc 100644 --- a/test/bootstrap.test.ts +++ b/test/bootstrap.test.ts @@ -47,6 +47,9 @@ mock.module('../src/routes/health', () => ({ mock.module('../src/utils/rateLimit', () => ({ cleanupRateLimitCache: mock(), + withRateLimit: ( + handler: (req: T) => Promise, + ): ((req: T) => Promise) => handler, })); describe('Bootstrap Server', () => { diff --git a/test/env.test.ts b/test/env.test.ts index 6072bb0..64e5056 100644 --- a/test/env.test.ts +++ b/test/env.test.ts @@ -39,7 +39,7 @@ describe('Environment Variables Validation', () => { expect(config.rateLimitWindowMs).toBe(60000); }); - it('rateLimitMaxRequests should default to 30 when not specified', () => { - expect(config.rateLimitMaxRequests).toBe(30); + it('rateLimitMaxRequests should default to 150 when not specified', () => { + expect(config.rateLimitMaxRequests).toBe(150); }); }); diff --git a/test/files.test.ts b/test/files.test.ts index fefcc0a..ee4cb12 100644 --- a/test/files.test.ts +++ b/test/files.test.ts @@ -60,25 +60,16 @@ mock.module('../src/db/files', () => ({ })); // Mock telegram utils -const mockGetFile = mock(() => Promise.resolve({ file_path: 'photos/file_0.jpg' })); -mock.module('../src/utils/telegram', () => ({ - getBot: () => ({ - telegram: { - getFile: mockGetFile, - }, - }), +const mockGetFileInfo = mock(async (_telegramFileId: string) => ({ + file_size: 98765, + mime_type: 'image/jpeg', + file_path: 'photos/file_0.jpg', + bot_token: '123456:ABC-DEF', })); -// Mock global fetch for proxy path -const originalFetch = globalThis.fetch; -const mockGlobalFetch = mock(async (_url: string) => - Promise.resolve( - new Response('fake-file-content', { - status: 200, - headers: { 'Content-Type': 'application/octet-stream' }, - }), - ), -); +mock.module('../src/utils/telegram', () => ({ + getFileInfo: mockGetFileInfo, +})); describe('File Route Handlers', () => { let handleFileRedirect: typeof import('../src/routes/files').handleFileRedirect; @@ -86,12 +77,10 @@ describe('File Route Handlers', () => { beforeEach(async () => { mockSelect.mockClear(); - mockGetFile.mockClear(); - mockGlobalFetch.mockClear(); + mockGetFileInfo.mockClear(); // Set up mock token process.env.BOT_TOKEN = '123456:ABC-DEF'; - globalThis.fetch = mockGlobalFetch as any; const filesRoute = await import('../src/routes/files'); handleFileRedirect = filesRoute.handleFileRedirect; @@ -100,7 +89,6 @@ describe('File Route Handlers', () => { afterAll(() => { mock.restore(); - globalThis.fetch = originalFetch; }); describe('handleFileRedirect', () => { diff --git a/test/telegram.test.ts b/test/telegram.test.ts index 95d3961..44ea404 100644 --- a/test/telegram.test.ts +++ b/test/telegram.test.ts @@ -28,6 +28,7 @@ mock.module('telegraf', () => { constructor(token) { this.token = token; this.telegram = { + token: token, sendPhoto: mock(() => Promise.resolve({ message_id: 12345,