feat(database): implement DatabaseWorkerClient and associated worker functionality

- Added DatabaseWorkerClient to manage database operations in a worker thread.
- Created database-worker-client.ts to handle requests and responses between the main process and the worker.
- Introduced database-worker.types.ts for type definitions related to worker messages and operations.
- Implemented database.worker.ts to execute various database operations, including CRUD operations for playlists, content, and favorites.
- Added database.worker-connection.ts to manage database connections and ensure proper handling of SQLite instances.
- Created tests for DatabaseWorkerClient to ensure correct request handling and error management.
- Updated xtream.events.ts to allow suppression of error logging for specific requests.
- Enhanced EPG parser worker to include a busy timeout for SQLite operations.
- Integrated shutdown logic for the DatabaseWorkerClient in the main application lifecycle.
This commit is contained in:
4gray committed 2026-03-28 15:40:06 +01:00
1 parent d2f3044ffa
commit 456173b772
32 files changed
+4347 -1763

No files matched your search

@@ -36,6 +36,10 @@ type ElectronFixtures = {
dataDir: string;
};
type LaunchElectronAppOptions = {
env?: Record<string, string | undefined>;
};
export type PortalDebugEvent = {
durationMs: number;
operation: string;
@@ -48,10 +52,26 @@ export type PortalDebugEvent = {
transport: 'electron-main' | 'electron-renderer' | 'pwa-http';
};
export type DbOperationEvent = {
operationId?: string;
operation: string;
playlistId?: string;
status: 'started' | 'progress' | 'completed' | 'cancelled' | 'error';
phase?: string;
current?: number;
total?: number;
increment?: number;
error?: string;
};
declare global {
interface Window {
__dbOperationEvents?: DbOperationEvent[];
__dbOperationUnsubscribe?: (() => void) | undefined;
__portalDebugEvents?: PortalDebugEvent[];
__portalDebugUnsubscribe?: (() => void) | undefined;
__rendererFrameCount?: number;
__rendererFrameRequestId?: number;
}
}
@@ -73,7 +93,8 @@ export const test = base.extend<ElectronFixtures>({
export { expect };
export async function launchElectronApp(
dataDir: string
dataDir: string,
options: LaunchElectronAppOptions = {}
): Promise<LaunchedElectronApp> {
if (!existsSync(electronMainPath)) {
throw new Error(
@@ -91,6 +112,7 @@ export async function launchElectronApp(
args,
env: {
...process.env,
...options.env,
ELECTRON_IS_DEV: '0',
IPTVNATOR_E2E_DATA_DIR: dataDir,
NODE_ENV: 'test',
@@ -100,6 +122,8 @@ export async function launchElectronApp(
const mainWindow = await findMainWindow(electronApp);
await waitForAppReady(mainWindow);
await startPortalDebugCapture(mainWindow);
await startDbOperationCapture(mainWindow);
await startRendererFrameCapture(mainWindow);
return {
electronApp,
@@ -406,6 +430,49 @@ async function startPortalDebugCapture(page: Page): Promise<void> {
});
}
async function startDbOperationCapture(page: Page): Promise<void> {
await page.evaluate(() => {
const electronApi = window.electron as typeof window.electron & {
onDbOperationEvent?: (
callback: (event: DbOperationEvent) => void
) => (() => void) | undefined;
};
window.__dbOperationUnsubscribe?.();
window.__dbOperationEvents = [];
if (!electronApi.onDbOperationEvent) {
return;
}
window.__dbOperationUnsubscribe = electronApi.onDbOperationEvent(
(event: DbOperationEvent) => {
window.__dbOperationEvents?.push(event);
}
);
});
}
async function startRendererFrameCapture(page: Page): Promise<void> {
await page.evaluate(() => {
if (window.__rendererFrameRequestId) {
window.cancelAnimationFrame(window.__rendererFrameRequestId);
}
window.__rendererFrameCount = 0;
const tick = () => {
window.__rendererFrameCount =
(window.__rendererFrameCount ?? 0) + 1;
window.__rendererFrameRequestId = window.requestAnimationFrame(
tick
);
};
window.__rendererFrameRequestId = window.requestAnimationFrame(tick);
});
}
export async function expectPortalDebugSuccess(
page: Page,
provider: PortalProvider
@@ -481,6 +548,101 @@ export async function waitForPortalDebugEvent(
);
}
export async function waitForDbOperationEvent(
page: Page,
options: {
operation: string;
operationId?: string;
phase?: string;
playlistId?: string;
predicate?: (event: DbOperationEvent) => boolean;
status?: DbOperationEvent['status'];
timeoutMs?: number;
}
): Promise<DbOperationEvent> {
const {
operation,
operationId,
phase,
playlistId,
predicate,
status,
timeoutMs = 20000,
} = options;
const startedAt = Date.now();
while (Date.now() - startedAt <= timeoutMs) {
const events = await page.evaluate(() => window.__dbOperationEvents ?? []);
const match = events.find((event) => {
if (event.operation !== operation) {
return false;
}
if (operationId && event.operationId !== operationId) {
return false;
}
if (status && event.status !== status) {
return false;
}
if (phase && event.phase !== phase) {
return false;
}
if (playlistId && event.playlistId !== playlistId) {
return false;
}
return predicate ? predicate(event) : true;
});
if (match) {
return match;
}
await page.waitForTimeout(200);
}
const recentEvents = await page.evaluate(() =>
(window.__dbOperationEvents ?? []).slice(-20)
);
throw new Error(
`Timed out waiting for DB event ${operation}. Recent events: ${JSON.stringify(
recentEvents,
null,
2
)}`
);
}
export async function getRendererFrameCount(page: Page): Promise<number> {
return page.evaluate(() => window.__rendererFrameCount ?? 0);
}
export async function expectRendererFramesAdvance(
page: Page,
options: {
minimumDelta?: number;
sampleMs?: number;
} = {}
): Promise<void> {
const {
minimumDelta = 4,
sampleMs = 300,
} = options;
const startCount = await getRendererFrameCount(page);
await page.waitForTimeout(sampleMs);
await expect
.poll(async () => (await getRendererFrameCount(page)) - startCount, {
timeout: sampleMs + 1500,
})
.toBeGreaterThanOrEqual(minimumDelta);
}
export async function resetMockServers(
request: APIRequestContext,
providers: PortalProvider[]
@@ -0,0 +1,171 @@
import {
addXtreamPortal,
closeElectronApp,
expect,
expectPortalDebugSuccess,
expectRendererFramesAdvance,
launchElectronApp,
resetMockServers,
test,
waitForDbOperationEvent,
waitForXtreamCatalog,
} from './electron-test-fixtures';
const stressPortalName = 'Stress Xtream Portal';
const stressXtreamUsername = 'stress';
const stressXtreamPassword = 'stress';
const dbWorkerBatchDelayMs = '20';
function playlistRowByName(name: string) {
return `app-playlist-item:has-text("${name}")`;
}
async function openSources(page: Parameters<typeof addXtreamPortal>[0]) {
await page.getByRole('link', { name: 'Sources', exact: true }).click();
await page.waitForURL(/\/workspace\/sources$/);
}
test.describe('Electron Xtream Responsiveness', () => {
test.describe.configure({ mode: 'serial' });
test.setTimeout(120000);
test('keeps the UI responsive during large Xtream imports', async ({
dataDir,
request,
}) => {
await resetMockServers(request, ['xtream']);
const app = await launchElectronApp(dataDir, {
env: {
IPTVNATOR_DB_WORKER_BATCH_DELAY_MS: dbWorkerBatchDelayMs,
},
});
try {
await addXtreamPortal(app.mainWindow, {
name: stressPortalName,
password: stressXtreamPassword,
username: stressXtreamUsername,
});
const overlay = app.mainWindow.locator('.workspace-loading-overlay');
await expect(overlay).toBeVisible({ timeout: 15000 });
await expect(
overlay.getByRole('button', { name: /cancel import/i })
).toBeVisible({ timeout: 10000 });
await waitForDbOperationEvent(app.mainWindow, {
operation: 'save-content',
predicate: (event) =>
event.status === 'progress' &&
(event.current ?? 0) > 0 &&
(event.total ?? 0) > 0,
timeoutMs: 30000,
});
await expect
.poll(
async () =>
(
await app.mainWindow.evaluate(() =>
(window.__dbOperationEvents ?? []).filter(
(event) =>
event.operation === 'save-content' &&
event.status === 'progress'
).length
)
) > 2,
{ timeout: 30000 }
)
.toBeTruthy();
await expectRendererFramesAdvance(app.mainWindow, {
minimumDelta: 5,
sampleMs: 400,
});
await expect
.poll(
async () =>
(await overlay.locator('p').allInnerTexts()).some(
(text) => /\d+\s*\/\s*\d+/.test(text)
),
{ timeout: 30000 }
)
.toBeTruthy();
await waitForXtreamCatalog(app.mainWindow);
await expectPortalDebugSuccess(app.mainWindow, 'xtream');
} finally {
await closeElectronApp(app);
}
});
test('keeps the UI responsive during large Xtream playlist deletion', async ({
dataDir,
request,
}) => {
await resetMockServers(request, ['xtream']);
const app = await launchElectronApp(dataDir, {
env: {
IPTVNATOR_DB_WORKER_BATCH_DELAY_MS: dbWorkerBatchDelayMs,
},
});
try {
await addXtreamPortal(app.mainWindow, {
name: stressPortalName,
password: stressXtreamPassword,
username: stressXtreamUsername,
});
await waitForXtreamCatalog(app.mainWindow);
await openSources(app.mainWindow);
const playlistRow = app.mainWindow.locator(
playlistRowByName(stressPortalName)
);
await expect(playlistRow).toBeVisible({ timeout: 20000 });
await playlistRow.locator('.delete-btn').click();
const confirmDialog = app.mainWindow.locator('mat-dialog-container');
await expect(confirmDialog).toBeVisible();
await confirmDialog
.getByRole('button', { name: 'Yes', exact: true })
.click();
await waitForDbOperationEvent(app.mainWindow, {
operation: 'delete-playlist',
status: 'started',
timeoutMs: 20000,
});
await expect(playlistRow.locator('.busy-state__message')).toBeVisible(
{ timeout: 20000 }
);
await expect(playlistRow.locator('.cancel-btn')).toBeVisible();
await waitForDbOperationEvent(app.mainWindow, {
operation: 'delete-playlist',
predicate: (event) =>
event.status === 'progress' &&
(event.current ?? 0) > 0 &&
(event.total ?? 0) > 0,
timeoutMs: 30000,
});
await expectRendererFramesAdvance(app.mainWindow, {
minimumDelta: 5,
sampleMs: 400,
});
await expect(playlistRow).toHaveCount(0, { timeout: 60000 });
await expect(
app.mainWindow.locator(playlistRowByName(stressPortalName))
).toHaveCount(0);
} finally {
await closeElectronApp(app);
}
});
});
+64 -22
View File
@@ -21,30 +21,72 @@ const isProduction = process.env.NODE_ENV === 'production';
async function buildWorker() {
try {
console.log(`Building EPG parser worker with esbuild (${isProduction ? 'production' : 'development'})...`);
await esbuild.build({
entryPoints: [path.join(__dirname, 'src/app/workers/epg-parser.worker.ts')],
bundle: true,
platform: 'node',
target: 'node18',
format: 'cjs',
outfile: path.join(__dirname, '../../dist/apps/electron-backend/workers/epg-parser.worker.js'),
external: [
...nodeBuiltins.map(m => `node:${m}`),
...nodeBuiltins,
...nativeModules,
],
sourcemap: !isProduction,
minify: isProduction,
// Resolve workspace libraries from tsconfig paths
alias: {
'shared-interfaces': path.join(__dirname, '../../libs/shared/interfaces/src/index.ts'),
'database': path.join(__dirname, '../../libs/shared/database/src/index.ts'),
const workers = [
{
label: 'EPG parser worker',
entry: path.join(
__dirname,
'src/app/workers/epg-parser.worker.ts'
),
outfile: path.join(
__dirname,
'../../dist/apps/electron-backend/workers/epg-parser.worker.js'
),
},
});
{
label: 'database worker',
entry: path.join(
__dirname,
'src/app/workers/database.worker.ts'
),
outfile: path.join(
__dirname,
'../../dist/apps/electron-backend/workers/database.worker.js'
),
},
];
console.log('✅ Worker built successfully!');
for (const worker of workers) {
console.log(
`Building ${worker.label} with esbuild (${isProduction ? 'production' : 'development'})...`
);
await esbuild.build({
entryPoints: [worker.entry],
bundle: true,
platform: 'node',
target: 'node18',
format: 'cjs',
outfile: worker.outfile,
external: [
...nodeBuiltins.map((m) => `node:${m}`),
...nodeBuiltins,
...nativeModules,
],
sourcemap: !isProduction,
minify: isProduction,
alias: {
'shared-interfaces': path.join(
__dirname,
'../../libs/shared/interfaces/src/index.ts'
),
'database': path.join(
__dirname,
'../../libs/shared/database/src/index.ts'
),
'database-schema': path.join(
__dirname,
'../../libs/shared/database/src/lib/schema.ts'
),
'database-path-utils': path.join(
__dirname,
'../../libs/shared/database/src/lib/path-utils.ts'
),
},
});
}
console.log('✅ Workers built successfully!');
} catch (error) {
console.error('❌ Worker build failed:', error);
process.exit(1);
@@ -3,6 +3,7 @@ import type { ExternalPlayerSession } from 'shared-interfaces';
const PORTAL_DEBUG_EVENT = 'PORTAL_DEBUG_EVENT';
const EXTERNAL_PLAYER_SESSION_UPDATE = 'EXTERNAL_PLAYER_SESSION_UPDATE';
const DB_OPERATION_EVENT = 'DB_OPERATION_EVENT';
type PortalDebugEvent = {
requestId: string;
@@ -17,6 +18,22 @@ type PortalDebugEvent = {
error?: unknown;
};
type DbOperationEvent = {
operationId?: string;
operation: string;
playlistId?: string;
status: 'started' | 'progress' | 'completed' | 'cancelled' | 'error';
phase?: string;
current?: number;
total?: number;
increment?: number;
error?: string;
};
const dbSaveContentProgressListeners = new Set<
(event: Electron.IpcRendererEvent, data: DbOperationEvent) => void
>();
contextBridge.exposeInMainWorld('electron', {
// Remote control channel change listener
onChannelChange: (
@@ -89,15 +106,37 @@ contextBridge.exposeInMainWorld('electron', {
ipcRenderer.on(EXTERNAL_PLAYER_SESSION_UPDATE, handler);
return () => ipcRenderer.off(EXTERNAL_PLAYER_SESSION_UPDATE, handler);
},
onDbOperationEvent: (callback: (data: DbOperationEvent) => void) => {
const handler = (_event: Electron.IpcRendererEvent, data: any) =>
callback(data as DbOperationEvent);
ipcRenderer.on(DB_OPERATION_EVENT, handler);
return () => ipcRenderer.off(DB_OPERATION_EVENT, handler);
},
// DB save content progress listener
onDbSaveContentProgress: (callback: (count: number) => void) => {
ipcRenderer.on('DB_SAVE_CONTENT_PROGRESS', (_event, count) =>
callback(count)
);
const handler = (
_event: Electron.IpcRendererEvent,
data: DbOperationEvent
) => {
if (
data.operation !== 'save-content' ||
data.status !== 'progress'
) {
return;
}
callback(data.increment ?? data.current ?? 0);
};
dbSaveContentProgressListeners.add(handler);
ipcRenderer.on(DB_OPERATION_EVENT, handler);
},
// Remove DB save content progress listener
removeDbSaveContentProgress: () => {
ipcRenderer.removeAllListeners('DB_SAVE_CONTENT_PROGRESS');
dbSaveContentProgressListeners.forEach((handler) => {
ipcRenderer.off(DB_OPERATION_EVENT, handler);
});
dbSaveContentProgressListeners.clear();
},
getAppVersion: () => ipcRenderer.invoke('get-app-version'),
platform: process.platform,
@@ -196,6 +235,7 @@ contextBridge.exposeInMainWorld('electron', {
url: string;
params: Record<string, string>;
requestId?: string;
suppressErrorLog?: boolean;
}) => ipcRenderer.invoke('XTREAM_REQUEST', payload),
// Database operations
dbCreatePlaylist: (playlist: any) =>
@@ -211,20 +251,26 @@ contextBridge.exposeInMainWorld('electron', {
ipcRenderer.invoke('DB_GET_APP_PLAYLIST', playlistId),
dbUpdatePlaylist: (playlistId: string, updates: any) =>
ipcRenderer.invoke('DB_UPDATE_PLAYLIST', playlistId, updates),
dbDeletePlaylist: (playlistId: string) =>
ipcRenderer.invoke('DB_DELETE_PLAYLIST', playlistId),
dbDeleteXtreamContent: (playlistId: string) =>
ipcRenderer.invoke('DB_DELETE_XTREAM_CONTENT', playlistId),
dbDeletePlaylist: (playlistId: string, operationId?: string) =>
ipcRenderer.invoke('DB_DELETE_PLAYLIST', playlistId, operationId),
dbDeleteXtreamContent: (playlistId: string, operationId?: string) =>
ipcRenderer.invoke(
'DB_DELETE_XTREAM_CONTENT',
playlistId,
operationId
),
dbRestoreXtreamUserData: (
playlistId: string,
favoritedXtreamIds: number[],
recentlyViewedXtreamIds: { xtreamId: number; viewedAt: string }[]
recentlyViewedXtreamIds: { xtreamId: number; viewedAt: string }[],
operationId?: string
) =>
ipcRenderer.invoke(
'DB_RESTORE_XTREAM_USER_DATA',
playlistId,
favoritedXtreamIds,
recentlyViewedXtreamIds
recentlyViewedXtreamIds,
operationId
),
dbHasCategories: (playlistId: string, type: string) =>
ipcRenderer.invoke('DB_HAS_CATEGORIES', playlistId, type),
@@ -255,8 +301,19 @@ contextBridge.exposeInMainWorld('electron', {
ipcRenderer.invoke('DB_HAS_CONTENT', playlistId, type),
dbGetContent: (playlistId: string, type: string) =>
ipcRenderer.invoke('DB_GET_CONTENT', playlistId, type),
dbSaveContent: (playlistId: string, streams: any[], type: string) =>
ipcRenderer.invoke('DB_SAVE_CONTENT', playlistId, streams, type),
dbSaveContent: (
playlistId: string,
streams: any[],
type: string,
operationId?: string
) =>
ipcRenderer.invoke(
'DB_SAVE_CONTENT',
playlistId,
streams,
type,
operationId
),
dbSearchContent: (
playlistId: string,
searchTerm: string,
@@ -281,6 +338,10 @@ contextBridge.exposeInMainWorld('electron', {
types,
excludeHidden
),
dbGetGlobalRecentlyAdded: (
kind: 'all' | 'vod' | 'series',
limit?: number
) => ipcRenderer.invoke('DB_GET_GLOBAL_RECENTLY_ADDED', kind, limit),
dbGetRecentlyViewed: () => ipcRenderer.invoke('DB_GET_RECENTLY_VIEWED'),
dbClearRecentlyViewed: () => ipcRenderer.invoke('DB_CLEAR_RECENTLY_VIEWED'),
// Favorites
@@ -308,7 +369,10 @@ contextBridge.exposeInMainWorld('electron', {
ipcRenderer.invoke('DB_REMOVE_RECENT_ITEM', contentId, playlistId),
dbGetContentByXtreamId: (xtreamId: number, playlistId: string) =>
ipcRenderer.invoke('DB_GET_CONTENT_BY_XTREAM_ID', xtreamId, playlistId),
dbDeleteAllPlaylists: () => ipcRenderer.invoke('DB_DELETE_ALL_PLAYLISTS'),
dbDeleteAllPlaylists: (operationId?: string) =>
ipcRenderer.invoke('DB_DELETE_ALL_PLAYLISTS', operationId),
dbCancelOperation: (operationId: string) =>
ipcRenderer.invoke('DB_CANCEL_OPERATION', operationId),
dbGetAppState: (key: string) => ipcRenderer.invoke('DB_GET_APP_STATE', key),
dbSetAppState: (key: string, value: string) =>
ipcRenderer.invoke('DB_SET_APP_STATE', key, value),
@@ -0,0 +1,4 @@
import type { BetterSQLite3Database } from 'drizzle-orm/better-sqlite3';
import * as schema from 'database-schema';
export type AppDatabase = BetterSQLite3Database<typeof schema>;
@@ -0,0 +1,101 @@
import { and, eq, inArray, sql } from 'drizzle-orm';
import * as schema from 'database-schema';
import type { AppDatabase } from '../database.types';
export async function hasCategories(
db: AppDatabase,
playlistId: string,
type: 'live' | 'movies' | 'series'
): Promise<boolean> {
const result = await db
.select({ count: sql<number>`count(*)` })
.from(schema.categories)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.categories.type, type)
)
);
return result[0].count > 0;
}
export async function getCategories(
db: AppDatabase,
playlistId: string,
type: 'live' | 'movies' | 'series'
) {
return db
.select()
.from(schema.categories)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.categories.type, type),
eq(schema.categories.hidden, false)
)
)
.orderBy(sql`name COLLATE NOCASE`);
}
export async function saveCategories(
db: AppDatabase,
playlistId: string,
categories: Array<{
category_name: string;
category_id: number;
}>,
type: 'live' | 'movies' | 'series',
hiddenCategoryXtreamIds?: number[]
): Promise<{ success: boolean }> {
if (!categories || categories.length === 0) {
return { success: true };
}
const hiddenSet = new Set(hiddenCategoryXtreamIds || []);
const values = categories.map((category) => ({
playlistId,
name: category.category_name,
type,
xtreamId: category.category_id,
hidden: hiddenSet.has(category.category_id),
}));
await db.insert(schema.categories).values(values);
return { success: true };
}
export async function getAllCategories(
db: AppDatabase,
playlistId: string,
type: 'live' | 'movies' | 'series'
) {
return db
.select()
.from(schema.categories)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.categories.type, type)
)
)
.orderBy(sql`name COLLATE NOCASE`);
}
export async function updateCategoryVisibility(
db: AppDatabase,
categoryIds: number[],
hidden: boolean
): Promise<{ success: boolean }> {
if (categoryIds.length === 0) {
return { success: true };
}
await db
.update(schema.categories)
.set({ hidden })
.where(inArray(schema.categories.id, categoryIds));
return { success: true };
}
@@ -0,0 +1,412 @@
import { and, asc, desc, eq, inArray, or, sql } from 'drizzle-orm';
import * as schema from 'database-schema';
import type { AppDatabase } from '../database.types';
import {
checkpointOperation,
type OperationControl,
reportOperationProgress,
} from './operation-control';
function escapeLikePattern(term: string): string {
return term.replace(/[%_\\]/g, '\\$&');
}
function buildLikePatterns(term: string): string[] {
const variants = new Set<string>();
const titleCase =
term.length > 0
? term.charAt(0).toLocaleUpperCase() +
term.slice(1).toLocaleLowerCase()
: term;
variants.add(term);
variants.add(term.toLocaleLowerCase());
variants.add(term.toLocaleUpperCase());
variants.add(titleCase);
return [...variants].map((value) => `%${escapeLikePattern(value)}%`);
}
export type GlobalRecentlyAddedKind = 'all' | 'vod' | 'series';
function getRecentlyAddedContentTypes(
kind: GlobalRecentlyAddedKind
): Array<'movie' | 'series'> {
if (kind === 'vod') {
return ['movie'];
}
if (kind === 'series') {
return ['series'];
}
return ['movie', 'series'];
}
export async function hasContent(
db: AppDatabase,
playlistId: string,
type: 'live' | 'movie' | 'series'
): Promise<boolean> {
const result = await db
.select({ count: sql<number>`count(*)` })
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.content.type, type)
)
);
return result[0].count > 0;
}
export async function getContent(
db: AppDatabase,
playlistId: string,
type: 'live' | 'movie' | 'series'
) {
const baseQuery = db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.content.type, type)
)
);
return type === 'live'
? baseQuery.orderBy(asc(schema.content.id))
: baseQuery.orderBy(desc(schema.content.added));
}
export async function getGlobalRecentlyAdded(
db: AppDatabase,
kind: GlobalRecentlyAddedKind = 'all',
limit = 200
) {
const contentTypes = getRecentlyAddedContentTypes(kind);
const normalizedLimit = Number.isFinite(limit)
? Math.min(Math.max(Math.trunc(limit), 1), 200)
: 200;
const addedOrder = sql<number>`CAST(${schema.content.added} AS INTEGER)`;
return db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
added_at: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.playlists.id,
playlist_name: schema.playlists.name,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.where(
and(
inArray(schema.content.type, contentTypes),
eq(schema.categories.hidden, false),
sql`${schema.content.added} <> ''`
)
)
.orderBy(desc(addedOrder))
.limit(normalizedLimit);
}
type XtreamContentValue = {
categoryId: number;
title: string;
rating: string;
added: string;
posterUrl: string;
xtreamId: number;
type: 'live' | 'movie' | 'series';
};
type XtreamContentSource = Record<string, unknown> & {
category_id?: string | number;
rating?: string | number;
rating_imdb?: string;
last_modified?: string;
added?: string;
stream_icon?: string;
poster?: string;
cover?: string;
name?: string;
title?: string;
series_id?: string | number;
stream_id?: string | number;
};
function toXtreamContentValue(
stream: Record<string, unknown>,
type: 'live' | 'movie' | 'series',
categoryMap: Map<number, number>
): XtreamContentValue | null {
const source = stream as XtreamContentSource;
const streamCategoryId =
type === 'series'
? parseInt(String(source.category_id || '0'))
: parseInt(String(source.category_id));
const categoryId = categoryMap.get(streamCategoryId);
if (!categoryId) {
return null;
}
const title =
type === 'series'
? source.title ||
source.name ||
`Unknown Series ${source.series_id ?? ''}`.trim()
: source.name ||
source.title ||
`Unknown Stream ${source.stream_id ?? ''}`.trim();
return {
categoryId,
title,
rating: String(source.rating || source.rating_imdb || ''),
added:
type === 'series'
? String(source.last_modified || '')
: String(source.added || ''),
posterUrl: String(
source.stream_icon || source.poster || source.cover || ''
),
xtreamId:
type === 'series'
? parseInt(String(source.series_id || '0'))
: parseInt(String(source.stream_id || '0')),
type,
};
}
export async function saveContent(
db: AppDatabase,
playlistId: string,
streams: Array<Record<string, unknown>>,
type: 'live' | 'movie' | 'series',
control?: OperationControl
): Promise<{ success: boolean; count: number }> {
console.log(
`>>> DB_SAVE_CONTENT called: ${streams.length} items, type: ${type}, playlist: ${playlistId}`
);
const dbType =
type === 'series' ? 'series' : type === 'movie' ? 'movies' : 'live';
const categories = await db
.select({
id: schema.categories.id,
xtreamId: schema.categories.xtreamId,
})
.from(schema.categories)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.categories.type, dbType)
)
);
const categoryMap = new Map(categories.map((category) => [
category.xtreamId,
category.id,
]));
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.insert(schema.content).values(chunk);
totalInserted += chunk.length;
await reportOperationProgress(control, {
phase: 'saving-content',
current: totalInserted,
total,
increment: chunk.length,
});
}
return { success: true, count: totalInserted };
}
export async function getContentByXtreamId(
db: AppDatabase,
xtreamId: number,
playlistId: string
) {
const result = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.content.xtreamId, xtreamId),
eq(schema.categories.playlistId, playlistId)
)
)
.limit(1);
return result[0] || null;
}
export async function searchContent(
db: AppDatabase,
playlistId: string,
searchTerm: string,
types: string[],
excludeHidden = false
) {
if (!types || types.length === 0) {
return [];
}
const searchTermLower = searchTerm.toLocaleLowerCase();
const likePatterns = buildLikePatterns(searchTerm);
const likeConditions = likePatterns.map(
(pattern) => sql`${schema.content.title} LIKE ${pattern} ESCAPE '\\'`
);
const conditions = [
eq(schema.categories.playlistId, playlistId),
inArray(schema.content.type, types as Array<'live' | 'movie' | 'series'>),
or(...likeConditions),
];
if (excludeHidden) {
conditions.push(eq(schema.categories.hidden, false));
}
const candidates = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(and(...conditions))
.limit(200);
return candidates
.filter((item) =>
item.title?.toLocaleLowerCase().includes(searchTermLower)
)
.slice(0, 50);
}
export async function globalSearch(
db: AppDatabase,
searchTerm: string,
types: string[],
excludeHidden = false
) {
if (!types || types.length === 0) {
return [];
}
const searchTermLower = searchTerm.toLocaleLowerCase();
const likePatterns = buildLikePatterns(searchTerm);
const likeConditions = likePatterns.map(
(pattern) => sql`${schema.content.title} LIKE ${pattern} ESCAPE '\\'`
);
const conditions = [
inArray(schema.content.type, types as Array<'live' | 'movie' | 'series'>),
or(...likeConditions),
];
if (excludeHidden) {
conditions.push(eq(schema.categories.hidden, false));
}
const candidates = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.categories.playlistId,
playlist_name: schema.playlists.name,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.where(and(...conditions))
.orderBy(schema.content.title)
.limit(200);
return candidates
.filter((item) =>
item.title?.toLocaleLowerCase().includes(searchTermLower)
)
.slice(0, 50);
}
@@ -0,0 +1,185 @@
import { and, asc, desc, eq, inArray, sql } from 'drizzle-orm';
import * as schema from 'database-schema';
import type { AppDatabase } from '../database.types';
import {
chunkValues,
checkpointOperation,
type OperationControl,
reportOperationProgress,
} from './operation-control';
const DEFAULT_BATCH_SIZE = 100;
export async function addFavorite(
db: AppDatabase,
contentId: number,
playlistId: string
): Promise<{ success: boolean }> {
await db.insert(schema.favorites).values({
contentId,
playlistId,
});
return { success: true };
}
export async function removeFavorite(
db: AppDatabase,
contentId: number,
playlistId: string
): Promise<{ success: boolean }> {
await db
.delete(schema.favorites)
.where(
and(
eq(schema.favorites.contentId, contentId),
eq(schema.favorites.playlistId, playlistId)
)
);
return { success: true };
}
export async function isFavorite(
db: AppDatabase,
contentId: number,
playlistId: string
): Promise<boolean> {
const result = await db
.select({ count: sql<number>`count(*)` })
.from(schema.favorites)
.where(
and(
eq(schema.favorites.contentId, contentId),
eq(schema.favorites.playlistId, playlistId)
)
);
return result[0].count > 0;
}
export async function getFavorites(
db: AppDatabase,
playlistId: string
) {
return db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
added_at: schema.favorites.addedAt,
})
.from(schema.favorites)
.innerJoin(
schema.content,
eq(schema.favorites.contentId, schema.content.id)
)
.where(eq(schema.favorites.playlistId, playlistId))
.orderBy(desc(schema.favorites.addedAt));
}
export async function getGlobalFavorites(db: AppDatabase) {
return db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.playlists.id,
playlist_name: schema.playlists.name,
added_at: schema.favorites.addedAt,
position: schema.favorites.position,
})
.from(schema.favorites)
.innerJoin(
schema.content,
eq(schema.favorites.contentId, schema.content.id)
)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.where(eq(schema.content.type, 'live'))
.orderBy(asc(schema.favorites.position), desc(schema.favorites.addedAt))
.limit(300);
}
export async function getAllGlobalFavorites(db: AppDatabase) {
return db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.playlists.id,
playlist_name: schema.playlists.name,
added_at: schema.favorites.addedAt,
position: schema.favorites.position,
})
.from(schema.favorites)
.innerJoin(
schema.content,
eq(schema.favorites.contentId, schema.content.id)
)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.orderBy(asc(schema.favorites.position), desc(schema.favorites.addedAt))
.limit(500);
}
export async function reorderGlobalFavorites(
db: AppDatabase,
updates: { content_id: number; position: number }[],
control?: OperationControl
): Promise<{ success: boolean }> {
if (!Array.isArray(updates) || updates.length === 0) {
return { success: true };
}
let current = 0;
const total = updates.length;
for (const chunk of chunkValues(updates, DEFAULT_BATCH_SIZE)) {
await checkpointOperation(control);
for (const { content_id, position } of chunk) {
await db
.update(schema.favorites)
.set({ position })
.where(eq(schema.favorites.contentId, content_id));
}
current += chunk.length;
await reportOperationProgress(control, {
phase: 'reordering-global-favorites',
current,
total,
increment: chunk.length,
});
}
return { success: true };
}
@@ -0,0 +1,40 @@
export interface OperationProgressUpdate {
phase: string;
current?: number;
total?: number;
increment?: number;
}
export interface OperationControl {
checkpoint?: () => void | Promise<void>;
onProgress?: (
progress: OperationProgressUpdate
) => void | Promise<void>;
}
export async function checkpointOperation(
control?: OperationControl
): Promise<void> {
await control?.checkpoint?.();
}
export async function reportOperationProgress(
control: OperationControl | undefined,
progress: OperationProgressUpdate
): Promise<void> {
await control?.onProgress?.(progress);
await checkpointOperation(control);
}
export function chunkValues<T>(
values: T[],
chunkSize: number
): T[][] {
const chunks: T[][] = [];
for (let index = 0; index < values.length; index += chunkSize) {
chunks.push(values.slice(index, index + chunkSize));
}
return chunks;
}
@@ -0,0 +1,173 @@
import { and, desc, eq, sql } from 'drizzle-orm';
import * as schema from 'database-schema';
import type { AppDatabase } from '../database.types';
type PlaylistType =
| 'xtream'
| 'stalker'
| 'm3u-file'
| 'm3u-text'
| 'm3u-url';
type PlaybackPositionPayload = {
contentXtreamId: number;
contentType: 'vod' | 'episode';
seriesXtreamId?: number;
seasonNumber?: number;
episodeNumber?: number;
positionSeconds: number;
durationSeconds?: number;
playlistType?: PlaylistType;
};
async function ensurePlaylistExists(
db: AppDatabase,
playlistId: string,
playlistType: PlaylistType = 'stalker'
): Promise<void> {
const existing = await db
.select()
.from(schema.playlists)
.where(eq(schema.playlists.id, playlistId))
.limit(1);
if (existing.length === 0) {
await db.insert(schema.playlists).values({
id: playlistId,
name: 'Imported Playlist',
type: playlistType,
});
}
}
export async function savePlaybackPosition(
db: AppDatabase,
playlistId: string,
data: PlaybackPositionPayload
): Promise<{ success: boolean }> {
await ensurePlaylistExists(db, playlistId, data.playlistType);
const values = {
playlistId,
contentXtreamId: data.contentXtreamId,
contentType: data.contentType,
seriesXtreamId: data.seriesXtreamId,
seasonNumber: data.seasonNumber,
episodeNumber: data.episodeNumber,
positionSeconds: data.positionSeconds,
durationSeconds: data.durationSeconds,
updatedAt: sql`CURRENT_TIMESTAMP`,
};
const existing = await db
.select()
.from(schema.playbackPositions)
.where(
and(
eq(schema.playbackPositions.playlistId, playlistId),
eq(
schema.playbackPositions.contentXtreamId,
data.contentXtreamId
),
eq(schema.playbackPositions.contentType, data.contentType)
)
)
.limit(1);
if (existing.length > 0) {
await db
.update(schema.playbackPositions)
.set(values)
.where(eq(schema.playbackPositions.id, existing[0].id));
} else {
await db.insert(schema.playbackPositions).values(values);
}
return { success: true };
}
export async function getPlaybackPosition(
db: AppDatabase,
playlistId: string,
contentXtreamId: number,
contentType: 'vod' | 'episode'
) {
const result = await db
.select()
.from(schema.playbackPositions)
.where(
and(
eq(schema.playbackPositions.playlistId, playlistId),
eq(
schema.playbackPositions.contentXtreamId,
contentXtreamId
),
eq(schema.playbackPositions.contentType, contentType)
)
)
.limit(1);
return result[0] || null;
}
export async function getSeriesPlaybackPositions(
db: AppDatabase,
playlistId: string,
seriesXtreamId: number
) {
return db
.select()
.from(schema.playbackPositions)
.where(
and(
eq(schema.playbackPositions.playlistId, playlistId),
eq(schema.playbackPositions.seriesXtreamId, seriesXtreamId),
eq(schema.playbackPositions.contentType, 'episode')
)
);
}
export async function getRecentPlaybackPositions(
db: AppDatabase,
playlistId: string,
limit = 20
) {
return db
.select()
.from(schema.playbackPositions)
.where(eq(schema.playbackPositions.playlistId, playlistId))
.orderBy(desc(schema.playbackPositions.updatedAt))
.limit(limit);
}
export async function getAllPlaybackPositions(
db: AppDatabase,
playlistId: string
) {
return db
.select()
.from(schema.playbackPositions)
.where(eq(schema.playbackPositions.playlistId, playlistId));
}
export async function clearPlaybackPosition(
db: AppDatabase,
playlistId: string,
contentXtreamId: number,
contentType: 'vod' | 'episode'
): Promise<{ success: boolean }> {
await db
.delete(schema.playbackPositions)
.where(
and(
eq(schema.playbackPositions.playlistId, playlistId),
eq(
schema.playbackPositions.contentXtreamId,
contentXtreamId
),
eq(schema.playbackPositions.contentType, contentType)
)
);
return { success: true };
}
@@ -0,0 +1,506 @@
import { eq, inArray } from 'drizzle-orm';
import * as schema from 'database-schema';
import type { AppDatabase } from '../database.types';
import {
checkpointOperation,
chunkValues,
type OperationControl,
reportOperationProgress,
} from './operation-control';
const PLAYLIST_TYPES = {
XTREAM: 'xtream',
STALKER: 'stalker',
M3U_FILE: 'm3u-file',
M3U_TEXT: 'm3u-text',
M3U_URL: 'm3u-url',
} as const;
const DEFAULT_BATCH_SIZE = 100;
type PlaylistType = (typeof PLAYLIST_TYPES)[keyof typeof PLAYLIST_TYPES];
const PLAYLIST_TYPE_VALUES = new Set<PlaylistType>(
Object.values(PLAYLIST_TYPES)
);
function getStringValue(value: unknown): string | undefined {
if (typeof value !== 'string') {
return undefined;
}
const normalized = value.trim();
return normalized.length > 0 ? normalized : undefined;
}
function getNumericValue(value: unknown): number | undefined {
if (typeof value === 'number' && Number.isFinite(value)) {
return value;
}
if (typeof value === 'string' && value.trim() !== '') {
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : undefined;
}
return undefined;
}
function parseJsonValue<T>(value: string | null | undefined, fallback: T): T {
if (!value) {
return fallback;
}
try {
return JSON.parse(value) as T;
} catch (error) {
console.warn('Failed to parse JSON value from DB:', error);
return fallback;
}
}
function inferPlaylistType(playlist: Record<string, unknown>): PlaylistType {
const explicitType = getStringValue(playlist.type);
if (
explicitType &&
PLAYLIST_TYPE_VALUES.has(explicitType as PlaylistType)
) {
return explicitType as PlaylistType;
}
if (getStringValue(playlist.serverUrl)) {
return PLAYLIST_TYPES.XTREAM;
}
if (getStringValue(playlist.macAddress)) {
return PLAYLIST_TYPES.STALKER;
}
if (getStringValue(playlist.filePath)) {
return PLAYLIST_TYPES.M3U_FILE;
}
if (getStringValue(playlist.url)) {
return PLAYLIST_TYPES.M3U_URL;
}
return PLAYLIST_TYPES.M3U_TEXT;
}
function buildPlaylistRow(
playlist: Record<string, unknown>
): schema.NewPlaylist | null {
const id = getStringValue(playlist._id) ?? getStringValue(playlist.id);
if (!id) {
return null;
}
const type = inferPlaylistType(playlist);
const portalUrl = getStringValue(playlist.portalUrl);
const url = getStringValue(playlist.url);
const nowIso = new Date().toISOString();
const updateDate = getNumericValue(playlist.updateDate);
return {
id,
name:
getStringValue(playlist.title) ??
getStringValue(playlist.name) ??
id,
serverUrl: getStringValue(playlist.serverUrl),
username: getStringValue(playlist.username),
password: getStringValue(playlist.password),
lastUpdated:
typeof updateDate === 'number'
? new Date(updateDate).toISOString()
: getStringValue(playlist.lastUpdated),
type,
userAgent: getStringValue(playlist.userAgent),
origin: getStringValue(playlist.origin),
referrer: getStringValue(playlist.referrer),
filePath: getStringValue(playlist.filePath),
autoRefresh: Boolean(playlist.autoRefresh),
macAddress: getStringValue(playlist.macAddress),
url: type === PLAYLIST_TYPES.STALKER ? (portalUrl ?? url) : url,
portalUrl,
count: getNumericValue(playlist.count),
importDate: getStringValue(playlist.importDate),
updateDate,
position: getNumericValue(playlist.position),
favorites:
playlist.favorites !== undefined
? JSON.stringify(playlist.favorites)
: undefined,
recentlyViewed:
playlist.recentlyViewed !== undefined
? JSON.stringify(playlist.recentlyViewed)
: undefined,
payload: JSON.stringify(playlist),
lastUsage: getStringValue(playlist.lastUsage) ?? nowIso,
};
}
function parseAppPlaylist(row: schema.Playlist): Record<string, unknown> {
const payload = parseJsonValue<Record<string, unknown> | null>(
row.payload,
null
);
if (payload && typeof payload === 'object') {
return {
...payload,
_id:
getStringValue(payload._id) ??
getStringValue(payload.id) ??
row.id,
title:
getStringValue(payload.title) ??
getStringValue(payload.name) ??
row.name,
};
}
const favorites = parseJsonValue<unknown[]>(row.favorites, []);
const recentlyViewed = parseJsonValue<unknown[]>(row.recentlyViewed, []);
const importDate =
row.importDate ?? row.dateCreated ?? new Date().toISOString();
const portalUrl =
row.portalUrl ??
(row.type === PLAYLIST_TYPES.STALKER ? row.url : null);
return {
_id: row.id,
title: row.name,
count: row.count ?? 0,
importDate,
lastUsage: row.lastUsage ?? importDate,
favorites,
recentlyViewed,
autoRefresh: row.autoRefresh ?? false,
url: row.type === PLAYLIST_TYPES.M3U_URL ? row.url : undefined,
filePath: row.filePath ?? undefined,
userAgent: row.userAgent ?? undefined,
referrer: row.referrer ?? undefined,
origin: row.origin ?? undefined,
updateDate:
row.updateDate ??
(row.lastUpdated ? new Date(row.lastUpdated).getTime() : undefined),
position: row.position ?? undefined,
serverUrl: row.serverUrl ?? undefined,
username: row.username ?? undefined,
password: row.password ?? undefined,
macAddress: row.macAddress ?? undefined,
portalUrl: portalUrl ?? undefined,
};
}
export async function createPlaylist(
db: AppDatabase,
playlist: {
id: string;
name: string;
serverUrl?: string;
username?: string;
password?: string;
macAddress?: string;
url?: string;
type: string;
}
): Promise<{ success: boolean }> {
await db.insert(schema.playlists).values({
id: playlist.id,
name: playlist.name,
serverUrl: playlist.serverUrl,
username: playlist.username,
password: playlist.password,
macAddress: playlist.macAddress,
url: playlist.url,
portalUrl:
playlist.type === PLAYLIST_TYPES.STALKER ? playlist.url : undefined,
type: playlist.type as PlaylistType,
});
return { success: true };
}
export async function upsertAppPlaylist(
db: AppDatabase,
playlist: Record<string, unknown>
): Promise<{ success: boolean }> {
const row = buildPlaylistRow(playlist);
if (!row) {
throw new Error('Playlist ID is required for upsert');
}
await db
.insert(schema.playlists)
.values(row)
.onConflictDoUpdate({
target: schema.playlists.id,
set: row,
});
return { success: true };
}
export async function upsertAppPlaylists(
db: AppDatabase,
playlists: Record<string, unknown>[]
): Promise<{ success: boolean; count: number }> {
if (!Array.isArray(playlists) || playlists.length === 0) {
return { success: true, count: 0 };
}
let count = 0;
for (const playlist of playlists) {
const row = buildPlaylistRow(playlist);
if (!row) {
continue;
}
await db
.insert(schema.playlists)
.values(row)
.onConflictDoUpdate({
target: schema.playlists.id,
set: row,
});
count += 1;
}
return { success: true, count };
}
export async function getAppPlaylists(db: AppDatabase) {
const rows = await db.select().from(schema.playlists);
return rows.map((row) => parseAppPlaylist(row));
}
export async function getAppPlaylist(db: AppDatabase, playlistId: string) {
const rows = await db
.select()
.from(schema.playlists)
.where(eq(schema.playlists.id, playlistId))
.limit(1);
return rows[0] ? parseAppPlaylist(rows[0]) : null;
}
export async function getPlaylist(db: AppDatabase, playlistId: string) {
const result = await db
.select()
.from(schema.playlists)
.where(eq(schema.playlists.id, playlistId))
.limit(1);
return result[0] || null;
}
export async function updatePlaylist(
db: AppDatabase,
playlistId: string,
updates: {
name?: string;
username?: string;
password?: string;
serverUrl?: string;
lastUpdated?: string;
}
): Promise<{ success: boolean }> {
await db
.update(schema.playlists)
.set(updates)
.where(eq(schema.playlists.id, playlistId));
return { success: true };
}
export async function deletePlaylist(
db: AppDatabase,
playlistId: string,
control?: OperationControl
): Promise<{ success: boolean }> {
const [favoriteRows, recentlyViewedRows, playbackPositionRows, downloadRows] =
await Promise.all([
db
.select({ id: schema.favorites.id })
.from(schema.favorites)
.where(eq(schema.favorites.playlistId, playlistId)),
db
.select({ id: schema.recentlyViewed.id })
.from(schema.recentlyViewed)
.where(eq(schema.recentlyViewed.playlistId, playlistId)),
db
.select({ id: schema.playbackPositions.id })
.from(schema.playbackPositions)
.where(eq(schema.playbackPositions.playlistId, playlistId)),
db
.select({ id: schema.downloads.id })
.from(schema.downloads)
.where(eq(schema.downloads.playlistId, playlistId)),
]);
const categoryRows = await db
.select({ id: schema.categories.id })
.from(schema.categories)
.where(eq(schema.categories.playlistId, playlistId));
const categoryIds = categoryRows.map((category) => category.id);
const contentRows =
categoryIds.length > 0
? await db
.select({ id: schema.content.id })
.from(schema.content)
.where(inArray(schema.content.categoryId, categoryIds))
: [];
for (const [phase, ids, column, table] of [
[
'deleting-favorites',
favoriteRows.map((row) => row.id),
schema.favorites.id,
schema.favorites,
],
[
'deleting-recently-viewed',
recentlyViewedRows.map((row) => row.id),
schema.recentlyViewed.id,
schema.recentlyViewed,
],
[
'deleting-playback-positions',
playbackPositionRows.map((row) => row.id),
schema.playbackPositions.id,
schema.playbackPositions,
],
[
'deleting-downloads',
downloadRows.map((row) => row.id),
schema.downloads.id,
schema.downloads,
],
[
'deleting-content',
contentRows.map((row) => row.id),
schema.content.id,
schema.content,
],
[
'deleting-categories',
categoryIds,
schema.categories.id,
schema.categories,
],
] as const) {
let current = 0;
const total = ids.length;
for (const chunk of chunkValues(ids, DEFAULT_BATCH_SIZE)) {
await checkpointOperation(control);
await db.delete(table).where(inArray(column, chunk));
current += chunk.length;
await reportOperationProgress(control, {
phase,
current,
total,
increment: chunk.length,
});
}
}
await checkpointOperation(control);
await db.delete(schema.playlists).where(eq(schema.playlists.id, playlistId));
await reportOperationProgress(control, {
phase: 'deleting-playlist',
current: 1,
total: 1,
increment: 1,
});
return { success: true };
}
export async function getAppState(db: AppDatabase, key: string) {
const rows = await db
.select()
.from(schema.appState)
.where(eq(schema.appState.key, key))
.limit(1);
return rows[0]?.value ?? null;
}
export async function setAppState(
db: AppDatabase,
key: string,
value: string
): Promise<{ success: boolean }> {
const updatedAt = new Date().toISOString();
await db
.insert(schema.appState)
.values({
key,
value,
updatedAt,
})
.onConflictDoUpdate({
target: schema.appState.key,
set: { value, updatedAt },
});
return { success: true };
}
export async function deleteAllPlaylists(
db: AppDatabase,
control?: OperationControl
): Promise<{ success: boolean }> {
const deleteStages = [
{
phase: 'deleting-favorites',
execute: () => db.delete(schema.favorites),
},
{
phase: 'deleting-recently-viewed',
execute: () => db.delete(schema.recentlyViewed),
},
{
phase: 'deleting-playback-positions',
execute: () => db.delete(schema.playbackPositions),
},
{
phase: 'deleting-downloads',
execute: () => db.delete(schema.downloads),
},
{
phase: 'deleting-content',
execute: () => db.delete(schema.content),
},
{
phase: 'deleting-categories',
execute: () => db.delete(schema.categories),
},
{
phase: 'deleting-playlists',
execute: () => db.delete(schema.playlists),
},
] as const;
let current = 0;
const total = deleteStages.length;
for (const stage of deleteStages) {
await checkpointOperation(control);
await stage.execute();
current += 1;
await reportOperationProgress(control, {
phase: stage.phase,
current,
total,
increment: 1,
});
}
return { success: true };
}
@@ -0,0 +1,146 @@
import { and, desc, eq, inArray, sql } from 'drizzle-orm';
import * as schema from 'database-schema';
import type { AppDatabase } from '../database.types';
export async function getRecentlyViewed(db: AppDatabase) {
return db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.categories.playlistId,
playlist_name: schema.playlists.name,
viewed_at: schema.recentlyViewed.viewedAt,
})
.from(schema.recentlyViewed)
.innerJoin(
schema.content,
eq(schema.recentlyViewed.contentId, schema.content.id)
)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.orderBy(desc(schema.recentlyViewed.viewedAt))
.limit(100);
}
export async function clearRecentlyViewed(
db: AppDatabase
): Promise<{ success: boolean }> {
await db.delete(schema.recentlyViewed);
return { success: true };
}
export async function getRecentItems(
db: AppDatabase,
playlistId: string
) {
return db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
viewed_at: schema.recentlyViewed.viewedAt,
})
.from(schema.recentlyViewed)
.innerJoin(
schema.content,
eq(schema.recentlyViewed.contentId, schema.content.id)
)
.where(eq(schema.recentlyViewed.playlistId, playlistId))
.orderBy(desc(schema.recentlyViewed.viewedAt))
.limit(100);
}
export async function addRecentItem(
db: AppDatabase,
contentId: number,
playlistId: string
): Promise<{ success: boolean }> {
const existing = await db
.select()
.from(schema.recentlyViewed)
.where(
and(
eq(schema.recentlyViewed.contentId, contentId),
eq(schema.recentlyViewed.playlistId, playlistId)
)
)
.limit(1);
if (existing.length > 0) {
await db
.update(schema.recentlyViewed)
.set({ viewedAt: sql`CURRENT_TIMESTAMP` })
.where(
and(
eq(schema.recentlyViewed.contentId, contentId),
eq(schema.recentlyViewed.playlistId, playlistId)
)
);
} else {
await db.insert(schema.recentlyViewed).values({
contentId,
playlistId,
});
}
return { success: true };
}
export async function clearPlaylistRecentItems(
db: AppDatabase,
playlistId: string
): Promise<{ success: boolean }> {
const contentIds = await db
.select({ id: schema.content.id })
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(eq(schema.categories.playlistId, playlistId));
if (contentIds.length > 0) {
await db.delete(schema.recentlyViewed).where(
inArray(
schema.recentlyViewed.contentId,
contentIds.map((content) => content.id)
)
);
}
return { success: true };
}
export async function removeRecentItem(
db: AppDatabase,
contentId: number,
playlistId: string
): Promise<{ success: boolean }> {
await db
.delete(schema.recentlyViewed)
.where(
and(
eq(schema.recentlyViewed.contentId, contentId),
eq(schema.recentlyViewed.playlistId, playlistId)
)
);
return { success: true };
}
@@ -0,0 +1,246 @@
import { and, eq, inArray } from 'drizzle-orm';
import * as schema from 'database-schema';
import type { AppDatabase } from '../database.types';
import {
checkpointOperation,
chunkValues,
type OperationControl,
reportOperationProgress,
} from './operation-control';
const DEFAULT_BATCH_SIZE = 100;
export async function deleteXtreamContent(
db: AppDatabase,
playlistId: string,
control?: OperationControl
): Promise<{
success: boolean;
favoritedXtreamIds: number[];
recentlyViewedXtreamIds: { xtreamId: number; viewedAt: string }[];
hiddenCategories: { xtreamId: number; type: string }[];
}> {
const categories = await db
.select({
id: schema.categories.id,
xtreamId: schema.categories.xtreamId,
type: schema.categories.type,
hidden: schema.categories.hidden,
})
.from(schema.categories)
.where(eq(schema.categories.playlistId, playlistId));
const categoryIds = categories.map((category) => category.id);
const hiddenCategories = categories
.filter((category) => category.hidden)
.map((category) => ({
xtreamId: category.xtreamId,
type: category.type,
}));
let favoritedXtreamIds: number[] = [];
let recentlyViewedXtreamIds: { xtreamId: number; viewedAt: string }[] = [];
if (categoryIds.length > 0) {
const favorites = await db
.select({
xtreamId: schema.content.xtreamId,
})
.from(schema.favorites)
.innerJoin(
schema.content,
eq(schema.favorites.contentId, schema.content.id)
)
.where(
and(
eq(schema.favorites.playlistId, playlistId),
inArray(schema.content.categoryId, categoryIds)
)
);
favoritedXtreamIds = favorites.map((favorite) => favorite.xtreamId);
const recentlyViewed = await db
.select({
xtreamId: schema.content.xtreamId,
viewedAt: schema.recentlyViewed.viewedAt,
})
.from(schema.recentlyViewed)
.innerJoin(
schema.content,
eq(schema.recentlyViewed.contentId, schema.content.id)
)
.where(
and(
eq(schema.recentlyViewed.playlistId, playlistId),
inArray(schema.content.categoryId, categoryIds)
)
);
recentlyViewedXtreamIds = recentlyViewed.map((item) => ({
xtreamId: item.xtreamId,
viewedAt: item.viewedAt || new Date().toISOString(),
}));
const contentRows = await db
.select({ id: schema.content.id })
.from(schema.content)
.where(inArray(schema.content.categoryId, categoryIds));
let deletedContent = 0;
const totalContent = contentRows.length;
for (const chunk of chunkValues(
contentRows.map((content) => content.id),
DEFAULT_BATCH_SIZE
)) {
await checkpointOperation(control);
await db
.delete(schema.content)
.where(inArray(schema.content.id, chunk));
deletedContent += chunk.length;
await reportOperationProgress(control, {
phase: 'deleting-content',
current: deletedContent,
total: totalContent,
increment: chunk.length,
});
}
}
let deletedCategories = 0;
const totalCategories = categoryIds.length;
for (const chunk of chunkValues(categoryIds, DEFAULT_BATCH_SIZE)) {
await checkpointOperation(control);
await db
.delete(schema.categories)
.where(inArray(schema.categories.id, chunk));
deletedCategories += chunk.length;
await reportOperationProgress(control, {
phase: 'deleting-categories',
current: deletedCategories,
total: totalCategories,
increment: chunk.length,
});
}
return {
success: true,
favoritedXtreamIds,
recentlyViewedXtreamIds,
hiddenCategories,
};
}
export async function restoreXtreamUserData(
db: AppDatabase,
playlistId: string,
favoritedXtreamIds: number[],
recentlyViewedXtreamIds: { xtreamId: number; viewedAt: string }[],
control?: OperationControl
): Promise<{ success: boolean }> {
if (favoritedXtreamIds.length > 0) {
const content = await db
.select({
id: schema.content.id,
xtreamId: schema.content.xtreamId,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.categories.playlistId, playlistId),
inArray(schema.content.xtreamId, favoritedXtreamIds)
)
);
let currentFavorites = 0;
const totalFavorites = content.length;
for (const chunk of chunkValues(content, DEFAULT_BATCH_SIZE)) {
await checkpointOperation(control);
await db.insert(schema.favorites).values(
chunk.map((item) => ({
contentId: item.id,
playlistId,
addedAt: new Date().toISOString(),
}))
);
currentFavorites += chunk.length;
await reportOperationProgress(control, {
phase: 'restoring-favorites',
current: currentFavorites,
total: totalFavorites,
increment: chunk.length,
});
}
}
if (recentlyViewedXtreamIds.length > 0) {
const xtreamIds = recentlyViewedXtreamIds.map((item) => item.xtreamId);
const content = await db
.select({
id: schema.content.id,
xtreamId: schema.content.xtreamId,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.categories.playlistId, playlistId),
inArray(schema.content.xtreamId, xtreamIds)
)
);
const xtreamIdToContentId = new Map(
content.map((item) => [item.xtreamId, item.id])
);
const values = recentlyViewedXtreamIds
.map((item) => {
const contentId = xtreamIdToContentId.get(item.xtreamId);
if (!contentId) {
return null;
}
return {
contentId,
playlistId,
viewedAt: item.viewedAt,
};
})
.filter(
(
value
): value is {
contentId: number;
playlistId: string;
viewedAt: string;
} => value !== null
);
let currentRecentlyViewed = 0;
const totalRecentlyViewed = values.length;
for (const chunk of chunkValues(values, DEFAULT_BATCH_SIZE)) {
await checkpointOperation(control);
await db.insert(schema.recentlyViewed).values(chunk);
currentRecentlyViewed += chunk.length;
await reportOperationProgress(control, {
phase: 'restoring-recently-viewed',
current: currentRecentlyViewed,
total: totalRecentlyViewed,
increment: chunk.length,
});
}
}
return { success: true };
}
@@ -3,71 +3,27 @@
* Operations for managing categories within playlists
*/
import { and, eq, inArray, sql } from 'drizzle-orm';
import { ipcMain } from 'electron';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import { handleWorkerRequest } from './worker-events.utils';
/**
* Check if categories exist for a playlist
*/
ipcMain.handle(
handleWorkerRequest(
'DB_HAS_CATEGORIES',
async (event, playlistId: string, type: 'live' | 'movies' | 'series') => {
try {
const db = await getDatabase();
const result = await db
.select({ count: sql<number>`count(*)` })
.from(schema.categories)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.categories.type, type)
)
);
return result[0].count > 0;
} catch (error) {
console.error('Error checking categories:', error);
throw error;
}
}
(playlistId: string, type: 'live' | 'movies' | 'series') => ({
playlistId,
type,
})
);
/**
* Get categories for a playlist (excluding hidden)
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GET_CATEGORIES',
async (event, playlistId: string, type: 'live' | 'movies' | 'series') => {
try {
const db = await getDatabase();
const result = await db
.select()
.from(schema.categories)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.categories.type, type),
eq(schema.categories.hidden, false)
)
)
.orderBy(sql`name COLLATE NOCASE`);
return result;
} catch (error) {
console.error('Error getting categories:', error);
throw error;
}
}
(playlistId: string, type: 'live' | 'movies' | 'series') => ({
playlistId,
type,
})
);
/**
* Save categories in bulk
* Optionally accepts hidden category data to restore visibility preferences
*/
ipcMain.handle(
handleWorkerRequest(
'DB_SAVE_CATEGORIES',
async (
event,
(
playlistId: string,
categories: Array<{
category_name: string;
@@ -75,81 +31,26 @@ ipcMain.handle(
}>,
type: 'live' | 'movies' | 'series',
hiddenCategoryXtreamIds?: number[]
) => {
try {
// Skip if no categories to save
if (!categories || categories.length === 0) {
return { success: true };
}
const db = await getDatabase();
// Create a Set of hidden xtreamIds for efficient lookup
const hiddenSet = new Set(hiddenCategoryXtreamIds || []);
const values = categories.map((cat) => ({
playlistId,
name: cat.category_name,
type,
xtreamId: cat.category_id,
// Restore hidden status if the category was previously hidden
hidden: hiddenSet.has(cat.category_id),
}));
await db.insert(schema.categories).values(values);
return { success: true };
} catch (error) {
console.error('Error saving categories:', error);
throw error;
}
}
) => ({
playlistId,
categories,
type,
hiddenCategoryXtreamIds,
})
);
/**
* Update category visibility (hidden status)
*/
ipcMain.handle(
handleWorkerRequest(
'DB_UPDATE_CATEGORY_VISIBILITY',
async (event, categoryIds: number[], hidden: boolean) => {
try {
if (categoryIds.length === 0) {
return { success: true };
}
const db = await getDatabase();
await db
.update(schema.categories)
.set({ hidden })
.where(inArray(schema.categories.id, categoryIds));
return { success: true };
} catch (error) {
console.error('Error updating category visibility:', error);
throw error;
}
}
(categoryIds: number[], hidden: boolean) => ({
categoryIds,
hidden,
})
);
/**
* Get all categories for a playlist (including hidden, for management dialog)
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GET_ALL_CATEGORIES',
async (event, playlistId: string, type: 'live' | 'movies' | 'series') => {
try {
const db = await getDatabase();
const result = await db
.select()
.from(schema.categories)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.categories.type, type)
)
)
.orderBy(sql`name COLLATE NOCASE`);
return result;
} catch (error) {
console.error('Error getting all categories:', error);
throw error;
}
}
(playlistId: string, type: 'live' | 'movies' | 'series') => ({
playlistId,
type,
})
);
@@ -3,394 +3,91 @@
* Operations for managing and searching content (streams, movies, series)
*/
import { and, asc, desc, eq, inArray, or, sql } from 'drizzle-orm';
import { ipcMain } from 'electron';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import {
handleWorkerRequest,
requestWorkerWithEvents,
} from './worker-events.utils';
/**
* Escape special LIKE pattern characters to prevent unexpected matching behavior.
* This escapes %, _, and \ which have special meaning in SQL LIKE patterns.
*/
function escapeLikePattern(term: string): string {
return term.replace(/[%_\\]/g, '\\$&');
}
/**
* Build LIKE patterns for multiple case variants to improve non-ASCII matching
* in SQLite (default LIKE case-insensitive behavior is ASCII-only).
*/
function buildLikePatterns(term: string): string[] {
const variants = new Set<string>();
const titleCase =
term.length > 0
? term.charAt(0).toLocaleUpperCase() + term.slice(1).toLocaleLowerCase()
: term;
variants.add(term);
variants.add(term.toLocaleLowerCase());
variants.add(term.toLocaleUpperCase());
variants.add(titleCase);
return [...variants].map(value => `%${escapeLikePattern(value)}%`);
}
/**
* Check if content exists
*/
ipcMain.handle(
handleWorkerRequest(
'DB_HAS_CONTENT',
async (event, playlistId: string, type: 'live' | 'movie' | 'series') => {
try {
const db = await getDatabase();
const result = await db
.select({ count: sql<number>`count(*)` })
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.content.type, type)
)
);
return result[0].count > 0;
} catch (error) {
console.error('Error checking content:', error);
throw error;
}
}
(playlistId: string, type: 'live' | 'movie' | 'series') => ({
playlistId,
type,
})
);
/**
* Get content for a playlist
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GET_CONTENT',
async (event, playlistId: string, type: 'live' | 'movie' | 'series') => {
try {
const db = await getDatabase();
const baseQuery = db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.content.type, type)
)
);
const result = await (type === 'live'
? baseQuery.orderBy(asc(schema.content.id))
: baseQuery.orderBy(desc(schema.content.added)));
return result;
} catch (error) {
console.error('Error getting content:', error);
throw error;
}
}
(playlistId: string, type: 'live' | 'movie' | 'series') => ({
playlistId,
type,
})
);
handleWorkerRequest(
'DB_GET_GLOBAL_RECENTLY_ADDED',
(kind: 'all' | 'vod' | 'series' = 'all', limit = 200) => ({
kind,
limit,
})
);
/**
* Save content in bulk with progress reporting
*/
ipcMain.handle(
'DB_SAVE_CONTENT',
async (
event,
playlistId: string,
streams: Array<Record<string, unknown>>,
type: 'live' | 'movie' | 'series'
type: 'live' | 'movie' | 'series',
operationId?: string
) => {
try {
console.log(
`>>> DB_SAVE_CONTENT called: ${streams.length} items, type: ${type}, playlist: ${playlistId}`
return await requestWorkerWithEvents(
event,
'DB_SAVE_CONTENT',
{
playlistId,
streams,
type,
operationId,
}
);
const db = await getDatabase();
const dbType =
type === 'series'
? 'series'
: type === 'movie'
? 'movies'
: 'live';
// Get categories with their IDs
const categories = await db
.select({
id: schema.categories.id,
xtreamId: schema.categories.xtreamId,
})
.from(schema.categories)
.where(
and(
eq(schema.categories.playlistId, playlistId),
eq(schema.categories.type, dbType)
)
);
const categoryMap = new Map(
categories.map((c) => [c.xtreamId, c.id])
);
// Prepare bulk insert data
const values = streams
.map((stream) => {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const s = stream as any;
const streamCategoryId =
type === 'series'
? parseInt(s.category_id || '0')
: parseInt(s.category_id);
const categoryId = categoryMap.get(streamCategoryId);
if (!categoryId) return null;
const title =
type === 'series'
? s.title ||
s.name ||
`Unknown Series ${s.series_id}`
: s.name ||
s.title ||
`Unknown Stream ${s.stream_id}`;
return {
categoryId,
title,
rating: s.rating || s.rating_imdb || '',
added:
type === 'series'
? s.last_modified || ''
: s.added || '',
posterUrl: s.stream_icon || s.poster || s.cover || '',
xtreamId:
type === 'series'
? parseInt(s.series_id || '0')
: parseInt(s.stream_id || '0'),
type,
};
})
.filter((data) => data !== null);
// Insert in chunks for better performance
const CHUNK_SIZE = 100;
let totalInserted = 0;
for (let i = 0; i < values.length; i += CHUNK_SIZE) {
const chunk = values.slice(i, i + CHUNK_SIZE);
await db.insert(schema.content).values(
chunk as Array<{
categoryId: number;
title: string;
rating: string;
added: string;
posterUrl: string;
xtreamId: number;
type: 'live' | 'movie' | 'series';
}>
);
totalInserted += chunk.length;
// Send incremental progress update (chunk size, not cumulative total)
event.sender.send('DB_SAVE_CONTENT_PROGRESS', chunk.length);
}
return { success: true, count: totalInserted };
} catch (error) {
console.error('Error saving content:', error);
console.error('Error handling DB_SAVE_CONTENT:', error);
throw error;
}
}
);
/**
* Get content by xtream ID
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GET_CONTENT_BY_XTREAM_ID',
async (event, xtreamId: number, playlistId: string) => {
try {
const db = await getDatabase();
const result = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.content.xtreamId, xtreamId),
eq(schema.categories.playlistId, playlistId)
)
)
.limit(1);
return result[0] || null;
} catch (error) {
console.error('Error getting content by xtream ID:', error);
throw error;
}
}
(xtreamId: number, playlistId: string) => ({
xtreamId,
playlistId,
})
);
/**
* Search content within a specific playlist
*/
ipcMain.handle(
handleWorkerRequest(
'DB_SEARCH_CONTENT',
async (event, playlistId: string, searchTerm: string, types: string[], excludeHidden = false) => {
try {
if (!types || types.length === 0) return [];
const db = await getDatabase();
const searchTermLower = searchTerm.toLocaleLowerCase();
const likePatterns = buildLikePatterns(searchTerm);
const likeConditions = likePatterns.map(
pattern => sql`${schema.content.title} LIKE ${pattern} ESCAPE '\\'`
);
// Pre-filter with SQL LIKE (case-insensitive for ASCII, handled in native C code).
// This avoids loading the entire content table into JS memory.
const conditions = [
eq(schema.categories.playlistId, playlistId),
inArray(
schema.content.type,
types as Array<'live' | 'movie' | 'series'>
),
or(...likeConditions),
];
if (excludeHidden) {
conditions.push(eq(schema.categories.hidden, false));
}
const candidates = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(and(...conditions))
.limit(200);
// Post-filter for proper Unicode case-insensitive matching
// (SQLite LIKE is only case-insensitive for ASCII)
const result = candidates
.filter(item =>
item.title?.toLocaleLowerCase().includes(searchTermLower)
)
.slice(0, 50);
return result;
} catch (error) {
console.error('Error searching content:', error);
throw error;
}
}
(
playlistId: string,
searchTerm: string,
types: string[],
excludeHidden = false
) => ({
playlistId,
searchTerm,
types,
excludeHidden,
})
);
/**
* Global search across all playlists
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GLOBAL_SEARCH',
async (event, searchTerm: string, types: string[], excludeHidden = false) => {
try {
if (!types || types.length === 0) return [];
const db = await getDatabase();
const searchTermLower = searchTerm.toLocaleLowerCase();
const likePatterns = buildLikePatterns(searchTerm);
const likeConditions = likePatterns.map(
pattern => sql`${schema.content.title} LIKE ${pattern} ESCAPE '\\'`
);
// Pre-filter with SQL LIKE (case-insensitive for ASCII, handled in native C code).
// This avoids loading the entire content table into JS memory.
// Previously this query loaded ALL content rows (potentially 100k+),
// created JS objects for each, then filtered in JS — blocking the main process.
const conditions = [
inArray(
schema.content.type,
types as Array<'live' | 'movie' | 'series'>
),
or(...likeConditions),
];
if (excludeHidden) {
conditions.push(eq(schema.categories.hidden, false));
}
const candidates = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.categories.playlistId,
playlist_name: schema.playlists.name,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.where(and(...conditions))
.orderBy(schema.content.title)
.limit(200);
// Post-filter for proper Unicode case-insensitive matching
// (SQLite LIKE is only case-insensitive for ASCII)
const result = candidates
.filter(item =>
item.title?.toLocaleLowerCase().includes(searchTermLower)
)
.slice(0, 50);
return result;
} catch (error) {
console.error('Error in global search:', error);
throw error;
}
}
(searchTerm: string, types: string[], excludeHidden = false) => ({
searchTerm,
types,
excludeHidden,
})
);
@@ -3,227 +3,42 @@
* Operations for managing user's favorite content
*/
import { and, asc, desc, eq, sql } from 'drizzle-orm';
import { ipcMain } from 'electron';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import { handleWorkerRequest } from './worker-events.utils';
/**
* Add content to favorites
*/
ipcMain.handle(
handleWorkerRequest(
'DB_ADD_FAVORITE',
async (event, contentId: number, playlistId: string) => {
try {
const db = await getDatabase();
await db.insert(schema.favorites).values({
contentId,
playlistId,
});
return { success: true };
} catch (error) {
console.error('Error adding favorite:', error);
throw error;
}
}
(contentId: number, playlistId: string) => ({
contentId,
playlistId,
})
);
/**
* Remove content from favorites
*/
ipcMain.handle(
handleWorkerRequest(
'DB_REMOVE_FAVORITE',
async (event, contentId: number, playlistId: string) => {
try {
const db = await getDatabase();
await db
.delete(schema.favorites)
.where(
and(
eq(schema.favorites.contentId, contentId),
eq(schema.favorites.playlistId, playlistId)
)
);
return { success: true };
} catch (error) {
console.error('Error removing favorite:', error);
throw error;
}
}
(contentId: number, playlistId: string) => ({
contentId,
playlistId,
})
);
/**
* Check if content is favorited
*/
ipcMain.handle(
handleWorkerRequest(
'DB_IS_FAVORITE',
async (event, contentId: number, playlistId: string) => {
try {
const db = await getDatabase();
const result = await db
.select({ count: sql<number>`count(*)` })
.from(schema.favorites)
.where(
and(
eq(schema.favorites.contentId, contentId),
eq(schema.favorites.playlistId, playlistId)
)
);
return result[0].count > 0;
} catch (error) {
console.error('Error checking favorite:', error);
throw error;
}
}
(contentId: number, playlistId: string) => ({
contentId,
playlistId,
})
);
/**
* Get all favorites for a playlist
*/
ipcMain.handle('DB_GET_FAVORITES', async (event, playlistId: string) => {
try {
const db = await getDatabase();
const result = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
added_at: schema.favorites.addedAt,
})
.from(schema.favorites)
.innerJoin(
schema.content,
eq(schema.favorites.contentId, schema.content.id)
)
.where(eq(schema.favorites.playlistId, playlistId))
.orderBy(desc(schema.favorites.addedAt));
return result;
} catch (error) {
console.error('Error getting favorites:', error);
throw error;
}
});
handleWorkerRequest('DB_GET_FAVORITES', (playlistId: string) => ({
playlistId,
}));
/**
* Get global favorites across all playlists (live TV channels only)
*/
ipcMain.handle('DB_GET_GLOBAL_FAVORITES', async () => {
try {
const db = await getDatabase();
const result = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.playlists.id,
playlist_name: schema.playlists.name,
added_at: schema.favorites.addedAt,
position: schema.favorites.position,
})
.from(schema.favorites)
.innerJoin(
schema.content,
eq(schema.favorites.contentId, schema.content.id)
)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.where(eq(schema.content.type, 'live'))
.orderBy(
asc(schema.favorites.position),
desc(schema.favorites.addedAt)
)
.limit(300);
return result;
} catch (error) {
console.error('Error getting global favorites:', error);
throw error;
}
});
handleWorkerRequest('DB_GET_GLOBAL_FAVORITES', () => ({}));
handleWorkerRequest('DB_GET_ALL_GLOBAL_FAVORITES', () => ({}));
/**
* Get global favorites across all playlists (all content types)
*/
ipcMain.handle('DB_GET_ALL_GLOBAL_FAVORITES', async () => {
try {
const db = await getDatabase();
const result = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.playlists.id,
playlist_name: schema.playlists.name,
added_at: schema.favorites.addedAt,
position: schema.favorites.position,
})
.from(schema.favorites)
.innerJoin(
schema.content,
eq(schema.favorites.contentId, schema.content.id)
)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.orderBy(
asc(schema.favorites.position),
desc(schema.favorites.addedAt)
)
.limit(500);
return result;
} catch (error) {
console.error('Error getting all global favorites:', error);
throw error;
}
});
/**
* Reorder global favorites by updating the position field on each Xtream favorite row.
* Accepts an array of { content_id, position } pairs.
*/
ipcMain.handle(
handleWorkerRequest(
'DB_REORDER_GLOBAL_FAVORITES',
async (event, updates: { content_id: number; position: number }[]) => {
try {
if (!Array.isArray(updates) || updates.length === 0) {
return { success: true };
}
const db = await getDatabase();
for (const { content_id, position } of updates) {
await db
.update(schema.favorites)
.set({ position })
.where(eq(schema.favorites.contentId, content_id));
}
return { success: true };
} catch (error) {
console.error('Error reordering global favorites:', error);
throw error;
}
}
(updates: { content_id: number; position: number }[]) => ({
updates,
})
);
@@ -3,236 +3,61 @@
* Operations for managing video playback progress
*/
import { and, desc, eq, sql } from 'drizzle-orm';
import { ipcMain } from 'electron';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import { handleWorkerRequest } from './worker-events.utils';
/**
* Ensure playlist exists in SQLite (auto-create if missing)
* This handles legacy playlists that weren't synced to SQLite
*/
async function ensurePlaylistExists(
db: any,
playlistId: string,
playlistType: 'xtream' | 'stalker' | 'm3u-file' | 'm3u-text' | 'm3u-url' = 'stalker'
): Promise<void> {
const existing = await db
.select()
.from(schema.playlists)
.where(eq(schema.playlists.id, playlistId))
.limit(1);
if (existing.length === 0) {
// Create a minimal placeholder entry for the playlist
await db.insert(schema.playlists).values({
id: playlistId,
name: 'Imported Playlist',
type: playlistType,
});
}
}
/**
* Save/update playback position for content
*/
ipcMain.handle(
handleWorkerRequest(
'DB_SAVE_PLAYBACK_POSITION',
async (event, playlistId: string, data: any) => {
try {
const db = await getDatabase();
// Ensure playlist exists (handles legacy playlists not in SQLite)
await ensurePlaylistExists(db, playlistId, data.playlistType);
// Prepare values
const values = {
playlistId,
contentXtreamId: data.contentXtreamId,
contentType: data.contentType,
seriesXtreamId: data.seriesXtreamId,
seasonNumber: data.seasonNumber,
episodeNumber: data.episodeNumber,
positionSeconds: data.positionSeconds,
durationSeconds: data.durationSeconds,
updatedAt: sql`CURRENT_TIMESTAMP`,
};
// Check if exists
const existing = await db
.select()
.from(schema.playbackPositions)
.where(
and(
eq(schema.playbackPositions.playlistId, playlistId),
eq(
schema.playbackPositions.contentXtreamId,
data.contentXtreamId
),
eq(
schema.playbackPositions.contentType,
data.contentType
)
)
)
.limit(1);
if (existing.length > 0) {
await db
.update(schema.playbackPositions)
.set(values)
.where(eq(schema.playbackPositions.id, existing[0].id));
} else {
await db.insert(schema.playbackPositions).values(values);
}
return { success: true };
} catch (error) {
console.error('Error saving playback position:', error);
throw error;
}
}
(playlistId: string, data: unknown) => ({
playlistId,
data,
})
);
/**
* Get playback position for a specific content item
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GET_PLAYBACK_POSITION',
async (
event,
(
playlistId: string,
contentXtreamId: number,
contentType: 'vod' | 'episode'
) => {
try {
const db = await getDatabase();
const result = await db
.select()
.from(schema.playbackPositions)
.where(
and(
eq(schema.playbackPositions.playlistId, playlistId),
eq(
schema.playbackPositions.contentXtreamId,
contentXtreamId
),
eq(schema.playbackPositions.contentType, contentType)
)
)
.limit(1);
return result[0] || null;
} catch (error) {
console.error('Error getting playback position:', error);
throw error;
}
}
) => ({
playlistId,
contentXtreamId,
contentType,
})
);
/**
* Get all episode positions for a series
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GET_SERIES_PLAYBACK_POSITIONS',
async (event, playlistId: string, seriesXtreamId: number) => {
try {
const db = await getDatabase();
const result = await db
.select()
.from(schema.playbackPositions)
.where(
and(
eq(schema.playbackPositions.playlistId, playlistId),
eq(
schema.playbackPositions.seriesXtreamId,
seriesXtreamId
),
eq(schema.playbackPositions.contentType, 'episode')
)
);
return result;
} catch (error) {
console.error('Error getting series playback positions:', error);
throw error;
}
}
(playlistId: string, seriesXtreamId: number) => ({
playlistId,
seriesXtreamId,
})
);
/**
* Get recently watched items with positions
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GET_RECENT_PLAYBACK_POSITIONS',
async (event, playlistId: string, limit: number = 20) => {
try {
const db = await getDatabase();
const result = await db
.select()
.from(schema.playbackPositions)
.where(eq(schema.playbackPositions.playlistId, playlistId))
.orderBy(desc(schema.playbackPositions.updatedAt))
.limit(limit);
return result;
} catch (error) {
console.error('Error getting recent playback positions:', error);
throw error;
}
}
(playlistId: string, limit?: number) => ({
playlistId,
limit,
})
);
/**
* Get all playback positions for a playlist (to populate grid view)
*/
ipcMain.handle(
handleWorkerRequest(
'DB_GET_ALL_PLAYBACK_POSITIONS',
async (event, playlistId: string) => {
try {
const db = await getDatabase();
const result = await db
.select()
.from(schema.playbackPositions)
.where(eq(schema.playbackPositions.playlistId, playlistId));
return result;
} catch (error) {
console.error('Error getting all playback positions:', error);
throw error;
}
}
(playlistId: string) => ({
playlistId,
})
);
/**
* Clear playback position
*/
ipcMain.handle(
handleWorkerRequest(
'DB_CLEAR_PLAYBACK_POSITION',
async (
event,
(
playlistId: string,
contentXtreamId: number,
contentType: 'vod' | 'episode'
) => {
try {
const db = await getDatabase();
await db
.delete(schema.playbackPositions)
.where(
and(
eq(schema.playbackPositions.playlistId, playlistId),
eq(
schema.playbackPositions.contentXtreamId,
contentXtreamId
),
eq(schema.playbackPositions.contentType, contentType)
)
);
return { success: true };
} catch (error) {
console.error('Error clearing playback position:', error);
throw error;
}
}
) => ({
playlistId,
contentXtreamId,
contentType,
})
);
@@ -3,362 +3,28 @@
* CRUD operations for playlists
*/
import { eq } from 'drizzle-orm';
import { ipcMain } from 'electron';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import { databaseWorkerClient } from '../../services/database-worker-client';
import {
handleWorkerRequest,
requestWorkerWithEvents,
} from './worker-events.utils';
const PLAYLIST_TYPES = {
XTREAM: 'xtream',
STALKER: 'stalker',
M3U_FILE: 'm3u-file',
M3U_TEXT: 'm3u-text',
M3U_URL: 'm3u-url',
} as const;
type PlaylistType = (typeof PLAYLIST_TYPES)[keyof typeof PLAYLIST_TYPES];
const PLAYLIST_TYPE_VALUES = new Set<PlaylistType>(
Object.values(PLAYLIST_TYPES)
);
function getStringValue(value: unknown): string | undefined {
if (typeof value !== 'string') {
return undefined;
}
const normalized = value.trim();
return normalized.length > 0 ? normalized : undefined;
}
function getNumericValue(value: unknown): number | undefined {
if (typeof value === 'number' && Number.isFinite(value)) {
return value;
}
if (typeof value === 'string' && value.trim() !== '') {
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : undefined;
}
return undefined;
}
function parseJsonValue<T>(value: string | null | undefined, fallback: T): T {
if (!value) {
return fallback;
}
try {
return JSON.parse(value) as T;
} catch (error) {
console.warn('Failed to parse JSON value from DB:', error);
return fallback;
}
}
function inferPlaylistType(playlist: Record<string, unknown>): PlaylistType {
const explicitType = getStringValue(playlist.type);
if (
explicitType &&
PLAYLIST_TYPE_VALUES.has(explicitType as PlaylistType)
) {
return explicitType as PlaylistType;
}
if (getStringValue(playlist.serverUrl)) {
return PLAYLIST_TYPES.XTREAM;
}
if (getStringValue(playlist.macAddress)) {
return PLAYLIST_TYPES.STALKER;
}
if (getStringValue(playlist.filePath)) {
return PLAYLIST_TYPES.M3U_FILE;
}
if (getStringValue(playlist.url)) {
return PLAYLIST_TYPES.M3U_URL;
}
return PLAYLIST_TYPES.M3U_TEXT;
}
function buildPlaylistRow(
playlist: Record<string, unknown>
): schema.NewPlaylist | null {
const id = getStringValue(playlist._id) ?? getStringValue(playlist.id);
if (!id) {
return null;
}
const type = inferPlaylistType(playlist);
const portalUrl = getStringValue(playlist.portalUrl);
const url = getStringValue(playlist.url);
const nowIso = new Date().toISOString();
const updateDate = getNumericValue(playlist.updateDate);
return {
id,
name:
getStringValue(playlist.title) ??
getStringValue(playlist.name) ??
id,
serverUrl: getStringValue(playlist.serverUrl),
username: getStringValue(playlist.username),
password: getStringValue(playlist.password),
lastUpdated:
typeof updateDate === 'number'
? new Date(updateDate).toISOString()
: getStringValue(playlist.lastUpdated),
type,
userAgent: getStringValue(playlist.userAgent),
origin: getStringValue(playlist.origin),
referrer: getStringValue(playlist.referrer),
filePath: getStringValue(playlist.filePath),
autoRefresh: Boolean(playlist.autoRefresh),
macAddress: getStringValue(playlist.macAddress),
// Keep historical semantics: stalker portal URL has been stored in "url"
url: type === PLAYLIST_TYPES.STALKER ? (portalUrl ?? url) : url,
portalUrl,
count: getNumericValue(playlist.count),
importDate: getStringValue(playlist.importDate),
updateDate,
position: getNumericValue(playlist.position),
favorites:
playlist.favorites !== undefined
? JSON.stringify(playlist.favorites)
: undefined,
recentlyViewed:
playlist.recentlyViewed !== undefined
? JSON.stringify(playlist.recentlyViewed)
: undefined,
payload: JSON.stringify(playlist),
lastUsage: getStringValue(playlist.lastUsage) ?? nowIso,
};
}
function parseAppPlaylist(row: schema.Playlist): Record<string, unknown> {
const payload = parseJsonValue<Record<string, unknown> | null>(
row.payload,
null
);
if (payload && typeof payload === 'object') {
return {
...payload,
_id:
getStringValue(payload._id) ??
getStringValue(payload.id) ??
row.id,
title:
getStringValue(payload.title) ??
getStringValue(payload.name) ??
row.name,
};
}
const favorites = parseJsonValue<unknown[]>(row.favorites, []);
const recentlyViewed = parseJsonValue<unknown[]>(row.recentlyViewed, []);
const importDate = row.importDate ?? row.dateCreated ?? new Date().toISOString();
const portalUrl =
row.portalUrl ??
(row.type === PLAYLIST_TYPES.STALKER ? row.url : null);
return {
_id: row.id,
title: row.name,
count: row.count ?? 0,
importDate,
lastUsage: row.lastUsage ?? importDate,
favorites,
recentlyViewed,
autoRefresh: row.autoRefresh ?? false,
url: row.type === PLAYLIST_TYPES.M3U_URL ? row.url : undefined,
filePath: row.filePath ?? undefined,
userAgent: row.userAgent ?? undefined,
referrer: row.referrer ?? undefined,
origin: row.origin ?? undefined,
updateDate:
row.updateDate ??
(row.lastUpdated ? new Date(row.lastUpdated).getTime() : undefined),
position: row.position ?? undefined,
serverUrl: row.serverUrl ?? undefined,
username: row.username ?? undefined,
password: row.password ?? undefined,
macAddress: row.macAddress ?? undefined,
portalUrl: portalUrl ?? undefined,
};
}
/**
* Create a new playlist
*/
ipcMain.handle(
'DB_CREATE_PLAYLIST',
async (
event,
playlist: {
id: string;
name: string;
serverUrl?: string;
username?: string;
password?: string;
macAddress?: string;
url?: string;
type: string;
}
) => {
try {
const db = await getDatabase();
await db.insert(schema.playlists).values({
id: playlist.id,
name: playlist.name,
serverUrl: playlist.serverUrl,
username: playlist.username,
password: playlist.password,
macAddress: playlist.macAddress,
url: playlist.url,
portalUrl:
playlist.type === PLAYLIST_TYPES.STALKER
? playlist.url
: undefined,
// enforce supported types
type: playlist.type as PlaylistType,
});
return { success: true };
} catch (error) {
console.error('Error creating playlist:', error);
throw error;
}
}
);
/**
* Upsert a full app playlist object (meta + optional content payload)
*/
ipcMain.handle(
handleWorkerRequest('DB_CREATE_PLAYLIST', (playlist: Record<string, unknown>) => playlist);
handleWorkerRequest(
'DB_UPSERT_APP_PLAYLIST',
async (event, playlist: Record<string, unknown>) => {
try {
const db = await getDatabase();
const row = buildPlaylistRow(playlist);
if (!row) {
throw new Error('Playlist ID is required for upsert');
}
await db
.insert(schema.playlists)
.values(row)
.onConflictDoUpdate({
target: schema.playlists.id,
set: row,
});
return { success: true };
} catch (error) {
console.error('Error upserting app playlist:', error);
throw error;
}
}
(playlist: Record<string, unknown>) => playlist
);
/**
* Bulk upsert full app playlist objects
*/
ipcMain.handle(
handleWorkerRequest(
'DB_UPSERT_APP_PLAYLISTS',
async (event, playlists: Record<string, unknown>[]) => {
try {
if (!Array.isArray(playlists) || playlists.length === 0) {
return { success: true, count: 0 };
}
const db = await getDatabase();
let count = 0;
for (const playlist of playlists) {
const row = buildPlaylistRow(playlist);
if (!row) {
continue;
}
await db
.insert(schema.playlists)
.values(row)
.onConflictDoUpdate({
target: schema.playlists.id,
set: row,
});
count += 1;
}
return { success: true, count };
} catch (error) {
console.error('Error bulk upserting app playlists:', error);
throw error;
}
}
(playlists: Record<string, unknown>[]) => playlists
);
/**
* Get all app playlists as playlist objects (parsed from payload)
*/
ipcMain.handle('DB_GET_APP_PLAYLISTS', async () => {
try {
const db = await getDatabase();
const rows = await db.select().from(schema.playlists);
return rows.map((row) => parseAppPlaylist(row));
} catch (error) {
console.error('Error getting app playlists:', error);
throw error;
}
});
/**
* Get app playlist by ID as playlist object
*/
ipcMain.handle('DB_GET_APP_PLAYLIST', async (event, playlistId: string) => {
try {
const db = await getDatabase();
const rows = await db
.select()
.from(schema.playlists)
.where(eq(schema.playlists.id, playlistId))
.limit(1);
return rows[0] ? parseAppPlaylist(rows[0]) : null;
} catch (error) {
console.error('Error getting app playlist:', error);
throw error;
}
});
/**
* Get playlist by ID
*/
ipcMain.handle('DB_GET_PLAYLIST', async (event, playlistId: string) => {
try {
const db = await getDatabase();
const result = await db
.select()
.from(schema.playlists)
.where(eq(schema.playlists.id, playlistId))
.limit(1);
return result[0] || null;
} catch (error) {
console.error('Error getting playlist:', error);
throw error;
}
});
/**
* Update playlist
*/
ipcMain.handle(
handleWorkerRequest('DB_GET_APP_PLAYLISTS', () => ({}));
handleWorkerRequest('DB_GET_APP_PLAYLIST', (playlistId: string) => ({ playlistId }));
handleWorkerRequest('DB_GET_PLAYLIST', (playlistId: string) => ({ playlistId }));
handleWorkerRequest(
'DB_UPDATE_PLAYLIST',
async (
event,
(
playlistId: string,
updates: {
name?: string;
@@ -367,112 +33,61 @@ ipcMain.handle(
serverUrl?: string;
lastUpdated?: string;
}
) => ({
playlistId,
updates,
})
);
handleWorkerRequest('DB_GET_APP_STATE', (key: string) => ({ key }));
handleWorkerRequest('DB_SET_APP_STATE', (key: string, value: string) => ({
key,
value,
}));
ipcMain.handle(
'DB_DELETE_PLAYLIST',
async (
event,
playlistId: string,
operationId?: string
) => {
try {
const db = await getDatabase();
await db
.update(schema.playlists)
.set(updates)
.where(eq(schema.playlists.id, playlistId));
return { success: true };
return await requestWorkerWithEvents(
event,
'DB_DELETE_PLAYLIST',
{
playlistId,
operationId,
}
);
} catch (error) {
console.error('Error updating playlist:', error);
console.error('Error handling DB_DELETE_PLAYLIST:', error);
throw error;
}
}
);
/**
* Delete playlist and all related data
*/
ipcMain.handle('DB_DELETE_PLAYLIST', async (event, playlistId: string) => {
try {
const db = await getDatabase();
// Delete playlist (cascade will handle related data)
await db
.delete(schema.playlists)
.where(eq(schema.playlists.id, playlistId));
return { success: true };
} catch (error) {
console.error('Error deleting playlist:', error);
throw error;
}
});
/**
* Read an app-level state value by key
*/
ipcMain.handle('DB_GET_APP_STATE', async (event, key: string) => {
try {
const db = await getDatabase();
const rows = await db
.select()
.from(schema.appState)
.where(eq(schema.appState.key, key))
.limit(1);
return rows[0]?.value ?? null;
} catch (error) {
console.error('Error getting app state:', error);
throw error;
}
});
/**
* Upsert an app-level state key/value pair
*/
ipcMain.handle(
'DB_SET_APP_STATE',
async (event, key: string, value: string) => {
'DB_DELETE_ALL_PLAYLISTS',
async (event, operationId?: string) => {
try {
const db = await getDatabase();
const updatedAt = new Date().toISOString();
await db
.insert(schema.appState)
.values({
key,
value,
updatedAt,
})
.onConflictDoUpdate({
target: schema.appState.key,
set: { value, updatedAt },
});
return { success: true };
return await requestWorkerWithEvents(
event,
'DB_DELETE_ALL_PLAYLISTS',
{ operationId }
);
} catch (error) {
console.error('Error setting app state:', error);
console.error('Error handling DB_DELETE_ALL_PLAYLISTS:', error);
throw error;
}
}
);
/**
* Delete all playlists and related data from SQLite
*/
ipcMain.handle('DB_DELETE_ALL_PLAYLISTS', async () => {
ipcMain.handle('DB_CANCEL_OPERATION', async (_event, operationId: string) => {
try {
const db = await getDatabase();
// Delete in order respecting foreign key constraints
// First delete favorites and recently_viewed (they reference content)
await db.delete(schema.favorites);
await db.delete(schema.recentlyViewed);
// Then delete content (references categories)
await db.delete(schema.content);
// Then delete categories (references playlists)
await db.delete(schema.categories);
// Finally delete playlists
await db.delete(schema.playlists);
return { success: true };
return await databaseWorkerClient.cancel(operationId);
} catch (error) {
console.error('Error deleting all playlists:', error);
console.error('Error handling DB_CANCEL_OPERATION:', error);
throw error;
}
});
@@ -3,204 +3,34 @@
* Operations for managing user's recently viewed content
*/
import { and, desc, eq, inArray, sql } from 'drizzle-orm';
import { ipcMain } from 'electron';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import { handleWorkerRequest } from './worker-events.utils';
/**
* Get recently viewed items across all playlists
*/
ipcMain.handle('DB_GET_RECENTLY_VIEWED', async () => {
try {
const db = await getDatabase();
const result = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
playlist_id: schema.categories.playlistId,
playlist_name: schema.playlists.name,
viewed_at: schema.recentlyViewed.viewedAt,
})
.from(schema.recentlyViewed)
.innerJoin(
schema.content,
eq(schema.recentlyViewed.contentId, schema.content.id)
)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.innerJoin(
schema.playlists,
eq(schema.categories.playlistId, schema.playlists.id)
)
.orderBy(desc(schema.recentlyViewed.viewedAt))
.limit(100);
return result;
} catch (error) {
console.error('Error getting recently viewed:', error);
throw error;
}
});
handleWorkerRequest('DB_GET_RECENTLY_VIEWED', () => ({}));
handleWorkerRequest('DB_CLEAR_RECENTLY_VIEWED', () => ({}));
/**
* Clear all recently viewed items
*/
ipcMain.handle('DB_CLEAR_RECENTLY_VIEWED', async () => {
try {
const db = await getDatabase();
await db.delete(schema.recentlyViewed);
return { success: true };
} catch (error) {
console.error('Error clearing recently viewed:', error);
throw error;
}
});
handleWorkerRequest('DB_GET_RECENT_ITEMS', (playlistId: string) => ({
playlistId,
}));
/**
* Get recently viewed items for a specific playlist
*/
ipcMain.handle('DB_GET_RECENT_ITEMS', async (event, playlistId: string) => {
try {
const db = await getDatabase();
const result = await db
.select({
id: schema.content.id,
category_id: schema.content.categoryId,
title: schema.content.title,
rating: schema.content.rating,
added: schema.content.added,
poster_url: schema.content.posterUrl,
xtream_id: schema.content.xtreamId,
type: schema.content.type,
viewed_at: schema.recentlyViewed.viewedAt,
})
.from(schema.recentlyViewed)
.innerJoin(
schema.content,
eq(schema.recentlyViewed.contentId, schema.content.id)
)
.where(eq(schema.recentlyViewed.playlistId, playlistId))
.orderBy(desc(schema.recentlyViewed.viewedAt))
.limit(100);
return result;
} catch (error) {
console.error('Error getting recent items:', error);
throw error;
}
});
/**
* Add item to recently viewed
*/
ipcMain.handle(
handleWorkerRequest(
'DB_ADD_RECENT_ITEM',
async (event, contentId: number, playlistId: string) => {
try {
const db = await getDatabase();
// Check if already exists
const existing = await db
.select()
.from(schema.recentlyViewed)
.where(
and(
eq(schema.recentlyViewed.contentId, contentId),
eq(schema.recentlyViewed.playlistId, playlistId)
)
)
.limit(1);
if (existing.length > 0) {
// Update viewed_at timestamp
await db
.update(schema.recentlyViewed)
.set({ viewedAt: sql`CURRENT_TIMESTAMP` })
.where(
and(
eq(schema.recentlyViewed.contentId, contentId),
eq(schema.recentlyViewed.playlistId, playlistId)
)
);
} else {
// Insert new entry
await db.insert(schema.recentlyViewed).values({
contentId,
playlistId,
});
}
return { success: true };
} catch (error) {
console.error('Error adding recent item:', error);
throw error;
}
}
(contentId: number, playlistId: string) => ({
contentId,
playlistId,
})
);
/**
* Clear recently viewed for a specific playlist
*/
ipcMain.handle(
handleWorkerRequest(
'DB_CLEAR_PLAYLIST_RECENT_ITEMS',
async (event, playlistId: string) => {
try {
const db = await getDatabase();
// Get content IDs that belong to this playlist
const contentIds = await db
.select({ id: schema.content.id })
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(eq(schema.categories.playlistId, playlistId));
if (contentIds.length > 0) {
await db.delete(schema.recentlyViewed).where(
inArray(
schema.recentlyViewed.contentId,
contentIds.map((c) => c.id)
)
);
}
return { success: true };
} catch (error) {
console.error('Error clearing playlist recent items:', error);
throw error;
}
}
(playlistId: string) => ({
playlistId,
})
);
/**
* Remove specific item from recently viewed
*/
ipcMain.handle(
handleWorkerRequest(
'DB_REMOVE_RECENT_ITEM',
async (event, contentId: number, playlistId: string) => {
try {
const db = await getDatabase();
await db
.delete(schema.recentlyViewed)
.where(
and(
eq(schema.recentlyViewed.contentId, contentId),
eq(schema.recentlyViewed.playlistId, playlistId)
)
);
return { success: true };
} catch (error) {
console.error('Error removing recent item:', error);
throw error;
}
}
(contentId: number, playlistId: string) => ({
contentId,
playlistId,
})
);
@@ -0,0 +1,44 @@
import { ipcMain } from 'electron';
import { databaseWorkerClient } from '../../services/database-worker-client';
import type {
DbOperationEvent,
DbWorkerOperation,
} from '../../workers/database-worker.types';
export function forwardWorkerEvent(
event: Electron.IpcMainInvokeEvent,
workerEvent: DbOperationEvent
): void {
if (event.sender.isDestroyed()) {
return;
}
event.sender.send('DB_OPERATION_EVENT', workerEvent);
}
export function handleWorkerRequest<TArgs extends unknown[]>(
channel: DbWorkerOperation,
buildPayload: (...args: TArgs) => unknown
): void {
ipcMain.handle(channel, async (_event, ...args: TArgs) => {
try {
return await databaseWorkerClient.request(
channel,
buildPayload(...args)
);
} catch (error) {
console.error(`Error handling ${channel}:`, error);
throw error;
}
});
}
export async function requestWorkerWithEvents<TResult, TPayload>(
event: Electron.IpcMainInvokeEvent,
channel: DbWorkerOperation,
payload: TPayload
): Promise<TResult> {
return databaseWorkerClient.request<TResult>(channel, payload, {
onEvent: (workerEvent) => forwardWorkerEvent(event, workerEvent),
});
}
@@ -3,209 +3,50 @@
* Operations for refreshing and managing Xtream playlist data
*/
import { and, eq, inArray } from 'drizzle-orm';
import { ipcMain } from 'electron';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
import { requestWorkerWithEvents } from './worker-events.utils';
/**
* Delete all content and categories for an Xtream playlist
* Keeps the playlist entry but removes all imported data
*/
ipcMain.handle(
'DB_DELETE_XTREAM_CONTENT',
async (event, playlistId: string) => {
async (event, playlistId: string, operationId?: string) => {
try {
const db = await getDatabase();
// First, get all categories for this playlist (including hidden status)
const categories = await db
.select({
id: schema.categories.id,
xtreamId: schema.categories.xtreamId,
type: schema.categories.type,
hidden: schema.categories.hidden,
})
.from(schema.categories)
.where(eq(schema.categories.playlistId, playlistId));
const categoryIds = categories.map((c) => c.id);
// Save hidden categories info to restore after refresh
const hiddenCategories = categories
.filter((c) => c.hidden)
.map((c) => ({
xtreamId: c.xtreamId,
type: c.type,
}));
// Before deleting content, save xtreamIds of favorited and recently viewed items
// so they can be restored after refresh
let favoritedXtreamIds: number[] = [];
let recentlyViewedXtreamIds: {
xtreamId: number;
viewedAt: string;
}[] = [];
if (categoryIds.length > 0) {
// Get favorites with their xtreamIds
const favorites = await db
.select({
xtreamId: schema.content.xtreamId,
})
.from(schema.favorites)
.innerJoin(
schema.content,
eq(schema.favorites.contentId, schema.content.id)
)
.where(
and(
eq(schema.favorites.playlistId, playlistId),
inArray(schema.content.categoryId, categoryIds)
)
);
favoritedXtreamIds = favorites.map((f) => f.xtreamId);
// Get recently viewed with their xtreamIds and timestamps
const recentlyViewed = await db
.select({
xtreamId: schema.content.xtreamId,
viewedAt: schema.recentlyViewed.viewedAt,
})
.from(schema.recentlyViewed)
.innerJoin(
schema.content,
eq(schema.recentlyViewed.contentId, schema.content.id)
)
.where(
and(
eq(schema.recentlyViewed.playlistId, playlistId),
inArray(schema.content.categoryId, categoryIds)
)
);
recentlyViewedXtreamIds = recentlyViewed.map((rv) => ({
xtreamId: rv.xtreamId,
viewedAt: rv.viewedAt || new Date().toISOString(),
}));
// Delete all content for these categories
// (This will cascade delete favorites and recently viewed)
await db
.delete(schema.content)
.where(inArray(schema.content.categoryId, categoryIds));
}
// Delete all categories for this playlist
await db
.delete(schema.categories)
.where(eq(schema.categories.playlistId, playlistId));
// Return the saved xtreamIds so favorites, recently viewed, and hidden categories can be restored
return {
success: true,
favoritedXtreamIds,
recentlyViewedXtreamIds,
hiddenCategories,
};
return await requestWorkerWithEvents(
event,
'DB_DELETE_XTREAM_CONTENT',
{
playlistId,
operationId,
}
);
} catch (error) {
console.error('Error deleting Xtream content:', error);
console.error('Error handling DB_DELETE_XTREAM_CONTENT:', error);
throw error;
}
}
);
/**
* Restore favorites and recently viewed items after Xtream refresh
* Matches content by xtreamId and re-creates favorites/recently viewed entries
*/
ipcMain.handle(
'DB_RESTORE_XTREAM_USER_DATA',
async (
event,
playlistId: string,
favoritedXtreamIds: number[],
recentlyViewedXtreamIds: { xtreamId: number; viewedAt: string }[]
recentlyViewedXtreamIds: { xtreamId: number; viewedAt: string }[],
operationId?: string
) => {
try {
const db = await getDatabase();
// Restore favorites
if (favoritedXtreamIds.length > 0) {
// Find content IDs for the xtreamIds
const content = await db
.select({
id: schema.content.id,
xtreamId: schema.content.xtreamId,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.categories.playlistId, playlistId),
inArray(schema.content.xtreamId, favoritedXtreamIds)
)
);
// Re-create favorite entries
for (const item of content) {
await db.insert(schema.favorites).values({
contentId: item.id,
playlistId: playlistId,
addedAt: new Date().toISOString(),
});
return await requestWorkerWithEvents(
event,
'DB_RESTORE_XTREAM_USER_DATA',
{
playlistId,
favoritedXtreamIds,
recentlyViewedXtreamIds,
operationId,
}
}
// Restore recently viewed
if (recentlyViewedXtreamIds.length > 0) {
const xtreamIds = recentlyViewedXtreamIds.map(
(rv) => rv.xtreamId
);
// Find content IDs for the xtreamIds
const content = await db
.select({
id: schema.content.id,
xtreamId: schema.content.xtreamId,
})
.from(schema.content)
.innerJoin(
schema.categories,
eq(schema.content.categoryId, schema.categories.id)
)
.where(
and(
eq(schema.categories.playlistId, playlistId),
inArray(schema.content.xtreamId, xtreamIds)
)
);
// Create a map of xtreamId -> contentId
const xtreamIdToContentId = new Map(
content.map((item) => [item.xtreamId, item.id])
);
// Re-create recently viewed entries with original timestamps
for (const item of recentlyViewedXtreamIds) {
const contentId = xtreamIdToContentId.get(item.xtreamId);
if (contentId) {
await db.insert(schema.recentlyViewed).values({
contentId: contentId,
playlistId: playlistId,
viewedAt: item.viewedAt,
});
}
}
}
return { success: true };
);
} catch (error) {
console.error('Error restoring Xtream user data:', error);
console.error('Error handling DB_RESTORE_XTREAM_USER_DATA:', error);
throw error;
}
}
@@ -430,12 +430,17 @@ export default class EpgEvents {
): Promise<EpgProgram[]> {
try {
const db = await getDatabase();
const trimmedChannelId = channelId.trim();
if (!trimmedChannelId) {
return [];
}
// Try exact channel ID match first
let results = await db
.select()
.from(schema.epgPrograms)
.where(eq(schema.epgPrograms.channelId, channelId))
.where(eq(schema.epgPrograms.channelId, trimmedChannelId))
.orderBy(schema.epgPrograms.start)
.limit(500);
@@ -443,15 +448,24 @@ export default class EpgEvents {
return results.map(this.transformDbRowToEpgProgram);
}
// Try to find channel by display name
const channel = await db
// Try exact display name match before giving up. Using wildcard LIKE
// here can scan the whole table on the Electron main process.
let channel = await db
.select()
.from(schema.epgChannels)
.where(
sql`LOWER(${schema.epgChannels.displayName}) LIKE LOWER(${'%' + channelId + '%'})`
)
.where(eq(schema.epgChannels.displayName, trimmedChannelId))
.limit(1);
if (channel.length === 0) {
channel = await db
.select()
.from(schema.epgChannels)
.where(
sql`${schema.epgChannels.displayName} = ${trimmedChannelId} COLLATE NOCASE`
)
.limit(1);
}
if (channel.length > 0) {
results = await db
.select()
@@ -0,0 +1,90 @@
jest.mock('electron', () => ({
BrowserWindow: {
getAllWindows: () => [],
},
}));
jest.mock('../../environments/environment', () => ({
environment: {
production: false,
},
}));
import { sanitizePortalDebugEvent } from './portal-debug.events';
describe('sanitizePortalDebugEvent', () => {
it('turns Error objects into plain serializable data', () => {
const error = new Error('boom');
(error as Error & { code?: string }).code = 'ECONNREFUSED';
const event = sanitizePortalDebugEvent({
requestId: 'req-1',
provider: 'xtream',
operation: 'get_account_info',
transport: 'electron-main',
startedAt: new Date().toISOString(),
durationMs: 12,
status: 'error',
request: { url: 'http://localhost:3211/player_api.php' },
error,
});
expect(event.error).toEqual(
expect.objectContaining({
name: 'Error',
message: 'boom',
code: 'ECONNREFUSED',
})
);
});
it('replaces circular references with markers', () => {
const request: Record<string, unknown> = {
method: 'GET',
};
request['self'] = request;
const event = sanitizePortalDebugEvent({
requestId: 'req-2',
provider: 'stalker',
operation: 'get_profile',
transport: 'electron-main',
startedAt: new Date().toISOString(),
durationMs: 4,
status: 'success',
request,
response: { ok: true },
});
expect(event.request).toEqual({
method: 'GET',
self: '[Circular]',
});
});
it('strips functions from nested debug payloads', () => {
const event = sanitizePortalDebugEvent({
requestId: 'req-3',
provider: 'xtream',
operation: 'test',
transport: 'electron-main',
startedAt: new Date().toISOString(),
durationMs: 1,
status: 'success',
request: {
headers: {
Accept: 'application/json',
toJSON: () => 'nope',
},
},
response: undefined,
});
expect(event.request).toEqual({
headers: {
Accept: 'application/json',
toJSON: undefined,
},
});
});
});
@@ -2,13 +2,127 @@ import { BrowserWindow } from 'electron';
import { PORTAL_DEBUG_EVENT, PortalDebugEvent } from 'shared-interfaces';
import { environment } from '../../environments/environment';
const MAX_DEBUG_DEPTH = 6;
function sanitizePortalDebugValue(
value: unknown,
seen = new WeakSet<object>(),
depth = 0
): unknown {
if (
value == null ||
typeof value === 'string' ||
typeof value === 'number' ||
typeof value === 'boolean'
) {
return value;
}
if (typeof value === 'bigint') {
return value.toString();
}
if (
typeof value === 'function' ||
typeof value === 'symbol'
) {
return undefined;
}
if (depth >= MAX_DEBUG_DEPTH) {
return '[MaxDepth]';
}
if (value instanceof Error) {
const baseError = {
name: value.name,
message: value.message,
stack: value.stack,
} as Record<string, unknown>;
for (const [key, entry] of Object.entries(
value as unknown as Record<string, unknown>
)) {
baseError[key] = sanitizePortalDebugValue(
entry,
seen,
depth + 1
);
}
return baseError;
}
if (value instanceof Date) {
return value.toISOString();
}
if (value instanceof URL) {
return value.toString();
}
if (Array.isArray(value)) {
return value.map((entry) =>
sanitizePortalDebugValue(entry, seen, depth + 1)
);
}
if (value instanceof Map) {
return Object.fromEntries(
[...value.entries()].map(([key, entry]) => [
String(key),
sanitizePortalDebugValue(entry, seen, depth + 1),
])
);
}
if (value instanceof Set) {
return [...value].map((entry) =>
sanitizePortalDebugValue(entry, seen, depth + 1)
);
}
if (typeof value === 'object') {
if (seen.has(value)) {
return '[Circular]';
}
seen.add(value);
const entries = Object.entries(value as Record<string, unknown>).map(
([key, entry]) => [
key,
sanitizePortalDebugValue(entry, seen, depth + 1),
]
);
seen.delete(value);
return Object.fromEntries(entries);
}
return String(value);
}
export function sanitizePortalDebugEvent(
event: PortalDebugEvent
): PortalDebugEvent {
return {
...event,
request: sanitizePortalDebugValue(event.request),
response: sanitizePortalDebugValue(event.response),
error: sanitizePortalDebugValue(event.error),
};
}
export function emitPortalDebugEvent(event: PortalDebugEvent): void {
if (environment.production) {
return;
}
const serializedEvent = sanitizePortalDebugEvent(event);
const windows = BrowserWindow.getAllWindows();
for (const window of windows) {
window.webContents.send(PORTAL_DEBUG_EVENT, event);
window.webContents.send(PORTAL_DEBUG_EVENT, serializedEvent);
}
}
@@ -62,6 +62,7 @@ ipcMain.handle(
url: string;
params: Record<string, string>;
requestId?: string;
suppressErrorLog?: boolean;
}
) => {
const startedAt = Date.now();
@@ -156,10 +157,12 @@ ipcMain.handle(
emitPortalDebugEvent(debugEvent);
}
console.error(
'[XTREAM_REQUEST] Failed',
formatXtreamError(error, payload.url, payload.params?.action)
);
if (!payload.suppressErrorLog) {
console.error(
'[XTREAM_REQUEST] Failed',
formatXtreamError(error, payload.url, payload.params?.action)
);
}
// Format error response
if (axios.isAxiosError(error)) {
@@ -0,0 +1,246 @@
import type { DatabaseWorkerClient as DatabaseWorkerClientType } from './database-worker-client';
const mockWorkerInstances: any[] = [];
jest.mock('electron', () => ({
app: {
isPackaged: false,
getAppPath: () => '/mock/app.asar',
},
}));
jest.mock('worker_threads', () => {
const { EventEmitter } = require('events');
class MockWorker extends EventEmitter {
postMessage = jest.fn();
terminate = jest.fn().mockResolvedValue(0);
}
return {
Worker: jest.fn().mockImplementation(() => {
const worker = new MockWorker();
mockWorkerInstances.push(worker);
return worker;
}),
};
});
describe('DatabaseWorkerClient', () => {
let DatabaseWorkerClient: typeof DatabaseWorkerClientType;
let clients: DatabaseWorkerClientType[];
beforeEach(async () => {
jest.resetModules();
mockWorkerInstances.length = 0;
clients = [];
({ DatabaseWorkerClient } = await import('./database-worker-client'));
});
afterEach(async () => {
await Promise.all(
clients.map((client) => client.shutdown().catch(() => undefined))
);
});
function createClient(): DatabaseWorkerClientType {
const client = new DatabaseWorkerClient();
clients.push(client);
return client;
}
async function flushPromises(): Promise<void> {
await new Promise((resolve) => setImmediate(resolve));
}
it('resolves requests after the worker reports ready and returns a response', async () => {
const client = createClient();
const requestPromise = client.request('DB_GLOBAL_SEARCH', {
searchTerm: 'matrix',
});
const worker = mockWorkerInstances[0];
worker.emit('message', { type: 'ready' });
await flushPromises();
expect(worker.postMessage).toHaveBeenCalledTimes(1);
const request = worker.postMessage.mock.calls[0][0];
worker.emit('message', {
type: 'response',
requestId: request.requestId,
success: true,
result: [{ id: 1 }],
});
await expect(requestPromise).resolves.toEqual([{ id: 1 }]);
});
it('forwards worker events to the pending request callback', async () => {
const client = createClient();
const onEvent = jest.fn();
const requestPromise = client.request(
'DB_SAVE_CONTENT',
{ playlistId: 'xtream-1', streams: [], type: 'movie' },
{ onEvent }
);
const worker = mockWorkerInstances[0];
worker.emit('message', { type: 'ready' });
await flushPromises();
const request = worker.postMessage.mock.calls[0][0];
const event = {
operationId: 'op-1',
operation: 'save-content',
status: 'progress',
current: 10,
total: 100,
};
worker.emit('message', {
type: 'event',
requestId: request.requestId,
event,
});
expect(onEvent).toHaveBeenCalledWith(event);
worker.emit('message', {
type: 'response',
requestId: request.requestId,
success: true,
result: { success: true, count: 10 },
});
await expect(requestPromise).resolves.toEqual({
success: true,
count: 10,
});
});
it('turns serialized worker errors into rejected Error instances', async () => {
const client = createClient();
const requestPromise = client.request('DB_DELETE_PLAYLIST', {
playlistId: 'xtream-1',
});
const worker = mockWorkerInstances[0];
worker.emit('message', { type: 'ready' });
await flushPromises();
const request = worker.postMessage.mock.calls[0][0];
worker.emit('message', {
type: 'response',
requestId: request.requestId,
success: false,
error: {
name: 'SqliteBusyError',
message: 'database is locked',
stack: 'stack-trace',
},
});
await expect(requestPromise).rejects.toMatchObject({
name: 'SqliteBusyError',
message: 'database is locked',
});
});
it('preserves AbortError responses from the worker', async () => {
const client = createClient();
const requestPromise = client.request('DB_SAVE_CONTENT', {
playlistId: 'xtream-1',
streams: [],
type: 'movie',
});
const worker = mockWorkerInstances[0];
worker.emit('message', { type: 'ready' });
await flushPromises();
const request = worker.postMessage.mock.calls[0][0];
worker.emit('message', {
type: 'response',
requestId: request.requestId,
success: false,
error: {
name: 'AbortError',
message: 'Operation "save-content" was cancelled',
},
});
await expect(requestPromise).rejects.toMatchObject({
name: 'AbortError',
message: 'Operation "save-content" was cancelled',
});
});
it('posts cancel messages to the current worker', async () => {
const client = createClient();
const requestPromise = client.request('DB_DELETE_PLAYLIST', {
playlistId: 'xtream-1',
});
const worker = mockWorkerInstances[0];
worker.emit('message', { type: 'ready' });
await flushPromises();
await expect(client.cancel('operation-123')).resolves.toEqual({
success: true,
});
expect(worker.postMessage).toHaveBeenCalledWith({
type: 'cancel',
operationId: 'operation-123',
});
const request = worker.postMessage.mock.calls[0][0];
worker.emit('message', {
type: 'response',
requestId: request.requestId,
success: true,
result: { success: true },
});
await expect(requestPromise).resolves.toEqual({ success: true });
});
it('rejects pending work on worker exit and starts a fresh worker for the next request', async () => {
const client = createClient();
const firstRequest = client.request('DB_SEARCH_CONTENT', {
playlistId: 'xtream-1',
searchTerm: 'matrix',
types: ['movie'],
});
const firstWorker = mockWorkerInstances[0];
firstWorker.emit('message', { type: 'ready' });
await flushPromises();
firstWorker.emit('exit', 1);
await expect(firstRequest).rejects.toThrow(
'Database worker stopped with exit code 1'
);
const secondRequest = client.request('DB_SEARCH_CONTENT', {
playlistId: 'xtream-1',
searchTerm: 'neo',
types: ['movie'],
});
const secondWorker = mockWorkerInstances[1];
secondWorker.emit('message', { type: 'ready' });
await flushPromises();
const request = secondWorker.postMessage.mock.calls[0][0];
secondWorker.emit('message', {
type: 'response',
requestId: request.requestId,
success: true,
result: [],
});
await expect(secondRequest).resolves.toEqual([]);
});
});
@@ -0,0 +1,233 @@
import { app } from 'electron';
import { randomUUID } from 'crypto';
import * as path from 'path';
import { pathToFileURL } from 'url';
import { Worker } from 'worker_threads';
import type {
DbOperationEvent,
DbWorkerMessage,
DbWorkerOperation,
} from '../workers/database-worker.types';
type PendingRequest = {
resolve: (value: unknown) => void;
reject: (reason?: unknown) => void;
onEvent?: (event: DbOperationEvent) => void;
};
type RequestOptions = {
onEvent?: (event: DbOperationEvent) => void;
};
function createWorkerError(error: {
message: string;
name?: string;
stack?: string;
}): Error {
const workerError = new Error(error.message);
workerError.name = error.name || 'DatabaseWorkerError';
workerError.stack = error.stack || workerError.stack;
return workerError;
}
export class DatabaseWorkerClient {
private worker: Worker | null = null;
private readyPromise: Promise<void> | null = null;
private readyResolve: (() => void) | null = null;
private readyReject: ((reason?: unknown) => void) | null = null;
private pendingRequests = new Map<string, PendingRequest>();
private shuttingDown = false;
async request<TResult>(
operation: DbWorkerOperation,
payload: unknown,
options?: RequestOptions
): Promise<TResult> {
await this.ensureWorker();
const requestId = randomUUID();
return new Promise<TResult>((resolve, reject) => {
this.pendingRequests.set(requestId, {
resolve: resolve as (value: unknown) => void,
reject,
onEvent: options?.onEvent,
});
this.worker?.postMessage({
type: 'request',
requestId,
operation,
payload,
});
});
}
async cancel(operationId: string): Promise<{ success: boolean }> {
if (!operationId) {
return { success: false };
}
await this.ensureWorker();
this.worker?.postMessage({
type: 'cancel',
operationId,
});
return { success: true };
}
async shutdown(): Promise<void> {
this.shuttingDown = true;
this.readyResolve = null;
this.readyReject = null;
this.readyPromise = null;
if (!this.worker) {
return;
}
const currentWorker = this.worker;
this.worker = null;
this.rejectAllPending(new Error('Database worker shut down'));
try {
await currentWorker.terminate();
} finally {
this.shuttingDown = false;
}
}
private async ensureWorker(): Promise<void> {
if (this.worker && this.readyPromise) {
return this.readyPromise;
}
this.createWorker();
return this.readyPromise!;
}
private createWorker(): void {
const workerPath = this.resolveWorkerPath();
const workerURL = pathToFileURL(workerPath);
const nativeModulesPath = app.isPackaged
? path.join(
path.dirname(app.getAppPath()),
'app.asar.unpacked',
'node_modules'
)
: undefined;
this.readyPromise = new Promise<void>((resolve, reject) => {
this.readyResolve = resolve;
this.readyReject = reject;
});
this.worker = new Worker(workerURL, {
workerData: { nativeModulesPath },
});
this.worker.on('message', (message: DbWorkerMessage) => {
this.handleMessage(message);
});
this.worker.on('error', (error) => {
this.handleWorkerFailure(error);
});
this.worker.on('exit', (code) => {
this.handleWorkerExit(code);
});
}
private resolveWorkerPath(): string {
if (app.isPackaged) {
const resourcesPath = path.dirname(app.getAppPath());
return path.join(
resourcesPath,
'dist',
'apps',
'electron-backend',
'workers',
'database.worker.js'
);
}
return path.join(__dirname, 'workers', 'database.worker.js');
}
private handleMessage(message: DbWorkerMessage): void {
if (message.type === 'ready') {
this.readyResolve?.();
this.readyResolve = null;
this.readyReject = null;
return;
}
if (message.type === 'event') {
this.pendingRequests.get(message.requestId)?.onEvent?.(message.event);
return;
}
const pendingRequest = this.pendingRequests.get(message.requestId);
if (!pendingRequest) {
return;
}
this.pendingRequests.delete(message.requestId);
if (message.success) {
pendingRequest.resolve(message.result);
return;
}
pendingRequest.reject(
createWorkerError(
message.error ?? { message: 'Database worker request failed' }
)
);
}
private handleWorkerFailure(error: Error): void {
if (this.readyReject) {
this.readyReject(error);
}
this.resetWorkerState();
this.rejectAllPending(error);
}
private handleWorkerExit(code: number): void {
if (this.shuttingDown) {
return;
}
const error =
code === 0
? new Error('Database worker exited unexpectedly')
: new Error(`Database worker stopped with exit code ${code}`);
if (this.readyReject) {
this.readyReject(error);
}
this.resetWorkerState();
this.rejectAllPending(error);
}
private resetWorkerState(): void {
this.worker = null;
this.readyPromise = null;
this.readyResolve = null;
this.readyReject = null;
}
private rejectAllPending(error: Error): void {
const pendingRequests = [...this.pendingRequests.values()];
this.pendingRequests.clear();
pendingRequests.forEach((request) => {
request.reject(error);
});
}
}
export const databaseWorkerClient = new DatabaseWorkerClient();
@@ -0,0 +1,139 @@
export const DB_WORKER_OPERATIONS = [
'DB_HAS_CATEGORIES',
'DB_GET_CATEGORIES',
'DB_SAVE_CATEGORIES',
'DB_GET_ALL_CATEGORIES',
'DB_UPDATE_CATEGORY_VISIBILITY',
'DB_HAS_CONTENT',
'DB_GET_CONTENT',
'DB_GET_GLOBAL_RECENTLY_ADDED',
'DB_SAVE_CONTENT',
'DB_GET_CONTENT_BY_XTREAM_ID',
'DB_SEARCH_CONTENT',
'DB_GLOBAL_SEARCH',
'DB_CREATE_PLAYLIST',
'DB_UPSERT_APP_PLAYLIST',
'DB_UPSERT_APP_PLAYLISTS',
'DB_GET_APP_PLAYLISTS',
'DB_GET_APP_PLAYLIST',
'DB_GET_PLAYLIST',
'DB_UPDATE_PLAYLIST',
'DB_DELETE_PLAYLIST',
'DB_GET_APP_STATE',
'DB_SET_APP_STATE',
'DB_DELETE_ALL_PLAYLISTS',
'DB_DELETE_XTREAM_CONTENT',
'DB_RESTORE_XTREAM_USER_DATA',
'DB_ADD_FAVORITE',
'DB_REMOVE_FAVORITE',
'DB_IS_FAVORITE',
'DB_GET_FAVORITES',
'DB_GET_GLOBAL_FAVORITES',
'DB_GET_ALL_GLOBAL_FAVORITES',
'DB_REORDER_GLOBAL_FAVORITES',
'DB_GET_RECENTLY_VIEWED',
'DB_CLEAR_RECENTLY_VIEWED',
'DB_GET_RECENT_ITEMS',
'DB_ADD_RECENT_ITEM',
'DB_CLEAR_PLAYLIST_RECENT_ITEMS',
'DB_REMOVE_RECENT_ITEM',
'DB_SAVE_PLAYBACK_POSITION',
'DB_GET_PLAYBACK_POSITION',
'DB_GET_SERIES_PLAYBACK_POSITIONS',
'DB_GET_RECENT_PLAYBACK_POSITIONS',
'DB_GET_ALL_PLAYBACK_POSITIONS',
'DB_CLEAR_PLAYBACK_POSITION',
] as const;
export type DbWorkerOperation = (typeof DB_WORKER_OPERATIONS)[number];
export const DB_OPERATION_NAMES = {
SAVE_CONTENT: 'save-content',
DELETE_XTREAM_CONTENT: 'delete-xtream-content',
RESTORE_XTREAM_USER_DATA: 'restore-xtream-user-data',
DELETE_PLAYLIST: 'delete-playlist',
DELETE_ALL_PLAYLISTS: 'delete-all-playlists',
} as const;
export type DbOperationName =
(typeof DB_OPERATION_NAMES)[keyof typeof DB_OPERATION_NAMES];
export const DB_OPERATION_PHASES = {
PREPARING_CONTENT: 'preparing-content',
SAVING_CONTENT: 'saving-content',
COLLECTING_USER_DATA: 'collecting-user-data',
DELETING_FAVORITES: 'deleting-favorites',
DELETING_RECENTLY_VIEWED: 'deleting-recently-viewed',
DELETING_PLAYBACK_POSITIONS: 'deleting-playback-positions',
DELETING_DOWNLOADS: 'deleting-downloads',
DELETING_CONTENT: 'deleting-content',
DELETING_CATEGORIES: 'deleting-categories',
DELETING_PLAYLIST: 'deleting-playlist',
DELETING_PLAYLISTS: 'deleting-playlists',
RESTORING_FAVORITES: 'restoring-favorites',
RESTORING_RECENTLY_VIEWED: 'restoring-recently-viewed',
} as const;
export type DbOperationStatus =
| 'started'
| 'progress'
| 'completed'
| 'cancelled'
| 'error';
export interface DbOperationEvent {
operationId?: string;
operation: DbOperationName | string;
playlistId?: string;
status: DbOperationStatus;
phase?: string;
current?: number;
total?: number;
increment?: number;
error?: string;
}
export interface SerializedWorkerError {
name?: string;
message: string;
stack?: string;
}
export interface DbWorkerRequestMessage<TPayload = unknown> {
type: 'request';
requestId: string;
operation: DbWorkerOperation;
payload: TPayload;
}
export interface DbWorkerCancelMessage {
type: 'cancel';
operationId: string;
}
export interface DbWorkerReadyMessage {
type: 'ready';
}
export interface DbWorkerEventMessage {
type: 'event';
requestId: string;
event: DbOperationEvent;
}
export interface DbWorkerResponseMessage<TResult = unknown> {
type: 'response';
requestId: string;
success: boolean;
result?: TResult;
error?: SerializedWorkerError;
}
export type DbWorkerIncomingMessage =
| DbWorkerRequestMessage
| DbWorkerCancelMessage;
export type DbWorkerMessage =
| DbWorkerReadyMessage
| DbWorkerEventMessage
| DbWorkerResponseMessage;
@@ -0,0 +1,91 @@
import type BetterSqlite3 from 'better-sqlite3';
import * as schema from 'database-schema';
import { getIptvnatorDatabasePath } from 'database-path-utils';
import { drizzle } from 'drizzle-orm/better-sqlite3';
import { existsSync } from 'fs';
import { createRequire } from 'module';
import { join } from 'path';
import { workerData } from 'worker_threads';
import type { AppDatabase } from '../database/database.types';
let Database: typeof BetterSqlite3;
function loadBetterSqlite3(): typeof BetterSqlite3 {
if (
workerData &&
typeof workerData === 'object' &&
'nativeModulesPath' in workerData &&
typeof workerData.nativeModulesPath === 'string' &&
existsSync(workerData.nativeModulesPath)
) {
try {
const nativeRequire = createRequire(
join(workerData.nativeModulesPath, 'index.js')
);
return nativeRequire('better-sqlite3');
} catch (error) {
console.error(
'[DB Worker] Failed to load better-sqlite3 from workerData path:',
error
);
}
}
if (
(process as NodeJS.Process & { resourcesPath?: string }).resourcesPath
) {
const resourcesPath = (
process as NodeJS.Process & { resourcesPath?: string }
).resourcesPath!;
const unpackedPath = join(
resourcesPath,
'app.asar.unpacked',
'node_modules'
);
if (existsSync(unpackedPath)) {
try {
const nativeRequire = createRequire(join(unpackedPath, 'index.js'));
return nativeRequire('better-sqlite3');
} catch (error) {
console.error(
'[DB Worker] Failed to load better-sqlite3 from resourcesPath:',
error
);
}
}
}
// eslint-disable-next-line @typescript-eslint/no-require-imports
return require('better-sqlite3');
}
Database = loadBetterSqlite3();
let db: AppDatabase | null = null;
let sqlite: BetterSqlite3.Database | null = null;
export async function getWorkerDatabase(): Promise<AppDatabase> {
if (db) {
return db;
}
const filePath = getIptvnatorDatabasePath();
sqlite = new Database(filePath);
sqlite.pragma('foreign_keys = ON');
sqlite.pragma('journal_mode = WAL');
sqlite.pragma('busy_timeout = 5000');
db = drizzle(sqlite, { schema });
return db;
}
export function closeWorkerDatabase(): void {
if (!sqlite) {
return;
}
sqlite.close();
sqlite = null;
db = null;
}
@@ -0,0 +1,828 @@
import {
closeWorkerDatabase,
getWorkerDatabase,
} from './database.worker-connection';
import { parentPort } from 'worker_threads';
import type {
DbOperationEvent,
DbWorkerIncomingMessage,
DbWorkerMessage,
DbWorkerRequestMessage,
} from './database-worker.types';
import {
DB_OPERATION_NAMES,
DB_OPERATION_PHASES,
} from './database-worker.types';
import {
getAllCategories,
getCategories,
hasCategories,
saveCategories,
updateCategoryVisibility,
} from '../database/operations/category.operations';
import {
addFavorite,
getAllGlobalFavorites,
getFavorites,
getGlobalFavorites,
isFavorite,
removeFavorite,
reorderGlobalFavorites,
} from '../database/operations/favorites.operations';
import {
getContent,
getContentByXtreamId,
getGlobalRecentlyAdded,
globalSearch,
hasContent,
saveContent,
searchContent,
} from '../database/operations/content.operations';
import {
clearPlaybackPosition,
getAllPlaybackPositions,
getPlaybackPosition,
getRecentPlaybackPositions,
getSeriesPlaybackPositions,
savePlaybackPosition,
} from '../database/operations/playback-position.operations';
import {
createPlaylist,
deleteAllPlaylists,
deletePlaylist,
getAppPlaylist,
getAppPlaylists,
getAppState,
getPlaylist,
setAppState,
updatePlaylist,
upsertAppPlaylist,
upsertAppPlaylists,
} from '../database/operations/playlist.operations';
import {
addRecentItem,
clearPlaylistRecentItems,
clearRecentlyViewed,
getRecentItems,
getRecentlyViewed,
removeRecentItem,
} from '../database/operations/recently-viewed.operations';
import {
deleteXtreamContent,
restoreXtreamUserData,
} from '../database/operations/xtream.operations';
const loggerLabel = '[DB Worker]';
const batchDelayMs = Number.parseInt(
process.env['IPTVNATOR_DB_WORKER_BATCH_DELAY_MS'] ?? '0',
10
);
type ActiveOperationState = {
cancelled: boolean;
};
type OperationController = {
control: {
checkpoint: () => Promise<void>;
onProgress: (progress: {
phase: string;
current?: number;
total?: number;
increment?: number;
}) => Promise<void>;
};
emitStarted: (event: Partial<DbOperationEvent> & { phase: string }) => void;
emitCompleted: (
event?: Partial<DbOperationEvent> & { phase?: string }
) => void;
emitCancelled: (
event?: Partial<DbOperationEvent> & { phase?: string }
) => void;
emitError: (
error: unknown,
event?: Partial<DbOperationEvent> & { phase?: string }
) => void;
cleanup: () => void;
};
const activeOperations = new Map<string, ActiveOperationState>();
if (!parentPort) {
throw new Error('Database worker must be started with a parent port');
}
function createAbortError(message: string): Error {
const error = new Error(message);
error.name = 'AbortError';
return error;
}
function isAbortError(error: unknown): boolean {
return error instanceof Error && error.name === 'AbortError';
}
function serializeError(error: unknown) {
if (error instanceof Error) {
return {
name: error.name,
message: error.message,
stack: error.stack,
};
}
return {
message: String(error),
};
}
function postMessage(message: DbWorkerMessage): void {
parentPort?.postMessage(message);
}
function postEvent(requestId: string, event: DbOperationEvent): void {
postMessage({
type: 'event',
requestId,
event,
});
}
async function pauseBetweenBatches(): Promise<void> {
if (batchDelayMs <= 0) {
return;
}
await new Promise((resolve) => setTimeout(resolve, batchDelayMs));
}
function createOperationController(config: {
operationId?: string;
operation: string;
playlistId?: string;
requestId: string;
cancellable?: boolean;
}): OperationController {
const { operationId, operation, playlistId, requestId } = config;
const cancellable = config.cancellable ?? true;
const activeState =
operationId && cancellable
? { cancelled: false }
: null;
if (operationId && activeState) {
activeOperations.set(operationId, activeState);
}
let lastEvent: Partial<DbOperationEvent> = {};
const send = (
status: DbOperationEvent['status'],
event: Partial<DbOperationEvent> = {}
): void => {
const mergedEvent: DbOperationEvent = {
operationId,
operation,
playlistId,
status,
phase: event.phase ?? lastEvent.phase,
current: event.current ?? lastEvent.current,
total: event.total ?? lastEvent.total,
increment: event.increment,
error: event.error,
};
lastEvent = {
phase: mergedEvent.phase,
current: mergedEvent.current,
total: mergedEvent.total,
};
postEvent(requestId, mergedEvent);
};
const checkpoint = async (): Promise<void> => {
if (activeState?.cancelled) {
throw createAbortError(`Operation "${operation}" was cancelled`);
}
await pauseBetweenBatches();
if (activeState?.cancelled) {
throw createAbortError(`Operation "${operation}" was cancelled`);
}
};
return {
control: {
checkpoint,
onProgress: async (progress) => {
send('progress', progress);
},
},
emitStarted: (event) => {
send('started', event);
},
emitCompleted: (event) => {
send('completed', event);
},
emitCancelled: (event) => {
send('cancelled', event);
},
emitError: (error, event) => {
send('error', {
...event,
error:
error instanceof Error ? error.message : String(error),
});
},
cleanup: () => {
if (operationId) {
activeOperations.delete(operationId);
}
},
};
}
async function executeTrackedOperation<TResult>(
config: Parameters<typeof createOperationController>[0],
handler: (controller: OperationController) => Promise<TResult>
): Promise<TResult> {
const controller = createOperationController(config);
try {
return await handler(controller);
} catch (error) {
if (isAbortError(error)) {
controller.emitCancelled();
} else {
controller.emitError(error);
}
throw error;
} finally {
controller.cleanup();
}
}
async function executeRequest(message: DbWorkerRequestMessage) {
const db = await getWorkerDatabase();
switch (message.operation) {
case 'DB_HAS_CATEGORIES': {
const payload = message.payload as {
playlistId: string;
type: 'live' | 'movies' | 'series';
};
return hasCategories(db, payload.playlistId, payload.type);
}
case 'DB_GET_CATEGORIES': {
const payload = message.payload as {
playlistId: string;
type: 'live' | 'movies' | 'series';
};
return getCategories(db, payload.playlistId, payload.type);
}
case 'DB_SAVE_CATEGORIES': {
const payload = message.payload as {
playlistId: string;
categories: Array<{
category_name: string;
category_id: number;
}>;
type: 'live' | 'movies' | 'series';
hiddenCategoryXtreamIds?: number[];
};
return saveCategories(
db,
payload.playlistId,
payload.categories,
payload.type,
payload.hiddenCategoryXtreamIds
);
}
case 'DB_GET_ALL_CATEGORIES': {
const payload = message.payload as {
playlistId: string;
type: 'live' | 'movies' | 'series';
};
return getAllCategories(db, payload.playlistId, payload.type);
}
case 'DB_UPDATE_CATEGORY_VISIBILITY': {
const payload = message.payload as {
categoryIds: number[];
hidden: boolean;
};
return updateCategoryVisibility(
db,
payload.categoryIds,
payload.hidden
);
}
case 'DB_HAS_CONTENT': {
const payload = message.payload as {
playlistId: string;
type: 'live' | 'movie' | 'series';
};
return hasContent(db, payload.playlistId, payload.type);
}
case 'DB_GET_CONTENT': {
const payload = message.payload as {
playlistId: string;
type: 'live' | 'movie' | 'series';
};
return getContent(db, payload.playlistId, payload.type);
}
case 'DB_GET_GLOBAL_RECENTLY_ADDED': {
const payload = message.payload as {
kind?: 'all' | 'vod' | 'series';
limit?: number;
};
return getGlobalRecentlyAdded(db, payload.kind, payload.limit);
}
case 'DB_SAVE_CONTENT': {
const payload = message.payload as {
playlistId: string;
streams: Array<Record<string, unknown>>;
type: 'live' | 'movie' | 'series';
operationId?: string;
};
return executeTrackedOperation(
{
requestId: message.requestId,
operation: DB_OPERATION_NAMES.SAVE_CONTENT,
operationId: payload.operationId,
playlistId: payload.playlistId,
},
async (controller) => {
controller.emitStarted({
phase: DB_OPERATION_PHASES.PREPARING_CONTENT,
current: 0,
total: payload.streams.length,
});
const result = await saveContent(
db,
payload.playlistId,
payload.streams,
payload.type,
controller.control
);
controller.emitCompleted({
phase: DB_OPERATION_PHASES.SAVING_CONTENT,
current: result.count,
total: result.count,
});
return result;
}
);
}
case 'DB_GET_CONTENT_BY_XTREAM_ID': {
const payload = message.payload as {
xtreamId: number;
playlistId: string;
};
return getContentByXtreamId(
db,
payload.xtreamId,
payload.playlistId
);
}
case 'DB_SEARCH_CONTENT': {
const payload = message.payload as {
playlistId: string;
searchTerm: string;
types: string[];
excludeHidden?: boolean;
};
return searchContent(
db,
payload.playlistId,
payload.searchTerm,
payload.types,
payload.excludeHidden
);
}
case 'DB_GLOBAL_SEARCH': {
const payload = message.payload as {
searchTerm: string;
types: string[];
excludeHidden?: boolean;
};
return globalSearch(
db,
payload.searchTerm,
payload.types,
payload.excludeHidden
);
}
case 'DB_CREATE_PLAYLIST': {
return createPlaylist(
db,
message.payload as {
id: string;
name: string;
serverUrl?: string;
username?: string;
password?: string;
macAddress?: string;
url?: string;
type: string;
}
);
}
case 'DB_UPSERT_APP_PLAYLIST': {
return upsertAppPlaylist(
db,
message.payload as Record<string, unknown>
);
}
case 'DB_UPSERT_APP_PLAYLISTS': {
return upsertAppPlaylists(
db,
message.payload as Record<string, unknown>[]
);
}
case 'DB_GET_APP_PLAYLISTS':
return getAppPlaylists(db);
case 'DB_GET_APP_PLAYLIST': {
const payload = message.payload as { playlistId: string };
return getAppPlaylist(db, payload.playlistId);
}
case 'DB_GET_PLAYLIST': {
const payload = message.payload as { playlistId: string };
return getPlaylist(db, payload.playlistId);
}
case 'DB_UPDATE_PLAYLIST': {
const payload = message.payload as {
playlistId: string;
updates: {
name?: string;
username?: string;
password?: string;
serverUrl?: string;
lastUpdated?: string;
};
};
return updatePlaylist(db, payload.playlistId, payload.updates);
}
case 'DB_DELETE_PLAYLIST': {
const payload = message.payload as {
playlistId: string;
operationId?: string;
};
return executeTrackedOperation(
{
requestId: message.requestId,
operation: DB_OPERATION_NAMES.DELETE_PLAYLIST,
operationId: payload.operationId,
playlistId: payload.playlistId,
},
async (controller) => {
controller.emitStarted({
phase: DB_OPERATION_PHASES.DELETING_FAVORITES,
current: 0,
});
const result = await deletePlaylist(
db,
payload.playlistId,
controller.control
);
controller.emitCompleted({
phase: DB_OPERATION_PHASES.DELETING_PLAYLIST,
current: 1,
total: 1,
});
return result;
}
);
}
case 'DB_GET_APP_STATE': {
const payload = message.payload as { key: string };
return getAppState(db, payload.key);
}
case 'DB_SET_APP_STATE': {
const payload = message.payload as { key: string; value: string };
return setAppState(db, payload.key, payload.value);
}
case 'DB_DELETE_ALL_PLAYLISTS': {
const payload = message.payload as { operationId?: string };
return executeTrackedOperation(
{
requestId: message.requestId,
operation: DB_OPERATION_NAMES.DELETE_ALL_PLAYLISTS,
operationId: payload.operationId,
cancellable: false,
},
async (controller) => {
controller.emitStarted({
phase: DB_OPERATION_PHASES.DELETING_FAVORITES,
current: 0,
total: 7,
});
const result = await deleteAllPlaylists(
db,
controller.control
);
controller.emitCompleted({
phase: DB_OPERATION_PHASES.DELETING_PLAYLISTS,
current: 7,
total: 7,
});
return result;
}
);
}
case 'DB_DELETE_XTREAM_CONTENT': {
const payload = message.payload as {
playlistId: string;
operationId?: string;
};
return executeTrackedOperation(
{
requestId: message.requestId,
operation: DB_OPERATION_NAMES.DELETE_XTREAM_CONTENT,
operationId: payload.operationId,
playlistId: payload.playlistId,
},
async (controller) => {
controller.emitStarted({
phase: DB_OPERATION_PHASES.COLLECTING_USER_DATA,
current: 0,
});
const result = await deleteXtreamContent(
db,
payload.playlistId,
controller.control
);
controller.emitCompleted({
phase: DB_OPERATION_PHASES.DELETING_CATEGORIES,
});
return result;
}
);
}
case 'DB_RESTORE_XTREAM_USER_DATA': {
const payload = message.payload as {
playlistId: string;
favoritedXtreamIds: number[];
recentlyViewedXtreamIds: { xtreamId: number; viewedAt: string }[];
operationId?: string;
};
return executeTrackedOperation(
{
requestId: message.requestId,
operation: DB_OPERATION_NAMES.RESTORE_XTREAM_USER_DATA,
operationId: payload.operationId,
playlistId: payload.playlistId,
},
async (controller) => {
const totalItems =
payload.favoritedXtreamIds.length +
payload.recentlyViewedXtreamIds.length;
controller.emitStarted({
phase: DB_OPERATION_PHASES.RESTORING_FAVORITES,
current: 0,
total: totalItems,
});
const result = await restoreXtreamUserData(
db,
payload.playlistId,
payload.favoritedXtreamIds,
payload.recentlyViewedXtreamIds,
controller.control
);
controller.emitCompleted({
phase: DB_OPERATION_PHASES.RESTORING_RECENTLY_VIEWED,
current: totalItems,
total: totalItems,
});
return result;
}
);
}
case 'DB_ADD_FAVORITE': {
const payload = message.payload as {
contentId: number;
playlistId: string;
};
return addFavorite(db, payload.contentId, payload.playlistId);
}
case 'DB_REMOVE_FAVORITE': {
const payload = message.payload as {
contentId: number;
playlistId: string;
};
return removeFavorite(db, payload.contentId, payload.playlistId);
}
case 'DB_IS_FAVORITE': {
const payload = message.payload as {
contentId: number;
playlistId: string;
};
return isFavorite(db, payload.contentId, payload.playlistId);
}
case 'DB_GET_FAVORITES': {
const payload = message.payload as { playlistId: string };
return getFavorites(db, payload.playlistId);
}
case 'DB_GET_GLOBAL_FAVORITES':
return getGlobalFavorites(db);
case 'DB_GET_ALL_GLOBAL_FAVORITES':
return getAllGlobalFavorites(db);
case 'DB_REORDER_GLOBAL_FAVORITES': {
const payload = message.payload as {
updates: { content_id: number; position: number }[];
};
return reorderGlobalFavorites(db, payload.updates);
}
case 'DB_GET_RECENTLY_VIEWED':
return getRecentlyViewed(db);
case 'DB_CLEAR_RECENTLY_VIEWED':
return clearRecentlyViewed(db);
case 'DB_GET_RECENT_ITEMS': {
const payload = message.payload as { playlistId: string };
return getRecentItems(db, payload.playlistId);
}
case 'DB_ADD_RECENT_ITEM': {
const payload = message.payload as {
contentId: number;
playlistId: string;
};
return addRecentItem(db, payload.contentId, payload.playlistId);
}
case 'DB_CLEAR_PLAYLIST_RECENT_ITEMS': {
const payload = message.payload as { playlistId: string };
return clearPlaylistRecentItems(db, payload.playlistId);
}
case 'DB_REMOVE_RECENT_ITEM': {
const payload = message.payload as {
contentId: number;
playlistId: string;
};
return removeRecentItem(db, payload.contentId, payload.playlistId);
}
case 'DB_SAVE_PLAYBACK_POSITION': {
const payload = message.payload as {
playlistId: string;
data: {
contentXtreamId: number;
contentType: 'vod' | 'episode';
seriesXtreamId?: number;
seasonNumber?: number;
episodeNumber?: number;
positionSeconds: number;
durationSeconds?: number;
playlistType?: 'xtream' | 'stalker' | 'm3u-file' | 'm3u-text' | 'm3u-url';
};
};
return savePlaybackPosition(db, payload.playlistId, payload.data);
}
case 'DB_GET_PLAYBACK_POSITION': {
const payload = message.payload as {
playlistId: string;
contentXtreamId: number;
contentType: 'vod' | 'episode';
};
return getPlaybackPosition(
db,
payload.playlistId,
payload.contentXtreamId,
payload.contentType
);
}
case 'DB_GET_SERIES_PLAYBACK_POSITIONS': {
const payload = message.payload as {
playlistId: string;
seriesXtreamId: number;
};
return getSeriesPlaybackPositions(
db,
payload.playlistId,
payload.seriesXtreamId
);
}
case 'DB_GET_RECENT_PLAYBACK_POSITIONS': {
const payload = message.payload as {
playlistId: string;
limit?: number;
};
return getRecentPlaybackPositions(
db,
payload.playlistId,
payload.limit
);
}
case 'DB_GET_ALL_PLAYBACK_POSITIONS': {
const payload = message.payload as { playlistId: string };
return getAllPlaybackPositions(db, payload.playlistId);
}
case 'DB_CLEAR_PLAYBACK_POSITION': {
const payload = message.payload as {
playlistId: string;
contentXtreamId: number;
contentType: 'vod' | 'episode';
};
return clearPlaybackPosition(
db,
payload.playlistId,
payload.contentXtreamId,
payload.contentType
);
}
}
}
parentPort.on('message', async (message: DbWorkerIncomingMessage) => {
if (message.type === 'cancel') {
const activeOperation = activeOperations.get(message.operationId);
if (activeOperation) {
activeOperation.cancelled = true;
}
return;
}
try {
const result = await executeRequest(message);
postMessage({
type: 'response',
requestId: message.requestId,
success: true,
result,
});
} catch (error) {
console.error(loggerLabel, `Error handling ${message.operation}:`, error);
postMessage({
type: 'response',
requestId: message.requestId,
success: false,
error: serializeError(error),
});
}
});
process.on('exit', () => {
closeWorkerDatabase();
});
postMessage({ type: 'ready' });
@@ -161,6 +161,7 @@ class EpgDatabase {
this.db = new Database(dbPath);
this.db.pragma('foreign_keys = ON');
this.db.pragma('journal_mode = WAL'); // Better concurrent write performance
this.db.pragma('busy_timeout = 5000');
// Prepare statements
// Use INSERT OR REPLACE to update existing channels and refresh updated_at
+5
View File
@@ -17,6 +17,7 @@ import SettingsEvents from './app/events/settings.events';
import SharedEvents from './app/events/shared.events';
import SquirrelEvents from './app/events/squirrel.events';
import StalkerEvents from './app/events/stalker.events';
import { databaseWorkerClient } from './app/services/database-worker-client';
import XtreamEvents from './app/events/xtream.events';
app.setName('iptvnator');
@@ -79,3 +80,7 @@ Main.bootstrapApp();
app.whenReady().then(async () => {
await Main.bootstrapAppEvents();
});
app.on('before-quit', () => {
void databaseWorkerClient.shutdown();
});