Phase 4: sync

Implements PLAN.md Phase 4: the read model. /sync, /messages and /context,
the filters that narrow them, and the streams a response carries that are
not room events — typing, receipts, presence and to-device messages. With
this a client can be used against Warp, which is milestone M1.

The one architectural choice is that the syncapi has no copy of the rooms:
it reads them through the roomserver. Dendrite gives its syncapi its own
tables and feeds them from the output stream, which is right for a design
where the two can be separate processes, and Dendrite's could be. Warp is
one process by decision, and there the copy buys nothing and costs the
hardest class of bug a homeserver has — a read model that disagrees with
the room, where a client is shown a message that was rejected and the only
way to find out is to compare two tables. So the roomserver grew a read
model (stream.go, roomsync.go, visibility.go) rather than exporting its
store: reading an event means parsing it under its room's version, and a
second copy of the version registry, the parser and the client projection
would be a second place for them to disagree.

syncapi: stream tokens naming a position in each of five independent
streams, so a client that has seen receipt 7 but only event 4 can say so in
one opaque string. A token with fewer positions than this build writes is
accepted and the rest read as zero, which is what lets a token stored
before a stream existed go on working. The PDU position comes from
MAX(stream_ordering) and not from the counter row the other four use: the
counter is advanced inside the writing transaction, so a token taken from
it would name a position whose event no reader can see yet, and the event
would be skipped for good once that transaction committed.

Long polling goes through a notifier that wakes by room and by user rather
than globally — an event wakes the listeners of its room, a membership
event also wakes the user it names, so an invite reaches a client that is
not in the room yet. A wake-up is only ever a hint: the listener re-reads
the streams and decides for itself what changed, which is what makes it
safe for the notifier to hold no state about what each client has seen. The
listener registers before the read, so an event landing in between leaves
the channel signalled and the sync reads again rather than blocking on a
change it just missed. EventPersisted must not block, because it is called
with the room's write lock still held.

"Something happened" is not "a token moved". A stream can move without this
user being concerned by it, and returning on every such move would turn a
long poll back into a busy loop. The same trap caught presence twice over:
a client says it is still online by syncing, so ?set_presence= makes every
sync a presence write, and a stream that moved each time would wake every
peer, which would sync, which would move it again. SetPresence therefore
reports whether anything a peer would notice actually changed, and a write
that only refreshes the last-active timestamp leaves the stream alone.

Timeline windows are named by their two ends and a direction, Low exclusive
and High inclusive in both directions, rather than by "from" and "to" as
the wire format spells them. Each of those would otherwise be the low bound
in one direction and the high bound in the other, which is exactly the
off-by-one every pagination bug is made of; the handlers translate once and
nothing below them thinks about it again. A sync's timeline is read
backwards and reported forwards, because a client wants the newest events
of the window it missed — a forward read with a limit hands it the oldest
and then calls the rest a gap.

visibility: history visibility is now a span rather than a yes or no, which
is what Phase 3 deferred here. A user who has left a room is served the
room as it was when they left — their state reads, their members and their
timeline — instead of being refused, and is bounded above at their own
membership event so a kicked client cannot go on streaming. Their span is
found from their own member events, so a user who joined, left and rejoined
sees their latest span. ?at= on /members is honoured for the same reason.
What is left is that the span is computed from the room's current
history_visibility where the spec judges each event by the one in force
when it was sent; that is a state read per event and belongs with open
question 1.

Typing is held in memory and nowhere else: notifications expire in seconds,
are worthless across a restart, and arrive one per keystroke-burst per
room, so persisting them would be the busiest table in the database holding
the least valuable data in it. Receipts only ever advance, compared on the
position of the event acknowledged rather than on arrival, because two of a
user's clients read at different speeds and the slower one must not drag
the marker backwards. To-device queues are per device and drained by the
next sync rather than by the one that delivered them, so a client that
crashes in between gets the batch again — the right way round, since a lost
key share leaves a message permanently undecryptable and a duplicate is
ignored. A wildcard addressee is expanded when sent, not stored, because
"every device" means the ones that existed then.

Presence is off by default, which answers PLAN.md open question 3. It is
the noisiest thing in Matrix and the one genuinely fan-out notification
here; a private homeserver gains little from it. With it off the endpoints
refuse rather than disappearing, because a client that gets a 404 from
/presence concludes the server is too old to have it and stops asking.

storage: one migration per driver adding the receipt, to-device, presence
and filter tables with a counter row each, and a stream position on the
existing account data table rather than a parallel one — there is exactly
one current value per (user, room, type), and a second table would be a
second thing to keep in step. Account data is reported in an event list but
is not an event, so it gets its own response type: a room event with an
empty sender and no timestamp would make a client decide what that meant.

Deliberately missing, each named in docs/sync.md: notification counts (they
need push rules, Phase 8), device lists (the keyserver, Phase 7), anything
remote (Phase 5), threads (Phase 10). device_lists is sent as two empty
lists because a client reads a missing one as malformed; the notification
counts are left out entirely, because a zero there is a claim and an absent
field is not.

Tested through HTTP rather than against the builder, because what is worth
pinning down is the response a client sees — which section a room lands in,
whether a token handed back can be handed in again, whether paging
terminates and loses nothing. The storage tests run against both SQL
drivers; the notifier has its own tests for the concurrent case. A live
server was driven end to end, which is what caught the two bugs the unit
tests had missed: a backwards /messages page must be in reverse
chronological order, and `start` must name where the page began rather than
position zero when ?from= was omitted.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

server/internal/syncapi/context.go

