feat: setup listener on input bucket

src/api.ts

@@ -7,6 +7,7 @@ import { Static, Type } from '@sinclair/typebox';
 import { FastifyPluginCallback } from 'fastify';
 import { Context } from '@osaas/client-core';
 import { apiPipeline } from './api_pipeline';
+import { onFileOnInput } from './orchestrator/events';
 
 const HelloWorld = Type.String({
   description: 'The magical words!'
@@ -67,7 +68,7 @@ export default (opts: ApiOptions) => {
 
   api.register(healthcheck, { title: opts.title });
   // register other API routes here
-  api.register(apiPipeline, { name: 'tvplus', ctx: opts.ctx });
+  api.register(apiPipeline, { name: 'tvplus', ctx: opts.ctx, onFileOnInput });
 
   return api;
 };

src/api_pipeline.ts

@@ -2,10 +2,12 @@ import { Context } from '@osaas/client-core';
 import { createVodPipeline, removeVodPipeline } from '@osaas/client-transcode';
 import { Type } from '@sinclair/typebox';
 import { FastifyPluginCallback } from 'fastify';
+import { setupListener } from './storage/listener';
 
 export interface PipelineOptions {
   name: string;
   ctx: Context;
+  onFileOnInput: (r: any, pipeline: any) => Promise<void>;
 }
 
 export const apiPipeline: FastifyPluginCallback<PipelineOptions> = (
@@ -26,9 +28,21 @@ export const apiPipeline: FastifyPluginCallback<PipelineOptions> = (
     },
     async (request, reply) => {
       try {
-        await createVodPipeline(opts.name, opts.ctx, {
+        const pipeline = await createVodPipeline(opts.name, opts.ctx, {
           createInputBucket: true
         });
+        if (pipeline.inputStorage) {
+          setupListener(
+            pipeline.inputStorage.name,
+            new URL(pipeline.inputStorage.endpoint),
+            pipeline.inputStorage.accessKeyId,
+            pipeline.inputStorage.secretAccessKey,
+            pipeline,
+            opts.onFileOnInput
+          );
+        } else {
+          throw new Error('Pipeline has no input storage');
+        }
         reply.status(201).send({ message: 'Pipeline created' });
       } catch (err: any) {
         reply.status(500).send({ message: err.message });

src/orchestrator/events.ts

@@ -0,0 +1,3 @@
+export async function onFileOnInput(rec: any, pipeline: any) {
+  console.log('Received file on input', rec);
+}

src/storage/listener.ts

@@ -0,0 +1,28 @@
+import * as Minio from 'minio';
+
+export async function setupListener(
+  bucketName: string,
+  endpoint: URL,
+  accessKeyId: string,
+  secretAccessKey: string,
+  pipeline: any,
+  onNotification: (r: any, pipeline: any) => Promise<void>
+) {
+  const client = new Minio.Client({
+    endPoint: endpoint.hostname,
+    accessKey: accessKeyId,
+    secretKey: secretAccessKey,
+    useSSL: endpoint.protocol === 'https:'
+  });
+  const poller = client.listenBucketNotification(bucketName, '', '.mp4', [
+    's3:ObjectCreated:*'
+  ]);
+  if (!poller) {
+    console.error('Failed to setup listener for bucket notifications');
+  }
+  console.log('Listening for notifications');
+  poller.on('notification', async (record) => {
+    await onNotification(record, pipeline);
+    poller.stop();
+  });
+}