Warp/server/internal/syncapi/sync.go

package syncapi

import (
	"context"
	"encoding/json"
	"errors"
	"slices"
	"time"

	"go.opentelemetry.io/otel/attribute"

	"github.com/dariuszjaniak/warp/server/internal/matrix/event"
	"github.com/dariuszjaniak/warp/server/internal/roomserver"
	"github.com/dariuszjaniak/warp/server/internal/storage"
	"github.com/dariuszjaniak/warp/server/internal/tracing"
)

// GET /sync: the endpoint a client spends its life in.
//
// The shape of the work is: fix the top of every stream, so that the whole
// response describes one moment; work out which rooms the user's view of
// has changed; read each of them; read the streams that are not about a
// room; and hand back a token naming where it all got to. If nothing had
// changed and the client asked to wait, block on the notifier and do it
// again.

// SyncRequest is one call to /sync.
type SyncRequest struct {
	// User is the full user ID syncing; Localpart and Device identify the
	// session, because a to-device message is addressed to a device rather
	// than to a user.
	User      string
	Localpart string
	Device    string

	// Since is the ?since= token. Empty is an initial sync.
	Since string

	// Filter is the ?filter= parameter: a filter ID, an inline filter as
	// JSON, or empty.
	Filter string

	// Timeout is how long to wait for something to happen before
	// answering with nothing. Zero returns immediately.
	Timeout time.Duration

	// FullState asks for every room's whole state even on an incremental
	// sync.
	FullState bool

	// SetPresence is the ?set_presence= parameter: a client says it is
	// online by syncing, and says it has gone away by syncing with
	// set_presence=unavailable.
	SetPresence string
}

// Response is the body of a /sync.
type Response struct {
	NextBatch string `json:"next_batch"`

	Rooms       Rooms           `json:"rooms"`
	Presence    PresenceList    `json:"presence"`
	AccountData AccountDataList `json:"account_data"`
	ToDevice    ToDeviceList    `json:"to_device"`

	// DeviceLists reports whose device list changed, which a client needs
	// in order to re-fetch their keys, and whose it can stop tracking.
	// Both lists are present even when empty: a client that finds no
	// device_lists treats the response as malformed rather than as
	// "nothing changed".
	DeviceLists DeviceLists `json:"device_lists"`

	// DeviceOneTimeKeysCount is how many one-time keys the syncing device
	// has left on the server, by algorithm, which is how it knows to
	// upload more.
	DeviceOneTimeKeysCount map[string]int `json:"device_one_time_keys_count,omitempty"`

	// DeviceUnusedFallbackKeyTypes lists the algorithms the device has an
	// unused fallback key for. It is a pointer so that an empty list goes
	// out as [] rather than not at all: a client reads the field's
	// absence as a server without fallback keys, and never uploads one.
	DeviceUnusedFallbackKeyTypes *[]string `json:"device_unused_fallback_key_types,omitempty"`
}

// Rooms is the room half of a sync, split by the user's membership.
type Rooms struct {
	Join   map[string]JoinedRoom  `json:"join"`
	Invite map[string]InvitedRoom `json:"invite"`
	Leave  map[string]LeftRoom    `json:"leave"`
	Knock  map[string]KnockedRoom `json:"knock"`
}

// JoinedRoom is a room the user is in.
type JoinedRoom struct {
	// Summary is left out for a room that has a name of its own: it exists
	// so that a client can name a room that does not.
	Summary *roomserver.RoomSyncSummary `json:"summary,omitempty"`

	State       EventList          `json:"state"`
	Timeline    Timeline           `json:"timeline"`
	Ephemeral   EphemeralEventList `json:"ephemeral"`
	AccountData AccountDataList    `json:"account_data"`

	// UnreadNotifications is the user's unread notification counts in the
	// room. It is a pointer so that a server without a push server leaves
	// it out rather than claiming zero: an absent count says nothing,
	// where a zero says "you have read everything".
	UnreadNotifications *UnreadNotifications `json:"unread_notifications,omitempty"`

	// UnreadThreadNotifications is the counts of each thread with
	// something unread, keyed by the thread root's event ID, for a filter
	// that asks for them; UnreadNotifications then counts the main
	// timeline alone.
	UnreadThreadNotifications map[string]UnreadNotifications `json:"unread_thread_notifications,omitempty"`
}

