zxyszx
468 lines
16 KiB
Go
468 lines
16 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/base64"
|
|
"errors"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
sendQueueStatusQueued = "queued"
|
|
sendQueueStatusSending = "sending"
|
|
sendQueueStatusDelivered = "delivered"
|
|
sendQueueStatusFailed = "failed"
|
|
sendQueueStatusCanceled = "canceled"
|
|
|
|
sendAuditAccepted = "accepted"
|
|
sendAuditQueued = "queued"
|
|
sendAuditDelivered = "delivered"
|
|
sendAuditFailed = "failed"
|
|
sendAuditRetry = "retry"
|
|
sendAuditCanceled = "canceled"
|
|
|
|
sendSourceWebmail = "webmail"
|
|
sendSourceSubmission = "submission"
|
|
sendSourceOpenAPI = "open_api"
|
|
sendSourceForwarding = "forwarding"
|
|
sendSourceRuleForwarding = "rule_forwarding"
|
|
sendSourceForwardingVerification = "forwarding_verification"
|
|
|
|
sendQueueStaleAfter = 15 * time.Minute
|
|
sendQueueConcurrency = 4
|
|
|
|
sendQueueDeliveredMarkerDir = "send_queue_delivered"
|
|
)
|
|
|
|
type sendQueueInput struct {
|
|
UserID string
|
|
MailboxID string
|
|
SentMessageID string
|
|
MessageID string
|
|
Source string
|
|
MailFrom string
|
|
HeaderFrom string
|
|
Recipients []string
|
|
MIMEBytes []byte
|
|
Now time.Time
|
|
}
|
|
|
|
type sendQueueItem struct {
|
|
ID string
|
|
UserID string
|
|
MailboxID string
|
|
SentMessageID string
|
|
MessageID string
|
|
Source string
|
|
MailFrom string
|
|
HeaderFrom string
|
|
Recipients []string
|
|
MIMEBytes []byte
|
|
AttemptCount int
|
|
MaxAttempts int
|
|
}
|
|
|
|
func (a *App) enqueueSend(ctx context.Context, in sendQueueInput) (string, error) {
|
|
if strings.TrimSpace(a.cfg.SMTPHost) == "" {
|
|
return "", nil
|
|
}
|
|
now := in.Now.UTC()
|
|
if now.IsZero() {
|
|
now = a.now().UTC()
|
|
}
|
|
id := newID("snd")
|
|
messageID := strings.TrimSpace(in.MessageID)
|
|
mimeBase64 := base64.StdEncoding.EncodeToString(in.MIMEBytes)
|
|
recipientsJSON := jsonEncode(dedupeEmails(in.Recipients))
|
|
_, err := a.db.ExecContext(ctx, `INSERT OR IGNORE INTO send_queue(id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,status,next_attempt_at,created_at,updated_at)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
|
id, in.UserID, in.MailboxID, in.SentMessageID, messageID, in.Source, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), recipientsJSON, mimeBase64, sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano))
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if strings.TrimSpace(messageID) != "" {
|
|
var existingID, status string
|
|
var attemptCount, maxAttempts int
|
|
if err := a.db.QueryRowContext(ctx, `SELECT id,status,attempt_count,max_attempts FROM send_queue WHERE mailbox_id=? AND source=? AND message_id=?`, in.MailboxID, in.Source, messageID).Scan(&existingID, &status, &attemptCount, &maxAttempts); err != nil {
|
|
return "", err
|
|
}
|
|
if existingID != id {
|
|
if status == sendQueueStatusDelivered || status == sendQueueStatusCanceled || (status == sendQueueStatusFailed && attemptCount >= maxAttempts) {
|
|
_, err := a.db.ExecContext(ctx, `UPDATE send_queue SET user_id=?,sent_message_id=?,mail_from=?,header_from=?,recipients_json=?,mime_base64=?,status=?,attempt_count=0,next_attempt_at=?,last_error='',updated_at=?,delivered_at=NULL WHERE id=? AND status=?`,
|
|
in.UserID, in.SentMessageID, normalizeEmail(in.MailFrom), normalizeEmail(in.HeaderFrom), recipientsJSON, mimeBase64, sendQueueStatusQueued, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano), existingID, status)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
a.deleteSendQueueDeliveredMarker(existingID)
|
|
a.recordSendAudit(ctx, sendAuditQueued, sendQueueStatusQueued, sendAuditInput{
|
|
QueueID: existingID,
|
|
UserID: in.UserID,
|
|
MailboxID: in.MailboxID,
|
|
SentMessageID: in.SentMessageID,
|
|
Source: in.Source,
|
|
MailFrom: in.MailFrom,
|
|
HeaderFrom: in.HeaderFrom,
|
|
Recipients: in.Recipients,
|
|
})
|
|
}
|
|
return existingID, nil
|
|
}
|
|
}
|
|
a.recordSendAudit(ctx, sendAuditQueued, sendQueueStatusQueued, sendAuditInput{
|
|
QueueID: id,
|
|
UserID: in.UserID,
|
|
MailboxID: in.MailboxID,
|
|
SentMessageID: in.SentMessageID,
|
|
Source: in.Source,
|
|
MailFrom: in.MailFrom,
|
|
HeaderFrom: in.HeaderFrom,
|
|
Recipients: in.Recipients,
|
|
})
|
|
return id, nil
|
|
}
|
|
|
|
func (a *App) sendQueueWorker(ctx context.Context) {
|
|
a.log.Info("send queue worker started")
|
|
ticker := time.NewTicker(10 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
a.log.Info("send queue worker stopped")
|
|
return
|
|
default:
|
|
}
|
|
if err := a.processDueSendQueue(ctx); err != nil {
|
|
a.log.Warn("send queue worker failed", "error", err)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
a.log.Info("send queue worker stopped")
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (a *App) processDueSendQueue(ctx context.Context) error {
|
|
if strings.TrimSpace(a.cfg.SMTPHost) == "" {
|
|
return nil
|
|
}
|
|
if err := a.recoverStaleSendQueueItems(ctx); err != nil {
|
|
return err
|
|
}
|
|
rows, err := a.db.QueryContext(ctx, `SELECT id FROM send_queue WHERE (status=? OR (status=? AND attempt_count<max_attempts)) AND next_attempt_at<=? ORDER BY next_attempt_at, created_at LIMIT 20`, sendQueueStatusQueued, sendQueueStatusFailed, a.now().UTC().Format(time.RFC3339Nano))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
var ids []string
|
|
for rows.Next() {
|
|
var id string
|
|
if err := rows.Scan(&id); err != nil {
|
|
return err
|
|
}
|
|
ids = append(ids, id)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return err
|
|
}
|
|
sem := make(chan struct{}, sendQueueConcurrency)
|
|
done := make(chan struct{}, len(ids))
|
|
for _, id := range ids {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case sem <- struct{}{}:
|
|
}
|
|
go func(id string) {
|
|
defer func() {
|
|
<-sem
|
|
done <- struct{}{}
|
|
}()
|
|
a.processSendQueueItem(ctx, id)
|
|
}(id)
|
|
}
|
|
for range ids {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-done:
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *App) recoverStaleSendQueueItems(ctx context.Context) error {
|
|
cutoff := a.now().UTC().Add(-sendQueueStaleAfter).Format(time.RFC3339Nano)
|
|
rows, err := a.db.QueryContext(ctx, `SELECT id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,attempt_count,max_attempts FROM send_queue WHERE status=? AND updated_at<=? AND attempt_count<max_attempts LIMIT 20`, sendQueueStatusSending, cutoff)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
var items []sendQueueItem
|
|
for rows.Next() {
|
|
var item sendQueueItem
|
|
var recipientsJSON, mimeBase64 string
|
|
if err := rows.Scan(&item.ID, &item.UserID, &item.MailboxID, &item.SentMessageID, &item.MessageID, &item.Source, &item.MailFrom, &item.HeaderFrom, &recipientsJSON, &mimeBase64, &item.AttemptCount, &item.MaxAttempts); err != nil {
|
|
return err
|
|
}
|
|
item.Recipients = jsonDecodeSlice(recipientsJSON)
|
|
delivered, err := a.hasSendQueueDeliveredMarker(item.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if delivered {
|
|
items = append(items, item)
|
|
continue
|
|
}
|
|
mimeBytes, err := base64.StdEncoding.DecodeString(mimeBase64)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
item.MIMEBytes = mimeBytes
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return err
|
|
}
|
|
now := a.now().UTC().Format(time.RFC3339Nano)
|
|
for _, item := range items {
|
|
delivered, err := a.hasSendQueueDeliveredMarker(item.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if delivered {
|
|
if err := a.markSendQueueDelivered(ctx, item); err != nil {
|
|
return err
|
|
}
|
|
continue
|
|
}
|
|
res, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,next_attempt_at=?,last_error=?,updated_at=? WHERE id=? AND status=?`, sendQueueStatusQueued, now, "send attempt interrupted", now, item.ID, sendQueueStatusSending)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n, _ := res.RowsAffected(); n > 0 {
|
|
a.recordSendAudit(ctx, sendAuditRetry, sendQueueStatusQueued, sendAuditInputFromQueue(item, "send attempt interrupted"))
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *App) processSendQueueItem(ctx context.Context, id string) {
|
|
item, err := a.claimSendQueueItem(ctx, id)
|
|
if err != nil {
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
a.log.Warn("failed to claim send queue item", "id", id, "error", err)
|
|
}
|
|
return
|
|
}
|
|
if err := a.sendSMTP(item.MailFrom, item.Recipients, item.MIMEBytes); err != nil {
|
|
a.markSendQueueFailed(ctx, item, err)
|
|
return
|
|
}
|
|
if err := a.writeSendQueueDeliveredMarker(item.ID); err != nil {
|
|
a.log.Warn("failed to persist send queue delivered marker", "id", item.ID, "error", err)
|
|
}
|
|
if err := a.markSendQueueDelivered(ctx, item); err != nil {
|
|
a.log.Warn("failed to mark send queue delivered", "id", item.ID, "error", err)
|
|
return
|
|
}
|
|
}
|
|
|
|
func (a *App) markSendQueueDelivered(ctx context.Context, item sendQueueItem) error {
|
|
now := a.now().UTC().Format(time.RFC3339Nano)
|
|
if _, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,delivered_at=?,updated_at=?,last_error='',mime_base64='' WHERE id=?`, sendQueueStatusDelivered, now, now, item.ID); err != nil {
|
|
return err
|
|
}
|
|
a.deleteSendQueueDeliveredMarker(item.ID)
|
|
a.recordSendAudit(ctx, sendAuditDelivered, sendQueueStatusDelivered, sendAuditInputFromQueue(item, ""))
|
|
return nil
|
|
}
|
|
|
|
func (a *App) claimSendQueueItem(ctx context.Context, id string) (sendQueueItem, error) {
|
|
now := a.now().UTC().Format(time.RFC3339Nano)
|
|
res, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,attempt_count=attempt_count+1,updated_at=? WHERE id=? AND (status=? OR (status=? AND attempt_count<max_attempts)) AND next_attempt_at<=?`, sendQueueStatusSending, now, id, sendQueueStatusQueued, sendQueueStatusFailed, now)
|
|
if err != nil {
|
|
return sendQueueItem{}, err
|
|
}
|
|
if n, _ := res.RowsAffected(); n == 0 {
|
|
return sendQueueItem{}, sql.ErrNoRows
|
|
}
|
|
var item sendQueueItem
|
|
var recipientsJSON, mimeBase64 string
|
|
row := a.db.QueryRowContext(ctx, `SELECT id,user_id,mailbox_id,sent_message_id,message_id,source,mail_from,header_from,recipients_json,mime_base64,attempt_count,max_attempts FROM send_queue WHERE id=?`, id)
|
|
if err := row.Scan(&item.ID, &item.UserID, &item.MailboxID, &item.SentMessageID, &item.MessageID, &item.Source, &item.MailFrom, &item.HeaderFrom, &recipientsJSON, &mimeBase64, &item.AttemptCount, &item.MaxAttempts); err != nil {
|
|
return sendQueueItem{}, err
|
|
}
|
|
item.Recipients = jsonDecodeSlice(recipientsJSON)
|
|
mimeBytes, err := base64.StdEncoding.DecodeString(mimeBase64)
|
|
if err != nil {
|
|
return sendQueueItem{}, err
|
|
}
|
|
item.MIMEBytes = mimeBytes
|
|
return item, nil
|
|
}
|
|
|
|
func (a *App) markSendQueueFailed(ctx context.Context, item sendQueueItem, sendErr error) {
|
|
now := a.now().UTC()
|
|
status := sendQueueStatusFailed
|
|
nextAttempt := now.Add(sendRetryDelay(item.AttemptCount))
|
|
if item.AttemptCount >= item.MaxAttempts {
|
|
nextAttempt = now.Add(365 * 24 * time.Hour)
|
|
}
|
|
_, err := a.db.ExecContext(ctx, `UPDATE send_queue SET status=?,next_attempt_at=?,last_error=?,updated_at=? WHERE id=?`, status, nextAttempt.Format(time.RFC3339Nano), sendErr.Error(), now.Format(time.RFC3339Nano), item.ID)
|
|
if err != nil {
|
|
a.log.Warn("failed to mark send queue failed", "id", item.ID, "error", err)
|
|
}
|
|
event := sendAuditRetry
|
|
if item.AttemptCount >= item.MaxAttempts {
|
|
event = sendAuditFailed
|
|
}
|
|
a.recordSendAudit(ctx, event, status, sendAuditInputFromQueue(item, sendErr.Error()))
|
|
}
|
|
|
|
func sendRetryDelay(attempt int) time.Duration {
|
|
if attempt < 1 {
|
|
attempt = 1
|
|
}
|
|
delays := []time.Duration{30 * time.Second, 2 * time.Minute, 10 * time.Minute, time.Hour, 6 * time.Hour}
|
|
if attempt > len(delays) {
|
|
return delays[len(delays)-1]
|
|
}
|
|
return delays[attempt-1]
|
|
}
|
|
|
|
type sendAuditInput struct {
|
|
QueueID string
|
|
UserID string
|
|
MailboxID string
|
|
SentMessageID string
|
|
Source string
|
|
MailFrom string
|
|
HeaderFrom string
|
|
Recipients []string
|
|
Error string
|
|
}
|
|
|
|
func sendAuditInputFromQueue(item sendQueueItem, errorText string) sendAuditInput {
|
|
return sendAuditInput{
|
|
QueueID: item.ID,
|
|
UserID: item.UserID,
|
|
MailboxID: item.MailboxID,
|
|
SentMessageID: item.SentMessageID,
|
|
Source: item.Source,
|
|
MailFrom: item.MailFrom,
|
|
HeaderFrom: item.HeaderFrom,
|
|
Recipients: item.Recipients,
|
|
Error: errorText,
|
|
}
|
|
}
|
|
|
|
func (a *App) recordSendAudit(ctx context.Context, event, status string, in sendAuditInput) {
|
|
source := strings.TrimSpace(in.Source)
|
|
if source == "" {
|
|
source = "unknown"
|
|
}
|
|
id := newID("audit")
|
|
createdAt := a.now().UTC()
|
|
item := SendAuditEvent{ID: id, QueueID: in.QueueID, MailboxID: in.MailboxID, SentMessageID: in.SentMessageID, Source: source, Event: event, Status: status, MailFrom: normalizeEmail(in.MailFrom), HeaderFrom: normalizeEmail(in.HeaderFrom), Recipients: dedupeEmails(in.Recipients), Error: in.Error, CreatedAt: createdAt}
|
|
tx, err := a.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
a.log.Warn("failed to start send audit transaction", "event", event, "error", err)
|
|
return
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO send_audit_events(id,queue_id,user_id,mailbox_id,sent_message_id,source,event,status,mail_from,header_from,recipients_json,error,created_at)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, in.QueueID, in.UserID, in.MailboxID, in.SentMessageID, source, event, status, item.MailFrom, item.HeaderFrom, jsonEncode(item.Recipients), in.Error, createdAt.Format(time.RFC3339Nano)); err != nil {
|
|
a.log.Warn("failed to record send audit", "event", event, "error", err)
|
|
return
|
|
}
|
|
if err := a.enqueueStatusWebhook(ctx, tx, "audit:"+id, "send."+event, in.MailboxID, item); err != nil {
|
|
a.log.Warn("failed to enqueue send status webhook", "event", event, "error", err)
|
|
return
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
a.log.Warn("failed to commit send audit", "event", event, "error", err)
|
|
}
|
|
}
|
|
|
|
func (a *App) sendQueueDeliveredMarkerPath(id string) string {
|
|
safeID := filepath.Base(strings.TrimSpace(id))
|
|
if safeID == "" || safeID == "." {
|
|
safeID = "unknown"
|
|
}
|
|
return filepath.Join(a.cfg.DataDir, sendQueueDeliveredMarkerDir, safeID+".marker")
|
|
}
|
|
|
|
func (a *App) writeSendQueueDeliveredMarker(id string) error {
|
|
path := a.sendQueueDeliveredMarkerPath(id)
|
|
dir := filepath.Dir(path)
|
|
if err := os.MkdirAll(dir, 0o700); err != nil {
|
|
return err
|
|
}
|
|
tmp := filepath.Join(dir, filepath.Base(path)+"."+newID("tmp"))
|
|
if err := os.WriteFile(tmp, []byte(a.now().UTC().Format(time.RFC3339Nano)), 0o600); err != nil {
|
|
return err
|
|
}
|
|
if err := os.Rename(tmp, path); err != nil {
|
|
_ = os.Remove(tmp)
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *App) hasSendQueueDeliveredMarker(id string) (bool, error) {
|
|
_, err := os.Stat(a.sendQueueDeliveredMarkerPath(id))
|
|
if err == nil {
|
|
return true, nil
|
|
}
|
|
if errors.Is(err, os.ErrNotExist) {
|
|
return false, nil
|
|
}
|
|
return false, err
|
|
}
|
|
|
|
func (a *App) deleteSendQueueDeliveredMarker(id string) {
|
|
err := os.Remove(a.sendQueueDeliveredMarkerPath(id))
|
|
if err != nil && !errors.Is(err, os.ErrNotExist) {
|
|
a.log.Warn("failed to remove send queue delivered marker", "id", id, "error", err)
|
|
}
|
|
}
|
|
|
|
func (a *App) authorizedSender(ctx context.Context, mb *Mailbox, from string) (string, string, error) {
|
|
from = normalizeEmail(from)
|
|
if from == "" {
|
|
from = normalizeEmail(mb.Address)
|
|
}
|
|
if from == normalizeEmail(mb.Address) {
|
|
return normalizeEmail(mb.Address), mb.DisplayName, nil
|
|
}
|
|
var displayName string
|
|
var enabled int
|
|
err := a.db.QueryRowContext(ctx, `SELECT display_name,enabled FROM send_as_grants WHERE mailbox_id=? AND address=?`, mb.ID, from).Scan(&displayName, &enabled)
|
|
if err == nil {
|
|
if enabled == 0 {
|
|
return "", "", errSenderNotAuthorized
|
|
}
|
|
return from, strings.TrimSpace(displayName), nil
|
|
}
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return "", "", err
|
|
}
|
|
var aliasDestination string
|
|
err = a.db.QueryRowContext(ctx, `SELECT destination FROM aliases WHERE source=? AND enabled=1`, from).Scan(&aliasDestination)
|
|
if err == nil {
|
|
for _, destination := range strings.Split(aliasDestination, ",") {
|
|
if normalizeEmail(destination) == normalizeEmail(mb.Address) {
|
|
return from, mb.DisplayName, nil
|
|
}
|
|
}
|
|
}
|
|
return "", "", errSenderNotAuthorized
|
|
}
|