@@ -0,0 +1,178 @@
+package syncapi
+
+import (
+	"context"
+
+	"github.com/dariuszjaniak/warp/server/internal/roomserver"
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// GET /rooms/{roomID}/context/{eventID}: one event and what surrounds it.
+//
+// It is what a permalink opens, and what jumping to a search result or a
+// reply does. The answer is the event, a few events either side of it, the
+// tokens to go on paging in both directions, and the state of the room as
+// of the event — because a client that jumps into the middle of a room's
+// history has none of it.
+
+// ContextRequest is one context read.
+type ContextRequest struct {
+	RoomID  string
+	EventID string
+
+	User      string
+	Localpart string
+
+	// Limit is how many events to return in total, including the event
+	// itself. The spec's default is 10, and the events either side of the
+	// target get half each.
+	Limit int
+
+	// Filter is the ?filter= parameter, a RoomEventFilter as for
+	// /messages.
+	Filter string
+}
+
+// ContextResponse is the event and its surroundings.
+type ContextResponse struct {
+	// Start and End are the tokens to page backwards and forwards from.
+	Start string `json:"start,omitempty"`
+	End   string `json:"end,omitempty"`
+
+	// EventsBefore and EventsAfter are oldest-first and oldest-first
+	// respectively — the spec says EventsBefore is in reverse
+	// chronological order, which is what a client scrolling up consumes.
+	EventsBefore []roomserver.ClientEvent `json:"events_before"`
+	Event        roomserver.ClientEvent   `json:"event"`
+	EventsAfter  []roomserver.ClientEvent `json:"events_after"`
+
+	// State is the room's state as of the event, which is what a client
+	// needs to render the names and avatars of the people in the chunk.
+	State []roomserver.ClientEvent `json:"state"`
+}
+
+// DefaultContextLimit is the spec's default for /context.
+const DefaultContextLimit = 10
+
+// Context answers one context read.
+func (s *API) Context(ctx context.Context, req ContextRequest) (ContextResponse, error) {
+	filter, err := s.messageFilter(ctx, req.Localpart, req.Filter)
+	if err != nil {
+		return ContextResponse{}, err
+	}
+
+	limit := req.Limit
+	if limit <= 0 {
+		limit = DefaultContextLimit
+	}
+	limit = min(limit, MaxTimelineLimit)
+
+	// The target event is read first, and through the roomserver, so that
+	// the visibility check happens before anything else: an event the user
+	// may not see has to come back as unknown rather than as a context
+	// with an empty middle.
+	target, err := s.rooms.Event(ctx, req.RoomID, req.User, req.EventID)
+	if err != nil {
+		return ContextResponse{}, err
+	}
+
+	// The target's own position. The roomserver's timeline reads are
+	// bounded by position rather than by event ID, so it has to be found
+	// before either side can be read — which is one extra read, and the
+	// reason /context costs more than /messages.
+	position, err := s.rooms.EventPosition(ctx, req.RoomID, req.User, req.EventID)
+	if err != nil {
+		return ContextResponse{}, err
+	}
+
+	// Half the budget each side, with the odd one going backwards: a
+	// client opening a permalink scrolls up from it more often than down.
+	after := limit / 2
+	before := limit - after - 1
+	if before < 0 {
+		before = 0
+	}
+
+	out := ContextResponse{
+		Event:        target,
+		EventsBefore: []roomserver.ClientEvent{},
+		EventsAfter:  []roomserver.ClientEvent{},
+		State:        []roomserver.ClientEvent{},
+	}
+
+	if before > 0 {
+		earlier, err := s.rooms.Timeline(ctx, roomserver.TimelineRequest{
+			RoomID: req.RoomID,
+			User:   req.User,
+			// Strictly before the target: High is inclusive, and the
+			// target itself is reported separately.
+			High:      position - 1,
+			Direction: storage.DirectionBackward,
+			Limit:     before,
+			Filter:    filter.TimelineFilter(),
+		})
+		if err != nil {
+			return ContextResponse{}, err
+		}
+		// The spec wants this half newest first: it is what a client
+		// prepends as the user scrolls up.
+		out.EventsBefore = reversed(earlier.Events)
+		out.Start = pduToken(earlier.Start).String()
+	}
+	if out.Start == "" {
+		out.Start = pduToken(position - 1).String()
+	}
+
+	if after > 0 {
+		later, err := s.rooms.Timeline(ctx, roomserver.TimelineRequest{
+			RoomID: req.RoomID,
+			User:   req.User,
+			// Strictly after the target: Low is exclusive.
+			Low:       position,
+			Direction: storage.DirectionForward,
+			Limit:     after,
+			Filter:    filter.TimelineFilter(),
+		})
+		if err != nil {
+			return ContextResponse{}, err
+		}
+		out.EventsAfter = later.Events
+		out.End = pduToken(later.End).String()
+	}
+	if out.End == "" {
+		out.End = pduToken(position).String()
+	}
+
+	state, err := s.contextState(ctx, req, filter, out)
+	if err != nil {
+		return ContextResponse{}, err
+	}
+	out.State = state
+	return out, nil
+}
+
+// contextState is the state a context read reports.
+//
+// With lazy loading it is the member events of everybody in the chunk,
+// which is all a client needs to render it. Without, it is the room's whole
+// state as of the event — which for a large room is most of the response,
+// and is why every client asks for lazy loading.
+func (s *API) contextState(ctx context.Context, req ContextRequest, filter Filter, resp ContextResponse) ([]roomserver.ClientEvent, error) {
+	if filter.LazyLoadMembers() {
+		chunk := make([]roomserver.ClientEvent, 0, len(resp.EventsBefore)+len(resp.EventsAfter)+1)
+		chunk = append(chunk, resp.EventsBefore...)
+		chunk = append(chunk, resp.Event)
+		chunk = append(chunk, resp.EventsAfter...)
+		return s.rooms.MemberEvents(ctx, req.RoomID, sendersOf(chunk))
+	}
+	return s.rooms.StateBeforeEvent(ctx, req.RoomID, req.User, req.EventID)
+}
+
+// reversed returns a copy of events in the opposite order.
+func reversed(events []roomserver.ClientEvent) []roomserver.ClientEvent {
+	out := make([]roomserver.ClientEvent, len(events))
+	for i, e := range events {
+		out[len(events)-1-i] = e
+	}
+	return out
+}

server/internal/syncapi/errors.go

@@ -0,0 +1,48 @@
+package syncapi
+
+import (
+	"fmt"
+
+	"github.com/dariuszjaniak/warp/server/internal/matrix/id"
+)
+
+// The component's errors are plain wrapped sentinels rather than HTTP
+// errors, the same as the roomserver's: this package is called from the
+// Client-Server handlers and, later, from the application service API, and
+// a component that returned HTTP statuses would be deciding something that
+// is not its to decide. The handler maps them; see clientapi.syncError.
+
+// errForbiddenf is a refusal the caller is told the reason for.
+//
+// The reason is passed through to the client on purpose, as it is in the
+// roomserver: "M_FORBIDDEN" on its own sends a user to their operator and
+// the operator to the logs, and nothing in these messages is anything the
+// caller could not work out from the room's state.
+func errForbiddenf(format string, args ...any) error {
+	return fmt.Errorf("%w: %s", ErrForbidden, fmt.Sprintf(format, args...))
+}
+
+// errInvalidParamf is a request parameter this server will not accept.
+func errInvalidParamf(format string, args ...any) error {
+	return fmt.Errorf("%w: %s", ErrInvalidParam, fmt.Sprintf(format, args...))
+}
+
+// errInvalidTokenf is a ?since= or ?from= that could not be read.
+func errInvalidTokenf(format string, args ...any) error {
+	return fmt.Errorf("%w: %s", ErrInvalidToken, fmt.Sprintf(format, args...))
+}
+
+// localpartOf returns the localpart of a user ID of this server.
+func (s *API) localpartOf(userID string) (string, bool) {
+	localpart, serverName, err := id.SplitUserID(userID)
+	if err != nil || serverName != s.serverName {
+		return "", false
+	}
+	return localpart, true
+}
+
+// isLocal reports whether a user ID belongs to this server.
+func (s *API) isLocal(userID string) bool {
+	_, ok := s.localpartOf(userID)
+	return ok
+}

server/internal/syncapi/filter.go

@@ -0,0 +1,374 @@
+package syncapi
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// Filters: what a client asks to be left out of a sync.
+//
+// A filter is uploaded once and referred to by ID on every later sync,
+// which is the point of it — the filter a client wants is long and it does
+// not change, so sending it on every request would be the largest part of
+// every request. The ?filter= parameter also accepts an inline filter as
+// JSON, which is what a client that does not want to upload one uses.
+
+// Errors a filter can fail with.
+var (
+	// ErrNoSuchFilter is an unknown filter ID.
+	ErrNoSuchFilter = storage.ErrNoSuchFilter
+
+	// ErrInvalidFilter is a filter that is not usable: not an object, or
+	// a limit that is not a number.
+	ErrInvalidFilter = errors.New("invalid filter")
+)
+
+// Filter is the whole of a Client-Server API filter.
+//
+// The fields Warp does not act on are still parsed, so that a client
+// uploading a complete filter gets it back unchanged and is not told its
+// filter was rejected because of a field this build ignores.
+type Filter struct {
+	// EventFields names the properties to keep on each event. Warp does
+	// not trim events: see trimToFields.
+	EventFields []string `json:"event_fields,omitempty"`
+
+	// EventFormat is "client" or "federation". Only "client" is offered —
+	// the federation format is a server's view of an event and the
+	// Client-Server API has no business handing it out.
+	EventFormat string `json:"event_format,omitempty"`
+
+	Presence    *EventFilter `json:"presence,omitempty"`
+	AccountData *EventFilter `json:"account_data,omitempty"`
+	Room        *RoomFilter  `json:"room,omitempty"`
+}
+
+// EventFilter narrows a stream of events that are not in a room.
+type EventFilter struct {
+	Limit      *int     `json:"limit,omitempty"`
+	Types      []string `json:"types,omitempty"`
+	NotTypes   []string `json:"not_types,omitempty"`
+	Senders    []string `json:"senders,omitempty"`
+	NotSenders []string `json:"not_senders,omitempty"`
+}
+
+// RoomFilter narrows the room half of a sync.
+type RoomFilter struct {
+	Rooms    []string `json:"rooms,omitempty"`
+	NotRooms []string `json:"not_rooms,omitempty"`
+
+	// IncludeLeave asks for the rooms the user has left. Without it a sync
+	// reports only the leaves that happened inside the window, which is
+	// what a client needs to notice one.
+	IncludeLeave bool `json:"include_leave,omitempty"`
+
+	Ephemeral   *RoomEventFilter `json:"ephemeral,omitempty"`
+	AccountData *RoomEventFilter `json:"account_data,omitempty"`
+	Timeline    *RoomEventFilter `json:"timeline,omitempty"`
+	State       *StateFilter     `json:"state,omitempty"`
+}
+
+// RoomEventFilter narrows a stream of room events.
+type RoomEventFilter struct {
+	Limit      *int     `json:"limit,omitempty"`
+	Types      []string `json:"types,omitempty"`
+	NotTypes   []string `json:"not_types,omitempty"`
+	Senders    []string `json:"senders,omitempty"`
+	NotSenders []string `json:"not_senders,omitempty"`
+	Rooms      []string `json:"rooms,omitempty"`
+	NotRooms   []string `json:"not_rooms,omitempty"`
+
+	// ContainsURL selects events whose content has a `url` property, which
+	// is how a client builds a room's media gallery. Warp does not act on
+	// it: see the note in Validate.
+	ContainsURL *bool `json:"contains_url,omitempty"`
+
+	// LazyLoadMembers leaves out the member events of everybody who does
+	// not appear in the timeline being returned. It is what makes a room
+	// with ten thousand members openable.
+	LazyLoadMembers bool `json:"lazy_load_members,omitempty"`
+
+	// IncludeRedundantMembers asks for those member events even when the
+	// client has already been sent them. Warp always behaves as though
+	// this were set; see roomserver.shapeState.
+	IncludeRedundantMembers bool `json:"include_redundant_members,omitempty"`
+
+	// UnreadThreadNotifications is a threads feature; threads are
+	// PLAN.md phase 10.
+	UnreadThreadNotifications bool `json:"unread_thread_notifications,omitempty"`
+}
+
+// StateFilter is a RoomEventFilter applied to a room's state. The spec
+// gives it the same shape, and the only reason it is a separate type is
+// that the state half of a response has no limit worth applying: a partial
+// state would be a lie about the room rather than a shorter answer.
+type StateFilter = RoomEventFilter
+
+// DefaultTimelineLimit is how many timeline events a room's half of a sync
+// carries when the client does not ask for a number. The spec's default is
+// 10; 20 is a screenful on a phone and still small.
+const DefaultTimelineLimit = 20
+
+// MaxTimelineLimit caps what a client may ask for in one room, on /sync and
+// on /messages alike.
+const MaxTimelineLimit = 1000
+
+// MaxFilterSize caps an uploaded filter. A filter is a list of types and
+// senders; anything larger is not a filter.
+const MaxFilterSize = 64 << 10
+
+// ParseFilter reads a filter from JSON and validates it.
+func ParseFilter(raw []byte) (Filter, error) {
+	if len(raw) > MaxFilterSize {
+		return Filter{}, fmt.Errorf("%w: a filter may be at most %d bytes", ErrInvalidFilter, MaxFilterSize)
+	}
+	var f Filter
+	if err := json.Unmarshal(raw, &f); err != nil {
+		return Filter{}, fmt.Errorf("%w: %v", ErrInvalidFilter, err)
+	}
+	if err := f.Validate(); err != nil {
+		return Filter{}, err
+	}
+	return f, nil
+}
+
+// unmarshalStrict reads JSON into v, refusing a body larger than a filter
+// may be. The size check is here as well as in ParseFilter because an
+// inline filter arrives in a query parameter, where nothing else bounds it.
+func unmarshalStrict(raw []byte, v any) error {
+	if len(raw) > MaxFilterSize {
+		return fmt.Errorf("a filter may be at most %d bytes", MaxFilterSize)
+	}
+	return json.Unmarshal(raw, v)
+}
+
+// Validate checks the fields this server acts on.
+//
+// A field Warp ignores is not an error. The spec's filters are a superset
+// of what any server acts on, and a client that uploads `contains_url`
+// would otherwise be unable to sync at all — whereas ignoring it returns
+// more events than asked for, which every client copes with because a
+// server is allowed to return a superset.
+func (f Filter) Validate() error {
+	if f.EventFormat != "" && f.EventFormat != "client" {
+		return fmt.Errorf("%w: event_format %q is not supported; only \"client\" is",
+			ErrInvalidFilter, f.EventFormat)
+	}
+	for _, check := range []struct {
+		name  string
+		limit *int
+	}{
+		{"presence.limit", limitOf(f.Presence)},
+		{"account_data.limit", limitOf(f.AccountData)},
+	} {
+		if check.limit != nil && *check.limit < 0 {
+			return fmt.Errorf("%w: %s must not be negative", ErrInvalidFilter, check.name)
+		}
+	}
+	if f.Room == nil {
+		return nil
+	}
+	for _, check := range []struct {
+		name   string
+		filter *RoomEventFilter
+	}{
+		{"room.timeline.limit", f.Room.Timeline},
+		{"room.ephemeral.limit", f.Room.Ephemeral},
+		{"room.account_data.limit", f.Room.AccountData},
+	} {
+		if check.filter != nil && check.filter.Limit != nil && *check.filter.Limit < 0 {
+			return fmt.Errorf("%w: %s must not be negative", ErrInvalidFilter, check.name)
+		}
+	}
+	return nil
+}
+
+func limitOf(f *EventFilter) *int {
+	if f == nil {
+		return nil
+	}
+	return f.Limit
+}
+
+// TimelineLimit is how many timeline events per room this filter asks for.
+func (f Filter) TimelineLimit() int {
+	if f.Room == nil || f.Room.Timeline == nil || f.Room.Timeline.Limit == nil {
+		return DefaultTimelineLimit
+	}
+	return clampLimit(*f.Room.Timeline.Limit)
+}
+
+// clampLimit holds a client's limit inside what this server will serve.
+func clampLimit(limit int) int {
+	switch {
+	case limit <= 0:
+		// Zero is a legitimate request for no timeline at all — a client
+		// that only wants state — and is honoured rather than defaulted.
+		return 0
+	case limit > MaxTimelineLimit:
+		return MaxTimelineLimit
+	}
+	return limit
+}
+
+// LazyLoadMembers reports whether the filter asks for lazy-loaded members.
+func (f Filter) LazyLoadMembers() bool {
+	return f.Room != nil && f.Room.Timeline != nil && f.Room.Timeline.LazyLoadMembers
+}
+
+// IncludeLeave reports whether the rooms the user has left should be
+// reported in full rather than only when the leave is new.
+func (f Filter) IncludeLeave() bool {
+	return f.Room != nil && f.Room.IncludeLeave
+}
+
+// TimelineFilter projects the timeline filter onto what the store applies.
+func (f Filter) TimelineFilter() storage.TimelineFilter {
+	if f.Room == nil {
+		return storage.TimelineFilter{}
+	}
+	return roomEventFilter(f.Room.Timeline)
+}
+
+// StateFilter projects the state filter onto what the roomserver applies.
+func (f Filter) StateFilter() storage.TimelineFilter {
+	if f.Room == nil {
+		return storage.TimelineFilter{}
+	}
+	return roomEventFilter(f.Room.State)
+}
+
+// EphemeralFilter projects the ephemeral filter: the typing and receipt
+// events of a room.
+func (f Filter) EphemeralFilter() storage.TimelineFilter {
+	if f.Room == nil {
+		return storage.TimelineFilter{}
+	}
+	return roomEventFilter(f.Room.Ephemeral)
+}
+
+// RoomAccountDataFilter projects the per-room account data filter.
+func (f Filter) RoomAccountDataFilter() storage.TimelineFilter {
+	if f.Room == nil {
+		return storage.TimelineFilter{}
+	}
+	return roomEventFilter(f.Room.AccountData)
+}
+
+// AccountDataFilter projects the global account data filter.
+func (f Filter) AccountDataFilter() storage.TimelineFilter {
+	return eventFilter(f.AccountData)
+}
+
+// PresenceFilter projects the presence filter.
+func (f Filter) PresenceFilter() storage.TimelineFilter {
+	return eventFilter(f.Presence)
+}
+
+func roomEventFilter(f *RoomEventFilter) storage.TimelineFilter {
+	if f == nil {
+		return storage.TimelineFilter{}
+	}
+	return storage.TimelineFilter{
+		Types:      f.Types,
+		NotTypes:   f.NotTypes,
+		Senders:    f.Senders,
+		NotSenders: f.NotSenders,
+	}
+}
+
+func eventFilter(f *EventFilter) storage.TimelineFilter {
+	if f == nil {
+		return storage.TimelineFilter{}
+	}
+	return storage.TimelineFilter{
+		Types:      f.Types,
+		NotTypes:   f.NotTypes,
+		Senders:    f.Senders,
+		NotSenders: f.NotSenders,
+	}
+}
+
+// AllowsRoom reports whether a room passes the filter's room lists.
+//
+// The lists are at two levels — room.rooms and, per section,
+// room.timeline.rooms — and both apply, with the exclusions winning, which
+// is what the spec requires.
+func (f Filter) AllowsRoom(roomID string) bool {
+	if f.Room == nil {
+		return true
+	}
+	if !allowsRoom(roomID, f.Room.Rooms, f.Room.NotRooms) {
+		return false
+	}
+	if f.Room.Timeline != nil {
+		return allowsRoom(roomID, f.Room.Timeline.Rooms, f.Room.Timeline.NotRooms)
+	}
+	return true
+}
+
+func allowsRoom(roomID string, include, exclude []string) bool {
+	for _, pattern := range exclude {
+		if matches(roomID, pattern) {
+			return false
+		}
+	}
+	if len(include) == 0 {
+		return true
+	}
+	for _, pattern := range include {
+		if matches(roomID, pattern) {
+			return true
+		}
+	}
+	return false
+}
+
+// matches applies one filter pattern, in which a trailing '*' is the
+// spec's wildcard.
+func matches(value, pattern string) bool {
+	if prefix, isWildcard := strings.CutSuffix(pattern, "*"); isWildcard {
+		return strings.HasPrefix(value, prefix)
+	}
+	return value == pattern
+}
+
+// SaveFilter stores a filter for a user and returns its ID.
+func (s *API) SaveFilter(ctx context.Context, localpart string, raw []byte) (string, error) {
+	if _, err := ParseFilter(raw); err != nil {
+		return "", err
+	}
+	return s.store.SaveFilter(ctx, localpart, raw, s.nowMS())
+}
+
+// StoredFilter returns an uploaded filter exactly as it was uploaded.
+//
+// Byte-for-byte is deliberate: a client that compares what it gets back
+// with what it sent would see a difference that is not there if the filter
+// were re-encoded through a Go struct, because the fields this build does
+// not act on would come back in a different order or not at all.
+func (s *API) StoredFilter(ctx context.Context, localpart, filterID string) ([]byte, error) {
+	return s.store.Filter(ctx, localpart, filterID)
+}
+
+// filterFor resolves the ?filter= parameter: a filter ID, an inline filter
+// as JSON, or nothing.
+func (s *API) filterFor(ctx context.Context, localpart, raw string) (Filter, error) {
+	switch {
+	case raw == "":
+		return Filter{}, nil
+	case strings.HasPrefix(raw, "{"):
+		return ParseFilter([]byte(raw))
+	}
+
+	stored, err := s.store.Filter(ctx, localpart, raw)
+	if err != nil {
+		return Filter{}, err
+	}
+	return ParseFilter(stored)
+}

server/internal/syncapi/filter_test.go

@@ -0,0 +1,144 @@
+package syncapi_test
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/dariuszjaniak/warp/server/internal/syncapi"
+)
+
+func TestParseFilterReadsTheFieldsThatMatter(t *testing.T) {
+	filter, err := syncapi.ParseFilter([]byte(`{
+		"room": {
+			"timeline": {"limit": 5, "types": ["m.room.message"], "lazy_load_members": true},
+			"include_leave": true
+		}
+	}`))
+	if err != nil {
+		t.Fatalf("ParseFilter: %v", err)
+	}
+
+	if got := filter.TimelineLimit(); got != 5 {
+		t.Errorf("TimelineLimit() = %d, want 5", got)
+	}
+	if !filter.LazyLoadMembers() {
+		t.Error("LazyLoadMembers() = false, want true")
+	}
+	if !filter.IncludeLeave() {
+		t.Error("IncludeLeave() = false, want true")
+	}
+	if types := filter.TimelineFilter().Types; len(types) != 1 || types[0] != "m.room.message" {
+		t.Errorf("timeline types = %v, want [m.room.message]", types)
+	}
+}
+
+func TestFilterDefaultsAreTheSpecsOwn(t *testing.T) {
+	var filter syncapi.Filter
+
+	if got := filter.TimelineLimit(); got != syncapi.DefaultTimelineLimit {
+		t.Errorf("TimelineLimit() = %d, want the default %d", got, syncapi.DefaultTimelineLimit)
+	}
+	if filter.LazyLoadMembers() {
+		t.Error("lazy loading must be off unless asked for")
+	}
+	if filter.IncludeLeave() {
+		t.Error("left rooms must be left out unless asked for")
+	}
+	if !filter.AllowsRoom("!anything:warp.example") {
+		t.Error("an empty filter must allow every room")
+	}
+}
+
+func TestFilterClampsAnAbsurdLimit(t *testing.T) {
+	filter, err := syncapi.ParseFilter([]byte(`{"room":{"timeline":{"limit":1000000}}}`))
+	if err != nil {
+		t.Fatalf("ParseFilter: %v", err)
+	}
+	if got := filter.TimelineLimit(); got != syncapi.MaxTimelineLimit {
+		t.Errorf("TimelineLimit() = %d, want it capped at %d", got, syncapi.MaxTimelineLimit)
+	}
+}
+
+func TestFilterHonoursAZeroLimit(t *testing.T) {
+	// Zero is a legitimate request for no timeline at all — a client that
+	// only wants state — and must not be read as "unset".
+	filter, err := syncapi.ParseFilter([]byte(`{"room":{"timeline":{"limit":0}}}`))
+	if err != nil {
+		t.Fatalf("ParseFilter: %v", err)
+	}
+	if got := filter.TimelineLimit(); got != 0 {
+		t.Errorf("TimelineLimit() = %d, want 0", got)
+	}
+}
+
+func TestFilterRoomLists(t *testing.T) {
+	filter, err := syncapi.ParseFilter([]byte(`{
+		"room": {"rooms": ["!kept:warp.example"], "not_rooms": ["!dropped:warp.example"]}
+	}`))
+	if err != nil {
+		t.Fatalf("ParseFilter: %v", err)
+	}
+
+	for _, tc := range []struct {
+		roomID string
+		want   bool
+	}{
+		{"!kept:warp.example", true},
+		{"!dropped:warp.example", false},
+		{"!unmentioned:warp.example", false},
+	} {
+		if got := filter.AllowsRoom(tc.roomID); got != tc.want {
+			t.Errorf("AllowsRoom(%q) = %v, want %v", tc.roomID, got, tc.want)
+		}
+	}
+}
+
+func TestFilterExclusionsWinOverInclusions(t *testing.T) {
+	// The spec is explicit: a value matching both lists is excluded.
+	filter, err := syncapi.ParseFilter([]byte(`{
+		"room": {"rooms": ["!both:warp.example"], "not_rooms": ["!both:warp.example"]}
+	}`))
+	if err != nil {
+		t.Fatalf("ParseFilter: %v", err)
+	}
+	if filter.AllowsRoom("!both:warp.example") {
+		t.Error("a room in both lists must be excluded")
+	}
+}
+
+func TestFilterRejectsAnUnsupportedEventFormat(t *testing.T) {
+	// The federation format is a server's view of an event, and the
+	// Client-Server API has no business handing it out.
+	_, err := syncapi.ParseFilter([]byte(`{"event_format":"federation"}`))
+	if !errors.Is(err, syncapi.ErrInvalidFilter) {
+		t.Errorf("error = %v, want ErrInvalidFilter", err)
+	}
+}
+
+func TestFilterRejectsNonsense(t *testing.T) {
+	for _, raw := range []string{
+		`not json`,
+		`[]`,
+		`{"room":{"timeline":{"limit":-1}}}`,
+	} {
+		if _, err := syncapi.ParseFilter([]byte(raw)); !errors.Is(err, syncapi.ErrInvalidFilter) {
+			t.Errorf("ParseFilter(%q) error = %v, want ErrInvalidFilter", raw, err)
+		}
+	}
+}
+
+func TestFilterIgnoresFieldsThisServerDoesNotActOn(t *testing.T) {
+	// A server is allowed to return a superset of what a filter asked
+	// for, and a client that uploads `contains_url` must still be able to
+	// sync. Refusing the filter would be the one outcome that breaks it.
+	filter, err := syncapi.ParseFilter([]byte(`{
+		"event_fields": ["type", "content"],
+		"room": {"timeline": {"contains_url": true, "unread_thread_notifications": true}}
+	}`))
+	if err != nil {
+		t.Fatalf("ParseFilter: %v", err)
+	}
+	if got := filter.TimelineLimit(); got != syncapi.DefaultTimelineLimit {
+		t.Errorf("TimelineLimit() = %d, want the default", got)
+	}
+}

