Refactor ffprobe to use signed url instead of piped object stream

I think this ends up being better in every way.

NodeJS doesn't have to worry about piping streams,
and the hangs that were occurring there.

Additionally, even when the previous process worked, we got a detected duration of 0.
This is some different with ffprobe.
Input streams would give this, but files/urls are detected correctly.

I upgraded the migration as well to re-detect any existing media
that has a duration of 0.

Author: CarsonF

src/components/file/file.repository.ts

@@ -154,7 +154,7 @@ export class FileRepository extends CommonRepository {
     return result!;
   }
 
-  private hydrate() {
+  hydrate() {
     return (query: Query) =>
       query
         .subQuery((sub) =>
@@ -463,13 +463,13 @@ export class FileRepository extends CommonRepository {
         }),
       )
       .apply(this.defaultPublicFromParent(input.public))
-      .return<{ id: ID }>('node.id as id');
+      .apply(this.hydrate());
 
     const result = await createFile.first();
     if (!result) {
       throw new ServerException('Failed to create file version');
     }
-    return result;
+    return result.dto as FileVersion;
   }
 
   async rename(fileNode: FileNode, newName: string): Promise<void> {

src/components/file/file.service.ts

@@ -14,7 +14,6 @@ import {
   DurationIn,
   generateId,
   ID,
-  IdOf,
   InputException,
   NotFoundException,
   ServerException,
@@ -318,7 +317,7 @@ export class FileService {
     const mimeType =
       mimeTypeOverride ?? upload?.ContentType ?? 'application/octet-stream';
 
-    await this.repo.createFileVersion(
+    const fv = await this.repo.createFileVersion(
       fileId,
       {
         id: uploadId,
@@ -357,16 +356,7 @@ export class FileService {
       }
     }
 
-    await this.mediaService.detectAndSave(
-      this.asDownloadable(
-        {
-          file: uploadId as IdOf<FileVersion>,
-          mimeType,
-          ...media,
-        },
-        uploadId,
-      ),
-    );
+    await this.mediaService.detectAndSave(fv, media);
 
     // Change the file's name to match the latest version name
     await this.rename({ id: fileId, name }, session);

src/components/file/media/detect-existing-media.migration.ts

@@ -1,13 +1,12 @@
 import { ModuleRef } from '@nestjs/core';
 import { node, relation } from 'cypher-query-builder';
-import { IdOf } from '~/common';
 import { BaseMigration, Migration } from '~/core/database';
 import { ACTIVE } from '~/core/database/query';
 import { FileVersion } from '../dto';
-import { FileService } from '../file.service';
+import { FileRepository } from '../file.repository';
 import { MediaService } from './media.service';
 
-@Migration('2023-08-24T15:00:00')
+@Migration('2023-09-01T18:00:00')
 export class DetectExistingMediaMigration extends BaseMigration {
   constructor(
     private readonly mediaService: MediaService,
@@ -17,12 +16,10 @@ export class DetectExistingMediaMigration extends BaseMigration {
   }
 
   async up() {
-    const fileService = this.moduleRef.get(FileService, { strict: false });
-    const detect = async (f: Row) => {
+    const detect = async (f: FileVersion) => {
       this.logger.info('Detecting', f);
       try {
-        const d = fileService.asDownloadable(f, f.file);
-        const result = await this.mediaService.detectAndSave(d);
+        const result = await this.mediaService.detectAndSave(f);
         this.logger.info('Detected and saved media', {
           ...f,
           ...(result ?? {}),
@@ -39,26 +36,40 @@ export class DetectExistingMediaMigration extends BaseMigration {
   }
 
   private async *grabFileVersionsToDetect(type: string) {
+    const fileRepo = this.moduleRef.get(FileRepository, { strict: false });
+
     let page = 0;
     const size = 100;
     do {
       this.logger.info(`Grabbing page of files to detect ${page}`);
 
       const currentPage = await this.db
         .query()
-        .match([
-          node('fv', 'FileVersion'),
-          relation('out', '', 'mimeType', ACTIVE),
-          node('mt', 'Property'),
-        ])
-        .raw(
-          `where not (fv)-[:media]->(:Media)
-            and mt.value starts with '${type}/'`,
+        // eslint-disable-next-line no-loop-func
+        .subQuery((sub) =>
+          sub
+            .match([
+              node('fv', 'FileVersion'),
+              relation('out', '', 'mimeType', ACTIVE),
+              node('mt', 'Property'),
+            ])
+            .optionalMatch([
+              node('fv'),
+              relation('out', '', 'media'),
+              node('media', 'Media'),
+            ])
+            .with('fv, mt, media')
+            .raw(
+              `where mt.value starts with '${type}/' and (media is null or media.duration = 0)`,
+            )
+            .return('fv')
+            .orderBy('fv.createdAt')
+            .skip(page * size)
+            .limit(size),
         )
-        .return<Row>('fv.id as file, mt.value as mimeType')
-        .orderBy('fv.createdAt')
-        .skip(page * size)
-        .limit(size)
+        .with('fv as node')
+        .apply(fileRepo.hydrate())
+        .map((row) => row.dto as FileVersion)
         .run();
 
       if (currentPage.length === 0) {
@@ -70,8 +81,3 @@ export class DetectExistingMediaMigration extends BaseMigration {
     } while (true);
   }
 }
-
-interface Row {
-  file: IdOf<FileVersion>;
-  mimeType: string;
-}

src/components/file/media/media-detector.service.ts

@@ -1,25 +1,29 @@
-import ffprobeBinary from '@ffprobe-installer/ffprobe';
-import { Injectable } from '@nestjs/common';
+import { path as ffprobeBinary } from '@ffprobe-installer/ffprobe';
+import { forwardRef, Inject, Injectable } from '@nestjs/common';
 import { execa } from 'execa';
 import { FFProbeResult } from 'ffprobe';
 import { imageSize } from 'image-size';
 import { ISize as ImageSize } from 'image-size/dist/types/interface';
-import { Readable } from 'stream';
 import { Except } from 'type-fest';
 import { retry } from '~/common/retry';
 import { ILogger, Logger } from '~/core';
-import { Downloadable } from '../dto';
+import { FileVersion } from '../dto';
+import { FileService } from '../file.service';
 import { AnyMedia, Media } from './media.dto';
 
 @Injectable()
 export class MediaDetector {
-  constructor(@Logger('media:detector') private readonly logger: ILogger) {}
+  constructor(
+    @Inject(forwardRef(() => FileService))
+    private readonly files: FileService & {},
+    @Logger('media:detector') private readonly logger: ILogger,
+  ) {}
 
   async detect(
-    file: Downloadable<{ mimeType: string }>,
+    file: FileVersion,
   ): Promise<Except<AnyMedia, Exclude<keyof Media, '__typename'>> | null> {
     if (file.mimeType.startsWith('image/')) {
-      const buffer = await file.download();
+      const buffer = await this.files.asDownloadable(file).download();
 
       let size: ImageSize = { width: 0, height: 0 };
       try {
@@ -43,9 +47,9 @@ export class MediaDetector {
       return null;
     }
 
-    const stream = await file.stream();
+    const url = await this.files.getDownloadUrl(file);
 
-    const result = await this.ffprobe(stream);
+    const result = await this.ffprobe(url);
     const { width, height, duration: rawDuration } = result.streams?.[0] ?? {};
 
     const d = rawDuration as string | number | undefined; // I've seen as string
@@ -64,35 +68,25 @@ export class MediaDetector {
     };
   }
 
-  private async ffprobe(stream: Readable): Promise<Partial<FFProbeResult>> {
+  private async ffprobe(url: string): Promise<Partial<FFProbeResult>> {
     try {
       return await retry(
         async () => {
           const probe = await execa(
-            ffprobeBinary.path,
+            ffprobeBinary,
             [
+              '-v',
+              'error',
               '-print_format',
               'json',
               '-show_format',
               '-show_streams',
-              '-i',
-              'pipe:0',
+              url,
             ],
             {
-              reject: false, // just return error instead of throwing
-              input: stream,
               timeout: 10_000,
             },
           );
-          this.logger.info('ffprobe result', probe);
-          // Ffprobe stops reading stdin & exits as soon as it has enough info.
-          // This causes the input stream to SIGPIPE error.
-          // In shells, this is normal and does not result in an error.
-          // However, NodeJS converts/interrupts this as an EPIPE error.
-          // https://github.com/sindresorhus/execa/issues/474#issuecomment-1640423498
-          if (probe instanceof Error && (probe as any).code !== 'EPIPE') {
-            throw probe;
-          }
           if (probe.stdout.trim() === '') {
             return {};
           }

src/components/file/media/media.module.ts

@@ -1,4 +1,5 @@
-import { Module } from '@nestjs/common';
+import { forwardRef, Module } from '@nestjs/common';
+import { FileModule } from '../file.module';
 import { DetectExistingMediaMigration } from './detect-existing-media.migration';
 import { DimensionsResolver } from './dimensions.resolver';
 import { MediaByFileVersionLoader } from './media-by-file-version.loader';
@@ -9,6 +10,7 @@ import { MediaResolver } from './media.resolver';
 import { MediaService } from './media.service';
 
 @Module({
+  imports: [forwardRef(() => FileModule)],
   providers: [
     DimensionsResolver,
     MediaByFileVersionLoader,

src/components/file/media/media.service.ts

@@ -1,9 +1,9 @@
 import { Injectable } from '@nestjs/common';
-import { Except, RequireAtLeastOne } from 'type-fest';
-import { NotFoundException, ServerException } from '~/common';
-import { Downloadable } from '../dto';
+import { RequireAtLeastOne } from 'type-fest';
+import { IdOf, NotFoundException, ServerException } from '~/common';
+import { FileVersion } from '../dto';
 import { MediaDetector } from './media-detector.service';
-import { AnyMedia, Media, MediaUserMetadata } from './media.dto';
+import { AnyMedia, MediaUserMetadata } from './media.dto';
 import { MediaRepository } from './media.repository';
 
 @Injectable()
@@ -13,12 +13,17 @@ export class MediaService {
     private readonly repo: MediaRepository,
   ) {}
 
-  async detectAndSave(input: Downloadable<Except<Media, 'id' | '__typename'>>) {
-    const media = await this.detector.detect(input);
+  async detectAndSave(file: FileVersion, metadata?: MediaUserMetadata) {
+    const media = await this.detector.detect(file);
     if (!media) {
       return null;
     }
-    return await this.repo.create({ ...input, ...media });
+    return await this.repo.create({
+      file: file.id as IdOf<FileVersion>,
+      mimeType: file.mimeType,
+      ...media,
+      ...metadata,
+    });
   }
 
   async updateUserMetadata(