mirror of
https://github.com/hoernschen/dendrite.git
synced 2025-07-31 13:22:46 +00:00
Roomserver API changes (#1118)
* s/QueryBackfill/PerformBackfill/g * OutputEvent now includes AddStateEvents which contain the full event of extra state events * Only include adds not the current event * Get adding state right
This commit is contained in:
parent
25cd2dd1c9
commit
ec7718e7f8
17 changed files with 152 additions and 287 deletions
|
@ -17,7 +17,6 @@ package consumers
|
|||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"github.com/Shopify/sarama"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
|
@ -105,17 +104,9 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
|
|||
"room_version": ev.RoomVersion,
|
||||
}).Info("received event from roomserver")
|
||||
|
||||
addsStateEvents, err := s.lookupStateEvents(msg.AddsStateEventIDs, ev)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"event": string(ev.JSON()),
|
||||
log.ErrorKey: err,
|
||||
"add": msg.AddsStateEventIDs,
|
||||
"del": msg.RemovesStateEventIDs,
|
||||
}).Panicf("roomserver output log: state event lookup failure")
|
||||
}
|
||||
addsStateEvents := msg.AddsState()
|
||||
|
||||
ev, err = s.updateStateEvent(ev)
|
||||
ev, err := s.updateStateEvent(ev)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -185,63 +176,6 @@ func (s *OutputRoomEventConsumer) onRetireInviteEvent(
|
|||
return nil
|
||||
}
|
||||
|
||||
// lookupStateEvents looks up the state events that are added by a new event.
|
||||
func (s *OutputRoomEventConsumer) lookupStateEvents(
|
||||
addsStateEventIDs []string, event gomatrixserverlib.HeaderedEvent,
|
||||
) ([]gomatrixserverlib.HeaderedEvent, error) {
|
||||
// Fast path if there aren't any new state events.
|
||||
if len(addsStateEventIDs) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Fast path if the only state event added is the event itself.
|
||||
if len(addsStateEventIDs) == 1 && addsStateEventIDs[0] == event.EventID() {
|
||||
return []gomatrixserverlib.HeaderedEvent{event}, nil
|
||||
}
|
||||
|
||||
// Check if this is re-adding a state events that we previously processed
|
||||
// If we have previously received a state event it may still be in
|
||||
// our event database.
|
||||
result, err := s.db.Events(context.TODO(), addsStateEventIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
missing := missingEventsFrom(result, addsStateEventIDs)
|
||||
|
||||
// Check if event itself is being added.
|
||||
for _, eventID := range missing {
|
||||
if eventID == event.EventID() {
|
||||
result = append(result, event)
|
||||
break
|
||||
}
|
||||
}
|
||||
missing = missingEventsFrom(result, addsStateEventIDs)
|
||||
|
||||
if len(missing) == 0 {
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// At this point the missing events are neither the event itself nor are
|
||||
// they present in our local database. Our only option is to fetch them
|
||||
// from the roomserver using the query API.
|
||||
eventReq := api.QueryEventsByIDRequest{EventIDs: missing}
|
||||
var eventResp api.QueryEventsByIDResponse
|
||||
if err := s.rsAPI.QueryEventsByID(context.TODO(), &eventReq, &eventResp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
result = append(result, eventResp.Events...)
|
||||
missing = missingEventsFrom(result, addsStateEventIDs)
|
||||
|
||||
if len(missing) != 0 {
|
||||
return nil, fmt.Errorf(
|
||||
"missing %d state events IDs at event %q", len(missing), event.EventID(),
|
||||
)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s *OutputRoomEventConsumer) updateStateEvent(event gomatrixserverlib.HeaderedEvent) (gomatrixserverlib.HeaderedEvent, error) {
|
||||
var stateKey string
|
||||
if event.StateKey() == nil {
|
||||
|
@ -270,17 +204,3 @@ func (s *OutputRoomEventConsumer) updateStateEvent(event gomatrixserverlib.Heade
|
|||
event.Event, err = event.SetUnsigned(prev)
|
||||
return event, err
|
||||
}
|
||||
|
||||
func missingEventsFrom(events []gomatrixserverlib.HeaderedEvent, required []string) []string {
|
||||
have := map[string]bool{}
|
||||
for _, event := range events {
|
||||
have[event.EventID()] = true
|
||||
}
|
||||
var missing []string
|
||||
for _, eventID := range required {
|
||||
if !have[eventID] {
|
||||
missing = append(missing, eventID)
|
||||
}
|
||||
}
|
||||
return missing
|
||||
}
|
||||
|
|
|
@ -375,15 +375,15 @@ func (e eventsByDepth) Less(i, j int) bool {
|
|||
// Returns an error if there was an issue with retrieving the list of servers in
|
||||
// the room or sending the request.
|
||||
func (r *messagesReq) backfill(roomID string, backwardsExtremities map[string][]string, limit int) ([]gomatrixserverlib.HeaderedEvent, error) {
|
||||
var res api.QueryBackfillResponse
|
||||
err := r.rsAPI.QueryBackfill(context.Background(), &api.QueryBackfillRequest{
|
||||
var res api.PerformBackfillResponse
|
||||
err := r.rsAPI.PerformBackfill(context.Background(), &api.PerformBackfillRequest{
|
||||
RoomID: roomID,
|
||||
BackwardsExtremities: backwardsExtremities,
|
||||
Limit: limit,
|
||||
ServerName: r.cfg.Matrix.ServerName,
|
||||
}, &res)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("QueryBackfill failed: %w", err)
|
||||
return nil, fmt.Errorf("PerformBackfill failed: %w", err)
|
||||
}
|
||||
util.GetLogger(r.ctx).WithField("new_events", len(res.Events)).Info("Storing new events from backfill")
|
||||
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue