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 }