// InvitedRoom is a room the user has been invited to. It carries no
// timeline at all: the invitee is not a member, and the stripped state is
// everything they are entitled to.
type InvitedRoom struct {
	InviteState StrippedStateList `json:"invite_state"`
}

// LeftRoom is a room the user has left, reported so that a client can
// notice the leave and close the room.
type LeftRoom struct {
	State       EventList       `json:"state"`
	Timeline    Timeline        `json:"timeline"`
	AccountData AccountDataList `json:"account_data"`
}

// KnockedRoom is a room the user has knocked on. Knocking needs no stream
// of its own; the stripped state is what tells the client which room it is
// waiting on.
type KnockedRoom struct {
	KnockState StrippedStateList `json:"knock_state"`
}

// Timeline is a room's events in this window.
type Timeline struct {
	Events []roomserver.ClientEvent `json:"events"`

	// Limited says there is a gap between PrevBatch and what the client
	// already had: it has to page backwards through /messages to close it.
	Limited bool `json:"limited"`

	// PrevBatch is the token to page backwards from.
	PrevBatch string `json:"prev_batch,omitempty"`
}

// The list wrappers. The Client-Server API wraps every collection of
// events in an object with an `events` key, and each of these carries a
// different event shape — which is the whole reason they are separate types
// rather than one.
type (
	// EventList holds room events: a timeline or a piece of room state.
	EventList struct {
		Events []roomserver.ClientEvent `json:"events"`
	}

	// AccountDataList holds account data.
	AccountDataList struct {
		Events []AccountDataEvent `json:"events"`
	}

	// EphemeralEventList holds typing and receipt EDUs, which have a type
	// and a content and nothing else.
	EphemeralEventList struct {
		Events []EphemeralEvent `json:"events"`
	}

	// PresenceList holds m.presence EDUs, which also have a sender.
	PresenceList struct {
		Events []presenceEvent `json:"events"`
	}

	// ToDeviceList holds to-device messages.
	ToDeviceList struct {
		Events []toDeviceEvent `json:"events"`
	}

	// StrippedStateList holds the stripped state of an invite or a knock.
	StrippedStateList struct {
		Events []roomserver.StrippedEvent `json:"events"`
	}
)

// EphemeralEvent is an EDU of a room: a typing notification or a set of
// receipts. It is not an event — no ID, no sender, no place in the room's
// history — which is why it has a type of its own.
type EphemeralEvent struct {
	Type    string          `json:"type"`
	Content json.RawMessage `json:"content"`
}

// AccountDataEvent is one account data blob as a sync reports it.
//
// It has a type and a content and nothing else — no sender, no ID, no room
// — because account data is not an event: it is a key-value store the
// Client-Server API happens to report in an event list, which is one of its
// older seams. Reporting it as a room event with empty fields would be
// worse than a type of its own: a client would see an event with no sender
// and no timestamp and have to decide what that meant.
type AccountDataEvent struct {
	Type    string          `json:"type"`
	Content json.RawMessage `json:"content"`
}

// DeviceLists is whose end-to-end keys a client should re-fetch.
type DeviceLists struct {
	Changed []string `json:"changed"`
	Left    []string `json:"left"`
}

