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

@@ -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"

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)
+		conn, tlserr := tls.Dial("tcp", addr, tlsConfig)
-		if err != nil {
+		if tlserr != nil {
-			return fmt.Errorf("smtp tls dial: %w", err)
+			return fmt.Errorf("smtp tls dial: %w", tlserr)
 		}
 		defer conn.Close()
-		client, err := smtp.NewClient(conn, host)
+		client, newClientErr := smtp.NewClient(conn, host)
-		if err != nil {
+		if newClientErr != nil {
-			return fmt.Errorf("smtp new client: %w", err)
+			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()
+		w, dataErr := client.Data()
-		if err != nil {
+		if dataErr != nil {
-			return fmt.Errorf("smtp data: %w", err)
+			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
+	// SenderInfo for allow-list and routing (canonical format: email:addr)
-	if !c.IsAllowed(senderID) {
+	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.

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{
+		msgBus := bus.NewMessageBus()
-			BaseChannel: &BaseChannel{
+		c, err := NewEmailChannel(config.EmailConfig{
-				bus: bus.NewMessageBus(),
-			},
-			config: config.EmailConfig{
 			Enabled:   true,
 			AllowFrom: config.FlexibleStringSlice{},
-			},
+		}, msgBus)
-			lastUID: 20,
+		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{
+		msgBus := bus.NewMessageBus()
-			BaseChannel: &BaseChannel{
+		c, err := NewEmailChannel(config.EmailConfig{
-				bus: bus.NewMessageBus(),
-			},
-			config: config.EmailConfig{
 			Enabled:   true,
 			AllowFrom: config.FlexibleStringSlice{},
-			},
+		}, msgBus)
-			lastUID: 20,
+		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{
+		msgBus := bus.NewMessageBus()
-			BaseChannel: &BaseChannel{
+		c, err := NewEmailChannel(config.EmailConfig{
-				bus: bus.NewMessageBus(),
-			},
-			config: 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)

pkg/channels/email/email_test.go

@@ -1,4 +1,4 @@
-package channels
+package email
 
 import (
 	"bytes"

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)
+	})
+}

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")
+		m.initChannel("email", "Email")
-		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")
-		}
 	}
 
 	if m.config.Channels.WeCom.Enabled && m.config.Channels.WeCom.Token != "" {