Implement program_external_id table (#468)

Motiviation:
Adding this table is a bit of a future-proofing mechanism. We want to
decouple the notion of a 'program' from its external references.
Currently, the only external reference Tunarr knows of is 'plex'.
Eventually, we will add different sources for programming. In theory, a
program could exist in multiple sources. And not only in multiple media
sources, but on other sites that contain more metadata, such as IMDB and
TMDB. The new DB table generalizes Program to have a 1:M relationship
with external references. Additionally, this allows us to maintain
stable and _unstable_ IDs for various sources (ex. plex rating key)
without losing metadata sourcing for a Program.

Changes:
* Migration to add `program_external_id` table
* Backfill fixer task to fill in details from existing programs in the
  DB
* Changes to program DB save path to save to new table and to start
  saving Plex GUID
* Self-heal when Plex rating key changes. This code is triggered when
  starting a stream for an item. If we have a Plex GUID available, we
will attempt to heal the item in the DB and start the stream
* Use the new table in the stream item calculator

TODO:
* Start saving Plex GUID for program_groupings, which have a separate
  external key table
* Start saving other external IDs for programs, such as IMDB ID and TMDB
  ID

server/mikro-orm.base.config.ts

@@ -14,11 +14,15 @@ import { FillerListContent } from './src/dao/entities/FillerListContent.js';
 import { FillerShow } from './src/dao/entities/FillerShow.js';
 import { PlexServerSettings } from './src/dao/entities/PlexServerSettings.js';
 import { Program } from './src/dao/entities/Program.js';
+import { DATABASE_LOCATION_ENV_VAR } from './src/util/constants.js';
 
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = dirname(__filename);
 
-const dbPath = path.join(process.env.DB_PATH ?? '.tunarr', 'db.db');
+const dbPath = path.join(
+  process.env[DATABASE_LOCATION_ENV_VAR] ?? '.tunarr',
+  'db.db',
+);
 
 export default defineConfig({
   dbName: dbPath,

server/package.json

@@ -20,9 +20,11 @@
     "make-exec:macos": "tsx scripts/makeExecutable.ts --target macos-x64-20.11.1",
     "make-exec:windows": "tsx scripts/makeExecutable.ts --target windows-x64-20.11.1 --python python",
     "generate-db-migration": "pnpm run mikro-orm migration:create",
+    "generate-db-migration2": "dotenv -e .env.development -- tsx src/index.ts db generate-migration",
     "generate-db-cache": "mikro-orm-esm cache:generate --combined --ts",
     "mikro-orm": "mikro-orm-esm",
     "preinstall": "npx only-allow pnpm",
+    "run-fixer": "dotenv -e .env.development -- tsx src/index.ts fixer",
     "test": "vitest"
   },
   "dependencies": {

server/src/dao/custom_types/ProgramExternalIdType.ts

@@ -0,0 +1,18 @@
+import { enumKeys } from '../../util/enumUtil.js';
+
+export enum ProgramExternalIdType {
+  PLEX = 'plex',
+  PLEX_GUID = 'plex-guid',
+}
+
+export function programExternalIdTypeFromString(
+  str: string,
+): ProgramExternalIdType | undefined {
+  for (const key of enumKeys(ProgramExternalIdType)) {
+    const value = ProgramExternalIdType[key];
+    if (key.toLowerCase() === str) {
+      return value;
+    }
+  }
+  return;
+}

server/src/dao/derived_types/StreamLineup.ts

@@ -65,9 +65,9 @@ const BaseContentBackedStreamLineupItemSchema =
   baseStreamLineupItemSchema.extend({
     programId: z.string().uuid(),
     // These are taken from the Program DB entity
-    plexFilePath: z.string(),
+    plexFilePath: z.string().optional(),
     externalSourceId: z.string(),
-    filePath: z.string(),
+    filePath: z.string().optional(),
     externalKey: z.string(),
     programType: ProgramTypeEnum,
   });

server/src/dao/entities/Program.ts

@@ -7,6 +7,7 @@ import {
   Index,
   ManyToMany,
   ManyToOne,
+  OneToMany,
   OptionalProps,
   Property,
   Unique,
@@ -23,6 +24,7 @@ import { CustomShow } from './CustomShow.js';
 import { FillerShow } from './FillerShow.js';
 import { ProgramGrouping } from './ProgramGrouping.js';
 import { createExternalId } from '@tunarr/shared';
+import { ProgramExternalId } from './ProgramExternalId.js';
 
 /**
  * Program represents a 'playable' entity. A movie, episode, or music track
@@ -70,6 +72,7 @@ export class Program extends BaseEntity {
 
   /**
    * Previously "key"
+   * @deprecated Use the external key selected from the external IDs relation
    */
   @Property()
   externalKey!: string;
@@ -83,6 +86,7 @@ export class Program extends BaseEntity {
 
   /**
    * Previously "plexFile"
+   * @deprecated Use the file path on the associated external ID
    */
   @Property({ nullable: true })
   plexFilePath?: string;
@@ -164,6 +168,9 @@ export class Program extends BaseEntity {
   @ManyToOne(() => ProgramGrouping, { nullable: true })
   artist?: Rel<ProgramGrouping>;
 
+  @OneToMany(() => ProgramExternalId, (eid) => eid.program)
+  externalIds = new Collection<ProgramExternalId>(this);
+
   toDTO(): ProgramDTO {
     return programDaoToDto(serialize(this as Program, { skipNull: true }));
   }

server/src/dao/entities/ProgramExternalId.ts

@@ -0,0 +1,51 @@
+import {
+  Entity,
+  Enum,
+  ManyToOne,
+  Property,
+  Unique,
+  type Rel,
+} from '@mikro-orm/core';
+import { ProgramExternalIdType } from '../custom_types/ProgramExternalIdType.js';
+import { BaseEntity } from './BaseEntity.js';
+import { Program } from './Program.js';
+
+/**
+ * References to external sources for a {@link Program}
+ *
+ * There are two flavors of IDs:
+ *   1. Source-specific. These are unique in 3 parts: [type, source_id, id]
+ *      e.x. a program's ID specific to a user's Plex server
+ *   2. Source-agnostic. These are unique in 2 parts: [type, id]
+ *      e.x. a program's ID on IMDB
+ */
+@Entity()
+@Unique({ properties: ['uuid', 'sourceType'] })
+export class ProgramExternalId extends BaseEntity {
+  @Enum(() => ProgramExternalIdType)
+  sourceType!: ProgramExternalIdType;
+
+  // Mappings:
+  // - Plex = server name
+  @Property({ nullable: true })
+  externalSourceId?: string;
+
+  // Mappings:
+  // - Plex = ratingKey
+  @Property()
+  externalKey!: string;
+
+  // Mappings:
+  // - Plex = Media.Part.key -- how to access the file via Plex server
+  @Property({ nullable: true })
+  externalFilePath?: string;
+
+  // Mappings:
+  // - Plex = Media.Part.file -- the file path _on_ the plex server
+  //          used in direct streaming mode
+  @Property({ nullable: true })
+  directFilePath?: string;
+
+  @ManyToOne(() => Program)
+  program!: Rel<Program>;
+}

server/src/dao/programDB.ts

@@ -1,9 +1,15 @@
-import { chunk, map } from 'lodash-es';
-import { flatMapAsyncSeq, groupByAndMapAsync } from '../util/index.js';
+import { chunk, isNil, map } from 'lodash-es';
+import {
+  flatMapAsyncSeq,
+  groupByAndMapAsync,
+  isNonEmptyString,
+} from '../util/index.js';
 import { ProgramConverter } from './converters/programConverters.js';
 import { programSourceTypeFromString } from './custom_types/ProgramSourceType';
 import { getEm } from './dataSource';
 import { Program } from './entities/Program';
+import { ProgramExternalId } from './entities/ProgramExternalId.js';
+import { ProgramExternalIdType } from './custom_types/ProgramExternalIdType.js';
 
 export class ProgramDB {
   async lookupByExternalIds(
@@ -61,4 +67,46 @@ export class ProgramDB {
       (r) => converter.partialEntityToContentProgram(r, { skipPopulate: true }),
     );
   }
+
+  async getProgramExternalIds(programId: string) {
+    const em = getEm();
+    return await em.find(ProgramExternalId, {
+      program: programId,
+    });
+  }
+
+  async updateProgramPlexRatingKey(
+    programId: string,
+    plexServerName: string,
+    details: Pick<
+      ProgramExternalId,
+      'externalKey' | 'directFilePath' | 'externalFilePath'
+    >,
+  ) {
+    const em = getEm();
+    const existingRatingKey = await em.findOne(ProgramExternalId, {
+      program: programId,
+      externalSourceId: plexServerName,
+      sourceType: ProgramExternalIdType.PLEX,
+    });
+
+    if (isNil(existingRatingKey)) {
+      const newEid = em.create(ProgramExternalId, {
+        program: em.getReference(Program, programId),
+        sourceType: ProgramExternalIdType.PLEX,
+        externalSourceId: plexServerName,
+        ...details,
+      });
+      em.persist(newEid);
+    } else {
+      existingRatingKey.externalKey = details.externalKey;
+      if (isNonEmptyString(details.externalFilePath)) {
+        existingRatingKey.externalFilePath = details.externalFilePath;
+      }
+      if (isNonEmptyString(details.directFilePath)) {
+        existingRatingKey.directFilePath = details.directFilePath;
+      }
+    }
+    await em.flush();
+  }
 }

server/src/dao/programHelpers.ts

@@ -113,10 +113,6 @@ export async function upsertContentPrograms(
     .compact()
     .value();
 
-  // We verified this is not nil above.
-
-  // TODO: Probably want to do this step in the background...
-  //
   const programsBySource = ld
     .chain(contentPrograms)
     .filter((p) => p.subtype === 'episode' || p.subtype === 'track')
@@ -128,19 +124,17 @@ export async function upsertContentPrograms(
 
   logger.debug('Upserting %d programs', programsToPersist.length);
 
-  const upsertedPrograms = flatten(
-    await mapAsyncSeq(chunk(programsToPersist, batchSize), (programs) =>
-      em.upsertMany(Program, programs, {
-        onConflictAction: 'merge',
-        onConflictFields: ['sourceType', 'externalSourceId', 'externalKey'],
-        onConflictExcludeFields: ['uuid'],
-      }),
-    ),
-  );
+  const upsertedPrograms = await em.upsertMany(Program, programsToPersist, {
+    onConflictAction: 'merge',
+    onConflictFields: ['sourceType', 'externalSourceId', 'externalKey'],
+    onConflictExcludeFields: ['uuid'],
+    batchSize,
+  });
 
   // Fork a new entity manager here so we don't attempt to persist anything
   // in the parent context. This function potentially does a lot of work
   // but we don't want to accidentally not do an upsert of a program.
+  // TODO: Probably want to do this step in the background...
   const programGroupingsBySource =
     await findAndUpdateProgramRelations(programsBySource);