send push notifications regardless of online status

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
 	}
 

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.
+//  2. Send a batched Expo request for every recipient's tokens.
-//  3. Send a batched Expo request for the remainder's tokens.
+//  3. Prune tokens Expo reports as DeviceNotRegistered.
-//  4. 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)
+	tokens, err := n.tokens.ListForHumans(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 {
+		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) {