From f13824b69653799a3e7d54462ef62ed57e1a1df0 Mon Sep 17 00:00:00 2001 From: zhouliang Date: Wed, 25 Feb 2026 20:49:36 +0800 Subject: [PATCH] feat(channels): add Email channel and refactor to latest structure Email: IMAP + optional SMTP, allow_from, attachments. Moved to pkg/channels/email/ with initChannel, identity.BuildCanonicalID, bus.Peer/SenderInfo. Bus and gateway updated for context-aware publish, media, and graceful shutdown. --- cmd/picoclaw/cmd_gateway.go | 1 + pkg/channels/{ => email}/email.go | 58 +++++++++------- pkg/channels/{ => email}/email_mockey_test.go | 67 +++++++++---------- pkg/channels/{ => email}/email_test.go | 2 +- pkg/channels/email/init.go | 13 ++++ pkg/channels/manager.go | 11 +-- 6 files changed, 82 insertions(+), 70 deletions(-) rename pkg/channels/{ => email}/email.go (94%) rename pkg/channels/{ => email}/email_mockey_test.go (86%) rename pkg/channels/{ => email}/email_test.go (99%) create mode 100644 pkg/channels/email/init.go diff --git a/cmd/picoclaw/cmd_gateway.go b/cmd/picoclaw/cmd_gateway.go index 798ad2813..e660e500d 100644 --- a/cmd/picoclaw/cmd_gateway.go +++ b/cmd/picoclaw/cmd_gateway.go @@ -16,6 +16,7 @@ import ( "github.com/sipeed/picoclaw/pkg/channels" _ "github.com/sipeed/picoclaw/pkg/channels/dingtalk" _ "github.com/sipeed/picoclaw/pkg/channels/discord" + _ "github.com/sipeed/picoclaw/pkg/channels/email" _ "github.com/sipeed/picoclaw/pkg/channels/feishu" _ "github.com/sipeed/picoclaw/pkg/channels/line" _ "github.com/sipeed/picoclaw/pkg/channels/maixcam" diff --git a/pkg/channels/email.go b/pkg/channels/email/email.go similarity index 94% rename from pkg/channels/email.go rename to pkg/channels/email/email.go index 6ba0a9ac9..fbbe72de2 100644 --- a/pkg/channels/email.go +++ b/pkg/channels/email/email.go @@ -1,4 +1,4 @@ -package channels +package email import ( "bytes" @@ -23,7 +23,9 @@ import ( "golang.org/x/text/encoding/simplifiedchinese" "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/identity" "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/utils" ) @@ -45,7 +47,7 @@ const ( ) type EmailChannel struct { - *BaseChannel + *channels.BaseChannel config config.EmailConfig imapClient *client.Client lastUID uint32 @@ -62,7 +64,7 @@ type EmailChannel struct { } func NewEmailChannel(cfg config.EmailConfig, bus *bus.MessageBus) (*EmailChannel, error) { - base := NewBaseChannel("email", cfg, bus, cfg.AllowFrom) + base := channels.NewBaseChannel("email", cfg, bus, cfg.AllowFrom) return &EmailChannel{ BaseChannel: base, config: cfg, @@ -182,14 +184,14 @@ func (c *EmailChannel) Send(ctx context.Context, msg bus.OutboundMessage) error if c.config.SMTPUseTLS { // Port 465: implicit TLS tlsConfig := &tls.Config{ServerName: host} - conn, err := tls.Dial("tcp", addr, tlsConfig) - if err != nil { - return fmt.Errorf("smtp tls dial: %w", err) + conn, tlserr := tls.Dial("tcp", addr, tlsConfig) + if tlserr != nil { + return fmt.Errorf("smtp tls dial: %w", tlserr) } defer conn.Close() - client, err := smtp.NewClient(conn, host) - if err != nil { - return fmt.Errorf("smtp new client: %w", err) + client, newClientErr := smtp.NewClient(conn, host) + if newClientErr != nil { + return fmt.Errorf("smtp new client: %w", newClientErr) } defer client.Close() auth := smtp.PlainAuth("", c.config.Username, c.config.Password, host) @@ -202,9 +204,9 @@ func (c *EmailChannel) Send(ctx context.Context, msg bus.OutboundMessage) error if err = client.Rcpt(toRaw); err != nil { return fmt.Errorf("smtp rcpt: %w", err) } - w, err := client.Data() - if err != nil { - return fmt.Errorf("smtp data: %w", err) + w, dataErr := client.Data() + if dataErr != nil { + return fmt.Errorf("smtp data: %w", dataErr) } if _, err = w.Write(body); err != nil { _ = w.Close() @@ -275,9 +277,9 @@ func (c *EmailChannel) connect() error { } // Login - if err := cl.Login(c.config.Username, c.config.Password); err != nil { + if loginErr := cl.Login(c.config.Username, c.config.Password); loginErr != nil { cl.Logout() - return err + return loginErr } c.mu.Lock() @@ -617,7 +619,7 @@ func (c *EmailChannel) CheckNewEmails(ctx context.Context) { } // Process the message - c.processEmail(msg) + c.processEmail(ctx, msg) // Mark as seen after fully read seenSet := new(imap.SeqSet) @@ -660,7 +662,7 @@ func (c *EmailChannel) CheckNewEmails(ctx context.Context) { } } -func (c *EmailChannel) processEmail(msg *imap.Message) { +func (c *EmailChannel) processEmail(ctx context.Context, msg *imap.Message) { if msg == nil { return } @@ -683,8 +685,15 @@ func (c *EmailChannel) processEmail(msg *imap.Message) { senderID = "unknown" } - // Check allowlist - if !c.IsAllowed(senderID) { + // SenderInfo for allow-list and routing (canonical format: email:addr) + senderInfo := bus.SenderInfo{ + Platform: "email", + PlatformID: senderID, + CanonicalID: identity.BuildCanonicalID("email", senderID), + } + + // Check allowlist (HandleMessage will also check; we avoid duplicate work by passing SenderInfo) + if !c.IsAllowedSender(senderInfo) { logger.DebugCF("email", "Email from unauthorized sender", map[string]any{ "sender": senderID, }) @@ -697,14 +706,16 @@ func (c *EmailChannel) processEmail(msg *imap.Message) { content = "[empty email body]" } - // ChatID is sender email + // ChatID is sender email (1:1 conversation) chatID := senderID + messageID := fmt.Sprintf("%d", msg.Uid) // Build metadata metadata := map[string]string{ "subject": envelope.Subject, - "message_id": fmt.Sprintf("%d", msg.Uid), + "message_id": messageID, "date": envelope.Date.Format(time.RFC3339), + "platform": "email", } if len(envelope.To) > 0 { @@ -712,14 +723,15 @@ func (c *EmailChannel) processEmail(msg *imap.Message) { metadata["to"] = fmt.Sprintf("%s@%s", to.MailboxName, to.HostName) } - logger.InfoCF("email", "Email received", map[string]any{ + logger.DebugCF("email", "Received message", map[string]any{ "sender_id": senderID, "subject": envelope.Subject, - "preview": utils.Truncate(content, 80), + "preview": utils.Truncate(content, 50), }) // Publish to message bus (attachment local paths in mediaPaths) - c.HandleMessage(senderID, chatID, content, mediaPaths, metadata) + peer := bus.Peer{Kind: "direct", ID: senderID} + c.HandleMessage(ctx, peer, messageID, senderID, chatID, content, mediaPaths, metadata, senderInfo) } // extractEmailBodyAndAttachments parses body and saves attachments to AttachmentDir; returns body text and local paths. diff --git a/pkg/channels/email_mockey_test.go b/pkg/channels/email/email_mockey_test.go similarity index 86% rename from pkg/channels/email_mockey_test.go rename to pkg/channels/email/email_mockey_test.go index 4ea9f39fe..402e67d91 100644 --- a/pkg/channels/email_mockey_test.go +++ b/pkg/channels/email/email_mockey_test.go @@ -1,9 +1,9 @@ //go:build mockey -package channels +package email // Tests in this file use github.com/bytedance/mockey and require -gcflags="all=-N -l" to run. -// Run with: go test -tags=mockey -gcflags="all=-N -l" ./pkg/channels/... +// Run with: go test -tags=mockey -gcflags="all=-N -l" ./pkg/channels/email/... // Without -tags=mockey they are not compiled; without -gcflags they may fail due to Mockey. import ( @@ -35,16 +35,15 @@ func TestEmailChannel_checkNewEmails(t *testing.T) { // mock connect to return mockClient mockey.PatchConvey("checkNewEmails", t, func() { // --------------- mock start --------------- - c := &EmailChannel{ - BaseChannel: &BaseChannel{ - bus: bus.NewMessageBus(), - }, - config: config.EmailConfig{ - Enabled: true, - AllowFrom: config.FlexibleStringSlice{}, - }, - lastUID: 20, + msgBus := bus.NewMessageBus() + c, err := NewEmailChannel(config.EmailConfig{ + Enabled: true, + AllowFrom: config.FlexibleStringSlice{}, + }, msgBus) + if err != nil { + t.Fatal(err) } + c.lastUID = 20 mockClient := &client.Client{} c.imapClient = mockClient // mock login and select to return mockClient @@ -97,7 +96,7 @@ func TestEmailChannel_checkNewEmails(t *testing.T) { c.CheckNewEmails(context.Background()) timeoutCtx, cancel := context.WithTimeout(ctx, time.Second) defer cancel() - messge, ok := c.bus.ConsumeInbound(timeoutCtx) + messge, ok := msgBus.ConsumeInbound(timeoutCtx) assert.True(t, ok) assert.True(t, strings.Contains(messge.Content, "Hello world")) }) @@ -115,19 +114,17 @@ func TestEmailChannel_runIdleLoop(t *testing.T) { mockey.PatchConvey("runIdleLoop", t, func() { // --------------- mock start --------------- hasCheckEmail := false - c := &EmailChannel{ - BaseChannel: &BaseChannel{ - bus: bus.NewMessageBus(), - }, - config: config.EmailConfig{ - Enabled: true, - AllowFrom: config.FlexibleStringSlice{}, - }, - lastUID: 20, + msgBus := bus.NewMessageBus() + c, err := NewEmailChannel(config.EmailConfig{ + Enabled: true, + AllowFrom: config.FlexibleStringSlice{}, + }, msgBus) + if err != nil { + t.Fatal(err) } + c.lastUID = 20 mockClient := &client.Client{} c.imapClient = mockClient - // mock login and select to return mockClient mockey.Mock(mockey.GetMethod(c, "CheckNewEmails")).To(func(*EmailChannel, context.Context) { hasCheckEmail = true }).Build() @@ -164,20 +161,18 @@ func TestEmailChannel_lifecycleCheck(t *testing.T) { mockey.PatchConvey("lifecycle test", t, func() { // --------------- mock start --------------- - c := &EmailChannel{ - BaseChannel: &BaseChannel{ - bus: bus.NewMessageBus(), - }, - config: config.EmailConfig{ - Enabled: true, - CheckInterval: 1, - ForcedPolling: true, - IMAPServer: "imap.example.com", - Username: "testuser", - Password: "testpassword", - }, + msgBus := bus.NewMessageBus() + c, err := NewEmailChannel(config.EmailConfig{ + Enabled: true, + CheckInterval: 1, + ForcedPolling: true, + IMAPServer: "imap.example.com", + Username: "testuser", + Password: "testpassword", + }, msgBus) + if err != nil { + t.Fatal(err) } - // mock login and select to return mockClient mockey.Mock(mockey.GetMethod(c, "connect")).To(func(*EmailChannel) error { return nil }).Build() @@ -186,7 +181,7 @@ func TestEmailChannel_lifecycleCheck(t *testing.T) { }).Build() // --------------- mock end --------------- ctx := context.Background() - err := c.Start(ctx) + err = c.Start(ctx) assert.NoError(t, err) wg := sync.WaitGroup{} wg.Add(1) diff --git a/pkg/channels/email_test.go b/pkg/channels/email/email_test.go similarity index 99% rename from pkg/channels/email_test.go rename to pkg/channels/email/email_test.go index f33959a2c..348c8d0a7 100644 --- a/pkg/channels/email_test.go +++ b/pkg/channels/email/email_test.go @@ -1,4 +1,4 @@ -package channels +package email import ( "bytes" diff --git a/pkg/channels/email/init.go b/pkg/channels/email/init.go new file mode 100644 index 000000000..ab6206118 --- /dev/null +++ b/pkg/channels/email/init.go @@ -0,0 +1,13 @@ +package email + +import ( + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" +) + +func init() { + channels.RegisterFactory("email", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { + return NewEmailChannel(cfg.Channels.Email, b) + }) +} diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index 67277bb87..32c9c5d2c 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -206,16 +206,7 @@ func (m *Manager) initChannels() error { m.initChannel("onebot", "OneBot") } if m.config.Channels.Email.Enabled { - logger.DebugC("channels", "Attempting to initialize Email channel") - email, err := NewEmailChannel(m.config.Channels.Email, m.bus) - if err != nil { - logger.ErrorCF("channels", "Failed to initialize Email channel", map[string]any{ - "error": err.Error(), - }) - } else { - m.channels["email"] = email - logger.InfoC("channels", "Email channel enabled successfully") - } + m.initChannel("email", "Email") } if m.config.Channels.WeCom.Enabled && m.config.Channels.WeCom.Token != "" {