mirror of
https://github.com/matrix-org/dendrite.git
synced 2025-12-16 19:33:09 -06:00
Squashed commit of the following:
commit067b875063Author: Neil Alexander <neilalexander@users.noreply.github.com> Date: Fri Apr 3 14:29:06 2020 +0100 Invites v2 endpoint (#952) * Start converting v1 invite endpoint to v2 * Update gomatrixserverlib * Early federationsender code for sending invites * Sending invites sorta happens now * Populate invite request with stripped state * Remodel a bit, don't reflect received invites * Handle invite_room_state * Handle room versions a bit better * Update gomatrixserverlib * Tweak order in destinationQueue.next * Revert check in processMessage * Tweak federation sender destination queue code a bit * Add comments commit955244c092Author: Ben B <benne@klimlive.de> Date: Fri Apr 3 12:40:50 2020 +0200 use custom http client instead of the http DefaultClient (#823) This commit replaces the default client from the http lib with a custom one. The previously used default client doesn't come with a timeout. This could cause unwanted locks. That solution chosen here creates a http client in the base component dendrite with a constant timeout of 30 seconds. If it should be necessary to overwrite this, we could include the timeout in the dendrite configuration. Here it would be a good idea to extend the type "Address" by a timeout and create an http client for each service. Closes #820 Signed-off-by: Benedikt Bongartz <benne@klimlive.de> Co-authored-by: Kegsay <kegan@matrix.org>
This commit is contained in:
Neil Alexander
parent
b8736256ad
commit
3f5c4a0047
|
|
@ -20,6 +20,7 @@ package api
|
|||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
|
||||
|
|
@ -97,15 +98,15 @@ type httpAppServiceQueryAPI struct {
|
|||
|
||||
// NewAppServiceQueryAPIHTTP creates a AppServiceQueryAPI implemented by talking
|
||||
// to a HTTP POST API.
|
||||
// If httpClient is nil then it uses http.DefaultClient
|
||||
// If httpClient is nil an error is returned
|
||||
func NewAppServiceQueryAPIHTTP(
|
||||
appserviceURL string,
|
||||
httpClient *http.Client,
|
||||
) AppServiceQueryAPI {
|
||||
) (AppServiceQueryAPI, error) {
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
return nil, errors.New("NewRoomserverAliasAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpAppServiceQueryAPI{appserviceURL, httpClient}
|
||||
return &httpAppServiceQueryAPI{appserviceURL, httpClient}, nil
|
||||
}
|
||||
|
||||
// RoomAliasExists implements AppServiceQueryAPI
|
||||
|
|
|
|||
|
|
@ -104,18 +104,14 @@ func (c *RoomserverProducer) SendInputRoomEvents(
|
|||
// This should only be needed for invite events that occur outside of a known room.
|
||||
// If we are in the room then the event should be sent using the SendEvents method.
|
||||
func (c *RoomserverProducer) SendInvite(
|
||||
ctx context.Context, inviteEvent gomatrixserverlib.Event,
|
||||
ctx context.Context, inviteEvent gomatrixserverlib.HeaderedEvent,
|
||||
inviteRoomState []gomatrixserverlib.InviteV2StrippedState,
|
||||
) error {
|
||||
verReq := api.QueryRoomVersionForRoomRequest{RoomID: inviteEvent.RoomID()}
|
||||
verRes := api.QueryRoomVersionForRoomResponse{}
|
||||
err := c.QueryAPI.QueryRoomVersionForRoom(ctx, &verReq, &verRes)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
request := api.InputRoomEventsRequest{
|
||||
InputInviteEvents: []api.InputInviteEvent{{
|
||||
Event: inviteEvent.Headered(verRes.RoomVersion),
|
||||
Event: inviteEvent,
|
||||
InviteRoomState: inviteRoomState,
|
||||
RoomVersion: inviteEvent.RoomVersion,
|
||||
}},
|
||||
}
|
||||
var response api.InputRoomEventsResponse
|
||||
|
|
|
|||
|
|
@ -260,6 +260,9 @@ func (r joinRoomReq) joinRoomUsingServers(
|
|||
}{roomID},
|
||||
}
|
||||
}
|
||||
// TODO: This needs to be re-thought, as in the case of an invite, the room
|
||||
// will exist in the database in roomserver_rooms but won't have any state
|
||||
// events, therefore this below check fails.
|
||||
if err != common.ErrRoomNoExists {
|
||||
util.GetLogger(r.req.Context()).WithError(err).Error("common.BuildEvent failed")
|
||||
return jsonerror.InternalServerError()
|
||||
|
|
|
|||
|
|
@ -44,6 +44,8 @@ var (
|
|||
// This needs to be high enough to account for the time it takes to create
|
||||
// the postgres database tables which can take a while on travis.
|
||||
timeoutString = defaulting(os.Getenv("TIMEOUT"), "60s")
|
||||
// Timeout for http client
|
||||
timeoutHTTPClient = defaulting(os.Getenv("TIMEOUT_HTTP"), "30s")
|
||||
// The name of maintenance database to connect to in order to create the test database.
|
||||
postgresDatabase = defaulting(os.Getenv("POSTGRES_DATABASE"), "postgres")
|
||||
// The name of the test database to create.
|
||||
|
|
@ -68,7 +70,10 @@ func defaulting(value, defaultValue string) string {
|
|||
return value
|
||||
}
|
||||
|
||||
var timeout time.Duration
|
||||
var (
|
||||
timeout time.Duration
|
||||
timeoutHTTP time.Duration
|
||||
)
|
||||
|
||||
func init() {
|
||||
var err error
|
||||
|
|
@ -76,6 +81,10 @@ func init() {
|
|||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
timeoutHTTP, err = time.ParseDuration(timeoutHTTPClient)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
func createDatabase(database string) error {
|
||||
|
|
@ -199,7 +208,10 @@ func writeToRoomServer(input []string, roomserverURL string) error {
|
|||
return err
|
||||
}
|
||||
}
|
||||
x := api.NewRoomserverInputAPIHTTP(roomserverURL, nil)
|
||||
x, err := api.NewRoomserverInputAPIHTTP(roomserverURL, &http.Client{Timeout: timeoutHTTP})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return x.InputRoomEvents(context.Background(), &request, &response)
|
||||
}
|
||||
|
||||
|
|
@ -258,7 +270,7 @@ func testRoomserver(input []string, wantOutput []string, checkQueries func(api.R
|
|||
cmd.Args = []string{"dendrite-room-server", "--config", filepath.Join(dir, test.ConfigFile)}
|
||||
|
||||
gotOutput, err := runAndReadFromTopic(cmd, cfg.RoomServerURL()+"/metrics", doInput, outputTopic, len(wantOutput), func() {
|
||||
queryAPI := api.NewRoomserverQueryAPIHTTP("http://"+string(cfg.Listen.RoomServer), nil)
|
||||
queryAPI, _ := api.NewRoomserverQueryAPIHTTP("http://"+string(cfg.Listen.RoomServer), &http.Client{Timeout: timeoutHTTP})
|
||||
checkQueries(queryAPI)
|
||||
})
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
"golang.org/x/crypto/ed25519"
|
||||
|
||||
|
|
@ -52,6 +53,7 @@ type BaseDendrite struct {
|
|||
|
||||
// APIMux should be used to register new public matrix api endpoints
|
||||
APIMux *mux.Router
|
||||
httpClient *http.Client
|
||||
Cfg *config.Dendrite
|
||||
KafkaConsumer sarama.Consumer
|
||||
KafkaProducer sarama.SyncProducer
|
||||
|
|
@ -77,11 +79,14 @@ func NewBaseDendrite(cfg *config.Dendrite, componentName string) *BaseDendrite {
|
|||
kafkaConsumer, kafkaProducer = setupKafka(cfg)
|
||||
}
|
||||
|
||||
const defaultHTTPTimeout = 30 * time.Second
|
||||
|
||||
return &BaseDendrite{
|
||||
componentName: componentName,
|
||||
tracerCloser: closer,
|
||||
Cfg: cfg,
|
||||
APIMux: mux.NewRouter().UseEncodedPath(),
|
||||
httpClient: &http.Client{Timeout: defaultHTTPTimeout},
|
||||
KafkaConsumer: kafkaConsumer,
|
||||
KafkaProducer: kafkaProducer,
|
||||
}
|
||||
|
|
@ -95,7 +100,11 @@ func (b *BaseDendrite) Close() error {
|
|||
// CreateHTTPAppServiceAPIs returns the QueryAPI for hitting the appservice
|
||||
// component over HTTP.
|
||||
func (b *BaseDendrite) CreateHTTPAppServiceAPIs() appserviceAPI.AppServiceQueryAPI {
|
||||
return appserviceAPI.NewAppServiceQueryAPIHTTP(b.Cfg.AppServiceURL(), nil)
|
||||
a, err := appserviceAPI.NewAppServiceQueryAPIHTTP(b.Cfg.AppServiceURL(), b.httpClient)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("CreateHTTPAppServiceAPIs failed")
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// CreateHTTPRoomserverAPIs returns the AliasAPI, InputAPI and QueryAPI for hitting
|
||||
|
|
@ -105,22 +114,40 @@ func (b *BaseDendrite) CreateHTTPRoomserverAPIs() (
|
|||
roomserverAPI.RoomserverInputAPI,
|
||||
roomserverAPI.RoomserverQueryAPI,
|
||||
) {
|
||||
alias := roomserverAPI.NewRoomserverAliasAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
input := roomserverAPI.NewRoomserverInputAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
query := roomserverAPI.NewRoomserverQueryAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
|
||||
alias, err := roomserverAPI.NewRoomserverAliasAPIHTTP(b.Cfg.RoomServerURL(), b.httpClient)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("NewRoomserverAliasAPIHTTP failed")
|
||||
}
|
||||
input, err := roomserverAPI.NewRoomserverInputAPIHTTP(b.Cfg.RoomServerURL(), b.httpClient)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("NewRoomserverInputAPIHTTP failed", b.httpClient)
|
||||
}
|
||||
query, err := roomserverAPI.NewRoomserverQueryAPIHTTP(b.Cfg.RoomServerURL(), nil)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("NewRoomserverQueryAPIHTTP failed", b.httpClient)
|
||||
}
|
||||
return alias, input, query
|
||||
}
|
||||
|
||||
// CreateHTTPEDUServerAPIs returns eduInputAPI for hitting the EDU
|
||||
// server over HTTP
|
||||
func (b *BaseDendrite) CreateHTTPEDUServerAPIs() eduServerAPI.EDUServerInputAPI {
|
||||
return eduServerAPI.NewEDUServerInputAPIHTTP(b.Cfg.EDUServerURL(), nil)
|
||||
e, err := eduServerAPI.NewEDUServerInputAPIHTTP(b.Cfg.EDUServerURL(), nil)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("NewEDUServerInputAPIHTTP failed", b.httpClient)
|
||||
}
|
||||
return e
|
||||
}
|
||||
|
||||
// CreateHTTPFederationSenderAPIs returns FederationSenderQueryAPI for hitting
|
||||
// the federation sender over HTTP
|
||||
func (b *BaseDendrite) CreateHTTPFederationSenderAPIs() federationSenderAPI.FederationSenderQueryAPI {
|
||||
return federationSenderAPI.NewFederationSenderQueryAPIHTTP(b.Cfg.FederationSenderURL(), nil)
|
||||
f, err := federationSenderAPI.NewFederationSenderQueryAPIHTTP(b.Cfg.FederationSenderURL(), nil)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panic("NewFederationSenderQueryAPIHTTP failed", b.httpClient)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// CreateDeviceDB creates a new instance of the device database. Should only be
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ package api
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
commonHTTP "github.com/matrix-org/dendrite/common/http"
|
||||
|
|
@ -57,11 +58,11 @@ type EDUServerInputAPI interface {
|
|||
const EDUServerInputTypingEventPath = "/api/eduserver/input"
|
||||
|
||||
// NewEDUServerInputAPIHTTP creates a EDUServerInputAPI implemented by talking to a HTTP POST API.
|
||||
func NewEDUServerInputAPIHTTP(eduServerURL string, httpClient *http.Client) EDUServerInputAPI {
|
||||
func NewEDUServerInputAPIHTTP(eduServerURL string, httpClient *http.Client) (EDUServerInputAPI, error) {
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
return nil, errors.New("NewTypingServerInputAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpEDUServerInputAPI{eduServerURL, httpClient}
|
||||
return &httpEDUServerInputAPI{eduServerURL, httpClient}, nil
|
||||
}
|
||||
|
||||
type httpEDUServerInputAPI struct {
|
||||
|
|
|
|||
|
|
@ -15,18 +15,17 @@
|
|||
package routing
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/jsonerror"
|
||||
"github.com/matrix-org/dendrite/clientapi/producers"
|
||||
"github.com/matrix-org/dendrite/common/config"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/util"
|
||||
)
|
||||
|
||||
// Invite implements /_matrix/federation/v1/invite/{roomID}/{eventID}
|
||||
// Invite implements /_matrix/federation/v2/invite/{roomID}/{eventID}
|
||||
func Invite(
|
||||
httpReq *http.Request,
|
||||
request *gomatrixserverlib.FederationRequest,
|
||||
|
|
@ -36,24 +35,14 @@ func Invite(
|
|||
producer *producers.RoomserverProducer,
|
||||
keys gomatrixserverlib.KeyRing,
|
||||
) util.JSONResponse {
|
||||
// Look up the room version for the room.
|
||||
verReq := api.QueryRoomVersionForRoomRequest{RoomID: roomID}
|
||||
verRes := api.QueryRoomVersionForRoomResponse{}
|
||||
if err := producer.QueryAPI.QueryRoomVersionForRoom(context.Background(), &verReq, &verRes); err != nil {
|
||||
inviteReq := gomatrixserverlib.InviteV2Request{}
|
||||
if err := json.Unmarshal(request.Content(), &inviteReq); err != nil {
|
||||
return util.JSONResponse{
|
||||
Code: http.StatusBadRequest,
|
||||
JSON: jsonerror.UnsupportedRoomVersion(err.Error()),
|
||||
}
|
||||
}
|
||||
|
||||
// Decode the event JSON from the request.
|
||||
event, err := gomatrixserverlib.NewEventFromUntrustedJSON(request.Content(), verRes.RoomVersion)
|
||||
if err != nil {
|
||||
return util.JSONResponse{
|
||||
Code: http.StatusBadRequest,
|
||||
JSON: jsonerror.NotJSON("The request body could not be decoded into valid JSON. " + err.Error()),
|
||||
JSON: jsonerror.NotJSON("The request body could not be decoded into an invite request. " + err.Error()),
|
||||
}
|
||||
}
|
||||
event := inviteReq.Event()
|
||||
|
||||
// Check that the room ID is correct.
|
||||
if event.RoomID() != roomID {
|
||||
|
|
@ -71,14 +60,6 @@ func Invite(
|
|||
}
|
||||
}
|
||||
|
||||
// Check that the event is from the server sending the request.
|
||||
if event.Origin() != request.Origin() {
|
||||
return util.JSONResponse{
|
||||
Code: http.StatusForbidden,
|
||||
JSON: jsonerror.Forbidden("The invite must be sent by the server it originated on"),
|
||||
}
|
||||
}
|
||||
|
||||
// Check that the event is signed by the server sending the request.
|
||||
redacted := event.Redact()
|
||||
verifyRequests := []gomatrixserverlib.VerifyJSONRequest{{
|
||||
|
|
@ -104,7 +85,11 @@ func Invite(
|
|||
)
|
||||
|
||||
// Add the invite event to the roomserver.
|
||||
if err = producer.SendInvite(httpReq.Context(), signedEvent); err != nil {
|
||||
if err = producer.SendInvite(
|
||||
httpReq.Context(),
|
||||
signedEvent.Headered(inviteReq.RoomVersion()),
|
||||
inviteReq.InviteRoomState(),
|
||||
); err != nil {
|
||||
util.GetLogger(httpReq.Context()).WithError(err).Error("producer.SendInvite failed")
|
||||
return jsonerror.InternalServerError()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -85,7 +85,7 @@ func Setup(
|
|||
},
|
||||
)).Methods(http.MethodPut, http.MethodOptions)
|
||||
|
||||
v1fedmux.Handle("/invite/{roomID}/{eventID}", common.MakeFedAPI(
|
||||
v2fedmux.Handle("/invite/{roomID}/{eventID}", common.MakeFedAPI(
|
||||
"federation_invite", cfg.Matrix.ServerName, keys,
|
||||
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest) util.JSONResponse {
|
||||
vars, err := common.URLDecodeMapValues(mux.Vars(httpReq))
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package api
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
commonHTTP "github.com/matrix-org/dendrite/common/http"
|
||||
|
|
@ -58,12 +59,12 @@ const FederationSenderQueryJoinedHostsInRoomPath = "/api/federationsender/queryJ
|
|||
const FederationSenderQueryJoinedHostServerNamesInRoomPath = "/api/federationsender/queryJoinedHostServerNamesInRoom"
|
||||
|
||||
// NewFederationSenderQueryAPIHTTP creates a FederationSenderQueryAPI implemented by talking to a HTTP POST API.
|
||||
// If httpClient is nil then it uses the http.DefaultClient
|
||||
func NewFederationSenderQueryAPIHTTP(federationSenderURL string, httpClient *http.Client) FederationSenderQueryAPI {
|
||||
// If httpClient is nil an error is returned
|
||||
func NewFederationSenderQueryAPIHTTP(federationSenderURL string, httpClient *http.Client) (FederationSenderQueryAPI, error) {
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
return nil, errors.New("NewFederationSenderQueryAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpFederationSenderQueryAPI{federationSenderURL, httpClient}
|
||||
return &httpFederationSenderQueryAPI{federationSenderURL, httpClient}, nil
|
||||
}
|
||||
|
||||
type httpFederationSenderQueryAPI struct {
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ import (
|
|||
|
||||
// OutputRoomEventConsumer consumes events that originated in the room server.
|
||||
type OutputRoomEventConsumer struct {
|
||||
cfg *config.Dendrite
|
||||
roomServerConsumer *common.ContinualConsumer
|
||||
db storage.Database
|
||||
queues *queue.OutgoingQueues
|
||||
|
|
@ -52,6 +53,7 @@ func NewOutputRoomEventConsumer(
|
|||
PartitionStore: store,
|
||||
}
|
||||
s := &OutputRoomEventConsumer{
|
||||
cfg: cfg,
|
||||
roomServerConsumer: &consumer,
|
||||
db: store,
|
||||
queues: queues,
|
||||
|
|
@ -79,29 +81,48 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
|
|||
log.WithError(err).Errorf("roomserver output log: message parse failure")
|
||||
return nil
|
||||
}
|
||||
if output.Type != api.OutputTypeNewRoomEvent {
|
||||
|
||||
switch output.Type {
|
||||
case api.OutputTypeNewRoomEvent:
|
||||
ev := &output.NewRoomEvent.Event
|
||||
log.WithFields(log.Fields{
|
||||
"event_id": ev.EventID(),
|
||||
"room_id": ev.RoomID(),
|
||||
"send_as_server": output.NewRoomEvent.SendAsServer,
|
||||
}).Info("received room event from roomserver")
|
||||
|
||||
if err := s.processMessage(*output.NewRoomEvent); err != nil {
|
||||
// panic rather than continue with an inconsistent database
|
||||
log.WithFields(log.Fields{
|
||||
"event": string(ev.JSON()),
|
||||
"add": output.NewRoomEvent.AddsStateEventIDs,
|
||||
"del": output.NewRoomEvent.RemovesStateEventIDs,
|
||||
log.ErrorKey: err,
|
||||
}).Panicf("roomserver output log: write room event failure")
|
||||
return nil
|
||||
}
|
||||
case api.OutputTypeNewInviteEvent:
|
||||
ev := &output.NewInviteEvent.Event
|
||||
log.WithFields(log.Fields{
|
||||
"event_id": ev.EventID(),
|
||||
"room_id": ev.RoomID(),
|
||||
"state_key": ev.StateKey(),
|
||||
}).Info("received invite event from roomserver")
|
||||
|
||||
if err := s.processInvite(*output.NewInviteEvent); err != nil {
|
||||
// panic rather than continue with an inconsistent database
|
||||
log.WithFields(log.Fields{
|
||||
"event": string(ev.JSON()),
|
||||
log.ErrorKey: err,
|
||||
}).Panicf("roomserver output log: write invite event failure")
|
||||
return nil
|
||||
}
|
||||
default:
|
||||
log.WithField("type", output.Type).Debug(
|
||||
"roomserver output log: ignoring unknown output type",
|
||||
)
|
||||
return nil
|
||||
}
|
||||
ev := &output.NewRoomEvent.Event
|
||||
log.WithFields(log.Fields{
|
||||
"event_id": ev.EventID(),
|
||||
"room_id": ev.RoomID(),
|
||||
"send_as_server": output.NewRoomEvent.SendAsServer,
|
||||
}).Info("received event from roomserver")
|
||||
|
||||
if err := s.processMessage(*output.NewRoomEvent); err != nil {
|
||||
// panic rather than continue with an inconsistent database
|
||||
log.WithFields(log.Fields{
|
||||
"event": string(ev.JSON()),
|
||||
log.ErrorKey: err,
|
||||
"add": output.NewRoomEvent.AddsStateEventIDs,
|
||||
"del": output.NewRoomEvent.RemovesStateEventIDs,
|
||||
}).Panicf("roomserver output log: write event failure")
|
||||
return nil
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -159,6 +180,69 @@ func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent) err
|
|||
)
|
||||
}
|
||||
|
||||
// processInvite handles an invite event for sending over federation.
|
||||
func (s *OutputRoomEventConsumer) processInvite(oie api.OutputNewInviteEvent) error {
|
||||
// Don't try to reflect and resend invites that didn't originate from us.
|
||||
if s.cfg.Matrix.ServerName != oie.Event.Origin() {
|
||||
return nil
|
||||
}
|
||||
|
||||
// When sending a v2 invite, the inviting server should try and include
|
||||
// a "stripped down" version of the room state. This is pretty much just
|
||||
// enough information for the remote side to show something useful to the
|
||||
// user, like the room name, aliases etc.
|
||||
strippedState := []gomatrixserverlib.InviteV2StrippedState{}
|
||||
stateWanted := []string{
|
||||
gomatrixserverlib.MRoomName, gomatrixserverlib.MRoomCanonicalAlias,
|
||||
gomatrixserverlib.MRoomAliases, gomatrixserverlib.MRoomJoinRules,
|
||||
}
|
||||
|
||||
// For each of the state keys that we want to try and send, ask the
|
||||
// roomserver if we have a state event for that room that matches the
|
||||
// state key.
|
||||
for _, wanted := range stateWanted {
|
||||
queryReq := api.QueryLatestEventsAndStateRequest{
|
||||
RoomID: oie.Event.RoomID(),
|
||||
StateToFetch: []gomatrixserverlib.StateKeyTuple{
|
||||
gomatrixserverlib.StateKeyTuple{
|
||||
EventType: wanted,
|
||||
StateKey: "",
|
||||
},
|
||||
},
|
||||
}
|
||||
// If this fails then we just move onto the next event - we don't
|
||||
// actually know at this point whether the room even has that type
|
||||
// of state.
|
||||
queryRes := api.QueryLatestEventsAndStateResponse{}
|
||||
if err := s.query.QueryLatestEventsAndState(context.TODO(), &queryReq, &queryRes); err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"room_id": queryReq.RoomID,
|
||||
"event_type": wanted,
|
||||
}).WithError(err).Info("couldn't find state to strip")
|
||||
continue
|
||||
}
|
||||
// Append the stripped down copy of the state to our list.
|
||||
for _, headeredEvent := range queryRes.StateEvents {
|
||||
event := headeredEvent.Unwrap()
|
||||
strippedState = append(strippedState, gomatrixserverlib.NewInviteV2StrippedState(&event))
|
||||
|
||||
log.WithFields(log.Fields{
|
||||
"room_id": queryReq.RoomID,
|
||||
"event_type": event.Type(),
|
||||
}).Info("adding stripped state")
|
||||
}
|
||||
}
|
||||
|
||||
// Build the invite request with the info we've got.
|
||||
inviteReq, err := gomatrixserverlib.NewInviteV2Request(&oie.Event, strippedState)
|
||||
if err != nil {
|
||||
return fmt.Errorf("gomatrixserverlib.NewInviteV2Request: %w", err)
|
||||
}
|
||||
|
||||
// Send the event.
|
||||
return s.queues.SendInvite(&inviteReq)
|
||||
}
|
||||
|
||||
// joinedHostsAtEvent works out a list of matrix servers that were joined to
|
||||
// the room at the event.
|
||||
// It is important to use the state at the event for sending messages because:
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ import (
|
|||
"github.com/matrix-org/gomatrixserverlib"
|
||||
"github.com/matrix-org/util"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"go.uber.org/atomic"
|
||||
)
|
||||
|
||||
// destinationQueue is a queue of events for a single destination.
|
||||
|
|
@ -34,14 +35,15 @@ type destinationQueue struct {
|
|||
client *gomatrixserverlib.FederationClient
|
||||
origin gomatrixserverlib.ServerName
|
||||
destination gomatrixserverlib.ServerName
|
||||
// The running mutex protects running, sentCounter, lastTransactionIDs and
|
||||
running atomic.Bool
|
||||
// The running mutex protects sentCounter, lastTransactionIDs and
|
||||
// pendingEvents, pendingEDUs.
|
||||
runningMutex sync.Mutex
|
||||
running bool
|
||||
sentCounter int
|
||||
lastTransactionIDs []gomatrixserverlib.TransactionID
|
||||
pendingEvents []*gomatrixserverlib.HeaderedEvent
|
||||
pendingEDUs []*gomatrixserverlib.EDU
|
||||
pendingInvites []*gomatrixserverlib.InviteV2Request
|
||||
}
|
||||
|
||||
// Send event adds the event to the pending queue for the destination.
|
||||
|
|
@ -51,29 +53,43 @@ func (oq *destinationQueue) sendEvent(ev *gomatrixserverlib.HeaderedEvent) {
|
|||
oq.runningMutex.Lock()
|
||||
defer oq.runningMutex.Unlock()
|
||||
oq.pendingEvents = append(oq.pendingEvents, ev)
|
||||
if !oq.running {
|
||||
oq.running = true
|
||||
if !oq.running.Load() {
|
||||
go oq.backgroundSend()
|
||||
}
|
||||
}
|
||||
|
||||
// sendEDU adds the EDU event to the pending queue for the destination.
|
||||
// If the queue is empty then it starts a background goroutine to
|
||||
// start sending event to that destination.
|
||||
// start sending events to that destination.
|
||||
func (oq *destinationQueue) sendEDU(e *gomatrixserverlib.EDU) {
|
||||
oq.runningMutex.Lock()
|
||||
defer oq.runningMutex.Unlock()
|
||||
oq.pendingEDUs = append(oq.pendingEDUs, e)
|
||||
if !oq.running {
|
||||
oq.running = true
|
||||
if !oq.running.Load() {
|
||||
go oq.backgroundSend()
|
||||
}
|
||||
}
|
||||
|
||||
// sendInvite adds the invite event to the pending queue for the
|
||||
// destination. If the queue is empty then it starts a background
|
||||
// goroutine to start sending events to that destination.
|
||||
func (oq *destinationQueue) sendInvite(ev *gomatrixserverlib.InviteV2Request) {
|
||||
oq.runningMutex.Lock()
|
||||
defer oq.runningMutex.Unlock()
|
||||
oq.pendingInvites = append(oq.pendingInvites, ev)
|
||||
if !oq.running.Load() {
|
||||
go oq.backgroundSend()
|
||||
}
|
||||
}
|
||||
|
||||
// backgroundSend is the worker goroutine for sending events.
|
||||
func (oq *destinationQueue) backgroundSend() {
|
||||
oq.running.Store(true)
|
||||
defer oq.running.Store(false)
|
||||
|
||||
for {
|
||||
t := oq.next()
|
||||
if t == nil {
|
||||
transaction, invites := oq.nextTransaction(), oq.nextInvites()
|
||||
if !transaction && !invites {
|
||||
// If the queue is empty then stop processing for this destination.
|
||||
// TODO: Remove this destination from the queue map.
|
||||
return
|
||||
|
|
@ -81,29 +97,18 @@ func (oq *destinationQueue) backgroundSend() {
|
|||
|
||||
// TODO: handle retries.
|
||||
// TODO: blacklist uncooperative servers.
|
||||
|
||||
util.GetLogger(context.TODO()).Infof("Sending transaction %q containing %d PDUs, %d EDUs", t.TransactionID, len(t.PDUs), len(t.EDUs))
|
||||
|
||||
_, err := oq.client.SendTransaction(context.TODO(), *t)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"destination": oq.destination,
|
||||
log.ErrorKey: err,
|
||||
}).Info("problem sending transaction")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// next creates a new transaction from the pending event queue
|
||||
// and flushes the queue.
|
||||
// Returns nil if the queue was empty.
|
||||
func (oq *destinationQueue) next() *gomatrixserverlib.Transaction {
|
||||
// nextTransaction creates a new transaction from the pending event
|
||||
// queue and sends it. Returns true if a transaction was sent or
|
||||
// false otherwise.
|
||||
func (oq *destinationQueue) nextTransaction() bool {
|
||||
oq.runningMutex.Lock()
|
||||
defer oq.runningMutex.Unlock()
|
||||
|
||||
if len(oq.pendingEvents) == 0 && len(oq.pendingEDUs) == 0 {
|
||||
oq.running = false
|
||||
return nil
|
||||
return false
|
||||
}
|
||||
|
||||
t := gomatrixserverlib.Transaction{
|
||||
|
|
@ -136,5 +141,46 @@ func (oq *destinationQueue) next() *gomatrixserverlib.Transaction {
|
|||
oq.pendingEDUs = nil
|
||||
oq.sentCounter += len(t.EDUs)
|
||||
|
||||
return &t
|
||||
util.GetLogger(context.TODO()).Infof("Sending transaction %q containing %d PDUs, %d EDUs", t.TransactionID, len(t.PDUs), len(t.EDUs))
|
||||
|
||||
_, err := oq.client.SendTransaction(context.TODO(), t)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"destination": oq.destination,
|
||||
log.ErrorKey: err,
|
||||
}).Info("problem sending transaction")
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// nextInvite takes pending invite events from the queue and sends
|
||||
// them. Returns true if a transaction was sent or false otherwise.
|
||||
func (oq *destinationQueue) nextInvites() bool {
|
||||
oq.runningMutex.Lock()
|
||||
defer oq.runningMutex.Unlock()
|
||||
|
||||
if len(oq.pendingInvites) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
for _, inviteReq := range oq.pendingInvites {
|
||||
ev := inviteReq.Event()
|
||||
|
||||
if _, err := oq.client.SendInviteV2(
|
||||
context.TODO(),
|
||||
oq.destination,
|
||||
*inviteReq,
|
||||
); err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"event_id": ev.EventID(),
|
||||
"state_key": ev.StateKey(),
|
||||
"destination": oq.destination,
|
||||
}).WithError(err).Error("failed to send invite")
|
||||
}
|
||||
}
|
||||
|
||||
oq.pendingInvites = nil
|
||||
|
||||
return true
|
||||
}
|
||||
|
|
|
|||
|
|
@ -80,6 +80,49 @@ func (oqs *OutgoingQueues) SendEvent(
|
|||
return nil
|
||||
}
|
||||
|
||||
// SendEvent sends an event to the destinations
|
||||
func (oqs *OutgoingQueues) SendInvite(
|
||||
inviteReq *gomatrixserverlib.InviteV2Request,
|
||||
) error {
|
||||
ev := inviteReq.Event()
|
||||
stateKey := ev.StateKey()
|
||||
if stateKey == nil {
|
||||
log.WithFields(log.Fields{
|
||||
"event_id": ev.EventID(),
|
||||
}).Info("invite had no state key, dropping")
|
||||
return nil
|
||||
}
|
||||
|
||||
_, destination, err := gomatrixserverlib.SplitID('@', *stateKey)
|
||||
if err != nil {
|
||||
log.WithFields(log.Fields{
|
||||
"event_id": ev.EventID(),
|
||||
"state_key": stateKey,
|
||||
}).Info("failed to split destination from state key")
|
||||
return nil
|
||||
}
|
||||
|
||||
log.WithFields(log.Fields{
|
||||
"event_id": ev.EventID(),
|
||||
}).Info("Sending invite")
|
||||
|
||||
oqs.queuesMutex.Lock()
|
||||
defer oqs.queuesMutex.Unlock()
|
||||
oq := oqs.queues[destination]
|
||||
if oq == nil {
|
||||
oq = &destinationQueue{
|
||||
origin: oqs.origin,
|
||||
destination: destination,
|
||||
client: oqs.client,
|
||||
}
|
||||
oqs.queues[destination] = oq
|
||||
}
|
||||
|
||||
oq.sendInvite(inviteReq)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// SendEDU sends an EDU event to the destinations
|
||||
func (oqs *OutgoingQueues) SendEDU(
|
||||
e *gomatrixserverlib.EDU, origin gomatrixserverlib.ServerName,
|
||||
|
|
|
|||
6
go.mod
6
go.mod
|
|
@ -1,6 +1,7 @@
|
|||
module github.com/matrix-org/dendrite
|
||||
|
||||
require (
|
||||
git.sr.ht/~fnux/yggdrasil-go-coap v0.0.0-20191202152715-614f652b70b2 // indirect
|
||||
github.com/gorilla/mux v1.7.3
|
||||
github.com/hashicorp/golang-lru v0.5.4
|
||||
github.com/lib/pq v1.2.0
|
||||
|
|
@ -9,7 +10,7 @@ require (
|
|||
github.com/matrix-org/go-http-js-libp2p v0.0.0-20200318135427-31631a9ef51f
|
||||
github.com/matrix-org/go-sqlite3-js v0.0.0-20200325174927-327088cdef10
|
||||
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200401100216-9ebd9f9fa5c3
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200402141635-4a6e1ade46f8
|
||||
github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1
|
||||
github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7
|
||||
github.com/mattn/go-sqlite3 v2.0.2+incompatible
|
||||
|
|
@ -23,9 +24,8 @@ require (
|
|||
github.com/tidwall/pretty v1.0.1 // indirect
|
||||
github.com/uber/jaeger-client-go v2.22.1+incompatible
|
||||
github.com/uber/jaeger-lib v2.2.0+incompatible
|
||||
go.uber.org/atomic v1.6.0 // indirect
|
||||
go.uber.org/atomic v1.6.0
|
||||
golang.org/x/crypto v0.0.0-20200115085410-6d4e4cb37c7d
|
||||
golang.org/x/net v0.0.0-20190909003024-a7b16738d86b // indirect
|
||||
gopkg.in/Shopify/sarama.v1 v1.20.1
|
||||
gopkg.in/h2non/bimg.v1 v1.0.18
|
||||
gopkg.in/yaml.v2 v2.2.5
|
||||
|
|
|
|||
36
go.sum
36
go.sum
|
|
@ -1,4 +1,8 @@
|
|||
cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
|
||||
git.sr.ht/~fnux/yggdrasil-go-coap v0.0.0-20191202152715-614f652b70b2 h1:S7Qd470lOSDq5ywxq6gYKdMqhKkU0/BqeBhiadlwDOQ=
|
||||
git.sr.ht/~fnux/yggdrasil-go-coap v0.0.0-20191202152715-614f652b70b2/go.mod h1:02J6pHPMVUn6zBUdU3CSkdZdi3Du65nvTa2XaYdQQoU=
|
||||
github.com/Arceliar/phony v0.0.0-20191006174943-d0c68492aca0 h1:p3puK8Sl2xK+2FnnIvY/C0N1aqJo2kbEsdAzU+Tnv48=
|
||||
github.com/Arceliar/phony v0.0.0-20191006174943-d0c68492aca0/go.mod h1:6Lkn+/zJilRMsKmbmG1RPoamiArC6HS73xbwRyp3UyI=
|
||||
github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
|
||||
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
|
||||
github.com/DataDog/zstd v1.4.4 h1:+IawcoXhCBylN7ccwdwf8LOH2jKq7NavGpEPanrlTzE=
|
||||
|
|
@ -65,6 +69,8 @@ github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs
|
|||
github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
|
||||
github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
|
||||
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
|
||||
github.com/gologme/log v0.0.0-20181207131047-4e5d8ccb38e8 h1:WD8iJ37bRNwvETMfVTusVSAi0WdXTpfNVGY2aHycNKY=
|
||||
github.com/gologme/log v0.0.0-20181207131047-4e5d8ccb38e8/go.mod h1:gq31gQ8wEHkR+WekdWsqDuf8pXTUZA9BnnzTuPz1Y9U=
|
||||
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
|
||||
github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
|
||||
github.com/google/go-cmp v0.3.1 h1:Xye71clBPdm5HgqGwUkwhbynsUJZhDbS20FvLhQ2izg=
|
||||
|
|
@ -76,8 +82,10 @@ github.com/gorilla/mux v1.7.3 h1:gnP5JzjVOuiZD07fKKToCAOjS0yOpj/qPETTXCCS6hw=
|
|||
github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
|
||||
github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542 h1:2VTzZjLZBgl62/EtslCrtky5vbi9dd7HrQPQIx6wqiw=
|
||||
github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542/go.mod h1:Ow0tF8D4Kplbc8s8sSb3V2oUCygFHVp8gC3Dn6U4MNI=
|
||||
github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=
|
||||
github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
|
||||
github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
|
||||
github.com/hjson/hjson-go v3.0.1-0.20190209023717-9147687966d9+incompatible/go.mod h1:qsetwF8NlsTsOTwZTApNlTCerV+b2GjYRRcIk4JMFio=
|
||||
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
|
||||
github.com/ipfs/go-cid v0.0.5 h1:o0Ix8e/ql7Zb5UVUJEUfjsWCIY8t48++9lR8qi6oiJU=
|
||||
github.com/ipfs/go-cid v0.0.5/go.mod h1:plgt+Y5MnOey4vO4UlUazGqdbEXuFYitED67FexhXog=
|
||||
|
|
@ -92,6 +100,7 @@ github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/u
|
|||
github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
|
||||
github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
|
||||
github.com/kami-zh/go-capturer v0.0.0-20171211120116-e492ea43421d/go.mod h1:P2viExyCEfeWGU259JnaQ34Inuec4R38JCyBx2edgD0=
|
||||
github.com/kardianos/minwinsvc v0.0.0-20151122163309-cad6b2b879b0/go.mod h1:rUi0/YffDo1oXBOGn1KRq7Fr07LX48XEBecQnmwjsAo=
|
||||
github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
|
||||
github.com/kisielk/gotool v1.0.0 h1:AV2c/EiW3KqPNT9ZKl07ehoAGi4C5/01Cfbblndcapg=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
|
|
@ -130,12 +139,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh
|
|||
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5 h1:kmRjpmFOenVpOaV/DRlo9p6z/IbOKlUC+hhKsAAh8Qg=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200124100636-0c2ec91d1df5/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200331153410-b4aa822258d9 h1:4KfTTOqT8lwXyqvHYHQ9jELpWsy9EilCpcz+pGLPpwY=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200331153410-b4aa822258d9/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200401082842-15690db5a292 h1:vg3wldYGfKsblOevioYymsHTcfGXjUrDHNTAGasPV5g=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200401082842-15690db5a292/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200401100216-9ebd9f9fa5c3 h1:38jIp2cRx20xt8XuLivN3DtkAomJiihIlbmZosRoZT8=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200401100216-9ebd9f9fa5c3/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200402141635-4a6e1ade46f8 h1:VZ7xGklSuzU9geMekuxKO4FvUBUaPjP+8IkcwzQtqOI=
|
||||
github.com/matrix-org/gomatrixserverlib v0.0.0-20200402141635-4a6e1ade46f8/go.mod h1:FsKa2pWE/bpQql9H7U4boOPXFoJX/QcqaZZ6ijLkaZI=
|
||||
github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1 h1:osLoFdOy+ChQqVUn2PeTDETFftVkl4w9t/OW18g3lnk=
|
||||
github.com/matrix-org/naffka v0.0.0-20200127221512-0716baaabaf1/go.mod h1:cXoYQIENbdWIQHt1SyCo6Bl3C3raHwJ0wgVrXHSqf+A=
|
||||
github.com/matrix-org/util v0.0.0-20171127121716-2e2df66af2f5 h1:W7l5CP4V7wPyPb4tYE11dbmeAOwtFQBTW0rf4OonOS8=
|
||||
|
|
@ -158,6 +163,7 @@ github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1/go.mod h1:pD8Rv
|
|||
github.com/minio/sha256-simd v0.1.1-0.20190913151208-6de447530771/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
|
||||
github.com/minio/sha256-simd v0.1.1 h1:5QHSlgo3nt5yKOJrC7W8w7X+NFl8cMPZm96iu8kKUJU=
|
||||
github.com/minio/sha256-simd v0.1.1/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
|
||||
github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
|
||||
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
|
||||
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
|
||||
github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
|
||||
|
|
@ -190,6 +196,11 @@ github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsq
|
|||
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
|
||||
github.com/pierrec/lz4 v2.4.1+incompatible h1:mFe7ttWaflA46Mhqh+jUfjp2qTbPYxLB2/OyBppH9dg=
|
||||
github.com/pierrec/lz4 v2.4.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
|
||||
github.com/pion/dtls v1.5.2 h1:cIVSR1GPGfUAnRS1nl7jSdpoB63WOLANSu4ewpwRHzg=
|
||||
github.com/pion/dtls v1.5.2/go.mod h1:v4ULmyyV65geAZQBBckCjgMhmngTqz7HQVsQVYnfkGo=
|
||||
github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
|
||||
github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
|
||||
github.com/pion/transport v0.8.9/go.mod h1:lpeSM6KJFejVtZf8k0fgeN7zE73APQpTF83WvA1FVP8=
|
||||
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
|
|
@ -224,6 +235,8 @@ github.com/sirupsen/logrus v1.3.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPx
|
|||
github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
|
||||
github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
|
||||
github.com/smola/gocompat v0.2.0/go.mod h1:1B0MlxbmoZNo3h8guHp8HztB3BSYR5itql9qtVc0ypY=
|
||||
github.com/songgao/packets v0.0.0-20160404182456-549a10cd4091/go.mod h1:N20Z5Y8oye9a7HmytmZ+tr8Q2vlP0tAHP13kTHzwvQY=
|
||||
github.com/songgao/water v0.0.0-20190725173103-fd331bda3f4b/go.mod h1:P5HUIBuIWKbyjl083/loAegFkfbFNx5i2qEP4CNbm7E=
|
||||
github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 h1:RC6RW7j+1+HkWaX/Yh71Ee5ZHaHYt7ZP4sQgUrm6cDU=
|
||||
github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572/go.mod h1:w0SWMsp6j9O/dk4/ZpIhL+3CkG8ofA2vuv7k+ltqUMc=
|
||||
github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
|
||||
|
|
@ -256,7 +269,14 @@ github.com/uber/jaeger-lib v1.5.0 h1:OHbgr8l656Ub3Fw5k9SWnBfIEwvoHQ+W2y+Aa9D1Uyo
|
|||
github.com/uber/jaeger-lib v1.5.0/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
|
||||
github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=
|
||||
github.com/uber/jaeger-lib v2.2.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
|
||||
github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
|
||||
github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
|
||||
github.com/vishvananda/netlink v1.0.0/go.mod h1:+SR5DhBJrl6ZM7CoCKvpw5BKroDKQ+PJqOg65H/2ktk=
|
||||
github.com/vishvananda/netns v0.0.0-20190625233234-7109fa855b0f/go.mod h1:ZjcWmFBXmLKZu9Nxj3WKYEafiSqer2rnvPr0en9UNpI=
|
||||
github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
|
||||
github.com/yggdrasil-network/water v0.0.0-20190812103929-c83fe40250f8/go.mod h1:R0SBCsugm+Sf1katgTb2t7GXMm+nRIv43tM4VDZbaOs=
|
||||
github.com/yggdrasil-network/yggdrasil-go v0.3.12 h1:LS8h5VNrBAVRtQKe8Pl+QJDD3gAdmpOd3U4dNaisPa0=
|
||||
github.com/yggdrasil-network/yggdrasil-go v0.3.12/go.mod h1:uZF8kl7k+KJKsav0b7mTTJFMOj5Aq6jU32XlYG5def0=
|
||||
go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
|
||||
go.uber.org/atomic v1.3.0 h1:vs7fgriifsPbGdK3bNuMWapNn3qnZhCRXc19NRdq010=
|
||||
go.uber.org/atomic v1.3.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
|
||||
|
|
@ -270,6 +290,7 @@ golang.org/x/crypto v0.0.0-20190131182504-b8fe1690c613/go.mod h1:6SG95UA2DQfeDnf
|
|||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2 h1:VklqNMn3ovrHsnt90PveolxSbWFaJdECFbxSq0Mqo2M=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20191001170739-f9e2070545dc/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550 h1:ObdrDkeb4kJdCP557AjRjq69pTHfNouLtWZG7j9rPN8=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200115085410-6d4e4cb37c7d h1:2+ZP7EfsZV7Vvmx3TIqSlSzATMkTAKqM14YGFPoSKjI=
|
||||
|
|
@ -294,6 +315,8 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwL
|
|||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20190909003024-a7b16738d86b h1:XfVGCX+0T4WOStkaOsJRllbsiImhB2jgVBGc9L0lPGc=
|
||||
golang.org/x/net v0.0.0-20190909003024-a7b16738d86b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20191021144547-ec77196f6094 h1:5O4U9trLjNpuhpynaDsqwCk+Tw6seqJz1EbqbnzHrc8=
|
||||
golang.org/x/net v0.0.0-20191021144547-ec77196f6094/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
|
||||
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
|
|
@ -315,6 +338,7 @@ golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7w
|
|||
golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191024172528-b4ff53e7a1cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200122134326-e047566fdf82 h1:ywK/j/KkyTHcdyYSZNXGjMwgmDSfjglYZ3vStQ/gSCU=
|
||||
golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ package api
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
commonHTTP "github.com/matrix-org/dendrite/common/http"
|
||||
|
|
@ -139,12 +140,12 @@ const RoomserverGetCreatorIDForAliasPath = "/api/roomserver/GetCreatorIDForAlias
|
|||
const RoomserverRemoveRoomAliasPath = "/api/roomserver/removeRoomAlias"
|
||||
|
||||
// NewRoomserverAliasAPIHTTP creates a RoomserverAliasAPI implemented by talking to a HTTP POST API.
|
||||
// If httpClient is nil then it uses the http.DefaultClient
|
||||
func NewRoomserverAliasAPIHTTP(roomserverURL string, httpClient *http.Client) RoomserverAliasAPI {
|
||||
// If httpClient is nil an error is returned
|
||||
func NewRoomserverAliasAPIHTTP(roomserverURL string, httpClient *http.Client) (RoomserverAliasAPI, error) {
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
return nil, errors.New("NewRoomserverAliasAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpRoomserverAliasAPI{roomserverURL, httpClient}
|
||||
return &httpRoomserverAliasAPI{roomserverURL, httpClient}, nil
|
||||
}
|
||||
|
||||
type httpRoomserverAliasAPI struct {
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ package api
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
commonHTTP "github.com/matrix-org/dendrite/common/http"
|
||||
|
|
@ -85,7 +86,9 @@ type TransactionID struct {
|
|||
// the usual context a matrix room event would have. We usually do not have
|
||||
// access to the events needed to check the event auth rules for the invite.
|
||||
type InputInviteEvent struct {
|
||||
Event gomatrixserverlib.HeaderedEvent `json:"event"`
|
||||
RoomVersion gomatrixserverlib.RoomVersion `json:"room_version"`
|
||||
Event gomatrixserverlib.HeaderedEvent `json:"event"`
|
||||
InviteRoomState []gomatrixserverlib.InviteV2StrippedState `json:"invite_room_state"`
|
||||
}
|
||||
|
||||
// InputRoomEventsRequest is a request to InputRoomEvents
|
||||
|
|
@ -112,12 +115,12 @@ type RoomserverInputAPI interface {
|
|||
const RoomserverInputRoomEventsPath = "/api/roomserver/inputRoomEvents"
|
||||
|
||||
// NewRoomserverInputAPIHTTP creates a RoomserverInputAPI implemented by talking to a HTTP POST API.
|
||||
// If httpClient is nil then it uses the http.DefaultClient
|
||||
func NewRoomserverInputAPIHTTP(roomserverURL string, httpClient *http.Client) RoomserverInputAPI {
|
||||
// If httpClient is nil an error is returned
|
||||
func NewRoomserverInputAPIHTTP(roomserverURL string, httpClient *http.Client) (RoomserverInputAPI, error) {
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
return nil, errors.New("NewRoomserverInputAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpRoomserverInputAPI{roomserverURL, httpClient}
|
||||
return &httpRoomserverInputAPI{roomserverURL, httpClient}, nil
|
||||
}
|
||||
|
||||
type httpRoomserverInputAPI struct {
|
||||
|
|
|
|||
|
|
@ -116,6 +116,8 @@ type OutputNewRoomEvent struct {
|
|||
// 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 room version of the invited room.
|
||||
RoomVersion gomatrixserverlib.RoomVersion `json:"room_version"`
|
||||
// The "m.room.member" invite event.
|
||||
Event gomatrixserverlib.HeaderedEvent `json:"event"`
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ package api
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
commonHTTP "github.com/matrix-org/dendrite/common/http"
|
||||
|
|
@ -406,12 +407,12 @@ const RoomserverQueryRoomVersionCapabilitiesPath = "/api/roomserver/queryRoomVer
|
|||
const RoomserverQueryRoomVersionForRoomPath = "/api/roomserver/queryRoomVersionForRoom"
|
||||
|
||||
// NewRoomserverQueryAPIHTTP creates a RoomserverQueryAPI implemented by talking to a HTTP POST API.
|
||||
// If httpClient is nil then it uses the http.DefaultClient
|
||||
func NewRoomserverQueryAPIHTTP(roomserverURL string, httpClient *http.Client) RoomserverQueryAPI {
|
||||
// If httpClient is nil an error is returned
|
||||
func NewRoomserverQueryAPIHTTP(roomserverURL string, httpClient *http.Client) (RoomserverQueryAPI, error) {
|
||||
if httpClient == nil {
|
||||
httpClient = http.DefaultClient
|
||||
return nil, errors.New("NewRoomserverQueryAPIHTTP: httpClient is <nil>")
|
||||
}
|
||||
return &httpRoomserverQueryAPI{roomserverURL, httpClient}
|
||||
return &httpRoomserverQueryAPI{roomserverURL, httpClient}, nil
|
||||
}
|
||||
|
||||
type httpRoomserverQueryAPI struct {
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ import (
|
|||
"github.com/matrix-org/dendrite/roomserver/state/database"
|
||||
"github.com/matrix-org/dendrite/roomserver/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
// A RoomEventDatabase has the storage APIs needed to store a room event.
|
||||
|
|
@ -64,6 +65,7 @@ type RoomEventDatabase interface {
|
|||
// Build a membership updater for the target user in a room.
|
||||
MembershipUpdater(
|
||||
ctx context.Context, roomID, targerUserID string,
|
||||
roomVersion gomatrixserverlib.RoomVersion,
|
||||
) (types.MembershipUpdater, error)
|
||||
// Look up event ID by transaction's info.
|
||||
// This is used to determine if the room event is processed/processing already.
|
||||
|
|
@ -193,7 +195,14 @@ func processInviteEvent(
|
|||
roomID := input.Event.RoomID()
|
||||
targetUserID := *input.Event.StateKey()
|
||||
|
||||
updater, err := db.MembershipUpdater(ctx, roomID, targetUserID)
|
||||
log.WithFields(log.Fields{
|
||||
"event_id": input.Event.EventID(),
|
||||
"room_id": roomID,
|
||||
"room_version": input.RoomVersion,
|
||||
"target_user_id": targetUserID,
|
||||
}).Info("processing invite event")
|
||||
|
||||
updater, err := db.MembershipUpdater(ctx, roomID, targetUserID, input.RoomVersion)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -237,7 +246,12 @@ func processInviteEvent(
|
|||
}
|
||||
|
||||
event := input.Event.Unwrap()
|
||||
outputUpdates, err := updateToInviteMembership(updater, &event, nil)
|
||||
|
||||
if err = event.SetUnsignedField("invite_room_state", input.InviteRoomState); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
outputUpdates, err := updateToInviteMembership(updater, &event, nil, input.Event.RoomVersion)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -112,7 +112,7 @@ func updateMembership(
|
|||
|
||||
switch newMembership {
|
||||
case gomatrixserverlib.Invite:
|
||||
return updateToInviteMembership(mu, add, updates)
|
||||
return updateToInviteMembership(mu, add, updates, updater.RoomVersion())
|
||||
case gomatrixserverlib.Join:
|
||||
return updateToJoinMembership(mu, add, updates)
|
||||
case gomatrixserverlib.Leave, gomatrixserverlib.Ban:
|
||||
|
|
@ -126,6 +126,7 @@ func updateMembership(
|
|||
|
||||
func updateToInviteMembership(
|
||||
mu types.MembershipUpdater, add *gomatrixserverlib.Event, updates []api.OutputEvent,
|
||||
roomVersion gomatrixserverlib.RoomVersion,
|
||||
) ([]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
|
||||
|
|
@ -136,14 +137,14 @@ func updateToInviteMembership(
|
|||
return nil, err
|
||||
}
|
||||
if needsSending {
|
||||
roomVersion := gomatrixserverlib.RoomVersionV1
|
||||
// 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).Headered(roomVersion),
|
||||
Event: (*add).Headered(roomVersion),
|
||||
RoomVersion: roomVersion,
|
||||
}
|
||||
updates = append(updates, api.OutputEvent{
|
||||
Type: api.OutputTypeNewInviteEvent,
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@ type Database interface {
|
|||
GetAliasesForRoomID(ctx context.Context, roomID string) ([]string, error)
|
||||
GetCreatorIDForAlias(ctx context.Context, alias string) (string, error)
|
||||
RemoveRoomAlias(ctx context.Context, alias string) error
|
||||
MembershipUpdater(ctx context.Context, roomID, targetUserID string) (types.MembershipUpdater, error)
|
||||
MembershipUpdater(ctx context.Context, roomID, targetUserID string, roomVersion gomatrixserverlib.RoomVersion) (types.MembershipUpdater, error)
|
||||
GetMembership(ctx context.Context, roomNID types.RoomNID, requestSenderUserID string) (membershipEventNID types.EventNID, stillInRoom bool, err error)
|
||||
GetMembershipEventNIDsForRoom(ctx context.Context, roomNID types.RoomNID, joinOnly bool) ([]types.EventNID, error)
|
||||
EventsFromIDs(ctx context.Context, eventIDs []string) ([]types.Event, error)
|
||||
|
|
|
|||
|
|
@ -393,6 +393,12 @@ type roomRecentEventsUpdater struct {
|
|||
currentStateSnapshotNID types.StateSnapshotNID
|
||||
}
|
||||
|
||||
// RoomVersion implements types.RoomRecentEventsUpdater
|
||||
func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) {
|
||||
version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID)
|
||||
return
|
||||
}
|
||||
|
||||
// LatestEvents implements types.RoomRecentEventsUpdater
|
||||
func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference {
|
||||
return u.latestEvents
|
||||
|
|
@ -534,6 +540,7 @@ func (d *Database) StateEntriesForTuples(
|
|||
// MembershipUpdater implements input.RoomEventDatabase
|
||||
func (d *Database) MembershipUpdater(
|
||||
ctx context.Context, roomID, targetUserID string,
|
||||
roomVersion gomatrixserverlib.RoomVersion,
|
||||
) (types.MembershipUpdater, error) {
|
||||
txn, err := d.db.Begin()
|
||||
if err != nil {
|
||||
|
|
@ -546,8 +553,7 @@ func (d *Database) MembershipUpdater(
|
|||
}
|
||||
}()
|
||||
|
||||
// TODO: Room version here
|
||||
roomNID, err := d.assignRoomNID(ctx, txn, roomID, "1")
|
||||
roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -486,6 +486,12 @@ type roomRecentEventsUpdater struct {
|
|||
currentStateSnapshotNID types.StateSnapshotNID
|
||||
}
|
||||
|
||||
// RoomVersion implements types.RoomRecentEventsUpdater
|
||||
func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) {
|
||||
version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID)
|
||||
return
|
||||
}
|
||||
|
||||
// LatestEvents implements types.RoomRecentEventsUpdater
|
||||
func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference {
|
||||
return u.latestEvents
|
||||
|
|
@ -657,6 +663,7 @@ func (d *Database) StateEntriesForTuples(
|
|||
// MembershipUpdater implements input.RoomEventDatabase
|
||||
func (d *Database) MembershipUpdater(
|
||||
ctx context.Context, roomID, targetUserID string,
|
||||
roomVersion gomatrixserverlib.RoomVersion,
|
||||
) (updater types.MembershipUpdater, err error) {
|
||||
var txn *sql.Tx
|
||||
txn, err = d.db.Begin()
|
||||
|
|
@ -682,8 +689,7 @@ func (d *Database) MembershipUpdater(
|
|||
}
|
||||
}()
|
||||
|
||||
// TODO: Room version here
|
||||
roomNID, err := d.assignRoomNID(ctx, txn, roomID, "1")
|
||||
roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -140,6 +140,8 @@ type StateEntryList struct {
|
|||
// (On postgresql this wraps a database transaction that holds a "FOR UPDATE"
|
||||
// lock on the row in the rooms table holding the latest events for the room.)
|
||||
type RoomRecentEventsUpdater interface {
|
||||
// The room version of the room.
|
||||
RoomVersion() gomatrixserverlib.RoomVersion
|
||||
// The latest event IDs and state in the room.
|
||||
LatestEvents() []StateAtEventAndReference
|
||||
// The event ID of the latest event written to the output log in the room.
|
||||
|
|
|
|||
|
|
@ -239,3 +239,4 @@ local user can join room with version 4
|
|||
remote user can join room with version 4
|
||||
Remote user can backfill in a room with version 4
|
||||
Ignore invite in incremental sync
|
||||
Outbound federation can send invites via v2 API
|
||||
|
|
|
|||
Loading…
Reference in a new issue