NOTSPEC: Make ?from= optional in /messages (#1647)

3 files changed, 16 insertions, 6 deletions

diff --git a/syncapi/routing/messages.go b/syncapi/routing/messages.go
index 865203a9..64e0fa68 100644
--- a/syncapi/routing/messages.go
+++ b/syncapi/routing/messages.go
@@ -25,6 +25,7 @@ import (
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/syncapi/storage"
+	"github.com/matrix-org/dendrite/syncapi/sync"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -65,6 +66,7 @@ func OnIncomingMessagesRequest(
 	federation *gomatrixserverlib.FederationClient,
 	rsAPI api.RoomserverInternalAPI,
 	cfg *config.SyncAPI,
+	srp *sync.RequestPool,
 ) util.JSONResponse {
 	var err error
 
@@ -84,9 +86,17 @@ func OnIncomingMessagesRequest(
 	// Extract parameters from the request's URL.
 	// Pagination tokens.
 	var fromStream *types.StreamingToken
-	from, err := types.NewTopologyTokenFromString(req.URL.Query().Get("from"))
+	fromQuery := req.URL.Query().Get("from")
+	if fromQuery == "" {
+		// NOTSPEC: We will pretend they used the latest sync token if no ?from= was provided.
+		// We do this to allow clients to get messages without having to call `/sync` e.g Cerulean
+		currPos := srp.Notifier.CurrentPosition()
+		fromQuery = currPos.String()
+	}
+
+	from, err := types.NewTopologyTokenFromString(fromQuery)
 	if err != nil {
-		fs, err2 := types.NewStreamTokenFromString(req.URL.Query().Get("from"))
+		fs, err2 := types.NewStreamTokenFromString(fromQuery)
 		fromStream = &fs
 		if err2 != nil {
 			return util.JSONResponse{
diff --git a/syncapi/routing/routing.go b/syncapi/routing/routing.go
index 20152b48..e2ff2739 100644
--- a/syncapi/routing/routing.go
+++ b/syncapi/routing/routing.go
@@ -51,7 +51,7 @@ func Setup(
 		if err != nil {
 			return util.ErrorResponse(err)
 		}
-		return OnIncomingMessagesRequest(req, syncDB, vars["roomID"], device, federation, rsAPI, cfg)
+		return OnIncomingMessagesRequest(req, syncDB, vars["roomID"], device, federation, rsAPI, cfg, srp)
 	})).Methods(http.MethodGet, http.MethodOptions)
 
 	r0mux.Handle("/user/{userId}/filter",
diff --git a/syncapi/sync/requestpool.go b/syncapi/sync/requestpool.go
index 32dfb2d6..3a31edd0 100644
--- a/syncapi/sync/requestpool.go
+++ b/syncapi/sync/requestpool.go
@@ -44,7 +44,7 @@ type RequestPool struct {
 	db       storage.Database
 	cfg      *config.SyncAPI
 	userAPI  userapi.UserInternalAPI
-	notifier *Notifier
+	Notifier *Notifier
 	keyAPI   keyapi.KeyInternalAPI
 	rsAPI    roomserverAPI.RoomserverInternalAPI
 	lastseen sync.Map
@@ -152,7 +152,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 
 	rp.updateLastSeen(req, device)
 
-	currPos := rp.notifier.CurrentPosition()
+	currPos := rp.Notifier.CurrentPosition()
 
 	if rp.shouldReturnImmediately(syncReq) {
 		syncData, err = rp.currentSyncForUser(*syncReq, currPos)
@@ -176,7 +176,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 	timer := time.NewTimer(syncReq.timeout) // case of timeout=0 is handled above
 	defer timer.Stop()
 
-	userStreamListener := rp.notifier.GetListener(*syncReq)
+	userStreamListener := rp.Notifier.GetListener(*syncReq)
 	defer userStreamListener.Close()
 
 	// We need the loop in case userStreamListener wakes up even if there isn't