Warp/server/internal/roomserver/roomserver.go

// Package roomserver owns rooms: it is the only thing in Warp that creates
// an event, and the only thing that decides whether one belongs in a room.
//
// Everything here runs through one pipeline — validate, authorise, resolve,
// persist, emit — because every event has to be judged the same way whether
// it came from a client, from `createRoom` or, from Phase 5, from another
// server. The rules themselves are not here: they are in internal/matrix,
// which is pure library code with no storage and no HTTP. This package is
// what connects those rules to a database and to a room's history.
//
// What the Client-Server API handlers get is a small set of operations —
// create a room, change a membership, send an event, redact one, read the
// state — each of which takes user IDs and content and returns an event ID
// or an error. Handlers do no authorisation of their own, and could not: a
// membership change is allowed or refused by the room's power levels and
// join rules, which only this package reads.
package roomserver

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

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

// Errors callers have to tell apart.
//
// The storage and auth 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
	ErrNoSuchAlias = storage.ErrNoSuchAlias
	ErrAliasExists = storage.ErrAliasExists

	// ErrRejected means the authorisation rules refused the event. The
	// wrapped message names the rule that said no, and is safe to show a
	// client: it says why the room would not accept what they sent.
	ErrRejected = auth.ErrRejected

	// ErrForbidden is a refusal from a check that sits above the
	// authorisation rules — not being in a room whose state you asked for,
	// or deleting somebody else's alias.
	ErrForbidden = errors.New("forbidden")

	// ErrFederationDisabled is an operation that would need to talk to
	// another server while federation is off.
	ErrFederationDisabled = errors.New("federation is disabled")

	// ErrInvalidContent is event content this server will not sign: not an
	// object, too large, or missing something its type requires.
	ErrInvalidContent = errors.New("invalid event content")

	// ErrUnsupportedRoomVersion is a room version Warp cannot take part in.
	ErrUnsupportedRoomVersion = roomversion.ErrUnsupportedVersion

	// ErrUnknownRoomVersion is a room version that does not exist.
	ErrUnknownRoomVersion = roomversion.ErrUnknownVersion
)

// Profiles supplies the display name and avatar that a membership event
// carries a copy of.
//
// Membership events duplicate the profile on purpose: a client showing a
// room's history has to render the name the sender had at the time, and the
// member event is the only place that is recorded. *userapi.API satisfies
// this as it stands.
type Profiles interface {
	Account(ctx context.Context, localpart string) (storage.Account, error)
}

// Notification is one persisted event, as the emit stage reports it.
type Notification struct {
	RoomID string

	// Event is the event as it was persisted.
	Event *event.Event

	// StreamOrdering is the position the event was given in this server's
	// event stream.
	StreamOrdering int64

	// StateChanged reports whether the room's current state moved.
	StateChanged bool
}

// Notifier is told about every event this server persists, in the order
// they were persisted.
//
// It is the emit stage of the input pipeline and the seam the sync API
// (PLAN.md phase 4) plugs into: a client long-polling /sync has to be woken
// the moment an event lands, and the input pipeline is the only place that
// knows one has. Phase 3 ships without a notifier, so this is nil and the
// stage is skipped.
type Notifier interface {
	EventPersisted(ctx context.Context, n Notification)
}

// API is the roomserver component.
type API struct {
	serverName string
	store      storage.RoomStore
	profiles   Profiles
	signingKey *signing.KeyPair
	notifier   Notifier
	log        *slog.Logger
	now        func() time.Time

	// federationEnabled reports the live state of the federation switch.
	// Several operations are refused outright while it is off, and the
	// answer has to be read per request because an admin can flip it.
	federationEnabled func() bool

	// defaultRoomVersion is what createRoom uses when the client does not
	// ask for one.
	defaultRoomVersion roomversion.Version

	// locks serialises the writers of each room. Two events racing into one
	// room would each pick up the same forward extremities and both become
	// one, forking the room for no reason — and then needing state
	// resolution to put it back together. Taking a lock per room instead
	// keeps every locally created event a child of the one before it.
	locks sync.Map // room ID -> *sync.Mutex
}

