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 16:50:19 +00: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 16:28:42 +00:00
|
|
|
package shared
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"database/sql"
|
2020-05-14 15:11:37 +00:00
|
|
|
"encoding/json"
|
|
|
|
"fmt"
|
2020-05-13 16:28:42 +00:00
|
|
|
|
2020-11-09 18:46:11 +00:00
|
|
|
eduAPI "github.com/matrix-org/dendrite/eduserver/api"
|
2020-06-16 13:10:55 +00:00
|
|
|
userapi "github.com/matrix-org/dendrite/userapi/api"
|
|
|
|
|
2020-07-08 16:45:39 +00:00
|
|
|
"github.com/matrix-org/dendrite/internal/eventutil"
|
2020-06-12 13:55:57 +00:00
|
|
|
"github.com/matrix-org/dendrite/internal/sqlutil"
|
2020-05-14 15:11:37 +00:00
|
|
|
"github.com/matrix-org/dendrite/roomserver/api"
|
2020-05-13 16:28:42 +00:00
|
|
|
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
|
|
|
"github.com/matrix-org/dendrite/syncapi/types"
|
|
|
|
"github.com/matrix-org/gomatrixserverlib"
|
2021-01-13 17:29:46 +00:00
|
|
|
"github.com/sirupsen/logrus"
|
2020-09-10 13:39:18 +00:00
|
|
|
log "github.com/sirupsen/logrus"
|
2020-05-13 16:28:42 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
// Database is a temporary struct until we have made syncserver.go the same for both pq/sqlite
|
|
|
|
// For now this contains the shared functions
|
|
|
|
type Database struct {
|
2020-05-14 15:11:37 +00:00
|
|
|
DB *sql.DB
|
2020-08-21 09:42:08 +00:00
|
|
|
Writer sqlutil.Writer
|
2020-05-14 15:11:37 +00:00
|
|
|
Invites tables.Invites
|
2020-09-10 13:39:18 +00:00
|
|
|
Peeks tables.Peeks
|
2020-05-14 15:11:37 +00:00
|
|
|
AccountData tables.AccountData
|
|
|
|
OutputEvents tables.Events
|
|
|
|
Topology tables.Topology
|
|
|
|
CurrentRoomState tables.CurrentRoomState
|
|
|
|
BackwardExtremities tables.BackwardsExtremities
|
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 16:50:19 +00:00
|
|
|
SendToDevice tables.SendToDevice
|
2020-06-26 14:34:41 +00:00
|
|
|
Filter tables.Filter
|
2020-11-09 18:46:11 +00:00
|
|
|
Receipts tables.Receipts
|
2021-01-20 20:43:20 +00:00
|
|
|
Memberships tables.Memberships
|
2021-01-08 16:59:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) readOnlySnapshot(ctx context.Context) (*sql.Tx, error) {
|
|
|
|
return d.DB.BeginTx(ctx, &sql.TxOptions{
|
|
|
|
// Set the isolation level so that we see a snapshot of the database.
|
|
|
|
// In PostgreSQL repeatable read transactions will see a snapshot taken
|
|
|
|
// at the first query, and since the transaction is read-only it can't
|
|
|
|
// run into any serialisation errors.
|
|
|
|
// https://www.postgresql.org/docs/9.5/static/transaction-iso.html#XACT-REPEATABLE-READ
|
|
|
|
Isolation: sql.LevelRepeatableRead,
|
|
|
|
ReadOnly: true,
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) MaxStreamPositionForPDUs(ctx context.Context) (types.StreamPosition, error) {
|
|
|
|
id, err := d.OutputEvents.SelectMaxEventID(ctx, nil)
|
|
|
|
if err != nil {
|
|
|
|
return 0, fmt.Errorf("d.OutputEvents.SelectMaxEventID: %w", err)
|
|
|
|
}
|
|
|
|
return types.StreamPosition(id), nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) MaxStreamPositionForReceipts(ctx context.Context) (types.StreamPosition, error) {
|
|
|
|
id, err := d.Receipts.SelectMaxReceiptID(ctx, nil)
|
|
|
|
if err != nil {
|
|
|
|
return 0, fmt.Errorf("d.Receipts.SelectMaxReceiptID: %w", err)
|
|
|
|
}
|
|
|
|
return types.StreamPosition(id), nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) MaxStreamPositionForInvites(ctx context.Context) (types.StreamPosition, error) {
|
|
|
|
id, err := d.Invites.SelectMaxInviteID(ctx, nil)
|
|
|
|
if err != nil {
|
|
|
|
return 0, fmt.Errorf("d.Invites.SelectMaxInviteID: %w", err)
|
|
|
|
}
|
|
|
|
return types.StreamPosition(id), nil
|
|
|
|
}
|
|
|
|
|
2021-01-13 17:29:46 +00:00
|
|
|
func (d *Database) MaxStreamPositionForSendToDeviceMessages(ctx context.Context) (types.StreamPosition, error) {
|
|
|
|
id, err := d.SendToDevice.SelectMaxSendToDeviceMessageID(ctx, nil)
|
|
|
|
if err != nil {
|
|
|
|
return 0, fmt.Errorf("d.SendToDevice.SelectMaxSendToDeviceMessageID: %w", err)
|
|
|
|
}
|
|
|
|
return types.StreamPosition(id), nil
|
|
|
|
}
|
|
|
|
|
2021-01-08 16:59:06 +00:00
|
|
|
func (d *Database) MaxStreamPositionForAccountData(ctx context.Context) (types.StreamPosition, error) {
|
|
|
|
id, err := d.AccountData.SelectMaxAccountDataID(ctx, nil)
|
|
|
|
if err != nil {
|
|
|
|
return 0, fmt.Errorf("d.Invites.SelectMaxAccountDataID: %w", err)
|
|
|
|
}
|
|
|
|
return types.StreamPosition(id), nil
|
|
|
|
}
|
|
|
|
|
2021-02-04 12:20:37 +00:00
|
|
|
func (d *Database) CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter, excludeEventIDs []string) ([]*gomatrixserverlib.HeaderedEvent, error) {
|
|
|
|
return d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilterPart, excludeEventIDs)
|
2021-01-08 16:59:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error) {
|
|
|
|
return d.CurrentRoomState.SelectRoomIDsWithMembership(ctx, nil, userID, membership)
|
|
|
|
}
|
|
|
|
|
2021-01-19 18:00:42 +00:00
|
|
|
func (d *Database) RecentEvents(ctx context.Context, roomID string, r types.Range, eventFilter *gomatrixserverlib.RoomEventFilter, chronologicalOrder bool, onlySyncEvents bool) ([]types.StreamEvent, bool, error) {
|
|
|
|
return d.OutputEvents.SelectRecentEvents(ctx, nil, roomID, r, eventFilter, chronologicalOrder, onlySyncEvents)
|
2021-01-08 16:59:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) PositionInTopology(ctx context.Context, eventID string) (pos types.StreamPosition, spos types.StreamPosition, err error) {
|
|
|
|
return d.Topology.SelectPositionInTopology(ctx, nil, eventID)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) InviteEventsInRange(ctx context.Context, targetUserID string, r types.Range) (map[string]*gomatrixserverlib.HeaderedEvent, map[string]*gomatrixserverlib.HeaderedEvent, error) {
|
|
|
|
return d.Invites.SelectInviteEventsInRange(ctx, nil, targetUserID, r)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) PeeksInRange(ctx context.Context, userID, deviceID string, r types.Range) (peeks []types.Peek, err error) {
|
|
|
|
return d.Peeks.SelectPeeksInRange(ctx, nil, userID, deviceID, r)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) RoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []eduAPI.OutputReceiptEvent, error) {
|
|
|
|
return d.Receipts.SelectRoomReceiptsAfter(ctx, roomIDs, streamPos)
|
2020-05-13 16:28:42 +00:00
|
|
|
}
|
|
|
|
|
2020-05-14 08:53:55 +00:00
|
|
|
// Events lookups a list of event by their event ID.
|
|
|
|
// Returns a list of events matching the requested IDs found in the database.
|
|
|
|
// If an event is not found in the database then it will be omitted from the list.
|
|
|
|
// Returns an error if there was a problem talking with the database.
|
|
|
|
// Does not include any transaction IDs in the returned events.
|
2020-11-16 15:44:53 +00:00
|
|
|
func (d *Database) Events(ctx context.Context, eventIDs []string) ([]*gomatrixserverlib.HeaderedEvent, error) {
|
2020-05-14 08:53:55 +00:00
|
|
|
streamEvents, err := d.OutputEvents.SelectEvents(ctx, nil, eventIDs)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// We don't include a device here as we only include transaction IDs in
|
|
|
|
// incremental syncs.
|
|
|
|
return d.StreamEventsToEvents(nil, streamEvents), nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// GetEventsInStreamingRange retrieves all of the events on a given ordering using the
|
|
|
|
// given extremities and limit.
|
|
|
|
func (d *Database) GetEventsInStreamingRange(
|
|
|
|
ctx context.Context,
|
|
|
|
from, to *types.StreamingToken,
|
2021-01-19 18:00:42 +00:00
|
|
|
roomID string, eventFilter *gomatrixserverlib.RoomEventFilter,
|
2020-05-14 08:53:55 +00:00
|
|
|
backwardOrdering bool,
|
|
|
|
) (events []types.StreamEvent, err error) {
|
2020-05-15 08:41:12 +00:00
|
|
|
r := types.Range{
|
2020-12-10 18:57:10 +00:00
|
|
|
From: from.PDUPosition,
|
|
|
|
To: to.PDUPosition,
|
2020-05-15 08:41:12 +00:00
|
|
|
Backwards: backwardOrdering,
|
|
|
|
}
|
2020-05-14 08:53:55 +00:00
|
|
|
if backwardOrdering {
|
|
|
|
// When using backward ordering, we want the most recent events first.
|
2020-06-26 14:34:41 +00:00
|
|
|
if events, _, err = d.OutputEvents.SelectRecentEvents(
|
2021-01-19 18:00:42 +00:00
|
|
|
ctx, nil, roomID, r, eventFilter, false, false,
|
2020-05-14 08:53:55 +00:00
|
|
|
); err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// When using forward ordering, we want the least recent events first.
|
|
|
|
if events, err = d.OutputEvents.SelectEarlyEvents(
|
2021-01-19 18:00:42 +00:00
|
|
|
ctx, nil, roomID, r, eventFilter,
|
2020-05-14 08:53:55 +00:00
|
|
|
); err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return events, err
|
|
|
|
}
|
|
|
|
|
2020-05-14 15:11:37 +00:00
|
|
|
func (d *Database) AllJoinedUsersInRooms(ctx context.Context) (map[string][]string, error) {
|
|
|
|
return d.CurrentRoomState.SelectJoinedUsers(ctx)
|
|
|
|
}
|
|
|
|
|
2020-09-10 13:39:18 +00:00
|
|
|
func (d *Database) AllPeekingDevicesInRooms(ctx context.Context) (map[string][]types.PeekingDevice, error) {
|
|
|
|
return d.Peeks.SelectPeekingDevices(ctx)
|
|
|
|
}
|
|
|
|
|
2020-05-14 15:11:37 +00:00
|
|
|
func (d *Database) GetStateEvent(
|
|
|
|
ctx context.Context, roomID, evType, stateKey string,
|
|
|
|
) (*gomatrixserverlib.HeaderedEvent, error) {
|
|
|
|
return d.CurrentRoomState.SelectStateEvent(ctx, roomID, evType, stateKey)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) GetStateEventsForRoom(
|
|
|
|
ctx context.Context, roomID string, stateFilter *gomatrixserverlib.StateFilter,
|
2020-11-16 15:44:53 +00:00
|
|
|
) (stateEvents []*gomatrixserverlib.HeaderedEvent, err error) {
|
2021-02-04 12:20:37 +00:00
|
|
|
stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilter, nil)
|
2020-05-14 15:11:37 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2020-05-14 08:53:55 +00:00
|
|
|
// AddInviteEvent stores a new invite event for a user.
|
|
|
|
// If the invite was successfully stored this returns the stream ID it was stored at.
|
|
|
|
// Returns an error if there was a problem communicating with the database.
|
2020-05-13 16:28:42 +00:00
|
|
|
func (d *Database) AddInviteEvent(
|
2020-11-16 15:44:53 +00:00
|
|
|
ctx context.Context, inviteEvent *gomatrixserverlib.HeaderedEvent,
|
2020-05-13 16:28:42 +00:00
|
|
|
) (sp types.StreamPosition, err error) {
|
2020-09-01 10:28:35 +00:00
|
|
|
_ = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
|
|
|
sp, err = d.Invites.InsertInviteEvent(ctx, txn, inviteEvent)
|
2020-09-10 13:39:18 +00:00
|
|
|
return err
|
2020-05-13 16:28:42 +00:00
|
|
|
})
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2020-05-14 08:53:55 +00:00
|
|
|
// RetireInviteEvent removes an old invite event from the database.
|
|
|
|
// Returns an error if there was a problem communicating with the database.
|
2020-05-13 16:28:42 +00:00
|
|
|
func (d *Database) RetireInviteEvent(
|
|
|
|
ctx context.Context, inviteEventID string,
|
2020-08-21 09:42:08 +00:00
|
|
|
) (sp types.StreamPosition, err error) {
|
2020-09-01 10:28:35 +00:00
|
|
|
_ = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
|
|
|
sp, err = d.Invites.DeleteInviteEvent(ctx, txn, inviteEventID)
|
2020-09-10 13:39:18 +00:00
|
|
|
return err
|
2020-08-21 09:42:08 +00:00
|
|
|
})
|
|
|
|
return
|
2020-05-13 16:28:42 +00:00
|
|
|
}
|
2020-05-14 08:53:55 +00:00
|
|
|
|
2020-09-10 13:39:18 +00:00
|
|
|
// AddPeek tracks the fact that a user has started peeking.
|
|
|
|
// If the peek was successfully stored this returns the stream ID it was stored at.
|
|
|
|
// Returns an error if there was a problem communicating with the database.
|
|
|
|
func (d *Database) AddPeek(
|
|
|
|
ctx context.Context, roomID, userID, deviceID string,
|
|
|
|
) (sp types.StreamPosition, err error) {
|
|
|
|
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
|
|
|
sp, err = d.Peeks.InsertPeek(ctx, txn, roomID, userID, deviceID)
|
|
|
|
return err
|
|
|
|
})
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2020-12-03 11:11:46 +00:00
|
|
|
// DeletePeeks tracks the fact that a user has stopped peeking from the specified
|
|
|
|
// device. If the peeks was successfully deleted this returns the stream ID it was
|
|
|
|
// stored at. Returns an error if there was a problem communicating with the database.
|
|
|
|
func (d *Database) DeletePeek(
|
|
|
|
ctx context.Context, roomID, userID, deviceID string,
|
|
|
|
) (sp types.StreamPosition, err error) {
|
|
|
|
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
|
|
|
sp, err = d.Peeks.DeletePeek(ctx, txn, roomID, userID, deviceID)
|
|
|
|
return err
|
|
|
|
})
|
|
|
|
if err == sql.ErrNoRows {
|
|
|
|
sp = 0
|
|
|
|
err = nil
|
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2020-09-10 13:39:18 +00:00
|
|
|
// DeletePeeks tracks the fact that a user has stopped peeking from all devices
|
|
|
|
// If the peeks was successfully deleted this returns the stream ID it was stored at.
|
|
|
|
// Returns an error if there was a problem communicating with the database.
|
|
|
|
func (d *Database) DeletePeeks(
|
|
|
|
ctx context.Context, roomID, userID string,
|
|
|
|
) (sp types.StreamPosition, err error) {
|
|
|
|
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
|
|
|
sp, err = d.Peeks.DeletePeeks(ctx, txn, roomID, userID)
|
|
|
|
return err
|
|
|
|
})
|
|
|
|
if err == sql.ErrNoRows {
|
|
|
|
sp = 0
|
|
|
|
err = nil
|
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2020-05-14 08:53:55 +00:00
|
|
|
// GetAccountDataInRange returns all account data for a given user inserted or
|
|
|
|
// updated between two given positions
|
|
|
|
// Returns a map following the format data[roomID] = []dataTypes
|
|
|
|
// If no data is retrieved, returns an empty map
|
|
|
|
// If there was an issue with the retrieval, returns an error
|
|
|
|
func (d *Database) GetAccountDataInRange(
|
2020-05-15 08:41:12 +00:00
|
|
|
ctx context.Context, userID string, r types.Range,
|
2020-05-14 08:53:55 +00:00
|
|
|
accountDataFilterPart *gomatrixserverlib.EventFilter,
|
|
|
|
) (map[string][]string, error) {
|
2020-05-15 08:41:12 +00:00
|
|
|
return d.AccountData.SelectAccountDataInRange(ctx, userID, r, accountDataFilterPart)
|
2020-05-14 08:53:55 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// UpsertAccountData keeps track of new or updated account data, by saving the type
|
|
|
|
// of the new/updated data, and the user ID and room ID the data is related to (empty)
|
|
|
|
// room ID means the data isn't specific to any room)
|
|
|
|
// If no data with the given type, user ID and room ID exists in the database,
|
|
|
|
// creates a new row, else update the existing one
|
|
|
|
// Returns an error if there was an issue with the upsert
|
|
|
|
func (d *Database) UpsertAccountData(
|
|
|
|
ctx context.Context, userID, roomID, dataType string,
|
|
|
|
) (sp types.StreamPosition, err error) {
|
2020-08-21 09:42:08 +00:00
|
|
|
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
2020-05-14 08:53:55 +00:00
|
|
|
sp, err = d.AccountData.InsertAccountData(ctx, txn, userID, roomID, dataType)
|
|
|
|
return err
|
|
|
|
})
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2020-11-16 15:44:53 +00:00
|
|
|
func (d *Database) StreamEventsToEvents(device *userapi.Device, in []types.StreamEvent) []*gomatrixserverlib.HeaderedEvent {
|
|
|
|
out := make([]*gomatrixserverlib.HeaderedEvent, len(in))
|
2020-05-14 08:53:55 +00:00
|
|
|
for i := 0; i < len(in); i++ {
|
|
|
|
out[i] = in[i].HeaderedEvent
|
|
|
|
if device != nil && in[i].TransactionID != nil {
|
|
|
|
if device.UserID == in[i].Sender() && device.SessionID == in[i].TransactionID.SessionID {
|
|
|
|
err := out[i].SetUnsignedField(
|
|
|
|
"transaction_id", in[i].TransactionID.TransactionID,
|
|
|
|
)
|
|
|
|
if err != nil {
|
2020-09-10 13:39:18 +00:00
|
|
|
log.WithFields(log.Fields{
|
2020-05-14 08:53:55 +00:00
|
|
|
"event_id": out[i].EventID(),
|
|
|
|
}).WithError(err).Warnf("Failed to add transaction ID to event")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return out
|
|
|
|
}
|
2020-05-14 15:11:37 +00:00
|
|
|
|
|
|
|
// handleBackwardExtremities adds this event as a backwards extremity if and only if we do not have all of
|
|
|
|
// the events listed in the event's 'prev_events'. This function also updates the backwards extremities table
|
|
|
|
// to account for the fact that the given event is no longer a backwards extremity, but may be marked as such.
|
2020-08-21 09:42:08 +00:00
|
|
|
// This function should always be called within a sqlutil.Writer for safety in SQLite.
|
2020-05-14 15:11:37 +00:00
|
|
|
func (d *Database) handleBackwardExtremities(ctx context.Context, txn *sql.Tx, ev *gomatrixserverlib.HeaderedEvent) error {
|
|
|
|
if err := d.BackwardExtremities.DeleteBackwardExtremity(ctx, txn, ev.RoomID(), ev.EventID()); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Check if we have all of the event's previous events. If an event is
|
|
|
|
// missing, add it to the room's backward extremities.
|
|
|
|
prevEvents, err := d.OutputEvents.SelectEvents(ctx, txn, ev.PrevEventIDs())
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
var found bool
|
|
|
|
for _, eID := range ev.PrevEventIDs() {
|
|
|
|
found = false
|
|
|
|
for _, prevEv := range prevEvents {
|
|
|
|
if eID == prevEv.EventID() {
|
|
|
|
found = true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// If the event is missing, consider it a backward extremity.
|
|
|
|
if !found {
|
|
|
|
if err = d.BackwardExtremities.InsertsBackwardExtremity(ctx, txn, ev.RoomID(), ev.EventID(), eID); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2020-10-22 09:39:16 +00:00
|
|
|
func (d *Database) PurgeRoomState(
|
2020-09-15 10:17:46 +00:00
|
|
|
ctx context.Context, roomID string,
|
|
|
|
) error {
|
|
|
|
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
|
|
|
// If the event is a create event then we'll delete all of the existing
|
|
|
|
// data for the room. The only reason that a create event would be replayed
|
|
|
|
// to us in this way is if we're about to receive the entire room state.
|
|
|
|
if err := d.CurrentRoomState.DeleteRoomStateForRoom(ctx, txn, roomID); err != nil {
|
|
|
|
return fmt.Errorf("d.CurrentRoomState.DeleteRoomStateForRoom: %w", err)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2020-05-14 15:11:37 +00:00
|
|
|
func (d *Database) WriteEvent(
|
|
|
|
ctx context.Context,
|
|
|
|
ev *gomatrixserverlib.HeaderedEvent,
|
2020-11-16 15:44:53 +00:00
|
|
|
addStateEvents []*gomatrixserverlib.HeaderedEvent,
|
2020-05-14 15:11:37 +00:00
|
|
|
addStateEventIDs, removeStateEventIDs []string,
|
|
|
|
transactionID *api.TransactionID, excludeFromSync bool,
|
|
|
|
) (pduPosition types.StreamPosition, returnErr error) {
|
2020-08-21 09:42:08 +00:00
|
|
|
returnErr = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
2020-05-14 15:11:37 +00:00
|
|
|
var err error
|
|
|
|
pos, err := d.OutputEvents.InsertEvent(
|
|
|
|
ctx, txn, ev, addStateEventIDs, removeStateEventIDs, transactionID, excludeFromSync,
|
|
|
|
)
|
|
|
|
if err != nil {
|
2020-07-21 14:48:21 +00:00
|
|
|
return fmt.Errorf("d.OutputEvents.InsertEvent: %w", err)
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
pduPosition = pos
|
2021-01-20 20:43:20 +00:00
|
|
|
var topoPosition types.StreamPosition
|
|
|
|
if topoPosition, err = d.Topology.InsertEventInTopology(ctx, txn, ev, pos); err != nil {
|
2020-07-21 14:48:21 +00:00
|
|
|
return fmt.Errorf("d.Topology.InsertEventInTopology: %w", err)
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
if err = d.handleBackwardExtremities(ctx, txn, ev); err != nil {
|
2020-07-21 14:48:21 +00:00
|
|
|
return fmt.Errorf("d.handleBackwardExtremities: %w", err)
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
if len(addStateEvents) == 0 && len(removeStateEventIDs) == 0 {
|
|
|
|
// Nothing to do, the event may have just been a message event.
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2021-01-20 20:43:20 +00:00
|
|
|
return d.updateRoomState(ctx, txn, removeStateEventIDs, addStateEvents, pduPosition, topoPosition)
|
2020-05-14 15:11:37 +00:00
|
|
|
})
|
|
|
|
|
|
|
|
return pduPosition, returnErr
|
|
|
|
}
|
|
|
|
|
2020-08-21 09:42:08 +00:00
|
|
|
// This function should always be called within a sqlutil.Writer for safety in SQLite.
|
2020-05-14 15:11:37 +00:00
|
|
|
func (d *Database) updateRoomState(
|
|
|
|
ctx context.Context, txn *sql.Tx,
|
|
|
|
removedEventIDs []string,
|
2020-11-16 15:44:53 +00:00
|
|
|
addedEvents []*gomatrixserverlib.HeaderedEvent,
|
2020-05-14 15:11:37 +00:00
|
|
|
pduPosition types.StreamPosition,
|
2021-01-20 20:43:20 +00:00
|
|
|
topoPosition types.StreamPosition,
|
2020-05-14 15:11:37 +00:00
|
|
|
) error {
|
|
|
|
// remove first, then add, as we do not ever delete state, but do replace state which is a remove followed by an add.
|
|
|
|
for _, eventID := range removedEventIDs {
|
|
|
|
if err := d.CurrentRoomState.DeleteRoomStateByEventID(ctx, txn, eventID); err != nil {
|
2020-07-21 14:48:21 +00:00
|
|
|
return fmt.Errorf("d.CurrentRoomState.DeleteRoomStateByEventID: %w", err)
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, event := range addedEvents {
|
|
|
|
if event.StateKey() == nil {
|
|
|
|
// ignore non state events
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
var membership *string
|
|
|
|
if event.Type() == "m.room.member" {
|
|
|
|
value, err := event.Membership()
|
|
|
|
if err != nil {
|
2020-07-21 14:48:21 +00:00
|
|
|
return fmt.Errorf("event.Membership: %w", err)
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
membership = &value
|
2021-01-20 20:43:20 +00:00
|
|
|
if err = d.Memberships.UpsertMembership(ctx, txn, event, pduPosition, topoPosition); err != nil {
|
|
|
|
return fmt.Errorf("d.Memberships.UpsertMembership: %w", err)
|
|
|
|
}
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
2020-05-20 15:04:31 +00:00
|
|
|
|
2020-05-14 15:11:37 +00:00
|
|
|
if err := d.CurrentRoomState.UpsertRoomState(ctx, txn, event, membership, pduPosition); err != nil {
|
2020-07-21 14:48:21 +00:00
|
|
|
return fmt.Errorf("d.CurrentRoomState.UpsertRoomState: %w", err)
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) GetEventsInTopologicalRange(
|
|
|
|
ctx context.Context,
|
|
|
|
from, to *types.TopologyToken,
|
|
|
|
roomID string, limit int,
|
|
|
|
backwardOrdering bool,
|
|
|
|
) (events []types.StreamEvent, err error) {
|
2020-05-14 16:30:16 +00:00
|
|
|
var minDepth, maxDepth, maxStreamPosForMaxDepth types.StreamPosition
|
2020-05-14 15:11:37 +00:00
|
|
|
if backwardOrdering {
|
2020-05-14 16:30:16 +00:00
|
|
|
// Backward ordering means the 'from' token has a higher depth than the 'to' token
|
2020-12-10 18:57:10 +00:00
|
|
|
minDepth = to.Depth
|
|
|
|
maxDepth = from.Depth
|
2020-05-14 16:30:16 +00:00
|
|
|
// for cases where we have say 5 events with the same depth, the TopologyToken needs to
|
|
|
|
// know which of the 5 the client has seen. This is done by using the PDU position.
|
|
|
|
// Events with the same maxDepth but less than this PDU position will be returned.
|
2020-12-10 18:57:10 +00:00
|
|
|
maxStreamPosForMaxDepth = from.PDUPosition
|
2020-05-14 15:11:37 +00:00
|
|
|
} else {
|
2020-05-14 16:30:16 +00:00
|
|
|
// Forward ordering means the 'from' token has a lower depth than the 'to' token.
|
2020-12-10 18:57:10 +00:00
|
|
|
minDepth = from.Depth
|
|
|
|
maxDepth = to.Depth
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Select the event IDs from the defined range.
|
|
|
|
var eIDs []string
|
|
|
|
eIDs, err = d.Topology.SelectEventIDsInRange(
|
2020-05-14 16:30:16 +00:00
|
|
|
ctx, nil, roomID, minDepth, maxDepth, maxStreamPosForMaxDepth, limit, !backwardOrdering,
|
2020-05-14 15:11:37 +00:00
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// Retrieve the events' contents using their IDs.
|
|
|
|
events, err = d.OutputEvents.SelectEvents(ctx, nil, eIDs)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) BackwardExtremitiesForRoom(
|
|
|
|
ctx context.Context, roomID string,
|
2020-05-20 15:04:31 +00:00
|
|
|
) (backwardExtremities map[string][]string, err error) {
|
2020-05-14 15:11:37 +00:00
|
|
|
return d.BackwardExtremities.SelectBackwardExtremitiesForRoom(ctx, roomID)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) MaxTopologicalPosition(
|
|
|
|
ctx context.Context, roomID string,
|
2020-05-14 16:30:16 +00:00
|
|
|
) (types.TopologyToken, error) {
|
|
|
|
depth, streamPos, err := d.Topology.SelectMaxPositionInTopology(ctx, nil, roomID)
|
2020-05-14 15:11:37 +00:00
|
|
|
if err != nil {
|
2020-12-10 18:57:10 +00:00
|
|
|
return types.TopologyToken{}, err
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
2020-12-10 18:57:10 +00:00
|
|
|
return types.TopologyToken{Depth: depth, PDUPosition: streamPos}, nil
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) EventPositionInTopology(
|
|
|
|
ctx context.Context, eventID string,
|
2020-05-14 16:30:16 +00:00
|
|
|
) (types.TopologyToken, error) {
|
|
|
|
depth, stream, err := d.Topology.SelectPositionInTopology(ctx, nil, eventID)
|
|
|
|
if err != nil {
|
2020-12-10 18:57:10 +00:00
|
|
|
return types.TopologyToken{}, err
|
2020-05-14 16:30:16 +00:00
|
|
|
}
|
2020-12-10 18:57:10 +00:00
|
|
|
return types.TopologyToken{Depth: depth, PDUPosition: stream}, nil
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
|
2020-06-26 14:34:41 +00:00
|
|
|
func (d *Database) GetFilter(
|
|
|
|
ctx context.Context, localpart string, filterID string,
|
|
|
|
) (*gomatrixserverlib.Filter, error) {
|
|
|
|
return d.Filter.SelectFilter(ctx, localpart, filterID)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) PutFilter(
|
|
|
|
ctx context.Context, localpart string, filter *gomatrixserverlib.Filter,
|
|
|
|
) (string, error) {
|
2020-09-01 19:35:38 +00:00
|
|
|
var filterID string
|
|
|
|
var err error
|
|
|
|
err = d.Writer.Do(nil, nil, func(txn *sql.Tx) error {
|
|
|
|
filterID, err = d.Filter.InsertFilter(ctx, filter, localpart)
|
|
|
|
return err
|
|
|
|
})
|
|
|
|
return filterID, err
|
2020-06-26 14:34:41 +00:00
|
|
|
}
|
|
|
|
|
2020-07-08 16:45:39 +00:00
|
|
|
func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, redactedBecause *gomatrixserverlib.HeaderedEvent) error {
|
|
|
|
redactedEvents, err := d.Events(ctx, []string{redactedEventID})
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if len(redactedEvents) == 0 {
|
2020-09-10 13:39:18 +00:00
|
|
|
log.WithField("event_id", redactedEventID).WithField("redaction_event", redactedBecause.EventID()).Warnf("missing redacted event for redaction")
|
2020-07-08 16:45:39 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
eventToRedact := redactedEvents[0].Unwrap()
|
|
|
|
redactionEvent := redactedBecause.Unwrap()
|
2020-11-16 15:44:53 +00:00
|
|
|
ev, err := eventutil.RedactEvent(redactionEvent, eventToRedact)
|
2020-07-08 16:45:39 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
newEvent := ev.Headered(redactedBecause.RoomVersion)
|
2020-09-01 19:35:38 +00:00
|
|
|
err = d.Writer.Do(nil, nil, func(txn *sql.Tx) error {
|
2020-11-16 15:44:53 +00:00
|
|
|
return d.OutputEvents.UpdateEventJSON(ctx, newEvent)
|
2020-09-01 19:35:38 +00:00
|
|
|
})
|
|
|
|
return err
|
2020-07-08 16:45:39 +00:00
|
|
|
}
|
|
|
|
|
2020-05-14 15:11:37 +00:00
|
|
|
// Retrieve the backward topology position, i.e. the position of the
|
|
|
|
// oldest event in the room's topology.
|
2021-01-08 16:59:06 +00:00
|
|
|
func (d *Database) GetBackwardTopologyPos(
|
|
|
|
ctx context.Context,
|
2020-05-14 15:11:37 +00:00
|
|
|
events []types.StreamEvent,
|
|
|
|
) (types.TopologyToken, error) {
|
2020-12-10 18:57:10 +00:00
|
|
|
zeroToken := types.TopologyToken{}
|
2020-05-14 15:11:37 +00:00
|
|
|
if len(events) == 0 {
|
|
|
|
return zeroToken, nil
|
|
|
|
}
|
2021-01-08 16:59:06 +00:00
|
|
|
pos, spos, err := d.Topology.SelectPositionInTopology(ctx, nil, events[0].EventID())
|
2020-05-14 15:11:37 +00:00
|
|
|
if err != nil {
|
|
|
|
return zeroToken, err
|
|
|
|
}
|
2020-12-10 18:57:10 +00:00
|
|
|
tok := types.TopologyToken{Depth: pos, PDUPosition: spos}
|
2020-05-14 15:11:37 +00:00
|
|
|
tok.Decrement()
|
|
|
|
return tok, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// fetchStateEvents converts the set of event IDs into a set of events. It will fetch any which are missing from the database.
|
|
|
|
// Returns a map of room ID to list of events.
|
|
|
|
func (d *Database) fetchStateEvents(
|
|
|
|
ctx context.Context, txn *sql.Tx,
|
|
|
|
roomIDToEventIDSet map[string]map[string]bool,
|
|
|
|
eventIDToEvent map[string]types.StreamEvent,
|
|
|
|
) (map[string][]types.StreamEvent, error) {
|
|
|
|
stateBetween := make(map[string][]types.StreamEvent)
|
|
|
|
missingEvents := make(map[string][]string)
|
|
|
|
for roomID, ids := range roomIDToEventIDSet {
|
|
|
|
events := stateBetween[roomID]
|
|
|
|
for id, need := range ids {
|
|
|
|
if !need {
|
|
|
|
continue // deleted state
|
|
|
|
}
|
|
|
|
e, ok := eventIDToEvent[id]
|
|
|
|
if ok {
|
|
|
|
events = append(events, e)
|
|
|
|
} else {
|
|
|
|
m := missingEvents[roomID]
|
|
|
|
m = append(m, id)
|
|
|
|
missingEvents[roomID] = m
|
|
|
|
}
|
|
|
|
}
|
|
|
|
stateBetween[roomID] = events
|
|
|
|
}
|
|
|
|
|
|
|
|
if len(missingEvents) > 0 {
|
|
|
|
// This happens when add_state_ids has an event ID which is not in the provided range.
|
|
|
|
// We need to explicitly fetch them.
|
|
|
|
allMissingEventIDs := []string{}
|
|
|
|
for _, missingEvIDs := range missingEvents {
|
|
|
|
allMissingEventIDs = append(allMissingEventIDs, missingEvIDs...)
|
|
|
|
}
|
|
|
|
evs, err := d.fetchMissingStateEvents(ctx, txn, allMissingEventIDs)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
// we know we got them all otherwise an error would've been returned, so just loop the events
|
|
|
|
for _, ev := range evs {
|
|
|
|
roomID := ev.RoomID()
|
|
|
|
stateBetween[roomID] = append(stateBetween[roomID], ev)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return stateBetween, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) fetchMissingStateEvents(
|
|
|
|
ctx context.Context, txn *sql.Tx, eventIDs []string,
|
|
|
|
) ([]types.StreamEvent, error) {
|
|
|
|
// Fetch from the events table first so we pick up the stream ID for the
|
|
|
|
// event.
|
|
|
|
events, err := d.OutputEvents.SelectEvents(ctx, txn, eventIDs)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
have := map[string]bool{}
|
|
|
|
for _, event := range events {
|
|
|
|
have[event.EventID()] = true
|
|
|
|
}
|
|
|
|
var missing []string
|
|
|
|
for _, eventID := range eventIDs {
|
|
|
|
if !have[eventID] {
|
|
|
|
missing = append(missing, eventID)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if len(missing) == 0 {
|
|
|
|
return events, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// If they are missing from the events table then they should be state
|
|
|
|
// events that we received from outside the main event stream.
|
|
|
|
// These should be in the room state table.
|
|
|
|
stateEvents, err := d.CurrentRoomState.SelectEventsWithEventIDs(ctx, txn, missing)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
if len(stateEvents) != len(missing) {
|
2020-12-21 14:27:01 +00:00
|
|
|
log.WithContext(ctx).Warnf("Failed to map all event IDs to events (got %d, wanted %d)", len(stateEvents), len(missing))
|
2020-12-09 18:07:17 +00:00
|
|
|
|
|
|
|
// TODO: Why is this happening? It's probably the roomserver. Uncomment
|
|
|
|
// this error again when we work out what it is and fix it, otherwise we
|
|
|
|
// just end up returning lots of 500s to the client and that breaks
|
|
|
|
// pretty much everything, rather than just sending what we have.
|
|
|
|
//return nil, fmt.Errorf("failed to map all event IDs to events: (got %d, wanted %d)", len(stateEvents), len(missing))
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
events = append(events, stateEvents...)
|
|
|
|
return events, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// getStateDeltas returns the state deltas between fromPos and toPos,
|
|
|
|
// exclusive of oldPos, inclusive of newPos, for the rooms in which
|
|
|
|
// the user has new membership events.
|
|
|
|
// A list of joined room IDs is also returned in case the caller needs it.
|
2021-01-08 16:59:06 +00:00
|
|
|
func (d *Database) GetStateDeltas(
|
|
|
|
ctx context.Context, device *userapi.Device,
|
2020-05-15 08:41:12 +00:00
|
|
|
r types.Range, userID string,
|
2020-05-14 15:11:37 +00:00
|
|
|
stateFilter *gomatrixserverlib.StateFilter,
|
2021-01-08 16:59:06 +00:00
|
|
|
) ([]types.StateDelta, []string, error) {
|
2020-05-14 15:11:37 +00:00
|
|
|
// Implement membership change algorithm: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L821
|
|
|
|
// - Get membership list changes for this user in this sync response
|
|
|
|
// - For each room which has membership list changes:
|
|
|
|
// * Check if the room is 'newly joined' (insufficient to just check for a join event because we allow dupe joins TODO).
|
|
|
|
// If it is, then we need to send the full room state down (and 'limited' is always true).
|
|
|
|
// * Check if user is still CURRENTLY invited to the room. If so, add room to 'invited' block.
|
|
|
|
// * Check if the user is CURRENTLY (TODO) left/banned. If so, add room to 'archived' block.
|
|
|
|
// - Get all CURRENTLY joined rooms, and add them to 'joined' block.
|
2021-01-08 16:59:06 +00:00
|
|
|
txn, err := d.readOnlySnapshot(ctx)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, fmt.Errorf("d.readOnlySnapshot: %w", err)
|
|
|
|
}
|
|
|
|
var succeeded bool
|
|
|
|
defer sqlutil.EndTransactionWithCheck(txn, &succeeded, &err)
|
|
|
|
|
|
|
|
var deltas []types.StateDelta
|
2020-05-14 15:11:37 +00:00
|
|
|
|
2020-09-10 13:39:18 +00:00
|
|
|
// get all the state events ever (i.e. for all available rooms) between these two positions
|
2020-05-15 08:41:12 +00:00
|
|
|
stateNeeded, eventMap, err := d.OutputEvents.SelectStateInRange(ctx, txn, r, stateFilter)
|
2020-05-14 15:11:37 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
state, err := d.fetchStateEvents(ctx, txn, stateNeeded, eventMap)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
2020-09-10 13:39:18 +00:00
|
|
|
// find out which rooms this user is peeking, if any.
|
|
|
|
// We do this before joins so any peeks get overwritten
|
|
|
|
peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, device.ID, r)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// add peek blocks
|
|
|
|
for _, peek := range peeks {
|
|
|
|
if peek.New {
|
|
|
|
// send full room state down instead of a delta
|
|
|
|
var s []types.StreamEvent
|
|
|
|
s, err = d.currentStateStreamEventsForRoom(ctx, txn, peek.RoomID, stateFilter)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
state[peek.RoomID] = s
|
|
|
|
}
|
|
|
|
if !peek.Deleted {
|
2021-01-08 16:59:06 +00:00
|
|
|
deltas = append(deltas, types.StateDelta{
|
|
|
|
Membership: gomatrixserverlib.Peek,
|
|
|
|
StateEvents: d.StreamEventsToEvents(device, state[peek.RoomID]),
|
|
|
|
RoomID: peek.RoomID,
|
2020-09-10 13:39:18 +00:00
|
|
|
})
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// handle newly joined rooms and non-joined rooms
|
2020-05-14 15:11:37 +00:00
|
|
|
for roomID, stateStreamEvents := range state {
|
|
|
|
for _, ev := range stateStreamEvents {
|
|
|
|
// TODO: Currently this will incorrectly add rooms which were ALREADY joined but they sent another no-op join event.
|
|
|
|
// We should be checking if the user was already joined at fromPos and not proceed if so. As a result of this,
|
|
|
|
// dupe join events will result in the entire room state coming down to the client again. This is added in
|
|
|
|
// the 'state' part of the response though, so is transparent modulo bandwidth concerns as it is not added to
|
|
|
|
// the timeline.
|
2020-11-16 15:44:53 +00:00
|
|
|
if membership := getMembershipFromEvent(ev.Event, userID); membership != "" {
|
2020-05-14 15:11:37 +00:00
|
|
|
if membership == gomatrixserverlib.Join {
|
|
|
|
// send full room state down instead of a delta
|
|
|
|
var s []types.StreamEvent
|
|
|
|
s, err = d.currentStateStreamEventsForRoom(ctx, txn, roomID, stateFilter)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
state[roomID] = s
|
|
|
|
continue // we'll add this room in when we do joined rooms
|
|
|
|
}
|
|
|
|
|
2021-01-08 16:59:06 +00:00
|
|
|
deltas = append(deltas, types.StateDelta{
|
|
|
|
Membership: membership,
|
|
|
|
MembershipPos: ev.StreamPosition,
|
|
|
|
StateEvents: d.StreamEventsToEvents(device, stateStreamEvents),
|
|
|
|
RoomID: roomID,
|
2020-05-14 15:11:37 +00:00
|
|
|
})
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Add in currently joined rooms
|
|
|
|
joinedRoomIDs, err := d.CurrentRoomState.SelectRoomIDsWithMembership(ctx, txn, userID, gomatrixserverlib.Join)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
for _, joinedRoomID := range joinedRoomIDs {
|
2021-01-08 16:59:06 +00:00
|
|
|
deltas = append(deltas, types.StateDelta{
|
|
|
|
Membership: gomatrixserverlib.Join,
|
|
|
|
StateEvents: d.StreamEventsToEvents(device, state[joinedRoomID]),
|
|
|
|
RoomID: joinedRoomID,
|
2020-05-14 15:11:37 +00:00
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2021-01-08 16:59:06 +00:00
|
|
|
succeeded = true
|
2020-05-14 15:11:37 +00:00
|
|
|
return deltas, joinedRoomIDs, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// getStateDeltasForFullStateSync is a variant of getStateDeltas used for /sync
|
|
|
|
// requests with full_state=true.
|
|
|
|
// Fetches full state for all joined rooms and uses selectStateInRange to get
|
|
|
|
// updates for other rooms.
|
2021-01-08 16:59:06 +00:00
|
|
|
func (d *Database) GetStateDeltasForFullStateSync(
|
|
|
|
ctx context.Context, device *userapi.Device,
|
2020-05-15 08:41:12 +00:00
|
|
|
r types.Range, userID string,
|
2020-05-14 15:11:37 +00:00
|
|
|
stateFilter *gomatrixserverlib.StateFilter,
|
2021-01-08 16:59:06 +00:00
|
|
|
) ([]types.StateDelta, []string, error) {
|
|
|
|
txn, err := d.readOnlySnapshot(ctx)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, fmt.Errorf("d.readOnlySnapshot: %w", err)
|
|
|
|
}
|
|
|
|
var succeeded bool
|
|
|
|
defer sqlutil.EndTransactionWithCheck(txn, &succeeded, &err)
|
|
|
|
|
2020-05-14 15:11:37 +00:00
|
|
|
// Use a reasonable initial capacity
|
2021-01-08 16:59:06 +00:00
|
|
|
deltas := make(map[string]types.StateDelta)
|
2020-05-14 15:11:37 +00:00
|
|
|
|
2020-09-10 13:39:18 +00:00
|
|
|
peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, device.ID, r)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Add full states for all peeking rooms
|
|
|
|
for _, peek := range peeks {
|
|
|
|
if !peek.Deleted {
|
|
|
|
s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, peek.RoomID, stateFilter)
|
|
|
|
if stateErr != nil {
|
|
|
|
return nil, nil, stateErr
|
|
|
|
}
|
2021-01-08 16:59:06 +00:00
|
|
|
deltas[peek.RoomID] = types.StateDelta{
|
|
|
|
Membership: gomatrixserverlib.Peek,
|
|
|
|
StateEvents: d.StreamEventsToEvents(device, s),
|
|
|
|
RoomID: peek.RoomID,
|
2020-09-10 13:39:18 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-05-14 15:11:37 +00:00
|
|
|
// Get all the state events ever between these two positions
|
2020-05-15 08:41:12 +00:00
|
|
|
stateNeeded, eventMap, err := d.OutputEvents.SelectStateInRange(ctx, txn, r, stateFilter)
|
2020-05-14 15:11:37 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
state, err := d.fetchStateEvents(ctx, txn, stateNeeded, eventMap)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
for roomID, stateStreamEvents := range state {
|
|
|
|
for _, ev := range stateStreamEvents {
|
2020-11-16 15:44:53 +00:00
|
|
|
if membership := getMembershipFromEvent(ev.Event, userID); membership != "" {
|
2020-05-14 15:11:37 +00:00
|
|
|
if membership != gomatrixserverlib.Join { // We've already added full state for all joined rooms above.
|
2021-01-08 16:59:06 +00:00
|
|
|
deltas[roomID] = types.StateDelta{
|
|
|
|
Membership: membership,
|
|
|
|
MembershipPos: ev.StreamPosition,
|
|
|
|
StateEvents: d.StreamEventsToEvents(device, stateStreamEvents),
|
|
|
|
RoomID: roomID,
|
2020-09-04 10:20:47 +00:00
|
|
|
}
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-09-04 10:20:47 +00:00
|
|
|
joinedRoomIDs, err := d.CurrentRoomState.SelectRoomIDsWithMembership(ctx, txn, userID, gomatrixserverlib.Join)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Add full states for all joined rooms
|
|
|
|
for _, joinedRoomID := range joinedRoomIDs {
|
|
|
|
s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, joinedRoomID, stateFilter)
|
|
|
|
if stateErr != nil {
|
|
|
|
return nil, nil, stateErr
|
|
|
|
}
|
2021-01-08 16:59:06 +00:00
|
|
|
deltas[joinedRoomID] = types.StateDelta{
|
|
|
|
Membership: gomatrixserverlib.Join,
|
|
|
|
StateEvents: d.StreamEventsToEvents(device, s),
|
|
|
|
RoomID: joinedRoomID,
|
2020-09-04 10:20:47 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Create a response array.
|
2021-01-08 16:59:06 +00:00
|
|
|
result := make([]types.StateDelta, len(deltas))
|
2020-09-04 10:20:47 +00:00
|
|
|
i := 0
|
|
|
|
for _, delta := range deltas {
|
|
|
|
result[i] = delta
|
|
|
|
i++
|
|
|
|
}
|
|
|
|
|
2021-01-08 16:59:06 +00:00
|
|
|
succeeded = true
|
2020-09-04 10:20:47 +00:00
|
|
|
return result, joinedRoomIDs, nil
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) currentStateStreamEventsForRoom(
|
|
|
|
ctx context.Context, txn *sql.Tx, roomID string,
|
|
|
|
stateFilter *gomatrixserverlib.StateFilter,
|
|
|
|
) ([]types.StreamEvent, error) {
|
2021-02-04 12:20:37 +00:00
|
|
|
allState, err := d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, stateFilter, nil)
|
2020-05-14 15:11:37 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
s := make([]types.StreamEvent, len(allState))
|
|
|
|
for i := 0; i < len(s); i++ {
|
|
|
|
s[i] = types.StreamEvent{HeaderedEvent: allState[i], StreamPosition: 0}
|
|
|
|
}
|
|
|
|
return s, nil
|
|
|
|
}
|
|
|
|
|
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 16:50:19 +00:00
|
|
|
func (d *Database) StoreNewSendForDeviceMessage(
|
2020-12-18 11:11:21 +00:00
|
|
|
ctx context.Context, userID, deviceID string, event gomatrixserverlib.SendToDeviceEvent,
|
|
|
|
) (newPos types.StreamPosition, 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 16:50:19 +00:00
|
|
|
j, err := json.Marshal(event)
|
|
|
|
if err != nil {
|
2020-12-18 11:11:21 +00:00
|
|
|
return 0, err
|
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 16:50:19 +00:00
|
|
|
}
|
|
|
|
// Delegate the database write task to the SendToDeviceWriter. It'll guarantee
|
|
|
|
// that we don't lock the table for writes in more than one place.
|
2020-08-21 09:42:08 +00:00
|
|
|
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
2020-12-18 11:11:21 +00:00
|
|
|
newPos, err = d.SendToDevice.InsertSendToDeviceMessage(
|
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 16:50:19 +00:00
|
|
|
ctx, txn, userID, deviceID, string(j),
|
|
|
|
)
|
2020-12-18 11:11:21 +00:00
|
|
|
return err
|
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 16:50:19 +00:00
|
|
|
})
|
|
|
|
if err != nil {
|
2020-12-18 11:11:21 +00:00
|
|
|
return 0, err
|
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 16:50:19 +00:00
|
|
|
}
|
2021-01-13 17:29:46 +00:00
|
|
|
return newPos, nil
|
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 16:50:19 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) SendToDeviceUpdatesForSync(
|
|
|
|
ctx context.Context,
|
|
|
|
userID, deviceID string,
|
2021-01-13 17:29:46 +00:00
|
|
|
from, to types.StreamPosition,
|
|
|
|
) (types.StreamPosition, []types.SendToDeviceEvent, 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 16:50:19 +00:00
|
|
|
// First of all, get our send-to-device updates for this user.
|
2021-01-13 17:29:46 +00:00
|
|
|
lastPos, events, err := d.SendToDevice.SelectSendToDeviceMessages(ctx, nil, userID, deviceID, from, to)
|
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 16:50:19 +00:00
|
|
|
if err != nil {
|
2021-01-13 17:29:46 +00:00
|
|
|
return from, nil, fmt.Errorf("d.SendToDevice.SelectSendToDeviceMessages: %w", err)
|
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 16:50:19 +00:00
|
|
|
}
|
|
|
|
// If there's nothing to do then stop here.
|
|
|
|
if len(events) == 0 {
|
2021-01-13 17:29:46 +00:00
|
|
|
return to, nil, nil
|
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 16:50:19 +00:00
|
|
|
}
|
2021-01-13 17:29:46 +00:00
|
|
|
return lastPos, events, nil
|
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 16:50:19 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) CleanSendToDeviceUpdates(
|
|
|
|
ctx context.Context,
|
2021-01-13 17:29:46 +00:00
|
|
|
userID, deviceID string, before types.StreamPosition,
|
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 16:50:19 +00:00
|
|
|
) (err error) {
|
2021-01-13 17:29:46 +00:00
|
|
|
if err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
|
|
|
return d.SendToDevice.DeleteSendToDeviceMessages(ctx, txn, userID, deviceID, before)
|
|
|
|
}); err != nil {
|
|
|
|
logrus.WithError(err).Errorf("Failed to clean up old send-to-device messages for user %q device %q", userID, deviceID)
|
|
|
|
return err
|
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 16:50:19 +00:00
|
|
|
}
|
2021-01-13 17:29:46 +00:00
|
|
|
return nil
|
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 16:50:19 +00:00
|
|
|
}
|
|
|
|
|
2020-05-14 15:11:37 +00:00
|
|
|
// getMembershipFromEvent returns the value of content.membership iff the event is a state event
|
|
|
|
// with type 'm.room.member' and state_key of userID. Otherwise, an empty string is returned.
|
|
|
|
func getMembershipFromEvent(ev *gomatrixserverlib.Event, userID string) string {
|
2020-08-25 20:04:35 +00:00
|
|
|
if ev.Type() != "m.room.member" || !ev.StateKeyEquals(userID) {
|
|
|
|
return ""
|
|
|
|
}
|
|
|
|
membership, err := ev.Membership()
|
|
|
|
if err != nil {
|
|
|
|
return ""
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
2020-08-25 20:04:35 +00:00
|
|
|
return membership
|
2020-05-14 15:11:37 +00:00
|
|
|
}
|
|
|
|
|
2020-11-09 18:46:11 +00:00
|
|
|
// StoreReceipt stores user receipts
|
|
|
|
func (d *Database) StoreReceipt(ctx context.Context, roomId, receiptType, userId, eventId string, timestamp gomatrixserverlib.Timestamp) (pos types.StreamPosition, err error) {
|
|
|
|
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
|
|
|
pos, err = d.Receipts.UpsertReceipt(ctx, txn, roomId, receiptType, userId, eventId, timestamp)
|
|
|
|
return err
|
|
|
|
})
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *Database) GetRoomReceipts(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) ([]eduAPI.OutputReceiptEvent, error) {
|
2020-12-18 11:11:21 +00:00
|
|
|
_, receipts, err := d.Receipts.SelectRoomReceiptsAfter(ctx, roomIDs, streamPos)
|
|
|
|
return receipts, err
|
2020-11-09 18:46:11 +00:00
|
|
|
}
|