Generic-based internal HTTP API (#2626)

* Generic-based internal HTTP API (tested out on a few endpoints in the federation API)

* Add `PerformInvite`

* More tweaks

* Fix metric name

* Fix LookupStateIDs

* Lots of changes to clients

* Some serverside stuff

* Some error handling

* Use paths as metric names

* Revert "Use paths as metric names"

This reverts commit a9323a6a343f5ce6461a2e5bd570fe06465f1b15.

* Namespace metric names

* Remove duplicate entry

* Remove another duplicate entry

* Tweak error handling

* Some more tweaks

* Update error behaviour

* Some more error tweaking

* Fix API path for `PerformDeleteKeys`

* Fix another path

* Tweak federation client proxying

* Fix another path

* Don't return typed nils

* Some more tweaks, not that it makes any difference

* Tweak federation client proxying

* Maybe fix the key backup test

roomserver/internal/input/input.go

@@ -337,18 +337,18 @@ func (r *Inputer) InputRoomEvents(
 	ctx context.Context,
 	request *api.InputRoomEventsRequest,
 	response *api.InputRoomEventsResponse,
-) {
+) error {
 	// Queue up the event into the roomserver.
 	replySub, err := r.queueInputRoomEvents(ctx, request)
 	if err != nil {
 		response.ErrMsg = err.Error()
-		return
+		return nil
 	}
 
 	// If we aren't waiting for synchronous responses then we can
 	// give up here, there is nothing further to do.
 	if replySub == nil {
-		return
+		return nil
 	}
 
 	// Otherwise, we'll want to sit and wait for the responses
@@ -360,12 +360,14 @@ func (r *Inputer) InputRoomEvents(
 		msg, err := replySub.NextMsgWithContext(ctx)
 		if err != nil {
 			response.ErrMsg = err.Error()
-			return
+			return nil
 		}
 		if len(msg.Data) > 0 {
 			response.ErrMsg = string(msg.Data)
 		}
 	}
+
+	return nil
 }
 
 var roomserverInputBackpressure = prometheus.NewGaugeVec(

roomserver/internal/perform/perform_admin.go

@@ -43,21 +43,21 @@ func (r *Admin) PerformAdminEvacuateRoom(
 	ctx context.Context,
 	req *api.PerformAdminEvacuateRoomRequest,
 	res *api.PerformAdminEvacuateRoomResponse,
-) {
+) error {
 	roomInfo, err := r.DB.RoomInfo(ctx, req.RoomID)
 	if err != nil {
 		res.Error = &api.PerformError{
 			Code: api.PerformErrorBadRequest,
 			Msg:  fmt.Sprintf("r.DB.RoomInfo: %s", err),
 		}
-		return
+		return nil
 	}
 	if roomInfo == nil || roomInfo.IsStub() {
 		res.Error = &api.PerformError{
 			Code: api.PerformErrorNoRoom,
 			Msg:  fmt.Sprintf("Room %s not found", req.RoomID),
 		}
-		return
+		return nil
 	}
 
 	memberNIDs, err := r.DB.GetMembershipEventNIDsForRoom(ctx, roomInfo.RoomNID, true, true)
@@ -66,7 +66,7 @@ func (r *Admin) PerformAdminEvacuateRoom(
 			Code: api.PerformErrorBadRequest,
 			Msg:  fmt.Sprintf("r.DB.GetMembershipEventNIDsForRoom: %s", err),
 		}
-		return
+		return nil
 	}
 
 	memberEvents, err := r.DB.Events(ctx, memberNIDs)
@@ -75,7 +75,7 @@ func (r *Admin) PerformAdminEvacuateRoom(
 			Code: api.PerformErrorBadRequest,
 			Msg:  fmt.Sprintf("r.DB.Events: %s", err),
 		}
-		return
+		return nil
 	}
 
 	inputEvents := make([]api.InputRoomEvent, 0, len(memberEvents))
@@ -89,7 +89,7 @@ func (r *Admin) PerformAdminEvacuateRoom(
 			Code: api.PerformErrorBadRequest,
 			Msg:  fmt.Sprintf("r.Queryer.QueryLatestEventsAndState: %s", err),
 		}
-		return
+		return nil
 	}
 
 	prevEvents := latestRes.LatestEvents
@@ -104,7 +104,7 @@ func (r *Admin) PerformAdminEvacuateRoom(
 				Code: api.PerformErrorBadRequest,
 				Msg:  fmt.Sprintf("json.Unmarshal: %s", err),
 			}
-			return
+			return nil
 		}
 		memberContent.Membership = gomatrixserverlib.Leave
 
@@ -122,7 +122,7 @@ func (r *Admin) PerformAdminEvacuateRoom(
 				Code: api.PerformErrorBadRequest,
 				Msg:  fmt.Sprintf("json.Marshal: %s", err),
 			}
-			return
+			return nil
 		}
 
 		eventsNeeded, err := gomatrixserverlib.StateNeededForEventBuilder(fledglingEvent)
@@ -131,7 +131,7 @@ func (r *Admin) PerformAdminEvacuateRoom(
 				Code: api.PerformErrorBadRequest,
 				Msg:  fmt.Sprintf("gomatrixserverlib.StateNeededForEventBuilder: %s", err),
 			}
