feat(services): server-synced user data storage

Synced Xtream maps live in a write-through cache under the
iptvnator-server: prefix; writes send only the playlist entries that
changed, debounced, and flush with keepalive when the page hides.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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';

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<string, string>();
+    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();
+    });
+});

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<string, unknown>
+    >();
+    private readonly pending = new Map<string, PendingEntry>();
+    private flushTimer: ReturnType<typeof setTimeout> | 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<void> {
+        await this.flush();
+        const { entries } =
+            await this.api.getJson<ServerUserStateResponse>('/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<void> {
+        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<string, unknown>
+    ): 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<void> {
+        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<string, unknown> {
+    try {
+        const parsed: unknown = JSON.parse(value);
+        return parsed && typeof parsed === 'object' && !Array.isArray(parsed)
+            ? (parsed as Record<string, unknown>)
+            : {};
+    } catch {
+        return {};
+    }
+}