feat: support download from OSC encore and upload to S3

README.md

@@ -29,6 +29,9 @@ package the output of the transcoding job referenced by the message.
 | `PACKAGE_CONCURRENCY`       | Number of concurrent packaging jobs                                                            | `1`                      |
 | `PACKAGE_LISTENER_PLUGIN`   | Optional path to a javascript file containing a custom listener for packaging event, see below |                          |
 | `ENCORE_PASSWORD`           | Optional password for the encore instance `user` user                                          |                          |
+| `OSC_ACCESS_TOKEN`          | Optional OSC access token for accessing Encore instance in OSC                                 |                          |
+| `AWS_ACCESS_KEY_ID`         | Optional AWS access key id when `PACKAGE_OUTPUT_FOLDER` is an AWS S3 bucket                    |                          |
+| `AWS_SECRET_ACCESS_KEY`     | Optional AWS secret access key when `PACKAGE_OUTPUT_FOLDER` is an AWS S3 bucket                |                          |
 
 ```bash
 npm run start

package.json

@@ -21,9 +21,10 @@
     "node": ">=18.15.0"
   },
   "dependencies": {
-    "@eyevinn/shaka-packager-s3": "^0.2.2",
+    "@eyevinn/shaka-packager-s3": "^0.3.0",
     "@fastify/cors": "^8.2.0",
     "@fastify/type-provider-typebox": "^2.4.0",
+    "@osaas/client-core": "^0.8.0",
     "@sinclair/typebox": "^0.25.24",
     "commander": "^12.1.0",
     "fastify": "^4.12.0",

src/config.ts

@@ -23,6 +23,7 @@ export interface PackagingConfig {
   shakaExecutable?: string;
   packageListenerPlugin?: string;
   encorePassword?: string;
+  oscAccessToken?: string;
 }
 
 function readRedisConfig(): RedisConfig {
@@ -34,11 +35,14 @@ function readRedisConfig(): RedisConfig {
 
 function readPackagingConfig(): PackagingConfig {
   return {
-    outputFolder: resolve(process.env.PACKAGE_OUTPUT_FOLDER || 'packaged'),
+    outputFolder: process.env.PACKAGE_OUTPUT_FOLDER?.match(/^s3:/)
+      ? new URL(process.env.PACKAGE_OUTPUT_FOLDER).toString()
+      : resolve(process.env.PACKAGE_OUTPUT_FOLDER || 'packaged'),
     shakaExecutable: process.env.SHAKA_PACKAGER_EXECUTABLE,
     concurrency: parseInt(process.env.PACKAGE_CONCURRENCY || '1'),
     packageListenerPlugin: process.env.PACKAGE_LISTENER_PLUGIN,
-    encorePassword: process.env.ENCORE_PASSWORD
+    encorePassword: process.env.ENCORE_PASSWORD,
+    oscAccessToken: process.env.OSC_ACCESS_TOKEN
   };
 }
 

src/encorePackager.test.ts

@@ -8,6 +8,7 @@ describe('Test parseInputsFromEncoreJob', () => {
       output: [
         {
           file: '/data/out/e5e76304-744c-41d6-85f7-69007b3b1a65/test3_x264_3100.mp4',
+          format: 'mp4',
           fileSize: 3757912,
           overallBitrate: 2982469,
           videoStreams: [
@@ -19,7 +20,8 @@ describe('Test parseInputsFromEncoreJob', () => {
           audioStreams: [],
           type: 'VideoFile'
         }
-      ]
+      ],
+      inputs: []
     };
     const inputs = parseInputsFromEncoreJob(job);
     console.log(inputs);

src/encorePackager.ts

@@ -2,6 +2,7 @@ import { Input, doPackage, PackageOptions } from '@eyevinn/shaka-packager-s3';
 import { resolve } from 'node:path';
 import { PackagingConfig } from './config';
 import { basename, extname } from 'node:path';
+import { Context } from '@osaas/client-core';
 
 export interface EncoreJob {
   id: string;
@@ -12,6 +13,7 @@ export interface EncoreJob {
 
 export interface Output {
   type: string;
+  format: string;
   file: string;
   fileSize: number;
   overallBitrate: number;
@@ -31,10 +33,19 @@ export class EncorePackager {
   async package(jobUrl: string) {
     const job = await this.getEncoreJob(jobUrl);
     const inputs = parseInputsFromEncoreJob(job);
+    let serviceAccessToken = undefined;
+    if (this.config.oscAccessToken) {
+      const ctx = new Context({
+        personalAccessToken: this.config.oscAccessToken
+      });
+      serviceAccessToken = await ctx.getServiceAccessToken('encore');
+    }
     const dest = this.getPackageDestination(job);
     await doPackage({
       dest,
       inputs,
+      source: this.config.oscAccessToken ? new URL(jobUrl).origin : undefined,
+      serviceAccessToken,
       noImplicitAudio: true,
       shakaExecutable: this.config.shakaExecutable
     } as PackageOptions);
@@ -44,7 +55,13 @@ export class EncorePackager {
   getPackageDestination(job: EncoreJob) {
     const inputUri = job.inputs[0].uri;
     const inputBasename = basename(inputUri, extname(inputUri));
-    return resolve(this.config.outputFolder, inputBasename, job.id);
+    if (this.config.outputFolder.match(/^s3:/)) {
+      return new URL(
+        this.config.outputFolder + inputBasename + '/' + job.id
+      ).toString();
+    } else {
+      return resolve(this.config.outputFolder, inputBasename, job.id);
+    }
   }
 
   async getEncoreJob(url: string): Promise<EncoreJob> {
@@ -58,8 +75,20 @@ export class EncorePackager {
             ).toString('base64')
         }
       : {};
+    let sat;
+    if (this.config.oscAccessToken) {
+      const ctx = new Context({
+        personalAccessToken: this.config.oscAccessToken
+      });
+      sat = await ctx.getServiceAccessToken('encore');
+    }
+    const jwtHeader: { 'x-jwt': string } | Record<string, never> = sat
+      ? {
+          'x-jwt': `Bearer ${sat}`
+        }
+      : {};
     const response = await fetch(url, {
-      headers: { ...authHeader }
+      headers: { ...authHeader, ...jwtHeader }
     });
     if (!response.ok) {
       throw new Error(
@@ -80,7 +109,9 @@ export function parseInputsFromEncoreJob(job: EncoreJob) {
     throw new Error('Encore job has no output');
   }
   const video = job.output
-    .filter((output) => output.type === 'VideoFile')
+    .filter(
+      (output) => output.type === 'VideoFile' && output.format.includes('mp4')
+    )
     .map((output) => ({ output, videoStream: output.videoStreams?.[0] }));
   const audio = job.output
     .filter((output) => output.type === 'AudioFile')