mirror of
https://github.com/4gray/iptvnator.git
synced 2026-10-14 13:16:16 -08:00
fix(server-storage): durable per-page journal, conflicts for edits on other tabs' changes, lock readability
- Unconfirmed writes and early edits are saved in localStorage per page load; the tab remembers its page load in sessionStorage, so a reload takes them over, and another tab replays a closed tab's writes once its record is stale. Live tabs renew their record every 30 s. - An edit made on another tab's unsent change, and an entry a re-read keeps from the cache, are sent on a version no entry has, so the write conflicts and is replayed instead of carrying another tab's change past a later removal. - Moving browser data counts the parental-lock step as failed while the lock store cannot be read, so the move stays on offer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
487c98ab96
commit
b1b05600d6
12 files changed
+387
-101
No files matched your search
@@ -33,14 +33,14 @@ describe('ServerMigrationService', () => {
|
||||
|
||||
const emptyLocks = () => ({ xtream: [], stalker: [], m3u: [] });
|
||||
let parentalLock: {
|
||||
initialize: jest.Mock;
|
||||
ensureLocksReadable: jest.Mock;
|
||||
locksFor: jest.Mock;
|
||||
replacePlaylistLocks: jest.Mock;
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
parentalLock = {
|
||||
initialize: jest.fn().mockResolvedValue(undefined),
|
||||
ensureLocksReadable: jest.fn().mockResolvedValue(true),
|
||||
locksFor: jest.fn(() => emptyLocks()),
|
||||
replacePlaylistLocks: jest.fn().mockResolvedValue(true),
|
||||
};
|
||||
@@ -367,4 +367,23 @@ describe('ServerMigrationService', () => {
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
it('keeps the move on offer while parental locks cannot be read', async () => {
|
||||
session.documentStore.dbGetAppPlaylistMetas.mockResolvedValue([
|
||||
{
|
||||
_id: 'server-x',
|
||||
title: 'Provider',
|
||||
serverUrl: 'http://panel.test',
|
||||
username: 'user',
|
||||
},
|
||||
]);
|
||||
parentalLock.ensureLocksReadable.mockResolvedValue(false);
|
||||
|
||||
const result = await TestBed.inject(
|
||||
ServerMigrationService
|
||||
).moveBrowserData();
|
||||
|
||||
expect(result.failed).toBeGreaterThan(0);
|
||||
expect(parentalLock.replacePlaylistLocks).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -279,7 +279,14 @@ export class ServerMigrationService {
|
||||
private async moveParentalLocks(
|
||||
targets: ReadonlyMap<string, string>
|
||||
): Promise<number> {
|
||||
await this.parentalLock.initialize();
|
||||
const moving = [...targets].filter(([from, to]) => from !== to);
|
||||
// An unreadable lock store reads as empty: keep the move on offer.
|
||||
if (
|
||||
moving.length > 0 &&
|
||||
!(await this.parentalLock.ensureLocksReadable())
|
||||
) {
|
||||
return moving.length;
|
||||
}
|
||||
let failed = 0;
|
||||
for (const [browserId, serverId] of targets) {
|
||||
const from = this.parentalLock.locksFor(browserId);
|
||||
|
||||
@@ -128,9 +128,10 @@ describe('ServerStorageSession', () => {
|
||||
});
|
||||
|
||||
it('sends no write before the reload checks the instance', async () => {
|
||||
// A write the last visit could not send, for instance i-1.
|
||||
sessionStorage.setItem(
|
||||
'iptvnator-server:unsent-writes:i-1',
|
||||
// A write this tab's last page load could not send.
|
||||
sessionStorage.setItem('iptvnator-server:page:i-1', 'last');
|
||||
localStorage.setItem(
|
||||
'iptvnator-server:unsent-writes:i-1:last',
|
||||
JSON.stringify([
|
||||
{
|
||||
namespace: 'xtream-favorites',
|
||||
|
||||
@@ -139,6 +139,7 @@ export class ServerStorageSession {
|
||||
this.userData.flushWithKeepalive()
|
||||
);
|
||||
setInterval(() => {
|
||||
this.userData.touch();
|
||||
if (this.status.state() === 'offline') {
|
||||
void this.refresh();
|
||||
}
|
||||
|
||||
@@ -6,20 +6,25 @@ import {
|
||||
import type { UserDataStorage } from '../user-data-storage';
|
||||
|
||||
/**
|
||||
* User-data writes the server has not confirmed, kept across reloads: a
|
||||
* failed write, or a page-hide write whose answer the page did not live to
|
||||
* see, is sent again on the next visit instead of being replaced by the
|
||||
* server's older value.
|
||||
* User-data writes the server has not confirmed, kept in `localStorage`
|
||||
* per page load: a failed write, or a page-hide write whose answer the page
|
||||
* did not live to see, is sent again after a reload (by the same tab) or
|
||||
* after the tab closed (by another, see `orphanedPages`), instead of being
|
||||
* replaced by the 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';
|
||||
/** The tab's current page load, in the tab's own storage. */
|
||||
export const PAGE_KEY = 'iptvnator-server:page';
|
||||
|
||||
const writesKey = (instanceId: string) => `${UNSENT_WRITES_KEY}:${instanceId}`;
|
||||
const writesKey = (instanceId: string, page: string) =>
|
||||
`${UNSENT_WRITES_KEY}:${instanceId}:${page}`;
|
||||
/**
|
||||
* Which entries each open tab has not had confirmed yet, so a re-read in
|
||||
* one tab keeps the cache values another tab is still sending. A record
|
||||
* older than `UNCONFIRMED_TTL_MS` belongs to a tab that is gone.
|
||||
* Which entries each page load has not had confirmed yet, so a re-read in
|
||||
* one tab keeps the cache values another tab is still sending. A live tab
|
||||
* renews its record; one older than `UNCONFIRMED_TTL_MS` belongs to a tab
|
||||
* that is gone. The records also list whose saved writes exist.
|
||||
*/
|
||||
export const UNCONFIRMED_KEY = 'iptvnator-server:unconfirmed';
|
||||
export const UNCONFIRMED_TTL_MS = 10 * 60 * 1000;
|
||||
@@ -28,8 +33,8 @@ const unconfirmedKey = (instanceId: string) =>
|
||||
|
||||
/** Edits made before the first read of a visit, kept as before → after. */
|
||||
export const DEFERRED_EDITS_KEY = 'iptvnator-server:deferred-edits';
|
||||
const deferredKey = (instanceId: string) =>
|
||||
`${DEFERRED_EDITS_KEY}:${instanceId}`;
|
||||
const deferredKey = (instanceId: string, page: string) =>
|
||||
`${DEFERRED_EDITS_KEY}:${instanceId}:${page}`;
|
||||
|
||||
/**
|
||||
* Another database behind this address (a new volume, another backend):
|
||||
@@ -46,9 +51,11 @@ export function claimCache(local: UserDataStorage, instanceId: string): void {
|
||||
local.removeItem(`iptvnator-server:${namespace}`);
|
||||
}
|
||||
if (owner) {
|
||||
for (const page of Object.keys(readRegistry(local, owner))) {
|
||||
local.removeItem(writesKey(owner, page));
|
||||
local.removeItem(deferredKey(owner, page));
|
||||
}
|
||||
local.removeItem(unconfirmedKey(owner));
|
||||
local.removeItem(writesKey(owner));
|
||||
local.removeItem(deferredKey(owner));
|
||||
}
|
||||
local.setItem(CACHE_OWNER_KEY, instanceId);
|
||||
} catch {
|
||||
@@ -80,13 +87,63 @@ interface StoredWrite {
|
||||
readonly prior?: { readonly before?: unknown; readonly after?: unknown };
|
||||
}
|
||||
|
||||
/**
|
||||
* Records `page` as the tab's current page load and returns the one
|
||||
* before it: a reload, whose saved writes this page takes over.
|
||||
*/
|
||||
export function swapPage(
|
||||
tabStore: UserDataStorage,
|
||||
instanceId: string,
|
||||
page: string
|
||||
): string | null {
|
||||
try {
|
||||
const previous = tabStore.getItem(`${PAGE_KEY}:${instanceId}`);
|
||||
tabStore.setItem(`${PAGE_KEY}:${instanceId}`, page);
|
||||
return previous;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Page loads whose record went stale: closed tabs nobody sends for. */
|
||||
export function orphanedPages(
|
||||
local: UserDataStorage,
|
||||
instanceId: string,
|
||||
page: string,
|
||||
now = Date.now()
|
||||
): string[] {
|
||||
return Object.entries(readRegistry(local, instanceId))
|
||||
.filter(
|
||||
([owner, record]) =>
|
||||
owner !== page &&
|
||||
!(now - Number(record?.at) < UNCONFIRMED_TTL_MS)
|
||||
)
|
||||
.map(([owner]) => owner);
|
||||
}
|
||||
|
||||
/** Forgets a page load's saved writes, early edits and record. */
|
||||
export function removePage(
|
||||
local: UserDataStorage,
|
||||
instanceId: string,
|
||||
page: string
|
||||
): void {
|
||||
try {
|
||||
local.removeItem(writesKey(instanceId, page));
|
||||
local.removeItem(deferredKey(instanceId, page));
|
||||
} catch {
|
||||
// Storage unavailable: nothing was saved there either.
|
||||
}
|
||||
markUnconfirmed(local, instanceId, page, []);
|
||||
}
|
||||
|
||||
export function readUnsentWrites(
|
||||
local: UserDataStorage,
|
||||
instanceId: string
|
||||
instanceId: string,
|
||||
page: string
|
||||
): UnsentWrite[] {
|
||||
let stored: unknown;
|
||||
try {
|
||||
stored = JSON.parse(local.getItem(writesKey(instanceId)) ?? '[]');
|
||||
stored = JSON.parse(local.getItem(writesKey(instanceId, page)) ?? '[]');
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
@@ -121,11 +178,12 @@ export function readUnsentWrites(
|
||||
export function saveUnsentWrites(
|
||||
local: UserDataStorage,
|
||||
instanceId: string,
|
||||
page: string,
|
||||
writes: readonly UnsentWrite[]
|
||||
): void {
|
||||
try {
|
||||
if (writes.length === 0) {
|
||||
local.removeItem(writesKey(instanceId));
|
||||
local.removeItem(writesKey(instanceId, page));
|
||||
return;
|
||||
}
|
||||
const stored: StoredWrite[] = writes.map(
|
||||
@@ -140,7 +198,7 @@ export function saveUnsentWrites(
|
||||
: { prior: { before: prior.before, after: prior.after } }),
|
||||
})
|
||||
);
|
||||
local.setItem(writesKey(instanceId), JSON.stringify(stored));
|
||||
local.setItem(writesKey(instanceId, page), JSON.stringify(stored));
|
||||
} catch {
|
||||
// Storage unavailable or full: the queue stays in memory only.
|
||||
}
|
||||
@@ -170,11 +228,14 @@ export interface SavedDeferredEdit {
|
||||
|
||||
export function readDeferredEdits(
|
||||
local: UserDataStorage,
|
||||
instanceId: string
|
||||
instanceId: string,
|
||||
page: string
|
||||
): SavedDeferredEdit[] {
|
||||
let stored: unknown;
|
||||
try {
|
||||
stored = JSON.parse(local.getItem(deferredKey(instanceId)) ?? '[]');
|
||||
stored = JSON.parse(
|
||||
local.getItem(deferredKey(instanceId, page)) ?? '[]'
|
||||
);
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
@@ -204,15 +265,16 @@ export function readDeferredEdits(
|
||||
export function saveDeferredEdits(
|
||||
local: UserDataStorage,
|
||||
instanceId: string,
|
||||
page: string,
|
||||
edits: readonly SavedDeferredEdit[]
|
||||
): void {
|
||||
try {
|
||||
if (edits.length === 0) {
|
||||
local.removeItem(deferredKey(instanceId));
|
||||
local.removeItem(deferredKey(instanceId, page));
|
||||
return;
|
||||
}
|
||||
local.setItem(
|
||||
deferredKey(instanceId),
|
||||
deferredKey(instanceId, page),
|
||||
JSON.stringify(
|
||||
edits.map((edit) => ({
|
||||
namespace: edit.namespace,
|
||||
|
||||
@@ -114,3 +114,18 @@ function isMapOrMissing(
|
||||
function isListOrMissing(value: unknown): value is unknown[] | undefined {
|
||||
return value === undefined || Array.isArray(value);
|
||||
}
|
||||
|
||||
/** A saved write's edits, its `prior` first, applied to `server`. */
|
||||
export function replayWrite(
|
||||
server: unknown,
|
||||
write: {
|
||||
readonly before?: unknown;
|
||||
readonly value: unknown;
|
||||
readonly prior?: { readonly before: unknown; readonly after: unknown };
|
||||
}
|
||||
): unknown {
|
||||
const base = write.prior
|
||||
? applyEdit(server, write.prior.before, write.prior.after)
|
||||
: server;
|
||||
return applyEdit(base, write.before, write.value);
|
||||
}
|
||||
@@ -30,6 +30,13 @@ export interface PriorEdit {
|
||||
readonly edit: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* A version no entry ever has: a write based on what the cache shows but
|
||||
* the server may not have (another tab's unsent change) always conflicts
|
||||
* and is replayed on the server's value.
|
||||
*/
|
||||
export const UNKNOWN_TAG = '~unknown';
|
||||
|
||||
/** `If-Match` / `If-None-Match` for a write based on `tag`. */
|
||||
export function preconditionHeaders(
|
||||
tag: string | null | undefined
|
||||
|
||||
@@ -10,8 +10,9 @@ import {
|
||||
type PendingEntry,
|
||||
preconditionHeaders,
|
||||
type PriorEdit,
|
||||
UNKNOWN_TAG,
|
||||
} from './server-user-data-entries';
|
||||
import { applyEdit } from './server-user-data-edit';
|
||||
import { replayWrite } from './server-user-data-edit';
|
||||
import { newestWrites } from './server-unsent-writes';
|
||||
|
||||
const FLUSH_DELAY_MS = 500;
|
||||
@@ -118,15 +119,20 @@ export class UserDataWriteQueue {
|
||||
/**
|
||||
* Queues an edit (`before` → `value`). A newer edit of an entry still
|
||||
* queued joins it; one behind a travelling write keeps its own edit,
|
||||
* carries the write's as `prior` and is based on the write's tag.
|
||||
* carries the write's as `prior` and is based on the write's tag. A
|
||||
* `foreign` edit was made on another tab's unsent change: its write
|
||||
* must conflict and be replayed.
|
||||
*/
|
||||
queue(
|
||||
namespace: ServerUserStateNamespace,
|
||||
playlistId: string,
|
||||
value: unknown,
|
||||
before: unknown,
|
||||
tag = this.tagFor(namespace, playlistId),
|
||||
prior?: Omit<PriorEdit, 'edit'>
|
||||
base: {
|
||||
readonly tag?: string | null;
|
||||
readonly prior?: Omit<PriorEdit, 'edit'>;
|
||||
readonly foreign?: boolean;
|
||||
} = {}
|
||||
): void {
|
||||
const key = `${namespace}|${playlistId}`;
|
||||
const queued = this.pending.get(key);
|
||||
@@ -152,11 +158,18 @@ export class UserDataWriteQueue {
|
||||
},
|
||||
};
|
||||
} else {
|
||||
const tag =
|
||||
base.tag !== undefined
|
||||
? base.tag
|
||||
: this.tagFor(namespace, playlistId);
|
||||
entry = { namespace, playlistId, value, edit, before, tag };
|
||||
if (prior) {
|
||||
entry = { ...entry, prior: { ...prior, edit: 0 } };
|
||||
if (base.prior) {
|
||||
entry = { ...entry, prior: { ...base.prior, edit: 0 } };
|
||||
}
|
||||
}
|
||||
if (base.foreign) {
|
||||
entry = { ...entry, tag: UNKNOWN_TAG };
|
||||
}
|
||||
this.pending.set(key, entry);
|
||||
this.edits.set(key, edit);
|
||||
}
|
||||
@@ -410,7 +423,12 @@ export class UserDataWriteQueue {
|
||||
const key = entryKey(entry);
|
||||
const next = this.pending.get(key);
|
||||
if (next?.prior?.edit === entry.edit) {
|
||||
this.pending.set(key, { ...next, prior: undefined, tag });
|
||||
// A foreign edit stays unknown: its baseline is not this value.
|
||||
this.pending.set(key, {
|
||||
...next,
|
||||
prior: undefined,
|
||||
tag: next.tag === UNKNOWN_TAG ? UNKNOWN_TAG : tag,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -432,7 +450,7 @@ export class UserDataWriteQueue {
|
||||
if (!latest) {
|
||||
return;
|
||||
}
|
||||
const merged = replay(current.value, latest);
|
||||
const merged = replayWrite(current.value, latest);
|
||||
this.hooks.show(entry.namespace, entry.playlistId, merged);
|
||||
if (
|
||||
latest === entry &&
|
||||
@@ -450,11 +468,3 @@ export class UserDataWriteQueue {
|
||||
this.scheduleFlush();
|
||||
}
|
||||
}
|
||||
|
||||
/** An entry's edits, its `prior` first, applied to the server's value. */
|
||||
function replay(server: unknown, entry: PendingEntry): unknown {
|
||||
const base = entry.prior
|
||||
? applyEdit(server, entry.prior.before, entry.prior.after)
|
||||
: server;
|
||||
return applyEdit(base, entry.before, entry.value);
|
||||
}
|
||||
@@ -12,6 +12,18 @@ function memoryStorage() {
|
||||
}
|
||||
|
||||
/** Lets chained promises run without advancing the fake timers. */
|
||||
/** The saved writes of the one page load that has any, for `instanceId`. */
|
||||
function savedWrites(
|
||||
local: ReturnType<typeof memoryStorage>,
|
||||
instanceId = 'default'
|
||||
): unknown[] {
|
||||
const prefix = `iptvnator-server:unsent-writes:${instanceId}:`;
|
||||
const key = [...local.values.keys()].find((name) =>
|
||||
name.startsWith(prefix)
|
||||
);
|
||||
return key ? JSON.parse(local.values.get(key) ?? '[]') : [];
|
||||
}
|
||||
|
||||
async function settle(): Promise<void> {
|
||||
for (let i = 0; i < 10; i += 1) {
|
||||
await Promise.resolve();
|
||||
@@ -470,13 +482,7 @@ describe('ServerUserDataStorage', () => {
|
||||
);
|
||||
await first.storage.flush();
|
||||
expect(first.storage.hasUnsentWrites()).toBe(true);
|
||||
expect(
|
||||
JSON.parse(
|
||||
first.local.values.get(
|
||||
'iptvnator-server:unsent-writes:default'
|
||||
) ?? '[]'
|
||||
)
|
||||
).toMatchObject([
|
||||
expect(savedWrites(first.local)).toMatchObject([
|
||||
{
|
||||
namespace: 'xtream-favorites',
|
||||
playlistId: 'p1',
|
||||
@@ -506,9 +512,7 @@ describe('ServerUserDataStorage', () => {
|
||||
{}
|
||||
);
|
||||
expect(reloaded.hasUnsentWrites()).toBe(false);
|
||||
expect(
|
||||
first.local.values.has('iptvnator-server:unsent-writes:default')
|
||||
).toBe(false);
|
||||
expect(savedWrites(first.local)).toEqual([]);
|
||||
});
|
||||
|
||||
it('saves a write still travelling when the page hides', async () => {
|
||||
@@ -521,12 +525,7 @@ describe('ServerUserDataStorage', () => {
|
||||
await settle();
|
||||
storage.flushWithKeepalive();
|
||||
|
||||
expect(
|
||||
JSON.parse(
|
||||
local.values.get('iptvnator-server:unsent-writes:default') ??
|
||||
'[]'
|
||||
)
|
||||
).toMatchObject([
|
||||
expect(savedWrites(local)).toMatchObject([
|
||||
{ namespace: 'xtream-favorites', playlistId: 'p1', value: ['x'] },
|
||||
]);
|
||||
});
|
||||
@@ -536,7 +535,11 @@ describe('ServerUserDataStorage', () => {
|
||||
local.setItem('iptvnator-server:cache-owner', 'old-db');
|
||||
local.setItem('iptvnator-server:xtream-favorites', '{"p1":["a"]}');
|
||||
local.setItem(
|
||||
'iptvnator-server:unsent-writes:old-db',
|
||||
'iptvnator-server:unconfirmed:old-db',
|
||||
JSON.stringify({ tab: { at: 0, keys: ['xtream-favorites|p1'] } })
|
||||
);
|
||||
local.setItem(
|
||||
'iptvnator-server:unsent-writes:old-db:tab',
|
||||
JSON.stringify([
|
||||
{
|
||||
namespace: 'xtream-favorites',
|
||||
@@ -557,7 +560,8 @@ describe('ServerUserDataStorage', () => {
|
||||
expect(local.values.has('iptvnator-server:xtream-favorites')).toBe(
|
||||
false
|
||||
);
|
||||
expect(local.values.has('iptvnator-server:unsent-writes:old-db')).toBe(
|
||||
expect(savedWrites(local, 'old-db')).toEqual([]);
|
||||
expect(local.values.has('iptvnator-server:unconfirmed:old-db')).toBe(
|
||||
false
|
||||
);
|
||||
expect(local.values.get('iptvnator-server:cache-owner')).toBe('new-db');
|
||||
@@ -582,12 +586,7 @@ describe('ServerUserDataStorage', () => {
|
||||
await settle();
|
||||
|
||||
expect(storage.hasUnsentWrites()).toBe(true);
|
||||
expect(
|
||||
JSON.parse(
|
||||
local.values.get('iptvnator-server:unsent-writes:default') ??
|
||||
'[]'
|
||||
)
|
||||
).toMatchObject([
|
||||
expect(savedWrites(local)).toMatchObject([
|
||||
{ namespace: 'xtream-favorites', playlistId: 'p1', value: ['new'] },
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -6,17 +6,28 @@ import {
|
||||
} from '@iptvnator/shared/interfaces';
|
||||
import type { UserDataStorage } from '../user-data-storage';
|
||||
import type { ServerApiClient } from './server-api-client';
|
||||
import { parseMap, splitEditKey } from './server-user-data-entries';
|
||||
import { type DeferredEdit, refreshedNamespace } from './server-user-data-edit';
|
||||
import {
|
||||
parseMap,
|
||||
splitEditKey,
|
||||
UNKNOWN_TAG,
|
||||
} from './server-user-data-entries';
|
||||
import {
|
||||
type DeferredEdit,
|
||||
refreshedNamespace,
|
||||
replayWrite,
|
||||
} from './server-user-data-edit';
|
||||
import { UserDataWriteQueue } from './server-user-data-queue';
|
||||
import {
|
||||
claimCache,
|
||||
markUnconfirmed,
|
||||
orphanedPages,
|
||||
othersUnconfirmed,
|
||||
readDeferredEdits,
|
||||
readUnsentWrites,
|
||||
removePage,
|
||||
saveDeferredEdits,
|
||||
saveUnsentWrites,
|
||||
swapPage,
|
||||
} from './server-unsent-writes';
|
||||
|
||||
/** Server-mode cache keys; the legacy unprefixed keys are never touched. */
|
||||
@@ -45,7 +56,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
*/
|
||||
private readonly deferred = new Map<string, DeferredEdit>();
|
||||
private hydrated = false;
|
||||
/** This tab, in the shared registry of unconfirmed entries. */
|
||||
/** This page load: names its saved writes and its registry record. */
|
||||
private readonly tabId = Math.random().toString(36).slice(2);
|
||||
/** Cache values that did not fit into browser storage. */
|
||||
private readonly overflow = new Map<ServerUserStateNamespace, string>();
|
||||
@@ -59,11 +70,10 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
/** Scopes the cache and saved writes to one server database. */
|
||||
private readonly instanceId = 'default',
|
||||
/**
|
||||
* Where this tab keeps what the server has not confirmed: its own
|
||||
* (`sessionStorage` in the app), so tabs never clear each other's,
|
||||
* and a reload of the tab finds it again.
|
||||
* The tab's own storage (`sessionStorage` in the app): remembers
|
||||
* its page load, so a reload takes over the writes it saved.
|
||||
*/
|
||||
private readonly tabStore: UserDataStorage = local
|
||||
tabStore: UserDataStorage = local
|
||||
) {
|
||||
this.writes = new UserDataWriteQueue(api, {
|
||||
changed: () => this.saveUnsent(),
|
||||
@@ -72,22 +82,81 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
writeFailed: onWriteFailure,
|
||||
});
|
||||
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(tabStore, instanceId)) {
|
||||
const previous = swapPage(tabStore, instanceId, this.tabId);
|
||||
if (previous) {
|
||||
this.takeOver(previous);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A reload of this tab: the writes its last page load could not confirm
|
||||
* go out with the first flush, and the first read keeps them.
|
||||
*/
|
||||
private takeOver(page: string): void {
|
||||
for (const write of readUnsentWrites(
|
||||
this.local,
|
||||
this.instanceId,
|
||||
page
|
||||
)) {
|
||||
this.writes.queue(
|
||||
write.namespace,
|
||||
write.playlistId,
|
||||
write.value,
|
||||
write.before,
|
||||
write.tag,
|
||||
write.prior
|
||||
{ tag: write.tag, prior: write.prior }
|
||||
);
|
||||
}
|
||||
this.unsentSaved = this.writes.hasUnsent();
|
||||
for (const edit of readDeferredEdits(tabStore, instanceId)) {
|
||||
for (const edit of readDeferredEdits(
|
||||
this.local,
|
||||
this.instanceId,
|
||||
page
|
||||
)) {
|
||||
this.deferred.set(`${edit.namespace}|${edit.playlistId}`, edit);
|
||||
}
|
||||
// Saved under this page load before the old one is forgotten.
|
||||
this.saveDeferred();
|
||||
this.saveUnsent();
|
||||
this.publishUnconfirmed();
|
||||
removePage(this.local, this.instanceId, page);
|
||||
}
|
||||
|
||||
/**
|
||||
* Saved writes and early edits of closed tabs: replayed as this tab's
|
||||
* own edits on what it shows now, then forgotten there.
|
||||
*/
|
||||
private adoptClosedTabs(): void {
|
||||
for (const page of orphanedPages(
|
||||
this.local,
|
||||
this.instanceId,
|
||||
this.tabId
|
||||
)) {
|
||||
const edits = [
|
||||
...readDeferredEdits(this.local, this.instanceId, page).map(
|
||||
(edit) => ({ ...edit, value: edit.after })
|
||||
),
|
||||
...readUnsentWrites(this.local, this.instanceId, page),
|
||||
];
|
||||
for (const edit of edits) {
|
||||
const before = this.baseline(edit.namespace);
|
||||
const after = { ...before };
|
||||
const value = replayWrite(before[edit.playlistId], edit);
|
||||
if (value === undefined) {
|
||||
delete after[edit.playlistId];
|
||||
} else {
|
||||
after[edit.playlistId] = value;
|
||||
}
|
||||
this.writeCache(edit.namespace, JSON.stringify(after));
|
||||
this.queueChanges(edit.namespace, after, before);
|
||||
}
|
||||
removePage(this.local, this.instanceId, page);
|
||||
}
|
||||
}
|
||||
|
||||
/** Tells other tabs this one is alive and still sending its writes. */
|
||||
touch(): void {
|
||||
if (this.hasUnsentWrites()) {
|
||||
this.publishUnconfirmed();
|
||||
}
|
||||
}
|
||||
|
||||
/** Whether a write is queued, being retried, or still travelling. */
|
||||
@@ -177,9 +246,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
this.deferred.delete(key);
|
||||
}
|
||||
}
|
||||
saveDeferredEdits(this.tabStore, this.instanceId, [
|
||||
...this.deferred.values(),
|
||||
]);
|
||||
this.saveDeferred();
|
||||
this.writes.drop(matches, playlistId);
|
||||
for (const namespace of SERVER_USER_STATE_NAMESPACES) {
|
||||
const map = this.baseline(namespace);
|
||||
@@ -232,7 +299,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
// next write must conflict and be replayed on the server's.
|
||||
const fresh = { ...tags[namespace] };
|
||||
for (const playlistId of keepCached) {
|
||||
delete fresh[playlistId];
|
||||
fresh[playlistId] = UNKNOWN_TAG;
|
||||
}
|
||||
this.writes.setTags(namespace, fresh);
|
||||
}
|
||||
@@ -250,9 +317,10 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
entries[edit.namespace]?.[edit.playlistId]
|
||||
);
|
||||
}
|
||||
saveDeferredEdits(this.tabStore, this.instanceId, []);
|
||||
this.saveDeferred();
|
||||
this.writes.scheduleFlush();
|
||||
}
|
||||
this.adoptClosedTabs();
|
||||
}
|
||||
|
||||
private readCache(namespace: ServerUserStateNamespace): string | null {
|
||||
@@ -314,13 +382,18 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
next: Record<string, unknown>,
|
||||
previous: Record<string, unknown>
|
||||
): void {
|
||||
// The cache shows other tabs' unsent changes too.
|
||||
const foreign = this.hydrated
|
||||
? othersUnconfirmed(this.local, this.instanceId, this.tabId)
|
||||
: new Set<string>();
|
||||
const change = (playlistId: string, value: unknown) => {
|
||||
if (this.hydrated) {
|
||||
this.writes.queue(
|
||||
namespace,
|
||||
playlistId,
|
||||
value,
|
||||
previous[playlistId]
|
||||
previous[playlistId],
|
||||
{ foreign: foreign.has(`${namespace}|${playlistId}`) }
|
||||
);
|
||||
return;
|
||||
}
|
||||
@@ -333,9 +406,7 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
before: earlier ? earlier.before : previous[playlistId],
|
||||
after: value,
|
||||
});
|
||||
saveDeferredEdits(this.tabStore, this.instanceId, [
|
||||
...this.deferred.values(),
|
||||
]);
|
||||
this.saveDeferred();
|
||||
this.publishUnconfirmed();
|
||||
};
|
||||
for (const [playlistId, value] of Object.entries(next)) {
|
||||
@@ -360,11 +431,17 @@ export class ServerUserDataStorage implements UserDataStorage {
|
||||
if (writes.length === 0 && !this.unsentSaved) {
|
||||
return;
|
||||
}
|
||||
saveUnsentWrites(this.tabStore, this.instanceId, writes);
|
||||
saveUnsentWrites(this.local, this.instanceId, this.tabId, writes);
|
||||
this.unsentSaved = writes.length > 0;
|
||||
this.publishUnconfirmed();
|
||||
}
|
||||
|
||||
private saveDeferred(): void {
|
||||
saveDeferredEdits(this.local, this.instanceId, this.tabId, [
|
||||
...this.deferred.values(),
|
||||
]);
|
||||
}
|
||||
|
||||
private publishUnconfirmed(): void {
|
||||
markUnconfirmed(this.local, this.instanceId, this.tabId, [
|
||||
...this.writes.unconfirmedKeys(),
|
||||
|
||||
@@ -452,4 +452,86 @@ describe('ServerUserDataStorage with other devices', () => {
|
||||
[...(server.state['xtream-favorites']['p1'] as string[])].sort()
|
||||
).toEqual(['S', 'X', 'Y', 'Z']);
|
||||
});
|
||||
|
||||
it('keeps a removal made after another tab built on the unsent item', async () => {
|
||||
const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
|
||||
const shared = memoryStorage();
|
||||
const tab = () =>
|
||||
new ServerUserDataStorage(
|
||||
server.api as never,
|
||||
shared,
|
||||
jest.fn(),
|
||||
'default',
|
||||
memoryStorage()
|
||||
);
|
||||
const tabA = tab();
|
||||
const tabB = tab();
|
||||
await tabA.hydrate();
|
||||
await tabB.hydrate();
|
||||
|
||||
tabA.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
|
||||
// Tab B sees X in the shared cache and adds Y next to it.
|
||||
tabB.setItem(
|
||||
'xtream-favorites',
|
||||
JSON.stringify({ p1: [...favorites(tabB).p1, 'Y'] })
|
||||
);
|
||||
// Tab A takes X back before anything was sent.
|
||||
tabA.setItem(
|
||||
'xtream-favorites',
|
||||
JSON.stringify({
|
||||
p1: favorites(tabA).p1.filter((id: string) => id !== 'X'),
|
||||
})
|
||||
);
|
||||
for (let round = 0; round < 4; round += 1) {
|
||||
await tabB.flush();
|
||||
await tabA.flush();
|
||||
}
|
||||
|
||||
expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'Y']);
|
||||
});
|
||||
|
||||
it('sends the saved writes of a tab that was closed offline', async () => {
|
||||
const server = fakeServer({ 'xtream-favorites': { p1: ['S'] } });
|
||||
const shared = memoryStorage();
|
||||
const offline = {
|
||||
...server.api,
|
||||
sendJson: jest
|
||||
.fn()
|
||||
.mockRejectedValue(new ServerApiError(0, 'network', 'offline')),
|
||||
};
|
||||
const closed = new ServerUserDataStorage(
|
||||
offline as never,
|
||||
shared,
|
||||
jest.fn(),
|
||||
'default',
|
||||
memoryStorage()
|
||||
);
|
||||
await closed.hydrate();
|
||||
closed.setItem('xtream-favorites', JSON.stringify({ p1: ['S', 'X'] }));
|
||||
await closed.flush();
|
||||
// The tab is gone: its record stops being renewed.
|
||||
const registry = 'iptvnator-server:unconfirmed:default';
|
||||
const records = JSON.parse(shared.values.get(registry) ?? '{}');
|
||||
for (const record of Object.values(records)) {
|
||||
(record as { at: number }).at = 0;
|
||||
}
|
||||
shared.setItem(registry, JSON.stringify(records));
|
||||
|
||||
const next = new ServerUserDataStorage(
|
||||
server.api as never,
|
||||
shared,
|
||||
jest.fn(),
|
||||
'default',
|
||||
memoryStorage()
|
||||
);
|
||||
await next.hydrate();
|
||||
await next.flush();
|
||||
|
||||
expect(server.state['xtream-favorites']['p1']).toEqual(['S', 'X']);
|
||||
expect(
|
||||
[...shared.values.keys()].filter((key) =>
|
||||
key.startsWith('iptvnator-server:unsent-writes:')
|
||||
)
|
||||
).toEqual([]);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user