Warp/server/internal/roomserver/input.go

package roomserver

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

	"go.opentelemetry.io/otel/attribute"

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

// sendRequest is one event this server is about to create.
//
// What the caller supplies is what the event says. Everything that makes it
// a valid link in the room's DAG — prev_events, depth, auth_events, the
// hashes and the signature — is worked out by the pipeline, because only the
// pipeline holds the room lock that makes those answers stable.
type sendRequest struct {
	RoomID string
	Type   string

	// StateKey makes this a state event when it is non-nil. The empty
	// string is a valid state key.
	StateKey *string

	// Sender is the full user ID sending the event. It must be a user of
	// this server: an event is signed by its sender's server, and this
	// server can only sign as itself.
	Sender string

	Content json.RawMessage

	// Redacts is the target of an m.room.redaction event. The builder puts
	// it wherever the room version keeps it.
	Redacts string

	// Transaction, when set, records the client transaction ID this event
	// came from, so that a retried request does not send a second event.
	Transaction *storage.TransactionRef

	// countersign, when set, is given the event once it has been built and
	// authorised and returns the copy to persist. An invite to a user of
	// another server goes through it: that server has to sign the invite
	// before it can be in the room, and the event persisted is the one with
	// both signatures. It runs under the room lock, so the event it signs
	// is still a child of the room's current extremities when it lands.
	countersign func(context.Context, *event.Event) (*event.Event, error)
}

// send is the input pipeline for an event this server creates.
//
// The stages are the ones PLAN.md names. Validate: the sender is ours and
// the content is something we will sign. Authorise: the event is built
// against the room's state and run through the authorisation rules, with
// exactly the auth_events the selection algorithm asks for. Resolve: the
// state the event is judged against is the state after the room's forward
// extremities, resolved when there is more than one of them. Persist: the
// event, its state snapshot, the new current state and the new extremity go
// down in one transaction. Emit: the sync API is told.
//
// Everything after the lock is taken is deterministic, which is what lets
// two clients sending into the same room at the same moment produce a
// straight chain rather than a fork.
func (r *API) send(ctx context.Context, req sendRequest) (*event.Event, error) {
	if !r.isLocal(req.Sender) {
		return nil, fmt.Errorf("%w: %s is not a user of this server", ErrForbidden, req.Sender)
	}
	content, err := validateContent(req.Content)
	if err != nil {
		return nil, err
	}
	req.Content = content

	mu := r.lock(req.RoomID)
	mu.Lock()
	defer mu.Unlock()

	room, err := r.store.Room(ctx, req.RoomID)
	if err != nil {
		return nil, err
	}
	rules, err := r.rules(room)
	if err != nil {
		return nil, err
	}

	// A retried request must not send a second event. The check is inside
	// the lock so that two copies of the same request racing each other
	// cannot both find the transaction unseen.
	if req.Transaction != nil {
		switch existing, err := r.store.TransactionEvent(ctx, *req.Transaction); {
		case err == nil:
			stored, err := r.store.Event(ctx, req.RoomID, existing)
			if err != nil {
				return nil, err
			}
			return parseEvent(stored, rules)
		case !isNotFound(err):
			return nil, err
		}
	}

	st, extremities, replacesWholeState, err := r.roomState(ctx, room, rules)
	if err != nil {
		return nil, err
	}

	authEvents, err := auth.SelectAuthEvents(rules, req.Type, req.Sender, req.StateKey, req.Content, st)
	if err != nil {
		return nil, fmt.Errorf("%w: %s", ErrRejected, err)
	}

	built, err := r.build(req, rules, authEvents, extremities)
	if err != nil {
		return nil, err
	}

	// Rule 3: the auth_events the event declares are well-formed and are
	// the ones the selection algorithm asks for. The pipeline chose them
	// itself a moment ago, so this can only fail on a bug in that choice —
	// which is exactly the bug worth catching before an event is in a room
	// and every other server has to agree with it.
	if err := auth.CheckAuthEvents(built, authEvents); err != nil {
		return nil, err
	}

	// The rules are applied against the state the event's own auth_events
	// name, which is what every other server in the room will do when it
	// sees this event. For an event created here that state is also the
	// room's current state, so there is no separate soft-failure check to
	// make: the two can only differ for an event that arrives from
	// somewhere else, and federated.go checks for that as soft failure.
	if err := auth.Check(ctx, built, checkState(authEvents, st), r.verifier()); err != nil {
		return nil, err
	}

	if req.countersign != nil {
		if built, err = req.countersign(ctx, built); err != nil {
			return nil, err
		}
	}

	if err := r.persist(ctx, persistRequest{
		Room:              room,
		Rules:             rules,
		Event:             built,
		StateBefore:       st,
		ReplaceWholeState: replacesWholeState,
		Transaction:       req.Transaction,
		RedactsEventID:    req.Redacts,
	}); err != nil {
		return nil, err
	}
	return built, nil
}