server/internal/syncapi/messages.go

@@ -0,0 +1,210 @@
+package syncapi
+
+import (
+	"context"
+
+	"github.com/dariuszjaniak/warp/server/internal/roomserver"
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// GET /rooms/{roomID}/messages: a room's history, a page at a time.
+//
+// It is the other half of /sync. A sync hands a client the live stream and,
+// when a client has been away, a `limited` timeline with a gap in it;
+// /messages is how the client closes that gap, and how it scrolls back
+// through a room it has just opened. The tokens are the same ones — a sync's
+// prev_batch is a valid ?from= here, and an ?end= from here is a valid
+// ?since= for /sync — which is what makes the two compose.
+
+// MessagesRequest is one page of a room's history.
+type MessagesRequest struct {
+	RoomID string
+
+	// User is the full user ID asking, and Localpart is their localpart,
+	// for resolving a filter ID.
+	User      string
+	Localpart string
+
+	// From is where to page from. Empty means the end of the room for a
+	// backwards page and the start of it for a forwards one, which is what
+	// a client opening a room sends.
+	From string
+
+	// To bounds the far end of the page. Empty is unbounded.
+	To string
+
+	// Direction is "b" to page backwards, which is what scrolling up does,
+	// or "f" to page forwards. The spec requires it.
+	Direction storage.TimelineDirection
+
+	// Limit caps the events returned.
+	Limit int
+
+	// Filter is the ?filter= parameter: a filter ID or an inline
+	// RoomEventFilter as JSON.
+	Filter string
+}
+
+// MessagesResponse is one page.
+type MessagesResponse struct {
+	// Start is the token this page began at, which is the ?from= the
+	// client sent back to it.
+	Start string `json:"start"`
+
+	// End is the token for the next page. It is left out when the page
+	// reached the end of what the client may see, which is how a client
+	// knows to stop scrolling.
+	End string `json:"end,omitempty"`
+
+	Chunk []roomserver.ClientEvent `json:"chunk"`
+
+	// State is the state of the events in the chunk: the member events of
+	// their senders, when the filter asked for lazy loading. A client
+	// needs them to render a name against a message from somebody who is
+	// no longer in the room.
+	State []roomserver.ClientEvent `json:"state,omitempty"`
+}
+
+// Messages answers one page of a room's history.
+func (s *API) Messages(ctx context.Context, req MessagesRequest) (MessagesResponse, error) {
+	from, err := ParseToken(req.From)
+	if err != nil {
+		return MessagesResponse{}, errInvalidTokenf("from: %v", err)
+	}
+	to, err := ParseToken(req.To)
+	if err != nil {
+		return MessagesResponse{}, errInvalidTokenf("to: %v", err)
+	}
+
+	filter, err := s.messageFilter(ctx, req.Localpart, req.Filter)
+	if err != nil {
+		return MessagesResponse{}, err
+	}
+
+	direction := req.Direction
+	if direction != storage.DirectionForward {
+		// Backwards is the default: the spec makes `dir` required, but
+		// every client that leaves it out means "scroll up".
+		direction = storage.DirectionBackward
+	}
+
+	limit := req.Limit
+	if limit <= 0 {
+		limit = DefaultTimelineLimit
+	}
+	limit = min(limit, MaxTimelineLimit)
+
+	// Which end of the window the client's two tokens are depends on which
+	// way it is paging: scrolling up, `from` is the ceiling and `to` the
+	// floor, and scrolling down it is the other way round. An unset bound
+	// is zero, which the read already takes as "unbounded at that end" —
+	// so paging backwards from the end of the room needs nothing special.
+	low, high := from.PDU, to.PDU
+	if direction == storage.DirectionBackward {
+		low, high = to.PDU, from.PDU
+	}
+
+	timeline, err := s.rooms.Timeline(ctx, roomserver.TimelineRequest{
+		RoomID:    req.RoomID,
+		User:      req.User,
+		Low:       low,
+		High:      high,
+		Direction: direction,
+		Limit:     limit,
+		Filter:    filter.TimelineFilter(),
+	})
+	if err != nil {
+		return MessagesResponse{}, err
+	}
+
+	// The chunk is in the direction asked for: the spec says a backwards
+	// page is in reverse chronological order, because that is the order a
+	// client scrolling up consumes it in. The read returns events oldest
+	// first — which is what /sync wants — so a backwards page is turned
+	// round here.
+	chunk := timeline.Events
+	if direction == storage.DirectionBackward {
+		chunk = reversed(chunk)
+	}
+	if chunk == nil {
+		chunk = []roomserver.ClientEvent{}
+	}
+
+	out := MessagesResponse{
+		Start: pduToken(from.PDU).String(),
+		Chunk: chunk,
+	}
+	// `from` is optional, and without it `start` has to name the end of
+	// the room the page actually began at rather than position zero: a
+	// client that paged backwards from the end and then used `start` to
+	// page forwards again would otherwise be handed the whole room.
+	if req.From == "" {
+		if direction == storage.DirectionBackward {
+			out.Start = pduToken(timeline.End).String()
+		} else {
+			out.Start = pduToken(timeline.Start).String()
+		}
+	}
+
+	// `end` is left out at the end of the room, which is how a client
+	// knows to stop asking. Reading backwards the far end is Start, the
+	// position below the oldest event; reading forwards it is End.
+	//
+	// The read fetches one more row than the limit, so `Limited` is the
+	// whole answer to "is there more": a page that came back exactly full
+	// and happened to be the last one still correctly has no `end`.
+	if timeline.Limited {
+		if direction == storage.DirectionBackward {
+			out.End = pduToken(timeline.Start).String()
+		} else {
+			out.End = pduToken(timeline.End).String()
+		}
+	}
+
+	if filter.LazyLoadMembers() {
+		state, err := s.rooms.MemberEvents(ctx, req.RoomID, sendersOf(chunk))
+		if err != nil {
+			return MessagesResponse{}, err
+		}
+		out.State = state
+	}
+	return out, nil
+}
+
+// messageFilter resolves the ?filter= of /messages.
+//
+// The parameter is a RoomEventFilter here rather than the whole filter that
+// /sync takes — the spec defines it that way, because a filter about
+// presence or account data would mean nothing to a room's history. A filter
+// ID, though, names a full filter, so both shapes have to be accepted: an
+// inline one is wrapped into the room's timeline slot, and a stored one is
+// used as it stands.
+func (s *API) messageFilter(ctx context.Context, localpart, raw string) (Filter, error) {
+	if raw == "" {
+		return Filter{}, nil
+	}
+	if raw[0] != '{' {
+		return s.filterFor(ctx, localpart, raw)
+	}
+
+	var roomEvents RoomEventFilter
+	if err := unmarshalStrict([]byte(raw), &roomEvents); err != nil {
+		return Filter{}, errInvalidParamf("filter: %v", err)
+	}
+	return Filter{Room: &RoomFilter{Timeline: &roomEvents}}, nil
+}
+
+// sendersOf lists the distinct senders of a chunk, in the order they first
+// appear.
+func sendersOf(events []roomserver.ClientEvent) []string {
+	seen := make(map[string]struct{}, len(events))
+	out := make([]string, 0, len(events))
+	for _, e := range events {
+		if _, dup := seen[e.Sender]; dup {
+			continue
+		}
+		seen[e.Sender] = struct{}{}
+		out = append(out, e.Sender)
+	}
+	return out
+}

server/internal/syncapi/notifier.go

@@ -0,0 +1,207 @@
+package syncapi
+
+import (
+	"context"
+	"sync"
+
+	"github.com/dariuszjaniak/warp/server/internal/matrix/event"
+	"github.com/dariuszjaniak/warp/server/internal/roomserver"
+)
+
+// Notifier wakes the long-polling syncs that something concerns.
+//
+// It is the other half of the roomserver's emit stage: the input pipeline
+// is the only thing that knows an event has landed, and a client blocked in
+// /sync for thirty seconds has to hear about it in milliseconds rather than
+// when its timeout expires. Polling the database instead would mean every
+// connected client querying every second, which is the cost this avoids.
+//
+// Waking is by room and by user rather than globally. A sync registers the
+// rooms it just read plus its own user ID; an event wakes the listeners of
+// its room, and a membership event also wakes the listener of the user it
+// names — which is how an invite reaches a client that is not in the room
+// yet and so is not listening for it. An event in a room nobody is polling
+// wakes nobody.
+//
+// A wake-up is only ever a hint. The listener re-reads the streams from the
+// database and works out for itself what changed, so a spurious wake-up
+// costs one query and a missed one costs at most the sync timeout. That is
+// what makes it safe for this to hold no state about what each client has
+// seen.
+type Notifier struct {
+	mu    sync.Mutex
+	rooms map[string]map[*Listener]struct{}
+	users map[string]map[*Listener]struct{}
+}
+
+// NewNotifier returns a notifier with no listeners.
+func NewNotifier() *Notifier {
+	return &Notifier{
+		rooms: map[string]map[*Listener]struct{}{},
+		users: map[string]map[*Listener]struct{}{},
+	}
+}
+
+// Listener is one blocked sync's wake-up channel.
+type Listener struct {
+	notifier *Notifier
+	wake     chan struct{}
+
+	userID string
+	rooms  []string
+}
+
+// Listen registers interest in a user and a set of rooms.
+//
+// The caller must Close the listener. Registering before reading the
+// streams is what closes the race: an event that lands between the
+// registration and the read leaves the channel already signalled, so the
+// following Wait returns at once and the sync reads again rather than
+// blocking on a change it just missed.
+func (n *Notifier) Listen(userID string, roomIDs []string) *Listener {
+	l := &Listener{
+		notifier: n,
+		// One slot, so a wake-up is never lost and never queued: the
+		// listener re-reads everything anyway, so two wake-ups and one are
+		// the same instruction.
+		wake:   make(chan struct{}, 1),
+		userID: userID,
+		rooms:  roomIDs,
+	}
+
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	if userID != "" {
+		add(n.users, userID, l)
+	}
+	for _, roomID := range roomIDs {
+		add(n.rooms, roomID, l)
+	}
+	return l
+}
+
+// Wait returns the channel a wake-up arrives on.
+func (l *Listener) Wait() <-chan struct{} { return l.wake }
+
+// Close deregisters the listener. It is safe to call more than once.
+func (l *Listener) Close() {
+	if l == nil || l.notifier == nil {
+		return
+	}
+	n := l.notifier
+	l.notifier = nil
+
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	if l.userID != "" {
+		remove(n.users, l.userID, l)
+	}
+	for _, roomID := range l.rooms {
+		remove(n.rooms, roomID, l)
+	}
+}
+
+// EventPersisted implements roomserver.Notifier: it is called once per
+// event the input pipeline has committed, in the order they were committed.
+//
+// It must not block. The call happens with the room's write lock still
+// held, so anything slow here would serialise the room's writers behind the
+// slowest reader of it — which is why waking is a non-blocking send onto a
+// one-slot channel and nothing else.
+func (n *Notifier) EventPersisted(_ context.Context, notification roomserver.Notification) {
+	n.NotifyRoom(notification.RoomID)
+
+	// A membership event also concerns the user it names, who for an
+	// invite is not in the room and so is not listening for it.
+	e := notification.Event
+	if e == nil || e.Type() != event.TypeMember {
+		return
+	}
+	if stateKey, ok := e.StateKey(); ok && stateKey != "" {
+		n.NotifyUser(stateKey)
+	}
+}
+
+// NotifyRoom wakes the listeners of one room. It is what the ephemeral
+// streams — typing and receipts — use, since those are about a room but are
+// not events.
+func (n *Notifier) NotifyRoom(roomID string) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	wakeAll(n.rooms[roomID])
+}
+
+// NotifyUser wakes the listeners of one user: their to-device messages,
+// their account data and the invites addressed to them.
+func (n *Notifier) NotifyUser(userID string) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	wakeAll(n.users[userID])
+}
+
+// NotifyUsers wakes several users at once, under one lock.
+func (n *Notifier) NotifyUsers(userIDs []string) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	for _, userID := range userIDs {
+		wakeAll(n.users[userID])
+	}
+}
+
+// Listeners reports how many syncs are currently blocked, for metrics and
+// for the tests that would otherwise have to guess.
+func (n *Notifier) Listeners() int {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	seen := map[*Listener]struct{}{}
+	for _, set := range n.rooms {
+		for l := range set {
+			seen[l] = struct{}{}
+		}
+	}
+	for _, set := range n.users {
+		for l := range set {
+			seen[l] = struct{}{}
+		}
+	}
+	return len(seen)
+}
+
+// wakeAll signals every listener in a set without blocking on any of them.
+func wakeAll(set map[*Listener]struct{}) {
+	for l := range set {
+		select {
+		case l.wake <- struct{}{}:
+		default:
+			// Already signalled and not yet read: the listener will see
+			// this change when it acts on the signal it has.
+		}
+	}
+}
+
+func add(index map[string]map[*Listener]struct{}, key string, l *Listener) {
+	set, ok := index[key]
+	if !ok {
+		set = map[*Listener]struct{}{}
+		index[key] = set
+	}
+	set[l] = struct{}{}
+}
+
+func remove(index map[string]map[*Listener]struct{}, key string, l *Listener) {
+	set, ok := index[key]
+	if !ok {
+		return
+	}
+	delete(set, l)
+	if len(set) == 0 {
+		// An empty set left behind would make the index grow with every
+		// room any client has ever polled.
+		delete(index, key)
+	}
+}
+
+// Compile-time proof that the notifier is what the roomserver's emit stage
+// expects.
+var _ roomserver.Notifier = (*Notifier)(nil)

