minstrel/internal/notifications/digest.go

package notifications

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

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/jackc/pgx/v5/pgxpool"

	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
	"git.fabledsword.com/bvandeusen/minstrel/internal/mailer"
)

// The email digest (#5346). Nothing is emailed per event. Operator,
// 2026-10-08: "music and stuff coming in should land as a summary email and
// not everytime same for a approvals the emails should be grouped by a
// reasonable amount of time like an hour or so."
//
// - EmailSummary kinds (new music) go out at most once a day, at the summary
//   hour in the user's own timezone, listing everything since the last one.
// - EmailBatch kinds go out a batch window after the first un-emailed item,
//   holding whatever accumulated.
// - A row the user has read is never emailed: they have seen it.

const (
	// digestInterval is how often the digest looks for due groups. It bounds
	// how late an email can be, not how often one is sent.
	digestInterval = 5 * time.Minute
	// maxEmailAge: an item older than this when the digest first gets to it is
	// not news worth an email. It covers a user who adds an address, or an
	// SMTP server set up, after weeks of unread notifications.
	maxEmailAge = 7 * 24 * time.Hour
	// A failed send waits retryBase, doubling with each failure in a row, up
	// to retryMax.
	retryBase = 5 * time.Minute
	retryMax  = 6 * time.Hour
)

const (
	groupBatch   = "batch"
	groupSummary = "summary"
)

func (g EmailGroup) key() string {
	if g == EmailSummary {
		return groupSummary
	}
	return groupBatch
}

// groupState is where one user's group stands; user_notification_email_state.
type groupState struct {
	BatchOpenedAt time.Time
	LastSentAt    time.Time
	Failures      int32
	RetryAfter    time.Time
}

// pendingItem is one unread, un-emailed notification.
type pendingItem struct {
	ID        pgtype.UUID
	Kind      Kind
	Payload   []byte
	CreatedAt time.Time
}

// userPlan is what the digest does for one user on one tick.
type userPlan struct {
	// Skip is stamped emailed without sending: email is off for its kind now,
	// or it is too old to be news.
	Skip []pendingItem

	Batch    []pendingItem
	BatchDue bool
	// BatchOpenedAt is the open batch's start, kept until it is sent. Zero
	// when nothing is pending in the batch group.
	BatchOpenedAt time.Time

	Summary    []pendingItem
	SummaryDue bool
}

// planUser decides, from the clock and stored state alone, what goes out for
// one user. items are oldest first. emailOn is the user's effective email
// channel per kind.
func planUser(now time.Time, cfg EmailSettings, loc *time.Location, items []pendingItem,
	emailOn map[Kind]bool, batch, summary groupState) userPlan {
	var p userPlan
	for _, it := range items {
		switch {
		case !emailOn[it.Kind] || now.Sub(it.CreatedAt) > maxEmailAge:
			p.Skip = append(p.Skip, it)
		case it.Kind.EmailGroup() == EmailSummary:
			p.Summary = append(p.Summary, it)
		default:
			p.Batch = append(p.Batch, it)
		}
	}
	if len(p.Batch) > 0 {
		p.BatchOpenedAt = batch.BatchOpenedAt
		if p.BatchOpenedAt.IsZero() {
			p.BatchOpenedAt = p.Batch[0].CreatedAt
		}
		p.BatchDue = !now.Before(batch.RetryAfter) && !now.Before(p.BatchOpenedAt.Add(cfg.BatchWindow()))
	}
	if len(p.Summary) > 0 {
		slot := summarySlot(now, loc, int(cfg.SummaryHour))
		p.SummaryDue = !now.Before(summary.RetryAfter) && !now.Before(slot) && summary.LastSentAt.Before(slot)
	}
	return p
}

// summarySlot is today's summary time in loc: the summary hour on the user's
// local calendar day. Across a DST change the hour keeps its local meaning,
// so the instant moves by the shift. An hour that does not exist that day
// (inside a spring-forward gap) is normalised by time.Date to one that does.
func summarySlot(now time.Time, loc *time.Location, hour int) time.Time {
	l := now.In(loc)
	return time.Date(l.Year(), l.Month(), l.Day(), hour, 0, 0, 0, loc)
}

// retryDelay is the wait after the failures-th failure in a row.
func retryDelay(failures int32) time.Duration {
	d := retryBase
	for i := int32(1); i < failures && d < retryMax; i++ {
		d *= 2
	}
	return min(d, retryMax)
}

// Digest sends the grouped emails. One per process, on a ticker.
type Digest struct {
	pool      *pgxpool.Pool
	sender    mailer.Sender
	logger    *slog.Logger
	publicURL func(context.Context) string
}

