llink/go/cmd/emailnotifierjob/main.go

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"time"

	"cloud.google.com/go/firestore"
	pbaero "github.com/flowy-live/llink/genproto/aero"
	pbpusher "github.com/flowy-live/llink/genproto/llink/pusher"
	"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/particle"
	"github.com/flowy-live/llink/internal/utils"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

const (
	maxActivityAge  = 24 * time.Hour  // ignore streams idle longer than this
	unreadThreshold = 10 * time.Minute // grace window before a message is "unread"
	emailCooldown   = 12 * time.Hour   // min gap between emails to the same user
)

func main() {
	ctx := context.Background()

	db.Init()
	defer db.Cleanup()

	gcpProject := utils.MustGetEnv("GCP_PROJECT")
	firestoreClient, err := firestore.NewClient(ctx, gcpProject)
	if err != nil {
		slog.Error("failed to create Firestore client", "error", err)
		os.Exit(1)
	}
	defer firestoreClient.Close()

	aeroAddr := utils.MustGetEnv("AERO_ADDR")
	aeroConn, err := grpc.NewClient(aeroAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		slog.Error("failed to connect to aero", "error", err)
		os.Exit(1)
	}
	defer aeroConn.Close()
	aeroSvc := pbaero.NewPrimaryClient(aeroConn)

	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()
	pusherSvc := pbpusher.NewPusherServiceClient(pusherConn)

	humanSvc := human.NewService(db.Pool())
	networkSvc := network.NewReader(db.Pool())

	slog.Info("starting email notification cycle")
	if err := runNotificationCycle(ctx, firestoreClient, aeroSvc, pusherSvc, humanSvc, networkSvc); err != nil {
		slog.Error("notification cycle failed", "error", err)
		os.Exit(1)
	}
	slog.Info("email notification cycle complete")
}

func runNotificationCycle(
	ctx context.Context,
	fsClient *firestore.Client,
	aeroSvc pbaero.PrimaryClient,
	pusherSvc pbpusher.PusherServiceClient,
	humanSvc human.Service,
	networkReader network.Reader,
) error {
	now := time.Now()

	networks, err := networkReader.ListAll(ctx)
	if err != nil {
		return fmt.Errorf("listing networks: %w", err)
	}

	allHumans, err := humanSvc.ListAll(ctx)
	if err != nil {
		return fmt.Errorf("listing humans: %w", err)
	}
	humansById := make(map[string]*human.Human, len(allHumans))
	for _, h := range allHumans {
		humansById[h.ID] = h
	}

	behindCounts := map[string]int{}
	latestActivity := map[string]time.Time{}

	allOnline := map[string]bool{}
	onlineResp, err := pusherSvc.GetOnlineHumanIds(ctx, &pbpusher.GetOnlineHumanIdsRequest{})
	if err != nil {
		return fmt.Errorf("failed to get online humans: %w", err)
	}
	slog.Info("gathered online presence", "onlineCount", len(onlineResp.HumanIds), "humanIds", onlineResp.HumanIds)
	for _, id := range onlineResp.HumanIds {
		allOnline[id] = true
	}

	for _, net := range networks {
		streams, err := getOpenStreams(ctx, fsClient, net.ID)
		if err != nil {
			slog.Error("failed to query streams", "networkId", net.ID, "error", err)
			continue
		}

		// net.MemberHumanIds already includes the admin.
		for _, stream := range streams {
			if stream.LastChildCreatedAt == nil {
				continue
			}
			if now.Sub(*stream.LastChildCreatedAt) > maxActivityAge {
				continue
			}
			if now.Sub(*stream.LastChildCreatedAt) < unreadThreshold {
				continue
			}

			for _, humanId := range network.ResolveVisibility(stream.VisibleTo, net.MemberHumanIds) {
				marker, hasMarker := stream.PlaybackMarkers[humanId]
				if hasMarker && !marker.Before(*stream.LastChildCreatedAt) {
					continue
				}
				behindCounts[humanId]++
				if stream.LastChildCreatedAt.After(latestActivity[humanId]) {
					latestActivity[humanId] = *stream.LastChildCreatedAt
				}
			}
		}

	}

	sentCount := 0
	for humanId, count := range behindCounts {
		if allOnline[humanId] {
			slog.Info("human online...skipping email", "humanId", humanId)
			continue
		}

		h, ok := humansById[humanId]
		if !ok {
			continue
		}

		if !h.EmailNotificationsEnabled {
			continue
		}

		// Skip if nothing new since the previous email.
		if h.LastEmailNotificationSentAt != nil && !latestActivity[humanId].After(*h.LastEmailNotificationSentAt) {
			continue
		}

		if h.LastEmailNotificationSentAt != nil && now.Sub(*h.LastEmailNotificationSentAt) < emailCooldown {
			continue
		}

		if err := sendNotificationEmail(ctx, aeroSvc, h, count); err != nil {
			slog.Error("failed to send email", "humanId", humanId, "error", err)
			continue
		}

		if err := humanSvc.UpdateLastEmailNotificationSentAt(ctx, humanId, now); err != nil {
			slog.Error("failed to update last_email_notification_sent_at", "humanId", humanId, "error", err)
		}

		sentCount++
	}

	slog.Info("notification cycle summary",
		"networks", len(networks),
		"humansBehind", len(behindCounts),
		"emailsSent", sentCount,
	)
	return nil
}

func getOpenStreams(ctx context.Context, client *firestore.Client, networkId string) ([]particle.FirestoreStreamParticle, error) {
	collPath := fmt.Sprintf("networks/%s/children", networkId)
	docs, err := client.Collection(collPath).
		Where("type", "==", "stream").
		Where("status", "==", "open").
		Documents(ctx).
		GetAll()
	if err != nil {
		return nil, err
	}

	streams := make([]particle.FirestoreStreamParticle, 0, len(docs))
	for _, doc := range docs {
		var s particle.FirestoreStreamParticle
		if err := doc.DataTo(&s); err != nil {
			slog.Warn("failed to unmarshal stream particle", "docId", doc.Ref.ID, "error", err)
			continue
		}
		streams = append(streams, s)
	}
	return streams, nil
}

func sendNotificationEmail(ctx context.Context, aeroSvc pbaero.PrimaryClient, h *human.Human, streamCount int) error {
	streamsWord := "stream"
	if streamCount != 1 {
		streamsWord = "streams"
	}

	subject := fmt.Sprintf("You have unseen messages in %d %s", streamCount, streamsWord)
	html := buildEmailHTML(h.EmailPrefix, streamCount, streamsWord)

	_, err := aeroSvc.ShootEmail(ctx, &pbaero.ShootEmailRequest{
		ToEmails: []string{h.Email},
		Subject:  subject,
		TemplateData: &pbaero.ShootEmailRequest_SimpleHtmlData{
			SimpleHtmlData: &pbaero.SimpleHtmlData{
				Html: html,
			},
		},
	})
	return err
}

func buildEmailHTML(name string, count int, streamsWord string) string {
	return fmt.Sprintf(`<!DOCTYPE html>
<html>
<body style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; color: #1a1a1a; max-width: 480px; margin: 0 auto; padding: 24px;">
  <p>Hi %s,</p>
  <p>You have unread messages in <strong>%d %s</strong> on Flowy.llink.</p>
  <p>Open the app to catch up with your team.</p>
  <p style="color: #666; font-size: 13px; margin-top: 32px;">
    Best,<br>Flowy Team <br>Note: disable email notifications from in-app settings.
  </p>
</body>
</html>`, name, count, streamsWord)
}