server/internal/syncapi/notifier_test.go

@@ -0,0 +1,150 @@
+package syncapi_test
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/dariuszjaniak/warp/server/internal/syncapi"
+)
+
+// The notifier is the one piece of this component that several goroutines
+// touch at once, so it is tested on its own rather than only through a
+// blocked /sync.
+
+// woken reports whether a listener has been signalled, without blocking.
+func woken(l *syncapi.Listener) bool {
+	select {
+	case <-l.Wait():
+		return true
+	default:
+		return false
+	}
+}
+
+func TestNotifierWakesOnlyWhoIsConcerned(t *testing.T) {
+	n := syncapi.NewNotifier()
+
+	inRoom := n.Listen("@alice:warp.example", []string{"!a:warp.example"})
+	elsewhere := n.Listen("@bob:warp.example", []string{"!b:warp.example"})
+	t.Cleanup(inRoom.Close)
+	t.Cleanup(elsewhere.Close)
+
+	n.NotifyRoom("!a:warp.example")
+
+	if !woken(inRoom) {
+		t.Error("a listener on the room was not woken")
+	}
+	if woken(elsewhere) {
+		t.Error("a listener on another room was woken")
+	}
+}
+
+func TestNotifierWakesByUser(t *testing.T) {
+	n := syncapi.NewNotifier()
+
+	// A user who is in no rooms at all still has to be reachable: an
+	// invite, a to-device message and account data all arrive without any
+	// room they are listening to.
+	listener := n.Listen("@alice:warp.example", nil)
+	t.Cleanup(listener.Close)
+
+	n.NotifyUser("@alice:warp.example")
+	if !woken(listener) {
+		t.Error("a listener was not woken by a notification addressed to its user")
+	}
+}
+
+func TestNotifierCoalescesWakeups(t *testing.T) {
+	n := syncapi.NewNotifier()
+	listener := n.Listen("@alice:warp.example", []string{"!a:warp.example"})
+	t.Cleanup(listener.Close)
+
+	// A listener re-reads everything when it wakes, so two wake-ups and
+	// one are the same instruction. What matters is that the second does
+	// not block the notifier.
+	for range 100 {
+		n.NotifyRoom("!a:warp.example")
+	}
+
+	if !woken(listener) {
+		t.Fatal("the listener was not woken at all")
+	}
+	if woken(listener) {
+		t.Error("wake-ups queued up instead of coalescing")
+	}
+}
+
+func TestNotifierForgetsAClosedListener(t *testing.T) {
+	n := syncapi.NewNotifier()
+
+	listener := n.Listen("@alice:warp.example", []string{"!a:warp.example"})
+	if n.Listeners() != 1 {
+		t.Fatalf("Listeners() = %d, want 1", n.Listeners())
+	}
+
+	listener.Close()
+	if n.Listeners() != 0 {
+		t.Errorf("Listeners() = %d after Close, want 0", n.Listeners())
+	}
+	// Closing twice must not panic or double-remove: a sync closes its
+	// listener on every path out, and some of those paths overlap.
+	listener.Close()
+
+	// Nothing is left behind to be woken, and notifying an unknown room is
+	// not an error.
+	n.NotifyRoom("!a:warp.example")
+	n.NotifyUser("@alice:warp.example")
+}
+
+func TestNotifierUnderConcurrentUse(t *testing.T) {
+	n := syncapi.NewNotifier()
+
+	// Listeners coming and going while notifications arrive is the normal
+	// state of a server: every /sync registers, reads and deregisters.
+	// Run under `go test -race` this is the test that would catch a lock
+	// held in the wrong place.
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+
+	for i := range 8 {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				l := n.Listen("@alice:warp.example", []string{"!a:warp.example"})
+				_ = woken(l)
+				l.Close()
+			}
+		}(i)
+	}
+	for range 4 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				n.NotifyRoom("!a:warp.example")
+				n.NotifyUser("@alice:warp.example")
+				n.NotifyUsers([]string{"@alice:warp.example", "@bob:warp.example"})
+			}
+		}()
+	}
+
+	time.Sleep(50 * time.Millisecond)
+	close(stop)
+	wg.Wait()
+
+	if n.Listeners() != 0 {
+		t.Errorf("Listeners() = %d after everything closed, want 0", n.Listeners())
+	}
+}

