create backfill for transcoding media particles

Closes #192

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"]

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).

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,
+			)
+		}
+	}
+}

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)
+}

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

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: {}