// Package aliyun 提供阿里云邮件推送(DirectMail)通道。 package aliyun import ( "context" "fmt" "strings" "sync" "time" mailx "code.yun.ink/pkg/mailx" openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client" dm20151123 "github.com/alibabacloud-go/dm-20151123/v2/client" util "github.com/alibabacloud-go/tea-utils/v2/service" "github.com/alibabacloud-go/tea/tea" ) // defaultTimeout 单次 API 调用的默认超时 const defaultTimeout = 30 * time.Second // Config 阿里云邮件推送通道配置 type Config struct { AccessKeyID string // AccessKey ID AccessKeySecret string // AccessKey Secret Endpoint string // Endpoint,默认 dm.aliyuncs.com AccountName string // 发信地址(必填) ReplyAddress string // 默认回复地址(可选) Timeout time.Duration // 单次 API 调用的超时,默认 30s } // dmMailer 阿里云 DM 客户端的最小接口(便于测试注入 mock) type dmMailer interface { SingleSendMailWithOptions(request *dm20151123.SingleSendMailRequest, runtime *util.RuntimeOptions) (*dm20151123.SingleSendMailResponse, error) SenderStatisticsDetailByParamWithOptions(request *dm20151123.SenderStatisticsDetailByParamRequest, runtime *util.RuntimeOptions) (*dm20151123.SenderStatisticsDetailByParamResponse, error) } // Aliyun 阿里云发送通道 type Aliyun struct { cfg Config initOnce sync.Once client dmMailer initErr error } // New 创建阿里云通道 func New(cfg Config) *Aliyun { if cfg.Endpoint == "" { cfg.Endpoint = "dm.aliyuncs.com" } return &Aliyun{cfg: cfg} } // Name 返回通道名称 func (a *Aliyun) Name() string { return "aliyun" } // Send 发送一封邮件 func (a *Aliyun) Send(ctx context.Context, msg *mailx.Message) error { logger := mailx.LoggerFromContext(ctx) if len(msg.To) > 100 { return fmt.Errorf("%w: aliyun up to 100 recipients allowed", mailx.ErrInvalidMessage) } if a.cfg.AccountName == "" { return fmt.Errorf("%w: aliyun AccountName is required", mailx.ErrInvalidConfig) } if len(msg.Inline) > 0 { return fmt.Errorf("%w: aliyun SingleSendMail does not support inline images", mailx.ErrInvalidConfig) } client, err := a.newClient() if err != nil { return fmt.Errorf("mailx/aliyun: create client: %w", err) } // 阿里云 DirectMail 的 HtmlBody 必填;纯文本也需以 HTML 形式传递 htmlBody := msg.Body if htmlBody == "" { htmlBody = mailx.EscapeHTML(msg.TextBody) } req := &dm20151123.SingleSendMailRequest{ AccountName: tea.String(a.cfg.AccountName), ToAddress: tea.String(strings.Join(msg.To, ",")), Subject: tea.String(msg.Subject), HtmlBody: tea.String(htmlBody), AddressType: tea.Int32(0), // 0:随机账号;1:发信地址 } replyTo := msg.ReplyTo if replyTo == "" { replyTo = a.cfg.ReplyAddress } if replyTo != "" { req.ReplyToAddress = tea.Bool(false) req.ReplyAddress = tea.String(replyTo) } else { req.ReplyToAddress = tea.Bool(true) } // 将 ctx 的 deadline 折算为超时,同时应用配置超时 runtime := a.runtimeOptions(ctx) if _, err := client.SingleSendMailWithOptions(req, runtime); err != nil { logger.Errorf(ctx, "mailx/aliyun: send failed: %v", err) return fmt.Errorf("%w: %v", mailx.ErrSendFailed, err) } logger.Infof(ctx, "mailx/aliyun: sent to %v subject=%q", msg.To, msg.Subject) return nil } // newClient 惰性创建并复用阿里云 SDK 客户端(线程安全)。 // 若已通过测试或其他方式注入 client,则直接返回注入的客户端。 func (a *Aliyun) newClient() (dmMailer, error) { if a.client != nil { return a.client, a.initErr } a.initOnce.Do(func() { config := &openapi.Config{ AccessKeyId: tea.String(a.cfg.AccessKeyID), AccessKeySecret: tea.String(a.cfg.AccessKeySecret), Endpoint: tea.String(a.cfg.Endpoint), } c, err := dm20151123.NewClient(config) if err != nil { a.initErr = fmt.Errorf("mailx/aliyun: create client: %w", err) return } a.client = c }) return a.client, a.initErr } // runtimeOptions 根据配置与 ctx 计算超时,返回 RuntimeOptions func (a *Aliyun) runtimeOptions(ctx context.Context) *util.RuntimeOptions { timeout := a.cfg.Timeout if timeout <= 0 { timeout = defaultTimeout } if dl, ok := ctx.Deadline(); ok { if remain := time.Until(dl); remain > 0 && remain < timeout { timeout = remain } } ro := &util.RuntimeOptions{} if timeout > 0 { ms := int(timeout / time.Millisecond) ro.SetConnectTimeout(ms) ro.SetReadTimeout(ms) } return ro } // SyncStatus 同步发送状态(阿里云特有能力,非 Sender 接口的一部分) func (a *Aliyun) SyncStatus(ctx context.Context) ([]mailx.EmailSendRecord, error) { logger := mailx.LoggerFromContext(ctx) client, err := a.newClient() if err != nil { return nil, fmt.Errorf("mailx/aliyun: create client: %w", err) } runtime := a.runtimeOptions(ctx) var records []mailx.EmailSendRecord start := "" // 分页拉取最近一天的数据 for { req := &dm20151123.SenderStatisticsDetailByParamRequest{ StartTime: tea.String(time.Now().AddDate(0, 0, -1).Format("2006-01-02 15:04")), EndTime: tea.String(time.Now().Format("2006-01-02 15:04")), Length: tea.Int32(100), } if start != "" { req.NextStart = tea.String(start) } resp, err := client.SenderStatisticsDetailByParamWithOptions(req, runtime) if err != nil { logger.Errorf(ctx, "mailx/aliyun: sync status failed: %v", err) return nil, fmt.Errorf("mailx/aliyun: sync status: %w", err) } if resp == nil || resp.Body == nil || resp.Body.Data == nil { break } for _, d := range resp.Body.Data.MailDetail { t, _ := time.ParseInLocation("2006-01-02T15:04Z", tea.StringValue(d.LastUpdateTime), time.Local) records = append(records, mailx.EmailSendRecord{ AccountName: tea.StringValue(d.AccountName), UpdateTime: t.UnixMilli(), ToUser: tea.StringValue(d.ToAddress), Subject: tea.StringValue(d.Subject), ErrorMessage: tea.StringValue(d.Message), Status: mapSendStatus(tea.Int32Value(d.Status)), }) } if resp.Body.NextStart == nil || *resp.Body.NextStart == "" { break } start = *resp.Body.NextStart } return records, nil } // mapSendStatus 阿里云状态码映射为统一状态 func mapSendStatus(code int32) mailx.EmailSendStatus { switch code { case 0: return mailx.EmailSendStatusSuccess case 2: return mailx.EmailSendStatusInvalidAddress case 3: return mailx.EmailSendStatusSpam case 4: return mailx.EmailSendStatusFailed default: return mailx.EmailSendStatusUnknown } }