// Options configure the component.
type Options struct {
	// ServerName is the Matrix server name. It is the domain part of the
	// user IDs this server may send events as, and of the room IDs and
	// aliases it mints.
	ServerName string

	// Store is the persistence layer.
	Store storage.RoomStore

	// Profiles supplies the display name and avatar of a local user.
	Profiles Profiles

	// SigningKey signs every event this server creates.
	SigningKey *signing.KeyPair

	// DefaultRoomVersion overrides the registry's default for createRoom.
	// Empty uses roomversion.Default.
	DefaultRoomVersion string

	// FederationEnabled reports the live federation switch. Nil means off,
	// which is the safe default for a half-wired server.
	FederationEnabled func() bool

	// Notifier receives every persisted event. Nil skips the emit stage.
	Notifier Notifier

	Logger *slog.Logger

	// Now is the clock, so that tests can stamp events deterministically.
	Now func() time.Time
}

// New builds the component.
func New(opts Options) (*API, error) {
	switch {
	case opts.ServerName == "":
		return nil, errors.New("roomserver: ServerName must not be empty")
	case opts.Store == nil:
		return nil, errors.New("roomserver: Store must not be nil")
	case opts.SigningKey == nil:
		// Without a key nothing here can produce an event, and finding that
		// out on the first createRoom rather than at startup would mean a
		// server that looks healthy and cannot be used.
		return nil, errors.New("roomserver: SigningKey must not be nil")
	}

	version := roomversion.Default
	if opts.DefaultRoomVersion != "" {
		if _, err := roomversion.Get(roomversion.Version(opts.DefaultRoomVersion)); err != nil {
			return nil, fmt.Errorf("roomserver: default room version %q: %w", opts.DefaultRoomVersion, err)
		}
		version = roomversion.Version(opts.DefaultRoomVersion)
	}
	if opts.FederationEnabled == nil {
		opts.FederationEnabled = func() bool { return false }
	}
	if opts.Logger == nil {
		opts.Logger = slog.Default()
	}
	if opts.Now == nil {
		opts.Now = time.Now
	}

	return &API{
		serverName:         opts.ServerName,
		store:              opts.Store,
		profiles:           opts.Profiles,
		signingKey:         opts.SigningKey,
		notifier:           opts.Notifier,
		log:                opts.Logger,
		now:                opts.Now,
		federationEnabled:  opts.FederationEnabled,
		defaultRoomVersion: version,
	}, nil
}

// ServerName returns the server name this roomserver signs events as.
func (r *API) ServerName() string { return r.serverName }

// DefaultRoomVersion returns the version createRoom uses by default.
func (r *API) DefaultRoomVersion() string { return string(r.defaultRoomVersion) }

// nowMS is the current time in the unit events carry.
func (r *API) nowMS() int64 { return r.now().UnixMilli() }

// lock serialises the writers of one room; see API.locks.
func (r *API) lock(roomID string) *sync.Mutex {
	actual, _ := r.locks.LoadOrStore(roomID, &sync.Mutex{})
	mu, ok := actual.(*sync.Mutex)
	if !ok {
		// Unreachable: nothing else ever stores into this map.
		panic("roomserver: room lock map holds something that is not a mutex")
	}
	return mu
}

// isLocal reports whether a user ID belongs to this server.
func (r *API) isLocal(userID string) bool {
	serverName, err := id.ServerNameOf(userID)
	return err == nil && serverName == r.serverName
}

// localpartOf returns the localpart of a user ID of this server.
func (r *API) localpartOf(userID string) (string, bool) {
	localpart, serverName, err := id.SplitUserID(userID)
	if err != nil || serverName != r.serverName {
		return "", false
	}
	return localpart, true
}

