zxyszx
415 lines
16 KiB
Go
415 lines
16 KiB
Go
package app
|
|
|
|
import (
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
)
|
|
|
|
const deliveryWebhookMaxAge = 5 * time.Minute
|
|
|
|
type deliveryWebhookEvent struct {
|
|
ID string `json:"id"`
|
|
Provider string `json:"provider"`
|
|
QueueID string `json:"queueId"`
|
|
MessageID string `json:"messageId"`
|
|
RFCMessageID string `json:"rfcMessageId"`
|
|
Recipient string `json:"recipient"`
|
|
Status string `json:"status"`
|
|
Reason string `json:"reason"`
|
|
OccurredAt string `json:"occurredAt"`
|
|
}
|
|
|
|
func (a *App) handleOpenAPIDeliveryWebhook(w http.ResponseWriter, r *http.Request) {
|
|
secret := strings.TrimSpace(a.config().DeliveryWebhookSecret)
|
|
if secret == "" {
|
|
respondError(w, http.StatusServiceUnavailable, "delivery webhook is not configured")
|
|
return
|
|
}
|
|
timestamp := strings.TrimSpace(r.Header.Get("X-LanQin-Timestamp"))
|
|
signature := strings.TrimPrefix(strings.TrimSpace(r.Header.Get("X-LanQin-Signature")), "sha256=")
|
|
unix, err := strconv.ParseInt(timestamp, 10, 64)
|
|
if err != nil || signature == "" {
|
|
respondError(w, http.StatusUnauthorized, "invalid webhook signature")
|
|
return
|
|
}
|
|
signedAt := time.Unix(unix, 0)
|
|
if delta := a.now().UTC().Sub(signedAt); delta < -deliveryWebhookMaxAge || delta > deliveryWebhookMaxAge {
|
|
respondError(w, http.StatusUnauthorized, "webhook timestamp is outside the allowed window")
|
|
return
|
|
}
|
|
body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20))
|
|
if err != nil {
|
|
badRequest(w, errors.New("invalid webhook body"))
|
|
return
|
|
}
|
|
mac := hmac.New(sha256.New, []byte(secret))
|
|
_, _ = mac.Write([]byte(timestamp + "."))
|
|
_, _ = mac.Write(body)
|
|
expected, err := hex.DecodeString(signature)
|
|
if err != nil || !hmac.Equal(mac.Sum(nil), expected) {
|
|
respondError(w, http.StatusUnauthorized, "invalid webhook signature")
|
|
return
|
|
}
|
|
var payload struct {
|
|
Events []deliveryWebhookEvent `json:"events"`
|
|
}
|
|
dec := json.NewDecoder(strings.NewReader(string(body)))
|
|
dec.DisallowUnknownFields()
|
|
if err := dec.Decode(&payload); err != nil || len(payload.Events) == 0 || len(payload.Events) > 100 {
|
|
badRequest(w, errors.New("events must contain between 1 and 100 items"))
|
|
return
|
|
}
|
|
accepted := 0
|
|
tx, err := a.db.BeginTx(r.Context(), nil)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to start delivery event transaction")
|
|
return
|
|
}
|
|
defer tx.Rollback()
|
|
for _, event := range payload.Events {
|
|
inserted, err := a.storeDeliveryEvent(r, tx, event)
|
|
if err != nil {
|
|
badRequest(w, err)
|
|
return
|
|
}
|
|
if inserted {
|
|
accepted++
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to store delivery events")
|
|
return
|
|
}
|
|
respondJSON(w, http.StatusOK, map[string]any{"ok": true, "accepted": accepted, "duplicates": len(payload.Events) - accepted})
|
|
}
|
|
|
|
func (a *App) storeDeliveryEvent(r *http.Request, tx *sql.Tx, event deliveryWebhookEvent) (bool, error) {
|
|
event.ID = strings.TrimSpace(event.ID)
|
|
event.Provider = strings.ToLower(strings.TrimSpace(event.Provider))
|
|
event.QueueID = strings.TrimSpace(event.QueueID)
|
|
event.MessageID = strings.TrimSpace(event.MessageID)
|
|
event.RFCMessageID = strings.TrimSpace(event.RFCMessageID)
|
|
event.Recipient = normalizeEmail(event.Recipient)
|
|
event.Status = strings.ToLower(strings.TrimSpace(event.Status))
|
|
if event.ID == "" || len(event.ID) > 200 || event.Provider == "" || len(event.Provider) > 80 || event.Recipient == "" || len(event.Recipient) > 320 || len(event.Reason) > 2000 || !validDeliveryEventStatus(event.Status) {
|
|
return false, errors.New("invalid delivery event")
|
|
}
|
|
if event.QueueID == "" && event.MessageID == "" && event.RFCMessageID == "" {
|
|
return false, errors.New("queueId, messageId, or rfcMessageId is required")
|
|
}
|
|
occurredAt, err := time.Parse(time.RFC3339Nano, event.OccurredAt)
|
|
if err != nil {
|
|
return false, errors.New("occurredAt must be an RFC3339 timestamp")
|
|
}
|
|
var queueID, sentMessageID, rfcMessageID string
|
|
err = tx.QueryRowContext(r.Context(), `SELECT id,sent_message_id,message_id FROM send_queue
|
|
WHERE (?<>'' AND id=?) OR (?<>'' AND sent_message_id=?) OR (?<>'' AND message_id=?)
|
|
ORDER BY created_at DESC LIMIT 1`, event.QueueID, event.QueueID, event.MessageID, event.MessageID, event.RFCMessageID, event.RFCMessageID).Scan(&queueID, &sentMessageID, &rfcMessageID)
|
|
if err != nil {
|
|
return false, errors.New("send item not found")
|
|
}
|
|
if (event.QueueID != "" && event.QueueID != queueID) || (event.MessageID != "" && event.MessageID != sentMessageID) || (event.RFCMessageID != "" && event.RFCMessageID != rfcMessageID) {
|
|
return false, errors.New("delivery event identifiers do not refer to the same send item")
|
|
}
|
|
var recipientsJSON string
|
|
if err := tx.QueryRowContext(r.Context(), `SELECT recipients_json FROM send_queue WHERE id=?`, queueID).Scan(&recipientsJSON); err != nil {
|
|
return false, errors.New("send item not found")
|
|
}
|
|
foundRecipient := false
|
|
for _, recipient := range jsonDecodeSlice(recipientsJSON) {
|
|
if normalizeEmail(recipient) == event.Recipient {
|
|
foundRecipient = true
|
|
break
|
|
}
|
|
}
|
|
if !foundRecipient {
|
|
return false, errors.New("delivery event recipient does not belong to the send item")
|
|
}
|
|
id := newID("dev")
|
|
createdAt := a.now().UTC()
|
|
reason := strings.TrimSpace(event.Reason)
|
|
res, err := tx.ExecContext(r.Context(), `INSERT OR IGNORE INTO delivery_events(id,external_id,provider,queue_id,sent_message_id,rfc_message_id,recipient,status,reason,occurred_at,created_at)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?)`, id, event.ID, event.Provider, queueID, sentMessageID, rfcMessageID, event.Recipient, event.Status, reason, occurredAt.UTC().Format(time.RFC3339Nano), createdAt.Format(time.RFC3339Nano))
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n > 0 {
|
|
item := DeliveryEvent{ID: id, ExternalID: event.ID, Provider: event.Provider, QueueID: queueID, MessageID: sentMessageID, RFCMessageID: rfcMessageID, Recipient: event.Recipient, Status: event.Status, Reason: reason, OccurredAt: occurredAt.UTC(), CreatedAt: createdAt}
|
|
var mailboxID string
|
|
if err := tx.QueryRowContext(r.Context(), `SELECT mailbox_id FROM send_queue WHERE id=?`, queueID).Scan(&mailboxID); err != nil {
|
|
return false, err
|
|
}
|
|
if err := a.enqueueStatusWebhook(r.Context(), tx, "delivery:"+event.Provider+":"+event.ID, "delivery."+event.Status, mailboxID, item); err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
return n > 0, nil
|
|
}
|
|
|
|
func validDeliveryEventStatus(status string) bool {
|
|
switch status {
|
|
case "delivered", "bounced", "complained", "rejected", "deferred":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (a *App) handleOpenAPIListSends(w http.ResponseWriter, r *http.Request) {
|
|
user := currentUser(r)
|
|
limit := parseOpenAPILimit(r, 30, 100)
|
|
where := "mb.user_id=?"
|
|
args := []any{user.ID}
|
|
if mailboxID := strings.TrimSpace(r.URL.Query().Get("mailboxId")); mailboxID != "" {
|
|
where += " AND sq.mailbox_id=?"
|
|
args = append(args, mailboxID)
|
|
}
|
|
if status := strings.TrimSpace(r.URL.Query().Get("status")); status != "" {
|
|
if !validSendQueueStatus(status) {
|
|
badRequest(w, errors.New("invalid send queue status"))
|
|
return
|
|
}
|
|
where += " AND sq.status=?"
|
|
args = append(args, status)
|
|
}
|
|
cursorCreatedAt, cursorID, _, err := parseSendQueueCursor(r.URL.Query().Get("cursor"))
|
|
if err != nil {
|
|
badRequest(w, err)
|
|
return
|
|
}
|
|
if cursorCreatedAt != "" {
|
|
where += " AND (sq.created_at<? OR (sq.created_at=? AND sq.id<?))"
|
|
args = append(args, cursorCreatedAt, cursorCreatedAt, cursorID)
|
|
}
|
|
args = append(args, limit+1)
|
|
rows, err := a.db.QueryContext(r.Context(), `SELECT sq.id,sq.mailbox_id,sq.sent_message_id,sq.message_id,COALESCE(m.subject,''),sq.source,sq.mail_from,sq.header_from,sq.recipients_json,sq.status,sq.attempt_count,sq.max_attempts,sq.next_attempt_at,sq.last_error,sq.created_at,sq.updated_at,sq.delivered_at
|
|
FROM send_queue sq JOIN mailboxes mb ON mb.id=sq.mailbox_id LEFT JOIN messages m ON m.id=sq.sent_message_id
|
|
WHERE `+where+` ORDER BY sq.created_at DESC,sq.id DESC LIMIT ?`, args...)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to list sends")
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
items := []openAPISendStatus{}
|
|
for rows.Next() {
|
|
item, err := scanSendQueueEntry(rows)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to scan sends")
|
|
return
|
|
}
|
|
status := openAPISendStatusFromQueue(item, item.MailFrom)
|
|
a.applyDeliveryStatus(r.Context(), &status)
|
|
items = append(items, status)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to list sends")
|
|
return
|
|
}
|
|
next := ""
|
|
if len(items) > limit {
|
|
items = items[:limit]
|
|
last := items[len(items)-1]
|
|
next = encodeSendQueueCursor(last.CreatedAt, last.QueueID)
|
|
}
|
|
respondJSON(w, http.StatusOK, map[string]any{"items": items, "nextCursor": next})
|
|
}
|
|
|
|
func (a *App) handleOpenAPISendEvents(w http.ResponseWriter, r *http.Request) {
|
|
item, err := a.resolveOpenAPISendQueue(r, chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
respondError(w, http.StatusNotFound, "send item not found")
|
|
return
|
|
}
|
|
audit, err := a.sendAuditEvents(r, item.ID)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to load send events")
|
|
return
|
|
}
|
|
delivery, err := a.deliveryEvents(r, item.SentMessageID)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to load delivery events")
|
|
return
|
|
}
|
|
respondJSON(w, http.StatusOK, map[string]any{"auditEvents": audit, "deliveryEvents": delivery})
|
|
}
|
|
|
|
func (a *App) handleOpenAPIRetrySend(w http.ResponseWriter, r *http.Request) {
|
|
item, err := a.resolveOpenAPISendQueue(r, chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
respondError(w, http.StatusNotFound, "send item not found")
|
|
return
|
|
}
|
|
if item.Status != sendQueueStatusFailed {
|
|
badRequest(w, errors.New("send item is not failed"))
|
|
return
|
|
}
|
|
now := a.now().UTC().Format(time.RFC3339Nano)
|
|
res, err := a.db.ExecContext(r.Context(), `UPDATE send_queue SET status=?,attempt_count=0,next_attempt_at=?,last_error='',updated_at=?,delivered_at=NULL WHERE id=? AND status=?`, sendQueueStatusQueued, now, now, item.ID, sendQueueStatusFailed)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to retry send item")
|
|
return
|
|
}
|
|
if affected, _ := res.RowsAffected(); affected == 0 {
|
|
respondError(w, http.StatusConflict, "send item status changed")
|
|
return
|
|
}
|
|
a.recordSendAudit(r.Context(), sendAuditRetry, sendQueueStatusQueued, sendAuditInputFromEntry(item, currentUser(r).ID, ""))
|
|
updated, _ := a.loadSendQueueEntryForUser(r.Context(), item.ID, currentUser(r).ID)
|
|
respondJSON(w, http.StatusOK, openAPISendStatusFromQueue(updated, updated.MailFrom))
|
|
}
|
|
|
|
func (a *App) handleOpenAPICancelSend(w http.ResponseWriter, r *http.Request) {
|
|
item, err := a.resolveOpenAPISendQueue(r, chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
respondError(w, http.StatusNotFound, "send item not found")
|
|
return
|
|
}
|
|
if item.Status != sendQueueStatusQueued && item.Status != sendQueueStatusFailed {
|
|
badRequest(w, errors.New("send item cannot be canceled"))
|
|
return
|
|
}
|
|
now := a.now().UTC().Format(time.RFC3339Nano)
|
|
res, err := a.db.ExecContext(r.Context(), `UPDATE send_queue SET status=?,last_error='',updated_at=? WHERE id=? AND status IN (?,?)`, sendQueueStatusCanceled, now, item.ID, sendQueueStatusQueued, sendQueueStatusFailed)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to cancel send item")
|
|
return
|
|
}
|
|
if affected, _ := res.RowsAffected(); affected == 0 {
|
|
respondError(w, http.StatusConflict, "send item status changed")
|
|
return
|
|
}
|
|
a.recordSendAudit(r.Context(), sendAuditCanceled, sendQueueStatusCanceled, sendAuditInputFromEntry(item, currentUser(r).ID, ""))
|
|
updated, _ := a.loadSendQueueEntryForUser(r.Context(), item.ID, currentUser(r).ID)
|
|
respondJSON(w, http.StatusOK, openAPISendStatusFromQueue(updated, updated.MailFrom))
|
|
}
|
|
|
|
func sendAuditInputFromEntry(item SendQueueEntry, userID, errorText string) sendAuditInput {
|
|
return sendAuditInput{QueueID: item.ID, UserID: userID, MailboxID: item.MailboxID, SentMessageID: item.SentMessageID, Source: item.Source, MailFrom: item.MailFrom, HeaderFrom: item.HeaderFrom, Recipients: item.Recipients, Error: errorText}
|
|
}
|
|
|
|
func (a *App) resolveOpenAPISendQueue(r *http.Request, id string) (SendQueueEntry, error) {
|
|
user := currentUser(r)
|
|
if item, err := a.loadSendQueueEntryForUser(r.Context(), strings.TrimSpace(id), user.ID); err == nil {
|
|
return item, nil
|
|
}
|
|
return a.loadLatestSendQueueForMessage(r.Context(), strings.TrimSpace(id), user.ID)
|
|
}
|
|
|
|
func (a *App) handleOpenAPIMessage(w http.ResponseWriter, r *http.Request) {
|
|
msg, err := a.loadMessageForRequest(r, chi.URLParam(r, "id"), true)
|
|
if err != nil {
|
|
respondError(w, http.StatusNotFound, "message not found")
|
|
return
|
|
}
|
|
respondJSON(w, http.StatusOK, msg)
|
|
}
|
|
|
|
func (a *App) handleOpenAPIListAliases(w http.ResponseWriter, r *http.Request) {
|
|
limit := parseOpenAPILimit(r, 50, 100)
|
|
sortValue, cursorID, err := parseOpenAPIListCursor(r.URL.Query().Get("cursor"))
|
|
if err != nil {
|
|
badRequest(w, err)
|
|
return
|
|
}
|
|
rows, err := a.db.QueryContext(r.Context(), `SELECT id,domain_id,source,destination,enabled,created_at FROM aliases
|
|
WHERE (?='' OR source>? OR (source=? AND id>?)) ORDER BY source,id LIMIT ?`, sortValue, sortValue, sortValue, cursorID, limit+1)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to list aliases")
|
|
return
|
|
}
|
|
defer rows.Close()
|
|
items := []Alias{}
|
|
for rows.Next() {
|
|
item, err := scanOpenAPIAlias(rows)
|
|
if err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to scan aliases")
|
|
return
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
respondError(w, http.StatusInternalServerError, "failed to list aliases")
|
|
return
|
|
}
|
|
next := ""
|
|
if len(items) > limit {
|
|
items = items[:limit]
|
|
last := items[len(items)-1]
|
|
next = encodeOpenAPIListCursor(last.Source, last.ID)
|
|
}
|
|
respondJSON(w, http.StatusOK, map[string]any{"items": items, "nextCursor": next})
|
|
}
|
|
|
|
func (a *App) handleOpenAPIGetAlias(w http.ResponseWriter, r *http.Request) {
|
|
item, err := scanOpenAPIAlias(a.db.QueryRowContext(r.Context(), `SELECT id,domain_id,source,destination,enabled,created_at FROM aliases WHERE id=?`, chi.URLParam(r, "id")))
|
|
if err != nil {
|
|
respondError(w, http.StatusNotFound, "alias not found")
|
|
return
|
|
}
|
|
respondJSON(w, http.StatusOK, item)
|
|
}
|
|
|
|
type aliasScanner interface{ Scan(...any) error }
|
|
|
|
func scanOpenAPIAlias(row aliasScanner) (Alias, error) {
|
|
var item Alias
|
|
var enabled int
|
|
var created string
|
|
err := row.Scan(&item.ID, &item.DomainID, &item.Source, &item.Destination, &enabled, &created)
|
|
item.Enabled = intBool(enabled)
|
|
item.CreatedAt = parseTime(created)
|
|
return item, err
|
|
}
|
|
|
|
func (a *App) sendAuditEvents(r *http.Request, queueID string) ([]SendAuditEvent, error) {
|
|
rows, err := a.db.QueryContext(r.Context(), `SELECT id,queue_id,mailbox_id,sent_message_id,source,event,status,mail_from,header_from,recipients_json,error,created_at FROM send_audit_events WHERE queue_id=? ORDER BY created_at,id`, queueID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []SendAuditEvent{}
|
|
for rows.Next() {
|
|
var item SendAuditEvent
|
|
var recipientsJSON, createdAt string
|
|
if err := rows.Scan(&item.ID, &item.QueueID, &item.MailboxID, &item.SentMessageID, &item.Source, &item.Event, &item.Status, &item.MailFrom, &item.HeaderFrom, &recipientsJSON, &item.Error, &createdAt); err != nil {
|
|
return nil, err
|
|
}
|
|
item.Recipients = jsonDecodeSlice(recipientsJSON)
|
|
item.CreatedAt = parseTime(createdAt)
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (a *App) deliveryEvents(r *http.Request, sentMessageID string) ([]DeliveryEvent, error) {
|
|
rows, err := a.db.QueryContext(r.Context(), `SELECT id,external_id,provider,queue_id,sent_message_id,rfc_message_id,recipient,status,reason,occurred_at,created_at FROM delivery_events WHERE sent_message_id=? ORDER BY occurred_at,id`, sentMessageID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []DeliveryEvent{}
|
|
for rows.Next() {
|
|
var item DeliveryEvent
|
|
var occurredAt, createdAt string
|
|
if err := rows.Scan(&item.ID, &item.ExternalID, &item.Provider, &item.QueueID, &item.MessageID, &item.RFCMessageID, &item.Recipient, &item.Status, &item.Reason, &occurredAt, &createdAt); err != nil {
|
|
return nil, err
|
|
}
|
|
item.OccurredAt = parseTime(occurredAt)
|
|
item.CreatedAt = parseTime(createdAt)
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|