fix: backfilling system events and reactions (#224)

* fix: backfilling system events and reactions

* fix: deprecated function

* fix: indent comments

Author: highesttt

pkg/connector/client.go

@@ -79,6 +79,7 @@ type LineClient struct {
 	generatedGroupNameCache map[string]bool     // chatMid -> true when Matrix name should be generated from member names
 	knownMemberChatMIDs     map[string]struct{} // chatMid -> current member chats returned by getAllChatMids
 	reactionIconMXC         map[int]string      // predefinedReactionType -> cached MXC URI
+	paidReactionIconMXC     map[string]string   // LINE sticon URL -> cached MXC URI
 	recentReactions         sync.Map            // "msgID\x00emoji" -> struct{} to dedup concurrent 139/140 events
 	unblockBackfills        sync.Map            // chat MID -> *unblockBackfillState while unblock history restoration is active
 
@@ -651,8 +652,7 @@ func (lc *LineClient) Connect(ctx context.Context) {
 	// deliver messages. Otherwise an existing LINE group whose Matrix room
 	// doesn't exist yet may be created by the first message, which makes the
 	// sender's existing membership look like a fresh join.
-	lc.wg.Add(1)
-	lc.syncChats(ctx)
+	lc.syncChatsNow(ctx)
 	if ctx.Err() != nil {
 		return
 	}

pkg/connector/consts.go

@@ -38,6 +38,7 @@ const (
 	ContentContact  ContentType = 13
 	ContentFile     ContentType = 14
 	ContentLocation ContentType = 15
+	ContentSystem   ContentType = 18
 	ContentFlex     ContentType = 22
 )
 

pkg/connector/handle_message.go

@@ -113,7 +113,7 @@ func (lc *LineClient) getChatInfoForIncomingMessage(ctx context.Context, portal
 	return info, nil
 }
 
-func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) {
+func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) bool {
 	// Only process known content types; skip system messages (group created, member invited, etc.)
 	if !isBridgeableContentType(msg) {
 		lc.UserLogin.Bridge.Log.Debug().
@@ -123,7 +123,7 @@ func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) {
 			Str("text", msg.Text).
 			Int("chunk_count", len(msg.Chunks)).
 			Msg("Skipping unsupported content type")
-		return
+		return false
 	}
 
 	portalIDStr := portalMIDForMessage(msg, opType)
@@ -164,7 +164,8 @@ func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) {
 		}
 	}
 
-	lc.UserLogin.Bridge.QueueRemoteEvent(lc.UserLogin, remoteEvent)
+	result := lc.UserLogin.Bridge.QueueRemoteEvent(lc.UserLogin, remoteEvent)
+	return result.Success && !result.Ignored
 }
 
 // isBridgeableContentType reports whether an inbound LINE message should be

pkg/connector/reaction.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net/url"
 	"slices"
 	"strconv"
