MythEclipse/TeleUploader
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: enhance file handling and metrics tracking; remove unused bot health tracker

Browse Source
This commit is contained in:
asepharyana
2026-07-06 01:25:18 +07:00
parent bbf420759b
commit 52bd704d21
17 changed files with 90 additions and 310 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+2 -1
View File
@@ -7,4 +7,5 @@ PORT=3000
NODE_ENV=production
LOG_LEVEL=info
RATE_LIMIT_WINDOW_MS=60000
RATE_LIMIT_MAX_REQUESTS=30
RATE_LIMIT_MAX_REQUESTS=30
# TRUST_PROXY=true # Uncomment when behind reverse proxy (Traefik, Nginx) for correct client IP detection
schema.sql
-8
View File
@@ -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);
src/bot.ts
+9 -8
View File
@@ -42,6 +42,15 @@ export const startBot = async (): Promise<Telegraf<Context>> => {
);
});
// 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<Telegraf<Context>> => {
},
);
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)}...` });
src/index.ts
+24
View File
@@ -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<void> => {
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');
src/routes/files.ts
+8 -14
View File
@@ -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<void> => {
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<Respon
void cleanupTempFile(tempZipPath);
});
return new Response(fileStream as any, {
return new Response(fileStream as unknown as ReadableStream, {
status: 200,
headers: {
'Content-Type': file.mimeType || 'application/octet-stream',
src/routes/upload.ts
+6 -9
View File
@@ -1,11 +1,11 @@
import { createWriteStream } from 'node:fs';
import { unlink } from 'node:fs/promises';
import { nanoid } from 'nanoid';
import { findFileByHash } from '../db/files';
import { config } from '../env';
import {
buildUploadResponse,
checkFileSize,
cleanupTempFile,
computeHash,
ensureExtension,
extractMimeType,
@@ -13,6 +13,7 @@ import {
getFileType,
} from '../utils/file';
import logger from '../utils/logger';
import { metricsCollector } from '../utils/metrics';
import { enqueuePreparedUpload, type PreparedUpload } from '../utils/uploadBatcher';
interface JsonUploadPayload {
@@ -56,14 +57,6 @@ const rejectOversizedRequest = (req: Request): Response | null => {
return null;
};
const cleanupTempFile = async (tempPath: string): Promise<void> => {
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<PreparedUpload> => {
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<Response> => {
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<Response> => {
{ 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);
}
};
src/utils/botHealth.ts
-129
View File
@@ -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<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 };
src/utils/cache.ts
-11
View File
@@ -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 };
src/utils/file.ts
+11
View File
@@ -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<void> => {
try {
await unlink(tempPath);
} catch (err) {
logger.warn('Failed to cleanup temp file', { tempPath, error: getErrorMessage(err) });
}
};
interface FileMetadata {
publicId: string;
telegramFileId: string;
src/utils/logger.ts
+7 -9
View File
@@ -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;
src/utils/metrics.ts
+1 -15
View File
@@ -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 };
src/utils/telegram.ts
-72
View File
@@ -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<TelegramMessageResult[]> => {
const mediaGroup = buildMediaGroup(items);
return executeWithBotRetry((activeBot) => {
const sendMediaGroup = activeBot.telegram.sendMediaGroup as unknown as (
chatId: number,
media: MediaGroupPayloadItem[],
) => Promise<TelegramMessageResult[]>;
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<TelegramFileInfo> => {
let lastError: unknown;
for (const activeBot of bots) {
src/utils/uploadBatcher.ts
+7 -11
View File
@@ -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<typeof setTimeout> | null = null;
const cleanupTempFile = async (tempPath: string): Promise<void> => {
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<void> => {
} 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);
}
}
};
test/bootstrap.test.ts
+3
View File
@@ -47,6 +47,9 @@ mock.module('../src/routes/health', () => ({
mock.module('../src/utils/rateLimit', () => ({
cleanupRateLimitCache: mock(),
withRateLimit: <T extends Request>(
handler: (req: T) => Promise<Response>,
): ((req: T) => Promise<Response>) => handler,
}));
describe('Bootstrap Server', () => {
test/env.test.ts
+2 -2
View File
@@ -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);
});
});
test/files.test.ts
+9 -21
View File
@@ -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', () => {
test/telegram.test.ts
+1
View File
@@ -28,6 +28,7 @@ mock.module('telegraf', () => {
constructor(token) {
this.token = token;
this.telegram = {
token: token,
sendPhoto: mock(() =>
Promise.resolve({
message_id: 12345,