// Sync answers one /sync, blocking for up to req.Timeout when there is
// nothing to report.
func (s *API) Sync(ctx context.Context, req SyncRequest) (Response, error) {
	since, err := ParseToken(req.Since)
	if err != nil {
		return Response{}, errInvalidTokenf("%v", err)
	}
	filter, err := s.filterFor(ctx, req.Localpart, req.Filter)
	if err != nil {
		return Response{}, err
	}
	if err := s.applySetPresence(ctx, req); err != nil {
		return Response{}, err
	}

	// A device acknowledges the to-device batch it was last sent by coming
	// back with the token that carried it, so this is where the previous
	// batch is dropped.
	s.ackToDevice(ctx, req.Localpart, req.Device, since.ToDevice)

	timeout := min(req.Timeout, s.longPollTimeout)
	deadline := s.now().Add(timeout)

	for {
		resp, changed, listener, err := s.attempt(ctx, req, filter, since)
		switch {
		case err != nil:
			listener.Close()
			return Response{}, err
		case changed || timeout <= 0:
			listener.Close()
			return resp, nil
		}

		wait := time.Until(deadline)
		if wait <= 0 {
			listener.Close()
			return resp, nil
		}

		woken, err := waitFor(ctx, listener, wait)
		listener.Close()
		switch {
		case err != nil:
			return Response{}, err
		case !woken:
			// Nothing happened in the time the client was willing to wait.
			// The empty response and its token are the right answer: the
			// client asks again.
			return resp, nil
		}
	}
}

// attempt builds one candidate response, with a listener registered before
// anything was read.
//
// Registering first is what closes the race: anything that lands between
// the registration and the read leaves the channel already signalled, so a
// wait on it returns at once and the caller reads again rather than
// blocking on a change it has just missed. The listener is returned rather
// than closed here because the caller is the one that knows whether it is
// about to wait on it.
func (s *API) attempt(ctx context.Context, req SyncRequest, filter Filter, since Token) (_ Response, _ bool, _ *Listener, err error) {
	ctx, span := tracing.Start(ctx, "syncapi.sync", attribute.Bool("matrix.initial", since.IsInitial()))
	defer func() { tracing.End(span, err) }()

	rooms, err := s.rooms.UserMemberships(ctx, req.User, nil)
	if err != nil {
		return Response{}, false, nil, err
	}
	listener := s.notifier.Listen(req.User, roomIDsOf(rooms))

	resp, changed, err := s.build(ctx, req, filter, since, rooms)
	if err != nil {
		return Response{}, false, listener, err
	}
	return resp, changed, listener, nil
}

// waitFor blocks until the listener is woken, the wait runs out, or the
// request is cancelled. It reports whether it was woken.
func waitFor(ctx context.Context, listener *Listener, wait time.Duration) (bool, error) {
	timer := time.NewTimer(wait)
	defer timer.Stop()

	select {
	case <-listener.Wait():
		return true, nil
	case <-timer.C:
		return false, nil
	case <-ctx.Done():
		return false, ctx.Err()
	}
}

