149 lines
3.8 KiB
Go
149 lines
3.8 KiB
Go
// 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,
|
|
)
|
|
}
|
|
}
|
|
}
|