crm.clientright.ru/crm_extensions/file_storage/s3_monitor_docker.js

#!/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);
});