Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions pkg/connector/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,9 @@ type LineClient struct {
sentReqSeqs map[int]time.Time
lastReqSeq int

deletedChatsMu sync.Mutex
deletedChats map[string]string

tokenMu sync.RWMutex
recoverMu sync.Mutex
recoverTime time.Time
Expand Down
8 changes: 6 additions & 2 deletions pkg/connector/connector.go
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ func (lc *LineConnector) Start(ctx context.Context) error {
}

func (lc *LineConnector) GetBridgeInfoVersion() (info, capabilities int) {
return 1, 3
return 1, 4
}

func (lc *LineConnector) GetCapabilities() *bridgev2.NetworkGeneralCapabilities {
Expand Down Expand Up @@ -138,14 +138,18 @@ func (lc *LineConnector) LoadUserLogin(ctx context.Context, login *bridgev2.User
if meta.SessionInvalidated {
accessToken = ""
}
login.Client = &LineClient{
client := &LineClient{
UserLogin: login,
AccessToken: accessToken,
RefreshToken: meta.RefreshToken,
Mid: meta.Mid,
HTTPClient: &http.Client{Timeout: 10 * time.Second},
sessionInvalidated: meta.SessionInvalidated,
}
if err := client.loadChatDeletions(ctx); err != nil {
return fmt.Errorf("failed to load chat deletions: %w", err)
}
login.Client = client
return nil
}

Expand Down
113 changes: 113 additions & 0 deletions pkg/connector/delete_chat.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
package connector

import (
"context"
"errors"
"fmt"
"math/big"

"maunium.net/go/mautrix/bridgev2"

"github.com/highesttt/matrix-line-messenger/pkg/line"
)

var _ bridgev2.DeleteChatHandlingNetworkAPI = (*LineClient)(nil)

func (lc *LineClient) HandleMatrixDeleteChat(ctx context.Context, msg *bridgev2.MatrixDeleteChat) error {
if msg.Content.DeleteForEveryone {
return errors.New("delete for everyone is not supported on LINE")
}
chatMid := string(msg.Portal.ID)
if msg.Portal.MessageRequest {
reqSeq := lc.nextUntrackedReqSeq()
_, err := lc.callLine(ctx, func(client *line.Client) error {
return client.RejectChatInvitationContext(ctx, int64(reqSeq), chatMid)
})
return err
}
_, boxes, err := callLineResult(lc, ctx, func(client *line.Client) (map[string]line.MessageBox, error) {
return client.GetMessageBoxesByIDsContext(ctx, []string{chatMid})
})
if err != nil {
return err
}
lastDelivered := boxes[chatMid].LastDeliveredMessageID
if lastDelivered == nil || lastDelivered.MessageID == "" || lastDelivered.MessageID == "-1" {
Comment thread
indent[bot] marked this conversation as resolved.
return lc.saveChatDeletion(ctx, chatMid, "0")
}
if cutoff, ok := new(big.Int).SetString(lastDelivered.MessageID, 10); !ok || cutoff.Sign() < 0 {
return errors.New("invalid last delivered message ID")
}
reqSeq := lc.nextReqSeq()
_, err = lc.callLine(ctx, func(client *line.Client) error {
return client.SendChatRemovedContext(ctx, int64(reqSeq), chatMid, lastDelivered.MessageID, 0)
})
if err != nil {
return err
}
return lc.saveChatDeletion(context.WithoutCancel(ctx), chatMid, lastDelivered.MessageID)
}

func (lc *LineClient) chatDeletionPrefix() string {
return fmt.Sprintf("line_deleted_chat:%+q:", lc.UserLogin.ID)
}

func (lc *LineClient) loadChatDeletions(ctx context.Context) error {
prefix := lc.chatDeletionPrefix()
rows, err := lc.UserLogin.Bridge.DB.KV.Query(ctx,
`SELECT key, value FROM kv_store WHERE bridge_id = $1 AND substr(key, 1, $2) = $3`,
lc.UserLogin.Bridge.ID, len(prefix), prefix)
if err != nil {
return err
}
defer rows.Close()
lc.deletedChats = make(map[string]string)
for rows.Next() {
var key, value string
if err := rows.Scan(&key, &value); err != nil {
return err
}
if cutoff, ok := new(big.Int).SetString(value, 10); !ok || cutoff.Sign() < 0 {
return errors.New("invalid stored chat deletion cutoff")
}
lc.deletedChats[key[len(prefix):]] = value
}
return rows.Err()
}

func (lc *LineClient) saveChatDeletion(ctx context.Context, mid, cutoff string) error {
lc.deletedChatsMu.Lock()
defer lc.deletedChatsMu.Unlock()
if previous, ok := lc.deletedChats[mid]; ok {
oldID, _ := new(big.Int).SetString(previous, 10)
newID, _ := new(big.Int).SetString(cutoff, 10)
if oldID.Cmp(newID) > 0 {
cutoff = previous
}
}
_, err := lc.UserLogin.Bridge.DB.KV.Exec(ctx,
`INSERT INTO kv_store (bridge_id, key, value) VALUES ($1, $2, $3)
ON CONFLICT (bridge_id, key) DO UPDATE SET value = $3`,
lc.UserLogin.Bridge.ID, lc.chatDeletionPrefix()+mid, cutoff)
if err != nil {
return err
}
if lc.deletedChats == nil {
lc.deletedChats = make(map[string]string)
}
lc.deletedChats[mid] = cutoff
return nil
}

func (lc *LineClient) shouldSkipDeletedChat(mid, messageID string) bool {
lc.deletedChatsMu.Lock()
cutoff, deleted := lc.deletedChats[mid]
lc.deletedChatsMu.Unlock()
if !deleted {
return false
}
oldID, _ := new(big.Int).SetString(cutoff, 10)
newID, ok := new(big.Int).SetString(messageID, 10)
// Retain the cutoff after reopening so backfill cannot restore deleted history.
return !ok || newID.Cmp(oldID) <= 0
}
3 changes: 3 additions & 0 deletions pkg/connector/handle_message.go
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,9 @@ func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) bool {
}

portalIDStr := portalMIDForMessage(msg, opType)
if lc.shouldSkipDeletedChat(portalIDStr, msg.ID) {
return false
}
portalKey := networkid.PortalKey{ID: makePortalID(portalIDStr), Receiver: lc.UserLogin.ID}
ts := lc.parseMessageTimestamp(msg)

Expand Down
4 changes: 2 additions & 2 deletions pkg/connector/reaction_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@ import (

func TestCapabilitiesAdvertiseSupportedReactions(t *testing.T) {
infoVersion, capabilityVersion := (&LineConnector{}).GetBridgeInfoVersion()
if infoVersion != 1 || capabilityVersion != 3 {
t.Fatalf("bridge info/capability versions = %d/%d, want 1/3", infoVersion, capabilityVersion)
if infoVersion != 1 || capabilityVersion != 4 {
t.Fatalf("bridge info/capability versions = %d/%d, want 1/4", infoVersion, capabilityVersion)
}

caps := (&LineClient{}).GetCapabilities(context.Background(), nil)
Expand Down
36 changes: 35 additions & 1 deletion pkg/connector/sync.go
Original file line number Diff line number Diff line change
Expand Up @@ -604,6 +604,9 @@ func (lc *LineClient) FetchMessages(ctx context.Context, params bridgev2.FetchMe
backfillMsgs := make([]*bridgev2.BackfillMessage, 0, len(msgs))
for i := len(msgs) - 1; i >= 0; i-- {
msg := msgs[i]
if lc.shouldSkipDeletedChat(chatMID, msg.ID) {
continue
}
lc.cacheGroupMembersFromMessage(chatMID, msg)
if !isBridgeableContentType(msg) {
continue
Expand Down Expand Up @@ -671,7 +674,15 @@ func (lc *LineClient) prefetchMessages(ctx context.Context) {
lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to prefetch message boxes")
return
}
chatMIDs := collectStartupBackfillChatMIDs(messageBoxes, lc.getKnownMemberChatMIDs(), lc.isUserBlocked)
latestMessageIDs := make(map[string]string, len(messageBoxes))
for _, box := range messageBoxes {
if box.LastDeliveredMessageID != nil {
latestMessageIDs[box.ID] = box.LastDeliveredMessageID.MessageID
}
}
chatMIDs := collectStartupBackfillChatMIDs(messageBoxes, lc.getKnownMemberChatMIDs(), func(mid string) bool {
return lc.isUserBlocked(mid) || lc.shouldSkipDeletedChat(mid, latestMessageIDs[mid])
})

workerCount := prefetchMessagesConcurrency
if len(chatMIDs) < workerCount {
Expand Down Expand Up @@ -770,6 +781,9 @@ func (lc *LineClient) backfillRecentMessages(ctx context.Context, chatMID string
// Reverse messages to process oldest first
for i := len(msgs) - 1; i >= 0; i-- {
msg := msgs[i]
if lc.shouldSkipDeletedChat(chatMID, msg.ID) {
continue
}
lc.cacheGroupMembersFromMessage(chatMID, msg)

existing, err := lc.UserLogin.Bridge.DB.Message.GetPartByID(ctx, lc.UserLogin.ID, networkid.MessageID(msg.ID), "")
Expand Down Expand Up @@ -868,6 +882,20 @@ func (lc *LineClient) syncChatsNow(ctx context.Context) {
if err != nil {
lc.UserLogin.Bridge.Log.Warn().Err(err).Str("chat_mid", chat.ChatMid).Msg("Failed to find existing group portal before sync")
}
if (existingPortal == nil || existingPortal.MXID == "") && lc.shouldSkipDeletedChat(chat.ChatMid, "") {
Comment thread
indent[bot] marked this conversation as resolved.
var boxes map[string]line.MessageBox
client, boxes, err = callLineResultUsing(lc, ctx, client, func(client *line.Client) (map[string]line.MessageBox, error) {
return client.GetMessageBoxesByIDsContext(ctx, []string{chat.ChatMid})
})
if err != nil {
lc.UserLogin.Bridge.Log.Warn().Err(err).Str("chat_mid", chat.ChatMid).Msg("Failed to fetch deleted group's latest message ID")
continue
}
latest := boxes[chat.ChatMid].LastDeliveredMessageID
if latest == nil || lc.shouldSkipDeletedChat(chat.ChatMid, latest.MessageID) {
continue
}
}

info := lc.chatToChatInfo(ctx, &chat, true)
// Existing member groups must have their Matrix portal before prefetch/SSE can
Expand Down Expand Up @@ -2043,6 +2071,9 @@ func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) {

case OpSendMessage, OpReceiveMessage:
if op.Message != nil {
if lc.shouldSkipDeletedChat(portalMIDForMessage(op.Message, op.Type), op.Message.ID) {
return
}
if ContentType(op.Message.ContentType) == ContentSystem {
lc.handleSystemMessage(op)
} else {
Expand Down Expand Up @@ -2602,6 +2633,9 @@ func (lc *LineClient) handleSystemMessage(op line.Operation) bool {
// while its converter applies the historical state event synchronously before
// the marker is inserted.
func (lc *LineClient) queueHistoricalSystemMessage(msg *line.Message, opType int) bool {
if lc.shouldSkipDeletedChat(msg.To, msg.ID) {
return false
}
if !isHandledSystemMessage(msg) {
lc.UserLogin.Bridge.Log.Debug().
Str("msg_id", msg.ID).
Expand Down
26 changes: 15 additions & 11 deletions pkg/connector/userinfo.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,16 +46,16 @@ func (lc *LineClient) HandleMatrixReadReceipt(ctx context.Context, read *bridgev

func (lc *LineClient) GetCapabilities(ctx context.Context, portal *bridgev2.Portal) *event.RoomFeatures {
return &event.RoomFeatures{
MaxTextLength: 5000,
Reply: event.CapLevelFullySupported,
Reaction: event.CapLevelPartialSupport,
ReactionCount: 1,
AllowedReactions: getLineAllowedReactions(),
ReadReceipts: true,
Delete: event.CapLevelFullySupported,
DeleteMaxAge: &jsontime.Seconds{Duration: 24 * time.Hour},
DeleteChatForEveryone: true,
LocationMessage: event.CapLevelPartialSupport,
MaxTextLength: 5000,
Reply: event.CapLevelFullySupported,
Reaction: event.CapLevelPartialSupport,
ReactionCount: 1,
AllowedReactions: getLineAllowedReactions(),
ReadReceipts: true,
Delete: event.CapLevelFullySupported,
DeleteMaxAge: &jsontime.Seconds{Duration: 24 * time.Hour},
DeleteChat: true,
LocationMessage: event.CapLevelPartialSupport,
// Pending LINE group invitations surface as Beeper message requests. AcceptWithMessage
// is intentionally left unset so that sending a message implicitly accepts the invite
// (bridgev2 autoAcceptMessageRequest) — LINE rejects messages to un-joined groups.
Expand Down Expand Up @@ -159,7 +159,11 @@ func (lc *LineClient) GetChatInfo(ctx context.Context, portal *bridgev2.Portal)
if len(res.Chats) == 0 {
return nil, fmt.Errorf("chat not found")
}
return lc.chatToChatInfo(ctx, &res.Chats[0], true), nil
info := lc.chatToChatInfo(ctx, &res.Chats[0], true)
if portal.MXID == "" && lc.shouldSkipDeletedChat(mid, "") {
lc.stripRemoteMembersFromInitialChatInfo(info)
}
return info, nil
}

contact := lc.getContact(ctx, string(portal.ID))
Expand Down
68 changes: 64 additions & 4 deletions pkg/line/methods.go
Original file line number Diff line number Diff line change
Expand Up @@ -823,6 +823,34 @@ func (c *Client) GetMessageBoxes(options MessageBoxesOptions) (*MessageBoxesResp
return &wrapper.Data, nil
}

func (c *Client) GetMessageBoxesByIDsContext(ctx context.Context, chatMids []string) (map[string]MessageBox, error) {
req := struct {
MessageBoxIDs []string `json:"messageBoxIds"`
WithUnreadCount bool `json:"withUnreadCount"`
LastMessagesCount int `json:"lastMessagesCount"`
}{chatMids, true, 1}
resp, err := c.callRPCContext(ctx, "TalkService", "getMessageBoxesByIds", req)
if err != nil {
return nil, err
}
var wrapper struct {
Code int `json:"code"`
Data struct {
MessageBoxesByIDs map[string]MessageBox `json:"messageBoxesByIds"`
} `json:"data"`
}
if err := json.Unmarshal(resp, &wrapper); err != nil {
return nil, err
}
if wrapper.Code != 0 {
return nil, fmt.Errorf("getMessageBoxesByIds failed: %s", resp)
}
if wrapper.Data.MessageBoxesByIDs == nil {
return nil, fmt.Errorf("getMessageBoxesByIds returned no message boxes")
}
return wrapper.Data.MessageBoxesByIDs, nil
}

func (c *Client) GetRecentMessagesV2(chatMid string, limit int) ([]*Message, error) {
resp, err := c.callRPC("TalkService", "getRecentMessagesV2", chatMid, limit)
if err != nil {
Expand All @@ -848,8 +876,24 @@ func (c *Client) UnsendMessage(reqSeq int64, messageID string) error {
}

func (c *Client) SendChatRemoved(reqSeq int64, chatMid, lastReadMessageId string, lastReadMessageTime int64) error {
_, err := c.callRPC("TalkService", "sendChatRemoved", reqSeq, chatMid, lastReadMessageId, lastReadMessageTime)
return err
return c.SendChatRemovedContext(context.Background(), reqSeq, chatMid, lastReadMessageId, lastReadMessageTime)
}

func (c *Client) SendChatRemovedContext(ctx context.Context, reqSeq int64, chatMid, lastReadMessageID string, lastReadMessageTime int64) error {
resp, err := c.callRPCContext(ctx, "TalkService", "sendChatRemoved", reqSeq, chatMid, lastReadMessageID, lastReadMessageTime)
if err != nil {
return err
}
var wrapper struct {
Code int `json:"code"`
}
if err := json.Unmarshal(resp, &wrapper); err != nil {
return err
}
if wrapper.Code != 0 {
return fmt.Errorf("sendChatRemoved failed: %s", resp)
}
return nil
}

// CreateChat creates a new LINE group chat with the given members and name.
Expand Down Expand Up @@ -948,8 +992,24 @@ func (c *Client) AcceptChatInvitation(reqSeq int64, chatMid string) error {

// RejectChatInvitation declines a pending invitation into a LINE group chat.
func (c *Client) RejectChatInvitation(reqSeq int64, chatMid string) error {
_, err := c.callRPC("TalkService", "rejectChatInvitation", ChatInvitationRequest{ReqSeq: reqSeq, ChatMid: chatMid})
return err
return c.RejectChatInvitationContext(context.Background(), reqSeq, chatMid)
}

func (c *Client) RejectChatInvitationContext(ctx context.Context, reqSeq int64, chatMid string) error {
resp, err := c.callRPCContext(ctx, "TalkService", "rejectChatInvitation", ChatInvitationRequest{ReqSeq: reqSeq, ChatMid: chatMid})
if err != nil {
return err
}
var wrapper struct {
Code int `json:"code"`
}
if err := json.Unmarshal(resp, &wrapper); err != nil {
return fmt.Errorf("failed to parse rejectChatInvitation response: %w", err)
}
if wrapper.Code != 0 {
return fmt.Errorf("rejectChatInvitation failed: %s", resp)
}
return nil
}

// FindContactByUserid looks up a LINE user by their user ID (not MID).
Expand Down
Loading