opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(whatsapp): enhance message handling and add media sending capabilities

Browse source
This commit is contained in:
Diego Fornalha 2026-03-30 11:23:38 -03:00
parent eca076c094
commit c823c3087a
1 changed files with 432 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

436
pkg/channels/whatsapp_native/whatsapp_native.go
View file

@ -13,6 +13,7 @@ import (
"fmt"
"os"
"path/filepath"
"regexp"
"strings"
"sync"
"sync/atomic"
@ -33,6 +34,7 @@ import (
"github.com/sipeed/picoclaw/pkg/config"
"github.com/sipeed/picoclaw/pkg/identity"
"github.com/sipeed/picoclaw/pkg/logger"
"github.com/sipeed/picoclaw/pkg/media"
"github.com/sipeed/picoclaw/pkg/utils"
)
@ -267,9 +269,17 @@ func (c *WhatsAppNativeChannel) Stop(ctx context.Context) error {
}
func (c *WhatsAppNativeChannel) eventHandler(evt any) {
switch evt.(type) {
switch v := evt.(type) {
case *events.Message:
c.handleIncoming(evt.(*events.Message))
c.handleIncoming(v)
case *events.CallOffer:
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client != nil {
_ = client.RejectCall(context.Background(), v.BasicCallMeta.From, v.BasicCallMeta.CallID)
logger.InfoCF("whatsapp", "Call rejected automatically", map[string]any{"from": v.BasicCallMeta.From.String()})
}
case *events.Disconnected:
logger.InfoCF("whatsapp", "WhatsApp disconnected, will attempt reconnection", nil)
c.reconnectMu.Lock()
@ -363,6 +373,23 @@ func (c *WhatsAppNativeChannel) handleIncoming(evt *events.Message) {
if evt.Info.PushName != "" {
metadata["user_name"] = evt.Info.PushName
}
// Resolve LID to phone number if available
if evt.Info.Sender.Server == types.HiddenUserServer {
metadata["lid"] = evt.Info.Sender.User
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client != nil && client.Store.LIDs != nil {
if pnJID, err := client.Store.LIDs.GetPNForLID(c.runCtx, evt.Info.Sender); err == nil && !pnJID.IsEmpty() {
metadata["phone_number"] = "+" + pnJID.User
senderID = pnJID.String()
logger.DebugCF("whatsapp", "LID resolved", map[string]any{"lid": evt.Info.Sender.User, "phone": "+" + pnJID.User})
} else {
logger.WarnCF("whatsapp", "LID not found in DB — unknown sender", map[string]any{"lid": evt.Info.Sender.User})
}
}
}
if evt.Info.Chat.Server == types.GroupServer {
metadata["peer_kind"] = "group"
metadata["peer_id"] = chatID
@ -393,6 +420,19 @@ func (c *WhatsAppNativeChannel) handleIncoming(evt *events.Message) {
"WhatsApp message received",
map[string]any{"sender_id": senderID, "content_preview": utils.Truncate(content, 50)},
)
// Send read receipt (blue ticks)
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client != nil {
senderJID := evt.Info.Sender
if evt.Info.Chat.Server != types.GroupServer {
senderJID = types.EmptyJID
}
_ = client.MarkRead(c.runCtx, []types.MessageID{evt.Info.ID}, time.Now(), evt.Info.Chat, senderJID)
}
c.HandleMessage(c.runCtx, peer, messageID, senderID, chatID, content, mediaPaths, metadata, sender)
}
@ -425,8 +465,16 @@ func (c *WhatsAppNativeChannel) Send(ctx context.Context, msg bus.OutboundMessag
return fmt.Errorf("invalid chat id %q: %w", msg.ChatID, err)
}
waMsg := &waE2E.Message{
Conversation: proto.String(msg.Content),
var waMsg *waE2E.Message
if url := extractURL(msg.Content); url != "" {
waMsg = &waE2E.Message{
ExtendedTextMessage: &waE2E.ExtendedTextMessage{
Text: proto.String(msg.Content),
MatchedText: proto.String(url),
},
}
} else {
waMsg = &waE2E.Message{Conversation: proto.String(msg.Content)}
}
if _, err = client.SendMessage(ctx, to, waMsg); err != nil {
@ -435,6 +483,386 @@ func (c *WhatsAppNativeChannel) Send(ctx context.Context, msg bus.OutboundMessag
return nil
}
// SendMedia implements the channels.MediaSender interface.
func (c *WhatsAppNativeChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error {
if !c.IsRunning() {
return channels.ErrNotRunning
}
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil || !client.IsConnected() {
return fmt.Errorf("whatsapp connection not established: %w", channels.ErrTemporary)
}
if client.Store.ID == nil {
return fmt.Errorf("whatsapp not yet paired (QR login pending): %w", channels.ErrTemporary)
}
to, err := parseJID(msg.ChatID)
if err != nil {
return fmt.Errorf("invalid chat id %q: %w", msg.ChatID, channels.ErrSendFailed)
}
store := c.GetMediaStore()
if store == nil {
return fmt.Errorf("no media store available: %w", channels.ErrSendFailed)
}
for _, part := range msg.Parts {
if err := c.sendMediaPart(ctx, client, to, store, part); err != nil {
return err
}
}
return nil
}
// sendMediaPart uploads and sends a single media part via whatsmeow.
func (c *WhatsAppNativeChannel) sendMediaPart(
ctx context.Context,
client *whatsmeow.Client,
to types.JID,
store media.MediaStore,
part bus.MediaPart,
) error {
localPath, err := store.Resolve(part.Ref)
if err != nil {
logger.ErrorCF("whatsapp", "Failed to resolve media ref", map[string]any{
"ref": part.Ref,
"error": err.Error(),
})
return fmt.Errorf("whatsapp resolve media ref %q: %w", part.Ref, channels.ErrSendFailed)
}
data, err := os.ReadFile(localPath)
if err != nil {
logger.ErrorCF("whatsapp", "Failed to read media file", map[string]any{
"path": localPath,
"error": err.Error(),
})
return fmt.Errorf("whatsapp read media: %w", channels.ErrSendFailed)
}
mimeType := part.ContentType
if mimeType == "" {
mimeType = "application/octet-stream"
}
var waMsg *waE2E.Message
switch part.Type {
case "image":
resp, err := client.Upload(ctx, data, whatsmeow.MediaImage)
if err != nil {
return fmt.Errorf("whatsapp upload image: %w", channels.ErrTemporary)
}
waMsg = &waE2E.Message{
ImageMessage: &waE2E.ImageMessage{
Caption: proto.String(part.Caption),
Mimetype: proto.String(mimeType),
URL: proto.String(resp.URL),
DirectPath: proto.String(resp.DirectPath),
MediaKey: resp.MediaKey,
FileEncSHA256: resp.FileEncSHA256,
FileSHA256: resp.FileSHA256,
FileLength: proto.Uint64(resp.FileLength),
},
}
case "video":
resp, err := client.Upload(ctx, data, whatsmeow.MediaVideo)
if err != nil {
return fmt.Errorf("whatsapp upload video: %w", channels.ErrTemporary)
}
waMsg = &waE2E.Message{
VideoMessage: &waE2E.VideoMessage{
Caption: proto.String(part.Caption),
Mimetype: proto.String(mimeType),
URL: proto.String(resp.URL),
DirectPath: proto.String(resp.DirectPath),
MediaKey: resp.MediaKey,
FileEncSHA256: resp.FileEncSHA256,
FileSHA256: resp.FileSHA256,
FileLength: proto.Uint64(resp.FileLength),
},
}
case "audio":
resp, err := client.Upload(ctx, data, whatsmeow.MediaAudio)
if err != nil {
return fmt.Errorf("whatsapp upload audio: %w", channels.ErrTemporary)
}
// Detect voice messages (PTT) by filename convention
fn := strings.ToLower(part.Filename)
isPTT := strings.Contains(fn, "voice") || strings.Contains(fn, "ptt")
waMsg = &waE2E.Message{
AudioMessage: &waE2E.AudioMessage{
Mimetype: proto.String(mimeType),
PTT: proto.Bool(isPTT),
URL: proto.String(resp.URL),
DirectPath: proto.String(resp.DirectPath),
MediaKey: resp.MediaKey,
FileEncSHA256: resp.FileEncSHA256,
FileSHA256: resp.FileSHA256,
FileLength: proto.Uint64(resp.FileLength),
},
}
default: // "file" or unknown → send as document
resp, err := client.Upload(ctx, data, whatsmeow.MediaDocument)
if err != nil {
return fmt.Errorf("whatsapp upload document: %w", channels.ErrTemporary)
}
filename := part.Filename
if filename == "" {
filename = filepath.Base(localPath)
}
waMsg = &waE2E.Message{
DocumentMessage: &waE2E.DocumentMessage{
Caption: proto.String(part.Caption),
Title: proto.String(filename),
FileName: proto.String(filename),
Mimetype: proto.String(mimeType),
URL: proto.String(resp.URL),
DirectPath: proto.String(resp.DirectPath),
MediaKey: resp.MediaKey,
FileEncSHA256: resp.FileEncSHA256,
FileSHA256: resp.FileSHA256,
FileLength: proto.Uint64(resp.FileLength),
},
}
}
if _, err := client.SendMessage(ctx, to, waMsg); err != nil {
logger.ErrorCF("whatsapp", "Failed to send media", map[string]any{
"type": part.Type,
"error": err.Error(),
})
return fmt.Errorf("whatsapp send media: %w", channels.ErrTemporary)
}
logger.DebugCF("whatsapp", "Media sent", map[string]any{
"type": part.Type,
"to": to.String(),
})
return nil
}
// extractURL returns the first URL found in text, or empty string.
var urlPattern = regexp.MustCompile(`https?://\S+`)
func extractURL(text string) string {
return urlPattern.FindString(text)
}
// StartTyping sends a "composing" chat presence to the given chat.
// It repeats every 4 seconds (WhatsApp indicator expires after ~5s).
// The returned stop function sends a "paused" presence and is idempotent.
func (c *WhatsAppNativeChannel) StartTyping(ctx context.Context, chatID string) (func(), error) {
jid, err := parseJID(chatID)
if err != nil {
return func() {}, fmt.Errorf("start typing: %w", err)
}
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil {
return func() {}, nil
}
_ = client.SendChatPresence(ctx, jid, types.ChatPresenceComposing, types.ChatPresenceMediaText)
typingCtx, typingCancel := context.WithCancel(ctx)
var once sync.Once
go func() {
ticker := time.NewTicker(4 * time.Second)
defer ticker.Stop()
for {
select {
case <-typingCtx.Done():
return
case <-ticker.C:
c.mu.Lock()
cl := c.client
c.mu.Unlock()
if cl != nil {
_ = cl.SendChatPresence(typingCtx, jid, types.ChatPresenceComposing, types.ChatPresenceMediaText)
}
}
}
}()
stop := func() {
once.Do(func() {
typingCancel()
c.mu.Lock()
cl := c.client
c.mu.Unlock()
if cl != nil {
_ = cl.SendChatPresence(context.Background(), jid, types.ChatPresencePaused, "")
}
})
}
return stop, nil
}
// EditMessage edits a previously sent message.
func (c *WhatsAppNativeChannel) EditMessage(ctx context.Context, chatID, messageID, content string) error {
jid, err := parseJID(chatID)
if err != nil {
return fmt.Errorf("edit message: %w", err)
}
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil {
return fmt.Errorf("whatsapp not connected")
}
edited := client.BuildEdit(jid, types.MessageID(messageID), &waE2E.Message{
Conversation: proto.String(content),
})
_, err = client.SendMessage(ctx, jid, edited)
return err
}
// DeleteMessage revokes (deletes for everyone) a previously sent message.
func (c *WhatsAppNativeChannel) DeleteMessage(ctx context.Context, chatID, messageID string) error {
jid, err := parseJID(chatID)
if err != nil {
return fmt.Errorf("delete message: %w", err)
}
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil {
return fmt.Errorf("whatsapp not connected")
}
revoke := client.BuildRevoke(jid, types.EmptyJID, types.MessageID(messageID))
_, err = client.SendMessage(ctx, jid, revoke)
return err
}
// ReactToMessage adds a 👀 reaction to an inbound message.
// The returned undo function removes the reaction (idempotent).
func (c *WhatsAppNativeChannel) ReactToMessage(ctx context.Context, chatID, messageID string) (func(), error) {
jid, err := parseJID(chatID)
if err != nil {
return func() {}, fmt.Errorf("react: %w", err)
}
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil {
return func() {}, nil
}
reaction := client.BuildReaction(jid, types.EmptyJID, types.MessageID(messageID), "👀")
_, _ = client.SendMessage(ctx, jid, reaction)
var once sync.Once
undo := func() {
once.Do(func() {
c.mu.Lock()
cl := c.client
c.mu.Unlock()
if cl != nil {
unreact := cl.BuildReaction(jid, types.EmptyJID, types.MessageID(messageID), "")
_, _ = cl.SendMessage(context.Background(), jid, unreact)
}
})
}
return undo, nil
}
// SendPlaceholder sends a temporary "thinking" message and returns its ID
// so it can be edited later with the real response.
func (c *WhatsAppNativeChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) {
jid, err := parseJID(chatID)
if err != nil {
return "", fmt.Errorf("placeholder: %w", err)
}
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil {
return "", fmt.Errorf("whatsapp not connected")
}
msg := &waE2E.Message{Conversation: proto.String("Buscando... ⏳")}
resp, err := client.SendMessage(ctx, jid, msg)
if err != nil {
return "", err
}
return resp.ID, nil
}
// IsOnWhatsApp checks if a phone number is registered on WhatsApp.
func (c *WhatsAppNativeChannel) IsOnWhatsApp(ctx context.Context, phone string) (bool, error) {
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil {
return false, fmt.Errorf("whatsapp not connected")
}
cleaned := "+" + strings.TrimLeft(phone, "+")
resp, err := client.IsOnWhatsApp(ctx, []string{cleaned})
if err != nil {
return false, err
}
if len(resp) == 0 {
return false, nil
}
return resp[0].IsIn, nil
}
// CreateNewsletter creates a WhatsApp Channel (newsletter) and returns its JID.
func (c *WhatsAppNativeChannel) CreateNewsletter(ctx context.Context, name, description string) (string, error) {
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil {
return "", fmt.Errorf("whatsapp not connected")
}
meta, err := client.CreateNewsletter(ctx, whatsmeow.CreateNewsletterParams{
Name: name,
Description: description,
})
if err != nil {
return "", err
}
return meta.ID.String(), nil
}
// SendToNewsletter sends a message to a WhatsApp Channel (newsletter).
func (c *WhatsAppNativeChannel) SendToNewsletter(ctx context.Context, newsletterJID, content string) error {
jid, err := types.ParseJID(newsletterJID)
if err != nil {
return fmt.Errorf("invalid newsletter JID: %w", err)
}
c.mu.Lock()
client := c.client
c.mu.Unlock()
if client == nil {
return fmt.Errorf("whatsapp not connected")
}
msg := &waE2E.Message{Conversation: proto.String(content)}
_, err = client.SendMessage(ctx, jid, msg)
return err
}
// parseJID converts a chat ID (phone number or JID string) to types.JID.
func parseJID(s string) (types.JID, error) {
s = strings.TrimSpace(s)