aboutsummaryrefslogtreecommitdiff
path: root/syncapi
diff options
context:
space:
mode:
Diffstat (limited to 'syncapi')
-rw-r--r--syncapi/consumers/clientapi.go9
-rw-r--r--syncapi/storage/interface.go3
-rw-r--r--syncapi/storage/postgres/ignores_table.go87
-rw-r--r--syncapi/storage/postgres/syncserver.go5
-rw-r--r--syncapi/storage/shared/syncserver.go9
-rw-r--r--syncapi/storage/sqlite3/ignores_table.go87
-rw-r--r--syncapi/storage/sqlite3/syncserver.go5
-rw-r--r--syncapi/storage/tables/interface.go5
-rw-r--r--syncapi/streams/stream_invite.go4
-rw-r--r--syncapi/streams/stream_pdu.go27
-rw-r--r--syncapi/streams/stream_receipt.go4
-rw-r--r--syncapi/streams/stream_sendtodevice.go4
-rw-r--r--syncapi/streams/stream_typing.go9
-rw-r--r--syncapi/streams/streams.go1
-rw-r--r--syncapi/types/provider.go2
-rw-r--r--syncapi/types/types.go4
16 files changed, 264 insertions, 1 deletions
diff --git a/syncapi/consumers/clientapi.go b/syncapi/consumers/clientapi.go
index c28da460..eec369c1 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 0b3ab235..841f6726 100644
--- a/syncapi/storage/interface.go
+++ b/syncapi/storage/interface.go
@@ -150,6 +150,9 @@ 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
}
type Presence interface {
diff --git a/syncapi/storage/postgres/ignores_table.go b/syncapi/storage/postgres/ignores_table.go
new file mode 100644
index 00000000..055a1a23
--- /dev/null
+++ b/syncapi/storage/postgres/ignores_table.go
@@ -0,0 +1,87 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// 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 54445c7e..b0382512 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
+ }
presence, err := NewPostgresPresenceTable(d.db)
if err != nil {
return nil, err
@@ -115,6 +119,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e
Receipts: receipts,
Memberships: memberships,
NotificationData: notificationData,
+ Ignores: ignores,
Presence: presence,
}
return &d, nil
diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go
index 7c4786fc..1c45d5d9 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
Presence tables.Presence
}
@@ -1004,6 +1005,14 @@ func (s *Database) SelectContextAfterEvent(ctx context.Context, id int, roomID s
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)
+}
+
func (s *Database) UpdatePresence(ctx context.Context, userID string, presence types.Presence, statusMsg *string, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (types.StreamPosition, error) {
return s.Presence.UpsertPresence(ctx, nil, userID, statusMsg, presence, lastActiveTS, fromSync)
}
diff --git a/syncapi/storage/sqlite3/ignores_table.go b/syncapi/storage/sqlite3/ignores_table.go
new file mode 100644
index 00000000..f4afca55
--- /dev/null
+++ b/syncapi/storage/sqlite3/ignores_table.go
@@ -0,0 +1,87 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// 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 cb2c3169..9d9d3598 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
+ }
presence, err := NewSqlitePresenceTable(d.db, &d.streamID)
if err != nil {
return err
@@ -125,6 +129,7 @@ func (d *SyncServerDatasource) prepare(dbProperties *config.DatabaseOptions) (er
Receipts: receipts,
Memberships: memberships,
NotificationData: notificationData,
+ Ignores: ignores,
Presence: presence,
}
return nil
diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go
index ef0587bb..8d368eec 100644
--- a/syncapi/storage/tables/interface.go
+++ b/syncapi/storage/tables/interface.go
@@ -183,6 +183,11 @@ type NotificationData interface {
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
+}
+
type Presence interface {
UpsertPresence(ctx context.Context, txn *sql.Tx, userID string, statusMsg *string, presence types.Presence, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (pos types.StreamPosition, err error)
GetPresenceForUser(ctx context.Context, txn *sql.Tx, userID string) (presence *types.PresenceInternal, err error)
diff --git a/syncapi/streams/stream_invite.go b/syncapi/streams/stream_invite.go
index 70374c6a..ddac9be2 100644
--- a/syncapi/streams/stream_invite.go
+++ b/syncapi/streams/stream_invite.go
@@ -54,6 +54,10 @@ func (p *InviteStreamProvider) IncrementalSync(
}
for roomID, inviteEvent := range invites {
+ // skip ignored user events
+ if _, ok := req.IgnoredUsers.List[inviteEvent.Sender()]; ok {
+ continue
+ }
ir := types.NewInviteResponse(inviteEvent)
req.Response.Rooms.Invite[roomID] = *ir
}
diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go
index d23209af..ab200e00 100644
--- a/syncapi/streams/stream_pdu.go
+++ b/syncapi/streams/stream_pdu.go
@@ -2,6 +2,7 @@ package streams
import (
"context"
+ "database/sql"
"sync"
"time"
@@ -25,6 +26,7 @@ type PDUStreamProvider struct {
tasks chan func()
workers atomic.Int32
+ userAPI userapi.UserInternalAPI
}
func (p *PDUStreamProvider) worker() {
@@ -87,6 +89,10 @@ func (p *PDUStreamProvider) CompleteSync(
stateFilter := req.Filter.Room.State
eventFilter := req.Filter.Room.Timeline
+ if err = p.addIgnoredUsersToFilter(ctx, req, &eventFilter); err != nil {
+ req.Log.WithError(err).Error("unable to update event filter with ignored users")
+ }
+
// Build up a /sync response. Add joined rooms.
var reqMutex sync.Mutex
var reqWaitGroup sync.WaitGroup
@@ -175,6 +181,10 @@ func (p *PDUStreamProvider) IncrementalSync(
return to
}
+ if err = p.addIgnoredUsersToFilter(ctx, req, &eventFilter); err != nil {
+ req.Log.WithError(err).Error("unable to update event filter with ignored users")
+ }
+
newPos = from
for _, delta := range stateDeltas {
var pos types.StreamPosition
@@ -402,6 +412,23 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
return jr, nil
}
+// 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 {
+ ignores, err := p.DB.IgnoresForUser(ctx, req.Device.UserID)
+ if err != nil {
+ if err == sql.ErrNoRows {
+ return nil
+ }
+ return err
+ }
+ req.IgnoredUsers = *ignores
+ for userID := range ignores.List {
+ eventFilter.NotSenders = append(eventFilter.NotSenders, userID)
+ }
+ return nil
+}
+
func removeDuplicates(stateEvents, recentEvents []*gomatrixserverlib.HeaderedEvent) []*gomatrixserverlib.HeaderedEvent {
for _, recentEv := range recentEvents {
if recentEv.StateKey() == nil {
diff --git a/syncapi/streams/stream_receipt.go b/syncapi/streams/stream_receipt.go
index 680f8cd8..9d7d479a 100644
--- a/syncapi/streams/stream_receipt.go
+++ b/syncapi/streams/stream_receipt.go
@@ -54,6 +54,10 @@ func (p *ReceiptStreamProvider) IncrementalSync(
// Group receipts by room, so we can create one ClientEvent for every room
receiptsByRoom := make(map[string][]types.OutputReceiptEvent)
for _, receipt := range receipts {
+ // skip ignored user events
+ 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 a3aaf3d7..6a18df50 100644
--- a/syncapi/streams/stream_sendtodevice.go
+++ b/syncapi/streams/stream_sendtodevice.go
@@ -48,6 +48,10 @@ func (p *SendToDeviceStreamProvider) IncrementalSync(
// Add the updates into the sync response.
for _, event := range events {
+ // skip ignored user events
+ 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 e46cd447..f781065b 100644
--- a/syncapi/streams/stream_typing.go
+++ b/syncapi/streams/stream_typing.go
@@ -40,11 +40,18 @@ func (p *TypingStreamProvider) IncrementalSync(
if users, updated := p.EDUCache.GetTypingUsersIfUpdatedAfter(
roomID, int64(from),
); updated {
+ typingUsers := make([]string, 0, len(users))
+ for i := range users {
+ // skip ignored user events
+ if _, ok := req.IgnoredUsers.List[users[i]]; !ok {
+ typingUsers = append(typingUsers, users[i])
+ }
+ }
ev := gomatrixserverlib.ClientEvent{
Type: gomatrixserverlib.MTyping,
}
ev.Content, err = json.Marshal(map[string]interface{}{
- "user_ids": users,
+ "user_ids": typingUsers,
})
if err != nil {
req.Log.WithError(err).Error("json.Marshal failed")
diff --git a/syncapi/streams/streams.go b/syncapi/streams/streams.go
index 07322388..c7d06a29 100644
--- a/syncapi/streams/streams.go
+++ b/syncapi/streams/streams.go
@@ -32,6 +32,7 @@ func NewSyncStreamProviders(
streams := &Streams{
PDUStreamProvider: &PDUStreamProvider{
StreamProvider: StreamProvider{DB: d},
+ userAPI: userAPI,
},
TypingStreamProvider: &TypingStreamProvider{
StreamProvider: StreamProvider{DB: d},
diff --git a/syncapi/types/provider.go b/syncapi/types/provider.go
index f6185fcb..e6777f64 100644
--- a/syncapi/types/provider.go
+++ b/syncapi/types/provider.go
@@ -21,6 +21,8 @@ type SyncRequest struct {
// Updated by the PDU stream.
Rooms map[string]string
+ // Updated by the PDU stream.
+ IgnoredUsers IgnoredUsers
}
type StreamProvider interface {
diff --git a/syncapi/types/types.go b/syncapi/types/types.go
index d21203b5..ba6b4f8c 100644
--- a/syncapi/types/types.go
+++ b/syncapi/types/types.go
@@ -518,3 +518,7 @@ type OutputSendToDeviceEvent struct {
DeviceID string `json:"device_id"`
gomatrixserverlib.SendToDeviceEvent
}
+
+type IgnoredUsers struct {
+ List map[string]interface{} `json:"ignored_users"`
+}
generated by cgit v1.2.3 (git 2.26.2) at 2024-12-25 11:47:28 +0000