fix(database): implement deduplication for Xtream categories and content

libs/portal/stalker/data-access/src/lib/stores/features/with-stalker-series.feature.ts

@@ -46,9 +46,11 @@ interface ResourceState<T> {
 }
 
 interface StalkerSeriesResponse<T> {
-    js?: {
-        data?: T[];
-    };
+    js?:
+        | {
+              data?: T[];
+          }
+        | T[];
 }
 
 interface StalkerSeriesStoreContext {
@@ -60,6 +62,18 @@ interface StalkerSeriesStoreContext {
     vodSeriesSeasonsResource: ResourceState<StalkerVodSeriesSeason[]>;
 }
 
+function extractSeriesItems<T>(response: StalkerSeriesResponse<T>): T[] {
+    if (Array.isArray(response?.js)) {
+        return response.js;
+    }
+
+    if (Array.isArray(response?.js?.data)) {
+        return response.js.data;
+    }
+
+    return [];
+}
+
 export function withStalkerSeries() {
     const logger = createLogger('withStalkerSeries');
     const toMovieId = (value: unknown): string => {
@@ -116,12 +130,14 @@ export function withStalkerSeries() {
                             );
                         }
 
-                        // Guard: ensure response has expected structure
-                        if (!response?.js?.data) {
+                        const seasons = extractSeriesItems(response);
+
+                        if (seasons.length === 0) {
                             logger.warn('Invalid seasons response', response);
                             return [];
                         }
-                        return sortByNumericValue(response.js.data);
+
+                        return sortByNumericValue(seasons);
                     },
                 }),
                 /**
@@ -181,20 +197,19 @@ export function withStalkerSeries() {
                             );
                         }
 
-                        if (!response?.js?.data) {
+                        const seasonItems = extractSeriesItems(response);
+
+                        if (seasonItems.length === 0) {
                             logger.debug(
                                 'vodSeriesSeasonsResource - no response data'
                             );
                             return [];
                         }
 
-                        logger.debug(
-                            'vodSeriesSeasonsResource response data',
-                            response.js.data
-                        );
+                        logger.debug('vodSeriesSeasonsResource response data', seasonItems);
 
                         // Filter for season items (is_season: true)
-                        const seasons = response.js.data.filter(
+                        const seasons = seasonItems.filter(
                             (item: StalkerVodSeriesSeason) =>
                                 item.is_season === true
                         );
@@ -278,13 +293,15 @@ export function withStalkerSeries() {
                             );
                         }
 
-                        if (!response?.js?.data) {
+                        const episodeItems = extractSeriesItems(response);
+
+                        if (episodeItems.length === 0) {
                             return [];
                         }
 
                         // Filter for episode items (is_episode: true) and sort by series_number
                         const episodes = sortEpisodesByNumber(
-                            response.js.data.filter(
+                            episodeItems.filter(
                                 (item: StalkerVodSeriesEpisode) =>
                                     item.is_episode === true
                             )

libs/portal/xtream/data-access/src/lib/data-sources/electron-xtream-data-source.ts

@@ -34,6 +34,14 @@ export class ElectronXtreamDataSource implements IXtreamDataSource {
     private readonly dbService = inject(DatabaseService);
     private readonly playbackService = inject(PlaybackPositionService);
     private readonly apiService = inject(XtreamApiService);
+    private readonly categoryRequests = new Map<
+        string,
+        Promise<XtreamCategoryFromDb[]>
+    >();
+    private readonly contentRequests = new Map<
+        string,
+        Promise<XtreamContentItem[]>
+    >();
 
     // =========================================================================
     // Playlist Operations
@@ -88,6 +96,28 @@ export class ElectronXtreamDataSource implements IXtreamDataSource {
         type: CategoryType
     ): Promise<XtreamCategoryFromDb[]> {
         const dbType = mapCategoryTypeToDbType(type);
+        const requestKey = `${playlistId}:${dbType}`;
+        const inFlightRequest = this.categoryRequests.get(requestKey);
+
+        if (inFlightRequest) {
+            return inFlightRequest;
+        }
+
+        const request = this.loadCategories(playlistId, credentials, type, dbType)
+            .finally(() => {
+                this.categoryRequests.delete(requestKey);
+            });
+
+        this.categoryRequests.set(requestKey, request);
+        return request;
+    }
+
+    private async loadCategories(
+        playlistId: string,
+        credentials: XtreamCredentials,
+        type: CategoryType,
+        dbType: DbCategoryType
+    ): Promise<XtreamCategoryFromDb[]> {
 
         // Fetch from DB directly — avoids a separate 'has' round-trip.
         // An empty result means the cache is cold; proceed to fetch from API.
@@ -198,6 +228,36 @@ export class ElectronXtreamDataSource implements IXtreamDataSource {
         onProgress?: (count: number) => void,
         onTotal?: (total: number) => void,
         options?: XtreamOperationOptions
+    ): Promise<XtreamContentItem[]> {
+        const requestKey = `${playlistId}:${type}`;
+        const inFlightRequest = this.contentRequests.get(requestKey);
+
+        if (inFlightRequest) {
+            return inFlightRequest;
+        }
+
+        const request = this.loadContent(
+            playlistId,
+            credentials,
+            type,
+            onProgress,
+            onTotal,
+            options
+        ).finally(() => {
+            this.contentRequests.delete(requestKey);
+        });
+
+        this.contentRequests.set(requestKey, request);
+        return request;
+    }
+
+    private async loadContent(
+        playlistId: string,
+        credentials: XtreamCredentials,
+        type: StreamType,
+        onProgress?: (count: number) => void,
+        onTotal?: (total: number) => void,
+        options?: XtreamOperationOptions
     ): Promise<XtreamContentItem[]> {
         // Fetch from DB directly — avoids a separate 'has' round-trip.
         // An empty result means the cache is cold; proceed to fetch from API.

libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.spec.ts

@@ -244,6 +244,40 @@ describe('withContent import state', () => {
         expect(dataSource.getContent).toHaveBeenCalledTimes(3);
     });
 
+    it('ignores concurrent initializeContent calls while an import is already running', async () => {
+        const pendingCategories = {
+            live: createDeferred<any[]>(),
+            vod: createDeferred<any[]>(),
+            series: createDeferred<any[]>(),
+        };
+
+        dataSource.getCategories.mockImplementation(
+            (
+                _playlistId: string,
+                _credentials: unknown,
+                type: 'live' | 'vod' | 'series'
+            ) => pendingCategories[type].promise
+        );
+        dataSource.getContent.mockResolvedValue([]);
+
+        const firstInitialization = store.initializeContent();
+        const secondInitialization = store.initializeContent();
+        await Promise.resolve();
+
+        expect(store.isImporting()).toBe(true);
+        expect(dataSource.getCategories).toHaveBeenCalledTimes(3);
+        expect(dataSource.getContent).not.toHaveBeenCalled();
+
+        pendingCategories.live.resolve([]);
+        pendingCategories.vod.resolve([]);
+        pendingCategories.series.resolve([]);
+
+        await Promise.all([firstInitialization, secondInitialization]);
+
+        expect(dataSource.getCategories).toHaveBeenCalledTimes(3);
+        expect(dataSource.getContent).toHaveBeenCalledTimes(3);
+    });
+
     it('cancels active imports and clears stale progress after worker aborts', async () => {
         const pendingCategories = {
             live: createDeferred<any[]>(),

libs/portal/xtream/data-access/src/lib/stores/features/with-content.feature.ts

@@ -310,8 +310,11 @@ export function withContent() {
                     const ctx = getCredentialsFromStore();
                     if (!ctx) return;
 
-                    // Skip if we already initialized content for this playlist
-                    if (store.isContentInitialized()) {
+                    // Skip duplicate route-session triggers while an import is already
+                    // running. The workspace session currently syncs from multiple
+                    // entry points during bootstrap, and without this guard Electron
+                    // can save the same Xtream categories/content multiple times.
+                    if (store.isContentInitialized() || store.isImporting()) {
                         return;
                     }
 

libs/portal/xtream/feature/src/lib/portal-channels-list/portal-channels-list.component.scss

@@ -10,7 +10,7 @@
 .channels-list {
     display: flex;
     flex-direction: column;
-    height: calc(100% - 56px);
+    height: 100%;
     overflow: hidden;
     width: 100%;
 }

libs/shared/database/src/lib/connection.ts

@@ -213,7 +213,7 @@ const CREATE_TABLE_STATEMENTS = [
  * Migration statements that may fail if already applied
  * These are run with try-catch to handle existing columns
  */
