forked from MirrorHub/mautrix-whatsapp
Mark all messages as read instead of only last one
This commit is contained in:
parent
04ab823a4d
commit
a909750dcc
7 changed files with 125 additions and 7 deletions
|
@ -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))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -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
|
||||
|
|
22
database/upgrades/2021-11-30-store-last-read-state.go
Normal file
22
database/upgrades/2021-11-30-store-last-read-state.go
Normal file
|
@ -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
|
||||
}}
|
||||
}
|
|
@ -39,7 +39,7 @@ type upgrade struct {
|
|||
fn upgradeFunc
|
||||
}
|
||||
|
||||
const NumberOfUpgrades = 30
|
||||
const NumberOfUpgrades = 31
|
||||
|
||||
var upgrades [NumberOfUpgrades]upgrade
|
||||
|
||||
|
|
55
database/userportal.go
Normal file
55
database/userportal.go
Normal file
|
@ -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 <https://www.gnu.org/licenses/>.
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
26
portal.go
26
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() {
|
||||
|
|
4
user.go
4
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)
|
||||
|
|
Loading…
Reference in a new issue