082cac36b2
Wire automation dispatcher to IMAP sync, drive mutations, and contact CRUD. Add webhook event_types and mail/drive/contacts scope filters (migration 30).
135 lines
3.4 KiB
Go
135 lines
3.4 KiB
Go
package imap
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log/slog"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
"github.com/ultisuite/ulti-backend/internal/mail/rules"
|
|
"github.com/ultisuite/ulti-backend/internal/realtime"
|
|
)
|
|
|
|
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
|
|
}
|
|
|
|
func newSyncPipeline(db *pgxpool.Pool, rulesEngine *rules.Engine, automation MailAutomation, hub *realtime.Hub) *syncPipeline {
|
|
return &syncPipeline{
|
|
db: db,
|
|
logger: slog.Default().With("component", "imap-pipeline"),
|
|
rules: rulesEngine,
|
|
automation: automation,
|
|
hub: hub,
|
|
}
|
|
}
|
|
|
|
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.broadcast(ev, event)
|
|
}
|
|
|
|
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
|
|
}
|