// NewDigest returns a Digest sending through sender. Links in its emails are
// built from the public address in admin Settings, never from a request.
func NewDigest(pool *pgxpool.Pool, sender mailer.Sender, logger *slog.Logger) *Digest {
	if logger == nil {
		logger = slog.Default()
	}
	d := &Digest{pool: pool, sender: sender, logger: logger}
	d.publicURL = func(ctx context.Context) string {
		row, err := dbq.New(pool).GetNetworkSettings(ctx)
		if err != nil {
			return ""
		}
		return row.PublicUrl
	}
	return d
}

// Run blocks until ctx is cancelled. It ticks once at startup, so a process
// that has been down past a batch or a summary hour catches up, then every
// few minutes.
func (d *Digest) Run(ctx context.Context) {
	d.Tick(ctx, time.Now())
	t := time.NewTicker(digestInterval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case now := <-t.C:
			d.Tick(ctx, now)
		}
	}
}

// Tick sends every group that is due as of now. Failures are logged and
// retried on a later tick; nothing here is fatal.
func (d *Digest) Tick(ctx context.Context, now time.Time) {
	q := dbq.New(d.pool)
	cfg, err := LoadEmailSettings(ctx, q)
	if err != nil {
		d.logger.Warn("notification digest: using default settings", "err", err)
	}
	rows, err := q.ListEmailPendingNotifications(ctx)
	if err != nil {
		d.logger.Warn("notification digest: list pending", "err", err)
		return
	}
	states, err := d.loadStates(ctx, q)
	if err != nil {
		d.logger.Warn("notification digest: load state", "err", err)
		return
	}
	base := d.publicURL(ctx)

	handled := map[pgtype.UUID]bool{}
	for start := 0; start < len(rows); {
		end := start + 1
		for end < len(rows) && rows[end].UserID == rows[start].UserID {
			end++
		}
		u := rows[start]
		handled[u.UserID] = true
		d.digestUser(ctx, q, now, cfg, base, rows[start:end], states[u.UserID])
		start = end
	}

	// A user with nothing pending at all whose batch is still open read
	// everything before it came due. Close it, so their next item opens a
	// fresh batch rather than going out at once.
	for userID, st := range states {
		if !handled[userID] && !st[groupBatch].BatchOpenedAt.IsZero() {
			closed := st[groupBatch]
			closed.BatchOpenedAt = time.Time{}
			d.saveState(ctx, q, userID, groupBatch, closed)
		}
	}
}

// digestUser handles one user's pending rows.
func (d *Digest) digestUser(ctx context.Context, q *dbq.Queries, now time.Time, cfg EmailSettings, base string,
	rows []dbq.ListEmailPendingNotificationsRow, st map[string]groupState) {
	u := rows[0]
	emailOn, err := emailChannels(ctx, q, u.UserID)
	if err != nil {
		d.logger.Warn("notification digest: read prefs", "err", err)
		return
	}
	items := make([]pendingItem, len(rows))
	for i, r := range rows {
		items[i] = pendingItem{ID: r.ID, Kind: Kind(r.Kind), Payload: r.Payload, CreatedAt: r.CreatedAt.Time}
	}
	plan := planUser(now, cfg, userLocation(u.Timezone), items, emailOn, st[groupBatch], st[groupSummary])
	readAsOf := u.ReadAsOf.Time

	if len(plan.Skip) > 0 {
		if _, err := q.MarkNotificationsEmailed(ctx, dbq.MarkNotificationsEmailedParams{
			Ids: itemIDs(plan.Skip), ReadAsOf: ts(readAsOf),
		}); err != nil {
			d.logger.Warn("notification digest: stamp skipped", "err", err)
		}
	}

	// The batch's start is recorded before anything is sent, so a failed send
	// below keeps it along with its retry. With nothing left in the batch
	// group (all read, or skipped), an open batch is closed.
	batchSt := st[groupBatch]
	if !plan.BatchOpenedAt.Equal(batchSt.BatchOpenedAt) {
		batchSt.BatchOpenedAt = plan.BatchOpenedAt
		d.saveState(ctx, q, u.UserID, groupBatch, batchSt)
	}

	to := recipient{email: u.Email, name: displayName(u)}
	if plan.SummaryDue {
		d.send(ctx, q, now, u.UserID, to, EmailSummary, plan.Summary, st[groupSummary], base, readAsOf)
	}
	if plan.BatchDue {
		d.send(ctx, q, now, u.UserID, to, EmailBatch, plan.Batch, batchSt, base, readAsOf)
	}
}

type recipient struct{ email, name string }