server/internal/syncapi/presence.go

@@ -0,0 +1,251 @@
+package syncapi
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// Presence: who is online, and the status message they set.
+//
+// It is off by default (PLAN.md open question 3). Presence is the single
+// noisiest thing in Matrix — every user's state change is sent to everybody
+// who shares a room with them, which over federation is quadratic in the
+// worst case, and Synapse deployments routinely turn it off for exactly
+// that reason. A private homeserver with a dozen users has no such problem,
+// so it is a switch rather than a decision.
+//
+// With it off, the endpoints answer M_FORBIDDEN and the presence section of
+// a sync response is empty. They are not unregistered: a client that gets
+// a 404 from /presence concludes the server is too old to have it and
+// stops asking, whereas a refusal says this server has it switched off.
+
+// The presence states the spec defines.
+const (
+	PresenceOnline      = "online"
+	PresenceOffline     = "offline"
+	PresenceUnavailable = "unavailable"
+)
+
+// maxStatusMsgLength caps a status message. The spec sets no limit; this is
+// long enough for a sentence and short enough that it cannot be used as
+// storage.
+const maxStatusMsgLength = 512
+
+// PresenceRequest is PUT /presence/{userID}/status.
+type PresenceRequest struct {
+	// User is the full user ID whose presence is being set, which the
+	// handler has already checked is the caller.
+	User string
+
+	Presence string
+
+	// StatusMsg is the free-text status. An explicitly empty string
+	// clears it, which is why this is a pointer: a request that does not
+	// mention it leaves it alone.
+	StatusMsg *string
+}
+
+// SetPresence records a user's presence.
+func (s *API) SetPresence(ctx context.Context, req PresenceRequest) error {
+	if !s.presenceEnabled {
+		return ErrPresenceDisabled
+	}
+	switch req.Presence {
+	case PresenceOnline, PresenceOffline, PresenceUnavailable:
+	default:
+		return errInvalidParamf("%q is not a presence state", req.Presence)
+	}
+	if req.StatusMsg != nil && len(*req.StatusMsg) > maxStatusMsgLength {
+		return errInvalidParamf("a status message may be at most %d bytes", maxStatusMsgLength)
+	}
+
+	state := storage.PresenceState{
+		UserID:       req.User,
+		Presence:     req.Presence,
+		LastActiveMS: s.nowMS(),
+	}
+	// Keep the status message a request did not mention.
+	if req.StatusMsg != nil {
+		state.StatusMsg = *req.StatusMsg
+	} else {
+		switch existing, err := s.store.Presence(ctx, req.User); {
+		case err == nil:
+			state.StatusMsg = existing.StatusMsg
+		case !errors.Is(err, storage.ErrNoSuchUser):
+			return err
+		}
+	}
+
+	_, moved, err := s.store.SetPresence(ctx, state)
+	if err != nil {
+		return err
+	}
+	// Nothing a peer would notice changed, so there is nobody to wake.
+	if !moved {
+		return nil
+	}
+	return s.notifyPresence(ctx, req.User)
+}
+
+// Presence reads a user's presence, which GET /presence/{userID}/status
+// answers.
+//
+// A user nothing has ever been recorded for is offline rather than unknown.
+// That is what the spec says to report and it is also the truth: a user who
+// has never told the server they were online is not online.
+func (s *API) Presence(ctx context.Context, userID string) (storage.PresenceState, error) {
+	if !s.presenceEnabled {
+		return storage.PresenceState{}, ErrPresenceDisabled
+	}
+	if !s.isLocal(userID) {
+		// Remote presence needs federation (Phase 5). Reporting offline
+		// would be a guess; saying so is not.
+		return storage.PresenceState{}, errForbiddenf(
+			"%s belongs to another server and federation is required to ask it", userID)
+	}
+
+	switch state, err := s.store.Presence(ctx, userID); {
+	case err == nil:
+		return state, nil
+	case errors.Is(err, storage.ErrNoSuchUser):
+		return storage.PresenceState{UserID: userID, Presence: PresenceOffline}, nil
+	default:
+		return storage.PresenceState{}, err
+	}
+}
+
+// notifyPresence wakes the users who share a room with this one.
+//
+// Presence is published to everybody a user shares a room with, which is
+// the set this walks: their rooms, and the members of each. It is the one
+// genuinely fan-out notification in the component, and the reason presence
+// is a switch.
+func (s *API) notifyPresence(ctx context.Context, userID string) error {
+	rooms, err := s.rooms.JoinedRooms(ctx, userID)
+	if err != nil {
+		return err
+	}
+
+	peers := map[string]struct{}{}
+	for _, roomID := range rooms {
+		members, err := s.rooms.JoinedMembers(ctx, roomID, userID)
+		if err != nil {
+			return err
+		}
+		for member := range members {
+			peers[member] = struct{}{}
+		}
+	}
+	delete(peers, userID)
+
+	s.notifier.NotifyUsers(slices.Sorted(maps.Keys(peers)))
+	// Their own other devices want it too.
+	s.notifier.NotifyUser(userID)
+	return nil
+}
+
+// presenceEvents reads the presence of everybody the user shares a room
+// with, between two positions.
+func (s *API) presenceEvents(ctx context.Context, userID string, rooms []string, from, to int64, filter storage.TimelineFilter) ([]presenceEvent, error) {
+	if !s.presenceEnabled || len(rooms) == 0 {
+		return nil, nil
+	}
+
+	peers := map[string]struct{}{userID: {}}
+	for _, roomID := range rooms {
+		entries, err := s.rooms.JoinedMembers(ctx, roomID, userID)
+		if err != nil {
+			// A room the user is no longer joined to is not an error here:
+			// their presence section simply does not include its members.
+			if errors.Is(err, ErrForbidden) {
+				continue
+			}
+			return nil, err
+		}
+		for member := range entries {
+			peers[member] = struct{}{}
+		}
+	}
+
+	states, err := s.store.PresenceForUsers(ctx, slices.Sorted(maps.Keys(peers)), from, to)
+	if err != nil {
+		return nil, err
+	}
+
+	out := make([]presenceEvent, 0, len(states))
+	for _, state := range states {
+		e, err := newPresenceEvent(state, s.nowMS())
+		if err != nil {
+			return nil, err
+		}
+		if !matchesEventFilter(e.Type, e.Sender, filter) {
+			continue
+		}
+		out = append(out, e)
+	}
+	return out, nil
+}
+
+// presenceEvent is the m.presence EDU of one user.
+type presenceEvent struct {
+	Type    string          `json:"type"`
+	Sender  string          `json:"sender"`
+	Content json.RawMessage `json:"content"`
+}
+
+// newPresenceEvent builds the EDU for one user's state.
+//
+// `last_active_ago` is computed on the way out rather than stored, for the
+// same reason `age` is on an event: a value written down at write time
+// would be wrong by however long the row sat in the database.
+func newPresenceEvent(state storage.PresenceState, nowMS int64) (presenceEvent, error) {
+	content := map[string]any{
+		"presence":         state.Presence,
+		"currently_active": state.Presence == PresenceOnline,
+	}
+	if state.StatusMsg != "" {
+		content["status_msg"] = state.StatusMsg
+	}
+	if state.LastActiveMS > 0 {
+		content["last_active_ago"] = max(nowMS-state.LastActiveMS, 0)
+	}
+
+	body, err := json.Marshal(content)
+	if err != nil {
+		return presenceEvent{}, fmt.Errorf("marshalling the presence of %s: %w", state.UserID, err)
+	}
+	return presenceEvent{Type: "m.presence", Sender: state.UserID, Content: body}, nil
+}
+
+// matchesEventFilter is the filter check for the streams whose events are
+// not room events: presence and global account data. Their filters have the
+// same types and senders fields, and exclusions win the same way.
+func matchesEventFilter(eventType, sender string, filter storage.TimelineFilter) bool {
+	switch {
+	case anyMatch(eventType, filter.NotTypes):
+		return false
+	case sender != "" && anyMatch(sender, filter.NotSenders):
+		return false
+	case len(filter.Types) > 0 && !anyMatch(eventType, filter.Types):
+		return false
+	case len(filter.Senders) > 0 && sender != "" && !anyMatch(sender, filter.Senders):
+		return false
+	}
+	return true
+}
+
+// anyMatch applies a list of filter patterns.
+func anyMatch(value string, patterns []string) bool {
+	for _, pattern := range patterns {
+		if matches(value, pattern) {
+			return true
+		}
+	}
+	return false
+}

server/internal/syncapi/receipts.go