// build assembles one response, reporting whether it found anything — which
// is what tells Sync whether to block.
func (s *API) build(ctx context.Context, req SyncRequest, filter Filter, since Token, rooms []roomserver.MembershipAt) (Response, bool, error) {
	// The top of every stream, read once, so that the whole response
	// describes one moment and the token it returns names that moment.
	// Read per section instead, a stream could move between two sections
	// and the token would skip whatever landed in the gap.
	positions, err := s.store.StreamPositions(ctx)
	if err != nil {
		return Response{}, false, err
	}
	to := tokenAt(positions)
	to.Typing = s.typing.position()

	resp := Response{
		Rooms: Rooms{
			Join:   map[string]JoinedRoom{},
			Invite: map[string]InvitedRoom{},
			Leave:  map[string]LeftRoom{},
			Knock:  map[string]KnockedRoom{},
		},
		Presence:    PresenceList{Events: []presenceEvent{}},
		AccountData: AccountDataList{Events: []AccountDataEvent{}},
		ToDevice:    ToDeviceList{Events: []toDeviceEvent{}},
		DeviceLists: DeviceLists{Changed: []string{}, Left: []string{}},
	}

	// The next token. Every stream advances to the top of the window: what
	// was in it has either been reported or is something this user is not
	// concerned by, and leaving a position behind would mean re-reading the
	// same window on every sync forever. The exception is to-device, whose
	// queue is drained a batch at a time and whose token therefore names
	// the last message delivered.
	next := to

	if err := s.buildRooms(ctx, req, filter, since, to, rooms, &resp); err != nil {
		return Response{}, false, err
	}
	if err := s.unreadNotifications(ctx, req.User, filter.UnreadThreadNotifications(), &resp); err != nil {
		return Response{}, false, err
	}

	accountData, err := s.globalAccountData(ctx, req, since.AccountData, to.AccountData,
		req.Since == "" || req.FullState, filter)
	if err != nil {
		return Response{}, false, err
	}
	resp.AccountData.Events = accountData

	toDevice, toDevicePos, err := s.toDeviceMessages(ctx, req.Localpart, req.Device, since.ToDevice, to.ToDevice, 0)
	if err != nil {
		return Response{}, false, err
	}
	resp.ToDevice.Events = toDevice
	next.ToDevice = toDevicePos

	// A room joined inside this window brings its members' presence with
	// it: their last presence change is older than the window, so a
	// window-bounded read would not find them and the client would show
	// everybody in its new room as offline until they next moved.
	presenceFrom := since.Presence
	if joinedInWindow(rooms, since.PDU, to.PDU) {
		presenceFrom = 0
	}
	presence, err := s.presenceEvents(ctx, req.User, joinedRoomIDs(rooms),
		presenceFrom, to.Presence, filter.PresenceFilter())
	if err != nil {
		return Response{}, false, err
	}
	resp.Presence.Events = presence

	if resp.DeviceLists, err = s.deviceLists(ctx, req.User, since, to, rooms); err != nil {
		return Response{}, false, err
	}
	if err := s.deviceKeyCounts(ctx, req, &resp); err != nil {
		return Response{}, false, err
	}

	resp.NextBatch = next.String()
	return resp, isInteresting(resp, req.Since == ""), nil
}

// isInteresting reports whether a response is worth returning rather than
// blocking on.
//
// Comparing the tokens would not do. A stream can move without this user
// being concerned by it — somebody else's receipt in a room they are not in
// — and a sync that returned on every such move would turn a long poll back
// into a busy loop. So the test is whether the response has anything in it,
// with an initial sync as the exception: a client that has just logged in
// needs an answer whatever is in it.
//
// Initial means the client sent no ?since=, not that the token it sent is
// all zeroes. On a server whose streams have not moved yet — no events, no
// receipts, nothing — "s0" is a client's every token, and a sync that took
// that for a first one would never wait.
func isInteresting(resp Response, initial bool) bool {
	if initial {
		return true
	}
	return len(resp.Rooms.Join) > 0 ||
		len(resp.Rooms.Invite) > 0 ||
		len(resp.Rooms.Leave) > 0 ||
		len(resp.Rooms.Knock) > 0 ||
		len(resp.AccountData.Events) > 0 ||
		len(resp.ToDevice.Events) > 0 ||
		len(resp.Presence.Events) > 0 ||
		len(resp.DeviceLists.Changed) > 0 ||
		len(resp.DeviceLists.Left) > 0
}

