/** * Сервис для работы с S3 (Yandex Object Storage) * Централизованные операции чтения/записи/кэширования */ const { S3Client, GetObjectCommand, HeadObjectCommand, PutObjectCommand, DeleteObjectCommand } = require('@aws-sdk/client-s3'); const { NodeHttpHandler } = require('@smithy/node-http-handler'); const http = require('http'); const https = require('https'); // S3 Client setup const s3 = new S3Client({ endpoint: 'https://storage.yandexcloud.net', region: process.env.AWS_REGION, forcePathStyle: true, maxAttempts: 3, requestHandler: new NodeHttpHandler({ httpAgent: new http.Agent({ keepAlive: true }), httpsAgent: new https.Agent({ keepAlive: true }) }), credentials: { accessKeyId: process.env.AWS_ACCESS_KEY_ID || process.env.S3_ACCESS_KEY_ID, secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY || process.env.S3_SECRET_ACCESS_KEY, } }); const BUCKET_NAME = process.env.S3_BUCKET_NAME; // In-memory cache const cache = { text: new Map(), head: new Map(), max: 100, ttlMs: 30_000 }; /** * Конвертирует stream в строку */ async function streamToString(stream) { if (!stream) return ''; if (typeof stream.transformToString === 'function') { return await stream.transformToString(); } return await new Promise((resolve, reject) => { let chunks = []; stream.on('data', (c) => chunks.push(Buffer.isBuffer(c) ? c : Buffer.from(String(c)))); stream.once('error', reject); stream.once('end', () => resolve(Buffer.concat(chunks).toString('utf-8'))); }); } /** * Получить значение из кэша */ function getCache(map, key) { const v = map.get(key); if (!v) return null; if (Date.now() > v.at + cache.ttlMs) { map.delete(key); return null; } return v.value; } /** * Установить значение в кэш */ function setCache(map, key, value) { if (map.size >= cache.max) { const firstKey = map.keys().next().value; if (firstKey) map.delete(firstKey); } map.set(key, { value, at: Date.now() }); } /** * Инвалидировать кэш для ключа */ function invalidateCacheForKey(key) { try { cache.text.delete(key); } catch {} try { cache.head.delete(key); } catch {} } /** * Прочитать текстовый объект из S3 */ async function readS3TextObject(key, s3Duration = null) { const cached = getCache(cache.text, key); if (cached) return cached; const s3Start = Date.now(); const data = await s3.send(new GetObjectCommand({ Bucket: BUCKET_NAME, Key: key })); if (s3Duration) { try { s3Duration.labels('getObject').observe((Date.now() - s3Start)/1000); } catch {} } const out = { body: await streamToString(data.Body), etag: data.ETag || undefined, lastModified: data.LastModified ? data.LastModified.toISOString() : undefined, contentLength: typeof data.ContentLength === 'number' ? data.ContentLength : undefined }; setCache(cache.text, key, out); return out; } /** * Получить ETag объекта из S3 */ async function headS3ObjectEtag(key, s3Duration = null) { const cached = getCache(cache.head, key); if (cached && cached.etag) return cached.etag; const s3Start = Date.now(); const head = await s3.send(new HeadObjectCommand({ Bucket: BUCKET_NAME, Key: key })); if (s3Duration) { try { s3Duration.labels('headObject').observe((Date.now() - s3Start)/1000); } catch {} } setCache(cache.head, key, { etag: head.ETag || undefined }); return head.ETag || undefined; } /** * Получить метаданные объекта (etag, lastModified, contentLength) */ async function headMeta(key) { try { const h = await s3.send(new HeadObjectCommand({ Bucket: BUCKET_NAME, Key: key })); return { etag: h.ETag || null, lastModified: h.LastModified ? new Date(h.LastModified).toISOString() : null, contentLength: typeof h.ContentLength === 'number' ? h.ContentLength : null, }; } catch (e) { return { etag: null, lastModified: null, contentLength: null }; } } /** * Записать текстовый объект в S3 */ async function writeS3TextObject(key, content, contentType = 'text/plain') { await s3.send(new PutObjectCommand({ Bucket: BUCKET_NAME, Key: key, Body: content, ContentType: contentType, })); invalidateCacheForKey(key); return await headMeta(key); } /** * Записать JSON объект в S3 */ async function writeS3JsonObject(key, data) { return await writeS3TextObject(key, JSON.stringify(data, null, 2), 'application/json'); } /** * Удалить объект из S3 */ async function deleteS3Object(key) { await s3.send(new DeleteObjectCommand({ Bucket: BUCKET_NAME, Key: key })); invalidateCacheForKey(key); } /** * Потоковое чтение с пагинацией больших текстовых файлов */ async function streamPaginatedText({ key, mapLine, q, offset = 0, limit = 0 }) { return new Promise(async (resolve, reject) => { let total = 0; const items = []; let sent = 0; let buffered = ''; const matchesQuery = (line) => { if (!q) return true; return line.toLowerCase().includes(String(q).toLowerCase()); }; try { const resp = await s3.send(new GetObjectCommand({ Bucket: BUCKET_NAME, Key: key })); const stream = resp.Body; if (!stream || typeof stream.on !== 'function') { const text = await streamToString(resp.Body); const lines = text.split('\n'); for (let i = 0; i < lines.length; i++) { const line = String(lines[i] || '').trim(); if (!line) continue; if (!matchesQuery(line)) continue; total++; const pos = total - 1; if (limit > 0) { if (pos >= offset && sent < limit) { items.push(mapLine(line)); sent++; } } else { items.push(mapLine(line)); } } return resolve({ items, total }); } stream.on('data', (chunk) => { buffered += chunk.toString('utf-8'); let lines = buffered.split('\n'); buffered = lines.pop(); for (const lnRaw of lines) { const line = lnRaw.trim(); if (!line) continue; if (!matchesQuery(line)) continue; total++; const pos = total - 1; if (limit > 0) { if (pos >= offset && sent < limit) { items.push(mapLine(line)); sent++; } } else { items.push(mapLine(line)); } } }); stream.on('end', () => { const last = (buffered || '').trim(); if (last) { if (!q || last.toLowerCase().includes(String(q).toLowerCase())) { total++; if (limit > 0) { const pos = total - 1; if (pos >= offset && items.length < limit) items.push(mapLine(last)); } else { items.push(mapLine(last)); } } } resolve({ items, total }); }); stream.on('error', reject); } catch (e) { reject(e); } }); } module.exports = { s3, BUCKET_NAME, streamToString, readS3TextObject, headS3ObjectEtag, headMeta, writeS3TextObject, writeS3JsonObject, deleteS3Object, streamPaginatedText, invalidateCacheForKey, };