From 6f476fe7738b912451c0ed4ad3bd1a708598f823 Mon Sep 17 00:00:00 2001 From: talksik Date: Mon, 18 May 2026 10:31:16 -0700 Subject: [PATCH] implement backend components for push notifications --- go/cmd/emailnotifierjob/main.go | 37 +--- go/cmd/orion/main.go | 8 +- go/cmd/particleprocessorworker/main.go | 223 ++++++++++++++++++--- go/internal/handler/handler.go | 4 + go/internal/handler/push_token.go | 85 ++++++++ go/internal/human/pushnotify/expo.go | 140 +++++++++++++ go/internal/human/pushnotify/notifier.go | 194 ++++++++++++++++++ go/internal/human/pushnotify/repository.go | 95 +++++++++ go/internal/human/pushnotify/service.go | 70 +++++++ go/internal/human/pushnotify/token.go | 43 ++++ go/internal/network/reader.go | 9 + go/internal/network/visibility.go | 41 ++++ go/k8s/dev/particleprocessorworker.yaml | 2 + go/k8s/prod/particleprocessorworker.yaml | 2 + go/migrations/000016_push_tokens.down.sql | 1 + go/migrations/000016_push_tokens.up.sql | 10 + 16 files changed, 904 insertions(+), 60 deletions(-) create mode 100644 go/internal/handler/push_token.go create mode 100644 go/internal/human/pushnotify/expo.go create mode 100644 go/internal/human/pushnotify/notifier.go create mode 100644 go/internal/human/pushnotify/repository.go create mode 100644 go/internal/human/pushnotify/service.go create mode 100644 go/internal/human/pushnotify/token.go create mode 100644 go/internal/network/visibility.go create mode 100644 go/migrations/000016_push_tokens.down.sql create mode 100644 go/migrations/000016_push_tokens.up.sql diff --git a/go/cmd/emailnotifierjob/main.go b/go/cmd/emailnotifierjob/main.go index 0cfeab8..b16ba04 100644 --- a/go/cmd/emailnotifierjob/main.go +++ b/go/cmd/emailnotifierjob/main.go @@ -5,7 +5,6 @@ import ( "fmt" "log/slog" "os" - "strings" "time" "cloud.google.com/go/firestore" @@ -119,13 +118,6 @@ func runNotificationCycle( } for _, net := range networks { - // Build set of all member humanIds for this network (members + admin) - networkMembers := make(map[string]bool, len(net.MemberHumanIds)+1) - for _, id := range net.MemberHumanIds { - networkMembers[id] = true - } - networkMembers[net.AdminHumanId] = true - // Query Firestore for open streams in this network streams, err := getOpenStreams(ctx, fsClient, net.ID) if err != nil { @@ -133,6 +125,8 @@ func runNotificationCycle( continue } + // net.MemberHumanIds already contains the admin (Create() adds them and + // RemoveMemberFromNetwork won't drop them) — no need to union separately. for _, stream := range streams { if stream.LastChildCreatedAt == nil { continue @@ -146,10 +140,8 @@ func runNotificationCycle( continue } - // Resolve members from visible_to - members := resolveMembers(stream.VisibleTo, networkMembers) - - for humanId := range members { + // Resolve members from visible_to using the shared helper + for _, humanId := range network.ResolveVisibility(stream.VisibleTo, net.MemberHumanIds) { marker, hasMarker := stream.PlaybackMarkers[humanId] if hasMarker && !marker.Before(*stream.LastChildCreatedAt) { continue // up to date @@ -239,27 +231,6 @@ func getOpenStreams(ctx context.Context, client *firestore.Client, networkId str return streams, nil } -// resolveMembers expands visible_to entries into a set of humanIds. -// "human:{id}" adds that id directly. "network:{id}" expands to all network members. -func resolveMembers(visibleTo []string, networkMembers map[string]bool) map[string]bool { - members := map[string]bool{} - for _, entry := range visibleTo { - if strings.HasPrefix(entry, "human:") { - humanId := strings.TrimPrefix(entry, "human:") - // human can be in visible_to, but no longer a member of the network - if _, ok := networkMembers[humanId]; ok { - members[humanId] = true - } - } else if strings.HasPrefix(entry, "network:") { - // Expand to all network members - for id := range networkMembers { - members[id] = true - } - } - } - return members -} - func sendNotificationEmail(ctx context.Context, aeroSvc pbaero.PrimaryClient, h *human.Human, streamCount int) error { streamsWord := "stream" if streamCount != 1 { diff --git a/go/cmd/orion/main.go b/go/cmd/orion/main.go index 205c324..0f6c4eb 100644 --- a/go/cmd/orion/main.go +++ b/go/cmd/orion/main.go @@ -18,6 +18,7 @@ import ( "github.com/flowy-live/llink/internal/depot" "github.com/flowy-live/llink/internal/handler" "github.com/flowy-live/llink/internal/human" + "github.com/flowy-live/llink/internal/human/pushnotify" "github.com/flowy-live/llink/internal/livekit" "github.com/flowy-live/llink/internal/livestore" "github.com/flowy-live/llink/internal/middleware" @@ -106,9 +107,10 @@ func main() { BucketName: gcsBucket, }) waitlistSvc := waitlist.NewService(db.Pool(), aeroSvc) + pushTokenSvc := pushnotify.NewService(db.Pool()) livekitClient := livekit.NewClient() - h := handler.NewHandler(authSvc, humanSvc, networkSvc, particleSvc, depotSvc, waitlistSvc, billingSvc, livekitClient, firestoreClient) + h := handler.NewHandler(authSvc, humanSvc, networkSvc, particleSvc, depotSvc, waitlistSvc, billingSvc, pushTokenSvc, livekitClient, firestoreClient) withAuth := func(hf http.HandlerFunc) http.Handler { return middleware.Auth(authSvc)(http.HandlerFunc(hf)) @@ -143,6 +145,10 @@ func main() { // Settings mux.Handle("PATCH /humans/me/settings", withAuth(h.UpdateSettings)) + // Push notification tokens (per-device) + mux.Handle("POST /humans/me/push-tokens", withAuth(h.RegisterPushToken)) + mux.Handle("DELETE /humans/me/push-tokens", withAuth(h.UnregisterPushToken)) + // Networks mux.Handle("POST /networks", withAuth(h.CreateNetwork)) mux.Handle("GET /networks", withAuth(h.ListNetworks)) diff --git a/go/cmd/particleprocessorworker/main.go b/go/cmd/particleprocessorworker/main.go index f7f34bd..b18706d 100644 --- a/go/cmd/particleprocessorworker/main.go +++ b/go/cmd/particleprocessorworker/main.go @@ -6,15 +6,22 @@ import ( "log" "log/slog" "os" + "strings" "time" + pbpusher "github.com/flowy-live/llink/genproto/llink/pusher" "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/db" "github.com/flowy-live/llink/internal/depot" + "github.com/flowy-live/llink/internal/human" + "github.com/flowy-live/llink/internal/human/pushnotify" + "github.com/flowy-live/llink/internal/network" "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" "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" "cloud.google.com/go/firestore" @@ -61,6 +68,24 @@ func main() { speechSvc := speech.NewSpeechService(ctx) + pusherAddr := utils.MustGetEnv("PUSHER_GRPC_ADDR") + pusherConn, err := grpc.NewClient(pusherAddr, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + slog.Error("failed to connect to pusher", "error", err) + os.Exit(1) + } + defer pusherConn.Close() + pusherClient := pbpusher.NewPusherServiceClient(pusherConn) + + humanSvc := human.NewService(db.Pool()) + networkReader := network.NewReader(db.Pool()) + pushTokenSvc := pushnotify.NewService(db.Pool()) + // EXPO_ACCESS_TOKEN is required: with Enhanced Security enabled on the Expo + // project, sends without it fail; without it, anyone holding one of our + // push tokens could spam our users via the public Expo endpoint. + expoClient := pushnotify.NewExpoClient(utils.MustGetEnv("EXPO_ACCESS_TOKEN")) + notifier := pushnotify.NewNotifier(networkReader, pushTokenSvc, pusherClient, expoClient) + client := createClient(ctx) defer client.Close() @@ -103,12 +128,13 @@ func main() { slog.Debug("processing particle", "particleID", particleID, "data", change.Doc.Data()) // --- Perform side effects --- - // All of them do not stop us from marking the particle as processed - - updateParentLastChildCreatedAt(ctx, change.Doc) - transcribeMediaParticle(ctx, depotSvc, speechSvc, change.Doc) + // All of them do not stop us from marking the particle as processed. + parentDoc := loadParentParticle(ctx, change.Doc) + updateParentLastChildCreatedAt(ctx, change.Doc, parentDoc) + transcript := transcribeMediaParticle(ctx, depotSvc, speechSvc, change.Doc) particle.Transcode(ctx, depotSvc, change.Doc) recordFreemiumUsage(ctx, billingSvc, change.Doc) + 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) @@ -117,35 +143,38 @@ func main() { } } -func transcribeMediaParticle(ctx context.Context, depotSvc depot.Service, speechSvc speech.SpeechService, doc *firestore.DocumentSnapshot) { +// transcribeMediaParticle transcribes a media particle, writes the structured +// transcript to Firestore, and returns the raw transcript text. Returns "" for +// non-media particles or on any error (errors are logged internally). +func transcribeMediaParticle(ctx context.Context, depotSvc depot.Service, speechSvc speech.SpeechService, doc *firestore.DocumentSnapshot) string { var mediaParticle particle.FirestoreMediaParticle err := doc.DataTo(&mediaParticle) if err != nil { slog.Error("unable to marshal particle data", "error", err) - return + return "" } particleType, err := particle.ParseParticleType(mediaParticle.Type) if err != nil { slog.Error("invalid particle type", "error", err) - return + return "" } if particleType != particle.TypeMedia { slog.Info("received a particle of type", "particle type", particleType) - return + 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) - return + return "" } result, err := speechSvc.Transcribe(ctx, downloadURL) if err != nil { slog.Error("failed to transcribe media", "error", err, "particleID", doc.Ref.ID) - return + return "" } transcript := toFirestoreTranscript(result) @@ -157,10 +186,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) - return + return "" } slog.Info("transcribed media particle", "particleID", doc.Ref.ID) + return transcript.Transcript } func toFirestoreTranscript(result *speech.TranscriptResult) particle.FirestoreTranscript { @@ -233,31 +263,36 @@ func recordFreemiumUsage(ctx context.Context, billingSvc billing.Service, doc *f } } -// updateParentLastChildCreatedAt updates the parent stream's last_child_created_at -// to the child's actual created_at timestamp, so it stays directly comparable with -// playback markers (which also store child created_at values). -func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.DocumentSnapshot) { +// loadParentParticle fetches the immediate parent particle doc for `doc`. +// Returns nil (and logs) if the path doesn't have a parent or the read fails. +func loadParentParticle(ctx context.Context, doc *firestore.DocumentSnapshot) *firestore.DocumentSnapshot { parentChildrenCollectionRef := doc.Ref.Parent if parentChildrenCollectionRef == nil { - return + return nil } - parentParticleDocRef := parentChildrenCollectionRef.Parent if parentParticleDocRef == nil { slog.Error("particle has no parent document", "particleID", doc.Ref.ID) - return + return nil } - parentParticleDoc, err := parentParticleDocRef.Get(ctx) if err != nil { - slog.Error("failed to get parent particle", "error", err) + slog.Error("failed to get parent particle", "error", err, "particleID", doc.Ref.ID) + return nil + } + return parentParticleDoc +} + +// updateParentLastChildCreatedAt updates the parent stream's last_child_created_at +// to the child's actual created_at timestamp, so it stays directly comparable with +// playback markers (which also store child created_at values). +func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.DocumentSnapshot, parent *firestore.DocumentSnapshot) { + if parent == nil { return } - slog.Info("parent particle is", "parent particle id", parentParticleDoc.Ref.ID) - var streamParticle particle.FirestoreStreamParticle - if err := parentParticleDoc.DataTo(&streamParticle); err != nil { + if err := parent.DataTo(&streamParticle); err != nil { slog.Error("failed to parse stream particle", "error", err) return } @@ -279,14 +314,150 @@ func updateParentLastChildCreatedAt(ctx context.Context, doc *firestore.Document return } - slog.Info("going to update the last_child_created_at for parent particle") - _, err = parentParticleDocRef.Update(ctx, []firestore.Update{ + _, err = parent.Ref.Update(ctx, []firestore.Update{ { Path: "last_child_created_at", Value: childCreatedAt, }, }) if err != nil { - slog.Error("unable to update parent particle `last_child_created_at`") + slog.Error("unable to update parent particle `last_child_created_at`", "error", err) } } + +// notifyForParticle dispatches a push notification for a newly-created particle. +// Skips containers (streams/folders) and particles whose parent isn't a stream +// (notifications are only sent for stream messages today). The transcript arg +// is used as the preview body for media particles when available. +func notifyForParticle( + ctx context.Context, + notifier *pushnotify.Notifier, + humanSvc human.Service, + doc *firestore.DocumentSnapshot, + parent *firestore.DocumentSnapshot, + transcript string, +) { + if parent == nil { + return + } + + typeStr, _ := doc.DataAt("type") + typeName, _ := typeStr.(string) + pType, err := particle.ParseParticleType(typeName) + if err != nil { + return + } + if pType == particle.TypeStream || pType == particle.TypeFolder { + return + } + + var parentStream particle.FirestoreStreamParticle + if err := parent.DataTo(&parentStream); err != nil { + slog.Error("notify: failed to parse parent stream", "error", err) + return + } + parentType, err := particle.ParseParticleType(parentStream.Type) + if err != nil || parentType != particle.TypeStream { + 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) + return + } + + senderHumanID := parentStream.CreatedByHumanId + if v, err := doc.DataAt("created_by_human_id"); err == nil { + if s, ok := v.(string); ok && s != "" { + senderHumanID = s + } + } + + streamName := "" + if v, err := parent.DataAt("properties.name"); err == nil { + if s, ok := v.(string); ok { + streamName = s + } + } + + senderEmailPrefix := "" + if senderHumanID != "" { + 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) + } + } + + if err := notifier.NotifyParticleCreated(ctx, pushnotify.NotifyInput{ + NetworkID: networkID, + SenderHumanID: senderHumanID, + SenderEmailPrefix: senderEmailPrefix, + ParticleID: doc.Ref.ID, + ParticleKind: string(pType), + StreamID: parent.Ref.ID, + StreamName: streamName, + StreamVisibleTo: parentStream.VisibleTo, + Body: previewForParticle(pType, doc, transcript), + }); err != nil { + slog.Error("notify: dispatch failed", "error", err, "particleID", doc.Ref.ID, "networkID", networkID) + } +} + +// previewForParticle builds the visible notification body. Kept short — push +// previews truncate aggressively on lockscreens. For media, prefers the +// transcript text (already computed by transcribeMediaParticle in the same +// processing step) and falls back to the generic "Sent a ..." line if speech +// recognition produced nothing. +func previewForParticle(pType particle.ParticleType, doc *firestore.DocumentSnapshot, transcript string) string { + switch pType { + case particle.TypeText: + if v, err := doc.DataAt("properties.content"); err == nil { + if s, ok := v.(string); ok { + return truncatePreview(s, 140) + } + } + return "Sent a message" + case particle.TypeMedia: + if t := strings.TrimSpace(transcript); t != "" { + return truncatePreview(t, 140) + } + mime := "" + if v, err := doc.DataAt("properties.mime_type"); err == nil { + if s, ok := v.(string); ok { + mime = s + } + } + if strings.HasPrefix(mime, "video/") { + return "Sent a video" + } + return "Sent a voice message" + case particle.TypeFile: + return "Sent a file" + case particle.TypeQuest: + if v, err := doc.DataAt("properties.title"); err == nil { + if s, ok := v.(string); ok && s != "" { + return "Quest: " + truncatePreview(s, 120) + } + } + return "Added a quest" + case particle.TypePaper: + if v, err := doc.DataAt("properties.title"); err == nil { + if s, ok := v.(string); ok && s != "" { + return "Paper: " + truncatePreview(s, 120) + } + } + return "Added a paper" + default: + return "New activity" + } +} + +func truncatePreview(s string, n int) string { + s = strings.TrimSpace(s) + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/go/internal/handler/handler.go b/go/internal/handler/handler.go index 51a8970..dd565dd 100644 --- a/go/internal/handler/handler.go +++ b/go/internal/handler/handler.go @@ -15,6 +15,7 @@ import ( "github.com/flowy-live/llink/internal/billing" "github.com/flowy-live/llink/internal/depot" "github.com/flowy-live/llink/internal/human" + "github.com/flowy-live/llink/internal/human/pushnotify" "github.com/flowy-live/llink/internal/livekit" "github.com/flowy-live/llink/internal/middleware" "github.com/flowy-live/llink/internal/network" @@ -32,6 +33,7 @@ type Handler struct { depotSvc depot.Service waitlistSvc waitlist.Service billingSvc billing.Service + pushTokenSvc pushnotify.Service livekitClient livekit.Client firestoreClient *firestore.Client } @@ -44,6 +46,7 @@ func NewHandler( depotSvc depot.Service, waitlistSvc waitlist.Service, billingSvc billing.Service, + pushTokenSvc pushnotify.Service, livekitClient livekit.Client, firestoreClient *firestore.Client, ) *Handler { @@ -55,6 +58,7 @@ func NewHandler( depotSvc: depotSvc, waitlistSvc: waitlistSvc, billingSvc: billingSvc, + pushTokenSvc: pushTokenSvc, livekitClient: livekitClient, firestoreClient: firestoreClient, } diff --git a/go/internal/handler/push_token.go b/go/internal/handler/push_token.go new file mode 100644 index 0000000..0c07b1e --- /dev/null +++ b/go/internal/handler/push_token.go @@ -0,0 +1,85 @@ +package handler + +import ( + "encoding/json" + "errors" + "log/slog" + "net/http" + + "github.com/flowy-live/llink/internal/human/pushnotify" + "github.com/flowy-live/llink/internal/middleware" +) + +type RegisterPushTokenRequest struct { + Token string `json:"token"` + Platform string `json:"platform"` + AppVersion string `json:"app_version"` +} + +type UnregisterPushTokenRequest struct { + Token string `json:"token"` +} + +// RegisterPushToken upserts an Expo push token against the authenticated human. +// Re-binding a token to a new human (e.g., after a device-level account switch) +// happens transparently via ON CONFLICT. +func (h *Handler) RegisterPushToken(w http.ResponseWriter, r *http.Request) { + humanId, ok := middleware.HumanIdFromContext(r.Context()) + if !ok { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + var req RegisterPushTokenRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body", http.StatusBadRequest) + return + } + + err := h.pushTokenSvc.Register(r.Context(), humanId, pushnotify.RegisterInput{ + Token: req.Token, + Platform: pushnotify.Platform(req.Platform), + AppVersion: req.AppVersion, + }) + if err != nil { + if errors.Is(err, pushnotify.ErrInvalidPlatform) || errors.Is(err, pushnotify.ErrInvalidToken) { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + slog.Error("failed to register push token", "error", err, "humanId", humanId) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + w.WriteHeader(http.StatusNoContent) +} + +// UnregisterPushToken removes a push token belonging to the authenticated human. +// Returns 204 even if the token wasn't found — idempotent from the client's POV. +func (h *Handler) UnregisterPushToken(w http.ResponseWriter, r *http.Request) { + humanId, ok := middleware.HumanIdFromContext(r.Context()) + if !ok { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + + var req UnregisterPushTokenRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "invalid request body", http.StatusBadRequest) + return + } + + if req.Token == "" { + http.Error(w, "token is required", http.StatusBadRequest) + return + } + + err := h.pushTokenSvc.Unregister(r.Context(), humanId, req.Token) + if err != nil && !errors.Is(err, pushnotify.ErrNotFound) { + slog.Error("failed to unregister push token", "error", err, "humanId", humanId) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + w.WriteHeader(http.StatusNoContent) +} diff --git a/go/internal/human/pushnotify/expo.go b/go/internal/human/pushnotify/expo.go new file mode 100644 index 0000000..546f7b6 --- /dev/null +++ b/go/internal/human/pushnotify/expo.go @@ -0,0 +1,140 @@ +package pushnotify + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +const ( + expoPushAPIURL = "https://exp.host/--/api/v2/push/send" + // expoMaxBatchSize is the documented per-request cap on push messages. + expoMaxBatchSize = 100 + + // Ticket error codes returned by Expo Push API. The only one we act on is + // DeviceNotRegistered — others are logged but not retried (per product call). + ExpoErrorDeviceNotRegistered = "DeviceNotRegistered" +) + +// Message is one push to one device. Sound defaults to "default" when empty. +type Message struct { + To string `json:"to"` + Title string `json:"title,omitempty"` + Body string `json:"body,omitempty"` + Data map[string]any `json:"data,omitempty"` + Sound string `json:"sound,omitempty"` +} + +// Ticket is the synchronous response Expo returns per message. Status is +// either "ok" or "error". On error, Details["error"] carries the code (e.g. +// "DeviceNotRegistered", "MessageTooBig", "InvalidCredentials"). +type Ticket struct { + Status string `json:"status"` + ID string `json:"id,omitempty"` + Message string `json:"message,omitempty"` + Details map[string]any `json:"details,omitempty"` +} + +// ExpoClient is a minimal HTTP client for the Expo Push API. It does NOT poll +// receipts and does NOT retry. +type ExpoClient struct { + http *http.Client + accessToken string +} + +func NewExpoClient(accessToken string) *ExpoClient { + return &ExpoClient{ + http: &http.Client{Timeout: 15 * time.Second}, + accessToken: accessToken, + } +} + +type expoSendResponse struct { + Data []Ticket `json:"data"` + Errors []map[string]any `json:"errors,omitempty"` +} + +// Send delivers messages in batches of up to expoMaxBatchSize. Returned tickets +// preserve the input order across batches: tickets[i] corresponds to msgs[i]. +// A request-level error (network, 5xx, malformed body) aborts the remaining +// batches and is returned to the caller along with whatever tickets succeeded. +func (c *ExpoClient) Send(ctx context.Context, msgs []Message) ([]Ticket, error) { + if len(msgs) == 0 { + return nil, nil + } + + for i := range msgs { + if msgs[i].Sound == "" { + msgs[i].Sound = "default" + } + } + + tickets := make([]Ticket, 0, len(msgs)) + for start := 0; start < len(msgs); start += expoMaxBatchSize { + end := start + expoMaxBatchSize + if end > len(msgs) { + end = len(msgs) + } + + batch := msgs[start:end] + batchTickets, err := c.sendBatch(ctx, batch) + tickets = append(tickets, batchTickets...) + if err != nil { + return tickets, fmt.Errorf("expo push batch [%d:%d]: %w", start, end, err) + } + } + return tickets, nil +} + +func (c *ExpoClient) sendBatch(ctx context.Context, batch []Message) ([]Ticket, error) { + body, err := json.Marshal(batch) + if err != nil { + return nil, fmt.Errorf("marshal batch: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, expoPushAPIURL, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("build request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + req.Header.Set("Accept-Encoding", "gzip, deflate") + if c.accessToken != "" { + req.Header.Set("Authorization", "Bearer "+c.accessToken) + } + + resp, err := c.http.Do(req) + if err != nil { + return nil, fmt.Errorf("do request: %w", err) + } + defer resp.Body.Close() + + raw, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("read response: %w", err) + } + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("expo push api returned %d: %s", resp.StatusCode, truncate(string(raw), 512)) + } + + var parsed expoSendResponse + if err := json.Unmarshal(raw, &parsed); err != nil { + return nil, fmt.Errorf("decode response: %w", err) + } + if len(parsed.Data) != len(batch) { + return parsed.Data, fmt.Errorf("expo returned %d tickets for %d messages", len(parsed.Data), len(batch)) + } + return parsed.Data, nil +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/go/internal/human/pushnotify/notifier.go b/go/internal/human/pushnotify/notifier.go new file mode 100644 index 0000000..913ff60 --- /dev/null +++ b/go/internal/human/pushnotify/notifier.go @@ -0,0 +1,194 @@ +package pushnotify + +import ( + "context" + "errors" + "fmt" + "log/slog" + + pbpusher "github.com/flowy-live/llink/genproto/llink/pusher" + "github.com/flowy-live/llink/internal/network" + "google.golang.org/grpc" +) + +// PusherClient is the subset of the pusher gRPC service the notifier needs. +// Defined here (rather than depending on the generated client interface) so +// tests can supply a fake without standing up a gRPC server. +type PusherClient interface { + IsOnline(ctx context.Context, in *pbpusher.IsOnlineRequest, opts ...grpc.CallOption) (*pbpusher.IsOnlineResponse, error) +} + +// NotifyInput is everything the notifier needs to know about a single newly-created particle. +type NotifyInput struct { + NetworkID string + SenderHumanID string + SenderEmailPrefix string + + ParticleID string + // One of "text", "media", "file", "quest", "paper". Containers (stream, + // folder) are dropped by the caller before reaching the notifier. + ParticleKind string + + // Parent stream context — drives the title and the recipient set. + StreamID string + StreamName string + StreamVisibleTo []string + + // Body — already formatted by the caller (e.g. truncated text, "Sent a + // voice message"). Title is derived inside the notifier. + Body string +} + +// Notifier orchestrates the per-particle fanout: +// 1. Resolve recipients (stream visibility ∩ network members, minus sender). +// 2. Filter out anyone with an active WebSocket connection. +// 3. Look up each remaining human's push tokens. +// 4. POST a single batched request to Expo. +// 5. Delete any token Expo reports as DeviceNotRegistered. +type Notifier struct { + networkR network.Reader + tokens Service + pusher PusherClient + expo *ExpoClient +} + +func NewNotifier(networkR network.Reader, tokens Service, pusher PusherClient, expo *ExpoClient) *Notifier { + return &Notifier{ + networkR: networkR, + tokens: tokens, + pusher: pusher, + expo: expo, + } +} + +func (n *Notifier) NotifyParticleCreated(ctx context.Context, in NotifyInput) error { + if in.NetworkID == "" || in.ParticleID == "" { + return nil + } + + members, err := n.networkR.ListMembers(ctx, in.NetworkID) + if err != nil { + return fmt.Errorf("list network members: %w", err) + } + + recipients := network.ResolveVisibility(in.StreamVisibleTo, members) + recipients = filterOut(recipients, in.SenderHumanID) + if len(recipients) == 0 { + return nil + } + + online, err := n.queryOnline(ctx, recipients) + if err != nil { + return fmt.Errorf("presence lookup: %w", err) + } + + offline := make([]string, 0, len(recipients)) + for _, id := range recipients { + if !online[id] { + offline = append(offline, id) + } + } + if len(offline) == 0 { + return nil + } + + tokens, err := n.tokens.ListForHumans(ctx, offline) + if err != nil { + return fmt.Errorf("token lookup: %w", err) + } + if len(tokens) == 0 { + return nil + } + + msgs := buildMessages(tokens, in) + tickets, sendErr := n.expo.Send(ctx, msgs) + slog.Info("pushnotify: dispatch", + "networkID", in.NetworkID, + "particleID", in.ParticleID, + "recipients", len(recipients), + "offline", len(offline), + "tokens", len(tokens), + "sent", len(tickets), + ) + + n.cleanupDeadTokens(ctx, msgs, tickets) + + if sendErr != nil { + return fmt.Errorf("expo send: %w", sendErr) + } + return nil +} + +func (n *Notifier) queryOnline(ctx context.Context, humanIDs []string) (map[string]bool, error) { + resp, err := n.pusher.IsOnline(ctx, &pbpusher.IsOnlineRequest{HumanIds: humanIDs}) + if err != nil { + return nil, err + } + return resp.Online, nil +} + +// cleanupDeadTokens DELETEs any token Expo reports as DeviceNotRegistered. +// This is the one feedback signal we honor — other ticket errors (e.g. +// MessageTooBig, RateLimit) are logged but never retried. +func (n *Notifier) cleanupDeadTokens(ctx context.Context, msgs []Message, tickets []Ticket) { + for i, t := range tickets { + if i >= len(msgs) { + break + } + if t.Status != "error" || t.Details == nil { + continue + } + code, _ := t.Details["error"].(string) + if code != ExpoErrorDeviceNotRegistered { + if t.Status == "error" { + slog.Warn("pushnotify: ticket error", "code", code, "message", t.Message, "to", msgs[i].To) + } + continue + } + if err := n.tokens.DeleteByToken(ctx, msgs[i].To); err != nil && !errors.Is(err, ErrNotFound) { + slog.Error("pushnotify: failed to delete dead token", "error", err, "token", msgs[i].To) + } else { + slog.Info("pushnotify: removed unregistered token", "token", msgs[i].To) + } + } +} + +func buildMessages(tokens []*PushToken, in NotifyInput) []Message { + title := in.SenderEmailPrefix + if in.StreamName != "" { + title = in.SenderEmailPrefix + " in " + in.StreamName + } + + data := map[string]any{ + "kind": "particle_created", + "network_id": in.NetworkID, + "stream_id": in.StreamID, + "particle_id": in.ParticleID, + "sender_human_id": in.SenderHumanID, + "particle_kind": in.ParticleKind, + } + + msgs := make([]Message, 0, len(tokens)) + for _, t := range tokens { + msgs = append(msgs, Message{ + To: t.Token, + Title: title, + Body: in.Body, + Data: data, + }) + } + return msgs +} + +func filterOut(ids []string, exclude string) []string { + if exclude == "" { + return ids + } + out := ids[:0:len(ids)] + for _, id := range ids { + if id != exclude { + out = append(out, id) + } + } + return out +} diff --git a/go/internal/human/pushnotify/repository.go b/go/internal/human/pushnotify/repository.go new file mode 100644 index 0000000..7f24bba --- /dev/null +++ b/go/internal/human/pushnotify/repository.go @@ -0,0 +1,95 @@ +package pushnotify + +import ( + "context" + "errors" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type repository interface { + upsert(ctx context.Context, t *PushToken) error + deleteForHuman(ctx context.Context, humanID, token string) error + deleteByToken(ctx context.Context, token string) error + listForHumans(ctx context.Context, humanIDs []string) ([]*PushToken, error) +} + +type repositoryImpl struct { + pool *pgxpool.Pool +} + +func newRepository(pool *pgxpool.Pool) repository { + return &repositoryImpl{pool: pool} +} + +func (r *repositoryImpl) upsert(ctx context.Context, t *PushToken) error { + _, err := r.pool.Exec(ctx, + `INSERT INTO push_tokens (token, human_id, platform, app_version) + VALUES ($1, $2, $3, NULLIF($4, '')) + ON CONFLICT (token) DO UPDATE SET + human_id = EXCLUDED.human_id, + platform = EXCLUDED.platform, + app_version = EXCLUDED.app_version, + last_seen_at = NOW()`, + t.Token, t.HumanID, string(t.Platform), t.AppVersion, + ) + return err +} + +func (r *repositoryImpl) deleteForHuman(ctx context.Context, humanID, token string) error { + result, err := r.pool.Exec(ctx, + `DELETE FROM push_tokens WHERE human_id = $1 AND token = $2`, + humanID, token, + ) + if err != nil { + return err + } + if result.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +func (r *repositoryImpl) deleteByToken(ctx context.Context, token string) error { + _, err := r.pool.Exec(ctx, + `DELETE FROM push_tokens WHERE token = $1`, + token, + ) + return err +} + +func (r *repositoryImpl) listForHumans(ctx context.Context, humanIDs []string) ([]*PushToken, error) { + if len(humanIDs) == 0 { + return nil, nil + } + rows, err := r.pool.Query(ctx, + `SELECT token, human_id, platform, app_version, created_at, last_seen_at + FROM push_tokens + WHERE human_id = ANY($1)`, + humanIDs, + ) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, err + } + defer rows.Close() + + var tokens []*PushToken + for rows.Next() { + var t PushToken + var appVersion *string + var platform string + if err := rows.Scan(&t.Token, &t.HumanID, &platform, &appVersion, &t.CreatedAt, &t.LastSeenAt); err != nil { + return nil, err + } + t.Platform = Platform(platform) + if appVersion != nil { + t.AppVersion = *appVersion + } + tokens = append(tokens, &t) + } + return tokens, rows.Err() +} diff --git a/go/internal/human/pushnotify/service.go b/go/internal/human/pushnotify/service.go new file mode 100644 index 0000000..3f7f2e0 --- /dev/null +++ b/go/internal/human/pushnotify/service.go @@ -0,0 +1,70 @@ +package pushnotify + +import ( + "context" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Service is the full surface for per-device push token storage. HTTP handlers +// use Register/Unregister; the worker's notifier uses ListForHumans and +// DeleteByToken. Both consumers share the same underlying repository. +type Service interface { + // Register upserts a token for the given human. Returns ErrInvalidToken / + // ErrInvalidPlatform on bad input. + Register(ctx context.Context, humanID string, in RegisterInput) error + // Unregister removes a token, scoped to the calling human so a user can't + // delete another user's token. Returns ErrNotFound if the token doesn't + // belong to humanID (or doesn't exist). + Unregister(ctx context.Context, humanID, token string) error + // ListForHumans returns every push token belonging to any of the given + // human IDs. Returns an empty slice when nothing matches. + ListForHumans(ctx context.Context, humanIDs []string) ([]*PushToken, error) + // DeleteByToken removes a token regardless of owning human. Used by the + // notifier to clean up after Expo returns DeviceNotRegistered. + DeleteByToken(ctx context.Context, token string) error +} + +type RegisterInput struct { + Token string + Platform Platform + AppVersion string +} + +type serviceImpl struct { + repo repository +} + +func NewService(pool *pgxpool.Pool) Service { + return &serviceImpl{repo: newRepository(pool)} +} + +func (s *serviceImpl) Register(ctx context.Context, humanID string, in RegisterInput) error { + if !in.Platform.Valid() { + return ErrInvalidPlatform + } + if !IsValidExpoToken(in.Token) { + return ErrInvalidToken + } + return s.repo.upsert(ctx, &PushToken{ + Token: in.Token, + HumanID: humanID, + Platform: in.Platform, + AppVersion: in.AppVersion, + }) +} + +func (s *serviceImpl) Unregister(ctx context.Context, humanID, token string) error { + if token == "" { + return ErrInvalidToken + } + return s.repo.deleteForHuman(ctx, humanID, token) +} + +func (s *serviceImpl) ListForHumans(ctx context.Context, humanIDs []string) ([]*PushToken, error) { + return s.repo.listForHumans(ctx, humanIDs) +} + +func (s *serviceImpl) DeleteByToken(ctx context.Context, token string) error { + return s.repo.deleteByToken(ctx, token) +} diff --git a/go/internal/human/pushnotify/token.go b/go/internal/human/pushnotify/token.go new file mode 100644 index 0000000..8a2d126 --- /dev/null +++ b/go/internal/human/pushnotify/token.go @@ -0,0 +1,43 @@ +// Package pushnotify owns mobile push notification delivery: storage of per-device +// Expo push tokens, and the worker-side orchestration of sending notifications +// to offline recipients via the Expo Push API. +package pushnotify + +import ( + "errors" + "strings" + "time" +) + +type Platform string + +const ( + PlatformIOS Platform = "ios" + PlatformAndroid Platform = "android" +) + +func (p Platform) Valid() bool { + return p == PlatformIOS || p == PlatformAndroid +} + +type PushToken struct { + Token string + HumanID string + Platform Platform + AppVersion string + CreatedAt time.Time + LastSeenAt time.Time +} + +var ( + ErrInvalidPlatform = errors.New("invalid platform") + ErrInvalidToken = errors.New("invalid expo push token") + ErrNotFound = errors.New("push token not found") +) + +// IsValidExpoToken matches the two prefix formats Expo currently uses. +// We don't validate the inner contents — Expo's server will reject malformed +// tokens with a per-message error and we'll clean those up via DeviceNotRegistered. +func IsValidExpoToken(token string) bool { + return strings.HasPrefix(token, "ExponentPushToken[") || strings.HasPrefix(token, "ExpoPushToken[") +} diff --git a/go/internal/network/reader.go b/go/internal/network/reader.go index d886533..9cfceec 100644 --- a/go/internal/network/reader.go +++ b/go/internal/network/reader.go @@ -16,6 +16,9 @@ type Reader interface { ListForHuman(ctx context.Context, humanId string) ([]*Network, error) // IsMember returns ErrInvalidHumanId if humanId is empty. IsMember(ctx context.Context, networkID, humanId string) (bool, error) + // ListMembers returns all human IDs belonging to the network. + // Returns an empty slice if the network doesn't exist. + ListMembers(ctx context.Context, networkID string) ([]string, error) // ListAll returns all networks with their members ListAll(ctx context.Context) ([]*Network, error) // ListAllMemberships returns humanId -> networkIds for every human with at @@ -69,6 +72,12 @@ func (r *readerImpl) IsMember(ctx context.Context, networkID, humanId string) (b return r.repo.isMember(ctx, networkID, humanId) } +func (r *readerImpl) ListMembers(ctx context.Context, networkID string) ([]string, error) { + // Admin is guaranteed to be in network_members: Create() calls AddMembers + // for the admin, and RemoveMemberFromNetwork rejects admin removal. + return r.repo.getMemberHumanIds(ctx, networkID) +} + func (r *readerImpl) ListAll(ctx context.Context) ([]*Network, error) { return r.repo.listAll(ctx) } diff --git a/go/internal/network/visibility.go b/go/internal/network/visibility.go new file mode 100644 index 0000000..d5c172b --- /dev/null +++ b/go/internal/network/visibility.go @@ -0,0 +1,41 @@ +package network + +import "strings" + +// ResolveVisibility expands a stream particle's visible_to entries into the set +// of human IDs that should see (and thus be notified about) activity in that +// stream. Entries are formatted as `human:{id}` for a specific human or +// `network:{id}` to expand to every member of the surrounding network. +// +// networkMembers must contain every human currently in the network (members + +// admin). visible_to entries that point to humans no longer in the network are +// dropped — they may have been removed since the stream was created. +// +// Returns a deduped slice; ordering is not stable. +func ResolveVisibility(visibleTo []string, networkMembers []string) []string { + memberSet := make(map[string]bool, len(networkMembers)) + for _, id := range networkMembers { + memberSet[id] = true + } + + result := make(map[string]bool) + for _, entry := range visibleTo { + switch { + case strings.HasPrefix(entry, "human:"): + id := strings.TrimPrefix(entry, "human:") + if memberSet[id] { + result[id] = true + } + case strings.HasPrefix(entry, "network:"): + for id := range memberSet { + result[id] = true + } + } + } + + out := make([]string, 0, len(result)) + for id := range result { + out = append(out, id) + } + return out +} diff --git a/go/k8s/dev/particleprocessorworker.yaml b/go/k8s/dev/particleprocessorworker.yaml index 7a69497..ccb4065 100644 --- a/go/k8s/dev/particleprocessorworker.yaml +++ b/go/k8s/dev/particleprocessorworker.yaml @@ -43,3 +43,5 @@ spec: secretKeyRef: name: shared-secrets key: DEEPGRAM_SECRET + - name: "PUSHER_GRPC_ADDR" + value: "pusher:50051" diff --git a/go/k8s/prod/particleprocessorworker.yaml b/go/k8s/prod/particleprocessorworker.yaml index ca6f3e0..eb6bae7 100644 --- a/go/k8s/prod/particleprocessorworker.yaml +++ b/go/k8s/prod/particleprocessorworker.yaml @@ -40,3 +40,5 @@ spec: secretKeyRef: name: shared-secrets key: DEEPGRAM_SECRET + - name: "PUSHER_GRPC_ADDR" + value: "pusher:50051" diff --git a/go/migrations/000016_push_tokens.down.sql b/go/migrations/000016_push_tokens.down.sql new file mode 100644 index 0000000..a159f1c --- /dev/null +++ b/go/migrations/000016_push_tokens.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS push_tokens; diff --git a/go/migrations/000016_push_tokens.up.sql b/go/migrations/000016_push_tokens.up.sql new file mode 100644 index 0000000..4b0c4ca --- /dev/null +++ b/go/migrations/000016_push_tokens.up.sql @@ -0,0 +1,10 @@ +CREATE TABLE push_tokens ( + token TEXT PRIMARY KEY, + human_id TEXT NOT NULL REFERENCES humans(id) ON DELETE CASCADE, + platform TEXT NOT NULL CHECK (platform IN ('ios', 'android')), + app_version TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX push_tokens_human_id_idx ON push_tokens (human_id);