// buildRooms fills the four room sections.
func (s *API) buildRooms(ctx context.Context, req SyncRequest, filter Filter, since, to Token, rooms []roomserver.MembershipAt, resp *Response) error {
	interesting, err := s.changedRooms(ctx, req, filter, since, to, rooms)
	if err != nil {
		return err
	}

	for _, room := range rooms {
		if _, ok := interesting[room.RoomID]; !ok {
			continue
		}
		if !filter.AllowsRoom(room.RoomID) {
			continue
		}

		// A room whose membership moved inside this window is read in full
		// rather than from the client's position: the ephemeral streams
		// and the account data of a room just joined are all older than
		// the window, and a window-bounded read of them would come back
		// empty.
		window := roomWindow{
			since: since,
			to:    to,
			full:  since.IsInitial() || req.FullState || room.Position > since.PDU,
		}
		if room.Membership == event.MembershipJoin && !since.IsInitial() && room.Position > since.PDU {
			joined, err := s.isNewJoin(ctx, req, room)
			if err != nil {
				return err
			}
			window.joined = joined
		}

		switch room.Membership {
		case event.MembershipJoin:
			joined, err := s.joinedRoom(ctx, req, filter, window, room)
			if err != nil {
				return err
			}
			resp.Rooms.Join[room.RoomID] = joined

		case event.MembershipInvite:
			state, err := s.rooms.StrippedState(ctx, room.RoomID, []string{req.User})
			if err != nil {
				return err
			}
			resp.Rooms.Invite[room.RoomID] = InvitedRoom{
				InviteState: StrippedStateList{Events: state},
			}

		case event.MembershipKnock:
			state, err := s.rooms.StrippedState(ctx, room.RoomID, []string{req.User})
			if err != nil {
				return err
			}
			resp.Rooms.Knock[room.RoomID] = KnockedRoom{
				KnockState: StrippedStateList{Events: state},
			}

		case event.MembershipLeave, event.MembershipBan:
			left, err := s.leftRoom(ctx, req, filter, window, room)
			if err != nil {
				return err
			}
			resp.Rooms.Leave[room.RoomID] = left
		}
	}
	return nil
}

// roomWindow is the span one room is read over.
type roomWindow struct {
	since Token
	to    Token

	// full reads the room's streams from the beginning rather than from
	// the client's position: an initial sync, ?full_state=true, or a room
	// the user has only just joined.
	full bool

	// joined is a room the user joined inside the window. To their client
	// the room is new, so it is read as an initial sync reads it: the
	// recent history they may see and the state before it, not only the
	// events since the client's position, which would be their own join.
	joined bool
}

// isNewJoin reports whether the user's membership event inside the window
// is a join into the room, rather than a profile change of a member who
// was already in it.
func (s *API) isNewJoin(ctx context.Context, req SyncRequest, room roomserver.MembershipAt) (bool, error) {
	before, err := s.rooms.MembershipBefore(ctx, room.RoomID, req.User, room.EventID)
	if err != nil {
		return false, err
	}
	return before != event.MembershipJoin, nil
}

// from is the position a stream is read from, which is the start of it for
// a full read.
func (w roomWindow) from(pos int64) int64 {
	if w.full {
		return 0
	}
	return pos
}

// timelineSince is the position a room's timeline and state are read from:
// the client's, or the start of the room for one joined in the window.
func (w roomWindow) timelineSince() int64 {
	if w.joined {
		return 0
	}
	return w.since.PDU
}