// build assembles and signs the event.
func (r *API) build(req sendRequest, rules roomversion.Rules, authEvents, prevEvents []*event.Event) (*event.Event, error) {
	builder := event.Builder{
		Type:       req.Type,
		StateKey:   req.StateKey,
		Sender:     req.Sender,
		Content:    req.Content,
		Redacts:    req.Redacts,
		AuthEvents: eventIDs(authEvents),
		PrevEvents: eventIDs(prevEvents),
		Depth:      nextDepth(prevEvents),
	}
	// A room version 12 create event carries no room_id, because the room
	// ID is its own event ID. Nothing else in a room may leave it out.
	if !rules.RoomIDIsCreateEventID || req.Type != event.TypeCreate {
		builder.RoomID = req.RoomID
	}

	built, err := builder.Build(r.now(), r.signingKey, rules)
	if err != nil {
		return nil, fmt.Errorf("building a %s event: %w", req.Type, err)
	}
	return built, nil
}

// checkState is the state an event is judged against: the events its
// auth_events name, plus the create event.
//
// The create event is added separately because from room version 12 it is
// not cited in auth_events at all — the room ID *is* its event ID, so
// naming it would be saying the same thing twice. The rules still need it:
// it is what says who the room's creators are and, in that same version,
// what the room's ID should be. In earlier versions it is already among the
// auth events and this changes nothing.
func checkState(authEvents []*event.Event, roomState auth.State) auth.State {
	st := auth.NewState(authEvents)
	if st.Create() == nil {
		if create := roomState.Create(); create != nil {
			st[auth.StateKey{Type: event.TypeCreate}] = create
		}
	}
	return st
}

// eventIDs projects events onto their IDs, in order.
func eventIDs(events []*event.Event) []string {
	out := make([]string, 0, len(events))
	for _, e := range events {
		out = append(out, e.EventID())
	}
	return out
}

// nextDepth is one more than the greatest depth among the prev_events.
//
// Depth is advisory — it orders events for display and is never used to
// authorise one — so a room that reaches the limit stays there rather than
// failing: refusing to accept any further events would be a worse outcome
// than a flat tail, and the limit is 2^53-1 deep.
func nextDepth(prevEvents []*event.Event) int64 {
	var depth int64
	for _, e := range prevEvents {
		if e.Depth() > depth {
			depth = e.Depth()
		}
	}
	if depth >= event.MaxDepth {
		return event.MaxDepth
	}
	return depth + 1
}

// roomState returns the state the next event in a room is authorised
// against, along with the room's forward extremities.
//
// The stored current state is the answer whenever a room has a single
// forward extremity, which is every room this server fills on its own: the
// lock in send makes each new event a child of the one before it. A room
// with several extremities has forked, and the state is then resolved from
// the state after each of them — the third return value says so, because a
// resolved state has to be written out whole: tuples the resolution dropped
// have to go, and an incremental update cannot express that.
//
// Only federation forks a room. The second path was written in Phase 3,
// before anything could reach it, because the alternative was a pipeline
// that authorised against one arbitrary branch of a fork without saying so.
func (r *API) roomState(ctx context.Context, room storage.Room, rules roomversion.Rules) (auth.State, []*event.Event, bool, error) {
	storedExtremities, err := r.store.Extremities(ctx, room.ID)
	if err != nil {
		return nil, nil, false, err
	}
	extremities, err := parseEvents(storedExtremities, rules)
	if err != nil {
		return nil, nil, false, err
	}

	if len(extremities) <= 1 {
		stored, err := r.store.CurrentStateEvents(ctx, room.ID)
		if err != nil {
			return nil, nil, false, err
		}
		events, err := parseEvents(stored, rules)
		if err != nil {
			return nil, nil, false, err
		}
		return auth.NewState(events), extremities, false, nil
	}

	stateSets := make([][]*event.Event, 0, len(extremities))
	for _, e := range extremities {
		set, err := r.stateAfter(ctx, room.ID, rules, e)
		if err != nil {
			return nil, nil, false, err
		}
		stateSets = append(stateSets, set)
	}
	resolved, err := stateres.Resolve(ctx, rules, stateSets,
		eventProvider{store: r.store, roomID: room.ID, rules: rules})
	if err != nil {
		return nil, nil, false, fmt.Errorf("resolving the state of %s: %w", room.ID, err)
	}
	return resolved, extremities, true, nil
}

// stateAfter returns the room state as of an event, the event included.
//
// The snapshot stored against an event is the state *before* it; adding the
// event itself when it is a state event is what turns that into the state a
// child of it would see.
func (r *API) stateAfter(ctx context.Context, roomID string, rules roomversion.Rules, e *event.Event) ([]*event.Event, error) {
	entries, err := r.store.StateBefore(ctx, roomID, e.EventID())
	if err != nil {
		return nil, err
	}
	ids := make([]string, 0, len(entries))
	for _, entry := range entries {
		ids = append(ids, entry.EventID)
	}
	stored, err := r.store.EventsByID(ctx, roomID, ids)
	if err != nil {
		return nil, err
	}
	events, err := parseEvents(stored, rules)
	if err != nil {
		return nil, err
	}
	if e.IsState() {
		events = append(events, e)
	}
	return events, nil
}

