feat(server): video streaming table definitions (#28147)

* video streaming table definitions

Co-authored-by: Copilot <copilot@github.com>

* update sql

* tetris

* use enum

Co-authored-by: Copilot <copilot@github.com>

* fix column name

---------

Co-authored-by: Copilot <copilot@github.com>

server/src/enum.ts

@@ -445,6 +445,12 @@ export enum VideoCodec {
 
 export const VideoCodecSchema = z.enum(VideoCodec).describe('Target video codec').meta({ id: 'VideoCodec' });
 
+export enum VideoSegmentCodec {
+  Av1 = 'av1',
+  Hevc = 'hevc',
+  H264 = 'h264',
+}
+
 export enum AudioCodec {
   Mp3 = 'mp3',
   Aac = 'aac',

server/src/queries/video.stream.repository.sql

@@ -0,0 +1,46 @@
+-- NOTE: This file is auto generated by ./sql-generator
+
+-- VideoStreamRepository.getSession
+select
+  *
+from
+  "video_stream_session"
+where
+  "id" = $1
+
+-- VideoStreamRepository.getVariant
+select
+  *
+from
+  "video_stream_variant"
+where
+  "id" = $1
+
+-- VideoStreamRepository.getSegment
+select
+  *
+from
+  "video_stream_segment"
+where
+  "variantId" = $1
+  and "index" = $2
+
+-- VideoStreamRepository.getExpiredSessions
+select
+  "id"
+from
+  "video_stream_session"
+where
+  "expiresAt" <= $1
+
+-- VideoStreamRepository.extendSession
+update "video_stream_session"
+set
+  "expiresAt" = $1
+where
+  "id" = $2
+
+-- VideoStreamRepository.deleteSession
+delete from "video_stream_session"
+where
+  "id" = $1

server/src/repositories/index.ts

@@ -46,6 +46,7 @@ import { TelemetryRepository } from 'src/repositories/telemetry.repository';
 import { TrashRepository } from 'src/repositories/trash.repository';
 import { UserRepository } from 'src/repositories/user.repository';
 import { VersionHistoryRepository } from 'src/repositories/version-history.repository';
+import { VideoStreamRepository } from 'src/repositories/video-stream.repository';
 import { ViewRepository } from 'src/repositories/view-repository';
 import { WebsocketRepository } from 'src/repositories/websocket.repository';
 import { WorkflowRepository } from 'src/repositories/workflow.repository';
@@ -100,6 +101,7 @@ export const repositories = [
   UserRepository,
   ViewRepository,
   VersionHistoryRepository,
+  VideoStreamRepository,
   WebsocketRepository,
   WorkflowRepository,
 ];

server/src/repositories/video-stream.repository.ts

@@ -0,0 +1,62 @@
+import { Injectable } from '@nestjs/common';
+import { Insertable, Kysely } from 'kysely';
+import { InjectKysely } from 'nestjs-kysely';
+import { DummyValue, GenerateSql } from 'src/decorators';
+import { DB } from 'src/schema';
+import {
+  VideoStreamSegmentTable,
+  VideoStreamSessionTable,
+  VideoStreamVariantTable,
+} from 'src/schema/tables/video-stream.table';
+
+@Injectable()
+export class VideoStreamRepository {
+  constructor(@InjectKysely() private db: Kysely<DB>) {}
+
+  createSession(session: Insertable<VideoStreamSessionTable>) {
+    return this.db.insertInto('video_stream_session').values(session).returning(['id']).executeTakeFirstOrThrow();
+  }
+
+  createVariant(variant: Insertable<VideoStreamVariantTable>) {
+    return this.db.insertInto('video_stream_variant').values(variant).returning(['id']).executeTakeFirstOrThrow();
+  }
+
+  async createSegment(segment: Insertable<VideoStreamSegmentTable>) {
+    await this.db.insertInto('video_stream_segment').values(segment).execute();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID] })
+  getSession(id: string) {
+    return this.db.selectFrom('video_stream_session').selectAll().where('id', '=', id).executeTakeFirst();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID] })
+  getVariant(id: string) {
+    return this.db.selectFrom('video_stream_variant').selectAll().where('id', '=', id).executeTakeFirst();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.NUMBER] })
+  getSegment(variantId: string, index: number) {
+    return this.db
+      .selectFrom('video_stream_segment')
+      .selectAll()
+      .where('variantId', '=', variantId)
+      .where('index', '=', index)
+      .executeTakeFirst();
+  }
+
+  @GenerateSql()
+  getExpiredSessions() {
+    return this.db.selectFrom('video_stream_session').select(['id']).where('expiresAt', '<=', new Date()).execute();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID, DummyValue.DATE] })
+  async extendSession(id: string, expiresAt: Date) {
+    await this.db.updateTable('video_stream_session').set({ expiresAt }).where('id', '=', id).execute();
+  }
+
+  @GenerateSql({ params: [DummyValue.UUID] })
+  async deleteSession(id: string) {
+    await this.db.deleteFrom('video_stream_session').where('id', '=', id).execute();
+  }
+}

