diff --git a/go/Dockerfile.particleprocessorworker b/go/Dockerfile.particleprocessorworker new file mode 100644 index 0000000..f68a2ac --- /dev/null +++ b/go/Dockerfile.particleprocessorworker @@ -0,0 +1,20 @@ +# golang two stage build +FROM golang:1.25 AS first-stage + +WORKDIR /app + +COPY go.mod go.sum ./ +RUN go mod download && go mod verify + +COPY . . + +WORKDIR /app/cmd/particleprocessorworker +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o main +RUN ls + +FROM alpine:latest AS second-stage + +WORKDIR /app +COPY --from=first-stage /app/cmd/particleprocessorworker . +RUN echo "copied over binary to production stage" +CMD ["./main"] diff --git a/go/cmd/particleprocessorworker/main.go b/go/cmd/particleprocessorworker/main.go new file mode 100644 index 0000000..c0474dc --- /dev/null +++ b/go/cmd/particleprocessorworker/main.go @@ -0,0 +1,67 @@ +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 + } +} diff --git a/go/go.mod b/go/go.mod index 07d9c46..bc5890b 100644 --- a/go/go.mod +++ b/go/go.mod @@ -23,7 +23,9 @@ require ( cloud.google.com/go/auth v0.17.0 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect cloud.google.com/go/compute/metadata v0.9.0 // indirect + cloud.google.com/go/firestore v1.21.0 // indirect cloud.google.com/go/iam v1.5.3 // indirect + cloud.google.com/go/longrunning v0.7.0 // indirect cloud.google.com/go/monitoring v1.24.2 // indirect dario.cat/mergo v1.0.2 // indirect github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect diff --git a/go/go.sum b/go/go.sum index a97c5d0..50b6c40 100644 --- a/go/go.sum +++ b/go/go.sum @@ -8,6 +8,8 @@ cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIi cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs= cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10= +cloud.google.com/go/firestore v1.21.0 h1:BhopUsx7kh6NFx77ccRsHhrtkbJUmDAxNY3uapWdjcM= +cloud.google.com/go/firestore v1.21.0/go.mod h1:1xH6HNcnkf/gGyR8udd6pFO4Z7GWJSwLKQMx/u6UrP4= cloud.google.com/go/iam v1.5.3 h1:+vMINPiDF2ognBJ97ABAYYwRgsaqxPbQDlMnbHMjolc= cloud.google.com/go/iam v1.5.3/go.mod h1:MR3v9oLkZCTlaqljW6Eb2d3HGDGK5/bDv93jhfISFvU= cloud.google.com/go/logging v1.13.0 h1:7j0HgAp0B94o1YRDqiqm26w4q1rDMH7XNRU34lJXHYc= diff --git a/go/k8s/dev/particleprocessorworker.yaml b/go/k8s/dev/particleprocessorworker.yaml new file mode 100644 index 0000000..641cdcf --- /dev/null +++ b/go/k8s/dev/particleprocessorworker.yaml @@ -0,0 +1,31 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: particleprocessorworker +spec: + selector: + matchLabels: + app: particleprocessorworker + replicas: 1 + template: + metadata: + labels: + app: particleprocessorworker + spec: + serviceAccountName: default-service-account + nodeSelector: + cloud.google.com/gke-spot: "true" + terminationGracePeriodSeconds: 15 + containers: + - name: particleprocessorworker + image: "particleprocessorworker" + resources: + requests: + memory: "52Mi" + cpu: 50m + limits: + memory: "52Mi" + cpu: 50m + env: + - name: "GCP_PROJECT" + value: "flowy-dev-440017" diff --git a/go/skaffold.yaml b/go/skaffold.yaml index 11e5f20..af0db9f 100644 --- a/go/skaffold.yaml +++ b/go/skaffold.yaml @@ -34,6 +34,10 @@ profiles: context: . docker: dockerfile: Dockerfile + - image: particleprocessorworker + context: . + docker: + dockerfile: Dockerfile.particleprocessorworker manifests: rawYaml: - k8s/dev/*