From 9df0dd635990bbd5c70e8b054b77e6b97f42d724 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 18 Mar 2026 09:16:33 +0000 Subject: [PATCH] feat: add native gmessages channel using libgm Added a new native `gmessages` channel implementation to JANE that connects directly to Google Messages via the `go.mau.fi/mautrix-gmessages` (libgm) library. This enables JANE to autonomously read/send SMS and RCS messages, parse incoming media, fetch contacts, and gracefully handle pairing by logging a QR code to the terminal if the session is not yet authenticated. A local SQLite database in WAL mode is included to sync and persist conversations, contacts, and message states. Key features: - Core integration with JANE's bus (`HandleMessage`, `Send`, `SendMedia`). - QR code pairing workflow in terminal. - Background sync of `contacts`, `conversations`, and `messages`. - Downloading incoming MMS/RCS attachments into JANE's `mediaStore`. - Uploading and sending rich media with retry logic. - Reaction support by converting and appending them to text bodies. Co-authored-by: hobbyistlabs-coder <267281733+hobbyistlabs-coder@users.noreply.github.com> --- cmd/picoclaw/internal/gateway/helpers.go | 1 + go.mod | 3 + pkg/channels/gmessages/client.go | 243 ++++++++++++++ pkg/channels/gmessages/db.go | 178 ++++++++++ pkg/channels/gmessages/events.go | 396 +++++++++++++++++++++++ pkg/channels/gmessages/gmessages.go | 66 ++++ pkg/channels/gmessages/init.go | 13 + pkg/channels/gmessages/messages.go | 226 +++++++++++++ pkg/config/channels.go | 17 +- 9 files changed, 1140 insertions(+), 3 deletions(-) create mode 100644 pkg/channels/gmessages/client.go create mode 100644 pkg/channels/gmessages/db.go create mode 100644 pkg/channels/gmessages/events.go create mode 100644 pkg/channels/gmessages/gmessages.go create mode 100644 pkg/channels/gmessages/init.go create mode 100644 pkg/channels/gmessages/messages.go diff --git a/cmd/picoclaw/internal/gateway/helpers.go b/cmd/picoclaw/internal/gateway/helpers.go index 19f5fb236..353883f1d 100644 --- a/cmd/picoclaw/internal/gateway/helpers.go +++ b/cmd/picoclaw/internal/gateway/helpers.go @@ -23,6 +23,7 @@ import ( _ "jane/pkg/channels/pico" _ "jane/pkg/channels/qq" _ "jane/pkg/channels/slack" + _ "jane/pkg/channels/gmessages" _ "jane/pkg/channels/telegram" _ "jane/pkg/channels/whatsapp" _ "jane/pkg/channels/whatsapp_native" diff --git a/go.mod b/go.mod index bab564939..50f084c4b 100644 --- a/go.mod +++ b/go.mod @@ -29,6 +29,7 @@ require ( github.com/stretchr/testify v1.11.1 github.com/tencent-connect/botgo v0.2.1 github.com/traefik/yaegi v0.16.1 + go.mau.fi/mautrix-gmessages v0.0.0-00010101000000-000000000000 go.mau.fi/whatsmeow v0.0.0-20260219150138-7ae702b1eed4 golang.org/x/oauth2 v0.35.0 golang.org/x/time v0.14.0 @@ -104,3 +105,5 @@ require ( golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.41.0 // indirect ) + +replace go.mau.fi/mautrix-gmessages => github.com/MaxGhenis/gmessages v0.2602.1-0.20260302032635-a6d7f9898a99 diff --git a/pkg/channels/gmessages/client.go b/pkg/channels/gmessages/client.go new file mode 100644 index 000000000..f86d20a96 --- /dev/null +++ b/pkg/channels/gmessages/client.go @@ -0,0 +1,243 @@ +package gmessages + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" + + "jane/pkg/logger" + "github.com/mdp/qrterminal/v3" + "github.com/rs/zerolog" + + "go.mau.fi/mautrix-gmessages/pkg/libgm" + "go.mau.fi/mautrix-gmessages/pkg/libgm/events" + "go.mau.fi/mautrix-gmessages/pkg/libgm/gmproto" +) + +type SessionData struct { + AuthDataJSON json.RawMessage `json:"auth_data"` + PushKeysJSON json.RawMessage `json:"push_keys,omitempty"` +} + +type GMClient struct { + GM *libgm.Client + SessionPath string +} + +func (c *GMessagesChannel) initClient(ctx context.Context) error { + dataDir := c.cfg.DataDir + if dataDir == "" { + home, err := os.UserHomeDir() + if err != nil { + return fmt.Errorf("could not get home dir and data_dir is empty: %w", err) + } + dataDir = filepath.Join(home, ".picoclaw", "gmessages") + } + + if err := os.MkdirAll(dataDir, 0755); err != nil { + return fmt.Errorf("failed to create gmessages data dir: %w", err) + } + + sessionPath := filepath.Join(dataDir, "session.json") + dbPath := filepath.Join(dataDir, "messages.db") + + // Initialize DB (we will fill this in db.go) + store, err := NewStore(dbPath) + if err != nil { + return fmt.Errorf("failed to initialize db: %w", err) + } + + c.state.store = store + + // Init LibGM Logger using zerolog (placeholder since GetLogger isn't exported in jane/pkg/logger) + // For production we can proxy mautrix logs to our system logger but keeping it simple for now. + gmLogger := zerolog.Nop() + + var client *GMClient + + // Try loading session + sessionData, err := loadSession(sessionPath) + if err == nil { + logger.InfoCF("channels.gmessages", "Loaded existing session", nil) + authData := libgm.NewAuthData() + if err := json.Unmarshal(sessionData.AuthDataJSON, authData); err != nil { + return fmt.Errorf("unmarshal auth data: %w", err) + } + + var pushKeys *libgm.PushKeys + if len(sessionData.PushKeysJSON) > 0 { + pushKeys = &libgm.PushKeys{} + if err := json.Unmarshal(sessionData.PushKeysJSON, pushKeys); err != nil { + return fmt.Errorf("unmarshal push keys: %w", err) + } + } + + gmClient := libgm.NewClient(authData, pushKeys, gmLogger) + client = &GMClient{GM: gmClient, SessionPath: sessionPath} + } else { + logger.InfoCF("channels.gmessages", "No session found or failed to load. Initiating pairing", map[string]any{"err": err.Error()}) + authData := libgm.NewAuthData() + gmClient := libgm.NewClient(authData, nil, gmLogger) + client = &GMClient{GM: gmClient, SessionPath: sessionPath} + + if err := c.handlePairing(ctx, client); err != nil { + return err + } + } + + c.state.client = client + + // Setup event handler + handler := &EventHandler{ + Store: store, + Channel: c, + SessionPath: sessionPath, + Client: client, + } + + client.GM.SetEventHandler(handler.Handle) + + // Connect + if err := client.GM.Connect(); err != nil { + return fmt.Errorf("failed to connect libgm: %w", err) + } + + logger.InfoCF("channels.gmessages", "gmessages client connected successfully", nil) + + return nil +} + +func (c *GMessagesChannel) stopClient(ctx context.Context) { + if c.state.client != nil { + client := c.state.client + if client.GM != nil { + client.GM.Disconnect() + } + } + if c.state.store != nil { + store := c.state.store + store.Close() + } +} + +func (c *GMClient) SessionData() (*SessionData, error) { + authJSON, err := json.Marshal(c.GM.AuthData) + if err != nil { + return nil, fmt.Errorf("marshal auth data: %w", err) + } + var pushJSON json.RawMessage + if c.GM.PushKeys != nil { + pushJSON, err = json.Marshal(c.GM.PushKeys) + if err != nil { + return nil, fmt.Errorf("marshal push keys: %w", err) + } + } + return &SessionData{ + AuthDataJSON: authJSON, + PushKeysJSON: pushJSON, + }, nil +} + +func loadSession(path string) (*SessionData, error) { + b, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var s SessionData + if err := json.Unmarshal(b, &s); err != nil { + return nil, err + } + return &s, nil +} + +func saveSession(path string, data *SessionData) error { + b, err := json.MarshalIndent(data, "", " ") + if err != nil { + return err + } + return os.WriteFile(path, b, 0600) +} + +func (c *GMessagesChannel) handlePairing(ctx context.Context, client *GMClient) error { + pairingCh := make(chan struct{}) + var pairErr error + + handler := func(rawEvt any) { + switch evt := rawEvt.(type) { + case *events.PairSuccessful: + logger.InfoCF("channels.gmessages", "Pairing successful", map[string]any{"phone_id": evt.PhoneID}) + + // Save session + sessionData, err := client.SessionData() + if err != nil { + logger.ErrorCF("channels.gmessages", "Failed to get session data", map[string]any{"error": err.Error()}) + } else { + if err := saveSession(client.SessionPath, sessionData); err != nil { + logger.ErrorCF("channels.gmessages", "Failed to save session", map[string]any{"error": err.Error()}) + } else { + logger.InfoCF("channels.gmessages", "Session saved to file", map[string]any{"path": client.SessionPath}) + } + } + + close(pairingCh) + case *events.ListenFatalError: + pairErr = evt.Error + close(pairingCh) + } + } + + client.GM.SetEventHandler(handler) + + logger.InfoCF("channels.gmessages", "Starting pairing process...", nil) + + var pairErr2 error + pairCB := func(data *gmproto.PairedData) { + logger.InfoCF("channels.gmessages", "Pairing successful", map[string]any{"phone_id": data.GetMobile().GetSourceID()}) + + sessionData, err := client.SessionData() + if err != nil { + logger.ErrorCF("channels.gmessages", "Failed to get session data", map[string]any{"error": err.Error()}) + } else { + if err := saveSession(client.SessionPath, sessionData); err != nil { + logger.ErrorCF("channels.gmessages", "Failed to save session", map[string]any{"error": err.Error()}) + } else { + logger.InfoCF("channels.gmessages", "Session saved to file", map[string]any{"path": client.SessionPath}) + } + } + + close(pairingCh) + } + client.GM.PairCallback.Store(&pairCB) + + qrURL, err := client.GM.StartLogin() + if err != nil { + return fmt.Errorf("failed to start pairing login: %w", err) + } + + fmt.Println("\n=== Scan this QR code in Google Messages (Device Pairing) ===") + qrterminal.GenerateHalfBlock(qrURL, qrterminal.L, os.Stdout) + fmt.Println("Waiting for pairing...") + + + select { + case <-pairingCh: + if pairErr != nil { + return fmt.Errorf("pairing failed: %w", pairErr) + } + if pairErr2 != nil { + return fmt.Errorf("pairing failed: %w", pairErr2) + } + // pairing successful, we disconnect so the main initClient can set the real event handler and reconnect + client.GM.Disconnect() + return nil + case <-ctx.Done(): + client.GM.Disconnect() + return ctx.Err() + case <-time.After(5 * time.Minute): + client.GM.Disconnect() + return fmt.Errorf("pairing timed out") + } +} diff --git a/pkg/channels/gmessages/db.go b/pkg/channels/gmessages/db.go new file mode 100644 index 000000000..4e6959c8a --- /dev/null +++ b/pkg/channels/gmessages/db.go @@ -0,0 +1,178 @@ +package gmessages + +import ( + "database/sql" + "fmt" + + _ "modernc.org/sqlite" +) + +type Message struct { + MessageID string + ConversationID string + SenderName string + SenderNumber string + Body string + MediaID string + MimeType string + DecryptionKey string + Reactions string + ReplyToID string + TimestampMS int64 + Status string + IsFromMe bool +} + +type Conversation struct { + ConversationID string + Name string + IsGroup bool + Participants string + LastMessageTS int64 + UnreadCount int +} + +type Contact struct { + Number string + Name string +} + +type Store struct { + db *sql.DB +} + +func NewStore(dbPath string) (*Store, error) { + // Use modernc.org/sqlite (pure Go) and WAL mode for concurrency + db, err := sql.Open("sqlite", dbPath+"?_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)") + if err != nil { + return nil, fmt.Errorf("open sqlite db: %w", err) + } + + if err := db.Ping(); err != nil { + return nil, fmt.Errorf("ping db: %w", err) + } + + if err := createSchema(db); err != nil { + return nil, fmt.Errorf("create schema: %w", err) + } + + return &Store{db: db}, nil +} + +func (s *Store) Close() error { + return s.db.Close() +} + +func createSchema(db *sql.DB) error { + schema := ` + CREATE TABLE IF NOT EXISTS messages ( + message_id TEXT PRIMARY KEY, + conversation_id TEXT NOT NULL, + sender_name TEXT, + sender_number TEXT, + body TEXT, + media_id TEXT, + mime_type TEXT, + decryption_key TEXT, + reactions TEXT, + reply_to_id TEXT, + timestamp_ms INTEGER NOT NULL, + status TEXT, + is_from_me BOOLEAN + ); + + CREATE INDEX IF NOT EXISTS idx_messages_conversation_id ON messages(conversation_id); + CREATE INDEX IF NOT EXISTS idx_messages_timestamp ON messages(timestamp_ms DESC); + CREATE INDEX IF NOT EXISTS idx_messages_body ON messages(body) WHERE body IS NOT NULL; + + CREATE TABLE IF NOT EXISTS conversations ( + conversation_id TEXT PRIMARY KEY, + name TEXT, + is_group BOOLEAN, + participants TEXT, + last_message_ts INTEGER, + unread_count INTEGER + ); + + CREATE INDEX IF NOT EXISTS idx_conversations_last_message_ts ON conversations(last_message_ts DESC); + + CREATE TABLE IF NOT EXISTS contacts ( + number TEXT PRIMARY KEY, + name TEXT + ); + ` + _, err := db.Exec(schema) + return err +} + +func (s *Store) UpsertMessage(m *Message) error { + query := ` + INSERT INTO messages ( + message_id, conversation_id, sender_name, sender_number, body, + media_id, mime_type, decryption_key, reactions, reply_to_id, + timestamp_ms, status, is_from_me + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(message_id) DO UPDATE SET + status=excluded.status, + reactions=excluded.reactions, + body=excluded.body, + media_id=excluded.media_id, + mime_type=excluded.mime_type + ` + _, err := s.db.Exec(query, + m.MessageID, m.ConversationID, m.SenderName, m.SenderNumber, m.Body, + m.MediaID, m.MimeType, m.DecryptionKey, m.Reactions, m.ReplyToID, + m.TimestampMS, m.Status, m.IsFromMe, + ) + return err +} + +func (s *Store) DeleteTmpMessages(conversationID string) (int64, error) { + res, err := s.db.Exec(`DELETE FROM messages WHERE conversation_id = ? AND message_id LIKE 'tmp_%'`, conversationID) + if err != nil { + return 0, err + } + return res.RowsAffected() +} + +func (s *Store) UpsertConversation(c *Conversation) error { + query := ` + INSERT INTO conversations ( + conversation_id, name, is_group, participants, last_message_ts, unread_count + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(conversation_id) DO UPDATE SET + name=excluded.name, + is_group=excluded.is_group, + participants=excluded.participants, + last_message_ts=excluded.last_message_ts, + unread_count=excluded.unread_count + ` + _, err := s.db.Exec(query, c.ConversationID, c.Name, c.IsGroup, c.Participants, c.LastMessageTS, c.UnreadCount) + return err +} + +func (s *Store) GetConversationIDByNumber(number string) (string, error) { + // Look for a 1:1 conversation that has this number in participants + query := ` + SELECT conversation_id + FROM conversations + WHERE is_group = false + AND participants LIKE ? + LIMIT 1 + ` + var convID string + err := s.db.QueryRow(query, "%"+number+"%").Scan(&convID) + if err == sql.ErrNoRows { + return "", nil // not found + } + return convID, err +} + +func (s *Store) UpsertContact(number, name string) error { + query := ` + INSERT INTO contacts (number, name) VALUES (?, ?) + ON CONFLICT(number) DO UPDATE SET name=excluded.name + ` + _, err := s.db.Exec(query, number, name) + return err +} diff --git a/pkg/channels/gmessages/events.go b/pkg/channels/gmessages/events.go new file mode 100644 index 000000000..bd577400b --- /dev/null +++ b/pkg/channels/gmessages/events.go @@ -0,0 +1,396 @@ +package gmessages + +import ( + "context" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "strings" + + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/logger" + mediaPkg "jane/pkg/media" + + "go.mau.fi/mautrix-gmessages/pkg/libgm" + "go.mau.fi/mautrix-gmessages/pkg/libgm/events" + "go.mau.fi/mautrix-gmessages/pkg/libgm/gmproto" +) + +type EventHandler struct { + Store *Store + Channel *GMessagesChannel + SessionPath string + Client *GMClient +} + +func (h *EventHandler) Handle(rawEvt any) { + switch evt := rawEvt.(type) { + case *events.ClientReady: + h.handleClientReady(evt) + case *libgm.WrappedMessage: + h.handleMessage(evt) + case *gmproto.Conversation: + h.handleConversation(evt) + case *events.AuthTokenRefreshed: + h.handleAuthRefresh() + case *events.ListenFatalError: + logger.ErrorCF("channels.gmessages", "Listen fatal error", map[string]any{"err": evt.Error}) + h.Channel.Stop(context.Background()) + case *events.ListenTemporaryError: + logger.WarnCF("channels.gmessages", "Listen temporary error", map[string]any{"err": evt.Error}) + case *events.ListenRecovered: + logger.InfoCF("channels.gmessages", "Listen recovered", nil) + case *events.PhoneNotResponding: + logger.WarnCF("channels.gmessages", "Phone not responding", nil) + case *events.PhoneRespondingAgain: + logger.InfoCF("channels.gmessages", "Phone responding again", nil) + default: + logger.DebugCF("channels.gmessages", "Unhandled event", map[string]any{"type": fmt.Sprintf("%T", evt)}) + } +} + +func (h *EventHandler) handleClientReady(evt *events.ClientReady) { + logger.InfoCF("channels.gmessages", "Client ready", map[string]any{ + "session_id": evt.SessionID, + "conversations": len(evt.Conversations), + }) + + for _, conv := range evt.Conversations { + h.handleConversation(conv) + } + + // Fetch contacts in background to populate our DB + go func() { + resp, err := h.Client.GM.ListContacts() + if err != nil { + logger.WarnCF("channels.gmessages", "Failed to list contacts", map[string]any{"err": err}) + return + } + + for _, contact := range resp.GetContacts() { + name := contact.GetName() + if num := contact.GetNumber(); num != nil && num.GetNumber() != "" { + h.Store.UpsertContact(num.GetNumber(), name) + } + } + logger.InfoCF("channels.gmessages", "Finished syncing contacts", map[string]any{"count": len(resp.GetContacts())}) + }() +} + +func (h *EventHandler) handleMessage(evt *libgm.WrappedMessage) { + msg := evt.Message + body := ExtractMessageBody(msg) + senderName, senderNumber := ExtractSenderInfo(msg) + + status := "unknown" + if ms := msg.GetMessageStatus(); ms != nil { + status = ms.GetStatus().String() + } + + dbMsg := &Message{ + MessageID: msg.GetMessageID(), + ConversationID: msg.GetConversationID(), + SenderName: senderName, + SenderNumber: senderNumber, + Body: body, + TimestampMS: msg.GetTimestamp() / 1000, + Status: status, + IsFromMe: msg.GetSenderParticipant() != nil && msg.GetSenderParticipant().GetIsMe(), + } + + var mediaURLs []string + + if media := ExtractMediaInfo(msg); media != nil { + dbMsg.MediaID = media.MediaID + dbMsg.MimeType = media.MimeType + dbMsg.DecryptionKey = hex.EncodeToString(media.DecryptionKey) + + saveToStore := func(data []byte, mimeType string) { + if h.Channel.GetMediaStore() == nil { + return + } + + // Save to a temporary file + tmpFile, err := os.CreateTemp("", "gmessages-media-*") + if err != nil { + logger.ErrorCF("channels.gmessages", "Failed to create temp file for media", map[string]any{"err": err}) + return + } + defer tmpFile.Close() + + if _, err := tmpFile.Write(data); err != nil { + logger.ErrorCF("channels.gmessages", "Failed to write media data to temp file", map[string]any{"err": err}) + return + } + + scope := channels.BuildMediaScope(h.Channel.Name(), msg.GetConversationID(), msg.GetMessageID()) + meta := mediaPkg.MediaMeta{ContentType: mimeType, Source: "gmessages"} + + ref, err := h.Channel.GetMediaStore().Store(tmpFile.Name(), meta, scope) + if err == nil { + mediaURLs = append(mediaURLs, ref) + } else { + logger.ErrorCF("channels.gmessages", "Failed to save media to store", map[string]any{"err": err}) + } + } + + if media.InlineData != nil { + saveToStore(media.InlineData, media.MimeType) + } else if media.MediaID != "" && media.DecryptionKey != nil { + data, err := h.Client.GM.DownloadMedia(media.MediaID, media.DecryptionKey) + if err == nil { + saveToStore(data, media.MimeType) + } else { + logger.ErrorCF("channels.gmessages", "Failed to download media from Google Messages", map[string]any{"err": err}) + } + } + } + + if reactions := ExtractReactions(msg); reactions != nil { + if b, err := json.Marshal(reactions); err == nil { + dbMsg.Reactions = string(b) + // Append reactions to body for bot visibility if body isn't empty + var reactionStrs []string + for _, r := range reactions { + reactionStrs = append(reactionStrs, r.Emoji) + } + if len(reactionStrs) > 0 { + body = body + "\n[Reactions: " + strings.Join(reactionStrs, " ") + "]" + dbMsg.Body = body + } + } + } + dbMsg.ReplyToID = ExtractReplyToID(msg) + + if err := h.Store.UpsertMessage(dbMsg); err != nil { + logger.ErrorCF("channels.gmessages", "Failed to store message", map[string]any{"err": err, "msg_id": dbMsg.MessageID}) + return + } + + if dbMsg.IsFromMe && !strings.HasPrefix(dbMsg.MessageID, "tmp_") { + h.Store.DeleteTmpMessages(dbMsg.ConversationID) + return // we don't dispatch our own messages usually, unless requested + } + + // Dispatch to bot + if !evt.IsOld && !dbMsg.IsFromMe { + senderID := senderNumber + if senderID == "" { + senderID = senderName + } + + // Map ChatID to phone number or conversation ID. In 1:1 we'll use phone number or conv ID + chatID := msg.GetConversationID() // fallback to group/thread ID + if chatID == "" { + chatID = senderID + } + + // determine if it's a group from DB (we do it in the next steps) + + peer := bus.Peer{ + ID: chatID, + } + + senderInfo := bus.SenderInfo{ + PlatformID: senderID, + CanonicalID: senderID, + DisplayName: senderName, + } + + metadata := map[string]string{ + "conversation_id": msg.GetConversationID(), + } + + h.Channel.HandleMessage( + context.Background(), + peer, + msg.GetMessageID(), + senderID, + chatID, + body, + mediaURLs, + metadata, + senderInfo, + ) + } + + logger.DebugCF("channels.gmessages", "Stored message", map[string]any{ + "msg_id": dbMsg.MessageID, + "from": senderName, + "is_old": evt.IsOld, + }) +} + +func (h *EventHandler) handleConversation(conv *gmproto.Conversation) { + participantsJSON := "[]" + if ps := conv.GetParticipants(); len(ps) > 0 { + type pInfo struct { + Name string `json:"name"` + Number string `json:"number"` + IsMe bool `json:"is_me,omitempty"` + } + var infos []pInfo + for _, p := range ps { + info := pInfo{ + Name: p.GetFullName(), + IsMe: p.GetIsMe(), + } + if id := p.GetID(); id != nil { + info.Number = id.GetNumber() + } + if info.Number == "" { + info.Number = p.GetFormattedNumber() + } + infos = append(infos, info) + } + if b, err := json.Marshal(infos); err == nil { + participantsJSON = string(b) + } + } + + unread := 0 + if conv.GetUnread() { + unread = 1 + } + + dbConv := &Conversation{ + ConversationID: conv.GetConversationID(), + Name: conv.GetName(), + IsGroup: conv.GetIsGroupChat(), + Participants: participantsJSON, + LastMessageTS: conv.GetLastMessageTimestamp() / 1000, + UnreadCount: unread, + } + + if err := h.Store.UpsertConversation(dbConv); err != nil { + logger.ErrorCF("channels.gmessages", "Failed to store conversation", map[string]any{"err": err, "conv_id": dbConv.ConversationID}) + return + } +} + +func (h *EventHandler) handleAuthRefresh() { + if h.Client == nil || h.SessionPath == "" { + return + } + sessionData, err := h.Client.SessionData() + if err != nil { + logger.ErrorCF("channels.gmessages", "Failed to get session data for save", map[string]any{"err": err}) + return + } + if err := saveSession(h.SessionPath, sessionData); err != nil { + logger.ErrorCF("channels.gmessages", "Failed to save refreshed session", map[string]any{"err": err}) + return + } + logger.DebugCF("channels.gmessages", "Saved refreshed auth token", nil) +} + +// Extractors mapped from libgm + +func ExtractMessageBody(msg *gmproto.Message) string { + for _, info := range msg.GetMessageInfo() { + if mc := info.GetMessageContent(); mc != nil { + return mc.GetContent() + } + } + return "" +} + +type MediaInfo struct { + MediaID string + MimeType string + MediaName string + DecryptionKey []byte + Size int64 + ThumbnailMediaID string + ThumbnailDecryptionKey []byte + InlineData []byte +} + +func ExtractMediaInfo(msg *gmproto.Message) *MediaInfo { + for _, info := range msg.GetMessageInfo() { + if mc := info.GetMediaContent(); mc != nil { + mime := mc.GetMimeType() + if mime == "" { + switch { + case mc.GetFormat() >= 1 && mc.GetFormat() <= 7: + mime = "image/jpeg" + default: + mime = "application/octet-stream" + } + } + + mi := &MediaInfo{ + MediaID: mc.GetMediaID(), + MimeType: mime, + MediaName: mc.GetMediaName(), + DecryptionKey: mc.GetDecryptionKey(), + Size: mc.GetSize(), + ThumbnailMediaID: mc.GetThumbnailMediaID(), + ThumbnailDecryptionKey: mc.GetThumbnailDecryptionKey(), + InlineData: mc.GetMediaData(), + } + + if mi.MediaID == "" && mi.ThumbnailMediaID != "" { + mi.MediaID = mi.ThumbnailMediaID + mi.DecryptionKey = mi.ThumbnailDecryptionKey + } + + return mi + } + } + return nil +} + +type Reaction struct { + Emoji string `json:"emoji"` + Count int `json:"count"` +} + +func ExtractReactions(msg *gmproto.Message) []Reaction { + entries := msg.GetReactions() + if len(entries) == 0 { + return nil + } + var reactions []Reaction + for _, entry := range entries { + if data := entry.GetData(); data != nil { + emoji := data.GetUnicode() + if emoji == "" { + continue + } + reactions = append(reactions, Reaction{ + Emoji: emoji, + Count: len(entry.GetParticipantIDs()), + }) + } + } + if len(reactions) == 0 { + return nil + } + return reactions +} + +func ExtractReplyToID(msg *gmproto.Message) string { + if rm := msg.GetReplyMessage(); rm != nil { + return rm.GetMessageID() + } + return "" +} + +func ExtractSenderInfo(msg *gmproto.Message) (name, number string) { + if p := msg.GetSenderParticipant(); p != nil { + name = p.GetFullName() + if name == "" { + name = p.GetFirstName() + } + if id := p.GetID(); id != nil { + number = id.GetNumber() + } + if number == "" { + number = p.GetFormattedNumber() + } + } + return +} diff --git a/pkg/channels/gmessages/gmessages.go b/pkg/channels/gmessages/gmessages.go new file mode 100644 index 000000000..aa88a6bce --- /dev/null +++ b/pkg/channels/gmessages/gmessages.go @@ -0,0 +1,66 @@ +package gmessages + +import ( + "context" + + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/config" +) + +type GMessagesChannel struct { + *channels.BaseChannel + cfg config.GMessagesConfig + state *ClientState +} + +// ClientState holds the internal dependencies like libgm Client and DB. +// It will be instantiated and injected when the channel starts. +type ClientState struct { + client *GMClient + store *Store +} + +func NewGMessagesChannel(cfg config.GMessagesConfig, b *bus.MessageBus) (*GMessagesChannel, error) { + bc := channels.NewBaseChannel( + "gmessages", + cfg, + b, + cfg.AllowFrom, + channels.WithGroupTrigger(cfg.GroupTrigger), + channels.WithReasoningChannelID(cfg.ReasoningChannelID), + ) + + ch := &GMessagesChannel{ + BaseChannel: bc, + cfg: cfg, + state: &ClientState{}, + } + bc.SetOwner(ch) + return ch, nil +} + +func (c *GMessagesChannel) Start(ctx context.Context) error { + if !c.cfg.Enabled { + return nil + } + + c.SetRunning(true) + + // Call the separate client initialization + if err := c.initClient(ctx); err != nil { + c.SetRunning(false) + return err + } + + return nil +} + +func (c *GMessagesChannel) Stop(ctx context.Context) error { + c.SetRunning(false) + + // Disconnect client if exists + c.stopClient(ctx) + + return nil +} diff --git a/pkg/channels/gmessages/init.go b/pkg/channels/gmessages/init.go new file mode 100644 index 000000000..478a8fc50 --- /dev/null +++ b/pkg/channels/gmessages/init.go @@ -0,0 +1,13 @@ +package gmessages + +import ( + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/config" +) + +func init() { + channels.RegisterFactory("gmessages", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { + return NewGMessagesChannel(cfg.Channels.GMessages, b) + }) +} diff --git a/pkg/channels/gmessages/messages.go b/pkg/channels/gmessages/messages.go new file mode 100644 index 000000000..77b6c4112 --- /dev/null +++ b/pkg/channels/gmessages/messages.go @@ -0,0 +1,226 @@ +package gmessages + +import ( + "context" + "fmt" + "math/rand" + "os" + "strings" + "time" + + "jane/pkg/bus" + "jane/pkg/logger" + + "go.mau.fi/mautrix-gmessages/pkg/libgm/gmproto" +) + +// ContactNumberMysteriousInt is the default value for the MysteriousInt field +const ContactNumberMysteriousInt = 7 + +func (c *GMessagesChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + client := c.state.client + store := c.state.store + + convID := msg.ChatID + + var sim *gmproto.SIMPayload + var participantID string + + if !strings.HasPrefix(convID, "bugle:") { + existingID, err := store.GetConversationIDByNumber(msg.ChatID) + if err == nil && existingID != "" { + convID = existingID + } else { + logger.InfoCF("channels.gmessages", "Creating new conversation", map[string]any{"number": msg.ChatID}) + + // Build numbers request + numbers := []*gmproto.ContactNumber{{ + MysteriousInt: ContactNumberMysteriousInt, + Number: msg.ChatID, + Number2: msg.ChatID, + }} + + req := &gmproto.GetOrCreateConversationRequest{ + Numbers: numbers, + } + newConv, err := client.GM.GetOrCreateConversation(req) + if err != nil { + return fmt.Errorf("failed to create conversation for %s: %w", msg.ChatID, err) + } + conv := newConv.GetConversation() + if conv == nil { + return fmt.Errorf("no conversation returned for %s", msg.ChatID) + } + convID = conv.GetConversationID() + + // Extract sim for new conv + for _, p := range conv.GetParticipants() { + if p.GetIsMe() { + if id := p.GetID(); id != nil { + participantID = id.GetNumber() + } + sim = p.GetSimPayload() + break + } + } + if sim == nil { + if sc := conv.GetSimCard(); sc != nil { + sim = sc.GetSIMData().GetSIMPayload() + } + } + } + } + + content := msg.Content + + tmpID := fmt.Sprintf("tmp_%012d", rand.Int63n(1e12)) + req := &gmproto.SendMessageRequest{ + ConversationID: convID, + MessagePayload: &gmproto.MessagePayload{ + TmpID: tmpID, + MessagePayloadContent: nil, + MessageInfo: []*gmproto.MessageInfo{{ + Data: &gmproto.MessageInfo_MessageContent{MessageContent: &gmproto.MessageContent{ + Content: content, + }}, + }}, + ConversationID: convID, + ParticipantID: participantID, + TmpID2: tmpID, + }, + SIMPayload: sim, + TmpID: tmpID, + } + if msg.ReplyToMessageID != "" { + req.Reply = &gmproto.ReplyPayload{ + MessageID: msg.ReplyToMessageID, + } + } + + _, err := client.GM.SendMessage(req) + if err != nil { + return fmt.Errorf("failed to send message: %w", err) + } + + logger.DebugCF("channels.gmessages", "Sent message", map[string]any{"conv_id": convID}) + return nil +} + +// SendMedia sends rich media using mautrix library. +func (c *GMessagesChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error { + client := c.state.client + store := c.state.store + + mediaStore := c.GetMediaStore() + if mediaStore == nil { + return fmt.Errorf("no media store available for channel %s", c.Name()) + } + + convID := msg.ChatID + + var sim *gmproto.SIMPayload + var participantID string + + // Ensure conversation exists or create one (similar logic to Send) + if !strings.HasPrefix(convID, "bugle:") { + existingID, err := store.GetConversationIDByNumber(msg.ChatID) + if err == nil && existingID != "" { + convID = existingID + } else { + numbers := []*gmproto.ContactNumber{{ + MysteriousInt: ContactNumberMysteriousInt, + Number: msg.ChatID, + Number2: msg.ChatID, + }} + req := &gmproto.GetOrCreateConversationRequest{Numbers: numbers} + newConv, err := client.GM.GetOrCreateConversation(req) + if err != nil { + return fmt.Errorf("failed to create conversation for %s: %w", msg.ChatID, err) + } + conv := newConv.GetConversation() + if conv == nil { + return fmt.Errorf("no conversation returned for %s", msg.ChatID) + } + convID = conv.GetConversationID() + + for _, p := range conv.GetParticipants() { + if p.GetIsMe() { + if id := p.GetID(); id != nil { + participantID = id.GetNumber() + } + sim = p.GetSimPayload() + break + } + } + if sim == nil { + if sc := conv.GetSimCard(); sc != nil { + sim = sc.GetSIMData().GetSIMPayload() + } + } + } + } + + for _, part := range msg.Parts { + localPath, err := mediaStore.Resolve(part.Ref) + if err != nil { + logger.ErrorCF("channels.gmessages", "Failed to resolve media ref", map[string]any{"ref": part.Ref, "err": err}) + continue + } + + fileData, err := os.ReadFile(localPath) + if err != nil { + logger.ErrorCF("channels.gmessages", "Failed to read media file", map[string]any{"path": localPath, "err": err}) + continue + } + + mimeType := part.ContentType + if mimeType == "" { + mimeType = "application/octet-stream" + } + + filename := part.Filename + if filename == "" { + filename = fmt.Sprintf("media-%d", time.Now().UnixNano()) + } + + // Retry media upload in case of transient network errors + var mediaContent *gmproto.MediaContent + var errUpload error + for attempt := 1; attempt <= 3; attempt++ { + mediaContent, errUpload = client.GM.UploadMedia(fileData, mimeType, filename) + if errUpload == nil { + break + } + time.Sleep(time.Duration(attempt*500) * time.Millisecond) + } + + if errUpload != nil { + logger.ErrorCF("channels.gmessages", "Failed to upload media after retries", map[string]any{"err": errUpload}) + continue + } + + tmpID := fmt.Sprintf("tmp_%012d", rand.Int63n(1e12)) + req := &gmproto.SendMessageRequest{ + ConversationID: convID, + MessagePayload: &gmproto.MessagePayload{ + TmpID: tmpID, + MessageInfo: []*gmproto.MessageInfo{{ + Data: &gmproto.MessageInfo_MediaContent{MediaContent: mediaContent}, + }}, + ConversationID: convID, + ParticipantID: participantID, + TmpID2: tmpID, + }, + SIMPayload: sim, + TmpID: tmpID, + } + + _, err = client.GM.SendMessage(req) + if err != nil { + logger.ErrorCF("channels.gmessages", "Failed to send media message", map[string]any{"err": err}) + continue + } + } + + return nil +} diff --git a/pkg/config/channels.go b/pkg/config/channels.go index cbc7d7db1..fa702d382 100644 --- a/pkg/config/channels.go +++ b/pkg/config/channels.go @@ -10,9 +10,10 @@ type ChannelsConfig struct { Slack SlackConfig `json:"slack"` Matrix MatrixConfig `json:"matrix"` LINE LINEConfig `json:"line"` - OneBot OneBotConfig `json:"onebot"` - Pico PicoConfig `json:"pico"` - IRC IRCConfig `json:"irc"` + OneBot OneBotConfig `json:"onebot"` + Pico PicoConfig `json:"pico"` + IRC IRCConfig `json:"irc"` + GMessages GMessagesConfig `json:"gmessages"` } // GroupTriggerConfig controls when the bot responds in group chats. @@ -176,3 +177,13 @@ type IRCConfig struct { Typing TypingConfig `json:"typing,omitempty"` ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_IRC_REASONING_CHANNEL_ID"` } + +type GMessagesConfig struct { + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_GMESSAGES_ENABLED"` + DataDir string `json:"data_dir" env:"PICOCLAW_CHANNELS_GMESSAGES_DATA_DIR"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_GMESSAGES_ALLOW_FROM"` + GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"` + Typing TypingConfig `json:"typing,omitempty"` + Placeholder PlaceholderConfig `json:"placeholder,omitempty"` + ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_GMESSAGES_REASONING_CHANNEL_ID"` +}