mirror of
https://github.com/dtzp555-max/ocp.git
synced 2026-07-19 09:44:07 +00:00
cli.js does not perform proxy-layer stampede protection. The singleflight layer is a value-add proxy operation that exists only inside OCP, between concurrent client requests and the single upstream cli.js spawn. It does not introduce, alter, or remove any endpoint, header, request field, or response field that cli.js emits or expects — no client-observable wire shape change. Justification under ALIGNMENT.md Rule 2: the singleflight Map deduplicates concurrent identical non-streaming cache-miss requests so only one cli.js spawn runs per unique hash window. All followers receive the same resolved (or rejected) content. This is a proxy-internal concurrency optimization, not a wire-level protocol change. Spec: docs/superpowers/specs/2026-05-07-cache-upgrade-design.md (D4) Changes: - keys.mjs: add singleflight(hash, fn) + getInflightStats() exports; in-memory Map cleared via Promise.finally() on each settlement - server.mjs: import singleflight + getInflightStats; wrap non-streaming cache-enabled path through singleflight with inner recheck; add TODO comment in callClaudeStreaming (streaming-path dedup is explicitly out of scope for v3.13.0, see spec D4 streaming caveat); extend /cache/stats to return inflight + requesters fields (additive, no removed fields) - test-features.mjs: 7 new PR-B singleflight tests covering basic dedup, failure fan-out, map cleanup (success + failure), different-hash independence, getInflightStats shape, and sequential-call non-sharing; all 31 tests pass (24 existing + 7 new) Streaming-path singleflight is explicitly out of scope; TODO left in callClaudeStreaming for a future follow-up ticket. Co-authored-by: dtzp555 <dtzp555@gmail.com> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
413 lines
15 KiB
JavaScript
413 lines
15 KiB
JavaScript
// keys.mjs — API key management and usage tracking for OCP LAN mode
|
|
// Uses Node.js built-in SQLite (node:sqlite) — zero external dependencies.
|
|
import { DatabaseSync } from "node:sqlite";
|
|
import { randomBytes, createHash } from "node:crypto";
|
|
import { join } from "node:path";
|
|
import { mkdirSync } from "node:fs";
|
|
import { homedir } from "node:os";
|
|
|
|
const OCP_DIR = join(homedir(), ".ocp");
|
|
mkdirSync(OCP_DIR, { recursive: true });
|
|
const DB_PATH = join(OCP_DIR, "ocp.db");
|
|
|
|
let db;
|
|
|
|
export function getDb() {
|
|
if (!db) {
|
|
db = new DatabaseSync(DB_PATH);
|
|
db.exec("PRAGMA journal_mode = WAL");
|
|
db.exec("PRAGMA foreign_keys = ON");
|
|
initSchema();
|
|
}
|
|
return db;
|
|
}
|
|
|
|
function initSchema() {
|
|
db.exec(`
|
|
CREATE TABLE IF NOT EXISTS api_keys (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
key TEXT UNIQUE NOT NULL,
|
|
name TEXT NOT NULL,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
revoked INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS usage_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
key_id INTEGER,
|
|
key_name TEXT NOT NULL DEFAULT 'anonymous',
|
|
model TEXT NOT NULL,
|
|
prompt_chars INTEGER NOT NULL DEFAULT 0,
|
|
response_chars INTEGER NOT NULL DEFAULT 0,
|
|
elapsed_ms INTEGER NOT NULL DEFAULT 0,
|
|
success INTEGER NOT NULL DEFAULT 1,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
FOREIGN KEY (key_id) REFERENCES api_keys(id)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_usage_created ON usage_log(created_at);
|
|
CREATE INDEX IF NOT EXISTS idx_usage_key ON usage_log(key_id);
|
|
|
|
CREATE TABLE IF NOT EXISTS response_cache (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
hash TEXT UNIQUE NOT NULL,
|
|
model TEXT NOT NULL,
|
|
response TEXT NOT NULL,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
last_hit_at TEXT,
|
|
hits INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_cache_hash ON response_cache(hash);
|
|
CREATE INDEX IF NOT EXISTS idx_cache_created ON response_cache(created_at);
|
|
`);
|
|
|
|
// Idempotent migrations: add quota columns if they don't exist yet.
|
|
for (const col of [
|
|
"ALTER TABLE api_keys ADD COLUMN quota_daily INTEGER DEFAULT NULL",
|
|
"ALTER TABLE api_keys ADD COLUMN quota_weekly INTEGER DEFAULT NULL",
|
|
"ALTER TABLE api_keys ADD COLUMN quota_monthly INTEGER DEFAULT NULL",
|
|
]) {
|
|
try { db.exec(col); } catch (e) {
|
|
// SQLite throws "duplicate column name" if already present — safe to ignore.
|
|
if (!e.message?.includes("duplicate column")) throw e;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Key CRUD ──
|
|
|
|
export function createKey(name) {
|
|
const key = "ocp_" + randomBytes(24).toString("base64url");
|
|
const d = getDb();
|
|
const stmt = d.prepare("INSERT INTO api_keys (key, name) VALUES (?, ?)");
|
|
const result = stmt.run(key, name);
|
|
return { id: result.lastInsertRowid, key, name };
|
|
}
|
|
|
|
export function listKeys() {
|
|
const d = getDb();
|
|
return d.prepare(
|
|
"SELECT id, key, name, created_at, revoked, quota_daily, quota_weekly, quota_monthly FROM api_keys ORDER BY created_at DESC"
|
|
).all().map(({ key, ...rest }) => ({
|
|
...rest,
|
|
keyPreview: key.slice(0, 8) + "..." + key.slice(-4),
|
|
}));
|
|
}
|
|
|
|
export function revokeKey(idOrName) {
|
|
const d = getDb();
|
|
const stmt = d.prepare(
|
|
"UPDATE api_keys SET revoked = 1 WHERE (id = ? OR name = ?) AND revoked = 0"
|
|
);
|
|
return stmt.run(idOrName, idOrName).changes > 0;
|
|
}
|
|
|
|
export function validateKey(key) {
|
|
const d = getDb();
|
|
const row = d.prepare(
|
|
"SELECT id, name FROM api_keys WHERE key = ? AND revoked = 0"
|
|
).get(key);
|
|
return row || null;
|
|
}
|
|
|
|
// ── Usage recording ──
|
|
|
|
export function recordUsage({ keyId, keyName, model, promptChars, responseChars, elapsedMs, success }) {
|
|
const d = getDb();
|
|
d.prepare(`
|
|
INSERT INTO usage_log (key_id, key_name, model, prompt_chars, response_chars, elapsed_ms, success)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
`).run(keyId ?? null, keyName || "anonymous", model, promptChars, responseChars, elapsedMs, success ? 1 : 0);
|
|
}
|
|
|
|
// ── Usage queries ──
|
|
|
|
export function getUsageByKey({ since, until } = {}) {
|
|
const d = getDb();
|
|
let where = "WHERE 1=1";
|
|
const params = [];
|
|
if (since) { where += " AND created_at >= ?"; params.push(since); }
|
|
if (until) { where += " AND created_at <= ?"; params.push(until); }
|
|
|
|
return d.prepare(`
|
|
SELECT
|
|
key_name,
|
|
COUNT(*) as requests,
|
|
SUM(CASE WHEN success = 1 THEN 1 ELSE 0 END) as successes,
|
|
SUM(CASE WHEN success = 0 THEN 1 ELSE 0 END) as errors,
|
|
SUM(prompt_chars) as total_prompt_chars,
|
|
SUM(response_chars) as total_response_chars,
|
|
SUM(elapsed_ms) as total_elapsed_ms,
|
|
AVG(elapsed_ms) as avg_elapsed_ms,
|
|
MIN(created_at) as first_request,
|
|
MAX(created_at) as last_request
|
|
FROM usage_log
|
|
${where}
|
|
GROUP BY key_name
|
|
ORDER BY requests DESC
|
|
`).all(...params);
|
|
}
|
|
|
|
export function getUsageTimeline({ keyName, hours = 24 } = {}) {
|
|
const d = getDb();
|
|
const since = new Date(Date.now() - hours * 3600000).toISOString();
|
|
let where = "WHERE created_at >= ?";
|
|
const params = [since];
|
|
if (keyName) { where += " AND key_name = ?"; params.push(keyName); }
|
|
|
|
return d.prepare(`
|
|
SELECT
|
|
strftime('%Y-%m-%dT%H:00:00', created_at) as hour,
|
|
COUNT(*) as requests,
|
|
SUM(prompt_chars) as prompt_chars,
|
|
SUM(response_chars) as response_chars,
|
|
AVG(elapsed_ms) as avg_elapsed_ms
|
|
FROM usage_log
|
|
${where}
|
|
GROUP BY hour
|
|
ORDER BY hour
|
|
`).all(...params);
|
|
}
|
|
|
|
export function getRecentUsage(limit = 50) {
|
|
const d = getDb();
|
|
return d.prepare(`
|
|
SELECT key_name, model, prompt_chars, response_chars, elapsed_ms, success, created_at
|
|
FROM usage_log
|
|
ORDER BY created_at DESC
|
|
LIMIT ?
|
|
`).all(limit);
|
|
}
|
|
|
|
// ── SQLite datetime helper ──
|
|
// SQLite datetime('now') stores as 'YYYY-MM-DD HH:MM:SS' (no T, no Z).
|
|
// JavaScript .toISOString() produces 'YYYY-MM-DDTHH:MM:SS.sssZ'.
|
|
// String comparison between the two breaks for same-day ranges (T > space).
|
|
// This helper formats Date to match SQLite's format for correct comparisons.
|
|
function sqliteDatetime(date) {
|
|
return date.toISOString().replace("T", " ").replace(/\.\d{3}Z$/, "");
|
|
}
|
|
|
|
// ── Quota management ──
|
|
|
|
// Returns { period, limit, used, resetsIn } if a quota is exceeded, null otherwise.
|
|
// Anonymous/admin callers (keyId === null) are never subject to quotas.
|
|
export function checkQuota(keyId, _keyName) {
|
|
if (keyId === null || keyId === undefined) return null;
|
|
|
|
const d = getDb();
|
|
const keyRow = d.prepare(
|
|
"SELECT quota_daily, quota_weekly, quota_monthly FROM api_keys WHERE id = ? AND revoked = 0"
|
|
).get(keyId);
|
|
if (!keyRow) return null;
|
|
|
|
const now = new Date();
|
|
|
|
// UTC period boundaries (SQLite-compatible format)
|
|
const startOfToday = sqliteDatetime(new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate())));
|
|
const sevenDaysAgo = sqliteDatetime(new Date(Date.now() - 7 * 86400000));
|
|
const thirtyDaysAgo = sqliteDatetime(new Date(Date.now() - 30 * 86400000));
|
|
|
|
// Next reset times for human display
|
|
const tomorrowUTC = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() + 1));
|
|
function msToHuman(ms) {
|
|
if (ms <= 0) return "now";
|
|
const h = Math.floor(ms / 3600000);
|
|
const m = Math.floor((ms % 3600000) / 60000);
|
|
if (h >= 24) { const d = Math.floor(h / 24); return `${d}d ${h % 24}h`; }
|
|
return h > 0 ? `${h}h ${m}m` : `${m}m`;
|
|
}
|
|
|
|
// Single query for all periods (widest window = monthly)
|
|
const row = d.prepare(`
|
|
SELECT
|
|
SUM(CASE WHEN created_at >= ? THEN 1 ELSE 0 END) as daily_cnt,
|
|
SUM(CASE WHEN created_at >= ? THEN 1 ELSE 0 END) as weekly_cnt,
|
|
COUNT(*) as monthly_cnt
|
|
FROM usage_log
|
|
WHERE key_id = ? AND success = 1 AND created_at >= ?
|
|
`).get(startOfToday, sevenDaysAgo, keyId, thirtyDaysAgo);
|
|
|
|
const checks = [
|
|
{ period: "daily", limit: keyRow.quota_daily, used: row?.daily_cnt ?? 0, resetsIn: msToHuman(tomorrowUTC - now) },
|
|
{ period: "weekly", limit: keyRow.quota_weekly, used: row?.weekly_cnt ?? 0, resetsIn: "rolling 7-day window" },
|
|
{ period: "monthly", limit: keyRow.quota_monthly, used: row?.monthly_cnt ?? 0, resetsIn: "rolling 30-day window" },
|
|
];
|
|
|
|
for (const { period, limit, used, resetsIn } of checks) {
|
|
if (limit === null || limit === undefined) continue;
|
|
if (used >= limit) {
|
|
return { period, limit, used, resetsIn };
|
|
}
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
// Set quota for a key. Only updates fields explicitly present in the input object.
|
|
// Pass null to clear a specific limit. Omit a field to leave it unchanged.
|
|
export function updateKeyQuota(idOrName, updates = {}) {
|
|
const d = getDb();
|
|
const setClauses = [];
|
|
const params = [];
|
|
if ("daily" in updates) { setClauses.push("quota_daily = ?"); params.push(updates.daily ?? null); }
|
|
if ("weekly" in updates) { setClauses.push("quota_weekly = ?"); params.push(updates.weekly ?? null); }
|
|
if ("monthly" in updates){ setClauses.push("quota_monthly = ?");params.push(updates.monthly ?? null); }
|
|
if (setClauses.length === 0) return false;
|
|
params.push(idOrName, idOrName);
|
|
const result = d.prepare(
|
|
`UPDATE api_keys SET ${setClauses.join(", ")} WHERE id = ? OR name = ?`
|
|
).run(...params);
|
|
return result.changes > 0;
|
|
}
|
|
|
|
// Returns { daily: { limit, used }, weekly: { limit, used }, monthly: { limit, used } }
|
|
export function getKeyQuota(keyId) {
|
|
const d = getDb();
|
|
const keyRow = d.prepare(
|
|
"SELECT quota_daily, quota_weekly, quota_monthly FROM api_keys WHERE id = ?"
|
|
).get(keyId);
|
|
if (!keyRow) return null;
|
|
|
|
const now = new Date();
|
|
const startOfToday = sqliteDatetime(new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate())));
|
|
const sevenDaysAgo = sqliteDatetime(new Date(Date.now() - 7 * 86400000));
|
|
const thirtyDaysAgo = sqliteDatetime(new Date(Date.now() - 30 * 86400000));
|
|
|
|
const row = d.prepare(`
|
|
SELECT
|
|
SUM(CASE WHEN created_at >= ? THEN 1 ELSE 0 END) as daily_cnt,
|
|
SUM(CASE WHEN created_at >= ? THEN 1 ELSE 0 END) as weekly_cnt,
|
|
COUNT(*) as monthly_cnt
|
|
FROM usage_log
|
|
WHERE key_id = ? AND success = 1 AND created_at >= ?
|
|
`).get(startOfToday, sevenDaysAgo, keyId, thirtyDaysAgo);
|
|
|
|
return {
|
|
daily: { limit: keyRow.quota_daily ?? null, used: row?.daily_cnt ?? 0 },
|
|
weekly: { limit: keyRow.quota_weekly ?? null, used: row?.weekly_cnt ?? 0 },
|
|
monthly: { limit: keyRow.quota_monthly ?? null, used: row?.monthly_cnt ?? 0 },
|
|
};
|
|
}
|
|
|
|
// ── Response cache ──
|
|
|
|
// Generate a cache key from model + messages + request params that affect output.
|
|
// opts.keyId isolates per-API-key cache pools (v2 hash format).
|
|
// When keyId is absent/null/empty, falls back to "anon" (shared anonymous pool).
|
|
export function cacheHash(model, messages, opts = {}) {
|
|
const keyId = opts.keyId || "anon";
|
|
const h = createHash("sha256");
|
|
h.update(`v2|k:${keyId}|`);
|
|
h.update(model);
|
|
if (opts.temperature != null) h.update(`t:${opts.temperature}`);
|
|
if (opts.max_tokens != null) h.update(`mt:${opts.max_tokens}`);
|
|
if (opts.top_p != null) h.update(`tp:${opts.top_p}`);
|
|
for (const m of messages) {
|
|
h.update(m.role || "");
|
|
h.update(typeof m.content === "string" ? m.content : JSON.stringify(m.content));
|
|
}
|
|
return h.digest("hex");
|
|
}
|
|
|
|
// Check whether any message (or content part) carries an Anthropic cache_control field.
|
|
// If true, OCP should skip its own cache to avoid interfering with prompt-caching intent.
|
|
export function hasCacheControl(messages) {
|
|
for (const m of messages || []) {
|
|
if (m && typeof m === "object") {
|
|
if (m.cache_control) return true;
|
|
if (Array.isArray(m.content)) {
|
|
for (const part of m.content) {
|
|
if (part && typeof part === "object" && part.cache_control) return true;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
// Look up a cached response. Returns { response, hits } or null.
|
|
// Also updates last_hit_at and increments hits counter on hit.
|
|
export function getCachedResponse(hash, ttlMs) {
|
|
const d = getDb();
|
|
const cutoff = sqliteDatetime(new Date(Date.now() - ttlMs));
|
|
const row = d.prepare(
|
|
"SELECT id, response, hits FROM response_cache WHERE hash = ? AND created_at >= ?"
|
|
).get(hash, cutoff);
|
|
if (!row) return null;
|
|
// Update hit stats
|
|
d.prepare("UPDATE response_cache SET hits = hits + 1, last_hit_at = datetime('now') WHERE id = ?").run(row.id);
|
|
return { response: row.response, hits: row.hits + 1 };
|
|
}
|
|
|
|
// Store a response in the cache
|
|
export function setCachedResponse(hash, model, response) {
|
|
const d = getDb();
|
|
// Upsert: if hash already exists (race condition), just update
|
|
d.prepare(`
|
|
INSERT INTO response_cache (hash, model, response) VALUES (?, ?, ?)
|
|
ON CONFLICT(hash) DO UPDATE SET response = excluded.response, created_at = datetime('now'), hits = 0
|
|
`).run(hash, model, response);
|
|
}
|
|
|
|
// Clear all cached responses, or expired ones only
|
|
export function clearCache(ttlMs = null) {
|
|
const d = getDb();
|
|
if (ttlMs === null) {
|
|
const result = d.prepare("DELETE FROM response_cache").run();
|
|
return result.changes;
|
|
}
|
|
const cutoff = sqliteDatetime(new Date(Date.now() - ttlMs));
|
|
const result = d.prepare("DELETE FROM response_cache WHERE created_at < ?").run(cutoff);
|
|
return result.changes;
|
|
}
|
|
|
|
// Get cache statistics
|
|
export function getCacheStats() {
|
|
const d = getDb();
|
|
const total = d.prepare("SELECT COUNT(*) as cnt FROM response_cache").get()?.cnt ?? 0;
|
|
const totalHits = d.prepare("SELECT SUM(hits) as total FROM response_cache").get()?.total ?? 0;
|
|
const sizeBytes = d.prepare("SELECT SUM(LENGTH(response)) as size FROM response_cache").get()?.size ?? 0;
|
|
return { entries: total, totalHits, sizeBytes };
|
|
}
|
|
|
|
// ── Singleflight stampede protection ──
|
|
|
|
// In-memory singleflight Map: hash → { promise, requesters }
|
|
// Deduplicates concurrent identical cache-miss flows so only one upstream call runs.
|
|
// Per ADR 0005 / spec D4: in-process scope only (single Node process per host).
|
|
const inflightMap = new Map();
|
|
|
|
export function singleflight(hash, fn) {
|
|
const existing = inflightMap.get(hash);
|
|
if (existing) {
|
|
existing.requesters++;
|
|
return existing.promise;
|
|
}
|
|
// Wrap fn() in Promise.resolve().then() so synchronous throws don't escape.
|
|
const promise = Promise.resolve().then(fn).finally(() => {
|
|
inflightMap.delete(hash);
|
|
});
|
|
inflightMap.set(hash, { promise, requesters: 1 });
|
|
return promise;
|
|
}
|
|
|
|
export function getInflightStats() {
|
|
let totalRequesters = 0;
|
|
for (const entry of inflightMap.values()) totalRequesters += entry.requesters;
|
|
return {
|
|
inflight: inflightMap.size,
|
|
requesters: totalRequesters,
|
|
};
|
|
}
|
|
|
|
// Find a key by id or name (returns { id, name } or null)
|
|
export function findKey(idOrName) {
|
|
const d = getDb();
|
|
return d.prepare("SELECT id, name FROM api_keys WHERE id = ? OR name = ?").get(idOrName, idOrName) || null;
|
|
}
|
|
|
|
export function closeDb() {
|
|
if (db) { db.close(); db = null; }
|
|
}
|