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,