Support PUTing typing status in clientapi (#550)

* Add handler for typing events

* Add typing events producer

* Setup typing server component

* Send one event per API call

dendrite-config.yaml

@@ -114,6 +114,7 @@ listen:
     public_rooms_api: "localhost:7775"
     federation_sender: "localhost:7776"
     appservice_api: "localhost:7777"
+    typing_server: "localhost:7778"
 
 # The configuration for tracing the dendrite components.
 tracing:

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

@@ -48,12 +48,16 @@ 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
+	selectMembershipInRoomByLocalpartStmt *sql.Stmt
 	selectMembershipsByLocalpartStmt      *sql.Stmt
 }
 
@@ -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/producers/typingserver.go

@@ -0,0 +1,54 @@
+// 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 producers
+
+import (
+	"context"
+	"time"
+
+	"github.com/matrix-org/dendrite/typingserver/api"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+// TypingServerProducer produces events for the typing server to consume
+type TypingServerProducer struct {
+	InputAPI api.TypingServerInputAPI
+}
+
+// NewTypingServerProducer creates a new TypingServerProducer
+func NewTypingServerProducer(inputAPI api.TypingServerInputAPI) *TypingServerProducer {
+	return &TypingServerProducer{
+		InputAPI: inputAPI,
+	}
+}
+
+// Send typing event to typing server
+func (p *TypingServerProducer) Send(
+	ctx context.Context, userID, roomID string,
+	typing bool, timeout int64,
+) error {
+	requestData := api.InputTypingEvent{
+		UserID:         userID,
+		RoomID:         roomID,
+		Typing:         typing,
+		Timeout:        timeout,
+		OriginServerTS: gomatrixserverlib.AsTimestamp(time.Now()),
+	}
+
+	var response api.InputTypingEventResponse
+	err := p.InputAPI.InputTypingEvent(
+		ctx, &api.InputTypingEventRequest{InputTypingEvent: requestData}, &response,
+	)
+
+	return err
+}

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{}{},
+	}
+}

src/github.com/matrix-org/dendrite/cmd/dendrite-client-api-server/main.go

