Warp/server/internal/clientapi/sync.go

package clientapi

import (
	"encoding/json"
	"errors"
	"net/http"
	"strconv"
	"time"

	"github.com/dariuszjaniak/warp/server/internal/httputil"
	"github.com/dariuszjaniak/warp/server/internal/storage"
	"github.com/dariuszjaniak/warp/server/internal/syncapi"
	"github.com/dariuszjaniak/warp/server/internal/userapi"
)

// syncError maps a syncapi error onto the Matrix error the spec answers it
// with.
//
// It defers to roomError for everything that came out of the roomserver,
// which is most of it: a sync read is mostly room reads, and the two have
// to answer "unknown room" the same way whichever endpoint asked.
func syncError(err error) error {
	var matrixErr *httputil.MatrixError
	switch {
	case err == nil:
		return nil
	case errors.As(err, &matrixErr):
		return err

	case errors.Is(err, syncapi.ErrNoSuchFilter):
		return httputil.NotFound("Unknown filter")
	case errors.Is(err, syncapi.ErrInvalidFilter),
		errors.Is(err, syncapi.ErrInvalidParam):
		return httputil.InvalidParam("%s", err.Error())
	case errors.Is(err, syncapi.ErrInvalidToken):
		return httputil.InvalidParam("%s", err.Error())
	case errors.Is(err, syncapi.ErrPresenceDisabled):
		return httputil.Forbidden("Presence is disabled on this server")

	default:
		return roomError(err)
	}
}

// handleSync implements GET /_matrix/client/v3/sync.
//
// It is the one endpoint that is expected to block. The handler holds the
// request open for up to the configured long-poll timeout, and the syncapi
// returns as soon as anything the client cares about happens — so the
// timeout decides how often an idle client re-establishes its request, not
// how quickly it hears about a message.
func (c *ClientAPI) handleSync(w http.ResponseWriter, r *http.Request, auth userapi.Authenticated) error {
	query := r.URL.Query()

	timeout, err := durationParam(query.Get("timeout"))
	if err != nil {
		return err
	}
	fullState, err := boolParam(query.Get("full_state"))
	if err != nil {
		return err
	}

	resp, err := c.sync.Sync(r.Context(), syncapi.SyncRequest{
		User:        auth.UserID,
		Localpart:   auth.Localpart,
		Device:      auth.DeviceID,
		Since:       query.Get("since"),
		Filter:      query.Get("filter"),
		Timeout:     timeout,
		FullState:   fullState,
		SetPresence: query.Get("set_presence"),
	})
	if err != nil {
		// A client that gives up on a long poll cancels the request, which
		// arrives here as a context error. Writing a response to a
		// connection nobody is reading would be the only thing a handler
		// could do about it, so the request is simply dropped.
		if ctxErr := r.Context().Err(); ctxErr != nil && errors.Is(err, ctxErr) {
			return nil //nolint:nilerr // the client is gone; there is nobody to answer
		}
		return syncError(err)
	}

	// A sync response is never cached: it is a position in a stream, and a
	// cached one would hand a client the same batch twice.
	httputil.NoCache(w)
	httputil.WriteJSON(w, http.StatusOK, resp)
	return nil
}

// handleMessages implements GET /_matrix/client/v3/rooms/{roomID}/messages.
func (c *ClientAPI) handleMessages(w http.ResponseWriter, r *http.Request, auth userapi.Authenticated) error {
	roomID, err := pathRoomID(r)
	if err != nil {
		return err
	}
	query := r.URL.Query()

	direction := storage.DirectionBackward
	switch query.Get("dir") {
	case "f":
		direction = storage.DirectionForward
	case "b", "":
		// The spec makes `dir` required; a client that leaves it out means
		// "scroll up", which is what every one of them is doing.
	default:
		return httputil.InvalidParam("dir must be \"b\" or \"f\"")
	}

	limit, err := intParam(query.Get("limit"))
	if err != nil {
		return err
	}

	resp, err := c.sync.Messages(r.Context(), syncapi.MessagesRequest{
		RoomID:    roomID,
		User:      auth.UserID,
		Localpart: auth.Localpart,
		From:      query.Get("from"),
		To:        query.Get("to"),
		Direction: direction,
		Limit:     limit,
		Filter:    query.Get("filter"),
	})
	if err != nil {
		return syncError(err)
	}
	httputil.WriteJSON(w, http.StatusOK, resp)
	return nil
}

