dendrite/roomserver/input/events.go

253 lines
8.8 KiB
Go
Raw Normal View History

// Copyright 2017 Vector Creations Ltd
// Copyright 2018 New Vector Ltd
// Copyright 2019-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.
package input
import (
"context"
"fmt"
"github.com/matrix-org/dendrite/common"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/state/database"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
)
// A RoomEventDatabase has the storage APIs needed to store a room event.
type RoomEventDatabase interface {
database.RoomStateDatabase
// Stores a matrix room event in the database
StoreEvent(
ctx context.Context,
event gomatrixserverlib.Event,
txnAndSessionID *api.TransactionID,
authEventNIDs []types.EventNID,
) (types.RoomNID, types.StateAtEvent, error)
// Look up the state entries for a list of string event IDs
// Returns an error if the there is an error talking to the database
// Returns a types.MissingEventError if the event IDs aren't in the database.
StateEntriesForEventIDs(
ctx context.Context, eventIDs []string,
) ([]types.StateEntry, error)
// Set the state at an event.
SetState(
ctx context.Context,
eventNID types.EventNID,
stateNID types.StateSnapshotNID,
) error
// Look up the latest events in a room in preparation for an update.
// The RoomRecentEventsUpdater must have Commit or Rollback called on it if this doesn't return an error.
// Returns the latest events in the room and the last eventID sent to the log along with an updater.
// If this returns an error then no further action is required.
GetLatestEventsForUpdate(
ctx context.Context, roomNID types.RoomNID,
) (updater types.RoomRecentEventsUpdater, err error)
// Look up the string event IDs for a list of numeric event IDs
EventIDs(
ctx context.Context, eventNIDs []types.EventNID,
) (map[types.EventNID]string, error)
// Build a membership updater for the target user in a room.
MembershipUpdater(
ctx context.Context, roomID, targerUserID string,
) (types.MembershipUpdater, error)
// Look up event ID by transaction's info.
// This is used to determine if the room event is processed/processing already.
// Returns an empty string if no such event exists.
GetTransactionEventID(
ctx context.Context, transactionID string,
sessionID int64, userID string,
) (string, error)
// Look up the room version for a given room.
GetRoomVersionForRoom(
ctx context.Context, roomID string,
) (gomatrixserverlib.RoomVersion, error)
}
// OutputRoomEventWriter has the APIs needed to write an event to the output logs.
type OutputRoomEventWriter interface {
// Write a list of events for a room
WriteOutputEvents(roomID string, updates []api.OutputEvent) error
}
// processRoomEvent can only be called once at a time
//
// TODO(#375): This should be rewritten to allow concurrent calls. The
// difficulty is in ensuring that we correctly annotate events with the correct
// state deltas when sending to kafka streams
func processRoomEvent(
ctx context.Context,
db RoomEventDatabase,
ow OutputRoomEventWriter,
input api.InputRoomEvent,
) (eventID string, err error) {
// Parse and validate the event JSON
event := input.Event
// Check that the event passes authentication checks and work out the numeric IDs for the auth events.
authEventNIDs, err := checkAuthEvents(ctx, db, event.Event, input.AuthEventIDs)
if err != nil {
return
}
if input.TransactionID != nil {
tdID := input.TransactionID
eventID, err = db.GetTransactionEventID(
ctx, tdID.TransactionID, tdID.SessionID, input.Event.Sender(),
)
// On error OR event with the transaction already processed/processesing
if err != nil || eventID != "" {
return
}
}
// Store the event
roomNID, stateAtEvent, err := db.StoreEvent(ctx, event.Event, input.TransactionID, authEventNIDs)
if err != nil {
return
}
if input.Kind == api.KindOutlier {
// For outliers we can stop after we've stored the event itself as it
// doesn't have any associated state to store and we don't need to
// notify anyone about it.
return event.EventID(), nil
}
if stateAtEvent.BeforeStateSnapshotNID == 0 {
// We haven't calculated a state for this event yet.
// Lets calculate one.
err = calculateAndSetState(ctx, db, input, roomNID, &stateAtEvent, event.Event)
if err != nil {
return
}
}
if input.Kind == api.KindBackfill {
// Backfill is not implemented.
panic("Not implemented")
}
// Update the extremities of the event graph for the room
return event.EventID(), updateLatestEvents(
ctx, db, ow, roomNID, stateAtEvent, event.Event, input.SendAsServer, input.TransactionID,
)
}
func calculateAndSetState(
ctx context.Context,
db RoomEventDatabase,
input api.InputRoomEvent,
roomNID types.RoomNID,
stateAtEvent *types.StateAtEvent,
event gomatrixserverlib.Event,
) error {
// TODO: get the correct room version
roomState, err := state.GetStateResolutionAlgorithm(state.StateResolutionAlgorithmV1, db)
if err != nil {
return err
}
if input.HasState {
// We've been told what the state at the event is so we don't need to calculate it.
// Check that those state events are in the database and store the state.
var entries []types.StateEntry
if entries, err = db.StateEntriesForEventIDs(ctx, input.StateEventIDs); err != nil {
return err
}
if stateAtEvent.BeforeStateSnapshotNID, err = db.AddState(ctx, roomNID, nil, entries); err != nil {
return err
}
} else {
// We haven't been told what the state at the event is so we need to calculate it from the prev_events
if stateAtEvent.BeforeStateSnapshotNID, err = roomState.CalculateAndStoreStateBeforeEvent(ctx, event, roomNID); err != nil {
return err
}
}
return db.SetState(ctx, stateAtEvent.EventNID, stateAtEvent.BeforeStateSnapshotNID)
}
func processInviteEvent(
ctx context.Context,
db RoomEventDatabase,
ow OutputRoomEventWriter,
input api.InputInviteEvent,
) (err error) {
if input.Event.StateKey() == nil {
return fmt.Errorf("invite must be a state event")
}
roomID := input.Event.RoomID()
targetUserID := *input.Event.StateKey()
updater, err := db.MembershipUpdater(ctx, roomID, targetUserID)
if err != nil {
return err
}
succeeded := false
Support sqlite in addition to postgres (#869) * Move current work into single branch * Initial massaging of clientapi etc (not working yet) * Interfaces for accounts/devices databases * Duplicate postgres package for sqlite3 (no changes made to it yet) * Some keydb, accountdb, devicedb, common partition fixes, some more syncapi tweaking * Fix accounts DB, device DB * Update naffka dependency for SQLite * Naffka SQLite * Update naffka to latest master * SQLite support for federationsender * Mostly not-bad support for SQLite in syncapi (although there are problems where lots of events get classed incorrectly as backward extremities, probably because of IN/ANY clauses that are badly supported) * Update Dockerfile -> Go 1.13.7, add build-base (as gcc and friends are needed for SQLite) * Implement GET endpoints for account_data in clientapi * Nuke filtering for now... * Revert "Implement GET endpoints for account_data in clientapi" This reverts commit 4d80dff4583d278620d9b3ed437e9fcd8d4674ee. * Implement GET endpoints for account_data in clientapi (#861) * Implement GET endpoints for account_data in clientapi * Fix accountDB parameter * Remove fmt.Println * Fix insertAccountData SQLite query * Fix accountDB storage interfaces * Add empty push rules into account data on account creation (#862) * Put SaveAccountData into the right function this time * Not sure if roomserver is better or worse now * sqlite work * Allow empty last sent ID for the first event * sqlite: room creation works * Support sending messages * Nuke fmt.println * Move QueryVariadic etc into common, other device fixes * Fix some linter issues * Fix bugs * Fix some linting errors * Fix errcheck lint errors * Make naffka use postgres as fallback, fix couple of compile errors * What on earth happened to the /rooms/{roomID}/send/{eventType} routing Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>
2020-02-13 17:27:33 +00:00
defer func() {
txerr := common.EndTransaction(updater, &succeeded)
if err == nil && txerr != nil {
err = txerr
}
}()
if updater.IsJoin() {
// If the user is joined to the room then that takes precedence over this
// invite event. It makes little sense to move a user that is already
// joined to the room into the invite state.
// This could plausibly happen if an invite request raced with a join
// request for a user. For example if a user was invited to a public
// room and they joined the room at the same time as the invite was sent.
// The other way this could plausibly happen is if an invite raced with
// a kick. For example if a user was kicked from a room in error and in
// response someone else in the room re-invited them then it is possible
// for the invite request to race with the leave event so that the
// target receives invite before it learns that it has been kicked.
// There are a few ways this could be plausibly handled in the roomserver.
// 1) Store the invite, but mark it as retired. That will result in the
// permanent rejection of that invite event. So even if the target
// user leaves the room and the invite is retransmitted it will be
// ignored. However a new invite with a new event ID would still be
// accepted.
// 2) Silently discard the invite event. This means that if the event
// was retransmitted at a later date after the target user had left
// the room we would accept the invite. However since we hadn't told
// the sending server that the invite had been discarded it would
// have no reason to attempt to retry.
// 3) Signal the sending server that the user is already joined to the
// room.
// For now we will implement option 2. Since in the abesence of a retry
// mechanism it will be equivalent to option 1, and we don't have a
// signalling mechanism to implement option 3.
return nil
}
outputUpdates, err := updateToInviteMembership(updater, &input.Event.Event, nil)
if err != nil {
return err
}
if err = ow.WriteOutputEvents(roomID, outputUpdates); err != nil {
return err
}
succeeded = true
return nil
}