diff --git a/libs/services/src/lib/server-storage/index.ts b/libs/services/src/lib/server-storage/index.ts index 4677a4de3..6c76c6fd7 100644 --- a/libs/services/src/lib/server-storage/index.ts +++ b/libs/services/src/lib/server-storage/index.ts @@ -1,3 +1,4 @@ export * from './server-storage.config'; export * from './server-api-client'; export * from './server-playlist-document-store'; +export * from './server-user-data-storage'; diff --git a/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts b/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts new file mode 100644 index 000000000..e0991df01 --- /dev/null +++ b/libs/services/src/lib/server-storage/server-user-data-storage.spec.ts @@ -0,0 +1,110 @@ +import { ServerUserDataStorage } from './server-user-data-storage'; + +function memoryStorage() { + const values = new Map(); + return { + values, + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => void values.set(key, value), + removeItem: (key: string) => void values.delete(key), + }; +} + +describe('ServerUserDataStorage', () => { + beforeEach(() => jest.useFakeTimers()); + afterEach(() => jest.useRealTimers()); + + function setup(useLocal = false) { + const local = memoryStorage(); + const api = { + getJson: jest.fn().mockResolvedValue({ + entries: { + 'xtream-favorites': { p1: ['live:1'] }, + 'xtream-recent-items': {}, + 'xtream-collection-items': {}, + 'xtream-playback-positions': {}, + }, + }), + sendJson: jest.fn().mockResolvedValue(undefined), + request: jest.fn().mockResolvedValue(undefined), + sendKeepalive: jest.fn(), + }; + const onWriteFailure = jest.fn(); + const storage = new ServerUserDataStorage( + api as never, + local, + () => useLocal, + onWriteFailure + ); + return { api, local, storage, onWriteFailure }; + } + + it('hydrates synced keys into a prefixed cache', async () => { + const { local, storage } = setup(); + + await storage.hydrate(); + + expect(storage.getItem('xtream-favorites')).toBe('{"p1":["live:1"]}'); + expect(local.values.get('iptvnator-server:xtream-favorites')).toBe( + '{"p1":["live:1"]}' + ); + expect(local.values.has('xtream-favorites')).toBe(false); + }); + + it('sends only changed and removed playlist entries', async () => { + const { api, storage } = setup(); + await storage.hydrate(); + + storage.setItem('xtream-favorites', JSON.stringify({ p2: ['live:9'] })); + await jest.runOnlyPendingTimersAsync(); + + expect(api.sendJson).toHaveBeenCalledWith( + 'PUT', + '/user-state/xtream-favorites/p2', + { value: ['live:9'] } + ); + expect(api.request).toHaveBeenCalledWith( + '/user-state/xtream-favorites/p1', + { method: 'DELETE' } + ); + expect(api.sendJson).toHaveBeenCalledTimes(1); + }); + + it('passes other keys and the browser override to plain storage', () => { + const { api, local, storage } = setup(true); + + storage.setItem('xtream-favorites', '{"legacy":[]}'); + storage.setItem('xtream-playlists', '[]'); + + expect(local.values.get('xtream-favorites')).toBe('{"legacy":[]}'); + expect(local.values.get('xtream-playlists')).toBe('[]'); + expect(api.sendJson).not.toHaveBeenCalled(); + }); + + it('flushes pending entries with keepalive', async () => { + const { api, storage } = setup(); + await storage.hydrate(); + + storage.setItem('xtream-recent-items', JSON.stringify({ p1: [] })); + storage.flushWithKeepalive(); + + expect(api.sendKeepalive).toHaveBeenCalledWith( + 'PUT', + '/user-state/xtream-recent-items/p1', + { value: [] } + ); + await jest.runOnlyPendingTimersAsync(); + expect(api.sendJson).not.toHaveBeenCalled(); + }); + + it('reports failed sends', async () => { + const { api, storage, onWriteFailure } = setup(); + await storage.hydrate(); + api.sendJson.mockRejectedValueOnce(new Error('offline')); + + storage.setItem('xtream-favorites', JSON.stringify({ p1: ['live:2'] })); + await jest.runOnlyPendingTimersAsync(); + + expect(onWriteFailure).toHaveBeenCalled(); + }); +}); diff --git a/libs/services/src/lib/server-storage/server-user-data-storage.ts b/libs/services/src/lib/server-storage/server-user-data-storage.ts new file mode 100644 index 000000000..e163b2657 --- /dev/null +++ b/libs/services/src/lib/server-storage/server-user-data-storage.ts @@ -0,0 +1,183 @@ +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; +} + +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, and every key while the browser override runs, 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; + + constructor( + private readonly api: ServerApi, + private readonly local: UserDataStorage, + private readonly useLocal: () => boolean, + 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; + } + this.local.setItem(SERVER_CACHE_PREFIX + key, value); + this.queueChanges(key, parseMap(value)); + } + + removeItem(key: string): void { + if (!this.isSynced(key)) { + this.local.removeItem(key); + return; + } + this.local.removeItem(SERVER_CACHE_PREFIX + key); + this.queueChanges(key, {}); + } + + /** Replaces the cache with the server's data, after sending pending writes. */ + async hydrate(): Promise { + await this.flush(); + const { entries } = + await this.api.getJson('/user-state'); + for (const namespace of SERVER_USER_STATE_NAMESPACES) { + const value = entries[namespace] ?? {}; + this.synced.set(namespace, value); + this.local.setItem( + SERVER_CACHE_PREFIX + namespace, + JSON.stringify(value) + ); + } + } + + async flush(): Promise { + const batch = this.takePending(); + await Promise.all(batch.map((entry) => this.send(entry))); + } + + flushWithKeepalive(): void { + for (const entry of this.takePending()) { + const path = entryPath(entry); + if (entry.value === undefined) { + this.api.sendKeepalive('DELETE', path); + } else { + this.api.sendKeepalive('PUT', path, { value: entry.value }); + } + } + } + + private isSynced(key: string): key is ServerUserStateNamespace { + return !this.useLocal() && isServerUserStateNamespace(key); + } + + private queueChanges( + namespace: ServerUserStateNamespace, + next: Record + ): void { + const previous = this.synced.get(namespace) ?? {}; + for (const [playlistId, value] of Object.entries(next)) { + if ( + JSON.stringify(previous[playlistId]) !== JSON.stringify(value) + ) { + this.pending.set(`${namespace}|${playlistId}`, { + namespace, + playlistId, + value, + }); + } + } + for (const playlistId of Object.keys(previous)) { + if (!(playlistId in next)) { + this.pending.set(`${namespace}|${playlistId}`, { + namespace, + playlistId, + value: undefined, + }); + } + } + this.synced.set(namespace, next); + this.scheduleFlush(); + } + + 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; + } + + 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.onWriteFailure(error); + } + } +} + +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 {}; + } +}