@@ -0,0 +1,267 @@
+package syncapi
+
+import (
+	"cmp"
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+
+	"github.com/dariuszjaniak/warp/server/internal/matrix/event"
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// The receipt types the Client-Server API defines.
+const (
+	// ReceiptRead is public: everybody in the room sees it.
+	ReceiptRead = "m.read"
+
+	// ReceiptReadPrivate is the user's own, and is reported only back to
+	// them. It exists so that a user can mark a room read without telling
+	// the room they read it.
+	ReceiptReadPrivate = "m.read.private"
+
+	// ReceiptFullyRead is not a receipt at all: it is the read marker, and
+	// is stored as the room account data type m.fully_read. It is named
+	// here because POST /read_markers takes all three in one body.
+	ReceiptFullyRead = "m.fully_read"
+)
+
+// requireJoined refuses a caller who is not in the room.
+//
+// Every write in this file is a claim about a room — "I am typing in it",
+// "I have read this far" — rather than data belonging to the caller, so
+// being in the room is the condition for all of them.
+func (s *API) requireJoined(ctx context.Context, roomID, userID string) error {
+	membership, err := s.rooms.Membership(ctx, roomID, userID)
+	if err != nil {
+		return err
+	}
+	if membership != event.MembershipJoin {
+		return errForbiddenf("%s is not joined to %s", userID, roomID)
+	}
+	return nil
+}
+
+// ReceiptRequest is POST /rooms/{roomID}/receipt/{receiptType}/{eventID}.
+type ReceiptRequest struct {
+	RoomID string
+
+	// User is the full user ID sending the receipt.
+	User string
+
+	ReceiptType string
+	EventID     string
+
+	// ThreadID is the thread the receipt is for, or empty for the room's
+	// main timeline. Threads are PLAN.md phase 10; the value is stored so
+	// that a client that sends one gets it back, rather than having its
+	// threaded receipt silently collapse onto the main timeline's.
+	ThreadID string
+}
+
+// SetReceipt records a read receipt.
+func (s *API) SetReceipt(ctx context.Context, req ReceiptRequest) error {
+	switch req.ReceiptType {
+	case ReceiptRead, ReceiptReadPrivate, ReceiptFullyRead:
+	default:
+		return errInvalidParamf("%q is not a receipt type this server accepts", req.ReceiptType)
+	}
+
+	if err := s.requireJoined(ctx, req.RoomID, req.User); err != nil {
+		return err
+	}
+
+	// The spec routes m.fully_read through this endpoint too, and it is
+	// account data rather than a receipt — but it is still a claim about a
+	// room the caller has to be in, which is why the check above comes
+	// first.
+	if req.ReceiptType == ReceiptFullyRead {
+		return s.setFullyRead(ctx, req.RoomID, req.User, req.EventID)
+	}
+
+	_, moved, err := s.store.SetReceipt(ctx, storage.Receipt{
+		RoomID:      req.RoomID,
+		UserID:      req.User,
+		ReceiptType: req.ReceiptType,
+		ThreadID:    req.ThreadID,
+		EventID:     req.EventID,
+		TimestampMS: s.nowMS(),
+	})
+	if err != nil {
+		return err
+	}
+
+	// Nothing moved means the user had already read further, and there is
+	// nothing for anybody to be woken about.
+	if !moved {
+		return nil
+	}
+	// A private receipt concerns only its sender; a public one is part of
+	// the room's ephemeral stream and everybody in it wants it.
+	if req.ReceiptType == ReceiptReadPrivate {
+		s.notifier.NotifyUser(req.User)
+		return nil
+	}
+	s.notifier.NotifyRoom(req.RoomID)
+	return nil
+}
+
+// ReadMarkersRequest is POST /rooms/{roomID}/read_markers.
+type ReadMarkersRequest struct {
+	RoomID string
+	User   string
+
+	// FullyRead is the event up to which the user has read, stored as the
+	// room account data m.fully_read. It is where a client draws the
+	// "unread from here" line, which is a different thing from a receipt:
+	// the line stays where the user left it even as they scroll past.
+	FullyRead string
+
+	// Read and ReadPrivate are ordinary receipts, sent in the same request
+	// because a client moving the marker almost always moves both.
+	Read        string
+	ReadPrivate string
+}
+
+// SetReadMarkers implements the read markers endpoint: up to three
+// independent writes from one request.
+//
+// They are not atomic, and cannot usefully be: the marker is account data
+// and the receipts are the receipt stream, which are two streams with two
+// positions. A failure partway through leaves the earlier writes in place,
+// which is the same state a client would reach by sending them separately
+// and losing its connection — and which the next call corrects.
+func (s *API) SetReadMarkers(ctx context.Context, req ReadMarkersRequest) error {
+	if err := s.requireJoined(ctx, req.RoomID, req.User); err != nil {
+		return err
+	}
+
+	if req.FullyRead != "" {
+		if err := s.setFullyRead(ctx, req.RoomID, req.User, req.FullyRead); err != nil {
+			return err
+		}
+	}
+	for _, receipt := range []struct {
+		receiptType string
+		eventID     string
+	}{
+		{ReceiptRead, req.Read},
+		{ReceiptReadPrivate, req.ReadPrivate},
+	} {
+		if receipt.eventID == "" {
+			continue
+		}
+		if err := s.SetReceipt(ctx, ReceiptRequest{
+			RoomID:      req.RoomID,
+			User:        req.User,
+			ReceiptType: receipt.receiptType,
+			EventID:     receipt.eventID,
+		}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// setFullyRead writes the read marker as room account data.
+func (s *API) setFullyRead(ctx context.Context, roomID, userID, eventID string) error {
+	localpart, ok := s.localpartOf(userID)
+	if !ok {
+		return errForbiddenf("%s is not a user of this server", userID)
+	}
+	content, err := json.Marshal(map[string]string{"event_id": eventID})
+	if err != nil {
+		return fmt.Errorf("marshalling the read marker: %w", err)
+	}
+	if err := s.accounts.SetAccountData(ctx, localpart, roomID, ReceiptFullyRead, content); err != nil {
+		return err
+	}
+	s.notifier.NotifyUser(userID)
+	return nil
+}
+
+// receiptsEvent builds the m.receipt EDU of one room.
+//
+// The shape is the spec's, and it is an awkward one: a map of event ID to
+// receipt type to user ID to the receipt's data. It is nested that way
+// because a client's job is to answer "who has read up to this event",
+// which this shape makes a lookup and a flat list would make a scan.
+func receiptsEvent(receipts []storage.Receipt) *EphemeralEvent {
+	if len(receipts) == 0 {
+		return nil
+	}
+
+	type receiptData struct {
+		TS       int64  `json:"ts"`
+		ThreadID string `json:"thread_id,omitempty"`
+	}
+	content := map[string]map[string]map[string]receiptData{}
+	for _, r := range receipts {
+		byType, ok := content[r.EventID]
+		if !ok {
+			byType = map[string]map[string]receiptData{}
+			content[r.EventID] = byType
+		}
+		byUser, ok := byType[r.ReceiptType]
+		if !ok {
+			byUser = map[string]receiptData{}
+			byType[r.ReceiptType] = byUser
+		}
+		byUser[r.UserID] = receiptData{TS: r.TimestampMS, ThreadID: r.ThreadID}
+	}
+
+	body, err := json.Marshal(content)
+	if err != nil {
+		// Every value here is a string or an integer, so this cannot
+		// happen; dropping the EDU beats failing the sync if it does.
+		return nil
+	}
+	return &EphemeralEvent{Type: "m.receipt", Content: body}
+}
+
+// roomReceipts reads the receipts a user is entitled to see in a set of
+// rooms, between two positions.
+//
+// Private receipts are filtered here rather than in SQL: the query is the
+// same one the public receipts come from, and a WHERE clause that had to
+// name the asking user would turn one query per sync into two.
+func (s *API) roomReceipts(ctx context.Context, userID string, roomIDs []string, from, to int64) (map[string][]storage.Receipt, error) {
+	if len(roomIDs) == 0 {
+		return nil, nil
+	}
+	receipts, err := s.store.Receipts(ctx, roomIDs, from, to)
+	if err != nil {
+		return nil, err
+	}
+	// An initial sync also wants the user's own receipts however old they
+	// are: a client that has just logged in has to know where it had read
+	// up to, and that receipt was set long before this sync's window.
+	if from == 0 {
+		own, err := s.store.ReceiptsForUser(ctx, userID, roomIDs)
+		if err != nil {
+			return nil, err
+		}
+		receipts = append(receipts, own...)
+	}
+
+	out := map[string][]storage.Receipt{}
+	seen := map[storage.Receipt]struct{}{}
+	for _, r := range receipts {
+		if r.ReceiptType == ReceiptReadPrivate && r.UserID != userID {
+			continue
+		}
+		if _, dup := seen[r]; dup {
+			// The user's own receipts can arrive from both reads above.
+			continue
+		}
+		seen[r] = struct{}{}
+		out[r.RoomID] = append(out[r.RoomID], r)
+	}
+	for roomID := range out {
+		slices.SortFunc(out[roomID], func(a, b storage.Receipt) int {
+			return cmp.Compare(a.StreamPos, b.StreamPos)
+		})
+	}
+	return out, nil
+}

server/internal/syncapi/sync.go

@@ -0,0 +1,759 @@
+package syncapi
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"slices"
+	"time"
+
+	"github.com/dariuszjaniak/warp/server/internal/matrix/event"
+	"github.com/dariuszjaniak/warp/server/internal/roomserver"
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// 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. The keyserver is PLAN.md phase 7,
+	// so both lists are always empty; the block is here all the same,
+	// because a client that finds no device_lists treats the response as
+	// malformed rather than as "nothing changed".
+	DeviceLists DeviceLists `json:"device_lists"`
+}
+
+// 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"`
+}
+
+// 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, error) {
+	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)
+
+	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{}},
+		// Both lists are empty until Phase 7 and both have to be present.
+		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
+	}
+
+	accountData, err := s.globalAccountData(ctx, req.Localpart, since.AccountData, to.AccountData, 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)
+	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
+
+	resp.NextBatch = next.String()
+	return resp, isInteresting(resp, 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.
+func isInteresting(resp Response, since Token) bool {
+	if since.IsInitial() {
+		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
+}
+
+// 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,
+		}
+
+		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
+}
+
+// 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
+}
+
+// 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 and has no position at all: a notification
+	// that started and stopped between two syncs is simply never reported,
+	// and one that is live is reported every time until it expires.
+	for _, roomID := range s.typing.activeRooms() {
+		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{}{}
+			}
+		}
+	}
+
+	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.since.PDU,
+		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()
+
+	if e := s.typingEvent(roomID); e != nil && matchesEventFilter(e.Type, "", ephemeralFilter) {
+		out = append(out, *e)
+	}
+
+	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.
+func (s *API) globalAccountData(ctx context.Context, localpart string, from, to int64, filter Filter) ([]AccountDataEvent, error) {
+	entries, err := s.accountDataEntries(ctx, localpart, from, to)
+	if err != nil {
+		return nil, err
+	}
+	return s.accountDataEvents(ctx, localpart, 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, 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, localpart string, 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
+		}
+		content, err := s.accounts.AccountData(ctx, 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
+}

server/internal/syncapi/syncapi.go

@@ -0,0 +1,216 @@
+// Package syncapi is the read model: it answers "what has happened that I
+// have not seen", which is the question a Matrix client spends its whole
+// life asking.
+//
+// Everything a client learns after it has logged in arrives through here —
+// /sync for the live stream, /messages for a room's history, /context for
+// one event and what surrounds it. The component writes almost nothing:
+// read receipts, to-device messages and presence are its own, and
+// everything else it serves is the roomserver's, read through the roomserver
+// rather than copied. That is the one real architectural choice in this
+// package, and it is a departure from Dendrite, which gives its syncapi its
+// own tables and feeds them from the output stream. In one process the copy
+// buys nothing and costs the hardest bug in a homeserver: a read model that
+// disagrees with the room.
+//
+// What this package owns instead is everything the roomserver has no
+// opinion about:
+//
+//   - Stream tokens: a position in each of the five streams a response
+//     carries, in one opaque string (token.go).
+//   - The notifier: waking a client blocked in a long poll the moment
+//     something concerns it, rather than when its timeout expires
+//     (notifier.go).
+//   - Filters: what a client asked to be left out (filter.go).
+//   - The ephemeral and device streams: typing, receipts, presence and
+//     to-device messages, none of which are events in a room.
+//
+// Phase 4 of PLAN.md. What it does not do yet is notification counts, which
+// need push rules (Phase 8), and device list tracking, which needs the
+// keyserver (Phase 7); both are named in docs/sync.md as the gaps they are.
+package syncapi
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"time"
+
+	"github.com/dariuszjaniak/warp/server/internal/roomserver"
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// Errors callers have to tell apart. The storage and roomserver sentinels
+// are re-exported rather than translated, so that errors.Is works whichever
+// package a caller reaches for.
+var (
+	ErrNoSuchRoom  = storage.ErrNoSuchRoom
+	ErrNoSuchEvent = storage.ErrNoSuchEvent
+
+	// ErrForbidden is a read the caller is not entitled to: a room they are
+	// not in, or somebody else's filter.
+	ErrForbidden = roomserver.ErrForbidden
+
+	// ErrPresenceDisabled is a presence operation on a server that has
+	// presence switched off.
+	ErrPresenceDisabled = errors.New("presence is disabled on this server")
+
+	// ErrInvalidParam is a request parameter this server will not accept:
+	// a receipt type it does not know, a presence state that is not one.
+	ErrInvalidParam = errors.New("invalid parameter")
+
+	// ErrInvalidToken is a ?since= or ?from= that is not a sync token.
+	ErrInvalidToken = errors.New("invalid sync token")
+)
+
+// Accounts is the part of the userapi the sync API needs: account data to
+// put in a response, and the devices of a user to address a to-device
+// message to. *userapi.API satisfies it as it stands.
+type Accounts interface {
+	AccountData(ctx context.Context, localpart, roomID, dataType string) ([]byte, error)
+	SetAccountData(ctx context.Context, localpart, roomID, dataType string, content []byte) error
+	Devices(ctx context.Context, localpart string) ([]storage.Device, error)
+}
+
+// API is the syncapi component.
+type API struct {
+	serverName string
+	rooms      *roomserver.API
+	accounts   Accounts
+	store      storage.SyncStore
+	notifier   *Notifier
+	typing     *typingTracker
+	log        *slog.Logger
+	now        func() time.Time
+
+	// presenceEnabled is the config switch of PLAN.md open question 3.
+	// With presence off the endpoints answer M_FORBIDDEN and the presence
+	// section of a sync is empty, rather than the endpoints disappearing:
+	// a client that finds no /presence assumes an old server and stops
+	// asking, where a clear refusal tells it this one has it turned off.
+	presenceEnabled bool
+
+	// longPollTimeout caps how long a /sync may block, whatever ?timeout=
+	// the client asked for. A client asking for ten minutes would hold a
+	// connection and a database handle for ten minutes.
+	longPollTimeout time.Duration
+}
+
+// Options configure the component.
+type Options struct {
+	// ServerName is the Matrix server name.
+	ServerName string
+
+	// Rooms is the roomserver, which every room read goes through.
+	Rooms *roomserver.API
+
+	// Accounts is the userapi.
+	Accounts Accounts
+
+	// Store is the sync half of storage.
+	Store storage.SyncStore
+
+	// Notifier wakes blocked syncs. It is the same object the roomserver
+	// was given as its Notifier; nil builds one, which is only useful in a
+	// test where nothing is being written.
+	Notifier *Notifier
+
+	// PresenceEnabled turns the presence stream on.
+	PresenceEnabled bool
+
+	// TypingTimeout is how long a typing notification lasts without being
+	// renewed, whatever timeout the client asked for.
+	TypingTimeout time.Duration
+
+	// LongPollTimeout caps a /sync's ?timeout=.
+	LongPollTimeout time.Duration
+
+	Logger *slog.Logger
+
+	// Now is the clock, so tests can expire a typing notification without
+	// waiting for it.
+	Now func() time.Time
+}
+
+// Defaults for the timeouts.
+const (
+	// DefaultTypingTimeout is the spec's own suggestion, and what every
+	// client sends.
+	DefaultTypingTimeout = 30 * time.Second
+
+	// DefaultLongPollTimeout is the timeout clients ask for. Holding a
+	// connection longer buys nothing: the notifier wakes a sync the moment
+	// anything happens, so the timeout only decides how often an idle
+	// client re-establishes its request.
+	DefaultLongPollTimeout = 30 * time.Second
+
+	// MaxTypingTimeout caps what a client may ask for. A typing
+	// notification that outlived its sender's session would leave a room
+	// permanently claiming somebody was typing.
+	MaxTypingTimeout = 2 * time.Minute
+)
+
+// New builds the component.
+func New(opts Options) (*API, error) {
+	switch {
+	case opts.ServerName == "":
+		return nil, errors.New("syncapi: ServerName must not be empty")
+	case opts.Rooms == nil:
+		return nil, errors.New("syncapi: Rooms must not be nil")
+	case opts.Store == nil:
+		return nil, errors.New("syncapi: Store must not be nil")
+	case opts.Accounts == nil:
+		return nil, errors.New("syncapi: Accounts must not be nil")
+	}
+
+	if opts.Notifier == nil {
+		opts.Notifier = NewNotifier()
+	}
+	if opts.TypingTimeout <= 0 {
+		opts.TypingTimeout = DefaultTypingTimeout
+	}
+	if opts.TypingTimeout > MaxTypingTimeout {
+		opts.TypingTimeout = MaxTypingTimeout
+	}
+	if opts.LongPollTimeout <= 0 {
+		opts.LongPollTimeout = DefaultLongPollTimeout
+	}
+	if opts.Logger == nil {
+		opts.Logger = slog.Default()
+	}
+	if opts.Now == nil {
+		opts.Now = time.Now
+	}
+
+	return &API{
+		serverName:      opts.ServerName,
+		rooms:           opts.Rooms,
+		accounts:        opts.Accounts,
+		store:           opts.Store,
+		notifier:        opts.Notifier,
+		typing:          newTypingTracker(opts.TypingTimeout, opts.Notifier, opts.Now),
+		log:             opts.Logger,
+		now:             opts.Now,
+		presenceEnabled: opts.PresenceEnabled,
+		longPollTimeout: opts.LongPollTimeout,
+	}, nil
+}
+
+// Notifier returns the notifier, which the roomserver is wired to as its
+// emit stage.
+func (s *API) Notifier() *Notifier { return s.notifier }
+
+// PresenceEnabled reports whether the presence stream is on, which
+// /capabilities and the admin API both report.
+func (s *API) PresenceEnabled() bool { return s.presenceEnabled }
+
+// LongPollTimeout is the cap a /sync's ?timeout= is held to.
+func (s *API) LongPollTimeout() time.Duration { return s.longPollTimeout }
+
+// nowMS is the current time in the unit the API uses on the wire.
+func (s *API) nowMS() int64 { return s.now().UnixMilli() }
+
+// Close releases what the component started. The typing tracker holds a
+// timer per typing user, and a server shutting down should not wait for
+// them.
+func (s *API) Close() { s.typing.close() }

server/internal/syncapi/todevice.go

@@ -0,0 +1,176 @@
+package syncapi
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// To-device messaging: a message from one device to another, which the
+// server stores and forwards without reading.
+//
+// It exists for end-to-end encryption — a device sending another one a
+// Megolm session key is a to-device message — and the server deliberately
+// cannot tell what is in one. Phase 7 builds the keyserver on top of this;
+// the transport itself belongs here, because the only way a device ever
+// receives one is in a /sync response.
+
+// wildcardDevice is the device ID that means "every device of this user".
+const wildcardDevice = "*"
+
+// SendToDeviceRequest is PUT /sendToDevice/{eventType}/{txnID}.
+type SendToDeviceRequest struct {
+	// Sender is the full user ID sending, which is the caller.
+	Sender string
+
+	EventType string
+
+	// Messages is user ID -> device ID -> content, as the request body
+	// nests it. A device ID of "*" addresses every device of that user.
+	Messages map[string]map[string]json.RawMessage
+}
+
+// SendToDevice queues messages for delivery.
+//
+// A message to a user of another server is dropped rather than refused.
+// That is deliberate: a client encrypting for a room with remote members
+// sends one message per device in one request, and failing the request
+// because federation is off would stop it from delivering to the local
+// devices as well. The dropped recipients simply never see the key and
+// their client reports an undecryptable message, which is what they would
+// see anyway. Phase 5 routes them.
+func (s *API) SendToDevice(ctx context.Context, req SendToDeviceRequest) error {
+	if req.EventType == "" {
+		return errInvalidParamf("an event type is required")
+	}
+
+	var (
+		queued []storage.NewToDeviceMessage
+		woken  []string
+		sentAt = s.nowMS()
+	)
+	for userID, byDevice := range req.Messages {
+		localpart, ok := s.localpartOf(userID)
+		if !ok {
+			s.log.Debug("dropping a to-device message for a remote user",
+				"sender", req.Sender, "recipient", userID)
+			continue
+		}
+
+		for deviceID, content := range byDevice {
+			if deviceID != wildcardDevice {
+				queued = append(queued, storage.NewToDeviceMessage{
+					Localpart: localpart,
+					DeviceID:  deviceID,
+					Sender:    req.Sender,
+					Type:      req.EventType,
+					Content:   contentOrEmpty(content),
+					SentAtMS:  sentAt,
+				})
+				continue
+			}
+
+			// A wildcard is expanded here rather than stored as one: the
+			// set of a user's devices changes, and a message addressed to
+			// "every device" means the ones that existed when it was sent.
+			// Storing the wildcard would deliver it to a device that was
+			// created afterwards, which for a key share is both useless
+			// and a leak.
+			devices, err := s.accounts.Devices(ctx, localpart)
+			if err != nil {
+				return err
+			}
+			for _, device := range devices {
+				queued = append(queued, storage.NewToDeviceMessage{
+					Localpart: localpart,
+					DeviceID:  device.ID,
+					Sender:    req.Sender,
+					Type:      req.EventType,
+					Content:   contentOrEmpty(content),
+					SentAtMS:  sentAt,
+				})
+			}
+		}
+		woken = append(woken, userID)
+	}
+
+	if len(queued) == 0 {
+		return nil
+	}
+	if _, err := s.store.SendToDevice(ctx, queued); err != nil {
+		return err
+	}
+	s.notifier.NotifyUsers(woken)
+	return nil
+}
+
+// contentOrEmpty keeps a null or absent content out of the store: the
+// to-device content is passed through untouched, and "untouched" has to
+// still be valid JSON on the way out.
+func contentOrEmpty(content json.RawMessage) []byte {
+	if len(content) == 0 {
+		return []byte("{}")
+	}
+	return content
+}
+
+// toDeviceEvent is one message in the shape a sync reports it.
+type toDeviceEvent struct {
+	Sender  string          `json:"sender"`
+	Type    string          `json:"type"`
+	Content json.RawMessage `json:"content"`
+}
+
+// toDeviceMessages reads a device's queue between two positions and returns
+// the events plus the position of the last one.
+//
+// The returned position is what the next token carries, and it is the
+// position of the last message *delivered* rather than the top of the
+// stream: a queue longer than one batch is drained over several syncs, and
+// a token naming the top would skip the rest.
+func (s *API) toDeviceMessages(ctx context.Context, localpart, deviceID string, from, to int64) ([]toDeviceEvent, int64, error) {
+	if deviceID == "" {
+		// An access token always names a device, so this cannot happen
+		// through the API; a guest session later might.
+		return nil, from, nil
+	}
+
+	msgs, err := s.store.ToDeviceMessages(ctx, localpart, deviceID, from, to, 0)
+	if err != nil {
+		return nil, 0, err
+	}
+	if len(msgs) == 0 {
+		return nil, from, nil
+	}
+
+	out := make([]toDeviceEvent, 0, len(msgs))
+	for _, m := range msgs {
+		out = append(out, toDeviceEvent{
+			Sender:  m.Sender,
+			Type:    m.Type,
+			Content: m.Content,
+		})
+	}
+	return out, msgs[len(msgs)-1].StreamPos, nil
+}
+
+// ackToDevice deletes the messages a device has acknowledged by syncing
+// past them.
+//
+// A device acknowledges a batch by sending back the token that batch came
+// with, so the deletion happens on the *next* sync rather than when the
+// batch is sent. A client that crashes between the two gets the batch
+// again, which is the right way round: a lost key share leaves a message
+// permanently undecryptable, and a duplicate one is ignored.
+func (s *API) ackToDevice(ctx context.Context, localpart, deviceID string, through int64) {
+	if deviceID == "" || through <= 0 {
+		return
+	}
+	if err := s.store.DeleteToDeviceMessages(ctx, localpart, deviceID, through); err != nil {
+		// Failing the sync over this would be worse than leaving the rows:
+		// the client has the messages, and the next sync will try again.
+		s.log.Warn("could not drop acknowledged to-device messages",
+			"localpart", localpart, "device_id", deviceID, "through", through, "error", err)
+	}
+}

server/internal/syncapi/token.go

@@ -0,0 +1,132 @@
+package syncapi
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/dariuszjaniak/warp/server/internal/storage"
+)
+
+// Token is a position in every stream a sync response carries.
+//
+// A sync response is assembled from streams that move independently: room
+// events, read receipts, to-device messages, account data and presence. A
+// client that has seen receipt 7 but only event 4 has to be able to say so
+// in the one opaque string the API gives it, so the token names a position
+// in each.
+//
+// The wire form is
+//
+//	s{pdu}_{receipt}_{to_device}_{account_data}_{presence}
+//
+// with trailing zeroes left off, so a client that has only ever seen room
+// events carries "s12" rather than "s12_0_0_0_0". The leading "s" is there
+// to make a token recognisable in a log and to leave room for a second
+// token shape later — Simplified Sliding Sync (PLAN.md phase 10) has its
+// own — without any ambiguity about which is which.
+//
+// The spec says only that the token is opaque, and clients treat it that
+// way. It is still worth being readable: a token is the first thing in
+// every bug report about /sync.
+type Token struct {
+	PDU         int64
+	Receipt     int64
+	ToDevice    int64
+	AccountData int64
+	Presence    int64
+}
+
+// tokenPrefix marks a classic /sync token.
+const tokenPrefix = "s"
+
+// IsInitial reports whether the token is the start of every stream, which
+// is what a sync with no ?since= means.
+func (t Token) IsInitial() bool { return t == Token{} }
+
+// String renders the wire form.
+func (t Token) String() string {
+	parts := []int64{t.PDU, t.Receipt, t.ToDevice, t.AccountData, t.Presence}
+	// Trailing zeroes carry no information, and leaving them off keeps the
+	// common token short.
+	last := 0
+	for i, p := range parts {
+		if p != 0 {
+			last = i
+		}
+	}
+
+	var b strings.Builder
+	b.WriteString(tokenPrefix)
+	for i, p := range parts[:last+1] {
+		if i > 0 {
+			b.WriteByte('_')
+		}
+		b.WriteString(strconv.FormatInt(p, 10))
+	}
+	return b.String()
+}
+
+// ParseToken reads a token a client sent back.
+//
+// An empty string is the initial token: /sync without ?since= and
+// /messages without ?from= both mean "from the beginning", and refusing
+// the empty string would make every caller special-case it.
+//
+// A token with fewer fields than this build writes is accepted and the
+// missing ones are zero. That is what lets a token stored by a client
+// before a stream existed go on working: the stream it did not know about
+// reads as "seen nothing", so the client is sent that stream in full once
+// and is in step from then on.
+func ParseToken(raw string) (Token, error) {
+	if raw == "" {
+		return Token{}, nil
+	}
+	body, ok := strings.CutPrefix(raw, tokenPrefix)
+	if !ok {
+		return Token{}, fmt.Errorf("%q is not a sync token: it must start with %q", raw, tokenPrefix)
+	}
+
+	fields := strings.Split(body, "_")
+	if len(fields) > 5 {
+		return Token{}, fmt.Errorf("%q is not a sync token: it has %d positions, this server writes at most 5",
+			raw, len(fields))
+	}
+
+	var positions [5]int64
+	for i, field := range fields {
+		value, err := strconv.ParseInt(field, 10, 64)
+		if err != nil || value < 0 {
+			return Token{}, fmt.Errorf("%q is not a sync token: position %d is %q", raw, i+1, field)
+		}
+		positions[i] = value
+	}
+	return Token{
+		PDU:         positions[0],
+		Receipt:     positions[1],
+		ToDevice:    positions[2],
+		AccountData: positions[3],
+		Presence:    positions[4],
+	}, nil
+}
+
+// tokenAt is the token naming the top of every stream, which is what a
+// sync response hands back when it has drained all of them.
+func tokenAt(positions storage.StreamPositions) Token {
+	return Token{
+		PDU:         positions.PDU,
+		Receipt:     positions.Receipt,
+		ToDevice:    positions.ToDevice,
+		AccountData: positions.AccountData,
+		Presence:    positions.Presence,
+	}
+}
+
+// pduToken is the token for a position in the room event stream alone.
+//
+// It is what /messages and a room's prev_batch hand out: both are about
+// room events only, and a token from either has to be usable as a ?since=
+// for /sync, where the other streams then read as unseen. That is the
+// conservative direction — the client is sent those streams again rather
+// than having them skipped.
+func pduToken(pos int64) Token { return Token{PDU: pos} }

