router-lists-ui/backend/services/s3Service.js

/**
 * Сервис для работы с S3 (Yandex Object Storage)
 * Централизованные операции чтения/записи/кэширования
 */

const { S3Client, GetObjectCommand, HeadObjectCommand, PutObjectCommand, DeleteObjectCommand, ListObjectsV2Command } = 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);
}

/**
 * Список объектов в S3 по префиксу
 * Используется для истории/бэкапов (количество ограничено для безопасности)
 */
async function listS3Objects(prefix, { maxKeys = 100 } = {}) {
  const out = [];
  let continuationToken = undefined;

  while (out.length < maxKeys) {
    const resp = await s3.send(new ListObjectsV2Command({
      Bucket: BUCKET_NAME,
      Prefix: prefix,
      ContinuationToken: continuationToken,
      MaxKeys: Math.min(1000, maxKeys - out.length),
    }));

    const contents = resp.Contents || [];
    for (const obj of contents) {
      out.push({
        key: obj.Key,
        size: typeof obj.Size === 'number' ? obj.Size : null,
        lastModified: obj.LastModified ? new Date(obj.LastModified).toISOString() : null,
        etag: obj.ETag || null,
      });
      if (out.length >= maxKeys) break;
    }

    if (!resp.IsTruncated || !resp.NextContinuationToken || out.length >= maxKeys) {
      break;
    }
    continuationToken = resp.NextContinuationToken;
  }

  return out;
}

/**
 * Потоковое чтение с пагинацией больших текстовых файлов
 */
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,
  listS3Objects,
};