diff --git a/go/Dockerfile.transcodebackfill b/go/Dockerfile.transcodebackfill new file mode 100644 index 0000000..1a546af --- /dev/null +++ b/go/Dockerfile.transcodebackfill @@ -0,0 +1,22 @@ +# golang two stage build +FROM golang:1.25 AS first-stage + +WORKDIR /app + +COPY go.mod go.sum ./ +RUN go mod download && go mod verify + +COPY . . + +WORKDIR /app/cmd/transcodebackfill +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o main +RUN ls + +FROM alpine:latest AS second-stage + +RUN apk add --no-cache ffmpeg ca-certificates + +WORKDIR /app +COPY --from=first-stage /app/cmd/transcodebackfill . +RUN echo "copied over binary to production stage" +CMD ["./main"] diff --git a/go/cmd/particleprocessorworker/main.go b/go/cmd/particleprocessorworker/main.go index 304c54c..f7f34bd 100644 --- a/go/cmd/particleprocessorworker/main.go +++ b/go/cmd/particleprocessorworker/main.go @@ -6,13 +6,11 @@ import ( "log" "log/slog" "os" - "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" @@ -109,7 +107,7 @@ func main() { updateParentLastChildCreatedAt(ctx, change.Doc) transcribeMediaParticle(ctx, depotSvc, speechSvc, change.Doc) - transcodeMediaParticle(ctx, depotSvc, change.Doc) + particle.Transcode(ctx, depotSvc, change.Doc) recordFreemiumUsage(ctx, billingSvc, change.Doc) if err := processingRepo.MarkProcessed(ctx, particleID); err != nil { @@ -165,98 +163,6 @@ func transcribeMediaParticle(ctx context.Context, depotSvc depot.Service, speech slog.Info("transcribed media particle", "particleID", doc.Ref.ID) } -// transcodeMediaParticle produces an iOS-playable MP4/m4a derivative for media -// particles whose original mime type AVPlayer can't decode (notably the WebM -// the desktop recorder emits today). Skips when the source is already in an -// iOS-playable family or when a transcoded variant has already been written. -// -// ffmpeg reads directly from the GCS signed URL and writes to a local temp -// file — `+faststart` requires seekable output, so a stdout pipe wouldn't work. -// The temp file is then streamed to GCS via depotSvc.CreateFromReader (no -// presigned-PUT round-trip — the worker has direct SDK access). -func transcodeMediaParticle(ctx context.Context, depotSvc depot.Service, doc *firestore.DocumentSnapshot) { - var mediaParticle particle.FirestoreMediaParticle - if err := doc.DataTo(&mediaParticle); err != nil { - slog.Error("transcode: unable to marshal particle data", "error", err) - return - } - - particleType, err := particle.ParseParticleType(mediaParticle.Type) - if err != nil { - slog.Error("transcode: invalid particle type", "error", err) - return - } - if particleType != particle.TypeMedia { - slog.Info("transcode: particle is not of type media") - return - } - - // Already transcoded — re-delivery within the 5-min Firestore window. - if mediaParticle.Properties.TranscodedObjectId != "" { - return - } - - // Source is already iOS-playable; nothing to do. - if media.IsIOSPlayableMime(mediaParticle.Properties.MimeType) { - slog.Info("transcode: skipping because already playable on ios") - return - } - - sourceURL, err := depotSvc.GetDownloadURL(ctx, mediaParticle.Properties.ObjectId) - if err != nil { - slog.Error("transcode: failed to get download URL", "error", err, "object_id", mediaParticle.Properties.ObjectId) - 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 - } - - 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: 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" + transcodeOutput.OutputExt, - ContentType: transcodeOutput.OutputMimeType, - }, f) - if err != nil { - slog.Error("transcode: failed to upload transcoded object", "error", err, "particleID", doc.Ref.ID) - return - } - - _, err = doc.Ref.Set(ctx, map[string]interface{}{ - "properties": map[string]interface{}{ - "transcoded_object_id": newObj.ID, - "transcoded_mime_type": transcodeOutput.OutputMimeType, - }, - }, firestore.MergeAll) - if err != nil { - slog.Error("transcode: failed to update particle in firestore", "error", err, "particleID", doc.Ref.ID) - return - } - - slog.Info("transcoded media particle", "particleID", doc.Ref.ID, "transcoded_object_id", newObj.ID, "mime", transcodeOutput.OutputMimeType) -} - func toFirestoreTranscript(result *speech.TranscriptResult) particle.FirestoreTranscript { words := make([]particle.FirestoreTranscriptWord, len(result.Words)) for i, w := range result.Words { @@ -316,7 +222,7 @@ func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *f return } - networkID, err := networkIDFromParticlePath(doc.Ref.Path) + networkID, err := particle.NetworkIDFromParticlePath(doc.Ref.Path) if err != nil { slog.Error("failed to derive network id", "error", err, "path", doc.Ref.Path) return @@ -327,23 +233,6 @@ func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *f } } -// networkIDFromParticlePath extracts the network id from a Firestore particle -// document path. Particles live at `networks/{network_id}/children/.../children/{id}` -// at arbitrary nesting depth, so the network id is always the second segment -// of the full doc path (which itself is rooted under the Firestore db path: -// `projects/.../documents/networks/{network_id}/...`). -func networkIDFromParticlePath(path string) (string, error) { - // doc.Ref.Path is the full resource path; find the "networks" collection - // and return the next segment. - segments := strings.Split(path, "/") - for i, seg := range segments { - if seg == "networks" && i+1 < len(segments) { - return segments[i+1], nil - } - } - return "", fmt.Errorf("no networks segment in path: %s", path) -} - // updateParentLastChildCreatedAt updates the parent stream's last_child_created_at // to the child's actual created_at timestamp, so it stays directly comparable with // playback markers (which also store child created_at values). diff --git a/go/cmd/transcodebackfill/main.go b/go/cmd/transcodebackfill/main.go new file mode 100644 index 0000000..258069e --- /dev/null +++ b/go/cmd/transcodebackfill/main.go @@ -0,0 +1,148 @@ +// transcodebackfill is a one-shot job that scans every doc under the +// "children" Firestore collection group, identifies media particles missing a +// transcoded variant, and re-runs the transcode + upload + Firestore-update +// flow against the configured GCS bucket. Idempotent — re-running the Job +// after a partial failure picks up where it left off via the existing +// `transcoded_object_id != ""` short-circuit inside particle.Transcode. +// +// Intended to back-fill the production iOS playback backlog created before we +// launched mobile / particle processor worker only transcodes new media. +package main + +import ( + "context" + "log/slog" + "os" + "time" + + "cloud.google.com/go/firestore" + "cloud.google.com/go/storage" + "google.golang.org/api/iterator" + + "github.com/flowy-live/llink/internal/db" + "github.com/flowy-live/llink/internal/depot" + "github.com/flowy-live/llink/internal/particle" + "github.com/flowy-live/llink/internal/utils" +) + +func main() { + ctx := context.Background() + + db.Init() + defer db.Cleanup() + + storageClient, err := storage.NewClient(ctx) + if err != nil { + slog.Error("failed to create GCS client", "error", err) + os.Exit(1) + } + defer storageClient.Close() + + depotSvc := depot.NewService(db.Pool(), storageClient, depot.Config{ + GoogleServiceAccountEmail: utils.MustGetEnv("GOOGLE_SERVICE_ACCOUNT_EMAIL"), + BucketName: utils.MustGetEnv("GCS_BUCKET"), + }) + + gcpProject := utils.MustGetEnv("GCP_PROJECT") + fs, err := firestore.NewClient(ctx, gcpProject) + if err != nil { + slog.Error("failed to create Firestore client", "error", err) + os.Exit(1) + } + defer fs.Close() + + started := time.Now() + stats, err := run(ctx, fs, depotSvc) + elapsed := time.Since(started) + + slog.Info("transcode_backfill_summary", + "scanned", stats.scanned, + "media", stats.media, + "transcoded", stats.transcoded, + "skippedAlreadyDone", stats.skippedAlreadyDone, + "skippedIOSPlayable", stats.skippedIOSPlayable, + "skippedNonMedia", stats.skippedNonMedia, + "failures", stats.failures, + "elapsed_seconds", elapsed.Seconds(), + ) + + if err != nil { + slog.Error("transcode backfill aborted", "error", err) + os.Exit(1) + } + if stats.failures > 0 { + os.Exit(1) + } +} + +type stats struct { + scanned int + media int + transcoded int + skippedAlreadyDone int + skippedIOSPlayable int + skippedNonMedia int + failures int +} + +func run(ctx context.Context, fs *firestore.Client, depotSvc depot.Service) (stats, error) { + var s stats + + it := fs.CollectionGroup("children"). + OrderBy(firestore.DocumentID, firestore.Asc). + Documents(ctx) + defer it.Stop() + + for { + doc, err := it.Next() + if err == iterator.Done { + return s, nil + } + if err != nil { + return s, err + } + + s.scanned++ + + // Cheap pre-filter: most docs under the "children" collection group + // are not media particles. DataAt avoids unmarshalling the full + // document for those. + rawType, err := doc.DataAt("type") + if err != nil { + s.skippedNonMedia++ + continue + } + typeStr, ok := rawType.(string) + if !ok || typeStr != string(particle.TypeMedia) { + s.skippedNonMedia++ + continue + } + + s.media++ + + result := particle.Transcode(ctx, depotSvc, doc) + switch { + case result.Err != nil: + s.failures++ + slog.Error("transcode_backfill_failure", + "particleID", doc.Ref.ID, + "path", doc.Ref.Path, + "error", result.Err, + ) + case result.Skipped && result.SkipReason == particle.SkipReasonAlreadyTranscoded: + s.skippedAlreadyDone++ + case result.Skipped && result.SkipReason == particle.SkipReasonIOSPlayable: + s.skippedIOSPlayable++ + case result.Skipped: + s.skippedNonMedia++ + default: + s.transcoded++ + slog.Info("transcode_backfill_progress", + "particleID", doc.Ref.ID, + "transcodedObjectID", result.TranscodedObjectID, + "outputMime", result.OutputMimeType, + "idx", s.transcoded, + ) + } + } +} diff --git a/go/internal/particle/transcode.go b/go/internal/particle/transcode.go new file mode 100644 index 0000000..fd75385 --- /dev/null +++ b/go/internal/particle/transcode.go @@ -0,0 +1,140 @@ +package particle + +import ( + "context" + "fmt" + "log/slog" + "os" + "strings" + "time" + + "cloud.google.com/go/firestore" + "github.com/flowy-live/llink/internal/depot" + "github.com/flowy-live/llink/internal/media" +) + +const ( + SkipReasonNotMedia = "not_media" + SkipReasonAlreadyTranscoded = "already_transcoded" + SkipReasonIOSPlayable = "ios_playable" +) + +// TranscodeResult reports the outcome of a single Transcode call. Skipped is +// true for any of the three idempotency short-circuits (with SkipReason set); +// Err is non-nil for real failures (download URL, ffmpeg, GCS upload, Firestore +// write). On success both flags are zero-value and TranscodedObjectID + +// OutputMimeType are populated. +type TranscodeResult struct { + Skipped bool + SkipReason string + TranscodedObjectID string + OutputMimeType string + Err error +} + +// Transcode produces an iOS-playable MP4/m4a derivative for media particles +// whose original mime type AVPlayer can't decode (notably the WebM the desktop +// recorder emits today). Skips when the particle is not of type media, when a +// transcoded variant has already been written, or when the source is already +// in an iOS-playable family. +// +// ffmpeg reads directly from the GCS signed URL and writes to a local temp +// file — `+faststart` requires seekable output, so a stdout pipe wouldn't +// work. The temp file is then streamed to GCS via depotSvc.CreateFromReader. +func Transcode(ctx context.Context, depotSvc depot.Service, doc *firestore.DocumentSnapshot) TranscodeResult { + var mp FirestoreMediaParticle + if err := doc.DataTo(&mp); err != nil { + slog.Error("transcode: unable to marshal particle data", "error", err) + return TranscodeResult{Err: fmt.Errorf("unmarshal particle: %w", err)} + } + + particleType, err := ParseParticleType(mp.Type) + if err != nil { + slog.Error("transcode: invalid particle type", "error", err) + return TranscodeResult{Err: fmt.Errorf("parse type: %w", err)} + } + if particleType != TypeMedia { + slog.Info("transcode: particle is not of type media") + return TranscodeResult{Skipped: true, SkipReason: SkipReasonNotMedia} + } + + if mp.Properties.TranscodedObjectId != "" { + return TranscodeResult{Skipped: true, SkipReason: SkipReasonAlreadyTranscoded} + } + + if media.IsIOSPlayableMime(mp.Properties.MimeType) { + slog.Info("transcode: skipping because already playable on ios") + return TranscodeResult{Skipped: true, SkipReason: SkipReasonIOSPlayable} + } + + sourceURL, err := depotSvc.GetDownloadURL(ctx, mp.Properties.ObjectId) + if err != nil { + slog.Error("transcode: failed to get download URL", "error", err, "object_id", mp.Properties.ObjectId) + return TranscodeResult{Err: fmt.Errorf("download URL: %w", err)} + } + + networkID, err := NetworkIDFromParticlePath(doc.Ref.Path) + if err != nil { + slog.Error("transcode: failed to derive network id", "error", err, "path", doc.Ref.Path) + return TranscodeResult{Err: fmt.Errorf("network id: %w", err)} + } + + transcodeCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) + defer cancel() + transcodeOutput, err := media.TranscodeToMp4(transcodeCtx, media.TranscodeInput{ + SourceURL: sourceURL, + MimeType: mp.Properties.MimeType, + }) + if err != nil { + slog.Error("transcode: ffmpeg failed", "error", err) + return TranscodeResult{Err: fmt.Errorf("ffmpeg: %w", err)} + } + 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 TranscodeResult{Err: fmt.Errorf("open temp file: %w", err)} + } + defer f.Close() + + newObj, err := depotSvc.CreateFromReader(ctx, depot.CreateFromReaderInput{ + Prefix: networkID, + 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) + return TranscodeResult{Err: fmt.Errorf("upload: %w", err)} + } + + if _, err := doc.Ref.Set(ctx, map[string]interface{}{ + "properties": map[string]interface{}{ + "transcoded_object_id": newObj.ID, + "transcoded_mime_type": transcodeOutput.OutputMimeType, + }, + }, firestore.MergeAll); err != nil { + slog.Error("transcode: failed to update particle in firestore", "error", err, "particleID", doc.Ref.ID) + return TranscodeResult{Err: fmt.Errorf("firestore update: %w", err)} + } + + slog.Info("transcoded media particle", "particleID", doc.Ref.ID, "transcoded_object_id", newObj.ID, "mime", transcodeOutput.OutputMimeType) + return TranscodeResult{ + TranscodedObjectID: newObj.ID, + OutputMimeType: transcodeOutput.OutputMimeType, + } +} + +// NetworkIDFromParticlePath extracts the network id from a Firestore particle +// document path. Particles live at `networks/{network_id}/children/.../children/{id}` +// at arbitrary nesting depth, so the network id is always the segment directly +// following the "networks" collection name in the full resource path. +func NetworkIDFromParticlePath(path string) (string, error) { + segments := strings.Split(path, "/") + for i, seg := range segments { + if seg == "networks" && i+1 < len(segments) { + return segments[i+1], nil + } + } + return "", fmt.Errorf("no networks segment in path: %s", path) +} diff --git a/go/k8s/prod/transcodebackfill.yaml b/go/k8s/prod/transcodebackfill.yaml new file mode 100644 index 0000000..ee26f0c --- /dev/null +++ b/go/k8s/prod/transcodebackfill.yaml @@ -0,0 +1,36 @@ +apiVersion: batch/v1 +kind: Job +metadata: + name: transcodebackfill +spec: + ttlSecondsAfterFinished: 86400 + backoffLimit: 0 + template: + metadata: + labels: + app: transcodebackfill + spec: + serviceAccountName: default-service-account + restartPolicy: Never + containers: + - name: transcodebackfill + image: transcodebackfill + resources: + requests: + memory: "2Gi" + cpu: 500m + limits: + memory: "2Gi" + cpu: 500m + env: + - name: "GCP_PROJECT" + value: "flowy-prod-440017" + - name: "GCS_BUCKET" + value: "flowy-llink-prod-bucket" + - name: "GOOGLE_SERVICE_ACCOUNT_EMAIL" + value: "iam-for-gke-sa@flowy-prod-440017.iam.gserviceaccount.com" + - name: "LLINK_POSTGRES_CONNECTION_URL" + valueFrom: + secretKeyRef: + name: shared-secrets + key: LLINK_POSTGRES_CONNECTION_URL diff --git a/go/skaffold.yaml b/go/skaffold.yaml index 613f22c..42cd884 100644 --- a/go/skaffold.yaml +++ b/go/skaffold.yaml @@ -195,3 +195,26 @@ profiles: - k8s/prod/pusher.yaml deploy: kubectl: {} +--- +apiVersion: skaffold/v4beta11 +kind: Config +metadata: + name: transcodebackfill +build: + local: {} + tagPolicy: + gitCommit: + variant: AbbrevCommitSha +profiles: + - name: prod + build: + artifacts: + - image: transcodebackfill + context: . + docker: + dockerfile: Dockerfile.transcodebackfill + manifests: + rawYaml: + - k8s/prod/transcodebackfill.yaml + deploy: + kubectl: {}