@@ -34,11 +34,12 @@ func main() {
 	keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
 
 	alias, input, query := base.CreateHTTPRoomserverAPIs()
+	typingInputAPI := base.CreateHTTPTypingServerAPIs()
 	cache := transactions.New()
 
 	clientapi.SetupClientAPIComponent(
 		base, deviceDB, accountDB, federation, &keyRing,
-		alias, input, query, cache,
+		alias, input, query, typingInputAPI, cache,
 	)
 
 	base.SetupAndServeHTTP(string(base.Cfg.Listen.ClientAPI))

src/github.com/matrix-org/dendrite/cmd/dendrite-monolith-server/main.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/matrix-org/dendrite/common/keydb"
 	"github.com/matrix-org/dendrite/common/transactions"
+	"github.com/matrix-org/dendrite/typingserver"
 
 	"github.com/matrix-org/dendrite/appservice"
 	"github.com/matrix-org/dendrite/clientapi"
@@ -55,10 +56,11 @@ func main() {
 	keyRing := keydb.CreateKeyRing(federation.Client, keyDB)
 
 	alias, input, query := roomserver.SetupRoomServerComponent(base)
+	typingInputAPI := typingserver.SetupTypingServerComponent(base)
 
 	clientapi.SetupClientAPIComponent(
 		base, deviceDB, accountDB,
-		federation, &keyRing, alias, input, query,
+		federation, &keyRing, alias, input, query, typingInputAPI,
 		transactions.New(),
 	)
 	federationapi.SetupFederationAPIComponent(base, accountDB, deviceDB, federation, &keyRing, alias, input, query)

src/github.com/matrix-org/dendrite/common/basecomponent/base.go

@@ -33,6 +33,7 @@ import (
 	appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
 	"github.com/matrix-org/dendrite/common/config"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
+	typingServerAPI "github.com/matrix-org/dendrite/typingserver/api"
 	"github.com/sirupsen/logrus"
 )
 
@@ -100,6 +101,12 @@ func (b *BaseDendrite) CreateHTTPRoomserverAPIs() (
 	return alias, input, query
 }
 
+// CreateHTTPTypingServerAPIs returns typingInputAPI for hitting the typing
+// server over HTTP
+func (b *BaseDendrite) CreateHTTPTypingServerAPIs() typingServerAPI.TypingServerInputAPI {
+	return typingServerAPI.NewTypingServerInputAPIHTTP(b.Cfg.TypingServerURL(), nil)
+}
+
 // CreateDeviceDB creates a new instance of the device database. Should only be
 // called once per component.
 func (b *BaseDendrite) CreateDeviceDB() *devices.Database {

src/github.com/matrix-org/dendrite/common/config/config.go

@@ -203,6 +203,7 @@ type Dendrite struct {
 		RoomServer       Address `yaml:"room_server"`
 		FederationSender Address `yaml:"federation_sender"`
 		PublicRoomsAPI   Address `yaml:"public_rooms_api"`
+		TypingServer     Address `yaml:"typing_server"`
 	} `yaml:"listen"`
 
 	// The config for tracing the dendrite servers.
@@ -546,6 +547,7 @@ func (config *Dendrite) checkListen(configErrs *configErrors) {
 	checkNotEmpty(configErrs, "listen.federation_api", string(config.Listen.FederationAPI))
 	checkNotEmpty(configErrs, "listen.sync_api", string(config.Listen.SyncAPI))
 	checkNotEmpty(configErrs, "listen.room_server", string(config.Listen.RoomServer))
+	checkNotEmpty(configErrs, "listen.typing_server", string(config.Listen.TypingServer))
 }
 
 // checkLogging verifies the parameters logging.* are valid.
@@ -659,6 +661,15 @@ func (config *Dendrite) RoomServerURL() string {
 	return "http://" + string(config.Listen.RoomServer)
 }
 
+// TypingServerURL returns an HTTP URL for where the typing server is listening.
+func (config *Dendrite) TypingServerURL() string {
+	// Hard code the typing server to talk HTTP for now.
+	// If we support HTTPS we need to think of a practical way to do certificate validation.
+	// People setting up servers shouldn't need to get a certificate valid for the public
+	// internet for an internal API.
+	return "http://" + string(config.Listen.TypingServer)
+}
+
 // SetupTracing configures the opentracing using the supplied configuration.
 func (config *Dendrite) SetupTracing(serviceName string) (closer io.Closer, err error) {
 	return config.Tracing.Jaeger.InitGlobalTracer(

src/github.com/matrix-org/dendrite/common/config/config_test.go

@@ -59,6 +59,7 @@ listen:
   federation_api: "localhost:7772"
   sync_api: "localhost:7773"
   media_api: "localhost:7774"
+  typing_server: "localhost:7778"
 logging:
   - type: "file"
     level: "info"

src/github.com/matrix-org/dendrite/common/test/config.go

@@ -103,6 +103,7 @@ func MakeConfig(configDir, kafkaURI, database, host string, startPort int) (*con
 	cfg.Listen.RoomServer = assignAddress()
 	cfg.Listen.SyncAPI = assignAddress()
 	cfg.Listen.PublicRoomsAPI = assignAddress()
+	cfg.Listen.TypingServer = assignAddress()
 
 	return &cfg, port, nil
 }

src/github.com/matrix-org/dendrite/typingserver/api/input.go

@@ -0,0 +1,83 @@
+// 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 api provides the types that are used to communicate with the typing server.
+package api
+
+import (
+	"context"
+	"net/http"
+
+	commonHTTP "github.com/matrix-org/dendrite/common/http"
+	"github.com/matrix-org/gomatrixserverlib"
+	opentracing "github.com/opentracing/opentracing-go"
+)
+
+// InputTypingEvent is an event for notifying the typing server about typing updates.
+type InputTypingEvent struct {
+	// UserID of the user to update typing status.
+	UserID string `json:"user_id"`
+	// RoomID of the room the user is typing (or has stopped).
+	RoomID string `json:"room_id"`
+	// Typing is true if the user is typing, false if they have stopped.
+	Typing bool `json:"typing"`
+	// Timeout is the interval for which the user should be marked as typing.
+	Timeout int64 `json:"timeout"`
+	// OriginServerTS when the server received the update.
+	OriginServerTS gomatrixserverlib.Timestamp `json:"origin_server_ts"`
+}
+
+// InputTypingEventRequest is a request to TypingServerInputAPI
+type InputTypingEventRequest struct {
+	InputTypingEvent InputTypingEvent `json:"input_typing_event"`
+}
+
+// InputTypingEventResponse is a response to InputTypingEvents
+type InputTypingEventResponse struct{}
+
+// TypingServerInputAPI is used to write events to the typing server.
+type TypingServerInputAPI interface {
+	InputTypingEvent(
+		ctx context.Context,
+		request *InputTypingEventRequest,
+		response *InputTypingEventResponse,
+	) error
+}
+
+// TypingServerInputTypingEventPath is the HTTP path for the InputTypingEvent API.
+const TypingServerInputTypingEventPath = "/api/typingserver/input"
+
+// NewTypingServerInputAPIHTTP creates a TypingServerInputAPI implemented by talking to a HTTP POST API.
+func NewTypingServerInputAPIHTTP(typingServerURL string, httpClient *http.Client) TypingServerInputAPI {
+	if httpClient == nil {
+		httpClient = http.DefaultClient
+	}
+	return &httpTypingServerInputAPI{typingServerURL, httpClient}
+}
+
+type httpTypingServerInputAPI struct {
+	typingServerURL string
+	httpClient      *http.Client
+}
+
+// InputRoomEvents implements TypingServerInputAPI
+func (h *httpTypingServerInputAPI) InputTypingEvent(
+	ctx context.Context,
+	request *InputTypingEventRequest,
+	response *InputTypingEventResponse,
+) error {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "InputTypingEvent")
+	defer span.Finish()
+
+	apiURL := h.typingServerURL + TypingServerInputTypingEventPath
+	return commonHTTP.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
+}

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

@@ -0,0 +1,29 @@
+// 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 typingserver
+
+import (
+	"github.com/matrix-org/dendrite/common/basecomponent"
+	"github.com/matrix-org/dendrite/typingserver/api"
+)
+
+// SetupTypingServerComponent sets up and registers HTTP handlers for the
+// TypingServer component. Returns instances of the various roomserver APIs,
+// allowing other components running in the same process to hit the query the
+// APIs directly instead of having to use HTTP.
+func SetupTypingServerComponent(
+	base *basecomponent.BaseDendrite,
+) api.TypingServerInputAPI {
+	// TODO: implement typing server
+	return base.CreateHTTPTypingServerAPIs()
+}