x15904151985/iptvnator
1
0 Fork 0
mirror of https://github.com/4gray/iptvnator.git synced 2026-10-14 05:06:16 -08:00
Code Issues Packages Projects Releases Wiki Activity

fix(server-storage): merge user data on move, keep failed writes, sign-in route gate

Browse Source 
- A merged playlist gains the browser document's favorites and history;
  Xtream values for one server entry are combined (lists unioned, maps
  merged, the server winning conflicts) and written once.
- A failed user-data write stays queued and survives a re-read; the
  page-hide flush waits for a batch still travelling.
- The start-up screen lets the sign-in route through whatever its check
  reports.
- The backend refuses IPTVNATOR_ACCESS_TOKEN without server storage.

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 676e3adcd4
commit fc32934ce3
17 files changed
+474 -90
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

apps/web-backend/src/app/storage/storage-config.spec.ts
+20 -1
View File
@@ -1,4 +1,7 @@
import { readServerStorageConfig } from './storage-config';
import {
readServerStorageConfig,
serverStorageConfigProblem,
} from './storage-config';
describe('readServerStorageConfig', () => {
it('keeps storage off and the API open by default', () => {
@@ -42,4 +45,20 @@ describe('readServerStorageConfig', () => {
.maxDocumentBytes
).toBe(64 * 1024 * 1024);
});
it('refuses an access token without server storage', () => {
const problem = (env: Record<string, string>) =>
serverStorageConfigProblem(readServerStorageConfig(env));
expect(problem({ IPTVNATOR_ACCESS_TOKEN: 's3cret' })).toContain(
'IPTVNATOR_STORAGE=server'
);
expect(
problem({
IPTVNATOR_ACCESS_TOKEN: 's3cret',
IPTVNATOR_STORAGE: 'server',
})
).toBeNull();
expect(problem({})).toBeNull();
});
});
apps/web-backend/src/app/storage/storage-config.ts
+13
View File
@@ -10,6 +10,19 @@ export interface ServerStorageConfig {
const DEFAULT_MAX_DOCUMENT_MB = 64;
const BYTES_PER_MB = 1024 * 1024;
/**
* A combination the browser app cannot use, as a start-up message, or
* `null`. The PWA asks for the access token only in server storage mode,
* so a token on a stateless backend would lock every browser out.
*/
export function serverStorageConfigProblem(
config: ServerStorageConfig
): string | null {
return config.accessToken && !config.enabled
? 'IPTVNATOR_ACCESS_TOKEN needs IPTVNATOR_STORAGE=server: the browser app asks for the token only when the server stores its playlists. Unset the token, or protect a stateless backend in your reverse proxy.'
: null;
}
export function readServerStorageConfig(
env: Record<string, string | undefined> = process.env
): ServerStorageConfig {
apps/web-backend/src/main.ts
+9 -1
View File
@@ -4,7 +4,10 @@ import {
} from '@iptvnator/shared/host-health';
import { closeGracefully } from './app/graceful-shutdown';
import { createWebBackendApp } from './app/web-backend-app';
import { readServerStorageConfig } from './app/storage/storage-config';
import {
readServerStorageConfig,
serverStorageConfigProblem,
} from './app/storage/storage-config';
import {
openServerStorage,
type ServerStorage,
@@ -20,6 +23,11 @@ async function start(): Promise<void> {
}
const config = readServerStorageConfig();
const problem = serverStorageConfigProblem(config);
if (problem) {
console.error(`[web-backend] ${problem}`);
process.exit(1);
}
let storage: ServerStorage | null = null;
if (config.enabled) {
try {
apps/web/src/app/app-startup-status.component.spec.ts
+7 -14
View File
@@ -6,7 +6,7 @@ import {
selectPlaylistsLoadFailed,
selectPlaylistsLoadingFlag,
} from '@iptvnator/m3u-state';
import { ServerStorageStatusService } from '@iptvnator/services';
import { provideRouter, Router } from '@angular/router';
import { AppStartupStatusComponent } from './app-startup-status.component';
describe('AppStartupStatusComponent', () => {
@@ -14,6 +14,7 @@ describe('AppStartupStatusComponent', () => {
TestBed.configureTestingModule({
imports: [AppStartupStatusComponent, TranslateModule.forRoot()],
providers: [
provideRouter([{ path: 'server-sign-in', children: [] }]),
provideMockStore({
selectors: [
{ selector: selectPlaylistsLoadingFlag, value: false },
@@ -49,26 +50,18 @@ describe('AppStartupStatusComponent', () => {
expect(fixture.componentInstance.complete()).toBe(true);
});
it('shows the server sign-in page although sources could not load', () => {
it('shows the server sign-in page although sources could not load', async () => {
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(false);
await TestBed.inject(Router).navigateByUrl('/server-sign-in');
fixture.detectChanges();
expect(fixture.componentInstance.complete()).toBe(true);
expect(fixture.nativeElement.querySelector('section')).toBeNull();
apps/web/src/app/app-startup-status.component.ts
+19 -7
View File
@@ -5,16 +5,19 @@ import {
inject,
input,
} from '@angular/core';
import { toSignal } from '@angular/core/rxjs-interop';
import { MatButtonModule } from '@angular/material/button';
import { MatProgressSpinnerModule } from '@angular/material/progress-spinner';
import { NavigationEnd, Router } from '@angular/router';
import { Store } from '@ngrx/store';
import { TranslatePipe } from '@ngx-translate/core';
import { filter, map } from 'rxjs';
import {
PlaylistActions,
selectPlaylistsLoadFailed,
selectPlaylistsLoadingFlag,
} from '@iptvnator/m3u-state';
import { ServerStorageStatusService } from '@iptvnator/services';
import { SERVER_SIGN_IN_PATH } from '@iptvnator/services';
/** The first route also waits for settings and XMLTV source reconciliation. */
@Component({
@@ -131,14 +134,23 @@ 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'
private readonly router = inject(Router);
private readonly url = toSignal(
this.router.events.pipe(
filter((event) => event instanceof NavigationEnd),
map((event) => event.urlAfterRedirects)
),
{ initialValue: this.router.url }
);
/**
* Server storage asking for a token: the playlists cannot load, and the
* sign-in page must stay up whatever its check reports.
*/
private readonly onSignInPage = computed(() =>
this.url().startsWith(SERVER_SIGN_IN_PATH)
);
readonly complete = computed(
() =>
this.routeReady() && (this.sourcesReady() || this.signInRequired())
() => this.routeReady() && (this.sourcesReady() || this.onSignInPage())
);
retry(): void {
apps/web/src/app/server-storage/server-access.guard.ts
+5 -2
View File
@@ -1,11 +1,14 @@
import { inject } from '@angular/core';
import { CanActivateFn, Router } from '@angular/router';
import { ServerStorageStatusService } from '@iptvnator/services';
import {
SERVER_SIGN_IN_PATH,
ServerStorageStatusService,
} from '@iptvnator/services';
/** Without a valid token the workspace would show an empty library. */
export const serverAccessGuard: CanActivateFn = () =>
inject(ServerStorageStatusService).state() === 'unauthorized'
? inject(Router).parseUrl('/server-sign-in')
? inject(Router).parseUrl(SERVER_SIGN_IN_PATH)
: true;
export const serverSignInGuard: CanActivateFn = () =>
docker/README.md
+1 -1
View File
@@ -103,7 +103,7 @@ safe local defaults shown below.
| `NODE_EXTRA_CA_CERTS` | unset | Optional Node.js CA bundle path for providers using private certificate authorities. Mount the CA file into the container and set this path. |
| `IPTVNATOR_STORAGE` | `server` in the image | `server` opens the server storage database under `IPTVNATOR_DATA_DIR` and its `/api/v1` routes; `none` keeps the backend stateless. |
| `IPTVNATOR_DATA_DIR` | `/data` in the image | Data root; the database is `<root>/databases/iptvnator.db`. Mount a volume here. |
| `IPTVNATOR_ACCESS_TOKEN` | unset | When set, every route except `/`, `/health`, `/config.js` and `/v1/instance` requires `Authorization: Bearer <token>`, proxy routes included. Each browser asks for the token once and keeps it. |
| `IPTVNATOR_ACCESS_TOKEN` | unset | When set, every route except `/`, `/health`, `/config.js` and `/v1/instance` requires `Authorization: Bearer <token>`, proxy routes included. Each browser asks for the token once and keeps it. Needs `IPTVNATOR_STORAGE=server`; the backend refuses to start otherwise. |
| `IPTVNATOR_MAX_DOCUMENT_MB` | `64` | Largest playlist document the backend accepts. Keep the nginx `client_max_body_size` in step. |
| `IPTVNATOR_BACKEND_START_TIMEOUT` | `300` | Seconds the entrypoint waits for the backend's `/health` before giving up. Database migrations run before the backend listens. |
docs/architecture/pwa-self-hosted.md
+12 -7
View File
@@ -196,8 +196,10 @@ settings and guards read them in every build.
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. Batches go out one after another,
and a re-read keeps entries edited while it was in flight. Start-up waits
`keepalive` flush when the page hides. Batches go out one after another
(the page-hide flush waits for a batch still travelling), a failed write
stays queued for the next flush, and a re-read keeps entries edited while
it was in flight or not yet saved. Start-up waits
up to 3 s for the first read.
- **Connection state.** `ServerStorageStatusService` is `online`, `offline`
(a request did not reach the server, or a proxy answered 502, 503 or 504
@@ -207,7 +209,8 @@ settings and guards read them in every build.
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
`auth: 'token'` (the backend refuses a token without server storage: the
PWA has no sign-in flow for a stateless backend); without a stored token, or after a 401, the guards route to
`/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.
@@ -218,10 +221,12 @@ settings and guards read them in every build.
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.
others upload with their ids. A merged playlist gains the browser
document's favorites and history, and the legacy Xtream maps follow under
the server id; `combineUserData` unions lists and merges maps, the
server's value winning on a conflict. 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.
docs/superpowers/specs/2026-10-08-server-storage-design.md
+3 -2
View File
@@ -83,8 +83,9 @@ on the server. Adding a playlist stays the ordinary "Add playlist" flow.
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
their ids. A merged playlist gains the browser's favorites and history,
and the Xtream maps follow under the server id, combined with what the
server has (lists unioned, maps merged, the server winning conflicts). 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.
libs/services/src/lib/server-storage-status.service.ts
+3
View File
@@ -4,6 +4,9 @@ import type { ServerInstanceInfo } from '@iptvnator/shared/interfaces';
export type ServerConnectionState =
'inactive' | 'online' | 'offline' | 'unauthorized';
/** Where server mode asks for the access token. */
export const SERVER_SIGN_IN_PATH = '/server-sign-in';
/**
* Connection state of the self-hosted server storage. `inactive` in Electron
* and in a PWA without server storage; the server session activates it.
libs/services/src/lib/server-storage/server-migration-merge.spec.ts
+42
View File
@@ -0,0 +1,42 @@
import { combineUserData, unionItems } from './server-migration-merge';
describe('combineUserData', () => {
it('unions lists, server entries first', () => {
expect(
combineUserData([
['live:1', 'live:2'],
['live:2', 'live:3'],
])
).toEqual(['live:1', 'live:2', 'live:3']);
});
it('merges maps and keeps the first value of a key', () => {
expect(
combineUserData([
{ a: { position: 10 } },
{ a: { position: 99 }, b: { position: 5 } },
])
).toEqual({ a: { position: 10 }, b: { position: 5 } });
});
it('ignores missing values and keeps the first of mixed kinds', () => {
expect(combineUserData([undefined, ['x']])).toEqual(['x']);
expect(combineUserData([undefined])).toBeUndefined();
expect(combineUserData([['x'], { a: 1 }])).toEqual(['x']);
});
});
describe('unionItems', () => {
it('identifies items by content id or URL', () => {
expect(
unionItems(
[{ id: 1, title: 'Server' }, { url: 'http://a' }],
[{ id: 1, title: 'Browser' }, { url: 'http://b' }]
)
).toEqual([
{ id: 1, title: 'Server' },
{ url: 'http://a' },
{ url: 'http://b' },
]);
});
});
libs/services/src/lib/server-storage/server-migration-merge.ts
+60
View File
@@ -0,0 +1,60 @@
/**
* How the move to the server combines user data that two places hold for
* one playlist: the server's value comes first and wins, the browser only
* adds. Lists are unioned, maps keep the first value of each key.
*/
export function combineUserData(values: readonly unknown[]): unknown {
const present = values.filter((value) => value !== undefined);
if (present.length === 0) {
return undefined;
}
if (present.every(Array.isArray)) {
return unionItems(...(present as unknown[][]));
}
if (present.every(isPlainObject)) {
const merged: Record<string, unknown> = {};
for (const value of present as Record<string, unknown>[]) {
for (const [key, entry] of Object.entries(value)) {
if (!(key in merged)) {
merged[key] = entry;
}
}
}
return merged;
}
return present[0];
}
/** Items in order of first appearance, without duplicates. */
export function unionItems<T>(...lists: readonly (readonly T[])[]): T[] {
const seen = new Set<string>();
const result: T[] = [];
for (const list of lists) {
for (const item of list) {
const key = itemKey(item);
if (!seen.has(key)) {
seen.add(key);
result.push(item);
}
}
}
return result;
}
/** Content ids or URLs identify favorites and history entries. */
function itemKey(item: unknown): string {
if (!isPlainObject(item)) {
return `value:${JSON.stringify(item)}`;
}
for (const field of ['id', 'stream_id', 'series_id', 'movie_id', 'url']) {
const value = item[field];
if (typeof value === 'string' || typeof value === 'number') {
return `${field}:${value}`;
}
}
return `json:${JSON.stringify(item)}`;
}
function isPlainObject(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object' && !Array.isArray(value);
}
libs/services/src/lib/server-storage/server-migration.service.spec.ts
+74 -3
View File
@@ -18,6 +18,7 @@ describe('ServerMigrationService', () => {
let session: {
fetchStatus: jest.Mock;
documentStore: {
dbGetAppPlaylist: jest.Mock;
dbGetAppPlaylistMetas: jest.Mock;
dbUpsertAppPlaylist: jest.Mock;
};
@@ -29,6 +30,7 @@ describe('ServerMigrationService', () => {
session = {
fetchStatus: jest.fn().mockResolvedValue({ playlists: 0 }),
documentStore: {
dbGetAppPlaylist: jest.fn().mockResolvedValue(null),
dbGetAppPlaylistMetas: jest.fn().mockResolvedValue([]),
dbUpsertAppPlaylist: jest
.fn()
@@ -106,7 +108,7 @@ describe('ServerMigrationService', () => {
expect(migration.browserDataMoved()).toBe(true);
});
it('merges into the server copy and moves Xtream data under its id', async () => {
it('merges into the server copy and combines Xtream data under its id', async () => {
session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([
{
_id: 'server-x',
@@ -131,7 +133,7 @@ describe('ServerMigrationService', () => {
);
localStorage.setItem(
'xtream-recent-items',
JSON.stringify({ x: ['vod:1'] })
JSON.stringify({ x: ['vod:1', 'vod:9'] })
);
const result = await TestBed.inject(
@@ -142,16 +144,85 @@ describe('ServerMigrationService', () => {
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'] },
],
[
'PUT',
'/user-state/xtream-recent-items/server-x',
{ value: ['vod:9', 'vod:1'] },
],
]);
});
it('combines browser duplicates of one source in one write', async () => {
TestBed.overrideProvider(NgxIndexedDBService, {
useValue: {
getAll: () =>
of([
browserPlaylists[1],
{ ...browserPlaylists[1], _id: 'x2' },
]),
},
});
localStorage.setItem(
'xtream-favorites',
JSON.stringify({ x: ['live:1'], x2: ['live:2'] })
);
const result = await TestBed.inject(
ServerMigrationService
).moveBrowserData();
expect(result).toEqual({ imported: 1, merged: 1, failed: 0 });
expect(session.api.sendJson.mock.calls).toEqual([
[
'PUT',
'/user-state/xtream-favorites/x',
{ value: ['live:1', 'live:2'] },
],
]);
});
it("adds a browser document's favorites to the matching server playlist", async () => {
TestBed.overrideProvider(NgxIndexedDBService, {
useValue: {
getAll: () =>
of([
{
...browserPlaylists[0],
favorites: ['http://a', 'http://b'],
},
]),
},
});
session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([
{ _id: 'server-m', title: 'Family', url: browserPlaylists[0].url },
]);
const stored = {
_id: 'server-m',
title: 'Family',
url: browserPlaylists[0].url,
favorites: ['http://a'],
recentlyViewed: [],
};
session.documentStore.dbGetAppPlaylist.mockResolvedValue(stored);
const result = await TestBed.inject(
ServerMigrationService
).moveBrowserData();
expect(result).toEqual({ imported: 0, merged: 1, failed: 0 });
expect(session.documentStore.dbUpsertAppPlaylist).toHaveBeenCalledWith({
...stored,
favorites: ['http://a', 'http://b'],
recentlyViewed: [],
});
});
it('keeps the move on offer when a playlist could not be uploaded', async () => {
session.documentStore.dbUpsertAppPlaylist
.mockRejectedValueOnce(new Error('413'))
libs/services/src/lib/server-storage/server-migration.service.ts
+83 -38
View File
@@ -7,6 +7,7 @@ import {
type Playlist,
type ServerUserStateResponse,
} from '@iptvnator/shared/interfaces';
import { combineUserData, unionItems } from './server-migration-merge';
import { planServerMigration } from './server-migration-plan';
import { SERVER_STORAGE_CONFIG } from './server-storage.config';
import { ServerStorageSession } from './server-storage-session';
@@ -25,19 +26,14 @@ 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. 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.
* merged into the server's copy, adding the browser's favorites and
* history to it. The legacy Xtream favorites, history, collections and
* playback positions follow under the server id, combined with what is
* there (`combineUserData`). Browser data is never deleted, and the move
* can run again after a failure.
*/
@Injectable({ providedIn: 'root' })
export class ServerMigrationService {
@@ -85,29 +81,34 @@ export class ServerMigrationService {
this.readBrowserPlaylists(),
this.session.documentStore.dbGetAppPlaylistMetas(),
]);
const targets = new Map<string, MigrationTarget>();
const byId = new Map(
browser.map((playlist) => [playlist._id, playlist])
);
/** Browser playlist id → its id on the server. */
const targets = new Map<string, string>();
let imported = 0;
let merged = 0;
let failed = 0;
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(
step.playlist
);
imported += 1;
targets.set(step.playlist._id, {
serverId: step.playlist._id,
uploaded: true,
});
if (step.kind === 'merge') {
const playlist = byId.get(step.browserId);
if (playlist) {
await this.mergeDocumentUserData(
playlist,
step.serverId
);
}
merged += 1;
targets.set(step.browserId, step.serverId);
} else {
await this.session.documentStore.dbUpsertAppPlaylist(
step.playlist
);
imported += 1;
targets.set(step.playlist._id, step.playlist._id);
}
} catch {
failed += 1;
}
@@ -122,9 +123,44 @@ export class ServerMigrationService {
return { imported, merged, failed };
}
/** Copies the legacy maps; returns how many entries failed. */
/** Adds a browser document's favorites and history to the server's. */
private async mergeDocumentUserData(
playlist: Playlist,
serverId: string
): Promise<void> {
const favorites = playlist.favorites ?? [];
const recentlyViewed = playlist.recentlyViewed ?? [];
if (favorites.length === 0 && recentlyViewed.length === 0) {
return;
}
const stored =
await this.session.documentStore.dbGetAppPlaylist(serverId);
if (!stored) {
throw new Error('The server playlist disappeared');
}
const next = {
...stored,
favorites: unionItems(stored.favorites ?? [], favorites),
recentlyViewed: unionItems(
stored.recentlyViewed ?? [],
recentlyViewed
),
};
if (
next.favorites.length !== (stored.favorites ?? []).length ||
next.recentlyViewed.length !== (stored.recentlyViewed ?? []).length
) {
await this.session.documentStore.dbUpsertAppPlaylist(next);
}
}
/**
* Copies the legacy maps, one write per server entry: browser copies
* of one source are combined with each other and with the server's
* value. Returns how many writes failed.
*/
private async copyXtreamUserData(
targets: ReadonlyMap<string, MigrationTarget>
targets: ReadonlyMap<string, string>
): Promise<number> {
if (targets.size === 0) {
return 0;
@@ -135,20 +171,29 @@ export class ServerMigrationService {
);
let failed = 0;
for (const namespace of SERVER_USER_STATE_NAMESPACES) {
const legacy = parseMap(readLocal(namespace));
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)
) {
const onServer: Record<string, unknown> = entries[namespace] ?? {};
const incoming = new Map<string, unknown[]>();
for (const [browserId, value] of Object.entries(
parseMap(readLocal(namespace))
)) {
const serverId = targets.get(browserId);
if (serverId) {
incoming.set(serverId, [
...(incoming.get(serverId) ?? []),
value,
]);
}
}
for (const [serverId, values] of incoming) {
const current = onServer[serverId];
const value = combineUserData([current, ...values]);
if (JSON.stringify(value) === JSON.stringify(current)) {
continue;
}
try {
await this.session.api.sendJson(
'PUT',
`/user-state/${namespace}/${encodeURIComponent(target.serverId)}`,
`/user-state/${namespace}/${encodeURIComponent(serverId)}`,
{ value }
);
} catch {
libs/services/src/lib/server-storage/server-storage-session.ts
+4 -2
View File
@@ -8,7 +8,10 @@ import {
clearServerAccessToken,
readServerAccessToken,
} from '../server-access-token';
import { ServerStorageStatusService } from '../server-storage-status.service';
import {
SERVER_SIGN_IN_PATH,
ServerStorageStatusService,
} from '../server-storage-status.service';
import { ServerApiClient, type ServerApiError } from './server-api-client';
import { ServerPlaylistDocumentStore } from './server-playlist-document-store';
import { SERVER_STORAGE_CONFIG } from './server-storage.config';
@@ -16,7 +19,6 @@ import { ServerUserDataStorage } from './server-user-data-storage';
const RETRY_INTERVAL_MS = 30_000;
const STARTUP_HYDRATE_WAIT_MS = 3_000;
export const SERVER_SIGN_IN_PATH = '/server-sign-in';
/** Server mode for this page: created only when the probe found server storage. */
@Injectable({ providedIn: 'root' })
libs/services/src/lib/server-storage/server-user-data-storage.spec.ts
+68
View File
@@ -191,4 +191,72 @@ describe('ServerUserDataStorage', () => {
await second;
expect(order).toEqual(['start ["a"]', 'end first', 'start ["b"]']);
});
it('keeps a failed write for the next flush and through a re-read', async () => {
const { api, storage, onWriteFailure } = setup();
await storage.hydrate();
api.sendJson.mockRejectedValueOnce(new Error('offline'));
storage.setItem(
'xtream-favorites',
JSON.stringify({ p1: ['live:1', 'live:2'] })
);
await storage.flush();
expect(onWriteFailure).toHaveBeenCalled();
// Back online: the re-read sends it first, so the server has it.
api.getJson.mockResolvedValueOnce({
entries: { 'xtream-favorites': { p1: ['live:1', 'live:2'] } },
});
await storage.hydrate();
expect(api.sendJson).toHaveBeenLastCalledWith(
'PUT',
'/user-state/xtream-favorites/p1',
{ value: ['live:1', 'live:2'] }
);
expect(JSON.parse(storage.getItem('xtream-favorites') ?? '')).toEqual({
p1: ['live:1', 'live:2'],
});
});
it('keeps an unsent edit when a re-read succeeds while writes fail', async () => {
const { api, storage } = setup();
await storage.hydrate();
api.sendJson.mockRejectedValue(new Error('500'));
storage.setItem(
'xtream-favorites',
JSON.stringify({ p1: ['live:1', 'live:2'] })
);
await storage.hydrate();
expect(JSON.parse(storage.getItem('xtream-favorites') ?? '')).toEqual({
p1: ['live:1', 'live:2'],
});
});
it('holds keepalive writes until a travelling batch finished', async () => {
const { api, storage } = setup();
await storage.hydrate();
let finish: () => void = () => undefined;
api.sendJson.mockReturnValueOnce(
new Promise<void>((resolve) => (finish = resolve))
);
storage.setItem('xtream-favorites', JSON.stringify({ p1: ['a'] }));
const travelling = storage.flush();
await settle();
storage.setItem('xtream-favorites', JSON.stringify({ p1: ['b'] }));
storage.flushWithKeepalive();
expect(api.sendKeepalive).not.toHaveBeenCalled();
finish();
await travelling;
await settle();
expect(api.sendKeepalive).toHaveBeenCalledWith(
'PUT',
'/user-state/xtream-favorites/p1',
{ value: ['b'] }
);
});
});
libs/services/src/lib/server-storage/server-user-data-storage.ts
+51 -12
View File
@@ -38,6 +38,7 @@ export class ServerUserDataStorage implements UserDataStorage {
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();
private batchesInFlight = 0;
/** Edit counter per `namespace|playlistId`, for hydrate to keep them. */
private readonly edits = new Map<string, number>();
private editCount = 0;
@@ -74,8 +75,8 @@ export class ServerUserDataStorage implements UserDataStorage {
/**
* 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.
* 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.
*/
async hydrate(): Promise<void> {
// Edits from here on may miss the read: queued, or still travelling.
@@ -90,9 +91,15 @@ export class ServerUserDataStorage implements UserDataStorage {
const current = parseMap(
this.local.getItem(SERVER_CACHE_PREFIX + namespace) ?? '{}'
);
const unsaved = new Set(this.pending.keys());
for (const [key, edit] of this.edits) {
if (edit > editsBefore) {
unsaved.add(key);
}
}
for (const key of unsaved) {
const [editedNamespace, playlistId] = splitEditKey(key);
if (edit <= editsBefore || editedNamespace !== namespace) {
if (editedNamespace !== namespace) {
continue;
}
if (playlistId in current) {
@@ -116,9 +123,18 @@ export class ServerUserDataStorage implements UserDataStorage {
/** 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)))
);
const run = this.sending.then(async () => {
const batch = this.takePending();
if (batch.length === 0) {
return;
}
this.batchesInFlight += 1;
try {
await Promise.all(batch.map((entry) => this.send(entry)));
} finally {
this.batchesInFlight -= 1;
}
});
this.sending = run.then(
() => undefined,
() => undefined
@@ -126,14 +142,31 @@ export class ServerUserDataStorage implements UserDataStorage {
return this.sending;
}
/**
* Page hide: `keepalive` requests survive the page. They go out at once
* unless a batch is still travelling, which they must not overtake.
*/
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 });
const entries = this.takePending();
if (entries.length === 0) {
return;
}
const sendAll = () => {
for (const entry of entries) {
const path = entryPath(entry);
if (entry.value === undefined) {
this.api.sendKeepalive('DELETE', path);
} else {
this.api.sendKeepalive('PUT', path, {
value: entry.value,
});
}
}
};
if (this.batchesInFlight === 0) {
sendAll();
} else {
this.sending = this.sending.then(sendAll);
}
}
@@ -198,6 +231,12 @@ export class ServerUserDataStorage implements UserDataStorage {
});
}
} catch (error) {
// Keep it for the next flush (the offline retry or the next
// edit) unless a newer value for the entry is already queued.
const key = `${entry.namespace}|${entry.playlistId}`;
if (!this.pending.has(key)) {
this.pending.set(key, entry);
}
this.onWriteFailure(error);
}
}