llink/go/internal/human/pushnotify/notifier.go

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 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
	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 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.
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
}

// 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) {
	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
}