// changedRooms is the set of rooms this sync has to look at.
//
// On an initial sync that is all of them. On an incremental one it is the
// rooms with a new event, plus the rooms whose ephemeral or account data
// changed — neither of which has an event to be found by — plus the rooms
// the user's own membership changed in, which is how a leave reaches a
// client even when it was the last thing to happen in the room.
func (s *API) changedRooms(ctx context.Context, req SyncRequest, filter Filter, since, to Token, rooms []roomserver.MembershipAt) (map[string]struct{}, error) {
	out := map[string]struct{}{}

	if since.IsInitial() || req.FullState {
		for _, room := range rooms {
			// A room left before this sync began is not reported unless
			// the filter asks for it: a client logging in does not want
			// every room it has ever left.
			if isDeparted(room.Membership) && !filter.IncludeLeave() {
				continue
			}
			out[room.RoomID] = struct{}{}
		}
		return out, nil
	}

	changed, err := s.rooms.RoomsWithEvents(ctx, roomIDsOf(rooms), since.PDU, to.PDU)
	if err != nil {
		return nil, err
	}
	for _, roomID := range changed {
		out[roomID] = struct{}{}
	}

	joined := joinedRoomIDs(rooms)
	if to.Receipt > since.Receipt {
		receipts, err := s.store.Receipts(ctx, joined, since.Receipt, to.Receipt)
		if err != nil {
			return nil, err
		}
		for _, r := range receipts {
			out[r.RoomID] = struct{}{}
		}
	}

	// Typing lives in memory, with a position of its own that counts
	// changes: a room is looked at when its typing changed in the window.
	if to.Typing != since.Typing {
		for _, roomID := range s.typing.changedRooms(since.Typing, to.Typing) {
			if slices.Contains(joined, roomID) {
				out[roomID] = struct{}{}
			}
		}
	}

	if to.AccountData > since.AccountData {
		entries, err := s.store.AccountDataChanges(ctx, req.Localpart, since.AccountData, to.AccountData)
		if err != nil {
			return nil, err
		}
		for _, entry := range entries {
			if entry.RoomID != "" {
				out[entry.RoomID] = struct{}{}
			}
		}
	}

	// A room whose unread counts moved without an event in this window —
	// read on another device, or counted after the event was already sent
	// to this client — is reported so that the client hears the new
	// counts.
	if s.push != nil && to.Notification > since.Notification {
		changed, err := s.push.NotificationCountChanges(ctx, req.User, since.Notification, to.Notification)
		if err != nil {
			return nil, err
		}
		for _, roomID := range changed {
			if slices.Contains(joined, roomID) {
				out[roomID] = struct{}{}
			}
		}
	}

	for _, room := range rooms {
		if room.Position > since.PDU && room.Position <= to.PDU {
			out[room.RoomID] = struct{}{}
		}
	}
	return out, nil
}

// joinedRoom reads one room the user is in.
func (s *API) joinedRoom(ctx context.Context, req SyncRequest, filter Filter, window roomWindow, room roomserver.MembershipAt) (JoinedRoom, error) {
	read, err := s.rooms.RoomSync(ctx, s.roomSyncRequest(req, filter, window, room))
	if err != nil {
		return JoinedRoom{}, err
	}

	out := JoinedRoom{
		State:    EventList{Events: withoutRoomID(read.State)},
		Timeline: s.timeline(read.Timeline),
	}
	// The summary exists to name a room that has no name of its own, so it
	// goes out only when there is somebody to name it after.
	if len(read.Summary.Heroes) > 0 {
		summary := read.Summary
		out.Summary = &summary
	}

	if out.Ephemeral.Events, err = s.roomEphemeral(ctx, req.User, room.RoomID, window, filter); err != nil {
		return JoinedRoom{}, err
	}
	if out.AccountData.Events, err = s.roomAccountData(ctx, req.Localpart, room.RoomID, window, filter); err != nil {
		return JoinedRoom{}, err
	}
	return out, nil
}

// leftRoom reads a room the user has left: everything up to their leave and
// nothing after it, which the roomserver's visible range takes care of.
func (s *API) leftRoom(ctx context.Context, req SyncRequest, filter Filter, window roomWindow, room roomserver.MembershipAt) (LeftRoom, error) {
	read, err := s.rooms.RoomSync(ctx, s.roomSyncRequest(req, filter, window, room))
	if err != nil {
		return LeftRoom{}, err
	}

	out := LeftRoom{
		State:    EventList{Events: withoutRoomID(read.State)},
		Timeline: s.timeline(read.Timeline),
	}
	if out.AccountData.Events, err = s.roomAccountData(ctx, req.Localpart, room.RoomID, window, filter); err != nil {
		return LeftRoom{}, err
	}
	return out, nil
}

// roomSyncRequest is the read the two room sections share.
func (s *API) roomSyncRequest(req SyncRequest, filter Filter, window roomWindow, room roomserver.MembershipAt) roomserver.RoomSyncRequest {
	return roomserver.RoomSyncRequest{
		RoomID:          room.RoomID,
		User:            req.User,
		Since:           window.timelineSince(),
		To:              window.to.PDU,
		TimelineLimit:   filter.TimelineLimit(),
		TimelineFilter:  filter.TimelineFilter(),
		StateFilter:     filter.StateFilter(),
		FullState:       req.FullState,
		LazyLoadMembers: filter.LazyLoadMembers(),
	}
}