// rules returns the room version rules a room's events are read under.
//
// A room whose version is no longer supported is a real possibility — a
// future build could drop one — and it has to fail as a room-level error
// rather than a parse error on every event in it.
func (r *API) rules(room storage.Room) (roomversion.Rules, error) {
	rules, err := roomversion.Get(roomversion.Version(room.Version))
	if err != nil {
		return roomversion.Rules{}, fmt.Errorf("room %s is version %q: %w", room.ID, room.Version, err)
	}
	return rules, nil
}

// parseEvent turns a stored event back into an Event.
//
// A redacted event parses from its redacted form and keeps its original
// event ID, because the ID is the reference hash of the redacted form in
// the first place.
func parseEvent(stored storage.StoredEvent, rules roomversion.Rules) (*event.Event, error) {
	e, err := event.Parse(stored.JSON, rules)
	if err != nil {
		return nil, fmt.Errorf("stored event %s is unreadable: %w", stored.EventID, err)
	}
	return e, nil
}

// parseEvents parses a batch, failing on the first unreadable one.
//
// An unreadable stored event is a corrupted database or a bug in this
// server's own encoder, never input: everything in the store was parsed and
// validated on the way in. It must not be skipped quietly, because the
// events being parsed here are a room's state and dropping one would
// silently widen what the room allows.
func parseEvents(stored []storage.StoredEvent, rules roomversion.Rules) ([]*event.Event, error) {
	out := make([]*event.Event, 0, len(stored))
	for _, s := range stored {
		e, err := parseEvent(s, rules)
		if err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, nil
}

// stateEntries projects a state map onto the rows the store keeps.
func stateEntries(st auth.State) []storage.StateEntry {
	out := make([]storage.StateEntry, 0, len(st))
	for key, e := range st {
		out = append(out, stateEntry(key.Type, key.StateKey, e))
	}
	return out
}

// stateEntry projects one piece of state onto a storage row, carrying the
// membership along for an m.room.member event so that the store can index
// it.
func stateEntry(eventType, stateKey string, e *event.Event) storage.StateEntry {
	entry := storage.StateEntry{Type: eventType, StateKey: stateKey, EventID: e.EventID()}
	if eventType == event.TypeMember {
		// A member event whose content has no membership would have been
		// refused on the way in, so an error here cannot happen; the empty
		// string it leaves behind is also the correct "not a member".
		entry.Membership, _ = e.Membership()
	}
	return entry
}

// eventProvider lets state resolution read a room's auth chains.
type eventProvider struct {
	store  storage.RoomStore
	roomID string
	rules  roomversion.Rules
}

// EventsByID implements stateres.EventProvider.
//
// Events this server does not hold are left out rather than reported as an
// error, which is what the algorithm requires: a rejected or missing auth
// event takes no part in resolution.
func (p eventProvider) EventsByID(ctx context.Context, ids []string) ([]*event.Event, error) {
	stored, err := p.store.EventsByID(ctx, p.roomID, ids)
	if err != nil {
		return nil, err
	}
	return parseEvents(stored, p.rules)
}

// validateContent checks event content before this server signs it.
//
// The size limit is the spec's limit on a whole event, applied to the
// content alone: the rest of an event is at most a few hundred bytes, so
// anything bigger than this cannot be made to fit and the client should hear
// about it before the event is built, hashed and signed.
func validateContent(content json.RawMessage) (json.RawMessage, error) {
	if len(content) == 0 {
		return json.RawMessage("{}"), nil
	}
	if len(content) > event.MaxEventSize {
		return nil, fmt.Errorf("%w: content is %d bytes, the limit for a whole event is %d",
			ErrInvalidContent, len(content), event.MaxEventSize)
	}
	var probe map[string]json.RawMessage
	if err := json.Unmarshal(content, &probe); err != nil {
		return nil, fmt.Errorf("%w: content must be a JSON object", ErrInvalidContent)
	}
	return content, nil
}