// resolveChatPrefs.js — ADR-0064 WP-A. // // 4-layer hierarchy resolver for chat notification preferences. // Order (private overrides general): per-chat → personal → space → global → DEFAULTS. // // Storage: // 1. conversation_participants.notification_overrides (JSONB, migration 064) // 2. user_settings.setting_value_encrypted (setting_key='chat_notifications') // 3. spaces.notification_defaults (JSONB, migration 064) // 4. _app_settings.value (key='chat_notifications_global', seeded by migration 064) // // Merge semantics: declared-keys-win. A layer "declares" a key when it sets // a non-null/non-undefined value for it. Inner layers win over outer layers // ONLY for keys they declare; undeclared keys fall through to the next layer. // // Cache: in-process Map, 60s TTL, keyed `(user_id, conversation_id)`. // Eviction: PG NOTIFY channel `chat_prefs_invalidate` with payload // `{user_id?, space_id?, conversation_id?}` from each PUT endpoint. // - conversation_id set → evict entries with that conv (any user) // - user_id set → evict entries with that user (any conv) // - space_id set → clear cache (rare; admin-driven; no userId→space map) // - empty payload → clear cache (manual invalidate) import pg from 'pg'; import { dbGet, safeJsonParse } from '../../database/connection.js'; import { apiLogger } from '../../utils/logger.js'; const log = apiLogger.child({ module: 'chat_prefs_resolver' }); const CACHE_TTL_MS = 60_000; const NOTIFY_CHANNEL = 'chat_prefs_invalidate'; const CACHE_SIZE_LIMIT = 10_000; export const CHAT_PREFS_DEFAULTS = Object.freeze({ enabled: true, sound_enabled: true, sound_volume: 0.6, humans: { sound: true, popup: true, badge: true }, agents: { sound: true, popup: true, badge: true }, }); const TOP_SCALARS = ['enabled', 'sound_enabled', 'sound_volume']; const NESTED_BLOCKS = ['humans', 'agents']; const NESTED_KEYS = ['sound', 'popup', 'badge']; // ─── Pure merge (testable without DB) ─────────────────────────────────────── /** * Walk layers in priority order; first layer that DECLARES a key wins. * A declaration is a non-undefined, non-null value at the right path. * * @param {...Object|null} layers Layers from MOST-private to LEAST-private. * Last entry should be CHAT_PREFS_DEFAULTS to guarantee a value. * @returns {Object} Fully-populated prefs object — no nulls, no undefineds. */ export function mergeDeclared(...layers) { const out = {}; // Top-level scalars for (const k of TOP_SCALARS) { for (const layer of layers) { if (layer && layer[k] !== undefined && layer[k] !== null) { out[k] = layer[k]; break; } } } // Nested per-sender-type blocks (humans, agents) — merge each key independently // so a layer that declares only `humans.popup` doesn't shadow another layer's // `humans.sound`. for (const block of NESTED_BLOCKS) { out[block] = {}; for (const k of NESTED_KEYS) { for (const layer of layers) { const v = layer && layer[block] ? layer[block][k] : undefined; if (v !== undefined && v !== null) { out[block][k] = v; break; } } } } return out; } // ─── DB fetchers (override-able for tests) ───────────────────────────────── async function _fetchPerChat(userId, conversationId) { if (!userId || !conversationId) return null; const row = await dbGet( `SELECT notification_overrides FROM conversation_participants WHERE user_id = $1 AND conversation_id = $2`, [userId, conversationId] ); return safeJsonParse(row?.notification_overrides, null); } async function _fetchPersonal(userId) { if (!userId) return null; const row = await dbGet( `SELECT setting_value_encrypted AS value FROM user_settings WHERE user_id = $1 AND setting_key = 'chat_notifications'`, [userId] ); return safeJsonParse(row?.value, null); } async function _fetchSpace(conversationId) { if (!conversationId) return null; // `conversations.space_id` references `spaces.id` directly (schema verified // 2026-05-16 — see migration 010 + ADR-0024). DMs have NULL space_id. const row = await dbGet( `SELECT s.notification_defaults AS prefs FROM conversations c JOIN spaces s ON s.id = c.space_id WHERE c.id = $1`, [conversationId] ); return safeJsonParse(row?.prefs, null); } async function _fetchGlobal() { const row = await dbGet( `SELECT value FROM _app_settings WHERE key = 'chat_notifications_global'` ); return safeJsonParse(row?.value, null); } // Injection points for unit tests — keep module-default fetchers swappable. let fetchPerChat = _fetchPerChat; let fetchPersonal = _fetchPersonal; let fetchSpace = _fetchSpace; let fetchGlobal = _fetchGlobal; /** Test-only helper. */ export function __setFetchers(overrides) { if (overrides.perChat) fetchPerChat = overrides.perChat; if (overrides.personal) fetchPersonal = overrides.personal; if (overrides.space) fetchSpace = overrides.space; if (overrides.global) fetchGlobal = overrides.global; } /** Test-only helper. */ export function __resetFetchers() { fetchPerChat = _fetchPerChat; fetchPersonal = _fetchPersonal; fetchSpace = _fetchSpace; fetchGlobal = _fetchGlobal; } // ─── Cache ────────────────────────────────────────────────────────────────── /** @type {Map} */ const _cache = new Map(); function cacheKey(userId, conversationId) { return `${userId}:${conversationId}`; } function cacheGet(userId, conversationId) { const k = cacheKey(userId, conversationId); const hit = _cache.get(k); if (!hit) return null; if (hit.expiresAt <= Date.now()) { _cache.delete(k); return null; } return hit.value; } function cacheSet(userId, conversationId, value) { if (_cache.size >= CACHE_SIZE_LIMIT) { // Cheap eviction: drop the oldest entry (Map preserves insertion order). const firstKey = _cache.keys().next().value; if (firstKey !== undefined) _cache.delete(firstKey); } _cache.set(cacheKey(userId, conversationId), { value, expiresAt: Date.now() + CACHE_TTL_MS, }); } /** * Evict cache entries matching a scope. * @param {object} scope { user_id?, space_id?, conversation_id? } * Also accepts `{ scope: 'global' }` (semantic alias for * a full clear, used by the admin/global PUT route). * Empty object → clear entire cache. */ export function invalidateCache(scope = {}) { if (!scope || Object.keys(scope).length === 0) { _cache.clear(); return; } // Semantic alias / unknown scope: any caller saying "global" → wipe all. if (scope.scope === 'global') { _cache.clear(); return; } // For space_id we have no userId→space map in-process — clear all. if (scope.space_id) { _cache.clear(); return; } const matchUser = scope.user_id != null ? String(scope.user_id) : null; const matchConv = scope.conversation_id != null ? String(scope.conversation_id) : null; // If the caller passed only unrecognised keys, fail safe by clearing. if (!matchUser && !matchConv) { _cache.clear(); return; } for (const k of _cache.keys()) { const [uid, cid] = k.split(':'); if (matchUser && uid === matchUser) { _cache.delete(k); continue; } if (matchConv && cid === matchConv) { _cache.delete(k); continue; } } } /** Test-only helper. */ export function __cacheSize() { return _cache.size; } /** Test-only helper. */ export function __cacheClear() { _cache.clear(); } // ─── Public API ───────────────────────────────────────────────────────────── /** * Resolve effective chat notification preferences for a (user, conversation). * * @param {object} args * @param {number} args.userId * @param {number} args.conversationId * @returns {Promise} Flattened prefs object — guaranteed non-null fields. */ export async function resolveChatPrefs({ userId, conversationId }) { const cached = cacheGet(userId, conversationId); if (cached) return cached; const [perChat, personal, spacePrefs, globalPrefs] = await Promise.all([ fetchPerChat(userId, conversationId), fetchPersonal(userId), fetchSpace(conversationId), fetchGlobal(), ]); const resolved = mergeDeclared(perChat, personal, spacePrefs, globalPrefs, CHAT_PREFS_DEFAULTS); cacheSet(userId, conversationId, resolved); return resolved; } /** * Fire pg_notify on the cache-invalidate channel. Wraps the dbRun shape * so callers can `await notifyInvalidate({ user_id: 5 })` from any PUT. * * @param {object} scope { user_id?, space_id?, conversation_id? } */ export async function notifyInvalidate(scope = {}) { try { const { dbRun } = await import('../../database/connection.js'); await dbRun(`SELECT pg_notify($1, $2)`, [NOTIFY_CHANNEL, JSON.stringify(scope || {})]); } catch (err) { log.warn({ err, scope }, 'notifyInvalidate failed — local-only eviction'); // Still evict locally so the originating process sees fresh data. invalidateCache(scope); } } // ─── LISTEN client (cluster-wide eviction) ───────────────────────────────── /** @type {pg.Client|null} */ let _listenClient = null; function buildConnectionConfig() { if (process.env.POSTGRES_URL) return { connectionString: process.env.POSTGRES_URL }; return { host: process.env.POSTGRES_HOST || 'localhost', port: parseInt(process.env.POSTGRES_PORT || '5432', 10), database: process.env.POSTGRES_DB || 'godcrm', user: process.env.POSTGRES_USER || 'godcrm', password: process.env.POSTGRES_PASSWORD, ssl: process.env.POSTGRES_SSL === 'true' ? { rejectUnauthorized: false } : false, }; } /** * Start a dedicated long-lived LISTEN client. Best-effort: if it fails, * eviction degrades to local-only + 60s TTL. */ export async function startInvalidationListener() { if (_listenClient) return _listenClient; const client = new pg.Client(buildConnectionConfig()); await client.connect(); client.on('notification', (msg) => { if (msg.channel !== NOTIFY_CHANNEL) return; let scope = {}; try { scope = msg.payload ? JSON.parse(msg.payload) : {}; } catch { scope = {}; } invalidateCache(scope); log.debug({ scope, cacheSize: _cache.size }, 'chat_prefs cache invalidated via NOTIFY'); }); client.on('error', (err) => { log.error({ err }, 'chat_prefs LISTEN client error — eviction degraded to TTL-only'); }); await client.query(`LISTEN ${NOTIFY_CHANNEL}`); _listenClient = client; log.info({ channel: NOTIFY_CHANNEL }, 'chat_prefs invalidation listener started'); return client; } /** Stop the LISTEN client; safe to call multiple times. */ export async function stopInvalidationListener() { if (!_listenClient) return; try { await _listenClient.end(); } catch { /* ignore */ } _listenClient = null; } export default { CHAT_PREFS_DEFAULTS, mergeDeclared, resolveChatPrefs, notifyInvalidate, invalidateCache, startInvalidationListener, stopInvalidationListener, };