ultisuite-backend/internal/mail/imap/pipeline.go

package imap

import (
	"context"
	"encoding/json"
	"log/slog"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/ultisuite/ulti-backend/internal/mail/rules"
	"github.com/ultisuite/ulti-backend/internal/realtime"
)

// Pusher fans a mobile push notification out to a user's registered devices on
// a newly received inbox message. Implemented by *push.Dispatcher.
type Pusher interface {
	NotifyNewMail(ctx context.Context, userID, messageID, accountID, sender, subject string)
}

type postSyncEvent struct {
	userID    string
	accountID string
	messageID string
	kind      string // created | updated | deleted
}

// syncPipeline runs rules and realtime notifications after message sync.
type syncPipeline struct {
	db         *pgxpool.Pool
	logger     *slog.Logger
	rules      *rules.Engine
	automation MailAutomation
	hub        *realtime.Hub
	push       Pusher
}

func newSyncPipeline(db *pgxpool.Pool, rulesEngine *rules.Engine, automation MailAutomation, hub *realtime.Hub, pusher Pusher) *syncPipeline {
	return &syncPipeline{
		db:         db,
		logger:     slog.Default().With("component", "imap-pipeline"),
		rules:      rulesEngine,
		automation: automation,
		hub:        hub,
		push:       pusher,
	}
}

func (p *syncPipeline) handle(ctx context.Context, ev postSyncEvent) {
	if ev.kind == "deleted" {
		p.broadcast(ev, realtime.NewMailDeletedEvent(ev.messageID, ev.accountID))
		return
	}

	if ev.kind == "created" {
		msg, err := p.loadRuleMessage(ctx, ev.messageID)
		if err != nil {
			p.logger.Error("load message for rules", "message_id", ev.messageID, "error", err)
		} else if p.automation != nil {
			p.automation.OnMailCreated(ctx, ev.userID, ev.accountID, ev.messageID, msg)
		} else if p.rules != nil {
			if err := p.rules.EvaluateMessage(ctx, ev.userID, msg); err != nil {
				p.logger.Error("rules evaluation failed", "message_id", ev.messageID, "error", err)
			}
		}
	}

	event := realtime.NewMailUpdatedEvent(ev.messageID, ev.accountID)
	if ev.kind == "created" {
		event = realtime.NewMailCreatedEvent(ev.messageID, ev.accountID)
		p.maybePush(ctx, ev)
	}
	p.broadcast(ev, event)
}

// maybePush fires a best-effort mobile push for a newly received inbox message.
// It skips non-inbox folders, already-read messages, and stale backfilled mail
// (older than 24h) to avoid notification storms during initial account sync.
func (p *syncPipeline) maybePush(ctx context.Context, ev postSyncEvent) {
	if p.push == nil || ev.userID == "" {
		return
	}

	var (
		fromJSON   []byte
		subject    string
		date       time.Time
		folderType string
		flags      []string
	)
	err := p.db.QueryRow(ctx, `
		SELECT m.from_addr, m.subject, m.date, COALESCE(mf.folder_type, ''), COALESCE(m.flags, '{}')
		FROM messages m
		LEFT JOIN mail_folders mf ON m.folder_id = mf.id
		WHERE m.id = $1
	`, ev.messageID).Scan(&fromJSON, &subject, &date, &folderType, &flags)
	if err != nil {
		p.logger.Error("load message for push", "message_id", ev.messageID, "error", err)
		return
	}

	if folderType != "inbox" {
		return
	}
	for _, f := range flags {
		if f == `\Seen` {
			return
		}
	}
	if !date.IsZero() && time.Since(date) > 24*time.Hour {
		return
	}

	sender := firstAddressDisplay(fromJSON)
	if subject == "" {
		subject = "(no subject)"
	}

	go p.push.NotifyNewMail(context.Background(), ev.userID, ev.messageID, ev.accountID, sender, subject)
}

func firstAddressDisplay(fromJSON []byte) string {
	var addrs []EmailAddress
	if err := json.Unmarshal(fromJSON, &addrs); err != nil || len(addrs) == 0 {
		return "New mail"
	}
	a := addrs[0]
	if a.Name != "" {
		return a.Name
	}
	if a.Address != "" {
		return a.Address
	}
	return "New mail"
}

func (p *syncPipeline) broadcast(ev postSyncEvent, event realtime.Event) {
	if p.hub == nil || ev.userID == "" {
		return
	}
	p.hub.Broadcast(ev.userID, event)
}

func (p *syncPipeline) loadRuleMessage(ctx context.Context, messageID string) (*rules.Message, error) {
	var (
		fromJSON []byte
		toJSON   []byte
		subject  string
		bodyText string
		hasAtt   bool
		accountID string
		folderID  *string
		labels    []string
	)
	err := p.db.QueryRow(ctx, `
		SELECT from_addr, to_addrs, subject, body_text, has_attachments, account_id, folder_id, labels
		FROM messages WHERE id = $1
	`, messageID).Scan(&fromJSON, &toJSON, &subject, &bodyText, &hasAtt, &accountID, &folderID, &labels)
	if err != nil {
		return nil, err
	}

	from := firstAddressString(fromJSON)
	to := addressListStrings(toJSON)

	msg := &rules.Message{
		ID:             messageID,
		From:           from,
		To:             to,
		Subject:        subject,
		BodyText:       bodyText,
		HasAttachments: hasAtt,
		AccountID:      accountID,
		Labels:         labels,
	}
	if folderID != nil {
		msg.FolderID = *folderID
	}
	return msg, nil
}

func firstAddressString(fromJSON []byte) string {
	var addrs []EmailAddress
	if err := json.Unmarshal(fromJSON, &addrs); err != nil || len(addrs) == 0 {
		return string(fromJSON)
	}
	a := addrs[0]
	if a.Name != "" {
		return a.Name + " <" + a.Address + ">"
	}
	return a.Address
}

func addressListStrings(toJSON []byte) []string {
	var addrs []EmailAddress
	if err := json.Unmarshal(toJSON, &addrs); err != nil {
		return nil
	}
	out := make([]string, 0, len(addrs))
	for _, a := range addrs {
		if a.Address != "" {
			out = append(out, a.Address)
		}
	}
	return out
}