Filtering in assets table (#101)

* Added transcodeOutcome job

* Added assets page

* feat: Added table for assets

* Added filtering to assets page

packages/api/src/db/index.ts

@@ -1,5 +1,5 @@
 import { Kysely, PostgresDialect } from "kysely";
-import { Pool } from "pg";
+import { Pool, types } from "pg";
 import { env } from "../env";
 import type { KyselyDatabase } from "./types.ts";
 
@@ -8,3 +8,7 @@ export const db = new Kysely<KyselyDatabase>({
     pool: new Pool({ connectionString: env.DATABASE }),
   }),
 });
+
+types.setTypeParser(/* INT8_TYPE_ID= */ 20, (val) => {
+  return parseInt(val, 10);
+});

packages/api/src/db/migrations/2024_10_26_init.ts

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 
-import { Kysely } from "kysely";
+import { Kysely, sql } from "kysely";
 
 export async function up(db: Kysely<any>) {
   await db.schema
@@ -10,7 +10,36 @@ export async function up(db: Kysely<any>) {
     .addColumn("username", "text", (col) => col.notNull())
     .addColumn("password", "text", (col) => col.notNull())
     // Settings
-    .addColumn("autoRefresh", "boolean", (col) => col.defaultTo(true))
+    .addColumn("autoRefresh", "boolean", (col) => col.notNull().defaultTo(true))
+    .execute();
+
+  await db.schema
+    .createTable("groups")
+    .addColumn("id", "serial", (col) => col.primaryKey())
+    .addColumn("name", "text", (col) => col.notNull().unique())
+    .execute();
+
+  await db.schema
+    .createTable("assets")
+    .addColumn("id", "uuid", (col) => col.primaryKey())
+    .addColumn("groupId", "integer", (col) =>
+      col.references("groups.id").onDelete("cascade"),
+    )
+    .addColumn("createdAt", "timestamp", (col) =>
+      col.notNull().defaultTo(sql`now()`),
+    )
+    .execute();
+
+  await db.schema
+    .createTable("playables")
+    .addColumn("assetId", "uuid", (col) =>
+      col.references("assets.id").onDelete("cascade"),
+    )
+    .addColumn("name", "text", (col) => col.notNull())
+    .addColumn("createdAt", "timestamp", (col) =>
+      col.notNull().defaultTo(sql`now()`),
+    )
+    .addPrimaryKeyConstraint("id", ["assetId", "name"])
     .execute();
 
   await db
@@ -20,14 +49,11 @@ export async function up(db: Kysely<any>) {
       password: await Bun.password.hash("admin"),
     })
     .execute();
-
-  await db.schema
-    .createTable("assets")
-    .addColumn("id", "uuid", (col) => col.primaryKey())
-    .execute();
 }
 
 export async function down(db: Kysely<any>) {
   await db.schema.dropTable("users").execute();
-  await db.schema.dropTable("assets");
+  await db.schema.dropTable("groups").execute();
+  await db.schema.dropTable("assets").execute();
+  await db.schema.dropTable("playables").execute();
 }

packages/api/src/db/types.ts

