Warp/server/internal/roomserver/stream.go

package roomserver

import (
	"context"
	"encoding/json"
	"fmt"
	"slices"

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

// The read model: the event stream as /sync, /messages and /context read
// it.
//
// It is here rather than in the syncapi because reading an event means
// parsing it under its room's version, and that is the roomserver's
// business — the syncapi would otherwise need its own copy of the room
// version registry, the event parser and the client projection, and a
// second copy of those would be a second place for them to disagree. What
// the syncapi gets instead is windows of ClientEvents with their positions,
// and it assembles the response shape around them.

// MembershipAt is a user's current membership in a room, with the position
// of the event that set it.
type MembershipAt struct {
	RoomID     string
	Membership string
	EventID    string

	// Position is the stream position of the membership event.
	Position int64
}

// TimelineRequest is one window of a room's event stream.
//
// The bounds are the store's: Low exclusive, High inclusive, both
// directions, and zero meaning unbounded at that end. Either can be
// negative, inside history backfilled from other servers. Direction decides
// only which end the limit trims and the order the store reads in — the
// events always come back oldest first.
type TimelineRequest struct {
	RoomID string

	// User is the full user ID the window is read on behalf of. The result
	// is clamped to what their membership and the room's history
	// visibility allow.
	User string

	Low  int64
	High int64

	Direction storage.TimelineDirection

	// Limit caps how many events come back.
	Limit int

	// Filter narrows which events are read. Applying it here rather than
	// after the limit is what keeps a filtered page from coming back short
	// and looking like the end of the room.
	Filter storage.TimelineFilter

	// Backfill allows a backwards read that runs out of history at the
	// bottom of what this server holds to ask the servers in the room for
	// more before it answers (see backfill.go). /messages sets it; /sync
	// and /context do not, because neither is a client asking to scroll
	// back, and neither should wait on another server.
	Backfill bool
}

// Timeline is a window of a room's event stream.
type Timeline struct {
	// Events are in stream order, oldest first, whichever direction was
	// asked for: a client renders a timeline forwards even when it paged
	// backwards to get it.
	Events []ClientEvent

	// Start and End are the positions bounding the window: Start below the
	// first event and End at the last. They are the two tokens a client
	// needs to ask for the next page in either direction, and either can
	// be handed back as a bound without re-reading the event it names.
	Start int64
	End   int64

	// Limited reports that the limit cut the window short, so there is
	// more of it beyond the end the limit trimmed.
	Limited bool
}

// StreamTop returns the top of this server's event stream.
func (r *API) StreamTop(ctx context.Context) (int64, error) {
	return r.store.MaxStreamOrdering(ctx)
}

// UserMemberships returns the rooms a user has one of the given memberships
// in, with the position of each membership event.
func (r *API) UserMemberships(ctx context.Context, userID string, memberships []string) ([]MembershipAt, error) {
	stored, err := r.store.MembershipsForUser(ctx, userID, memberships)
	if err != nil {
		return nil, err
	}
	out := make([]MembershipAt, 0, len(stored))
	for _, m := range stored {
		out = append(out, MembershipAt{
			RoomID:     m.RoomID,
			Membership: m.Membership,
			EventID:    m.EventID,
			Position:   m.StreamOrdering,
		})
	}
	return out, nil
}

// RoomsWithEvents returns the subset of rooms that have an event after from
// and at or before to.
func (r *API) RoomsWithEvents(ctx context.Context, roomIDs []string, from, to int64) ([]string, error) {
	return r.store.RoomsWithEvents(ctx, roomIDs, from, to)
}

// Timeline reads one window of a room's event stream, as far as the user is
// allowed to see it.
func (r *API) Timeline(ctx context.Context, req TimelineRequest) (Timeline, error) {
	room, rules, err := r.roomAndRules(ctx, req.RoomID)
	if err != nil {
		return Timeline{}, err
	}
	visible, err := r.requireReadable(ctx, room.ID, req.User)
	if err != nil {
		return Timeline{}, err
	}
	page, err := r.timeline(ctx, rules, visible, req)
	if err != nil {
		return Timeline{}, err
	}

	// A page that came back short at the bottom of the room is either the
	// beginning of the room or the beginning of what this server holds of
	// it. Telling a client it has reached the end — no `end` token — when
	// it is the second would stop it scrolling for good, so the servers in
	// the room are asked first.
	for range backfillRounds {
		if !r.wantsBackfill(req, visible, page) {
			break
		}
		added, err := r.backfill(ctx, room, rules, req.Limit)
		if err != nil {
			// The page this server has is still an answer. The client
			// will try again the next time it scrolls up.
			r.log.Info("could not backfill a room", "room_id", room.ID, "error", err)
			break
		}
		if added == 0 {
			break
		}
		if page, err = r.timeline(ctx, rules, visible, req); err != nil {
			return Timeline{}, err
		}
	}
	return page, nil
}

// timeline is Timeline with the room's rules and the user's visible range
// already in hand, for the sync loop that reads many rooms at once.
func (r *API) timeline(ctx context.Context, rules roomversion.Rules, visible VisibleRange, req TimelineRequest) (Timeline, error) {
	// An empty window is reported at the bound the caller came in on, so
	// that the tokens it hands back are still usable.
	empty := Timeline{Start: req.Low, End: req.Low}
	if req.Direction == storage.DirectionBackward && req.High != 0 {
		empty = Timeline{Start: req.High, End: req.High}
	}
	if req.Limit <= 0 {
		return empty, nil
	}

	low, high, any := visible.Clamp(req.Low, req.High)
	if !any {
		return empty, nil
	}

	// One more than asked for, so that "there is more" can be told from
	// "that was all" without a second query.
	stored, err := r.store.Timeline(ctx, storage.TimelineRequest{
		RoomID:    req.RoomID,
		Low:       low,
		High:      high,
		Direction: req.Direction,
		Limit:     req.Limit + 1,
		Filter:    req.Filter,
	})
	if err != nil {
		return Timeline{}, err
	}

	out := empty
	if len(stored) > req.Limit {
		out.Limited = true
		stored = stored[:req.Limit]
	}
	if len(stored) == 0 {
		return out, nil
	}

	// The store returned them in the direction asked for; a client always
	// renders a timeline oldest first.
	if req.Direction == storage.DirectionBackward {
		slices.Reverse(stored)
	}

	out.Events = make([]ClientEvent, 0, len(stored))
	for _, s := range stored {
		e, err := parseEvent(s, rules)
		if err != nil {
			return Timeline{}, err
		}
		out.Events = append(out.Events, r.clientEvent(s, e, req.RoomID))
	}
	if err := r.bundleAggregations(ctx, req.RoomID, rules, req.User, visible, stored, out.Events); err != nil {
		return Timeline{}, err
	}
	// Start sits below the oldest event, so that handing it back as a High
	// bound returns the page before this one; End is at the newest, so that
	// handing it back as a Low bound returns the page after. Below the
	// event at 1 is -1, not 0: zero would read as "unbounded" and hand the
	// client the whole room again rather than the history beneath it.
	out.Start = storage.PositionBefore(stored[0].StreamOrdering)
	out.End = stored[len(stored)-1].StreamOrdering
	return out, nil
}

// EventPosition returns an event's position in the stream, for a user who
// may see it.
//
// /context needs it: the timeline reads are bounded by position rather than
// by event ID, so the event a context is centred on has to be located
// before either side of it can be read.
func (r *API) EventPosition(ctx context.Context, roomID, userID, eventID string) (int64, error) {
	visible, err := r.requireReadable(ctx, roomID, userID)
	if err != nil {
		return 0, err
	}
	stored, err := r.store.Event(ctx, roomID, eventID)
	if err != nil {
		return 0, err
	}
	if !visible.Contains(stored.StreamOrdering) {
		return 0, fmt.Errorf("event %q: %w", eventID, ErrNoSuchEvent)
	}
	return stored.StreamOrdering, nil
}

// StateAt returns a room's state as of a stream position, in the client
// shape.
//
// It is what a sync with a gap in it and a leaver's state read both need:
// the state current state cannot give, because the room has moved on since.
func (r *API) StateAt(ctx context.Context, roomID, userID string, pos int64) ([]ClientEvent, error) {
	room, rules, err := r.roomAndRules(ctx, roomID)
	if err != nil {
		return nil, err
	}
	visible, err := r.requireReadable(ctx, room.ID, userID)
	if err != nil {
		return nil, err
	}
	if visible.To != 0 && (pos == 0 || pos > visible.To) {
		pos = visible.To
	}
	return r.stateAt(ctx, roomID, rules, pos)
}

// stateAt is StateAt with the checks already done.
//
// A position of zero is the room's current state, which is both cheaper to
// read and what every caller that is up to date wants.
func (r *API) stateAt(ctx context.Context, roomID string, rules roomversion.Rules, pos int64) ([]ClientEvent, error) {
	var (
		entries []storage.StateEntry
		err     error
	)
	if pos == 0 {
		stored, err := r.store.CurrentStateEvents(ctx, roomID)
		if err != nil {
			return nil, err
		}
		return r.clientEvents(stored, rules, roomID)
	}
	if entries, err = r.store.StateAtStream(ctx, roomID, pos); err != nil {
		return nil, err
	}
	return r.clientEventsByID(ctx, roomID, rules, stateEntryIDs(entries))
}

// StateBeforeEvent returns the state of a room as of an event's
// prev_events: the state /context reports alongside it.
func (r *API) StateBeforeEvent(ctx context.Context, roomID, userID, eventID string) ([]ClientEvent, error) {
	room, rules, err := r.roomAndRules(ctx, roomID)
	if err != nil {
		return nil, err
	}
	visible, err := r.requireReadable(ctx, room.ID, userID)
	if err != nil {
		return nil, err
	}

	stored, err := r.store.Event(ctx, roomID, eventID)
	if err != nil {
		return nil, err
	}
	if !visible.Contains(stored.StreamOrdering) {
		return nil, fmt.Errorf("%w: %s may not see %s", ErrForbidden, userID, eventID)
	}

	entries, err := r.store.StateBefore(ctx, roomID, eventID)
	if err != nil {
		return nil, err
	}
	state, err := r.clientEventsByID(ctx, roomID, rules, stateEntryIDs(entries))
	if err != nil {
		return nil, err
	}
	// The event itself, when it is a state event, is part of the state
	// after it but not before it. /context wants the state the event was
	// sent against, so it is left out — except that a client rendering the
	// event needs the sender's member event, which is in the snapshot
	// already.
	return state, nil
}

// MemberEvents returns the m.room.member events of specific users in a
// room, in the client shape.
//
// Lazy-loading is what this is for: a client that asked for it is sent the
// members who appear in the timeline it was given, rather than all of them.
func (r *API) MemberEvents(ctx context.Context, roomID string, userIDs []string) ([]ClientEvent, error) {
	if len(userIDs) == 0 {
		return nil, nil
	}
	_, rules, err := r.roomAndRules(ctx, roomID)
	if err != nil {
		return nil, err
	}
	return r.memberEvents(ctx, roomID, rules, userIDs)
}

// memberEvents is MemberEvents with the rules in hand.
func (r *API) memberEvents(ctx context.Context, roomID string, rules roomversion.Rules, userIDs []string) ([]ClientEvent, error) {
	out := make([]ClientEvent, 0, len(userIDs))
	for _, userID := range userIDs {
		stored, err := r.store.CurrentStateEvent(ctx, roomID, event.TypeMember, userID)
		switch {
		case isNotFound(err):
			// A member who appears in the timeline but is not in the
			// room's current state: their membership was reset by state
			// resolution, or the event is old enough to predate it. Not an
			// error, and nothing to send.
			continue
		case err != nil:
			return nil, err
		}
		e, err := parseEvent(stored, rules)
		if err != nil {
			return nil, err
		}
		out = append(out, r.clientEvent(stored, e, roomID))
	}
	return out, nil
}

// StrippedEvent is a piece of room state as it travels with an invite:
// enough for a client to show the user what they are being asked to join,
// and nothing that would let them read the room.
//
// It has no event ID and no timestamp on purpose — the spec's "stripped
// state" is explicitly not an event, and a client must not put it in a
// timeline.
type StrippedEvent struct {
	Type     string          `json:"type"`
	StateKey string          `json:"state_key"`
	Sender   string          `json:"sender"`
	Content  json.RawMessage `json:"content"`
}

// strippedStateTypes is what the spec says an invite carries: the state a
// client needs to render the room in an invite list.
var strippedStateTypes = []string{
	event.TypeCreate,
	event.TypeJoinRules,
	TypeCanonicalAlias,
	TypeName,
	TypeAvatar,
	TypeTopic,
	TypeEncryption,
	event.TypeHistoryVisibility,
}

// StrippedState returns the stripped state of a room, plus the member
// events of the users named.
//
// The member events are what makes an invite legible: a client showing
// "Alice invited you" needs Alice's display name, and the invitee's own
// member event carries the invite's reason.
//
// For a room this server is not in — an invite that came over federation —
// the room's state is not here, and what is returned is the stripped state
// the inviting server sent along, plus the invite itself.
func (r *API) StrippedState(ctx context.Context, roomID string, members []string) ([]StrippedEvent, error) {
	_, rules, err := r.roomAndRules(ctx, roomID)
	if err != nil {
		return nil, err
	}
	if _, err := r.store.CurrentStateEvent(ctx, roomID, event.TypeCreate, ""); isNotFound(err) {
		return r.outOfBandStrippedState(ctx, roomID, rules, members)
	} else if err != nil {
		return nil, err
	}
	return r.strippedState(ctx, roomID, rules, members)
}

// outOfBandStrippedState is StrippedState for a room known only from an
// invite.
func (r *API) outOfBandStrippedState(ctx context.Context, roomID string, rules roomversion.Rules, members []string) ([]StrippedEvent, error) {
	var out []StrippedEvent
	for _, member := range members {
		raw, err := r.store.InviteState(ctx, roomID, member)
		if err != nil {
			return nil, err
		}
		if len(raw) > 0 {
			var stripped []StrippedEvent
			if err := json.Unmarshal(raw, &stripped); err != nil {
				// What another server sent is not ours to fail a sync over.
				r.log.Warn("stored invite state is unreadable", "room_id", roomID, "user_id", member, "error", err)
			}
			for _, s := range stripped {
				if s.Type == event.TypeMember && s.StateKey == member {
					continue // the invite itself is added below
				}
				out = append(out, s)
			}
		}
	}
	for _, member := range members {
		stored, err := r.store.CurrentStateEvent(ctx, roomID, event.TypeMember, member)
		if isNotFound(err) {
			continue
		} else if err != nil {
			return nil, err
		}
		e, err := parseEvent(stored, rules)
		if err != nil {
			return nil, err
		}
		out = append(out, StrippedEvent{Type: e.Type(), StateKey: member, Sender: e.Sender(), Content: e.Content()})
	}
	return out, nil
}

// strippedState is StrippedState with the rules in hand.
func (r *API) strippedState(ctx context.Context, roomID string, rules roomversion.Rules, members []string) ([]StrippedEvent, error) {
	var out []StrippedEvent

	appendState := func(eventType, stateKey string) error {
		stored, err := r.store.CurrentStateEvent(ctx, roomID, eventType, stateKey)
		switch {
		case isNotFound(err):
			return nil
		case err != nil:
			return err
		}
		e, err := parseEvent(stored, rules)
		if err != nil {
			return err
		}
		key, _ := e.StateKey()
		out = append(out, StrippedEvent{
			Type:     e.Type(),
			StateKey: key,
			Sender:   e.Sender(),
			Content:  e.Content(),
		})
		return nil
	}

	for _, eventType := range strippedStateTypes {
		if err := appendState(eventType, ""); err != nil {
			return nil, err
		}
	}
	for _, member := range members {
		if err := appendState(event.TypeMember, member); err != nil {
			return nil, err
		}
	}
	return out, nil
}

// RoomSyncSummary is the `summary` block of a room in a sync response.
//
// It exists for rooms with no name: a client showing a direct message has
// to call it something, and the heroes are the members it names it after.
type RoomSyncSummary struct {
	// Heroes are up to five members of the room other than the user
	// syncing, which is what a client falls back to for a room with no
	// name and no canonical alias.
	Heroes []string `json:"m.heroes,omitempty"`

	JoinedMemberCount  int `json:"m.joined_member_count"`
	InvitedMemberCount int `json:"m.invited_member_count"`
}

// maxHeroes is how many members a summary names. The spec says five.
const maxHeroes = 5

// RoomSummaryFor builds the summary block of a room for one user.
func (r *API) RoomSummaryFor(ctx context.Context, roomID, userID string) (RoomSyncSummary, error) {
	entries, err := r.store.MembershipEntries(ctx, roomID,
		[]string{event.MembershipJoin, event.MembershipInvite})
	if err != nil {
		return RoomSyncSummary{}, err
	}

	var summary RoomSyncSummary
	for _, entry := range entries {
		switch entry.Membership {
		case event.MembershipJoin:
			summary.JoinedMemberCount++
		case event.MembershipInvite:
			summary.InvitedMemberCount++
		}
		if entry.UserID != userID && len(summary.Heroes) < maxHeroes {
			summary.Heroes = append(summary.Heroes, entry.UserID)
		}
	}
	return summary, nil
}

// clientEvents projects stored events into the client shape.
func (r *API) clientEvents(stored []storage.StoredEvent, rules roomversion.Rules, roomID string) ([]ClientEvent, error) {
	out := make([]ClientEvent, 0, len(stored))
	for _, s := range stored {
		e, err := parseEvent(s, rules)
		if err != nil {
			return nil, err
		}
		out = append(out, r.clientEvent(s, e, roomID))
	}
	return out, nil
}

// clientEventsByID reads events by ID and projects them.
func (r *API) clientEventsByID(ctx context.Context, roomID string, rules roomversion.Rules, eventIDs []string) ([]ClientEvent, error) {
	if len(eventIDs) == 0 {
		return nil, nil
	}
	stored, err := r.store.EventsByID(ctx, roomID, eventIDs)
	if err != nil {
		return nil, err
	}
	return r.clientEvents(stored, rules, roomID)
}

// stateEntryIDs projects state entries onto the event IDs behind them.
func stateEntryIDs(entries []storage.StateEntry) []string {
	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		out = append(out, entry.EventID)
	}
	return out
}

