Handle state with input event as new events (#1415)
* SendEventWithState events as new * Use cumulative state IDs for final event * Error wrapping in calculateAndSetState * Handle overwriting same event type and state key * Hacky way to spot historical events * Don't exclude from sync * Don't generate output events when rewriting forward extremities * Update output event check * Historical output events * Define output room event type * Notify key changes on state * Don't send our membership event twice * Deduplicate state entries * Tweaks * Remove unnecessary nolint * Fix current state upsert in sync API * Send auth events as outliers, state events as rewrite * Sync API don't consume state events * Process events actually * Improve outlier check * Fix local room check * Remove extra room check, it seems to break the whole damn world * Fix federated join check * Fix nil pointer exception * Better comments on DeduplicateStateEntries * Reflow forced federated joins * Don't force federated join for possibly even local invites * Comment SendEventWithState better * Rewrite room state in sync API storage * Add TODO * Clean up all room data when receiving create event * Don't generate output events for rewrites, but instead notify that state is rewritten on the final new event * Rename to PurgeRoom * Exclude backfilled messages from /sync * Split out rewriting state from updating state from state res Co-authored-by: Kegan Dougal <kegan@matrix.org>main
Neil Alexander
GitHub
parent
8dc9506210
commit
965f068d1a
|
|
@ -98,7 +98,10 @@ func (r *FederationSenderInternalAPI) PerformJoin(
|
|||
response.LastError = &gomatrix.HTTPError{
|
||||
Code: 0,
|
||||
WrappedError: nil,
|
||||
Message: lastErr.Error(),
|
||||
Message: "Unknown HTTP error",
|
||||
}
|
||||
if lastErr != nil {
|
||||
response.LastError.Message = lastErr.Error()
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -195,7 +198,7 @@ func (r *FederationSenderInternalAPI) performJoinUsingServer(
|
|||
// If we successfully performed a send_join above then the other
|
||||
// server now thinks we're a part of the room. Send the newly
|
||||
// returned state to the roomserver to update our local view.
|
||||
if err = roomserverAPI.SendEventWithState(
|
||||
if err = roomserverAPI.SendEventWithRewrite(
|
||||
ctx, r.rsAPI,
|
||||
respState,
|
||||
event.Headered(respMakeJoin.RoomVersion),
|
||||
|
|
|
|||
|
|
@ -33,6 +33,10 @@ const (
|
|||
// KindBackfill event extend the contiguous graph going backwards.
|
||||
// They always have state.
|
||||
KindBackfill = 3
|
||||
// KindRewrite events are used when rewriting the head of the room
|
||||
// graph with entirely new state. The output events generated will
|
||||
// be state events rather than timeline events.
|
||||
KindRewrite = 4
|
||||
)
|
||||
|
||||
// DoNotSendToOtherServers tells us not to send the event to other matrix
|
||||
|
|
|
|||
|
|
@ -68,6 +68,17 @@ type OutputEvent struct {
|
|||
NewPeek *OutputNewPeek `json:"new_peek,omitempty"`
|
||||
}
|
||||
|
||||
// Type of the OutputNewRoomEvent.
|
||||
type OutputRoomEventType int
|
||||
|
||||
const (
|
||||
// The event is a timeline event and likely just happened.
|
||||
OutputRoomTimeline OutputRoomEventType = iota
|
||||
|
||||
// The event is a state event and quite possibly happened in the past.
|
||||
OutputRoomState
|
||||
)
|
||||
|
||||
// An OutputNewRoomEvent is written when the roomserver receives a new event.
|
||||
// It contains the full matrix room event and enough information for a
|
||||
// consumer to construct the current state of the room and the state before the
|
||||
|
|
@ -80,6 +91,9 @@ type OutputEvent struct {
|
|||
type OutputNewRoomEvent struct {
|
||||
// The Event.
|
||||
Event gomatrixserverlib.HeaderedEvent `json:"event"`
|
||||
// Does the event completely rewrite the room state? If so, then AddsStateEventIDs
|
||||
// will contain the entire room state.
|
||||
RewritesState bool `json:"rewrites_state"`
|
||||
// The latest events in the room after this event.
|
||||
// This can be used to set the prev events for new events in the room.
|
||||
// This also can be used to get the full current state after this event.
|
||||
|
|
|
|||
|
|
@ -80,6 +80,107 @@ func SendEventWithState(
|
|||
return SendInputRoomEvents(ctx, rsAPI, ires)
|
||||
}
|
||||
|
||||
// SendEventWithRewrite writes an event with KindNew to the roomserver along
|
||||
// with a number of rewrite and outlier events for state and auth events
|
||||
// respectively.
|
||||
func SendEventWithRewrite(
|
||||
ctx context.Context, rsAPI RoomserverInternalAPI, state *gomatrixserverlib.RespState,
|
||||
event gomatrixserverlib.HeaderedEvent, haveEventIDs map[string]bool,
|
||||
) error {
|
||||
isCurrentState := map[string]struct{}{}
|
||||
for _, se := range state.StateEvents {
|
||||
isCurrentState[se.EventID()] = struct{}{}
|
||||
}
|
||||
|
||||
authAndStateEvents, err := state.Events()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var ires []InputRoomEvent
|
||||
var stateIDs []string
|
||||
|
||||
// This function generates three things:
|
||||
// A - A set of "rewrite" events, which will form the newly rewritten
|
||||
// state before the event, which includes every rewrite event that
|
||||
// came before it in its state
|
||||
// B - A set of "outlier" events, which are auth events but not part
|
||||
// of the rewritten state
|
||||
// C - A "new" event, which include all of the rewrite events in its
|
||||
// state
|
||||
for _, authOrStateEvent := range authAndStateEvents {
|
||||
if authOrStateEvent.StateKey() == nil {
|
||||
continue
|
||||
}
|
||||
if haveEventIDs[authOrStateEvent.EventID()] {
|
||||
continue
|
||||
}
|
||||
if event.StateKey() == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// We will handle an event as if it's an outlier if one of the
|
||||
// following conditions is true:
|
||||
storeAsOutlier := false
|
||||
if authOrStateEvent.Type() == event.Type() && *authOrStateEvent.StateKey() == *event.StateKey() {
|
||||
// The event is a state event but the input event is going to
|
||||
// replace it, therefore it can't be added to the state or we'll
|
||||
// get duplicate state keys in the state block. We'll send it
|
||||
// as an outlier because we don't know if something will be
|
||||
// referring to it as an auth event, but need it to be stored
|
||||
// just in case.
|
||||
storeAsOutlier = true
|
||||
} else if _, ok := isCurrentState[authOrStateEvent.EventID()]; !ok {
|
||||
// The event is an auth event and isn't a part of the state set.
|
||||
// We'll send it as an outlier because we need it to be stored
|
||||
// in case something is referring to it as an auth event.
|
||||
storeAsOutlier = true
|
||||
}
|
||||
|
||||
if storeAsOutlier {
|
||||
ires = append(ires, InputRoomEvent{
|
||||
Kind: KindOutlier,
|
||||
Event: authOrStateEvent.Headered(event.RoomVersion),
|
||||
AuthEventIDs: authOrStateEvent.AuthEventIDs(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
// If the event isn't an outlier then we'll instead send it as a
|
||||
// rewrite event, so that it'll form part of the rewritten state.
|
||||
// These events will go through the membership and latest event
|
||||
// updaters and we will generate output events, but they will be
|
||||
// flagged as non-current (i.e. didn't just happen) events.
|
||||
// Each of these rewrite events includes all of the rewrite events
|
||||
// that came before in their StateEventIDs.
|
||||
ires = append(ires, InputRoomEvent{
|
||||
Kind: KindRewrite,
|
||||
Event: authOrStateEvent.Headered(event.RoomVersion),
|
||||
AuthEventIDs: authOrStateEvent.AuthEventIDs(),
|
||||
HasState: true,
|
||||
StateEventIDs: stateIDs,
|
||||
})
|
||||
|
||||
// Add the event ID into the StateEventIDs of all subsequent
|
||||
// rewrite events, and the new event.
|
||||
stateIDs = append(stateIDs, authOrStateEvent.EventID())
|
||||
}
|
||||
|
||||
// Send the final event as a new event, which will generate
|
||||
// a timeline output event for it. All of the rewrite events
|
||||
// that came before will be sent as StateEventIDs, forming a
|
||||
// new clean state before the event.
|
||||
ires = append(ires, InputRoomEvent{
|
||||
Kind: KindNew,
|
||||
Event: event,
|
||||
AuthEventIDs: event.AuthEventIDs(),
|
||||
HasState: true,
|
||||
StateEventIDs: stateIDs,
|
||||
})
|
||||
|
||||
return SendInputRoomEvents(ctx, rsAPI, ires)
|
||||
}
|
||||
|
||||
// SendInputRoomEvents to the roomserver.
|
||||
func SendInputRoomEvents(
|
||||
ctx context.Context, rsAPI RoomserverInternalAPI, ires []InputRoomEvent,
|
||||
|
|
|
|||
|
|
@ -36,7 +36,7 @@ func CheckAuthEvents(
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// TODO: check for duplicate state keys here.
|
||||
authStateEntries = types.DeduplicateStateEntries(authStateEntries)
|
||||
|
||||
// Work out which of the state events we actually need.
|
||||
stateNeeded := gomatrixserverlib.StateNeededForAuth([]gomatrixserverlib.Event{event.Unwrap()})
|
||||
|
|
|
|||
|
|
@ -86,7 +86,7 @@ func (r *Inputer) processRoomEvent(
|
|||
"event_id": event.EventID(),
|
||||
"type": event.Type(),
|
||||
"room": event.RoomID(),
|
||||
}).Info("Stored outlier")
|
||||
}).Debug("Stored outlier")
|
||||
return event.EventID(), nil
|
||||
}
|
||||
|
||||
|
|
@ -107,6 +107,15 @@ func (r *Inputer) processRoomEvent(
|
|||
}
|
||||
}
|
||||
|
||||
if input.Kind == api.KindRewrite {
|
||||
logrus.WithFields(logrus.Fields{
|
||||
"event_id": event.EventID(),
|
||||
"type": event.Type(),
|
||||
"room": event.RoomID(),
|
||||
}).Debug("Stored rewrite")
|
||||
return event.EventID(), nil
|
||||
}
|
||||
|
||||
if err = r.updateLatestEvents(
|
||||
ctx, // context
|
||||
roomInfo, // room info for the room being updated
|
||||
|
|
@ -114,6 +123,7 @@ func (r *Inputer) processRoomEvent(
|
|||
event, // event
|
||||
input.SendAsServer, // send as server
|
||||
input.TransactionID, // transaction ID
|
||||
input.HasState, // rewrites state?
|
||||
); err != nil {
|
||||
return "", fmt.Errorf("r.updateLatestEvents: %w", err)
|
||||
}
|
||||
|
|
@ -167,19 +177,25 @@ func (r *Inputer) calculateAndSetState(
|
|||
// Check that those state events are in the database and store the state.
|
||||
var entries []types.StateEntry
|
||||
if entries, err = r.DB.StateEntriesForEventIDs(ctx, input.StateEventIDs); err != nil {
|
||||
return err
|
||||
return fmt.Errorf("r.DB.StateEntriesForEventIDs: %w", err)
|
||||
}
|
||||
entries = types.DeduplicateStateEntries(entries)
|
||||
|
||||
if stateAtEvent.BeforeStateSnapshotNID, err = r.DB.AddState(ctx, roomInfo.RoomNID, nil, entries); err != nil {
|
||||
return err
|
||||
return fmt.Errorf("r.DB.AddState: %w", err)
|
||||
}
|
||||
} else {
|
||||
stateAtEvent.Overwrite = false
|
||||
|
||||
// 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); err != nil {
|
||||
return err
|
||||
return fmt.Errorf("roomState.CalculateAndStoreStateBeforeEvent: %w", err)
|
||||
}
|
||||
}
|
||||
return r.DB.SetState(ctx, stateAtEvent.EventNID, stateAtEvent.BeforeStateSnapshotNID)
|
||||
|
||||
err = r.DB.SetState(ctx, stateAtEvent.EventNID, stateAtEvent.BeforeStateSnapshotNID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("r.DB.SetState: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -54,6 +54,7 @@ func (r *Inputer) updateLatestEvents(
|
|||
event gomatrixserverlib.Event,
|
||||
sendAsServer string,
|
||||
transactionID *api.TransactionID,
|
||||
rewritesState bool,
|
||||
) (err error) {
|
||||
updater, err := r.DB.GetLatestEventsForUpdate(ctx, *roomInfo)
|
||||
if err != nil {
|
||||
|
|
@ -71,6 +72,7 @@ func (r *Inputer) updateLatestEvents(
|
|||
event: event,
|
||||
sendAsServer: sendAsServer,
|
||||
transactionID: transactionID,
|
||||
rewritesState: rewritesState,
|
||||
}
|
||||
|
||||
if err = u.doUpdateLatestEvents(); err != nil {
|
||||
|
|
@ -93,6 +95,7 @@ type latestEventsUpdater struct {
|
|||
stateAtEvent types.StateAtEvent
|
||||
event gomatrixserverlib.Event
|
||||
transactionID *api.TransactionID
|
||||
rewritesState bool
|
||||
// Which server to send this event as.
|
||||
sendAsServer string
|
||||
// The eventID of the event that was processed before this one.
|
||||
|
|
@ -178,7 +181,8 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
|
|||
return fmt.Errorf("u.api.updateMemberships: %w", err)
|
||||
}
|
||||
|
||||
update, err := u.makeOutputNewRoomEvent()
|
||||
var update *api.OutputEvent
|
||||
update, err = u.makeOutputNewRoomEvent()
|
||||
if err != nil {
|
||||
return fmt.Errorf("u.makeOutputNewRoomEvent: %w", err)
|
||||
}
|
||||
|
|
@ -305,6 +309,7 @@ func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error)
|
|||
|
||||
ore := api.OutputNewRoomEvent{
|
||||
Event: u.event.Headered(u.roomInfo.RoomVersion),
|
||||
RewritesState: u.rewritesState,
|
||||
LastSentEventID: u.lastEventIDSent,
|
||||
LatestEventIDs: latestEventIDs,
|
||||
TransactionID: u.transactionID,
|
||||
|
|
@ -337,6 +342,11 @@ func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error)
|
|||
return nil, fmt.Errorf("failed to load add_state_events from db: %w", err)
|
||||
}
|
||||
}
|
||||
// State is rewritten if the input room event HasState and we actually produced a delta on state events.
|
||||
// Without this check, /get_missing_events which produce events with associated (but not complete) state
|
||||
// will incorrectly purge the room and set it to no state. TODO: This is likely flakey, as if /gme produced
|
||||
// a state conflict res which just so happens to include 2+ events we might purge the room state downstream.
|
||||
ore.RewritesState = len(ore.AddsStateEventIDs) > 1
|
||||
|
||||
return &api.OutputEvent{
|
||||
Type: api.OutputTypeNewRoomEvent,
|
||||
|
|
|
|||
|
|
@ -183,33 +183,33 @@ func (r *Joiner) performJoinRoomByID(
|
|||
return "", fmt.Errorf("eb.SetContent: %w", err)
|
||||
}
|
||||
|
||||
// First work out if this is in response to an existing invite
|
||||
// from a federated server. If it is then we avoid the situation
|
||||
// where we might think we know about a room in the following
|
||||
// section but don't know the latest state as all of our users
|
||||
// have left.
|
||||
// Force a federated join if we aren't in the room and we've been
|
||||
// given some server names to try joining by.
|
||||
serverInRoom, _ := helpers.IsServerCurrentlyInRoom(ctx, r.DB, r.ServerName, req.RoomIDOrAlias)
|
||||
forceFederatedJoin := len(req.ServerNames) > 0 && !serverInRoom
|
||||
|
||||
// Force a federated join if we're dealing with a pending invite
|
||||
// and we aren't in the room.
|
||||
isInvitePending, inviteSender, _, err := helpers.IsInvitePending(ctx, r.DB, req.RoomIDOrAlias, req.UserID)
|
||||
if err == nil && isInvitePending && !serverInRoom {
|
||||
// Check if there's an invite pending.
|
||||
if err == nil && isInvitePending {
|
||||
_, inviterDomain, ierr := gomatrixserverlib.SplitID('@', inviteSender)
|
||||
if ierr != nil {
|
||||
return "", fmt.Errorf("gomatrixserverlib.SplitID: %w", err)
|
||||
}
|
||||
|
||||
// Check that the domain isn't ours. If it's local then we don't
|
||||
// need to do anything as our own copy of the room state will be
|
||||
// up-to-date.
|
||||
// If we were invited by someone from another server then we can
|
||||
// assume they are in the room so we can join via them.
|
||||
if inviterDomain != r.Cfg.Matrix.ServerName {
|
||||
// Add the server of the person who invited us to the server list,
|
||||
// as they should be a fairly good bet.
|
||||
req.ServerNames = append(req.ServerNames, inviterDomain)
|
||||
|
||||
// Perform a federated room join.
|
||||
return req.RoomIDOrAlias, r.performFederatedJoinRoomByID(ctx, req)
|
||||
forceFederatedJoin = true
|
||||
}
|
||||
}
|
||||
|
||||
// If we should do a forced federated join then do that.
|
||||
if forceFederatedJoin {
|
||||
return req.RoomIDOrAlias, r.performFederatedJoinRoomByID(ctx, req)
|
||||
}
|
||||
|
||||
// Try to construct an actual join event from the template.
|
||||
// If this succeeds then it is a sign that the room already exists
|
||||
// locally on the homeserver.
|
||||
|
|
|
|||
|
|
@ -1,12 +1,15 @@
|
|||
package roomserver
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/ed25519"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Shopify/sarama"
|
||||
"github.com/matrix-org/dendrite/internal/caching"
|
||||
|
|
@ -80,7 +83,73 @@ func deleteDatabase() {
|
|||
}
|
||||
}
|
||||
|
||||
func mustLoadEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []json.RawMessage) []gomatrixserverlib.HeaderedEvent {
|
||||
type fledglingEvent struct {
|
||||
Type string
|
||||
StateKey *string
|
||||
Content interface{}
|
||||
Sender string
|
||||
RoomID string
|
||||
}
|
||||
|
||||
func mustCreateEvents(t *testing.T, roomVer gomatrixserverlib.RoomVersion, events []fledglingEvent) (result []gomatrixserverlib.HeaderedEvent) {
|
||||
t.Helper()
|
||||
depth := int64(1)
|
||||
seed := make([]byte, ed25519.SeedSize) // zero seed
|
||||
key := ed25519.NewKeyFromSeed(seed)
|
||||
var prevs []string
|
||||
roomState := make(map[gomatrixserverlib.StateKeyTuple]string) // state -> event ID
|
||||
for _, ev := range events {
|
||||
eb := gomatrixserverlib.EventBuilder{
|
||||
Sender: ev.Sender,
|
||||
Depth: depth,
|
||||
Type: ev.Type,
|
||||
StateKey: ev.StateKey,
|
||||
RoomID: ev.RoomID,
|
||||
PrevEvents: prevs,
|
||||
}
|
||||
err := eb.SetContent(ev.Content)
|
||||
if err != nil {
|
||||
t.Fatalf("mustCreateEvent: failed to marshal event content %+v", ev.Content)
|
||||
}
|
||||
stateNeeded, err := gomatrixserverlib.StateNeededForEventBuilder(&eb)
|
||||
if err != nil {
|
||||
t.Fatalf("mustCreateEvent: failed to work out auth_events : %s", err)
|
||||
}
|
||||
var authEvents []string
|
||||
for _, tuple := range stateNeeded.Tuples() {
|
||||
eventID := roomState[tuple]
|
||||
if eventID != "" {
|
||||
authEvents = append(authEvents, eventID)
|
||||
}
|
||||
}
|
||||
eb.AuthEvents = authEvents
|
||||
signedEvent, err := eb.Build(time.Now(), testOrigin, "ed25519:test", key, roomVer)
|
||||
if err != nil {
|
||||
t.Fatalf("mustCreateEvent: failed to sign event: %s", err)
|
||||
}
|
||||
depth++
|
||||
prevs = []string{signedEvent.EventID()}
|
||||
if ev.StateKey != nil {
|
||||
roomState[gomatrixserverlib.StateKeyTuple{
|
||||
EventType: ev.Type,
|
||||
StateKey: *ev.StateKey,
|
||||
}] = signedEvent.EventID()
|
||||
}
|
||||
result = append(result, signedEvent.Headered(roomVer))
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func eventsJSON(events []gomatrixserverlib.Event) []json.RawMessage {
|
||||
result := make([]json.RawMessage, len(events))
|
||||
for i := range events {
|
||||
result[i] = events[i].JSON()
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func mustLoadRawEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []json.RawMessage) []gomatrixserverlib.HeaderedEvent {
|
||||
t.Helper()
|
||||
hs := make([]gomatrixserverlib.HeaderedEvent, len(events))
|
||||
for i := range events {
|
||||
e, err := gomatrixserverlib.NewEventFromTrustedJSON(events[i], false, ver)
|
||||
|
|
@ -93,7 +162,8 @@ func mustLoadEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []js
|
|||
return hs
|
||||
}
|
||||
|
||||
func mustSendEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []json.RawMessage) (api.RoomserverInternalAPI, *dummyProducer, []gomatrixserverlib.HeaderedEvent) {
|
||||
func mustCreateRoomserverAPI(t *testing.T) (api.RoomserverInternalAPI, *dummyProducer) {
|
||||
t.Helper()
|
||||
cfg := &config.Dendrite{}
|
||||
cfg.Defaults()
|
||||
cfg.Global.ServerName = testOrigin
|
||||
|
|
@ -112,9 +182,14 @@ func mustSendEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []js
|
|||
Cfg: cfg,
|
||||
}
|
||||
|
||||
rsAPI := NewInternalAPI(base, &test.NopJSONVerifier{})
|
||||
hevents := mustLoadEvents(t, ver, events)
|
||||
if err = api.SendEvents(ctx, rsAPI, hevents, testOrigin, nil); err != nil {
|
||||
return NewInternalAPI(base, &test.NopJSONVerifier{}), dp
|
||||
}
|
||||
|
||||
func mustSendEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []json.RawMessage) (api.RoomserverInternalAPI, *dummyProducer, []gomatrixserverlib.HeaderedEvent) {
|
||||
t.Helper()
|
||||
rsAPI, dp := mustCreateRoomserverAPI(t)
|
||||
hevents := mustLoadRawEvents(t, ver, events)
|
||||
if err := api.SendEvents(ctx, rsAPI, hevents, testOrigin, nil); err != nil {
|
||||
t.Errorf("failed to SendEvents: %s", err)
|
||||
}
|
||||
return rsAPI, dp, hevents
|
||||
|
|
@ -170,3 +245,163 @@ func TestOutputRedactedEvent(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// This tests that rewriting state via KindRewrite works correctly.
|
||||
// This creates a small room with a create/join/name state, then replays it
|
||||
// with a new room name. We expect the output events to contain the original events,
|
||||
// followed by a single OutputNewRoomEvent with RewritesState set to true with the
|
||||
// rewritten state events (with the 2nd room name).
|
||||
func TestOutputRewritesState(t *testing.T) {
|
||||
roomID := "!foo:" + string(testOrigin)
|
||||
alice := "@alice:" + string(testOrigin)
|
||||
emptyKey := ""
|
||||
originalEvents := mustCreateEvents(t, gomatrixserverlib.RoomVersionV6, []fledglingEvent{
|
||||
{
|
||||
RoomID: roomID,
|
||||
Sender: alice,
|
||||
Content: map[string]interface{}{
|
||||
"creator": alice,
|
||||
"room_version": "6",
|
||||
},
|
||||
StateKey: &emptyKey,
|
||||
Type: gomatrixserverlib.MRoomCreate,
|
||||
},
|
||||
{
|
||||
RoomID: roomID,
|
||||
Sender: alice,
|
||||
Content: map[string]interface{}{
|
||||
"membership": "join",
|
||||
},
|
||||
StateKey: &alice,
|
||||
Type: gomatrixserverlib.MRoomMember,
|
||||
},
|
||||
{
|
||||
RoomID: roomID,
|
||||
Sender: alice,
|
||||
Content: map[string]interface{}{
|
||||
"body": "hello world",
|
||||
},
|
||||
StateKey: nil,
|
||||
Type: "m.room.message",
|
||||
},
|
||||
{
|
||||
RoomID: roomID,
|
||||
Sender: alice,
|
||||
Content: map[string]interface{}{
|
||||
"name": "Room Name",
|
||||
},
|
||||
StateKey: &emptyKey,
|
||||
Type: "m.room.name",
|
||||
},
|
||||
})
|
||||
rewriteEvents := mustCreateEvents(t, gomatrixserverlib.RoomVersionV6, []fledglingEvent{
|
||||
{
|
||||
RoomID: roomID,
|
||||
Sender: alice,
|
||||
Content: map[string]interface{}{
|
||||
"creator": alice,
|
||||
},
|
||||
StateKey: &emptyKey,
|
||||
Type: gomatrixserverlib.MRoomCreate,
|
||||
},
|
||||
{
|
||||
RoomID: roomID,
|
||||
Sender: alice,
|
||||
Content: map[string]interface{}{
|
||||
"membership": "join",
|
||||
},
|
||||
StateKey: &alice,
|
||||
Type: gomatrixserverlib.MRoomMember,
|
||||
},
|
||||
{
|
||||
RoomID: roomID,
|
||||
Sender: alice,
|
||||
Content: map[string]interface{}{
|
||||
"name": "Room Name 2",
|
||||
},
|
||||
StateKey: &emptyKey,
|
||||
Type: "m.room.name",
|
||||
},
|
||||
{
|
||||
RoomID: roomID,
|
||||
Sender: alice,
|
||||
Content: map[string]interface{}{
|
||||
"body": "hello world 2",
|
||||
},
|
||||
StateKey: nil,
|
||||
Type: "m.room.message",
|
||||
},
|
||||
})
|
||||
deleteDatabase()
|
||||
rsAPI, producer := mustCreateRoomserverAPI(t)
|
||||
defer deleteDatabase()
|
||||
err := api.SendEvents(context.Background(), rsAPI, originalEvents, testOrigin, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("failed to send original events: %s", err)
|
||||
}
|
||||
// assert we got them produced, this is just a sanity check and isn't the intention of this test
|
||||
if len(producer.producedMessages) != len(originalEvents) {
|
||||
t.Fatalf("SendEvents didn't result in same number of produced output events: got %d want %d", len(producer.producedMessages), len(originalEvents))
|
||||
}
|
||||
producer.producedMessages = nil // we aren't actually interested in these events, just the rewrite ones
|
||||
|
||||
var inputEvents []api.InputRoomEvent
|
||||
// slowly build up the state IDs again, we're basically telling the roomserver what to store as a snapshot
|
||||
var stateIDs []string
|
||||
// skip the last event, we'll use this to tie together the rewrite as the KindNew event
|
||||
for i := 0; i < len(rewriteEvents)-1; i++ {
|
||||
ev := rewriteEvents[i]
|
||||
inputEvents = append(inputEvents, api.InputRoomEvent{
|
||||
Kind: api.KindRewrite,
|
||||
Event: ev,
|
||||
AuthEventIDs: ev.AuthEventIDs(),
|
||||
HasState: true,
|
||||
StateEventIDs: stateIDs,
|
||||
})
|
||||
if ev.StateKey() != nil {
|
||||
stateIDs = append(stateIDs, ev.EventID())
|
||||
}
|
||||
}
|
||||
lastEv := rewriteEvents[len(rewriteEvents)-1]
|
||||
inputEvents = append(inputEvents, api.InputRoomEvent{
|
||||
Kind: api.KindNew,
|
||||
Event: lastEv,
|
||||
AuthEventIDs: lastEv.AuthEventIDs(),
|
||||
HasState: true,
|
||||
StateEventIDs: stateIDs,
|
||||
})
|
||||
if err := api.SendInputRoomEvents(context.Background(), rsAPI, inputEvents); err != nil {
|
||||
t.Fatalf("SendInputRoomEvents returned error for rewrite events: %s", err)
|
||||
}
|
||||
// we should just have one output event with the entire state of the room in it
|
||||
if len(producer.producedMessages) != 1 {
|
||||
t.Fatalf("Rewritten events got output, want only 1 got %d", len(producer.producedMessages))
|
||||
}
|
||||
outputEvent := producer.producedMessages[0]
|
||||
if !outputEvent.NewRoomEvent.RewritesState {
|
||||
t.Errorf("RewritesState flag not set on output event")
|
||||
}
|
||||
if !reflect.DeepEqual(stateIDs, outputEvent.NewRoomEvent.AddsStateEventIDs) {
|
||||
t.Errorf("Output event is missing room state event IDs, got %v want %v", outputEvent.NewRoomEvent.AddsStateEventIDs, stateIDs)
|
||||
}
|
||||
if !bytes.Equal(outputEvent.NewRoomEvent.Event.JSON(), lastEv.JSON()) {
|
||||
t.Errorf(
|
||||
"Output event isn't the latest KindNew event:\ngot %s\nwant %s",
|
||||
string(outputEvent.NewRoomEvent.Event.JSON()),
|
||||
string(lastEv.JSON()),
|
||||
)
|
||||
}
|
||||
if len(outputEvent.NewRoomEvent.AddStateEvents) != len(stateIDs) {
|
||||
t.Errorf("Output event is missing room state events themselves, got %d want %d", len(outputEvent.NewRoomEvent.AddStateEvents), len(stateIDs))
|
||||
}
|
||||
// make sure the state got overwritten, check the room name
|
||||
hasRoomName := false
|
||||
for _, ev := range outputEvent.NewRoomEvent.AddStateEvents {
|
||||
if ev.Type() == "m.room.name" {
|
||||
hasRoomName = string(ev.Content()) == `{"name":"Room Name 2"}`
|
||||
}
|
||||
}
|
||||
if !hasRoomName {
|
||||
t.Errorf("Output event did not overwrite room state")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,6 +16,8 @@
|
|||
package types
|
||||
|
||||
import (
|
||||
"sort"
|
||||
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
|
|
@ -72,6 +74,25 @@ func (a StateEntry) LessThan(b StateEntry) bool {
|
|||
return a.EventNID < b.EventNID
|
||||
}
|
||||
|
||||
// Deduplicate takes a set of state entries and ensures that there are no
|
||||
// duplicate (event type, state key) tuples. If there are then we dedupe
|
||||
// them, making sure that the latest/highest NIDs are always chosen.
|
||||
func DeduplicateStateEntries(a []StateEntry) []StateEntry {
|
||||
if len(a) < 2 {
|
||||
return a
|
||||
}
|
||||
sort.SliceStable(a, func(i, j int) bool {
|
||||
return a[i].LessThan(a[j])
|
||||
})
|
||||
for i := 0; i < len(a)-1; i++ {
|
||||
if a[i].StateKeyTuple == a[i+1].StateKeyTuple {
|
||||
a = append(a[:i], a[i+1:]...)
|
||||
i--
|
||||
}
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// StateAtEvent is the state before and after a matrix event.
|
||||
type StateAtEvent struct {
|
||||
// Should this state overwrite the latest events and memberships of the room?
|
||||
|
|
|
|||
|
|
@ -0,0 +1,26 @@
|
|||
package types
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestDeduplicateStateEntries(t *testing.T) {
|
||||
entries := []StateEntry{
|
||||
{StateKeyTuple{1, 1}, 1},
|
||||
{StateKeyTuple{1, 1}, 2},
|
||||
{StateKeyTuple{1, 1}, 3},
|
||||
{StateKeyTuple{2, 2}, 4},
|
||||
{StateKeyTuple{2, 3}, 5},
|
||||
{StateKeyTuple{3, 3}, 6},
|
||||
}
|
||||
expected := []EventNID{3, 4, 5, 6}
|
||||
entries = DeduplicateStateEntries(entries)
|
||||
if len(entries) != 4 {
|
||||
t.Fatalf("Expected 4 entries, got %d entries", len(entries))
|
||||
}
|
||||
for i, v := range entries {
|
||||
if v.EventNID != expected[i] {
|
||||
t.Fatalf("Expected position %d to be %d but got %d", i, expected[i], v.EventNID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -149,6 +149,12 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
|
|||
}
|
||||
}
|
||||
|
||||
if msg.RewritesState {
|
||||
if err = s.db.PurgeRoom(ctx, ev.RoomID()); err != nil {
|
||||
return fmt.Errorf("s.db.PurgeRoom: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
pduPos, err := s.db.WriteEvent(
|
||||
ctx,
|
||||
&ev,
|
||||
|
|
|
|||
|
|
@ -43,6 +43,9 @@ type Database interface {
|
|||
// Returns an error if there was a problem inserting this event.
|
||||
WriteEvent(ctx context.Context, ev *gomatrixserverlib.HeaderedEvent, addStateEvents []gomatrixserverlib.HeaderedEvent,
|
||||
addStateEventIDs []string, removeStateEventIDs []string, transactionID *api.TransactionID, excludeFromSync bool) (types.StreamPosition, error)
|
||||
// PurgeRoom completely purges room state from the sync API. This is done when
|
||||
// receiving an output event that completely resets the state.
|
||||
PurgeRoom(ctx context.Context, roomID string) error
|
||||
// GetStateEvent returns the Matrix state event of a given type for a given room with a given state key
|
||||
// If no event could be found, returns nil
|
||||
// If there was an issue during the retrieval, returns an error
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
)
|
||||
|
||||
|
|
@ -46,10 +47,14 @@ const selectBackwardExtremitiesForRoomSQL = "" +
|
|||
const deleteBackwardExtremitySQL = "" +
|
||||
"DELETE FROM syncapi_backward_extremities WHERE room_id = $1 AND prev_event_id = $2"
|
||||
|
||||
const deleteBackwardExtremitiesForRoomSQL = "" +
|
||||
"DELETE FROM syncapi_backward_extremities WHERE room_id = $1"
|
||||
|
||||
type backwardExtremitiesStatements struct {
|
||||
insertBackwardExtremityStmt *sql.Stmt
|
||||
selectBackwardExtremitiesForRoomStmt *sql.Stmt
|
||||
deleteBackwardExtremityStmt *sql.Stmt
|
||||
deleteBackwardExtremitiesForRoomStmt *sql.Stmt
|
||||
}
|
||||
|
||||
func NewPostgresBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremities, error) {
|
||||
|
|
@ -67,6 +72,9 @@ func NewPostgresBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremiti
|
|||
if s.deleteBackwardExtremityStmt, err = db.Prepare(deleteBackwardExtremitySQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.deleteBackwardExtremitiesForRoomStmt, err = db.Prepare(deleteBackwardExtremitiesForRoomSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
|
|
@ -105,3 +113,10 @@ func (s *backwardExtremitiesStatements) DeleteBackwardExtremity(
|
|||
_, err = txn.Stmt(s.deleteBackwardExtremityStmt).ExecContext(ctx, roomID, knownEventID)
|
||||
return
|
||||
}
|
||||
|
||||
func (s *backwardExtremitiesStatements) DeleteBackwardExtremitiesForRoom(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) (err error) {
|
||||
_, err = sqlutil.TxStmt(txn, s.deleteBackwardExtremitiesForRoomStmt).ExecContext(ctx, roomID)
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -69,6 +69,9 @@ const upsertRoomStateSQL = "" +
|
|||
const deleteRoomStateByEventIDSQL = "" +
|
||||
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
|
||||
|
||||
const DeleteRoomStateForRoomSQL = "" +
|
||||
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
|
||||
|
||||
const selectRoomIDsWithMembershipSQL = "" +
|
||||
"SELECT DISTINCT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
|
||||
|
||||
|
|
@ -98,6 +101,7 @@ const selectEventsWithEventIDsSQL = "" +
|
|||
type currentRoomStateStatements struct {
|
||||
upsertRoomStateStmt *sql.Stmt
|
||||
deleteRoomStateByEventIDStmt *sql.Stmt
|
||||
DeleteRoomStateForRoomStmt *sql.Stmt
|
||||
selectRoomIDsWithMembershipStmt *sql.Stmt
|
||||
selectCurrentStateStmt *sql.Stmt
|
||||
selectJoinedUsersStmt *sql.Stmt
|
||||
|
|
@ -117,6 +121,9 @@ func NewPostgresCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, erro
|
|||
if s.deleteRoomStateByEventIDStmt, err = db.Prepare(deleteRoomStateByEventIDSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.DeleteRoomStateForRoomStmt, err = db.Prepare(DeleteRoomStateForRoomSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.selectRoomIDsWithMembershipStmt, err = db.Prepare(selectRoomIDsWithMembershipSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -214,6 +221,14 @@ func (s *currentRoomStateStatements) DeleteRoomStateByEventID(
|
|||
return err
|
||||
}
|
||||
|
||||
func (s *currentRoomStateStatements) DeleteRoomStateForRoom(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) error {
|
||||
stmt := sqlutil.TxStmt(txn, s.DeleteRoomStateForRoomStmt)
|
||||
_, err := stmt.ExecContext(ctx, roomID)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *currentRoomStateStatements) UpsertRoomState(
|
||||
ctx context.Context, txn *sql.Tx,
|
||||
event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition,
|
||||
|
|
|
|||
|
|
@ -115,6 +115,9 @@ const selectStateInRangeSQL = "" +
|
|||
" ORDER BY id ASC" +
|
||||
" LIMIT $8"
|
||||
|
||||
const deleteEventsForRoomSQL = "" +
|
||||
"DELETE FROM syncapi_output_room_events WHERE room_id = $1"
|
||||
|
||||
type outputRoomEventsStatements struct {
|
||||
insertEventStmt *sql.Stmt
|
||||
selectEventsStmt *sql.Stmt
|
||||
|
|
@ -124,6 +127,7 @@ type outputRoomEventsStatements struct {
|
|||
selectEarlyEventsStmt *sql.Stmt
|
||||
selectStateInRangeStmt *sql.Stmt
|
||||
updateEventJSONStmt *sql.Stmt
|
||||
deleteEventsForRoomStmt *sql.Stmt
|
||||
}
|
||||
|
||||
func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
|
||||
|
|
@ -156,6 +160,9 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
|
|||
if s.updateEventJSONStmt, err = db.Prepare(updateEventJSONSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.deleteEventsForRoomStmt, err = db.Prepare(deleteEventsForRoomSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
|
|
@ -395,6 +402,13 @@ func (s *outputRoomEventsStatements) SelectEvents(
|
|||
return rowsToStreamEvents(rows)
|
||||
}
|
||||
|
||||
func (s *outputRoomEventsStatements) DeleteEventsForRoom(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) (err error) {
|
||||
_, err = sqlutil.TxStmt(txn, s.deleteEventsForRoomStmt).ExecContext(ctx, roomID)
|
||||
return err
|
||||
}
|
||||
|
||||
func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
|
||||
var result []types.StreamEvent
|
||||
for rows.Next() {
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
|
@ -71,12 +72,16 @@ const selectMaxPositionInTopologySQL = "" +
|
|||
"SELECT MAX(topological_position) FROM syncapi_output_room_events_topology WHERE room_id=$1" +
|
||||
") ORDER BY stream_position DESC LIMIT 1"
|
||||
|
||||
const deleteTopologyForRoomSQL = "" +
|
||||
"DELETE FROM syncapi_output_room_events_topology WHERE room_id = $1"
|
||||
|
||||
type outputRoomEventsTopologyStatements struct {
|
||||
insertEventInTopologyStmt *sql.Stmt
|
||||
selectEventIDsInRangeASCStmt *sql.Stmt
|
||||
selectEventIDsInRangeDESCStmt *sql.Stmt
|
||||
selectPositionInTopologyStmt *sql.Stmt
|
||||
selectMaxPositionInTopologyStmt *sql.Stmt
|
||||
deleteTopologyForRoomStmt *sql.Stmt
|
||||
}
|
||||
|
||||
func NewPostgresTopologyTable(db *sql.DB) (tables.Topology, error) {
|
||||
|
|
@ -100,6 +105,9 @@ func NewPostgresTopologyTable(db *sql.DB) (tables.Topology, error) {
|
|||
if s.selectMaxPositionInTopologyStmt, err = db.Prepare(selectMaxPositionInTopologySQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.deleteTopologyForRoomStmt, err = db.Prepare(deleteTopologyForRoomSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
|
|
@ -167,3 +175,10 @@ func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology(
|
|||
err = s.selectMaxPositionInTopologyStmt.QueryRowContext(ctx, roomID).Scan(&pos, &spos)
|
||||
return
|
||||
}
|
||||
|
||||
func (s *outputRoomEventsTopologyStatements) DeleteTopologyForRoom(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) (err error) {
|
||||
_, err = sqlutil.TxStmt(txn, s.deleteTopologyForRoomStmt).ExecContext(ctx, roomID)
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -276,6 +276,29 @@ func (d *Database) handleBackwardExtremities(ctx context.Context, txn *sql.Tx, e
|
|||
return nil
|
||||
}
|
||||
|
||||
func (d *Database) PurgeRoom(
|
||||
ctx context.Context, roomID string,
|
||||
) error {
|
||||
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
|
||||
// If the event is a create event then we'll delete all of the existing
|
||||
// data for the room. The only reason that a create event would be replayed
|
||||
// to us in this way is if we're about to receive the entire room state.
|
||||
if err := d.CurrentRoomState.DeleteRoomStateForRoom(ctx, txn, roomID); err != nil {
|
||||
return fmt.Errorf("d.CurrentRoomState.DeleteRoomStateForRoom: %w", err)
|
||||
}
|
||||
if err := d.OutputEvents.DeleteEventsForRoom(ctx, txn, roomID); err != nil {
|
||||
return fmt.Errorf("d.Events.DeleteEventsForRoom: %w", err)
|
||||
}
|
||||
if err := d.Topology.DeleteTopologyForRoom(ctx, txn, roomID); err != nil {
|
||||
return fmt.Errorf("d.Topology.DeleteTopologyForRoom: %w", err)
|
||||
}
|
||||
if err := d.BackwardExtremities.DeleteBackwardExtremitiesForRoom(ctx, txn, roomID); err != nil {
|
||||
return fmt.Errorf("d.BackwardExtremities.DeleteBackwardExtremitiesForRoom: %w", err)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
func (d *Database) WriteEvent(
|
||||
ctx context.Context,
|
||||
ev *gomatrixserverlib.HeaderedEvent,
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
)
|
||||
|
||||
|
|
@ -46,11 +47,15 @@ const selectBackwardExtremitiesForRoomSQL = "" +
|
|||
const deleteBackwardExtremitySQL = "" +
|
||||
"DELETE FROM syncapi_backward_extremities WHERE room_id = $1 AND prev_event_id = $2"
|
||||
|
||||
const deleteBackwardExtremitiesForRoomSQL = "" +
|
||||
"DELETE FROM syncapi_backward_extremities WHERE room_id = $1"
|
||||
|
||||
type backwardExtremitiesStatements struct {
|
||||
db *sql.DB
|
||||
insertBackwardExtremityStmt *sql.Stmt
|
||||
selectBackwardExtremitiesForRoomStmt *sql.Stmt
|
||||
deleteBackwardExtremityStmt *sql.Stmt
|
||||
deleteBackwardExtremitiesForRoomStmt *sql.Stmt
|
||||
}
|
||||
|
||||
func NewSqliteBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremities, error) {
|
||||
|
|
@ -70,6 +75,9 @@ func NewSqliteBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremities
|
|||
if s.deleteBackwardExtremityStmt, err = db.Prepare(deleteBackwardExtremitySQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.deleteBackwardExtremitiesForRoomStmt, err = db.Prepare(deleteBackwardExtremitiesForRoomSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
|
|
@ -108,3 +116,10 @@ func (s *backwardExtremitiesStatements) DeleteBackwardExtremity(
|
|||
_, err = txn.Stmt(s.deleteBackwardExtremityStmt).ExecContext(ctx, roomID, knownEventID)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *backwardExtremitiesStatements) DeleteBackwardExtremitiesForRoom(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) (err error) {
|
||||
_, err = sqlutil.TxStmt(txn, s.deleteBackwardExtremitiesForRoomStmt).ExecContext(ctx, roomID)
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -51,12 +51,15 @@ CREATE UNIQUE INDEX IF NOT EXISTS syncapi_event_id_idx ON syncapi_current_room_s
|
|||
const upsertRoomStateSQL = "" +
|
||||
"INSERT INTO syncapi_current_room_state (room_id, event_id, type, sender, contains_url, state_key, headered_event_json, membership, added_at)" +
|
||||
" VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)" +
|
||||
" ON CONFLICT (event_id, room_id, type, sender, contains_url)" +
|
||||
" ON CONFLICT (room_id, type, state_key)" +
|
||||
" DO UPDATE SET event_id = $2, sender=$4, contains_url=$5, headered_event_json = $7, membership = $8, added_at = $9"
|
||||
|
||||
const deleteRoomStateByEventIDSQL = "" +
|
||||
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
|
||||
|
||||
const DeleteRoomStateForRoomSQL = "" +
|
||||
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
|
||||
|
||||
const selectRoomIDsWithMembershipSQL = "" +
|
||||
"SELECT DISTINCT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
|
||||
|
||||
|
|
@ -88,6 +91,7 @@ type currentRoomStateStatements struct {
|
|||
streamIDStatements *streamIDStatements
|
||||
upsertRoomStateStmt *sql.Stmt
|
||||
deleteRoomStateByEventIDStmt *sql.Stmt
|
||||
DeleteRoomStateForRoomStmt *sql.Stmt
|
||||
selectRoomIDsWithMembershipStmt *sql.Stmt
|
||||
selectCurrentStateStmt *sql.Stmt
|
||||
selectJoinedUsersStmt *sql.Stmt
|
||||
|
|
@ -109,6 +113,9 @@ func NewSqliteCurrentRoomStateTable(db *sql.DB, streamID *streamIDStatements) (t
|
|||
if s.deleteRoomStateByEventIDStmt, err = db.Prepare(deleteRoomStateByEventIDSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.DeleteRoomStateForRoomStmt, err = db.Prepare(DeleteRoomStateForRoomSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.selectRoomIDsWithMembershipStmt, err = db.Prepare(selectRoomIDsWithMembershipSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -203,6 +210,14 @@ func (s *currentRoomStateStatements) DeleteRoomStateByEventID(
|
|||
return err
|
||||
}
|
||||
|
||||
func (s *currentRoomStateStatements) DeleteRoomStateForRoom(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) error {
|
||||
stmt := sqlutil.TxStmt(txn, s.DeleteRoomStateForRoomStmt)
|
||||
_, err := stmt.ExecContext(ctx, roomID)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *currentRoomStateStatements) UpsertRoomState(
|
||||
ctx context.Context, txn *sql.Tx,
|
||||
event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition,
|
||||
|
|
|
|||
|
|
@ -103,6 +103,9 @@ const selectStateInRangeSQL = "" +
|
|||
" ORDER BY id ASC" +
|
||||
" LIMIT $8" // limit
|
||||
|
||||
const deleteEventsForRoomSQL = "" +
|
||||
"DELETE FROM syncapi_output_room_events WHERE room_id = $1"
|
||||
|
||||
type outputRoomEventsStatements struct {
|
||||
db *sql.DB
|
||||
streamIDStatements *streamIDStatements
|
||||
|
|
@ -114,6 +117,7 @@ type outputRoomEventsStatements struct {
|
|||
selectEarlyEventsStmt *sql.Stmt
|
||||
selectStateInRangeStmt *sql.Stmt
|
||||
updateEventJSONStmt *sql.Stmt
|
||||
deleteEventsForRoomStmt *sql.Stmt
|
||||
}
|
||||
|
||||
func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Events, error) {
|
||||
|
|
@ -149,6 +153,9 @@ func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Even
|
|||
if s.updateEventJSONStmt, err = db.Prepare(updateEventJSONSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.deleteEventsForRoomStmt, err = db.Prepare(deleteEventsForRoomSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
|
|
@ -410,6 +417,13 @@ func (s *outputRoomEventsStatements) SelectEvents(
|
|||
return returnEvents, nil
|
||||
}
|
||||
|
||||
func (s *outputRoomEventsStatements) DeleteEventsForRoom(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) (err error) {
|
||||
_, err = sqlutil.TxStmt(txn, s.deleteEventsForRoomStmt).ExecContext(ctx, roomID)
|
||||
return err
|
||||
}
|
||||
|
||||
func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
|
||||
var result []types.StreamEvent
|
||||
for rows.Next() {
|
||||
|
|
|
|||
|
|
@ -65,6 +65,9 @@ const selectMaxPositionInTopologySQL = "" +
|
|||
"SELECT MAX(topological_position), stream_position FROM syncapi_output_room_events_topology" +
|
||||
" WHERE room_id = $1 ORDER BY stream_position DESC"
|
||||
|
||||
const deleteTopologyForRoomSQL = "" +
|
||||
"DELETE FROM syncapi_output_room_events_topology WHERE room_id = $1"
|
||||
|
||||
type outputRoomEventsTopologyStatements struct {
|
||||
db *sql.DB
|
||||
insertEventInTopologyStmt *sql.Stmt
|
||||
|
|
@ -72,6 +75,7 @@ type outputRoomEventsTopologyStatements struct {
|
|||
selectEventIDsInRangeDESCStmt *sql.Stmt
|
||||
selectPositionInTopologyStmt *sql.Stmt
|
||||
selectMaxPositionInTopologyStmt *sql.Stmt
|
||||
deleteTopologyForRoomStmt *sql.Stmt
|
||||
}
|
||||
|
||||
func NewSqliteTopologyTable(db *sql.DB) (tables.Topology, error) {
|
||||
|
|
@ -97,6 +101,9 @@ func NewSqliteTopologyTable(db *sql.DB) (tables.Topology, error) {
|
|||
if s.selectMaxPositionInTopologyStmt, err = db.Prepare(selectMaxPositionInTopologySQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if s.deleteTopologyForRoomStmt, err = db.Prepare(deleteTopologyForRoomSQL); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
|
|
@ -164,3 +171,10 @@ func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology(
|
|||
err = stmt.QueryRowContext(ctx, roomID).Scan(&pos, &spos)
|
||||
return
|
||||
}
|
||||
|
||||
func (s *outputRoomEventsTopologyStatements) DeleteTopologyForRoom(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) (err error) {
|
||||
_, err = sqlutil.TxStmt(txn, s.deleteTopologyForRoomStmt).ExecContext(ctx, roomID)
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -60,6 +60,8 @@ type Events interface {
|
|||
SelectEarlyEvents(ctx context.Context, txn *sql.Tx, roomID string, r types.Range, limit int) ([]types.StreamEvent, error)
|
||||
SelectEvents(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]types.StreamEvent, error)
|
||||
UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error
|
||||
// DeleteEventsForRoom removes all event information for a room. This should only be done when removing the room entirely.
|
||||
DeleteEventsForRoom(ctx context.Context, txn *sql.Tx, roomID string) (err error)
|
||||
}
|
||||
|
||||
// Topology keeps track of the depths and stream positions for all events.
|
||||
|
|
@ -77,6 +79,8 @@ type Topology interface {
|
|||
SelectPositionInTopology(ctx context.Context, txn *sql.Tx, eventID string) (depth, spos types.StreamPosition, err error)
|
||||
// SelectMaxPositionInTopology returns the event which has the highest depth, and if there are multiple, the event with the highest stream position.
|
||||
SelectMaxPositionInTopology(ctx context.Context, txn *sql.Tx, roomID string) (depth types.StreamPosition, spos types.StreamPosition, err error)
|
||||
// DeleteTopologyForRoom removes all topological information for a room. This should only be done when removing the room entirely.
|
||||
DeleteTopologyForRoom(ctx context.Context, txn *sql.Tx, roomID string) (err error)
|
||||
}
|
||||
|
||||
type CurrentRoomState interface {
|
||||
|
|
@ -84,6 +88,7 @@ type CurrentRoomState interface {
|
|||
SelectEventsWithEventIDs(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]types.StreamEvent, error)
|
||||
UpsertRoomState(ctx context.Context, txn *sql.Tx, event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition) error
|
||||
DeleteRoomStateByEventID(ctx context.Context, txn *sql.Tx, eventID string) error
|
||||
DeleteRoomStateForRoom(ctx context.Context, txn *sql.Tx, roomID string) error
|
||||
// SelectCurrentState returns all the current state events for the given room.
|
||||
SelectCurrentState(ctx context.Context, txn *sql.Tx, roomID string, stateFilter *gomatrixserverlib.StateFilter) ([]gomatrixserverlib.HeaderedEvent, error)
|
||||
// SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state.
|
||||
|
|
@ -118,6 +123,8 @@ type BackwardsExtremities interface {
|
|||
SelectBackwardExtremitiesForRoom(ctx context.Context, roomID string) (bwExtrems map[string][]string, err error)
|
||||
// DeleteBackwardExtremity removes a backwards extremity for a room, if one existed.
|
||||
DeleteBackwardExtremity(ctx context.Context, txn *sql.Tx, roomID, knownEventID string) (err error)
|
||||
// DeleteBackwardExtremitiesFoorRoomID removes all backward extremities for a room. This should only be done when removing the room entirely.
|
||||
DeleteBackwardExtremitiesForRoom(ctx context.Context, txn *sql.Tx, roomID string) (err error)
|
||||
}
|
||||
|
||||
// SendToDevice tracks send-to-device messages which are sent to individual
|
||||
|
|
|
|||
Loading…
Reference in New Issue