From a909750dcc14304c9c06c36caf4b1dcac627ea22 Mon Sep 17 00:00:00 2001 From: Tulir Asokan Date: Tue, 30 Nov 2021 16:38:37 +0200 Subject: [PATCH] Mark all messages as read instead of only last one --- custompuppet.go | 8 +-- database/message.go | 15 +++++ .../2021-11-30-store-last-read-state.go | 22 ++++++++ database/upgrades/upgrades.go | 2 +- database/userportal.go | 55 +++++++++++++++++++ portal.go | 26 +++++++++ user.go | 4 ++ 7 files changed, 125 insertions(+), 7 deletions(-) create mode 100644 database/upgrades/2021-11-30-store-last-read-state.go create mode 100644 database/userportal.go diff --git a/custompuppet.go b/custompuppet.go index 643de1f..e8b4e4f 100644 --- a/custompuppet.go +++ b/custompuppet.go @@ -256,12 +256,8 @@ func (puppet *Puppet) handleReceiptEvent(portal *Portal, event *event.Event) { } else if isDoublePuppeted, _ := receipt.Extra[doublePuppetField].(bool); isDoublePuppeted { puppet.customUser.log.Debugfln("Ignoring double puppeted read receipt %+v", event.Content.Raw) // Ignore double puppeted read receipts. - } else if message := puppet.bridge.DB.Message.GetByMXID(eventID); message != nil { - puppet.customUser.log.Debugfln("Marking %s/%s in %s/%s as read", message.JID, message.MXID, portal.Key.JID, portal.MXID) - err := puppet.customUser.Client.MarkRead([]types.MessageID{message.JID}, time.UnixMilli(receipt.Timestamp), portal.Key.JID, message.Sender) - if err != nil { - puppet.customUser.log.Warnln("Error marking read:", err) - } + } else { + portal.HandleMatrixReadReceipt(puppet.customUser, eventID, time.UnixMilli(receipt.Timestamp)) } } } diff --git a/database/message.go b/database/message.go index 382894c..b6ef12b 100644 --- a/database/message.go +++ b/database/message.go @@ -62,6 +62,10 @@ const ( SELECT chat_jid, chat_receiver, jid, mxid, sender, timestamp, sent, decryption_error FROM message WHERE chat_jid=$1 AND chat_receiver=$2 AND sent=true ORDER BY timestamp ASC LIMIT 1 ` + getMessagesBetweenQuery = ` + SELECT chat_jid, chat_receiver, jid, mxid, sender, timestamp, sent, decryption_error FROM message + WHERE chat_jid=$1 AND chat_receiver=$2 AND timestamp>$3 AND timestamp<=$4 AND sent=true ORDER BY timestamp ASC + ` ) func (mq *MessageQuery) GetAll(chat PortalKey) (messages []*Message) { @@ -100,6 +104,17 @@ func (mq *MessageQuery) GetFirstInChat(chat PortalKey) *Message { return mq.maybeScan(mq.db.QueryRow(getFirstMessageInChatQuery, chat.JID, chat.Receiver)) } +func (mq *MessageQuery) GetMessagesBetween(chat PortalKey, minTimestamp, maxTimestamp time.Time) (messages []*Message) { + rows, err := mq.db.Query(getMessagesBetweenQuery, chat.JID, chat.Receiver, minTimestamp.Unix(), maxTimestamp.Unix()) + if err != nil || rows == nil { + return nil + } + for rows.Next() { + messages = append(messages, mq.New().Scan(rows)) + } + return +} + func (mq *MessageQuery) maybeScan(row *sql.Row) *Message { if row == nil { return nil diff --git a/database/upgrades/2021-11-30-store-last-read-state.go b/database/upgrades/2021-11-30-store-last-read-state.go new file mode 100644 index 0000000..b5c3ec6 --- /dev/null +++ b/database/upgrades/2021-11-30-store-last-read-state.go @@ -0,0 +1,22 @@ +package upgrades + +import ( + "database/sql" +) + +func init() { + upgrades[30] = upgrade{"Store last read message timestamp in database", func(tx *sql.Tx, ctx context) error { + _, err := tx.Exec(`CREATE TABLE user_portal ( + user_mxid TEXT, + portal_jid TEXT, + portal_receiver TEXT, + + last_read_ts BIGINT NOT NULL DEFAULT 0, + + PRIMARY KEY (user_mxid, portal_jid, portal_receiver), + FOREIGN KEY (user_mxid) REFERENCES "user"(mxid) ON DELETE CASCADE ON UPDATE CASCADE, + FOREIGN KEY (portal_jid, portal_receiver) REFERENCES portal(jid, receiver) ON DELETE CASCADE ON UPDATE CASCADE + )`) + return err + }} +} diff --git a/database/upgrades/upgrades.go b/database/upgrades/upgrades.go index 87d2490..3c996c6 100644 --- a/database/upgrades/upgrades.go +++ b/database/upgrades/upgrades.go @@ -39,7 +39,7 @@ type upgrade struct { fn upgradeFunc } -const NumberOfUpgrades = 30 +const NumberOfUpgrades = 31 var upgrades [NumberOfUpgrades]upgrade diff --git a/database/userportal.go b/database/userportal.go new file mode 100644 index 0000000..8c6768b --- /dev/null +++ b/database/userportal.go @@ -0,0 +1,55 @@ +// mautrix-whatsapp - A Matrix-WhatsApp puppeting bridge. +// Copyright (C) 2021 Tulir Asokan +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package database + +import ( + "database/sql" + "errors" + "fmt" + "time" +) + +func (user *User) GetLastReadTS(portal PortalKey) time.Time { + var ts int64 + err := user.db.QueryRow("SELECT last_read_ts FROM user_portal WHERE user_mxid=$1 AND portal_jid=$2 AND portal_receiver=$3", user.MXID, portal.JID, portal.Receiver).Scan(&ts) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + user.log.Warnfln("Failed to scan last read timestamp from user portal table: %v", err) + } + if ts == 0 { + return time.Time{} + } + return time.Unix(ts, 0) +} + +func (user *User) SetLastReadTS(portal PortalKey, ts time.Time) { + var err error + if user.db.dialect == "postgres" { + _, err = user.db.Exec(` + INSERT INTO user_portal (user_mxid, portal_jid, portal_receiver, last_read_ts) VALUES ($1, $2, $3, $4) + ON CONFLICT (user_mxid, portal_jid, portal_receiver) DO UPDATE SET last_read_ts=$4 + `, user.MXID, portal.JID, portal.Receiver, ts.Unix()) + } else if user.db.dialect == "sqlite3" { + _, err = user.db.Exec( + "INSERT OR REPLACE INTO user_portal (user_mxid, portal_jid, portal_receiver, last_read_ts) VALUES (?, ?, ?, ?)", + user.MXID, portal.JID, portal.Receiver, ts.Unix()) + } else { + err = fmt.Errorf("unsupported dialect %s", user.db.dialect) + } + if err != nil { + user.log.Warnfln("Failed to update last read timestamp: %v", err) + } +} diff --git a/portal.go b/portal.go index df23b07..422d133 100644 --- a/portal.go +++ b/portal.go @@ -2192,6 +2192,32 @@ func (portal *Portal) HandleMatrixRedaction(sender *User, evt *event.Event) { } } +func (portal *Portal) HandleMatrixReadReceipt(sender *User, eventID id.EventID, receiptTimestamp time.Time) { + maxTimestamp := receiptTimestamp + if message := portal.bridge.DB.Message.GetByMXID(eventID); message != nil { + maxTimestamp = message.Timestamp + } + + prevTimestamp := sender.GetLastReadTS(portal.Key) + if prevTimestamp.IsZero() { + prevTimestamp = maxTimestamp.Add(-2 * time.Second) + } + + messages := portal.bridge.DB.Message.GetMessagesBetween(portal.Key, prevTimestamp, maxTimestamp) + sender.SetLastReadTS(portal.Key, messages[len(messages)-1].Timestamp) + groupedMessages := make(map[types.JID][]types.MessageID) + for _, msg := range messages { + groupedMessages[msg.Sender] = append(groupedMessages[msg.Sender], msg.JID) + } + portal.log.Debugfln("Sending read receipts by %s: %v", sender.JID, groupedMessages) + for messageSender, ids := range groupedMessages { + err := sender.Client.MarkRead(ids, receiptTimestamp, portal.Key.JID, messageSender) + if err != nil { + portal.log.Warnfln("Failed to mark %v as read by %s: %v", ids, sender.JID, err) + } + } +} + func (portal *Portal) canBridgeFrom(sender *User, evtType string) bool { if !sender.IsLoggedIn() { if portal.HasRelaybot() { diff --git a/user.go b/user.go index 79a37e5..c275a26 100644 --- a/user.go +++ b/user.go @@ -747,6 +747,9 @@ func (user *User) handleReceipt(receipt *events.Receipt) { markAsRead = append(markAsRead, msg) } } + if receipt.Sender.User == user.JID.User { + user.SetLastReadTS(portal.Key, markAsRead[0].Timestamp) + } intent := user.bridge.GetPuppetByJID(receipt.Sender).IntentFor(portal) for _, msg := range markAsRead { err := intent.MarkReadWithContent(portal.MXID, msg.MXID, &CustomReadReceipt{DoublePuppet: intent.IsCustomPuppet}) @@ -767,6 +770,7 @@ func (user *User) markSelfReadFull(portal *Portal) { if lastMessage == nil { return } + user.SetLastReadTS(portal.Key, lastMessage.Timestamp) err := puppet.CustomIntent().MarkReadWithContent(portal.MXID, lastMessage.MXID, &CustomReadReceipt{DoublePuppet: true}) if err != nil { user.log.Warnfln("Failed to mark %s (last message) in %s as read: %v", lastMessage.MXID, portal.MXID, err)