import { isServerUserStateNamespace, SERVER_USER_STATE_NAMESPACES, type ServerUserStateNamespace, type ServerUserStateResponse, } from '@iptvnator/shared/interfaces'; import type { UserDataStorage } from '../user-data-storage'; import type { ServerApiClient } from './server-api-client'; /** Server-mode cache keys; the legacy unprefixed keys are never touched. */ export const SERVER_CACHE_PREFIX = 'iptvnator-server:'; const FLUSH_DELAY_MS = 500; interface PendingEntry { readonly namespace: ServerUserStateNamespace; readonly playlistId: string; /** `undefined` deletes the entry. */ readonly value: unknown; /** The edit this value came from; a retry never replaces a newer one. */ readonly edit: number; } type ServerApi = Pick< ServerApiClient, 'getJson' | 'sendJson' | 'request' | 'sendKeepalive' >; /** * Write-through `USER_DATA_STORAGE` for server mode. Synced namespaces are * maps keyed by playlist id; a write compares the new map with the last one * the server saw and sends only the playlist entries that changed. Other * keys use plain storage. */ export class ServerUserDataStorage implements UserDataStorage { private readonly synced = new Map< ServerUserStateNamespace, Record >(); private readonly pending = new Map(); private flushTimer: ReturnType | null = null; /** Batches go out one at a time, so an older write never lands last. */ private sending: Promise = Promise.resolve(); private writesInFlight = 0; /** Page-hide writes still travelling, by `namespace|playlistId`. */ private readonly keepaliveKeys = new Set(); /** Re-reads run one at a time, so an older one never lands last. */ private hydration: Promise = Promise.resolve(); /** Edit counter per `namespace|playlistId`, for hydrate to keep them. */ private readonly edits = new Map(); private editCount = 0; constructor( private readonly api: ServerApi, private readonly local: UserDataStorage, private readonly onWriteFailure: (error: unknown) => void ) {} getItem(key: string): string | null { return this.isSynced(key) ? this.local.getItem(SERVER_CACHE_PREFIX + key) : this.local.getItem(key); } setItem(key: string, value: string): void { if (!this.isSynced(key)) { this.local.setItem(key, value); return; } const before = this.baseline(key); this.local.setItem(SERVER_CACHE_PREFIX + key, value); this.queueChanges(key, parseMap(value), before); } removeItem(key: string): void { if (!this.isSynced(key)) { this.local.removeItem(key); return; } const before = this.baseline(key); this.local.removeItem(SERVER_CACHE_PREFIX + key); this.queueChanges(key, {}, before); } /** * Replaces the cache with the server's data, after sending pending * writes. Entries edited while the read was in flight, and entries whose * write failed, keep their local value: the server does not have it yet. */ hydrate(): Promise { const run = this.hydration.then(() => this.hydrateNow()); this.hydration = run.then( () => undefined, () => undefined ); return run; } private async hydrateNow(): Promise { // Edits from here on may miss the read: queued, or still travelling. const editsBefore = this.editCount; await this.flush(); const { entries } = await this.api.getJson('/user-state'); for (const namespace of SERVER_USER_STATE_NAMESPACES) { const value: Record = { ...(entries[namespace] ?? {}), }; const current = parseMap( this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}' ); const unsaved = new Set([ ...this.pending.keys(), ...this.keepaliveKeys, ]); for (const [key, edit] of this.edits) { if (edit > editsBefore) { unsaved.add(key); } } for (const key of unsaved) { const [editedNamespace, playlistId] = splitEditKey(key); if (editedNamespace !== namespace) { continue; } if (playlistId in current) { value[playlistId] = current[playlistId]; } else { delete value[playlistId]; } } try { this.local.setItem( SERVER_CACHE_PREFIX + namespace, JSON.stringify(value) ); } catch { // Storage full: keep the old cache and the baseline it was // diffed against together, or an edit would read as deletes. continue; } this.synced.set(namespace, value); } for (const [key, edit] of this.edits) { if ( edit <= editsBefore && !this.pending.has(key) && !this.keepaliveKeys.has(key) ) { this.edits.delete(key); } } } /** Sends the queued entries after any batch still travelling. */ flush(): Promise { const run = this.sending.then(async () => { const batch = this.takePending(); if (batch.length === 0) { return; } this.writesInFlight += 1; try { await Promise.all(batch.map((entry) => this.send(entry))); } finally { this.writesInFlight -= 1; } }); this.sending = run.then( () => undefined, () => undefined ); return this.sending; } /** * Page hide: `keepalive` requests survive the page. They go out at once * unless a write is still travelling, which they must not overtake, and * later writes wait for them in turn. */ flushWithKeepalive(): void { const entries = this.takePending(); if (entries.length === 0) { return; } const sendAll = async (): Promise => { this.writesInFlight += 1; try { await Promise.all( entries.map((entry) => this.sendKeepalive(entry)) ); } finally { this.writesInFlight -= 1; } }; // Idle: send synchronously, while the hiding page still runs. const run = this.writesInFlight === 0 ? sendAll() : this.sending.then(sendAll); this.sending = run.then( () => undefined, () => undefined ); } private isSynced(key: string): key is ServerUserStateNamespace { return isServerUserStateNamespace(key); } /** * What a write is compared against: the server's last value, or before * the first read finishes, the cache, so an edit sends only its entry. */ private baseline( namespace: ServerUserStateNamespace ): Record { return ( this.synced.get(namespace) ?? parseMap( this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}' ) ); } private queueChanges( namespace: ServerUserStateNamespace, next: Record, previous: Record ): void { for (const [playlistId, value] of Object.entries(next)) { if ( JSON.stringify(previous[playlistId]) !== JSON.stringify(value) ) { this.queue(namespace, playlistId, value); } } for (const playlistId of Object.keys(previous)) { if (!(playlistId in next)) { this.queue(namespace, playlistId, undefined); } } this.synced.set(namespace, next); this.scheduleFlush(); } private queue( namespace: ServerUserStateNamespace, playlistId: string, value: unknown ): void { const entry: PendingEntry = { namespace, playlistId, value, edit: ++this.editCount, }; this.pending.set(entryKey(entry), entry); this.edits.set(entryKey(entry), entry.edit); } private scheduleFlush(): void { if (this.flushTimer !== null || this.pending.size === 0) { return; } this.flushTimer = setTimeout(() => { this.flushTimer = null; void this.flush(); }, FLUSH_DELAY_MS); } private takePending(): PendingEntry[] { if (this.flushTimer !== null) { clearTimeout(this.flushTimer); this.flushTimer = null; } const batch = [...this.pending.values()]; this.pending.clear(); return batch; } /** * Back into the queue after a failed write, for the offline retry or * the next edit, unless the entry was edited since it was sent. */ private requeue(entry: PendingEntry): void { const key = entryKey(entry); if (this.edits.get(key) === entry.edit && !this.pending.has(key)) { this.pending.set(key, entry); } } private async sendKeepalive(entry: PendingEntry): Promise { const key = entryKey(entry); const path = entryPath(entry); this.keepaliveKeys.add(key); const ok = await (entry.value === undefined ? this.api.sendKeepalive('DELETE', path) : this.api.sendKeepalive('PUT', path, { value: entry.value })); this.keepaliveKeys.delete(key); if (!ok) { this.requeue(entry); } } private async send(entry: PendingEntry): Promise { try { if (entry.value === undefined) { await this.api.request(entryPath(entry), { method: 'DELETE' }); } else { await this.api.sendJson('PUT', entryPath(entry), { value: entry.value, }); } } catch (error) { this.requeue(entry); this.onWriteFailure(error); } } } function entryKey(entry: PendingEntry): string { return `${entry.namespace}|${entry.playlistId}`; } function splitEditKey(key: string): [string, string] { const separator = key.indexOf('|'); return [key.slice(0, separator), key.slice(separator + 1)]; } function entryPath(entry: PendingEntry): string { return `/user-state/${entry.namespace}/${encodeURIComponent(entry.playlistId)}`; } function parseMap(value: string): Record { try { const parsed: unknown = JSON.parse(value); return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? (parsed as Record) : {}; } catch { return {}; } }