feat/ffmpeg execute ffmpeg through temporal

activities/transcode.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/bcc-code/bccm-flows/common"
+	"github.com/bcc-code/bccm-flows/services/ffmpeg"
 	"github.com/bcc-code/bccm-flows/services/transcode"
 	"go.temporal.io/sdk/activity"
 )
@@ -182,3 +183,22 @@ func TranscodeMux(ctx context.Context, input common.MuxInput) (*common.MuxResult
 	}
 	return result, nil
 }
+
+type ExecuteFFmpegInput struct {
+	Arguments []string
+}
+
+func ExecuteFFmpeg(ctx context.Context, input ExecuteFFmpegInput) error {
+	log := activity.GetLogger(ctx)
+	activity.RecordHeartbeat(ctx, "ExecuteFFmpeg")
+	log.Info("Starting ExecuteFFmpeg")
+
+	stopChan, progressCallback := registerProgressCallback(ctx)
+	defer close(stopChan)
+
+	_, err := ffmpeg.Do(input.Arguments, ffmpeg.StreamInfo{}, progressCallback)
+	if err != nil {
+		return err
+	}
+	return nil
+}

cmd/httpin/main.go

@@ -94,6 +94,17 @@ func triggerHandler(ctx *gin.Context) {
 			VXID:      vxID,
 			WithFiles: getParamFromCtx(ctx, "withFiles") == "true",
 		})
+	case "ExecuteFFmpeg":
+		var input struct {
+			Arguments []string `json:"arguments"`
+		}
+		if err = ctx.BindJSON(&input); err != nil {
+			ctx.Status(http.StatusBadRequest)
+			return
+		}
+		res, err = wfClient.ExecuteWorkflow(ctx, workflowOptions, workflows.ExecuteFFmpeg, workflows.ExecuteFFmpegInput{
+			Arguments: input.Arguments,
+		})
 	}
 
 	if err != nil {

cmd/worker/main.go

@@ -41,6 +41,7 @@ var transcodeActivities = []any{
 	activities.TranscodeToVideoH264,
 	activities.TranscodeToAudioAac,
 	activities.TranscodeMux,
+	activities.ExecuteFFmpeg,
 }
 
 var workerWorkflows = []any{
@@ -53,6 +54,7 @@ var workerWorkflows = []any{
 	workflows.MergeExportData,
 	workflows.MuxFiles,
 	workflows.PrepareFiles,
+	workflows.ExecuteFFmpeg,
 }
 
 func main() {

workflows/ffmpeg.go

@@ -0,0 +1,45 @@
+package workflows
+
+import (
+	"github.com/bcc-code/bccm-flows/activities"
+	"github.com/bcc-code/bccm-flows/utils"
+	"go.temporal.io/sdk/temporal"
+	"go.temporal.io/sdk/workflow"
+	"time"
+)
+
+type ExecuteFFmpegInput struct {
+	Arguments []string
+}
+
+func ExecuteFFmpeg(
+	ctx workflow.Context,
+	params ExecuteFFmpegInput,
+) error {
+	logger := workflow.GetLogger(ctx)
+	options := workflow.ActivityOptions{
+		RetryPolicy: &temporal.RetryPolicy{
+			InitialInterval: time.Minute * 1,
+			MaximumAttempts: 10,
+			MaximumInterval: time.Hour * 1,
+		},
+		StartToCloseTimeout:    time.Hour * 4,
+		ScheduleToCloseTimeout: time.Hour * 48,
+		HeartbeatTimeout:       time.Minute * 1,
+		TaskQueue:              utils.GetTranscodeQueue(),
+	}
+
+	ctx = workflow.WithActivityOptions(ctx, options)
+
+	logger.Info("Starting ExecuteFFmpeg")
+
+	err := workflow.ExecuteActivity(ctx, activities.ExecuteFFmpeg, activities.ExecuteFFmpegInput{
+		Arguments: params.Arguments,
+	}).Get(ctx, nil)
+
+	if err != nil {
+		return err
+	}
+
+	return err
+}