server/internal/syncapi/token_test.go

@@ -0,0 +1,99 @@
+package syncapi_test
+
+import (
+	"testing"
+
+	"github.com/dariuszjaniak/warp/server/internal/syncapi"
+)
+
+func TestTokenRoundTrips(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		token syncapi.Token
+		want  string
+	}{
+		{"the initial token", syncapi.Token{}, "s0"},
+		{"events only", syncapi.Token{PDU: 12}, "s12"},
+		{"events and receipts", syncapi.Token{PDU: 12, Receipt: 3}, "s12_3"},
+		{
+			"every stream",
+			syncapi.Token{PDU: 1, Receipt: 2, ToDevice: 3, AccountData: 4, Presence: 5},
+			"s1_2_3_4_5",
+		},
+		{
+			// A zero in the middle has to be written, because the
+			// positions are read by place.
+			"a gap in the middle",
+			syncapi.Token{PDU: 1, Presence: 5},
+			"s1_0_0_0_5",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := tc.token.String(); got != tc.want {
+				t.Fatalf("String() = %q, want %q", got, tc.want)
+			}
+			parsed, err := syncapi.ParseToken(tc.want)
+			if err != nil {
+				t.Fatalf("ParseToken(%q): %v", tc.want, err)
+			}
+			if parsed != tc.token {
+				t.Errorf("ParseToken(%q) = %+v, want %+v", tc.want, parsed, tc.token)
+			}
+		})
+	}
+}
+
+func TestParseTokenAcceptsAnEmptyString(t *testing.T) {
+	// /sync without ?since= and /messages without ?from= both mean "from
+	// the beginning"; refusing the empty string would make every caller
+	// special-case it.
+	parsed, err := syncapi.ParseToken("")
+	if err != nil {
+		t.Fatalf("ParseToken(\"\"): %v", err)
+	}
+	if !parsed.IsInitial() {
+		t.Errorf("ParseToken(\"\") = %+v, want the initial token", parsed)
+	}
+}
+
+func TestParseTokenAcceptsAShortToken(t *testing.T) {
+	// A token a client stored before a stream existed must go on working:
+	// the stream it does not mention reads as "seen nothing", so the
+	// client is sent that stream in full once and is in step afterwards.
+	parsed, err := syncapi.ParseToken("s7")
+	if err != nil {
+		t.Fatalf("ParseToken(\"s7\"): %v", err)
+	}
+	if parsed.PDU != 7 {
+		t.Errorf("PDU = %d, want 7", parsed.PDU)
+	}
+	if parsed.Receipt != 0 || parsed.ToDevice != 0 || parsed.AccountData != 0 || parsed.Presence != 0 {
+		t.Errorf("the streams a short token omits must read as zero: %+v", parsed)
+	}
+}
+
+func TestParseTokenRejectsNonsense(t *testing.T) {
+	for _, raw := range []string{
+		"nonsense",     // no prefix
+		"t1_2",         // a prefix this build does not write
+		"s",            // a prefix and nothing else
+		"s-1",          // a negative position
+		"sx",           // not a number
+		"s1_",          // a trailing separator
+		"s1_2_3_4_5_6", // more positions than there are streams
+		"s1.5",
+	} {
+		if _, err := syncapi.ParseToken(raw); err == nil {
+			t.Errorf("ParseToken(%q) was accepted", raw)
+		}
+	}
+}
+
+func TestParseTokenIsNotFooledByOverflow(t *testing.T) {
+	// A position larger than an int64 is not a position this server ever
+	// handed out, and silently wrapping it would mean a client being sent
+	// the whole of every room again.
+	if _, err := syncapi.ParseToken("s99999999999999999999"); err == nil {
+		t.Error("a position that does not fit in an int64 was accepted")
+	}
+}

