mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-26 08:13:09 -06:00
Code Issues Packages Projects Releases Wiki Activity

Tweaks

Browse source
This commit is contained in:
Neil Alexander 2021-01-12 14:33:28 +00:00
parent 53abc2c0ec
commit 6ae9c1fea8
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
6 changed files with 25 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
syncapi/consumers/eduserver_sendtodevice.go
View file

@ -105,6 +105,8 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
return err return err
} }
util.GetLogger(context.TODO()).Infof("Stored send-to-device message at position %d", streamPos)
s.stream.Advance(streamPos) s.stream.Advance(streamPos)
s.notifier.OnNewSendToDevice( s.notifier.OnNewSendToDevice(
output.UserID, output.UserID,

14
syncapi/storage/postgres/send_to_device_table.go
View file

@ -50,7 +50,7 @@ const insertSendToDeviceMessageSQL = `
const selectSendToDeviceMessagesSQL = ` const selectSendToDeviceMessagesSQL = `
SELECT id, user_id, device_id, content SELECT id, user_id, device_id, content
FROM syncapi_send_to_device FROM syncapi_send_to_device
WHERE user_id = $1 AND device_id = $2 AND id > $3 AND id <= $4 WHERE user_id = $1 AND device_id = $2 AND id <= $3
ORDER BY id DESC ORDER BY id DESC
` `
@ -98,16 +98,16 @@ func (s *sendToDeviceStatements) InsertSendToDeviceMessage(
} }
func (s *sendToDeviceStatements) SelectSendToDeviceMessages( func (s *sendToDeviceStatements) SelectSendToDeviceMessages(
ctx context.Context, txn *sql.Tx, userID, deviceID string, from, to types.StreamPosition, ctx context.Context, txn *sql.Tx, userID, deviceID string, to types.StreamPosition,
) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) { ) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) {
rows, err := sqlutil.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID, from, to) rows, err := sqlutil.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID, to)
if err != nil { if err != nil {
return return
} }
defer internal.CloseAndLogIfError(ctx, rows, "SelectSendToDeviceMessages: rows.close() failed") defer internal.CloseAndLogIfError(ctx, rows, "SelectSendToDeviceMessages: rows.close() failed")
for rows.Next() { for rows.Next() {
var id types.SendToDeviceNID var id types.StreamPosition
var userID, deviceID, content string var userID, deviceID, content string
if err = rows.Scan(&id, &userID, &deviceID, &content); err != nil { if err = rows.Scan(&id, &userID, &deviceID, &content); err != nil {
return return
@ -118,11 +118,11 @@ func (s *sendToDeviceStatements) SelectSendToDeviceMessages(
DeviceID: deviceID, DeviceID: deviceID,
} }
if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil { if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil {
return continue
} }
events = append(events, event) events = append(events, event)
if types.StreamPosition(id) > lastPos { if id > lastPos {
lastPos = types.StreamPosition(id) lastPos = id
} }
} }

11
syncapi/storage/shared/syncserver.go
View file

@ -29,6 +29,7 @@ import (
"github.com/matrix-org/dendrite/syncapi/storage/tables" "github.com/matrix-org/dendrite/syncapi/storage/tables"
"github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
) )
@ -893,7 +894,7 @@ func (d *Database) StoreNewSendForDeviceMessage(
if err != nil { if err != nil {
return 0, err return 0, err
} }
return return newPos, nil
} }
func (d *Database) SendToDeviceUpdatesForSync( func (d *Database) SendToDeviceUpdatesForSync(
@ -902,14 +903,14 @@ func (d *Database) SendToDeviceUpdatesForSync(
from, to types.StreamPosition, from, to types.StreamPosition,
) (types.StreamPosition, []types.SendToDeviceEvent, error) { ) (types.StreamPosition, []types.SendToDeviceEvent, error) {
// First of all, get our send-to-device updates for this user. // First of all, get our send-to-device updates for this user.
lastPos, events, err := d.SendToDevice.SelectSendToDeviceMessages(ctx, nil, userID, deviceID, from, to) lastPos, events, err := d.SendToDevice.SelectSendToDeviceMessages(ctx, nil, userID, deviceID, to)
if err != nil { if err != nil {
return 0, nil, fmt.Errorf("d.SendToDevice.SelectSendToDeviceMessages: %w", err) return from, nil, fmt.Errorf("d.SendToDevice.SelectSendToDeviceMessages: %w", err)
} }
// If there's nothing to do then stop here. // If there's nothing to do then stop here.
if len(events) == 0 { if len(events) == 0 {
return 0, nil, nil return to, nil, nil
} }
// If we've advanced past this stream position for this // If we've advanced past this stream position for this
@ -917,7 +918,7 @@ func (d *Database) SendToDeviceUpdatesForSync(
if err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error { if err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
return d.SendToDevice.DeleteSendToDeviceMessages(ctx, txn, userID, deviceID, from) return d.SendToDevice.DeleteSendToDeviceMessages(ctx, txn, userID, deviceID, from)
}); err != nil { }); err != nil {
return 0, nil, fmt.Errorf("d.Writer.Do: %w", err) logrus.WithError(err).Errorf("Failed to clean up old send-to-device messages for user %q device %q", userID, deviceID)
} }
return lastPos, events, nil return lastPos, events, nil

14
syncapi/storage/sqlite3/send_to_device_table.go
View file

@ -47,7 +47,7 @@ const insertSendToDeviceMessageSQL = `
const selectSendToDeviceMessagesSQL = ` const selectSendToDeviceMessagesSQL = `
SELECT id, user_id, device_id, content SELECT id, user_id, device_id, content
FROM syncapi_send_to_device FROM syncapi_send_to_device
WHERE user_id = $1 AND device_id = $2 AND id > $3 AND id <= $4 WHERE user_id = $1 AND device_id = $2 AND id <= $3
ORDER BY id DESC ORDER BY id DESC
` `
@ -104,16 +104,16 @@ func (s *sendToDeviceStatements) InsertSendToDeviceMessage(
} }
func (s *sendToDeviceStatements) SelectSendToDeviceMessages( func (s *sendToDeviceStatements) SelectSendToDeviceMessages(
ctx context.Context, txn *sql.Tx, userID, deviceID string, from, to types.StreamPosition, ctx context.Context, txn *sql.Tx, userID, deviceID string, to types.StreamPosition,
) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) { ) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) {
rows, err := sqlutil.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID, from, to) rows, err := sqlutil.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID, to)
if err != nil { if err != nil {
return return
} }
defer internal.CloseAndLogIfError(ctx, rows, "SelectSendToDeviceMessages: rows.close() failed") defer internal.CloseAndLogIfError(ctx, rows, "SelectSendToDeviceMessages: rows.close() failed")
for rows.Next() { for rows.Next() {
var id types.SendToDeviceNID var id types.StreamPosition
var userID, deviceID, content string var userID, deviceID, content string
if err = rows.Scan(&id, &userID, &deviceID, &content); err != nil { if err = rows.Scan(&id, &userID, &deviceID, &content); err != nil {
return return
@ -124,11 +124,11 @@ func (s *sendToDeviceStatements) SelectSendToDeviceMessages(
DeviceID: deviceID, DeviceID: deviceID,
} }
if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil { if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil {
return continue
} }
events = append(events, event) events = append(events, event)
if types.StreamPosition(id) > lastPos { if id > lastPos {
lastPos = types.StreamPosition(id) lastPos = id
} }
} }

4
syncapi/storage/tables/interface.go
View file

@ -147,8 +147,8 @@ type BackwardsExtremities interface {
// sync response, as the client is seemingly trying to repeat the same /sync. // sync response, as the client is seemingly trying to repeat the same /sync.
type SendToDevice interface { type SendToDevice interface {
InsertSendToDeviceMessage(ctx context.Context, txn *sql.Tx, userID, deviceID, content string) (pos types.StreamPosition, err error) InsertSendToDeviceMessage(ctx context.Context, txn *sql.Tx, userID, deviceID, content string) (pos types.StreamPosition, err error)
SelectSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string, from, to types.StreamPosition) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) SelectSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string, to types.StreamPosition) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error)
DeleteSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string, pos types.StreamPosition) (err error) DeleteSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string, from types.StreamPosition) (err error)
SelectMaxSendToDeviceMessageID(ctx context.Context, txn *sql.Tx) (id int64, err error) SelectMaxSendToDeviceMessageID(ctx context.Context, txn *sql.Tx) (id int64, err error)
} }

4
syncapi/types/types.go
View file

@ -492,11 +492,9 @@ func NewLeaveResponse() *LeaveResponse {
return &res return &res
} }
type SendToDeviceNID int
type SendToDeviceEvent struct { type SendToDeviceEvent struct {
gomatrixserverlib.SendToDeviceEvent gomatrixserverlib.SendToDeviceEvent
ID SendToDeviceNID ID StreamPosition
UserID string UserID string
DeviceID string DeviceID string
} }