Backend: SQLite storage, EvoBGP integration, filters in SQL

Made-with: Cursor
2026-04-20 01:17:00 +07:00
parent 7f3eaea40a
commit 1c6e6ab24a
27 changed files with 2406 additions and 2542 deletions
@@ -0,0 +1,167 @@
+/**
+ * SQLite blob store (logical S3 keys → rows in `blobs`).
+ */
+
+const crypto = require('crypto');
+const { openDatabase } = require('../db/sqliteDb');
+
+function sha256Etag(buf) {
+  const h = crypto.createHash('sha256').update(buf).digest('hex');
+  return `"${h}"`;
+}
+
+function namespaceFromKey(objectKey) {
+  const k = String(objectKey || '');
+  if (k.startsWith('ping-cache/') || k.startsWith('speed-test-cache/') || k.startsWith('network-map-cache/') ||
+      k.startsWith('uptime-monitor-cache/') || k.startsWith('ping-services/')) {
+    return 'cache';
+  }
+  if (k.startsWith('filter-manager/config-') || k.startsWith('filter-manager/server-filters-')) {
+    return 'per_server';
+  }
+  if (k === 'mikrotik-frouting-config.txt') {
+    return 'derived';
+  }
+  return 'config';
+}
+
+function isVersionedKey(db, objectKey) {
+  const row = db.prepare('SELECT 1 FROM versioned_keys WHERE object_key = ?').get(objectKey);
+  return Boolean(row);
+}
+
+function trimVersions(db, objectKey, keep = 50) {
+  db.prepare(
+    `DELETE FROM blob_versions WHERE object_key = ? AND id NOT IN (
+       SELECT id FROM blob_versions WHERE object_key = ? ORDER BY created_at DESC, id DESC LIMIT ?
+     )`
+  ).run(objectKey, objectKey, keep);
+}
+
+/**
+ * @param {Buffer|string} body
+ */
+function writeBlobTx(db, objectKey, body, contentType) {
+  const buf = Buffer.isBuffer(body) ? body : Buffer.from(String(body), 'utf8');
+  const ns = namespaceFromKey(objectKey);
+  const etag = sha256Etag(buf);
+  const now = Date.now();
+  const size = buf.length;
+
+  const existing = db.prepare('SELECT body, content_type, etag, updated_at, created_at FROM blobs WHERE object_key = ?').get(objectKey);
+
+  const run = () => {
+    if (existing && isVersionedKey(db, objectKey)) {
+      const oldBuf = existing.body;
+      db.prepare(
+        `INSERT INTO blob_versions (object_key, body, byte_size, etag, created_at)
+         VALUES (?, ?, ?, ?, ?)`
+      ).run(objectKey, oldBuf, oldBuf.length, existing.etag, existing.updated_at);
+      trimVersions(db, objectKey);
+    }
+
+    if (existing) {
+      db.prepare(
+        `UPDATE blobs SET namespace = ?, content_type = ?, body = ?, byte_size = ?, etag = ?, updated_at = ?
+         WHERE object_key = ?`
+      ).run(ns, contentType, buf, size, etag, now, objectKey);
+    } else {
+      db.prepare(
+        `INSERT INTO blobs (object_key, namespace, content_type, body, byte_size, etag, updated_at, created_at)
+         VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
+      ).run(objectKey, ns, contentType, buf, size, etag, now, now);
+    }
+  };
+
+  db.transaction(run)();
+  return {
+    etag,
+    lastModified: new Date(now).toISOString(),
+    contentLength: size,
+  };
+}
+
+function readBlob(objectKey) {
+  const db = openDatabase();
+  const row = db.prepare(
+    'SELECT body, etag, updated_at, byte_size FROM blobs WHERE object_key = ?'
+  ).get(objectKey);
+  if (!row) return null;
+  const bodyStr = row.body.toString('utf8');
+  return {
+    body: bodyStr,
+    etag: row.etag,
+    lastModified: new Date(row.updated_at).toISOString(),
+    contentLength: row.byte_size,
+  };
+}
+
+function headMetaRow(objectKey) {
+  const db = openDatabase();
+  const row = db.prepare('SELECT etag, updated_at, byte_size FROM blobs WHERE object_key = ?').get(objectKey);
+  if (!row) {
+    return { etag: null, lastModified: null, contentLength: null };
+  }
+  return {
+    etag: row.etag,
+    lastModified: new Date(row.updated_at).toISOString(),
+    contentLength: row.byte_size,
+  };
+}
+
+function headEtag(objectKey) {
+  const m = headMetaRow(objectKey);
+  return m.etag || undefined;
+}
+
+function deleteBlob(objectKey) {
+  const db = openDatabase();
+  db.prepare('DELETE FROM blobs WHERE object_key = ?').run(objectKey);
+}
+
+function listByPrefix(prefix, maxKeys = 100) {
+  const db = openDatabase();
+  const rows = db.prepare(
+    `SELECT object_key AS key, byte_size AS size, updated_at, etag FROM blobs
+     WHERE object_key >= ? AND object_key < ? ORDER BY object_key LIMIT ?`
+  ).all(prefix, prefix + '\uffff', maxKeys);
+  return rows.map((r) => ({
+    key: r.key,
+    size: r.size,
+    lastModified: new Date(r.updated_at).toISOString(),
+    etag: r.etag,
+  }));
+}
+
+function listVersions(objectKey, maxKeys = 50) {
+  const db = openDatabase();
+  return db.prepare(
+    `SELECT id, object_key, byte_size, etag, created_at FROM blob_versions
+     WHERE object_key = ? ORDER BY created_at DESC, id DESC LIMIT ?`
+  ).all(objectKey, maxKeys);
+}
+
+function rollbackToVersionId(objectKey, versionId) {
+  const db = openDatabase();
+  const vid = Number(versionId);
+  if (!Number.isFinite(vid)) throw new Error('Invalid versionId');
+  const ver = db.prepare('SELECT body, etag, created_at FROM blob_versions WHERE id = ? AND object_key = ?').get(vid, objectKey);
+  if (!ver) throw new Error('Version not found');
+  const buf = ver.body;
+  const contentType = objectKey.endsWith('.json') ? 'application/json' : 'text/plain';
+  writeBlobTx(db, objectKey, buf, contentType);
+  return headMetaRow(objectKey);
+}
+
+module.exports = {
+  namespaceFromKey,
+  writeBlobTx,
+  readBlob,
+  headMetaRow,
+  headEtag,
+  deleteBlob,
+  listByPrefix,
+  listVersions,
+  rollbackToVersionId,
+  sha256Etag,
+};