// UsersSharingRooms lists the users joined to a room the given user is
// joined to, that user included. It is who hears about the user's device
// list changes.
func (r *API) UsersSharingRooms(ctx context.Context, userID string) ([]string, error) {
	return r.store.UsersSharingRooms(ctx, userID)
}

// MemberChanges lists the membership events of the given rooms in a window
// of the event stream, each with the user's membership now.
func (r *API) MemberChanges(ctx context.Context, roomIDs []string, from, to int64) ([]storage.MemberChange, error) {
	return r.store.MemberChanges(ctx, roomIDs, from, to)
}

// JoinedUserIDs lists the users joined to a room, as this server last
// knew it.
func (r *API) JoinedUserIDs(ctx context.Context, roomID string) ([]string, error) {
	entries, err := r.store.MembershipEntries(ctx, roomID, []string{event.MembershipJoin})
	if err != nil {
		return nil, err
	}
	out := make([]string, 0, len(entries))
	for _, e := range entries {
		out = append(out, e.UserID)
	}
	return out, nil
}

// StreamEvent is one event of the server-wide stream, at its position.
type StreamEvent struct {
	Position int64
	Event    ClientEvent
}

// StreamEvents reads the server's event stream across every room: the
// events after position `after`, up to `upTo` (0 for no bound), at most
// limit of them, in the client shape with their room IDs.
//
// Nothing is filtered by who may see what. Its one reader is the
// application service sender, which decides for itself which of them an
// application service is told about.
func (r *API) StreamEvents(ctx context.Context, after, upTo int64, limit int) ([]StreamEvent, error) {
	stored, err := r.store.StreamEvents(ctx, after, upTo, limit)
	if err != nil {
		return nil, err
	}
	rules := map[string]roomversion.Rules{}
	out := make([]StreamEvent, 0, len(stored))
	for _, s := range stored {
		roomRules, ok := rules[s.RoomID]
		if !ok {
			_, roomRules, err = r.roomAndRules(ctx, s.RoomID)
			if err != nil {
				return nil, err
			}
			rules[s.RoomID] = roomRules
		}
		e, err := parseEvent(s, roomRules)
		if err != nil {
			return nil, err
		}
		out = append(out, StreamEvent{Position: s.StreamOrdering, Event: r.clientEvent(s, e, s.RoomID)})
	}
	return out, nil
}

// LocalAliases lists the aliases on this server that point at a room,
// without asking who wants to know: the application service sender matches
// them against its alias namespaces.
func (r *API) LocalAliases(ctx context.Context, roomID string) ([]string, error) {
	return r.store.AliasesForRoom(ctx, roomID)
}