From 89906dba7cea8ddb10ea73189c59011d0df6df75 Mon Sep 17 00:00:00 2001 From: talksik Date: Thu, 30 Apr 2026 07:30:55 -0700 Subject: [PATCH] refactor: abstraction for transcoding --- go/cmd/particleprocessorworker/main.go | 100 ++++++----------------- go/internal/media/helpers.go | 107 +++++++++++++++++++++++++ 2 files changed, 130 insertions(+), 77 deletions(-) create mode 100644 go/internal/media/helpers.go diff --git a/go/cmd/particleprocessorworker/main.go b/go/cmd/particleprocessorworker/main.go index fe58a22..304c54c 100644 --- a/go/cmd/particleprocessorworker/main.go +++ b/go/cmd/particleprocessorworker/main.go @@ -1,19 +1,18 @@ package main import ( - "bytes" "context" "fmt" "log" "log/slog" "os" - "os/exec" "strings" "time" "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/db" "github.com/flowy-live/llink/internal/depot" + "github.com/flowy-live/llink/internal/media" "github.com/flowy-live/llink/internal/particle" "github.com/flowy-live/llink/internal/speech" "github.com/flowy-live/llink/internal/utils" @@ -106,6 +105,7 @@ func main() { slog.Debug("processing particle", "particleID", particleID, "data", change.Doc.Data()) // --- Perform side effects --- + // All of them do not stop us from marking the particle as processed updateParentLastChildCreatedAt(ctx, change.Doc) transcribeMediaParticle(ctx, depotSvc, speechSvc, change.Doc) @@ -187,7 +187,7 @@ func transcodeMediaParticle(ctx context.Context, depotSvc depot.Service, doc *fi return } if particleType != particle.TypeMedia { - slog.Info("transcode: particle is not of type media") + slog.Info("transcode: particle is not of type media") return } @@ -197,8 +197,8 @@ func transcodeMediaParticle(ctx context.Context, depotSvc depot.Service, doc *fi } // Source is already iOS-playable; nothing to do. - if isIOSPlayableMime(mediaParticle.Properties.MimeType) { - slog.Info("transcode: skipping because already playable on ios") + if media.IsIOSPlayableMime(mediaParticle.Properties.MimeType) { + slog.Info("transcode: skipping because already playable on ios") return } @@ -208,81 +208,35 @@ func transcodeMediaParticle(ctx context.Context, depotSvc depot.Service, doc *fi return } - isAudio := strings.HasPrefix(mediaParticle.Properties.MimeType, "audio/") - var outputExt, outputMime string - if isAudio { - outputExt = ".m4a" - outputMime = "audio/mp4" - } else { - outputExt = ".mp4" - outputMime = "video/mp4" - } - - tmp, err := os.CreateTemp("", "transcode-*"+outputExt) - if err != nil { - slog.Error("transcode: failed to create temp file", "error", err) - return - } - tmpPath := tmp.Name() - tmp.Close() - defer os.Remove(tmpPath) - - transcodeCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) - defer cancel() - - var args []string - if isAudio { - args = []string{ - "-y", "-i", sourceURL, - "-vn", - "-c:a", "aac", "-b:a", "128k", - "-movflags", "+faststart", - tmpPath, - } - } else { - // Cap encoder parallelism and lookahead to keep memory bounded — screen - // recordings come in at native display resolution (often 1440p–4K) and - // libx264's per-thread lookahead/reference buffers blow past the worker's - // memory limit otherwise. Output is also downscaled to 1080p max, which - // mobile playback won't notice; the original WebM stays in GCS untouched. - args = []string{ - "-y", "-i", sourceURL, - "-vf", "scale='min(1920,iw)':-2:flags=lanczos", - "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", - "-pix_fmt", "yuv420p", "-profile:v", "baseline", "-level", "3.1", - "-x264-params", "rc-lookahead=20:ref=2", - "-threads", "2", "-filter_threads", "2", - "-c:a", "aac", "-b:a", "128k", - "-movflags", "+faststart", - tmpPath, - } - } - - cmd := exec.CommandContext(transcodeCtx, "ffmpeg", args...) - var stderr bytes.Buffer - cmd.Stderr = &stderr - if err := cmd.Run(); err != nil { - slog.Error("transcode: ffmpeg failed", "error", err, "stderr", stderr.String(), "particleID", doc.Ref.ID) - return - } - networkID, err := networkIDFromParticlePath(doc.Ref.Path) if err != nil { slog.Error("transcode: failed to derive network id", "error", err, "path", doc.Ref.Path) return } - f, err := os.Open(tmpPath) + transcodeCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) + defer cancel() + transcodeOutput, err := media.TranscodeToMp4(transcodeCtx, media.TranscodeInput{ + SourceURL: sourceURL, + MimeType: mediaParticle.Properties.MimeType, + }) if err != nil { - slog.Error("transcode: failed to open transcoded file", "error", err, "path", tmpPath) + slog.Error("transcode: ffmpeg failed", "error", err) + return + } + defer os.Remove(transcodeOutput.TempLocalFilePath) + + f, err := os.Open(transcodeOutput.TempLocalFilePath) + if err != nil { + slog.Error("transcode: failed to open transcoded file", "error", err, "path", transcodeOutput.TempLocalFilePath) return } defer f.Close() newObj, err := depotSvc.CreateFromReader(ctx, depot.CreateFromReaderInput{ Prefix: networkID, - Name: "transcoded" + outputExt, - ContentType: outputMime, + Name: "transcoded" + transcodeOutput.OutputExt, + ContentType: transcodeOutput.OutputMimeType, }, f) if err != nil { slog.Error("transcode: failed to upload transcoded object", "error", err, "particleID", doc.Ref.ID) @@ -292,7 +246,7 @@ func transcodeMediaParticle(ctx context.Context, depotSvc depot.Service, doc *fi _, err = doc.Ref.Set(ctx, map[string]interface{}{ "properties": map[string]interface{}{ "transcoded_object_id": newObj.ID, - "transcoded_mime_type": outputMime, + "transcoded_mime_type": transcodeOutput.OutputMimeType, }, }, firestore.MergeAll) if err != nil { @@ -300,15 +254,7 @@ func transcodeMediaParticle(ctx context.Context, depotSvc depot.Service, doc *fi return } - slog.Info("transcoded media particle", "particleID", doc.Ref.ID, "transcoded_object_id", newObj.ID, "mime", outputMime) -} - -func isIOSPlayableMime(mime string) bool { - switch mime { - case "video/mp4", "video/quicktime", "audio/mp4", "audio/aac", "audio/x-m4a", "audio/mpeg": - return true - } - return false + slog.Info("transcoded media particle", "particleID", doc.Ref.ID, "transcoded_object_id", newObj.ID, "mime", transcodeOutput.OutputMimeType) } func toFirestoreTranscript(result *speech.TranscriptResult) particle.FirestoreTranscript { diff --git a/go/internal/media/helpers.go b/go/internal/media/helpers.go new file mode 100644 index 0000000..08eb06b --- /dev/null +++ b/go/internal/media/helpers.go @@ -0,0 +1,107 @@ +package media + +import ( + "bytes" + "context" + "errors" + "log/slog" + "os" + "os/exec" + "strings" +) + +func IsIOSPlayableMime(mime string) bool { + switch mime { + case "video/mp4", "video/quicktime", "audio/mp4", "audio/aac", "audio/x-m4a", "audio/mpeg": + return true + } + return false +} + +func IsAudio(mime string) bool { + return strings.HasPrefix(mime, "audio/") +} + +type TranscodeInput struct { + SourceURL string + MimeType string +} + +type TranscodeOutput struct { + TempLocalFilePath string + OutputMimeType string + // Extension such as ".m4a" or ".mp4" + OutputExt string +} + +var ( + ErrInvalidInput error = errors.New("invalid input") +) + +// TranscodeToMp4 takes in any audio or video source URL and +// returns the filepath of the transcoded media +// WARNING: caller responsible for deleting TempLocalFilePath +func TranscodeToMp4(ctx context.Context, input TranscodeInput) (*TranscodeOutput, error) { + if input.SourceURL == "" || input.MimeType == "" { + return nil, ErrInvalidInput + } + + isAudio := IsAudio(input.MimeType) + var outputExt, outputMime string + if isAudio { + outputExt = ".m4a" + outputMime = "audio/mp4" + } else { + outputExt = ".mp4" + outputMime = "video/mp4" + } + + tmp, err := os.CreateTemp("", "transcode-*"+outputExt) + if err != nil { + slog.Error("transcode: failed to create temp file", "error", err) + return nil, err + } + tmpPath := tmp.Name() + tmp.Close() + + var args []string + if isAudio { + args = []string{ + "-y", "-i", input.SourceURL, + "-vn", + "-c:a", "aac", "-b:a", "128k", + "-movflags", "+faststart", + tmpPath, + } + } else { + // Cap encoder parallelism and lookahead to keep memory bounded — screen + // recordings come in at native display resolution (often 1440p–4K) and + // libx264's per-thread lookahead/reference buffers blow past the worker's + // memory limit otherwise. Output is also downscaled to 1080p max, which + // mobile playback won't notice; the original WebM stays in GCS untouched. + args = []string{ + "-y", "-i", input.SourceURL, + "-vf", "scale='min(1920,iw)':-2:flags=lanczos", + "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", + "-pix_fmt", "yuv420p", "-profile:v", "baseline", "-level", "3.1", + "-x264-params", "rc-lookahead=20:ref=2", + "-threads", "2", "-filter_threads", "2", + "-c:a", "aac", "-b:a", "128k", + "-movflags", "+faststart", + tmpPath, + } + } + + cmd := exec.CommandContext(ctx, "ffmpeg", args...) + var stderr bytes.Buffer + cmd.Stderr = &stderr + if err := cmd.Run(); err != nil { + return nil, errors.Join(err, errors.New(stderr.String())) + } + + return &TranscodeOutput{ + TempLocalFilePath: tmpPath, + OutputMimeType: outputMime, + OutputExt: outputExt, + }, nil +}