diff --git a/src/github.com/matrix-org/dendrite/roomserver/api/output.go b/src/github.com/matrix-org/dendrite/roomserver/api/output.go index f1b40231..953fe3c8 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/api/output.go +++ b/src/github.com/matrix-org/dendrite/roomserver/api/output.go @@ -21,8 +21,14 @@ import ( // An OutputType is a type of roomserver output. type OutputType string -// OutputTypeNewRoomEvent indicates that the event is an OutputNewRoomEvent -const OutputTypeNewRoomEvent OutputType = "new_room_event" +const ( + // OutputTypeNewRoomEvent indicates that the event is an OutputNewRoomEvent + OutputTypeNewRoomEvent OutputType = "new_room_event" + // OutputTypeNewInviteEvent indicates that the event is an OutputNewInviteEvent + OutputTypeNewInviteEvent OutputType = "new_invite_event" + // OutputTypeRetireInviteEvent indicates that the event is an OutputRetireInviteEvent + OutputTypeRetireInviteEvent OutputType = "retire_invite_event" +) // An OutputEvent is an entry in the roomserver output kafka log. // Consumers should check the type field when consuming this event. @@ -31,6 +37,10 @@ type OutputEvent struct { Type OutputType `json:"type"` // The content of event with type OutputTypeNewRoomEvent NewRoomEvent *OutputNewRoomEvent `json:"new_room_event,omitempty"` + // The content of event with type OutputTypeNewInviteEvent + NewInviteEvent *OutputNewInviteEvent `json:"new_invite_event,omitempty"` + // The content of event with type OutputTypeRetireInviteEvent + RetireInviteEvent *OutputRetireInviteEvent `json:"retire_invite_event,omitempty"` } // An OutputNewRoomEvent is written when the roomserver receives a new event. @@ -98,3 +108,26 @@ type OutputNewRoomEvent struct { // future proof the API for virtual hosting. SendAsServer string `json:"send_as_server"` } + +// An OutputNewInviteEvent is written whenever an invite becomes active. +// Invite events can be received outside of an existing room so have to be +// tracked separately from the room events themselves. +type OutputNewInviteEvent struct { + // The "m.room.member" invite event. + Event gomatrixserverlib.Event `json:"event"` +} + +// An OutputRetireInviteEvent is written whenever an existing invite is no longer +// active. An invite stops being active if the user joins the room or if the +// invite is rejected by the user. +type OutputRetireInviteEvent struct { + // The ID of the "m.room.member" invite event. + EventID string + // Optional event ID of the event that replaced the invite. + // This can be empty if the invite was rejected locally and we were unable + // to reach the server that originally sent the invite. + RetiredByEventID string + // The "membership" of the user after retiring the invite. One of "join" + // "leave" or "ban". + Membership string +} diff --git a/src/github.com/matrix-org/dendrite/roomserver/input/events.go b/src/github.com/matrix-org/dendrite/roomserver/input/events.go index f8acff47..c1eee4c9 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/input/events.go +++ b/src/github.com/matrix-org/dendrite/roomserver/input/events.go @@ -43,8 +43,8 @@ type RoomEventDatabase interface { // OutputRoomEventWriter has the APIs needed to write an event to the output logs. type OutputRoomEventWriter interface { - // Write an event. - WriteOutputRoomEvent(output api.OutputNewRoomEvent) error + // Write a list of events for a room + WriteOutputEvents(roomID string, updates []api.OutputEvent) error } func processRoomEvent(db RoomEventDatabase, ow OutputRoomEventWriter, input api.InputRoomEvent) error { diff --git a/src/github.com/matrix-org/dendrite/roomserver/input/input.go b/src/github.com/matrix-org/dendrite/roomserver/input/input.go index ffbebd0c..c8ac58d3 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/input/input.go +++ b/src/github.com/matrix-org/dendrite/roomserver/input/input.go @@ -46,22 +46,21 @@ type RoomserverInputAPI struct { processed int64 } -// WriteOutputRoomEvent implements OutputRoomEventWriter -func (r *RoomserverInputAPI) WriteOutputRoomEvent(output api.OutputNewRoomEvent) error { - var m sarama.ProducerMessage - oe := api.OutputEvent{ - Type: api.OutputTypeNewRoomEvent, - NewRoomEvent: &output, +// WriteOutputEvents implements OutputRoomEventWriter +func (r *RoomserverInputAPI) WriteOutputEvents(roomID string, updates []api.OutputEvent) error { + messages := make([]*sarama.ProducerMessage, len(updates)) + for i := range updates { + value, err := json.Marshal(updates[i]) + if err != nil { + return err + } + messages[i] = &sarama.ProducerMessage{ + Topic: r.OutputRoomEventTopic, + Key: sarama.StringEncoder(roomID), + Value: sarama.ByteEncoder(value), + } } - value, err := json.Marshal(oe) - if err != nil { - return err - } - m.Topic = r.OutputRoomEventTopic - m.Key = sarama.StringEncoder("") - m.Value = sarama.ByteEncoder(value) - _, _, err = r.Producer.SendMessage(&m) - return err + return r.Producer.SendMessages(messages) } // InputRoomEvents implements api.RoomserverInputAPI diff --git a/src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go b/src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go index 6b5f3967..9328ecf3 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go +++ b/src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go @@ -66,69 +66,88 @@ func updateLatestEvents( } }() - err = doUpdateLatestEvents(db, updater, ow, roomNID, stateAtEvent, event, sendAsServer) - return + u := latestEventsUpdater{ + db: db, updater: updater, ow: ow, roomNID: roomNID, + stateAtEvent: stateAtEvent, event: event, sendAsServer: sendAsServer, + } + return u.doUpdateLatestEvents() } -func doUpdateLatestEvents( - db RoomEventDatabase, - updater types.RoomRecentEventsUpdater, - ow OutputRoomEventWriter, - roomNID types.RoomNID, - stateAtEvent types.StateAtEvent, - event gomatrixserverlib.Event, - sendAsServer string, -) error { +// latestEventsUpdater tracks the state used to update the latest events in the +// room. It mostly just ferries state between the various function calls. +// The state could be passed using function arguments, but it becomes impractical +// when there are so many variables to pass around. +type latestEventsUpdater struct { + db RoomEventDatabase + updater types.RoomRecentEventsUpdater + ow OutputRoomEventWriter + roomNID types.RoomNID + stateAtEvent types.StateAtEvent + event gomatrixserverlib.Event + // Which server to send this event as. + sendAsServer string + // The eventID of the event that was processed before this one. + lastEventIDSent string + // The latest events in the room after processing this event. + latest []types.StateAtEventAndReference + // The state entries removed from and added to the current state of the + // room as a result of processing this event. They are sorted lists. + removed []types.StateEntry + added []types.StateEntry + // The state entries that are removed and added to recover the state before + // the event being processed. They are sorted lists. + stateBeforeEventRemoves []types.StateEntry + stateBeforeEventAdds []types.StateEntry + // The snapshots of current state before and after processing this event + oldStateNID types.StateSnapshotNID + newStateNID types.StateSnapshotNID +} + +func (u *latestEventsUpdater) doUpdateLatestEvents() error { var err error var prevEvents []gomatrixserverlib.EventReference - prevEvents = event.PrevEvents() - oldLatest := updater.LatestEvents() - lastEventIDSent := updater.LastEventIDSent() - oldStateNID := updater.CurrentStateSnapshotNID() + prevEvents = u.event.PrevEvents() + oldLatest := u.updater.LatestEvents() + u.lastEventIDSent = u.updater.LastEventIDSent() + u.oldStateNID = u.updater.CurrentStateSnapshotNID() - if hasBeenSent, err := updater.HasEventBeenSent(stateAtEvent.EventNID); err != nil { + if hasBeenSent, err := u.updater.HasEventBeenSent(u.stateAtEvent.EventNID); err != nil { return err } else if hasBeenSent { // Already sent this event so we can stop processing return nil } - if err = updater.StorePreviousEvents(stateAtEvent.EventNID, prevEvents); err != nil { + if err = u.updater.StorePreviousEvents(u.stateAtEvent.EventNID, prevEvents); err != nil { return err } - eventReference := event.EventReference() + eventReference := u.event.EventReference() // Check if this event is already referenced by another event in the room. var alreadyReferenced bool - if alreadyReferenced, err = updater.IsReferenced(eventReference); err != nil { + if alreadyReferenced, err = u.updater.IsReferenced(eventReference); err != nil { return err } - newLatest := calculateLatest(oldLatest, alreadyReferenced, prevEvents, types.StateAtEventAndReference{ + u.latest = calculateLatest(oldLatest, alreadyReferenced, prevEvents, types.StateAtEventAndReference{ EventReference: eventReference, - StateAtEvent: stateAtEvent, + StateAtEvent: u.stateAtEvent, }) - latestStateAtEvents := make([]types.StateAtEvent, len(newLatest)) - for i := range newLatest { - latestStateAtEvents[i] = newLatest[i].StateAtEvent + if err = u.latestState(); err != nil { + return err } - newStateNID, err := state.CalculateAndStoreStateAfterEvents(db, roomNID, latestStateAtEvents) + + updates, err := updateMemberships(u.db, u.updater, u.removed, u.added) if err != nil { return err } - removed, added, err := state.DifferenceBetweeenStateSnapshots(db, oldStateNID, newStateNID) - if err != nil { - return err - } - - stateBeforeEventRemoves, stateBeforeEventAdds, err := state.DifferenceBetweeenStateSnapshots( - db, newStateNID, stateAtEvent.BeforeStateSnapshotNID, - ) + update, err := u.makeOutputNewRoomEvent() if err != nil { return err } + updates = append(updates, *update) // Send the event to the output logs. // We do this inside the database transaction to ensure that we only mark an event as sent if we sent it. @@ -138,24 +157,47 @@ func doUpdateLatestEvents( // send the event asynchronously but we would need to ensure that 1) the events are written to the log in // the correct order, 2) that pending writes are resent across restarts. In order to avoid writing all the // necessary bookkeeping we'll keep the event sending synchronous for now. - if err = writeEvent( - db, ow, lastEventIDSent, event, newLatest, removed, added, - stateBeforeEventRemoves, stateBeforeEventAdds, sendAsServer, - ); err != nil { + if err = u.ow.WriteOutputEvents(u.event.RoomID(), updates); err != nil { return err } - if err = updater.SetLatestEvents(roomNID, newLatest, stateAtEvent.EventNID, newStateNID); err != nil { + if err = u.updater.SetLatestEvents(u.roomNID, u.latest, u.stateAtEvent.EventNID, u.newStateNID); err != nil { return err } - if err = updater.MarkEventAsSent(stateAtEvent.EventNID); err != nil { + if err = u.updater.MarkEventAsSent(u.stateAtEvent.EventNID); err != nil { return err } return nil } +func (u *latestEventsUpdater) latestState() error { + var err error + + latestStateAtEvents := make([]types.StateAtEvent, len(u.latest)) + for i := range u.latest { + latestStateAtEvents[i] = u.latest[i].StateAtEvent + } + u.newStateNID, err = state.CalculateAndStoreStateAfterEvents(u.db, u.roomNID, latestStateAtEvents) + if err != nil { + return err + } + + u.removed, u.added, err = state.DifferenceBetweeenStateSnapshots(u.db, u.oldStateNID, u.newStateNID) + if err != nil { + return err + } + + u.stateBeforeEventRemoves, u.stateBeforeEventAdds, err = state.DifferenceBetweeenStateSnapshots( + u.db, u.newStateNID, u.stateAtEvent.BeforeStateSnapshotNID, + ) + if err != nil { + return err + } + return nil +} + func calculateLatest(oldLatest []types.StateAtEventAndReference, alreadyReferenced bool, prevEvents []gomatrixserverlib.EventReference, newEvent types.StateAtEventAndReference) []types.StateAtEventAndReference { var alreadyInLatest bool var newLatest []types.StateAtEventAndReference @@ -189,57 +231,55 @@ func calculateLatest(oldLatest []types.StateAtEventAndReference, alreadyReferenc return newLatest } -func writeEvent( - db RoomEventDatabase, ow OutputRoomEventWriter, lastEventIDSent string, - event gomatrixserverlib.Event, latest []types.StateAtEventAndReference, - removed, added []types.StateEntry, - stateBeforeEventRemoves, stateBeforeEventAdds []types.StateEntry, - sendAsServer string, -) error { +func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error) { - latestEventIDs := make([]string, len(latest)) - for i := range latest { - latestEventIDs[i] = latest[i].EventID + latestEventIDs := make([]string, len(u.latest)) + for i := range u.latest { + latestEventIDs[i] = u.latest[i].EventID } ore := api.OutputNewRoomEvent{ - Event: event, - LastSentEventID: lastEventIDSent, + Event: u.event, + LastSentEventID: u.lastEventIDSent, LatestEventIDs: latestEventIDs, } var stateEventNIDs []types.EventNID - for _, entry := range added { + for _, entry := range u.added { stateEventNIDs = append(stateEventNIDs, entry.EventNID) } - for _, entry := range removed { + for _, entry := range u.removed { stateEventNIDs = append(stateEventNIDs, entry.EventNID) } - for _, entry := range stateBeforeEventRemoves { + for _, entry := range u.stateBeforeEventRemoves { stateEventNIDs = append(stateEventNIDs, entry.EventNID) } - for _, entry := range stateBeforeEventAdds { + for _, entry := range u.stateBeforeEventAdds { stateEventNIDs = append(stateEventNIDs, entry.EventNID) } stateEventNIDs = stateEventNIDs[:util.SortAndUnique(eventNIDSorter(stateEventNIDs))] - eventIDMap, err := db.EventIDs(stateEventNIDs) + eventIDMap, err := u.db.EventIDs(stateEventNIDs) if err != nil { - return err + return nil, err } - for _, entry := range added { + for _, entry := range u.added { ore.AddsStateEventIDs = append(ore.AddsStateEventIDs, eventIDMap[entry.EventNID]) } - for _, entry := range removed { + for _, entry := range u.removed { ore.RemovesStateEventIDs = append(ore.RemovesStateEventIDs, eventIDMap[entry.EventNID]) } - for _, entry := range stateBeforeEventRemoves { + for _, entry := range u.stateBeforeEventRemoves { ore.StateBeforeRemovesEventIDs = append(ore.StateBeforeRemovesEventIDs, eventIDMap[entry.EventNID]) } - for _, entry := range stateBeforeEventAdds { + for _, entry := range u.stateBeforeEventAdds { ore.StateBeforeAddsEventIDs = append(ore.StateBeforeAddsEventIDs, eventIDMap[entry.EventNID]) } - ore.SendAsServer = sendAsServer - return ow.WriteOutputRoomEvent(ore) + ore.SendAsServer = u.sendAsServer + + return &api.OutputEvent{ + Type: api.OutputTypeNewRoomEvent, + NewRoomEvent: &ore, + }, nil } type eventNIDSorter []types.EventNID diff --git a/src/github.com/matrix-org/dendrite/roomserver/input/membership.go b/src/github.com/matrix-org/dendrite/roomserver/input/membership.go new file mode 100644 index 00000000..f306697f --- /dev/null +++ b/src/github.com/matrix-org/dendrite/roomserver/input/membership.go @@ -0,0 +1,297 @@ +// Copyright 2017 Vector Creations 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 input + +import ( + "fmt" + + "github.com/matrix-org/dendrite/roomserver/api" + "github.com/matrix-org/dendrite/roomserver/types" + "github.com/matrix-org/gomatrixserverlib" +) + +// updateMembership updates the current membership and the invites for each +// user affected by a change in the current state of the room. +// Returns a list of output events to write to the kafka log to inform the +// consumers about the invites added or retired by the change in current state. +func updateMemberships( + db RoomEventDatabase, updater types.RoomRecentEventsUpdater, removed, added []types.StateEntry, +) ([]api.OutputEvent, error) { + changes := membershipChanges(removed, added) + var eventNIDs []types.EventNID + for _, change := range changes { + if change.addedEventNID != 0 { + eventNIDs = append(eventNIDs, change.addedEventNID) + } + if change.removedEventNID != 0 { + eventNIDs = append(eventNIDs, change.removedEventNID) + } + } + + // Load the event JSON so we can look up the "membership" key. + // TODO: Maybe add a membership key to the events table so we can load that + // key without having to load the entire event JSON? + events, err := db.Events(eventNIDs) + if err != nil { + return nil, err + } + + var updates []api.OutputEvent + + for _, change := range changes { + var ae *gomatrixserverlib.Event + var re *gomatrixserverlib.Event + targetUserNID := change.EventStateKeyNID + if change.removedEventNID != 0 { + ev, _ := eventMap(events).lookup(change.removedEventNID) + if ev != nil { + re = &ev.Event + } + } + if change.addedEventNID != 0 { + ev, _ := eventMap(events).lookup(change.addedEventNID) + if ev != nil { + ae = &ev.Event + } + } + if updates, err = updateMembership(updater, targetUserNID, re, ae, updates); err != nil { + return nil, err + } + } + return nil, nil +} + +func updateMembership( + updater types.RoomRecentEventsUpdater, targetUserNID types.EventStateKeyNID, + remove, add *gomatrixserverlib.Event, + updates []api.OutputEvent, +) ([]api.OutputEvent, error) { + var err error + // Default the membership to "leave" if no event was added or removed. + old := "leave" + new := "leave" + + if remove != nil { + old, err = remove.Membership() + if err != nil { + return nil, err + } + } + if add != nil { + new, err = add.Membership() + if err != nil { + return nil, err + } + } + if old == new { + // If the membership is the same then nothing changed and we can return + // immediately. This should help speed up processing for display name + // changes where the membership is "join" both before and after. + return updates, nil + } + + mu, err := updater.MembershipUpdater(targetUserNID) + if err != nil { + return nil, err + } + + switch new { + case "invite": + return updateToInviteMembership(mu, add, updates) + case "join": + return updateToJoinMembership(mu, add, updates) + case "leave", "ban": + return updateToLeaveMembership(mu, add, new, updates) + default: + panic(fmt.Errorf( + "input: membership %q is not one of the allowed values", new, + )) + } +} + +func updateToInviteMembership( + mu types.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent, +) ([]api.OutputEvent, error) { + // We may have already sent the invite to the user, either because we are + // reprocessing this event, or because the we received this invite from a + // remote server via the federation invite API. In those cases we don't need + // to send the event. + needsSending, err := mu.SetToInvite(*add) + if err != nil { + return nil, err + } + if needsSending { + // We notify the consumers using a special event even though we will + // notify them about the change in current state as part of the normal + // room event stream. This ensures that the consumers only have to + // consider a single stream of events when determining whether a user + // is invited, rather than having to combine multiple streams themselves. + onie := api.OutputNewInviteEvent{ + Event: *add, + } + updates = append(updates, api.OutputEvent{ + Type: api.OutputTypeNewInviteEvent, + NewInviteEvent: &onie, + }) + } + return updates, nil +} + +func updateToJoinMembership( + mu types.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent, +) ([]api.OutputEvent, error) { + // If the user is already marked as being joined then we can return immediately. + // TODO: Is this code reachable given the "old != new" guard in updateMembership? + if mu.IsJoin() { + return updates, nil + } + // When we mark a user as being joined we will invalidate any invites that + // are active for that user. We notify the consumers that the invites have + // been retired using a special event, even though they could infer this + // by studying the state changes in the room event stream. + retired, err := mu.SetToJoin(add.Sender()) + if err != nil { + return nil, err + } + for _, eventID := range retired { + orie := api.OutputRetireInviteEvent{ + EventID: eventID, + Membership: "join", + } + if add != nil { + orie.RetiredByEventID = add.EventID() + } + updates = append(updates, api.OutputEvent{ + Type: api.OutputTypeRetireInviteEvent, + RetireInviteEvent: &orie, + }) + } + return updates, nil +} + +func updateToLeaveMembership( + mu types.MembershipUpdater, add *gomatrixserverlib.Event, + newMembership string, updates []api.OutputEvent, +) ([]api.OutputEvent, error) { + // If the user is already neither joined, nor invited to the room then we + // can return immediately. + if mu.IsLeave() { + return updates, nil + } + // When we mark a user as having left we will invalidate any invites that + // are active for that user. We notify the consumers that the invites have + // been retired using a special event, even though they could infer this + // by studying the state changes in the room event stream. + retired, err := mu.SetToLeave(add.Sender()) + if err != nil { + return nil, err + } + for _, eventID := range retired { + orie := api.OutputRetireInviteEvent{ + EventID: eventID, + Membership: newMembership, + } + if add != nil { + orie.RetiredByEventID = add.EventID() + } + updates = append(updates, api.OutputEvent{ + Type: api.OutputTypeRetireInviteEvent, + RetireInviteEvent: &orie, + }) + } + return updates, nil +} + +// membershipChanges pairs up the membership state changes from a sorted list +// of state removed and a sorted list of state added. +func membershipChanges(removed, added []types.StateEntry) []stateChange { + changes := pairUpChanges(removed, added) + var result []stateChange + for _, c := range changes { + if c.EventTypeNID == types.MRoomMemberNID { + result = append(result, c) + } + } + return result +} + +type stateChange struct { + types.StateKeyTuple + removedEventNID types.EventNID + addedEventNID types.EventNID +} + +// pairUpChanges pairs up the state events added and removed for each type, +// state key tuple. Assumes that removed and added are sorted. +func pairUpChanges(removed, added []types.StateEntry) []stateChange { + var ai int + var ri int + var result []stateChange + for { + switch { + case ai == len(added): + // We've reached the end of the added entries. + // The rest of the removed list are events that were removed without + // an event with the same state key being added. + for _, s := range removed[ri:] { + result = append(result, stateChange{ + StateKeyTuple: s.StateKeyTuple, + removedEventNID: s.EventNID, + }) + } + return result + case ri == len(removed): + // We've reached the end of the removed entries. + // The rest of the added list are events that were added without + // an event with the same state key being removed. + for _, s := range added[ai:] { + result = append(result, stateChange{ + StateKeyTuple: s.StateKeyTuple, + addedEventNID: s.EventNID, + }) + } + return result + case added[ai].StateKeyTuple == removed[ri].StateKeyTuple: + // The tuple is in both lists so an event with that key is being + // removed and another event with the same key is being added. + result = append(result, stateChange{ + StateKeyTuple: added[ai].StateKeyTuple, + removedEventNID: removed[ri].EventNID, + addedEventNID: added[ai].EventNID, + }) + ai++ + ri++ + case added[ai].StateKeyTuple.LessThan(removed[ri].StateKeyTuple): + // The lists are sorted so the added entry being less than the + // removed entry means that the added event was added without an + // event with the same key being removed. + result = append(result, stateChange{ + StateKeyTuple: added[ai].StateKeyTuple, + addedEventNID: added[ai].EventNID, + }) + ai++ + default: + // Reaching the default case implies that the removed entry is less + // than the added entry. Since the lists are sorted this means that + // the removed event was removed without an event with the same + // key being added. + result = append(result, stateChange{ + StateKeyTuple: removed[ai].StateKeyTuple, + removedEventNID: removed[ri].EventNID, + }) + ri++ + } + } +} diff --git a/src/github.com/matrix-org/dendrite/roomserver/storage/event_state_keys_table.go b/src/github.com/matrix-org/dendrite/roomserver/storage/event_state_keys_table.go index da5b8a4e..d30e4581 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/storage/event_state_keys_table.go +++ b/src/github.com/matrix-org/dendrite/roomserver/storage/event_state_keys_table.go @@ -76,15 +76,23 @@ func (s *eventStateKeyStatements) prepare(db *sql.DB) (err error) { }.prepare(db) } -func (s *eventStateKeyStatements) insertEventStateKeyNID(eventStateKey string) (types.EventStateKeyNID, error) { +func (s *eventStateKeyStatements) insertEventStateKeyNID(txn *sql.Tx, eventStateKey string) (types.EventStateKeyNID, error) { var eventStateKeyNID int64 - err := s.insertEventStateKeyNIDStmt.QueryRow(eventStateKey).Scan(&eventStateKeyNID) + stmt := s.insertEventStateKeyNIDStmt + if txn != nil { + stmt = txn.Stmt(stmt) + } + err := stmt.QueryRow(eventStateKey).Scan(&eventStateKeyNID) return types.EventStateKeyNID(eventStateKeyNID), err } -func (s *eventStateKeyStatements) selectEventStateKeyNID(eventStateKey string) (types.EventStateKeyNID, error) { +func (s *eventStateKeyStatements) selectEventStateKeyNID(txn *sql.Tx, eventStateKey string) (types.EventStateKeyNID, error) { var eventStateKeyNID int64 - err := s.selectEventStateKeyNIDStmt.QueryRow(eventStateKey).Scan(&eventStateKeyNID) + stmt := s.selectEventStateKeyNIDStmt + if txn != nil { + stmt = txn.Stmt(stmt) + } + err := stmt.QueryRow(eventStateKey).Scan(&eventStateKeyNID) return types.EventStateKeyNID(eventStateKeyNID), err } diff --git a/src/github.com/matrix-org/dendrite/roomserver/storage/invite_table.go b/src/github.com/matrix-org/dendrite/roomserver/storage/invite_table.go new file mode 100644 index 00000000..9e0860b4 --- /dev/null +++ b/src/github.com/matrix-org/dendrite/roomserver/storage/invite_table.go @@ -0,0 +1,149 @@ +// Copyright 2017 Vector Creations 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 storage + +import ( + "database/sql" + + "github.com/matrix-org/dendrite/roomserver/types" +) + +const inviteSchema = ` +CREATE TABLE IF NOT EXISTS roomserver_invites ( + -- The string ID of the invite event itself. + -- We can't use a numeric event ID here because we don't always have + -- enough information to store an invite in the event table. + -- In particular we don't always have a chain of auth_events for invites + -- received over federation. + invite_event_id TEXT PRIMARY KEY, + -- The numeric ID of the room the invite m.room.member event is in. + room_nid BIGINT NOT NULL, + -- The numeric ID for the state key of the invite m.room.member event. + -- This tells us who the invite is for. + -- This is used to query the active invites for a user. + target_nid BIGINT NOT NULL, + -- The numeric ID for the sender of the invite m.room.member event. + -- This tells us who sent the invite. + -- This is used to work out which matrix server we should talk to when + -- we try to join the room. + sender_nid BIGINT NOT NULL DEFAULT 0, + -- This is used to track whether the invite is still active. + -- This is set implicitly when processing new join and leave events and + -- explicitly when rejecting events over federation. + retired BOOLEAN NOT NULL DEFAULT FALSE, + -- The invite event JSON. + invite_event_json TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS roomserver_invites_active_idx ON roomserver_invites (target_nid, room_nid) + WHERE NOT retired; +` +const insertInviteEventSQL = "" + + "INSERT INTO roomserver_invites (invite_event_id, room_nid, target_nid," + + " sender_nid, invite_event_json) VALUES ($1, $2, $3, $4, $5)" + + " ON CONFLICT DO NOTHING" + +const selectInviteActiveForUserInRoomSQL = "" + + "SELECT sender_nid FROM roomserver_invites" + + " WHERE target_nid = $1 AND room_nid = $2" + + " AND NOT retired" + +// Retire every active invite for a user in a room. +// Ideally we'd know which invite events were retired by a given update so we +// wouldn't need to remove every active invite. +// However the matrix protocol doesn't give us a way to reliably identify the +// invites that were retired, so we are forced to retire all of them. +const updateInviteRetiredSQL = "" + + "UPDATE roomserver_invites SET retired = TRUE" + + " WHERE room_nid = $1 AND target_nid = $2 AND NOT retired" + + " RETURNING invite_event_id" + +type inviteStatements struct { + insertInviteEventStmt *sql.Stmt + selectInviteActiveForUserInRoomStmt *sql.Stmt + updateInviteRetiredStmt *sql.Stmt +} + +func (s *inviteStatements) prepare(db *sql.DB) (err error) { + _, err = db.Exec(inviteSchema) + if err != nil { + return + } + + return statementList{ + {&s.insertInviteEventStmt, insertInviteEventSQL}, + {&s.selectInviteActiveForUserInRoomStmt, selectInviteActiveForUserInRoomSQL}, + {&s.updateInviteRetiredStmt, updateInviteRetiredSQL}, + }.prepare(db) +} + +func (s *inviteStatements) insertInviteEvent( + txn *sql.Tx, inviteEventID string, roomNID types.RoomNID, + targetUserNID, senderUserNID types.EventStateKeyNID, + inviteEventJSON []byte, +) (bool, error) { + result, err := txn.Stmt(s.insertInviteEventStmt).Exec( + inviteEventID, roomNID, targetUserNID, senderUserNID, inviteEventJSON, + ) + if err != nil { + return false, err + } + count, err := result.RowsAffected() + if err != nil { + return false, err + } + return count != 0, nil +} + +func (s *inviteStatements) updateInviteRetired( + txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID, +) ([]string, error) { + rows, err := txn.Stmt(s.updateInviteRetiredStmt).Query(roomNID, targetUserNID) + if err != nil { + return nil, err + } + defer rows.Close() + var result []string + for rows.Next() { + var inviteEventID string + if err := rows.Scan(&inviteEventID); err != nil { + return nil, err + } + result = append(result, inviteEventID) + } + return result, nil +} + +// selectInviteActiveForUserInRoom returns a list of sender state key NIDs +func (s *inviteStatements) selectInviteActiveForUserInRoom( + targetUserNID types.EventStateKeyNID, roomNID types.RoomNID, +) ([]types.EventStateKeyNID, error) { + rows, err := s.selectInviteActiveForUserInRoomStmt.Query( + targetUserNID, roomNID, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var result []types.EventStateKeyNID + for rows.Next() { + var senderUserNID int64 + if err := rows.Scan(&senderUserNID); err != nil { + return nil, err + } + result = append(result, types.EventStateKeyNID(senderUserNID)) + } + return result, nil +} diff --git a/src/github.com/matrix-org/dendrite/roomserver/storage/membership_table.go b/src/github.com/matrix-org/dendrite/roomserver/storage/membership_table.go new file mode 100644 index 00000000..725e5b8d --- /dev/null +++ b/src/github.com/matrix-org/dendrite/roomserver/storage/membership_table.go @@ -0,0 +1,111 @@ +// Copyright 2017 Vector Creations 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 storage + +import ( + "database/sql" + + "github.com/matrix-org/dendrite/roomserver/types" +) + +type membershipState int64 + +const ( + membershipStateLeaveOrBan membershipState = 1 + membershipStateInvite membershipState = 2 + membershipStateJoin membershipState = 3 +) + +const membershipSchema = ` +-- The membership table is used to coordinate updates between the invite table +-- and the room state tables. +-- This table is updated in one of 3 ways: +-- 1) The membership of a user changes within the current state of the room. +-- 2) An invite is received outside of a room over federation. +-- 3) An invite is rejected outside of a room over federation. +CREATE TABLE IF NOT EXISTS roomserver_membership ( + room_nid BIGINT NOT NULL, + -- Numeric state key ID for the user ID this state is for. + target_nid BIGINT NOT NULL, + -- Numeric state key ID for the user ID who changed the state. + -- This may be 0 since it is not always possible to identify the user that + -- changed the state. + sender_nid BIGINT NOT NULL DEFAULT 0, + -- The state the user is in within this room. + -- Default value is "membershipStateLeaveOrBan" + membership_nid BIGINT NOT NULL DEFAULT 1, + UNIQUE (room_nid, target_nid) +); +` + +// Insert a row in to membership table so that it can be locked by the +// SELECT FOR UPDATE +const insertMembershipSQL = "" + + "INSERT INTO roomserver_membership (room_nid, target_nid)" + + " VALUES ($1, $2)" + + " ON CONFLICT DO NOTHING" + +const selectMembershipForUpdateSQL = "" + + "SELECT membership_nid FROM roomserver_membership" + + " WHERE room_nid = $1 AND target_nid = $2 FOR UPDATE" + +const updateMembershipSQL = "" + + "UPDATE roomserver_membership SET sender_nid = $3, membership_nid = $4" + + " WHERE room_nid = $1 AND target_nid = $2" + +type membershipStatements struct { + insertMembershipStmt *sql.Stmt + selectMembershipForUpdateStmt *sql.Stmt + updateMembershipStmt *sql.Stmt +} + +func (s *membershipStatements) prepare(db *sql.DB) (err error) { + _, err = db.Exec(membershipSchema) + if err != nil { + return + } + + return statementList{ + {&s.insertMembershipStmt, insertMembershipSQL}, + {&s.selectMembershipForUpdateStmt, selectMembershipForUpdateSQL}, + {&s.updateMembershipStmt, updateMembershipSQL}, + }.prepare(db) +} + +func (s *membershipStatements) insertMembership( + txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID, +) error { + _, err := txn.Stmt(s.insertMembershipStmt).Exec(roomNID, targetUserNID) + return err +} + +func (s *membershipStatements) selectMembershipForUpdate( + txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID, +) (membership membershipState, err error) { + err = txn.Stmt(s.selectMembershipForUpdateStmt).QueryRow( + roomNID, targetUserNID, + ).Scan(&membership) + return +} + +func (s *membershipStatements) updateMembership( + txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID, + senderUserNID types.EventStateKeyNID, membership membershipState, +) error { + _, err := txn.Stmt(s.updateMembershipStmt).Exec( + roomNID, targetUserNID, senderUserNID, membership, + ) + return err +} diff --git a/src/github.com/matrix-org/dendrite/roomserver/storage/sql.go b/src/github.com/matrix-org/dendrite/roomserver/storage/sql.go index ddab2356..a24dbb1d 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/storage/sql.go +++ b/src/github.com/matrix-org/dendrite/roomserver/storage/sql.go @@ -28,6 +28,8 @@ type statements struct { stateBlockStatements previousEventStatements roomAliasesStatements + inviteStatements + membershipStatements } func (s *statements) prepare(db *sql.DB) error { @@ -43,6 +45,8 @@ func (s *statements) prepare(db *sql.DB) error { s.stateBlockStatements.prepare, s.previousEventStatements.prepare, s.roomAliasesStatements.prepare, + s.inviteStatements.prepare, + s.membershipStatements.prepare, } { if err = prepare(db); err != nil { return err diff --git a/src/github.com/matrix-org/dendrite/roomserver/storage/storage.go b/src/github.com/matrix-org/dendrite/roomserver/storage/storage.go index 3f99e7d8..d323fd13 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/storage/storage.go +++ b/src/github.com/matrix-org/dendrite/roomserver/storage/storage.go @@ -16,6 +16,7 @@ package storage import ( "database/sql" + // Import the postgres database driver. _ "github.com/lib/pq" "github.com/matrix-org/dendrite/roomserver/types" @@ -64,7 +65,7 @@ func (d *Database) StoreEvent(event gomatrixserverlib.Event, authEventNIDs []typ // Assigned a numeric ID for the state_key if there is one present. // Otherwise set the numeric ID for the state_key to 0. if eventStateKey != nil { - if eventStateKeyNID, err = d.assignStateKeyNID(*eventStateKey); err != nil { + if eventStateKeyNID, err = d.assignStateKeyNID(nil, *eventStateKey); err != nil { return 0, types.StateAtEvent{}, err } } @@ -131,15 +132,15 @@ func (d *Database) assignEventTypeNID(eventType string) (types.EventTypeNID, err return eventTypeNID, err } -func (d *Database) assignStateKeyNID(eventStateKey string) (types.EventStateKeyNID, error) { +func (d *Database) assignStateKeyNID(txn *sql.Tx, eventStateKey string) (types.EventStateKeyNID, error) { // Check if we already have a numeric ID in the database. - eventStateKeyNID, err := d.statements.selectEventStateKeyNID(eventStateKey) + eventStateKeyNID, err := d.statements.selectEventStateKeyNID(txn, eventStateKey) if err == sql.ErrNoRows { // We don't have a numeric ID so insert one into the database. - eventStateKeyNID, err = d.statements.insertEventStateKeyNID(eventStateKey) + eventStateKeyNID, err = d.statements.insertEventStateKeyNID(txn, eventStateKey) if err == sql.ErrNoRows { // We raced with another insert so run the select again. - eventStateKeyNID, err = d.statements.selectEventStateKeyNID(eventStateKey) + eventStateKeyNID, err = d.statements.selectEventStateKeyNID(txn, eventStateKey) } } return eventStateKeyNID, err @@ -249,12 +250,15 @@ func (d *Database) GetLatestEventsForUpdate(roomNID types.RoomNID) (types.RoomRe return nil, err } } - return &roomRecentEventsUpdater{txn, d, stateAndRefs, lastEventIDSent, currentStateSnapshotNID}, nil + return &roomRecentEventsUpdater{ + transaction{txn}, d, roomNID, stateAndRefs, lastEventIDSent, currentStateSnapshotNID, + }, nil } type roomRecentEventsUpdater struct { - txn *sql.Tx + transaction d *Database + roomNID types.RoomNID latestEvents []types.StateAtEventAndReference lastEventIDSent string currentStateSnapshotNID types.StateSnapshotNID @@ -319,14 +323,8 @@ func (u *roomRecentEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error return u.d.statements.updateEventSentToOutput(u.txn, eventNID) } -// Commit implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) Commit() error { - return u.txn.Commit() -} - -// Rollback implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) Rollback() error { - return u.txn.Rollback() +func (u *roomRecentEventsUpdater) MembershipUpdater(targetUserNID types.EventStateKeyNID) (types.MembershipUpdater, error) { + return u.d.membershipUpdaterTxn(u.txn, u.roomNID, targetUserNID) } // RoomNID implements query.RoomserverQueryAPIDB @@ -381,3 +379,124 @@ func (d *Database) StateEntriesForTuples( ) ([]types.StateEntryList, error) { return d.statements.bulkSelectFilteredStateBlockEntries(stateBlockNIDs, stateKeyTuples) } + +type membershipUpdater struct { + transaction + d *Database + roomNID types.RoomNID + targetUserNID types.EventStateKeyNID + membership membershipState +} + +func (d *Database) membershipUpdaterTxn( + txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID, +) (types.MembershipUpdater, error) { + + if err := d.statements.insertMembership(txn, roomNID, targetUserNID); err != nil { + return nil, err + } + + membership, err := d.statements.selectMembershipForUpdate(txn, roomNID, targetUserNID) + if err != nil { + return nil, err + } + + return &membershipUpdater{ + transaction{txn}, d, roomNID, targetUserNID, membership, + }, nil +} + +// IsInvite implements types.MembershipUpdater +func (u *membershipUpdater) IsInvite() bool { + return u.membership == membershipStateInvite +} + +// IsJoin implements types.MembershipUpdater +func (u *membershipUpdater) IsJoin() bool { + return u.membership == membershipStateJoin +} + +// IsLeave implements types.MembershipUpdater +func (u *membershipUpdater) IsLeave() bool { + return u.membership == membershipStateLeaveOrBan +} + +// SetToInvite implements types.MembershipUpdater +func (u *membershipUpdater) SetToInvite(event gomatrixserverlib.Event) (bool, error) { + senderUserNID, err := u.d.assignStateKeyNID(u.txn, event.Sender()) + if err != nil { + return false, err + } + inserted, err := u.d.statements.insertInviteEvent( + u.txn, event.EventID(), u.roomNID, u.targetUserNID, senderUserNID, event.JSON(), + ) + if err != nil { + return false, err + } + if u.membership != membershipStateInvite { + if err = u.d.statements.updateMembership( + u.txn, u.roomNID, u.targetUserNID, senderUserNID, membershipStateInvite, + ); err != nil { + return false, err + } + } + return inserted, nil +} + +// SetToJoin implements types.MembershipUpdater +func (u *membershipUpdater) SetToJoin(senderUserID string) ([]string, error) { + senderUserNID, err := u.d.assignStateKeyNID(u.txn, senderUserID) + if err != nil { + return nil, err + } + inviteEventIDs, err := u.d.statements.updateInviteRetired( + u.txn, u.roomNID, u.targetUserNID, + ) + if err != nil { + return nil, err + } + if u.membership != membershipStateJoin { + if err = u.d.statements.updateMembership( + u.txn, u.roomNID, u.targetUserNID, senderUserNID, membershipStateJoin, + ); err != nil { + return nil, err + } + } + return inviteEventIDs, nil +} + +// SetToLeave implements types.MembershipUpdater +func (u *membershipUpdater) SetToLeave(senderUserID string) ([]string, error) { + senderUserNID, err := u.d.assignStateKeyNID(u.txn, senderUserID) + if err != nil { + return nil, err + } + inviteEventIDs, err := u.d.statements.updateInviteRetired( + u.txn, u.roomNID, u.targetUserNID, + ) + if err != nil { + return nil, err + } + if u.membership != membershipStateLeaveOrBan { + if err = u.d.statements.updateMembership( + u.txn, u.roomNID, u.targetUserNID, senderUserNID, membershipStateLeaveOrBan, + ); err != nil { + return nil, err + } + } + return inviteEventIDs, nil +} + +type transaction struct { + txn *sql.Tx +} + +// Commit implements types.Transaction +func (t *transaction) Commit() error { + return t.txn.Commit() +} + +// Rollback implements types.Transaction +func (t *transaction) Rollback() error { + return t.txn.Rollback() +} diff --git a/src/github.com/matrix-org/dendrite/roomserver/types/types.go b/src/github.com/matrix-org/dendrite/roomserver/types/types.go index b255b64b..809b6e57 100644 --- a/src/github.com/matrix-org/dendrite/roomserver/types/types.go +++ b/src/github.com/matrix-org/dendrite/roomserver/types/types.go @@ -135,9 +135,17 @@ type StateEntryList struct { StateEntries []StateEntry } +// A Transaction is something that can be committed or rolledback. +type Transaction interface { + // Commit the transaction + Commit() error + // Rollback the transaction. + Rollback() error +} + // A RoomRecentEventsUpdater is used to update the recent events in a room. // (On postgresql this wraps a database transaction that holds a "FOR UPDATE" -// lock on the row holding the latest events for the room.) +// lock on the row in the rooms table holding the latest events for the room.) type RoomRecentEventsUpdater interface { // The latest event IDs and state in the room. LatestEvents() []StateAtEventAndReference @@ -163,10 +171,36 @@ type RoomRecentEventsUpdater interface { HasEventBeenSent(eventNID EventNID) (bool, error) // Mark the event as having been sent to the output logs. MarkEventAsSent(eventNID EventNID) error - // Commit the transaction - Commit() error - // Rollback the transaction. - Rollback() error + // Build a membership updater for the target user in this room. + // It will share the same transaction as this updater. + MembershipUpdater(targetUserNID EventStateKeyNID) (MembershipUpdater, error) + // Implements Transaction so it can be committed or rolledback + Transaction +} + +// A MembershipUpdater is used to update the membership of a user in a room. +// (On postgresql this wraps a database transaction that holds a "FOR UPDATE" +// lock on the row in the membership table for this user in the room) +// The caller should call one of SetToInvite, SetToJoin or SetToLeave once to +// make the update, or none of them if no update is required. +type MembershipUpdater interface { + // True if the target user is invited to the room before updating. + IsInvite() bool + // True if the target user is joined to the room before updating. + IsJoin() bool + // True if the target user is not invited or joined to the room before updating. + IsLeave() bool + // Set the state to invite. + // Returns whether this invite needs to be sent + SetToInvite(event gomatrixserverlib.Event) (needsSending bool, err error) + // Set the state to join. + // Returns a list of invite event IDs that this state change retired. + SetToJoin(senderUserID string) (inviteEventIDs []string, err error) + // Set the state to leave. + // Returns a list of invite event IDs that this state change retired. + SetToLeave(senderUserID string) (inviteEventIDs []string, err error) + // Implements Transaction so it can be committed or rolledback. + Transaction } // A MissingEventError is an error that happened because the roomserver was