Implement remote runner jobs in server

Move ffmpeg functions to @shared

7 files changed, 607 insertions, 22 deletions

diff --git a/server/models/runner/runner-job.ts b/server/models/runner/runner-job.ts
new file mode 100644
index 000000000..add6f9a43
--- /dev/null
+++ b/server/models/runner/runner-job.ts
@@ -0,0 +1,347 @@
+import { FindOptions, Op, Transaction } from 'sequelize'
+import {
+  AllowNull,
+  BelongsTo,
+  Column,
+  CreatedAt,
+  DataType,
+  Default,
+  ForeignKey,
+  IsUUID,
+  Model,
+  Scopes,
+  Table,
+  UpdatedAt
+} from 'sequelize-typescript'
+import { isUUIDValid } from '@server/helpers/custom-validators/misc'
+import { CONSTRAINTS_FIELDS, RUNNER_JOB_STATES } from '@server/initializers/constants'
+import { MRunnerJob, MRunnerJobRunner, MRunnerJobRunnerParent } from '@server/types/models/runners'
+import { RunnerJob, RunnerJobAdmin, RunnerJobPayload, RunnerJobPrivatePayload, RunnerJobState, RunnerJobType } from '@shared/models'
+import { AttributesOnly } from '@shared/typescript-utils'
+import { getSort, searchAttribute } from '../shared'
+import { RunnerModel } from './runner'
+
+enum ScopeNames {
+  WITH_RUNNER = 'WITH_RUNNER',
+  WITH_PARENT = 'WITH_PARENT'
+}
+
+@Scopes(() => ({
+  [ScopeNames.WITH_RUNNER]: {
+    include: [
+      {
+        model: RunnerModel.unscoped(),
+        required: false
+      }
+    ]
+  },
+  [ScopeNames.WITH_PARENT]: {
+    include: [
+      {
+        model: RunnerJobModel.unscoped(),
+        required: false
+      }
+    ]
+  }
+}))
+@Table({
+  tableName: 'runnerJob',
+  indexes: [
+    {
+      fields: [ 'uuid' ],
+      unique: true
+    },
+    {
+      fields: [ 'processingJobToken' ],
+      unique: true
+    },
+    {
+      fields: [ 'runnerId' ]
+    }
+  ]
+})
+export class RunnerJobModel extends Model<Partial<AttributesOnly<RunnerJobModel>>> {
+
+  @AllowNull(false)
+  @IsUUID(4)
+  @Column(DataType.UUID)
+  uuid: string
+
+  @AllowNull(false)
+  @Column
+  type: RunnerJobType
+
+  @AllowNull(false)
+  @Column(DataType.JSONB)
+  payload: RunnerJobPayload
+
+  @AllowNull(false)
+  @Column(DataType.JSONB)
+  privatePayload: RunnerJobPrivatePayload
+
+  @AllowNull(false)
+  @Column
+  state: RunnerJobState
+
+  @AllowNull(false)
+  @Default(0)
+  @Column
+  failures: number
+
+  @AllowNull(true)
+  @Column(DataType.STRING(CONSTRAINTS_FIELDS.RUNNER_JOBS.ERROR_MESSAGE.max))
+  error: string
+
+  // Less has priority
+  @AllowNull(false)
+  @Column
+  priority: number
+
+  // Used to fetch the appropriate job when the runner wants to post the result
+  @AllowNull(true)
+  @Column
+  processingJobToken: string
+
+  @AllowNull(true)
+  @Column
+  progress: number
+
+  @AllowNull(true)
+  @Column
+  startedAt: Date
+
+  @AllowNull(true)
+  @Column
+  finishedAt: Date
+
+  @CreatedAt
+  createdAt: Date
+
+  @UpdatedAt
+  updatedAt: Date
+
+  @ForeignKey(() => RunnerJobModel)
+  @Column
+  dependsOnRunnerJobId: number
+
+  @BelongsTo(() => RunnerJobModel, {
+    foreignKey: {
+      name: 'dependsOnRunnerJobId',
+      allowNull: true
+    },
+    onDelete: 'cascade'
+  })
+  DependsOnRunnerJob: RunnerJobModel
+
+  @ForeignKey(() => RunnerModel)
+  @Column
+  runnerId: number
+
+  @BelongsTo(() => RunnerModel, {
+    foreignKey: {
+      name: 'runnerId',
+      allowNull: true
+    },
+    onDelete: 'SET NULL'
+  })
+  Runner: RunnerModel
+
+  // ---------------------------------------------------------------------------
+
+  static loadWithRunner (uuid: string) {
+    const query = {
+      where: { uuid }
+    }
+
+    return RunnerJobModel.scope(ScopeNames.WITH_RUNNER).findOne<MRunnerJobRunner>(query)
+  }
+
+  static loadByRunnerAndJobTokensWithRunner (options: {
+    uuid: string
+    runnerToken: string
+    jobToken: string
+  }) {
+    const { uuid, runnerToken, jobToken } = options
+
+    const query = {
+      where: {
+        uuid,
+        processingJobToken: jobToken
+      },
+      include: {
+        model: RunnerModel.unscoped(),
+        required: true,
+        where: {
+          runnerToken
+        }
+      }
+    }
+
+    return RunnerJobModel.findOne<MRunnerJobRunner>(query)
+  }
+
+  static listAvailableJobs () {
+    const query = {
+      limit: 10,
+      order: getSort('priority'),
+      where: {
+        state: RunnerJobState.PENDING
+      }
+    }
+
+    return RunnerJobModel.findAll<MRunnerJob>(query)
+  }
+
+  static listStalledJobs (options: {
+    staleTimeMS: number
+    types: RunnerJobType[]
+  }) {
+    const before = new Date(Date.now() - options.staleTimeMS)
+
+    return RunnerJobModel.findAll<MRunnerJob>({
+      where: {
+        type: {
+          [Op.in]: options.types
+        },
+        state: RunnerJobState.PROCESSING,
+        updatedAt: {
+          [Op.lt]: before
+        }
+      }
+    })
+  }
+
+  static listChildrenOf (job: MRunnerJob, transaction?: Transaction) {
+    const query = {
+      where: {
+        dependsOnRunnerJobId: job.id
+      },
+      transaction
+    }
+
+    return RunnerJobModel.findAll<MRunnerJob>(query)
+  }
+
+  static listForApi (options: {
+    start: number
+    count: number
+    sort: string
+    search?: string
+  }) {
+    const { start, count, sort, search } = options
+
+    const query: FindOptions = {
+      offset: start,
+      limit: count,
+      order: getSort(sort)
+    }
+
+    if (search) {
+      if (isUUIDValid(search)) {
+        query.where = { uuid: search }
+      } else {
+        query.where = {
+          [Op.or]: [
+            searchAttribute(search, 'type'),
+            searchAttribute(search, '$Runner.name$')
+          ]
+        }
+      }
+    }
+
+    return Promise.all([
+      RunnerJobModel.scope([ ScopeNames.WITH_RUNNER ]).count(query),
+      RunnerJobModel.scope([ ScopeNames.WITH_RUNNER, ScopeNames.WITH_PARENT ]).findAll<MRunnerJobRunnerParent>(query)
+    ]).then(([ total, data ]) => ({ total, data }))
+  }
+
+  static updateDependantJobsOf (runnerJob: MRunnerJob) {
+    const where = {
+      dependsOnRunnerJobId: runnerJob.id
+    }
+
+    return RunnerJobModel.update({ state: RunnerJobState.PENDING }, { where })
+  }
+
+  static cancelAllJobs (options: { type: RunnerJobType }) {
+    const where = {
+      type: options.type
+    }
+
+    return RunnerJobModel.update({ state: RunnerJobState.CANCELLED }, { where })
+  }
+
+  // ---------------------------------------------------------------------------
+
+  resetToPending () {
+    this.state = RunnerJobState.PENDING
+    this.processingJobToken = null
+    this.progress = null
+    this.startedAt = null
+    this.runnerId = null
+  }
+
+  setToErrorOrCancel (
+    state: RunnerJobState.PARENT_ERRORED | RunnerJobState.ERRORED | RunnerJobState.CANCELLED | RunnerJobState.PARENT_CANCELLED
+  ) {
+    this.state = state
+    this.processingJobToken = null
+    this.finishedAt = new Date()
+  }
+
+  toFormattedJSON (this: MRunnerJobRunnerParent): RunnerJob {
+    const runner = this.Runner
+      ? {
+        id: this.Runner.id,
+        name: this.Runner.name,
+        description: this.Runner.description
+      }
+      : null
+
+    const parent = this.DependsOnRunnerJob
+      ? {
+        id: this.DependsOnRunnerJob.id,
+        uuid: this.DependsOnRunnerJob.uuid,
+        type: this.DependsOnRunnerJob.type,
+        state: {
+          id: this.DependsOnRunnerJob.state,
+          label: RUNNER_JOB_STATES[this.DependsOnRunnerJob.state]
+        }
+      }
+      : undefined
+
+    return {
+      uuid: this.uuid,
+      type: this.type,
+
+      state: {
+        id: this.state,
+        label: RUNNER_JOB_STATES[this.state]
+      },
+
+      progress: this.progress,
+      priority: this.priority,
+      failures: this.failures,
+      error: this.error,
+
+      payload: this.payload,
+
+      startedAt: this.startedAt?.toISOString(),
+      finishedAt: this.finishedAt?.toISOString(),
+
+      createdAt: this.createdAt.toISOString(),
+      updatedAt: this.updatedAt.toISOString(),
+
+      parent,
+      runner
+    }
+  }
+
+  toFormattedAdminJSON (this: MRunnerJobRunnerParent): RunnerJobAdmin {
+    return {
+      ...this.toFormattedJSON(),
+
+      privatePayload: this.privatePayload
+    }
+  }
+}
diff --git a/server/models/runner/runner-registration-token.ts b/server/models/runner/runner-registration-token.ts
new file mode 100644
index 000000000..b2ae6c9eb
--- /dev/null
+++ b/server/models/runner/runner-registration-token.ts
@@ -0,0 +1,103 @@
+import { FindOptions, literal } from 'sequelize'
+import { AllowNull, Column, CreatedAt, HasMany, Model, Table, UpdatedAt } from 'sequelize-typescript'
+import { MRunnerRegistrationToken } from '@server/types/models/runners'
+import { RunnerRegistrationToken } from '@shared/models'
+import { AttributesOnly } from '@shared/typescript-utils'
+import { getSort } from '../shared'
+import { RunnerModel } from './runner'
+
+/**
+ *
+ * Tokens used by PeerTube runners to register themselves to the PeerTube instance
+ *
+ */
+
+@Table({
+  tableName: 'runnerRegistrationToken',
+  indexes: [
+    {
+      fields: [ 'registrationToken' ],
+      unique: true
+    }
+  ]
+})
+export class RunnerRegistrationTokenModel extends Model<Partial<AttributesOnly<RunnerRegistrationTokenModel>>> {
+
+  @AllowNull(false)
+  @Column
+  registrationToken: string
+
+  @CreatedAt
+  createdAt: Date
+
+  @UpdatedAt
+  updatedAt: Date
+
+  @HasMany(() => RunnerModel, {
+    foreignKey: {
+      allowNull: true
+    },
+    onDelete: 'cascade'
+  })
+  Runners: RunnerModel[]
+
+  static load (id: number) {
+    return RunnerRegistrationTokenModel.findByPk(id)
+  }
+
+  static loadByRegistrationToken (registrationToken: string) {
+    const query = {
+      where: { registrationToken }
+    }
+
+    return RunnerRegistrationTokenModel.findOne(query)
+  }
+
+  static countTotal () {
+    return RunnerRegistrationTokenModel.unscoped().count()
+  }
+
+  static listForApi (options: {
+    start: number
+    count: number
+    sort: string
+  }) {
+    const { start, count, sort } = options
+
+    const query: FindOptions = {
+      attributes: {
+        include: [
+          [
+            literal('(SELECT COUNT(*) FROM "runner" WHERE "runner"."runnerRegistrationTokenId" = "RunnerRegistrationTokenModel"."id")'),
+            'registeredRunnersCount'
+          ]
+        ]
+      },
+      offset: start,
+      limit: count,
+      order: getSort(sort)
+    }
+
+    return Promise.all([
+      RunnerRegistrationTokenModel.count(query),
+      RunnerRegistrationTokenModel.findAll<MRunnerRegistrationToken>(query)
+    ]).then(([ total, data ]) => ({ total, data }))
+  }
+
+  // ---------------------------------------------------------------------------
+
+  toFormattedJSON (this: MRunnerRegistrationToken): RunnerRegistrationToken {
+    const registeredRunnersCount = this.get('registeredRunnersCount') as number
+
+    return {
+      id: this.id,
+
+      registrationToken: this.registrationToken,
+
+      createdAt: this.createdAt,
+      updatedAt: this.updatedAt,
+
+      registeredRunnersCount
+    }
+  }
+}
diff --git a/server/models/runner/runner.ts b/server/models/runner/runner.ts
new file mode 100644
index 000000000..1ef0018b4
--- /dev/null
+++ b/server/models/runner/runner.ts
@@ -0,0 +1,112 @@
+import { FindOptions } from 'sequelize'
+import { AllowNull, BelongsTo, Column, CreatedAt, DataType, ForeignKey, Model, Table, UpdatedAt } from 'sequelize-typescript'
+import { MRunner } from '@server/types/models/runners'
+import { Runner } from '@shared/models'
+import { AttributesOnly } from '@shared/typescript-utils'
+import { getSort } from '../shared'
+import { RunnerRegistrationTokenModel } from './runner-registration-token'
+import { CONSTRAINTS_FIELDS } from '@server/initializers/constants'
+
+@Table({
+  tableName: 'runner',
+  indexes: [
+    {
+      fields: [ 'runnerToken' ],
+      unique: true
+    },
+    {
+      fields: [ 'runnerRegistrationTokenId' ]
+    }
+  ]
+})
+export class RunnerModel extends Model<Partial<AttributesOnly<RunnerModel>>> {
+
+  // Used to identify the appropriate runner when it uses the runner REST API
+  @AllowNull(false)
+  @Column
+  runnerToken: string
+
+  @AllowNull(false)
+  @Column
+  name: string
+
+  @AllowNull(true)
+  @Column(DataType.STRING(CONSTRAINTS_FIELDS.RUNNERS.DESCRIPTION.max))
+  description: string
+
+  @AllowNull(false)
+  @Column
+  lastContact: Date
+
+  @AllowNull(false)
+  @Column
+  ip: string
+
+  @CreatedAt
+  createdAt: Date
+
+  @UpdatedAt
+  updatedAt: Date
+
+  @ForeignKey(() => RunnerRegistrationTokenModel)
+  @Column
+  runnerRegistrationTokenId: number
+
+  @BelongsTo(() => RunnerRegistrationTokenModel, {
+    foreignKey: {
+      allowNull: false
+    },
+    onDelete: 'cascade'
+  })
+  RunnerRegistrationToken: RunnerRegistrationTokenModel
+
+  // ---------------------------------------------------------------------------
+
+  static load (id: number) {
+    return RunnerModel.findByPk(id)
+  }
+
+  static loadByToken (runnerToken: string) {
+    const query = {
+      where: { runnerToken }
+    }
+
+    return RunnerModel.findOne(query)
+  }
+
+  static listForApi (options: {
+    start: number
+    count: number
+    sort: string
+  }) {
+    const { start, count, sort } = options
+
+    const query: FindOptions = {
+      offset: start,
+      limit: count,
+      order: getSort(sort)
+    }
+
+    return Promise.all([
+      RunnerModel.count(query),
+      RunnerModel.findAll<MRunner>(query)
+    ]).then(([ total, data ]) => ({ total, data }))
+  }
+
+  // ---------------------------------------------------------------------------
+
+  toFormattedJSON (this: MRunner): Runner {
+    return {
+      id: this.id,
+
+      name: this.name,
+      description: this.description,
+
+      ip: this.ip,
+      lastContact: this.lastContact,
+
+      createdAt: this.createdAt,
+      updatedAt: this.updatedAt
+    }
+  }
+}
diff --git a/server/models/shared/update.ts b/server/models/shared/update.ts
index d02c4535d..96db43730 100644
--- a/server/models/shared/update.ts
+++ b/server/models/shared/update.ts
@@ -1,22 +1,32 @@
 import { QueryTypes, Sequelize, Transaction } from 'sequelize'
 