server/internal/syncapi/typing.go

@@ -0,0 +1,220 @@
+package syncapi
+
+import (
+	"context"
+	"encoding/json"
+	"slices"
+	"sync"
+	"time"
+)
+
+// Typing notifications are held in memory and nowhere else.
+//
+// They expire in seconds, they are worthless after a restart — nobody is
+// still typing the message they were typing when the server went down —
+// and they arrive one per keystroke-burst per room. Writing them to disk
+// would be the busiest table in the database, holding the least valuable
+// data in it. Dendrite reaches the same conclusion.
+//
+// The cost of the choice is that a restart clears them, which a client
+// cannot tell from everybody having stopped typing, and that a second
+// process would not see them. The second only matters when Warp stops
+// being one process, which PLAN.md says is not the plan.
+
+// typingTracker holds who is typing in which room, and expires them.
+type typingTracker struct {
+	timeout  time.Duration
+	notifier *Notifier
+	now      func() time.Time
+
+	mu sync.Mutex
+	// rooms is room ID -> user ID -> when the notification expires.
+	rooms map[string]map[string]time.Time
+	// timers hold the expiry wake-ups, so that a notification running out
+	// reaches a blocked sync rather than waiting for one to ask.
+	timers map[typingKey]*time.Timer
+	closed bool
+}
+
+type typingKey struct {
+	roomID string
+	userID string
+}
+
+func newTypingTracker(timeout time.Duration, notifier *Notifier, now func() time.Time) *typingTracker {
+	return &typingTracker{
+		timeout:  timeout,
+		notifier: notifier,
+		now:      now,
+		rooms:    map[string]map[string]time.Time{},
+		timers:   map[typingKey]*time.Timer{},
+	}
+}
+
+// set records that a user is typing in a room for a while, or stops them.
+//
+// The timeout a client asks for is honoured up to the configured cap: a
+// client that asks for an hour would leave the room claiming somebody was
+// typing long after they closed it.
+func (t *typingTracker) set(roomID, userID string, typing bool, timeout time.Duration) {
+	t.mu.Lock()
+	if t.closed {
+		t.mu.Unlock()
+		return
+	}
+
+	key := typingKey{roomID: roomID, userID: userID}
+	if timer, ok := t.timers[key]; ok {
+		timer.Stop()
+		delete(t.timers, key)
+	}
+
+	switch {
+	case !typing:
+		if users, ok := t.rooms[roomID]; ok {
+			delete(users, userID)
+			if len(users) == 0 {
+				delete(t.rooms, roomID)
+			}
+		}
+	default:
+		if timeout <= 0 || timeout > t.timeout {
+			timeout = t.timeout
+		}
+		users, ok := t.rooms[roomID]
+		if !ok {
+			users = map[string]time.Time{}
+			t.rooms[roomID] = users
+		}
+		users[userID] = t.now().Add(timeout)
+		t.timers[key] = time.AfterFunc(timeout, func() { t.expire(key) })
+	}
+	t.mu.Unlock()
+
+	// Outside the lock: the notifier takes its own, and a sync woken by it
+	// will want to read this tracker.
+	t.notifier.NotifyRoom(roomID)
+}
+
+// expire drops a notification whose timer has fired and wakes the room.
+func (t *typingTracker) expire(key typingKey) {
+	t.mu.Lock()
+	delete(t.timers, key)
+	users, ok := t.rooms[key.roomID]
+	if ok {
+		delete(users, key.userID)
+		if len(users) == 0 {
+			delete(t.rooms, key.roomID)
+		}
+	}
+	t.mu.Unlock()
+
+	if ok {
+		t.notifier.NotifyRoom(key.roomID)
+	}
+}
+
+// users lists who is typing in a room, sorted, skipping anyone whose
+// notification has run out.
+//
+// The expiry check is here as well as in the timer because the two answer
+// different failure modes: the timer wakes a blocked client, and this makes
+// sure a notification is never *reported* after its deadline even if the
+// timer has not run yet.
+func (t *typingTracker) users(roomID string) []string {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	now := t.now()
+	out := make([]string, 0, len(t.rooms[roomID]))
+	for userID, expires := range t.rooms[roomID] {
+		if expires.After(now) {
+			out = append(out, userID)
+		}
+	}
+	slices.Sort(out)
+	return out
+}
+
+// rooms with anybody typing, which is how a sync knows a room has an
+// ephemeral change without asking per room.
+func (t *typingTracker) activeRooms() []string {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	now := t.now()
+	out := make([]string, 0, len(t.rooms))
+	for roomID, users := range t.rooms {
+		for _, expires := range users {
+			if expires.After(now) {
+				out = append(out, roomID)
+				break
+			}
+		}
+	}
+	slices.Sort(out)
+	return out
+}
+
+// close stops every timer. A server shutting down should not be held up by
+// a typing notification.
+func (t *typingTracker) close() {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	t.closed = true
+	for key, timer := range t.timers {
+		timer.Stop()
+		delete(t.timers, key)
+	}
+	t.rooms = map[string]map[string]time.Time{}
+}
+
+// TypingRequest is PUT /rooms/{roomID}/typing/{userID}.
+type TypingRequest struct {
+	RoomID string
+
+	// User is the full user ID said to be typing, which the handler has
+	// already checked is the caller.
+	User string
+
+	Typing bool
+
+	// Timeout is how long the notification should last. Zero takes the
+	// server's own.
+	Timeout time.Duration
+}
+
+// SetTyping records a typing notification.
+//
+// Being in the room is required: a typing notification is published to
+// everybody in it, so it is a write to the room even though it is not an
+// event in it.
+func (s *API) SetTyping(ctx context.Context, req TypingRequest) error {
+	if err := s.requireJoined(ctx, req.RoomID, req.User); err != nil {
+		return err
+	}
+
+	s.typing.set(req.RoomID, req.User, req.Typing, req.Timeout)
+	return nil
+}
+
+// typingEvent is the m.typing EDU of a room, or nil when nobody is typing.
+//
+// The event is the whole list rather than a delta, which is what the spec
+// defines: a client replaces what it had, so a missed update corrects
+// itself on the next one.
+func (s *API) typingEvent(roomID string) *EphemeralEvent {
+	users := s.typing.users(roomID)
+	if len(users) == 0 {
+		return nil
+	}
+	content, err := json.Marshal(map[string][]string{"user_ids": users})
+	if err != nil {
+		// Marshalling a list of strings cannot fail; if it somehow did,
+		// dropping the notification is better than failing the sync.
+		s.log.Error("marshalling a typing notification failed", "room_id", roomID, "error", err)
+		return nil
+	}
+	return &EphemeralEvent{Type: "m.typing", Content: content}
+}