feat(db): add EPG schema and IPC handlers for EPG storage

Add tables, indexes, FTS5 virtual table, and triggers to support EPG
(electronic program guide) data in the shared SQLite connection. The
changes introduce epg_channels and epg_programs tables, multiple
indexes for efficient lookups (by source, name, channel, start, and
time range), and an epg_programs_fts virtual table with triggers to
keep the full-text index synchronized with epg_programs.

Register EPG database event handlers in the Electron backend and
add an IPC handler for bulk-saving channels. The handler deletes
existing channels for a source and inserts new channels in chunks to
avoid large single inserts, returning the inserted count.

These changes enable storing, querying, and full-text searching EPG
data and prepare backend IPC endpoints for importing program guide
data.
This commit is contained in:
4gray committed 2026-01-02 17:22:09 +01:00
1 parent 87bf0f6178
commit 895a269200
10 files changed
+1396 -327

No files matched your search

+11 -1
View File
@@ -12,6 +12,11 @@ const nodeBuiltins = [
'v8', 'vm', 'worker_threads', 'zlib'
];
// Native modules that must be externalized (loaded at runtime)
const nativeModules = [
'better-sqlite3',
];
const isProduction = process.env.NODE_ENV === 'production';
async function buildWorker() {
@@ -25,12 +30,17 @@ async function buildWorker() {
target: 'node18',
format: 'cjs',
outfile: path.join(__dirname, '../../dist/apps/electron-backend/workers/epg-parser.worker.js'),
external: nodeBuiltins.map(m => `node:${m}`).concat(nodeBuiltins),
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'),
},
});
@@ -17,6 +17,17 @@ contextBridge.exposeInMainWorld('electron', {
) => {
ipcRenderer.on('player-error', (_event, data) => callback(data));
},
// EPG progress listener
onEpgProgress: (
callback: (data: {
url: string;
status: 'loading' | 'complete' | 'error';
stats?: { totalChannels: number; totalPrograms: number };
error?: string;
}) => void
) => {
ipcRenderer.on('EPG_PROGRESS_UPDATE', (_event, data) => callback(data));
},
getAppVersion: () => ipcRenderer.invoke('get-app-version'),
platform: process.platform,
fetchPlaylistByUrl: (url: string, title?: string) =>
@@ -72,6 +83,11 @@ contextBridge.exposeInMainWorld('electron', {
getEpgChannelsByRange: (skip: number, limit: number) =>
ipcRenderer.invoke('EPG_GET_CHANNELS_BY_RANGE', { skip, limit }),
forceFetchEpg: (url: string) => ipcRenderer.invoke('EPG_FORCE_FETCH', url),
clearEpgData: () => ipcRenderer.invoke('EPG_CLEAR_ALL'),
checkEpgFreshness: (urls: string[], maxAgeHours?: number) =>
ipcRenderer.invoke('EPG_CHECK_FRESHNESS', { urls, maxAgeHours }),
searchEpgPrograms: (searchTerm: string, limit?: number) =>
ipcRenderer.invoke('EPG_DB_SEARCH_PROGRAMS', searchTerm, limit),
setMpvPlayerPath: (mpvPlayerPath: string) =>
ipcRenderer.invoke('SET_MPV_PLAYER_PATH', mpvPlayerPath),
setVlcPlayerPath: (vlcPlayerPath: string) =>
@@ -10,6 +10,8 @@ export {
content,
recentlyViewed,
favorites,
epgChannels,
epgPrograms,
// Types
type Playlist,
type NewPlaylist,
@@ -21,5 +23,9 @@ export {
type NewRecentlyViewed,
type Favorite,
type NewFavorite,
type EpgChannel,
type NewEpgChannel,
type EpgProgramDb,
type NewEpgProgramDb,
} from 'database';
@@ -7,6 +7,7 @@ import { ipcMain } from 'electron';
import './database/category.events';
import './database/content.events';
import './database/epg-db.events';
import './database/favorites.events';
import './database/playlist.events';
import './database/recently-viewed.events';
@@ -0,0 +1,391 @@
/**
* EPG Database IPC event handlers
* Operations for storing and querying EPG data in SQLite with FTS5 support
*/
import { and, eq, gte, lte, sql } from 'drizzle-orm';
import { ipcMain } from 'electron';
import { getDatabase } from '../../database/connection';
import * as schema from '../../database/schema';
const loggerLabel = '[EPG DB]';
/**
* Save EPG channels in bulk (upsert)
*/
ipcMain.handle(
'EPG_DB_SAVE_CHANNELS',
async (
_event,
sourceUrl: string,
channels: Array<{
id: string;
displayName: string;
iconUrl?: string;
url?: string;
}>
) => {
try {
const db = await getDatabase();
// Delete existing channels from this source first
await db
.delete(schema.epgChannels)
.where(eq(schema.epgChannels.sourceUrl, sourceUrl));
if (channels.length === 0) return { success: true, count: 0 };
// Insert in chunks
const CHUNK_SIZE = 500;
let totalInserted = 0;
for (let i = 0; i < channels.length; i += CHUNK_SIZE) {
const chunk = channels.slice(i, i + CHUNK_SIZE);
const values = chunk.map((ch) => ({
id: ch.id,
displayName: ch.displayName,
iconUrl: ch.iconUrl || null,
url: ch.url || null,
sourceUrl,
}));
await db.insert(schema.epgChannels).values(values);
totalInserted += chunk.length;
}
console.log(
loggerLabel,
`Saved ${totalInserted} channels from ${sourceUrl}`
);
return { success: true, count: totalInserted };
} catch (error) {
console.error(loggerLabel, 'Error saving EPG channels:', error);
throw error;
}
}
);
/**
* Save EPG programs in bulk
*/
ipcMain.handle(
'EPG_DB_SAVE_PROGRAMS',
async (
_event,
programs: Array<{
channelId: string;
start: string;
stop: string;
title: string;
description?: string;
category?: string;
iconUrl?: string;
rating?: string;
episodeNum?: string;
}>
) => {
try {
const db = await getDatabase();
if (programs.length === 0) return { success: true, count: 0 };
// Insert in chunks for better performance
const CHUNK_SIZE = 500;
let totalInserted = 0;
for (let i = 0; i < programs.length; i += CHUNK_SIZE) {
const chunk = programs.slice(i, i + CHUNK_SIZE);
const values = chunk.map((prog) => ({
channelId: prog.channelId,
start: prog.start,
stop: prog.stop,
title: prog.title,
description: prog.description || null,
category: prog.category || null,
iconUrl: prog.iconUrl || null,
rating: prog.rating || null,
episodeNum: prog.episodeNum || null,
}));
await db.insert(schema.epgPrograms).values(values);
totalInserted += chunk.length;
}
console.log(loggerLabel, `Saved ${totalInserted} programs`);
return { success: true, count: totalInserted };
} catch (error) {
console.error(loggerLabel, 'Error saving EPG programs:', error);
throw error;
}
}
);
/**
* Get programs for a specific channel
*/
ipcMain.handle(
'EPG_DB_GET_CHANNEL_PROGRAMS',
async (_event, channelId: string, fromTime?: string, toTime?: string) => {
try {
const db = await getDatabase();
const now = new Date().toISOString();
let query = db
.select()
.from(schema.epgPrograms)
.where(eq(schema.epgPrograms.channelId, channelId));
// Apply time filters if provided
if (fromTime && toTime) {
query = db
.select()
.from(schema.epgPrograms)
.where(
and(
eq(schema.epgPrograms.channelId, channelId),
gte(schema.epgPrograms.stop, fromTime),
lte(schema.epgPrograms.start, toTime)
)
);
} else {
// Default: from now onwards
query = db
.select()
.from(schema.epgPrograms)
.where(
and(
eq(schema.epgPrograms.channelId, channelId),
gte(schema.epgPrograms.stop, now)
)
);
}
const results = await query.orderBy(schema.epgPrograms.start);
return results;
} catch (error) {
console.error(
loggerLabel,
'Error getting channel programs:',
error
);
throw error;
}
}
);
/**
* Get current program for a channel (what's on now)
*/
ipcMain.handle('EPG_DB_GET_CURRENT_PROGRAM', async (_event, channelId: string) => {
try {
const db = await getDatabase();
const now = new Date().toISOString();
const result = await db
.select()
.from(schema.epgPrograms)
.where(
and(
eq(schema.epgPrograms.channelId, channelId),
lte(schema.epgPrograms.start, now),
gte(schema.epgPrograms.stop, now)
)
)
.limit(1);
return result[0] || null;
} catch (error) {
console.error(loggerLabel, 'Error getting current program:', error);
throw error;
}
});
/**
* Full-text search EPG programs using FTS5 with LIKE fallback
* Handles Cyrillic and other Unicode text properly
* Includes channel display name via JOIN
*/
ipcMain.handle(
'EPG_DB_SEARCH_PROGRAMS',
async (_event, searchTerm: string, limit = 50) => {
try {
const db = await getDatabase();
const now = new Date().toISOString();
const trimmedTerm = searchTerm.trim();
if (!trimmedTerm) {
return [];
}
// Use LIKE for substring matching (works better with Cyrillic)
// This is more intuitive for users expecting exact substring matches
const likePattern = `%${trimmedTerm}%`;
// JOIN with epg_channels to get channel display name
const results = await db.all(sql`
SELECT
p.*,
c.display_name as channel_name
FROM epg_programs p
LEFT JOIN epg_channels c ON p.channel_id = c.id
WHERE (
p.title LIKE ${likePattern}
OR p.description LIKE ${likePattern}
OR p.category LIKE ${likePattern}
)
AND p.stop >= ${now}
ORDER BY p.start
LIMIT ${limit}
`);
return results;
} catch (error) {
console.error(loggerLabel, 'Error searching EPG programs:', error);
throw error;
}
}
);
/**
* Get all EPG channels
*/
ipcMain.handle('EPG_DB_GET_CHANNELS', async () => {
try {
const db = await getDatabase();
const results = await db
.select()
.from(schema.epgChannels)
.orderBy(schema.epgChannels.displayName);
return results;
} catch (error) {
console.error(loggerLabel, 'Error getting EPG channels:', error);
throw error;
}
});
/**
* Get channel by ID or display name
*/
ipcMain.handle('EPG_DB_GET_CHANNEL', async (_event, channelIdOrName: string) => {
try {
const db = await getDatabase();
// Try exact ID match first
let result = await db
.select()
.from(schema.epgChannels)
.where(eq(schema.epgChannels.id, channelIdOrName))
.limit(1);
if (result.length > 0) return result[0];
// Try display name match (case-insensitive)
result = await db
.select()
.from(schema.epgChannels)
.where(
sql`LOWER(${schema.epgChannels.displayName}) = LOWER(${channelIdOrName})`
)
.limit(1);
return result[0] || null;
} catch (error) {
console.error(loggerLabel, 'Error getting EPG channel:', error);
throw error;
}
});
/**
* Cleanup expired programs (older than specified hours)
*/
ipcMain.handle('EPG_DB_CLEANUP_EXPIRED', async (_event, hoursToKeep = 24) => {
try {
const db = await getDatabase();
const cutoff = new Date(
Date.now() - hoursToKeep * 60 * 60 * 1000
).toISOString();
const result = await db
.delete(schema.epgPrograms)
.where(lte(schema.epgPrograms.stop, cutoff));
console.log(
loggerLabel,
`Cleaned up expired programs (older than ${hoursToKeep}h)`
);
return { success: true };
} catch (error) {
console.error(loggerLabel, 'Error cleaning up EPG programs:', error);
throw error;
}
});
/**
* Clear all EPG data for a specific source URL
*/
ipcMain.handle('EPG_DB_CLEAR_SOURCE', async (_event, sourceUrl: string) => {
try {
const db = await getDatabase();
// Deleting channels will cascade delete programs due to foreign key
await db
.delete(schema.epgChannels)
.where(eq(schema.epgChannels.sourceUrl, sourceUrl));
console.log(loggerLabel, `Cleared EPG data for source: ${sourceUrl}`);
return { success: true };
} catch (error) {
console.error(loggerLabel, 'Error clearing EPG source:', error);
throw error;
}
});
/**
* Clear all EPG data
*/
ipcMain.handle('EPG_DB_CLEAR_ALL', async () => {
try {
const db = await getDatabase();
await db.delete(schema.epgPrograms);
await db.delete(schema.epgChannels);
console.log(loggerLabel, 'Cleared all EPG data');
return { success: true };
} catch (error) {
console.error(loggerLabel, 'Error clearing all EPG data:', error);
throw error;
}
});
/**
* Get EPG statistics
*/
ipcMain.handle('EPG_DB_GET_STATS', async () => {
try {
const db = await getDatabase();
const channelCount = await db
.select({ count: sql<number>`count(*)` })
.from(schema.epgChannels);
const programCount = await db
.select({ count: sql<number>`count(*)` })
.from(schema.epgPrograms);
const now = new Date().toISOString();
const futureProgramCount = await db
.select({ count: sql<number>`count(*)` })
.from(schema.epgPrograms)
.where(gte(schema.epgPrograms.stop, now));
return {
channels: channelCount[0].count,
programs: programCount[0].count,
futurePrograms: futureProgramCount[0].count,
};
} catch (error) {
console.error(loggerLabel, 'Error getting EPG stats:', error);
throw error;
}
});
+431 -220
View File
@@ -1,112 +1,220 @@
import { app, ipcMain } from 'electron';
import { and, eq, gte, sql } from 'drizzle-orm';
import { app, BrowserWindow, ipcMain } from 'electron';
import * as path from 'path';
import {
EpgChannel,
EpgChannelWithPrograms,
EpgData,
EpgProgram,
} from 'shared-interfaces';
import { Worker } from 'worker_threads';
import { pathToFileURL } from 'url';
import { getDatabase } from '../database/connection';
import * as schema from '../database/schema';
import { EpgProgram } from 'shared-interfaces';
/**
* EPG Events Handler
* Manages EPG data fetching and querying using worker threads
* Manages EPG data fetching and querying using worker threads.
* Database operations are performed in the worker thread to avoid blocking the main thread.
*/
export default class EpgEvents {
private static epgData: EpgData = {
channels: [],
programs: [],
};
private static epgDataMerged: Map<string, EpgChannelWithPrograms> =
new Map();
private static fetchedUrls: Set<string> = new Set();
private static workers: Map<string, Worker> = new Map();
private static readonly loggerLabel = '[EPG Events]';
// Display name index for O(1) lookups (lowercase name -> channel id)
private static displayNameIndex: Map<string, string> = new Map();
/**
* Send EPG progress to all renderer windows
*/
private static sendProgressToRenderer(
url: string,
status: 'queued' | 'loading' | 'complete' | 'error',
stats?: { totalChannels: number; totalPrograms: number },
error?: string,
queuePosition?: number
): void {
const windows = BrowserWindow.getAllWindows();
windows.forEach((win) => {
win.webContents.send('EPG_PROGRESS_UPDATE', {
url,
status,
stats,
error,
queuePosition,
});
});
}
/**
* Bootstrap EPG events
*/
static bootstrapEpgEvents(): Electron.IpcMain {
// Fetch EPG from URLs
ipcMain.handle('FETCH_EPG', async (event, args: { url: string[] }) => {
ipcMain.handle('FETCH_EPG', async (_event, args: { url: string[] }) => {
return await this.handleFetchEpg(args.url);
});
// Get programs for a specific channel
ipcMain.handle(
'GET_CHANNEL_PROGRAMS',
async (event, args: { channelId: string }) => {
async (_event, args: { channelId: string }) => {
return this.handleGetChannelPrograms(args.channelId);
}
);
// Get all channels
// Get all channels from database
ipcMain.handle('EPG_GET_CHANNELS', async () => {
return this.epgData;
return this.handleGetAllChannels();
});
// Get channels by range (pagination)
ipcMain.handle(
'EPG_GET_CHANNELS_BY_RANGE',
async (event, args: { skip: number; limit: number }) => {
async (_event, args: { skip: number; limit: number }) => {
return this.handleGetChannelsByRange(args.skip, args.limit);
}
);
// Force fetch (ignore cache)
ipcMain.handle('EPG_FORCE_FETCH', async (event, url: string) => {
ipcMain.handle('EPG_FORCE_FETCH', async (_event, url: string) => {
this.fetchedUrls.delete(url);
return await this.handleFetchEpg([url]);
});
// Cleanup expired programs
// Cleanup expired programs (uses worker thread)
ipcMain.handle(
'EPG_CLEANUP_EXPIRED',
async (event, hoursToKeep?: number) => {
return this.cleanupExpiredPrograms(hoursToKeep);
async (_event, hoursToKeep?: number) => {
return this.runCleanupInWorker(hoursToKeep ?? 24);
}
);
// Clear all EPG data
ipcMain.handle('EPG_CLEAR_ALL', async () => {
await this.clearEpgData();
return { success: true };
});
// Check if EPG data for URLs is fresh (not stale)
ipcMain.handle(
'EPG_CHECK_FRESHNESS',
async (
_event,
args: { urls: string[]; maxAgeHours?: number }
): Promise<{ staleUrls: string[]; freshUrls: string[] }> => {
return this.checkEpgFreshness(args.urls, args.maxAgeHours ?? 12);
}
);
return ipcMain;
}
/**
* Check which EPG URLs have fresh data vs stale/missing data
* @param urls - EPG source URLs to check
* @param maxAgeHours - Maximum age in hours before data is considered stale
*/
private static async checkEpgFreshness(
urls: string[],
maxAgeHours: number
): Promise<{ staleUrls: string[]; freshUrls: string[] }> {
const staleUrls: string[] = [];
const freshUrls: string[] = [];
const cutoffTime = new Date(
Date.now() - maxAgeHours * 60 * 60 * 1000
).toISOString();
try {
const db = await getDatabase();
for (const url of urls) {
if (!url?.trim()) continue;
const result = await db
.select({ updatedAt: schema.epgChannels.updatedAt })
.from(schema.epgChannels)
.where(eq(schema.epgChannels.sourceUrl, url))
.limit(1);
const isFresh =
result.length > 0 &&
result[0].updatedAt &&
result[0].updatedAt >= cutoffTime;
if (isFresh) {
freshUrls.push(url);
this.fetchedUrls.add(url);
} else {
staleUrls.push(url);
}
}
} catch (error) {
console.error(this.loggerLabel, 'Error checking EPG freshness:', error);
return { staleUrls: urls, freshUrls: [] };
}
// Log summary
if (freshUrls.length > 0) {
console.log(this.loggerLabel, `EPG fresh (skipping): ${freshUrls.length} source(s)`);
}
if (staleUrls.length > 0) {
console.log(this.loggerLabel, `EPG stale (will fetch): ${staleUrls.length} source(s)`);
}
return { staleUrls, freshUrls };
}
/**
* Handle EPG fetch from URLs
* Automatically skips URLs with fresh data (less than 12 hours old)
* Processes URLs sequentially to avoid SQLite database locking issues
*/
private static async handleFetchEpg(
urls: string[]
): Promise<{ success: boolean; message?: string }> {
): Promise<{ success: boolean; message?: string; skipped?: string[] }> {
const validUrls = urls.filter((url) => url?.trim());
if (validUrls.length === 0) {
return { success: false, message: 'No valid URLs provided' };
}
const promises = validUrls.map((url) => this.fetchEpgFromUrl(url));
// Check which URLs have fresh data and can be skipped
const { staleUrls, freshUrls } = await this.checkEpgFreshness(validUrls, 12);
try {
await Promise.all(promises);
return { success: true };
} catch (error) {
console.error(this.loggerLabel, 'Error fetching EPG:', error);
if (staleUrls.length === 0) {
return { success: true, message: 'All EPG data is fresh', skipped: freshUrls };
}
// Send queued status for all stale URLs first
staleUrls.forEach((url, index) => {
this.sendProgressToRenderer(url, 'queued', undefined, undefined, index + 1);
});
// Process only stale URLs sequentially to avoid database locking
const errors: string[] = [];
for (let i = 0; i < staleUrls.length; i++) {
const url = staleUrls[i];
try {
await this.fetchEpgFromUrl(url);
} catch (error) {
console.error(this.loggerLabel, `Error fetching EPG from ${url}:`, error);
errors.push(error instanceof Error ? error.message : String(error));
}
}
if (errors.length > 0) {
return {
success: false,
message: error instanceof Error ? error.message : String(error),
success: errors.length < staleUrls.length, // Partial success if some worked
message: errors.join('; '),
skipped: freshUrls,
};
}
return { success: true, skipped: freshUrls };
}
/**
* Fetch EPG from a single URL using worker thread
* The worker handles both parsing AND database operations
*/
private static async fetchEpgFromUrl(url: string): Promise<void> {
// Skip if already fetched
// Skip if already fetched this session
if (this.fetchedUrls.has(url)) {
console.log(this.loggerLabel, `Skipping already fetched URL: ${url}`);
return;
}
@@ -114,7 +222,6 @@ export default class EpgEvents {
let workerPath: string;
if (app.isPackaged) {
// In packaged app: Resources/dist/apps/electron-backend/workers/epg-parser.worker.js
const resourcesPath = path.dirname(app.getAppPath());
workerPath = path.join(
resourcesPath,
@@ -125,27 +232,20 @@ export default class EpgEvents {
'epg-parser.worker.js'
);
} else {
// In development: dist/apps/electron-backend/workers/epg-parser.worker.js
workerPath = path.join(__dirname, 'workers', 'epg-parser.worker.js');
}
let worker: Worker;
try {
// Worker threads require file:// URLs wrapped in URL object for packaged apps
const workerURL = pathToFileURL(workerPath);
worker = new Worker(workerURL, {
// Increase memory limits for large EPG files
resourceLimits: {
maxOldGenerationSizeMb: 4096, // 4GB for parsed data
maxYoungGenerationSizeMb: 512, // 512MB for temporary allocations
maxOldGenerationSizeMb: 4096,
maxYoungGenerationSizeMb: 512,
},
});
} catch (error) {
console.error(
this.loggerLabel,
'Failed to create worker:',
error
);
console.error(this.loggerLabel, 'Failed to create worker:', error);
reject(error);
return;
}
@@ -154,68 +254,70 @@ export default class EpgEvents {
worker.on(
'message',
(message: {
async (message: {
type: string;
channels?: EpgChannel[];
programs?: EpgProgram[];
error?: string;
url?: string;
stats?: { totalChannels: number; totalPrograms: number };
}) => {
switch (message.type) {
case 'READY':
worker.postMessage({ type: 'FETCH_EPG', url });
break;
try {
switch (message.type) {
case 'READY':
// Notify renderer that loading started
this.sendProgressToRenderer(url, 'loading', { totalChannels: 0, totalPrograms: 0 });
worker.postMessage({ type: 'FETCH_EPG', url });
break;
case 'EPG_CHANNELS_BATCH':
// Add channels incrementally
if (message.channels) {
this.addChannelsBatch(message.channels);
}
break;
case 'EPG_PROGRESS':
if (message.stats) {
console.log(
this.loggerLabel,
`Progress: ${message.stats.totalChannels} channels, ${message.stats.totalPrograms} programs`
);
// Forward progress to renderer
this.sendProgressToRenderer(url, 'loading', message.stats);
}
break;
case 'EPG_PROGRAMS_BATCH':
// Add programs incrementally
if (message.programs) {
this.addProgramsBatch(message.programs);
}
break;
case 'EPG_PROGRESS':
// Log progress for large files
if (message.stats) {
case 'EPG_COMPLETE':
console.log(
this.loggerLabel,
`Progress: ${message.stats.totalChannels} channels, ${message.stats.totalPrograms} programs`
`EPG parsing complete for ${url}:`,
message.stats
);
}
break;
// Notify renderer of completion
this.sendProgressToRenderer(url, 'complete', message.stats);
this.fetchedUrls.add(url);
// Trigger cleanup in worker thread (non-blocking)
worker.postMessage({ type: 'CLEANUP_EXPIRED', hoursToKeep: 24 });
break;
case 'EPG_COMPLETE':
console.log(
this.loggerLabel,
`EPG parsing complete for ${url}:`,
message.stats
);
this.rebuildMergedData();
// Auto-cleanup programs older than 24 hours
this.cleanupExpiredPrograms(24);
this.fetchedUrls.add(url);
worker.terminate();
this.workers.delete(url);
resolve();
break;
case 'CLEANUP_COMPLETE':
console.log(
this.loggerLabel,
'Cleanup complete, terminating worker'
);
worker.terminate();
this.workers.delete(url);
resolve();
break;
case 'EPG_ERROR':
console.error(
this.loggerLabel,
'Worker error:',
message.error
);
worker.terminate();
this.workers.delete(url);
reject(new Error(message.error || 'Unknown error'));
break;
case 'EPG_ERROR':
console.error(
this.loggerLabel,
'Worker error:',
message.error
);
// Notify renderer of error
this.sendProgressToRenderer(url, 'error', undefined, message.error);
worker.terminate();
this.workers.delete(url);
reject(new Error(message.error || 'Unknown error'));
break;
}
} catch (err) {
console.error(this.loggerLabel, 'Error handling message:', err);
reject(err);
}
}
);
@@ -239,154 +341,263 @@ export default class EpgEvents {
}
/**
* Add a batch of channels (streaming mode)
* Transform database row to flat EpgProgram interface
*/
private static addChannelsBatch(channels: EpgChannel[]): void {
// Use Set for O(1) duplicate checking
const existingIds = new Set(this.epgData.channels.map((c) => c.id));
const newChannels = channels.filter((c) => !existingIds.has(c.id));
this.epgData.channels = this.epgData.channels.concat(newChannels);
private static transformDbRowToEpgProgram(row: {
id: number;
channelId: string;
start: string;
stop: string;
title: string;
description: string | null;
category: string | null;
iconUrl: string | null;
rating: string | null;
episodeNum: string | null;
}) {
return {
start: row.start,
stop: row.stop,
channel: row.channelId,
title: row.title,
desc: row.description,
category: row.category,
iconUrl: row.iconUrl,
rating: row.rating,
episodeNum: row.episodeNum,
};
}
/**
* Add a batch of programs (streaming mode)
* Get programs for a specific channel from database
*/
private static addProgramsBatch(programs: EpgProgram[]): void {
// Simply append programs - no duplicate check needed as each program is unique
this.epgData.programs = this.epgData.programs.concat(programs);
private static async handleGetChannelPrograms(channelId: string): Promise<EpgProgram[]> {
try {
const db = await getDatabase();
const now = new Date().toISOString();
// Try exact channel ID match first
let results = await db
.select()
.from(schema.epgPrograms)
.where(
and(
eq(schema.epgPrograms.channelId, channelId),
gte(schema.epgPrograms.stop, now)
)
)
.orderBy(schema.epgPrograms.start)
.limit(100);
if (results.length > 0) {
return results.map(this.transformDbRowToEpgProgram);
}
// Try to find channel by display name
const channel = await db
.select()
.from(schema.epgChannels)
.where(
sql`LOWER(${schema.epgChannels.displayName}) LIKE LOWER(${'%' + channelId + '%'})`
)
.limit(1);
if (channel.length > 0) {
results = await db
.select()
.from(schema.epgPrograms)
.where(
and(
eq(schema.epgPrograms.channelId, channel[0].id),
gte(schema.epgPrograms.stop, now)
)
)
.orderBy(schema.epgPrograms.start)
.limit(100);
return results.map(this.transformDbRowToEpgProgram);
}
return [];
} catch (error) {
console.error(this.loggerLabel, 'Error getting channel programs:', error);
return [];
}
}
/**
* Rebuild merged data structure (channels with programs)
* Optimized for large datasets
* Get all channels from database
*/
private static rebuildMergedData(): void {
this.epgDataMerged.clear();
this.displayNameIndex.clear();
private static async handleGetAllChannels(): Promise<{
channels: Array<{ id: string; displayName: string }>;
programs: never[];
}> {
try {
const db = await getDatabase();
const channels = await db
.select({
id: schema.epgChannels.id,
displayName: schema.epgChannels.displayName,
})
.from(schema.epgChannels)
.orderBy(schema.epgChannels.displayName);
// Create channel lookup map for O(1) access
const channelMap = new Map(this.epgData.channels.map((c) => [c.id, c]));
// Build display name index for fast lookups
for (const channel of this.epgData.channels) {
for (const name of channel.displayName) {
if (name.value) {
this.displayNameIndex.set(
name.value.toLowerCase(),
channel.id
);
}
}
return { channels, programs: [] };
} catch (error) {
console.error(this.loggerLabel, 'Error getting all channels:', error);
return { channels: [], programs: [] };
}
// Group programs by channel in a single pass
const programsByChannel = new Map<string, EpgProgram[]>();
for (const program of this.epgData.programs) {
const channelId = program.channel;
if (!programsByChannel.has(channelId)) {
programsByChannel.set(channelId, []);
}
programsByChannel.get(channelId)!.push(program);
}
// Build merged structure
for (const [channelId, programs] of programsByChannel.entries()) {
const channel = channelMap.get(channelId);
if (channel) {
this.epgDataMerged.set(channelId, {
...channel,
programs,
});
}
}
// Free memory - programs are now stored in epgDataMerged
this.epgData.programs = [];
console.log(
this.loggerLabel,
`Merged data rebuilt: ${this.epgDataMerged.size} channels, ${this.displayNameIndex.size} display names indexed`
);
}
/**
* Get programs for a specific channel
* Get channels by range (for pagination) with their programs
*/
private static handleGetChannelPrograms(channelId: string): EpgProgram[] {
// First try exact ID match
let channelData = this.epgDataMerged.get(channelId);
// If not found, try display name index (O(1) lookup)
if (!channelData) {
const indexedId = this.displayNameIndex.get(channelId.toLowerCase());
if (indexedId) {
channelData = this.epgDataMerged.get(indexedId);
}
}
// Fallback: partial match (only if index lookup failed)
if (!channelData) {
const searchTerm = channelId.toLowerCase();
for (const [name, id] of this.displayNameIndex.entries()) {
if (name.includes(searchTerm) || searchTerm.includes(name)) {
channelData = this.epgDataMerged.get(id);
if (channelData) break;
}
}
}
return channelData?.programs || [];
}
/**
* Get channels by range (for pagination)
*/
private static handleGetChannelsByRange(
private static async handleGetChannelsByRange(
skip: number,
limit: number
): EpgChannelWithPrograms[] {
const channels = Array.from(this.epgDataMerged.values());
return channels.slice(skip, skip + limit);
}
): Promise<Array<{ id: string; displayName: string; iconUrl: string | null; programs: EpgProgram[] }>> {
try {
const db = await getDatabase();
const channels = await db
.select({
id: schema.epgChannels.id,
displayName: schema.epgChannels.displayName,
iconUrl: schema.epgChannels.iconUrl,
})
.from(schema.epgChannels)
.orderBy(schema.epgChannels.displayName)
.offset(skip)
.limit(limit);
/**
* Clean up expired programs (older than specified hours)
* Call this periodically to free memory from stale EPG data
*/
static cleanupExpiredPrograms(hoursToKeep = 24): number {
const cutoff = new Date(
Date.now() - hoursToKeep * 60 * 60 * 1000
).toISOString();
let removedCount = 0;
// Fetch programs for each channel
const channelsWithPrograms = await Promise.all(
channels.map(async (channel) => {
const programs = await db
.select()
.from(schema.epgPrograms)
.where(eq(schema.epgPrograms.channelId, channel.id))
.orderBy(schema.epgPrograms.start);
for (const channel of this.epgDataMerged.values()) {
const originalLength = channel.programs.length;
channel.programs = channel.programs.filter((p) => p.stop > cutoff);
removedCount += originalLength - channel.programs.length;
}
if (removedCount > 0) {
console.log(
this.loggerLabel,
`Cleaned up ${removedCount} expired programs (older than ${hoursToKeep}h)`
return {
...channel,
programs: programs.map(this.transformDbRowToEpgProgram),
};
})
);
}
return removedCount;
return channelsWithPrograms;
} catch (error) {
console.error(this.loggerLabel, 'Error getting channels by range:', error);
return [];
}
}
/**
* Clear all EPG data
* Run cleanup in worker thread to avoid blocking main thread
*/
static clearEpgData(): void {
this.epgData = { channels: [], programs: [] };
this.epgDataMerged.clear();
this.displayNameIndex.clear();
this.fetchedUrls.clear();
private static async runCleanupInWorker(hoursToKeep: number): Promise<{ success: boolean }> {
return new Promise((resolve, reject) => {
let workerPath: string;
// Terminate all workers
this.workers.forEach((worker) => worker.terminate());
this.workers.clear();
if (app.isPackaged) {
const resourcesPath = path.dirname(app.getAppPath());
workerPath = path.join(
resourcesPath,
'dist',
'apps',
'electron-backend',
'workers',
'epg-parser.worker.js'
);
} else {
workerPath = path.join(__dirname, 'workers', 'epg-parser.worker.js');
}
let worker: Worker;
try {
const workerURL = pathToFileURL(workerPath);
worker = new Worker(workerURL);
} catch (error) {
console.error(this.loggerLabel, 'Failed to create worker for cleanup:', error);
reject(error);
return;
}
worker.on('message', (message: { type: string; error?: string }) => {
if (message.type === 'READY') {
worker.postMessage({ type: 'CLEANUP_EXPIRED', hoursToKeep });
} else if (message.type === 'CLEANUP_COMPLETE') {
worker.terminate();
resolve({ success: true });
} else if (message.type === 'EPG_ERROR') {
console.error(this.loggerLabel, 'Worker cleanup error:', message.error);
worker.terminate();
reject(new Error(message.error || 'Cleanup failed'));
}
});
worker.on('error', (error) => {
console.error(this.loggerLabel, 'Worker error during cleanup:', error);
worker.terminate();
reject(error);
});
});
}
/**
* Clear all EPG data using worker thread to avoid blocking main thread
*/
static async clearEpgData(): Promise<void> {
return new Promise((resolve, reject) => {
let workerPath: string;
if (app.isPackaged) {
const resourcesPath = path.dirname(app.getAppPath());
workerPath = path.join(
resourcesPath,
'dist',
'apps',
'electron-backend',
'workers',
'epg-parser.worker.js'
);
} else {
workerPath = path.join(__dirname, 'workers', 'epg-parser.worker.js');
}
let worker: Worker;
try {
const workerURL = pathToFileURL(workerPath);
worker = new Worker(workerURL);
} catch (error) {
console.error(this.loggerLabel, 'Failed to create worker for clear:', error);
reject(error);
return;
}
worker.on('message', (message: { type: string; error?: string }) => {
if (message.type === 'READY') {
worker.postMessage({ type: 'CLEAR_EPG' });
} else if (message.type === 'CLEAR_COMPLETE') {
console.log(this.loggerLabel, 'EPG data cleared via worker');
this.fetchedUrls.clear();
// Terminate any running fetch workers
this.workers.forEach((w) => w.terminate());
this.workers.clear();
worker.terminate();
resolve();
} else if (message.type === 'EPG_ERROR') {
console.error(this.loggerLabel, 'Worker clear error:', message.error);
worker.terminate();
reject(new Error(message.error || 'Clear failed'));
}
});
worker.on('error', (error) => {
console.error(this.loggerLabel, 'Worker error during clear:', error);
worker.terminate();
reject(error);
});
});
}
}
@@ -1,30 +1,78 @@
import { EpgChannel, EpgProgram } from 'shared-interfaces';
import { parentPort } from 'worker_threads';
import { createGunzip } from 'zlib';
import { SaxesParser, SaxesTagPlain } from 'saxes';
import { Readable } from 'stream';
import Database from 'better-sqlite3';
import { existsSync, mkdirSync } from 'fs';
import { homedir } from 'os';
import { join } from 'path';
/**
* Internal parsing types with arrays for XML parsing
* These are different from the flat EpgProgram interface used by the frontend
*/
interface ParsedTextValue {
lang: string;
value: string;
}
interface ParsedIcon {
src: string;
width?: number;
height?: number;
}
interface ParsedRating {
system: string;
value: string;
}
interface ParsedEpisodeNum {
system: string;
value: string;
}
interface ParsedChannel {
id: string;
displayName: ParsedTextValue[];
icon: ParsedIcon[];
url: string[];
}
interface ParsedProgram {
start: string;
stop: string;
channel: string;
title: ParsedTextValue[];
desc: ParsedTextValue[];
category: ParsedTextValue[];
date: string;
episodeNum: ParsedEpisodeNum[];
icon: ParsedIcon[];
rating: ParsedRating[];
}
/**
* Streaming EPG Parser Worker
* Uses SAX parsing to process XML incrementally without loading entire file into memory.
* Supports both regular and gzipped EPG files.
* Performs database operations directly to avoid blocking the main thread.
*/
interface WorkerMessage {
type: 'FETCH_EPG' | 'FORCE_FETCH';
url: string;
type: 'FETCH_EPG' | 'FORCE_FETCH' | 'CLEAR_EPG' | 'CLEANUP_EXPIRED';
url?: string;
hoursToKeep?: number;
}
interface WorkerResponse {
type:
| 'EPG_CHANNELS_BATCH'
| 'EPG_PROGRAMS_BATCH'
| 'EPG_COMPLETE'
| 'EPG_ERROR'
| 'EPG_PROGRESS'
| 'CLEAR_COMPLETE'
| 'CLEANUP_COMPLETE'
| 'READY';
channels?: EpgChannel[];
programs?: EpgProgram[];
error?: string;
url?: string;
stats?: {
@@ -35,10 +83,174 @@ interface WorkerResponse {
const loggerLabel = '[EPG Worker]';
// Batch size for sending data back to main thread
// Batch size for database inserts
const CHANNEL_BATCH_SIZE = 100;
const PROGRAM_BATCH_SIZE = 1000;
// Skip programs that ended more than this many hours ago
const SKIP_PROGRAMS_OLDER_THAN_HOURS = 2;
/**
* Calculate the cutoff time for filtering old programs
* Programs that ended before this time will be skipped
*/
function getProgramCutoffTime(): Date {
return new Date(Date.now() - SKIP_PROGRAMS_OLDER_THAN_HOURS * 60 * 60 * 1000);
}
/**
* Check if a program has already ended (is expired)
* Handles timezone-aware ISO date strings
*/
function isProgramExpired(stopTime: string, cutoff: Date): boolean {
if (!stopTime) return true; // Skip programs without stop time
try {
const stopDate = new Date(stopTime);
// Check if the date is valid
if (isNaN(stopDate.getTime())) return false; // Don't skip if we can't parse
return stopDate < cutoff;
} catch {
return false; // Don't skip if parsing fails
}
}
/**
* Get database file path (same as main app)
*/
function getDatabasePath(): string {
const dbDir = join(homedir(), '.iptvnator', 'databases');
if (!existsSync(dbDir)) {
mkdirSync(dbDir, { recursive: true });
}
return join(dbDir, 'iptvnator.db');
}
/**
* Database helper class for EPG operations
* Creates its own connection to avoid blocking main thread
*/
class EpgDatabase {
private db: Database.Database;
private knownChannelIds: Set<string> = new Set();
// Prepared statements for better performance
private insertChannelStmt: Database.Statement;
private insertProgramStmt: Database.Statement;
private deleteChannelsStmt: Database.Statement;
constructor() {
const dbPath = getDatabasePath();
this.db = new Database(dbPath);
this.db.pragma('foreign_keys = ON');
this.db.pragma('journal_mode = WAL'); // Better concurrent write performance
// Prepare statements
// Use INSERT OR REPLACE to update existing channels and refresh updated_at
// Use strftime with ISO format for consistent date comparison
this.insertChannelStmt = this.db.prepare(`
INSERT INTO epg_channels (id, display_name, icon_url, url, source_url, updated_at)
VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
ON CONFLICT(id) DO UPDATE SET
display_name = excluded.display_name,
icon_url = excluded.icon_url,
url = excluded.url,
source_url = excluded.source_url,
updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')
`);
this.insertProgramStmt = this.db.prepare(`
INSERT INTO epg_programs (channel_id, start, stop, title, description, category, icon_url, rating, episode_num)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
`);
this.deleteChannelsStmt = this.db.prepare(`
DELETE FROM epg_channels WHERE source_url = ?
`);
}
/**
* Clear existing EPG data for a source URL
*/
clearSourceData(sourceUrl: string): void {
this.deleteChannelsStmt.run(sourceUrl);
this.knownChannelIds.clear();
}
/**
* Insert a batch of channels
*/
insertChannels(channels: ParsedChannel[], sourceUrl: string): void {
const insertMany = this.db.transaction((channels: ParsedChannel[]) => {
for (const channel of channels) {
const displayName = channel.displayName?.[0]?.value || channel.id;
const iconUrl = channel.icon?.[0]?.src || null;
const url = channel.url?.[0] || null;
this.insertChannelStmt.run(
channel.id,
displayName,
iconUrl,
url,
sourceUrl
);
this.knownChannelIds.add(channel.id);
}
});
insertMany(channels);
}
/**
* Insert a batch of programs
* Only inserts programs for known channels to avoid FK constraint failures
*/
insertPrograms(programs: ParsedProgram[]): number {
let insertedCount = 0;
const insertMany = this.db.transaction((programs: ParsedProgram[]) => {
for (const prog of programs) {
// Skip if channel not known
if (!this.knownChannelIds.has(prog.channel)) continue;
const title = prog.title?.[0]?.value || 'Unknown';
const description = prog.desc?.[0]?.value || null;
const category = prog.category?.[0]?.value || null;
const iconUrl = prog.icon?.[0]?.src || null;
const rating = prog.rating?.[0]?.value || null;
const episodeNum = prog.episodeNum?.[0]?.value || null;
try {
this.insertProgramStmt.run(
prog.channel,
prog.start,
prog.stop,
title,
description,
category,
iconUrl,
rating,
episodeNum
);
insertedCount++;
} catch (err) {
// Skip individual failures (e.g., FK constraint)
}
}
});
insertMany(programs);
return insertedCount;
}
/**
* Close the database connection
*/
close(): void {
this.db.close();
}
}
/**
* Parse XMLTV datetime format to ISO string
* Format: YYYYMMDDHHmmss +HHMM or YYYYMMDDHHmmss
@@ -46,7 +258,6 @@ const PROGRAM_BATCH_SIZE = 1000;
function parseXmltvDate(dateStr: string): string {
if (!dateStr) return '';
// Extract date parts: 20231225060000 +0000
const match = dateStr.match(
/^(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})\s*([+-]\d{4})?$/
);
@@ -55,11 +266,9 @@ function parseXmltvDate(dateStr: string): string {
const [, year, month, day, hour, minute, second, tz] = match;
// Build ISO string
let isoString = `${year}-${month}-${day}T${hour}:${minute}:${second}`;
if (tz) {
// Convert +0000 to +00:00
isoString += `${tz.slice(0, 3)}:${tz.slice(3)}`;
} else {
isoString += 'Z';
@@ -73,22 +282,31 @@ function parseXmltvDate(dateStr: string): string {
*/
class StreamingEpgParser {
private parser: SaxesParser;
private channels: EpgChannel[] = [];
private programs: EpgProgram[] = [];
private channels: ParsedChannel[] = [];
private programs: ParsedProgram[] = [];
private totalChannels = 0;
private totalPrograms = 0;
private skippedPrograms = 0;
// Cutoff time for filtering old programs (calculated once at start)
private readonly cutoffTime: Date;
// Current element being parsed
private currentChannel: Partial<EpgChannel> | null = null;
private currentProgram: Partial<EpgProgram> | null = null;
private currentChannel: Partial<ParsedChannel> | null = null;
private currentProgram: Partial<ParsedProgram> | null = null;
private currentTextContent = '';
private currentLang = '';
// For nested elements
private elementStack: string[] = [];
constructor(private onChannelsBatch: (channels: EpgChannel[]) => void, private onProgramsBatch: (programs: EpgProgram[]) => void, private onProgress: (channels: number, programs: number) => void) {
constructor(
private onChannelsBatch: (channels: ParsedChannel[]) => void,
private onProgramsBatch: (programs: ParsedProgram[]) => void,
private onProgress: (channels: number, programs: number) => void
) {
this.parser = new SaxesParser();
this.cutoffTime = getProgramCutoffTime();
this.setupParser();
}
@@ -121,12 +339,8 @@ class StreamingEpgParser {
category: [],
date: '',
episodeNum: [],
previouslyShown: [],
subtitles: [],
icon: [],
rating: [],
credits: [],
audio: [],
};
break;
@@ -163,10 +377,8 @@ class StreamingEpgParser {
case 'rating':
if (this.currentProgram) {
// Rating has system attribute, value comes as child <value> element
const system =
(tag.attributes['system'] as string) || '';
// We'll complete this when we see the value element
this.currentProgram.rating!.push({ system, value: '' });
}
break;
@@ -203,8 +415,7 @@ class StreamingEpgParser {
if (text) this.currentChannel.url!.push(text);
break;
case 'channel':
// Channel complete, add to batch
this.channels.push(this.currentChannel as EpgChannel);
this.channels.push(this.currentChannel as ParsedChannel);
this.totalChannels++;
this.currentChannel = null;
@@ -239,7 +450,6 @@ class StreamingEpgParser {
this.currentProgram.date = text;
break;
case 'value':
// This is for rating value
if (
this.elementStack.includes('rating') &&
this.currentProgram.rating!.length > 0
@@ -257,14 +467,18 @@ class StreamingEpgParser {
}
break;
case 'programme':
// Program complete, add to batch
this.programs.push(this.currentProgram as EpgProgram);
this.totalPrograms++;
this.currentProgram = null;
// Skip programs that have already ended
if (isProgramExpired(this.currentProgram.stop!, this.cutoffTime)) {
this.skippedPrograms++;
} else {
this.programs.push(this.currentProgram as ParsedProgram);
this.totalPrograms++;
if (this.programs.length >= PROGRAM_BATCH_SIZE) {
this.flushPrograms();
if (this.programs.length >= PROGRAM_BATCH_SIZE) {
this.flushPrograms();
}
}
this.currentProgram = null;
break;
}
}
@@ -298,20 +512,29 @@ class StreamingEpgParser {
this.parser.write(chunk);
}
finish(): { totalChannels: number; totalPrograms: number } {
finish(): { totalChannels: number; totalPrograms: number; skippedPrograms: number } {
this.parser.close();
// Flush any remaining data
this.flushChannels();
this.flushPrograms();
if (this.skippedPrograms > 0) {
console.log(
loggerLabel,
`Skipped ${this.skippedPrograms} expired programs (ended more than ${SKIP_PROGRAMS_OLDER_THAN_HOURS}h ago)`
);
}
return {
totalChannels: this.totalChannels,
totalPrograms: this.totalPrograms,
skippedPrograms: this.skippedPrograms,
};
}
}
/**
* Fetches and parses EPG data from URL using streaming
* Inserts directly into SQLite to avoid blocking main thread
*/
async function fetchAndParseEpgStreaming(url: string): Promise<void> {
const isGzipped = url.endsWith('.gz');
@@ -321,90 +544,176 @@ async function fetchAndParseEpgStreaming(url: string): Promise<void> {
`Fetching EPG from ${url} (gzipped: ${isGzipped})`
);
const response = await fetch(url.trim());
// Create database connection in worker
const epgDb = new EpgDatabase();
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
try {
// Clear existing data for this source
console.log(loggerLabel, `Clearing existing data for ${url}`);
epgDb.clearSourceData(url);
if (!response.body) {
throw new Error('Response body is null');
}
const response = await fetch(url.trim());
const parser = new StreamingEpgParser(
(channels) => {
const response: WorkerResponse = {
type: 'EPG_CHANNELS_BATCH',
channels,
};
parentPort?.postMessage(response);
},
(programs) => {
const response: WorkerResponse = {
type: 'EPG_PROGRAMS_BATCH',
programs,
};
parentPort?.postMessage(response);
},
(totalChannels, totalPrograms) => {
const response: WorkerResponse = {
type: 'EPG_PROGRESS',
stats: { totalChannels, totalPrograms },
};
parentPort?.postMessage(response);
}
);
// Convert web stream to Node.js stream
const nodeStream = Readable.fromWeb(response.body as any);
return new Promise((resolve, reject) => {
let dataStream: Readable = nodeStream;
if (isGzipped) {
const gunzip = createGunzip();
dataStream = nodeStream.pipe(gunzip);
gunzip.on('error', (err) => {
console.error(loggerLabel, 'Gunzip error:', err);
reject(err);
});
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
dataStream.on('data', (chunk: Buffer) => {
try {
parser.write(chunk.toString('utf-8'));
} catch (err) {
console.error(loggerLabel, 'Parse error:', err);
reject(err);
}
});
dataStream.on('end', () => {
try {
const stats = parser.finish();
console.log(
loggerLabel,
`Parsing complete: ${stats.totalChannels} channels, ${stats.totalPrograms} programs`
);
if (!response.body) {
throw new Error('Response body is null');
}
const parser = new StreamingEpgParser(
(channels) => {
// Insert channels directly into database
epgDb.insertChannels(channels, url);
},
(programs) => {
// Insert programs directly into database
epgDb.insertPrograms(programs);
},
(totalChannels, totalPrograms) => {
// Send progress to main thread (lightweight)
const response: WorkerResponse = {
type: 'EPG_COMPLETE',
url,
stats,
type: 'EPG_PROGRESS',
stats: { totalChannels, totalPrograms },
};
parentPort?.postMessage(response);
resolve();
} catch (err) {
reject(err);
}
});
);
dataStream.on('error', (err) => {
console.error(loggerLabel, 'Stream error:', err);
reject(err);
// Convert web stream to Node.js stream
const nodeStream = Readable.fromWeb(response.body as any);
return new Promise((resolve, reject) => {
let dataStream: Readable = nodeStream;
if (isGzipped) {
const gunzip = createGunzip();
dataStream = nodeStream.pipe(gunzip);
gunzip.on('error', (err) => {
console.error(loggerLabel, 'Gunzip error:', err);
epgDb.close();
reject(err);
});
}
dataStream.on('data', (chunk: Buffer) => {
try {
parser.write(chunk.toString('utf-8'));
} catch (err) {
console.error(loggerLabel, 'Parse error:', err);
epgDb.close();
reject(err);
}
});
dataStream.on('end', () => {
try {
const stats = parser.finish();
console.log(
loggerLabel,
`Parsing complete: ${stats.totalChannels} channels, ${stats.totalPrograms} programs` +
(stats.skippedPrograms > 0
? ` (skipped ${stats.skippedPrograms} expired)`
: '')
);
// Close database connection
epgDb.close();
const response: WorkerResponse = {
type: 'EPG_COMPLETE',
url,
stats: {
totalChannels: stats.totalChannels,
totalPrograms: stats.totalPrograms,
},
};
parentPort?.postMessage(response);
resolve();
} catch (err) {
epgDb.close();
reject(err);
}
});
dataStream.on('error', (err) => {
console.error(loggerLabel, 'Stream error:', err);
epgDb.close();
reject(err);
});
});
});
} catch (error) {
epgDb.close();
throw error;
}
}
/**
* Clears all EPG data from the database
* Runs in worker thread to avoid blocking main thread
*/
function clearAllEpgData(): void {
const dbPath = getDatabasePath();
const db = new Database(dbPath);
try {
console.log(loggerLabel, 'Clearing all EPG data...');
// Delete programs first (foreign key constraint)
db.exec('DELETE FROM epg_programs');
// Then delete channels
db.exec('DELETE FROM epg_channels');
console.log(loggerLabel, 'All EPG data cleared');
const response: WorkerResponse = { type: 'CLEAR_COMPLETE' };
parentPort?.postMessage(response);
} catch (error) {
console.error(loggerLabel, 'Error clearing EPG data:', error);
const errorResponse: WorkerResponse = {
type: 'EPG_ERROR',
error: error instanceof Error ? error.message : String(error),
};
parentPort?.postMessage(errorResponse);
} finally {
db.close();
}
}
/**
* Cleans up expired EPG programs from the database
* Runs in worker thread to avoid blocking main thread
*/
function cleanupExpiredPrograms(hoursToKeep: number): void {
const dbPath = getDatabasePath();
const db = new Database(dbPath);
try {
const cutoff = new Date(
Date.now() - hoursToKeep * 60 * 60 * 1000
).toISOString();
console.log(loggerLabel, `Cleaning up programs older than ${hoursToKeep}h (before ${cutoff})...`);
const stmt = db.prepare('DELETE FROM epg_programs WHERE stop <= ?');
const result = stmt.run(cutoff);
console.log(loggerLabel, `Cleaned up ${result.changes} expired programs`);
const response: WorkerResponse = { type: 'CLEANUP_COMPLETE' };
parentPort?.postMessage(response);
} catch (error) {
console.error(loggerLabel, 'Error cleaning up expired programs:', error);
const errorResponse: WorkerResponse = {
type: 'EPG_ERROR',
error: error instanceof Error ? error.message : String(error),
};
parentPort?.postMessage(errorResponse);
} finally {
db.close();
}
}
/**
@@ -417,7 +726,11 @@ if (parentPort) {
message.type === 'FETCH_EPG' ||
message.type === 'FORCE_FETCH'
) {
await fetchAndParseEpgStreaming(message.url);
await fetchAndParseEpgStreaming(message.url!);
} else if (message.type === 'CLEAR_EPG') {
clearAllEpgData();
} else if (message.type === 'CLEANUP_EXPIRED') {
cleanupExpiredPrograms(message.hoursToKeep ?? 24);
}
} catch (error) {
console.error(loggerLabel, 'Worker error:', error);
Vendored
+19
View File
@@ -57,6 +57,15 @@ declare global {
forceFetchEpg: (
url: string
) => Promise<{ success: boolean; message?: string }>;
clearEpgData: () => Promise<{ success: boolean }>;
checkEpgFreshness: (
urls: string[],
maxAgeHours?: number
) => Promise<{ staleUrls: string[]; freshUrls: string[] }>;
searchEpgPrograms: (
searchTerm: string,
limit?: number
) => Promise<any[]>;
updateSettings: (settings: any) => Promise<void>;
setMpvPlayerPath: (mpvPlayerPath: string) => Promise<void>;
setVlcPlayerPath: (vlcPlayerPath: string) => Promise<void>;
@@ -173,6 +182,16 @@ declare global {
}) => void
) => void;
getLocalIpAddresses: () => Promise<string[]>;
// EPG progress listener
onEpgProgress?: (
callback: (data: {
url: string;
status: 'loading' | 'complete' | 'error';
stats?: { totalChannels: number; totalPrograms: number };
error?: string;
}) => void
) => void;
dbDeleteAllPlaylists: () => Promise<{ success: boolean }>;
};
process: NodeJS.Process;
require: NodeRequire;
@@ -105,6 +105,57 @@ const CREATE_TABLE_STATEMENTS = [
`CREATE UNIQUE INDEX IF NOT EXISTS recently_viewed_content_playlist_unique ON recently_viewed(content_id, playlist_id)`,
`CREATE INDEX IF NOT EXISTS recently_viewed_playlist_idx ON recently_viewed(playlist_id)`,
`CREATE INDEX IF NOT EXISTS recently_viewed_viewed_at_idx ON recently_viewed(viewed_at)`,
// EPG tables
`CREATE TABLE IF NOT EXISTS epg_channels (
id TEXT PRIMARY KEY,
display_name TEXT NOT NULL,
icon_url TEXT,
url TEXT,
source_url TEXT NOT NULL,
updated_at TEXT DEFAULT (datetime('now'))
)`,
`CREATE TABLE IF NOT EXISTS epg_programs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel_id TEXT NOT NULL,
start TEXT NOT NULL,
stop TEXT NOT NULL,
title TEXT NOT NULL,
description TEXT,
category TEXT,
icon_url TEXT,
rating TEXT,
episode_num TEXT,
FOREIGN KEY (channel_id) REFERENCES epg_channels(id) ON DELETE CASCADE
)`,
// EPG indexes
`CREATE INDEX IF NOT EXISTS idx_epg_channels_source ON epg_channels(source_url)`,
`CREATE INDEX IF NOT EXISTS idx_epg_channels_name ON epg_channels(display_name)`,
`CREATE INDEX IF NOT EXISTS idx_epg_programs_channel ON epg_programs(channel_id)`,
`CREATE INDEX IF NOT EXISTS idx_epg_programs_start ON epg_programs(start)`,
`CREATE INDEX IF NOT EXISTS idx_epg_programs_time_range ON epg_programs(channel_id, start, stop)`,
// FTS5 virtual table for full-text search on EPG programs
`CREATE VIRTUAL TABLE IF NOT EXISTS epg_programs_fts USING fts5(
title,
description,
category,
content='epg_programs',
content_rowid='id'
)`,
// Triggers to keep FTS index in sync with epg_programs table
`CREATE TRIGGER IF NOT EXISTS epg_programs_ai AFTER INSERT ON epg_programs BEGIN
INSERT INTO epg_programs_fts(rowid, title, description, category)
VALUES (new.id, new.title, new.description, new.category);
END`,
`CREATE TRIGGER IF NOT EXISTS epg_programs_ad AFTER DELETE ON epg_programs BEGIN
INSERT INTO epg_programs_fts(epg_programs_fts, rowid, title, description, category)
VALUES ('delete', old.id, old.title, old.description, old.category);
END`,
`CREATE TRIGGER IF NOT EXISTS epg_programs_au AFTER UPDATE ON epg_programs BEGIN
INSERT INTO epg_programs_fts(epg_programs_fts, rowid, title, description, category)
VALUES ('delete', old.id, old.title, old.description, old.category);
INSERT INTO epg_programs_fts(rowid, title, description, category)
VALUES (new.id, new.title, new.description, new.category);
END`,
];
/**
+51
View File
@@ -123,6 +123,51 @@ export const favorites = sqliteTable(
})
);
// EPG Channels table
export const epgChannels = sqliteTable(
'epg_channels',
{
id: text('id').primaryKey(), // Channel ID from EPG source
displayName: text('display_name').notNull(),
iconUrl: text('icon_url'),
url: text('url'),
sourceUrl: text('source_url').notNull(), // Which EPG URL this came from
updatedAt: text('updated_at').default(sql`CURRENT_TIMESTAMP`),
},
(table) => ({
sourceIdx: index('idx_epg_channels_source').on(table.sourceUrl),
nameIdx: index('idx_epg_channels_name').on(table.displayName),
})
);
// EPG Programs table
export const epgPrograms = sqliteTable(
'epg_programs',
{
id: integer('id').primaryKey({ autoIncrement: true }),
channelId: text('channel_id')
.notNull()
.references(() => epgChannels.id, { onDelete: 'cascade' }),
start: text('start').notNull(), // ISO datetime
stop: text('stop').notNull(), // ISO datetime
title: text('title').notNull(),
description: text('description'),
category: text('category'),
iconUrl: text('icon_url'),
rating: text('rating'),
episodeNum: text('episode_num'),
},
(table) => ({
channelIdx: index('idx_epg_programs_channel').on(table.channelId),
startIdx: index('idx_epg_programs_start').on(table.start),
timeRangeIdx: index('idx_epg_programs_time_range').on(
table.channelId,
table.start,
table.stop
),
})
);
// Type exports for TypeScript
export type Playlist = typeof playlists.$inferSelect;
export type NewPlaylist = typeof playlists.$inferInsert;
@@ -138,3 +183,9 @@ export type NewRecentlyViewed = typeof recentlyViewed.$inferInsert;
export type Favorite = typeof favorites.$inferSelect;
export type NewFavorite = typeof favorites.$inferInsert;
export type EpgChannel = typeof epgChannels.$inferSelect;
export type NewEpgChannel = typeof epgChannels.$inferInsert;
export type EpgProgramDb = typeof epgPrograms.$inferSelect;
export type NewEpgProgramDb = typeof epgPrograms.$inferInsert;