package main
import (
"context"
"fmt"
"os"
"time"
"github.com/flowy-live/llink/internal/utils/flog"
"cloud.google.com/go/firestore"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
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"
)
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 {
flog.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 {
flog.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 {
flog.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())
flog.Info("starting email notification cycle")
if err := runNotificationCycle(ctx, firestoreClient, aeroSvc, pusherSvc, humanSvc, networkSvc); err != nil {
flog.Error("notification cycle failed", "error", err)
os.Exit(1)
}
flog.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)
}
flog.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 {
flog.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] {
flog.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 {
flog.Error("failed to send email", "humanId", humanId, "error", err)
continue
}
if err := humanSvc.UpdateLastEmailNotificationSentAt(ctx, humanId, now); err != nil {
flog.Error("failed to update last_email_notification_sent_at", "humanId", humanId, "error", err)
}
sentCount++
}
flog.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 {
flog.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(`
Hi %s,
You have unread messages in %d %s on Flowy.llink.
Open the app to catch up with your team.
Open Flowy.llink
Best,
Flowy Team
Note: disable email notifications from in-app settings.
`, name, count, streamsWord)
}