-			return
+			return nil
 		}
 
 		event, err := eventutil.BuildEvent(ctx, fledglingEvent, r.Cfg.Matrix, time.Now(), &eventsNeeded, latestRes)
@@ -140,7 +140,7 @@ func (r *Admin) PerformAdminEvacuateRoom(
 				Code: api.PerformErrorBadRequest,
 				Msg:  fmt.Sprintf("eventutil.BuildEvent: %s", err),
 			}
-			return
+			return nil
 		}
 
 		inputEvents = append(inputEvents, api.InputRoomEvent{
@@ -160,28 +160,28 @@ func (r *Admin) PerformAdminEvacuateRoom(
 		Asynchronous:    true,
 	}
 	inputRes := &api.InputRoomEventsResponse{}
-	r.Inputer.InputRoomEvents(ctx, inputReq, inputRes)
+	return r.Inputer.InputRoomEvents(ctx, inputReq, inputRes)
 }
 
 func (r *Admin) PerformAdminEvacuateUser(
 	ctx context.Context,
 	req *api.PerformAdminEvacuateUserRequest,
 	res *api.PerformAdminEvacuateUserResponse,
-) {
+) error {
 	_, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
 	if err != nil {
 		res.Error = &api.PerformError{
 			Code: api.PerformErrorBadRequest,
 			Msg:  fmt.Sprintf("Malformed user ID: %s", err),
 		}
-		return
+		return nil
 	}
 	if domain != r.Cfg.Matrix.ServerName {
 		res.Error = &api.PerformError{
 			Code: api.PerformErrorBadRequest,
 			Msg:  "Can only evacuate local users using this endpoint",
 		}
-		return
+		return nil
 	}
 
 	roomIDs, err := r.DB.GetRoomsByMembership(ctx, req.UserID, gomatrixserverlib.Join)
@@ -190,7 +190,7 @@ func (r *Admin) PerformAdminEvacuateUser(
 			Code: api.PerformErrorBadRequest,
 			Msg:  fmt.Sprintf("r.DB.GetRoomsByMembership: %s", err),
 		}
-		return
+		return nil
 	}
 
 	inviteRoomIDs, err := r.DB.GetRoomsByMembership(ctx, req.UserID, gomatrixserverlib.Invite)
@@ -199,7 +199,7 @@ func (r *Admin) PerformAdminEvacuateUser(
 			Code: api.PerformErrorBadRequest,
 			Msg:  fmt.Sprintf("r.DB.GetRoomsByMembership: %s", err),
 		}
-		return
+		return nil
 	}
 
 	for _, roomID := range append(roomIDs, inviteRoomIDs...) {
@@ -214,7 +214,7 @@ func (r *Admin) PerformAdminEvacuateUser(
 				Code: api.PerformErrorBadRequest,
 				Msg:  fmt.Sprintf("r.Leaver.PerformLeave: %s", err),
 			}
-			return
+			return nil
 		}
 		if len(outputEvents) == 0 {
 			continue
@@ -224,9 +224,10 @@ func (r *Admin) PerformAdminEvacuateUser(
 				Code: api.PerformErrorBadRequest,
 				Msg:  fmt.Sprintf("r.Inputer.WriteOutputEvents: %s", err),
 			}
-			return
+			return nil
 		}
 
 		res.Affected = append(res.Affected, roomID)
 	}
+	return nil
 }

roomserver/internal/perform/perform_invite.go

@@ -241,7 +241,9 @@ func (r *Inviter) PerformInvite(
 		},
 	}
 	inputRes := &api.InputRoomEventsResponse{}
-	r.Inputer.InputRoomEvents(context.Background(), inputReq, inputRes)
+	if err = r.Inputer.InputRoomEvents(context.Background(), inputReq, inputRes); err != nil {
+		return nil, fmt.Errorf("r.Inputer.InputRoomEvents: %w", err)
+	}
 	if err = inputRes.Err(); err != nil {
 		res.Error = &api.PerformError{
 			Msg:  fmt.Sprintf("r.InputRoomEvents: %s", err.Error()),

roomserver/internal/perform/perform_join.go

@@ -52,7 +52,7 @@ func (r *Joiner) PerformJoin(
 	ctx context.Context,
 	req *rsAPI.PerformJoinRequest,
 	res *rsAPI.PerformJoinResponse,
-) {
+) error {
 	logger := logrus.WithContext(ctx).WithFields(logrus.Fields{
 		"room_id": req.RoomIDOrAlias,
 		"user_id": req.UserID,
@@ -71,11 +71,12 @@ func (r *Joiner) PerformJoin(
 				Msg: err.Error(),
 			}
 		}
-		return
+		return nil
 	}
 	logger.Info("User joined room successfully")
 	res.RoomID = roomID
 	res.JoinedVia = joinedVia
+	return nil
 }
 
 func (r *Joiner) performJoin(
@@ -291,7 +292,12 @@ func (r *Joiner) performJoinRoomByID(
 				},
 			}
 			inputRes := rsAPI.InputRoomEventsResponse{}
-			r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes)
+			if err = r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
+				return "", "", &rsAPI.PerformError{
+					Code: rsAPI.PerformErrorNoOperation,
+					Msg:  fmt.Sprintf("InputRoomEvents failed: %s", err),
+				}
+			}
 			if err = inputRes.Err(); err != nil {
 				return "", "", &rsAPI.PerformError{
 					Code: rsAPI.PerformErrorNotAllowed,

roomserver/internal/perform/perform_leave.go

@@ -186,7 +186,9 @@ func (r *Leaver) performLeaveRoomByID(
 		},
 	}
 	inputRes := api.InputRoomEventsResponse{}
-	r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes)
+	if err = r.Inputer.InputRoomEvents(ctx, &inputReq, &inputRes); err != nil {
+		return nil, fmt.Errorf("r.Inputer.InputRoomEvents: %w", err)
+	}
 	if err = inputRes.Err(); err != nil {
 		return nil, fmt.Errorf("r.InputRoomEvents: %w", err)
 	}