// timeline projects a read timeline into the response shape.
func (s *API) timeline(read roomserver.Timeline) Timeline {
	out := Timeline{
		Events:  withoutRoomID(read.Events),
		Limited: read.Limited,
	}
	// prev_batch is what the client pages backwards from, so it names the
	// position below the first event rather than the window asked for.
	if len(read.Events) > 0 {
		out.PrevBatch = pduToken(read.Start).String()
	}
	return out
}

// roomEphemeral builds a room's typing and receipt EDUs.
func (s *API) roomEphemeral(ctx context.Context, userID, roomID string, window roomWindow, filter Filter) ([]EphemeralEvent, error) {
	out := []EphemeralEvent{}
	ephemeralFilter := filter.EphemeralFilter()

	// A full read reports whoever is typing now; an incremental one
	// reports the room's list when it changed, even when it is now empty.
	var typing *EphemeralEvent
	if window.full {
		typing = s.typingEvent(roomID, false)
	} else if window.to.Typing != window.since.Typing &&
		s.typing.changedIn(roomID, window.since.Typing, window.to.Typing) {
		typing = s.typingEvent(roomID, true)
	}
	if typing != nil && matchesEventFilter(typing.Type, "", ephemeralFilter) {
		out = append(out, *typing)
	}

	byRoom, err := s.roomReceipts(ctx, userID, []string{roomID},
		window.from(window.since.Receipt), window.to.Receipt)
	if err != nil {
		return nil, err
	}
	if e := receiptsEvent(byRoom[roomID]); e != nil && matchesEventFilter(e.Type, "", ephemeralFilter) {
		out = append(out, *e)
	}
	return out, nil
}

// globalAccountData reads the account data that is not about a room.
// initial is whether this is the client's first read of it.
func (s *API) globalAccountData(ctx context.Context, req SyncRequest, from, to int64, initial bool, filter Filter) ([]AccountDataEvent, error) {
	entries, err := s.accountDataEntries(ctx, req.Localpart, from, to)
	if err != nil {
		return nil, err
	}
	// Every user has push rules, written or not: a user who has never
	// changed one has the server defaults, and a client reads its rules
	// from the account data of its first sync whichever it is.
	//
	// "First" is the client's first sync, not a token whose account data
	// position is zero: on a server where nobody has written any account
	// data yet every token has a zero there, and treating each sync as the
	// first would send the rules every time — which makes every sync
	// "interesting", and turns every long poll into a busy loop.
	if initial && s.push != nil && !slices.ContainsFunc(entries, func(e storage.AccountDataEntry) bool {
		return e.RoomID == "" && e.DataType == pushRulesType
	}) {
		entries = append(entries, storage.AccountDataEntry{DataType: pushRulesType})
	}
	return s.accountDataEvents(ctx, req, entries, "", filter.AccountDataFilter())
}

// roomAccountData reads one room's account data.
func (s *API) roomAccountData(ctx context.Context, localpart, roomID string, window roomWindow, filter Filter) ([]AccountDataEvent, error) {
	entries, err := s.accountDataEntries(ctx, localpart,
		window.from(window.since.AccountData), window.to.AccountData)
	if err != nil {
		return nil, err
	}
	return s.accountDataEvents(ctx, SyncRequest{Localpart: localpart}, entries, roomID, filter.RoomAccountDataFilter())
}

// accountDataEntries lists which of a user's account data changed in a
// window, or all of it when the window starts at the beginning.
func (s *API) accountDataEntries(ctx context.Context, localpart string, from, to int64) ([]storage.AccountDataEntry, error) {
	if from == 0 {
		return s.store.AccountDataAt(ctx, localpart, to)
	}
	return s.store.AccountDataChanges(ctx, localpart, from, to)
}

