142 lines
3.5 KiB
Go
142 lines
3.5 KiB
Go
// 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 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/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())
|
|
networkSvc := network.NewReader(db.Pool())
|
|
|
|
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,
|
|
networkReader network.Reader,
|
|
) (written, scanned int, err error) {
|
|
humans, err := humanSvc.ListAll(ctx)
|
|
if err != nil {
|
|
return 0, 0, err
|
|
}
|
|
|
|
memberships, err := networkReader.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)
|
|
}
|