Send-to-device support (#1072)
* Groundwork for send-to-device messaging
* Update sample config
* Add unstable routing for now
* Send to device consumer in sync API
* Start the send-to-device consumer
* fix indentation in dendrite-config.yaml
* Create send-to-device database tables, other tweaks
* Add some logic for send-to-device messages, add them into sync stream
* Handle incoming send-to-device messages, count them with EDU stream pos
* Undo changes to test
* pq.Array
* Fix sync
* Logging
* Fix a couple of transaction things, fix client API
* Add send-to-device test, hopefully fix bugs
* Comments
* Refactor a bit
* Fix schema
* Fix queries
* Debug logging
* Fix storing and retrieving of send-to-device messages
* Try to avoid database locks
* Update sync position
* Use latest sync position
* Jiggle about sync a bit
* Fix tests
* Break out the retrieval from the update/delete behaviour
* Comments
* nolint on getResponseWithPDUsForCompleteSync
* Try to line up sync tokens again
* Implement wildcard
* Add all send-to-device tests to whitelist, what could possibly go wrong?
* Only care about wildcard when targeted locally
* Deduplicate transactions
* Handle tokens properly, return immediately if waiting send-to-device messages
* Fix sync
* Update sytest-whitelist
* Fix copyright notice (need to do more of this)
* Comments, copyrights
* Return errors from Do, fix dendritejs
* Review comments
* Comments
* Constructor for TransactionWriter
* defletions
* Update gomatrixserverlib, sytest-blacklist
2020-06-01 17:50:19 +01:00
// Copyright 2020 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.
2020-05-13 17:28:42 +01:00
package tables
import (
"context"
"database/sql"
2022-03-03 11:40:53 +00:00
"github.com/matrix-org/dendrite/internal/eventutil"
2020-05-14 09:53:55 +01:00
"github.com/matrix-org/dendrite/roomserver/api"
2020-05-13 17:28:42 +01:00
"github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/gomatrixserverlib"
)
2020-05-14 09:53:55 +01:00
type AccountData interface {
InsertAccountData ( ctx context . Context , txn * sql . Tx , userID , roomID , dataType string ) ( pos types . StreamPosition , err error )
2020-05-15 09:41:12 +01:00
// SelectAccountDataInRange returns a map of room ID to a list of `dataType`.
2022-04-26 14:50:56 +01:00
SelectAccountDataInRange ( ctx context . Context , userID string , r types . Range , accountDataEventFilter * gomatrixserverlib . EventFilter ) ( data map [ string ] [ ] string , pos types . StreamPosition , err error )
2020-05-14 09:53:55 +01:00
SelectMaxAccountDataID ( ctx context . Context , txn * sql . Tx ) ( id int64 , err error )
}
2020-05-13 17:28:42 +01:00
type Invites interface {
2020-11-16 15:44:53 +00:00
InsertInviteEvent ( ctx context . Context , txn * sql . Tx , inviteEvent * gomatrixserverlib . HeaderedEvent ) ( streamPos types . StreamPosition , err error )
2020-09-01 11:28:35 +01:00
DeleteInviteEvent ( ctx context . Context , txn * sql . Tx , inviteEventID string ) ( types . StreamPosition , error )
2020-08-25 18:59:00 +01:00
// SelectInviteEventsInRange returns a map of room ID to invite events. If multiple invite/retired invites exist in the given range, return the latest value
// for the room.
2020-11-16 15:44:53 +00:00
SelectInviteEventsInRange ( ctx context . Context , txn * sql . Tx , targetUserID string , r types . Range ) ( invites map [ string ] * gomatrixserverlib . HeaderedEvent , retired map [ string ] * gomatrixserverlib . HeaderedEvent , err error )
2020-05-13 17:28:42 +01:00
SelectMaxInviteID ( ctx context . Context , txn * sql . Tx ) ( id int64 , err error )
}
2020-05-14 09:53:55 +01:00
2020-09-10 14:39:18 +01:00
type Peeks interface {
InsertPeek ( ctx context . Context , txn * sql . Tx , roomID , userID , deviceID string ) ( streamPos types . StreamPosition , err error )
DeletePeek ( ctx context . Context , txn * sql . Tx , roomID , userID , deviceID string ) ( streamPos types . StreamPosition , err error )
DeletePeeks ( ctx context . Context , txn * sql . Tx , roomID , userID string ) ( streamPos types . StreamPosition , err error )
SelectPeeksInRange ( ctxt context . Context , txn * sql . Tx , userID , deviceID string , r types . Range ) ( peeks [ ] types . Peek , err error )
SelectPeekingDevices ( ctxt context . Context ) ( peekingDevices map [ string ] [ ] types . PeekingDevice , err error )
SelectMaxPeekID ( ctx context . Context , txn * sql . Tx ) ( id int64 , err error )
}
2020-05-14 09:53:55 +01:00
type Events interface {
2022-03-11 12:48:45 +00:00
SelectStateInRange ( ctx context . Context , txn * sql . Tx , r types . Range , stateFilter * gomatrixserverlib . StateFilter , roomIDs [ ] string ) ( map [ string ] map [ string ] bool , map [ string ] types . StreamEvent , error )
2020-05-14 09:53:55 +01:00
SelectMaxEventID ( ctx context . Context , txn * sql . Tx ) ( id int64 , err error )
2022-07-18 13:46:15 +01:00
InsertEvent (
ctx context . Context , txn * sql . Tx ,
event * gomatrixserverlib . HeaderedEvent ,
addState , removeState [ ] string ,
transactionID * api . TransactionID ,
excludeFromSync bool ,
historyVisibility gomatrixserverlib . HistoryVisibility ,
) ( streamPos types . StreamPosition , err error )
2020-05-15 09:41:12 +01:00
// SelectRecentEvents returns events between the two stream positions: exclusive of low and inclusive of high.
2020-05-14 17:30:16 +01:00
// If onlySyncEvents has a value of true, only returns the events that aren't marked as to exclude from sync.
2020-06-26 15:34:41 +01:00
// Returns up to `limit` events. Returns `limited=true` if there are more events in this range but we hit the `limit`.
2021-01-19 18:00:42 +00:00
SelectRecentEvents ( ctx context . Context , txn * sql . Tx , roomID string , r types . Range , eventFilter * gomatrixserverlib . RoomEventFilter , chronologicalOrder bool , onlySyncEvents bool ) ( [ ] types . StreamEvent , bool , error )
2020-05-15 09:41:12 +01:00
// SelectEarlyEvents returns the earliest events in the given room.
2021-01-19 18:00:42 +00:00
SelectEarlyEvents ( ctx context . Context , txn * sql . Tx , roomID string , r types . Range , eventFilter * gomatrixserverlib . RoomEventFilter ) ( [ ] types . StreamEvent , error )
2022-04-13 12:16:02 +01:00
SelectEvents ( ctx context . Context , txn * sql . Tx , eventIDs [ ] string , filter * gomatrixserverlib . RoomEventFilter , preserveOrder bool ) ( [ ] types . StreamEvent , error )
2020-07-08 17:45:39 +01:00
UpdateEventJSON ( ctx context . Context , event * gomatrixserverlib . HeaderedEvent ) error
2020-09-15 11:17:46 +01:00
// DeleteEventsForRoom removes all event information for a room. This should only be done when removing the room entirely.
DeleteEventsForRoom ( ctx context . Context , txn * sql . Tx , roomID string ) ( err error )
2022-02-21 16:12:22 +00:00
SelectContextEvent ( ctx context . Context , txn * sql . Tx , roomID , eventID string ) ( int , gomatrixserverlib . HeaderedEvent , error )
SelectContextBeforeEvent ( ctx context . Context , txn * sql . Tx , id int , roomID string , filter * gomatrixserverlib . RoomEventFilter ) ( [ ] * gomatrixserverlib . HeaderedEvent , error )
SelectContextAfterEvent ( ctx context . Context , txn * sql . Tx , id int , roomID string , filter * gomatrixserverlib . RoomEventFilter ) ( int , [ ] * gomatrixserverlib . HeaderedEvent , error )
2020-05-14 09:53:55 +01:00
}
2020-05-14 16:11:37 +01:00
2020-05-14 17:30:16 +01:00
// Topology keeps track of the depths and stream positions for all events.
// These positions are used as types.TopologyToken when backfilling events locally.
2020-05-14 16:11:37 +01:00
type Topology interface {
2020-05-14 17:30:16 +01:00
// InsertEventInTopology inserts the given event in the room's topology, based on the event's depth.
// `pos` is the stream position of this event in the events table, and is used to order events which have the same depth.
2021-01-20 20:43:20 +00:00
InsertEventInTopology ( ctx context . Context , txn * sql . Tx , event * gomatrixserverlib . HeaderedEvent , pos types . StreamPosition ) ( topoPos types . StreamPosition , err error )
2020-05-14 17:30:16 +01:00
// SelectEventIDsInRange selects the IDs of events whose depths are within a given range in a given room's topological order.
// Events with `minDepth` are *exclusive*, as is the event which has exactly `minDepth`,`maxStreamPos`.
// `maxStreamPos` is only used when events have the same depth as `maxDepth`, which results in events less than `maxStreamPos` being returned.
2020-05-14 16:11:37 +01:00
// Returns an empty slice if no events match the given range.
2020-05-14 17:30:16 +01:00
SelectEventIDsInRange ( ctx context . Context , txn * sql . Tx , roomID string , minDepth , maxDepth , maxStreamPos types . StreamPosition , limit int , chronologicalOrder bool ) ( eventIDs [ ] string , err error )
// SelectPositionInTopology returns the depth and stream position of a given event in the topology of the room it belongs to.
SelectPositionInTopology ( ctx context . Context , txn * sql . Tx , eventID string ) ( depth , spos types . StreamPosition , err error )
// SelectMaxPositionInTopology returns the event which has the highest depth, and if there are multiple, the event with the highest stream position.
SelectMaxPositionInTopology ( ctx context . Context , txn * sql . Tx , roomID string ) ( depth types . StreamPosition , spos types . StreamPosition , err error )
2022-03-18 10:40:01 +00:00
// SelectStreamToTopologicalPosition converts a stream position to a topological position by finding the nearest topological position in the room.
SelectStreamToTopologicalPosition ( ctx context . Context , txn * sql . Tx , roomID string , streamPos types . StreamPosition , forward bool ) ( topoPos types . StreamPosition , err error )
2020-05-14 16:11:37 +01:00
}
type CurrentRoomState interface {
SelectStateEvent ( ctx context . Context , roomID , evType , stateKey string ) ( * gomatrixserverlib . HeaderedEvent , error )
SelectEventsWithEventIDs ( ctx context . Context , txn * sql . Tx , eventIDs [ ] string ) ( [ ] types . StreamEvent , error )
2020-11-16 15:44:53 +00:00
UpsertRoomState ( ctx context . Context , txn * sql . Tx , event * gomatrixserverlib . HeaderedEvent , membership * string , addedAt types . StreamPosition ) error
2020-05-14 16:11:37 +01:00
DeleteRoomStateByEventID ( ctx context . Context , txn * sql . Tx , eventID string ) error
2020-09-15 11:17:46 +01:00
DeleteRoomStateForRoom ( ctx context . Context , txn * sql . Tx , roomID string ) error
2020-05-14 16:11:37 +01:00
// SelectCurrentState returns all the current state events for the given room.
2021-02-04 12:20:37 +00:00
SelectCurrentState ( ctx context . Context , txn * sql . Tx , roomID string , stateFilter * gomatrixserverlib . StateFilter , excludeEventIDs [ ] string ) ( [ ] * gomatrixserverlib . HeaderedEvent , error )
2020-05-14 16:11:37 +01:00
// SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state.
SelectRoomIDsWithMembership ( ctx context . Context , txn * sql . Tx , userID string , membership string ) ( [ ] string , error )
2022-03-11 12:48:45 +00:00
// SelectRoomIDsWithAnyMembership returns a map of all memberships for the given user.
SelectRoomIDsWithAnyMembership ( ctx context . Context , txn * sql . Tx , userID string ) ( map [ string ] string , error )
2020-05-14 16:11:37 +01:00
// SelectJoinedUsers returns a map of room ID to a list of joined user IDs.
SelectJoinedUsers ( ctx context . Context ) ( map [ string ] [ ] string , error )
2022-04-28 17:53:28 +01:00
// SelectJoinedUsersInRoom returns a map of room ID to a list of joined user IDs for a given room.
SelectJoinedUsersInRoom ( ctx context . Context , roomIDs [ ] string ) ( map [ string ] [ ] string , error )
2022-07-15 16:25:26 +01:00
// SelectSharedUsers returns a subset of otherUserIDs that share a room with userID.
SelectSharedUsers ( ctx context . Context , txn * sql . Tx , userID string , otherUserIDs [ ] string ) ( [ ] string , error )
2020-05-14 16:11:37 +01:00
}
// BackwardsExtremities keeps track of backwards extremities for a room.
// Backwards extremities are the earliest (DAG-wise) known events which we have
// the entire event JSON. These event IDs are used in federation requests to fetch
// even earlier events.
//
// We persist the previous event IDs as well, one per row, so when we do fetch even
// earlier events we can simply delete rows which referenced it. Consider the graph:
// A
// | Event C has 1 prev_event ID: A.
// B C
// |___| Event D has 2 prev_event IDs: B and C.
// |
// D
// The earliest known event we have is D, so this table has 2 rows.
// A backfill request gives us C but not B. We delete rows where prev_event=C. This
// still means that D is a backwards extremity as we do not have event B. However, event
// C is *also* a backwards extremity at this point as we do not have event A. Later,
// when we fetch event B, we delete rows where prev_event=B. This then removes D as
// a backwards extremity because there are no more rows with event_id=B.
type BackwardsExtremities interface {
// InsertsBackwardExtremity inserts a new backwards extremity.
InsertsBackwardExtremity ( ctx context . Context , txn * sql . Tx , roomID , eventID string , prevEventID string ) ( err error )
2020-05-20 16:04:31 +01:00
// SelectBackwardExtremitiesForRoom retrieves all backwards extremities for the room, as a map of event_id to list of prev_event_ids.
SelectBackwardExtremitiesForRoom ( ctx context . Context , roomID string ) ( bwExtrems map [ string ] [ ] string , err error )
2020-05-14 16:11:37 +01:00
// DeleteBackwardExtremity removes a backwards extremity for a room, if one existed.
DeleteBackwardExtremity ( ctx context . Context , txn * sql . Tx , roomID , knownEventID string ) ( err error )
}
Send-to-device support (#1072)
* Groundwork for send-to-device messaging
* Update sample config
* Add unstable routing for now
* Send to device consumer in sync API
* Start the send-to-device consumer
* fix indentation in dendrite-config.yaml
* Create send-to-device database tables, other tweaks
* Add some logic for send-to-device messages, add them into sync stream
* Handle incoming send-to-device messages, count them with EDU stream pos
* Undo changes to test
* pq.Array
* Fix sync
* Logging
* Fix a couple of transaction things, fix client API
* Add send-to-device test, hopefully fix bugs
* Comments
* Refactor a bit
* Fix schema
* Fix queries
* Debug logging
* Fix storing and retrieving of send-to-device messages
* Try to avoid database locks
* Update sync position
* Use latest sync position
* Jiggle about sync a bit
* Fix tests
* Break out the retrieval from the update/delete behaviour
* Comments
* nolint on getResponseWithPDUsForCompleteSync
* Try to line up sync tokens again
* Implement wildcard
* Add all send-to-device tests to whitelist, what could possibly go wrong?
* Only care about wildcard when targeted locally
* Deduplicate transactions
* Handle tokens properly, return immediately if waiting send-to-device messages
* Fix sync
* Update sytest-whitelist
* Fix copyright notice (need to do more of this)
* Comments, copyrights
* Return errors from Do, fix dendritejs
* Review comments
* Comments
* Constructor for TransactionWriter
* defletions
* Update gomatrixserverlib, sytest-blacklist
2020-06-01 17:50:19 +01:00
// SendToDevice tracks send-to-device messages which are sent to individual
// clients. Each message gets inserted into this table at the point that we
// receive it from the EDU server.
//
// We're supposed to try and do our best to deliver send-to-device messages
// once, but the only way that we can really guarantee that they have been
// delivered is if the client successfully requests the next sync as given
// in the next_batch. Each time the device syncs, we will request all of the
// updates that either haven't been sent yet, along with all updates that we
// *have* sent but we haven't confirmed to have been received yet. If it's the
// first time we're sending a given update then we update the table to say
// what the "since" parameter was when we tried to send it.
//
// When the client syncs again, if their "since" parameter is *later* than
// the recorded one, we drop the entry from the DB as it's "sent". If the
// sync parameter isn't later then we will keep including the updates in the
// sync response, as the client is seemingly trying to repeat the same /sync.
type SendToDevice interface {
2020-12-18 11:11:21 +00:00
InsertSendToDeviceMessage ( ctx context . Context , txn * sql . Tx , userID , deviceID , content string ) ( pos types . StreamPosition , err error )
2021-01-13 17:29:46 +00:00
SelectSendToDeviceMessages ( ctx context . Context , txn * sql . Tx , userID , deviceID string , from , to types . StreamPosition ) ( lastPos types . StreamPosition , events [ ] types . SendToDeviceEvent , 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 )
Send-to-device support (#1072)
* Groundwork for send-to-device messaging
* Update sample config
* Add unstable routing for now
* Send to device consumer in sync API
* Start the send-to-device consumer
* fix indentation in dendrite-config.yaml
* Create send-to-device database tables, other tweaks
* Add some logic for send-to-device messages, add them into sync stream
* Handle incoming send-to-device messages, count them with EDU stream pos
* Undo changes to test
* pq.Array
* Fix sync
* Logging
* Fix a couple of transaction things, fix client API
* Add send-to-device test, hopefully fix bugs
* Comments
* Refactor a bit
* Fix schema
* Fix queries
* Debug logging
* Fix storing and retrieving of send-to-device messages
* Try to avoid database locks
* Update sync position
* Use latest sync position
* Jiggle about sync a bit
* Fix tests
* Break out the retrieval from the update/delete behaviour
* Comments
* nolint on getResponseWithPDUsForCompleteSync
* Try to line up sync tokens again
* Implement wildcard
* Add all send-to-device tests to whitelist, what could possibly go wrong?
* Only care about wildcard when targeted locally
* Deduplicate transactions
* Handle tokens properly, return immediately if waiting send-to-device messages
* Fix sync
* Update sytest-whitelist
* Fix copyright notice (need to do more of this)
* Comments, copyrights
* Return errors from Do, fix dendritejs
* Review comments
* Comments
* Constructor for TransactionWriter
* defletions
* Update gomatrixserverlib, sytest-blacklist
2020-06-01 17:50:19 +01:00
}
2020-06-26 15:34:41 +01:00
type Filter interface {
2022-04-26 16:58:20 +01:00
SelectFilter ( ctx context . Context , target * gomatrixserverlib . Filter , localpart string , filterID string ) error
2020-06-26 15:34:41 +01:00
InsertFilter ( ctx context . Context , filter * gomatrixserverlib . Filter , localpart string ) ( filterID string , err error )
}
2020-11-09 18:46:11 +00:00
type Receipts interface {
UpsertReceipt ( ctx context . Context , txn * sql . Tx , roomId , receiptType , userId , eventId string , timestamp gomatrixserverlib . Timestamp ) ( pos types . StreamPosition , err error )
2022-03-29 13:14:35 +01:00
SelectRoomReceiptsAfter ( ctx context . Context , roomIDs [ ] string , streamPos types . StreamPosition ) ( types . StreamPosition , [ ] types . OutputReceiptEvent , error )
2020-12-11 14:02:17 +00:00
SelectMaxReceiptID ( ctx context . Context , txn * sql . Tx ) ( id int64 , err error )
2020-11-09 18:46:11 +00:00
}
2021-01-20 20:43:20 +00:00
type Memberships interface {
UpsertMembership ( ctx context . Context , txn * sql . Tx , event * gomatrixserverlib . HeaderedEvent , streamPos , topologicalPos types . StreamPosition ) error
2022-04-01 16:14:38 +01:00
SelectMembershipCount ( ctx context . Context , txn * sql . Tx , roomID , membership string , pos types . StreamPosition ) ( count int , err error )
2022-04-26 09:53:17 +01:00
SelectHeroes ( ctx context . Context , txn * sql . Tx , roomID , userID string , memberships [ ] string ) ( heroes [ ] string , err error )
2021-01-20 20:43:20 +00:00
}
2022-03-03 11:40:53 +00:00
type NotificationData interface {
UpsertRoomUnreadCounts ( ctx context . Context , userID , roomID string , notificationCount , highlightCount int ) ( types . StreamPosition , error )
SelectUserUnreadCounts ( ctx context . Context , userID string , fromExcl , toIncl types . StreamPosition ) ( map [ string ] * eventutil . NotificationData , error )
SelectMaxID ( ctx context . Context ) ( int64 , error )
}
2022-04-06 12:11:19 +01:00
2022-04-07 15:08:19 +01:00
type Ignores interface {
SelectIgnores ( ctx context . Context , userID string ) ( * types . IgnoredUsers , error )
UpsertIgnores ( ctx context . Context , userID string , ignores * types . IgnoredUsers ) error
}
2022-04-06 12:11:19 +01:00
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 )
GetMaxPresenceID ( ctx context . Context , txn * sql . Tx ) ( pos types . StreamPosition , err error )
2022-04-28 15:12:40 +01:00
GetPresenceAfter ( ctx context . Context , txn * sql . Tx , after types . StreamPosition , filter gomatrixserverlib . EventFilter ) ( presences map [ string ] * types . PresenceInternal , err error )
2022-04-06 12:11:19 +01:00
}