From c192c6096b20bcbd8e15700fdd72a21243dbd652 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Thu, 7 Apr 2022 14:44:35 +0100 Subject: [PATCH] Store ignore knowledge in the sync API --- clientapi/producers/syncapi.go | 9 +-- clientapi/routing/account_data.go | 11 ++- clientapi/routing/room_tagging.go | 4 +- internal/eventutil/types.go | 9 ++- syncapi/consumers/clientapi.go | 9 +++ syncapi/storage/interface.go | 3 + syncapi/storage/postgres/ignores_table.go | 87 +++++++++++++++++++++++ syncapi/storage/postgres/syncserver.go | 5 ++ syncapi/storage/shared/syncserver.go | 9 +++ syncapi/storage/sqlite3/ignores_table.go | 87 +++++++++++++++++++++++ syncapi/storage/sqlite3/syncserver.go | 5 ++ syncapi/storage/tables/interface.go | 5 ++ syncapi/streams/stream_invite.go | 2 +- syncapi/streams/stream_pdu.go | 23 +++--- syncapi/streams/stream_receipt.go | 2 +- syncapi/streams/stream_sendtodevice.go | 2 +- syncapi/streams/stream_typing.go | 2 +- syncapi/types/provider.go | 2 +- syncapi/types/types.go | 4 ++ userapi/consumers/syncapi_streamevent.go | 6 +- 20 files changed, 250 insertions(+), 36 deletions(-) create mode 100644 syncapi/storage/postgres/ignores_table.go create mode 100644 syncapi/storage/sqlite3/ignores_table.go diff --git a/clientapi/producers/syncapi.go b/clientapi/producers/syncapi.go index 2dee04e3a..e5dbbedf2 100644 --- a/clientapi/producers/syncapi.go +++ b/clientapi/producers/syncapi.go @@ -40,7 +40,7 @@ type SyncAPIProducer struct { } // SendData sends account data to the sync API server -func (p *SyncAPIProducer) SendData(userID string, roomID string, dataType string, readMarker *eventutil.ReadMarkerJSON) error { +func (p *SyncAPIProducer) SendData(userID string, roomID string, dataType string, readMarker *eventutil.ReadMarkerJSON, ignoredUsers *types.IgnoredUsers) error { m := &nats.Msg{ Subject: p.TopicClientData, Header: nats.Header{}, @@ -48,9 +48,10 @@ func (p *SyncAPIProducer) SendData(userID string, roomID string, dataType string m.Header.Set(jetstream.UserID, userID) data := eventutil.AccountData{ - RoomID: roomID, - Type: dataType, - ReadMarker: readMarker, + RoomID: roomID, + Type: dataType, + ReadMarker: readMarker, + IgnoredUsers: ignoredUsers, } var err error m.Data, err = json.Marshal(data) diff --git a/clientapi/routing/account_data.go b/clientapi/routing/account_data.go index 873ffaf5d..9399fd0bb 100644 --- a/clientapi/routing/account_data.go +++ b/clientapi/routing/account_data.go @@ -25,6 +25,7 @@ import ( "github.com/matrix-org/dendrite/clientapi/producers" "github.com/matrix-org/dendrite/internal/eventutil" roomserverAPI "github.com/matrix-org/dendrite/roomserver/api" + "github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/dendrite/userapi/api" "github.com/matrix-org/util" @@ -126,8 +127,14 @@ func SaveAccountData( return util.ErrorResponse(err) } + var ignoredUsers *types.IgnoredUsers + if dataType == "m.ignored_user_list" { + ignoredUsers = &types.IgnoredUsers{} + _ = json.Unmarshal(body, ignoredUsers) + } + // TODO: user API should do this since it's account data - if err := syncProducer.SendData(userID, roomID, dataType, nil); err != nil { + if err := syncProducer.SendData(userID, roomID, dataType, nil, ignoredUsers); err != nil { util.GetLogger(req.Context()).WithError(err).Error("syncProducer.SendData failed") return jsonerror.InternalServerError() } @@ -184,7 +191,7 @@ func SaveReadMarker( return util.ErrorResponse(err) } - if err := syncProducer.SendData(device.UserID, roomID, "m.fully_read", &r); err != nil { + if err := syncProducer.SendData(device.UserID, roomID, "m.fully_read", &r, nil); err != nil { util.GetLogger(req.Context()).WithError(err).Error("syncProducer.SendData failed") return jsonerror.InternalServerError() } diff --git a/clientapi/routing/room_tagging.go b/clientapi/routing/room_tagging.go index 83294b180..ce173613e 100644 --- a/clientapi/routing/room_tagging.go +++ b/clientapi/routing/room_tagging.go @@ -98,7 +98,7 @@ func PutTag( return jsonerror.InternalServerError() } - if err = syncProducer.SendData(userID, roomID, "m.tag", nil); err != nil { + if err = syncProducer.SendData(userID, roomID, "m.tag", nil, nil); err != nil { logrus.WithError(err).Error("Failed to send m.tag account data update to syncapi") } @@ -151,7 +151,7 @@ func DeleteTag( } // TODO: user API should do this since it's account data - if err := syncProducer.SendData(userID, roomID, "m.tag", nil); err != nil { + if err := syncProducer.SendData(userID, roomID, "m.tag", nil, nil); err != nil { logrus.WithError(err).Error("Failed to send m.tag account data update to syncapi") } diff --git a/internal/eventutil/types.go b/internal/eventutil/types.go index 17861d6c5..afc62d8c2 100644 --- a/internal/eventutil/types.go +++ b/internal/eventutil/types.go @@ -17,6 +17,8 @@ package eventutil import ( "errors" "strconv" + + "github.com/matrix-org/dendrite/syncapi/types" ) // ErrProfileNoExists is returned when trying to lookup a user's profile that @@ -26,9 +28,10 @@ var ErrProfileNoExists = errors.New("no known profile for given user ID") // AccountData represents account data sent from the client API server to the // sync API server type AccountData struct { - RoomID string `json:"room_id"` - Type string `json:"type"` - ReadMarker *ReadMarkerJSON `json:"read_marker,omitempty"` // optional + RoomID string `json:"room_id"` + Type string `json:"type"` + ReadMarker *ReadMarkerJSON `json:"read_marker,omitempty"` // optional + IgnoredUsers *types.IgnoredUsers `json:"ignored_users,omitempty"` // optional } type ReadMarkerJSON struct { diff --git a/syncapi/consumers/clientapi.go b/syncapi/consumers/clientapi.go index c28da4600..eec369c1a 100644 --- a/syncapi/consumers/clientapi.go +++ b/syncapi/consumers/clientapi.go @@ -119,6 +119,15 @@ func (s *OutputClientDataConsumer) onMessage(ctx context.Context, msg *nats.Msg) return false } + if output.IgnoredUsers != nil { + if err := s.db.UpdateIgnoresForUser(ctx, userID, output.IgnoredUsers); err != nil { + log.WithError(err).WithFields(logrus.Fields{ + "user_id": userID, + }).Errorf("Failed to update ignored users") + sentry.CaptureException(err) + } + } + s.stream.Advance(streamPos) s.notifier.OnNewAccountData(userID, types.StreamingToken{AccountDataPosition: streamPos}) diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go index 03313ec6e..d6cf0f253 100644 --- a/syncapi/storage/interface.go +++ b/syncapi/storage/interface.go @@ -149,4 +149,7 @@ type Database interface { SelectContextAfterEvent(ctx context.Context, id int, roomID string, filter *gomatrixserverlib.RoomEventFilter) (int, []*gomatrixserverlib.HeaderedEvent, error) StreamToTopologicalPosition(ctx context.Context, roomID string, streamPos types.StreamPosition, backwardOrdering bool) (types.TopologyToken, error) + + IgnoresForUser(ctx context.Context, userID string) (*types.IgnoredUsers, error) + UpdateIgnoresForUser(ctx context.Context, userID string, ignores *types.IgnoredUsers) error } diff --git a/syncapi/storage/postgres/ignores_table.go b/syncapi/storage/postgres/ignores_table.go new file mode 100644 index 000000000..c58e71997 --- /dev/null +++ b/syncapi/storage/postgres/ignores_table.go @@ -0,0 +1,87 @@ +// Copyright 2017 Jan Christian Grünhage +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package postgres + +import ( + "context" + "database/sql" + "encoding/json" + + "github.com/matrix-org/dendrite/syncapi/storage/tables" + "github.com/matrix-org/dendrite/syncapi/types" +) + +const ignoresSchema = ` +-- Stores data about ignoress +CREATE TABLE IF NOT EXISTS syncapi_ignores ( + -- The user ID whose ignore list this belongs to. + user_id TEXT NOT NULL, + ignores_json TEXT NOT NULL, + PRIMARY KEY(user_id) +); +` + +const selectIgnoresSQL = "" + + "SELECT ignores_json FROM syncapi_ignores WHERE user_id = $1" + +const upsertIgnoresSQL = "" + + "INSERT INTO syncapi_ignores (user_id, ignores_json) VALUES ($1, $2)" + + " ON CONFLICT (user_id) DO UPDATE set ignores_json = $2" + +type ignoresStatements struct { + selectIgnoresStmt *sql.Stmt + upsertIgnoresStmt *sql.Stmt +} + +func NewPostgresIgnoresTable(db *sql.DB) (tables.Ignores, error) { + _, err := db.Exec(ignoresSchema) + if err != nil { + return nil, err + } + s := &ignoresStatements{} + if s.selectIgnoresStmt, err = db.Prepare(selectIgnoresSQL); err != nil { + return nil, err + } + if s.upsertIgnoresStmt, err = db.Prepare(upsertIgnoresSQL); err != nil { + return nil, err + } + return s, nil +} + +func (s *ignoresStatements) SelectIgnores( + ctx context.Context, userID string, +) (*types.IgnoredUsers, error) { + var ignoresData []byte + err := s.selectIgnoresStmt.QueryRowContext(ctx, userID).Scan(&ignoresData) + if err != nil { + return nil, err + } + var ignores types.IgnoredUsers + if err = json.Unmarshal(ignoresData, &ignores); err != nil { + return nil, err + } + return &ignores, nil +} + +func (s *ignoresStatements) UpsertIgnores( + ctx context.Context, userID string, ignores *types.IgnoredUsers, +) error { + ignoresJSON, err := json.Marshal(ignores) + if err != nil { + return err + } + _, err = s.upsertIgnoresStmt.ExecContext(ctx, userID, ignoresJSON) + return err +} diff --git a/syncapi/storage/postgres/syncserver.go b/syncapi/storage/postgres/syncserver.go index 4e4b5c0bb..d549dac75 100644 --- a/syncapi/storage/postgres/syncserver.go +++ b/syncapi/storage/postgres/syncserver.go @@ -90,6 +90,10 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e if err != nil { return nil, err } + ignores, err := NewPostgresIgnoresTable(d.db) + if err != nil { + return nil, err + } m := sqlutil.NewMigrations() deltas.LoadFixSequences(m) deltas.LoadRemoveSendToDeviceSentColumn(m) @@ -111,6 +115,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e Receipts: receipts, Memberships: memberships, NotificationData: notificationData, + Ignores: ignores, } return &d, nil } diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index de43678d7..3f13fb005 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/syncapi/storage/shared/syncserver.go @@ -48,6 +48,7 @@ type Database struct { Receipts tables.Receipts Memberships tables.Memberships NotificationData tables.NotificationData + Ignores tables.Ignores } func (d *Database) readOnlySnapshot(ctx context.Context) (*sql.Tx, error) { @@ -1002,3 +1003,11 @@ func (s *Database) SelectContextBeforeEvent(ctx context.Context, id int, roomID func (s *Database) SelectContextAfterEvent(ctx context.Context, id int, roomID string, filter *gomatrixserverlib.RoomEventFilter) (int, []*gomatrixserverlib.HeaderedEvent, error) { return s.OutputEvents.SelectContextAfterEvent(ctx, nil, id, roomID, filter) } + +func (s *Database) IgnoresForUser(ctx context.Context, userID string) (*types.IgnoredUsers, error) { + return s.Ignores.SelectIgnores(ctx, userID) +} + +func (s *Database) UpdateIgnoresForUser(ctx context.Context, userID string, ignores *types.IgnoredUsers) error { + return s.Ignores.UpsertIgnores(ctx, userID, ignores) +} diff --git a/syncapi/storage/sqlite3/ignores_table.go b/syncapi/storage/sqlite3/ignores_table.go new file mode 100644 index 000000000..4eab46b71 --- /dev/null +++ b/syncapi/storage/sqlite3/ignores_table.go @@ -0,0 +1,87 @@ +// Copyright 2017 Jan Christian Grünhage +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sqlite3 + +import ( + "context" + "database/sql" + "encoding/json" + + "github.com/matrix-org/dendrite/syncapi/storage/tables" + "github.com/matrix-org/dendrite/syncapi/types" +) + +const ignoresSchema = ` +-- Stores data about ignoress +CREATE TABLE IF NOT EXISTS syncapi_ignores ( + -- The user ID whose ignore list this belongs to. + user_id TEXT NOT NULL, + ignores_json TEXT NOT NULL, + PRIMARY KEY(user_id) +); +` + +const selectIgnoresSQL = "" + + "SELECT ignores_json FROM syncapi_ignores WHERE user_id = $1" + +const upsertIgnoresSQL = "" + + "INSERT INTO syncapi_ignores (user_id, ignores_json) VALUES ($1, $2)" + + " ON CONFLICT DO UPDATE set ignores_json = $2" + +type ignoresStatements struct { + selectIgnoresStmt *sql.Stmt + upsertIgnoresStmt *sql.Stmt +} + +func NewSqliteIgnoresTable(db *sql.DB) (tables.Ignores, error) { + _, err := db.Exec(ignoresSchema) + if err != nil { + return nil, err + } + s := &ignoresStatements{} + if s.selectIgnoresStmt, err = db.Prepare(selectIgnoresSQL); err != nil { + return nil, err + } + if s.upsertIgnoresStmt, err = db.Prepare(upsertIgnoresSQL); err != nil { + return nil, err + } + return s, nil +} + +func (s *ignoresStatements) SelectIgnores( + ctx context.Context, userID string, +) (*types.IgnoredUsers, error) { + var ignoresData []byte + err := s.selectIgnoresStmt.QueryRowContext(ctx, userID).Scan(&ignoresData) + if err != nil { + return nil, err + } + var ignores types.IgnoredUsers + if err = json.Unmarshal(ignoresData, &ignores); err != nil { + return nil, err + } + return &ignores, nil +} + +func (s *ignoresStatements) UpsertIgnores( + ctx context.Context, userID string, ignores *types.IgnoredUsers, +) error { + ignoresJSON, err := json.Marshal(ignores) + if err != nil { + return err + } + _, err = s.upsertIgnoresStmt.ExecContext(ctx, userID, ignoresJSON) + return err +} diff --git a/syncapi/storage/sqlite3/syncserver.go b/syncapi/storage/sqlite3/syncserver.go index cb7e3b46f..1d0214c9f 100644 --- a/syncapi/storage/sqlite3/syncserver.go +++ b/syncapi/storage/sqlite3/syncserver.go @@ -100,6 +100,10 @@ func (d *SyncServerDatasource) prepare(dbProperties *config.DatabaseOptions) (er if err != nil { return err } + ignores, err := NewSqliteIgnoresTable(d.db) + if err != nil { + return err + } m := sqlutil.NewMigrations() deltas.LoadFixSequences(m) deltas.LoadRemoveSendToDeviceSentColumn(m) @@ -121,6 +125,7 @@ func (d *SyncServerDatasource) prepare(dbProperties *config.DatabaseOptions) (er Receipts: receipts, Memberships: memberships, NotificationData: notificationData, + Ignores: ignores, } return nil } diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index 2c29888d3..264dd160c 100644 --- a/syncapi/storage/tables/interface.go +++ b/syncapi/storage/tables/interface.go @@ -182,3 +182,8 @@ type NotificationData interface { SelectUserUnreadCounts(ctx context.Context, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) SelectMaxID(ctx context.Context) (int64, error) } + +type Ignores interface { + SelectIgnores(ctx context.Context, userID string) (*types.IgnoredUsers, error) + UpsertIgnores(ctx context.Context, userID string, ignores *types.IgnoredUsers) error +} diff --git a/syncapi/streams/stream_invite.go b/syncapi/streams/stream_invite.go index 147f33fc7..ddac9be2c 100644 --- a/syncapi/streams/stream_invite.go +++ b/syncapi/streams/stream_invite.go @@ -55,7 +55,7 @@ func (p *InviteStreamProvider) IncrementalSync( for roomID, inviteEvent := range invites { // skip ignored user events - if _, ok := req.IgnoredUsers[inviteEvent.Sender()]; ok { + if _, ok := req.IgnoredUsers.List[inviteEvent.Sender()]; ok { continue } ir := types.NewInviteResponse(inviteEvent) diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go index ae3e7b5fa..5fe01e819 100644 --- a/syncapi/streams/stream_pdu.go +++ b/syncapi/streams/stream_pdu.go @@ -2,7 +2,7 @@ package streams import ( "context" - "encoding/json" + "database/sql" "sync" "time" @@ -415,23 +415,16 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync( // addIgnoredUsersToFilter adds ignored users to the eventfilter and // the syncreq itself for further use in streams. func (p *PDUStreamProvider) addIgnoredUsersToFilter(ctx context.Context, req *types.SyncRequest, eventFilter *gomatrixserverlib.RoomEventFilter) error { - accountData := userapi.QueryAccountDataResponse{} - err := p.userAPI.QueryAccountData(ctx, &userapi.QueryAccountDataRequest{ - UserID: req.Device.UserID, RoomID: "", DataType: "m.ignored_user_list", - }, &accountData) + ignores, err := p.DB.SelectIgnores(ctx, req.Device.UserID) if err != nil { - req.Log.WithError(err).Error("unable to query ignored users") + if err == sql.ErrNoRows { + return nil + } return err } - if data, ok := accountData.GlobalAccountData["m.ignored_user_list"]; ok { - err = json.Unmarshal(data, &req) - if err != nil { - req.Log.WithError(err).Error("unable to parse json") - return err - } - for userID := range req.IgnoredUsers { - eventFilter.NotSenders = append(eventFilter.NotSenders, userID) - } + req.IgnoredUsers = *ignores + for userID := range ignores.List { + eventFilter.NotSenders = append(eventFilter.NotSenders, userID) } return nil } diff --git a/syncapi/streams/stream_receipt.go b/syncapi/streams/stream_receipt.go index d65459530..9d7d479a2 100644 --- a/syncapi/streams/stream_receipt.go +++ b/syncapi/streams/stream_receipt.go @@ -55,7 +55,7 @@ func (p *ReceiptStreamProvider) IncrementalSync( receiptsByRoom := make(map[string][]types.OutputReceiptEvent) for _, receipt := range receipts { // skip ignored user events - if _, ok := req.IgnoredUsers[receipt.UserID]; ok { + if _, ok := req.IgnoredUsers.List[receipt.UserID]; ok { continue } receiptsByRoom[receipt.RoomID] = append(receiptsByRoom[receipt.RoomID], receipt) diff --git a/syncapi/streams/stream_sendtodevice.go b/syncapi/streams/stream_sendtodevice.go index 1eddeb9af..6a18df506 100644 --- a/syncapi/streams/stream_sendtodevice.go +++ b/syncapi/streams/stream_sendtodevice.go @@ -49,7 +49,7 @@ func (p *SendToDeviceStreamProvider) IncrementalSync( // Add the updates into the sync response. for _, event := range events { // skip ignored user events - if _, ok := req.IgnoredUsers[event.Sender]; ok { + if _, ok := req.IgnoredUsers.List[event.Sender]; ok { continue } req.Response.ToDevice.Events = append(req.Response.ToDevice.Events, event.SendToDeviceEvent) diff --git a/syncapi/streams/stream_typing.go b/syncapi/streams/stream_typing.go index 276aad9ba..f781065be 100644 --- a/syncapi/streams/stream_typing.go +++ b/syncapi/streams/stream_typing.go @@ -43,7 +43,7 @@ func (p *TypingStreamProvider) IncrementalSync( typingUsers := make([]string, 0, len(users)) for i := range users { // skip ignored user events - if _, ok := req.IgnoredUsers[users[i]]; !ok { + if _, ok := req.IgnoredUsers.List[users[i]]; !ok { typingUsers = append(typingUsers, users[i]) } } diff --git a/syncapi/types/provider.go b/syncapi/types/provider.go index 4c30fce89..e6777f643 100644 --- a/syncapi/types/provider.go +++ b/syncapi/types/provider.go @@ -22,7 +22,7 @@ type SyncRequest struct { // Updated by the PDU stream. Rooms map[string]string // Updated by the PDU stream. - IgnoredUsers map[string]interface{} `json:"ignored_users"` + IgnoredUsers IgnoredUsers } type StreamProvider interface { diff --git a/syncapi/types/types.go b/syncapi/types/types.go index d0efa1bbb..be300ebda 100644 --- a/syncapi/types/types.go +++ b/syncapi/types/types.go @@ -510,3 +510,7 @@ type OutputSendToDeviceEvent struct { DeviceID string `json:"device_id"` gomatrixserverlib.SendToDeviceEvent } + +type IgnoredUsers struct { + List map[string]interface{} `json:"ignored_users"` +} diff --git a/userapi/consumers/syncapi_streamevent.go b/userapi/consumers/syncapi_streamevent.go index 1205af268..9ef7b5083 100644 --- a/userapi/consumers/syncapi_streamevent.go +++ b/userapi/consumers/syncapi_streamevent.go @@ -395,10 +395,6 @@ func (s *OutputStreamEventConsumer) notifyLocal(ctx context.Context, event *goma return nil } -type ignoredUsers struct { - List map[string]interface{} `json:"ignored_users"` -} - // evaluatePushRules fetches and evaluates the push rules of a local // user. Returns actions (including dont_notify). func (s *OutputStreamEventConsumer) evaluatePushRules(ctx context.Context, event *gomatrixserverlib.HeaderedEvent, mem *localMembership, roomSize int) ([]*pushrules.Action, error) { @@ -414,7 +410,7 @@ func (s *OutputStreamEventConsumer) evaluatePushRules(ctx context.Context, event return nil, err } if data != nil { - ignored := ignoredUsers{} + ignored := types.IgnoredUsers{} err = json.Unmarshal(data, &ignored) if err != nil { return nil, err