Add handler for typing events

src/github.com/matrix-org/dendrite/clientapi/auth/storage/accounts/membership_table.go

@@ -48,13 +48,17 @@ const insertMembershipSQL = `
 const selectMembershipsByLocalpartSQL = "" +
 	"SELECT room_id, event_id FROM account_memberships WHERE localpart = $1"
 
+const selectMembershipInRoomByLocalpartSQL = "" +
+	"SELECT event_id FROM account_memberships WHERE localpart = $1 AND room_id = $2"
+
 const deleteMembershipsByEventIDsSQL = "" +
 	"DELETE FROM account_memberships WHERE event_id = ANY($1)"
 
 type membershipStatements struct {
-	deleteMembershipsByEventIDsStmt  *sql.Stmt
-	insertMembershipStmt             *sql.Stmt
-	selectMembershipsByLocalpartStmt *sql.Stmt
+	deleteMembershipsByEventIDsStmt       *sql.Stmt
+	insertMembershipStmt                  *sql.Stmt
+	selectMembershipInRoomByLocalpartStmt *sql.Stmt
+	selectMembershipsByLocalpartStmt      *sql.Stmt
 }
 
 func (s *membershipStatements) prepare(db *sql.DB) (err error) {
@@ -68,6 +72,9 @@ func (s *membershipStatements) prepare(db *sql.DB) (err error) {
 	if s.insertMembershipStmt, err = db.Prepare(insertMembershipSQL); err != nil {
 		return
 	}
+	if s.selectMembershipInRoomByLocalpartStmt, err = db.Prepare(selectMembershipInRoomByLocalpartSQL); err != nil {
+		return
+	}
 	if s.selectMembershipsByLocalpartStmt, err = db.Prepare(selectMembershipsByLocalpartSQL); err != nil {
 		return
 	}
@@ -90,6 +97,16 @@ func (s *membershipStatements) deleteMembershipsByEventIDs(
 	return
 }
 
+func (s *membershipStatements) selectMembershipInRoomByLocalpart(
+	ctx context.Context, localpart, roomID string,
+) (authtypes.Membership, error) {
+	membership := authtypes.Membership{Localpart: localpart, RoomID: roomID}
+	stmt := s.selectMembershipInRoomByLocalpartStmt
+	err := stmt.QueryRowContext(ctx, localpart, roomID).Scan(&membership.EventID)
+
+	return membership, err
+}
+
 func (s *membershipStatements) selectMembershipsByLocalpart(
 	ctx context.Context, localpart string,
 ) (memberships []authtypes.Membership, err error) {

src/github.com/matrix-org/dendrite/clientapi/auth/storage/accounts/storage.go

@@ -185,6 +185,16 @@ func (d *Database) UpdateMemberships(
 	})
 }
 
+// GetMembershipInRoomByLocalpart returns the membership for an user
+// matching the given localpart if he is a member of the room matching roomID,
+// if not sql.ErrNoRows is returned.
+// If there was an issue during the retrieval, returns the SQL error
+func (d *Database) GetMembershipInRoomByLocalpart(
+	ctx context.Context, localpart, roomID string,
+) (authtypes.Membership, error) {
+	return d.memberships.selectMembershipInRoomByLocalpart(ctx, localpart, roomID)
+}
+
 // GetMembershipsByLocalpart returns an array containing the memberships for all
 // the rooms a user matching a given localpart is a member of
 // If no membership match the given localpart, returns an empty array

src/github.com/matrix-org/dendrite/clientapi/clientapi.go

@@ -23,6 +23,7 @@ import (
 	"github.com/matrix-org/dendrite/common/basecomponent"
 	"github.com/matrix-org/dendrite/common/transactions"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
+	typingServerAPI "github.com/matrix-org/dendrite/typingserver/api"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/sirupsen/logrus"
 )
@@ -38,9 +39,11 @@ func SetupClientAPIComponent(
 	aliasAPI roomserverAPI.RoomserverAliasAPI,
 	inputAPI roomserverAPI.RoomserverInputAPI,
 	queryAPI roomserverAPI.RoomserverQueryAPI,
+	typingInputAPI typingServerAPI.TypingServerInputAPI,
 	transactionsCache *transactions.Cache,
 ) {
 	roomserverProducer := producers.NewRoomserverProducer(inputAPI)
+	typingProducer := producers.NewTypingServerProducer(typingInputAPI)
 
 	userUpdateProducer := &producers.UserUpdateProducer{
 		Producer: base.KafkaProducer,
@@ -62,6 +65,6 @@ func SetupClientAPIComponent(
 	routing.Setup(
 		base.APIMux, *base.Cfg, roomserverProducer, queryAPI, aliasAPI,
 		accountsDB, deviceDB, federation, *keyRing, userUpdateProducer,
-		syncProducer, transactionsCache,
+		syncProducer, typingProducer, transactionsCache,
 	)
 }

src/github.com/matrix-org/dendrite/clientapi/routing/routing.go

@@ -50,6 +50,7 @@ func Setup(
 	keyRing gomatrixserverlib.KeyRing,
 	userUpdateProducer *producers.UserUpdateProducer,
 	syncProducer *producers.SyncAPIProducer,
+	typingProducer *producers.TypingServerProducer,
 	transactionsCache *transactions.Cache,
 ) {
 
@@ -173,6 +174,13 @@ func Setup(
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
 
+	r0mux.Handle("/rooms/{roomID}/typing/{userID}",
+		common.MakeAuthAPI("rooms_typing", authData, func(req *http.Request, device *authtypes.Device) util.JSONResponse {
+			vars := mux.Vars(req)
+			return SendTyping(req, device, vars["roomID"], vars["userID"], accountDB, typingProducer)
+		}),
+	).Methods(http.MethodPut, http.MethodOptions)
+
 	// Stub endpoints required by Riot
 
 	r0mux.Handle("/login",
@@ -351,13 +359,6 @@ func Setup(
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
 
-	r0mux.Handle("/rooms/{roomID}/typing/{userID}",
-		common.MakeExternalAPI("rooms_typing", func(req *http.Request) util.JSONResponse {
-			// TODO: handling typing
-			return util.JSONResponse{Code: http.StatusOK, JSON: struct{}{}}
-		}),
-	).Methods(http.MethodPut, http.MethodOptions)
-
 	r0mux.Handle("/devices",
 		common.MakeAuthAPI("get_devices", authData, func(req *http.Request, device *authtypes.Device) util.JSONResponse {
 			return GetDevicesByLocalpart(req, deviceDB, device)

src/github.com/matrix-org/dendrite/clientapi/routing/sendtyping.go

@@ -0,0 +1,80 @@
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing
+
+import (
+	"database/sql"
+	"net/http"
+
+	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
+	"github.com/matrix-org/dendrite/clientapi/auth/storage/accounts"
+	"github.com/matrix-org/dendrite/clientapi/httputil"
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/clientapi/producers"
+	"github.com/matrix-org/dendrite/clientapi/userutil"
+	"github.com/matrix-org/util"
+)
+
+type typingContentJSON struct {
+	Typing  bool  `json:"typing"`
+	Timeout int64 `json:"timeout"`
+}
+
+// SendTyping handles PUT /rooms/{roomID}/typing/{userID}
+// sends the typing events to client API typingProducer
+func SendTyping(
+	req *http.Request, device *authtypes.Device, roomID string,
+	userID string, accountDB *accounts.Database,
+	typingProducer *producers.TypingServerProducer,
+) util.JSONResponse {
+	if device.UserID != userID {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("Cannot set another user's typing state"),
+		}
+	}
+
+	localpart, err := userutil.ParseUsernameParam(userID, nil)
+	if err != nil {
+		return httputil.LogThenError(req, err)
+	}
+
+	// Verify that the user is a member of this room
+	_, err = accountDB.GetMembershipInRoomByLocalpart(req.Context(), localpart, roomID)
+	if err == sql.ErrNoRows {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("User not in this room"),
+		}
+	} else if err != nil {
+		return httputil.LogThenError(req, err)
+	}
+
+	// parse the incoming http request
+	var r typingContentJSON
+	resErr := httputil.UnmarshalJSONRequest(req, &r)
+	if resErr != nil {
+		return *resErr
+	}
+
+	if err = typingProducer.Send(
+		req.Context(), userID, roomID, r.Typing, r.Timeout,
+	); err != nil {
+		return httputil.LogThenError(req, err)
+	}
+
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: struct{}{},
+	}
+}