diff --git a/go/cmd/particleprocessorworker/main.go b/go/cmd/particleprocessorworker/main.go index a9a3e97..51e5da0 100644 --- a/go/cmd/particleprocessorworker/main.go +++ b/go/cmd/particleprocessorworker/main.go @@ -9,7 +9,6 @@ import ( "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" @@ -19,9 +18,7 @@ import ( "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" @@ -64,22 +61,13 @@ 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: Enhanced Security is on for our Expo // project (otherwise anyone holding one of our push tokens could spam users). expoClient := pushnotify.NewExpoClient(utils.MustGetEnv("EXPO_ACCESS_TOKEN")) - notifier := pushnotify.NewNotifier(networkReader, pushTokenSvc, pusherClient, expoClient) + notifier := pushnotify.NewNotifier(networkReader, pushTokenSvc, expoClient) client := createClient(ctx) defer client.Close() @@ -327,6 +315,7 @@ func notifyForParticle( transcript string, ) { if parent == nil { + slog.Info("notify: skip — no parent", "particleID", doc.Ref.ID) return } @@ -334,9 +323,13 @@ func notifyForParticle( typeName, _ := typeStr.(string) pType, err := particle.ParseParticleType(typeName) if err != nil { + slog.Info("notify: skip — unparseable particle type", + "particleID", doc.Ref.ID, "type", typeName, "error", err) return } if pType == particle.TypeStream || pType == particle.TypeFolder { + slog.Info("notify: skip — container particle", + "particleID", doc.Ref.ID, "type", pType) return } @@ -347,6 +340,8 @@ func notifyForParticle( } parentType, err := particle.ParseParticleType(parentStream.Type) if err != nil || parentType != particle.TypeStream { + slog.Info("notify: skip — parent isn't a stream", + "particleID", doc.Ref.ID, "parentType", parentType, "parseErr", err) return } diff --git a/go/internal/human/pushnotify/notifier.go b/go/internal/human/pushnotify/notifier.go index a06c59a..ee5be37 100644 --- a/go/internal/human/pushnotify/notifier.go +++ b/go/internal/human/pushnotify/notifier.go @@ -6,17 +6,9 @@ import ( "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 a narrow subset of the pusher gRPC service so tests can -// supply a fake without standing up a real server. -type PusherClient interface { - IsOnline(ctx context.Context, in *pbpusher.IsOnlineRequest, opts ...grpc.CallOption) (*pbpusher.IsOnlineResponse, error) -} - type NotifyInput struct { NetworkID string SenderHumanID string @@ -39,27 +31,34 @@ type NotifyInput struct { // Notifier fans out one particle to Expo: // 1. Resolve recipients (visibility ∩ network members, minus sender). -// 2. Drop anyone currently connected via WebSocket. -// 3. Send a batched Expo request for the remainder's tokens. -// 4. Prune tokens Expo reports as DeviceNotRegistered. +// 2. Send a batched Expo request for every recipient's tokens. +// 3. Prune tokens Expo reports as DeviceNotRegistered. +// +// Online/offline presence is intentionally NOT consulted: a live WebSocket +// is a poor proxy for "user is actively consuming this particle right now" +// (backgrounded apps, idle desktops, etc. all look online), and the resulting +// false-negatives outweigh the duplicate-notification cost on a focused +// device, which the OS handles via Focus modes and per-app settings. type Notifier struct { networkR network.Reader tokens Service - pusher PusherClient expo *ExpoClient } -func NewNotifier(networkR network.Reader, tokens Service, pusher PusherClient, expo *ExpoClient) *Notifier { +func NewNotifier(networkR network.Reader, tokens Service, 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 == "" { + slog.Info("pushnotify: skip — missing ids", + "networkID", in.NetworkID, + "particleID", in.ParticleID, + ) return nil } @@ -69,31 +68,31 @@ func (n *Notifier) NotifyParticleCreated(ctx context.Context, in NotifyInput) er } recipients := network.ResolveVisibility(in.StreamVisibleTo, members) + recipientsBeforeSenderFilter := len(recipients) recipients = filterOut(recipients, in.SenderHumanID) if len(recipients) == 0 { + slog.Info("pushnotify: skip — no recipients", + "networkID", in.NetworkID, + "particleID", in.ParticleID, + "senderHumanID", in.SenderHumanID, + "members", len(members), + "visibleTo", in.StreamVisibleTo, + "resolved", recipientsBeforeSenderFilter, + ) 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) + tokens, err := n.tokens.ListForHumans(ctx, recipients) if err != nil { return fmt.Errorf("token lookup: %w", err) } if len(tokens) == 0 { + slog.Info("pushnotify: skip — no tokens for recipients", + "networkID", in.NetworkID, + "particleID", in.ParticleID, + "recipients", len(recipients), + "recipientIDs", recipients, + ) return nil } @@ -103,7 +102,6 @@ func (n *Notifier) NotifyParticleCreated(ctx context.Context, in NotifyInput) er "networkID", in.NetworkID, "particleID", in.ParticleID, "recipients", len(recipients), - "offline", len(offline), "tokens", len(tokens), "sent", len(tickets), ) @@ -116,14 +114,6 @@ func (n *Notifier) NotifyParticleCreated(ctx context.Context, in NotifyInput) er 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 -} - // DeviceNotRegistered is the one feedback signal we honor; other ticket // errors (MessageTooBig, RateLimit, …) are logged and dropped. func (n *Notifier) cleanupDeadTokens(ctx context.Context, msgs []Message, tickets []Ticket) {