Added plex library access

services/request/.env.example

@@ -36,3 +36,16 @@ TORRENT_CLIENT_IPC_SOCKET = /var/autoplex/ipc/torrent_client.sock
 
 # Web server port
 WEBSERVER_PORT = 3200
+
+# Other Services -----------------------------------------------------------------------------------
+
+# The URL to the Plex server
+PLEX_URL = https://plex.dlii.tech
+
+# The X-Plex-Token for the Plex
+PLEX_TOKEN_FILE = /run/secrets/plex_token
+
+
+# Library Keys
+PLEX_LIBRARY_MOVIES_KEY =
+PLEX_LIBRARY_TV_KEY =

services/request/src/lib/plex/index.ts

@@ -0,0 +1,29 @@
+import { LibraryType, IRawLibrariesResponse } from "./schema";
+import ApiRequestManager from "./request";
+import { readFileSync } from "fs";
+
+/**
+ * An interface to the a Plex server
+ */
+export default class Plex
+{
+	/**
+	 * The API request manager instance
+	 */
+	protected requestManager: ApiRequestManager;
+
+	/**
+	 * Create a new instance of the Plex interface
+	 */
+	public constructor(plexToken: string) {
+		this.requestManager = new ApiRequestManager(plexToken);
+	}
+
+	/**
+	 * Fetch the IMDb ID set from the Plex movies library
+	 */
+	public async movies(key: string|number) {
+		let response = await this.requestManager.get(`/library/sections/${key}/all`);
+		return new Set(response.match(/(?<=guid="com\.plexapp\.agents\.imdb\:\/\/)tt\d+/ig));
+	}
+}

services/request/src/lib/plex/request.ts

@@ -0,0 +1,112 @@
+import https, { RequestOptions } from "https";
+
+/**
+ * The URL to the Plex server
+ */
+const API_URL = <string>process.env["PLEX_URL"];
+
+/**
+ * A status error is used to indicate responses with non-200 status codes
+ */
+export class StatusError<T = any> extends Error
+{
+	/**
+	 * The resulting body of a response
+	 */
+	public readonly response: T;
+
+	/**
+	 * The resulting status code of a response
+	 */
+	public readonly statusCode?: number;
+
+	/**
+	 * Create a new error indicating non-200 status
+	 */
+	public constructor(response: T, statusCode: number) {
+		super();
+		Object.setPrototypeOf(this, StatusError.prototype);
+		this.response = response;
+		this.statusCode = statusCode;
+	}
+}
+
+/**
+ * A request manager with atomic/persistent request options
+ */
+export default class ApiRequestManager
+{
+	/**
+	 * The authorization token
+	 */
+	private __plexToken: string;
+
+	/**
+	 * Store additional request options
+	 */
+	protected options: RequestOptions;
+
+	/**
+	 * Create a new API request manager
+	 *
+	 * @param options Additional request options
+	 */
+	public constructor(plexToken: string, options: RequestOptions = {}) {
+		this.__plexToken = plexToken;
+		this.options = options;
+	}
+
+	/**
+	 * Perform a generic HTTPS request
+	 *
+	 * @param method    The HTTP method
+	 * @param url       The URL to request
+	 * @param apiKey An optional bearer token
+	 * @param params    Optional parameters
+	 * @param body      Optional body
+	 */
+	protected request(method: string, url: string, params?: any)
+	{
+		return new Promise<string>((resolve, reject) => {
+			// Create request options
+			let options = Object.assign({ method, headers: {} }, this.options);
+
+			// Add search parameters if necessary
+			let requestUrl = new URL(url);
+			requestUrl.searchParams.set("X-Plex-Token", this.__plexToken);
+			if (params) {
+				Object.keys(params).forEach((key) => {
+					if (params[key] !== undefined) {
+						requestUrl.searchParams.set(key, params[key]);
+					}
+				});
+			}
+
+			// Create the request
+			let request = https.request(<any>requestUrl, options, (res) => {
+				let rawData: string = "";
+				res.setEncoding("utf8");
+				res.on("data", chunk => {rawData += chunk});
+				res.on("error", reject);
+				res.on("end", async () => {
+					rawData;
+					if (res.statusCode == 200) {
+						resolve(rawData)
+					} else {
+						reject(new StatusError(rawData, <number>res.statusCode));
+					}
+				});
+			})
+			.on("error", reject)
+			.on("timeout", () => reject("timeout"));
+			request.end();
+		});
+	}
+
+	/**
+	 * Perform a generic GET request
+	 */
+	public async get(path: string, params?: any) {
+		return await this.request("GET", `${API_URL}${path}`, params);
+	}
+}

services/request/src/lib/plex/schema.ts

@@ -0,0 +1,34 @@
+/**
+ * The supported library types
+ */
+export enum LibraryType {
+	Movie  = "movie",
+	TvShow = "show"
+}
+
+/**
+ * Library directory informaiton
+ */
+export interface IDirectory {
+	$: {
+		key : string,
+		type: LibraryType
+	}
+}
+
+/**
+ * The response generated from a library listing request
+ */
+export interface IRawLibrariesResponse {
+	MediaContainer: {
+		Directory: IDirectory[]
+	}
+}
+
+/**
+ * The parsed library type
+ */
+export interface ILibrary {
+	key : string,
+	type: LibraryType
+}

services/request/src/server/database/entities/PlexMovie.ts

@@ -0,0 +1,47 @@
+import Plex from "@lib/plex";
+import { BaseEntity, Column, Entity, In, PrimaryColumn } from "typeorm";
+
+@Entity()
+export class PlexMovie extends BaseEntity
+{
+	@PrimaryColumn({ length: 27 })
+	imdbId!: string;
+
+	@Column({ type: "int", nullable: true, unique: true })
+	tmdbId!: number|null;
+
+	/**
+	 * Check if a movie is on Plex given its TMDb ID
+	 */
+	public static async isOnPlex(tmdbId: number|string) {
+		return await PlexMovie.count({ where: { tmdbId } }) > 0;
+	}
+
+	/**
+	 * Get the set of IMDb IDs stored in the library
+	 */
+	public static async imdbSet() {
+		let rows = await PlexMovie.createQueryBuilder("plex_movie")
+			.select("imdbId")
+			.getRawMany();
+		return new Set(rows.map(row => <string>row.imdbId));
+	}
+
+	/**
+	 * Insert a set of IMDb IDs into the database
+	 */
+	public static async insertImdbSet(imdbIds: Set<string>) {
+		await PlexMovie.createQueryBuilder()
+			.insert()
+			.into("plex_movie")
+			.values([...imdbIds].map(imdbId => ({ imdbId, tmdbId: null })))
+			.execute();
+	}
+
+	/**
+	 * Remove the given set of IMDb IDs from the library
+	 */
+	public static async removeImdbSet(imdbIds: Set<string>) {
+		await PlexMovie.delete({ imdbId: In([...imdbIds]) });
+	}
+}

services/request/src/server/database/entities/index.ts

@@ -6,5 +6,6 @@ export * from "./MovieInfo";
 export * from "./MovieQuota";
 export * from "./MovieTicket";
 export * from "./MovieTorrent";
+export * from "./PlexMovie";
 export * from "./RegisterToken";
 export * from "./User";

services/request/src/server/services/MovieSearch.ts

@@ -5,7 +5,7 @@ import { request } from "https";
 import Service from "./Service";
 import TvDb from "./TvDb";
 import { IApiMovie, IApiMovieDetails, IApiPaginatedResponse } from "@common/api_schema";
-import { MovieTicket } from "@server/database/entities";
+import { MovieTicket, PlexMovie } from "@server/database/entities";
 import { IMovieSearchResult } from "@lib/tmdb/schema";
 
 const CACHE_CLEAR_INTERVAL = 1000*60; // 60 seconds
@@ -97,11 +97,6 @@ export default class MovieSearch extends Service
 	 */
 	public verifyImdbId(id: string) {
 		return new Promise<boolean>((resolve, reject) => {
-			// If the ID is cached, no need to fetch it
-			// if (id in this.imdbCache) {
-			// 	resolve(true);
-			// }
-			// Verify the movie exists on IMDb by checking for a 404
 			let req = request({ method: "HEAD", host: "www.imdb.com", path: `/title/${id}/` }, (response) => {
 				response.resume();
 				if (response.statusCode == undefined) {
@@ -144,7 +139,7 @@ export default class MovieSearch extends Service
 			runtime     : movie.runtime,
 			title       : movie.title,
 			ticketId    : ticket?.id ?? null,
-			isOnPlex    : false,
+			isOnPlex    : await PlexMovie.isOnPlex(id),
 			requestedBy : (ticket ? {
 				id     : ticket.user.id,
 				isAdmin: ticket.user.isAdmin,
@@ -168,10 +163,13 @@ export default class MovieSearch extends Service
 		let movieFetchRequest = this.tmdb.searchMovie(query, year);
 		let activeTickets = await MovieTicket.activeTicketMap();
 		let results = await movieFetchRequest;
+		let isOnPlex = await Promise.all(results.results.map(
+			movie => PlexMovie.isOnPlex(movie.id)
+		));
 		return <IApiPaginatedResponse<IApiMovie>>{
 			page: results.page,
-			results: results.results.map(movie => <IApiMovie>{
-				isOnPlex   : false,
+			results: results.results.map((movie, index) => <IApiMovie>{
+				isOnPlex   : isOnPlex[index],
 				posterPath : movie.poster_path,
 				releaseDate: movie.release_date,
 				ticketId   : activeTickets[movie.id] ?? null,

services/request/src/server/services/PlexLibrary.ts

@@ -0,0 +1,103 @@
+import Plex from "@lib/plex";
+import Application from "@server/Application";
+import { PlexMovie } from "@server/database/entities";
+import { env, secret, sleep } from "@server/util";
+import MovieSearch from "./MovieSearch";
+import Service from "./Service";
+
+/**
+ * Throttle requests when updating the movie database
+ */
+const API_DATABASE_THROTTLE = 500;
+
+/**
+ * A service for maintaining an internal representation of the Plex library
+ */
+export default class PlexLibrary extends Service
+{
+	/**
+	 * A reference to the Plex library
+	 */
+	protected plex!: Plex;
+
+	/**
+	 * The key for the movies library
+	 */
+	protected moviesKey!: string;
+
+	/**
+	 * The key for the TV shows library
+	 */
+	protected tvKey!: string;
+
+	/**
+	 * Indicate if the plex library is currently being updated
+	 */
+	protected isUpdating: boolean;
+
+	/**
+	 * Create a new Plex library service instance
+	 */
+	public constructor(app: Application) {
+		super("Plex Library", app);
+		this.isUpdating = false;
+	}
+
+	/**
+	 * Boot the Plex library service
+	*/
+	public async boot() {
+		let token = await secret(env("PLEX_TOKEN_FILE"));
+		this.moviesKey = env("PLEX_LIBRARY_MOVIES_KEY");
+		this.tvKey = env("PLEX_LIBRARY_TV_KEY");
+		this.plex = new Plex(token);
+	}
+
+	public start() {
+		this.updateMovies();
+	}
+
+	/**
+	 * Shutdown the Plex library service
+	 */
+	public async shutdown() {}
+
+	/**
+	 * Update the movie catalog
+	 */
+	public async updateMovies() {
+		if (this.isUpdating) {
+			return;
+		}
+		this.isUpdating = true;
+		let start = Date.now();
+		this.log("Updating movies...");
+
+		// Fetch the current and new sets of IMDb IDs
+		let [currentIds, newIds] = await Promise.all([
+			PlexMovie.imdbSet(),
+			this.plex.movies(this.moviesKey),
+		]);
+
+		// Calculate the updates
+		let toInsert = new Set([...newIds].filter(id => !currentIds.has(id)));
+		let toRemove = new Set([...currentIds].filter(id => !newIds.has(id)));
+
+		// Remove ald Ids and insert new ones
+		await PlexMovie.removeImdbSet(toRemove);
+		await PlexMovie.insertImdbSet(toInsert);
+
+		// Update TMDb IDs
+		for (let movie of await PlexMovie.find({ where: { tmdbId: null } })) {
+			let result = await this.app.service<MovieSearch>("Movie Search").findImdb(movie.imdbId);
+			if (result !== null) {
+				movie.tmdbId = result.id;
+				await movie.save();
+			}
+			await sleep(API_DATABASE_THROTTLE);
+		}
+
+		this.log("Movies updated. Took", (Date.now() - start)/1000, "seconds");
+		this.isUpdating = false;
+	}
+}

services/request/src/server/services/index.ts

@@ -1,6 +1,7 @@
 import Database         from "./Database";
 import DiscordBot       from "./DiscordBot";
 import MovieSearch      from "./MovieSearch";
+import PlexLibrary      from "./PlexLibrary";
 import SeekerIpcClient  from "./Ipc/SeekerIpcClient";
 import TorrentIpcClient from "./Ipc/TorrentIpcClient";
 import TvDb             from "./TvDb";
@@ -10,6 +11,7 @@ export default {
 	Database,
 	// DiscordBot,
 	MovieSearch,
+	PlexLibrary,
 	// TorrentIpcClient,
 	SeekerIpcClient,
 	TvDb,

services/request/src/server/util.ts

@@ -3,6 +3,15 @@ import { readFile } from "fs/promises";
 import { readFileSync } from "fs";
 import { randomBytes } from "crypto";
 
+/**
+ * Await for the given number of milliseconds
+ */
+export async function sleep(msecs: number) {
+	return new Promise(resolve => {
+		setTimeout(resolve, msecs);
+	});
+}
+
 /**
  * Fetch an environment variable
  */