fix(server-storage): keep Xtream data, ordered sync, sign-in and stale-cache fixes

- Moving browser playlists matches them to the server's by id or source and
  carries the legacy Xtream maps under the server id; it no longer goes
  through the backup format, whose PWA export has no Xtream user data. A
  partial move stays on offer.
- User-data batches go out in order, and a re-read keeps entries edited
  while it was in flight.
- A keepPayload write keeps its cached copy only when the server replaced
  the version its payload came from (new X-Replaced-ETag header).
- 502/503/504 from the proxy count as offline.
- The start-up screen lets the sign-in page through while a token is
  required.
- Settings → Server offers a reload after a move.
- The probe detects Electron through RuntimeCapabilitiesService.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
4grayandClaude Opus 5.5 committed 2026-10-12 00:16:56 +02:00
1 parent 3bfd5394ef
commit 676e3adcd4
24 files changed
+732 -198

No files matched your search

@@ -1,4 +1,5 @@
import express, { Router, type Response } from 'express';
import { SERVER_REPLACED_ETAG_HEADER as REPLACED_ETAG_HEADER } from '@iptvnator/shared/interfaces';
import { matchesEtag } from './etag-match';
import {
deleteAllPlaylistsWithState,
@@ -53,11 +54,16 @@ export function createPlaylistsRouter(
router.put('/playlists/:id', json, (req, res) => {
const id = requirePlaylistId(req.params.id);
const replaced = revisions.etag(id);
playlists.upsert(
{ ...requireDocument(req.body), _id: id },
{ keepPayload: req.query['keepPayload'] === '1' }
);
withEtag(res, revisions.bump(id)).json({ success: true });
// With keepPayload a client caches its copy only if the payload it
// holds is the one this write replaced.
withEtag(res, revisions.bump(id))
.set(REPLACED_ETAG_HEADER, replaced)
.json({ success: true });
});
router.delete('/playlists/:id', (req, res) => {
@@ -76,7 +82,7 @@ export function createPlaylistsRouter(
function withEtag(res: Response, etag: string): Response {
return res
.set('ETag', etag)
.set('Access-Control-Expose-Headers', 'ETag')
.set('Access-Control-Expose-Headers', `ETag, ${REPLACED_ETAG_HEADER}`)
.set('Cache-Control', 'no-cache');
}
@@ -91,6 +91,17 @@ describe('server storage routes', () => {
headers: { 'if-none-match': `W/${etag}` },
});
expect(weakened.status).toBe(304);
// A write names the version it replaced.
const next = await fetch(
`${baseUrl}/v1/playlists/p1?keepPayload=1`,
json({ _id: 'p1', title: 'Renamed' }, { method: 'PUT' })
);
expect(next.headers.get('x-replaced-etag')).toBe(etag);
expect(next.headers.get('etag')).not.toBe(etag);
expect(
next.headers.get('access-control-expose-headers')
).toContain('X-Replaced-ETag');
})
));
@@ -6,6 +6,7 @@ import {
selectPlaylistsLoadFailed,
selectPlaylistsLoadingFlag,
} from '@iptvnator/m3u-state';
import { ServerStorageStatusService } from '@iptvnator/services';
import { AppStartupStatusComponent } from './app-startup-status.component';
describe('AppStartupStatusComponent', () => {
@@ -48,6 +49,31 @@ describe('AppStartupStatusComponent', () => {
expect(fixture.componentInstance.complete()).toBe(true);
});
it('shows the server sign-in page although sources could not load', () => {
TestBed.inject(MockStore).overrideSelector(
selectPlaylistsLoadFailed,
true
);
const status = TestBed.inject(ServerStorageStatusService);
status.activate(
{
service: 'iptvnator',
apiVersion: 1,
instanceId: 'i',
storage: 'server',
auth: 'token',
},
() => undefined
);
status.reportUnauthorized();
const fixture = TestBed.createComponent(AppStartupStatusComponent);
fixture.componentRef.setInput('routeReady', true);
fixture.detectChanges();
expect(fixture.componentInstance.complete()).toBe(true);
expect(fixture.nativeElement.querySelector('section')).toBeNull();
});
it('offers a retry on failure without rendering raw storage errors', () => {
TestBed.inject(MockStore).overrideSelector(
selectPlaylistsLoadFailed,
@@ -14,6 +14,7 @@ import {
selectPlaylistsLoadFailed,
selectPlaylistsLoadingFlag,
} from '@iptvnator/m3u-state';
import { ServerStorageStatusService } from '@iptvnator/services';
/** The first route also waits for settings and XMLTV source reconciliation. */
@Component({
@@ -130,8 +131,14 @@ export class AppStartupStatusComponent {
selectPlaylistsLoadingFlag
);
readonly failed = this.store.selectSignal(selectPlaylistsLoadFailed);
private readonly serverStatus = inject(ServerStorageStatusService);
/** Server storage waiting for a token: only its sign-in page can load. */
private readonly signInRequired = computed(
() => this.serverStatus.state() === 'unauthorized'
);
readonly complete = computed(
() => this.routeReady() && this.sourcesReady()
() =>
this.routeReady() && (this.sourcesReady() || this.signInRequired())
);
retry(): void {
@@ -1,3 +1,4 @@
import { RuntimeCapabilitiesService } from '@iptvnator/services';
import type { ServerInstanceInfo } from '@iptvnator/shared/interfaces';
import type { IptvnatorRuntimeConfig } from '../services/runtime-config';
@@ -29,7 +30,10 @@ export async function probeServerInstance(
): Promise<ServerInstanceProbe> {
const runtimeConfig =
options.runtimeConfig ?? globalThis.window?.__IPTVNATOR_CONFIG__;
const isElectron = options.isElectron ?? !!globalThis.window?.electron;
// The capability service has no dependencies, so it works before
// bootstrap and keeps one definition of the desktop runtime.
const isElectron =
options.isElectron ?? new RuntimeCapabilitiesService().isElectron;
const baseUrl = runtimeConfig?.BACKEND_URL?.trim();
if (isElectron || !baseUrl) {
return { kind: 'skipped' };
@@ -91,8 +91,19 @@
}
</p>
</div>
@if (browserPlaylistCount() > 0) {
<div class="setting-item__control">
<div class="setting-item__control inline-actions">
@if (movedSomething()) {
<!-- The library on screen predates the move. -->
<button
mat-stroked-button
type="button"
(click)="reload()"
data-test-id="server-import-reload"
>
{{ 'SERVER_STORAGE.MIGRATION_RELOAD' | translate }}
</button>
}
@if (browserPlaylistCount() > 0) {
<button
mat-flat-button
type="button"
@@ -110,8 +121,8 @@
}
{{ 'SERVER_STORAGE.MIGRATION_MOVE' | translate }}
</button>
</div>
}
}
</div>
</div>
}
@@ -92,6 +92,7 @@ describe('SettingsServerSectionComponent', () => {
expect(migration.moveBrowserData).toHaveBeenCalled();
expect(byTestId('server-import-browser')).toBeNull();
expect(byTestId('server-import-reload')).not.toBeNull();
expect(fixture.nativeElement.textContent).toContain(
'SERVER_STORAGE.MIGRATION_DONE'
);
@@ -61,6 +61,10 @@ export class SettingsServerSectionComponent {
readonly showBrowserImport = computed(
() => this.browserPlaylistCount() > 0 || this.importResult() !== null
);
readonly movedSomething = computed(() => {
const result = this.importResult();
return !!result && result.imported + result.merged > 0;
});
constructor() {
void this.load();
@@ -99,6 +103,10 @@ export class SettingsServerSectionComponent {
signOut(): void {
this.session.signOut();
}
reload(): void {
window.location.reload();
}
}
function backendOrigin(baseUrl: string): string {
+20 -13
View File
@@ -83,7 +83,9 @@ Server storage adds versioned routes under `/v1`, which the PWA uses in
`DELETE /v1/playlists`: whole playlist documents; `GET` sends an `ETag`
and answers `304` to a matching `If-None-Match`, and
`PUT ...?keepPayload=1` keeps the stored `playlist`, `items` and
`header` fields (`409 payload_missing` without a stored row)
`header` fields (`409 payload_missing` without a stored row); every `PUT`
answers with the new `ETag` and `X-Replaced-ETag`, the version it
replaced
- `GET /v1/user-state`, `PUT|DELETE /v1/user-state/:namespace/:playlistId`:
Xtream favorites, recently viewed, collection snapshots and playback
positions, one JSON value per namespace and playlist
@@ -187,34 +189,39 @@ settings and guards read them in every build.
`electron`, `server`, `indexeddb`), so `PlaylistsService` is unchanged. The
store caches full documents by ETag and uploads a document whose payload it
served unchanged with `keepPayload=1`, so a favorite or a rename does not
resend a large M3U.
resend a large M3U. It keeps its copy after such a write only when
`X-Replaced-ETag` names the version its payload came from; otherwise
another device wrote in between and the next read fetches the document.
- **Xtream user data.** `USER_DATA_STORAGE` becomes `ServerUserDataStorage`:
the four synced namespaces are cached under the `iptvnator-server:` key
prefix (the legacy unprefixed keys are never written in server mode) and
each change sends only the playlist entries that differ, debounced, with a
`keepalive` flush when the page hides. Start-up waits up to 3 s for the
first read.
`keepalive` flush when the page hides. Batches go out one after another,
and a re-read keeps entries edited while it was in flight. Start-up waits
up to 3 s for the first read.
- **Connection state.** `ServerStorageStatusService` is `online`, `offline`
(a request did not reach the server; retried every 30 s and when the tab
returns) or
(a request did not reach the server, or a proxy answered 502, 503 or 504
for it; retried every 30 s and when the tab returns) or
`unauthorized` (a 401 clears the stored token). The rail shows a status
shortcut, an offline banner sits above the workspace content, and failed
writes raise a snackbar (`SERVER_STORAGE.WRITE_FAILED`, or `TOO_LARGE` for
a 413).
- **Access token.** With `IPTVNATOR_ACCESS_TOKEN` the instance reports
`auth: 'token'`; without a stored token, or after a 401, the guards route to
`/server-sign-in`, which checks the token against `/v1/instance/status`
`/server-sign-in` (the start-up screen lets it through although the
playlists could not load), which checks the token against `/v1/instance/status`
before it stores it under `iptvnator-server:access-token` and reloads.
`ServerAccessTokenInterceptor` adds the header to `HttpClient` calls to the
backend, and `PwaService` adds it to its two raw `fetch` calls.
- **Moving browser data.** `ServerMigrationService` offers, once per server
instance (marker `iptvnator-server:migration:<instanceId>`), to move the
playlists this browser kept in IndexedDB. An empty server receives them
with their ids plus the legacy Xtream maps for those ids; a server that
already has playlists goes through the backup export and import, whose
fingerprints merge duplicates. Browser data is never deleted. Settings →
Server repeats the offer while the browser has playlists that were not
moved.
playlists this browser kept in IndexedDB (`planServerMigration`). A
playlist the server already has, by id or by source (Xtream server and
user, Stalker portal and MAC, M3U URL), merges into the server's copy; the
others upload with their ids. The legacy Xtream maps follow under the
server id without replacing what a merged playlist already has. Browser
data is never deleted, and the marker is written only when nothing
failed, so Settings → Server keeps offering a partial move.
What stays per device: settings, parental lock PINs and locked groups,
the global favorites order and other UI preferences in browser storage.
@@ -79,9 +79,15 @@ on the server. Adding a playlist stays the ordinary "Add playlist" flow.
this instance): a dialog offers "Move to server". Playlists are copied
with their ids, then the four Xtream user-data maps for those ids.
- Any later device (server already has playlists): Settings → Server
offers "Import this browser's playlists", which runs the existing backup
export (browser) and import (server) with fingerprint merging, so the
same provider added on two devices is not duplicated.
offers "Import this browser's playlists". A browser playlist the server
already has (same id, or same source: Xtream server and user, Stalker
portal and MAC, M3U URL) merges into the server's copy, so the same
provider added on two devices is not duplicated; the others upload with
their ids. The Xtream maps follow under the server id and never replace
what a merged playlist already has. The backup format is not used: its
PWA export carries no Xtream user data.
- The decision marker is written only when nothing failed, so a partial
move stays on offer and can run again.
- Browser data is never deleted. The server-mode cache uses its own
localStorage key prefix, so legacy keys stay intact.
5. Offline: reads and writes fail with a visible state. The rail shows an
@@ -143,8 +149,7 @@ closes it on `SIGTERM`. Storage errors are logged by name and route only.
- `PlaylistStorageBackend` (libs/services) resolves the document store:
server store in server mode, Electron bridge when `supportsSqlite`, else
IndexedDB. It can switch to browser storage temporarily for the migration
export.
IndexedDB.
- `@iptvnator/services/server-storage` (sub-entry, kept out of the eager
barrel): API client, document store, user-data storage, session, migration.
- `ServerStorageStatusService` (main barrel, tiny): connection state for the
@@ -52,19 +52,4 @@ describe('PlaylistStorageBackend', () => {
expect(backend.kind()).toBe('server');
expect(backend.documentStore()).toBe(store);
});
it('falls back to browser storage while an override task runs', async () => {
testWindow.electron = undefined;
const backend = create(false);
backend.useServerStore({ dbGetAppPlaylists: jest.fn() } as never);
const seen = await backend.runWithBrowserStorage(async () => [
backend.kind(),
backend.usesBrowserStorageOverride(),
]);
expect(seen).toEqual(['indexeddb', true]);
expect(backend.kind()).toBe('server');
expect(backend.usesBrowserStorageOverride()).toBe(false);
});
});
@@ -35,32 +35,14 @@ export type PlaylistStorageKind = 'electron' | 'server' | 'indexeddb';
export class PlaylistStorageBackend {
private readonly runtime = inject(RuntimeCapabilitiesService);
private serverStore: PlaylistDocumentStore | null = null;
private browserOverrideDepth = 0;
/** Server mode: set once by the server storage session at startup. */
useServerStore(store: PlaylistDocumentStore): void {
this.serverStore = store;
}
usesBrowserStorageOverride(): boolean {
return this.browserOverrideDepth > 0;
}
/**
* Runs `task` against browser storage although server mode is active.
* Only the migration export uses it, inside a modal flow.
*/
async runWithBrowserStorage<T>(task: () => Promise<T>): Promise<T> {
this.browserOverrideDepth += 1;
try {
return await task();
} finally {
this.browserOverrideDepth -= 1;
}
}
kind(): PlaylistStorageKind {
if (this.serverStore && this.browserOverrideDepth === 0) {
if (this.serverStore) {
return 'server';
}
return this.runtime.supportsSqlite ? 'electron' : 'indexeddb';
@@ -62,6 +62,28 @@ describe('ServerApiClient', () => {
expect(onWriteFailure).toHaveBeenCalledTimes(1);
});
it('reports offline when a proxy answers for a down backend', async () => {
for (const code of [502, 503, 504]) {
const fetchFn = jest
.fn()
.mockResolvedValue(response(code, 'Bad Gateway'));
await expect(
client(fetchFn).getJson('/user-state')
).rejects.toMatchObject({ status: code });
}
expect(status.reportOffline).toHaveBeenCalledTimes(3);
// A backend error is not an outage.
status.reportOffline.mockClear();
await expect(
client(
jest.fn().mockResolvedValue(response(500, { code: 'x' }))
).getJson('/user-state')
).rejects.toMatchObject({ status: 500 });
expect(status.reportOffline).not.toHaveBeenCalled();
});
it('clears the token and reports unauthorized on 401', async () => {
localStorage.setItem(SERVER_ACCESS_TOKEN_STORAGE_KEY, 'old');
const fetchFn = jest
@@ -7,6 +7,9 @@ import type { ServerStorageStatusService } from '../server-storage-status.servic
const REQUEST_TIMEOUT_MS = 20_000;
/** nginx answers these while the backend behind it is down. */
const GATEWAY_UNAVAILABLE = new Set([502, 503, 504]);
export class ServerApiError extends Error {
constructor(
readonly status: number,
@@ -78,6 +81,11 @@ export class ServerApiClient {
);
}
if (GATEWAY_UNAVAILABLE.has(response.status)) {
// A proxy in front of a backend that is down or restarting.
this.options.status.reportOffline();
}
if (!response.ok && response.status !== 304) {
const body = await readErrorBody(response);
const error = new ServerApiError(
@@ -0,0 +1,79 @@
import type { Playlist } from '@iptvnator/shared/interfaces';
import { planServerMigration, sourceIdentity } from './server-migration-plan';
const playlist = (fields: Partial<Playlist> & { _id: string }) =>
({ title: fields._id, ...fields }) as Playlist;
describe('planServerMigration', () => {
it('uploads everything to an empty server', () => {
const browser = [
playlist({ _id: 'm', url: 'http://a.test/list.m3u' }),
playlist({ _id: 'file' }),
];
expect(planServerMigration(browser, [])).toEqual([
{ kind: 'upload', playlist: browser[0] },
{ kind: 'upload', playlist: browser[1] },
]);
});
it('merges by id first, then by source', () => {
const server = [
playlist({ _id: 'moved' }),
playlist({
_id: 's-x',
serverUrl: 'HTTP://Panel.test/',
username: 'u',
}),
playlist({
_id: 's-st',
portalUrl: 'http://portal.test/c',
macAddress: '00:1A:79:AA:BB:CC',
}),
playlist({ _id: 's-m', url: 'http://a.test/list.m3u' }),
];
const browser = [
playlist({ _id: 'moved' }),
playlist({
_id: 'b-x',
serverUrl: 'http://panel.test',
username: 'u',
}),
playlist({
_id: 'b-st',
url: 'http://portal.test/c/',
macAddress: '00:1a:79:aa:bb:cc',
}),
playlist({ _id: 'b-m', url: 'http://a.test/list.m3u' }),
playlist({
_id: 'b-other',
serverUrl: 'http://panel.test',
username: 'other',
}),
];
expect(planServerMigration(browser, server)).toEqual([
{ kind: 'merge', browserId: 'moved', serverId: 'moved' },
{ kind: 'merge', browserId: 'b-x', serverId: 's-x' },
{ kind: 'merge', browserId: 'b-st', serverId: 's-st' },
{ kind: 'merge', browserId: 'b-m', serverId: 's-m' },
{ kind: 'upload', playlist: browser[4] },
]);
});
it('collapses browser duplicates of one source onto the first', () => {
const browser = [
playlist({ _id: 'one', url: 'http://a.test/list.m3u' }),
playlist({ _id: 'two', url: 'http://A.test/list.m3u' }),
];
expect(planServerMigration(browser, [])).toEqual([
{ kind: 'upload', playlist: browser[0] },
{ kind: 'merge', browserId: 'two', serverId: 'one' },
]);
});
it('has no source for a file or pasted M3U', () => {
expect(sourceIdentity(playlist({ _id: 'file' }))).toBeNull();
});
});
@@ -0,0 +1,94 @@
import type { Playlist } from '@iptvnator/shared/interfaces';
/** Where one browser playlist goes on the server. */
export type ServerMigrationStep =
| {
readonly kind: 'upload';
readonly playlist: Playlist;
}
| {
/** The server already has this source; it keeps its document. */
readonly kind: 'merge';
readonly browserId: string;
readonly serverId: string;
};
/**
* Matches browser playlists to the server's: the same id first (a playlist
* moved before), then the same source (Xtream server and user, Stalker
* portal and MAC, M3U URL). A file or pasted M3U has no source to compare
* and is uploaded unless its id is taken. Browser duplicates of one source
* collapse onto the first.
*/
export function planServerMigration(
browser: readonly Playlist[],
server: readonly Playlist[]
): ServerMigrationStep[] {
const serverIds = new Set(server.map((playlist) => playlist._id));
const bySource = new Map<string, string>();
for (const playlist of server) {
const source = sourceIdentity(playlist);
if (source && !bySource.has(source)) {
bySource.set(source, playlist._id);
}
}
const steps: ServerMigrationStep[] = [];
for (const playlist of browser) {
const source = sourceIdentity(playlist);
const match = serverIds.has(playlist._id)
? playlist._id
: source
? bySource.get(source)
: undefined;
if (match) {
steps.push({
kind: 'merge',
browserId: playlist._id,
serverId: match,
});
continue;
}
steps.push({ kind: 'upload', playlist });
serverIds.add(playlist._id);
if (source) {
bySource.set(source, playlist._id);
}
}
return steps;
}
/** The provider account or URL a playlist reads from, if it has one. */
export function sourceIdentity(playlist: Playlist): string | null {
if (playlist.serverUrl && playlist.username) {
return [
'xtream',
normalizeUrl(playlist.serverUrl),
playlist.username.trim(),
].join('|');
}
const portalUrl = playlist.portalUrl ?? playlist.url;
if (playlist.macAddress && portalUrl) {
return [
'stalker',
normalizeUrl(portalUrl),
playlist.macAddress.trim().toLowerCase(),
].join('|');
}
return playlist.url ? `m3u|${normalizeUrl(playlist.url)}` : null;
}
/** Same rules as the backup import's fingerprints. */
function normalizeUrl(value: string): string {
const trimmed = value.trim();
try {
const url = new URL(trimmed);
const pathname =
url.pathname === '/'
? '/'
: url.pathname.replace(/\/+$/, '') || '/';
return `${url.protocol.toLowerCase()}//${url.host.toLowerCase()}${pathname}${url.search}${url.hash}`;
} catch {
return trimmed.toLowerCase();
}
}
@@ -1,46 +1,45 @@
import { TestBed } from '@angular/core/testing';
import { NgxIndexedDBService } from 'ngx-indexed-db';
import { of } from 'rxjs';
import { PlaylistStorageBackend } from '../playlist-storage-backend';
import { PlaylistBackupService } from '../playlist-backup.service';
import { SERVER_STORAGE_CONFIG } from './server-storage.config';
import { ServerStorageSession } from './server-storage-session';
import { ServerMigrationService } from './server-migration.service';
describe('ServerMigrationService', () => {
const browserPlaylists = [
{ _id: 'm', title: 'Family' },
{ _id: 'x', title: 'Provider', serverUrl: 'http://panel.test' },
{ _id: 'm', title: 'Family', url: 'http://lists.test/family.m3u' },
{
_id: 'x',
title: 'Provider',
serverUrl: 'http://panel.test',
username: 'user',
},
];
let session: {
fetchStatus: jest.Mock;
documentStore: { dbUpsertAppPlaylist: jest.Mock };
api: { sendJson: jest.Mock };
documentStore: {
dbGetAppPlaylistMetas: jest.Mock;
dbUpsertAppPlaylist: jest.Mock;
};
api: { getJson: jest.Mock; sendJson: jest.Mock };
userData: { hydrate: jest.Mock };
};
let backup: { exportBackup: jest.Mock; importBackup: jest.Mock };
beforeEach(() => {
session = {
fetchStatus: jest.fn().mockResolvedValue({ playlists: 0 }),
documentStore: {
dbGetAppPlaylistMetas: jest.fn().mockResolvedValue([]),
dbUpsertAppPlaylist: jest
.fn()
.mockResolvedValue({ success: true }),
},
api: { sendJson: jest.fn().mockResolvedValue(undefined) },
api: {
getJson: jest.fn().mockResolvedValue({ entries: {} }),
sendJson: jest.fn().mockResolvedValue(undefined),
},
userData: { hydrate: jest.fn().mockResolvedValue(undefined) },
};
backup = {
exportBackup: jest.fn().mockResolvedValue({ json: '{}' }),
importBackup: jest.fn().mockResolvedValue({
imported: 1,
merged: 1,
skipped: 0,
failed: 0,
errors: [],
}),
};
TestBed.configureTestingModule({
providers: [
{
@@ -48,14 +47,6 @@ describe('ServerMigrationService', () => {
useValue: { getAll: () => of(browserPlaylists) },
},
{ provide: ServerStorageSession, useValue: session },
{ provide: PlaylistBackupService, useValue: backup },
{
provide: PlaylistStorageBackend,
useValue: {
runWithBrowserStorage: (task: () => Promise<unknown>) =>
task(),
},
},
{
provide: SERVER_STORAGE_CONFIG,
useValue: {
@@ -111,18 +102,76 @@ describe('ServerMigrationService', () => {
{ value: ['live:1'] }
);
expect(session.api.sendJson).toHaveBeenCalledTimes(1);
expect(session.userData.hydrate).toHaveBeenCalled();
expect(migration.browserDataMoved()).toBe(true);
});
it('merges through the backup format when the server already has playlists', async () => {
session.fetchStatus.mockResolvedValue({ playlists: 4 });
it('merges into the server copy and moves Xtream data under its id', async () => {
session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([
{
_id: 'server-x',
title: 'Provider',
serverUrl: 'http://panel.test/',
username: 'user',
},
{
_id: 'server-m',
title: 'Family',
url: 'http://lists.test/family.m3u',
},
]);
session.api.getJson.mockResolvedValue({
entries: {
'xtream-recent-items': { 'server-x': ['vod:9'] },
},
});
localStorage.setItem(
'xtream-favorites',
JSON.stringify({ x: ['live:1'] })
);
localStorage.setItem(
'xtream-recent-items',
JSON.stringify({ x: ['vod:1'] })
);
const result = await TestBed.inject(
ServerMigrationService
).moveBrowserData();
expect(backup.exportBackup).toHaveBeenCalled();
expect(backup.importBackup).toHaveBeenCalledWith('{}');
expect(result).toEqual({ imported: 1, merged: 1, failed: 0 });
expect(result).toEqual({ imported: 0, merged: 2, failed: 0 });
expect(
session.documentStore.dbUpsertAppPlaylist
).not.toHaveBeenCalled();
// Favorites the server copy lacks follow; its own history stays.
expect(session.api.sendJson.mock.calls).toEqual([
[
'PUT',
'/user-state/xtream-favorites/server-x',
{ value: ['live:1'] },
],
]);
});
it('keeps the move on offer when a playlist could not be uploaded', async () => {
session.documentStore.dbUpsertAppPlaylist
.mockRejectedValueOnce(new Error('413'))
.mockResolvedValue({ success: true });
const migration = TestBed.inject(ServerMigrationService);
const result = await migration.moveBrowserData();
expect(result).toEqual({ imported: 1, merged: 0, failed: 1 });
expect(migration.browserDataMoved()).toBe(false);
// A second run merges the moved one and retries the other.
session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([
browserPlaylists[1],
]);
await expect(migration.moveBrowserData()).resolves.toEqual({
imported: 1,
merged: 1,
failed: 0,
});
expect(migration.browserDataMoved()).toBe(true);
});
});
@@ -5,9 +5,9 @@ import {
DbStores,
SERVER_USER_STATE_NAMESPACES,
type Playlist,
type ServerUserStateResponse,
} from '@iptvnator/shared/interfaces';
import { PlaylistBackupService } from '../playlist-backup.service';
import { PlaylistStorageBackend } from '../playlist-storage-backend';
import { planServerMigration } from './server-migration-plan';
import { SERVER_STORAGE_CONFIG } from './server-storage.config';
import { ServerStorageSession } from './server-storage-session';
@@ -25,18 +25,24 @@ export interface ServerMigrationResult {
readonly failed: number;
}
/** A browser playlist's id on the server, and whether it arrived just now. */
interface MigrationTarget {
readonly serverId: string;
readonly uploaded: boolean;
}
/**
* Moves this browser's playlists to the server. An empty server receives
* them with their ids plus the legacy Xtream user-data maps; a server that
* already has playlists goes through the backup export/import, whose
* fingerprints merge duplicates. Browser data is never deleted.
* Moves this browser's playlists to the server. New playlists keep their
* ids; one the server already has (same id, or same provider source) is
* merged into the server's copy. The legacy Xtream favorites, history,
* collections and playback positions follow under the server id, without
* replacing what a merged playlist already has there. Browser data is
* never deleted, and the move can run again after a failure.
*/
@Injectable({ providedIn: 'root' })
export class ServerMigrationService {
private readonly dbService = inject(NgxIndexedDBService);
private readonly session = inject(ServerStorageSession);
private readonly backend = inject(PlaylistStorageBackend);
private readonly backup = inject(PlaylistBackupService);
private readonly instanceId = inject(SERVER_STORAGE_CONFIG).instance
.instanceId;
@@ -75,59 +81,82 @@ export class ServerMigrationService {
}
async moveBrowserData(): Promise<ServerMigrationResult> {
const { playlists: serverCount } = await this.session.fetchStatus();
const result =
serverCount === 0
? await this.copyWithIds()
: await this.mergeThroughBackup();
this.writeMarker('done');
return result;
}
private async copyWithIds(): Promise<ServerMigrationResult> {
const playlists = await this.readBrowserPlaylists();
const [browser, server] = await Promise.all([
this.readBrowserPlaylists(),
this.session.documentStore.dbGetAppPlaylistMetas(),
]);
const targets = new Map<string, MigrationTarget>();
let imported = 0;
let merged = 0;
let failed = 0;
for (const playlist of playlists) {
for (const step of planServerMigration(browser, server)) {
if (step.kind === 'merge') {
merged += 1;
targets.set(step.browserId, {
serverId: step.serverId,
uploaded: false,
});
continue;
}
try {
await this.session.documentStore.dbUpsertAppPlaylist(playlist);
await this.session.documentStore.dbUpsertAppPlaylist(
step.playlist
);
imported += 1;
targets.set(step.playlist._id, {
serverId: step.playlist._id,
uploaded: true,
});
} catch {
failed += 1;
}
}
await this.copyXtreamUserData(new Set(playlists.map((p) => p._id)));
return { imported, merged: 0, failed };
failed += await this.copyXtreamUserData(targets);
await this.session.userData.hydrate();
if (failed === 0) {
// A partial move stays on offer, in Settings → Server.
this.writeMarker('done');
}
return { imported, merged, failed };
}
/** Copies the legacy maps; returns how many entries failed. */
private async copyXtreamUserData(
playlistIds: ReadonlySet<string>
): Promise<void> {
targets: ReadonlyMap<string, MigrationTarget>
): Promise<number> {
if (targets.size === 0) {
return 0;
}
const { entries } =
await this.session.api.getJson<ServerUserStateResponse>(
'/user-state'
);
let failed = 0;
for (const namespace of SERVER_USER_STATE_NAMESPACES) {
const legacy = parseMap(readLocal(namespace));
for (const [playlistId, value] of Object.entries(legacy)) {
if (playlistIds.has(playlistId)) {
const onServer = entries[namespace] ?? {};
for (const [browserId, value] of Object.entries(legacy)) {
const target = targets.get(browserId);
if (
!target ||
(!target.uploaded && target.serverId in onServer)
) {
continue;
}
try {
await this.session.api.sendJson(
'PUT',
`/user-state/${namespace}/${encodeURIComponent(playlistId)}`,
`/user-state/${namespace}/${encodeURIComponent(target.serverId)}`,
{ value }
);
} catch {
failed += 1;
}
}
}
await this.session.userData.hydrate();
}
private async mergeThroughBackup(): Promise<ServerMigrationResult> {
const { json } = await this.backend.runWithBrowserStorage(() =>
this.backup.exportBackup()
);
const summary = await this.backup.importBackup(json);
return {
imported: summary.imported,
merged: summary.merged,
failed: summary.failed,
};
return failed;
}
private readBrowserPlaylists(): Promise<Playlist[]> {
@@ -11,6 +11,15 @@ function fakeResponse(status: number, body?: unknown, etag?: string) {
} as unknown as Response;
}
function putResponse(etag: string, replaced: string) {
return {
status: 200,
ok: true,
headers: new Headers({ etag, 'x-replaced-etag': replaced }),
json: async () => ({ success: true }),
} as unknown as Response;
}
describe('ServerPlaylistDocumentStore', () => {
const doc = {
_id: 'p1',
@@ -99,4 +108,36 @@ describe('ServerPlaylistDocumentStore', () => {
expect(api.sendJson.mock.calls[1][1]).toBe('/playlists/p1');
expect(api.sendJson.mock.calls[1][2]).toBe(served);
});
describe('after a keepPayload write', () => {
async function readThenEdit(replaced: string) {
const { api, store } = setup();
// nginx gzip weakens the read's ETag.
api.request.mockResolvedValueOnce(
fakeResponse(200, doc, 'W/"b.0.1"')
);
const served = await store.dbGetAppPlaylist('p1');
if (!served) throw new Error('expected a document');
api.sendJson.mockResolvedValueOnce(
putResponse('"b.0.2"', replaced)
);
await store.dbUpsertAppPlaylist({ ...served, favorites: ['a'] });
api.request.mockResolvedValueOnce(
fakeResponse(200, doc, '"b.0.3"')
);
await store.dbGetAppPlaylist('p1');
return api.request.mock.calls[1][1];
}
it('keeps its copy when it replaced the version it read', async () => {
await expect(readThenEdit('"b.0.1"')).resolves.toEqual({
headers: { 'If-None-Match': '"b.0.2"' },
});
});
it('drops its copy when another device wrote in between', async () => {
// The server kept a newer payload than the one this client has.
await expect(readThenEdit('"b.0.4"')).resolves.toEqual({});
});
});
});
@@ -1,5 +1,6 @@
import {
SERVER_PAYLOAD_FIELDS,
SERVER_REPLACED_ETAG_HEADER,
type ElectronBridgeCountResult,
type ElectronBridgeResult,
type Playlist,
@@ -16,14 +17,15 @@ type ServerApi = Pick<ServerApiClient, 'request' | 'getJson' | 'sendJson'>;
/**
* `PlaylistDocumentStore` over the server storage API. Full documents are
* cached by ETag; every read hands out a private copy and remembers its
* payload objects, so an upsert whose payload is one of them omits it
* (`keepPayload=1`). `PlaylistsService` builds a new object per mutation,
* so cached documents are never mutated in place.
* cached by ETag; every read hands out a private copy and remembers which
* ETag its payload objects came from, so an upsert whose payload is one of
* them omits it (`keepPayload=1`). `PlaylistsService` builds a new object
* per mutation, so cached documents are never mutated in place.
*/
export class ServerPlaylistDocumentStore implements PlaylistDocumentStore {
private readonly cache = new Map<string, CachedDocument>();
private readonly servedPayloads = new WeakSet<object>();
/** Served payload object → the ETag of the document it came from. */
private readonly servedPayloads = new WeakMap<object, string>();
constructor(private readonly api: ServerApi) {}
@@ -52,37 +54,46 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore {
}
if (response.status === 304 && cached) {
return this.serve(cached.document);
return this.serve(cached.document, cached.etag);
}
const document = (await response.json()) as Playlist;
this.remember(playlistId, response, document);
return this.serve(document);
const etag = this.remember(playlistId, response, document);
return this.serve(document, etag);
}
async dbUpsertAppPlaylist(
playlist: Playlist
): Promise<ElectronBridgeResult> {
const path = `/playlists/${encodeURIComponent(playlist._id)}`;
const keepPayload = this.carriesServedPayload(playlist);
let response: Response;
try {
response = keepPayload
? await this.api.sendJson(
'PUT',
`${path}?keepPayload=1`,
withoutPayload(playlist)
)
: await this.api.sendJson('PUT', path, playlist);
} catch (error) {
if (!(
keepPayload &&
error instanceof ServerApiError &&
error.code === 'payload_missing'
)) {
throw error;
const payloadEtag = this.servedPayloadEtag(playlist);
if (payloadEtag !== null) {
try {
const response = await this.api.sendJson(
'PUT',
`${path}?keepPayload=1`,
withoutPayload(playlist)
);
// The server kept its stored payload; it is ours only if
// nothing replaced it since we read it.
const replaced = response.headers.get(
SERVER_REPLACED_ETAG_HEADER
);
if (replaced && sameEtag(replaced, payloadEtag)) {
this.remember(playlist._id, response, playlist);
} else {
this.cache.delete(playlist._id);
}
return { success: true };
} catch (error) {
if (!(
error instanceof ServerApiError &&
error.code === 'payload_missing'
)) {
throw error;
}
}
response = await this.api.sendJson('PUT', path, playlist);
}
const response = await this.api.sendJson('PUT', path, playlist);
this.remember(playlist._id, response, playlist);
return { success: true };
}
@@ -131,43 +142,60 @@ export class ServerPlaylistDocumentStore implements PlaylistDocumentStore {
return { success: true };
}
private remember(id: string, response: Response, document: Playlist): void {
private remember(
id: string,
response: Response,
document: Playlist
): string | null {
const etag = response.headers.get('etag');
if (etag) {
this.cache.set(id, { etag, document });
} else {
this.cache.delete(id);
}
return etag;
}
private serve(document: Playlist): Playlist {
private serve(document: Playlist, etag: string | null): Playlist {
const copy = structuredClone(document);
if (!etag) {
return copy;
}
const record = copy as unknown as Record<string, unknown>;
for (const field of SERVER_PAYLOAD_FIELDS) {
const value = record[field];
if (value && typeof value === 'object') {
this.servedPayloads.add(value);
this.servedPayloads.set(value, etag);
}
}
return copy;
}
private carriesServedPayload(playlist: Playlist): boolean {
/** The ETag every payload field of `playlist` was served under, if any. */
private servedPayloadEtag(playlist: Playlist): string | null {
const record = playlist as unknown as Record<string, unknown>;
const present = SERVER_PAYLOAD_FIELDS.filter(
(field) => record[field] !== undefined && record[field] !== null
);
return (
present.length > 0 &&
present.every(
(field) =>
typeof record[field] === 'object' &&
this.servedPayloads.has(record[field] as object)
)
);
const etags = new Set<string | undefined>();
for (const field of SERVER_PAYLOAD_FIELDS) {
const value = record[field];
if (value === undefined || value === null) {
continue;
}
etags.add(
typeof value === 'object'
? this.servedPayloads.get(value)
: undefined
);
}
const [etag] = etags;
return etags.size === 1 && etag !== undefined ? etag : null;
}
}
/** nginx's gzip turns ETags weak; the revision stays the same. */
function sameEtag(left: string, right: string): boolean {
return left.replace(/^W\//, '') === right.replace(/^W\//, '');
}
function withoutPayload(playlist: Playlist): Record<string, unknown> {
const copy = { ...playlist } as Record<string, unknown>;
for (const field of SERVER_PAYLOAD_FIELDS) {
@@ -38,7 +38,6 @@ export class ServerStorageSession {
readonly userData = new ServerUserDataStorage(
this.api,
globalThis.localStorage,
() => this.backend.usesBrowserStorageOverride(),
() => this.announceWriteFailure()
);
@@ -10,11 +10,18 @@ function memoryStorage() {
};
}
/** Lets chained promises run without advancing the fake timers. */
async function settle(): Promise<void> {
for (let i = 0; i < 10; i += 1) {
await Promise.resolve();
}
}
describe('ServerUserDataStorage', () => {
beforeEach(() => jest.useFakeTimers());
afterEach(() => jest.useRealTimers());
function setup(useLocal = false) {
function setup() {
const local = memoryStorage();
const api = {
getJson: jest.fn().mockResolvedValue({
@@ -33,7 +40,6 @@ describe('ServerUserDataStorage', () => {
const storage = new ServerUserDataStorage(
api as never,
local,
() => useLocal,
onWriteFailure
);
return { api, local, storage, onWriteFailure };
@@ -70,13 +76,11 @@ describe('ServerUserDataStorage', () => {
expect(api.sendJson).toHaveBeenCalledTimes(1);
});
it('passes other keys and the browser override to plain storage', () => {
const { api, local, storage } = setup(true);
it('passes other keys to plain storage', () => {
const { api, local, storage } = setup();
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();
});
@@ -107,4 +111,84 @@ describe('ServerUserDataStorage', () => {
expect(onWriteFailure).toHaveBeenCalled();
});
it('keeps an edit made while a refresh is reading', async () => {
const { api, storage } = setup();
await storage.hydrate();
let answer: (value: unknown) => void = () => undefined;
api.getJson.mockReturnValueOnce(
new Promise((resolve) => (answer = resolve))
);
const refresh = storage.hydrate();
await settle();
storage.setItem(
'xtream-favorites',
JSON.stringify({ p1: ['live:1'], p2: ['live:2'] })
);
// The server's older snapshot arrives after the edit.
answer({ entries: { 'xtream-favorites': { p1: ['live:1'] } } });
await refresh;
expect(JSON.parse(storage.getItem('xtream-favorites') ?? '')).toEqual({
p1: ['live:1'],
p2: ['live:2'],
});
await jest.runOnlyPendingTimersAsync();
expect(api.sendJson).toHaveBeenCalledWith(
'PUT',
'/user-state/xtream-favorites/p2',
{ value: ['live:2'] }
);
// The next edit does not send a delete for the kept entry.
storage.setItem(
'xtream-favorites',
JSON.stringify({ p1: ['live:1', 'live:3'], p2: ['live:2'] })
);
await jest.runOnlyPendingTimersAsync();
expect(api.request).not.toHaveBeenCalledWith(
'/user-state/xtream-favorites/p2',
{ method: 'DELETE' }
);
});
it('sends a newer batch only after the older one finished', async () => {
const { api, storage } = setup();
await storage.hydrate();
const order: string[] = [];
let finishFirst: () => void = () => undefined;
api.sendJson
.mockImplementationOnce(
(_method: string, _path: string, body: { value: unknown }) => {
order.push(`start ${JSON.stringify(body.value)}`);
return new Promise<void>((resolve) => {
finishFirst = () => {
order.push('end first');
resolve();
};
});
}
)
.mockImplementation(
(_method: string, _path: string, body: { value: unknown }) => {
order.push(`start ${JSON.stringify(body.value)}`);
return Promise.resolve();
}
);
storage.setItem('xtream-favorites', JSON.stringify({ p1: ['a'] }));
const first = storage.flush();
await settle();
// A newer edit while the first batch is still travelling.
storage.setItem('xtream-favorites', JSON.stringify({ p1: ['b'] }));
const second = storage.flush();
await settle();
expect(order).toEqual(['start ["a"]']);
finishFirst();
await first;
await second;
expect(order).toEqual(['start ["a"]', 'end first', 'start ["b"]']);
});
});
@@ -27,7 +27,7 @@ type ServerApi = Pick<
* 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.
* keys use plain storage.
*/
export class ServerUserDataStorage implements UserDataStorage {
private readonly synced = new Map<
@@ -36,11 +36,15 @@ export class ServerUserDataStorage implements UserDataStorage {
>();
private readonly pending = new Map<string, PendingEntry>();
private flushTimer: ReturnType<typeof setTimeout> | null = null;
/** Batches go out one at a time, so an older write never lands last. */
private sending: Promise<void> = Promise.resolve();
/** Edit counter per `namespace|playlistId`, for hydrate to keep them. */
private readonly edits = new Map<string, number>();
private editCount = 0;
constructor(
private readonly api: ServerApi,
private readonly local: UserDataStorage,
private readonly useLocal: () => boolean,
private readonly onWriteFailure: (error: unknown) => void
) {}
@@ -68,24 +72,58 @@ export class ServerUserDataStorage implements UserDataStorage {
this.queueChanges(key, {});
}
/** Replaces the cache with the server's data, after sending pending writes. */
/**
* Replaces the cache with the server's data, after sending pending
* writes. Entries edited while the read was in flight keep their local
* value: their write is queued or travelling and will reach the server.
*/
async hydrate(): Promise<void> {
// 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<ServerUserStateResponse>('/user-state');
for (const namespace of SERVER_USER_STATE_NAMESPACES) {
const value = entries[namespace] ?? {};
const value: Record<string, unknown> = {
...(entries[namespace] ?? {}),
};
const current = parseMap(
this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}'
);
for (const [key, edit] of this.edits) {
const [editedNamespace, playlistId] = splitEditKey(key);
if (edit <= editsBefore || editedNamespace !== namespace) {
continue;
}
if (playlistId in current) {
value[playlistId] = current[playlistId];
} else {
delete value[playlistId];
}
}
this.synced.set(namespace, value);
this.local.setItem(
SERVER_CACHE_PREFIX + namespace,
JSON.stringify(value)
);
}
for (const [key, edit] of this.edits) {
if (edit <= editsBefore) {
this.edits.delete(key);
}
}
}
async flush(): Promise<void> {
const batch = this.takePending();
await Promise.all(batch.map((entry) => this.send(entry)));
/** Sends the queued entries after any batch still travelling. */
flush(): Promise<void> {
const run = this.sending.then(() =>
Promise.all(this.takePending().map((entry) => this.send(entry)))
);
this.sending = run.then(
() => undefined,
() => undefined
);
return this.sending;
}
flushWithKeepalive(): void {
@@ -100,7 +138,7 @@ export class ServerUserDataStorage implements UserDataStorage {
}
private isSynced(key: string): key is ServerUserStateNamespace {
return !this.useLocal() && isServerUserStateNamespace(key);
return isServerUserStateNamespace(key);
}
private queueChanges(
@@ -112,26 +150,24 @@ export class ServerUserDataStorage implements UserDataStorage {
if (
JSON.stringify(previous[playlistId]) !== JSON.stringify(value)
) {
this.pending.set(`${namespace}|${playlistId}`, {
namespace,
playlistId,
value,
});
this.queue({ namespace, playlistId, value });
}
}
for (const playlistId of Object.keys(previous)) {
if (!(playlistId in next)) {
this.pending.set(`${namespace}|${playlistId}`, {
namespace,
playlistId,
value: undefined,
});
this.queue({ namespace, playlistId, value: undefined });
}
}
this.synced.set(namespace, next);
this.scheduleFlush();
}
private queue(entry: PendingEntry): void {
const key = `${entry.namespace}|${entry.playlistId}`;
this.pending.set(key, entry);
this.edits.set(key, ++this.editCount);
}
private scheduleFlush(): void {
if (this.flushTimer !== null || this.pending.size === 0) {
return;
@@ -167,6 +203,11 @@ export class ServerUserDataStorage implements UserDataStorage {
}
}
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)}`;
}
@@ -36,6 +36,13 @@ export interface ServerInstanceInfo {
readonly auth: 'none' | 'token';
}
/**
* Sent with every `PUT /v1/playlists/:id`: the document's ETag before that
* write. A client that kept the stored payload (`keepPayload=1`) may cache
* its copy under the new ETag only if its payload came from this one.
*/
export const SERVER_REPLACED_ETAG_HEADER = 'X-Replaced-ETag';
export interface ServerStorageStatusResponse {
readonly playlists: number;
readonly databaseBytes: number;