From 986a3896069814c347ad174dcc36873983673de5 Mon Sep 17 00:00:00 2001 From: talksik Date: Wed, 25 Mar 2026 14:43:47 -0700 Subject: [PATCH] feat: generate transcript and event-driven particle processing This generates the transcript and shows the caption experience on the client side for media particles. It also simplifies other side effects that we must perform such as updating the `last_child_created_at` field for stream and container particles. --- go/cmd/particleprocessorworker/main.go | 211 ++++++++++++++++-- go/go.mod | 9 + go/go.sum | 21 ++ go/internal/particle/firestore_types.go | 53 +++++ go/internal/particle/processing_repository.go | 37 +++ go/internal/speech/service.go | 99 ++++++++ go/k8s/dev/orion.yaml | 7 +- go/k8s/dev/particleprocessorworker.yaml | 11 + go/k8s/migrations.yaml | 5 +- .../000009_processed_particles.down.sql | 1 + .../000009_processed_particles.up.sql | 4 + js/src/api/types.ts | 26 +++ .../particles/media-particle-view.tsx | 37 ++- .../features/particles/transcript-overlay.tsx | 53 +++++ js/src/hooks/use-create-particle.ts | 8 +- js/src/hooks/use-transcript-playback.ts | 40 ++++ js/src/lib/firestore-particles.ts | 10 - 17 files changed, 579 insertions(+), 53 deletions(-) create mode 100644 go/internal/particle/firestore_types.go create mode 100644 go/internal/particle/processing_repository.go create mode 100644 go/internal/speech/service.go create mode 100644 go/migrations/000009_processed_particles.down.sql create mode 100644 go/migrations/000009_processed_particles.up.sql create mode 100644 js/src/features/particles/transcript-overlay.tsx create mode 100644 js/src/hooks/use-transcript-playback.ts diff --git a/go/cmd/particleprocessorworker/main.go b/go/cmd/particleprocessorworker/main.go index c0474dc..03cc79e 100644 --- a/go/cmd/particleprocessorworker/main.go +++ b/go/cmd/particleprocessorworker/main.go @@ -5,12 +5,19 @@ import ( "fmt" "log" "log/slog" + "os" + "time" + "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/speech" "github.com/flowy-live/llink/internal/utils" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "cloud.google.com/go/firestore" + "cloud.google.com/go/storage" ) func createClient(ctx context.Context) *firestore.Client { @@ -28,13 +35,38 @@ func createClient(ctx context.Context) *firestore.Client { // - generate transcript if the particle is of type media // - send mobile notifications if a client is offline // - update the parent stream's `last_child_created_at` +// - generate vector embedding +// - synthesize and decide whether ai should generate a particle as a response func main() { ctx := context.Background() + + db.Init() + defer db.Cleanup() + processingRepo := particle.NewProcessingRepository(db.Pool()) + + storageClient, err := storage.NewClient(ctx) + if err != nil { + slog.Error("failed to create GCS client", "error", err) + os.Exit(1) + } + defer storageClient.Close() + + gcsBucket := utils.MustGetEnv("GCS_BUCKET") + depotSvc := depot.NewService(db.Pool(), storageClient, depot.Config{ + GoogleServiceAccountEmail: utils.MustGetEnv("GOOGLE_SERVICE_ACCOUNT_EMAIL"), + BucketName: gcsBucket, + }) + + speechSvc := speech.NewSpeechService(ctx) + client := createClient(ctx) defer client.Close() - it := client.CollectionGroup("children").Snapshots(ctx) - var initialLoad = true + cutoff := time.Now().Add(-5 * time.Minute) + it := client.CollectionGroup("children"). + Where("created_at", ">", cutoff). + Snapshots(ctx) + for { snap, err := it.Next() if e := status.Code(err); e == codes.DeadlineExceeded || e == codes.Canceled { @@ -42,26 +74,167 @@ func main() { } if err != nil { slog.Error("error in processing snapshot", "error", err) + continue } - if snap != nil { - if initialLoad { - slog.Info("initial load", "changeCount", len(snap.Changes)) - } else { - for _, change := range snap.Changes { - switch change.Kind { - case firestore.DocumentAdded: - slog.Info("document added: ") - case firestore.DocumentModified: - slog.Info("document modified") - case firestore.DocumentRemoved: - slog.Info("document removed") - } - slog.Info("received document snapshot", "data", change.Doc.Data()) - } + if snap == nil { + continue + } + + for _, change := range snap.Changes { + if change.Kind != firestore.DocumentAdded { + continue + } + + particleID := change.Doc.Ref.ID + + processed, err := processingRepo.IsProcessed(ctx, particleID) + if err != nil { + slog.Error("failed to check processing status", "particleID", particleID, "error", err) + continue + } + if processed { + slog.Debug("skipping already processed particle", "particleID", particleID) + continue + } + + slog.Debug("processing particle", "particleID", particleID, "data", change.Doc.Data()) + + // --- Perform side effects --- + + updateParentLastChildCreatedAt(ctx, change.Doc.Ref) + transcribeMediaParticle(ctx, depotSvc, speechSvc, change.Doc) + + if err := processingRepo.MarkProcessed(ctx, particleID); err != nil { + slog.Error("failed to mark particle as processed", "particleID", particleID, "error", err) } } - - initialLoad = false + } +} + +func transcribeMediaParticle(ctx context.Context, depotSvc depot.Service, speechSvc speech.SpeechService, doc *firestore.DocumentSnapshot) { + var mediaParticle particle.FirestoreMediaParticle + err := doc.DataTo(&mediaParticle) + if err != nil { + slog.Error("unable to marshal particle data", "error", err) + return + } + + particleType, err := particle.ParseParticleType(mediaParticle.Type) + if err != nil { + slog.Error("invalid particle type", "error", err) + return + } + + if particleType != particle.TypeMedia { + slog.Info("received a particle of type", "particle type", particleType) + return + } + + downloadURL, err := depotSvc.GetDownloadURL(ctx, mediaParticle.Properties.ObjectId) + if err != nil { + slog.Error("failed to get download URL", "error", err, "object_id", mediaParticle.Properties.ObjectId) + return + } + + result, err := speechSvc.Transcribe(ctx, downloadURL) + if err != nil { + slog.Error("failed to transcribe media", "error", err, "particleID", doc.Ref.ID) + return + } + + transcript := toFirestoreTranscript(result) + + _, err = doc.Ref.Set(ctx, map[string]interface{}{ + "properties": map[string]interface{}{ + "transcript": transcript, + }, + }, firestore.MergeAll) + if err != nil { + slog.Error("failed to update transcript in firestore", "error", err, "particleID", doc.Ref.ID) + return + } + + slog.Info("transcribed media particle", "particleID", doc.Ref.ID) +} + +func toFirestoreTranscript(result *speech.TranscriptResult) particle.FirestoreTranscript { + words := make([]particle.FirestoreTranscriptWord, len(result.Words)) + for i, w := range result.Words { + words[i] = particle.FirestoreTranscriptWord{ + Word: w.Word, + Start: w.Start, + End: w.End, + } + } + + paragraphs := make([]particle.FirestoreTranscriptParagraph, len(result.Paragraphs)) + for i, p := range result.Paragraphs { + sentences := make([]particle.FirestoreTranscriptSentence, len(p.Sentences)) + for j, s := range p.Sentences { + sentences[j] = particle.FirestoreTranscriptSentence{ + Text: s.Text, + Start: s.Start, + End: s.End, + } + } + paragraphs[i] = particle.FirestoreTranscriptParagraph{ + Sentences: sentences, + Start: p.Start, + End: p.End, + } + } + + return particle.FirestoreTranscript{ + Transcript: result.Transcript, + Words: words, + Paragraphs: paragraphs, + } +} + +// updateParentLastChildCreatedAt updates the parent particle's field only if it's a particle of type stream +func updateParentLastChildCreatedAt(ctx context.Context, docRef *firestore.DocumentRef) { + parentChildrenCollectionRef := docRef.Parent + if parentChildrenCollectionRef == nil { + return + } + + parentParticleDocRef := parentChildrenCollectionRef.Parent + if parentParticleDocRef == nil { + slog.Error("particle has no parent document", "particleID", docRef.ID) + return + } + + parentParticleDoc, err := parentParticleDocRef.Get(ctx) + if err != nil { + slog.Error("failed to get parent particle", "error", err) + return + } + + slog.Info("parent particle is", "parent particle id", parentParticleDoc.Ref.ID) + + var streamParticle particle.FirestoreStreamParticle + if err := parentParticleDoc.DataTo(&streamParticle); err != nil { + slog.Error("failed to parse stream particle", "error", err) + return + } + + particleType, err := particle.ParseParticleType(streamParticle.Type) + if err != nil { + slog.Error("invalid particle type", "error", err) + return + } + + if particleType == particle.TypeStream { + slog.Info("going to update the last_child_created_at for parent particle") + _, err = parentParticleDocRef.Update(ctx, []firestore.Update{ + { + Path: "last_child_created_at", + Value: firestore.ServerTimestamp, + }, + }) + if err != nil { + slog.Error("unable to update parent particle `last_child_created_at`") + } } } diff --git a/go/go.mod b/go/go.mod index bc5890b..81fefdc 100644 --- a/go/go.mod +++ b/go/go.mod @@ -42,14 +42,18 @@ require ( github.com/containerd/platforms v0.2.1 // indirect github.com/cpuguy83/dockercfg v0.3.2 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/deepgram/deepgram-go-sdk v1.9.0 // indirect + github.com/deepgram/deepgram-go-sdk/v3 v3.5.0 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/distribution/reference v0.6.0 // indirect github.com/docker/docker v28.5.1+incompatible // indirect github.com/docker/go-connections v0.6.0 // indirect github.com/docker/go-units v0.5.0 // indirect + github.com/dvonthenen/websocket v1.5.1-dyv.2 // indirect github.com/ebitengine/purego v0.8.4 // indirect github.com/envoyproxy/go-control-plane/envoy v1.35.0 // indirect github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect + github.com/fatih/color v1.15.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/go-jose/go-jose/v4 v4.1.3 // indirect github.com/go-logr/logr v1.4.3 // indirect @@ -59,7 +63,9 @@ require ( github.com/google/s2a-go v0.1.9 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.7 // indirect github.com/googleapis/gax-go/v2 v2.15.0 // indirect + github.com/gorilla/schema v1.3.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.4 // indirect + github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect @@ -67,6 +73,8 @@ require ( github.com/lib/pq v1.10.9 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect + github.com/mattn/go-colorable v0.1.13 // indirect + github.com/mattn/go-isatty v0.0.17 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/go-archive v0.1.0 // indirect github.com/moby/patternmatcher v0.6.0 // indirect @@ -107,4 +115,5 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20251222181119-0a764e51fe1b // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20251222181119-0a764e51fe1b // indirect gopkg.in/yaml.v3 v3.0.1 // indirect + k8s.io/klog/v2 v2.110.1 // indirect ) diff --git a/go/go.sum b/go/go.sum index 50b6c40..0a47f91 100644 --- a/go/go.sum +++ b/go/go.sum @@ -64,6 +64,10 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/deepgram/deepgram-go-sdk v1.9.0 h1:FlJ1iJ//+Cz0goWGWU/Ms2jjM/wMBqyNAk+5bcAsptU= +github.com/deepgram/deepgram-go-sdk v1.9.0/go.mod h1:il+6HLmvxa47EG12LG6VwzaHcyI8Lo+yfBsOcDq3R8s= +github.com/deepgram/deepgram-go-sdk/v3 v3.5.0 h1:ug48j1DVNRKrkXti18/aFT3NP5HV2Q2CN3QMwTvHmy4= +github.com/deepgram/deepgram-go-sdk/v3 v3.5.0/go.mod h1:wVr0PDvlJFWVLUmf65u+K80SJVf/PUWvkFFubGPW/As= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/dhui/dktest v0.4.6 h1:+DPKyScKSEp3VLtbMDHcUq6V5Lm5zfZZVb0Sk7Ahom4= @@ -76,6 +80,8 @@ github.com/docker/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pM github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/dvonthenen/websocket v1.5.1-dyv.2 h1:OXlWJJkeHt8k4+MEI0Y8SQjY2ihHYD2z/tI7sZZfsnA= +github.com/dvonthenen/websocket v1.5.1-dyv.2/go.mod h1:q2GbopbpFJvBP4iqVvqwwahVmvu2HnCfdqCWDoQVKMM= github.com/ebitengine/purego v0.8.4 h1:CF7LEKg5FFOsASUj0+QwaXf8Ht6TlFxg09+S9wz0omw= github.com/ebitengine/purego v0.8.4/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= github.com/envoyproxy/go-control-plane v0.13.5-0.20251024222203-75eaa193e329 h1:K+fnvUM0VZ7ZFJf0n4L/BRlnsb9pL/GuDG6FqaH+PwM= @@ -86,11 +92,14 @@ github.com/envoyproxy/go-control-plane/ratelimit v0.1.0 h1:/G9QYbddjL25KvtKTv3an github.com/envoyproxy/go-control-plane/ratelimit v0.1.0/go.mod h1:Wk+tMFAFbCXaJPzVVHnPgRKdUdwW/KdbRt94AzgRee4= github.com/envoyproxy/protoc-gen-validate v1.2.1 h1:DEo3O99U8j4hBFwbJfrz9VtgcDfUKS7KJ7spH3d86P8= github.com/envoyproxy/protoc-gen-validate v1.2.1/go.mod h1:d/C80l/jxXLdfEIhX1W2TmLfsJ31lvEjwamM4DxlWXU= +github.com/fatih/color v1.15.0 h1:kOqh6YHBtK8aywxGerMG2Eq3H6Qgoqeo13Bk2Mv/nBs= +github.com/fatih/color v1.15.0/go.mod h1:0h5ZqXfHYED7Bhv2ZJamyIOUej9KtShiJESRwBDUSsw= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/go-jose/go-jose/v4 v4.1.3 h1:CVLmWDhDVRa6Mi/IgCgaopNosCaHz7zrMeF9MlZRkrs= github.com/go-jose/go-jose/v4 v4.1.3/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= @@ -116,8 +125,12 @@ github.com/googleapis/enterprise-certificate-proxy v0.3.7 h1:zrn2Ee/nWmHulBx5sAV github.com/googleapis/enterprise-certificate-proxy v0.3.7/go.mod h1:MkHOF77EYAE7qfSuSS9PU6g4Nt4e11cnsDUowfwewLA= github.com/googleapis/gax-go/v2 v2.15.0 h1:SyjDc1mGgZU5LncH8gimWo9lW1DtIfPibOG81vgd/bo= github.com/googleapis/gax-go/v2 v2.15.0/go.mod h1:zVVkkxAQHa1RQpg9z2AUCMnKhi0Qld9rcmyfL1OZhoc= +github.com/gorilla/schema v1.3.0 h1:rbciOzXAx3IB8stEFnfTwO3sYa6EWlQk79XdyustPDA= +github.com/gorilla/schema v1.3.0/go.mod h1:Dg5SSm5PV60mhF2NFaTV1xuYYj8tV8NOPRo4FggUMnM= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.4 h1:kEISI/Gx67NzH3nJxAmY/dGac80kKZgZt134u7Y/k1s= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.4/go.mod h1:6Nz966r3vQYCqIzWsuEl9d7cf7mRhtDmm++sOxlnfxI= +github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f h1:7LYC+Yfkj3CTRcShK0KOL/w6iTiKyqqBA9a41Wnggw8= +github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f/go.mod h1:pFlLw2CfqZiIBOx6BuCeRLCrfxBJipTY0nIOF/VbGcI= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= @@ -138,6 +151,11 @@ github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= +github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= +github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= +github.com/mattn/go-isatty v0.0.17 h1:BTarxUcIeDqL27Mc+vyvdWYSL28zpIhv3RoTdsLMPng= +github.com/mattn/go-isatty v0.0.17/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= github.com/mdelapenya/tlscert v0.2.0 h1:7H81W6Z/4weDvZBNOfQte5GpIMo0lGYEeWbkGp5LJHI= github.com/mdelapenya/tlscert v0.2.0/go.mod h1:O4njj3ELLnJjGdkN7M/vIVCpZ+Cf0L6muqOG4tLSl8o= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= @@ -238,6 +256,7 @@ golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= @@ -273,3 +292,5 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA= +k8s.io/klog/v2 v2.110.1 h1:U/Af64HJf7FcwMcXyKm2RPM22WZzyR7OSpYj5tg3cL0= +k8s.io/klog/v2 v2.110.1/go.mod h1:YGtd1984u+GgbuZ7e08/yBuAfKLSO0+uR1Fhi6ExXjo= diff --git a/go/internal/particle/firestore_types.go b/go/internal/particle/firestore_types.go new file mode 100644 index 0000000..dbae226 --- /dev/null +++ b/go/internal/particle/firestore_types.go @@ -0,0 +1,53 @@ +package particle + +import "time" + +type FirestoreMediaParticle struct { + CreatedByHumanId string `firestore:"created_by_human_id"` + Type string `firestore:"type"` + Properties FirestoreMediaParticleProperties `firestore:"properties"` + CreatedAt time.Time `firestore:"created_at,serverTimestamp"` + UpdatedAt *time.Time `firestore:"updated_at,omitempty"` +} + +type FirestoreTranscriptWord struct { + Word string `firestore:"word"` + Start float64 `firestore:"start"` + End float64 `firestore:"end"` +} + +type FirestoreTranscriptSentence struct { + Text string `firestore:"text"` + Start float64 `firestore:"start"` + End float64 `firestore:"end"` +} + +type FirestoreTranscriptParagraph struct { + Sentences []FirestoreTranscriptSentence `firestore:"sentences"` + Start float64 `firestore:"start"` + End float64 `firestore:"end"` +} + +type FirestoreTranscript struct { + Transcript string `firestore:"transcript"` + Words []FirestoreTranscriptWord `firestore:"words"` + Paragraphs []FirestoreTranscriptParagraph `firestore:"paragraphs"` +} + +type FirestoreMediaParticleProperties struct { + ObjectId string `firestore:"object_id"` + MimeType string `firestore:"mime_type"` + DurationMs int `firestore:"duration_ms"` + SizeBytes int `firestore:"size_bytes"` + Transcript *FirestoreTranscript `firestore:"transcript,omitempty"` +} + +type FirestoreStreamParticle struct { + CreatedByHumanId string `firestore:"created_by_human_id"` + Type string `firestore:"type"` + // Properties FirestoreStreamParticleProperties `firestore:"properties"` + CreatedAt time.Time `firestore:"created_at,serverTimestamp"` + LastChildCreatedAt *time.Time `firestore:"last_child_created_at,omitempty"` + VisibleTo []string `firestore:"visible_to"` + UpdatedAt *time.Time `firestore:"updated_at,omitempty"` +} diff --git a/go/internal/particle/processing_repository.go b/go/internal/particle/processing_repository.go new file mode 100644 index 0000000..dc6c3f5 --- /dev/null +++ b/go/internal/particle/processing_repository.go @@ -0,0 +1,37 @@ +package particle + +import ( + "context" + + "github.com/jackc/pgx/v5/pgxpool" +) + +type ProcessingRepository interface { + IsProcessed(ctx context.Context, particleID string) (bool, error) + MarkProcessed(ctx context.Context, particleID string) error +} + +type processingRepositoryImpl struct { + pool *pgxpool.Pool +} + +func NewProcessingRepository(pool *pgxpool.Pool) ProcessingRepository { + return &processingRepositoryImpl{pool: pool} +} + +func (r *processingRepositoryImpl) IsProcessed(ctx context.Context, particleID string) (bool, error) { + var exists bool + err := r.pool.QueryRow(ctx, + `SELECT EXISTS(SELECT 1 FROM processed_particles WHERE particle_id = $1)`, + particleID, + ).Scan(&exists) + return exists, err +} + +func (r *processingRepositoryImpl) MarkProcessed(ctx context.Context, particleID string) error { + _, err := r.pool.Exec(ctx, + `INSERT INTO processed_particles (particle_id) VALUES ($1) ON CONFLICT (particle_id) DO NOTHING`, + particleID, + ) + return err +} diff --git a/go/internal/speech/service.go b/go/internal/speech/service.go new file mode 100644 index 0000000..d14ea66 --- /dev/null +++ b/go/internal/speech/service.go @@ -0,0 +1,99 @@ +package speech + +import ( + "context" + "log/slog" + + dgapi "github.com/deepgram/deepgram-go-sdk/v3/pkg/api/listen/v1/rest" + interfaces "github.com/deepgram/deepgram-go-sdk/v3/pkg/client/interfaces" + client "github.com/deepgram/deepgram-go-sdk/v3/pkg/client/listen" +) + +type TranscriptWord struct { + Word string + Start float64 + End float64 +} + +type TranscriptSentence struct { + Text string + Start float64 + End float64 +} + +type TranscriptParagraph struct { + Sentences []TranscriptSentence + Start float64 + End float64 +} + +type TranscriptResult struct { + Transcript string + Words []TranscriptWord + Paragraphs []TranscriptParagraph +} + +type SpeechService interface { + Transcribe(ctx context.Context, mediaUrl string) (*TranscriptResult, error) +} + +type speechServiceImpl struct { + deepgramClient *dgapi.Client +} + +func NewSpeechService(ctx context.Context) SpeechService { + return &speechServiceImpl{ + deepgramClient: dgapi.New(client.NewRESTWithDefaults()), + } +} + +func (s *speechServiceImpl) Transcribe(ctx context.Context, mediaUrl string) (*TranscriptResult, error) { + options := &interfaces.PreRecordedTranscriptionOptions{ + Model: "nova-3", + SmartFormat: true, + Paragraphs: true, + } + + response, err := s.deepgramClient.FromURL(ctx, mediaUrl, options) + if err != nil { + slog.Error("failed to transcribe prerecorded media", "error", err) + return nil, err + } + + alt := response.Results.Channels[0].Alternatives[0] + + words := make([]TranscriptWord, len(alt.Words)) + for i, w := range alt.Words { + words[i] = TranscriptWord{ + Word: w.PunctuatedWord, + Start: w.Start, + End: w.End, + } + } + + var paragraphs []TranscriptParagraph + if alt.Paragraphs != nil { + paragraphs = make([]TranscriptParagraph, len(alt.Paragraphs.Paragraphs)) + for i, p := range alt.Paragraphs.Paragraphs { + sentences := make([]TranscriptSentence, len(p.Sentences)) + for j, s := range p.Sentences { + sentences[j] = TranscriptSentence{ + Text: s.Text, + Start: s.Start, + End: s.End, + } + } + paragraphs[i] = TranscriptParagraph{ + Sentences: sentences, + Start: p.Start, + End: p.End, + } + } + } + + return &TranscriptResult{ + Transcript: alt.Transcript, + Words: words, + Paragraphs: paragraphs, + }, nil +} diff --git a/go/k8s/dev/orion.yaml b/go/k8s/dev/orion.yaml index b1bcabf..a5b28ab 100644 --- a/go/k8s/dev/orion.yaml +++ b/go/k8s/dev/orion.yaml @@ -33,15 +33,10 @@ spec: value: "8080" - name: "AERO_ADDR" value: "aero:50051" - - name: "IAM_FOR_GKE_SERVICE_ACCOUNT" - value: "iam-for-gke-sa@flowy-dev-440017.iam.gserviceaccount.com" - name: "GCS_BUCKET" value: "flowy-llink-bucket" - name: "LLINK_POSTGRES_CONNECTION_URL" - valueFrom: - secretKeyRef: - name: shared-secrets - key: LLINK_POSTGRES_CONNECTION_URL + value: "postgresql://neondb_owner:npg_ysiLkYo3Ez9G@ep-aged-feather-amfusyhe-pooler.c-5.us-east-1.aws.neon.tech/neondb?sslmode=require&channel_binding=require" - name: "DEEPGRAM_SECRET" valueFrom: secretKeyRef: diff --git a/go/k8s/dev/particleprocessorworker.yaml b/go/k8s/dev/particleprocessorworker.yaml index 641cdcf..1da47f3 100644 --- a/go/k8s/dev/particleprocessorworker.yaml +++ b/go/k8s/dev/particleprocessorworker.yaml @@ -29,3 +29,14 @@ spec: env: - name: "GCP_PROJECT" value: "flowy-dev-440017" + - name: "GCS_BUCKET" + value: "flowy-llink-bucket" + - name: "GOOGLE_SERVICE_ACCOUNT_EMAIL" + value: "iam-for-gke-sa@flowy-dev-440017.iam.gserviceaccount.com" + - name: "LLINK_POSTGRES_CONNECTION_URL" + value: "postgresql://neondb_owner:npg_ysiLkYo3Ez9G@ep-aged-feather-amfusyhe-pooler.c-5.us-east-1.aws.neon.tech/neondb?sslmode=require&channel_binding=require" + - name: "DEEPGRAM_API_KEY" + valueFrom: + secretKeyRef: + name: shared-secrets + key: DEEPGRAM_SECRET diff --git a/go/k8s/migrations.yaml b/go/k8s/migrations.yaml index 0a67a4b..b490a58 100644 --- a/go/k8s/migrations.yaml +++ b/go/k8s/migrations.yaml @@ -12,9 +12,6 @@ spec: args: ["-path", "/migrations", "-database", "$(LLINK_POSTGRES_CONNECTION_URL)", "up"] env: - name: "LLINK_POSTGRES_CONNECTION_URL" - valueFrom: - secretKeyRef: - name: shared-secrets - key: LLINK_POSTGRES_CONNECTION_URL + value: "postgresql://neondb_owner:npg_ysiLkYo3Ez9G@ep-aged-feather-amfusyhe-pooler.c-5.us-east-1.aws.neon.tech/neondb?sslmode=require&channel_binding=require" restartPolicy: Never backoffLimit: 0 diff --git a/go/migrations/000009_processed_particles.down.sql b/go/migrations/000009_processed_particles.down.sql new file mode 100644 index 0000000..0db4ab7 --- /dev/null +++ b/go/migrations/000009_processed_particles.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS processed_particles; diff --git a/go/migrations/000009_processed_particles.up.sql b/go/migrations/000009_processed_particles.up.sql new file mode 100644 index 0000000..aed9657 --- /dev/null +++ b/go/migrations/000009_processed_particles.up.sql @@ -0,0 +1,4 @@ +CREATE TABLE processed_particles ( + particle_id TEXT PRIMARY KEY, + processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/js/src/api/types.ts b/js/src/api/types.ts index 7eaf334..748d9a4 100644 --- a/js/src/api/types.ts +++ b/js/src/api/types.ts @@ -91,11 +91,37 @@ export const FolderPropertiesSchema = z.object({ }); export type FolderProperties = z.infer; +const TranscriptWordSchema = z.object({ + word: z.string(), + start: z.number(), + end: z.number(), +}); + +const TranscriptSentenceSchema = z.object({ + text: z.string(), + start: z.number(), + end: z.number(), +}); + +const TranscriptParagraphSchema = z.object({ + sentences: z.array(TranscriptSentenceSchema), + start: z.number(), + end: z.number(), +}); + +export const TranscriptSchema = z.object({ + transcript: z.string(), + words: z.array(TranscriptWordSchema), + paragraphs: z.array(TranscriptParagraphSchema), +}); +export type Transcript = z.infer; + export const MediaPropertiesSchema = z.object({ object_id: z.string(), mime_type: z.string(), duration_ms: z.number(), size_bytes: z.number(), + transcript: TranscriptSchema.optional(), }); export type MediaProperties = z.infer; diff --git a/js/src/features/particles/media-particle-view.tsx b/js/src/features/particles/media-particle-view.tsx index b2cfd1d..1193261 100644 --- a/js/src/features/particles/media-particle-view.tsx +++ b/js/src/features/particles/media-particle-view.tsx @@ -1,6 +1,8 @@ import { useEffect, useRef, useState } from "react"; import type { Particle } from "@/api/types"; import { useDownloadUrl } from "@/hooks/use-download-url"; +import { useTranscriptPlayback } from "@/hooks/use-transcript-playback"; +import { TranscriptOverlay } from "@/features/particles/transcript-overlay"; import { Skeleton } from "@/components/ui/skeleton"; import { AudioLevelBars } from "@/components/audio/audio-level-bars"; import { useAudioSource } from "@/components/audio/use-audio-source"; @@ -25,6 +27,13 @@ export function MediaParticleView({ const videoRef = useRef(null); const audioRef = useRef(null); const isAudio = particle.properties.mime_type?.startsWith("audio/"); + const [currentTime, setCurrentTime] = useState(0); + + const transcript = particle.properties.transcript; + const { activeParagraph, activeWordIndex } = useTranscriptPlayback( + transcript, + currentTime, + ); // WORKAROUND: useAudioSource needs an audio element, but audioRef is only set after mount const [audioEl, setAudioEl] = useState(null); @@ -55,6 +64,20 @@ export function MediaParticleView({ return ; } + const handleTimeUpdate = (e: React.SyntheticEvent) => { + const { currentTime: time, duration } = e.currentTarget; + setCurrentTime(time); + if (duration > 0) onProgress?.(time / duration); + }; + + const captionOverlay = transcript ? ( + + ) : null; + if (isAudio) { return (
@@ -67,10 +90,7 @@ export function MediaParticleView({ src={url} autoPlay onEnded={onEnded} - onTimeUpdate={(e) => { - const { currentTime, duration } = e.currentTarget; - if (duration > 0) onProgress?.(currentTime / duration); - }} + onTimeUpdate={handleTimeUpdate} /> {audioSource && ( @@ -78,6 +98,8 @@ export function MediaParticleView({
)} + + {captionOverlay} ); } @@ -90,12 +112,11 @@ export function MediaParticleView({ autoPlay playsInline onEnded={onEnded} - onTimeUpdate={(e) => { - const { currentTime, duration } = e.currentTarget; - if (duration > 0) onProgress?.(currentTime / duration); - }} + onTimeUpdate={handleTimeUpdate} className="h-full w-full object-cover" /> + + {captionOverlay} ); } diff --git a/js/src/features/particles/transcript-overlay.tsx b/js/src/features/particles/transcript-overlay.tsx new file mode 100644 index 0000000..e1cefd2 --- /dev/null +++ b/js/src/features/particles/transcript-overlay.tsx @@ -0,0 +1,53 @@ +import { useMemo } from "react"; +import type { Transcript } from "@/api/types"; + +interface TranscriptOverlayProps { + transcript: Transcript; + activeParagraph: Transcript["paragraphs"][number] | null; + activeWordIndex: number | null; +} + +export function TranscriptOverlay({ + transcript, + activeParagraph, + activeWordIndex, +}: TranscriptOverlayProps) { + // Find the words that belong to the active paragraph by time range + const paragraphWords = useMemo(() => { + if (!activeParagraph) return []; + return transcript.words.filter( + (w) => w.start >= activeParagraph.start && w.end <= activeParagraph.end, + ); + }, [transcript.words, activeParagraph]); + + if (!activeParagraph || paragraphWords.length === 0) return null; + + // The active word from the flat array — find it by index to compare + const activeWord = + activeWordIndex !== null ? transcript.words[activeWordIndex] : null; + + return ( +
+

+ {paragraphWords.map((word, i) => { + const isSpoken = + activeWord !== null && word.start <= activeWord.end; + + return ( + + {i > 0 ? " " : ""} + {word.word} + + ); + })} +

+
+ ); +} diff --git a/js/src/hooks/use-create-particle.ts b/js/src/hooks/use-create-particle.ts index a108450..263e18e 100644 --- a/js/src/hooks/use-create-particle.ts +++ b/js/src/hooks/use-create-particle.ts @@ -1,5 +1,5 @@ import { useMutation } from "@tanstack/react-query"; -import { createParticle, updateStreamLastChildAt } from "@/lib/firestore-particles"; +import { createParticle } from "@/lib/firestore-particles"; import type { ParticleType, ParticlePropertiesMap } from "@/api/types"; import { particlePath, ParticlePath, toFirestoreChildrenPath, toFirestoreDocPath } from "@/lib/particle-path"; @@ -15,16 +15,12 @@ export function useCreateParticle() { return useMutation({ mutationFn: async (params: CreateParticleParams) => { const collectionPath = toFirestoreChildrenPath(params.path); - const result = await createParticle( + return await createParticle( collectionPath, params.type, params.properties, params.createdByHumanId, ); - - const streamDocPath = toFirestoreDocPath(params.path); - await updateStreamLastChildAt(streamDocPath); - return result; } }); } diff --git a/js/src/hooks/use-transcript-playback.ts b/js/src/hooks/use-transcript-playback.ts new file mode 100644 index 0000000..33ae8fd --- /dev/null +++ b/js/src/hooks/use-transcript-playback.ts @@ -0,0 +1,40 @@ +import { useMemo } from "react"; +import type { Transcript } from "@/api/types"; + +interface TranscriptPlaybackState { + /** The paragraph currently being spoken, or null if before/after speech */ + activeParagraph: Transcript["paragraphs"][number] | null; + /** Index of the active word within the transcript's flat words array */ + activeWordIndex: number | null; +} + +export function useTranscriptPlayback( + transcript: Transcript | undefined, + currentTime: number, +): TranscriptPlaybackState { + return useMemo(() => { + if (!transcript) return { activeParagraph: null, activeWordIndex: null }; + + const activeParagraph = + transcript.paragraphs.find( + (p) => currentTime >= p.start && currentTime <= p.end, + ) ?? null; + + // Binary search for active word + const words = transcript.words; + let activeWordIndex: number | null = null; + let lo = 0; + let hi = words.length - 1; + while (lo <= hi) { + const mid = (lo + hi) >>> 1; + if (currentTime < words[mid].start) hi = mid - 1; + else if (currentTime > words[mid].end) lo = mid + 1; + else { + activeWordIndex = mid; + break; + } + } + + return { activeParagraph, activeWordIndex }; + }, [transcript, currentTime]); +} diff --git a/js/src/lib/firestore-particles.ts b/js/src/lib/firestore-particles.ts index 9414c4d..831fc8f 100644 --- a/js/src/lib/firestore-particles.ts +++ b/js/src/lib/firestore-particles.ts @@ -263,16 +263,6 @@ export async function updateParticle( }); } -export async function updateStreamLastChildAt( - docPath: string, -): Promise { - const particleRef = typedDoc(docPath); - await updateDoc(particleRef, { - last_child_created_at: serverTimestamp(), - updated_at: serverTimestamp(), - }); -} - export async function updateStreamPlaybackMarker( docPath: string, humanId: string,