// send renders and sends one group's email. Only once the mailer has accepted
// it are its rows stamped and the group's state reset, in one transaction, so
// a failure sends again later and a success never sends twice.
func (d *Digest) send(ctx context.Context, q *dbq.Queries, now time.Time, userID pgtype.UUID, to recipient,
	group EmailGroup, items []pendingItem, st groupState, base string, readAsOf time.Time) {
	msg, err := renderDigest(group, to.name, items, base)
	if err != nil {
		d.logger.Error("notification digest: render", "group", group.key(), "err", err)
		return
	}
	err = d.sender.Send(ctx, to.email, msg.Subject, msg.Text, msg.HTML)
	if errors.Is(err, mailer.ErrNotConfigured) {
		// Not a failure to back off from: nothing can go until SMTP is set
		// up, and the rows wait, unread and un-emailed, until it is.
		return
	}
	if err != nil {
		st.Failures++
		st.RetryAfter = now.Add(retryDelay(st.Failures))
		d.logger.Warn("notification digest: send failed",
			"group", group.key(), "failures", st.Failures, "retry_after", st.RetryAfter, "err", err)
		d.saveState(ctx, q, userID, group.key(), st)
		return
	}

	sent := groupState{LastSentAt: now}
	if err := pgx.BeginFunc(ctx, d.pool, func(tx pgx.Tx) error {
		tq := dbq.New(tx)
		if _, err := tq.MarkNotificationsEmailed(ctx, dbq.MarkNotificationsEmailedParams{
			Ids: itemIDs(items), ReadAsOf: ts(readAsOf),
		}); err != nil {
			return err
		}
		return tq.UpsertNotificationEmailState(ctx, stateParams(userID, group.key(), sent))
	}); err != nil {
		// The email is out but its rows are not stamped, so the next due
		// group would repeat them. Logged at ERROR: it needs a look.
		d.logger.Error("notification digest: sent but not recorded", "group", group.key(), "err", err)
	}
}

func (d *Digest) loadStates(ctx context.Context, q *dbq.Queries) (map[pgtype.UUID]map[string]groupState, error) {
	rows, err := q.ListNotificationEmailState(ctx)
	if err != nil {
		return nil, err
	}
	out := make(map[pgtype.UUID]map[string]groupState, len(rows))
	for _, r := range rows {
		if out[r.UserID] == nil {
			out[r.UserID] = map[string]groupState{}
		}
		out[r.UserID][r.EmailGroup] = groupState{
			BatchOpenedAt: tsTime(r.BatchOpenedAt),
			LastSentAt:    tsTime(r.LastSentAt),
			Failures:      r.Failures,
			RetryAfter:    tsTime(r.RetryAfter),
		}
	}
	return out, nil
}

func (d *Digest) saveState(ctx context.Context, q *dbq.Queries, userID pgtype.UUID, group string, st groupState) {
	if err := q.UpsertNotificationEmailState(ctx, stateParams(userID, group, st)); err != nil {
		d.logger.Warn("notification digest: save state", "group", group, "err", err)
	}
}

// emailChannels is the user's effective email channel for every kind: their
// stored preference, or the kind's default.
func emailChannels(ctx context.Context, q *dbq.Queries, userID pgtype.UUID) (map[Kind]bool, error) {
	rows, err := q.ListNotificationPrefsForUser(ctx, userID)
	if err != nil {
		return nil, fmt.Errorf("prefs: %w", err)
	}
	out := make(map[Kind]bool, len(order))
	for _, k := range order {
		out[k] = k.Defaults().Effective().Email
	}
	for _, r := range rows {
		out[Kind(r.Kind)] = Channels{Inbox: r.Inbox, Phone: r.Phone, Email: r.Email}.Effective().Email
	}
	return out, nil
}

// userLocation is the user's timezone, or UTC when it does not parse.
func userLocation(tz string) *time.Location {
	if loc, err := time.LoadLocation(tz); err == nil && tz != "" {
		return loc
	}
	return time.UTC
}

func displayName(u dbq.ListEmailPendingNotificationsRow) string {
	if u.DisplayName != nil && *u.DisplayName != "" {
		return *u.DisplayName
	}
	return u.Username
}

func itemIDs(items []pendingItem) []pgtype.UUID {
	out := make([]pgtype.UUID, len(items))
	for i, it := range items {
		out[i] = it.ID
	}
	return out
}

func stateParams(userID pgtype.UUID, group string, st groupState) dbq.UpsertNotificationEmailStateParams {
	return dbq.UpsertNotificationEmailStateParams{
		UserID:        userID,
		EmailGroup:    group,
		BatchOpenedAt: ts(st.BatchOpenedAt),
		LastSentAt:    ts(st.LastSentAt),
		Failures:      st.Failures,
		RetryAfter:    ts(st.RetryAfter),
	}
}

func ts(t time.Time) pgtype.Timestamptz { return pgtype.Timestamptz{Time: t, Valid: !t.IsZero()} }

func tsTime(t pgtype.Timestamptz) time.Time {
	if !t.Valid {
		return time.Time{}
	}
	return t.Time
}