server/src/schema/enums.ts

@@ -1,5 +1,12 @@
 import { registerEnum } from '@immich/sql-tools';
-import { AlbumUserRole, AssetStatus, AssetVisibility, ChecksumAlgorithm, SourceType } from 'src/enum';
+import {
+  AlbumUserRole,
+  AssetStatus,
+  AssetVisibility,
+  ChecksumAlgorithm,
+  SourceType,
+  VideoSegmentCodec,
+} from 'src/enum';
 
 export const album_user_role_enum = registerEnum({
   name: 'album_user_role_enum',
@@ -25,3 +32,8 @@ export const asset_checksum_algorithm_enum = registerEnum({
   name: 'asset_checksum_algorithm_enum',
   values: Object.values(ChecksumAlgorithm),
 });
+
+export const video_stream_variant_codec_enum = registerEnum({
+  name: 'video_stream_variant_codec_enum',
+  values: Object.values(VideoSegmentCodec),
+});

server/src/schema/index.ts

@@ -76,6 +76,11 @@ import { UserMetadataAuditTable } from 'src/schema/tables/user-metadata-audit.ta
 import { UserMetadataTable } from 'src/schema/tables/user-metadata.table';
 import { UserTable } from 'src/schema/tables/user.table';
 import { VersionHistoryTable } from 'src/schema/tables/version-history.table';
+import {
+  VideoStreamSegmentTable,
+  VideoStreamSessionTable,
+  VideoStreamVariantTable,
+} from 'src/schema/tables/video-stream.table';
 import { WorkflowActionTable, WorkflowFilterTable, WorkflowTable } from 'src/schema/tables/workflow.table';
 
 @Extensions(['uuid-ossp', 'unaccent', 'cube', 'earthdistance', 'pg_trgm', 'plpgsql'])
@@ -133,6 +138,9 @@ export class ImmichDatabase {
     UserMetadataAuditTable,
     UserTable,
     VersionHistoryTable,
+    VideoStreamSessionTable,
+    VideoStreamVariantTable,
+    VideoStreamSegmentTable,
     PluginTable,
     PluginFilterTable,
     PluginActionTable,
@@ -247,6 +255,10 @@ export interface DB {
 
   version_history: VersionHistoryTable;
 
+  video_stream_session: VideoStreamSessionTable;
+  video_stream_variant: VideoStreamVariantTable;
+  video_stream_segment: VideoStreamSegmentTable;
+
   plugin: PluginTable;
   plugin_filter: PluginFilterTable;
   plugin_action: PluginActionTable;

server/src/schema/migrations/1777415973792-AddVideoStreamTables.ts

@@ -0,0 +1,40 @@
+import { Kysely, sql } from 'kysely';
+
+export async function up(db: Kysely<any>): Promise<void> {
+  await sql`CREATE TYPE "video_stream_variant_codec_enum" AS ENUM ('av1','hevc','h264');`.execute(db);
+  await sql`CREATE TABLE "video_stream_session" (
+  "id" uuid NOT NULL DEFAULT uuid_generate_v4(),
+  "assetId" uuid NOT NULL,
+  "expiresAt" timestamp with time zone NOT NULL,
+  "createdAt" timestamp with time zone NOT NULL DEFAULT now(),
+  CONSTRAINT "video_stream_session_assetId_fkey" FOREIGN KEY ("assetId") REFERENCES "asset" ("id") ON UPDATE NO ACTION ON DELETE CASCADE,
+  CONSTRAINT "video_stream_session_pkey" PRIMARY KEY ("id")
+);`.execute(db);
+  await sql`CREATE INDEX "video_stream_session_assetId_idx" ON "video_stream_session" ("assetId");`.execute(db);
+  await sql`CREATE INDEX "video_stream_session_expiresAt_idx" ON "video_stream_session" ("expiresAt");`.execute(db);
+  await sql`CREATE TABLE "video_stream_variant" (
+  "id" uuid NOT NULL DEFAULT uuid_generate_v4(),
+  "sessionId" uuid NOT NULL,
+  "createdAt" timestamp with time zone NOT NULL DEFAULT now(),
+  "bitrate" integer NOT NULL,
+  "codec" video_stream_variant_codec_enum NOT NULL,
+  "resolution" smallint NOT NULL,
+  CONSTRAINT "video_stream_variant_sessionId_fkey" FOREIGN KEY ("sessionId") REFERENCES "video_stream_session" ("id") ON UPDATE NO ACTION ON DELETE CASCADE,
+  CONSTRAINT "video_stream_variant_pkey" PRIMARY KEY ("id")
+);`.execute(db);
+  await sql`CREATE UNIQUE INDEX "video_stream_variant_sessionId_bitrate_resolution_codec_idx" ON "video_stream_variant" ("sessionId", "bitrate", "resolution", "codec");`.execute(db);
+  await sql`CREATE TABLE "video_stream_segment" (
+  "variantId" uuid NOT NULL,
+  "index" integer NOT NULL,
+  "durationUs" integer NOT NULL,
+  CONSTRAINT "video_stream_segment_variantId_fkey" FOREIGN KEY ("variantId") REFERENCES "video_stream_variant" ("id") ON UPDATE NO ACTION ON DELETE CASCADE,
+  CONSTRAINT "video_stream_segment_pkey" PRIMARY KEY ("variantId", "index")
+);`.execute(db);
+}
+
+export async function down(db: Kysely<any>): Promise<void> {
+  await sql`DROP TABLE "video_stream_segment";`.execute(db);
+  await sql`DROP TABLE "video_stream_variant";`.execute(db);
+  await sql`DROP TABLE "video_stream_session";`.execute(db);
+  await sql`DROP TYPE "asset_checksum_algorithm_enum";`.execute(db);
+}

server/src/schema/tables/video-stream.table.ts

@@ -0,0 +1,63 @@
+import {
+  Column,
+  CreateDateColumn,
+  ForeignKeyColumn,
+  Generated,
+  Index,
+  PrimaryColumn,
+  PrimaryGeneratedColumn,
+  Table,
+  Timestamp,
+} from '@immich/sql-tools';
+import { VideoSegmentCodec } from 'src/enum';
+import { video_stream_variant_codec_enum } from 'src/schema/enums';
+import { AssetTable } from 'src/schema/tables/asset.table';
+
+@Table('video_stream_session')
+export class VideoStreamSessionTable {
+  @PrimaryGeneratedColumn()
+  id!: Generated<string>;
+
+  @ForeignKeyColumn(() => AssetTable, { onDelete: 'CASCADE' })
+  assetId!: string;
+
+  @Column({ type: 'timestamp with time zone', index: true })
+  expiresAt!: Timestamp;
+
+  @CreateDateColumn()
+  createdAt!: Generated<Timestamp>;
+}
+
+@Index({ columns: ['sessionId', 'bitrate', 'resolution', 'codec'], unique: true })
+@Table('video_stream_variant')
+export class VideoStreamVariantTable {
+  @PrimaryGeneratedColumn()
+  id!: Generated<string>;
+
+  @ForeignKeyColumn(() => VideoStreamSessionTable, { onDelete: 'CASCADE', index: false })
+  sessionId!: string;
+
+  @CreateDateColumn()
+  createdAt!: Generated<Timestamp>;
+
+  @Column({ type: 'integer' })
+  bitrate!: number;
+
+  @Column({ enum: video_stream_variant_codec_enum })
+  codec!: VideoSegmentCodec;
+
+  @Column({ type: 'smallint' })
+  resolution!: number;
+}
+
+@Table('video_stream_segment')
+export class VideoStreamSegmentTable {
+  @ForeignKeyColumn(() => VideoStreamVariantTable, { onDelete: 'CASCADE', primary: true, index: false })
+  variantId!: string;
+
+  @PrimaryColumn({ type: 'integer' })
+  index!: number;
+
+  @Column({ type: 'integer' })
+  durationUs!: number;
+}

server/src/services/base.service.ts

@@ -53,6 +53,7 @@ import { TelemetryRepository } from 'src/repositories/telemetry.repository';
 import { TrashRepository } from 'src/repositories/trash.repository';
 import { UserRepository } from 'src/repositories/user.repository';
 import { VersionHistoryRepository } from 'src/repositories/version-history.repository';
+import { VideoStreamRepository } from 'src/repositories/video-stream.repository';
 import { ViewRepository } from 'src/repositories/view-repository';
 import { WebsocketRepository } from 'src/repositories/websocket.repository';
 import { WorkflowRepository } from 'src/repositories/workflow.repository';
@@ -109,6 +110,7 @@ export const BASE_SERVICE_DEPENDENCIES = [
   TrashRepository,
   UserRepository,
   VersionHistoryRepository,
+  VideoStreamRepository,
   ViewRepository,
   WebsocketRepository,
   WorkflowRepository,
@@ -167,6 +169,7 @@ export class BaseService {
     protected trashRepository: TrashRepository,
     protected userRepository: UserRepository,
     protected versionRepository: VersionHistoryRepository,
+    protected videoStreamRepository: VideoStreamRepository,
     protected viewRepository: ViewRepository,
     protected websocketRepository: WebsocketRepository,
     protected workflowRepository: WorkflowRepository,

server/test/utils.ts

@@ -64,6 +64,7 @@ import { TelemetryRepository } from 'src/repositories/telemetry.repository';
 import { TrashRepository } from 'src/repositories/trash.repository';
 import { UserRepository } from 'src/repositories/user.repository';
 import { VersionHistoryRepository } from 'src/repositories/version-history.repository';
+import { VideoStreamRepository } from 'src/repositories/video-stream.repository';
 import { ViewRepository } from 'src/repositories/view-repository';
 import { WebsocketRepository } from 'src/repositories/websocket.repository';
 import { WorkflowRepository } from 'src/repositories/workflow.repository';
@@ -260,6 +261,7 @@ export type ServiceOverrides = {
   trash: TrashRepository;
   user: UserRepository;
   versionHistory: VersionHistoryRepository;
+  videoStream: VideoStreamRepository;
   view: ViewRepository;
   websocket: WebsocketRepository;
   workflow: WorkflowRepository;
@@ -344,6 +346,7 @@ export const getMocks = () => {
     trash: automock(TrashRepository),
     user: automock(UserRepository, { strict: false }),
     versionHistory: automock(VersionHistoryRepository),
+    videoStream: automock(VideoStreamRepository),
     view: automock(ViewRepository),
     // eslint-disable-next-line no-sparse-arrays
     websocket: automock(WebsocketRepository, { args: [, loggerMock], strict: false }),
@@ -408,6 +411,7 @@ export const newTestService = <T extends BaseService>(
     overrides.trash || (mocks.trash as As<TrashRepository>),
     overrides.user || (mocks.user as As<UserRepository>),
     overrides.versionHistory || (mocks.versionHistory as As<VersionHistoryRepository>),
+    overrides.videoStream || (mocks.videoStream as As<VideoStreamRepository>),
     overrides.view || (mocks.view as As<ViewRepository>),
     overrides.websocket || (mocks.websocket as As<WebsocketRepository>),
     overrides.workflow || (mocks.workflow as As<WorkflowRepository>),