@@ -1,8 +1,10 @@
-import type { Generated, Insertable, Updateable } from "kysely";
+import type { Generated, Insertable, Updateable, ColumnType } from "kysely";
 
 export interface KyselyDatabase {
   users: UsersTable;
+  groups: GroupsTable;
   assets: AssetsTable;
+  playables: PlayablesTable;
 }
 
 export interface UsersTable {
@@ -14,8 +16,25 @@ export interface UsersTable {
 
 export type UserUpdate = Updateable<UsersTable>;
 
+export interface GroupsTable {
+  id: Generated<number>;
+  name: string;
+}
+
+export type GroupInsert = Insertable<GroupsTable>;
+
 export interface AssetsTable {
   id: string;
+  groupId: number | null;
+  createdAt: ColumnType<Date, never, never>;
 }
 
 export type AssetInsert = Insertable<AssetsTable>;
+
+export interface PlayablesTable {
+  assetId: string;
+  name: string;
+  createdAt: ColumnType<Date, never, never>;
+}
+
+export type PlayableInsert = Insertable<PlayablesTable>;

packages/api/src/errors.ts

@@ -77,7 +77,14 @@ export const errors = () =>
 function mapValidationError(
   error: ValidationError,
 ): ApiError<"ERR_VALIDATION"> {
-  const first = error.validator.Errors(error.value).First();
+  const first = error.validator?.Errors(error.value).First();
+  if (!first) {
+    return {
+      type: "ERR_VALIDATION",
+      path: "/",
+      fail: error.message,
+    };
+  }
   return {
     type: "ERR_VALIDATION",
     path: first.path,

packages/api/src/index.ts

@@ -6,6 +6,7 @@ import { token } from "./routes/token";
 import { user } from "./routes/user";
 import { jobs } from "./routes/jobs";
 import { storage } from "./routes/storage";
+import { assets } from "./routes/assets";
 import { errors } from "./errors";
 import {
   LangCodeSchema,
@@ -18,8 +19,13 @@ import {
   JobSchema,
   StorageFolderSchema,
   StorageFileSchema,
+  AssetSchema,
+  GroupSchema,
 } from "./types";
 
+// Import workers and they'll start running immediately.
+import "./workers";
+
 export type App = typeof app;
 
 const app = new Elysia()
@@ -90,11 +96,14 @@ const app = new Elysia()
     Job: JobSchema,
     StorageFolder: StorageFolderSchema,
     StorageFile: StorageFileSchema,
+    Asset: AssetSchema,
+    Group: GroupSchema,
   })
   .use(token)
   .use(user)
   .use(jobs)
-  .use(storage);
+  .use(storage)
+  .use(assets);
 
 app.on("stop", () => {
   process.exit(0);

packages/api/src/repositories/assets.ts

@@ -4,3 +4,57 @@ import type { AssetInsert } from "../db/types";
 export async function createAsset(fields: AssetInsert) {
   return await db.insertInto("assets").values(fields).executeTakeFirstOrThrow();
 }
+
+export async function getAssetsCount() {
+  const { count } = await db
+    .selectFrom("assets")
+    .select((eb) => eb.fn.count<number>("id").as("count"))
+    .executeTakeFirstOrThrow();
+  return count;
+}
+
+export async function getAssets(filter: {
+  page: number;
+  perPage: number;
+  orderBy: string;
+  direction: string;
+}) {
+  const orderBy = mapOrderBy(filter.orderBy);
+  const direction = mapDirection(filter.direction);
+
+  const assets = await db
+    .selectFrom("assets")
+    .leftJoin("playables", "playables.assetId", "assets.id")
+    .select(({ fn }) => [
+      "assets.id",
+      "assets.groupId",
+      "assets.createdAt",
+      fn.count<number>("playables.assetId").as("playablesCount"),
+    ])
+    .groupBy("assets.id")
+    .limit(filter.perPage)
+    .offset((filter.page - 1) * filter.perPage)
+    .orderBy(orderBy, direction)
+    .execute();
+
+  return assets.map((asset) => {
+    return {
+      ...asset,
+      name: asset.id,
+    };
+  });
+}
+
+function mapOrderBy(orderBy: string) {
+  if (orderBy === "name") {
+    return "id";
+  }
+  if (orderBy === "createdAt") {
+    return "createdAt";
+  }
+  return "createdAt";
+}
+
+function mapDirection(direction: string) {
+  return direction === "asc" || direction === "desc" ? direction : "desc";
+}

packages/api/src/repositories/groups.ts

@@ -0,0 +1,21 @@
+import { db } from "../db";
+
+export async function getGroups() {
+  return await db.selectFrom("groups").select(["id", "name"]).execute();
+}
+
+export async function getOrCreateGroup(name: string) {
+  let group = await db
+    .selectFrom("groups")
+    .select(["id", "name"])
+    .where("name", "=", name)
+    .executeTakeFirst();
+  if (!group) {
+    group = await db
+      .insertInto("groups")
+      .values({ name })
+      .returning(["id", "name"])
+      .executeTakeFirstOrThrow();
+  }
+  return group;
+}

packages/api/src/repositories/jobs.ts

@@ -3,14 +3,21 @@ import {
   packageQueue,
   ffmpegQueue,
   ffprobeQueue,
+  outcomeQueue,
   flowProducer,
 } from "bolt";
 import { Job as RawJob } from "bullmq";
 import { isRecordWithNumbers } from "../utils/type-guard";
 import type { JobNode, JobState, Queue } from "bullmq";
 import type { Job } from "../types";
 
-const allQueus = [transcodeQueue, packageQueue, ffmpegQueue, ffprobeQueue];
+const allQueus = [
+  transcodeQueue,
+  packageQueue,
+  ffmpegQueue,
+  ffprobeQueue,
+  outcomeQueue,
+];
 
 function findQueueByName(name: string): Queue {
   const queue = allQueus.find((queue) => queue.name === name);
@@ -152,12 +159,6 @@ async function formatJobNode(node: JobNode): Promise<Job> {
       ? job.finishedOn - processedOn
       : undefined;
 
-  let tag: string | undefined;
-  const potentialTag = job.data?.tag;
-  if (typeof potentialTag === "string") {
-    tag = potentialTag;
-  }
-
   let progress: Record<string, number> | undefined;
   if (isRecordWithNumbers(job.progress)) {
     progress = job.progress;
@@ -175,7 +176,6 @@ async function formatJobNode(node: JobNode): Promise<Job> {
     inputData: JSON.stringify(job.data),
     outputData: job.returnvalue ? JSON.stringify(job.returnvalue) : undefined,
     failedReason,
-    tag,
     children: jobChildren,
   };
 }

packages/api/src/repositories/playables.ts

@@ -0,0 +1,9 @@
+import { db } from "../db";
+import type { PlayableInsert } from "../db/types";
+
+export async function createPlayable(fields: PlayableInsert) {
+  return await db
+    .insertInto("playables")
+    .values(fields)
+    .executeTakeFirstOrThrow();
+}

packages/api/src/routes/assets.ts

@@ -0,0 +1,55 @@
+import { Elysia, t } from "elysia";
+import { authUser } from "./token";
+import { getAssets, getAssetsCount } from "../repositories/assets";
+import { getGroups } from "../repositories/groups";
+import { AssetSchema } from "../types";
+
+export const assets = new Elysia()
+  .use(authUser)
+  .get(
+    "/assets",
+    async ({ query }) => {
+      const assets = await getAssets(query);
+
+      const count = await getAssetsCount();
+      const totalPages = Math.ceil(count / query.perPage);
+
+      return {
+        page: query.page,
+        totalPages,
+        assets,
+      };
+    },
+    {
+      detail: {
+        summary: "Get all assets",
+        tags: ["Assets"],
+      },
+
+      query: t.Object({
+        page: t.Number(),
+        perPage: t.Number(),
+        orderBy: t.String(),
+        direction: t.String(),
+      }),
+      response: {
+        200: t.Object({
+          page: t.Number(),
+          totalPages: t.Number(),
+          assets: t.Array(AssetSchema),
+        }),
+      },
+    },
+  )
+  .get(
+    "/groups",
+    async () => {
+      return await getGroups();
+    },
+    {
+      detail: {
+        summary: "Get all groups",
+        tags: ["Assets"],
+      },
+    },
+  );