jellyfin/jellyfin-audio-player
1
0
Fork 0
mirror of https://github.com/leinelissen/jellyfin-audio-player.git synced 2026-09-02 21:03:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix: update entity columns

Browse Source
This commit is contained in:
Lei Nelissen
2026-03-01 10:30:43 +01:00
parent 7d71a79996
commit 343f8f321f
33 changed files with 2310 additions and 830 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/store/album-artists/db.ts
+29
View File
@@ -0,0 +1,29 @@
import { db, sqliteDb } from '@/store';
import albumArtists from './entity';
import type { Artist } from '../sources/types';
/**
* Upsert album-artist relations for a given album.
* Preserves order from the source by storing orderIndex.
*/
export async function upsertAlbumArtists(
sourceId: string,
albumId: string,
artistItems: Pick<Artist, 'id'>[],
): Promise<void> {
if (artistItems.length === 0) return;
for (const [orderIndex, artist] of artistItems.entries()) {
await db.insert(albumArtists).values({
sourceId,
albumId,
artistId: artist.id,
orderIndex,
}).onConflictDoUpdate({
target: [albumArtists.sourceId, albumArtists.albumId, albumArtists.artistId],
set: { orderIndex },
});
}
sqliteDb.flushPendingReactiveQueries();
}
src/store/album-similar/db.ts
+20
View File
@@ -0,0 +1,20 @@
import { db, sqliteDb } from '@/store';
import albumSimilar from './entity';
export async function upsertAlbumSimilar(
sourceId: string,
albumId: string,
similarAlbumIds: string[],
): Promise<void> {
if (similarAlbumIds.length === 0) return;
for (const similarAlbumId of similarAlbumIds) {
await db.insert(albumSimilar).values({
sourceId,
albumId,
similarAlbumId,
}).onConflictDoNothing();
}
sqliteDb.flushPendingReactiveQueries();
}
src/store/albums/actions.ts
+27 -7
View File
@@ -6,26 +6,41 @@ import { db, sqliteDb } from '@/store';
import albums from './entity';
import { eq } from 'drizzle-orm';
import type { InsertAlbum } from './types';
import { getAllSourceDrivers } from '../sources/actions';
export async function upsertAlbum(album: InsertAlbum): Promise<void> {
const now = Date.now();
/**
* createdAt and updatedAt are optional — they reflect server-side timestamps
* and are stored as-is when provided, or left null when the server omits them.
* firstSyncedAt and lastSyncedAt are fully managed by the schema and must never
* be set manually: firstSyncedAt is set once on insert; lastSyncedAt is
* updated automatically on every insert and update via $defaultFn/$onUpdateFn.
*/
type UpsertAlbum = Omit<InsertAlbum, 'firstSyncedAt' | 'lastSyncedAt'>;
export async function upsertAlbum(album: UpsertAlbum): Promise<void> {
await db.insert(albums).values({
...album,
createdAt: now,
updatedAt: now,
}).onConflictDoUpdate({
target: albums.id,
set: {
...album,
updatedAt: now,
sourceId: album.sourceId,
name: album.name,
productionYear: album.productionYear,
isFolder: album.isFolder,
albumArtist: album.albumArtist,
metadata: album.metadata,
// Use the server-provided timestamps when available, otherwise null.
// firstSyncedAt is intentionally excluded — it is set once on insert
// and must never be overwritten.
createdAt: album.createdAt ?? null,
updatedAt: album.updatedAt ?? null,
},
});
sqliteDb.flushPendingReactiveQueries();
}
export async function upsertAlbums(albumList: InsertAlbum[]): Promise<void> {
export async function upsertAlbums(albumList: UpsertAlbum[]): Promise<void> {
for (const album of albumList) {
await upsertAlbum(album);
}
@@ -40,3 +55,8 @@ export async function deleteAlbumsBySource(sourceId: string): Promise<void> {
await db.delete(albums).where(eq(albums.sourceId, sourceId));
sqliteDb.flushPendingReactiveQueries();
}
export async function refreshAlbums(): Promise<void> {
// TODO: implement per-driver refresh logic
await getAllSourceDrivers();
}
src/store/albums/entity.ts
+29 -6
View File
@@ -5,20 +5,43 @@ import sources from '../sources/entity';
* Albums table
*/
const albums = sqliteTable('albums', {
/** The source this album belongs to. */
sourceId: text('source_id').notNull().references(() => sources.id, { onDelete: 'cascade' }),
/** Item ID assigned by the source — stable across syncs and used as the primary key. */
id: text('id').primaryKey(),
/** Display name of the album. */
name: text('name').notNull(),
/** Release year as reported by the server, if available. */
productionYear: integer('production_year'),
/** Whether this album is a folder-type item rather than a true album. */
isFolder: integer('is_folder', { mode: 'boolean' }).notNull(),
/** Primary album artist name as reported by the server, if available. */
albumArtist: text('album_artist'),
dateCreated: integer('date_created'),
lastRefreshed: integer('last_refreshed'),
metadataJson: text('metadata_json'), // JSON-encoded additional fields
createdAt: integer('created_at').notNull(),
updatedAt: integer('updated_at').notNull(),
/** Full source API response serialised as JSON, for fields not mapped to dedicated columns. */
metadata: text('metadata'),
/**
* Local timestamp (ms) of the first time this record was synced from the server.
* Set once on insert and never updated — useful as a stable sort fallback.
*/
firstSyncedAt: integer('first_synced_at').notNull().$defaultFn(() => Date.now()),
/**
* Local timestamp (ms) of the most recent sync that touched this record.
* Set automatically on every insert and update.
*/
lastSyncedAt: integer('last_synced_at').notNull().$defaultFn(() => Date.now()).$onUpdateFn(() => Date.now()),
/**
* Server-reported creation timestamp (ms), if provided by the source.
* Null when the server does not supply this value. Used for sorting.
*/
createdAt: integer('created_at'),
/**
* Server-reported last-modified timestamp (ms), if provided by the source.
* Null when the server does not supply this value. Used for sorting.
*/
updatedAt: integer('updated_at'),
}, (table) => [
index('albums_source_name_idx').on(table.sourceId, table.name),
index('albums_source_year_idx').on(table.sourceId, table.productionYear),
]);
export default albums;
export default albums;
src/store/artists/actions.ts
+20 -8
View File
@@ -7,25 +7,37 @@ import artists from './entity';
import { eq } from 'drizzle-orm';
import type { InsertArtist } from './types';
export async function upsertArtist(artist: InsertArtist): Promise<void> {
const now = Date.now();
/**
* createdAt and updatedAt are optional — they reflect server-reported dates and
* are stored as-is (null when the source does not provide them).
* firstSyncedAt and lastSyncedAt are always managed by the schema: firstSyncedAt
* is set once on insert and never overwritten; lastSyncedAt is set automatically
* on every insert and update via $defaultFn/$onUpdateFn.
*/
type UpsertArtist = Omit<InsertArtist, 'firstSyncedAt' | 'lastSyncedAt'>;
export async function upsertArtist(artist: UpsertArtist): Promise<void> {
await db.insert(artists).values({
...artist,
createdAt: now,
updatedAt: now,
}).onConflictDoUpdate({
target: artists.id,
set: {
...artist,
updatedAt: now,
sourceId: artist.sourceId,
name: artist.name,
isFolder: artist.isFolder,
metadata: artist.metadata,
// Use the source-provided dates as-is; null if the source omits them.
createdAt: artist.createdAt,
updatedAt: artist.updatedAt,
// firstSyncedAt is intentionally excluded — preserve the original insert value.
// lastSyncedAt is handled automatically by $onUpdateFn.
},
});
sqliteDb.flushPendingReactiveQueries();
}
export async function upsertArtists(artistList: InsertArtist[]): Promise<void> {
export async function upsertArtists(artistList: UpsertArtist[]): Promise<void> {
for (const artist of artistList) {
await upsertArtist(artist);
}
@@ -39,4 +51,4 @@ export async function deleteArtist(id: string): Promise<void> {
export async function deleteArtistsBySource(sourceId: string): Promise<void> {
await db.delete(artists).where(eq(artists.sourceId, sourceId));
sqliteDb.flushPendingReactiveQueries();
}
}
src/store/artists/entity.ts
+21 -4
View File
@@ -5,15 +5,32 @@ import sources from '../sources/entity';
* Artists table
*/
const artists = sqliteTable('artists', {
/** Foreign key to the source this artist belongs to. */
sourceId: text('source_id').notNull().references(() => sources.id, { onDelete: 'cascade' }),
/** Item ID assigned by the source — stable identifier used as the primary key. */
id: text('id').primaryKey(),
/** Display name of the artist. */
name: text('name').notNull(),
/** Whether this artist is represented as a folder on the server. */
isFolder: integer('is_folder', { mode: 'boolean' }).notNull(),
metadataJson: text('metadata_json'), // JSON-encoded additional fields
createdAt: integer('created_at').notNull(),
updatedAt: integer('updated_at').notNull(),
/** Full source API response serialized as JSON. Preserves all fields not mapped to dedicated columns. */
metadata: text('metadata'),
/**
* Timestamp of when this record was first synced from the server (local time).
* Set once on insert and never overwritten — use this as a stable sort fallback.
*/
firstSyncedAt: integer('first_synced_at').notNull().$defaultFn(() => Date.now()),
/**
* Timestamp of the most recent sync that touched this record (local time).
* Set automatically on every insert and update — never set manually.
*/
lastSyncedAt: integer('last_synced_at').notNull().$defaultFn(() => Date.now()).$onUpdateFn(() => Date.now()),
/** Server-reported creation date. Null if the server did not provide one. */
createdAt: integer('created_at'),
/** Server-reported last-modified date. Null if the server did not provide one. */
updatedAt: integer('updated_at'),
}, (table) => [
index('artists_source_name_idx').on(table.sourceId, table.name),
]);
export default artists;
export default artists;
src/store/database/migrations/0001_tiny_warbound.sql
+104
View File
@@ -0,0 +1,104 @@
ALTER TABLE `app_settings` RENAME TO `settings`;--> statement-breakpoint
ALTER TABLE `downloads` RENAME COLUMN "metadata_json" TO "metadata";--> statement-breakpoint
ALTER TABLE `search_queries` RENAME COLUMN "metadata_json" TO "metadata";--> statement-breakpoint
PRAGMA foreign_keys=OFF;--> statement-breakpoint
CREATE TABLE `__new_albums` (
`source_id` text NOT NULL,
`id` text PRIMARY KEY NOT NULL,
`name` text NOT NULL,
`production_year` integer,
`is_folder` integer NOT NULL,
`album_artist` text,
`metadata` text,
`first_synced_at` integer NOT NULL,
`last_synced_at` integer NOT NULL,
`created_at` integer,
`updated_at` integer,
FOREIGN KEY (`source_id`) REFERENCES `sources`(`id`) ON UPDATE no action ON DELETE cascade
);
--> statement-breakpoint
INSERT INTO `__new_albums`("source_id", "id", "name", "production_year", "is_folder", "album_artist", "metadata", "first_synced_at", "last_synced_at", "created_at", "updated_at") SELECT "source_id", "id", "name", "production_year", "is_folder", "album_artist", "metadata", "first_synced_at", "last_synced_at", "created_at", "updated_at" FROM `albums`;--> statement-breakpoint
DROP TABLE `albums`;--> statement-breakpoint
ALTER TABLE `__new_albums` RENAME TO `albums`;--> statement-breakpoint
PRAGMA foreign_keys=ON;--> statement-breakpoint
CREATE INDEX `albums_source_name_idx` ON `albums` (`source_id`,`name`);--> statement-breakpoint
CREATE INDEX `albums_source_year_idx` ON `albums` (`source_id`,`production_year`);--> statement-breakpoint
CREATE TABLE `__new_artists` (
`source_id` text NOT NULL,
`id` text PRIMARY KEY NOT NULL,
`name` text NOT NULL,
`is_folder` integer NOT NULL,
`metadata` text,
`first_synced_at` integer NOT NULL,
`last_synced_at` integer NOT NULL,
`created_at` integer,
`updated_at` integer,
FOREIGN KEY (`source_id`) REFERENCES `sources`(`id`) ON UPDATE no action ON DELETE cascade
);
--> statement-breakpoint
INSERT INTO `__new_artists`("source_id", "id", "name", "is_folder", "metadata", "first_synced_at", "last_synced_at", "created_at", "updated_at") SELECT "source_id", "id", "name", "is_folder", "metadata", "first_synced_at", "last_synced_at", "created_at", "updated_at" FROM `artists`;--> statement-breakpoint
DROP TABLE `artists`;--> statement-breakpoint
ALTER TABLE `__new_artists` RENAME TO `artists`;--> statement-breakpoint
CREATE INDEX `artists_source_name_idx` ON `artists` (`source_id`,`name`);--> statement-breakpoint
CREATE TABLE `__new_playlists` (
`source_id` text NOT NULL,
`id` text PRIMARY KEY NOT NULL,
`name` text NOT NULL,
`can_delete` integer NOT NULL,
`child_count` integer,
`metadata` text,
`first_synced_at` integer NOT NULL,
`last_synced_at` integer NOT NULL,
`created_at` integer,
`updated_at` integer,
FOREIGN KEY (`source_id`) REFERENCES `sources`(`id`) ON UPDATE no action ON DELETE cascade
);
--> statement-breakpoint
INSERT INTO `__new_playlists`("source_id", "id", "name", "can_delete", "child_count", "metadata", "first_synced_at", "last_synced_at", "created_at", "updated_at") SELECT "source_id", "id", "name", "can_delete", "child_count", "metadata", "first_synced_at", "last_synced_at", "created_at", "updated_at" FROM `playlists`;--> statement-breakpoint
DROP TABLE `playlists`;--> statement-breakpoint
ALTER TABLE `__new_playlists` RENAME TO `playlists`;--> statement-breakpoint
CREATE INDEX `playlists_source_name_idx` ON `playlists` (`source_id`,`name`);--> statement-breakpoint
CREATE TABLE `__new_tracks` (
`source_id` text NOT NULL,
`id` text PRIMARY KEY NOT NULL,
`name` text NOT NULL,
`album_id` text,
`album` text,
`album_artist` text,
`production_year` integer,
`index_number` integer,
`parent_index_number` integer,
`run_time_ticks` integer,
`lyrics` text,
`metadata` text,
`first_synced_at` integer NOT NULL,
`last_synced_at` integer NOT NULL,
`created_at` integer,
`updated_at` integer,
FOREIGN KEY (`source_id`) REFERENCES `sources`(`id`) ON UPDATE no action ON DELETE cascade
);
--> statement-breakpoint
INSERT INTO `__new_tracks`("source_id", "id", "name", "album_id", "album", "album_artist", "production_year", "index_number", "parent_index_number", "run_time_ticks", "lyrics", "metadata", "first_synced_at", "last_synced_at", "created_at", "updated_at") SELECT "source_id", "id", "name", "album_id", "album", "album_artist", "production_year", "index_number", "parent_index_number", "run_time_ticks", "lyrics", "metadata", "first_synced_at", "last_synced_at", "created_at", "updated_at" FROM `tracks`;--> statement-breakpoint
DROP TABLE `tracks`;--> statement-breakpoint
ALTER TABLE `__new_tracks` RENAME TO `tracks`;--> statement-breakpoint
CREATE INDEX `tracks_source_album_idx` ON `tracks` (`source_id`,`album_id`);--> statement-breakpoint
CREATE INDEX `tracks_source_name_idx` ON `tracks` (`source_id`,`name`);--> statement-breakpoint
CREATE TABLE `__new_sync_cursors` (
`source_id` text NOT NULL,
`entity_type` text NOT NULL,
`parent_entity_id` text DEFAULT '' NOT NULL,
`parent_entity_type` text,
`start_index` integer NOT NULL,
`page_size` integer NOT NULL,
`completed` integer DEFAULT false NOT NULL,
`attempts` integer DEFAULT 0 NOT NULL,
`failed_at` integer,
`last_error` text,
`updated_at` integer NOT NULL,
PRIMARY KEY(`source_id`, `entity_type`, `parent_entity_id`),
FOREIGN KEY (`source_id`) REFERENCES `sources`(`id`) ON UPDATE no action ON DELETE cascade
);
--> statement-breakpoint
INSERT INTO `__new_sync_cursors`("source_id", "entity_type", "parent_entity_id", "parent_entity_type", "start_index", "page_size", "completed", "attempts", "failed_at", "last_error", "updated_at") SELECT "source_id", "entity_type", "parent_entity_id", "parent_entity_type", "start_index", "page_size", "completed", "attempts", "failed_at", "last_error", "updated_at" FROM `sync_cursors`;--> statement-breakpoint
DROP TABLE `sync_cursors`;--> statement-breakpoint
ALTER TABLE `__new_sync_cursors` RENAME TO `sync_cursors`;
src/store/database/migrations/meta/0001_snapshot.json
+1237
View File
File diff suppressed because it is too large Load Diff
src/store/database/migrations/meta/_journal.json
+7
View File
@@ -8,6 +8,13 @@
"when": 1770556950693,
"tag": "0000_cuddly_captain_cross",
"breakpoints": true
},
{
"idx": 1,
"version": "6",
"when": 1772357379037,
"tag": "0001_tiny_warbound",
"breakpoints": true
}
]
}
src/store/database/migrations/migrations.js
+5 -3
View File
@@ -2,11 +2,13 @@
import journal from './meta/_journal.json';
import m0000 from './0000_cuddly_captain_cross.sql';
import m0001 from './0001_tiny_warbound.sql';
export default {
export default {
journal,
migrations: {
m0000
m0000,
m0001
}
}
}
src/store/database/prefill/prefill.ts
-669
View File
@@ -1,669 +0,0 @@
/**
* Unified Prefill System
*
* Manages automated data synchronization from media sources to local database.
* Uses p-queue for bounded concurrency with recursive task spawning for pagination.
* Handles both basic entities (artists, albums, playlists) and dependent entities
* (album tracks, playlist tracks, similar albums, lyrics).
*/
import PQueue from 'p-queue';
import { eq, and } from 'drizzle-orm';
import { db } from '@/store/db';
import type { SourceDriver } from '@/store/sources/types';
import syncCursors from '@/store/sync-cursors/entity';
import artists from '@/store/artists/entity';
import albums from '@/store/albums/entity';
import playlists from '@/store/playlists/entity';
import tracks from '@/store/tracks/entity';
import playlistTracks from '@/store/playlist-tracks/entity';
import albumSimilar from '@/store/album-similar/entity';
/**
* Entity types that can be prefilled
*/
export enum EntityType {
ARTISTS = 'artists',
ALBUMS = 'albums',
PLAYLISTS = 'playlists',
ALBUM_TRACKS = 'album_tracks',
PLAYLIST_TRACKS = 'playlist_tracks',
SIMILAR_ALBUMS = 'similar_albums',
LYRICS = 'lyrics',
}
/**
* Progress information for a single entity type
*/
export interface EntityProgress {
entityType: EntityType;
totalFetched: number;
totalInserted: number;
currentPage: number;
isComplete: boolean;
error?: Error;
}
/**
* Overall prefill progress
*/
export interface PrefillProgress {
entities: Record<EntityType, EntityProgress>;
queueSize: number;
pendingTasks: number;
}
export type PrefillProgressCallback = (progress: PrefillProgress) => void;
/**
* Prefill configuration
*/
export interface PrefillConfig {
/** Maximum concurrent requests */
concurrency?: number;
/** Page size for pagination */
pageSize?: number;
/** Progress callback */
onProgress?: PrefillProgressCallback;
/** Whether to fetch dependent entities */
includeDependents?: boolean;
}
const DEFAULT_CONFIG: Required<Omit<PrefillConfig, 'onProgress'>> = {
concurrency: 5,
pageSize: 500,
includeDependents: true,
};
/**
* Unified Prefill Manager
* Uses recursive task spawning for pagination with p-queue for concurrency control
*/
export class PrefillManager {
private queue: PQueue;
private config: Required<Omit<PrefillConfig, 'onProgress'>> & Pick<PrefillConfig, 'onProgress'>;
private sourceId: string;
private driver: SourceDriver;
private progress: Record<EntityType, EntityProgress>;
constructor(sourceId: string, driver: SourceDriver, config: PrefillConfig = {}) {
this.sourceId = sourceId;
this.driver = driver;
this.config = {
...DEFAULT_CONFIG,
...config,
};
this.queue = new PQueue({ concurrency: this.config.concurrency });
// Initialize progress tracking
this.progress = {} as Record<EntityType, EntityProgress>;
Object.values(EntityType).forEach(type => {
this.progress[type] = {
entityType: type,
totalFetched: 0,
totalInserted: 0,
currentPage: 0,
isComplete: false,
};
});
}
/**
* Report current progress
*/
private reportProgress() {
if (this.config.onProgress) {
this.config.onProgress({
entities: { ...this.progress },
queueSize: this.queue.size,
pendingTasks: this.queue.pending,
});
}
}
/**
* Update progress for an entity
*/
private updateProgress(entityType: EntityType, update: Partial<EntityProgress>) {
this.progress[entityType] = {
...this.progress[entityType],
...update,
};
this.reportProgress();
}
/**
* Get sync cursor for resuming prefill
*/
private async getSyncCursor(entityType: EntityType): Promise<number> {
const cursor = await db
.select()
.from(syncCursors)
.where(
and(
eq(syncCursors.sourceId, this.sourceId),
eq(syncCursors.entityType, entityType)
)
)
.limit(1);
return cursor[0]?.startIndex || 0;
}
/**
* Update sync cursor
*/
private async updateSyncCursor(
entityType: EntityType,
startIndex: number,
completed: boolean
) {
const now = Date.now();
await db
.insert(syncCursors)
.values({
sourceId: this.sourceId,
entityType,
startIndex,
pageSize: this.config.pageSize,
completed,
updatedAt: now,
})
.onConflictDoUpdate({
target: [syncCursors.sourceId, syncCursors.entityType],
set: {
startIndex,
pageSize: this.config.pageSize,
completed,
updatedAt: now,
},
});
}
/**
* Recursively fetch and store artists (one page at a time)
*/
private async fetchArtistsPage(offset: number): Promise<void> {
const entityType = EntityType.ARTISTS;
try {
const artistsData = await this.driver.getArtists({
offset,
limit: this.config.pageSize,
});
if (artistsData.length === 0) {
this.updateProgress(entityType, { isComplete: true });
await this.updateSyncCursor(entityType, offset, true);
return;
}
// Insert artists
const now = Date.now();
await db.insert(artists).values(
artistsData.map(artist => ({
sourceId: this.sourceId,
id: artist.id,
name: artist.name,
isFolder: artist.isFolder,
metadataJson: artist.metadataJson,
createdAt: now,
updatedAt: now,
}))
).onConflictDoUpdate({
target: [artists.id],
set: {
name: artistsData[0].name,
isFolder: artistsData[0].isFolder,
metadataJson: artistsData[0].metadataJson,
updatedAt: now,
},
});
const newOffset = offset + artistsData.length;
this.updateProgress(entityType, {
totalFetched: this.progress[entityType].totalFetched + artistsData.length,
totalInserted: this.progress[entityType].totalInserted + artistsData.length,
currentPage: Math.floor(newOffset / this.config.pageSize),
});
await this.updateSyncCursor(entityType, newOffset, false);
// If we got a full page, recursively spawn the next page fetch
if (artistsData.length === this.config.pageSize) {
this.queue.add(() => this.fetchArtistsPage(newOffset));
} else {
this.updateProgress(entityType, { isComplete: true });
await this.updateSyncCursor(entityType, newOffset, true);
}
} catch (error) {
this.updateProgress(entityType, {
error: error instanceof Error ? error : new Error(String(error)),
isComplete: true,
});
throw error;
}
}
/**
* Recursively fetch and store albums (one page at a time)
*/
private async fetchAlbumsPage(offset: number): Promise<void> {
const entityType = EntityType.ALBUMS;
try {
const albumsData = await this.driver.getAlbums({
offset,
limit: this.config.pageSize,
});
if (albumsData.length === 0) {
this.updateProgress(entityType, { isComplete: true });
await this.updateSyncCursor(entityType, offset, true);
return;
}
// Insert albums
const now = Date.now();
await db.insert(albums).values(
albumsData.map(album => ({
sourceId: this.sourceId,
id: album.id,
name: album.name,
productionYear: album.productionYear ?? null,
isFolder: album.isFolder,
albumArtist: album.albumArtist ?? null,
dateCreated: album.dateCreated ?? null,
lastRefreshed: null,
metadataJson: album.metadataJson,
createdAt: now,
updatedAt: now,
}))
).onConflictDoUpdate({
target: [albums.id],
set: {
name: albumsData[0].name,
productionYear: albumsData[0].productionYear ?? null,
isFolder: albumsData[0].isFolder,
albumArtist: albumsData[0].albumArtist ?? null,
dateCreated: albumsData[0].dateCreated ?? null,
metadataJson: albumsData[0].metadataJson,
updatedAt: now,
},
});
const newOffset = offset + albumsData.length;
this.updateProgress(entityType, {
totalFetched: this.progress[entityType].totalFetched + albumsData.length,
totalInserted: this.progress[entityType].totalInserted + albumsData.length,
currentPage: Math.floor(newOffset / this.config.pageSize),
});
await this.updateSyncCursor(entityType, newOffset, false);
// If we got a full page, recursively spawn the next page fetch
if (albumsData.length === this.config.pageSize) {
this.queue.add(() => this.fetchAlbumsPage(newOffset));
} else {
this.updateProgress(entityType, { isComplete: true });
await this.updateSyncCursor(entityType, newOffset, true);
}
} catch (error) {
this.updateProgress(entityType, {
error: error instanceof Error ? error : new Error(String(error)),
isComplete: true,
});
throw error;
}
}
/**
* Recursively fetch and store playlists (one page at a time)
*/
private async fetchPlaylistsPage(offset: number): Promise<void> {
const entityType = EntityType.PLAYLISTS;
try {
const playlistsData = await this.driver.getPlaylists({
offset,
limit: this.config.pageSize,
});
if (playlistsData.length === 0) {
this.updateProgress(entityType, { isComplete: true });
await this.updateSyncCursor(entityType, offset, true);
return;
}
// Insert playlists
const now = Date.now();
await db.insert(playlists).values(
playlistsData.map(playlist => ({
sourceId: this.sourceId,
id: playlist.id,
name: playlist.name,
canDelete: playlist.canDelete,
childCount: playlist.childCount ?? null,
lastRefreshed: null,
metadataJson: playlist.metadataJson,
createdAt: now,
updatedAt: now,
}))
).onConflictDoUpdate({
target: [playlists.id],
set: {
name: playlistsData[0].name,
canDelete: playlistsData[0].canDelete,
childCount: playlistsData[0].childCount ?? null,
metadataJson: playlistsData[0].metadataJson,
updatedAt: now,
},
});
const newOffset = offset + playlistsData.length;
this.updateProgress(entityType, {
totalFetched: this.progress[entityType].totalFetched + playlistsData.length,
totalInserted: this.progress[entityType].totalInserted + playlistsData.length,
currentPage: Math.floor(newOffset / this.config.pageSize),
});
await this.updateSyncCursor(entityType, newOffset, false);
// If we got a full page, recursively spawn the next page fetch
if (playlistsData.length === this.config.pageSize) {
this.queue.add(() => this.fetchPlaylistsPage(newOffset));
} else {
this.updateProgress(entityType, { isComplete: true });
await this.updateSyncCursor(entityType, newOffset, true);
}
} catch (error) {
this.updateProgress(entityType, {
error: error instanceof Error ? error : new Error(String(error)),
isComplete: true,
});
throw error;
}
}
/**
* Fetch tracks for a single album (with pagination)
*/
private async fetchAlbumTracksPage(albumId: string, offset: number): Promise<void> {
const entityType = EntityType.ALBUM_TRACKS;
try {
const tracksData = await this.driver.getTracksByAlbum(albumId, {
offset,
limit: this.config.pageSize,
});
if (tracksData.length === 0) {
return;
}
// Insert tracks
const now = Date.now();
await db.insert(tracks).values(
tracksData.map(track => ({
sourceId: this.sourceId,
id: track.id,
name: track.name,
albumId: track.albumId ?? null,
indexNumber: track.indexNumber ?? null,
parentIndexNumber: track.parentIndexNumber ?? null,
productionYear: track.productionYear ?? null,
runTimeTicks: track.runTimeTicks ?? null,
dateCreated: track.dateCreated ?? null,
lastRefreshed: null,
metadataJson: track.metadataJson,
createdAt: now,
updatedAt: now,
}))
).onConflictDoUpdate({
target: [tracks.id],
set: {
name: tracksData[0].name,
albumId: tracksData[0].albumId ?? null,
indexNumber: tracksData[0].indexNumber ?? null,
parentIndexNumber: tracksData[0].parentIndexNumber ?? null,
productionYear: tracksData[0].productionYear ?? null,
runTimeTicks: tracksData[0].runTimeTicks ?? null,
dateCreated: tracksData[0].dateCreated ?? null,
metadataJson: tracksData[0].metadataJson,
updatedAt: now,
},
});
this.updateProgress(entityType, {
totalFetched: this.progress[entityType].totalFetched + tracksData.length,
totalInserted: this.progress[entityType].totalInserted + tracksData.length,
});
// Recursively fetch next page if this was a full page
if (tracksData.length === this.config.pageSize) {
this.queue.add(() => this.fetchAlbumTracksPage(albumId, offset + this.config.pageSize));
}
} catch (error) {
// Don't throw - just log and continue with other albums
console.error(`Error fetching tracks for album ${albumId}:`, error);
}
}
/**
* Fetch tracks for a single playlist (with pagination)
*/
private async fetchPlaylistTracksPage(playlistId: string, offset: number): Promise<void> {
const entityType = EntityType.PLAYLIST_TRACKS;
try {
const tracksData = await this.driver.getTracksByPlaylist(playlistId, {
offset,
limit: this.config.pageSize,
});
if (tracksData.length === 0) {
return;
}
// Insert tracks first
const now = Date.now();
await db.insert(tracks).values(
tracksData.map(track => ({
sourceId: this.sourceId,
id: track.id,
name: track.name,
albumId: track.albumId ?? null,
indexNumber: track.indexNumber ?? null,
parentIndexNumber: track.parentIndexNumber ?? null,
productionYear: track.productionYear ?? null,
runTimeTicks: track.runTimeTicks ?? null,
dateCreated: track.dateCreated ?? null,
lastRefreshed: null,
metadataJson: track.metadataJson,
createdAt: now,
updatedAt: now,
}))
).onConflictDoUpdate({
target: [tracks.id],
set: {
name: tracksData[0].name,
albumId: tracksData[0].albumId ?? null,
indexNumber: tracksData[0].indexNumber ?? null,
parentIndexNumber: tracksData[0].parentIndexNumber ?? null,
productionYear: tracksData[0].productionYear ?? null,
runTimeTicks: tracksData[0].runTimeTicks ?? null,
dateCreated: tracksData[0].dateCreated ?? null,
metadataJson: tracksData[0].metadataJson,
updatedAt: now,
},
});
// Insert playlist-track relationships
await db.insert(playlistTracks).values(
tracksData.map((track, index) => ({
playlistId,
trackId: track.id,
position: offset + index,
createdAt: now,
}))
).onConflictDoUpdate({
target: [playlistTracks.playlistId, playlistTracks.trackId],
set: {
position: offset,
},
});
this.updateProgress(entityType, {
totalFetched: this.progress[entityType].totalFetched + tracksData.length,
totalInserted: this.progress[entityType].totalInserted + tracksData.length,
});
// Recursively fetch next page if this was a full page
if (tracksData.length === this.config.pageSize) {
this.queue.add(() => this.fetchPlaylistTracksPage(playlistId, offset + this.config.pageSize));
}
} catch (error) {
// Don't throw - just log and continue with other playlists
console.error(`Error fetching tracks for playlist ${playlistId}:`, error);
}
}
/**
* Fetch similar albums for a single album
*/
private async fetchSimilarAlbums(albumId: string): Promise<void> {
try {
const similarAlbums = await this.driver.getSimilarAlbums(albumId, { limit: 20 });
if (similarAlbums.length === 0) {
return;
}
// Insert similar album relationships
const now = Date.now();
await db.insert(albumSimilar).values(
similarAlbums.map((similarAlbum, index) => ({
albumId,
similarAlbumId: similarAlbum.id,
rank: index,
createdAt: now,
}))
).onConflictDoUpdate({
target: [albumSimilar.albumId, albumSimilar.similarAlbumId],
set: {
rank: 0,
},
});
this.updateProgress(EntityType.SIMILAR_ALBUMS, {
totalFetched: this.progress[EntityType.SIMILAR_ALBUMS].totalFetched + similarAlbums.length,
totalInserted: this.progress[EntityType.SIMILAR_ALBUMS].totalInserted + similarAlbums.length,
});
} catch (error) {
// Similar albums are optional, silently fail
console.debug(`Could not fetch similar albums for ${albumId}:`, error);
}
}
/**
* Prefill basic entities (artists, albums, playlists)
*/
async prefillBasicEntities(): Promise<void> {
// Get starting offsets from cursors
const artistsOffset = await this.getSyncCursor(EntityType.ARTISTS);
const albumsOffset = await this.getSyncCursor(EntityType.ALBUMS);
const playlistsOffset = await this.getSyncCursor(EntityType.PLAYLISTS);
// Queue initial page fetches for each entity type
// These will recursively spawn more tasks as needed
this.queue.add(() => this.fetchArtistsPage(artistsOffset));
this.queue.add(() => this.fetchAlbumsPage(albumsOffset));
this.queue.add(() => this.fetchPlaylistsPage(playlistsOffset));
await this.queue.onIdle();
}
/**
* Prefill dependent entities (album tracks, playlist tracks, etc.)
*/
async prefillDependentEntities(): Promise<void> {
// Wait for basic entities to complete first
await this.queue.onIdle();
// Fetch all albums and spawn track fetch tasks
const albumsList = await db.query.albums.findMany({
where: (albums, { eq }) => eq(albums.sourceId, this.sourceId),
});
albumsList.forEach(album => {
this.queue.add(() => this.fetchAlbumTracksPage(album.id, 0));
});
// Fetch all playlists and spawn track fetch tasks
const playlistsList = await db.query.playlists.findMany({
where: (playlists, { eq }) => eq(playlists.sourceId, this.sourceId),
});
playlistsList.forEach(playlist => {
this.queue.add(() => this.fetchPlaylistTracksPage(playlist.id, 0));
});
await this.queue.onIdle();
// Optional: Fetch similar albums (limit to first 100 albums)
const albumsForSimilar = albumsList.slice(0, 100);
albumsForSimilar.forEach(album => {
this.queue.add(() => this.fetchSimilarAlbums(album.id));
});
await this.queue.onIdle();
this.updateProgress(EntityType.ALBUM_TRACKS, { isComplete: true });
this.updateProgress(EntityType.PLAYLIST_TRACKS, { isComplete: true });
this.updateProgress(EntityType.SIMILAR_ALBUMS, { isComplete: true });
}
/**
* Run complete prefill process
*/
async runPrefill(): Promise<void> {
// Prefill basic entities first
await this.prefillBasicEntities();
// Then prefill dependent entities if configured
if (this.config.includeDependents) {
await this.prefillDependentEntities();
}
// Final progress report
this.reportProgress();
}
/**
* Get current progress
*/
getProgress(): PrefillProgress {
return {
entities: { ...this.progress },
queueSize: this.queue.size,
pendingTasks: this.queue.pending,
};
}
}
/**
* Helper function to run prefill
*/
export async function runPrefill(
sourceId: string,
driver: SourceDriver,
config?: PrefillConfig
): Promise<void> {
const manager = new PrefillManager(sourceId, driver, config);
await manager.runPrefill();
}
src/store/downloads/actions.ts
+1 -1
View File
@@ -36,7 +36,7 @@ export async function initializeDownload(
progress: 0,
isFailed: false,
isComplete: false,
metadataJson: null,
metadata: null,
createdAt: now,
updatedAt: now,
}).onConflictDoUpdate({
src/store/downloads/entity.ts
+1 -1
View File
@@ -13,7 +13,7 @@ const downloads = sqliteTable('downloads', {
progress: integer('progress'),
isFailed: integer('is_failed', { mode: 'boolean' }).notNull(),
isComplete: integer('is_complete', { mode: 'boolean' }).notNull(),
metadataJson: text('metadata_json'), // JSON-encoded additional fields
metadata: text('metadata'), // JSON-encoded additional fields
createdAt: integer('created_at').notNull(),
updatedAt: integer('updated_at').notNull(),
});
src/store/downloads/queue.ts
+2 -2
View File
@@ -39,11 +39,11 @@ async function getDriver(): Promise<{ driver: SourceDriver; source: Source } | n
async function updateDownloadMetadata(id: string, updates: Record<string, unknown>): Promise<void> {
const existing = await db.select().from(downloads).where(eq(downloads.id, id)).limit(1);
const current = existing[0];
const currentMetadata = current?.metadataJson ? JSON.parse(current.metadataJson) : {};
const currentMetadata = current?.metadata ? JSON.parse(current.metadata) : {};
await db.update(downloads)
.set({
metadataJson: JSON.stringify({ ...currentMetadata, ...updates }),
metadata: JSON.stringify({ ...currentMetadata, ...updates }),
updatedAt: Date.now(),
})
.where(eq(downloads.id, id));
src/store/playlist-tracks/db.ts
+33
View File
@@ -0,0 +1,33 @@
import { db, sqliteDb } from '@/store';
import { eq, and } from 'drizzle-orm';
import playlistTracks from './entity';
/**
* Replaces all tracks for a playlist with the provided ordered list of track IDs.
* Deletes existing entries first to handle removals and reordering.
*/
export async function setPlaylistTracks(
sourceId: string,
playlistId: string,
trackIds: string[],
): Promise<void> {
await db.delete(playlistTracks).where(
and(
eq(playlistTracks.sourceId, sourceId),
eq(playlistTracks.playlistId, playlistId),
)
);
if (trackIds.length > 0) {
await db.insert(playlistTracks).values(
trackIds.map((trackId, position) => ({
sourceId,
playlistId,
trackId,
position,
}))
);
}
sqliteDb.flushPendingReactiveQueries();
}
src/store/playlists/actions.ts
+21 -32
View File
@@ -4,29 +4,40 @@
import { db, sqliteDb } from '@/store';
import playlists from './entity';
import playlistTracks from '@/store/playlist-tracks/entity';
import { eq, and } from 'drizzle-orm';
import { eq } from 'drizzle-orm';
import type { InsertPlaylist } from './types';
export async function upsertPlaylist(playlist: InsertPlaylist): Promise<void> {
const now = Date.now();
/**
* createdAt and updatedAt are optional — they reflect server-provided timestamps
* and may be null if the server does not supply them.
* firstSyncedAt and lastSyncedAt are omitted from the input type and managed
* entirely by the schema: firstSyncedAt is set once on insert and never
* overwritten; lastSyncedAt is set automatically on every insert and update.
*/
type UpsertPlaylist = Omit<InsertPlaylist, 'firstSyncedAt' | 'lastSyncedAt'>;
export async function upsertPlaylist(playlist: UpsertPlaylist): Promise<void> {
await db.insert(playlists).values({
...playlist,
createdAt: now,
updatedAt: now,
}).onConflictDoUpdate({
target: playlists.id,
set: {
...playlist,
updatedAt: now,
sourceId: playlist.sourceId,
name: playlist.name,
canDelete: playlist.canDelete,
childCount: playlist.childCount,
metadata: playlist.metadata,
// Take server-provided timestamps when available, otherwise leave null.
// firstSyncedAt is intentionally excluded — it is set once on insert and never changed.
createdAt: playlist.createdAt ?? null,
updatedAt: playlist.updatedAt ?? null,
},
});
sqliteDb.flushPendingReactiveQueries();
}
export async function upsertPlaylists(playlistList: InsertPlaylist[]): Promise<void> {
export async function upsertPlaylists(playlistList: UpsertPlaylist[]): Promise<void> {
for (const playlist of playlistList) {
await upsertPlaylist(playlist);
}
@@ -40,26 +51,4 @@ export async function deletePlaylist(id: string): Promise<void> {
export async function deletePlaylistsBySource(sourceId: string): Promise<void> {
await db.delete(playlists).where(eq(playlists.sourceId, sourceId));
sqliteDb.flushPendingReactiveQueries();
}
export async function setPlaylistTracks(sourceId: string, playlistId: string, trackIds: string[]): Promise<void> {
await db.delete(playlistTracks).where(
and(
eq(playlistTracks.sourceId, sourceId),
eq(playlistTracks.playlistId, playlistId)
)
);
if (trackIds.length > 0) {
await db.insert(playlistTracks).values(
trackIds.map((trackId, index) => ({
sourceId,
playlistId,
trackId,
position: index,
}))
);
}
sqliteDb.flushPendingReactiveQueries();
}
}
src/store/playlists/entity.ts
+22 -5
View File
@@ -5,17 +5,34 @@ import sources from '../sources/entity';
* Playlists table
*/
const playlists = sqliteTable('playlists', {
/** Foreign key to the source this playlist belongs to. */
sourceId: text('source_id').notNull().references(() => sources.id, { onDelete: 'cascade' }),
/** Item ID assigned by the source. Used as the primary key. */
id: text('id').primaryKey(),
/** Display name of the playlist. */
name: text('name').notNull(),
/** Whether the current user is allowed to delete this playlist on the server. */
canDelete: integer('can_delete', { mode: 'boolean' }).notNull(),
/** Number of tracks in the playlist, if provided by the server. */
childCount: integer('child_count'),
lastRefreshed: integer('last_refreshed'),
metadataJson: text('metadata_json'), // JSON-encoded additional fields
createdAt: integer('created_at').notNull(),
updatedAt: integer('updated_at').notNull(),
/** Full source API response serialized as JSON, for fields not mapped to dedicated columns. */
metadata: text('metadata'),
/**
* When this record was first synced from the server into the local database.
* Set once on insert and never updated. Useful as a stable sort fallback.
*/
firstSyncedAt: integer('first_synced_at').notNull().$defaultFn(() => Date.now()),
/**
* When this record was most recently synced from the server.
* Set automatically on every insert and update — never set manually.
*/
lastSyncedAt: integer('last_synced_at').notNull().$defaultFn(() => Date.now()).$onUpdateFn(() => Date.now()),
/** When the playlist was created on the server. Null if the server did not provide it. */
createdAt: integer('created_at'),
/** When the playlist was last updated on the server. Null if the server did not provide it. */
updatedAt: integer('updated_at'),
}, (table) => [
index('playlists_source_name_idx').on(table.sourceId, table.name),
]);
export default playlists;
export default playlists;
src/store/search-queries/entity.ts
+1 -1
View File
@@ -10,7 +10,7 @@ const searchQueries = sqliteTable('search_queries', {
query: text('query').notNull(),
timestamp: integer('timestamp').notNull(),
localPlaybackOnly: integer('local_playback_only', { mode: 'boolean' }).notNull(),
metadataJson: text('metadata_json'), // JSON-encoded additional fields
metadata: text('metadata'), // JSON-encoded additional fields
createdAt: integer('created_at').notNull(),
updatedAt: integer('updated_at').notNull(),
}, (table) => [
src/store/sources/actions.ts
+27
View File
@@ -0,0 +1,27 @@
import { db } from "..";
import getDriverBySource from "./drivers";
import { SourceType } from "./types";
/**
* Retrieve all sources from the database
*/
export async function getSources() {
return db.query.sources.findMany();
}
/**
* Instantiate all sources with their respective drivers
*/
export async function getAllSourceDrivers() {
// Retrieve all sources first
const sources = await getSources();
// Then, loop through all sources
return sources.map((source) => {
// Retrieve the appropriate driver class for this source type
const Driver = getDriverBySource(source.type as SourceType);
// Instantiate and return the driver for this source
return new Driver(source);
});
}
src/store/sources/drivers/emby/driver.ts
+37 -29
View File
@@ -134,7 +134,9 @@ export class EmbyDriver extends SourceDriver {
id: item.Id,
name: item.Name,
isFolder: item.IsFolder || false,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
})),
total: response.TotalRecordCount,
offset,
@@ -169,16 +171,15 @@ export class EmbyDriver extends SourceDriver {
productionYear: item.ProductionYear ?? null,
isFolder: item.IsFolder || false,
albumArtist: item.AlbumArtist ?? null,
dateCreated: item.DateCreated
? new Date(item.DateCreated).getTime()
: null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems:
item.ArtistItems?.map((artist) => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -201,16 +202,15 @@ export class EmbyDriver extends SourceDriver {
productionYear: item.ProductionYear ?? null,
isFolder: item.IsFolder || false,
albumArtist: item.AlbumArtist ?? null,
dateCreated: item.DateCreated
? new Date(item.DateCreated).getTime()
: null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems:
item.ArtistItems?.map((artist) => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
};
}
@@ -248,13 +248,15 @@ export class EmbyDriver extends SourceDriver {
indexNumber: item.IndexNumber ?? null,
parentIndexNumber: item.ParentIndexNumber ?? null,
runTimeTicks: item.RunTimeTicks ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems:
item.ArtistItems?.map((artist) => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -288,7 +290,9 @@ export class EmbyDriver extends SourceDriver {
name: item.Name,
canDelete: item.CanDelete || false,
childCount: item.ChildCount ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
})),
total: response.TotalRecordCount,
offset,
@@ -309,7 +313,9 @@ export class EmbyDriver extends SourceDriver {
name: item.Name,
canDelete: item.CanDelete || false,
childCount: item.ChildCount ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
};
}
@@ -345,13 +351,15 @@ export class EmbyDriver extends SourceDriver {
indexNumber: item.IndexNumber ?? null,
parentIndexNumber: item.ParentIndexNumber ?? null,
runTimeTicks: item.RunTimeTicks ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems:
item.ArtistItems?.map((artist) => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -448,16 +456,15 @@ export class EmbyDriver extends SourceDriver {
productionYear: item.ProductionYear ?? null,
isFolder: item.IsFolder || false,
albumArtist: item.AlbumArtist ?? null,
dateCreated: item.DateCreated
? new Date(item.DateCreated).getTime()
: null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems:
item.ArtistItems?.map((artist) => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -493,16 +500,15 @@ export class EmbyDriver extends SourceDriver {
productionYear: item.ProductionYear ?? null,
isFolder: item.IsFolder || false,
albumArtist: item.AlbumArtist ?? null,
dateCreated: item.DateCreated
? new Date(item.DateCreated).getTime()
: null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems:
item.ArtistItems?.map((artist) => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -539,13 +545,15 @@ export class EmbyDriver extends SourceDriver {
indexNumber: item.IndexNumber ?? null,
parentIndexNumber: item.ParentIndexNumber ?? null,
runTimeTicks: item.RunTimeTicks ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems:
item.ArtistItems?.map((artist) => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
src/store/sources/drivers/emby/types.ts
+3
View File
@@ -32,6 +32,7 @@ export interface EmbyBaseItem {
*/
export interface EmbyArtist extends EmbyBaseItem {
IsFolder: boolean;
DateCreated?: string;
}
/**
@@ -56,6 +57,7 @@ export interface EmbyTrack extends EmbyBaseItem {
IndexNumber?: number;
ParentIndexNumber?: number;
RunTimeTicks?: number;
DateCreated?: string;
ArtistItems?: EmbyArtist[];
}
@@ -65,6 +67,7 @@ export interface EmbyTrack extends EmbyBaseItem {
export interface EmbyPlaylist extends EmbyBaseItem {
CanDelete: boolean;
ChildCount?: number;
DateCreated?: string;
}
/**
src/store/sources/drivers/index.ts
+19
View File
@@ -0,0 +1,19 @@
import { Source, SourceDriver, SourceType } from '../types'
import { EmbyDriver } from './emby/driver'
import { JellyfinDriver } from './jellyfin/driver'
/**
* Get the appropriate driver class based on the source type
*/
type ConcreteSourceDriver = new (source: Source) => SourceDriver;
export default function getDriverBySource(type: SourceType): ConcreteSourceDriver {
switch (type) {
case SourceType.JELLYFIN_V1:
return JellyfinDriver;
case SourceType.EMBY_V1:
return EmbyDriver;
default:
throw new Error(`Unsupported source type: ${type}`);
}
}
src/store/sources/drivers/jellyfin/driver.ts
+37 -21
View File
@@ -133,7 +133,9 @@ export class JellyfinDriver extends SourceDriver {
id: item.Id,
name: item.Name,
isFolder: item.IsFolder || false,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
})),
total: response.TotalRecordCount,
offset,
@@ -168,13 +170,14 @@ export class JellyfinDriver extends SourceDriver {
productionYear: item.ProductionYear ?? null,
isFolder: item.IsFolder || false,
albumArtist: item.AlbumArtist ?? null,
dateCreated: item.DateCreated ? new Date(item.DateCreated).getTime() : null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems: item.ArtistItems?.map(artist => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -197,13 +200,14 @@ export class JellyfinDriver extends SourceDriver {
productionYear: item.ProductionYear ?? null,
isFolder: item.IsFolder || false,
albumArtist: item.AlbumArtist ?? null,
dateCreated: item.DateCreated ? new Date(item.DateCreated).getTime() : null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems: item.ArtistItems?.map(artist => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
};
}
@@ -238,12 +242,14 @@ export class JellyfinDriver extends SourceDriver {
indexNumber: item.IndexNumber ?? null,
parentIndexNumber: item.ParentIndexNumber ?? null,
runTimeTicks: item.RunTimeTicks ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems: item.ArtistItems?.map(artist => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -277,7 +283,9 @@ export class JellyfinDriver extends SourceDriver {
name: item.Name,
canDelete: item.CanDelete || false,
childCount: item.ChildCount ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
})),
total: response.TotalRecordCount,
offset,
@@ -298,7 +306,9 @@ export class JellyfinDriver extends SourceDriver {
name: item.Name,
canDelete: item.CanDelete || false,
childCount: item.ChildCount ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
};
}
@@ -331,12 +341,14 @@ export class JellyfinDriver extends SourceDriver {
indexNumber: item.IndexNumber ?? null,
parentIndexNumber: item.ParentIndexNumber ?? null,
runTimeTicks: item.RunTimeTicks ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems: item.ArtistItems?.map(artist => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -410,13 +422,14 @@ export class JellyfinDriver extends SourceDriver {
productionYear: item.ProductionYear ?? null,
isFolder: item.IsFolder || false,
albumArtist: item.AlbumArtist ?? null,
dateCreated: item.DateCreated ? new Date(item.DateCreated).getTime() : null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems: item.ArtistItems?.map(artist => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -449,13 +462,14 @@ export class JellyfinDriver extends SourceDriver {
productionYear: item.ProductionYear ?? null,
isFolder: item.IsFolder || false,
albumArtist: item.AlbumArtist ?? null,
dateCreated: item.DateCreated ? new Date(item.DateCreated).getTime() : null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems: item.ArtistItems?.map(artist => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
@@ -492,12 +506,14 @@ export class JellyfinDriver extends SourceDriver {
indexNumber: item.IndexNumber ?? null,
parentIndexNumber: item.ParentIndexNumber ?? null,
runTimeTicks: item.RunTimeTicks ?? null,
metadataJson: JSON.stringify(item),
metadata: JSON.stringify(item),
createdAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
updatedAt: item.DateCreated ? new Date(item.DateCreated).getTime() : undefined,
artistItems: item.ArtistItems?.map(artist => ({
id: artist.Id,
name: artist.Name,
isFolder: artist.IsFolder,
metadataJson: JSON.stringify(artist),
metadata: JSON.stringify(artist),
})) || [],
})),
total: response.TotalRecordCount,
src/store/sources/drivers/jellyfin/types.ts
+3
View File
@@ -31,6 +31,7 @@ export interface JellyfinBaseItem {
*/
export interface JellyfinArtist extends JellyfinBaseItem {
IsFolder: boolean;
DateCreated?: string;
}
/**
@@ -55,6 +56,7 @@ export interface JellyfinTrack extends JellyfinBaseItem {
IndexNumber?: number;
ParentIndexNumber?: number;
RunTimeTicks?: number;
DateCreated?: string;
ArtistItems?: JellyfinArtist[];
}
@@ -64,6 +66,7 @@ export interface JellyfinTrack extends JellyfinBaseItem {
export interface JellyfinPlaylist extends JellyfinBaseItem {
CanDelete: boolean;
ChildCount?: number;
DateCreated?: string;
}
/**
src/store/sources/sync-manager.ts
+372
View File
@@ -0,0 +1,372 @@
import PQueue from 'p-queue';
import type { SourceDriver } from './types';
import { EntityType, type SyncCursor } from '../sync-cursors/types';
import {
getIncompleteCursors,
createCursorIfNotExists,
updateCursorOffset,
markCursorComplete,
} from '../sync-cursors/db';
import { upsertArtists } from '../artists/actions';
import { upsertAlbums } from '../albums/actions';
import { upsertTracks } from '../tracks/actions';
import { upsertPlaylists } from '../playlists/actions';
import { upsertAlbumArtists } from '../album-artists/db';
import { upsertTrackArtists } from '../track-artists/db';
import { upsertAlbumSimilar } from '../album-similar/db';
import { setPlaylistTracks } from '../playlist-tracks/db';
import { updateTrackLyrics } from '../tracks/db';
const PAGE_SIZE = 500;
/**
* SourceSyncManager
*
* Accepts multiple instantiated SourceDrivers and manages syncing their data into
* the local SQLite database.
*
* Work is persisted as sync_cursor rows before execution, so it survives restarts
* and can be resumed from the last successful page. Dependent tasks (e.g. album
* tracks after albums) are enqueued as cursors during execution and picked up by
* the next iteration of run().
*
* Usage:
* const manager = new SourceSyncManager([jellyfinDriver, embyDriver]);
* await manager.enqueueAlbumsSync(sourceId);
* await manager.run();
*/
export class SourceSyncManager {
private queue: PQueue;
private drivers: Map<string, SourceDriver>;
constructor(drivers: SourceDriver[], concurrency = 5) {
this.drivers = new Map(drivers.map(d => [d.getSourceId(), d]));
this.queue = new PQueue({ concurrency });
}
// -------------------------------------------------------------------------
// Public enqueue methods
// Each method writes a cursor row if one does not already exist, making them
// safe to call multiple times (idempotent). Actual execution only happens
// when run() is called.
// -------------------------------------------------------------------------
async enqueueArtistsSync(sourceId: string): Promise<void> {
await createCursorIfNotExists(sourceId, EntityType.ARTISTS);
}
async enqueueAlbumsSync(sourceId: string): Promise<void> {
await createCursorIfNotExists(sourceId, EntityType.ALBUMS);
}
async enqueueAlbumTracksSync(sourceId: string, albumId: string): Promise<void> {
await createCursorIfNotExists(sourceId, EntityType.ALBUM_TRACKS, albumId, EntityType.ALBUMS);
}
async enqueuePlaylistsSync(sourceId: string): Promise<void> {
await createCursorIfNotExists(sourceId, EntityType.PLAYLISTS);
}
async enqueuePlaylistTracksSync(sourceId: string, playlistId: string): Promise<void> {
await createCursorIfNotExists(sourceId, EntityType.PLAYLIST_TRACKS, playlistId, EntityType.PLAYLISTS);
}
async enqueueSimilarAlbumsSync(sourceId: string, albumId: string): Promise<void> {
await createCursorIfNotExists(sourceId, EntityType.SIMILAR_ALBUMS, albumId, EntityType.ALBUMS);
}
async enqueueLyricsSync(sourceId: string, trackId: string): Promise<void> {
await createCursorIfNotExists(sourceId, EntityType.LYRICS, trackId, EntityType.ALBUM_TRACKS);
}
// -------------------------------------------------------------------------
// run()
//
// Loads all incomplete cursors and drains them through p-queue. Loops until
// no incomplete cursors remain, which handles the case where task execution
// itself creates new cursors (e.g. album pages creating album_tracks cursors).
// -------------------------------------------------------------------------
async run(): Promise<void> {
while (true) {
const cursors = await getIncompleteCursors();
if (cursors.length === 0) break;
for (const cursor of cursors) {
this.queue.add(() => this.executeTask(cursor));
}
await this.queue.onIdle();
}
}
// -------------------------------------------------------------------------
// Task dispatch
// -------------------------------------------------------------------------
private async executeTask(cursor: SyncCursor): Promise<void> {
const driver = this.drivers.get(cursor.sourceId);
if (!driver) return;
const { sourceId, entityType, parentEntityId, startIndex } = cursor;
switch (entityType as EntityType) {
case EntityType.ARTISTS:
await this.executeArtistsPage(driver, sourceId, startIndex);
break;
case EntityType.ALBUMS:
await this.executeAlbumsPage(driver, sourceId, startIndex);
break;
case EntityType.ALBUM_TRACKS:
await this.executeAlbumTracksPage(driver, sourceId, parentEntityId, startIndex);
break;
case EntityType.PLAYLISTS:
await this.executePlaylistsPage(driver, sourceId, startIndex);
break;
case EntityType.PLAYLIST_TRACKS:
await this.executePlaylistTracks(driver, sourceId, parentEntityId);
break;
case EntityType.SIMILAR_ALBUMS:
await this.executeSimilarAlbums(driver, sourceId, parentEntityId);
break;
case EntityType.LYRICS:
await this.executeLyrics(driver, sourceId, parentEntityId);
break;
}
}
// -------------------------------------------------------------------------
// Entity executors
//
// Pagination is handled inline: if a page is full, the cursor offset is
// updated and the next page is added directly to the queue. This keeps all
// pages of a single entity type within one run() iteration.
//
// Dependent tasks (e.g. album_tracks after albums) are written as cursor
// rows only — not added to the current queue. The outer loop in run() picks
// them up in the next iteration, naturally sequencing parents before children.
// -------------------------------------------------------------------------
private async executeArtistsPage(
driver: SourceDriver,
sourceId: string,
offset: number,
): Promise<void> {
const result = await driver.getArtists({ offset, limit: PAGE_SIZE });
await upsertArtists(result.items.map(a => ({
...a,
sourceId,
})));
const newOffset = offset + result.items.length;
if (result.items.length === PAGE_SIZE) {
await updateCursorOffset(sourceId, EntityType.ARTISTS, newOffset);
this.queue.add(() => this.executeArtistsPage(driver, sourceId, newOffset));
} else {
await markCursorComplete(sourceId, EntityType.ARTISTS);
}
}
private async executeAlbumsPage(
driver: SourceDriver,
sourceId: string,
offset: number,
): Promise<void> {
const result = await driver.getAlbums({ offset, limit: PAGE_SIZE });
// Upsert artists embedded in album responses before inserting albums,
// so they exist before album_artists relations reference them.
const embeddedArtists = result.items.flatMap(a => a.artistItems ?? []);
if (embeddedArtists.length > 0) {
await upsertArtists(embeddedArtists.map(a => ({
...a,
sourceId,
})));
}
await upsertAlbums(result.items.map(({ artistItems: _artistItems, ...album }) => ({
...album,
sourceId,
})));
for (const album of result.items) {
if (album.artistItems?.length) {
await upsertAlbumArtists(sourceId, album.id, album.artistItems);
}
}
// Create album_tracks cursors for every album in this page.
// These are picked up in the next run() iteration, after all album pages complete.
for (const album of result.items) {
await this.enqueueAlbumTracksSync(sourceId, album.id);
}
const newOffset = offset + result.items.length;
if (result.items.length === PAGE_SIZE) {
await updateCursorOffset(sourceId, EntityType.ALBUMS, newOffset);
this.queue.add(() => this.executeAlbumsPage(driver, sourceId, newOffset));
} else {
await markCursorComplete(sourceId, EntityType.ALBUMS);
}
}
private async executeAlbumTracksPage(
driver: SourceDriver,
sourceId: string,
albumId: string,
offset: number,
): Promise<void> {
const result = await driver.getTracksByAlbum(albumId, { offset, limit: PAGE_SIZE });
const embeddedArtists = result.items.flatMap(t => t.artistItems ?? []);
if (embeddedArtists.length > 0) {
await upsertArtists(embeddedArtists.map(a => ({
...a,
sourceId,
})));
}
await upsertTracks(result.items.map(({ artistItems: _artistItems, ...track }) => ({
...track,
sourceId,
})));
for (const track of result.items) {
if (track.artistItems?.length) {
await upsertTrackArtists(sourceId, track.id, track.artistItems);
}
}
const newOffset = offset + result.items.length;
if (result.items.length === PAGE_SIZE) {
await updateCursorOffset(sourceId, EntityType.ALBUM_TRACKS, newOffset, albumId);
this.queue.add(() => this.executeAlbumTracksPage(driver, sourceId, albumId, newOffset));
} else {
await markCursorComplete(sourceId, EntityType.ALBUM_TRACKS, albumId);
}
}
private async executePlaylistsPage(
driver: SourceDriver,
sourceId: string,
offset: number,
): Promise<void> {
const result = await driver.getPlaylists({ offset, limit: PAGE_SIZE });
await upsertPlaylists(result.items.map(p => ({
...p,
sourceId,
})));
for (const playlist of result.items) {
await this.enqueuePlaylistTracksSync(sourceId, playlist.id);
}
const newOffset = offset + result.items.length;
if (result.items.length === PAGE_SIZE) {
await updateCursorOffset(sourceId, EntityType.PLAYLISTS, newOffset);
this.queue.add(() => this.executePlaylistsPage(driver, sourceId, newOffset));
} else {
await markCursorComplete(sourceId, EntityType.PLAYLISTS);
}
}
/**
* Fetches all pages of tracks for a playlist in a single execution slot,
* then replaces the playlist's tracks atomically via setPlaylistTracks.
* Playlist tracks are not paginated across cursors because setPlaylistTracks
* requires the complete ordered list.
*/
private async executePlaylistTracks(
driver: SourceDriver,
sourceId: string,
playlistId: string,
): Promise<void> {
const trackIds: string[] = [];
let offset = 0;
while (true) {
const result = await driver.getTracksByPlaylist(playlistId, { offset, limit: PAGE_SIZE });
const embeddedArtists = result.items.flatMap(t => t.artistItems ?? []);
if (embeddedArtists.length > 0) {
await upsertArtists(embeddedArtists.map(a => ({
...a,
sourceId,
})));
}
await upsertTracks(result.items.map(({ artistItems: _artistItems, ...track }) => ({
...track,
sourceId,
})));
for (const track of result.items) {
if (track.artistItems?.length) {
await upsertTrackArtists(sourceId, track.id, track.artistItems);
}
trackIds.push(track.id);
}
offset += result.items.length;
if (result.items.length < PAGE_SIZE) break;
}
await setPlaylistTracks(sourceId, playlistId, trackIds);
await markCursorComplete(sourceId, EntityType.PLAYLIST_TRACKS, playlistId);
}
/**
* Fetches similar albums for a single album. No pagination — the API returns
* a bounded list and we take the first page.
*/
private async executeSimilarAlbums(
driver: SourceDriver,
sourceId: string,
albumId: string,
): Promise<void> {
const result = await driver.getSimilarAlbums(albumId, { limit: PAGE_SIZE });
const embeddedArtists = result.items.flatMap(a => a.artistItems ?? []);
if (embeddedArtists.length > 0) {
await upsertArtists(embeddedArtists.map(a => ({
...a,
sourceId,
})));
}
await upsertAlbums(result.items.map(({ artistItems: _artistItems, ...album }) => ({
...album,
sourceId,
})));
for (const album of result.items) {
if (album.artistItems?.length) {
await upsertAlbumArtists(sourceId, album.id, album.artistItems);
}
}
await upsertAlbumSimilar(sourceId, albumId, result.items.map(a => a.id));
await markCursorComplete(sourceId, EntityType.SIMILAR_ALBUMS, albumId);
}
/**
* Fetches lyrics for a single track and stores the result.
* If the source returns null, lyrics is set to null on the track.
*/
private async executeLyrics(
driver: SourceDriver,
sourceId: string,
trackId: string,
): Promise<void> {
const result = await driver.getTrackLyrics(trackId);
await updateTrackLyrics(trackId, result?.lyrics ?? null);
await markCursorComplete(sourceId, EntityType.LYRICS, trackId);
}
}
src/store/sources/types.ts
+39 -25
View File
@@ -9,6 +9,8 @@ import type { Artist as SchemaArtist } from '../artists/types';
import type { Album as SchemaAlbum } from '../albums/types';
import type { Track as SchemaTrack } from '../tracks/types';
import type { Playlist as SchemaPlaylist } from '../playlists/types';
import { InferSelectModel } from 'drizzle-orm';
import sources from './entity';
/**
* Source types enum
@@ -18,17 +20,7 @@ export enum SourceType {
EMBY_V1 = 'emby.v1',
}
/**
* Source information
*/
export interface Source {
id: string;
uri: string;
userId?: string;
accessToken?: string;
deviceId?: string;
type: SourceType;
}
export type Source = InferSelectModel<typeof sources>;
/**
* Source info returned during connection
@@ -67,34 +59,52 @@ export interface ListResult<T> {
}
/**
* Artist entity returned from drivers
* Compatible with schema but without sourceId, timestamps
* Artist entity returned from drivers.
* sourceId, firstSyncedAt, and lastSyncedAt are omitted — they are managed
* locally by the schema and actions, never supplied by drivers.
* createdAt/updatedAt are optional — drivers provide source-side dates where available.
*/
export type Artist = Omit<SchemaArtist, 'sourceId' | 'createdAt' | 'updatedAt'>;
export type Artist = Omit<SchemaArtist, 'sourceId' | 'firstSyncedAt' | 'lastSyncedAt' | 'createdAt' | 'updatedAt'> & {
createdAt?: number;
updatedAt?: number;
};
/**
* Album entity returned from drivers
* Compatible with schema but without sourceId, timestamps
* Includes temporary artistItems field for relationship data
* Album entity returned from drivers.
* sourceId, firstSyncedAt, and lastSyncedAt are omitted — they are managed
* locally by the schema and actions, never supplied by drivers.
* createdAt/updatedAt are optional — drivers provide source-side dates where available.
* Includes temporary artistItems field for relationship data.
*/
export type Album = Omit<SchemaAlbum, 'sourceId' | 'createdAt' | 'updatedAt' | 'lastRefreshed'> & {
export type Album = Omit<SchemaAlbum, 'sourceId' | 'firstSyncedAt' | 'lastSyncedAt' | 'createdAt' | 'updatedAt'> & {
createdAt?: number;
updatedAt?: number;
artistItems?: Artist[];
};
/**
* Track entity returned from drivers
* Compatible with schema but without sourceId, timestamps
* Includes temporary artistItems field for relationship data
* Track entity returned from drivers.
* sourceId, firstSyncedAt, lastSyncedAt, and lyrics are omitted —
* the sync timestamps are managed by the schema; lyrics are managed locally.
* createdAt/updatedAt are optional — drivers provide source-side dates where available.
* Includes temporary artistItems field for relationship data.
*/
export type Track = Omit<SchemaTrack, 'sourceId' | 'createdAt' | 'updatedAt' | 'hasLyrics' | 'lyrics'> & {
export type Track = Omit<SchemaTrack, 'sourceId' | 'firstSyncedAt' | 'lastSyncedAt' | 'lyrics' | 'createdAt' | 'updatedAt'> & {
createdAt?: number;
updatedAt?: number;
artistItems?: Artist[];
};
/**
* Playlist entity returned from drivers
* Compatible with schema but without sourceId, timestamps
* Playlist entity returned from drivers.
* sourceId, firstSyncedAt, and lastSyncedAt are omitted — they are managed
* locally by the schema and actions, never supplied by drivers.
* createdAt/updatedAt are optional — drivers provide source-side dates where available.
*/
export type Playlist = Omit<SchemaPlaylist, 'sourceId' | 'createdAt' | 'updatedAt' | 'lastRefreshed'>;
export type Playlist = Omit<SchemaPlaylist, 'sourceId' | 'firstSyncedAt' | 'lastSyncedAt' | 'createdAt' | 'updatedAt'> & {
createdAt?: number;
updatedAt?: number;
};
/**
* Search filter types
@@ -179,6 +189,10 @@ export abstract class SourceDriver {
this.source = source;
}
getSourceId(): string {
return this.source.id;
}
/**
* Connect to the source and retrieve server info
*/
src/store/sync-cursors/db.ts
+61
View File
@@ -0,0 +1,61 @@
import { db, sqliteDb } from '@/store';
import { and, eq } from 'drizzle-orm';
import syncCursors from './entity';
import { EntityType, type SyncCursor } from './types';
export async function getIncompleteCursors(): Promise<SyncCursor[]> {
return db.query.syncCursors.findMany({
where: (c) => eq(c.completed, false),
});
}
export async function createCursorIfNotExists(
sourceId: string,
entityType: EntityType,
parentEntityId: string = '',
parentEntityType: EntityType | null = null,
pageSize: number = 500,
): Promise<void> {
await db.insert(syncCursors).values({
sourceId,
entityType,
parentEntityId,
parentEntityType,
startIndex: 0,
pageSize,
completed: false,
updatedAt: Date.now(),
}).onConflictDoNothing();
sqliteDb.flushPendingReactiveQueries();
}
export async function updateCursorOffset(
sourceId: string,
entityType: EntityType,
newOffset: number,
parentEntityId: string = '',
): Promise<void> {
await db.update(syncCursors)
.set({ startIndex: newOffset, updatedAt: Date.now() })
.where(and(
eq(syncCursors.sourceId, sourceId),
eq(syncCursors.entityType, entityType),
eq(syncCursors.parentEntityId, parentEntityId),
));
}
export async function markCursorComplete(
sourceId: string,
entityType: EntityType,
parentEntityId: string = '',
): Promise<void> {
await db.update(syncCursors)
.set({ completed: true, updatedAt: Date.now() })
.where(and(
eq(syncCursors.sourceId, sourceId),
eq(syncCursors.entityType, entityType),
eq(syncCursors.parentEntityId, parentEntityId),
));
}
src/store/sync-cursors/entity.ts
+11 -4
View File
@@ -2,17 +2,24 @@ import { sqliteTable, text, integer, primaryKey } from 'drizzle-orm/sqlite-core'
import sources from '../sources/entity';
/**
* Sync cursors table - tracks prefill progress
* Sync cursors table - tracks sync progress per source, entity type, and optional parent entity.
* The composite PK (sourceId, entityType, parentEntityId) allows independent cursors for
* both top-level entities (parentEntityId = '') and dependent entities (e.g. tracks per album).
*/
const syncCursors = sqliteTable('sync_cursors', {
sourceId: text('source_id').notNull().references(() => sources.id, { onDelete: 'cascade' }),
entityType: text('entity_type').notNull(), // 'artists', 'albums', 'tracks', 'playlists', etc.
entityType: text('entity_type').notNull(),
parentEntityId: text('parent_entity_id').notNull().default(''),
parentEntityType: text('parent_entity_type'),
startIndex: integer('start_index').notNull(),
pageSize: integer('page_size').notNull(),
completed: integer('completed', { mode: 'boolean' }).notNull(),
completed: integer('completed', { mode: 'boolean' }).notNull().default(false),
attempts: integer('attempts').notNull().default(0),
failedAt: integer('failed_at'),
lastError: text('last_error'),
updatedAt: integer('updated_at').notNull(),
}, (table) => [
primaryKey({ columns: [table.sourceId, table.entityType] }),
primaryKey({ columns: [table.sourceId, table.entityType, table.parentEntityId] }),
]);
export default syncCursors;
src/store/sync-cursors/types.ts
+23
View File
@@ -0,0 +1,23 @@
import type { InferSelectModel } from 'drizzle-orm';
import syncCursors from './entity';
export type SyncCursor = InferSelectModel<typeof syncCursors>;
export type InsertSyncCursor = typeof syncCursors.$inferInsert;
/**
* All entity types that can be synced. These values are stored as-is in the
* entity_type column of sync_cursors, so they must remain stable.
*
* Top-level entities (no parent): ARTISTS, ALBUMS, PLAYLISTS
* Dependent entities (require a parentEntityId): ALBUM_TRACKS, PLAYLIST_TRACKS,
* SIMILAR_ALBUMS, LYRICS
*/
export enum EntityType {
ARTISTS = 'artists',
ALBUMS = 'albums',
ALBUM_TRACKS = 'album_tracks',
PLAYLISTS = 'playlists',
PLAYLIST_TRACKS = 'playlist_tracks',
SIMILAR_ALBUMS = 'similar_albums',
LYRICS = 'lyrics',
}
src/store/track-artists/db.ts
+24
View File
@@ -0,0 +1,24 @@
import { db, sqliteDb } from '@/store';
import trackArtists from './entity';
export async function upsertTrackArtists(
sourceId: string,
trackId: string,
artistItems: { id: string }[],
): Promise<void> {
if (artistItems.length === 0) return;
for (const [orderIndex, artist] of artistItems.entries()) {
await db.insert(trackArtists).values({
sourceId,
trackId,
artistId: artist.id,
orderIndex,
}).onConflictDoUpdate({
target: [trackArtists.sourceId, trackArtists.trackId, trackArtists.artistId],
set: { orderIndex },
});
}
sqliteDb.flushPendingReactiveQueries();
}
src/store/tracks/actions.ts
+26 -7
View File
@@ -6,25 +6,44 @@ import { db, sqliteDb } from '@/store';
import tracks from './entity';
import type { InsertTrack } from './types';
export async function upsertTrack(track: InsertTrack): Promise<void> {
const now = Date.now();
/**
* createdAt and updatedAt are optional — they reflect server-provided timestamps
* and are stored as-is. Null when the server does not supply them.
*
* firstSyncedAt and lastSyncedAt are always managed by the schema and are
* excluded from the upsert type: firstSyncedAt is set once on insert and never
* overwritten; lastSyncedAt is set automatically on every insert and update.
*/
type UpsertTrack = Omit<InsertTrack, 'firstSyncedAt' | 'lastSyncedAt'>;
export async function upsertTrack(track: UpsertTrack): Promise<void> {
await db.insert(tracks).values({
...track,
createdAt: now,
updatedAt: now,
}).onConflictDoUpdate({
target: tracks.id,
set: {
...track,
updatedAt: now,
sourceId: track.sourceId,
name: track.name,
albumId: track.albumId,
album: track.album,
albumArtist: track.albumArtist,
productionYear: track.productionYear,
indexNumber: track.indexNumber,
parentIndexNumber: track.parentIndexNumber,
runTimeTicks: track.runTimeTicks,
metadata: track.metadata,
// createdAt and updatedAt reflect source-provided values; store as-is.
createdAt: track.createdAt,
updatedAt: track.updatedAt,
// firstSyncedAt is intentionally absent — preserve the original insert value.
// lastSyncedAt is intentionally absent — the schema $onUpdateFn handles it.
},
});
sqliteDb.flushPendingReactiveQueries();
}
export async function upsertTracks(trackList: InsertTrack[]): Promise<void> {
export async function upsertTracks(trackList: UpsertTrack[]): Promise<void> {
for (const track of trackList) {
await upsertTrack(track);
}
src/store/tracks/db.ts
+20
View File
@@ -0,0 +1,20 @@
import { db, sqliteDb } from '@/store';
import { eq } from 'drizzle-orm';
import tracks from './entity';
/**
* Update lyrics content for a track.
* Called after fetching lyrics from the source driver.
* Note: updatedAt is intentionally not set here — it reflects the server-reported
* last-modified date, not local operations. lastSyncedAt is bumped automatically
* by the schema $onUpdateFn.
*/
export async function updateTrackLyrics(trackId: string, lyrics: string | null): Promise<void> {
await db.update(tracks)
.set({
lyrics,
})
.where(eq(tracks.id, trackId));
sqliteDb.flushPendingReactiveQueries();
}
src/store/tracks/entity.ts
+28 -5
View File
@@ -5,24 +5,47 @@ import sources from '../sources/entity';
* Tracks table
*/
const tracks = sqliteTable('tracks', {
/** Foreign key to the source this track belongs to. */
sourceId: text('source_id').notNull().references(() => sources.id, { onDelete: 'cascade' }),
/** Item ID assigned by the source. */
id: text('id').primaryKey(),
/** Display name of the track. */
name: text('name').notNull(),
/** Item ID of the parent album as assigned by the source, if any. */
albumId: text('album_id'),
/** Display name of the parent album, if any. */
album: text('album'),
/** Display name of the album artist, if any. */
albumArtist: text('album_artist'),
/** Year the track was produced, if known. */
productionYear: integer('production_year'),
/** Track number within its disc/album, if known. */
indexNumber: integer('index_number'),
/** Disc number within the album, if known. */
parentIndexNumber: integer('parent_index_number'),
hasLyrics: integer('has_lyrics', { mode: 'boolean' }).notNull().default(false),
/** Duration of the track in ticks (1 tick = 100 nanoseconds), if known. */
runTimeTicks: integer('run_time_ticks'),
/** Cached lyrics text, populated on demand. */
lyrics: text('lyrics'),
metadataJson: text('metadata_json'), // JSON-encoded additional fields
createdAt: integer('created_at').notNull(),
updatedAt: integer('updated_at').notNull(),
/** Full source API response serialised as JSON, for fields not promoted to dedicated columns. */
metadata: text('metadata'),
/**
* When this record was first synced from the server locally. Set once on
* insert and never updated — use for stable sorting when remote dates are absent.
*/
firstSyncedAt: integer('first_synced_at').notNull().$defaultFn(() => Date.now()),
/**
* When this record was most recently synced from the server.
* Set automatically on every insert and update — never set manually.
*/
lastSyncedAt: integer('last_synced_at').notNull().$defaultFn(() => Date.now()).$onUpdateFn(() => Date.now()),
/** Source-reported creation date. Null if the source did not provide one. */
createdAt: integer('created_at'),
/** Source-reported last-updated date. Null if the source did not provide one. */
updatedAt: integer('updated_at'),
}, (table) => [
index('tracks_source_album_idx').on(table.sourceId, table.albumId),
index('tracks_source_name_idx').on(table.sourceId, table.name),
]);
export default tracks;
export default tracks;