Implement read receipts (#1528)

* fix conversion from int to string yields a string of one rune, not a string of digits

* Add receipts table to syncapi

* Use StreamingToken as the since value

* Add required method to testEDUProducer

* Make receipt json creation "easier" to read

* Add receipts api to the eduserver

* Add receipts endpoint

* Add eduserver kafka consumer

* Add missing kafka config

* Add passing tests to whitelist

Signed-off-by: Till Faelligen <tfaelligen@gmail.com>

* Fix copy & paste error

* Fix column count error

* Make outbound federation receipts pass

* Make "Inbound federation rejects receipts from wrong remote" pass

* Don't use errors package

* - Add TODO for batching requests
- Rename variable

* Return a better error message

* - Use OutputReceiptEvent instead of InputReceiptEvent as result
- Don't use the errors package for errors
- Defer CloseAndLogIfError to close rows
- Fix Copyright

* Better creation/usage of JoinResponse

* Query all joined rooms instead of just one

* Update gomatrixserverlib

* Add sqlite3 migration

* Add postgres migration

* Ensure required sequence exists before running migrations

* Clarification on comment

* - Fix a bug when creating client receipts
- Use concrete types instead of interface{}

* Remove dead code
Use key for timestamp

* Fix postgres query...

* Remove single purpose struct

* Use key/value directly

* Only apply receipts on initial sync or if edu positions differ,
otherwise we'll be sending the same receipts over and over again.

* Actually update the id, so it is correctly send in syncs

* Set receipt on request to /read_markers

* Fix issue with receipts getting overwritten

* Use fmt.Errorf instead of pkg/errors

* Revert "Add postgres migration"

This reverts commit 722fe5a04628882b787d096942459961db159b06.

* Revert "Add sqlite3 migration"

This reverts commit d113b03f6495a4b8f8bcf158a3d00b510b4240cc.

* Fix selectRoomReceipts query

* Make golangci-lint happy

Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>

3 files changed, 75 insertions, 4 deletions

diff --git a/clientapi/routing/account_data.go b/clientapi/routing/account_data.go
index 48303c97..22e63513 100644
--- a/clientapi/routing/account_data.go
+++ b/clientapi/routing/account_data.go
@@ -23,6 +23,7 @@ import (
 	"github.com/matrix-org/dendrite/clientapi/httputil"
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	"github.com/matrix-org/dendrite/clientapi/producers"
+	eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/userapi/api"
 
@@ -148,7 +149,8 @@ type fullyReadEvent struct {
 
 // SaveReadMarker implements POST /rooms/{roomId}/read_markers
 func SaveReadMarker(
-	req *http.Request, userAPI api.UserInternalAPI, rsAPI roomserverAPI.RoomserverInternalAPI,
+	req *http.Request,
+	userAPI api.UserInternalAPI, rsAPI roomserverAPI.RoomserverInternalAPI, eduAPI eduserverAPI.EDUServerInputAPI,
 	syncProducer *producers.SyncAPIProducer, device *api.Device, roomID string,
 ) util.JSONResponse {
 	// Verify that the user is a member of this room
@@ -192,8 +194,10 @@ func SaveReadMarker(
 		return jsonerror.InternalServerError()
 	}
 
-	// TODO handle the read receipt that may be included in the read marker
-	// See https://matrix.org/docs/spec/client_server/r0.6.0#post-matrix-client-r0-rooms-roomid-read-markers
+	// Handle the read receipt that may be included in the read marker
+	if r.Read != "" {
+		return SetReceipt(req, eduAPI, device, roomID, "m.read", r.Read)
+	}
 
 	return util.JSONResponse{
 		Code: http.StatusOK,
diff --git a/clientapi/routing/receipt.go b/clientapi/routing/receipt.go
new file mode 100644
index 00000000..fe8fe765
--- /dev/null
+++ b/clientapi/routing/receipt.go
@@ -0,0 +1,54 @@
+// Copyright 2020 The Matrix.org Foundation C.I.C.
+//
+// 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 (
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/matrix-org/gomatrixserverlib"
+
+	"github.com/matrix-org/dendrite/eduserver/api"
+
+	userapi "github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/util"
+	"github.com/sirupsen/logrus"
+)
+
+func SetReceipt(req *http.Request, eduAPI api.EDUServerInputAPI, device *userapi.Device, roomId, receiptType, eventId string) util.JSONResponse {
+	timestamp := gomatrixserverlib.AsTimestamp(time.Now())
+	logrus.WithFields(logrus.Fields{
+		"roomId":      roomId,
+		"receiptType": receiptType,
+		"eventId":     eventId,
+		"userId":      device.UserID,
+		"timestamp":   timestamp,
+	}).Debug("Setting receipt")
+
+	// currently only m.read is accepted
+	if receiptType != "m.read" {
+		return util.MessageResponse(400, fmt.Sprintf("receipt type must be m.read not '%s'", receiptType))
+	}
+
+	if err := api.SendReceipt(req.Context(), eduAPI, device.UserID, roomId, eventId, receiptType, timestamp); err != nil {
+		return util.ErrorResponse(err)
+	}
+
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: struct{}{},
+	}
+}
diff --git a/clientapi/routing/routing.go b/clientapi/routing/routing.go
index 369ed571..99d1bd09 100644
--- a/clientapi/routing/routing.go
+++ b/clientapi/routing/routing.go
@@ -705,7 +705,7 @@ func Setup(
 			if err != nil {
 				return util.ErrorResponse(err)
 			}
-			return SaveReadMarker(req, userAPI, rsAPI, syncProducer, device, vars["roomID"])
+			return SaveReadMarker(req, userAPI, rsAPI, eduAPI, syncProducer, device, vars["roomID"])
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
 
@@ -843,4 +843,17 @@ func Setup(
 			return ClaimKeys(req, keyAPI)
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
+	r0mux.Handle("/rooms/{roomId}/receipt/{receiptType}/{eventId}",
+		httputil.MakeAuthAPI(gomatrixserverlib.Join, userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			if r := rateLimits.rateLimit(req); r != nil {
+				return *r
+			}
+			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
+			if err != nil {
+				return util.ErrorResponse(err)
+			}
+
+			return SetReceipt(req, eduAPI, device, vars["roomId"], vars["receiptType"], vars["eventId"])
+		}),
+	).Methods(http.MethodPost, http.MethodOptions)
 }