-const MIGRATION_STATEMENTS = [
+const COLUMN_MIGRATION_STATEMENTS = [
     // v1.0.0 -> v1.1.0: Add hidden column to categories for category management
     `ALTER TABLE categories ADD COLUMN hidden INTEGER DEFAULT 0`,
     // v1.1.0 -> v1.2.0: Add playlist metadata/payload columns for M3U + unified playlist persistence
@@ -229,6 +229,12 @@ const MIGRATION_STATEMENTS = [
     `ALTER TABLE favorites ADD COLUMN position INTEGER DEFAULT 0`,
 ];
 
+const INDEX_MIGRATION_STATEMENTS = [
+    // v1.3.0 -> v1.4.0: Prevent duplicate Xtream categories/content rows
+    `CREATE UNIQUE INDEX IF NOT EXISTS categories_playlist_type_xtream_unique ON categories(playlist_id, type, xtream_id)`,
+    `CREATE UNIQUE INDEX IF NOT EXISTS content_category_type_xtream_unique ON content(category_id, type, xtream_id)`,
+];
+
 /**
  * Create tables if they don't exist
  */
@@ -247,11 +253,157 @@ function isDuplicateColumnError(error: unknown): boolean {
     return message.toLowerCase().includes('duplicate column name');
 }
 
-/**
- * Run migrations that may fail if already applied
- */
-function runMigrations(sqliteDb: Database.Database): void {
-    for (const stmt of MIGRATION_STATEMENTS) {
+type XtreamCategoryDuplicateGroup = {
+    playlistId: string;
+    type: 'live' | 'movies' | 'series';
+    xtreamId: number;
+};
+
+type XtreamCategoryCandidate = {
+    id: number;
+    hidden: number;
+    contentCount: number;
+};
+
+type XtreamContentDuplicateGroup = {
+    categoryId: number;
+    type: 'live' | 'movie' | 'series';
+    xtreamId: number;
+};
+
+type XtreamContentCandidate = {
+    id: number;
+};
+
+function deduplicateXtreamCache(sqliteDb: Database.Database): void {
+    const executeCleanup = sqliteDb.transaction(() => {
+        const duplicateCategoryGroups = sqliteDb
+            .prepare(
+                `SELECT
+                    playlist_id AS playlistId,
+                    type,
+                    xtream_id AS xtreamId
+                 FROM categories
+                 GROUP BY playlist_id, type, xtream_id
+                 HAVING COUNT(*) > 1`
+            )
+            .all() as XtreamCategoryDuplicateGroup[];
+
+        const selectCategoryCandidates = sqliteDb.prepare(
+            `SELECT
+                categories.id AS id,
+                COALESCE(categories.hidden, 0) AS hidden,
+                COUNT(content.id) AS contentCount
+             FROM categories
+             LEFT JOIN content ON content.category_id = categories.id
+             WHERE categories.playlist_id = ?
+               AND categories.type = ?
+               AND categories.xtream_id = ?
+             GROUP BY categories.id, categories.hidden
+             ORDER BY COUNT(content.id) DESC, COALESCE(categories.hidden, 0) ASC, categories.id ASC`
+        );
+        const updateContentCategory = sqliteDb.prepare(
+            `UPDATE content SET category_id = ? WHERE category_id = ?`
+        );
+        const deleteCategory = sqliteDb.prepare(
+            `DELETE FROM categories WHERE id = ?`
+        );
+
+        for (const group of duplicateCategoryGroups) {
+            const candidates = selectCategoryCandidates.all(
+                group.playlistId,
+                group.type,
+                group.xtreamId
+            ) as XtreamCategoryCandidate[];
+            const canonicalCategoryId = candidates[0]?.id;
+
+            if (!canonicalCategoryId) {
+                continue;
+            }
+
+            for (const candidate of candidates.slice(1)) {
+                updateContentCategory.run(canonicalCategoryId, candidate.id);
+                deleteCategory.run(candidate.id);
+            }
+        }
+
+        const duplicateContentGroups = sqliteDb
+            .prepare(
+                `SELECT
+                    category_id AS categoryId,
+                    type,
+                    xtream_id AS xtreamId
+                 FROM content
+                 GROUP BY category_id, type, xtream_id
+                 HAVING COUNT(*) > 1`
+            )
+            .all() as XtreamContentDuplicateGroup[];
+
+        const selectContentCandidates = sqliteDb.prepare(
+            `SELECT id
+             FROM content
+             WHERE category_id = ?
+               AND type = ?
+               AND xtream_id = ?
+             ORDER BY id ASC`
+        );
+        const moveFavorites = sqliteDb.prepare(
+            `INSERT INTO favorites (content_id, playlist_id, added_at, position)
+             SELECT ?, playlist_id, added_at, COALESCE(position, 0)
+             FROM favorites
+             WHERE content_id = ?
+             ON CONFLICT(content_id, playlist_id) DO NOTHING`
+        );
+        const deleteFavorites = sqliteDb.prepare(
+            `DELETE FROM favorites WHERE content_id = ?`
+        );
+        const moveRecentlyViewed = sqliteDb.prepare(
+            `INSERT INTO recently_viewed (content_id, playlist_id, viewed_at)
+             SELECT ?, playlist_id, viewed_at
+             FROM recently_viewed
+             WHERE content_id = ?
+             ON CONFLICT(content_id, playlist_id) DO UPDATE SET
+                viewed_at = CASE
+                    WHEN excluded.viewed_at > recently_viewed.viewed_at
+                        THEN excluded.viewed_at
+                    ELSE recently_viewed.viewed_at
+                END`
+        );
+        const deleteRecentlyViewed = sqliteDb.prepare(
+            `DELETE FROM recently_viewed WHERE content_id = ?`
+        );
+        const deleteContent = sqliteDb.prepare(`DELETE FROM content WHERE id = ?`);
+
+        for (const group of duplicateContentGroups) {
+            const candidates = selectContentCandidates.all(
+                group.categoryId,
+                group.type,
+                group.xtreamId
+            ) as XtreamContentCandidate[];
+            const canonicalContentId = candidates[0]?.id;
+
+            if (!canonicalContentId) {
+                continue;
+            }
+
+            for (const candidate of candidates.slice(1)) {
+                moveFavorites.run(canonicalContentId, candidate.id);
+                deleteFavorites.run(candidate.id);
+                moveRecentlyViewed.run(canonicalContentId, candidate.id);
+                deleteRecentlyViewed.run(candidate.id);
+                deleteContent.run(candidate.id);
+            }
+        }
+    });
+
+    executeCleanup();
+}
+
+function runMigrationStatements(
+    sqliteDb: Database.Database,
+    statements: string[]
+): void {
+    for (const stmt of statements) {
         try {
             sqliteDb.exec(stmt);
         } catch (error) {
@@ -275,6 +427,15 @@ function runMigrations(sqliteDb: Database.Database): void {
     }
 }
 
+/**
+ * Run migrations that may fail if already applied
+ */
+function runMigrations(sqliteDb: Database.Database): void {
+    runMigrationStatements(sqliteDb, COLUMN_MIGRATION_STATEMENTS);
+    deduplicateXtreamCache(sqliteDb);
+    runMigrationStatements(sqliteDb, INDEX_MIGRATION_STATEMENTS);
+}
+
 export interface DatabaseOptions {
     /** Open database in read-only mode (for agent-backend) */
     readonly?: boolean;

libs/shared/database/src/lib/schema.ts

@@ -69,6 +69,9 @@ export const categories = sqliteTable(
     (table) => ({
         playlistIdx: index('idx_categories_playlist').on(table.playlistId),
         typeIdx: index('idx_categories_type').on(table.type),
+        playlistTypeXtreamUnique: uniqueIndex(
+            'categories_playlist_type_xtream_unique'
+        ).on(table.playlistId, table.type, table.xtreamId),
     })
 );
 
@@ -92,6 +95,9 @@ export const content = sqliteTable(
         categoryIdx: index('idx_content_category').on(table.categoryId),
         titleIdx: index('idx_content_title').on(table.title),
         xtreamIdx: index('idx_content_xtream').on(table.xtreamId),
+        categoryTypeXtreamUnique: uniqueIndex(
+            'content_category_type_xtream_unique'
+        ).on(table.categoryId, table.type, table.xtreamId),
         typeAddedIdx: index('idx_content_type_added').on(
             table.type,
             table.added