// persistRequest is everything the persist stage needs.
type persistRequest struct {
	Room  storage.Room
	Rules roomversion.Rules
	Event *event.Event

	// StateBefore is the room state the event was authorised against.
	StateBefore auth.State

	// ReplaceWholeState says StateBefore came out of state resolution, so
	// the new current state has to be written out in full.
	ReplaceWholeState bool

	Transaction    *storage.TransactionRef
	RedactsEventID string

	// Extremities replaces the room's forward extremities. Nil means the
	// event alone, which is right for every event created here: it
	// descends from every extremity there was.
	Extremities []string

	// CurrentState, when set, is the room's whole current state after the
	// event, written as a replacement. An event from another server sets
	// it whenever the room's state after it is not simply the state before
	// it plus the event: when it forks the room, or when it was built on
	// something other than the room's single extremity.
	CurrentState auth.State

	// Origin and Relay are passed to the notifier; see Notification.
	Origin string
	Relay  bool
}

// persist writes the event and everything its arrival changes, then emits.
func (r *API) persist(ctx context.Context, req persistRequest) (err error) {
	ctx, span := tracing.Start(ctx, "roomserver.persist",
		attribute.String("matrix.room_id", req.Room.ID),
		attribute.String("matrix.event_type", req.Event.Type()))
	defer func() { tracing.End(span, err) }()

	e := req.Event
	stateKey, isState := e.StateKey()

	write := storage.RoomWrite{
		Event:       storedOf(req.Room.ID, e),
		StateBefore: stateEntries(req.StateBefore),
		Extremities: req.Extremities,
		Transaction: req.Transaction,
	}
	if write.Extremities == nil {
		write.Extremities = []string{e.EventID()}
	}

	// The state after this event. For the ordinary case that is one tuple
	// to upsert; after a resolution it is the whole state, which the store
	// is told to write as a replacement.
	stateAfter := req.StateBefore
	stateChanged := isState
	switch {
	case req.CurrentState != nil:
		write.StateDeltaIsWholeState = true
		write.StateDelta = stateEntries(req.CurrentState)
		stateAfter = req.CurrentState
		stateChanged = true
	case req.ReplaceWholeState:
		write.StateDeltaIsWholeState = true
		write.StateDelta = stateEntries(req.StateBefore)
		stateChanged = true
	}
	if isState && req.CurrentState == nil {
		entry := stateEntry(e.Type(), stateKey, e)
		if req.ReplaceWholeState {
			write.StateDelta = upsertEntry(write.StateDelta, entry)
		} else {
			write.StateDelta = []storage.StateEntry{entry}
		}
		stateAfter = withState(req.StateBefore, e.Type(), stateKey, e)
	}

	// The denormalised listing columns are recomputed only when this event
	// is one of the ones behind them, which keeps a room full of messages
	// from rewriting its own row on every line of chat. A whole-state
	// write may have changed any of them, so it always recomputes.
	if (isState && affectsRoomProfile(e.Type())) || write.StateDeltaIsWholeState {
		profile := roomProfile(stateAfter)
		write.Profile = &profile
	}

	if req.RedactsEventID != "" {
		if err := r.prepareRedaction(ctx, &write, req); err != nil {
			return err
		}
	}

	ordering, err := r.store.Write(ctx, write)
	if err != nil {
		return err
	}

	r.log.Debug("persisted event",
		"room_id", req.Room.ID, "event_id", e.EventID(), "type", e.Type(),
		"sender", e.Sender(), "stream_ordering", ordering)

	if r.notifier != nil {
		r.notifier.EventPersisted(ctx, Notification{
			RoomID:         req.Room.ID,
			Event:          e,
			StreamOrdering: ordering,
			StateChanged:   stateChanged,
			Origin:         req.Origin,
			Relay:          req.Relay,
		})
	}
	return nil
}

// withState returns st with one piece of state replaced, leaving st alone.
func withState(st auth.State, eventType, stateKey string, e *event.Event) auth.State {
	out := make(auth.State, len(st)+1)
	for key, value := range st {
		out[key] = value
	}
	out[auth.StateKey{Type: eventType, StateKey: stateKey}] = e
	return out
}

// upsertEntry replaces the matching tuple in entries, or appends it.
func upsertEntry(entries []storage.StateEntry, entry storage.StateEntry) []storage.StateEntry {
	for i, existing := range entries {
		if existing.Type == entry.Type && existing.StateKey == entry.StateKey {
			entries[i] = entry
			return entries
		}
	}
	return append(entries, entry)
}

// isNotFound reports whether err is one of the store's "nothing there"
// sentinels, which several callers treat as an ordinary answer rather than
// as a failure.
func isNotFound(err error) bool {
	return errors.Is(err, ErrNoSuchEvent) ||
		errors.Is(err, ErrNoSuchRoom) ||
		errors.Is(err, ErrNoSuchAlias)
}