// accountDataEvents turns the entries of one scope — global, or one room —
// into the events a sync reports.
func (s *API) accountDataEvents(ctx context.Context, req SyncRequest, entries []storage.AccountDataEntry, roomID string, filter storage.TimelineFilter) ([]AccountDataEvent, error) {
	out := []AccountDataEvent{}
	for _, entry := range entries {
		if entry.RoomID != roomID {
			continue
		}
		if !matchesEventFilter(entry.DataType, "", filter) {
			continue
		}
		if entry.RoomID == "" && entry.DataType == pushRulesType && s.push != nil {
			content, err := s.push.Document(ctx, req.Localpart, req.User)
			if err != nil {
				return nil, err
			}
			out = append(out, AccountDataEvent{Type: entry.DataType, Content: content})
			continue
		}
		content, err := s.accounts.AccountData(ctx, req.Localpart, entry.RoomID, entry.DataType)
		switch {
		case errors.Is(err, storage.ErrNoSuchAccountData):
			// Deleted between the two reads: nothing to report.
			continue
		case err != nil:
			return nil, err
		}
		out = append(out, AccountDataEvent{Type: entry.DataType, Content: content})
	}
	return out, nil
}

// applySetPresence acts on ?set_presence=.
//
// With presence off the parameter is ignored rather than refused: it is on
// every sync a client sends, and failing them all would make a server with
// presence off unusable.
func (s *API) applySetPresence(ctx context.Context, req SyncRequest) error {
	if !s.presenceEnabled {
		return nil
	}
	state := req.SetPresence
	if state == "" {
		state = PresenceOnline
	}
	if state == PresenceOffline {
		// A client that is syncing is not offline, whatever it says.
		return nil
	}

	err := s.SetPresence(ctx, PresenceRequest{User: req.User, Presence: state})
	if errors.Is(err, ErrInvalidParam) {
		// An unusable value in a parameter that rides on every request is
		// not worth failing the sync for.
		s.log.Debug("ignoring an unusable set_presence", "user_id", req.User, "value", req.SetPresence)
		return nil
	}
	return err
}

// withoutRoomID strips the room ID from the events of a sync response.
//
// The spec's sync carries events without a room_id: it would be the same
// value on every event of the section they are already nested in, and on a
// client with a thousand rooms that is a measurable part of every response.
// /messages, /context and the single-event endpoint do carry it, because
// there the event is not nested under its room.
func withoutRoomID(events []roomserver.ClientEvent) []roomserver.ClientEvent {
	out := make([]roomserver.ClientEvent, 0, len(events))
	for _, e := range events {
		e.RoomID = ""
		out = append(out, e)
	}
	return out
}

// roomIDsOf projects memberships onto room IDs.
func roomIDsOf(rooms []roomserver.MembershipAt) []string {
	out := make([]string, 0, len(rooms))
	for _, room := range rooms {
		out = append(out, room.RoomID)
	}
	return out
}

// joinedRoomIDs is roomIDsOf for the rooms the user is actually in, which
// is the set the ephemeral and presence streams are about.
func joinedRoomIDs(rooms []roomserver.MembershipAt) []string {
	out := make([]string, 0, len(rooms))
	for _, room := range rooms {
		if room.Membership == event.MembershipJoin {
			out = append(out, room.RoomID)
		}
	}
	slices.Sort(out)
	return out
}

// joinedInWindow reports whether the user joined a room inside this
// window.
func joinedInWindow(rooms []roomserver.MembershipAt, from, to int64) bool {
	for _, room := range rooms {
		if room.Membership != event.MembershipJoin {
			continue
		}
		if room.Position > from && (to == 0 || room.Position <= to) {
			return true
		}
	}
	return false
}

// isDeparted reports whether a membership means the user is out of the
// room.
func isDeparted(membership string) bool {
	return membership == event.MembershipLeave || membership == event.MembershipBan
}