roomserver/internal/perform/perform_peek.go

@@ -44,7 +44,7 @@ func (r *Peeker) PerformPeek(
 	ctx context.Context,
 	req *api.PerformPeekRequest,
 	res *api.PerformPeekResponse,
-) {
+) error {
 	roomID, err := r.performPeek(ctx, req)
 	if err != nil {
 		perr, ok := err.(*api.PerformError)
@@ -57,6 +57,7 @@ func (r *Peeker) PerformPeek(
 		}
 	}
 	res.RoomID = roomID
+	return nil
 }
 
 func (r *Peeker) performPeek(

roomserver/internal/perform/perform_publish.go

@@ -29,11 +29,12 @@ func (r *Publisher) PerformPublish(
 	ctx context.Context,
 	req *api.PerformPublishRequest,
 	res *api.PerformPublishResponse,
-) {
+) error {
 	err := r.DB.PublishRoom(ctx, req.RoomID, req.Visibility == "public")
 	if err != nil {
 		res.Error = &api.PerformError{
 			Msg: err.Error(),
 		}
 	}
+	return nil
 }

roomserver/internal/perform/perform_unpeek.go

@@ -41,7 +41,7 @@ func (r *Unpeeker) PerformUnpeek(
 	ctx context.Context,
 	req *api.PerformUnpeekRequest,
 	res *api.PerformUnpeekResponse,
-) {
+) error {
 	if err := r.performUnpeek(ctx, req); err != nil {
 		perr, ok := err.(*api.PerformError)
 		if ok {
@@ -52,6 +52,7 @@ func (r *Unpeeker) PerformUnpeek(
 			}
 		}
 	}
+	return nil
 }
 
 func (r *Unpeeker) performUnpeek(

roomserver/internal/perform/perform_upgrade.go

@@ -45,12 +45,13 @@ func (r *Upgrader) PerformRoomUpgrade(
 	ctx context.Context,
 	req *api.PerformRoomUpgradeRequest,
 	res *api.PerformRoomUpgradeResponse,
-) {
+) error {
 	res.NewRoomID, res.Error = r.performRoomUpgrade(ctx, req)
 	if res.Error != nil {
 		res.NewRoomID = ""
 		logrus.WithContext(ctx).WithError(res.Error).Error("Room upgrade failed")
 	}
+	return nil
 }
 
 func (r *Upgrader) performRoomUpgrade(
@@ -286,22 +287,24 @@ func publishNewRoomAndUnpublishOldRoom(
 ) {
 	// expose this room in the published room list
 	var pubNewRoomRes api.PerformPublishResponse
-	URSAPI.PerformPublish(ctx, &api.PerformPublishRequest{
+	if err := URSAPI.PerformPublish(ctx, &api.PerformPublishRequest{
 		RoomID:     newRoomID,
 		Visibility: "public",
-	}, &pubNewRoomRes)
-	if pubNewRoomRes.Error != nil {
+	}, &pubNewRoomRes); err != nil {
+		util.GetLogger(ctx).WithError(err).Error("failed to reach internal API")
+	} else if pubNewRoomRes.Error != nil {
 		// treat as non-fatal since the room is already made by this point
 		util.GetLogger(ctx).WithError(pubNewRoomRes.Error).Error("failed to visibility:public")
 	}
 
 	var unpubOldRoomRes api.PerformPublishResponse
 	// remove the old room from the published room list
-	URSAPI.PerformPublish(ctx, &api.PerformPublishRequest{
+	if err := URSAPI.PerformPublish(ctx, &api.PerformPublishRequest{
 		RoomID:     oldRoomID,
 		Visibility: "private",
-	}, &unpubOldRoomRes)
-	if unpubOldRoomRes.Error != nil {
+	}, &unpubOldRoomRes); err != nil {
+		util.GetLogger(ctx).WithError(err).Error("failed to reach internal API")
+	} else if unpubOldRoomRes.Error != nil {
 		// treat as non-fatal since the room is already made by this point
 		util.GetLogger(ctx).WithError(unpubOldRoomRes.Error).Error("failed to visibility:private")
 	}