import fs from 'fs/promises'; import path from 'path'; import type { Logger } from '../../utils/logger.js'; import type { ActiveChat, ChatCacheFile, InvitationCooldown } from './types.js'; import type { AccountContentRepository } from '../../database/account-repositories.js'; export interface SyncMarkers { nextBatch: string | null; lastFullSyncAt: number | null; lastDeltaSyncAt: number | null; } /** * In-memory + on-disk cache of Reddit chats plus the Matrix delta cursor * so the bot can resume incremental sync across restarts. */ export class ChatCache { private chats: ActiveChat[] = []; private lastFetchTime = 0; readonly cacheDurationMs = 60 * 1000; private readonly filePath: string; private nextBatch: string | null = null; private lastFullSyncAt: number | null = null; private lastDeltaSyncAt: number | null = null; private cooldown: InvitationCooldown | null = null; constructor( private logger: Logger, filePath?: string, private readonly databaseStore?: AccountContentRepository, private accountId: string | null = null, ) { if (!filePath && !databaseStore && process.env.NODE_ENV === 'test') { // Without this guard a test that constructs a ChatCache (or a // ChatService that builds one) would silently write to the prod // ./data/chat-cache.json — concurrent test runs have already torn // that file once. Force every test to opt into a tmp path. throw new Error( 'ChatCache: refusing to fall back to ./data/chat-cache.json under NODE_ENV=test — pass an explicit filePath', ); } this.filePath = filePath ?? path.join(process.cwd(), 'data', 'chat-cache.json'); } async switchAccount(accountId: string): Promise { if (this.accountId === accountId) return; this.accountId = accountId; this.chats = []; this.lastFetchTime = 0; this.nextBatch = null; this.lastFullSyncAt = null; this.lastDeltaSyncAt = null; this.cooldown = null; await this.load(); } get(): ActiveChat[] { return this.chats; } set(chats: ActiveChat[]): void { this.chats = chats; this.lastFetchTime = Date.now(); } isFresh(): boolean { return Date.now() - this.lastFetchTime < this.cacheDurationMs && this.chats.length > 0; } getMarkers(): SyncMarkers { return { nextBatch: this.nextBatch, lastFullSyncAt: this.lastFullSyncAt, lastDeltaSyncAt: this.lastDeltaSyncAt, }; } setNextBatch(token: string | null): void { this.nextBatch = token; } markFullSync(at: number = Date.now()): void { this.lastFullSyncAt = at; this.lastDeltaSyncAt = at; } markDeltaSync(at: number = Date.now()): void { this.lastDeltaSyncAt = at; } getCooldown(): InvitationCooldown | null { return this.cooldown; } setCooldown(cooldown: InvitationCooldown | null): void { this.cooldown = cooldown; } async load(): Promise { if (this.databaseStore && this.accountId) { const rows = this.databaseStore.listConversations(this.accountId); this.chats = rows.flatMap((row) => { try { return [JSON.parse(row.payloadJson) as ActiveChat]; } catch { return []; } }); const state = this.databaseStore.getSyncState(this.accountId); this.nextBatch = state?.nextBatch ?? null; this.lastFullSyncAt = state?.lastFullSyncAt ?? null; this.lastDeltaSyncAt = state?.lastDeltaSyncAt ?? null; try { this.cooldown = state?.cooldownJson ? JSON.parse(state.cooldownJson) as InvitationCooldown : null; } catch { this.cooldown = null; } this.lastFetchTime = Date.now(); this.logger.info(`Loaded ${this.chats.length} account-owned chats from the database`); return; } try { const raw = await fs.readFile(this.filePath, 'utf-8'); const parsed = JSON.parse(raw) as ChatCacheFile; if (!parsed || !Array.isArray(parsed.chats)) { this.logger.warn('Chat cache file is invalid, ignoring.'); return; } this.chats = parsed.chats; this.lastFetchTime = Date.now(); this.nextBatch = parsed.nextBatch ?? null; this.lastFullSyncAt = parsed.lastFullSyncAt ?? null; this.lastDeltaSyncAt = parsed.lastDeltaSyncAt ?? null; this.cooldown = parsed.cooldown ?? null; this.logger.info( `Loaded ${parsed.chats.length} chats from cache` + (this.lastFullSyncAt ? ` (last full sync ${new Date(this.lastFullSyncAt).toISOString()})` : ' (no full sync recorded)'), ); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') { this.logger.info('No chat cache found to load'); return; } this.logger.warn('Failed to load chat cache:', error); } } /** * Persist only the cooldown field, preserving whatever chats / sync markers * are already on disk. Used when a rate-limit fires before the in-memory * cache has been populated — a plain `save()` would clobber a prior full * sync with `chats: []` and null markers. */ async persistCooldown(): Promise { if (this.databaseStore && this.accountId) { this.persistDatabaseState(); return; } let existing: ChatCacheFile | null = null; try { const raw = await fs.readFile(this.filePath, 'utf-8'); existing = JSON.parse(raw) as ChatCacheFile; } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { this.logger.warn('Failed to read chat cache for cooldown update:', error); } } const payload: ChatCacheFile = existing ? { ...existing, cooldown: this.cooldown, savedAt: Date.now() } : { version: 1, savedAt: Date.now(), chats: [], nextBatch: null, lastFullSyncAt: null, lastDeltaSyncAt: null, cooldown: this.cooldown, }; try { await this.atomicWrite(payload); } catch (error) { this.logger.warn('Failed to save chat cache cooldown:', error); } } async save(chats: ActiveChat[] = this.chats): Promise { if (this.databaseStore && this.accountId) { const retained = new Set(); for (const chat of chats) { const conversationId = chat.roomId ?? chat.username; retained.add(conversationId); this.databaseStore.upsertConversation({ accountId: this.accountId, conversationId, username: chat.username, roomName: chat.roomName ?? null, payloadJson: JSON.stringify(chat), }); } for (const existing of this.databaseStore.listConversations(this.accountId)) { if (!retained.has(existing.conversationId)) { this.databaseStore.deleteConversation(this.accountId, existing.conversationId); } } this.persistDatabaseState(); return; } const payload: ChatCacheFile = { version: 1, savedAt: Date.now(), chats, nextBatch: this.nextBatch, lastFullSyncAt: this.lastFullSyncAt, lastDeltaSyncAt: this.lastDeltaSyncAt, cooldown: this.cooldown, }; try { await this.atomicWrite(payload); } catch (error) { this.logger.warn('Failed to save chat cache:', error); } } private persistDatabaseState(): void { if (!this.databaseStore || !this.accountId) return; this.databaseStore.saveSyncState({ accountId: this.accountId, nextBatch: this.nextBatch, lastFullSyncAt: this.lastFullSyncAt, lastDeltaSyncAt: this.lastDeltaSyncAt, cooldownJson: this.cooldown ? JSON.stringify(this.cooldown) : null, }); } /** * Write the cache file via a temp+rename so a crashed/interrupted/concurrent * writer can never leave the on-disk file half-written. rename(2) is atomic * on the same filesystem, so any reader sees either the prior complete file * or the new complete file — never NULs or truncated JSON. * * Serialises against itself so two concurrent saves don't fight over the * same temp path or trample each other's rename. */ private writeQueue: Promise = Promise.resolve(); private async atomicWrite(payload: ChatCacheFile): Promise { const task = this.writeQueue.then(async () => { await fs.mkdir(path.dirname(this.filePath), { recursive: true }); const tmp = `${this.filePath}.${process.pid}.${Date.now()}.${Math.random().toString(36).slice(2, 8)}.tmp`; await fs.writeFile(tmp, JSON.stringify(payload, null, 2), 'utf-8'); await fs.rename(tmp, this.filePath); }); this.writeQueue = task.catch(() => undefined); return task; } }