Warp/server/internal/syncapi/syncapi.go

// 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 six 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,
//     to-device messages and device list changes, none of which are
//     events in a room.
//
// Phase 4 of PLAN.md, with device lists from Phase 7: the keyserver writes
// the device list stream, and this package works out which of its changes
// concern whom (devicelists.go). Notification counts, from Phase 8, are
// the push server's: it writes them and their stream, and a sync reports
// them per joined room (notifications.go).
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)
}

// Keys is the part of the keyserver a sync reads: a device's key counts,
// and the device list stream. *keyserver.API satisfies it.
type Keys interface {
	OneTimeKeyCounts(ctx context.Context, userID, deviceID string) (map[string]int, error)
	UnusedFallbackKeyTypes(ctx context.Context, userID, deviceID string) ([]string, error)
	DeviceListChanges(ctx context.Context, userIDs []string, from, to int64) ([]storage.DeviceListChange, error)
}

// Push is the part of the push server a sync reads: the push rules it
// reports as account data, the unread counts of each room (and of each
// thread in it, for a client that asks), and which
// rooms' counts moved. It is also told about read receipts, which are what
// clear the counts. *pushserver.API satisfies it.
type Push interface {
	Document(ctx context.Context, localpart, userID string) ([]byte, error)
	NotificationCounts(ctx context.Context, userID string, roomIDs []string) (map[string]storage.NotificationCounts, error)
	ThreadNotificationCounts(ctx context.Context, userID string, roomIDs []string) (map[string]map[string]storage.NotificationCounts, error)
	NotificationCountChanges(ctx context.Context, userID string, from, to int64) ([]string, error)
	ReceiptSet(ctx context.Context, userID, roomID, receiptType, threadID, eventID string)
}

// API is the syncapi component.
type API struct {
	serverName string
	rooms      *roomserver.API
	accounts   Accounts
	keys       Keys
	push       Push
	store      storage.SyncStore
	notifier   *Notifier
	typing     *typingTracker

	// conns remembers what each sliding sync connection has been sent.
	conns *connStore
	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

	// outbound passes local typing, receipts and remote to-device
	// messages to the federation queue; nil without federation.
	outbound Outbound

	// 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

	// Keys is the keyserver. Nil leaves the key counts out of a sync and
	// both device lists empty.
	Keys Keys

	// Push is the push server. Nil leaves the notification counts out of
	// a sync, and m.push_rules is then whatever account data holds.
	Push Push

	// 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,
		keys:            opts.Keys,
		push:            opts.Push,
		store:           opts.Store,
		notifier:        opts.Notifier,
		typing:          newTypingTracker(opts.TypingTimeout, opts.Notifier, opts.Now),
		conns:           newConnStore(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() }