FlowyLabs/llink
1
0
Fork 0
Code Issues 71 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

infra: add logging wrapping

Browse Source 
Resolves issues with gcp cloud logging quirks such as field names
This commit is contained in:
Arjun Patel
2026-05-27 15:49:27 -07:00
parent e804cee229
commit 563c91e7d5
30 changed files with 347 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/cmd/particleprocessorworker/main.go
+37 -35
View File
@@ -4,11 +4,15 @@ import (
"context"
"fmt"
"log"
"log/slog"
"os"
"strings"
"time"
"github.com/flowy-live/llink/internal/utils/flog"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"github.com/flowy-live/llink/internal/billing"
"github.com/flowy-live/llink/internal/db"
"github.com/flowy-live/llink/internal/depot"
@@ -18,8 +22,6 @@ import (
"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"
@@ -48,7 +50,7 @@ func main() {
storageClient, err := storage.NewClient(ctx)
if err != nil {
slog.Error("failed to create GCS client", "error", err)
flog.Error("failed to create GCS client", "error", err)
os.Exit(1)
}
defer storageClient.Close()
@@ -83,7 +85,7 @@ func main() {
panic(fmt.Errorf("error: %w", err))
}
if err != nil {
slog.Error("error in processing snapshot", "error", err)
flog.Error("error in processing snapshot", "error", err)
continue
}
@@ -100,15 +102,15 @@ func main() {
processed, err := processingRepo.IsProcessed(ctx, particleID)
if err != nil {
slog.Error("failed to check processing status", "particleID", particleID, "error", err)
flog.Error("failed to check processing status", "particleID", particleID, "error", err)
continue
}
if processed {
slog.Debug("skipping already processed particle", "particleID", particleID)
flog.Debug("skipping already processed particle", "particleID", particleID)
continue
}
slog.Debug("processing particle", "particleID", particleID, "data", change.Doc.Data())
flog.Debug("processing particle", "particleID", particleID, "data", change.Doc.Data())
// Side effects below are best-effort — failures don't prevent
// marking the particle as processed.
@@ -120,7 +122,7 @@ func main() {
notifyForParticle(ctx, notifier, humanSvc, change.Doc, parentDoc, transcript)
if err := processingRepo.MarkProcessed(ctx, particleID); err != nil {
slog.Error("failed to mark particle as processed", "particleID", particleID, "error", err)
flog.Error("failed to mark particle as processed", "particleID", particleID, "error", err)
}
}
}
@@ -132,30 +134,30 @@ func transcribeMediaParticle(ctx context.Context, depotSvc depot.Service, speech
var mediaParticle particle.FirestoreMediaParticle
err := doc.DataTo(&mediaParticle)
if err != nil {
slog.Error("unable to marshal particle data", "error", err)
flog.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)
flog.Error("invalid particle type", "error", err)
return ""
}
if particleType != particle.TypeMedia {
slog.Info("received a particle of type", "particle type", particleType)
flog.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)
flog.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)
flog.Error("failed to transcribe media", "error", err, "particleID", doc.Ref.ID)
return ""
}
@@ -167,11 +169,11 @@ func transcribeMediaParticle(ctx context.Context, depotSvc depot.Service, speech
},
}, firestore.MergeAll)
if err != nil {
slog.Error("failed to update transcript in firestore", "error", err, "particleID", doc.Ref.ID)
flog.Error("failed to update transcript in firestore", "error", err, "particleID", doc.Ref.ID)
return ""
}
slog.Info("transcribed media particle", "particleID", doc.Ref.ID)
flog.Info("transcribed media particle", "particleID", doc.Ref.ID)
return transcript.Transcript
}
@@ -215,17 +217,17 @@ func toFirestoreTranscript(result *speech.TranscriptResult) particle.FirestoreTr
func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *firestore.DocumentSnapshot) {
rawType, err := doc.DataAt("type")
if err != nil {
slog.Error("failed to read particle type", "error", err, "particleID", doc.Ref.ID)
flog.Error("failed to read particle type", "error", err, "particleID", doc.Ref.ID)
return
}
typeStr, ok := rawType.(string)
if !ok {
slog.Error("particle type is not a string", "particleID", doc.Ref.ID, "type", rawType)
flog.Error("particle type is not a string", "particleID", doc.Ref.ID, "type", rawType)
return
}
particleType, err := particle.ParseParticleType(typeStr)
if err != nil {
slog.Error("invalid particle type", "error", err, "particleID", doc.Ref.ID)
flog.Error("invalid particle type", "error", err, "particleID", doc.Ref.ID)
return
}
// Containers don't count toward the daily message cap.
@@ -235,12 +237,12 @@ func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *f
networkID, err := particle.NetworkIDFromParticlePath(doc.Ref.Path)
if err != nil {
slog.Error("failed to derive network id", "error", err, "path", doc.Ref.Path)
flog.Error("failed to derive network id", "error", err, "path", doc.Ref.Path)
return
}
if err := billingSvc.IncrementDailyUsage(ctx, networkID, doc.CreateTime); err != nil {
slog.Error("failed to increment daily usage", "error", err, "networkID", networkID, "particleID", doc.Ref.ID)
flog.Error("failed to increment daily usage", "error", err, "networkID", networkID, "particleID", doc.Ref.ID)
}
}
@@ -252,12 +254,12 @@ func loadParentParticle(ctx context.Context, doc *firestore.DocumentSnapshot) *f
}
parentParticleDocRef := parentChildrenCollectionRef.Parent
if parentParticleDocRef == nil {
slog.Error("particle has no parent document", "particleID", doc.Ref.ID)
flog.Error("particle has no parent document", "particleID", doc.Ref.ID)
return nil
}
parentParticleDoc, err := parentParticleDocRef.Get(ctx)
if err != nil {
slog.Error("failed to get parent particle", "error", err, "particleID", doc.Ref.ID)
flog.Error("failed to get parent particle", "error", err, "particleID", doc.Ref.ID)
return nil
}
return parentParticleDoc
@@ -272,13 +274,13 @@ func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.Document
var streamParticle particle.FirestoreStreamParticle
if err := parent.DataTo(&streamParticle); err != nil {
slog.Error("failed to parse stream particle", "error", err)
flog.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)
flog.Error("invalid particle type", "error", err)
return
}
@@ -288,7 +290,7 @@ func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.Document
childCreatedAt, err := doc.DataAt("created_at")
if err != nil {
slog.Error("failed to read child created_at", "error", err, "particleID", doc.Ref.ID)
flog.Error("failed to read child created_at", "error", err, "particleID", doc.Ref.ID)
return
}
@@ -299,7 +301,7 @@ func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.Document
},
})
if err != nil {
slog.Error("unable to update parent particle `last_child_created_at`", "error", err)
flog.Error("unable to update parent particle `last_child_created_at`", "error", err)
}
}
@@ -315,7 +317,7 @@ func notifyForParticle(
transcript string,
) {
if parent == nil {
slog.Info("notify: skip — no parent", "particleID", doc.Ref.ID)
flog.Info("notify: skip — no parent", "particleID", doc.Ref.ID)
return
}
@@ -323,31 +325,31 @@ func notifyForParticle(
typeName, _ := typeStr.(string)
pType, err := particle.ParseParticleType(typeName)
if err != nil {
slog.Info("notify: skip — unparseable particle type",
flog.Info("notify: skip — unparseable particle type",
"particleID", doc.Ref.ID, "type", typeName, "error", err)
return
}
if pType == particle.TypeStream || pType == particle.TypeFolder {
slog.Info("notify: skip — container particle",
flog.Info("notify: skip — container particle",
"particleID", doc.Ref.ID, "type", pType)
return
}
var parentStream particle.FirestoreStreamParticle
if err := parent.DataTo(&parentStream); err != nil {
slog.Error("notify: failed to parse parent stream", "error", err)
flog.Error("notify: failed to parse parent stream", "error", err)
return
}
parentType, err := particle.ParseParticleType(parentStream.Type)
if err != nil || parentType != particle.TypeStream {
slog.Info("notify: skip — parent isn't a stream",
flog.Info("notify: skip — parent isn't a stream",
"particleID", doc.Ref.ID, "parentType", parentType, "parseErr", err)
return
}
networkID, err := particle.NetworkIDFromParticlePath(doc.Ref.Path)
if err != nil {
slog.Error("notify: failed to derive network id", "error", err, "path", doc.Ref.Path)
flog.Error("notify: failed to derive network id", "error", err, "path", doc.Ref.Path)
return
}
@@ -370,7 +372,7 @@ func notifyForParticle(
if sender, err := humanSvc.GetByID(ctx, senderHumanID); err == nil {
senderEmailPrefix = sender.EmailPrefix
} else {
slog.Warn("notify: failed to look up sender", "error", err, "humanID", senderHumanID)
flog.Warn("notify: failed to look up sender", "error", err, "humanID", senderHumanID)
}
}
@@ -385,7 +387,7 @@ func notifyForParticle(
StreamVisibleTo: parentStream.VisibleTo,
Body: previewForParticle(pType, doc, transcript),
}); err != nil {
slog.Error("notify: dispatch failed", "error", err, "particleID", doc.Ref.ID, "networkID", networkID)
flog.Error("notify: dispatch failed", "error", err, "particleID", doc.Ref.ID, "networkID", networkID)
}
}