#!/usr/bin/env node /** * S3 Monitor - отслеживает изменения в S3 bucket через polling * Работает в Docker контейнере */ const { S3Client, ListObjectsV2Command } = require('@aws-sdk/client-s3'); const Redis = require('ioredis'); const CONFIG = { s3: { endpoint: 'https://s3.twcstorage.ru', region: 'ru-1', credentials: { accessKeyId: process.env.S3_ACCESS_KEY, secretAccessKey: process.env.S3_SECRET_KEY }, bucket: process.env.S3_BUCKET }, redis: { host: process.env.REDIS_HOST || '147.45.146.17', port: parseInt(process.env.REDIS_PORT || '6379'), password: process.env.REDIS_PASSWORD }, pollInterval: parseInt(process.env.POLL_INTERVAL || '30000'), stateKey: 'crm:s3:monitor:state' }; const s3Client = new S3Client({ endpoint: CONFIG.s3.endpoint, region: CONFIG.s3.region, credentials: CONFIG.s3.credentials, forcePathStyle: true }); const redis = new Redis({ host: CONFIG.redis.host, port: CONFIG.redis.port, password: CONFIG.redis.password }); // Хранилище состояния файлов (ETag/LastModified) let fileState = {}; async function loadState() { try { const data = await redis.get(CONFIG.stateKey); if (data) { fileState = JSON.parse(data); console.log(`📥 Загружено ${Object.keys(fileState).length} файлов из состояния`); } else { console.log('📥 Состояние пустое (первый запуск)'); } } catch (err) { console.error('⚠️ Ошибка загрузки состояния:', err.message); } } async function saveState() { try { await redis.set(CONFIG.stateKey, JSON.stringify(fileState)); } catch (err) { console.error('⚠️ Ошибка сохранения состояния:', err.message); } } async function scanS3() { try { console.log(`\n🔍 Сканирование S3... (${new Date().toISOString()})`); const currentFiles = {}; let newCount = 0; let modifiedCount = 0; let deletedCount = 0; let continuationToken = null; let pageCount = 0; // Pagination - получаем ВСЕ файлы do { pageCount++; const command = new ListObjectsV2Command({ Bucket: CONFIG.s3.bucket, ContinuationToken: continuationToken }); const response = await s3Client.send(command); console.log(` Страница ${pageCount}: ${response.Contents?.length || 0} файлов`); // Обрабатываем файлы for (const object of response.Contents || []) { const key = object.Key; const etag = object.ETag; const lastModified = object.LastModified.toISOString(); currentFiles[key] = { etag, lastModified }; // Проверяем изменения if (!fileState[key]) { // Новый файл (только если это не первый запуск) if (Object.keys(fileState).length > 0) { await publishEvent('created', key, object); newCount++; } } else if (fileState[key].etag !== etag) { // Файл изменён await publishEvent('modified', key, object); modifiedCount++; } } // Проверяем есть ли ещё страницы continuationToken = response.NextContinuationToken; } while (continuationToken); console.log(` ✅ Загружено ${pageCount} страниц`); // Проверяем удалённые файлы for (const key in fileState) { if (!currentFiles[key]) { await publishEvent('deleted', key, null); deletedCount++; } } fileState = currentFiles; await saveState(); console.log(`✅ Сканирование завершено:`); console.log(` 📊 Всего файлов: ${Object.keys(currentFiles).length}`); console.log(` 🆕 Новых: ${newCount}`); console.log(` ✏️ Изменённых: ${modifiedCount}`); console.log(` 🗑️ Удалённых: ${deletedCount}`); } catch (err) { console.error('❌ Ошибка сканирования S3:', err.message); console.error(' Детали:', err.name, err.$metadata?.httpStatusCode); console.error(' Stack:', err.stack); } } async function publishEvent(action, key, object) { const timestamp = new Date().toISOString(); const event = { type: 'file_' + action, source: 's3_monitor', timestamp: timestamp, path: key, filename: key.split('/').pop(), action: action }; if (object) { event.size = object.Size; event.etag = object.ETag; event.last_modified = object.LastModified.toISOString(); } console.log(`\n 📢 ${action.toUpperCase()}: ${event.filename}`); try { await redis.publish('crm:file:events', JSON.stringify(event)); console.log(` ✅ Опубликовано в Redis`); } catch (err) { console.error(` ❌ Ошибка публикации:`, err.message); } } async function start() { console.log('🚀 S3 Monitor (Docker)'); console.log('════════════════════════════════════════════════════════════════════════════════'); console.log(`📦 Bucket: ${CONFIG.s3.bucket}`); console.log(`📡 Redis: ${CONFIG.redis.host}:${CONFIG.redis.port}`); console.log(`🔄 Интервал: ${CONFIG.pollInterval / 1000}с`); console.log('════════════════════════════════════════════════════════════════════════════════\n'); await loadState(); console.log('👂 Начинаем мониторинг...\n'); // Первое сканирование await scanS3(); // Периодическое сканирование setInterval(scanS3, CONFIG.pollInterval); } // Запуск redis.on('connect', () => { console.log('✅ Подключились к Redis\n'); start(); }); redis.on('error', (err) => { console.error('❌ Redis ошибка:', err.message); }); process.on('SIGINT', () => { console.log('\n\n⛔ Остановка мониторинга...'); redis.disconnect(); process.exit(0); });