// handleContext implements
// GET /_matrix/client/v3/rooms/{roomID}/context/{eventID}.
func (c *ClientAPI) handleContext(w http.ResponseWriter, r *http.Request, auth userapi.Authenticated) error {
	roomID, err := pathRoomID(r)
	if err != nil {
		return err
	}
	eventID := r.PathValue("eventID")
	if eventID == "" {
		return httputil.MissingParam("An event ID is required")
	}

	limit, err := intParam(r.URL.Query().Get("limit"))
	if err != nil {
		return err
	}

	resp, err := c.sync.Context(r.Context(), syncapi.ContextRequest{
		RoomID:    roomID,
		EventID:   eventID,
		User:      auth.UserID,
		Localpart: auth.Localpart,
		Limit:     limit,
		Filter:    r.URL.Query().Get("filter"),
	})
	if err != nil {
		return syncError(err)
	}
	httputil.WriteJSON(w, http.StatusOK, resp)
	return nil
}

// filterResponse is what POST /user/{userID}/filter returns.
type filterResponse struct {
	FilterID string `json:"filter_id"`
}

// handlePostFilter implements POST /_matrix/client/v3/user/{userID}/filter.
func (c *ClientAPI) handlePostFilter(w http.ResponseWriter, r *http.Request, auth userapi.Authenticated) error {
	localpart, err := c.requireOwnAccountData(auth, r.PathValue("userID"))
	if err != nil {
		return err
	}

	// The filter is stored and returned byte for byte, so it is read as
	// raw JSON rather than decoded and re-encoded: a client that compares
	// what it gets back with what it sent must see the same thing.
	var body json.RawMessage
	if err := httputil.DecodeJSON(w, r, &body); err != nil {
		return err
	}

	filterID, err := c.sync.SaveFilter(r.Context(), localpart, body)
	if err != nil {
		return syncError(err)
	}
	httputil.WriteJSON(w, http.StatusOK, filterResponse{FilterID: filterID})
	return nil
}

// handleGetFilter implements
// GET /_matrix/client/v3/user/{userID}/filter/{filterID}.
func (c *ClientAPI) handleGetFilter(w http.ResponseWriter, r *http.Request, auth userapi.Authenticated) error {
	localpart, err := c.requireOwnAccountData(auth, r.PathValue("userID"))
	if err != nil {
		return err
	}
	filterID := r.PathValue("filterID")
	if filterID == "" {
		return httputil.MissingParam("A filter ID is required")
	}

	stored, err := c.sync.StoredFilter(r.Context(), localpart, filterID)
	if err != nil {
		return syncError(err)
	}
	httputil.WriteRawJSON(w, http.StatusOK, stored)
	return nil
}

// ParseSyncPosition reads a sync token from a query parameter and returns
// the event-stream position in it, or 0 for an absent one.
//
// It lives here rather than in the roomserver because the token format is
// the syncapi's, and /rooms/{id}/members?at= is the one room endpoint that
// takes one.
func ParseSyncPosition(raw string) (int64, error) {
	if raw == "" {
		return 0, nil
	}
	token, err := syncapi.ParseToken(raw)
	if err != nil {
		return 0, httputil.InvalidParam("%v", err)
	}
	return token.PDU, nil
}

// durationParam reads a millisecond duration from a query parameter.
func durationParam(raw string) (time.Duration, error) {
	if raw == "" {
		return 0, nil
	}
	ms, err := strconv.ParseInt(raw, 10, 64)
	if err != nil || ms < 0 {
		return 0, httputil.InvalidParam("timeout must be a non-negative number of milliseconds")
	}
	return time.Duration(ms) * time.Millisecond, nil
}

// intParam reads a non-negative integer from a query parameter.
func intParam(raw string) (int, error) {
	if raw == "" {
		return 0, nil
	}
	value, err := strconv.Atoi(raw)
	if err != nil || value < 0 {
		return 0, httputil.InvalidParam("limit must be a non-negative integer")
	}
	return value, nil
}

// boolParam reads a boolean from a query parameter.
func boolParam(raw string) (bool, error) {
	if raw == "" {
		return false, nil
	}
	value, err := strconv.ParseBool(raw)
	if err != nil {
		return false, httputil.InvalidParam("%q is not a boolean", raw)
	}
	return value, nil
}