diff --git a/pkg/channels/matrix/cache.go b/pkg/channels/matrix/cache.go new file mode 100644 index 000000000..13b3f6846 --- /dev/null +++ b/pkg/channels/matrix/cache.go @@ -0,0 +1,129 @@ +package matrix + +import ( + "context" + "sync" + "time" +) + +type roomKindCacheEntry struct { + isGroup bool + expiresAt time.Time + touchedAt time.Time +} + +type roomKindCache struct { + mu sync.Mutex + entries map[string]roomKindCacheEntry + maxEntries int + ttl time.Duration +} + +func newRoomKindCache(maxEntries int, ttl time.Duration) *roomKindCache { + if maxEntries <= 0 { + maxEntries = roomKindCacheMaxEntries + } + if ttl <= 0 { + ttl = roomKindCacheTTL + } + + return &roomKindCache{ + entries: make(map[string]roomKindCacheEntry), + maxEntries: maxEntries, + ttl: ttl, + } +} + +func (c *roomKindCache) get(roomID string, now time.Time) (bool, bool) { + c.mu.Lock() + defer c.mu.Unlock() + + entry, ok := c.entries[roomID] + if !ok { + return false, false + } + if !entry.expiresAt.After(now) { + delete(c.entries, roomID) + return false, false + } + + return entry.isGroup, true +} + +func (c *roomKindCache) set(roomID string, isGroup bool, now time.Time) { + c.mu.Lock() + defer c.mu.Unlock() + + if entry, ok := c.entries[roomID]; ok { + entry.isGroup = isGroup + entry.expiresAt = now.Add(c.ttl) + entry.touchedAt = now + c.entries[roomID] = entry + return + } + + c.cleanupExpiredLocked(now) + for len(c.entries) >= c.maxEntries { + if !c.evictOldestLocked() { + break + } + } + + c.entries[roomID] = roomKindCacheEntry{ + isGroup: isGroup, + expiresAt: now.Add(c.ttl), + touchedAt: now, + } +} + +func (c *roomKindCache) cleanupExpired(now time.Time) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.cleanupExpiredLocked(now) +} + +func (c *roomKindCache) cleanupExpiredLocked(now time.Time) int { + removed := 0 + for roomID, entry := range c.entries { + if !entry.expiresAt.After(now) { + delete(c.entries, roomID) + removed++ + } + } + return removed +} + +func (c *roomKindCache) evictOldestLocked() bool { + if len(c.entries) == 0 { + return false + } + + var ( + oldestRoomID string + oldestAt time.Time + ) + + for roomID, entry := range c.entries { + if oldestRoomID == "" || entry.touchedAt.Before(oldestAt) { + oldestRoomID = roomID + oldestAt = entry.touchedAt + } + } + + delete(c.entries, oldestRoomID) + return true +} + +func (c *MatrixChannel) runRoomKindCacheJanitor(ctx context.Context) { + ticker := time.NewTicker(roomKindCacheCleanupPeriod) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case now := <-ticker.C: + c.roomKindCache.cleanupExpired(now) + } + } +} diff --git a/pkg/channels/matrix/events.go b/pkg/channels/matrix/events.go new file mode 100644 index 000000000..4227d9040 --- /dev/null +++ b/pkg/channels/matrix/events.go @@ -0,0 +1,179 @@ +package matrix + +import ( + "context" + "fmt" + "strings" + "time" + + "maunium.net/go/mautrix/event" + + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/identity" + "jane/pkg/logger" +) + +func (c *MatrixChannel) handleMemberEvent(ctx context.Context, evt *event.Event) { + if !c.config.JoinOnInvite { + return + } + if evt == nil { + return + } + + member := evt.Content.AsMember() + if member.Membership != event.MembershipInvite { + return + } + if evt.GetStateKey() != c.client.UserID.String() { + return + } + + _, err := c.client.JoinRoomByID(c.baseContext(), evt.RoomID) + if err != nil { + logger.WarnCF("matrix", "Failed to auto-join invited room", map[string]any{ + "room_id": evt.RoomID.String(), + "error": err.Error(), + }) + return + } + + logger.InfoCF("matrix", "Joined room after invite", map[string]any{ + "room_id": evt.RoomID.String(), + }) +} + +func (c *MatrixChannel) handleMessageEvent(ctx context.Context, evt *event.Event) { + if evt == nil { + return + } + + // Ignore our own messages. + if evt.Sender == c.client.UserID { + return + } + + // Ignore historical events on first sync. + if time.UnixMilli(evt.Timestamp).Before(c.startTime) { + return + } + + msgEvt := evt.Content.AsMessage() + if msgEvt == nil { + return + } + + // Ignore edits. + if msgEvt.RelatesTo != nil && msgEvt.RelatesTo.GetReplaceID() != "" { + return + } + + roomID := evt.RoomID.String() + scope := channels.BuildMediaScope("matrix", roomID, evt.ID.String()) + + content, mediaPaths, ok := c.extractInboundContent(ctx, msgEvt, scope) + if !ok { + return + } + content = strings.TrimSpace(content) + if content == "" && len(mediaPaths) == 0 { + return + } + + senderID := evt.Sender.String() + sender := bus.SenderInfo{ + Platform: "matrix", + PlatformID: senderID, + CanonicalID: identity.BuildCanonicalID("matrix", senderID), + Username: senderID, + DisplayName: senderID, + } + + if !c.IsAllowedSender(sender) { + logger.DebugCF("matrix", "Message rejected by allowlist", map[string]any{ + "sender_id": senderID, + }) + return + } + + isGroup := c.isGroupRoom(ctx, evt.RoomID) + if isGroup { + isMentioned := c.isBotMentioned(msgEvt) + if isMentioned { + content = c.stripSelfMention(content) + } + respond, cleaned := c.ShouldRespondInGroup(isMentioned, content) + if !respond { + logger.DebugCF("matrix", "Ignoring group message by trigger rules", map[string]any{ + "room_id": roomID, + "is_mentioned": isMentioned, + "mention_only": c.config.GroupTrigger.MentionOnly, + "prefixes": c.config.GroupTrigger.Prefixes, + }) + return + } + content = cleaned + } else { + content = c.stripSelfMention(content) + } + + content = strings.TrimSpace(content) + if content == "" { + return + } + + peerKind := "direct" + peerID := senderID + if isGroup { + peerKind = "group" + peerID = roomID + } + + metadata := map[string]string{ + "room_id": roomID, + "timestamp": fmt.Sprintf("%d", evt.Timestamp), + "is_group": fmt.Sprintf("%t", isGroup), + "sender_raw": senderID, + } + + logger.DebugCF("matrix", "Received message", map[string]any{ + "sender_id": senderID, + "room_id": roomID, + "is_group": isGroup, + }) + + if replyTo := msgEvt.GetRelatesTo().GetReplyTo(); replyTo != "" { + metadata["reply_to_msg_id"] = replyTo.String() + } + + c.HandleMessage( + c.baseContext(), + bus.Peer{Kind: peerKind, ID: peerID}, + evt.ID.String(), + senderID, + roomID, + content, + mediaPaths, + metadata, + sender, + ) +} + +func (c *MatrixChannel) extractInboundContent( + ctx context.Context, + msgEvt *event.MessageEventContent, + scope string, +) (string, []string, bool) { + switch msgEvt.MsgType { + case event.MsgText, event.MsgNotice: + return msgEvt.Body, nil, true + case event.MsgImage, event.MsgAudio, event.MsgVideo, event.MsgFile: + return c.extractInboundMedia(ctx, msgEvt, scope) + default: + logger.DebugCF("matrix", "Ignoring unsupported matrix msgtype", map[string]any{ + "msgtype": msgEvt.MsgType, + }) + return "", nil, false + } +} diff --git a/pkg/channels/matrix/matrix.go b/pkg/channels/matrix/matrix.go index 9025639bb..28de51995 100644 --- a/pkg/channels/matrix/matrix.go +++ b/pkg/channels/matrix/matrix.go @@ -3,20 +3,11 @@ package matrix import ( "context" "fmt" - "html" - "io" - "mime" - "net/url" - "os" - "path/filepath" "regexp" "strings" "sync" "time" - "github.com/gomarkdown/markdown" - mdhtml "github.com/gomarkdown/markdown/html" - "github.com/gomarkdown/markdown/parser" "maunium.net/go/mautrix" "maunium.net/go/mautrix/event" "maunium.net/go/mautrix/id" @@ -24,9 +15,7 @@ import ( "jane/pkg/bus" "jane/pkg/channels" "jane/pkg/config" - "jane/pkg/identity" "jane/pkg/logger" - "jane/pkg/media" ) const ( @@ -35,137 +24,8 @@ const ( roomKindCacheTTL = 5 * time.Minute roomKindCacheCleanupPeriod = 1 * time.Minute roomKindCacheMaxEntries = 2048 - - matrixMediaTempDirName = "picoclaw_media" ) -var matrixMentionHrefRegexp = regexp.MustCompile(`(?i)]+href=["']([^"']+)["']`) - -type roomKindCacheEntry struct { - isGroup bool - expiresAt time.Time - touchedAt time.Time -} - -type roomKindCache struct { - mu sync.Mutex - entries map[string]roomKindCacheEntry - maxEntries int - ttl time.Duration -} - -func newRoomKindCache(maxEntries int, ttl time.Duration) *roomKindCache { - if maxEntries <= 0 { - maxEntries = roomKindCacheMaxEntries - } - if ttl <= 0 { - ttl = roomKindCacheTTL - } - - return &roomKindCache{ - entries: make(map[string]roomKindCacheEntry), - maxEntries: maxEntries, - ttl: ttl, - } -} - -func (c *roomKindCache) get(roomID string, now time.Time) (bool, bool) { - c.mu.Lock() - defer c.mu.Unlock() - - entry, ok := c.entries[roomID] - if !ok { - return false, false - } - if !entry.expiresAt.After(now) { - delete(c.entries, roomID) - return false, false - } - - return entry.isGroup, true -} - -func (c *roomKindCache) set(roomID string, isGroup bool, now time.Time) { - c.mu.Lock() - defer c.mu.Unlock() - - if entry, ok := c.entries[roomID]; ok { - entry.isGroup = isGroup - entry.expiresAt = now.Add(c.ttl) - entry.touchedAt = now - c.entries[roomID] = entry - return - } - - c.cleanupExpiredLocked(now) - for len(c.entries) >= c.maxEntries { - if !c.evictOldestLocked() { - break - } - } - - c.entries[roomID] = roomKindCacheEntry{ - isGroup: isGroup, - expiresAt: now.Add(c.ttl), - touchedAt: now, - } -} - -func (c *roomKindCache) cleanupExpired(now time.Time) int { - c.mu.Lock() - defer c.mu.Unlock() - return c.cleanupExpiredLocked(now) -} - -func (c *roomKindCache) cleanupExpiredLocked(now time.Time) int { - removed := 0 - for roomID, entry := range c.entries { - if !entry.expiresAt.After(now) { - delete(c.entries, roomID) - removed++ - } - } - return removed -} - -func (c *roomKindCache) evictOldestLocked() bool { - if len(c.entries) == 0 { - return false - } - - var ( - oldestRoomID string - oldestAt time.Time - ) - - for roomID, entry := range c.entries { - if oldestRoomID == "" || entry.touchedAt.Before(oldestAt) { - oldestRoomID = roomID - oldestAt = entry.touchedAt - } - } - - delete(c.entries, oldestRoomID) - return true -} - -type typingSession struct { - stopCh chan struct{} - once sync.Once -} - -func newTypingSession() *typingSession { - return &typingSession{ - stopCh: make(chan struct{}), - } -} - -func (s *typingSession) stop() { - s.once.Do(func() { - close(s.stopCh) - }) -} - // MatrixChannel implements the Channel interface for Matrix. type MatrixChannel struct { *channels.BaseChannel @@ -272,656 +132,6 @@ func (c *MatrixChannel) Stop(ctx context.Context) error { return nil } -func markdownToHTML(md string) string { - p := parser.NewWithExtensions(parser.CommonExtensions | parser.AutoHeadingIDs) - renderer := mdhtml.NewRenderer(mdhtml.RendererOptions{Flags: mdhtml.CommonFlags}) - return strings.TrimSpace(string(markdown.ToHTML([]byte(md), p, renderer))) -} - -func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { - if !c.IsRunning() { - return channels.ErrNotRunning - } - - roomID := id.RoomID(strings.TrimSpace(msg.ChatID)) - if roomID == "" { - return fmt.Errorf("matrix room ID is empty: %w", channels.ErrSendFailed) - } - - content := strings.TrimSpace(msg.Content) - if content == "" { - return nil - } - - _, err := c.client.SendMessageEvent(ctx, roomID, event.EventMessage, c.messageContent(content)) - if err != nil { - return fmt.Errorf("matrix send: %w", channels.ErrTemporary) - } - return nil -} - -func (c *MatrixChannel) messageContent(text string) *event.MessageEventContent { - mc := &event.MessageEventContent{MsgType: event.MsgText, Body: text} - if c.config.MessageFormat != "plain" { - mc.Format = event.FormatHTML - mc.FormattedBody = markdownToHTML(text) - } - return mc -} - -// SendMedia implements channels.MediaSender. -func (c *MatrixChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error { - if !c.IsRunning() { - return channels.ErrNotRunning - } - sendCtx := ctx - if sendCtx == nil { - sendCtx = context.Background() - } - - roomID := id.RoomID(strings.TrimSpace(msg.ChatID)) - if roomID == "" { - return fmt.Errorf("matrix room ID is empty: %w", 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 := sendCtx.Err(); err != nil { - return err - } - - localPath, meta, err := store.ResolveWithMeta(part.Ref) - if err != nil { - logger.ErrorCF("matrix", "Failed to resolve media ref", map[string]any{ - "ref": part.Ref, - "error": err.Error(), - }) - continue - } - - fileInfo, err := os.Stat(localPath) - if err != nil { - logger.ErrorCF("matrix", "Failed to stat media file", map[string]any{ - "path": localPath, - "error": err.Error(), - }) - continue - } - - file, err := os.Open(localPath) - if err != nil { - logger.ErrorCF("matrix", "Failed to open media file", map[string]any{ - "path": localPath, - "error": err.Error(), - }) - continue - } - - filename := strings.TrimSpace(part.Filename) - if filename == "" { - filename = strings.TrimSpace(meta.Filename) - } - if filename == "" { - filename = filepath.Base(localPath) - } - if filename == "" { - filename = "file" - } - - contentType := strings.TrimSpace(part.ContentType) - if contentType == "" { - contentType = strings.TrimSpace(meta.ContentType) - } - if contentType == "" { - contentType = mime.TypeByExtension(strings.ToLower(filepath.Ext(filename))) - } - if contentType == "" { - contentType = "application/octet-stream" - } - - uploadResp, err := c.client.UploadMedia(sendCtx, mautrix.ReqUploadMedia{ - Content: file, - ContentLength: fileInfo.Size(), - ContentType: contentType, - FileName: filename, - }) - file.Close() - if err != nil { - logger.ErrorCF("matrix", "Failed to upload media", map[string]any{ - "path": localPath, - "type": part.Type, - "error": err.Error(), - }) - return fmt.Errorf("matrix upload media: %w", channels.ErrTemporary) - } - - msgType := matrixOutboundMsgType(part.Type, filename, contentType) - content := matrixOutboundContent( - part.Caption, - filename, - msgType, - contentType, - fileInfo.Size(), - uploadResp.ContentURI.CUString(), - ) - - if _, err := c.client.SendMessageEvent(sendCtx, roomID, event.EventMessage, content); err != nil { - logger.ErrorCF("matrix", "Failed to send media message", map[string]any{ - "room_id": roomID.String(), - "type": msgType, - "error": err.Error(), - }) - return fmt.Errorf("matrix send media: %w", channels.ErrTemporary) - } - } - - return nil -} - -// StartTyping implements channels.TypingCapable. -func (c *MatrixChannel) StartTyping(ctx context.Context, chatID string) (func(), error) { - if !c.IsRunning() { - return func() {}, nil - } - - roomID := id.RoomID(strings.TrimSpace(chatID)) - if roomID == "" { - return func() {}, fmt.Errorf("matrix room ID is empty") - } - - session := newTypingSession() - - c.typingMu.Lock() - if prev := c.typingSessions[chatID]; prev != nil { - prev.stop() - } - c.typingSessions[chatID] = session - c.typingMu.Unlock() - - parent := c.baseContext() - go c.typingLoop(parent, roomID, session) - - var once sync.Once - stop := func() { - once.Do(func() { - session.stop() - c.typingMu.Lock() - if current := c.typingSessions[chatID]; current == session { - delete(c.typingSessions, chatID) - } - c.typingMu.Unlock() - _, _ = c.client.UserTyping(context.Background(), roomID, false, 0) - }) - } - - return stop, nil -} - -// SendPlaceholder implements channels.PlaceholderCapable. -func (c *MatrixChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) { - if !c.config.Placeholder.Enabled { - return "", nil - } - - roomID := id.RoomID(strings.TrimSpace(chatID)) - if roomID == "" { - return "", fmt.Errorf("matrix room ID is empty") - } - - text := strings.TrimSpace(c.config.Placeholder.Text) - if text == "" { - text = "Thinking... 💭" - } - - resp, err := c.client.SendMessageEvent(ctx, roomID, event.EventMessage, &event.MessageEventContent{ - MsgType: event.MsgNotice, - Body: text, - }) - if err != nil { - return "", err - } - - return resp.EventID.String(), nil -} - -// EditMessage implements channels.MessageEditor. -func (c *MatrixChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error { - roomID := id.RoomID(strings.TrimSpace(chatID)) - if roomID == "" { - return fmt.Errorf("matrix room ID is empty") - } - if strings.TrimSpace(messageID) == "" { - return fmt.Errorf("matrix message ID is empty") - } - - editContent := c.messageContent(content) - editContent.SetEdit(id.EventID(messageID)) - - _, err := c.client.SendMessageEvent(ctx, roomID, event.EventMessage, editContent) - return err -} - -func (c *MatrixChannel) handleMemberEvent(ctx context.Context, evt *event.Event) { - if !c.config.JoinOnInvite { - return - } - if evt == nil { - return - } - - member := evt.Content.AsMember() - if member.Membership != event.MembershipInvite { - return - } - if evt.GetStateKey() != c.client.UserID.String() { - return - } - - _, err := c.client.JoinRoomByID(c.baseContext(), evt.RoomID) - if err != nil { - logger.WarnCF("matrix", "Failed to auto-join invited room", map[string]any{ - "room_id": evt.RoomID.String(), - "error": err.Error(), - }) - return - } - - logger.InfoCF("matrix", "Joined room after invite", map[string]any{ - "room_id": evt.RoomID.String(), - }) -} - -func (c *MatrixChannel) handleMessageEvent(ctx context.Context, evt *event.Event) { - if evt == nil { - return - } - - // Ignore our own messages. - if evt.Sender == c.client.UserID { - return - } - - // Ignore historical events on first sync. - if time.UnixMilli(evt.Timestamp).Before(c.startTime) { - return - } - - msgEvt := evt.Content.AsMessage() - if msgEvt == nil { - return - } - - // Ignore edits. - if msgEvt.RelatesTo != nil && msgEvt.RelatesTo.GetReplaceID() != "" { - return - } - - roomID := evt.RoomID.String() - scope := channels.BuildMediaScope("matrix", roomID, evt.ID.String()) - - content, mediaPaths, ok := c.extractInboundContent(ctx, msgEvt, scope) - if !ok { - return - } - content = strings.TrimSpace(content) - if content == "" && len(mediaPaths) == 0 { - return - } - - senderID := evt.Sender.String() - sender := bus.SenderInfo{ - Platform: "matrix", - PlatformID: senderID, - CanonicalID: identity.BuildCanonicalID("matrix", senderID), - Username: senderID, - DisplayName: senderID, - } - - if !c.IsAllowedSender(sender) { - logger.DebugCF("matrix", "Message rejected by allowlist", map[string]any{ - "sender_id": senderID, - }) - return - } - - isGroup := c.isGroupRoom(ctx, evt.RoomID) - if isGroup { - isMentioned := c.isBotMentioned(msgEvt) - if isMentioned { - content = c.stripSelfMention(content) - } - respond, cleaned := c.ShouldRespondInGroup(isMentioned, content) - if !respond { - logger.DebugCF("matrix", "Ignoring group message by trigger rules", map[string]any{ - "room_id": roomID, - "is_mentioned": isMentioned, - "mention_only": c.config.GroupTrigger.MentionOnly, - "prefixes": c.config.GroupTrigger.Prefixes, - }) - return - } - content = cleaned - } else { - content = c.stripSelfMention(content) - } - - content = strings.TrimSpace(content) - if content == "" { - return - } - - peerKind := "direct" - peerID := senderID - if isGroup { - peerKind = "group" - peerID = roomID - } - - metadata := map[string]string{ - "room_id": roomID, - "timestamp": fmt.Sprintf("%d", evt.Timestamp), - "is_group": fmt.Sprintf("%t", isGroup), - "sender_raw": senderID, - } - if replyTo := msgEvt.GetRelatesTo().GetReplyTo(); replyTo != "" { - metadata["reply_to_msg_id"] = replyTo.String() - } - - c.HandleMessage( - c.baseContext(), - bus.Peer{Kind: peerKind, ID: peerID}, - evt.ID.String(), - senderID, - roomID, - content, - mediaPaths, - metadata, - sender, - ) -} - -func (c *MatrixChannel) extractInboundContent( - ctx context.Context, - msgEvt *event.MessageEventContent, - scope string, -) (string, []string, bool) { - switch msgEvt.MsgType { - case event.MsgText, event.MsgNotice: - return msgEvt.Body, nil, true - case event.MsgImage, event.MsgAudio, event.MsgVideo, event.MsgFile: - return c.extractInboundMedia(ctx, msgEvt, scope) - default: - logger.DebugCF("matrix", "Ignoring unsupported matrix msgtype", map[string]any{ - "msgtype": msgEvt.MsgType, - }) - return "", nil, false - } -} - -func (c *MatrixChannel) extractInboundMedia( - ctx context.Context, - msgEvt *event.MessageEventContent, - scope string, -) (string, []string, bool) { - mediaKind := matrixMediaKind(msgEvt.MsgType) - label := matrixMediaLabel(msgEvt, mediaKind) - content := fmt.Sprintf("[%s: %s]", mediaKind, label) - if caption := strings.TrimSpace(msgEvt.GetCaption()); caption != "" { - content = caption + "\n" + content - } - - localPath, err := c.downloadMedia(ctx, msgEvt, mediaKind) - if err != nil { - logger.WarnCF("matrix", "Failed to download media; forwarding as text-only marker", map[string]any{ - "msgtype": msgEvt.MsgType, - "error": err.Error(), - }) - return content, nil, true - } - - filename := matrixMediaFilename(label, mediaKind, matrixContentType(msgEvt)) - ref := c.storeMedia(localPath, media.MediaMeta{ - Filename: filename, - ContentType: matrixContentType(msgEvt), - Source: "matrix", - }, scope) - return content, []string{ref}, true -} - -func (c *MatrixChannel) storeMedia(localPath string, meta media.MediaMeta, scope string) string { - if store := c.GetMediaStore(); store != nil { - ref, err := store.Store(localPath, meta, scope) - if err == nil { - return ref - } - logger.WarnCF("matrix", "Failed to store media in MediaStore, falling back to local path", map[string]any{ - "path": localPath, - "error": err.Error(), - }) - } - return localPath -} - -func (c *MatrixChannel) downloadMedia( - ctx context.Context, - msgEvt *event.MessageEventContent, - mediaKind string, -) (string, error) { - uri := matrixMediaURI(msgEvt) - if uri == "" { - return "", fmt.Errorf("empty matrix media URL") - } - parsed := uri.ParseOrIgnore() - if parsed.IsEmpty() { - return "", fmt.Errorf("invalid matrix media URL: %s", uri) - } - - dlCtx := c.baseContext() - if ctx != nil { - dlCtx = ctx - } - reqCtx, cancel := context.WithTimeout(dlCtx, 20*time.Second) - defer cancel() - - resp, err := c.client.Download(reqCtx, parsed) - if err != nil { - return "", err - } - defer resp.Body.Close() - - reader := resp.Body - readerClose := func() error { return nil } - - // Encrypted attachments put URL in msgEvt.File and require client-side decryption. - if msgEvt != nil && msgEvt.File != nil && msgEvt.URL == "" { - if err = msgEvt.File.PrepareForDecryption(); err != nil { - return "", fmt.Errorf("decrypt matrix media: %w", err) - } - decryptReader := msgEvt.File.DecryptStream(resp.Body) - reader = decryptReader - readerClose = decryptReader.Close - } - - label := matrixMediaLabel(msgEvt, mediaKind) - ext := matrixMediaExt(label, matrixContentType(msgEvt), mediaKind) - mediaDir, err := matrixMediaTempDir() - if err != nil { - return "", fmt.Errorf("create matrix media directory: %w", err) - } - tmp, err := os.CreateTemp(mediaDir, "matrix-media-*"+ext) - if err != nil { - return "", err - } - tmpPath := tmp.Name() - cleanup := true - defer func() { - _ = tmp.Close() - if cleanup { - _ = os.Remove(tmpPath) - } - }() - - _, err = io.Copy(tmp, reader) - if err != nil { - return "", err - } - if err = readerClose(); err != nil { - return "", fmt.Errorf("decrypt matrix media: %w", err) - } - if err = tmp.Close(); err != nil { - return "", err - } - - cleanup = false - return tmpPath, nil -} - -func matrixContentType(msgEvt *event.MessageEventContent) string { - if msgEvt != nil && msgEvt.Info != nil { - return strings.TrimSpace(msgEvt.Info.MimeType) - } - return "" -} - -func matrixMediaURI(msgEvt *event.MessageEventContent) id.ContentURIString { - if msgEvt == nil { - return "" - } - if msgEvt.URL != "" { - return msgEvt.URL - } - if msgEvt.File != nil { - return msgEvt.File.URL - } - return "" -} - -func matrixMediaKind(msgType event.MessageType) string { - switch msgType { - case event.MsgAudio: - return "audio" - case event.MsgVideo: - return "video" - case event.MsgFile: - return "file" - default: - return "image" - } -} - -func matrixOutboundMsgType(partType, filename, contentType string) event.MessageType { - switch strings.ToLower(strings.TrimSpace(partType)) { - case "image": - return event.MsgImage - case "audio", "voice": - return event.MsgAudio - case "video": - return event.MsgVideo - case "file", "document": - return event.MsgFile - } - - ct := strings.ToLower(strings.TrimSpace(contentType)) - switch { - case strings.HasPrefix(ct, "image/"): - return event.MsgImage - case strings.HasPrefix(ct, "audio/"), ct == "application/ogg", ct == "application/x-ogg": - return event.MsgAudio - case strings.HasPrefix(ct, "video/"): - return event.MsgVideo - } - - switch strings.ToLower(strings.TrimSpace(filepath.Ext(filename))) { - case ".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".svg": - return event.MsgImage - case ".mp3", ".wav", ".ogg", ".m4a", ".flac", ".aac", ".wma", ".opus": - return event.MsgAudio - case ".mp4", ".avi", ".mov", ".webm", ".mkv": - return event.MsgVideo - default: - return event.MsgFile - } -} - -func matrixOutboundContent( - caption, filename string, - msgType event.MessageType, - contentType string, - size int64, - uri id.ContentURIString, -) *event.MessageEventContent { - body := strings.TrimSpace(caption) - if body == "" { - body = filename - } - if body == "" { - body = matrixMediaKind(msgType) - } - - info := &event.FileInfo{MimeType: strings.TrimSpace(contentType)} - if size > 0 && size <= int64(int(^uint(0)>>1)) { - info.Size = int(size) - } - - content := &event.MessageEventContent{ - MsgType: msgType, - Body: body, - URL: uri, - FileName: filename, - Info: info, - } - return content -} - -func matrixMediaLabel(msgEvt *event.MessageEventContent, fallback string) string { - if msgEvt == nil { - return fallback - } - if v := strings.TrimSpace(msgEvt.FileName); v != "" { - return v - } - if v := strings.TrimSpace(msgEvt.Body); v != "" { - return v - } - return fallback -} - -func matrixMediaFilename(label, mediaKind, contentType string) string { - filename := strings.TrimSpace(label) - if filename == "" { - filename = mediaKind - } - if filepath.Ext(filename) == "" { - filename += matrixMediaExt("", contentType, mediaKind) - } - return filename -} - -func matrixMediaExt(filename, contentType, mediaKind string) string { - if ext := strings.TrimSpace(filepath.Ext(filename)); ext != "" { - return ext - } - if contentType != "" { - if exts, err := mime.ExtensionsByType(contentType); err == nil && len(exts) > 0 { - return exts[0] - } - } - switch mediaKind { - case "audio": - return ".ogg" - case "video": - return ".mp4" - case "file": - return ".bin" - default: - return ".jpg" - } -} - func (c *MatrixChannel) isGroupRoom(ctx context.Context, roomID id.RoomID) bool { now := time.Now() if isGroup, ok := c.roomKindCache.get(roomID.String(), now); ok { @@ -949,200 +159,9 @@ func (c *MatrixChannel) isGroupRoom(ctx context.Context, roomID id.RoomID) bool return isGroup } -func (c *MatrixChannel) isBotMentioned(msgEvt *event.MessageEventContent) bool { - if msgEvt == nil { - return false - } - - if msgEvt.Mentions != nil && msgEvt.Mentions.Has(c.client.UserID) { - return true - } - - userID := c.client.UserID.String() - if userID != "" && strings.Contains(msgEvt.Body, userID) { - return true - } - if mentionsUserInFormattedBody(msgEvt.FormattedBody, c.client.UserID) { - return true - } - - mentionR := c.localpartMentionR - if mentionR == nil { - mentionR = localpartMentionRegexp(matrixLocalpart(c.client.UserID)) - } - if mentionR == nil { - return false - } - - // Matrix users are addressed as MXID "@localpart:server", but many clients - // emit plain-text mentions as "@localpart". Both forms are handled here. - return mentionR.MatchString(msgEvt.Body) || mentionR.MatchString(msgEvt.FormattedBody) -} - -func mentionsUserInFormattedBody(formattedBody string, userID id.UserID) bool { - target := strings.ToLower(strings.TrimSpace(userID.String())) - if target == "" { - return false - } - - formattedBody = strings.TrimSpace(formattedBody) - if formattedBody == "" { - return false - } - - if strings.Contains(strings.ToLower(formattedBody), target) { - return true - } - - matches := matrixMentionHrefRegexp.FindAllStringSubmatch(formattedBody, -1) - for _, match := range matches { - if len(match) < 2 { - continue - } - decoded := decodeMatrixMentionHref(match[1]) - if strings.Contains(strings.ToLower(decoded), target) { - return true - } - - u, err := url.Parse(decoded) - if err != nil { - continue - } - - if strings.Contains(strings.ToLower(u.Path), target) || strings.Contains(strings.ToLower(u.Fragment), target) { - return true - } - if strings.Contains(strings.ToLower(decodeMatrixMentionHref(u.Fragment)), target) { - return true - } - } - - return false -} - -func decodeMatrixMentionHref(v string) string { - decoded := html.UnescapeString(strings.TrimSpace(v)) - if decoded == "" { - return "" - } - - for i := 0; i < 2; i++ { - next, err := url.QueryUnescape(decoded) - if err != nil || next == decoded { - break - } - decoded = next - } - return decoded -} - -func (c *MatrixChannel) typingLoop(ctx context.Context, roomID id.RoomID, session *typingSession) { - sendTyping := func() { - _, err := c.client.UserTyping(ctx, roomID, true, typingServerTTL) - if err != nil { - logger.DebugCF("matrix", "Failed to send typing status", map[string]any{ - "room_id": roomID.String(), - "error": err.Error(), - }) - } - } - - sendTyping() - ticker := time.NewTicker(typingRefreshInterval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-session.stopCh: - return - case <-ticker.C: - sendTyping() - } - } -} - -func (c *MatrixChannel) stopTypingSessions(ctx context.Context) { - c.typingMu.Lock() - sessions := c.typingSessions - c.typingSessions = make(map[string]*typingSession) - c.typingMu.Unlock() - - stopCtx := ctx - if stopCtx == nil { - stopCtx = context.Background() - } - for roomID, session := range sessions { - session.stop() - _, _ = c.client.UserTyping(stopCtx, id.RoomID(roomID), false, 0) - } -} - func (c *MatrixChannel) baseContext() context.Context { if c.ctx != nil { return c.ctx } return context.Background() } - -func (c *MatrixChannel) runRoomKindCacheJanitor(ctx context.Context) { - ticker := time.NewTicker(roomKindCacheCleanupPeriod) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case now := <-ticker.C: - c.roomKindCache.cleanupExpired(now) - } - } -} - -func (c *MatrixChannel) stripSelfMention(text string) string { - return stripUserMentionWithRegexp(text, c.client.UserID, c.localpartMentionR) -} - -func matrixMediaTempDir() (string, error) { - mediaDir := filepath.Join(os.TempDir(), matrixMediaTempDirName) - if err := os.MkdirAll(mediaDir, 0o700); err != nil { - return "", err - } - return mediaDir, nil -} - -func matrixLocalpart(userID id.UserID) string { - s := strings.TrimPrefix(userID.String(), "@") - localpart, _, _ := strings.Cut(s, ":") - return strings.TrimSpace(localpart) -} - -func localpartMentionRegexp(localpart string) *regexp.Regexp { - localpart = strings.TrimSpace(localpart) - if localpart == "" { - return nil - } - - // Match Matrix mentions in plain text while avoiding false positives: - // "@picoclaw" and "@picoclaw:matrix.org" should match, - // "test@example.com" and "hellopicoclawworld" should not. - pattern := `(?i)(^|[^[:alnum:]_])@` + regexp.QuoteMeta(localpart) + `(?::[A-Za-z0-9._:-]+)?([^[:alnum:]_]|$)` - return regexp.MustCompile(pattern) -} - -func stripUserMention(text string, userID id.UserID) string { - return stripUserMentionWithRegexp(text, userID, localpartMentionRegexp(matrixLocalpart(userID))) -} - -func stripUserMentionWithRegexp(text string, userID id.UserID, mentionR *regexp.Regexp) string { - cleaned := strings.ReplaceAll(text, userID.String(), "") - - if mentionR != nil { - cleaned = mentionR.ReplaceAllString(cleaned, "$1$2") - } - - cleaned = strings.TrimSpace(cleaned) - cleaned = strings.TrimLeft(cleaned, ",:; ") - return strings.TrimSpace(cleaned) -} diff --git a/pkg/channels/matrix/media.go b/pkg/channels/matrix/media.go new file mode 100644 index 000000000..ac75a2ec2 --- /dev/null +++ b/pkg/channels/matrix/media.go @@ -0,0 +1,406 @@ +package matrix + +import ( + "context" + "fmt" + "io" + "mime" + "os" + "path/filepath" + "strings" + "time" + + "maunium.net/go/mautrix" + "maunium.net/go/mautrix/event" + "maunium.net/go/mautrix/id" + + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/logger" + "jane/pkg/media" +) + +const ( + matrixMediaTempDirName = "picoclaw_media" +) + +// SendMedia implements channels.MediaSender. +func (c *MatrixChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) error { + if !c.IsRunning() { + return channels.ErrNotRunning + } + sendCtx := ctx + if sendCtx == nil { + sendCtx = context.Background() + } + + roomID := id.RoomID(strings.TrimSpace(msg.ChatID)) + if roomID == "" { + return fmt.Errorf("matrix room ID is empty: %w", 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 := sendCtx.Err(); err != nil { + return err + } + + localPath, meta, err := store.ResolveWithMeta(part.Ref) + if err != nil { + logger.ErrorCF("matrix", "Failed to resolve media ref", map[string]any{ + "ref": part.Ref, + "error": err.Error(), + }) + continue + } + + fileInfo, err := os.Stat(localPath) + if err != nil { + logger.ErrorCF("matrix", "Failed to stat media file", map[string]any{ + "path": localPath, + "error": err.Error(), + }) + continue + } + + file, err := os.Open(localPath) + if err != nil { + logger.ErrorCF("matrix", "Failed to open media file", map[string]any{ + "path": localPath, + "error": err.Error(), + }) + continue + } + + filename := strings.TrimSpace(part.Filename) + if filename == "" { + filename = strings.TrimSpace(meta.Filename) + } + if filename == "" { + filename = filepath.Base(localPath) + } + if filename == "" { + filename = "file" + } + + contentType := strings.TrimSpace(part.ContentType) + if contentType == "" { + contentType = strings.TrimSpace(meta.ContentType) + } + if contentType == "" { + contentType = mime.TypeByExtension(strings.ToLower(filepath.Ext(filename))) + } + if contentType == "" { + contentType = "application/octet-stream" + } + + uploadResp, err := c.client.UploadMedia(sendCtx, mautrix.ReqUploadMedia{ + Content: file, + ContentLength: fileInfo.Size(), + ContentType: contentType, + FileName: filename, + }) + file.Close() + if err != nil { + logger.ErrorCF("matrix", "Failed to upload media", map[string]any{ + "path": localPath, + "type": part.Type, + "error": err.Error(), + }) + return fmt.Errorf("matrix upload media: %w", channels.ErrTemporary) + } + + msgType := matrixOutboundMsgType(part.Type, filename, contentType) + content := matrixOutboundContent( + part.Caption, + filename, + msgType, + contentType, + fileInfo.Size(), + uploadResp.ContentURI.CUString(), + ) + + if _, err := c.client.SendMessageEvent(sendCtx, roomID, event.EventMessage, content); err != nil { + logger.ErrorCF("matrix", "Failed to send media message", map[string]any{ + "room_id": roomID.String(), + "type": msgType, + "error": err.Error(), + }) + return fmt.Errorf("matrix send media: %w", channels.ErrTemporary) + } + } + + return nil +} + +func (c *MatrixChannel) extractInboundMedia( + ctx context.Context, + msgEvt *event.MessageEventContent, + scope string, +) (string, []string, bool) { + mediaKind := matrixMediaKind(msgEvt.MsgType) + label := matrixMediaLabel(msgEvt, mediaKind) + content := fmt.Sprintf("[%s: %s]", mediaKind, label) + if caption := strings.TrimSpace(msgEvt.GetCaption()); caption != "" { + content = caption + "\n" + content + } + + localPath, err := c.downloadMedia(ctx, msgEvt, mediaKind) + if err != nil { + logger.WarnCF("matrix", "Failed to download media; forwarding as text-only marker", map[string]any{ + "msgtype": msgEvt.MsgType, + "error": err.Error(), + }) + return content, nil, true + } + + filename := matrixMediaFilename(label, mediaKind, matrixContentType(msgEvt)) + ref := c.storeMedia(localPath, media.MediaMeta{ + Filename: filename, + ContentType: matrixContentType(msgEvt), + Source: "matrix", + }, scope) + return content, []string{ref}, true +} + +func (c *MatrixChannel) storeMedia(localPath string, meta media.MediaMeta, scope string) string { + if store := c.GetMediaStore(); store != nil { + ref, err := store.Store(localPath, meta, scope) + if err == nil { + return ref + } + logger.WarnCF("matrix", "Failed to store media in MediaStore, falling back to local path", map[string]any{ + "path": localPath, + "error": err.Error(), + }) + } + return localPath +} + +func (c *MatrixChannel) downloadMedia( + ctx context.Context, + msgEvt *event.MessageEventContent, + mediaKind string, +) (string, error) { + uri := matrixMediaURI(msgEvt) + if uri == "" { + return "", fmt.Errorf("empty matrix media URL") + } + parsed := uri.ParseOrIgnore() + if parsed.IsEmpty() { + return "", fmt.Errorf("invalid matrix media URL: %s", uri) + } + + dlCtx := c.baseContext() + if ctx != nil { + dlCtx = ctx + } + reqCtx, cancel := context.WithTimeout(dlCtx, 20*time.Second) + defer cancel() + + resp, err := c.client.Download(reqCtx, parsed) + if err != nil { + return "", err + } + defer resp.Body.Close() + + reader := resp.Body + readerClose := func() error { return nil } + + // Encrypted attachments put URL in msgEvt.File and require client-side decryption. + if msgEvt != nil && msgEvt.File != nil && msgEvt.URL == "" { + if err = msgEvt.File.PrepareForDecryption(); err != nil { + return "", fmt.Errorf("decrypt matrix media: %w", err) + } + decryptReader := msgEvt.File.DecryptStream(resp.Body) + reader = decryptReader + readerClose = decryptReader.Close + } + + label := matrixMediaLabel(msgEvt, mediaKind) + ext := matrixMediaExt(label, matrixContentType(msgEvt), mediaKind) + mediaDir, err := matrixMediaTempDir() + if err != nil { + return "", fmt.Errorf("create matrix media directory: %w", err) + } + tmp, err := os.CreateTemp(mediaDir, "matrix-media-*"+ext) + if err != nil { + return "", err + } + tmpPath := tmp.Name() + cleanup := true + defer func() { + _ = tmp.Close() + if cleanup { + _ = os.Remove(tmpPath) + } + }() + + _, err = io.Copy(tmp, reader) + if err != nil { + return "", err + } + if err = readerClose(); err != nil { + return "", fmt.Errorf("decrypt matrix media: %w", err) + } + if err = tmp.Close(); err != nil { + return "", err + } + + cleanup = false + return tmpPath, nil +} + +func matrixContentType(msgEvt *event.MessageEventContent) string { + if msgEvt != nil && msgEvt.Info != nil { + return strings.TrimSpace(msgEvt.Info.MimeType) + } + return "" +} + +func matrixMediaURI(msgEvt *event.MessageEventContent) id.ContentURIString { + if msgEvt == nil { + return "" + } + if msgEvt.URL != "" { + return msgEvt.URL + } + if msgEvt.File != nil { + return msgEvt.File.URL + } + return "" +} + +func matrixMediaKind(msgType event.MessageType) string { + switch msgType { + case event.MsgAudio: + return "audio" + case event.MsgVideo: + return "video" + case event.MsgFile: + return "file" + default: + return "image" + } +} + +func matrixOutboundMsgType(partType, filename, contentType string) event.MessageType { + switch strings.ToLower(strings.TrimSpace(partType)) { + case "image": + return event.MsgImage + case "audio", "voice": + return event.MsgAudio + case "video": + return event.MsgVideo + case "file", "document": + return event.MsgFile + } + + ct := strings.ToLower(strings.TrimSpace(contentType)) + switch { + case strings.HasPrefix(ct, "image/"): + return event.MsgImage + case strings.HasPrefix(ct, "audio/"), ct == "application/ogg", ct == "application/x-ogg": + return event.MsgAudio + case strings.HasPrefix(ct, "video/"): + return event.MsgVideo + } + + switch strings.ToLower(strings.TrimSpace(filepath.Ext(filename))) { + case ".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".svg": + return event.MsgImage + case ".mp3", ".wav", ".ogg", ".m4a", ".flac", ".aac", ".wma", ".opus": + return event.MsgAudio + case ".mp4", ".avi", ".mov", ".webm", ".mkv": + return event.MsgVideo + default: + return event.MsgFile + } +} + +func matrixOutboundContent( + caption, filename string, + msgType event.MessageType, + contentType string, + size int64, + uri id.ContentURIString, +) *event.MessageEventContent { + body := strings.TrimSpace(caption) + if body == "" { + body = filename + } + if body == "" { + body = matrixMediaKind(msgType) + } + + info := &event.FileInfo{MimeType: strings.TrimSpace(contentType)} + if size > 0 && size <= int64(int(^uint(0)>>1)) { + info.Size = int(size) + } + + content := &event.MessageEventContent{ + MsgType: msgType, + Body: body, + URL: uri, + FileName: filename, + Info: info, + } + return content +} + +func matrixMediaLabel(msgEvt *event.MessageEventContent, fallback string) string { + if msgEvt == nil { + return fallback + } + if v := strings.TrimSpace(msgEvt.FileName); v != "" { + return v + } + if v := strings.TrimSpace(msgEvt.Body); v != "" { + return v + } + return fallback +} + +func matrixMediaFilename(label, mediaKind, contentType string) string { + filename := strings.TrimSpace(label) + if filename == "" { + filename = mediaKind + } + if filepath.Ext(filename) == "" { + filename += matrixMediaExt("", contentType, mediaKind) + } + return filename +} + +func matrixMediaExt(filename, contentType, mediaKind string) string { + if ext := strings.TrimSpace(filepath.Ext(filename)); ext != "" { + return ext + } + if contentType != "" { + if exts, err := mime.ExtensionsByType(contentType); err == nil && len(exts) > 0 { + return exts[0] + } + } + switch mediaKind { + case "audio": + return ".ogg" + case "video": + return ".mp4" + case "file": + return ".bin" + default: + return ".jpg" + } +} + +func matrixMediaTempDir() (string, error) { + mediaDir := filepath.Join(os.TempDir(), matrixMediaTempDirName) + if err := os.MkdirAll(mediaDir, 0o700); err != nil { + return "", err + } + return mediaDir, nil +} diff --git a/pkg/channels/matrix/mentions.go b/pkg/channels/matrix/mentions.go new file mode 100644 index 000000000..b4642ad86 --- /dev/null +++ b/pkg/channels/matrix/mentions.go @@ -0,0 +1,139 @@ +package matrix + +import ( + "html" + "net/url" + "regexp" + "strings" + + "maunium.net/go/mautrix/event" + "maunium.net/go/mautrix/id" +) + +var matrixMentionHrefRegexp = regexp.MustCompile(`(?i)]+href=["']([^"']+)["']`) + +func (c *MatrixChannel) isBotMentioned(msgEvt *event.MessageEventContent) bool { + if msgEvt == nil { + return false + } + + if msgEvt.Mentions != nil && msgEvt.Mentions.Has(c.client.UserID) { + return true + } + + userID := c.client.UserID.String() + if userID != "" && strings.Contains(msgEvt.Body, userID) { + return true + } + if mentionsUserInFormattedBody(msgEvt.FormattedBody, c.client.UserID) { + return true + } + + mentionR := c.localpartMentionR + if mentionR == nil { + mentionR = localpartMentionRegexp(matrixLocalpart(c.client.UserID)) + } + if mentionR == nil { + return false + } + + // Matrix users are addressed as MXID "@localpart:server", but many clients + // emit plain-text mentions as "@localpart". Both forms are handled here. + return mentionR.MatchString(msgEvt.Body) || mentionR.MatchString(msgEvt.FormattedBody) +} + +func mentionsUserInFormattedBody(formattedBody string, userID id.UserID) bool { + target := strings.ToLower(strings.TrimSpace(userID.String())) + if target == "" { + return false + } + + formattedBody = strings.TrimSpace(formattedBody) + if formattedBody == "" { + return false + } + + if strings.Contains(strings.ToLower(formattedBody), target) { + return true + } + + matches := matrixMentionHrefRegexp.FindAllStringSubmatch(formattedBody, -1) + for _, match := range matches { + if len(match) < 2 { + continue + } + decoded := decodeMatrixMentionHref(match[1]) + if strings.Contains(strings.ToLower(decoded), target) { + return true + } + + u, err := url.Parse(decoded) + if err != nil { + continue + } + + if strings.Contains(strings.ToLower(u.Path), target) || strings.Contains(strings.ToLower(u.Fragment), target) { + return true + } + if strings.Contains(strings.ToLower(decodeMatrixMentionHref(u.Fragment)), target) { + return true + } + } + + return false +} + +func decodeMatrixMentionHref(v string) string { + decoded := html.UnescapeString(strings.TrimSpace(v)) + if decoded == "" { + return "" + } + + for i := 0; i < 2; i++ { + next, err := url.QueryUnescape(decoded) + if err != nil || next == decoded { + break + } + decoded = next + } + return decoded +} + +func (c *MatrixChannel) stripSelfMention(text string) string { + return stripUserMentionWithRegexp(text, c.client.UserID, c.localpartMentionR) +} + +func matrixLocalpart(userID id.UserID) string { + s := strings.TrimPrefix(userID.String(), "@") + localpart, _, _ := strings.Cut(s, ":") + return strings.TrimSpace(localpart) +} + +func localpartMentionRegexp(localpart string) *regexp.Regexp { + localpart = strings.TrimSpace(localpart) + if localpart == "" { + return nil + } + + // Match Matrix mentions in plain text while avoiding false positives: + // "@picoclaw" and "@picoclaw:matrix.org" should match, + // "test@example.com" and "hellopicoclawworld" should not. + pattern := `(?i)(^|[^[:alnum:]_])@` + regexp.QuoteMeta(localpart) + `(?::[A-Za-z0-9._:-]+)?([^[:alnum:]_]|$)` + return regexp.MustCompile(pattern) +} + +func stripUserMention(text string, userID id.UserID) string { + return stripUserMentionWithRegexp(text, userID, localpartMentionRegexp(matrixLocalpart(userID))) +} + +func stripUserMentionWithRegexp(text string, userID id.UserID, mentionR *regexp.Regexp) string { + cleaned := strings.ReplaceAll(text, userID.String(), "") + + if mentionR != nil { + cleaned = mentionR.ReplaceAllString(cleaned, "$1$2") + } + + cleaned = strings.TrimSpace(cleaned) + cleaned = strings.TrimLeft(cleaned, ",:; ") + return strings.TrimSpace(cleaned) +} diff --git a/pkg/channels/matrix/messages.go b/pkg/channels/matrix/messages.go new file mode 100644 index 000000000..9a4643adc --- /dev/null +++ b/pkg/channels/matrix/messages.go @@ -0,0 +1,118 @@ +package matrix + +import ( + "context" + "fmt" + "strings" + + "github.com/gomarkdown/markdown" + mdhtml "github.com/gomarkdown/markdown/html" + "github.com/gomarkdown/markdown/parser" + "maunium.net/go/mautrix/event" + "maunium.net/go/mautrix/id" + + "jane/pkg/bus" + "jane/pkg/channels" + "jane/pkg/logger" +) + +func markdownToHTML(md string) string { + p := parser.NewWithExtensions(parser.CommonExtensions | parser.AutoHeadingIDs) + renderer := mdhtml.NewRenderer(mdhtml.RendererOptions{Flags: mdhtml.CommonFlags}) + return strings.TrimSpace(string(markdown.ToHTML([]byte(md), p, renderer))) +} + +func (c *MatrixChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + if !c.IsRunning() { + return channels.ErrNotRunning + } + + roomID := id.RoomID(strings.TrimSpace(msg.ChatID)) + if roomID == "" { + return fmt.Errorf("matrix room ID is empty: %w", channels.ErrSendFailed) + } + + content := strings.TrimSpace(msg.Content) + if content == "" { + return nil + } + + _, err := c.client.SendMessageEvent(ctx, roomID, event.EventMessage, c.messageContent(content)) + if err != nil { + logger.ErrorCF("matrix", "Failed to send message", map[string]any{ + "room_id": roomID.String(), + "error": err.Error(), + }) + return fmt.Errorf("matrix send: %w", channels.ErrTemporary) + } + + logger.DebugCF("matrix", "Sent message", map[string]any{ + "room_id": roomID.String(), + }) + return nil +} + +func (c *MatrixChannel) messageContent(text string) *event.MessageEventContent { + mc := &event.MessageEventContent{MsgType: event.MsgText, Body: text} + if c.config.MessageFormat != "plain" { + mc.Format = event.FormatHTML + mc.FormattedBody = markdownToHTML(text) + } + return mc +} + +// SendPlaceholder implements channels.PlaceholderCapable. +func (c *MatrixChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) { + if !c.config.Placeholder.Enabled { + return "", nil + } + + roomID := id.RoomID(strings.TrimSpace(chatID)) + if roomID == "" { + return "", fmt.Errorf("matrix room ID is empty") + } + + text := strings.TrimSpace(c.config.Placeholder.Text) + if text == "" { + text = "Thinking... 💭" + } + + resp, err := c.client.SendMessageEvent(ctx, roomID, event.EventMessage, &event.MessageEventContent{ + MsgType: event.MsgNotice, + Body: text, + }) + if err != nil { + return "", err + } + + return resp.EventID.String(), nil +} + +// EditMessage implements channels.MessageEditor. +func (c *MatrixChannel) EditMessage(ctx context.Context, chatID string, messageID string, content string) error { + roomID := id.RoomID(strings.TrimSpace(chatID)) + if roomID == "" { + return fmt.Errorf("matrix room ID is empty") + } + if strings.TrimSpace(messageID) == "" { + return fmt.Errorf("matrix message ID is empty") + } + + editContent := c.messageContent(content) + editContent.SetEdit(id.EventID(messageID)) + + _, err := c.client.SendMessageEvent(ctx, roomID, event.EventMessage, editContent) + if err != nil { + logger.ErrorCF("matrix", "Failed to edit message", map[string]any{ + "room_id": roomID.String(), + "message_id": messageID, + "error": err.Error(), + }) + } else { + logger.DebugCF("matrix", "Edited message", map[string]any{ + "room_id": roomID.String(), + "message_id": messageID, + }) + } + return err +} diff --git a/pkg/channels/matrix/typing.go b/pkg/channels/matrix/typing.go new file mode 100644 index 000000000..fa6c2a101 --- /dev/null +++ b/pkg/channels/matrix/typing.go @@ -0,0 +1,112 @@ +package matrix + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + "maunium.net/go/mautrix/id" + + "jane/pkg/logger" +) + +type typingSession struct { + stopCh chan struct{} + once sync.Once +} + +func newTypingSession() *typingSession { + return &typingSession{ + stopCh: make(chan struct{}), + } +} + +func (s *typingSession) stop() { + s.once.Do(func() { + close(s.stopCh) + }) +} + +// StartTyping implements channels.TypingCapable. +func (c *MatrixChannel) StartTyping(ctx context.Context, chatID string) (func(), error) { + if !c.IsRunning() { + return func() {}, nil + } + + roomID := id.RoomID(strings.TrimSpace(chatID)) + if roomID == "" { + return func() {}, fmt.Errorf("matrix room ID is empty") + } + + session := newTypingSession() + + c.typingMu.Lock() + if prev := c.typingSessions[chatID]; prev != nil { + prev.stop() + } + c.typingSessions[chatID] = session + c.typingMu.Unlock() + + parent := c.baseContext() + go c.typingLoop(parent, roomID, session) + + var once sync.Once + stop := func() { + once.Do(func() { + session.stop() + c.typingMu.Lock() + if current := c.typingSessions[chatID]; current == session { + delete(c.typingSessions, chatID) + } + c.typingMu.Unlock() + _, _ = c.client.UserTyping(context.Background(), roomID, false, 0) + }) + } + + return stop, nil +} + +func (c *MatrixChannel) typingLoop(ctx context.Context, roomID id.RoomID, session *typingSession) { + sendTyping := func() { + _, err := c.client.UserTyping(ctx, roomID, true, typingServerTTL) + if err != nil { + logger.DebugCF("matrix", "Failed to send typing status", map[string]any{ + "room_id": roomID.String(), + "error": err.Error(), + }) + } + } + + sendTyping() + ticker := time.NewTicker(typingRefreshInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-session.stopCh: + return + case <-ticker.C: + sendTyping() + } + } +} + +func (c *MatrixChannel) stopTypingSessions(ctx context.Context) { + c.typingMu.Lock() + sessions := c.typingSessions + c.typingSessions = make(map[string]*typingSession) + c.typingMu.Unlock() + + stopCtx := ctx + if stopCtx == nil { + stopCtx = context.Background() + } + for roomID, session := range sessions { + session.stop() + _, _ = c.client.UserTyping(stopCtx, id.RoomID(roomID), false, 0) + } +}