feat: send email notifications for missed messages
- New cron job in cluster - Handle presence and other concerns - Toggle in app to disable email notifications - Handles other edge cases such as cooldown period - Simple html email with simple message Closes #117
This commit is contained in:
talksik
@@ -0,0 +1,293 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"strings"
|
||||
"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 (
|
||||
// Only notify about streams with activity in the last 24 hours
|
||||
maxActivityAge = 24 * time.Hour
|
||||
// Minimum time a message must be unread before we consider notifying
|
||||
unreadThreshold = 10 * time.Minute
|
||||
// Minimum time between emails to the same user
|
||||
emailCooldown = 1 * time.Hour
|
||||
)
|
||||
|
||||
func main() {
|
||||
ctx := context.Background()
|
||||
|
||||
// Initialize Postgres
|
||||
db.Init()
|
||||
defer db.Cleanup()
|
||||
|
||||
// Initialize Firestore
|
||||
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()
|
||||
|
||||
// Initialize aero (email) gRPC client
|
||||
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)
|
||||
|
||||
// Initialize pusher gRPC client
|
||||
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)
|
||||
|
||||
// Initialize services
|
||||
humanSvc := human.NewService(db.Pool())
|
||||
networkSvc := network.NewService(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,
|
||||
networkSvc network.Service,
|
||||
) error {
|
||||
now := time.Now()
|
||||
|
||||
// Load all networks
|
||||
networks, err := networkSvc.ListAll(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("listing networks: %w", err)
|
||||
}
|
||||
|
||||
// Load all humans into a lookup map
|
||||
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
|
||||
}
|
||||
|
||||
// Track which streams each human is behind on, and the latest activity across those streams
|
||||
behindCounts := map[string]int{}
|
||||
latestActivity := map[string]time.Time{}
|
||||
|
||||
// Get all currently connected humans (empty channel_ids = all connections)
|
||||
allOnline := map[string]bool{}
|
||||
presenceResp, err := pusherSvc.BulkGetPresence(ctx, &pbpusher.BulkGetPresenceRequest{})
|
||||
if err != nil {
|
||||
slog.Error("failed to get connected humans", "error", err)
|
||||
// Continue without presence data — worst case we send an email to someone who's online
|
||||
} else if ch, ok := presenceResp.Presences["_all"]; ok {
|
||||
for _, id := range ch.HumanIds {
|
||||
allOnline[id] = true
|
||||
}
|
||||
}
|
||||
|
||||
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 {
|
||||
slog.Error("failed to query streams", "networkId", net.ID, "error", err)
|
||||
continue
|
||||
}
|
||||
|
||||
for _, stream := range streams {
|
||||
if stream.LastChildCreatedAt == nil {
|
||||
continue
|
||||
}
|
||||
// Skip streams with no recent activity
|
||||
if now.Sub(*stream.LastChildCreatedAt) > maxActivityAge {
|
||||
continue
|
||||
}
|
||||
// Skip if the latest message is too fresh (within threshold)
|
||||
if now.Sub(*stream.LastChildCreatedAt) < unreadThreshold {
|
||||
continue
|
||||
}
|
||||
|
||||
// Resolve members from visible_to
|
||||
members := resolveMembers(stream.VisibleTo, networkMembers)
|
||||
|
||||
for humanId := range members {
|
||||
marker, hasMarker := stream.PlaybackMarkers[humanId]
|
||||
if hasMarker && !marker.Before(*stream.LastChildCreatedAt) {
|
||||
continue // up to date
|
||||
}
|
||||
// No marker or marker is behind → this human is behind on this stream
|
||||
behindCounts[humanId]++
|
||||
if stream.LastChildCreatedAt.After(latestActivity[humanId]) {
|
||||
latestActivity[humanId] = *stream.LastChildCreatedAt
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// Send notifications
|
||||
sentCount := 0
|
||||
for humanId, count := range behindCounts {
|
||||
// Skip online users
|
||||
if allOnline[humanId] {
|
||||
continue
|
||||
}
|
||||
|
||||
h, ok := humansById[humanId]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
// Skip if notifications disabled
|
||||
if !h.EmailNotificationsEnabled {
|
||||
continue
|
||||
}
|
||||
|
||||
// Skip if no new activity since last notification
|
||||
if h.LastEmailNotificationSentAt != nil && !latestActivity[humanId].After(*h.LastEmailNotificationSentAt) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Enforce cooldown between emails to the same user
|
||||
if h.LastEmailNotificationSentAt != nil && now.Sub(*h.LastEmailNotificationSentAt) < emailCooldown {
|
||||
continue
|
||||
}
|
||||
|
||||
// Send email
|
||||
if err := sendNotificationEmail(ctx, aeroSvc, h, count); err != nil {
|
||||
slog.Error("failed to send email", "humanId", humanId, "error", err)
|
||||
continue
|
||||
}
|
||||
|
||||
// Update last sent timestamp
|
||||
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
|
||||
}
|
||||
|
||||
// getOpenStreams queries Firestore for all open stream particles in a network.
|
||||
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
|
||||
}
|
||||
|
||||
// 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:")
|
||||
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 {
|
||||
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
|
||||
</p>
|
||||
</body>
|
||||
</html>`, name, count, streamsWord)
|
||||
}
|
||||
@@ -122,6 +122,9 @@ func main() {
|
||||
mux.Handle("POST /auth/sign-out", withAuth(h.SignOut))
|
||||
mux.Handle("GET /auth/me", withAuth(h.GetCurrentHuman))
|
||||
|
||||
// Settings
|
||||
mux.Handle("PATCH /humans/me/settings", withAuth(h.UpdateSettings))
|
||||
|
||||
// Networks
|
||||
mux.Handle("POST /networks", withAuth(h.CreateNetwork))
|
||||
mux.Handle("GET /networks", withAuth(h.ListNetworks))
|
||||
|
||||
Reference in New Issue
Block a user