import { upsertClientRows, type ClientDbRow } from "./db"; import { metrics } from "./metrics"; function positiveNumberEnv(name: string, fallback: number): number { const value = Number(process.env[name] || fallback); return Number.isFinite(value) && value > 0 ? value : fallback; } export const CLIENT_DB_SYNC_INTERVAL_MS = positiveNumberEnv("OVERLORD_CLIENT_DB_SYNC_MS", 5000); export const CLIENT_PRESENCE_SYNC_INTERVAL_MS = Math.max( CLIENT_DB_SYNC_INTERVAL_MS, positiveNumberEnv("OVERLORD_CLIENT_PRESENCE_SYNC_MS", 60_000), ); export const CLIENT_DB_SYNC_BATCH_SIZE = Math.max( 1, Number(process.env.OVERLORD_CLIENT_DB_SYNC_BATCH_SIZE || 500), ); export const CLIENT_DB_SYNC_FLUSH_DELAY_MS = Math.max( 1, Number(process.env.OVERLORD_CLIENT_DB_SYNC_FLUSH_DELAY_MS || 100), ); const lastClientDbSync = new Map(); const pendingClientDbUpdates = new Map(); let flushSoonTimer: ReturnType | null = null; export function queueClientDbUpdate(partial: ClientDbRow): void { const existing = pendingClientDbUpdates.get(partial.id); if (!existing) { pendingClientDbUpdates.set(partial.id, { ...partial }); return; } pendingClientDbUpdates.set(partial.id, { ...existing, ...partial, id: partial.id, lastSeen: partial.lastSeen ?? existing.lastSeen, online: partial.online ?? existing.online, pingMs: partial.pingMs ?? existing.pingMs, }); } export function scheduleQueuedClientDbFlush(delayMs = CLIENT_DB_SYNC_FLUSH_DELAY_MS): void { if (pendingClientDbUpdates.size === 0 || flushSoonTimer) return; flushSoonTimer = setTimeout(() => { flushSoonTimer = null; flushQueuedClientDbUpdates(); }, Math.max(1, Math.floor(delayMs))); } export function flushQueuedClientDbUpdates(): void { if (pendingClientDbUpdates.size === 0) return; const startedAt = Date.now(); let processed = 0; const updates: ClientDbRow[] = []; for (const [clientId, update] of pendingClientDbUpdates.entries()) { updates.push(update); pendingClientDbUpdates.delete(clientId); processed += 1; if (processed >= CLIENT_DB_SYNC_BATCH_SIZE) { break; } } upsertClientRows(updates); if (pendingClientDbUpdates.size > 0) { scheduleQueuedClientDbFlush(25); } metrics.recordInternalTask("client-db-flush", Date.now() - startedAt); } setInterval(flushQueuedClientDbUpdates, CLIENT_DB_SYNC_INTERVAL_MS); export function shouldSyncClientToDb(clientId: string, now: number): boolean { const last = lastClientDbSync.get(clientId) || 0; if (now - last < CLIENT_PRESENCE_SYNC_INTERVAL_MS) return false; lastClientDbSync.set(clientId, now); return true; } export function markClientDbSynced(clientId: string, now: number): void { lastClientDbSync.set(clientId, now); } export function clearClientSyncState(clientId: string): void { lastClientDbSync.delete(clientId); pendingClientDbUpdates.delete(clientId); } export function getClientDbSyncStats(): { trackedClients: number; pendingUpdates: number; flushScheduled: boolean; } { return { trackedClients: lastClientDbSync.size, pendingUpdates: pendingClientDbUpdates.size, flushScheduled: flushSoonTimer !== null, }; }