diff options
Diffstat (limited to 'syncapi/storage/postgres')
| -rw-r--r-- | syncapi/storage/postgres/backwards_extremities_table.go | 106 | ||||
| -rw-r--r-- | syncapi/storage/postgres/current_room_state_table.go | 42 | ||||
| -rw-r--r-- | syncapi/storage/postgres/output_room_events_topology_table.go | 50 | ||||
| -rw-r--r-- | syncapi/storage/postgres/syncserver.go | 903 |
4 files changed, 169 insertions, 932 deletions
diff --git a/syncapi/storage/postgres/backwards_extremities_table.go b/syncapi/storage/postgres/backwards_extremities_table.go new file mode 100644 index 00000000..3dd1b234 --- /dev/null +++ b/syncapi/storage/postgres/backwards_extremities_table.go @@ -0,0 +1,106 @@ +// Copyright 2018 New Vector Ltd +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package postgres + +import ( + "context" + "database/sql" + + "github.com/matrix-org/dendrite/common" + "github.com/matrix-org/dendrite/syncapi/storage/tables" +) + +const backwardExtremitiesSchema = ` +-- Stores output room events received from the roomserver. +CREATE TABLE IF NOT EXISTS syncapi_backward_extremities ( + -- The 'room_id' key for the event. + room_id TEXT NOT NULL, + -- The event ID for the last known event. This is the backwards extremity. + event_id TEXT NOT NULL, + -- The prev_events for the last known event. This is used to update extremities. + prev_event_id TEXT NOT NULL, + PRIMARY KEY(room_id, event_id, prev_event_id) +); +` + +const insertBackwardExtremitySQL = "" + + "INSERT INTO syncapi_backward_extremities (room_id, event_id, prev_event_id)" + + " VALUES ($1, $2, $3)" + + " ON CONFLICT DO NOTHING" + +const selectBackwardExtremitiesForRoomSQL = "" + + "SELECT DISTINCT event_id FROM syncapi_backward_extremities WHERE room_id = $1" + +const deleteBackwardExtremitySQL = "" + + "DELETE FROM syncapi_backward_extremities WHERE room_id = $1 AND prev_event_id = $2" + +type backwardExtremitiesStatements struct { + insertBackwardExtremityStmt *sql.Stmt + selectBackwardExtremitiesForRoomStmt *sql.Stmt + deleteBackwardExtremityStmt *sql.Stmt +} + +func NewPostgresBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremities, error) { + s := &backwardExtremitiesStatements{} + _, err := db.Exec(backwardExtremitiesSchema) + if err != nil { + return nil, err + } + if s.insertBackwardExtremityStmt, err = db.Prepare(insertBackwardExtremitySQL); err != nil { + return nil, err + } + if s.selectBackwardExtremitiesForRoomStmt, err = db.Prepare(selectBackwardExtremitiesForRoomSQL); err != nil { + return nil, err + } + if s.deleteBackwardExtremityStmt, err = db.Prepare(deleteBackwardExtremitySQL); err != nil { + return nil, err + } + return s, nil +} + +func (s *backwardExtremitiesStatements) InsertsBackwardExtremity( + ctx context.Context, txn *sql.Tx, roomID, eventID string, prevEventID string, +) (err error) { + _, err = txn.Stmt(s.insertBackwardExtremityStmt).ExecContext(ctx, roomID, eventID, prevEventID) + return +} + +func (s *backwardExtremitiesStatements) SelectBackwardExtremitiesForRoom( + ctx context.Context, roomID string, +) (eventIDs []string, err error) { + rows, err := s.selectBackwardExtremitiesForRoomStmt.QueryContext(ctx, roomID) + if err != nil { + return + } + defer common.CloseAndLogIfError(ctx, rows, "selectBackwardExtremitiesForRoom: rows.close() failed") + + for rows.Next() { + var eID string + if err = rows.Scan(&eID); err != nil { + return + } + + eventIDs = append(eventIDs, eID) + } + + return eventIDs, rows.Err() +} + +func (s *backwardExtremitiesStatements) DeleteBackwardExtremity( + ctx context.Context, txn *sql.Tx, roomID, knownEventID string, +) (err error) { + _, err = txn.Stmt(s.deleteBackwardExtremityStmt).ExecContext(ctx, roomID, knownEventID) + return +} diff --git a/syncapi/storage/postgres/current_room_state_table.go b/syncapi/storage/postgres/current_room_state_table.go index ab8f07b2..b6ab7221 100644 --- a/syncapi/storage/postgres/current_room_state_table.go +++ b/syncapi/storage/postgres/current_room_state_table.go @@ -22,6 +22,7 @@ import ( "github.com/lib/pq" "github.com/matrix-org/dendrite/common" + "github.com/matrix-org/dendrite/syncapi/storage/tables" "github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/gomatrixserverlib" ) @@ -103,37 +104,38 @@ type currentRoomStateStatements struct { selectStateEventStmt *sql.Stmt } -func (s *currentRoomStateStatements) prepare(db *sql.DB) (err error) { - _, err = db.Exec(currentRoomStateSchema) +func NewPostgresCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, error) { + s := ¤tRoomStateStatements{} + _, err := db.Exec(currentRoomStateSchema) if err != nil { - return + return nil, err } if s.upsertRoomStateStmt, err = db.Prepare(upsertRoomStateSQL); err != nil { - return + return nil, err } if s.deleteRoomStateByEventIDStmt, err = db.Prepare(deleteRoomStateByEventIDSQL); err != nil { - return + return nil, err } if s.selectRoomIDsWithMembershipStmt, err = db.Prepare(selectRoomIDsWithMembershipSQL); err != nil { - return + return nil, err } if s.selectCurrentStateStmt, err = db.Prepare(selectCurrentStateSQL); err != nil { - return + return nil, err } if s.selectJoinedUsersStmt, err = db.Prepare(selectJoinedUsersSQL); err != nil { - return + return nil, err } if s.selectEventsWithEventIDsStmt, err = db.Prepare(selectEventsWithEventIDsSQL); err != nil { - return + return nil, err } if s.selectStateEventStmt, err = db.Prepare(selectStateEventSQL); err != nil { - return + return nil, err } - return + return s, nil } -// JoinedMemberLists returns a map of room ID to a list of joined user IDs. -func (s *currentRoomStateStatements) selectJoinedUsers( +// SelectJoinedUsers returns a map of room ID to a list of joined user IDs. +func (s *currentRoomStateStatements) SelectJoinedUsers( ctx context.Context, ) (map[string][]string, error) { rows, err := s.selectJoinedUsersStmt.QueryContext(ctx) @@ -157,7 +159,7 @@ func (s *currentRoomStateStatements) selectJoinedUsers( } // SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state. -func (s *currentRoomStateStatements) selectRoomIDsWithMembership( +func (s *currentRoomStateStatements) SelectRoomIDsWithMembership( ctx context.Context, txn *sql.Tx, userID string, @@ -181,8 +183,8 @@ func (s *currentRoomStateStatements) selectRoomIDsWithMembership( return result, rows.Err() } -// CurrentState returns all the current state events for the given room. -func (s *currentRoomStateStatements) selectCurrentState( +// SelectCurrentState returns all the current state events for the given room. +func (s *currentRoomStateStatements) SelectCurrentState( ctx context.Context, txn *sql.Tx, roomID string, stateFilter *gomatrixserverlib.StateFilter, ) ([]gomatrixserverlib.HeaderedEvent, error) { @@ -203,7 +205,7 @@ func (s *currentRoomStateStatements) selectCurrentState( return rowsToEvents(rows) } -func (s *currentRoomStateStatements) deleteRoomStateByEventID( +func (s *currentRoomStateStatements) DeleteRoomStateByEventID( ctx context.Context, txn *sql.Tx, eventID string, ) error { stmt := common.TxStmt(txn, s.deleteRoomStateByEventIDStmt) @@ -211,7 +213,7 @@ func (s *currentRoomStateStatements) deleteRoomStateByEventID( return err } -func (s *currentRoomStateStatements) upsertRoomState( +func (s *currentRoomStateStatements) UpsertRoomState( ctx context.Context, txn *sql.Tx, event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition, ) error { @@ -245,7 +247,7 @@ func (s *currentRoomStateStatements) upsertRoomState( return err } -func (s *currentRoomStateStatements) selectEventsWithEventIDs( +func (s *currentRoomStateStatements) SelectEventsWithEventIDs( ctx context.Context, txn *sql.Tx, eventIDs []string, ) ([]types.StreamEvent, error) { stmt := common.TxStmt(txn, s.selectEventsWithEventIDsStmt) @@ -274,7 +276,7 @@ func rowsToEvents(rows *sql.Rows) ([]gomatrixserverlib.HeaderedEvent, error) { return result, rows.Err() } -func (s *currentRoomStateStatements) selectStateEvent( +func (s *currentRoomStateStatements) SelectStateEvent( ctx context.Context, roomID, evType, stateKey string, ) (*gomatrixserverlib.HeaderedEvent, error) { stmt := s.selectStateEventStmt diff --git a/syncapi/storage/postgres/output_room_events_topology_table.go b/syncapi/storage/postgres/output_room_events_topology_table.go index 51cbd50d..48ebcf25 100644 --- a/syncapi/storage/postgres/output_room_events_topology_table.go +++ b/syncapi/storage/postgres/output_room_events_topology_table.go @@ -20,6 +20,7 @@ import ( "github.com/matrix-org/dendrite/common" + "github.com/matrix-org/dendrite/syncapi/storage/tables" "github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/gomatrixserverlib" ) @@ -84,36 +85,37 @@ type outputRoomEventsTopologyStatements struct { selectEventIDsFromPositionStmt *sql.Stmt } -func (s *outputRoomEventsTopologyStatements) prepare(db *sql.DB) (err error) { - _, err = db.Exec(outputRoomEventsTopologySchema) +func NewPostgresTopologyTable(db *sql.DB) (tables.Topology, error) { + s := &outputRoomEventsTopologyStatements{} + _, err := db.Exec(outputRoomEventsTopologySchema) if err != nil { - return + return nil, err } if s.insertEventInTopologyStmt, err = db.Prepare(insertEventInTopologySQL); err != nil { - return + return nil, err } if s.selectEventIDsInRangeASCStmt, err = db.Prepare(selectEventIDsInRangeASCSQL); err != nil { - return + return nil, err } if s.selectEventIDsInRangeDESCStmt, err = db.Prepare(selectEventIDsInRangeDESCSQL); err != nil { - return + return nil, err } if s.selectPositionInTopologyStmt, err = db.Prepare(selectPositionInTopologySQL); err != nil { - return + return nil, err } if s.selectMaxPositionInTopologyStmt, err = db.Prepare(selectMaxPositionInTopologySQL); err != nil { - return + return nil, err } if s.selectEventIDsFromPositionStmt, err = db.Prepare(selectEventIDsFromPositionSQL); err != nil { - return + return nil, err } - return + return s, nil } -// insertEventInTopology inserts the given event in the room's topology, based +// InsertEventInTopology inserts the given event in the room's topology, based // on the event's depth. -func (s *outputRoomEventsTopologyStatements) insertEventInTopology( - ctx context.Context, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition, +func (s *outputRoomEventsTopologyStatements) InsertEventInTopology( + ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition, ) (err error) { _, err = s.insertEventInTopologyStmt.ExecContext( ctx, event.EventID(), event.Depth(), event.RoomID(), pos, @@ -121,11 +123,11 @@ func (s *outputRoomEventsTopologyStatements) insertEventInTopology( return } -// selectEventIDsInRange selects the IDs of events which positions are within a +// SelectEventIDsInRange selects the IDs of events which positions are within a // given range in a given room's topological order. // Returns an empty slice if no events match the given range. -func (s *outputRoomEventsTopologyStatements) selectEventIDsInRange( - ctx context.Context, roomID string, fromPos, toPos, toMicroPos types.StreamPosition, +func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange( + ctx context.Context, txn *sql.Tx, roomID string, fromPos, toPos, toMicroPos types.StreamPosition, limit int, chronologicalOrder bool, ) (eventIDs []string, err error) { // Decide on the selection's order according to whether chronological order @@ -159,26 +161,26 @@ func (s *outputRoomEventsTopologyStatements) selectEventIDsInRange( return eventIDs, rows.Err() } -// selectPositionInTopology returns the position of a given event in the +// SelectPositionInTopology returns the position of a given event in the // topology of the room it belongs to. -func (s *outputRoomEventsTopologyStatements) selectPositionInTopology( - ctx context.Context, eventID string, +func (s *outputRoomEventsTopologyStatements) SelectPositionInTopology( + ctx context.Context, txn *sql.Tx, eventID string, ) (pos, spos types.StreamPosition, err error) { err = s.selectPositionInTopologyStmt.QueryRowContext(ctx, eventID).Scan(&pos, &spos) return } -func (s *outputRoomEventsTopologyStatements) selectMaxPositionInTopology( - ctx context.Context, roomID string, +func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology( + ctx context.Context, txn *sql.Tx, roomID string, ) (pos types.StreamPosition, spos types.StreamPosition, err error) { err = s.selectMaxPositionInTopologyStmt.QueryRowContext(ctx, roomID).Scan(&pos, &spos) return } -// selectEventIDsFromPosition returns the IDs of all events that have a given +// SelectEventIDsFromPosition returns the IDs of all events that have a given // position in the topology of a given room. -func (s *outputRoomEventsTopologyStatements) selectEventIDsFromPosition( - ctx context.Context, roomID string, pos types.StreamPosition, +func (s *outputRoomEventsTopologyStatements) SelectEventIDsFromPosition( + ctx context.Context, txn *sql.Tx, roomID string, pos types.StreamPosition, ) (eventIDs []string, err error) { // Query the event IDs. rows, err := s.selectEventIDsFromPositionStmt.QueryContext(ctx, roomID, pos) diff --git a/syncapi/storage/postgres/syncserver.go b/syncapi/storage/postgres/syncserver.go index 4fa08ce5..d44f8b7e 100644 --- a/syncapi/storage/postgres/syncserver.go +++ b/syncapi/storage/postgres/syncserver.go @@ -16,47 +16,27 @@ package postgres import ( - "context" "database/sql" - "encoding/json" - "fmt" - "github.com/matrix-org/dendrite/clientapi/auth/authtypes" "github.com/matrix-org/dendrite/internal/sqlutil" - "github.com/matrix-org/dendrite/roomserver/api" // Import the postgres database driver. _ "github.com/lib/pq" "github.com/matrix-org/dendrite/common" "github.com/matrix-org/dendrite/eduserver/cache" "github.com/matrix-org/dendrite/syncapi/storage/shared" - "github.com/matrix-org/dendrite/syncapi/storage/tables" - "github.com/matrix-org/dendrite/syncapi/types" - "github.com/matrix-org/gomatrixserverlib" ) -type stateDelta struct { - roomID string - stateEvents []gomatrixserverlib.HeaderedEvent - membership string - // The PDU stream position of the latest membership event for this user, if applicable. - // Can be 0 if there is no membership event in this delta. - membershipPos types.StreamPosition -} - // SyncServerDatasource represents a sync server datasource which manages // both the database for PDUs and caches for EDUs. type SyncServerDatasource struct { shared.Database db *sql.DB common.PartitionOffsetStatements - roomstate currentRoomStateStatements - topology outputRoomEventsTopologyStatements - backwardExtremities tables.BackwardsExtremities } -// NewSyncServerDatasource creates a new sync server database -func NewSyncServerDatasource(dbDataSourceName string, dbProperties common.DbProperties) (*SyncServerDatasource, error) { +// NewDatabase creates a new sync server database +func NewDatabase(dbDataSourceName string, dbProperties common.DbProperties) (*SyncServerDatasource, error) { var d SyncServerDatasource var err error if d.db, err = sqlutil.Open("postgres", dbDataSourceName, dbProperties); err != nil { @@ -73,884 +53,31 @@ func NewSyncServerDatasource(dbDataSourceName string, dbProperties common.DbProp if err != nil { return nil, err } - if err = d.roomstate.prepare(d.db); err != nil { + currState, err := NewPostgresCurrentRoomStateTable(d.db) + if err != nil { return nil, err } invites, err := NewPostgresInvitesTable(d.db) if err != nil { return nil, err } - if err = d.topology.prepare(d.db); err != nil { + topology, err := NewPostgresTopologyTable(d.db) + if err != nil { return nil, err } - d.backwardExtremities, err = tables.NewBackwardsExtremities(d.db, &tables.PostgresBackwardsExtremitiesStatements{}) + backwardExtremities, err := NewPostgresBackwardsExtremitiesTable(d.db) if err != nil { return nil, err } d.Database = shared.Database{ - DB: d.db, - Invites: invites, - AccountData: accountData, - OutputEvents: events, - EDUCache: cache.New(), + DB: d.db, + Invites: invites, + AccountData: accountData, + OutputEvents: events, + Topology: topology, + CurrentRoomState: currState, + BackwardExtremities: backwardExtremities, + EDUCache: cache.New(), } return &d, nil } - -func (d *SyncServerDatasource) AllJoinedUsersInRooms(ctx context.Context) (map[string][]string, error) { - return d.roomstate.selectJoinedUsers(ctx) -} - -// 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. -func (d *SyncServerDatasource) 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.Database.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 -} - -func (d *SyncServerDatasource) WriteEvent( - ctx context.Context, - ev *gomatrixserverlib.HeaderedEvent, - addStateEvents []gomatrixserverlib.HeaderedEvent, - addStateEventIDs, removeStateEventIDs []string, - transactionID *api.TransactionID, excludeFromSync bool, -) (pduPosition types.StreamPosition, returnErr error) { - returnErr = common.WithTransaction(d.db, func(txn *sql.Tx) error { - var err error - pos, err := d.Database.OutputEvents.InsertEvent( - ctx, txn, ev, addStateEventIDs, removeStateEventIDs, transactionID, excludeFromSync, - ) - if err != nil { - return err - } - pduPosition = pos - - if err = d.topology.insertEventInTopology(ctx, ev, pos); err != nil { - return err - } - - if err = d.handleBackwardExtremities(ctx, txn, ev); err != nil { - return err - } - - if len(addStateEvents) == 0 && len(removeStateEventIDs) == 0 { - // Nothing to do, the event may have just been a message event. - return nil - } - - return d.updateRoomState(ctx, txn, removeStateEventIDs, addStateEvents, pduPosition) - }) - - return pduPosition, returnErr -} - -func (d *SyncServerDatasource) updateRoomState( - ctx context.Context, txn *sql.Tx, - removedEventIDs []string, - addedEvents []gomatrixserverlib.HeaderedEvent, - pduPosition types.StreamPosition, -) 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.roomstate.deleteRoomStateByEventID(ctx, txn, eventID); err != nil { - return err - } - } - - 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 { - return err - } - membership = &value - } - if err := d.roomstate.upsertRoomState(ctx, txn, event, membership, pduPosition); err != nil { - return err - } - } - - return nil -} - -func (d *SyncServerDatasource) GetStateEvent( - ctx context.Context, roomID, evType, stateKey string, -) (*gomatrixserverlib.HeaderedEvent, error) { - return d.roomstate.selectStateEvent(ctx, roomID, evType, stateKey) -} - -func (d *SyncServerDatasource) GetStateEventsForRoom( - ctx context.Context, roomID string, stateFilter *gomatrixserverlib.StateFilter, -) (stateEvents []gomatrixserverlib.HeaderedEvent, err error) { - err = common.WithTransaction(d.db, func(txn *sql.Tx) error { - stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID, stateFilter) - return err - }) - return -} - -func (d *SyncServerDatasource) GetEventsInTopologicalRange( - ctx context.Context, - from, to *types.TopologyToken, - roomID string, limit int, - backwardOrdering bool, -) (events []types.StreamEvent, err error) { - // Determine the backward and forward limit, i.e. the upper and lower - // limits to the selection in the room's topology, from the direction. - var backwardLimit, forwardLimit, forwardMicroLimit types.StreamPosition - if backwardOrdering { - // Backward ordering is antichronological (latest event to oldest - // one). - backwardLimit = to.Depth() - forwardLimit = from.Depth() - forwardMicroLimit = from.PDUPosition() - } else { - // Forward ordering is chronological (oldest event to latest one). - backwardLimit = from.Depth() - forwardLimit = to.Depth() - } - - // Select the event IDs from the defined range. - var eIDs []string - eIDs, err = d.topology.selectEventIDsInRange( - ctx, roomID, backwardLimit, forwardLimit, forwardMicroLimit, limit, !backwardOrdering, - ) - if err != nil { - return - } - - // Retrieve the events' contents using their IDs. - events, err = d.Database.OutputEvents.SelectEvents(ctx, nil, eIDs) - return -} - -func (d *SyncServerDatasource) SyncPosition(ctx context.Context) (types.StreamingToken, error) { - return d.syncPositionTx(ctx, nil) -} - -func (d *SyncServerDatasource) BackwardExtremitiesForRoom( - ctx context.Context, roomID string, -) (backwardExtremities []string, err error) { - return d.backwardExtremities.SelectBackwardExtremitiesForRoom(ctx, roomID) -} - -func (d *SyncServerDatasource) MaxTopologicalPosition( - ctx context.Context, roomID string, -) (depth types.StreamPosition, stream types.StreamPosition, err error) { - return d.topology.selectMaxPositionInTopology(ctx, roomID) -} - -func (d *SyncServerDatasource) EventsAtTopologicalPosition( - ctx context.Context, roomID string, pos types.StreamPosition, -) ([]types.StreamEvent, error) { - eIDs, err := d.topology.selectEventIDsFromPosition(ctx, roomID, pos) - if err != nil { - return nil, err - } - - return d.Database.OutputEvents.SelectEvents(ctx, nil, eIDs) -} - -func (d *SyncServerDatasource) EventPositionInTopology( - ctx context.Context, eventID string, -) (depth types.StreamPosition, stream types.StreamPosition, err error) { - return d.topology.selectPositionInTopology(ctx, eventID) -} - -func (d *SyncServerDatasource) syncPositionTx( - ctx context.Context, txn *sql.Tx, -) (sp types.StreamingToken, err error) { - - maxEventID, err := d.Database.OutputEvents.SelectMaxEventID(ctx, txn) - if err != nil { - return sp, err - } - maxAccountDataID, err := d.Database.AccountData.SelectMaxAccountDataID(ctx, txn) - if err != nil { - return sp, err - } - if maxAccountDataID > maxEventID { - maxEventID = maxAccountDataID - } - maxInviteID, err := d.Database.Invites.SelectMaxInviteID(ctx, txn) - if err != nil { - return sp, err - } - if maxInviteID > maxEventID { - maxEventID = maxInviteID - } - sp = types.NewStreamToken(types.StreamPosition(maxEventID), types.StreamPosition(d.Database.EDUCache.GetLatestSyncPosition())) - return -} - -// addPDUDeltaToResponse adds all PDU deltas to a sync response. -// IDs of all rooms the user joined are returned so EDU deltas can be added for them. -func (d *SyncServerDatasource) addPDUDeltaToResponse( - ctx context.Context, - device authtypes.Device, - fromPos, toPos types.StreamPosition, - numRecentEventsPerRoom int, - wantFullState bool, - res *types.Response, -) (joinedRoomIDs []string, err error) { - txn, err := d.db.BeginTx(ctx, &txReadOnlySnapshot) - if err != nil { - return nil, err - } - var succeeded bool - defer func() { - txerr := common.EndTransaction(txn, &succeeded) - if err == nil && txerr != nil { - err = txerr - } - }() - - stateFilter := gomatrixserverlib.DefaultStateFilter() // TODO: use filter provided in request - - // Work out which rooms to return in the response. This is done by getting not only the currently - // joined rooms, but also which rooms have membership transitions for this user between the 2 PDU stream positions. - // This works out what the 'state' key should be for each room as well as which membership block - // to put the room into. - var deltas []stateDelta - if !wantFullState { - deltas, joinedRoomIDs, err = d.getStateDeltas( - ctx, &device, txn, fromPos, toPos, device.UserID, &stateFilter, - ) - } else { - deltas, joinedRoomIDs, err = d.getStateDeltasForFullStateSync( - ctx, &device, txn, fromPos, toPos, device.UserID, &stateFilter, - ) - } - if err != nil { - return nil, err - } - - for _, delta := range deltas { - err = d.addRoomDeltaToResponse(ctx, &device, txn, fromPos, toPos, delta, numRecentEventsPerRoom, res) - if err != nil { - return nil, err - } - } - - // TODO: This should be done in getStateDeltas - if err = d.addInvitesToResponse(ctx, txn, device.UserID, fromPos, toPos, res); err != nil { - return nil, err - } - - succeeded = true - return joinedRoomIDs, nil -} - -// addTypingDeltaToResponse adds all typing notifications to a sync response -// since the specified position. -func (d *SyncServerDatasource) addTypingDeltaToResponse( - since types.StreamingToken, - joinedRoomIDs []string, - res *types.Response, -) error { - var jr types.JoinResponse - var ok bool - var err error - for _, roomID := range joinedRoomIDs { - if typingUsers, updated := d.Database.EDUCache.GetTypingUsersIfUpdatedAfter( - roomID, int64(since.EDUPosition()), - ); updated { - ev := gomatrixserverlib.ClientEvent{ - Type: gomatrixserverlib.MTyping, - } - ev.Content, err = json.Marshal(map[string]interface{}{ - "user_ids": typingUsers, - }) - if err != nil { - return err - } - - if jr, ok = res.Rooms.Join[roomID]; !ok { - jr = *types.NewJoinResponse() - } - jr.Ephemeral.Events = append(jr.Ephemeral.Events, ev) - res.Rooms.Join[roomID] = jr - } - } - return nil -} - -// addEDUDeltaToResponse adds updates for EDUs of each type since fromPos if -// the positions of that type are not equal in fromPos and toPos. -func (d *SyncServerDatasource) addEDUDeltaToResponse( - fromPos, toPos types.StreamingToken, - joinedRoomIDs []string, - res *types.Response, -) (err error) { - - if fromPos.EDUPosition() != toPos.EDUPosition() { - err = d.addTypingDeltaToResponse( - fromPos, joinedRoomIDs, res, - ) - } - - return -} - -func (d *SyncServerDatasource) IncrementalSync( - ctx context.Context, - device authtypes.Device, - fromPos, toPos types.StreamingToken, - numRecentEventsPerRoom int, - wantFullState bool, -) (*types.Response, error) { - nextBatchPos := fromPos.WithUpdates(toPos) - res := types.NewResponse(nextBatchPos) - - var joinedRoomIDs []string - var err error - if fromPos.PDUPosition() != toPos.PDUPosition() || wantFullState { - joinedRoomIDs, err = d.addPDUDeltaToResponse( - ctx, device, fromPos.PDUPosition(), toPos.PDUPosition(), numRecentEventsPerRoom, wantFullState, res, - ) - } else { - joinedRoomIDs, err = d.roomstate.selectRoomIDsWithMembership( - ctx, nil, device.UserID, gomatrixserverlib.Join, - ) - } - if err != nil { - return nil, err - } - - err = d.addEDUDeltaToResponse( - fromPos, toPos, joinedRoomIDs, res, - ) - if err != nil { - return nil, err - } - - return res, nil -} - -// getResponseWithPDUsForCompleteSync creates a response and adds all PDUs needed -// to it. It returns toPos and joinedRoomIDs for use of adding EDUs. -func (d *SyncServerDatasource) getResponseWithPDUsForCompleteSync( - ctx context.Context, - userID string, - numRecentEventsPerRoom int, -) ( - res *types.Response, - toPos types.StreamingToken, - joinedRoomIDs []string, - err error, -) { - // This needs to be all done in a transaction as we need to do multiple SELECTs, and we need to have - // a consistent view of the database throughout. This includes extracting the sync position. - // This does have the unfortunate side-effect that all the matrixy logic resides in this function, - // but it's better to not hide the fact that this is being done in a transaction. - txn, err := d.db.BeginTx(ctx, &txReadOnlySnapshot) - if err != nil { - return - } - var succeeded bool - defer func() { - txerr := common.EndTransaction(txn, &succeeded) - if err == nil && txerr != nil { - err = txerr - } - }() - - // Get the current sync position which we will base the sync response on. - toPos, err = d.syncPositionTx(ctx, txn) - if err != nil { - return - } - - res = types.NewResponse(toPos) - - // Extract room state and recent events for all rooms the user is joined to. - joinedRoomIDs, err = d.roomstate.selectRoomIDsWithMembership(ctx, txn, userID, gomatrixserverlib.Join) - if err != nil { - return - } - - stateFilter := gomatrixserverlib.DefaultStateFilter() // TODO: use filter provided in request - - // Build up a /sync response. Add joined rooms. - for _, roomID := range joinedRoomIDs { - var stateEvents []gomatrixserverlib.HeaderedEvent - stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID, &stateFilter) - if err != nil { - return - } - // TODO: When filters are added, we may need to call this multiple times to get enough events. - // See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316 - var recentStreamEvents []types.StreamEvent - recentStreamEvents, err = d.Database.OutputEvents.SelectRecentEvents( - ctx, txn, roomID, types.StreamPosition(0), toPos.PDUPosition(), - numRecentEventsPerRoom, true, true, - ) - if err != nil { - return - } - - // Retrieve the backward topology position, i.e. the position of the - // oldest event in the room's topology. - var prevBatchStr string - if len(recentStreamEvents) > 0 { - var backwardTopologyPos, backwardStreamPos types.StreamPosition - backwardTopologyPos, backwardStreamPos, err = d.topology.selectPositionInTopology(ctx, recentStreamEvents[0].EventID()) - if err != nil { - return - } - prevBatch := types.NewTopologyToken(backwardTopologyPos, backwardStreamPos) - prevBatch.Decrement() - prevBatchStr = prevBatch.String() - } - - // We don't include a device here as we don't need to send down - // transaction IDs for complete syncs - recentEvents := d.StreamEventsToEvents(nil, recentStreamEvents) - stateEvents = removeDuplicates(stateEvents, recentEvents) - jr := types.NewJoinResponse() - jr.Timeline.PrevBatch = prevBatchStr - jr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(recentEvents, gomatrixserverlib.FormatSync) - jr.Timeline.Limited = true - jr.State.Events = gomatrixserverlib.HeaderedToClientEvents(stateEvents, gomatrixserverlib.FormatSync) - res.Rooms.Join[roomID] = *jr - } - - if err = d.addInvitesToResponse(ctx, txn, userID, 0, toPos.PDUPosition(), res); err != nil { - return - } - - succeeded = true - return res, toPos, joinedRoomIDs, err -} - -func (d *SyncServerDatasource) CompleteSync( - ctx context.Context, userID string, numRecentEventsPerRoom int, -) (*types.Response, error) { - res, toPos, joinedRoomIDs, err := d.getResponseWithPDUsForCompleteSync( - ctx, userID, numRecentEventsPerRoom, - ) - if err != nil { - return nil, err - } - - // Use a zero value SyncPosition for fromPos so all EDU states are added. - err = d.addEDUDeltaToResponse( - types.NewStreamToken(0, 0), toPos, joinedRoomIDs, res, - ) - if err != nil { - return nil, err - } - - return res, nil -} - -var txReadOnlySnapshot = 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 *SyncServerDatasource) addInvitesToResponse( - ctx context.Context, txn *sql.Tx, - userID string, - fromPos, toPos types.StreamPosition, - res *types.Response, -) error { - invites, err := d.Database.Invites.SelectInviteEventsInRange( - ctx, txn, userID, fromPos, toPos, - ) - if err != nil { - return err - } - for roomID, inviteEvent := range invites { - ir := types.NewInviteResponse(inviteEvent) - res.Rooms.Invite[roomID] = *ir - } - return nil -} - -// Retrieve the backward topology position, i.e. the position of the -// oldest event in the room's topology. -func (d *SyncServerDatasource) getBackwardTopologyPos( - ctx context.Context, - events []types.StreamEvent, -) (pos, spos types.StreamPosition) { - if len(events) > 0 { - pos, spos, _ = d.topology.selectPositionInTopology(ctx, events[0].EventID()) - } - if pos-1 <= 0 { - pos = types.StreamPosition(1) - } else { - pos = pos - 1 - } - return -} - -// addRoomDeltaToResponse adds a room state delta to a sync response -func (d *SyncServerDatasource) addRoomDeltaToResponse( - ctx context.Context, - device *authtypes.Device, - txn *sql.Tx, - fromPos, toPos types.StreamPosition, - delta stateDelta, - numRecentEventsPerRoom int, - res *types.Response, -) error { - endPos := toPos - if delta.membershipPos > 0 && delta.membership == gomatrixserverlib.Leave { - // make sure we don't leak recent events after the leave event. - // TODO: History visibility makes this somewhat complex to handle correctly. For example: - // TODO: This doesn't work for join -> leave in a single /sync request (see events prior to join). - // TODO: This will fail on join -> leave -> sensitive msg -> join -> leave - // in a single /sync request - // This is all "okay" assuming history_visibility == "shared" which it is by default. - endPos = delta.membershipPos - } - recentStreamEvents, err := d.Database.OutputEvents.SelectRecentEvents( - ctx, txn, delta.roomID, types.StreamPosition(fromPos), types.StreamPosition(endPos), - numRecentEventsPerRoom, true, true, - ) - if err != nil { - return err - } - recentEvents := d.StreamEventsToEvents(device, recentStreamEvents) - delta.stateEvents = removeDuplicates(delta.stateEvents, recentEvents) // roll back - backwardTopologyPos, backwardStreamPos := d.getBackwardTopologyPos(ctx, recentStreamEvents) - prevBatch := types.NewTopologyToken( - backwardTopologyPos, backwardStreamPos, - ) - - switch delta.membership { - case gomatrixserverlib.Join: - jr := types.NewJoinResponse() - - jr.Timeline.PrevBatch = prevBatch.String() - jr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(recentEvents, gomatrixserverlib.FormatSync) - jr.Timeline.Limited = false // TODO: if len(events) >= numRecents + 1 and then set limited:true - jr.State.Events = gomatrixserverlib.HeaderedToClientEvents(delta.stateEvents, gomatrixserverlib.FormatSync) - res.Rooms.Join[delta.roomID] = *jr - case gomatrixserverlib.Leave: - fallthrough // transitions to leave are the same as ban - case gomatrixserverlib.Ban: - // TODO: recentEvents may contain events that this user is not allowed to see because they are - // no longer in the room. - lr := types.NewLeaveResponse() - lr.Timeline.PrevBatch = prevBatch.String() - lr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(recentEvents, gomatrixserverlib.FormatSync) - lr.Timeline.Limited = false // TODO: if len(events) >= numRecents + 1 and then set limited:true - lr.State.Events = gomatrixserverlib.HeaderedToClientEvents(delta.stateEvents, gomatrixserverlib.FormatSync) - res.Rooms.Leave[delta.roomID] = *lr - } - - return 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 *SyncServerDatasource) 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 *SyncServerDatasource) 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.Database.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.roomstate.selectEventsWithEventIDs(ctx, txn, missing) - - if err != nil { - return nil, err - } - if len(stateEvents) != len(missing) { - return nil, fmt.Errorf("failed to map all event IDs to events: (got %d, wanted %d)", len(stateEvents), len(missing)) - } - 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. -func (d *SyncServerDatasource) getStateDeltas( - ctx context.Context, device *authtypes.Device, txn *sql.Tx, - fromPos, toPos types.StreamPosition, userID string, - stateFilter *gomatrixserverlib.StateFilter, -) ([]stateDelta, []string, error) { - // 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. - var deltas []stateDelta - - // get all the state events ever between these two positions - stateNeeded, eventMap, err := d.Database.OutputEvents.SelectStateInRange(ctx, txn, fromPos, toPos, stateFilter) - 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 { - // 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. - if membership := getMembershipFromEvent(&ev.Event, userID); membership != "" { - 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 - } - - deltas = append(deltas, stateDelta{ - membership: membership, - membershipPos: ev.StreamPosition, - stateEvents: d.StreamEventsToEvents(device, stateStreamEvents), - roomID: roomID, - }) - break - } - } - } - - // Add in currently joined rooms - joinedRoomIDs, err := d.roomstate.selectRoomIDsWithMembership(ctx, txn, userID, gomatrixserverlib.Join) - if err != nil { - return nil, nil, err - } - for _, joinedRoomID := range joinedRoomIDs { - deltas = append(deltas, stateDelta{ - membership: gomatrixserverlib.Join, - stateEvents: d.StreamEventsToEvents(device, state[joinedRoomID]), - roomID: joinedRoomID, - }) - } - - 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. -func (d *SyncServerDatasource) getStateDeltasForFullStateSync( - ctx context.Context, device *authtypes.Device, txn *sql.Tx, - fromPos, toPos types.StreamPosition, userID string, - stateFilter *gomatrixserverlib.StateFilter, -) ([]stateDelta, []string, error) { - joinedRoomIDs, err := d.roomstate.selectRoomIDsWithMembership(ctx, txn, userID, gomatrixserverlib.Join) - if err != nil { - return nil, nil, err - } - - // Use a reasonable initial capacity - deltas := make([]stateDelta, 0, len(joinedRoomIDs)) - - // 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 - } - deltas = append(deltas, stateDelta{ - membership: gomatrixserverlib.Join, - stateEvents: d.StreamEventsToEvents(device, s), - roomID: joinedRoomID, - }) - } - - // Get all the state events ever between these two positions - stateNeeded, eventMap, err := d.Database.OutputEvents.SelectStateInRange(ctx, txn, fromPos, toPos, stateFilter) - 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 { - if membership := getMembershipFromEvent(&ev.Event, userID); membership != "" { - if membership != gomatrixserverlib.Join { // We've already added full state for all joined rooms above. - deltas = append(deltas, stateDelta{ - membership: membership, - membershipPos: ev.StreamPosition, - stateEvents: d.StreamEventsToEvents(device, stateStreamEvents), - roomID: roomID, - }) - } - - break - } - } - } - - return deltas, joinedRoomIDs, nil -} - -func (d *SyncServerDatasource) currentStateStreamEventsForRoom( - ctx context.Context, txn *sql.Tx, roomID string, - stateFilter *gomatrixserverlib.StateFilter, -) ([]types.StreamEvent, error) { - allState, err := d.roomstate.selectCurrentState(ctx, txn, roomID, stateFilter) - 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 -} - -// There may be some overlap where events in stateEvents are already in recentEvents, so filter -// them out so we don't include them twice in the /sync response. They should be in recentEvents -// only, so clients get to the correct state once they have rolled forward. -func removeDuplicates(stateEvents, recentEvents []gomatrixserverlib.HeaderedEvent) []gomatrixserverlib.HeaderedEvent { - for _, recentEv := range recentEvents { - if recentEv.StateKey() == nil { - continue // not a state event - } - // TODO: This is a linear scan over all the current state events in this room. This will - // be slow for big rooms. We should instead sort the state events by event ID (ORDER BY) - // then do a binary search to find matching events, similar to what roomserver does. - for j := 0; j < len(stateEvents); j++ { - if stateEvents[j].EventID() == recentEv.EventID() { - // overwrite the element to remove with the last element then pop the last element. - // This is orders of magnitude faster than re-slicing, but doesn't preserve ordering - // (we don't care about the order of stateEvents) - stateEvents[j] = stateEvents[len(stateEvents)-1] - stateEvents = stateEvents[:len(stateEvents)-1] - break // there shouldn't be multiple events with the same event ID - } - } - } - return stateEvents -} - -// 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 { - if ev.Type() == "m.room.member" && ev.StateKeyEquals(userID) { - membership, err := ev.Membership() - if err != nil { - return "" - } - return membership - } - return "" -} |