Mux Video Migration tool

functions.ts

import { inngest } from './client';
/*
1. Submit migration job details include video source platform, video destination platform, authentication details, and import settings, and get back a job id (provided by Inngest?)
2. Inngest starts the migration by listing all videos in the source platform
3. A request is then made to each individual video to get that specific video’s details, including a signed URL to access that video’s master file
4. A request is then made to PUT that video over to the destination platform either by providing the signed URL directly to the platform or by pushing chunks of the video to the destination platform as a multipart file upload
5. The status of the video upload/ingest is monitored via multipart upload progress or by listening for a webhook from the destination platform indicating success or failure
6. Each video’s status is then reported back to the client based off of the initial job ID that the client received in step one
7. The job is considered complete when all videos have been processed
*/

// #1 - This is what your POST API handler would look like to kick off the job
async function POST(request) {
  // NOTE - You'd typically get all of the info in the event payload from the request body ;)
  const { ids } = await inngest.send({
    name: 'app/migration.created',
    data: {
      sourcePlatform: 'vimeo',
      destinationPlatform: 'mux',
      settings: {
        authentication: {
          //...
        },
      },
    },
  });
  const eventId = ids[0];
  // TODO - Save the eventId to a database to fetch job info later via REST API
  // Here we just return the event ID for the client for the convenience of the example
  return {
    status: 200,
    body: {
      eventId,
      message: `Migration job started!`,
    },
  }
}

// #2
const runMigration = inngest.createFunction(
  { id: 'run-migration', name: 'Run migration' },
  { event: 'app/migration.initiated' },
  async ({ event, step }) => {
    // #3
    const videos = await step.run('get-all-videos', async () => {
      // Use the source platform's API to get all videos
      // NOTE - If this might be lots of paginated API calls,
      // this might be better broken into multiple steps or
      // possibly a separate function using step.invoke
      return await fetchVideosFromSourcePlatform(
        event.data.sourcePlatform,
        event.data.settings.authentication
      );
    });

    // #4
    const videoCopyJobs = videos.map((video) => {
      return step.invoke(`copy-video-${video.id}`, {
        function: copyVideo,
        data: {
          video,
          destinationPlatform: event.data.destinationPlatform,
          settings: event.data.settings,
        },
      });
    });

    // Wait for all videos to be copied this will resolve (or reject)
    // when all the videoCopyJobs are done
    await Promise.all(videoCopyJobs)

    // NOTE - Could also push something via Websockets to the client here as well

    return { message: 'migration complete', videosMigrated: videos.length };
  }
);

// We decouple the logic of the copy video function from the run migration function
// to allow this to be re-used and tested independently.
// This also allows the system to define different configuration for each part,
// like limits on concurrency for how many videos should be copied in parallel
// A separate function could be created for each destination platform
const copyVideo = inngest.createFunction(
  { id: 'copy-video', name: 'Copy video', concurrency: 10 },
  { event: 'app/copy.initiated' },
  async ({ event, step }) => {

    let destinationVideoURL = null;

    // Check if the destination platform supports passing a signed URL
    if (doesPlatformSupportSignedURLs(event.data.destinationPlatform)) {
      destinationVideoURL = await step.run('copy-video-via-signed-url', async () => {
        // This is the business logic that copies the video
        const newVideo = await copyVideoViaSignedUrl(event.data.video, event.data.destinationPlatform, event.data.settings.authentication);
        return newVideo.url;
      })
    } else {
      // If the platform doesn't support signed URLs, we need to download and then upload the video

      // NOTE - This step may take a while. Typically, you would want to break this into multiple step.run calls,
      // but in serverless, there is no guarantee that the same instance will be used for each step.run call,
      // so the file may be gone. If this doesn't run on serverless you could download the video in one step and
      // upload in parts in others.
      //
      // ALTERNATE - Dave's idea about a step that breaks the video into chunks and then a series of steps that handle each chunk!
      destinationVideoURL = const await step.run('copy-video-via-file', async () => {
        // Download file
        const filePath = downloadVideo(event.data.video, event.data.destinationPlatform, event.data.settings.authentication);

        // Upload file, multi-part
        const newVideo = await uploadVideo(event.data.video, event.data.destinationPlatform, event.data.settings.authentication);
        return newVideo.url;
      });
    }


    // #6 - Option A - Could add pushing updates to the browser with something like Websockets

    return { status: 'success', videoId: event.data.video.id, destinationVideoURL };
  }
);

// #5 - TBD - More complex depending on the webhooks

// #6 - Option B - Use the Inngest REST API to fetch the job status by Event ID (stored in step 1)
// Docs: https://api-docs.inngest.com/docs/inngest-api/yoyeen3mu7wj0-list-event-function-runs

// Use the Dev Server URL for local testing
const inngestAPI = process.env.NODE_ENV === 'production' ? 'https://api.inngest.com' : 'http://localhost:8288';

async function GET(request) {
  // NOTE - Need to ensure this user should have access to this event
  const eventID = request.params.eventID;

  const res = await fetch(`${inngestAPI}/v1/events/${eventID}/runs`, {
    headers: {
      'Content-Type': 'application/json',
      Authorization: 'Bearer ' + process.env.INNGEST_SIGNING_KEY,
    },
  });
  const body = await res.json();
  /** Example response:
    {
      data: [
        {
          run_id: '01HQ93SBR10951E86XAADX85YB',
          run_started_at: '2024-02-22T19:13:28.833Z',
          function_id: '1e839882-7e98-4a43-b5c2-6f9f00f0b91c',
          function_version: 8,
          environment_id: '899fcf3b-575f-4524-9bf0-ec28ca434fdb',
          event_id: '01HQ93SBHT3TPHSG6YSGY8ZMM6',
          status: 'Completed',
          ended_at: '2024-02-22T19:13:30.176Z',
          output: {
            message: 'success',
            videosMigrated: 34
          }
        }
      ],
      metadata: {
        fetched_at: '2024-02-22T20:00:23.28775122Z',
        cached_until: '2024-02-22T20:00:28.28775131Z'
      }
    }
   */

  // We assume here that there will only be one run triggered for this event
  const run = body.data[0];

  // Return the response to the browser - NOTE - pseudo code
  return {
    status: 200,
    body: {
      migrationStatus: run.state,
      message: `Migrated ${run.output.videosMigrated} videos!`,
    },
  }
}