test(performance): add end-to-end Xtream benchmark harness (#1300)

* docs(performance): plan Xtream benchmark

* feat(xtream-mock-server): add deterministic 100k fixture

* style(xtream-mock-server): apply repository formatting

* fix(xtream-mock-server): harden performance fixture data

* feat(xtream-mock-server): add performance control plane

* docs(performance): correct Xtream capture plan

* fix(xtream-mock-server): harden performance controls

* fix(xtream-mock-server): harden control lifecycle

* feat(performance): add Xtream preload markers

* feat(performance): trace Xtream main phases

* feat(performance): mark Xtream store publications

* feat(performance): trace Xtream database phases

* feat(performance): trace Xtream delete cancellation

* feat(performance): capture Xtream phase attribution

* feat(performance): mark Sources Xtream refresh

* test(performance): define Xtream benchmark evidence contracts

* test(performance): add Xtream benchmark runner

* test(performance): surface failure evidence writes

* test(performance): align database read clock

* test(performance): preserve capture failure contracts
This commit is contained in:
4gray authored and GitHub committed 2026-07-28 08:08:07 +02:00
1 parent 5e725a06f3
commit a2fafcfc08
256 files changed
+43760 -883

No files matched your search

@@ -0,0 +1,110 @@
import type { AppDatabase } from '../database.types';
import { XTREAM_DATABASE_PERFORMANCE_PHASE } from '@iptvnator/shared/interfaces';
import { getCategories, saveCategories } from './category.operations';
import { createRecordingOperationPhaseCapture } from './performance-phase-capture.test-helpers';
function createDeferred<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((resolvePromise) => {
resolve = resolvePromise;
});
return { promise, resolve };
}
describe('category operation performance phases', () => {
it('keeps the categories read phase open until the query resolves', async () => {
const rows = [{ id: 1, name: 'News' }];
const pending = createDeferred<typeof rows>();
const orderBy = jest.fn(() => pending.promise);
const where = jest.fn(() => ({ orderBy }));
const from = jest.fn(() => ({ where }));
const db = {
select: jest.fn(() => ({ from })),
} as unknown as AppDatabase;
const recording = createRecordingOperationPhaseCapture();
const result = getCategories(
db,
'playlist-1',
'live',
recording.capture
);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CATEGORIES_READ,
},
]);
pending.resolve(rows);
await expect(result).resolves.toBe(rows);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CATEGORIES_READ,
},
{
boundary: 'end',
metadata: { itemCount: 1 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CATEGORIES_READ,
},
]);
});
it('captures normalization and the existing category write separately', async () => {
const existingWhere = jest.fn().mockResolvedValue([{ count: 0 }]);
const select = jest.fn(() => ({
from: jest.fn(() => ({ where: existingWhere })),
}));
const onConflictDoNothing = jest.fn().mockResolvedValue(undefined);
const values = jest.fn(() => ({ onConflictDoNothing }));
const insert = jest.fn(() => ({ values }));
const db = { insert, select } as unknown as AppDatabase;
const recording = createRecordingOperationPhaseCapture();
await expect(
saveCategories(
db,
'playlist-1',
[
{ category_name: 'News', category_id: '101' },
{ category_name: 'Broken', category_id: 'not-a-number' },
{ category_name: 'Sports', category_id: 102 },
],
'live',
[102],
recording.capture
)
).resolves.toEqual({ success: true });
expect(values).toHaveBeenCalledWith([
expect.objectContaining({ name: 'News', xtreamId: 101 }),
expect.objectContaining({
hidden: true,
name: 'Sports',
xtreamId: 102,
}),
]);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_CATEGORIES,
},
{
boundary: 'end',
metadata: { itemCount: 2 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_CATEGORIES,
},
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CATEGORIES_WRITE_TRANSACTIONS,
},
{
boundary: 'end',
metadata: { itemCount: 2 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CATEGORIES_WRITE_TRANSACTIONS,
},
]);
});
});
@@ -1,12 +1,16 @@
import { and, eq, inArray, sql } from 'drizzle-orm';
import * as schema from '@iptvnator/shared/database/schema';
import { XTREAM_DATABASE_PERFORMANCE_PHASE } from '@iptvnator/shared/interfaces';
import type { AppDatabase } from '../database.types';
import type { DatabaseOperationPerformancePhaseCapture } from './performance-phase-capture';
type XtreamCategoryInput = {
category_name: string;
category_id: string | number;
};
type XtreamCategoryValue = typeof schema.categories.$inferInsert;
// Category rows cross the DB-worker IPC boundary in the snake_case wire
// shape declared by XCategoryFromDb/XtreamCategoryFromDb. A bare select()
// would leak Drizzle's camelCase property names (xtreamId, playlistId)
@@ -29,6 +33,49 @@ function normalizeXtreamCategoryId(
return Number.isNaN(xtreamId) ? null : xtreamId;
}
function normalizeXtreamCategories(
playlistId: string,
categories: XtreamCategoryInput[],
type: 'live' | 'movies' | 'series',
hiddenCategoryXtreamIds?: number[]
): XtreamCategoryValue[] {
const hiddenSet = new Set(hiddenCategoryXtreamIds || []);
return categories.flatMap((category) => {
const xtreamId = normalizeXtreamCategoryId(category.category_id);
if (xtreamId === null) {
return [];
}
return [
{
playlistId,
name: category.category_name,
type,
xtreamId,
hidden: hiddenSet.has(xtreamId),
},
];
});
}
async function insertXtreamCategories(
db: AppDatabase,
values: XtreamCategoryValue[]
): Promise<void> {
await db
.insert(schema.categories)
.values(values)
.onConflictDoNothing({
target: [
schema.categories.playlistId,
schema.categories.type,
schema.categories.xtreamId,
],
});
}
export async function hasCategories(
db: AppDatabase,
playlistId: string,
@@ -50,13 +97,14 @@ export async function hasCategories(
export async function getCategories(
db: AppDatabase,
playlistId: string,
type: 'live' | 'movies' | 'series'
type: 'live' | 'movies' | 'series',
capturePhase?: DatabaseOperationPerformancePhaseCapture
) {
// Xtream categories are inserted once in provider order and existing
// xtream IDs are preserved, so row id order represents server order.
// If partial category re-inserts are added later, persist a provider
// sort index instead of relying on the insertion id.
return db
const query = db
.select(categoryWireShape)
.from(schema.categories)
.where(
@@ -67,6 +115,14 @@ export async function getCategories(
)
)
.orderBy(schema.categories.id);
return capturePhase
? capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CATEGORIES_READ,
async () => query,
(rows) => ({ itemCount: rows.length })
)
: query;
}
export async function saveCategories(
@@ -74,7 +130,8 @@ export async function saveCategories(
playlistId: string,
categories: XtreamCategoryInput[],
type: 'live' | 'movies' | 'series',
hiddenCategoryXtreamIds?: number[]
hiddenCategoryXtreamIds?: number[],
capturePhase?: DatabaseOperationPerformancePhaseCapture
): Promise<{ success: boolean }> {
if (!categories || categories.length === 0) {
return { success: true };
@@ -94,39 +151,38 @@ export async function saveCategories(
return { success: true };
}
const hiddenSet = new Set(hiddenCategoryXtreamIds || []);
const values = categories.flatMap((category) => {
const xtreamId = normalizeXtreamCategoryId(category.category_id);
if (xtreamId === null) {
return [];
}
return [
{
playlistId,
name: category.category_name,
type,
xtreamId,
hidden: hiddenSet.has(xtreamId),
},
];
});
const values = capturePhase
? capturePhase.captureSync(
XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_CATEGORIES,
() =>
normalizeXtreamCategories(
playlistId,
categories,
type,
hiddenCategoryXtreamIds
),
(result) => ({ itemCount: result.length })
)
: normalizeXtreamCategories(
playlistId,
categories,
type,
hiddenCategoryXtreamIds
);
if (values.length === 0) {
return { success: true };
}
await db
.insert(schema.categories)
.values(values)
.onConflictDoNothing({
target: [
schema.categories.playlistId,
schema.categories.type,
schema.categories.xtreamId,
],
});
if (capturePhase) {
await capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CATEGORIES_WRITE_TRANSACTIONS,
() => insertXtreamCategories(db, values),
() => ({ itemCount: values.length })
);
} else {
await insertXtreamCategories(db, values);
}
return { success: true };
}
@@ -0,0 +1,269 @@
import { XTREAM_DATABASE_PERFORMANCE_PHASE } from '@iptvnator/shared/interfaces';
import type { AppDatabase } from '../database.types';
import {
clearXtreamImportCache,
getContent,
saveContent,
} from './content.operations';
import { createRecordingOperationPhaseCapture } from './performance-phase-capture.test-helpers';
function createContentDb(streamCount: number, timeline: string[]) {
const existingWhere = jest.fn().mockResolvedValue([{ count: 0 }]);
const categoriesWhere = jest
.fn()
.mockResolvedValue([{ id: 42, xtreamId: 201 }]);
const select = jest
.fn()
.mockReturnValueOnce({
from: jest.fn(() => ({
innerJoin: jest.fn(() => ({ where: existingWhere })),
})),
})
.mockReturnValueOnce({
from: jest.fn(() => ({ where: categoriesWhere })),
});
const run = jest.fn();
const values = jest.fn((_chunk: unknown[]) => ({
onConflictDoNothing: jest.fn(() => ({ run })),
}));
const insert = jest.fn(() => ({ values }));
const transactionResults: unknown[] = [];
const transaction = jest.fn((callback: (tx: unknown) => unknown) => {
timeline.push('transaction');
const result = callback({ insert });
transactionResults.push(result);
return result;
});
const streams = Array.from({ length: streamCount }, (_, index) => ({
category_id: '201',
name: `Channel ${index}`,
stream_id: String(index + 1),
}));
return {
db: { select, transaction } as unknown as AppDatabase,
streams,
transaction,
transactionResults,
values,
};
}
describe('content import performance phases', () => {
it('keeps the content read phase open until the ordered query resolves', async () => {
const rows = [{ id: 1, title: 'One' }];
let resolveRows!: (value: typeof rows) => void;
const pending = new Promise<typeof rows>((resolve) => {
resolveRows = resolve;
});
const orderBy = jest.fn(() => pending);
const where = jest.fn(() => ({ orderBy }));
const innerJoin = jest.fn(() => ({ where }));
const from = jest.fn(() => ({ innerJoin }));
const db = {
select: jest.fn(() => ({ from })),
} as unknown as AppDatabase;
const recording = createRecordingOperationPhaseCapture();
const result = getContent(db, 'playlist-1', 'live', recording.capture);
expect(recording.events).toHaveLength(1);
resolveRows(rows);
await expect(result).resolves.toBe(rows);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_READ,
},
{
boundary: 'end',
metadata: { itemCount: 1 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_READ,
},
]);
});
it('uses one aggregate write pair around every existing batch commit and progress callback', async () => {
const timeline: string[] = [];
const recording = createRecordingOperationPhaseCapture((event) =>
timeline.push(`${event.phase}:${event.boundary}`)
);
const harness = createContentDb(205, timeline);
const checkpoint = jest.fn(() => {
timeline.push('checkpoint');
});
const onProgress = jest.fn((progress: { current?: number }) => {
timeline.push(`progress:${progress.current}`);
});
await expect(
saveContent(
harness.db,
'playlist-1',
harness.streams,
'live',
{ checkpoint, onProgress },
recording.capture
)
).resolves.toEqual({ success: true, count: 205 });
expect(harness.transaction).toHaveBeenCalledTimes(3);
expect(harness.transactionResults).toEqual([
undefined,
undefined,
undefined,
]);
expect(
harness.values.mock.calls.map(([chunk]) => chunk.length)
).toEqual([100, 100, 5]);
expect(onProgress.mock.calls.map(([value]) => value.current)).toEqual([
100, 200, 205,
]);
const writeStart = timeline.indexOf(
'sqlite.content.write-transactions:start'
);
const writeEnd = timeline.indexOf(
'sqlite.content.write-transactions:end'
);
expect(writeStart).toBeGreaterThanOrEqual(0);
expect(writeEnd).toBeGreaterThan(timeline.lastIndexOf('progress:205'));
expect(
timeline
.map((entry, index) => ({ entry, index }))
.filter(({ entry }) => entry === 'transaction')
.every(({ index }) => index > writeStart && index < writeEnd)
).toBe(true);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_CATEGORY_MAP_READ,
},
{
boundary: 'end',
metadata: { itemCount: 1 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_CATEGORY_MAP_READ,
},
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_CONTENT,
},
{
boundary: 'end',
metadata: { itemCount: 205 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_CONTENT,
},
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_WRITE_TRANSACTIONS,
},
{
boundary: 'end',
metadata: { itemCount: 205 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_WRITE_TRANSACTIONS,
},
]);
});
it('closes the aggregate write phase on cancellation without changing the error', async () => {
const timeline: string[] = [];
const recording = createRecordingOperationPhaseCapture();
const harness = createContentDb(101, timeline);
const cancellation = new Error('cancelled');
cancellation.name = 'AbortError';
let checkpointCount = 0;
await expect(
saveContent(
harness.db,
'playlist-1',
harness.streams,
'live',
{
checkpoint: () => {
checkpointCount += 1;
if (checkpointCount === 2) throw cancellation;
},
},
recording.capture
)
).rejects.toBe(cancellation);
expect(harness.transaction).toHaveBeenCalledTimes(1);
expect(recording.events.at(-1)).toEqual({
boundary: 'end',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_WRITE_TRANSACTIONS,
});
});
it('captures cache deletion as one pair across content and category chunks', async () => {
const categoryRows = Array.from({ length: 150 }, (_, id) => ({ id }));
const contentRows = Array.from({ length: 205 }, (_, id) => ({ id }));
const select = jest
.fn()
.mockReturnValueOnce({
from: jest.fn(() => ({
where: jest.fn().mockResolvedValue(categoryRows),
})),
})
.mockReturnValueOnce({
from: jest.fn(() => ({
where: jest.fn().mockResolvedValue(contentRows),
})),
});
const run = jest.fn();
const deleteRows = jest.fn(() => ({
where: jest.fn(() => ({ run })),
}));
const transaction = jest.fn((callback: (tx: unknown) => unknown) =>
callback({ delete: deleteRows })
);
const db = { select, transaction } as unknown as AppDatabase;
const recording = createRecordingOperationPhaseCapture();
await clearXtreamImportCache(
db,
'playlist-1',
'live',
recording.capture
);
expect(transaction).toHaveBeenCalledTimes(5);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_CACHE_CLEAR_WRITE_TRANSACTIONS,
},
{
boundary: 'end',
metadata: { itemCount: 355 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_CACHE_CLEAR_WRITE_TRANSACTIONS,
},
]);
});
it('emits one zero-count cache-clear pair when no categories exist', async () => {
const where = jest.fn().mockResolvedValue([]);
const from = jest.fn(() => ({ where }));
const select = jest.fn(() => ({ from }));
const transaction = jest.fn();
const db = { select, transaction } as unknown as AppDatabase;
const recording = createRecordingOperationPhaseCapture();
await expect(
clearXtreamImportCache(db, 'playlist-1', 'live', recording.capture)
).resolves.toEqual({ success: true });
expect(transaction).not.toHaveBeenCalled();
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_CACHE_CLEAR_WRITE_TRANSACTIONS,
},
{
boundary: 'end',
metadata: { itemCount: 0 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_CACHE_CLEAR_WRITE_TRANSACTIONS,
},
]);
});
});
@@ -0,0 +1,124 @@
import { XTREAM_DATABASE_PERFORMANCE_PHASE } from '@iptvnator/shared/interfaces';
import type { AppDatabase } from '../database.types';
import { searchContent } from './content.operations';
import { createRecordingOperationPhaseCapture } from './performance-phase-capture.test-helpers';
function createSearchDb(query: Promise<unknown[]>) {
const limit = jest.fn(() => query);
const where = jest.fn(() => ({ limit }));
const innerJoin = jest.fn(() => ({ where }));
const from = jest.fn(() => ({ innerJoin }));
const select = jest.fn(() => ({ from }));
return {
db: { select } as unknown as AppDatabase,
select,
};
}
describe('in-playlist search performance phases', () => {
it('closes query before measuring synchronous filtering and ranking', async () => {
const candidates = [
{
id: 1,
title: 'News One',
xtream_id: 101,
type: 'live',
},
{
id: 2,
title: 'Movies',
xtream_id: 102,
type: 'live',
},
];
let resolveQuery!: (value: typeof candidates) => void;
const query = new Promise<typeof candidates>((resolve) => {
resolveQuery = resolve;
});
const { db } = createSearchDb(query);
const recording = createRecordingOperationPhaseCapture();
const result = searchContent(
db,
'playlist-1',
'news',
['live'],
false,
recording.capture
);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_SEARCH_QUERY,
},
]);
resolveQuery(candidates);
await expect(result).resolves.toEqual([candidates[0]]);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_SEARCH_QUERY,
},
{
boundary: 'end',
metadata: { itemCount: 2 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_SEARCH_QUERY,
},
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_SEARCH_RANK,
},
{
boundary: 'end',
metadata: { itemCount: 1 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_SEARCH_RANK,
},
]);
});
it('closes a rejected query without starting rank and preserves identity', async () => {
const queryError = new Error('search query failed');
const { db } = createSearchDb(Promise.reject(queryError));
const recording = createRecordingOperationPhaseCapture();
await expect(
searchContent(
db,
'playlist-1',
'news',
['live'],
false,
recording.capture
)
).rejects.toBe(queryError);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_SEARCH_QUERY,
},
{
boundary: 'end',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_SEARCH_QUERY,
},
]);
});
it('keeps invalid input as zero-work with no query or phases', async () => {
const { db, select } = createSearchDb(Promise.resolve([]));
const recording = createRecordingOperationPhaseCapture();
await expect(
searchContent(
db,
'playlist-1',
'',
['live'],
false,
recording.capture
)
).resolves.toEqual([]);
expect(select).not.toHaveBeenCalled();
expect(recording.events).toEqual([]);
});
});
@@ -9,6 +9,7 @@ import {
GlobalSearchResult,
GlobalSearchResultSource,
M3uGlobalSearchResult,
XTREAM_DATABASE_PERFORMANCE_PHASE,
getXtreamRecentlyAddedMaxEpochSeconds,
toXtreamRecentlyAddedEpochSeconds,
XtreamGlobalSearchResult,
@@ -31,12 +32,15 @@ import {
shouldUseContentTitleFts,
shouldUseContentTitlePrefixIndex,
} from './content-search.util';
import type { OperationControl } from './operation-control';
import type { DatabaseOperationPerformancePhaseCapture } from './performance-phase-capture';
import {
checkpointOperation,
chunkValues,
type OperationControl,
reportOperationProgress,
} from './operation-control';
deleteXtreamCacheRows,
normalizeXtreamContentValues,
rankSearchCandidates,
writeXtreamContentValues,
type XtreamContentValue,
} from './xtream-content-operation-steps';
export { scoreSearchTextMatch } from './content-search.util';
@@ -673,7 +677,8 @@ export async function hasContent(
export async function getContent(
db: AppDatabase,
playlistId: string,
type: 'live' | 'movie' | 'series'
type: 'live' | 'movie' | 'series',
capturePhase?: DatabaseOperationPerformancePhaseCapture
) {
const baseQuery = db
.select(selectContentFields())
@@ -689,17 +694,22 @@ export async function getContent(
)
);
return type === 'live'
? baseQuery.orderBy(asc(schema.content.id))
: baseQuery.orderBy(desc(schema.content.added));
const query =
type === 'live'
? baseQuery.orderBy(asc(schema.content.id))
: baseQuery.orderBy(desc(schema.content.added));
return capturePhase
? capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_READ,
async () => query,
(rows) => ({ itemCount: rows.length })
)
: query;
}
export type RecentlyAddedPlaylistType =
| 'xtream'
| 'stalker'
| 'm3u-file'
| 'm3u-text'
| 'm3u-url';
'xtream' | 'stalker' | 'm3u-file' | 'm3u-text' | 'm3u-url';
export async function getGlobalRecentlyAdded(
db: AppDatabase,
@@ -796,20 +806,6 @@ function getGlobalRecentlyAddedByType(
.limit(limit);
}
type XtreamContentValue = {
categoryId: number;
title: string;
rating: string;
added: string;
posterUrl: string;
epgChannelId?: string | null;
tvArchive?: number | null;
tvArchiveDuration?: number | null;
directSource?: string | null;
xtreamId: number;
type: 'live' | 'movie' | 'series';
};
type XtreamContentSource = Record<string, unknown> & {
category_id?: string | number;
rating?: string | number;
@@ -898,7 +894,8 @@ export async function saveContent(
playlistId: string,
streams: Array<Record<string, unknown>>,
type: 'live' | 'movie' | 'series',
control?: OperationControl
control?: OperationControl,
capturePhase?: DatabaseOperationPerformancePhaseCapture
): Promise<{ success: boolean; count: number }> {
const dbType =
type === 'series' ? 'series' : type === 'movie' ? 'movies' : 'live';
@@ -922,7 +919,7 @@ export async function saveContent(
return { success: true, count: existingContent[0].count };
}
const categories = await db
const categoriesQuery = db
.select({
id: schema.categories.id,
xtreamId: schema.categories.xtreamId,
@@ -935,49 +932,46 @@ export async function saveContent(
)
);
const categoryMap = new Map(
categories.map((category) => [category.xtreamId, category.id])
);
const categories = capturePhase
? await capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_CATEGORY_MAP_READ,
async () => categoriesQuery,
(rows) => ({ itemCount: rows.length })
)
: await categoriesQuery;
const values = capturePhase
? capturePhase.captureSync(
XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_CONTENT,
() =>
normalizeXtreamContentValues(
streams,
type,
categories,
toXtreamContentValue
),
(result) => ({ itemCount: result.length })
)
: normalizeXtreamContentValues(
streams,
type,
categories,
toXtreamContentValue
);
const values = streams
.map((stream) => toXtreamContentValue(stream, type, categoryMap))
.filter((value): value is XtreamContentValue => value !== null);
const total = values.length;
const chunkSize = 100;
let totalInserted = 0;
for (let index = 0; index < values.length; index += chunkSize) {
await checkpointOperation(control);
const chunk = values.slice(index, index + chunkSize);
await db.transaction((tx) => {
tx.insert(schema.content)
.values(chunk)
.onConflictDoNothing({
target: [
schema.content.categoryId,
schema.content.type,
schema.content.xtreamId,
],
})
.run();
});
totalInserted += chunk.length;
await reportOperationProgress(control, {
phase: 'saving-content',
current: totalInserted,
total,
increment: chunk.length,
});
}
return { success: true, count: totalInserted };
return capturePhase
? capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_CONTENT_WRITE_TRANSACTIONS,
() => writeXtreamContentValues(db, values, control),
(result) => ({ itemCount: result.count })
)
: writeXtreamContentValues(db, values, control);
}
export async function clearXtreamImportCache(
db: AppDatabase,
playlistId: string,
type: 'live' | 'movie' | 'series'
type: 'live' | 'movie' | 'series',
capturePhase?: DatabaseOperationPerformancePhaseCapture
): Promise<{ success: boolean }> {
const dbType =
type === 'series' ? 'series' : type === 'movie' ? 'movies' : 'live';
@@ -994,7 +988,13 @@ export async function clearXtreamImportCache(
const categoryIds = categoryRows.map((category) => category.id);
if (categoryIds.length === 0) {
return { success: true };
return capturePhase
? capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_CACHE_CLEAR_WRITE_TRANSACTIONS,
async () => ({ success: true }),
() => ({ itemCount: 0 })
)
: { success: true };
}
const contentRows = await db
@@ -1002,26 +1002,13 @@ export async function clearXtreamImportCache(
.from(schema.content)
.where(inArray(schema.content.categoryId, categoryIds));
for (const chunk of chunkValues(
contentRows.map((row) => row.id),
100
)) {
await db.transaction((tx) => {
tx.delete(schema.content)
.where(inArray(schema.content.id, chunk))
.run();
});
}
for (const chunk of chunkValues(categoryIds, 100)) {
await db.transaction((tx) => {
tx.delete(schema.categories)
.where(inArray(schema.categories.id, chunk))
.run();
});
}
return { success: true };
return capturePhase
? capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_CACHE_CLEAR_WRITE_TRANSACTIONS,
() => deleteXtreamCacheRows(db, contentRows, categoryIds),
() => ({ itemCount: contentRows.length + categoryIds.length })
)
: deleteXtreamCacheRows(db, contentRows, categoryIds);
}
export async function getContentByXtreamId(
@@ -1057,7 +1044,8 @@ export async function searchContent(
playlistId: string,
searchTerm: string,
types: string[],
excludeHidden = false
excludeHidden = false,
capturePhase?: DatabaseOperationPerformancePhaseCapture
) {
if (!types || types.length === 0 || !normalizeSearchMatchText(searchTerm)) {
return [];
@@ -1076,7 +1064,7 @@ export async function searchContent(
conditions.push(eq(schema.categories.hidden, false));
}
const candidates = await db
const query = db
.select(selectContentFields())
.from(schema.content)
.innerJoin(
@@ -1086,12 +1074,21 @@ export async function searchContent(
.where(and(...conditions))
.limit(200);
return candidates
.filter(
(item) =>
scoreSearchTextMatch(item.title ?? '', searchTerm) !== null
)
.slice(0, 50);
const candidates = capturePhase
? await capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_SEARCH_QUERY,
async () => query,
(rows) => ({ itemCount: rows.length })
)
: await query;
return capturePhase
? capturePhase.captureSync(
XTREAM_DATABASE_PERFORMANCE_PHASE.NORMALIZE_SEARCH_RANK,
() => rankSearchCandidates(candidates, searchTerm),
(result) => ({ itemCount: result.length })
)
: rankSearchCandidates(candidates, searchTerm);
}
export async function globalSearch(
@@ -0,0 +1,68 @@
import type {
XtreamDatabasePerformancePhase,
PerformancePhaseMetadata,
} from '@iptvnator/shared/interfaces';
import type { DatabaseOperationPerformancePhaseCapture } from './performance-phase-capture';
export interface RecordedOperationPhase {
readonly boundary: 'end' | 'start';
readonly metadata?: PerformancePhaseMetadata;
readonly phase: XtreamDatabasePerformancePhase;
}
export function createRecordingOperationPhaseCapture(
onBoundary?: (event: RecordedOperationPhase) => void
): {
readonly capture: DatabaseOperationPerformancePhaseCapture;
readonly events: RecordedOperationPhase[];
} {
const events: RecordedOperationPhase[] = [];
const record = (event: RecordedOperationPhase): void => {
events.push(event);
onBoundary?.(event);
};
return {
capture: {
async captureAsync<TResult>(
phase: XtreamDatabasePerformancePhase,
execute: () => Promise<TResult>,
metadata?: (result: TResult) => PerformancePhaseMetadata
): Promise<TResult> {
record({ boundary: 'start', phase });
try {
const result = await execute();
record({
boundary: 'end',
metadata: metadata?.(result),
phase,
});
return result;
} catch (error) {
record({ boundary: 'end', phase });
throw error;
}
},
captureSync<TResult>(
phase: XtreamDatabasePerformancePhase,
execute: () => TResult,
metadata?: (result: TResult) => PerformancePhaseMetadata
): TResult {
record({ boundary: 'start', phase });
try {
const result = execute();
record({
boundary: 'end',
metadata: metadata?.(result),
phase,
});
return result;
} catch (error) {
record({ boundary: 'end', phase });
throw error;
}
},
},
events,
};
}
@@ -0,0 +1,17 @@
import type {
PerformancePhaseMetadata,
XtreamDatabasePerformancePhase,
} from '@iptvnator/shared/interfaces';
export interface DatabaseOperationPerformancePhaseCapture {
captureAsync: <TResult>(
phase: XtreamDatabasePerformancePhase,
execute: () => Promise<TResult>,
metadata?: (result: TResult) => PerformancePhaseMetadata
) => Promise<TResult>;
captureSync: <TResult>(
phase: XtreamDatabasePerformancePhase,
execute: () => TResult,
metadata?: (result: TResult) => PerformancePhaseMetadata
) => TResult;
}
@@ -0,0 +1,165 @@
import { XTREAM_DATABASE_PERFORMANCE_PHASE } from '@iptvnator/shared/interfaces';
import type { AppDatabase } from '../database.types';
import { createRecordingOperationPhaseCapture } from './performance-phase-capture.test-helpers';
import { deletePlaylist } from './playlist.operations';
function createSelectStep(
rows: readonly unknown[],
label: string,
timeline: string[]
) {
const where = jest.fn(() => {
timeline.push(label);
return Promise.resolve(rows);
});
const from = jest.fn(() => ({ where }));
return { from };
}
function createDeleteHarness() {
const timeline: string[] = [];
const selections = [
['select:favorites', [{ id: 1 }, { id: 2 }]],
['select:recently-viewed', [{ id: 3 }]],
['select:playback-positions', []],
['select:downloads', [{ id: 4 }]],
['select:categories', [{ id: 10 }, { id: 11 }]],
[
'select:content',
Array.from({ length: 205 }, (_, id) => ({ id: id + 100 })),
],
] as const;
const select = jest.fn();
for (const [label, rows] of selections) {
select.mockReturnValueOnce(createSelectStep(rows, label, timeline));
}
const transactionRun = jest.fn();
const transactionDelete = jest.fn(() => ({
where: jest.fn(() => ({ run: transactionRun })),
}));
const transaction = jest.fn((execute: (tx: unknown) => unknown) => {
timeline.push('transaction');
return execute({ delete: transactionDelete });
});
const playlistWhere = jest.fn(() => {
timeline.push('delete:playlist');
return Promise.resolve(undefined);
});
const deletePlaylistRow = jest.fn(() => ({ where: playlistWhere }));
return {
db: {
delete: deletePlaylistRow,
select,
transaction,
} as unknown as AppDatabase,
timeline,
transaction,
};
}
describe('playlist delete performance phases', () => {
it('counts collected dependent IDs and includes the final playlist row in writes', async () => {
const harness = createDeleteHarness();
const recording = createRecordingOperationPhaseCapture((event) => {
harness.timeline.push(`${event.phase}:${event.boundary}`);
});
const checkpoint = jest.fn(() => {
harness.timeline.push('checkpoint');
});
const onProgress = jest.fn((progress: { phase: string }) => {
harness.timeline.push(`progress:${progress.phase}`);
});
await expect(
deletePlaylist(
harness.db,
'playlist-1',
{ checkpoint, onProgress },
recording.capture
)
).resolves.toEqual({ success: true });
expect(harness.transaction).toHaveBeenCalledTimes(7);
// Seven committed chunks plus the final playlist row each keep both
// the pre-write and post-progress cooperative checkpoints.
expect(checkpoint).toHaveBeenCalledTimes(16);
expect(harness.timeline.slice(1, 7)).toEqual([
'select:favorites',
'select:recently-viewed',
'select:playback-positions',
'select:downloads',
'select:categories',
'select:content',
]);
const writeStart = harness.timeline.indexOf(
'sqlite.playlist-delete.write-transactions:start'
);
const writeEnd = harness.timeline.indexOf(
'sqlite.playlist-delete.write-transactions:end'
);
expect(harness.timeline.indexOf('delete:playlist')).toBeGreaterThan(
writeStart
);
expect(writeEnd).toBeGreaterThan(
harness.timeline.indexOf('progress:deleting-playlist')
);
expect(
harness.timeline
.map((entry, index) => ({ entry, index }))
.filter(({ entry }) => entry === 'transaction')
.every(({ index }) => index > writeStart && index < writeEnd)
).toBe(true);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_PLAYLIST_DELETE_COLLECT_IDS,
},
{
boundary: 'end',
metadata: { itemCount: 211 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_PLAYLIST_DELETE_COLLECT_IDS,
},
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_PLAYLIST_DELETE_WRITE_TRANSACTIONS,
},
{
boundary: 'end',
metadata: { itemCount: 212 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_PLAYLIST_DELETE_WRITE_TRANSACTIONS,
},
]);
});
it('closes the write phase on cancellation before deleting the playlist row', async () => {
const harness = createDeleteHarness();
const recording = createRecordingOperationPhaseCapture();
const cancellation = new Error('synthetic cancellation');
cancellation.name = 'AbortError';
let checkpointCount = 0;
await expect(
deletePlaylist(
harness.db,
'playlist-1',
{
checkpoint: () => {
checkpointCount += 1;
if (checkpointCount === 2) {
throw cancellation;
}
},
},
recording.capture
)
).rejects.toBe(cancellation);
expect(harness.transaction).toHaveBeenCalledTimes(1);
expect(harness.timeline).not.toContain('delete:playlist');
expect(recording.events.at(-1)).toEqual({
boundary: 'end',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_PLAYLIST_DELETE_WRITE_TRANSACTIONS,
});
});
});
@@ -4,6 +4,7 @@ import type { Channel, M3uFavoriteChannel } from '@iptvnator/shared/interfaces';
import {
APP_PLAYLIST_GET_PERFORMANCE_PHASE,
APP_PLAYLIST_UPSERT_PERFORMANCE_PHASE,
XTREAM_DATABASE_PERFORMANCE_PHASE,
type AppPlaylistGetPerformancePhase,
type AppPlaylistUpsertPerformancePhase,
type PerformancePhaseMetadata,
@@ -15,6 +16,7 @@ import {
type OperationControl,
reportOperationProgress,
} from './operation-control';
import type { DatabaseOperationPerformancePhaseCapture } from './performance-phase-capture';
const PLAYLIST_TYPES = {
XTREAM: 'xtream',
@@ -365,13 +367,10 @@ export async function upsertAppPlaylist(
}
const write = async () => {
await db
.insert(schema.playlists)
.values(row)
.onConflictDoUpdate({
target: schema.playlists.id,
set: row,
});
await db.insert(schema.playlists).values(row).onConflictDoUpdate({
target: schema.playlists.id,
set: row,
});
};
if (capturePhase) {
await capturePhase.captureAsync(
@@ -483,17 +482,14 @@ export async function getAppPlaylist(
(value) => ({ itemCount: value.length })
)
: await read();
const deserialize = () =>
rows[0] ? parseAppPlaylist(rows[0]) : null;
const deserialize = () => (rows[0] ? parseAppPlaylist(rows[0]) : null);
return capturePhase
? capturePhase.captureSync(
APP_PLAYLIST_GET_PERFORMANCE_PHASE.DESERIALIZE_PLAYLIST,
deserialize,
(value) => ({
itemCount: value
? (getPlaylistItemCount(value) ?? 0)
: 0,
itemCount: value ? (getPlaylistItemCount(value) ?? 0) : 0,
})
)
: deserialize();
@@ -592,11 +588,19 @@ export async function updatePlaylist(
return { success: true };
}
export async function deletePlaylist(
interface PlaylistDeletionCollection {
readonly categoryIds: number[];
readonly contentRows: Array<{ id: number }>;
readonly downloadRows: Array<{ id: number }>;
readonly favoriteRows: Array<{ id: number }>;
readonly playbackPositionRows: Array<{ id: number }>;
readonly recentlyViewedRows: Array<{ id: number }>;
}
async function collectPlaylistDeletionRows(
db: AppDatabase,
playlistId: string,
control?: OperationControl
): Promise<{ success: boolean }> {
playlistId: string
): Promise<PlaylistDeletionCollection> {
const [
favoriteRows,
recentlyViewedRows,
@@ -634,40 +638,69 @@ export async function deletePlaylist(
.where(inArray(schema.content.categoryId, categoryIds))
: [];
return {
categoryIds,
contentRows,
downloadRows,
favoriteRows,
playbackPositionRows,
recentlyViewedRows,
};
}
function countPlaylistDeletionRows(
collection: PlaylistDeletionCollection
): number {
return (
collection.favoriteRows.length +
collection.recentlyViewedRows.length +
collection.playbackPositionRows.length +
collection.downloadRows.length +
collection.contentRows.length +
collection.categoryIds.length
);
}
async function deleteCollectedPlaylistRows(
db: AppDatabase,
playlistId: string,
collection: PlaylistDeletionCollection,
control?: OperationControl
): Promise<number> {
for (const [phase, ids, column, table] of [
[
'deleting-favorites',
favoriteRows.map((row) => row.id),
collection.favoriteRows.map((row) => row.id),
schema.favorites.id,
schema.favorites,
],
[
'deleting-recently-viewed',
recentlyViewedRows.map((row) => row.id),
collection.recentlyViewedRows.map((row) => row.id),
schema.recentlyViewed.id,
schema.recentlyViewed,
],
[
'deleting-playback-positions',
playbackPositionRows.map((row) => row.id),
collection.playbackPositionRows.map((row) => row.id),
schema.playbackPositions.id,
schema.playbackPositions,
],
[
'deleting-downloads',
downloadRows.map((row) => row.id),
collection.downloadRows.map((row) => row.id),
schema.downloads.id,
schema.downloads,
],
[
'deleting-content',
contentRows.map((row) => row.id),
collection.contentRows.map((row) => row.id),
schema.content.id,
schema.content,
],
[
'deleting-categories',
categoryIds,
collection.categoryIds,
schema.categories.id,
schema.categories,
],
@@ -701,6 +734,41 @@ export async function deletePlaylist(
increment: 1,
});
return countPlaylistDeletionRows(collection) + 1;
}
export async function deletePlaylist(
db: AppDatabase,
playlistId: string,
control?: OperationControl,
capturePhase?: DatabaseOperationPerformancePhaseCapture
): Promise<{ success: boolean }> {
const collection = capturePhase
? await capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_PLAYLIST_DELETE_COLLECT_IDS,
() => collectPlaylistDeletionRows(db, playlistId),
(result) => ({
itemCount: countPlaylistDeletionRows(result),
})
)
: await collectPlaylistDeletionRows(db, playlistId);
if (capturePhase) {
await capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_PLAYLIST_DELETE_WRITE_TRANSACTIONS,
() =>
deleteCollectedPlaylistRows(
db,
playlistId,
collection,
control
),
(itemCount) => ({ itemCount })
);
} else {
await deleteCollectedPlaylistRows(db, playlistId, collection, control);
}
return { success: true };
}
@@ -0,0 +1,117 @@
import { inArray } from 'drizzle-orm';
import * as schema from '@iptvnator/shared/database/schema';
import type { AppDatabase } from '../database.types';
import { scoreSearchTextMatch } from './content-search.util';
import {
checkpointOperation,
chunkValues,
type OperationControl,
reportOperationProgress,
} from './operation-control';
export type XtreamContentValue = {
categoryId: number;
title: string;
rating: string;
added: string;
posterUrl: string;
epgChannelId?: string | null;
tvArchive?: number | null;
tvArchiveDuration?: number | null;
directSource?: string | null;
xtreamId: number;
type: 'live' | 'movie' | 'series';
};
export function normalizeXtreamContentValues(
streams: Array<Record<string, unknown>>,
type: 'live' | 'movie' | 'series',
categories: Array<{ id: number; xtreamId: number }>,
normalizeValue: (
stream: Record<string, unknown>,
type: 'live' | 'movie' | 'series',
categoryMap: Map<number, number>
) => XtreamContentValue | null
): XtreamContentValue[] {
const categoryMap = new Map(
categories.map((category) => [category.xtreamId, category.id])
);
return streams
.map((stream) => normalizeValue(stream, type, categoryMap))
.filter((value): value is XtreamContentValue => value !== null);
}
export async function writeXtreamContentValues(
db: AppDatabase,
values: XtreamContentValue[],
control?: OperationControl
): Promise<{ success: boolean; count: number }> {
const total = values.length;
const chunkSize = 100;
let totalInserted = 0;
for (let index = 0; index < values.length; index += chunkSize) {
await checkpointOperation(control);
const chunk = values.slice(index, index + chunkSize);
await db.transaction((tx) => {
tx.insert(schema.content)
.values(chunk)
.onConflictDoNothing({
target: [
schema.content.categoryId,
schema.content.type,
schema.content.xtreamId,
],
})
.run();
});
totalInserted += chunk.length;
await reportOperationProgress(control, {
phase: 'saving-content',
current: totalInserted,
total,
increment: chunk.length,
});
}
return { success: true, count: totalInserted };
}
export async function deleteXtreamCacheRows(
db: AppDatabase,
contentRows: Array<{ id: number }>,
categoryIds: number[]
): Promise<{ success: boolean }> {
for (const chunk of chunkValues(
contentRows.map((row) => row.id),
100
)) {
await db.transaction((tx) => {
tx.delete(schema.content)
.where(inArray(schema.content.id, chunk))
.run();
});
}
for (const chunk of chunkValues(categoryIds, 100)) {
await db.transaction((tx) => {
tx.delete(schema.categories)
.where(inArray(schema.categories.id, chunk))
.run();
});
}
return { success: true };
}
export function rankSearchCandidates<
TCandidate extends { title: string | null },
>(candidates: TCandidate[], searchTerm: string): TCandidate[] {
return candidates
.filter(
(item) =>
scoreSearchTextMatch(item.title ?? '', searchTerm) !== null
)
.slice(0, 50);
}
@@ -0,0 +1,217 @@
import { XTREAM_DATABASE_PERFORMANCE_PHASE } from '@iptvnator/shared/interfaces';
import type { AppDatabase } from '../database.types';
import { createRecordingOperationPhaseCapture } from './performance-phase-capture.test-helpers';
import { deleteXtreamContent } from './xtream.operations';
function createSelectStep(
rows: readonly unknown[],
label: string,
timeline: string[]
) {
const where = jest.fn(() => {
timeline.push(label);
return Promise.resolve(rows);
});
const innerJoin = jest.fn(() => ({ where }));
const from = jest.fn(() => ({ innerJoin, where }));
return { from };
}
function createDeleteHarness() {
const timeline: string[] = [];
const categories = [
{ hidden: true, id: 11, type: 'live', xtreamId: 101 },
{ hidden: false, id: 12, type: 'movies', xtreamId: 102 },
];
const favorites = [
{
addedAt: '2026-01-01T00:00:00.000Z',
contentType: 'live',
position: 3,
xtreamId: 201,
},
];
const recentlyViewed = [
{
contentType: 'movie',
viewedAt: '2026-01-02T00:00:00.000Z',
xtreamId: 202,
},
];
const content = Array.from({ length: 205 }, (_, id) => ({ id: id + 1 }));
const select = jest
.fn()
.mockReturnValueOnce(
createSelectStep(categories, 'select:categories', timeline)
)
.mockReturnValueOnce(
createSelectStep(favorites, 'select:favorites', timeline)
)
.mockReturnValueOnce(
createSelectStep(recentlyViewed, 'select:recently-viewed', timeline)
)
.mockReturnValueOnce(
createSelectStep(content, 'select:content', timeline)
);
const run = jest.fn();
const deleteRows = jest.fn(() => ({
where: jest.fn(() => ({ run })),
}));
const transaction = jest.fn((execute: (tx: unknown) => unknown) => {
timeline.push('transaction');
return execute({ delete: deleteRows });
});
return {
content,
db: { select, transaction } as unknown as AppDatabase,
timeline,
transaction,
};
}
describe('Xtream delete performance phases', () => {
it('times all user-data collection but counts only content/category deletion candidates', async () => {
const harness = createDeleteHarness();
const recording = createRecordingOperationPhaseCapture((event) => {
harness.timeline.push(`${event.phase}:${event.boundary}`);
});
const checkpoint = jest.fn(() => {
harness.timeline.push('checkpoint');
});
const onProgress = jest.fn((progress: { current?: number }) => {
harness.timeline.push(`progress:${progress.current}`);
});
await expect(
deleteXtreamContent(
harness.db,
'playlist-1',
{ checkpoint, onProgress },
recording.capture
)
).resolves.toEqual({
favorites: [
{
addedAt: '2026-01-01T00:00:00.000Z',
contentType: 'live',
position: 3,
xtreamId: 201,
},
],
hiddenCategories: [{ categoryType: 'live', xtreamId: 101 }],
recentlyViewed: [
{
contentType: 'movie',
viewedAt: '2026-01-02T00:00:00.000Z',
xtreamId: 202,
},
],
success: true,
});
expect(harness.transaction).toHaveBeenCalledTimes(4);
// Existing progress reporting performs a second cooperative
// checkpoint after each committed batch.
expect(checkpoint).toHaveBeenCalledTimes(8);
expect(onProgress.mock.calls.map(([value]) => value.current)).toEqual([
100, 200, 205, 2,
]);
expect(harness.timeline.slice(1, 5)).toEqual([
'select:categories',
'select:favorites',
'select:recently-viewed',
'select:content',
]);
const writeStart = harness.timeline.indexOf(
'sqlite.xtream-delete.write-transactions:start'
);
const writeEnd = harness.timeline.indexOf(
'sqlite.xtream-delete.write-transactions:end'
);
expect(writeEnd).toBeGreaterThan(
harness.timeline.lastIndexOf('progress:2')
);
expect(
harness.timeline
.map((entry, index) => ({ entry, index }))
.filter(({ entry }) => entry === 'transaction')
.every(({ index }) => index > writeStart && index < writeEnd)
).toBe(true);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_COLLECT_USER_DATA,
},
{
boundary: 'end',
metadata: { itemCount: 207 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_COLLECT_USER_DATA,
},
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_WRITE_TRANSACTIONS,
},
{
boundary: 'end',
metadata: { itemCount: 207 },
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_WRITE_TRANSACTIONS,
},
]);
});
it('closes the write phase on cancellation and preserves error identity', async () => {
const harness = createDeleteHarness();
const recording = createRecordingOperationPhaseCapture();
const cancellation = new Error('synthetic cancellation');
cancellation.name = 'AbortError';
let checkpointCount = 0;
await expect(
deleteXtreamContent(
harness.db,
'playlist-1',
{
checkpoint: () => {
checkpointCount += 1;
if (checkpointCount === 2) {
throw cancellation;
}
},
},
recording.capture
)
).rejects.toBe(cancellation);
expect(harness.transaction).toHaveBeenCalledTimes(1);
expect(recording.events.at(-1)).toEqual({
boundary: 'end',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_WRITE_TRANSACTIONS,
});
});
it('closes collection on query failure and preserves error identity', async () => {
const failure = new Error('synthetic select failure');
const where = jest.fn().mockRejectedValue(failure);
const from = jest.fn(() => ({ where }));
const db = {
select: jest.fn(() => ({ from })),
} as unknown as AppDatabase;
const recording = createRecordingOperationPhaseCapture();
await expect(
deleteXtreamContent(db, 'playlist-1', undefined, recording.capture)
).rejects.toBe(failure);
expect(recording.events).toEqual([
{
boundary: 'start',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_COLLECT_USER_DATA,
},
{
boundary: 'end',
phase: XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_COLLECT_USER_DATA,
},
]);
});
});
@@ -1,9 +1,10 @@
import { and, eq, inArray } from 'drizzle-orm';
import * as schema from '@iptvnator/shared/database/schema';
import type {
XtreamBackupFavoriteItem,
XtreamBackupHiddenCategory,
XtreamBackupRecentlyViewedItem,
import {
XTREAM_DATABASE_PERFORMANCE_PHASE,
type XtreamBackupFavoriteItem,
type XtreamBackupHiddenCategory,
type XtreamBackupRecentlyViewedItem,
} from '@iptvnator/shared/interfaces';
import type { AppDatabase } from '../database.types';
import {
@@ -12,6 +13,7 @@ import {
type OperationControl,
reportOperationProgress,
} from './operation-control';
import type { DatabaseOperationPerformancePhaseCapture } from './performance-phase-capture';
const DEFAULT_BATCH_SIZE = 100;
@@ -28,16 +30,18 @@ function toContentIdentityKey(
return `${contentType}:${xtreamId}`;
}
export async function deleteXtreamContent(
interface XtreamDeletionCollection {
readonly categoryIds: number[];
readonly contentRows: Array<{ id: number }>;
readonly favorites: XtreamBackupFavoriteItem[];
readonly hiddenCategories: XtreamBackupHiddenCategory[];
readonly recentlyViewed: XtreamBackupRecentlyViewedItem[];
}
async function collectXtreamDeletionRows(
db: AppDatabase,
playlistId: string,
control?: OperationControl
): Promise<{
success: boolean;
favorites: XtreamBackupFavoriteItem[];
recentlyViewed: XtreamBackupRecentlyViewedItem[];
hiddenCategories: XtreamBackupHiddenCategory[];
}> {
playlistId: string
): Promise<XtreamDeletionCollection> {
const categories = await db
.select({
id: schema.categories.id,
@@ -58,6 +62,7 @@ export async function deleteXtreamContent(
let favorites: XtreamBackupFavoriteItem[] = [];
let recentlyViewed: XtreamBackupRecentlyViewedItem[] = [];
let contentRows: Array<{ id: number }> = [];
if (categoryIds.length > 0) {
const favoritedContent = await db
@@ -110,43 +115,58 @@ export async function deleteXtreamContent(
viewedAt: item.viewedAt || new Date().toISOString(),
}));
const contentRows = await db
contentRows = await db
.select({ id: schema.content.id })
.from(schema.content)
.where(inArray(schema.content.categoryId, categoryIds));
}
let deletedContent = 0;
const totalContent = contentRows.length;
return {
categoryIds,
contentRows,
favorites,
hiddenCategories,
recentlyViewed,
};
}
for (const chunk of chunkValues(
contentRows.map((content) => content.id),
DEFAULT_BATCH_SIZE
)) {
await checkpointOperation(control);
await db.transaction((tx) => {
tx
.delete(schema.content)
.where(inArray(schema.content.id, chunk))
.run();
});
deletedContent += chunk.length;
await reportOperationProgress(control, {
phase: 'deleting-content',
current: deletedContent,
total: totalContent,
increment: chunk.length,
});
}
async function deleteCollectedXtreamRows(
db: AppDatabase,
collection: XtreamDeletionCollection,
control?: OperationControl
): Promise<number> {
let deletedContent = 0;
const totalContent = collection.contentRows.length;
for (const chunk of chunkValues(
collection.contentRows.map((content) => content.id),
DEFAULT_BATCH_SIZE
)) {
await checkpointOperation(control);
await db.transaction((tx) => {
tx.delete(schema.content)
.where(inArray(schema.content.id, chunk))
.run();
});
deletedContent += chunk.length;
await reportOperationProgress(control, {
phase: 'deleting-content',
current: deletedContent,
total: totalContent,
increment: chunk.length,
});
}
let deletedCategories = 0;
const totalCategories = categoryIds.length;
const totalCategories = collection.categoryIds.length;
for (const chunk of chunkValues(categoryIds, DEFAULT_BATCH_SIZE)) {
for (const chunk of chunkValues(
collection.categoryIds,
DEFAULT_BATCH_SIZE
)) {
await checkpointOperation(control);
await db.transaction((tx) => {
tx
.delete(schema.categories)
tx.delete(schema.categories)
.where(inArray(schema.categories.id, chunk))
.run();
});
@@ -159,11 +179,46 @@ export async function deleteXtreamContent(
});
}
return totalContent + totalCategories;
}
export async function deleteXtreamContent(
db: AppDatabase,
playlistId: string,
control?: OperationControl,
capturePhase?: DatabaseOperationPerformancePhaseCapture
): Promise<{
success: boolean;
favorites: XtreamBackupFavoriteItem[];
recentlyViewed: XtreamBackupRecentlyViewedItem[];
hiddenCategories: XtreamBackupHiddenCategory[];
}> {
const collection = capturePhase
? await capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_COLLECT_USER_DATA,
() => collectXtreamDeletionRows(db, playlistId),
(result) => ({
itemCount:
result.contentRows.length + result.categoryIds.length,
})
)
: await collectXtreamDeletionRows(db, playlistId);
if (capturePhase) {
await capturePhase.captureAsync(
XTREAM_DATABASE_PERFORMANCE_PHASE.SQLITE_XTREAM_DELETE_WRITE_TRANSACTIONS,
() => deleteCollectedXtreamRows(db, collection, control),
(itemCount) => ({ itemCount })
);
} else {
await deleteCollectedXtreamRows(db, collection, control);
}
return {
success: true,
favorites,
recentlyViewed,
hiddenCategories,
favorites: collection.favorites,
recentlyViewed: collection.recentlyViewed,
hiddenCategories: collection.hiddenCategories,
};
}