@@ -15,6 +16,7 @@ import (
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/bridgev2/database"
 	"maunium.net/go/mautrix/bridgev2/networkid"
+	"maunium.net/go/mautrix/bridgev2/simplevent"
 	"maunium.net/go/mautrix/event"
 
 	"github.com/highesttt/matrix-line-messenger/pkg/line"
@@ -201,6 +203,201 @@ func lineSticonURL(productID, emojiID string) string {
 	return fmt.Sprintf("https://stickershop.line-scdn.net/sticonshop/v1/sticon/%s/android/%s.png", productID, emojiID)
 }
 
+func reactionUploadMXC(uploadedMXC string, uploadedFile *event.EncryptedFileInfo) (string, error) {
+	if uploadedFile != nil {
+		return "", errors.New("reaction icon upload returned encrypted media")
+	}
+	if uploadedMXC == "" {
+		return "", errors.New("reaction icon upload returned an empty MXC URI")
+	}
+	return uploadedMXC, nil
+}
+
+func (lc *LineClient) getPredefinedReactionMXC(ctx context.Context, prt int) (string, error) {
+	if _, ok := line.PredefinedReactionEmoji[prt]; !ok {
+		return "", fmt.Errorf("unknown predefined reaction type %d", prt)
+	}
+
+	lc.cacheMu.Lock()
+	mxc := lc.reactionIconMXC[prt]
+	lc.cacheMu.Unlock()
+	if mxc != "" {
+		return mxc, nil
+	}
+
+	pngData, err := getReactionIconData(prt)
+	if err != nil {
+		return "", fmt.Errorf("get reaction icon data: %w", err)
+	}
+	uploadedMXC, uploadedFile, err := lc.UserLogin.Bridge.Bot.UploadMedia(ctx, "", pngData, "reaction.png", "image/png")
+	if err != nil {
+		return "", fmt.Errorf("upload reaction icon: %w", err)
+	}
+	mxc, err = reactionUploadMXC(string(uploadedMXC), uploadedFile)
+	if err != nil {
+		return "", err
+	}
+
+	lc.cacheMu.Lock()
+	if lc.reactionIconMXC == nil {
+		lc.reactionIconMXC = make(map[int]string)
+	}
+	if cached := lc.reactionIconMXC[prt]; cached != "" {
+		mxc = cached
+	} else {
+		lc.reactionIconMXC[prt] = mxc
+	}
+	lc.cacheMu.Unlock()
+	return mxc, nil
+}
+
+func (lc *LineClient) getPaidReactionMXC(ctx context.Context, prt *line.PaidReactionType) (string, error) {
+	if prt == nil || prt.ProductID == "" || prt.EmojiID == "" {
+		return "", errors.New("paid reaction is missing product or emoji ID")
+	}
+	iconURL := lineSticonURL(prt.ProductID, prt.EmojiID)
+
+	lc.cacheMu.Lock()
+	mxc := lc.paidReactionIconMXC[iconURL]
+	lc.cacheMu.Unlock()
+	if mxc != "" {
+		return mxc, nil
+	}
+
+	resp, err := lc.HTTPClient.Get(iconURL)
+	if err != nil {
+		return "", fmt.Errorf("download paid reaction icon: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != 200 {
+		return "", fmt.Errorf("download paid reaction icon: HTTP %d", resp.StatusCode)
+	}
+	data, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", fmt.Errorf("read paid reaction icon: %w", err)
+	}
+	mimeType := resp.Header.Get("Content-Type")
+	if mimeType == "" {
+		mimeType = "image/png"
+	}
+	uploadedMXC, uploadedFile, err := lc.UserLogin.Bridge.Bot.UploadMedia(ctx, "", data, "reaction.png", mimeType)
+	if err != nil {
+		return "", fmt.Errorf("upload paid reaction icon: %w", err)
+	}
+	mxc, err = reactionUploadMXC(string(uploadedMXC), uploadedFile)
+	if err != nil {
+		return "", fmt.Errorf("paid %w", err)
+	}
+
+	lc.cacheMu.Lock()
+	if lc.paidReactionIconMXC == nil {
+		lc.paidReactionIconMXC = make(map[string]string)
+	}
+	if cached := lc.paidReactionIconMXC[iconURL]; cached != "" {
+		mxc = cached
+	} else {
+		lc.paidReactionIconMXC[iconURL] = mxc
+	}
+	lc.cacheMu.Unlock()
+	return mxc, nil
+}
+
+func (lc *LineClient) convertMessageReactions(ctx context.Context, msg *line.Message) ([]*bridgev2.BackfillReaction, bool) {
+	if msg == nil || msg.Reactions == nil {
+		return nil, false
+	}
+
+	converted := make([]*bridgev2.BackfillReaction, 0, len(msg.Reactions))
+	complete := true
+	for _, reaction := range msg.Reactions {
+		if !isUserMID(reaction.FromUserMID) {
+			complete = false
+			lc.UserLogin.Bridge.Log.Warn().
+				Str("msg_id", msg.ID).
+				Str("reaction_sender", reaction.FromUserMID).
+				Msg("Skipping historical reaction without a valid sender MID")
+			continue
+		}
+
+		var (
+			mxc string
+			err error
+		)
+		switch {
+		case reaction.ReactionType.PaidReactionType != nil:
+			mxc, err = lc.getPaidReactionMXC(ctx, reaction.ReactionType.PaidReactionType)
+		case reaction.ReactionType.PredefinedReactionType != 0:
+			mxc, err = lc.getPredefinedReactionMXC(ctx, reaction.ReactionType.PredefinedReactionType)
+		default:
+			err = errors.New("reaction type is missing")
+		}
+		if err != nil {
+			complete = false
+			lc.UserLogin.Bridge.Log.Warn().
+				Err(err).
+				Str("msg_id", msg.ID).
+				Str("reaction_sender", reaction.FromUserMID).
+				Msg("Skipping unsupported historical reaction")
+			continue
+		}
+
+		var timestamp time.Time
+		if timestampMillis, err := reaction.AtMillis.Int64(); err == nil && timestampMillis > 0 {
+			timestamp = time.UnixMilli(timestampMillis)
+		}
+		converted = append(converted, &bridgev2.BackfillReaction{
+			Timestamp: timestamp,
+			Sender:    lc.eventSenderForMID(reaction.FromUserMID),
+			Emoji:     mxc,
+		})
+	}
+	return converted, complete
+}
+
+func (lc *LineClient) queueMessageReactionSync(ctx context.Context, chatMID string, msg *line.Message) bool {
+	if msg == nil || ContentType(msg.ContentType) == ContentSystem || msg.Reactions == nil {
+		return false
+	}
+
+	converted, complete := lc.convertMessageReactions(ctx, msg)
+	if !complete {
+		// The embedded list is authoritative, but an upload/parse failure
+		// means our converted view is incomplete. Do not accidentally redact
+		// reactions that LINE still reports.
+		return false
+	}
+	users := make(map[networkid.UserID]*bridgev2.ReactionSyncUser, len(converted))
+	var timestamp time.Time
+	for _, reaction := range converted {
+		if reaction.Timestamp.After(timestamp) {
+			timestamp = reaction.Timestamp
+		}
+		// LINE allows one reaction per user per message. Keeping the last
+		// record also makes malformed duplicate entries deterministic.
+		users[reaction.Sender.Sender] = &bridgev2.ReactionSyncUser{
+			Reactions:       []*bridgev2.BackfillReaction{reaction},
+			HasAllReactions: true,
+		}
+	}
+	if timestamp.IsZero() {
+		timestamp = lc.parseMessageTimestamp(msg)
+	}
+
+	result := lc.UserLogin.Bridge.QueueRemoteEvent(lc.UserLogin, &simplevent.ReactionSync{
+		EventMeta: simplevent.EventMeta{
+			Type:      bridgev2.RemoteEventReactionSync,
+			PortalKey: networkid.PortalKey{ID: makePortalID(chatMID), Receiver: lc.UserLogin.ID},
+			Timestamp: timestamp,
+		},
+		TargetMessage: networkid.MessageID(msg.ID),
+		Reactions: &bridgev2.ReactionSyncData{
+			Users:       users,
+			HasAllUsers: true,
+		},
+	})
+	return result.Success && !result.Ignored
+}
+
 func parseLineSticonURL(rawURL string) (linePaidReactionRef, error) {
 	parsed, err := url.Parse(rawURL)
 	if err != nil {

pkg/connector/reaction_test.go

@@ -2,11 +2,14 @@ package connector
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"slices"
 	"testing"
 	"time"
 
+	"github.com/rs/zerolog"
+
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/bridgev2/database"
 	"maunium.net/go/mautrix/bridgev2/networkid"
@@ -516,6 +519,90 @@ func TestEventSenderForMIDMarksOwnAccount(t *testing.T) {
 	}
 }
 
+func TestConvertMessageReactionsUsesEmbeddedHistory(t *testing.T) {
+	paidType := &line.PaidReactionType{
+		ProductID: "paid-product",
+		EmojiID:   "paid-emoji",
+	}
+	lc := &LineClient{
+		Mid: "Uself",
+		UserLogin: &bridgev2.UserLogin{
+			UserLogin: &database.UserLogin{ID: "Uself"},
+			Bridge:    &bridgev2.Bridge{Log: zerolog.Nop()},
+		},
+		reactionIconMXC: map[int]string{
+			2: "mxc://line/like",
+		},
+		paidReactionIconMXC: map[string]string{
+			lineSticonURL(paidType.ProductID, paidType.EmojiID): "mxc://line/paid",
+		},
+	}
+	msg := &line.Message{
+		ID: "message-id",
+		Reactions: []line.MessageReaction{
+			{
+				FromUserMID: "Uself",
+				AtMillis:    json.Number("1784930400123"),
+				ReactionType: line.ReactionType{
+					PredefinedReactionType: 2,
+				},
+			},
+			{
+				FromUserMID:  "Uother",
+				AtMillis:     json.Number("1784930400456"),
+				ReactionType: line.ReactionType{PaidReactionType: paidType},
+			},
+		},
+	}
+
+	reactions, complete := lc.convertMessageReactions(context.Background(), msg)
+	if !complete {
+		t.Fatal("reaction conversion was unexpectedly incomplete")
+	}
+	if len(reactions) != 2 {
+		t.Fatalf("reaction count = %d, want 2", len(reactions))
+	}
+	if reactions[0].Emoji != "mxc://line/like" || reactions[0].Sender.Sender != "Uself" || !reactions[0].Sender.IsFromMe {
+		t.Fatalf("predefined reaction = %#v", reactions[0])
+	}
+	if want := time.UnixMilli(1784930400123); !reactions[0].Timestamp.Equal(want) {
+		t.Fatalf("predefined reaction timestamp = %s, want %s", reactions[0].Timestamp, want)
+	}
+	if reactions[1].Emoji != "mxc://line/paid" || reactions[1].Sender.Sender != "Uother" || reactions[1].Sender.IsFromMe {
+		t.Fatalf("paid reaction = %#v", reactions[1])
+	}
+}
+
+func TestReactionUploadMXCRejectsEncryptedMedia(t *testing.T) {
+	if mxc, err := reactionUploadMXC("mxc://line/plain", nil); err != nil || mxc != "mxc://line/plain" {
+		t.Fatalf("plain upload = %q, %v", mxc, err)
+	}
+	if _, err := reactionUploadMXC("mxc://line/encrypted", &event.EncryptedFileInfo{}); err == nil {
+		t.Fatal("encrypted upload was accepted without decryption metadata")
+	}
+	if _, err := reactionUploadMXC("", nil); err == nil {
+		t.Fatal("empty upload was accepted")
+	}
+}
+
+func TestQueueMessageReactionSyncSkipsSystemMarkers(t *testing.T) {
+	lc := &LineClient{}
+	msg := &line.Message{
+		ID:          "system-message",
+		ContentType: int(ContentSystem),
+		Reactions: []line.MessageReaction{{
+			FromUserMID: "Ureactor",
+			ReactionType: line.ReactionType{
+				PredefinedReactionType: 2,
+			},
+		}},
+	}
+
+	if lc.queueMessageReactionSync(context.Background(), "Cgroup", msg) {
+		t.Fatal("system-message marker unexpectedly queued a reaction sync")
+	}
+}
+
 func TestResolveReactionSenderMID(t *testing.T) {
 	lc := &LineClient{
 		UserLogin: &bridgev2.UserLogin{