+const updating = new Set<string>()
+
 // Sequelize always skip the update if we only update updatedAt field
-function setAsUpdated (options: {
+async function setAsUpdated (options: {
   sequelize: Sequelize
   table: string
   id: number
   transaction?: Transaction
 }) {
   const { sequelize, table, id, transaction } = options
+  const key = table + '-' + id
+
+  if (updating.has(key)) return
+  updating.add(key)
 
-  return sequelize.query(
-    `UPDATE "${table}" SET "updatedAt" = :updatedAt WHERE id = :id`,
-    {
-      replacements: { table, id, updatedAt: new Date() },
-      type: QueryTypes.UPDATE,
-      transaction
-    }
-  )
+  try {
+    await sequelize.query(
+      `UPDATE "${table}" SET "updatedAt" = :updatedAt WHERE id = :id`,
+      {
+        replacements: { table, id, updatedAt: new Date() },
+        type: QueryTypes.UPDATE,
+        transaction
+      }
+    )
+  } finally {
+    updating.delete(key)
+  }
 }
 
 export {
diff --git a/server/models/video/video-job-info.ts b/server/models/video/video-job-info.ts
index 740f6b5c6..5845b8c74 100644
--- a/server/models/video/video-job-info.ts
+++ b/server/models/video/video-job-info.ts
@@ -1,5 +1,6 @@
 import { Op, QueryTypes, Transaction } from 'sequelize'
 import { AllowNull, BelongsTo, Column, CreatedAt, Default, ForeignKey, IsInt, Model, Table, Unique, UpdatedAt } from 'sequelize-typescript'
+import { forceNumber } from '@shared/core-utils'
 import { AttributesOnly } from '@shared/typescript-utils'
 import { VideoModel } from './video'
 
@@ -59,32 +60,33 @@ export class VideoJobInfoModel extends Model<Partial<AttributesOnly<VideoJobInfo
     return VideoJobInfoModel.findOne({ where, transaction })
   }
 
-  static async increaseOrCreate (videoUUID: string, column: VideoJobInfoColumnType): Promise<number> {
+  static async increaseOrCreate (videoUUID: string, column: VideoJobInfoColumnType, amountArg = 1): Promise<number> {
     const options = { type: QueryTypes.SELECT as QueryTypes.SELECT, bind: { videoUUID } }
+    const amount = forceNumber(amountArg)
 
-    const [ { pendingMove } ] = await VideoJobInfoModel.sequelize.query<{ pendingMove: number }>(`
+    const [ result ] = await VideoJobInfoModel.sequelize.query<{ pendingMove: number }>(`
     INSERT INTO "videoJobInfo" ("videoId", "${column}", "createdAt", "updatedAt")
     SELECT
-      "video"."id" AS "videoId", 1, NOW(), NOW()
+      "video"."id" AS "videoId", ${amount}, NOW(), NOW()
     FROM
       "video"
     WHERE
       "video"."uuid" = $videoUUID
     ON CONFLICT ("videoId") DO UPDATE
     SET
-      "${column}" = "videoJobInfo"."${column}" + 1,
+      "${column}" = "videoJobInfo"."${column}" + ${amount},
       "updatedAt" = NOW()
     RETURNING
       "${column}"
     `, options)
 
-    return pendingMove
+    return result[column]
   }
 
   static async decrease (videoUUID: string, column: VideoJobInfoColumnType): Promise<number> {
     const options = { type: QueryTypes.SELECT as QueryTypes.SELECT, bind: { videoUUID } }
 
-    const result = await VideoJobInfoModel.sequelize.query<{ pendingMove: number }>(`
+    const result = await VideoJobInfoModel.sequelize.query(`
     UPDATE
       "videoJobInfo"
     SET
@@ -99,7 +101,7 @@ export class VideoJobInfoModel extends Model<Partial<AttributesOnly<VideoJobInfo
 
     if (result.length === 0) return undefined
 
-    return result[0].pendingMove
+    return result[0][column]
   }
 
   static async abortAllTasks (videoUUID: string, column: VideoJobInfoColumnType): Promise<void> {
diff --git a/server/models/video/video-live-session.ts b/server/models/video/video-live-session.ts
index dcded7872..9426f5d11 100644
--- a/server/models/video/video-live-session.ts
+++ b/server/models/video/video-live-session.ts
@@ -147,12 +147,21 @@ export class VideoLiveSessionModel extends Model<Partial<AttributesOnly<VideoLiv
     return VideoLiveSessionModel.scope(ScopeNames.WITH_REPLAY).findOne(query)
   }
 
-  static findCurrentSessionOf (videoId: number) {
+  static findCurrentSessionOf (videoUUID: string) {
     return VideoLiveSessionModel.findOne({
       where: {
-        liveVideoId: videoId,
         endDate: null
       },
+      include: [
+        {
+          model: VideoModel.unscoped(),
+          as: 'LiveVideo',
+          required: true,
+          where: {
+            uuid: videoUUID
+          }
+        }
+      ],
       order: [ [ 'startDate', 'DESC' ] ]
     })
   }
diff --git a/server/models/video/video.ts b/server/models/video/video.ts
index f817c4a33..baa8c120a 100644
--- a/server/models/video/video.ts
+++ b/server/models/video/video.ts
@@ -29,12 +29,14 @@ import { LiveManager } from '@server/lib/live/live-manager'
 import { removeHLSFileObjectStorageByFilename, removeHLSObjectStorage, removeWebTorrentObjectStorage } from '@server/lib/object-storage'
 import { tracer } from '@server/lib/opentelemetry/tracing'
 import { getHLSDirectory, getHLSRedundancyDirectory, getHlsResolutionPlaylistFilename } from '@server/lib/paths'
+import { Hooks } from '@server/lib/plugins/hooks'
 import { VideoPathManager } from '@server/lib/video-path-manager'
 import { isVideoInPrivateDirectory } from '@server/lib/video-privacy'
 import { getServerActor } from '@server/models/application/application'
 import { ModelCache } from '@server/models/shared/model-cache'
 import { buildVideoEmbedPath, buildVideoWatchPath, pick } from '@shared/core-utils'
-import { ffprobePromise, getAudioStream, hasAudioStream, uuidToShort } from '@shared/extra-utils'
+import { uuidToShort } from '@shared/extra-utils'
+import { ffprobePromise, getAudioStream, getVideoStreamDimensionsInfo, getVideoStreamFPS, hasAudioStream } from '@shared/ffmpeg'
 import {
   ResultList,
   ThumbnailType,
@@ -62,7 +64,6 @@ import {
   isVideoStateValid,
   isVideoSupportValid
 } from '../../helpers/custom-validators/videos'
-import { getVideoStreamDimensionsInfo } from '../../helpers/ffmpeg'
 import { logger } from '../../helpers/logger'
 import { CONFIG } from '../../initializers/config'
 import { ACTIVITY_PUB, API_VERSION, CONSTRAINTS_FIELDS, LAZY_STATIC_PATHS, STATIC_PATHS, WEBSERVER } from '../../initializers/constants'
@@ -137,7 +138,6 @@ import { VideoShareModel } from './video-share'
 import { VideoSourceModel } from './video-source'
 import { VideoStreamingPlaylistModel } from './video-streaming-playlist'
 import { VideoTagModel } from './video-tag'
-import { Hooks } from '@server/lib/plugins/hooks'
 
 export enum ScopeNames {
   FOR_API = 'FOR_API',
@@ -798,7 +798,7 @@ export class VideoModel extends Model<Partial<AttributesOnly<VideoModel>>> {
 
     logger.info('Stopping live of video %s after video deletion.', instance.uuid)
 
-    LiveManager.Instance.stopSessionOf(instance.id, null)
+    LiveManager.Instance.stopSessionOf(instance.uuid, null)
   }
 
   @BeforeDestroy
@@ -1763,10 +1763,12 @@ export class VideoModel extends Model<Partial<AttributesOnly<VideoModel>>> {
 
       const { audioStream } = await getAudioStream(originalFilePath, probe)
       const hasAudio = await hasAudioStream(originalFilePath, probe)
+      const fps = await getVideoStreamFPS(originalFilePath, probe)
 
       return {
         audioStream,
         hasAudio,
+        fps,
 
         ...await getVideoStreamDimensionsInfo(originalFilePath, probe)
       }