hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-07-30 21:12:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Move MakeJoin logic to GMSL (#3081)

Browse source
This commit is contained in:
devonh 2023-05-17 00:33:27 +00:00 • committed by GitHub
parent 0489d16f95
commit 67d6876857
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
80 changed files with 1158 additions and 494 deletions
Download patch file Download diff file Expand all files Collapse all files

45
syncapi/routing/context.go
View file

@ -56,7 +56,10 @@ func Context(
) util.JSONResponse {
snapshot, err := syncDB.NewDatabaseSnapshot(req.Context())
if err != nil {
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
var succeeded bool
defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
@ -87,7 +90,10 @@ func Context(
membershipReq := roomserver.QueryMembershipForUserRequest{UserID: device.UserID, RoomID: roomID}
if err = rsAPI.QueryMembershipForUser(ctx, &membershipReq, &membershipRes); err != nil {
logrus.WithError(err).Error("unable to query membership")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
if !membershipRes.RoomExists {
return util.JSONResponse{
@ -117,7 +123,10 @@ func Context(
}
}
logrus.WithError(err).WithField("eventID", eventID).Error("unable to find requested event")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
// verify the user is allowed to see the context for this room/event
@ -125,7 +134,10 @@ func Context(
filteredEvents, err := internal.ApplyHistoryVisibilityFilter(ctx, snapshot, rsAPI, []*rstypes.HeaderedEvent{&requestedEvent}, nil, device.UserID, "context")
if err != nil {
logrus.WithError(err).Error("unable to apply history visibility filter")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
logrus.WithFields(logrus.Fields{
"duration": time.Since(startTime),
@ -141,20 +153,29 @@ func Context(
eventsBefore, err := snapshot.SelectContextBeforeEvent(ctx, id, roomID, filter)
if err != nil && err != sql.ErrNoRows {
logrus.WithError(err).Error("unable to fetch before events")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
_, eventsAfter, err := snapshot.SelectContextAfterEvent(ctx, id, roomID, filter)
if err != nil && err != sql.ErrNoRows {
logrus.WithError(err).Error("unable to fetch after events")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
startTime = time.Now()
eventsBeforeFiltered, eventsAfterFiltered, err := applyHistoryVisibilityOnContextEvents(ctx, snapshot, rsAPI, eventsBefore, eventsAfter, device.UserID)
if err != nil {
logrus.WithError(err).Error("unable to apply history visibility filter")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
logrus.WithFields(logrus.Fields{
@ -166,7 +187,10 @@ func Context(
state, err := snapshot.CurrentState(ctx, roomID, &stateFilter, nil)
if err != nil {
logrus.WithError(err).Error("unable to fetch current room state")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
eventsBeforeClient := synctypes.ToClientEvents(gomatrixserverlib.ToPDUs(eventsBeforeFiltered), synctypes.FormatAll)
@ -180,7 +204,10 @@ func Context(
newState, err = applyLazyLoadMembers(ctx, device, snapshot, roomID, evs, lazyLoadCache)
if err != nil {
logrus.WithError(err).Error("unable to load membership events")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
}

15
syncapi/routing/filter.go
View file

@ -43,7 +43,10 @@ func GetFilter(
localpart, _, err := gomatrixserverlib.SplitID('@', userID)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("gomatrixserverlib.SplitID failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
filter := synctypes.DefaultFilter()
@ -83,7 +86,10 @@ func PutFilter(
localpart, _, err := gomatrixserverlib.SplitID('@', userID)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("gomatrixserverlib.SplitID failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
var filter synctypes.Filter
@ -122,7 +128,10 @@ func PutFilter(
filterID, err := syncDB.PutFilter(req.Context(), localpart, &filter)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("syncDB.PutFilter failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
return util.JSONResponse{

12
syncapi/routing/getevent.go
View file

@ -51,13 +51,19 @@ func GetEvent(
})
if err != nil {
logger.WithError(err).Error("GetEvent: syncDB.NewDatabaseTransaction failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
events, err := db.Events(ctx, []string{eventID})
if err != nil {
logger.WithError(err).Error("GetEvent: syncDB.Events failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
// The requested event does not exist in our database
@ -81,7 +87,7 @@ func GetEvent(
logger.WithError(err).Error("GetEvent: internal.ApplyHistoryVisibilityFilter failed")
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError(),
JSON: spec.InternalServerError{},
}
}

30
syncapi/routing/memberships.go
View file

@ -67,7 +67,10 @@ func GetMemberships(
var queryRes api.QueryMembershipForUserResponse
if err := rsAPI.QueryMembershipForUser(req.Context(), &queryReq, &queryRes); err != nil {
util.GetLogger(req.Context()).WithError(err).Error("rsAPI.QueryMembershipsForRoom failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
if !queryRes.HasBeenInRoom {
@ -86,7 +89,10 @@ func GetMemberships(
db, err := syncDB.NewDatabaseSnapshot(req.Context())
if err != nil {
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
defer db.Rollback() // nolint: errcheck
@ -98,7 +104,10 @@ func GetMemberships(
atToken, err = db.EventPositionInTopology(req.Context(), queryRes.EventID)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("unable to get 'atToken'")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
}
}
@ -106,13 +115,19 @@ func GetMemberships(
eventIDs, err := db.SelectMemberships(req.Context(), roomID, atToken, membership, notMembership)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("db.SelectMemberships failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
qryRes := &api.QueryEventsByIDResponse{}
if err := rsAPI.QueryEventsByID(req.Context(), &api.QueryEventsByIDRequest{EventIDs: eventIDs, RoomID: roomID}, qryRes); err != nil {
util.GetLogger(req.Context()).WithError(err).Error("rsAPI.QueryEventsByID failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
result := qryRes.Events
@ -124,7 +139,10 @@ func GetMemberships(
var content databaseJoinedMember
if err := json.Unmarshal(ev.Content(), &content); err != nil {
util.GetLogger(req.Context()).WithError(err).Error("failed to unmarshal event content")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
res.Joined[ev.Sender()] = joinedMember(content)
}

30
syncapi/routing/messages.go
View file

@ -81,7 +81,10 @@ func OnIncomingMessagesRequest(
// request that requires backfilling from the roomserver or federation.
snapshot, err := db.NewDatabaseTransaction(req.Context())
if err != nil {
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
var succeeded bool
defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
@ -89,7 +92,10 @@ func OnIncomingMessagesRequest(
// check if the user has already forgotten about this room
membershipResp, err := getMembershipForUser(req.Context(), roomID, device.UserID, rsAPI)
if err != nil {
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
if !membershipResp.RoomExists {
return util.JSONResponse{
@ -151,7 +157,10 @@ func OnIncomingMessagesRequest(
from, err = snapshot.StreamToTopologicalPosition(req.Context(), roomID, streamToken.PDUPosition, backwardOrdering)
if err != nil {
logrus.WithError(err).Errorf("Failed to get topological position for streaming token %v", streamToken)
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
}
}
@ -173,7 +182,10 @@ func OnIncomingMessagesRequest(
to, err = snapshot.StreamToTopologicalPosition(req.Context(), roomID, streamToken.PDUPosition, !backwardOrdering)
if err != nil {
logrus.WithError(err).Errorf("Failed to get topological position for streaming token %v", streamToken)
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
}
}
@ -232,7 +244,10 @@ func OnIncomingMessagesRequest(
clientEvents, start, end, err := mReq.retrieveEvents()
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("mreq.retrieveEvents failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
util.GetLogger(req.Context()).WithFields(logrus.Fields{
@ -253,7 +268,10 @@ func OnIncomingMessagesRequest(
membershipEvents, err := applyLazyLoadMembers(req.Context(), device, snapshot, roomID, clientEvents, lazyLoadCache)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("failed to apply lazy loading")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
res.State = append(res.State, synctypes.ToClientEvents(gomatrixserverlib.ToPDUs(membershipEvents), synctypes.FormatAll)...)
}

5
syncapi/routing/relations.go
View file

@ -80,7 +80,10 @@ func Relations(
snapshot, err := syncDB.NewDatabaseSnapshot(req.Context())
if err != nil {
logrus.WithError(err).Error("Failed to get snapshot for relations")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
var succeeded bool
defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)

5
syncapi/routing/routing.go
View file

@ -162,7 +162,10 @@ func Setup(
}
var nextBatch *string
if err := req.ParseForm(); err != nil {
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
if req.Form.Has("next_batch") {
nb := req.FormValue("next_batch")

40
syncapi/routing/search.go
View file

@ -55,7 +55,10 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
if from != nil && *from != "" {
nextBatch, err = strconv.Atoi(*from)
if err != nil {
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
}
@ -65,7 +68,10 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
snapshot, err := syncDB.NewDatabaseSnapshot(req.Context())
if err != nil {
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
var succeeded bool
defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
@ -73,7 +79,10 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
// only search rooms the user is actually joined to
joinedRooms, err := snapshot.RoomIDsWithMembership(ctx, device.UserID, "join")
if err != nil {
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
if len(joinedRooms) == 0 {
return util.JSONResponse{
@ -115,7 +124,10 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
)
if err != nil {
logrus.WithError(err).Error("failed to search fulltext")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
logrus.Debugf("Search took %s", result.Took)
@ -155,7 +167,10 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
evs, err := syncDB.Events(ctx, wantEvents)
if err != nil {
logrus.WithError(err).Error("failed to get events from database")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
groups := make(map[string]RoomResult)
@ -173,12 +188,18 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
eventsBefore, eventsAfter, err := contextEvents(ctx, snapshot, event, roomFilter, searchReq)
if err != nil {
logrus.WithError(err).Error("failed to get context events")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
startToken, endToken, err := getStartEnd(ctx, snapshot, eventsBefore, eventsAfter)
if err != nil {
logrus.WithError(err).Error("failed to get start/end")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
profileInfos := make(map[string]ProfileInfoResponse)
@ -221,7 +242,10 @@ func Search(req *http.Request, device *api.Device, syncDB storage.Database, fts
state, err := snapshot.CurrentState(ctx, event.RoomID(), &stateFilter, nil)
if err != nil {
logrus.WithError(err).Error("unable to get current state")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
stateForRooms[event.RoomID()] = synctypes.ToClientEvents(gomatrixserverlib.ToPDUs(state), synctypes.FormatSync)
}

15
syncapi/sync/requestpool.go
View file

@ -536,12 +536,18 @@ func (rp *RequestPool) OnIncomingKeyChangeRequest(req *http.Request, device *use
syncReq, err := newSyncRequest(req, *device, rp.db)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("newSyncRequest failed")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
snapshot, err := rp.db.NewDatabaseSnapshot(req.Context())
if err != nil {
logrus.WithError(err).Error("Failed to acquire database snapshot for key change")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
var succeeded bool
defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
@ -552,7 +558,10 @@ func (rp *RequestPool) OnIncomingKeyChangeRequest(req *http.Request, device *use
)
if err != nil {
util.GetLogger(req.Context()).WithError(err).Error("Failed to DeviceListCatchup info")
return spec.InternalServerError()
return util.JSONResponse{
Code: http.StatusInternalServerError,
JSON: spec.InternalServerError{},
}
}
succeeded = true
return util.JSONResponse{