mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2024-11-22 22:31:56 -06:00
Code Issues Packages Projects Releases Wiki Activity

More writer fixes in the Sync API

Browse source
This commit is contained in:
Neil Alexander 2022-09-09 13:06:42 +01:00
parent 34e1dc210b
commit 646de03d60
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
6 changed files with 41 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
syncapi/storage/postgres/ignores_table.go
View file

@ -19,6 +19,7 @@ import (
"database/sql" "database/sql"
"encoding/json" "encoding/json"
"github.com/matrix-org/dendrite/internal/sqlutil"
"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"
) )
@ -61,10 +62,10 @@ func NewPostgresIgnoresTable(db *sql.DB) (tables.Ignores, error) {
} }
func (s *ignoresStatements) SelectIgnores( func (s *ignoresStatements) SelectIgnores(
ctx context.Context, userID string, ctx context.Context, txn *sql.Tx, userID string,
) (*types.IgnoredUsers, error) { ) (*types.IgnoredUsers, error) {
var ignoresData []byte var ignoresData []byte
err := s.selectIgnoresStmt.QueryRowContext(ctx, userID).Scan(&ignoresData) err := sqlutil.TxStmt(txn, s.selectIgnoresStmt).QueryRowContext(ctx, userID).Scan(&ignoresData)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -76,12 +77,12 @@ func (s *ignoresStatements) SelectIgnores(
} }
func (s *ignoresStatements) UpsertIgnores( func (s *ignoresStatements) UpsertIgnores(
ctx context.Context, userID string, ignores *types.IgnoredUsers, ctx context.Context, txn *sql.Tx, userID string, ignores *types.IgnoredUsers,
) error { ) error {
ignoresJSON, err := json.Marshal(ignores) ignoresJSON, err := json.Marshal(ignores)
if err != nil { if err != nil {
return err return err
} }
_, err = s.upsertIgnoresStmt.ExecContext(ctx, userID, ignoresJSON) _, err = sqlutil.TxStmt(txn, s.upsertIgnoresStmt).ExecContext(ctx, userID, ignoresJSON)
return err return err
} }

12
syncapi/storage/postgres/notification_data_table.go
View file

@ -70,13 +70,13 @@ const selectUserUnreadNotificationCountsSQL = `SELECT
const selectMaxNotificationIDSQL = `SELECT CASE COUNT(*) WHEN 0 THEN 0 ELSE MAX(id) END FROM syncapi_notification_data` const selectMaxNotificationIDSQL = `SELECT CASE COUNT(*) WHEN 0 THEN 0 ELSE MAX(id) END FROM syncapi_notification_data`
func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) { func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, txn *sql.Tx, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
err = r.upsertRoomUnreadCounts.QueryRowContext(ctx, userID, roomID, notificationCount, highlightCount).Scan(&pos) err = sqlutil.TxStmt(txn, r.upsertRoomUnreadCounts).QueryRowContext(ctx, userID, roomID, notificationCount, highlightCount).Scan(&pos)
return return
} }
func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) { func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context, txn *sql.Tx, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) {
rows, err := r.selectUserUnreadCounts.QueryContext(ctx, userID, fromExcl, toIncl) rows, err := sqlutil.TxStmt(txn, r.selectUserUnreadCounts).QueryContext(ctx, userID, fromExcl, toIncl)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -101,8 +101,8 @@ func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context,
return roomCounts, rows.Err() return roomCounts, rows.Err()
} }
func (r *notificationDataStatements) SelectMaxID(ctx context.Context) (int64, error) { func (r *notificationDataStatements) SelectMaxID(ctx context.Context, txn *sql.Tx) (int64, error) {
var id int64 var id int64
err := r.selectMaxID.QueryRowContext(ctx).Scan(&id) err := sqlutil.TxStmt(txn, r.selectMaxID).QueryRowContext(ctx).Scan(&id)
return id, err return id, err
} }

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

@ -108,7 +108,7 @@ func (d *Database) MaxStreamPositionForAccountData(ctx context.Context) (types.S
} }
func (d *Database) MaxStreamPositionForNotificationData(ctx context.Context) (types.StreamPosition, error) { func (d *Database) MaxStreamPositionForNotificationData(ctx context.Context) (types.StreamPosition, error) {
id, err := d.NotificationData.SelectMaxID(ctx) id, err := d.NotificationData.SelectMaxID(ctx, nil)
if err != nil { if err != nil {
return 0, fmt.Errorf("d.NotificationData.SelectMaxID: %w", err) return 0, fmt.Errorf("d.NotificationData.SelectMaxID: %w", err)
} }
@ -1029,15 +1029,15 @@ func (d *Database) GetRoomReceipts(ctx context.Context, roomIDs []string, stream
} }
func (d *Database) UpsertRoomUnreadNotificationCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) { func (d *Database) UpsertRoomUnreadNotificationCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
err = d.Writer.Do(nil, nil, func(_ *sql.Tx) error { err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
pos, err = d.NotificationData.UpsertRoomUnreadCounts(ctx, userID, roomID, notificationCount, highlightCount) pos, err = d.NotificationData.UpsertRoomUnreadCounts(ctx, txn, userID, roomID, notificationCount, highlightCount)
return err return err
}) })
return return
} }
func (d *Database) GetUserUnreadNotificationCounts(ctx context.Context, userID string, from, to types.StreamPosition) (map[string]*eventutil.NotificationData, error) { func (d *Database) GetUserUnreadNotificationCounts(ctx context.Context, userID string, from, to types.StreamPosition) (map[string]*eventutil.NotificationData, error) {
return d.NotificationData.SelectUserUnreadCounts(ctx, userID, from, to) return d.NotificationData.SelectUserUnreadCounts(ctx, nil, userID, from, to)
} }
func (d *Database) SelectContextEvent(ctx context.Context, roomID, eventID string) (int, gomatrixserverlib.HeaderedEvent, error) { func (d *Database) SelectContextEvent(ctx context.Context, roomID, eventID string) (int, gomatrixserverlib.HeaderedEvent, error) {
@ -1052,15 +1052,23 @@ func (d *Database) SelectContextAfterEvent(ctx context.Context, id int, roomID s
} }
func (d *Database) IgnoresForUser(ctx context.Context, userID string) (*types.IgnoredUsers, error) { func (d *Database) IgnoresForUser(ctx context.Context, userID string) (*types.IgnoredUsers, error) {
return d.Ignores.SelectIgnores(ctx, userID) return d.Ignores.SelectIgnores(ctx, nil, userID)
} }
func (d *Database) UpdateIgnoresForUser(ctx context.Context, userID string, ignores *types.IgnoredUsers) error { func (d *Database) UpdateIgnoresForUser(ctx context.Context, userID string, ignores *types.IgnoredUsers) error {
return d.Ignores.UpsertIgnores(ctx, userID, ignores) return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
return d.Ignores.UpsertIgnores(ctx, txn, userID, ignores)
})
} }
func (d *Database) UpdatePresence(ctx context.Context, userID string, presence types.Presence, statusMsg *string, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (types.StreamPosition, error) { func (d *Database) UpdatePresence(ctx context.Context, userID string, presence types.Presence, statusMsg *string, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (types.StreamPosition, error) {
return d.Presence.UpsertPresence(ctx, nil, userID, statusMsg, presence, lastActiveTS, fromSync) var pos types.StreamPosition
var err error
_ = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
pos, err = d.Presence.UpsertPresence(ctx, txn, userID, statusMsg, presence, lastActiveTS, fromSync)
return nil
})
return pos, err
} }
func (d *Database) GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error) { func (d *Database) GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error) {

9
syncapi/storage/sqlite3/ignores_table.go
View file

@ -19,6 +19,7 @@ import (
"database/sql" "database/sql"
"encoding/json" "encoding/json"
"github.com/matrix-org/dendrite/internal/sqlutil"
"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"
) )
@ -61,10 +62,10 @@ func NewSqliteIgnoresTable(db *sql.DB) (tables.Ignores, error) {
} }
func (s *ignoresStatements) SelectIgnores( func (s *ignoresStatements) SelectIgnores(
ctx context.Context, userID string, ctx context.Context, txn *sql.Tx, userID string,
) (*types.IgnoredUsers, error) { ) (*types.IgnoredUsers, error) {
var ignoresData []byte var ignoresData []byte
err := s.selectIgnoresStmt.QueryRowContext(ctx, userID).Scan(&ignoresData) err := sqlutil.TxStmt(txn, s.selectIgnoresStmt).QueryRowContext(ctx, userID).Scan(&ignoresData)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -76,12 +77,12 @@ func (s *ignoresStatements) SelectIgnores(
} }
func (s *ignoresStatements) UpsertIgnores( func (s *ignoresStatements) UpsertIgnores(
ctx context.Context, userID string, ignores *types.IgnoredUsers, ctx context.Context, txn *sql.Tx, userID string, ignores *types.IgnoredUsers,
) error { ) error {
ignoresJSON, err := json.Marshal(ignores) ignoresJSON, err := json.Marshal(ignores)
if err != nil { if err != nil {
return err return err
} }
_, err = s.upsertIgnoresStmt.ExecContext(ctx, userID, ignoresJSON) _, err = sqlutil.TxStmt(txn, s.upsertIgnoresStmt).ExecContext(ctx, userID, ignoresJSON)
return err return err
} }

10
syncapi/storage/sqlite3/notification_data_table.go
View file

@ -72,7 +72,7 @@ const selectUserUnreadNotificationCountsSQL = `SELECT
const selectMaxNotificationIDSQL = `SELECT CASE COUNT(*) WHEN 0 THEN 0 ELSE MAX(id) END FROM syncapi_notification_data` const selectMaxNotificationIDSQL = `SELECT CASE COUNT(*) WHEN 0 THEN 0 ELSE MAX(id) END FROM syncapi_notification_data`
func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) { func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, txn *sql.Tx, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
pos, err = r.streamIDStatements.nextNotificationID(ctx, nil) pos, err = r.streamIDStatements.nextNotificationID(ctx, nil)
if err != nil { if err != nil {
return return
@ -81,8 +81,8 @@ func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context,
return return
} }
func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) { func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context, txn *sql.Tx, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) {
rows, err := r.selectUserUnreadCounts.QueryContext(ctx, userID, fromExcl, toIncl) rows, err := sqlutil.TxStmt(txn, r.selectUserUnreadCounts).QueryContext(ctx, userID, fromExcl, toIncl)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -107,8 +107,8 @@ func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context,
return roomCounts, rows.Err() return roomCounts, rows.Err()
} }
func (r *notificationDataStatements) SelectMaxID(ctx context.Context) (int64, error) { func (r *notificationDataStatements) SelectMaxID(ctx context.Context, txn *sql.Tx) (int64, error) {
var id int64 var id int64
err := r.selectMaxID.QueryRowContext(ctx).Scan(&id) err := sqlutil.TxStmt(txn, r.selectMaxID).QueryRowContext(ctx).Scan(&id)
return id, err return id, err
} }

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

@ -189,14 +189,14 @@ type Memberships interface {
} }
type NotificationData interface { type NotificationData interface {
UpsertRoomUnreadCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (types.StreamPosition, error) UpsertRoomUnreadCounts(ctx context.Context, txn *sql.Tx, userID, roomID string, notificationCount, highlightCount int) (types.StreamPosition, error)
SelectUserUnreadCounts(ctx context.Context, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) SelectUserUnreadCounts(ctx context.Context, txn *sql.Tx, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error)
SelectMaxID(ctx context.Context) (int64, error) SelectMaxID(ctx context.Context, txn *sql.Tx) (int64, error)
} }
type Ignores interface { type Ignores interface {
SelectIgnores(ctx context.Context, userID string) (*types.IgnoredUsers, error) SelectIgnores(ctx context.Context, txn *sql.Tx, userID string) (*types.IgnoredUsers, error)
UpsertIgnores(ctx context.Context, userID string, ignores *types.IgnoredUsers) error UpsertIgnores(ctx context.Context, txn *sql.Tx, userID string, ignores *types.IgnoredUsers) error
} }
type Presence interface { type Presence interface {