talksik
d8a45b22d3
This sets up an example of listening to firestore data and reacting. It also sets up our first worker in the kubernetes cluster.
68 lines
1.7 KiB
Go
68 lines
1.7 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
"log/slog"
|
|
|
|
"github.com/flowy-live/llink/internal/utils"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
|
|
"cloud.google.com/go/firestore"
|
|
)
|
|
|
|
func createClient(ctx context.Context) *firestore.Client {
|
|
projectId := utils.MustGetEnv("GCP_PROJECT")
|
|
|
|
client, err := firestore.NewClient(ctx, projectId)
|
|
if err != nil {
|
|
log.Fatalf("Failed to create client: %v", err)
|
|
}
|
|
return client
|
|
}
|
|
|
|
// The purpose of the particle processor worker is to listen for new particles
|
|
// across all streams and perform side effects such as
|
|
// - 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`
|
|
func main() {
|
|
ctx := context.Background()
|
|
client := createClient(ctx)
|
|
defer client.Close()
|
|
|
|
it := client.CollectionGroup("children").Snapshots(ctx)
|
|
var initialLoad = true
|
|
for {
|
|
snap, err := it.Next()
|
|
if e := status.Code(err); e == codes.DeadlineExceeded || e == codes.Canceled {
|
|
panic(fmt.Errorf("error: %w", err))
|
|
}
|
|
if err != nil {
|
|
slog.Error("error in processing snapshot", "error", err)
|
|
}
|
|
|
|
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())
|
|
}
|
|
}
|
|
}
|
|
|
|
initialLoad = false
|
|
}
|
|
}
|