diff --git a/go/Dockerfile.memberreconciler b/go/Dockerfile.memberreconciler new file mode 100644 index 0000000..75afb40 --- /dev/null +++ b/go/Dockerfile.memberreconciler @@ -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/memberreconciler +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/memberreconciler . +RUN echo "copied over binary to production stage" +CMD ["./main"] diff --git a/go/cmd/emailnotifierjob/main.go b/go/cmd/emailnotifierjob/main.go index 84871d9..e115ac2 100644 --- a/go/cmd/emailnotifierjob/main.go +++ b/go/cmd/emailnotifierjob/main.go @@ -68,7 +68,7 @@ func main() { // Initialize services humanSvc := human.NewService(db.Pool()) - networkSvc := network.NewService(db.Pool(), aeroSvc, billing.Noop()) + networkSvc := network.NewService(db.Pool(), aeroSvc, billing.Noop(), nil) slog.Info("starting email notification cycle") if err := runNotificationCycle(ctx, firestoreClient, aeroSvc, pusherSvc, humanSvc, networkSvc); err != nil { diff --git a/go/cmd/memberreconciler/main.go b/go/cmd/memberreconciler/main.go new file mode 100644 index 0000000..aa66823 --- /dev/null +++ b/go/cmd/memberreconciler/main.go @@ -0,0 +1,144 @@ +// memberreconciler is a one-shot job (also safe to run on a cron) that makes +// the Firestore membership mirror (humans/{humanId}.networks) match the +// authoritative Postgres network_members table. +// +// Run before flipping Stage 1 Firestore rules so existing humans have their +// mirror docs populated, and on a cron to heal any drift from a dropped mirror +// write in network.Service. +// +// The reconciler reads the current Firestore state and only writes humans +// whose mirrored networks differ from Postgres. Writes cost ~3x reads, and in +// steady state drift is rare, so read-first keeps the cron nearly free. +package main + +import ( + "context" + "log/slog" + "os" + "slices" + "time" + + "cloud.google.com/go/firestore" + "github.com/flowy-live/llink/internal/billing" + "github.com/flowy-live/llink/internal/db" + "github.com/flowy-live/llink/internal/human" + "github.com/flowy-live/llink/internal/network" + "github.com/flowy-live/llink/internal/utils" + "google.golang.org/api/iterator" +) + +func main() { + ctx := context.Background() + + db.Init() + defer db.Cleanup() + + gcpProject := utils.MustGetEnv("GCP_PROJECT") + fs, err := firestore.NewClient(ctx, gcpProject) + if err != nil { + slog.Error("failed to create Firestore client", "error", err) + os.Exit(1) + } + defer fs.Close() + + humanSvc := human.NewService(db.Pool()) + // billing/aero/fs unused here — we only call ListAllMemberships. + networkSvc := network.NewService(db.Pool(), nil, billing.Noop(), nil) + + started := time.Now() + written, scanned, err := reconcile(ctx, fs, humanSvc, networkSvc) + if err != nil { + slog.Error("reconciliation failed", "error", err, "elapsed", time.Since(started)) + os.Exit(1) + } + slog.Info("reconciliation complete", + "humans_scanned", scanned, + "humans_written", written, + "elapsed", time.Since(started), + ) +} + +func reconcile( + ctx context.Context, + fs *firestore.Client, + humanSvc human.Service, + networkSvc network.Service, +) (written, scanned int, err error) { + humans, err := humanSvc.ListAll(ctx) + if err != nil { + return 0, 0, err + } + + memberships, err := networkSvc.ListAllMemberships(ctx) + if err != nil { + return 0, 0, err + } + + current, err := snapshotMirror(ctx, fs) + if err != nil { + return 0, 0, err + } + + bulk := fs.BulkWriter(ctx) + for _, h := range humans { + scanned++ + desired := memberships[h.ID] + if desired == nil { + desired = []string{} + } + if sameSet(current[h.ID], desired) { + continue + } + if _, err := bulk.Set(fs.Collection("humans").Doc(h.ID), map[string]any{ + "networks": desired, + "updated_at": firestore.ServerTimestamp, + }, firestore.MergeAll); err != nil { + return written, scanned, err + } + written++ + } + bulk.End() + return written, scanned, nil +} + +// snapshotMirror streams the humans collection once and returns a map of +// humanId -> current networks array. One iterator, N billed reads. +func snapshotMirror(ctx context.Context, fs *firestore.Client) (map[string][]string, error) { + out := map[string][]string{} + iter := fs.Collection("humans").Documents(ctx) + defer iter.Stop() + for { + doc, err := iter.Next() + if err == iterator.Done { + break + } + if err != nil { + return nil, err + } + var data struct { + Networks []string `firestore:"networks"` + } + if err := doc.DataTo(&data); err != nil { + slog.Warn("skipping malformed mirror doc", "id", doc.Ref.ID, "error", err) + continue + } + out[doc.Ref.ID] = data.Networks + } + return out, nil +} + +// sameSet reports whether a and b contain the same elements, ignoring order +// and duplicates. Firestore array ops don't preserve order, so set equality is +// the right comparison for the networks array. +func sameSet(a, b []string) bool { + if len(a) == 0 && len(b) == 0 { + return true + } + ac := slices.Clone(a) + bc := slices.Clone(b) + slices.Sort(ac) + slices.Sort(bc) + ac = slices.Compact(ac) + bc = slices.Compact(bc) + return slices.Equal(ac, bc) +} diff --git a/go/cmd/orion/main.go b/go/cmd/orion/main.go index b45faf0..c6fda4e 100644 --- a/go/cmd/orion/main.go +++ b/go/cmd/orion/main.go @@ -9,6 +9,7 @@ import ( "cloud.google.com/go/firestore" "cloud.google.com/go/storage" + firebase "firebase.google.com/go/v4" pbaero "github.com/flowy-live/llink/genproto/aero" "github.com/flowy-live/llink/internal" "github.com/flowy-live/llink/internal/auth" @@ -62,7 +63,19 @@ func main() { defer aeroServer.Close() aeroSvc := pbaero.NewPrimaryClient(aeroServer) - authSvc := auth.NewAuthService(redisClient, aeroSvc) + gcpProject := utils.MustGetEnv("GCP_PROJECT") + fbApp, err := firebase.NewApp(ctx, &firebase.Config{ProjectID: gcpProject}) + if err != nil { + slog.Error("failed to init Firebase Admin app", "error", err) + os.Exit(1) + } + fbAuth, err := fbApp.Auth(ctx) + if err != nil { + slog.Error("failed to create Firebase auth client", "error", err) + os.Exit(1) + } + + authSvc := auth.NewAuthService(redisClient, aeroSvc, fbAuth) humanSvc := human.NewService(db.Pool()) billingSvc, err := billing.NewService(ctx, db.Pool(), billing.Config{ @@ -78,7 +91,14 @@ func main() { os.Exit(1) } - networkSvc := network.NewService(db.Pool(), aeroSvc, billingSvc) + firestoreClient, err := firestore.NewClient(ctx, gcpProject) + if err != nil { + slog.Error("failed to create Firestore client", "error", err) + os.Exit(1) + } + defer firestoreClient.Close() + + networkSvc := network.NewService(db.Pool(), aeroSvc, billingSvc, firestoreClient) particleSvc := particle.NewService(db.Pool(), networkSvc) depotSvc := depot.NewService(db.Pool(), storageClient, depot.Config{ GoogleServiceAccountEmail: utils.MustGetEnv("GOOGLE_SERVICE_ACCOUNT_EMAIL"), @@ -87,14 +107,6 @@ func main() { waitlistSvc := waitlist.NewService(db.Pool(), aeroSvc) livekitClient := livekit.NewClient() - gcpProject := utils.MustGetEnv("GCP_PROJECT") - firestoreClient, err := firestore.NewClient(ctx, gcpProject) - if err != nil { - slog.Error("failed to create Firestore client", "error", err) - os.Exit(1) - } - defer firestoreClient.Close() - h := handler.NewHandler(authSvc, humanSvc, networkSvc, particleSvc, depotSvc, waitlistSvc, billingSvc, livekitClient, firestoreClient) withAuth := func(hf http.HandlerFunc) http.Handler { @@ -125,6 +137,7 @@ func main() { // Auth mux.Handle("POST /auth/sign-out", withAuth(h.SignOut)) mux.Handle("GET /auth/me", withAuth(h.GetCurrentHuman)) + mux.Handle("POST /auth/firebase-token", withAuth(h.FirebaseToken)) // Settings mux.Handle("PATCH /humans/me/settings", withAuth(h.UpdateSettings)) diff --git a/go/cmd/pusherservice/main.go b/go/cmd/pusherservice/main.go index 0dab5cb..21fd9b7 100644 --- a/go/cmd/pusherservice/main.go +++ b/go/cmd/pusherservice/main.go @@ -37,8 +37,8 @@ func main() { pusherRedis := internal.ConnectAndTestRedis(db.RedisDBPusher) // Services - authSvc := auth.NewAuthService(authRedis, nil) // nil aeroSvc — pusher only calls GetSession - networkSvc := network.NewService(db.Pool(), nil, billing.Noop()) // nil aeroSvc / noop billing — pusher never mutates membership + authSvc := auth.NewAuthService(authRedis, nil, nil) // nil aeroSvc / fbAuth — pusher only calls GetSession + networkSvc := network.NewService(db.Pool(), nil, billing.Noop(), nil) // nil aeroSvc / noop billing / nil firestore — pusher never mutates membership // Pod identity (use hostname in k8s, which is the pod name) podID, err := os.Hostname() diff --git a/go/go.mod b/go/go.mod index e9326e0..d9b8116 100644 --- a/go/go.mod +++ b/go/go.mod @@ -37,10 +37,12 @@ require ( 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 + firebase.google.com/go/v4 v4.19.0 // indirect github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.54.0 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0 // indirect + github.com/MicahParks/keyfunc v1.9.0 // indirect github.com/Microsoft/go-winio v0.6.2 // indirect github.com/antlr4-go/antlr/v4 v4.13.1 // indirect github.com/benbjohnson/clock v1.3.5 // indirect @@ -77,6 +79,8 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/gofrs/uuid/v5 v5.2.0 // indirect + github.com/golang-jwt/jwt/v4 v4.5.2 // indirect + github.com/golang/protobuf v1.5.4 // indirect github.com/google/cel-go v0.27.0 // indirect github.com/google/s2a-go v0.1.9 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.7 // indirect @@ -173,6 +177,7 @@ require ( golang.org/x/text v0.34.0 // indirect golang.org/x/time v0.14.0 // indirect google.golang.org/api v0.256.0 // indirect + google.golang.org/appengine/v2 v2.0.6 // indirect google.golang.org/genproto v0.0.0-20250922171735-9219d122eba9 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect diff --git a/go/go.sum b/go/go.sum index db9694f..39a66c2 100644 --- a/go/go.sum +++ b/go/go.sum @@ -30,6 +30,8 @@ cloud.google.com/go/trace v1.11.6 h1:2O2zjPzqPYAHrn3OKl029qlqG6W8ZdYaOWRyr8NgMT4 cloud.google.com/go/trace v1.11.6/go.mod h1:GA855OeDEBiBMzcckLPE2kDunIpC72N+Pq8WFieFjnI= dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= +firebase.google.com/go/v4 v4.19.0 h1:f5NMlC2YHFsncz00c2+ecBr+ZYlRMhKIhj1z8Iz0lD8= +firebase.google.com/go/v4 v4.19.0/go.mod h1:P7UfBpzc8+Z3MckX79+zsWzKVfpGryr6HLbAe7gCWfs= github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= @@ -42,6 +44,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.54.0/go.mod h1:vB2GH9GAYYJTO3mEn8oYwzEdhlayZIdQz6zdzgUIRvA= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0 h1:s0WlVbf9qpvkh1c/uDAPElam0WrL7fHRIidgZJ7UqZI= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.54.0/go.mod h1:Mf6O40IAyB9zR/1J8nGDDPirZQQPbYJni8Yisy7NTMc= +github.com/MicahParks/keyfunc v1.9.0 h1:lhKd5xrFHLNOWrDc4Tyb/Q1AJ4LCzQ48GVJyVIID3+o= +github.com/MicahParks/keyfunc v1.9.0/go.mod h1:IdnCilugA0O/99dW+/MkvlyrsX8+L8+x95xuVNtM5jw= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw= @@ -142,12 +146,17 @@ github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9L github.com/go-viper/mapstructure/v2 v2.4.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM= github.com/gofrs/uuid/v5 v5.2.0 h1:qw1GMx6/y8vhVsx626ImfKMuS5CvJmhIKKtuyvfajMM= github.com/gofrs/uuid/v5 v5.2.0/go.mod h1:CDOjlDMVAtN56jqyRUZh58JT31Tiw7/oQyEXZV+9bD8= +github.com/golang-jwt/jwt/v4 v4.4.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= +github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI= +github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= github.com/golang-migrate/migrate/v4 v4.19.1 h1:OCyb44lFuQfYXYLx1SCxPZQGU7mcaZ7gH9yH4jSFbBA= github.com/golang-migrate/migrate/v4 v4.19.1/go.mod h1:CTcgfjxhaUtsLipnLoQRWCrjYXycRz/g5+RWDuYgPrE= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/cel-go v0.27.0 h1:e7ih85+4qVrBuqQWTW4FKSqZYokVuc3HnhH5keboFTo= github.com/google/cel-go v0.27.0/go.mod h1:tTJ11FWqnhw5KKpnWpvW9CJC3Y9GK4EIS0WXnBbebzw= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= @@ -456,6 +465,7 @@ golang.org/x/term v0.40.0/go.mod h1:w2P8uVp06p2iyKKuvXIm7N/y0UCRt3UfJTfZ7oOpglM= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= @@ -473,6 +483,9 @@ gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= google.golang.org/api v0.256.0 h1:u6Khm8+F9sxbCTYNoBHg6/Hwv0N/i+V94MvkOSor6oI= google.golang.org/api v0.256.0/go.mod h1:KIgPhksXADEKJlnEoRa9qAII4rXcy40vfI8HRqcU964= +google.golang.org/appengine v1.6.8 h1:IhEN5q69dyKagZPYMSdIjS2HqprW324FRQZJcGqPAsM= +google.golang.org/appengine/v2 v2.0.6 h1:LvPZLGuchSBslPBp+LAhihBeGSiRh1myRoYK4NtuBIw= +google.golang.org/appengine/v2 v2.0.6/go.mod h1:WoEXGoXNfa0mLvaH5sV3ZSGXwVmy8yf7Z1JKf3J3wLI= google.golang.org/genproto v0.0.0-20250922171735-9219d122eba9 h1:LvZVVaPE0JSqL+ZWb6ErZfnEOKIqqFWUJE2D0fObSmc= google.golang.org/genproto v0.0.0-20250922171735-9219d122eba9/go.mod h1:QFOrLhdAe2PsTp3vQY4quuLKTi9j3XG3r6JPPaw7MSc= google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 h1:JLQynH/LBHfCTSbDWl+py8C+Rg/k1OVH3xfcaiANuF0= @@ -481,6 +494,8 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 h1: google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ= google.golang.org/grpc v1.79.1 h1:zGhSi45ODB9/p3VAawt9a+O/MULLl9dpizzNNpq7flY= google.golang.org/grpc v1.79.1/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/go/internal/auth/service.go b/go/internal/auth/service.go index 0d6a85b..24222b9 100644 --- a/go/internal/auth/service.go +++ b/go/internal/auth/service.go @@ -9,6 +9,7 @@ import ( "strings" "time" + firebaseauth "firebase.google.com/go/v4/auth" "github.com/flowy-live/llink/genproto/aero" "github.com/flowy-live/llink/internal/utils" "github.com/redis/go-redis/v9" @@ -56,16 +57,30 @@ type AuthService interface { ExtendSession(ctx context.Context, sessionToken string) error SignOut(ctx context.Context, sessionToken string) error + // MintFirebaseCustomToken returns a Firebase custom token with uid=humanId + // and no custom claims. Network membership is enforced via a Firestore + // mirror (humans/{humanId}.networks) read at rule eval time, so the token + // only needs to carry identity. + MintFirebaseCustomToken(ctx context.Context, humanId string) (string, error) + IsSystemAdmin(ctx context.Context, email string) bool } type authServiceImpl struct { redisClient *redis.Client aeroSvc pbaero.PrimaryClient + fbAuth *firebaseauth.Client } -func NewAuthService(redisClient *redis.Client, aeroSvc pbaero.PrimaryClient) AuthService { - return &authServiceImpl{redisClient: redisClient, aeroSvc: aeroSvc} +func NewAuthService(redisClient *redis.Client, aeroSvc pbaero.PrimaryClient, fbAuth *firebaseauth.Client) AuthService { + return &authServiceImpl{redisClient: redisClient, aeroSvc: aeroSvc, fbAuth: fbAuth} +} + +func (a *authServiceImpl) MintFirebaseCustomToken(ctx context.Context, humanId string) (string, error) { + if humanId == "" { + return "", errors.New("humanId is required") + } + return a.fbAuth.CustomToken(ctx, humanId) } func (a *authServiceImpl) IsSystemAdmin(ctx context.Context, email string) bool { diff --git a/go/internal/handler/handler.go b/go/internal/handler/handler.go index 85feeec..7989edc 100644 --- a/go/internal/handler/handler.go +++ b/go/internal/handler/handler.go @@ -94,6 +94,10 @@ type SignInResponse struct { Token string `json:"token"` } +type FirebaseTokenResponse struct { + Token string `json:"token"` +} + // Network Request DTOs type CreateNetworkRequest struct { @@ -240,6 +244,27 @@ func (h *Handler) SignIn(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(resp) } +// FirebaseToken mints a Firebase custom token for the authenticated human so +// the client can signInWithCustomToken and have request.auth.uid populated in +// Firestore security rules. +func (h *Handler) FirebaseToken(w http.ResponseWriter, r *http.Request) { + humanId, ok := middleware.HumanIdFromContext(r.Context()) + if !ok { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + token, err := h.authSvc.MintFirebaseCustomToken(r.Context(), humanId) + if err != nil { + slog.Error("failed to mint Firebase custom token", "error", err, "humanId", humanId) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(FirebaseTokenResponse{Token: token}) +} + // SignOut deletes the session from the token in headers func (h *Handler) SignOut(w http.ResponseWriter, r *http.Request) { token := extractBearerToken(r) diff --git a/go/internal/network/repository.go b/go/internal/network/repository.go index 523a89b..d755ce5 100644 --- a/go/internal/network/repository.go +++ b/go/internal/network/repository.go @@ -47,6 +47,7 @@ type repository interface { getNetworksForHuman(ctx context.Context, humanId string) ([]*Network, error) isMember(ctx context.Context, networkID, humanId string) (bool, error) listAll(ctx context.Context) ([]*Network, error) + listAllMemberships(ctx context.Context) (map[string][]string, error) // Invitations createInvitation(ctx context.Context, networkID, email string) error @@ -232,6 +233,27 @@ func (r *repositoryImpl) isMember(ctx context.Context, networkID, humanId string return isMember, err } +// listAllMemberships returns humanId -> networkIds for every human with at +// least one membership. Humans with zero memberships are absent from the map; +// callers layer them in separately. +func (r *repositoryImpl) listAllMemberships(ctx context.Context) (map[string][]string, error) { + rows, err := r.pool.Query(ctx, `SELECT human_id, network_id FROM network_members`) + if err != nil { + return nil, err + } + defer rows.Close() + + out := map[string][]string{} + for rows.Next() { + var humanId, networkId string + if err := rows.Scan(&humanId, &networkId); err != nil { + return nil, err + } + out[humanId] = append(out[humanId], networkId) + } + return out, rows.Err() +} + func (r *repositoryImpl) listAll(ctx context.Context) ([]*Network, error) { rows, err := r.pool.Query(ctx, `SELECT `+networkColumns+` FROM networks`, diff --git a/go/internal/network/service.go b/go/internal/network/service.go index c06392a..1db969a 100644 --- a/go/internal/network/service.go +++ b/go/internal/network/service.go @@ -8,6 +8,7 @@ import ( "log/slog" "strings" + "cloud.google.com/go/firestore" pbaero "github.com/flowy-live/llink/genproto/aero" "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/utils" @@ -37,6 +38,10 @@ type Service interface { IsMember(ctx context.Context, networkID, humanId string) (bool, error) // ListAll returns all networks with their members ListAll(ctx context.Context) ([]*Network, error) + // ListAllMemberships returns humanId -> networkIds for every human with at + // least one membership. Humans with zero memberships are absent from the map. + // Used by the membership reconciler to diff the Firestore mirror. + ListAllMemberships(ctx context.Context) (map[string][]string, error) // Invitations (email-based, for users who haven't registered yet) InviteByEmail(ctx context.Context, networkID string, emails []string) error @@ -51,14 +56,19 @@ type serviceImpl struct { repo repository aeroSvc pbaero.PrimaryClient billingSvc billing.Service + // fs mirrors network membership into humans/{humanId}.networks so + // Firestore security rules can check membership at rule-eval time. + // May be nil in services that never mutate membership (pusher). + fs *firestore.Client } -func NewService(pool *pgxpool.Pool, aeroSvc pbaero.PrimaryClient, billingSvc billing.Service) Service { +func NewService(pool *pgxpool.Pool, aeroSvc pbaero.PrimaryClient, billingSvc billing.Service, fs *firestore.Client) Service { return &serviceImpl{ pool: pool, repo: newRepository(pool), aeroSvc: aeroSvc, billingSvc: billingSvc, + fs: fs, } } @@ -105,23 +115,66 @@ func (s *serviceImpl) AddMembers(ctx context.Context, networkID string, humanIds if slices.Contains(humanIds, "") { return fmt.Errorf("invalid humanId") } - return s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { + if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { for _, humanId := range humanIds { if err := s.repo.addMember(ctx, tx, networkID, humanId); err != nil { return err } } return nil - }) + }); err != nil { + return err + } + + for _, humanId := range humanIds { + s.mirrorAddMembership(ctx, humanId, networkID) + } + return nil } func (s *serviceImpl) RemoveMember(ctx context.Context, networkID, humanId string) error { if humanId == "" { return fmt.Errorf("invalid humanId") } - return s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { + if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { return s.repo.removeMember(ctx, tx, networkID, humanId) - }) + }); err != nil { + return err + } + + s.mirrorRemoveMembership(ctx, humanId, networkID) + return nil +} + +// mirrorAddMembership / mirrorRemoveMembership keep the Firestore membership +// mirror (humans/{humanId}.networks) in sync with Postgres. Called after the +// Postgres transaction commits. Failures are logged but not returned: Postgres +// is the source of truth and the reconciler will heal drift. A missing fs +// client (pusher) no-ops. +func (s *serviceImpl) mirrorAddMembership(ctx context.Context, humanId, networkID string) { + if s.fs == nil { + return + } + _, err := s.fs.Collection("humans").Doc(humanId).Set(ctx, map[string]any{ + "networks": firestore.ArrayUnion(networkID), + "updated_at": firestore.ServerTimestamp, + }, firestore.MergeAll) + if err != nil { + slog.Error("firestore mirror add failed", "error", err, "humanId", humanId, "networkID", networkID) + } +} + +func (s *serviceImpl) mirrorRemoveMembership(ctx context.Context, humanId, networkID string) { + if s.fs == nil { + return + } + _, err := s.fs.Collection("humans").Doc(humanId).Set(ctx, map[string]any{ + "networks": firestore.ArrayRemove(networkID), + "updated_at": firestore.ServerTimestamp, + }, firestore.MergeAll) + if err != nil { + slog.Error("firestore mirror remove failed", "error", err, "humanId", humanId, "networkID", networkID) + } } // mutateMembers runs fn in a tx, recounts seats, calls billing.SyncSeats, @@ -174,6 +227,10 @@ func (s *serviceImpl) ListAll(ctx context.Context) ([]*Network, error) { return s.repo.listAll(ctx) } +func (s *serviceImpl) ListAllMemberships(ctx context.Context) (map[string][]string, error) { + return s.repo.listAllMemberships(ctx) +} + func (s *serviceImpl) InviteByEmail(ctx context.Context, networkID string, emails []string) error { network, err := s.repo.getByID(ctx, networkID) if err != nil { @@ -226,13 +283,18 @@ func (s *serviceImpl) AcceptInvitation(ctx context.Context, networkID, email, hu return fmt.Errorf("invalid humanId") } - return s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { + if err := s.mutateMembers(ctx, networkID, func(tx pgx.Tx) error { err := s.repo.deleteInvitation(ctx, tx, networkID, normalized) if err != nil { return err } return s.repo.addMember(ctx, tx, networkID, humanId) - }) + }); err != nil { + return err + } + + s.mirrorAddMembership(ctx, humanId, networkID) + return nil } func (s *serviceImpl) RevokeInvitation(ctx context.Context, networkID, email string) error { diff --git a/go/internal/network/service_test.go b/go/internal/network/service_test.go index c14e4fa..5901267 100644 --- a/go/internal/network/service_test.go +++ b/go/internal/network/service_test.go @@ -32,7 +32,7 @@ func newTestService(t *testing.T) network.Service { ShootEmail(gomock.Any(), gomock.Any()). Return(&pbaero.ShootEmailResponse{}, nil). AnyTimes() - return network.NewService(dbPool, mockAero, billing.Noop()) + return network.NewService(dbPool, mockAero, billing.Noop(), nil) } func TestNetworkService(t *testing.T) { diff --git a/go/internal/particle/service_test.go b/go/internal/particle/service_test.go index 61a6946..f128705 100644 --- a/go/internal/particle/service_test.go +++ b/go/internal/particle/service_test.go @@ -34,7 +34,7 @@ func getStreamStatus(data json.RawMessage) string { func TestParticleService_CreateAndGet(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil, billing.Noop()) + networkSvc := network.NewService(dbPool, nil, billing.Noop(), nil) svc := particle.NewService(dbPool, networkSvc) // Create a network first @@ -73,7 +73,7 @@ func TestParticleService_CreateAndGet(t *testing.T) { func TestParticleService_NestedParticles(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil, billing.Noop()) + networkSvc := network.NewService(dbPool, nil, billing.Noop(), nil) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -119,7 +119,7 @@ func TestParticleService_NestedParticles(t *testing.T) { func TestParticleService_CustomVisibility(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil, billing.Noop()) + networkSvc := network.NewService(dbPool, nil, billing.Noop(), nil) svc := particle.NewService(dbPool, networkSvc) // Create a network with a member @@ -161,7 +161,7 @@ func TestParticleService_CustomVisibility(t *testing.T) { func TestParticleService_UpdateAndDelete(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil, billing.Noop()) + networkSvc := network.NewService(dbPool, nil, billing.Noop(), nil) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -199,7 +199,7 @@ func TestParticleService_UpdateAndDelete(t *testing.T) { func TestParticleService_ListRootParticles(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil, billing.Noop()) + networkSvc := network.NewService(dbPool, nil, billing.Noop(), nil) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -225,7 +225,7 @@ func TestParticleService_ListRootParticles(t *testing.T) { func TestParticleService_OpenCloseStream(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil, billing.Noop()) + networkSvc := network.NewService(dbPool, nil, billing.Noop(), nil) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -273,7 +273,7 @@ func TestParticleService_OpenCloseStream(t *testing.T) { func TestParticleService_NotAStream(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil, billing.Noop()) + networkSvc := network.NewService(dbPool, nil, billing.Noop(), nil) svc := particle.NewService(dbPool, networkSvc) // Create a network @@ -302,7 +302,7 @@ func TestParticleService_NotAStream(t *testing.T) { func TestParticleService_AccessInheritance(t *testing.T) { ctx := context.Background() - networkSvc := network.NewService(dbPool, nil, billing.Noop()) + networkSvc := network.NewService(dbPool, nil, billing.Noop(), nil) svc := particle.NewService(dbPool, networkSvc) // Create a network with members diff --git a/go/k8s/dev/memberreconciler.yaml b/go/k8s/dev/memberreconciler.yaml new file mode 100644 index 0000000..b876596 --- /dev/null +++ b/go/k8s/dev/memberreconciler.yaml @@ -0,0 +1,38 @@ +apiVersion: batch/v1 +kind: CronJob +metadata: + name: memberreconciler +spec: + schedule: "0 2 * * *" + concurrencyPolicy: Forbid + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 3 + jobTemplate: + spec: + template: + metadata: + labels: + app: memberreconciler + spec: + serviceAccountName: default-service-account + nodeSelector: + cloud.google.com/gke-spot: "true" + restartPolicy: Never + containers: + - name: memberreconciler + image: "memberreconciler" + resources: + requests: + memory: "64Mi" + cpu: 50m + limits: + memory: "64Mi" + cpu: 50m + env: + - name: "GCP_PROJECT" + value: "flowy-dev-440017" + - name: "LLINK_POSTGRES_CONNECTION_URL" + valueFrom: + secretKeyRef: + name: shared-secrets + key: LLINK_POSTGRES_CONNECTION_URL diff --git a/go/k8s/prod/memberreconciler.yaml b/go/k8s/prod/memberreconciler.yaml new file mode 100644 index 0000000..88c7d8e --- /dev/null +++ b/go/k8s/prod/memberreconciler.yaml @@ -0,0 +1,36 @@ +apiVersion: batch/v1 +kind: CronJob +metadata: + name: memberreconciler +spec: + schedule: "0 2 * * *" + concurrencyPolicy: Forbid + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 3 + jobTemplate: + spec: + template: + metadata: + labels: + app: memberreconciler + spec: + serviceAccountName: default-service-account + restartPolicy: Never + containers: + - name: memberreconciler + image: "memberreconciler" + resources: + requests: + memory: "64Mi" + cpu: 50m + limits: + memory: "64Mi" + cpu: 50m + env: + - name: "GCP_PROJECT" + value: "flowy-prod-440017" + - name: "LLINK_POSTGRES_CONNECTION_URL" + valueFrom: + secretKeyRef: + name: shared-secrets + key: LLINK_POSTGRES_CONNECTION_URL diff --git a/go/skaffold.yaml b/go/skaffold.yaml index 34bbe9c..1669737 100644 --- a/go/skaffold.yaml +++ b/go/skaffold.yaml @@ -128,6 +128,41 @@ profiles: --- apiVersion: skaffold/v4beta11 kind: Config +metadata: + name: memberreconciler +build: + local: {} + tagPolicy: + gitCommit: + variant: AbbrevCommitSha +profiles: + - name: dev + build: + artifacts: + - image: memberreconciler + context: . + docker: + dockerfile: Dockerfile.memberreconciler + manifests: + rawYaml: + - k8s/dev/memberreconciler.yaml + deploy: + kubectl: {} + - name: prod + build: + artifacts: + - image: memberreconciler + context: . + docker: + dockerfile: Dockerfile.memberreconciler + manifests: + rawYaml: + - k8s/prod/memberreconciler.yaml + deploy: + kubectl: {} +--- +apiVersion: skaffold/v4beta11 +kind: Config metadata: name: pusher build: diff --git a/js/src/api/client.ts b/js/src/api/client.ts index d1671e0..bb6699d 100644 --- a/js/src/api/client.ts +++ b/js/src/api/client.ts @@ -5,6 +5,7 @@ import { BillingStatusSchema, CheckoutSessionResponseSchema, DepotObjectSchema, + FirebaseTokenResponseSchema, GetLivekitTokenResponseSchema, HumanSchema, ListInvitationsResponseSchema, @@ -121,6 +122,14 @@ class ApiClient { await this.requestVoid("POST", "/auth/sign-out"); } + async getFirebaseToken() { + return this.request( + FirebaseTokenResponseSchema, + "POST", + "/auth/firebase-token", + ); + } + // TODO: security: require passing in the particle id once api deprecates this async getParticleDownloadUrl(objectId: string): Promise { const response = await this.fetch( diff --git a/js/src/api/types.ts b/js/src/api/types.ts index d1b1a60..42e014a 100644 --- a/js/src/api/types.ts +++ b/js/src/api/types.ts @@ -263,6 +263,11 @@ export const SignInResponseSchema = z.object({ }); export type SignInResponse = z.infer; +export const FirebaseTokenResponseSchema = z.object({ + token: z.string(), +}); +export type FirebaseTokenResponse = z.infer; + // --- Billing types --- export const BillingCadenceSchema = z.enum(["monthly", "annual"]); diff --git a/js/src/firebase.ts b/js/src/firebase.ts index 8825944..a8e931f 100644 --- a/js/src/firebase.ts +++ b/js/src/firebase.ts @@ -1,9 +1,12 @@ import { initializeApp } from 'firebase/app'; +import { getAuth } from "firebase/auth"; import { getFirestore } from "firebase/firestore"; import { appConfig } from "@/config/env"; export const firebaseApp = initializeApp(appConfig.firebase); +export const firebaseAuth = getAuth(firebaseApp); + export const firestoreDb = getFirestore(firebaseApp); // simplifying setup to debug production issues // export const firestoreDb = initializeFirestore(firebaseApp, diff --git a/js/src/stores/auth-store.ts b/js/src/stores/auth-store.ts index 252b5bd..426cb20 100644 --- a/js/src/stores/auth-store.ts +++ b/js/src/stores/auth-store.ts @@ -1,8 +1,19 @@ import { create } from "zustand"; +import { signInWithCustomToken, signOut as firebaseSignOut } from "firebase/auth"; import { apiClient, ApiError } from "@/api/client"; import type { Human } from "@/api/types"; +import { firebaseAuth } from "@/firebase"; import { useSessionStore } from "./session-store"; +async function signInToFirebase() { + try { + const { token } = await apiClient.getFirebaseToken(); + await signInWithCustomToken(firebaseAuth, token); + } catch (e) { + console.error("Failed to sign in to Firebase", e); + } +} + type AuthStatus = "idle" | "restoring" | "unauthenticated" | "authenticated"; interface AuthState { @@ -37,6 +48,7 @@ export const useAuthStore = create((set) => ({ set({ status: "restoring" }); try { const user = await apiClient.me(); + await signInToFirebase(); set({ status: "authenticated", user }); } catch { useSessionStore.getState().clearToken(); @@ -63,6 +75,7 @@ export const useAuthStore = create((set) => ({ try { const { human, token } = await apiClient.signIn({ email, code }); useSessionStore.getState().setToken(token); + await signInToFirebase(); set({ status: "authenticated", user: human }); } catch (e) { const message = @@ -81,6 +94,9 @@ export const useAuthStore = create((set) => ({ } catch { // Best-effort — sign out locally regardless } finally { + await firebaseSignOut(firebaseAuth).catch((e) => + console.error("Firebase sign-out failed", e), + ); useSessionStore.getState().clearToken(); set({ status: "unauthenticated",