feat(db): migrate from lowdb to SQLite with repos pattern
- Add modular DB layer (adapters, migrations, repos, helpers) - Replace localDb/usageDb/requestDetailsDb monoliths with repos - Add Tailscale tunnel integration & status check API - Add /api/cli-tools/all-statuses aggregated endpoint - Add settingsStore (Zustand) and mitm/dbReader - Add DB unit tests (benchmark, concurrent, migration, vs-lowdb)
This commit is contained in:
decolua
200
src/lib/db/repos/requestDetailsRepo.js
Normal file
200
src/lib/db/repos/requestDetailsRepo.js
Normal file
@@ -0,0 +1,200 @@
|
||||
import { getAdapter } from "../driver.js";
|
||||
import { parseJson, stringifyJson } from "../helpers/jsonCol.js";
|
||||
|
||||
const DEFAULT_MAX_RECORDS = 200;
|
||||
const DEFAULT_BATCH_SIZE = 20;
|
||||
const DEFAULT_FLUSH_INTERVAL_MS = 5000;
|
||||
const DEFAULT_MAX_JSON_SIZE = 5 * 1024;
|
||||
const CONFIG_CACHE_TTL_MS = 5000;
|
||||
|
||||
let cachedConfig = null;
|
||||
let cachedConfigTs = 0;
|
||||
|
||||
async function getObservabilityConfig() {
|
||||
if (cachedConfig && (Date.now() - cachedConfigTs) < CONFIG_CACHE_TTL_MS) return cachedConfig;
|
||||
try {
|
||||
const { getSettings } = await import("./settingsRepo.js");
|
||||
const settings = await getSettings();
|
||||
const envEnabled = process.env.OBSERVABILITY_ENABLED !== "false";
|
||||
const enabled = typeof settings.enableObservability === "boolean"
|
||||
? settings.enableObservability
|
||||
: envEnabled;
|
||||
cachedConfig = {
|
||||
enabled,
|
||||
maxRecords: settings.observabilityMaxRecords || parseInt(process.env.OBSERVABILITY_MAX_RECORDS || String(DEFAULT_MAX_RECORDS), 10),
|
||||
batchSize: settings.observabilityBatchSize || parseInt(process.env.OBSERVABILITY_BATCH_SIZE || String(DEFAULT_BATCH_SIZE), 10),
|
||||
flushIntervalMs: settings.observabilityFlushIntervalMs || parseInt(process.env.OBSERVABILITY_FLUSH_INTERVAL_MS || String(DEFAULT_FLUSH_INTERVAL_MS), 10),
|
||||
maxJsonSize: (settings.observabilityMaxJsonSize || parseInt(process.env.OBSERVABILITY_MAX_JSON_SIZE || "5", 10)) * 1024,
|
||||
};
|
||||
} catch {
|
||||
cachedConfig = {
|
||||
enabled: false,
|
||||
maxRecords: DEFAULT_MAX_RECORDS,
|
||||
batchSize: DEFAULT_BATCH_SIZE,
|
||||
flushIntervalMs: DEFAULT_FLUSH_INTERVAL_MS,
|
||||
maxJsonSize: DEFAULT_MAX_JSON_SIZE,
|
||||
};
|
||||
}
|
||||
cachedConfigTs = Date.now();
|
||||
return cachedConfig;
|
||||
}
|
||||
|
||||
let writeBuffer = [];
|
||||
let flushTimer = null;
|
||||
let isFlushing = false;
|
||||
|
||||
function sanitizeHeaders(headers) {
|
||||
if (!headers || typeof headers !== "object") return {};
|
||||
const sensitiveKeys = ["authorization", "x-api-key", "cookie", "token", "api-key"];
|
||||
const sanitized = { ...headers };
|
||||
for (const key of Object.keys(sanitized)) {
|
||||
if (sensitiveKeys.some((s) => key.toLowerCase().includes(s))) delete sanitized[key];
|
||||
}
|
||||
return sanitized;
|
||||
}
|
||||
|
||||
function generateDetailId(model) {
|
||||
const timestamp = new Date().toISOString();
|
||||
const random = Math.random().toString(36).substring(2, 8);
|
||||
const modelPart = model ? model.replace(/[^a-zA-Z0-9-]/g, "-") : "unknown";
|
||||
return `${timestamp}-${random}-${modelPart}`;
|
||||
}
|
||||
|
||||
function truncateField(obj, maxSize) {
|
||||
const str = JSON.stringify(obj || {});
|
||||
if (str.length > maxSize) {
|
||||
return { _truncated: true, _originalSize: str.length, _preview: str.substring(0, 200) };
|
||||
}
|
||||
return obj || {};
|
||||
}
|
||||
|
||||
async function flushToDatabase() {
|
||||
if (isFlushing) return;
|
||||
if (writeBuffer.length === 0) return;
|
||||
isFlushing = true;
|
||||
try {
|
||||
// Drain entire buffer (loop in case more pushed during await)
|
||||
while (writeBuffer.length > 0) {
|
||||
const items = writeBuffer.splice(0, writeBuffer.length);
|
||||
const db = await getAdapter();
|
||||
const config = await getObservabilityConfig();
|
||||
|
||||
db.transaction(() => {
|
||||
for (const item of items) {
|
||||
if (!item.id) item.id = generateDetailId(item.model);
|
||||
if (!item.timestamp) item.timestamp = new Date().toISOString();
|
||||
if (item.request?.headers) item.request.headers = sanitizeHeaders(item.request.headers);
|
||||
|
||||
const record = {
|
||||
id: item.id,
|
||||
provider: item.provider || null,
|
||||
model: item.model || null,
|
||||
connectionId: item.connectionId || null,
|
||||
timestamp: item.timestamp,
|
||||
status: item.status || null,
|
||||
latency: item.latency || {},
|
||||
tokens: item.tokens || {},
|
||||
request: truncateField(item.request, config.maxJsonSize),
|
||||
providerRequest: truncateField(item.providerRequest, config.maxJsonSize),
|
||||
providerResponse: truncateField(item.providerResponse, config.maxJsonSize),
|
||||
response: truncateField(item.response, config.maxJsonSize),
|
||||
};
|
||||
|
||||
db.run(
|
||||
`INSERT INTO requestDetails(id, timestamp, provider, model, connectionId, status, data) VALUES(?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET timestamp = excluded.timestamp, provider = excluded.provider, model = excluded.model, connectionId = excluded.connectionId, status = excluded.status, data = excluded.data`,
|
||||
[record.id, record.timestamp, record.provider, record.model, record.connectionId, record.status, stringifyJson(record)]
|
||||
);
|
||||
}
|
||||
|
||||
const cnt = db.get(`SELECT COUNT(*) as c FROM requestDetails`);
|
||||
if (cnt && cnt.c > config.maxRecords) {
|
||||
db.run(
|
||||
`DELETE FROM requestDetails WHERE id IN (SELECT id FROM requestDetails ORDER BY timestamp ASC LIMIT ?)`,
|
||||
[cnt.c - config.maxRecords]
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
} catch (e) {
|
||||
console.error("[requestDetailsRepo] Batch write failed:", e);
|
||||
} finally {
|
||||
isFlushing = false;
|
||||
}
|
||||
}
|
||||
|
||||
export async function saveRequestDetail(detail) {
|
||||
const config = await getObservabilityConfig();
|
||||
if (!config.enabled) return;
|
||||
|
||||
writeBuffer.push(detail);
|
||||
|
||||
// Trigger immediate flush if batch threshold reached.
|
||||
// flushToDatabase() drains entire buffer in a loop, so all pushes during await are persisted.
|
||||
if (writeBuffer.length >= config.batchSize) {
|
||||
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
|
||||
flushToDatabase().catch((e) => console.error("[requestDetailsRepo] flush err:", e));
|
||||
} else if (!flushTimer) {
|
||||
flushTimer = setTimeout(() => {
|
||||
flushTimer = null;
|
||||
flushToDatabase().catch(() => {});
|
||||
}, config.flushIntervalMs);
|
||||
}
|
||||
}
|
||||
|
||||
export async function getRequestDetails(filter = {}) {
|
||||
const db = await getAdapter();
|
||||
const conds = [];
|
||||
const params = [];
|
||||
|
||||
if (filter.provider) { conds.push("provider = ?"); params.push(filter.provider); }
|
||||
if (filter.model) { conds.push("model = ?"); params.push(filter.model); }
|
||||
if (filter.connectionId) { conds.push("connectionId = ?"); params.push(filter.connectionId); }
|
||||
if (filter.status) { conds.push("status = ?"); params.push(filter.status); }
|
||||
if (filter.startDate) { conds.push("timestamp >= ?"); params.push(new Date(filter.startDate).toISOString()); }
|
||||
if (filter.endDate) { conds.push("timestamp <= ?"); params.push(new Date(filter.endDate).toISOString()); }
|
||||
|
||||
const where = conds.length ? `WHERE ${conds.join(" AND ")}` : "";
|
||||
const cntRow = db.get(`SELECT COUNT(*) as c FROM requestDetails ${where}`, params);
|
||||
const totalItems = cntRow ? cntRow.c : 0;
|
||||
|
||||
const page = filter.page || 1;
|
||||
const pageSize = filter.pageSize || 50;
|
||||
const totalPages = Math.ceil(totalItems / pageSize);
|
||||
const offset = (page - 1) * pageSize;
|
||||
|
||||
const rows = db.all(
|
||||
`SELECT data FROM requestDetails ${where} ORDER BY timestamp DESC LIMIT ? OFFSET ?`,
|
||||
[...params, pageSize, offset]
|
||||
);
|
||||
const details = rows.map((r) => parseJson(r.data, {}));
|
||||
|
||||
return {
|
||||
details,
|
||||
pagination: { page, pageSize, totalItems, totalPages, hasNext: page < totalPages, hasPrev: page > 1 },
|
||||
};
|
||||
}
|
||||
|
||||
export async function getRequestDetailById(id) {
|
||||
const db = await getAdapter();
|
||||
const row = db.get(`SELECT data FROM requestDetails WHERE id = ?`, [id]);
|
||||
return row ? parseJson(row.data, null) : null;
|
||||
}
|
||||
|
||||
const _shutdownHandler = async () => {
|
||||
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
|
||||
if (writeBuffer.length > 0) await flushToDatabase();
|
||||
};
|
||||
|
||||
function ensureShutdownHandler() {
|
||||
process.off("beforeExit", _shutdownHandler);
|
||||
process.off("SIGINT", _shutdownHandler);
|
||||
process.off("SIGTERM", _shutdownHandler);
|
||||
process.off("exit", _shutdownHandler);
|
||||
|
||||
process.on("beforeExit", _shutdownHandler);
|
||||
process.on("SIGINT", _shutdownHandler);
|
||||
process.on("SIGTERM", _shutdownHandler);
|
||||
process.on("exit", _shutdownHandler);
|
||||
}
|
||||
|
||||
ensureShutdownHandler();
|
||||
Reference in New Issue
Block a user