mirror of
https://github.com/4gray/iptvnator.git
synced 2026-10-14 05:06:16 -08:00
fix(server-storage): locks follow merged playlists, typed legacy Xtream ids, instance-scoped cache
- Moving a playlist under another server id takes its parental locks along (they stay in the browser). - Bare legacy Xtream ids in each copy are typed from that copy's own snapshots before favorites, history and snapshots are combined, so a live channel and a movie sharing an id both survive. - Writes still travelling are saved on page hide too; saved writes and the cache belong to one server database (cache owner = instanceId). - A late successful answer no longer turns unauthorized back to online. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
abae207d02
commit
21f640ccc1
11 files changed
+388
-46
No files matched your search
@@ -200,8 +200,12 @@ settings and guards read them in every build.
|
||||
each change sends only the playlist entries that differ, debounced, with a
|
||||
`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,
|
||||
including a refused `keepalive` one, stays queued for the next flush and
|
||||
survives a reload under `iptvnator-server:unsent-writes`, and
|
||||
including a refused `keepalive` one, stays queued for the next flush, and
|
||||
the page-hide flush saves every unconfirmed write (queued or travelling)
|
||||
under `iptvnator-server:unsent-writes:<instanceId>` for the next visit.
|
||||
Cache and saved writes belong to one server database
|
||||
(`iptvnator-server:cache-owner`); another one behind the same address
|
||||
starts clean. Also,
|
||||
re-reads run one at a time and keep entries edited while they were in
|
||||
flight or not yet saved. A re-read whose result does not fit the cache
|
||||
keeps the old cache and baseline together. A response body cut off after
|
||||
@@ -213,7 +217,8 @@ settings and guards read them in every build.
|
||||
- **Connection state.** `ServerStorageStatusService` is `online`, `offline`
|
||||
(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
|
||||
`unauthorized` (a 401 clears the stored token; only a reload after a new
|
||||
token leaves it, so a late answer cannot undo sign-out). 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).
|
||||
@@ -236,9 +241,12 @@ settings and guards read them in every build.
|
||||
document's favorites and history (M3U favorites name channels by stream
|
||||
URL; older ones by a channel id that is random per import and is
|
||||
translated through the URL; nothing is dropped), and the legacy Xtream maps follow under
|
||||
the server id; `combineUserData` unions lists (playback positions by
|
||||
content type and Xtream id, renamed to the server playlist) and merges
|
||||
maps, the server's value winning on a conflict. Browser data is never
|
||||
the server id; each copy's bare legacy Xtream ids are typed from its own
|
||||
snapshots first (`typeLegacyXtreamRefs`), then `combineUserData` unions
|
||||
lists (playback positions by content type and Xtream id, renamed to the
|
||||
server playlist) and merges maps, the server's value winning on a
|
||||
conflict. Parental locks stay in the browser but move to the server
|
||||
playlist id with it. Browser data is never
|
||||
deleted, the move does not start while user-data writes are unsent, and
|
||||
the marker is written only when nothing failed, so Settings → Server
|
||||
keeps offering a partial move.
|
||||
|
||||
@@ -42,5 +42,9 @@ describe('ServerStorageStatusService', () => {
|
||||
|
||||
status.reportUnauthorized();
|
||||
expect(status.state()).toBe('unauthorized');
|
||||
|
||||
// A late answer to a request sent with the old token.
|
||||
status.reportSuccess();
|
||||
expect(status.state()).toBe('unauthorized');
|
||||
});
|
||||
});
|
||||
@@ -31,7 +31,9 @@ export class ServerStorageStatusService {
|
||||
}
|
||||
|
||||
reportSuccess(): void {
|
||||
if (!this.isActive()) {
|
||||
// A late answer to a request sent with the old token must not undo
|
||||
// sign-out; only a reload after a new token leaves this state.
|
||||
if (!this.isActive() || this.stateSignal() === 'unauthorized') {
|
||||
return;
|
||||
}
|
||||
this.stateSignal.set('online');
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
import { combineUserData } from './server-migration-merge';
|
||||
import { typeLegacyXtreamRefs } from './server-migration-xtream';
|
||||
|
||||
describe('typeLegacyXtreamRefs', () => {
|
||||
it('types bare ids from the same copy snapshots', () => {
|
||||
expect(
|
||||
typeLegacyXtreamRefs({
|
||||
'xtream-favorites': [42, 'movie:7', 99],
|
||||
'xtream-recent-items': [{ id: 42, viewedAt: 't' }],
|
||||
'xtream-collection-items': {
|
||||
'42': { xtream_id: 42, type: 'live', title: 'News' },
|
||||
'movie:7': { xtream_id: 7, type: 'movie' },
|
||||
},
|
||||
})
|
||||
).toEqual({
|
||||
'xtream-favorites': ['live:42', 'movie:7', 99],
|
||||
'xtream-recent-items': [{ id: 'live:42', viewedAt: 't' }],
|
||||
'xtream-collection-items': {
|
||||
'live:42': { xtream_id: 42, type: 'live', title: 'News' },
|
||||
'movie:7': { xtream_id: 7, type: 'movie' },
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('keeps a live and a movie favorite that share an id apart', () => {
|
||||
const browser = typeLegacyXtreamRefs({
|
||||
'xtream-favorites': [42],
|
||||
'xtream-collection-items': {
|
||||
'42': { xtream_id: 42, type: 'live' },
|
||||
},
|
||||
});
|
||||
const server = typeLegacyXtreamRefs({
|
||||
'xtream-favorites': [42],
|
||||
'xtream-collection-items': {
|
||||
'42': { xtream_id: 42, type: 'movie' },
|
||||
},
|
||||
});
|
||||
|
||||
expect(
|
||||
combineUserData([
|
||||
server['xtream-favorites'],
|
||||
browser['xtream-favorites'],
|
||||
])
|
||||
).toEqual(['movie:42', 'live:42']);
|
||||
expect(
|
||||
combineUserData([
|
||||
server['xtream-collection-items'],
|
||||
browser['xtream-collection-items'],
|
||||
])
|
||||
).toEqual({
|
||||
'movie:42': { xtream_id: 42, type: 'movie' },
|
||||
'live:42': { xtream_id: 42, type: 'live' },
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,53 @@
|
||||
import type { ServerUserStateNamespace } from '@iptvnator/shared/interfaces';
|
||||
|
||||
/** One playlist's Xtream user data from one place (browser or server). */
|
||||
export type XtreamUserData = Partial<Record<ServerUserStateNamespace, unknown>>;
|
||||
|
||||
const SNAPSHOTS = 'xtream-collection-items';
|
||||
|
||||
/**
|
||||
* Older favorites and history name content by its bare Xtream id, which
|
||||
* repeats across live, movie and series. Before two copies are combined,
|
||||
* each one's bare ids become typed keys (`live:42`) using that copy's own
|
||||
* collection snapshots, the same evidence the PWA reads them with; ids
|
||||
* without a snapshot stay as they are.
|
||||
*/
|
||||
export function typeLegacyXtreamRefs(data: XtreamUserData): XtreamUserData {
|
||||
const snapshots = isRecord(data[SNAPSHOTS]) ? data[SNAPSHOTS] : {};
|
||||
const typed = (key: unknown): unknown => {
|
||||
if (typeof key !== 'number') {
|
||||
return key;
|
||||
}
|
||||
const snapshot = snapshots[String(key)];
|
||||
return isRecord(snapshot) &&
|
||||
snapshot['xtream_id'] === key &&
|
||||
typeof snapshot['type'] === 'string'
|
||||
? `${snapshot['type']}:${key}`
|
||||
: key;
|
||||
};
|
||||
|
||||
const result: XtreamUserData = { ...data };
|
||||
const favorites = data['xtream-favorites'];
|
||||
if (Array.isArray(favorites)) {
|
||||
result['xtream-favorites'] = favorites.map(typed);
|
||||
}
|
||||
const recent = data['xtream-recent-items'];
|
||||
if (Array.isArray(recent)) {
|
||||
result['xtream-recent-items'] = recent.map((item) =>
|
||||
isRecord(item) ? { ...item, id: typed(item['id']) } : item
|
||||
);
|
||||
}
|
||||
if (isRecord(data[SNAPSHOTS])) {
|
||||
const rekeyed: Record<string, unknown> = {};
|
||||
for (const [key, snapshot] of Object.entries(snapshots)) {
|
||||
rekeyed[String(/^\d+$/.test(key) ? typed(Number(key)) : key)] =
|
||||
snapshot;
|
||||
}
|
||||
result[SNAPSHOTS] = rekeyed;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return !!value && typeof value === 'object' && !Array.isArray(value);
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
import { TestBed } from '@angular/core/testing';
|
||||
import { NgxIndexedDBService } from 'ngx-indexed-db';
|
||||
import { of } from 'rxjs';
|
||||
import { ParentalLockService } from '../parental-lock/parental-lock.service';
|
||||
import { SERVER_STORAGE_CONFIG } from './server-storage.config';
|
||||
import { ServerStorageSession } from './server-storage-session';
|
||||
import { ServerMigrationService } from './server-migration.service';
|
||||
@@ -30,7 +31,19 @@ describe('ServerMigrationService', () => {
|
||||
};
|
||||
};
|
||||
|
||||
const emptyLocks = () => ({ xtream: [], stalker: [], m3u: [] });
|
||||
let parentalLock: {
|
||||
initialize: jest.Mock;
|
||||
locksFor: jest.Mock;
|
||||
replacePlaylistLocks: jest.Mock;
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
parentalLock = {
|
||||
initialize: jest.fn().mockResolvedValue(undefined),
|
||||
locksFor: jest.fn(() => emptyLocks()),
|
||||
replacePlaylistLocks: jest.fn().mockResolvedValue(true),
|
||||
};
|
||||
session = {
|
||||
fetchStatus: jest.fn().mockResolvedValue({ playlists: 0 }),
|
||||
documentStore: {
|
||||
@@ -57,6 +70,7 @@ describe('ServerMigrationService', () => {
|
||||
useValue: { getAll: () => of(browserPlaylists) },
|
||||
},
|
||||
{ provide: ServerStorageSession, useValue: session },
|
||||
{ provide: ParentalLockService, useValue: parentalLock },
|
||||
{
|
||||
provide: SERVER_STORAGE_CONFIG,
|
||||
useValue: {
|
||||
@@ -308,4 +322,45 @@ describe('ServerMigrationService', () => {
|
||||
expect(result).toEqual({ imported: 0, merged: 0, failed: 2 });
|
||||
expect(session.api.sendJson).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('takes parental locks along to the server playlist id', async () => {
|
||||
session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([
|
||||
{
|
||||
_id: 'server-x',
|
||||
title: 'Provider',
|
||||
serverUrl: 'http://panel.test',
|
||||
username: 'user',
|
||||
},
|
||||
]);
|
||||
parentalLock.locksFor.mockImplementation((id: string) =>
|
||||
id === 'x'
|
||||
? {
|
||||
xtream: [{ categoryType: 'live', xtreamId: 5 }],
|
||||
stalker: [],
|
||||
m3u: [],
|
||||
}
|
||||
: {
|
||||
xtream: [{ categoryType: 'vod', xtreamId: 9 }],
|
||||
stalker: [],
|
||||
m3u: [],
|
||||
}
|
||||
);
|
||||
|
||||
const result = await TestBed.inject(
|
||||
ServerMigrationService
|
||||
).moveBrowserData();
|
||||
|
||||
expect(result.failed).toBe(0);
|
||||
expect(parentalLock.replacePlaylistLocks).toHaveBeenCalledWith(
|
||||
'server-x',
|
||||
{
|
||||
xtream: [
|
||||
{ categoryType: 'vod', xtreamId: 9 },
|
||||
{ categoryType: 'live', xtreamId: 5 },
|
||||
],
|
||||
stalker: [],
|
||||
m3u: [],
|
||||
}
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -14,6 +14,11 @@ import {
|
||||
unionItems,
|
||||
} from './server-migration-merge';
|
||||
import { planServerMigration } from './server-migration-plan';
|
||||
import {
|
||||
typeLegacyXtreamRefs,
|
||||
type XtreamUserData,
|
||||
} from './server-migration-xtream';
|
||||
import { ParentalLockService } from '../parental-lock/parental-lock.service';
|
||||
import { SERVER_STORAGE_CONFIG } from './server-storage.config';
|
||||
import { ServerStorageSession } from './server-storage-session';
|
||||
|
||||
@@ -44,6 +49,7 @@ export interface ServerMigrationResult {
|
||||
export class ServerMigrationService {
|
||||
private readonly dbService = inject(NgxIndexedDBService);
|
||||
private readonly session = inject(ServerStorageSession);
|
||||
private readonly parentalLock = inject(ParentalLockService);
|
||||
private readonly instanceId = inject(SERVER_STORAGE_CONFIG).instance
|
||||
.instanceId;
|
||||
|
||||
@@ -131,6 +137,7 @@ export class ServerMigrationService {
|
||||
}
|
||||
|
||||
failed += await this.copyXtreamUserData(targets);
|
||||
failed += await this.moveParentalLocks(targets);
|
||||
await this.session.userData.hydrate();
|
||||
if (failed === 0) {
|
||||
// A partial move stays on offer, in Settings → Server.
|
||||
@@ -184,9 +191,10 @@ export class ServerMigrationService {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* Copies the legacy maps, one write per server entry: each copy's
|
||||
* legacy ids are typed first, then the 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, string>
|
||||
@@ -198,25 +206,47 @@ export class ServerMigrationService {
|
||||
await this.session.api.getJson<ServerUserStateResponse>(
|
||||
'/user-state'
|
||||
);
|
||||
let failed = 0;
|
||||
for (const namespace of SERVER_USER_STATE_NAMESPACES) {
|
||||
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) ?? []),
|
||||
retargetUserData(value, browserId, serverId),
|
||||
]);
|
||||
const legacy = new Map(
|
||||
SERVER_USER_STATE_NAMESPACES.map((namespace) => [
|
||||
namespace,
|
||||
parseMap(readLocal(namespace)),
|
||||
])
|
||||
);
|
||||
const copies = new Map<string, XtreamUserData[]>();
|
||||
for (const [browserId, serverId] of targets) {
|
||||
const copy: XtreamUserData = {};
|
||||
for (const [namespace, map] of legacy) {
|
||||
if (browserId in map) {
|
||||
copy[namespace] = retargetUserData(
|
||||
map[browserId],
|
||||
browserId,
|
||||
serverId
|
||||
);
|
||||
}
|
||||
}
|
||||
for (const [serverId, values] of incoming) {
|
||||
const current = onServer[serverId];
|
||||
const value = combineUserData([current, ...values]);
|
||||
if (JSON.stringify(value) === JSON.stringify(current)) {
|
||||
if (Object.keys(copy).length > 0) {
|
||||
copies.set(serverId, [...(copies.get(serverId) ?? []), copy]);
|
||||
}
|
||||
}
|
||||
|
||||
let failed = 0;
|
||||
for (const [serverId, browserCopies] of copies) {
|
||||
const stored: XtreamUserData = {};
|
||||
for (const namespace of SERVER_USER_STATE_NAMESPACES) {
|
||||
const value = entries[namespace]?.[serverId];
|
||||
if (value !== undefined) {
|
||||
stored[namespace] = value;
|
||||
}
|
||||
}
|
||||
const typed = [stored, ...browserCopies].map(typeLegacyXtreamRefs);
|
||||
for (const namespace of SERVER_USER_STATE_NAMESPACES) {
|
||||
const value = combineUserData(
|
||||
typed.map((copy) => copy[namespace])
|
||||
);
|
||||
if (
|
||||
value === undefined ||
|
||||
JSON.stringify(value) === JSON.stringify(stored[namespace])
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
@@ -233,6 +263,40 @@ export class ServerMigrationService {
|
||||
return failed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parental locks stay in this browser, keyed by playlist id: a playlist
|
||||
* merged under another id takes its locks along, or its locked
|
||||
* categories would show after the move. Returns how many failed.
|
||||
*/
|
||||
private async moveParentalLocks(
|
||||
targets: ReadonlyMap<string, string>
|
||||
): Promise<number> {
|
||||
await this.parentalLock.initialize();
|
||||
let failed = 0;
|
||||
for (const [browserId, serverId] of targets) {
|
||||
const from = this.parentalLock.locksFor(browserId);
|
||||
if (
|
||||
browserId === serverId ||
|
||||
from.xtream.length + from.stalker.length + from.m3u.length === 0
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const to = this.parentalLock.locksFor(serverId);
|
||||
const moved = await this.parentalLock.replacePlaylistLocks(
|
||||
serverId,
|
||||
{
|
||||
xtream: unionItems(to.xtream, from.xtream),
|
||||
stalker: unionItems(to.stalker, from.stalker),
|
||||
m3u: unionItems(to.m3u, from.m3u),
|
||||
}
|
||||
);
|
||||
if (!moved) {
|
||||
failed += 1;
|
||||
}
|
||||
}
|
||||
return failed;
|
||||
}
|
||||
|
||||
private readBrowserPlaylists(): Promise<Playlist[]> {
|
||||
return firstValueFrom(
|
||||
this.dbService.getAll<Playlist>(DbStores.Playlists)
|
||||
|
||||
@@ -42,7 +42,8 @@ export class ServerStorageSession {
|
||||
this.api,
|
||||
// Lazy: a browser that denies storage must not stop the app here.
|
||||
createBrowserUserDataStorage(),
|
||||
() => this.announceWriteFailure()
|
||||
() => this.announceWriteFailure(),
|
||||
this.config.instance.instanceId ?? undefined
|
||||
);
|
||||
|
||||
constructor() {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import {
|
||||
isServerUserStateNamespace,
|
||||
SERVER_USER_STATE_NAMESPACES,
|
||||
type ServerUserStateNamespace,
|
||||
} from '@iptvnator/shared/interfaces';
|
||||
import type { UserDataStorage } from '../user-data-storage';
|
||||
@@ -11,6 +12,33 @@ import type { UserDataStorage } from '../user-data-storage';
|
||||
* server's older value.
|
||||
*/
|
||||
export const UNSENT_WRITES_KEY = 'iptvnator-server:unsent-writes';
|
||||
/** Which server database the cached user data belongs to. */
|
||||
export const CACHE_OWNER_KEY = 'iptvnator-server:cache-owner';
|
||||
|
||||
const writesKey = (instanceId: string) => `${UNSENT_WRITES_KEY}:${instanceId}`;
|
||||
|
||||
/**
|
||||
* Another database behind this address (a new volume, another backend):
|
||||
* its cache and unsent writes belong to the old one and must not be sent
|
||||
* to or shown for this one.
|
||||
*/
|
||||
export function claimCache(local: UserDataStorage, instanceId: string): void {
|
||||
try {
|
||||
const owner = local.getItem(CACHE_OWNER_KEY);
|
||||
if (owner === instanceId) {
|
||||
return;
|
||||
}
|
||||
for (const namespace of SERVER_USER_STATE_NAMESPACES) {
|
||||
local.removeItem(`iptvnator-server:${namespace}`);
|
||||
}
|
||||
if (owner) {
|
||||
local.removeItem(writesKey(owner));
|
||||
}
|
||||
local.setItem(CACHE_OWNER_KEY, instanceId);
|
||||
} catch {
|
||||
// Storage unavailable: nothing cached to mix up.
|
||||
}
|
||||
}
|
||||
|
||||
export interface UnsentWrite {
|
||||
readonly namespace: ServerUserStateNamespace;
|
||||
@@ -26,10 +54,13 @@ interface StoredWrite {
|
||||
readonly remove?: true;
|
||||
}
|
||||
|
||||
export function readUnsentWrites(local: UserDataStorage): UnsentWrite[] {
|
||||
export function readUnsentWrites(
|
||||
local: UserDataStorage,
|
||||
instanceId: string
|
||||
): UnsentWrite[] {
|
||||
let stored: unknown;
|
||||
try {
|
||||
stored = JSON.parse(local.getItem(UNSENT_WRITES_KEY) ?? '[]');
|
||||
stored = JSON.parse(local.getItem(writesKey(instanceId)) ?? '[]');
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
@@ -54,11 +85,12 @@ export function readUnsentWrites(local: UserDataStorage): UnsentWrite[] {
|
||||
|
||||
export function saveUnsentWrites(
|
||||
local: UserDataStorage,
|
||||
instanceId: string,
|
||||
writes: readonly UnsentWrite[]
|
||||
): void {
|
||||
try {
|
||||
if (writes.length === 0) {
|
||||
local.removeItem(UNSENT_WRITES_KEY);
|
||||
local.removeItem(writesKey(instanceId));
|
||||
return;
|
||||
}
|
||||
const stored: StoredWrite[] = writes.map(
|
||||
@@ -67,7 +99,7 @@ export function saveUnsentWrites(
|
||||
? { namespace, playlistId, remove: true }
|
||||
: { namespace, playlistId, value }
|
||||
);
|
||||
local.setItem(UNSENT_WRITES_KEY, JSON.stringify(stored));
|
||||
local.setItem(writesKey(instanceId), JSON.stringify(stored));
|
||||
} catch {
|
||||
// Storage unavailable or full: the queue stays in memory only.
|
||||
}
|
||||
|
||||
@@ -417,7 +417,9 @@ describe('ServerUserDataStorage', () => {
|
||||
);
|
||||
await first.storage.flush();
|
||||
expect(first.storage.hasUnsentWrites()).toBe(true);
|
||||
expect(first.local.values.get('iptvnator-server:unsent-writes')).toBe(
|
||||
expect(
|
||||
first.local.values.get('iptvnator-server:unsent-writes:default')
|
||||
).toBe(
|
||||
JSON.stringify([
|
||||
{
|
||||
namespace: 'xtream-favorites',
|
||||
@@ -448,8 +450,60 @@ describe('ServerUserDataStorage', () => {
|
||||
{ value: ['live:1', 'live:9'] }
|
||||
);
|
||||
expect(reloaded.hasUnsentWrites()).toBe(false);
|
||||
expect(first.local.values.has('iptvnator-server:unsent-writes')).toBe(
|
||||
expect(
|
||||
first.local.values.has('iptvnator-server:unsent-writes:default')
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it('saves a write still travelling when the page hides', async () => {
|
||||
const { api, local, storage } = setup();
|
||||
await storage.hydrate();
|
||||
api.sendJson.mockReturnValueOnce(new Promise(() => undefined));
|
||||
|
||||
storage.setItem('xtream-favorites', JSON.stringify({ p1: ['x'] }));
|
||||
void storage.flush();
|
||||
await settle();
|
||||
storage.flushWithKeepalive();
|
||||
|
||||
expect(
|
||||
JSON.parse(
|
||||
local.values.get('iptvnator-server:unsent-writes:default') ??
|
||||
'[]'
|
||||
)
|
||||
).toEqual([
|
||||
{ namespace: 'xtream-favorites', playlistId: 'p1', value: ['x'] },
|
||||
]);
|
||||
});
|
||||
|
||||
it('drops cache and saved writes of another server database', () => {
|
||||
const local = memoryStorage();
|
||||
local.setItem('iptvnator-server:cache-owner', 'old-db');
|
||||
local.setItem('iptvnator-server:xtream-favorites', '{"p1":["a"]}');
|
||||
local.setItem(
|
||||
'iptvnator-server:unsent-writes:old-db',
|
||||
JSON.stringify([
|
||||
{
|
||||
namespace: 'xtream-favorites',
|
||||
playlistId: 'p1',
|
||||
value: ['a'],
|
||||
},
|
||||
])
|
||||
);
|
||||
|
||||
const storage = new ServerUserDataStorage(
|
||||
{} as never,
|
||||
local,
|
||||
jest.fn(),
|
||||
'new-db'
|
||||
);
|
||||
|
||||
expect(storage.hasUnsentWrites()).toBe(false);
|
||||
expect(local.values.has('iptvnator-server:xtream-favorites')).toBe(
|
||||
false
|
||||
);
|
||||
expect(local.values.has('iptvnator-server:unsent-writes:old-db')).toBe(
|
||||
false
|
||||
);
|
||||
expect(local.values.get('iptvnator-server:cache-owner')).toBe('new-db');
|
||||
});
|
||||
});
|
||||
@@ -13,7 +13,11 @@ import {
|
||||
type PendingEntry,
|
||||
splitEditKey,
|
||||
} from './server-user-data-entries';
|
||||
import { readUnsentWrites, saveUnsentWrites } from './server-unsent-writes';
|
||||
import {
|
||||
claimCache,
|
||||
readUnsentWrites,
|
||||
saveUnsentWrites,
|
||||
} from './server-unsent-writes';
|
||||
|
||||
/** Server-mode cache keys; the legacy unprefixed keys are never touched. */
|
||||
export const SERVER_CACHE_PREFIX = 'iptvnator-server:';
|
||||
@@ -40,8 +44,8 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
/** Batches go out one at a time, so an older write never lands last. */
|
||||
private sending: Promise<void> = Promise.resolve();
|
||||
private writesInFlight = 0;
|
||||
/** Page-hide writes still travelling, by `namespace|playlistId`. */
|
||||
private readonly keepalive = new Map<string, PendingEntry>();
|
||||
/** Writes still travelling (normal or page-hide), by entry key. */
|
||||
private readonly inFlight = new Map<string, PendingEntry>();
|
||||
private unsentSaved = false;
|
||||
/** Re-reads run one at a time, so an older one never lands last. */
|
||||
private hydration: Promise<void> = Promise.resolve();
|
||||
@@ -52,11 +56,14 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
constructor(
|
||||
private readonly api: ServerApi,
|
||||
private readonly local: UserDataStorage,
|
||||
private readonly onWriteFailure: (error: unknown) => void
|
||||
private readonly onWriteFailure: (error: unknown) => void,
|
||||
/** Scopes the cache and saved writes to one server database. */
|
||||
private readonly instanceId = 'default'
|
||||
) {
|
||||
claimCache(local, instanceId);
|
||||
// Writes a previous visit could not confirm go out with the first
|
||||
// flush, and the first read keeps them.
|
||||
for (const write of readUnsentWrites(local)) {
|
||||
for (const write of readUnsentWrites(local, instanceId)) {
|
||||
this.queue(write.namespace, write.playlistId, write.value);
|
||||
}
|
||||
this.unsentSaved = this.pending.size > 0;
|
||||
@@ -64,7 +71,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
|
||||
/** Whether a write is queued, being retried, or still travelling. */
|
||||
hasUnsentWrites(): boolean {
|
||||
return this.pending.size > 0 || this.keepalive.size > 0;
|
||||
return this.pending.size > 0 || this.inFlight.size > 0;
|
||||
}
|
||||
|
||||
getItem(key: string): string | null {
|
||||
@@ -122,7 +129,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
);
|
||||
const unsaved = new Set([
|
||||
...this.pending.keys(),
|
||||
...this.keepalive.keys(),
|
||||
...this.inFlight.keys(),
|
||||
]);
|
||||
for (const [key, edit] of this.edits) {
|
||||
if (edit > editsBefore) {
|
||||
@@ -156,7 +163,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
if (
|
||||
edit <= editsBefore &&
|
||||
!this.pending.has(key) &&
|
||||
!this.keepalive.has(key)
|
||||
!this.inFlight.has(key)
|
||||
) {
|
||||
this.edits.delete(key);
|
||||
}
|
||||
@@ -170,6 +177,9 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
if (batch.length === 0) {
|
||||
return;
|
||||
}
|
||||
for (const entry of batch) {
|
||||
this.inFlight.set(entryKey(entry), entry);
|
||||
}
|
||||
this.writesInFlight += 1;
|
||||
try {
|
||||
await Promise.all(batch.map((entry) => this.send(entry)));
|
||||
@@ -192,6 +202,8 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
flushWithKeepalive(): void {
|
||||
const entries = this.takePending();
|
||||
if (entries.length === 0) {
|
||||
// Writes still travelling may be cut off with the page.
|
||||
this.saveUnsent();
|
||||
return;
|
||||
}
|
||||
const sendAll = async (): Promise<void> => {
|
||||
@@ -205,7 +217,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
}
|
||||
};
|
||||
for (const entry of entries) {
|
||||
this.keepalive.set(entryKey(entry), entry);
|
||||
this.inFlight.set(entryKey(entry), entry);
|
||||
}
|
||||
this.saveUnsent();
|
||||
// Idle: send synchronously, while the hiding page still runs.
|
||||
@@ -308,7 +320,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
const ok = await (entry.value === undefined
|
||||
? this.api.sendKeepalive('DELETE', path)
|
||||
: this.api.sendKeepalive('PUT', path, { value: entry.value }));
|
||||
this.keepalive.delete(entryKey(entry));
|
||||
this.inFlight.delete(entryKey(entry));
|
||||
if (!ok) {
|
||||
this.requeue(entry);
|
||||
}
|
||||
@@ -324,8 +336,10 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
value: entry.value,
|
||||
});
|
||||
}
|
||||
this.inFlight.delete(entryKey(entry));
|
||||
this.saveUnsent();
|
||||
} catch (error) {
|
||||
this.inFlight.delete(entryKey(entry));
|
||||
this.requeue(entry);
|
||||
this.saveUnsent();
|
||||
this.onWriteFailure(error);
|
||||
@@ -334,11 +348,11 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
|
||||
/** Persists what the server has not confirmed; skips idle rewrites. */
|
||||
private saveUnsent(): void {
|
||||
const writes = [...this.pending.values(), ...this.keepalive.values()];
|
||||
const writes = [...this.pending.values(), ...this.inFlight.values()];
|
||||
if (writes.length === 0 && !this.unsentSaved) {
|
||||
return;
|
||||
}
|
||||
saveUnsentWrites(this.local, writes);
|
||||
saveUnsentWrites(this.local, this.instanceId, writes);
|
||||
this.unsentSaved = writes.length > 0;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user