Warp/server/internal/delayedevents/delayedevents.go

// Package delayedevents implements MSC4140 delayed events: an event a client
// hands the server now, to be sent later on the client's behalf unless the
// client pushes it back or cancels it first.
//
// It is what makes MatrixRTC group calls robust. A member joining a call
// sends its call.member state, and schedules an empty one to replace it a
// few seconds out; while it is in the call it keeps restarting that delay.
// A client that crashes, loses its network or is closed stops restarting
// it, and the server sends the "left" for it.
//
// Events are kept in the database, so a restart of the server does not
// lose them, and are sent by a single scheduler goroutine that sleeps until
// the next one is due.
package delayedevents

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

	"github.com/dariuszjaniak/warp/server/internal/roomserver"
	"github.com/dariuszjaniak/warp/server/internal/storage"
)

// Outcomes and reasons a finalised event is reported with, as MSC4140
// names them.
const (
	OutcomeSend   = "send"
	OutcomeCancel = "cancel"

	ReasonDelay  = "delay"
	ReasonAction = "action"
	ReasonError  = "error"
)

var (
	// ErrDisabled is returned when delayed events are switched off.
	ErrDisabled = errors.New("delayed events are disabled")

	// ErrMaxDelayExceeded is returned for a delay longer than the
	// configured maximum.
	ErrMaxDelayExceeded = errors.New("the requested delay exceeds the allowed maximum")

	// ErrTooMany is returned when the user already has as many delayed
	// events scheduled as they may.
	ErrTooMany = errors.New("too many delayed events are scheduled")

	// ErrNoSuchDelayedEvent is returned for a delay ID that is unknown or
	// no longer scheduled.
	ErrNoSuchDelayedEvent = storage.ErrNoSuchDelayedEvent
)

// finalisedRetention is how long a sent or cancelled event is kept for a
// client to look up.
const finalisedRetention = 24 * time.Hour

// idleCheck is how long the scheduler sleeps with nothing scheduled. It
// also paces the purge of finalised events.
const idleCheck = time.Hour

// dueBatch is how many due events are read at once.
const dueBatch = 100

// Sender sends an event into a room as a user: the roomserver.
type Sender interface {
	SendEvent(ctx context.Context, req roomserver.SendEventRequest) (string, error)
}

// Options configure the service.
type Options struct {
	Store storage.DelayedEventStore
	Rooms Sender

	// MaxDelay is the longest delay allowed. Zero disables the service.
	MaxDelay time.Duration

	// MaxPerUser caps the events one user may have scheduled.
	MaxPerUser int

	// DescribeError turns the error a send failed with into the Matrix
	// error a client is shown. Nil reports every failure as M_UNKNOWN.
	DescribeError func(error) json.RawMessage

	Logger *slog.Logger
	Now    func() time.Time
}

// Service schedules and sends delayed events.
type Service struct {
	store         storage.DelayedEventStore
	rooms         Sender
	maxDelay      time.Duration
	maxPerUser    int
	describeError func(error) json.RawMessage
	log           *slog.Logger
	now           func() time.Time

	// mu serialises everything that finalises or moves an event, so the
	// scheduler sending an event and its owner restarting or cancelling
	// it cannot both win.
	mu sync.Mutex

	wake    chan struct{}
	stop    chan struct{}
	done    chan struct{}
	started bool
	closing sync.Once

	// cancels tracks the background cancellations the emit stage starts,
	// so Close can wait for them.
	cancels sync.WaitGroup
}

// New builds the service. It does not start the scheduler.
func New(opts Options) (*Service, error) {
	if opts.Store == nil {
		return nil, errors.New("delayedevents: Store must not be nil")
	}
	if opts.Rooms == nil {
		return nil, errors.New("delayedevents: Rooms must not be nil")
	}
	if opts.Logger == nil {
		opts.Logger = slog.Default()
	}
	if opts.Now == nil {
		opts.Now = time.Now
	}
	if opts.DescribeError == nil {
		opts.DescribeError = func(err error) json.RawMessage {
			raw, _ := json.Marshal(map[string]string{"errcode": "M_UNKNOWN", "error": err.Error()})
			return raw
		}
	}
	return &Service{
		store:         opts.Store,
		rooms:         opts.Rooms,
		maxDelay:      opts.MaxDelay,
		maxPerUser:    opts.MaxPerUser,
		describeError: opts.DescribeError,
		log:           opts.Logger.With("component", "delayedevents"),
		now:           opts.Now,
		wake:          make(chan struct{}, 1),
		stop:          make(chan struct{}),
		done:          make(chan struct{}),
	}, nil
}

// Enabled reports whether clients may schedule delayed events.
func (s *Service) Enabled() bool { return s != nil && s.maxDelay > 0 }

// MaxDelay is the longest delay a client may ask for.
func (s *Service) MaxDelay() time.Duration { return s.maxDelay }

// ScheduleRequest is an event to send later.
type ScheduleRequest struct {
	UserID   string
	DeviceID string
	RoomID   string
	Type     string

	// StateKey makes this a state event when it is non-nil.
	StateKey *string
	Content  json.RawMessage
	Delay    time.Duration
}

// Schedule stores an event to be sent after req.Delay and returns its
// delay ID.
func (s *Service) Schedule(ctx context.Context, req ScheduleRequest) (string, error) {
	if !s.Enabled() {
		return "", ErrDisabled
	}
	if req.Delay < 0 {
		return "", fmt.Errorf("%w: the delay must not be negative", roomserver.ErrInvalidContent)
	}
	if req.Delay > s.maxDelay {
		return "", ErrMaxDelayExceeded
	}
	n, err := s.store.CountScheduledDelayedEvents(ctx, req.UserID)
	if err != nil {
		return "", err
	}
	if n >= s.maxPerUser {
		return "", ErrTooMany
	}
	delayID, err := newDelayID()
	if err != nil {
		return "", err
	}
	nowMS := s.now().UnixMilli()
	delayMS := req.Delay.Milliseconds()
	if err := s.store.InsertDelayedEvent(ctx, storage.DelayedEvent{
		DelayID:        delayID,
		UserID:         req.UserID,
		DeviceID:       req.DeviceID,
		RoomID:         req.RoomID,
		Type:           req.Type,
		StateKey:       req.StateKey,
		Content:        req.Content,
		DelayMS:        delayMS,
		RunningSinceMS: nowMS,
		SendAtMS:       nowMS + delayMS,
	}); err != nil {
		return "", err
	}
	s.poke()
	return delayID, nil
}

// Get returns one of a user's delayed events, scheduled or finalised.
func (s *Service) Get(ctx context.Context, userID, delayID string) (storage.DelayedEvent, error) {
	return s.store.DelayedEvent(ctx, userID, delayID)
}

// List returns a user's delayed events.
func (s *Service) List(ctx context.Context, userID string) ([]storage.DelayedEvent, error) {
	return s.store.DelayedEvents(ctx, userID)
}

// Restart starts a scheduled event's delay again from now.
//
// The action endpoints may be called without an access token: the delay
// ID is the capability. That is what lets a MatrixRTC client hand its
// delayed leave to a service that keeps it alive on its behalf.
func (s *Service) Restart(ctx context.Context, delayID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.store.RestartDelayedEvent(ctx, delayID, s.now().UnixMilli()); err != nil {
		return err
	}
	s.poke()
	return nil
}

// Cancel drops a scheduled event without sending it.
func (s *Service) Cancel(ctx context.Context, delayID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	e, err := s.store.ScheduledDelayedEvent(ctx, delayID)
	if err != nil {
		return err
	}
	return s.finalise(ctx, e, OutcomeCancel, ReasonAction, "", nil)
}

// Send sends a scheduled event now, without waiting for its delay. The
// error of a failed send is returned as well as recorded.
func (s *Service) Send(ctx context.Context, delayID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	e, err := s.store.ScheduledDelayedEvent(ctx, delayID)
	if err != nil {
		return err
	}
	return s.sendLocked(ctx, e, ReasonAction)
}

// sendLocked sends e and records the outcome. s.mu must be held.
func (s *Service) sendLocked(ctx context.Context, e storage.DelayedEvent, reason string) error {
	eventID, sendErr := s.rooms.SendEvent(ctx, roomserver.SendEventRequest{
		RoomID:   e.RoomID,
		Sender:   e.UserID,
		Type:     e.Type,
		StateKey: e.StateKey,
		Content:  e.Content,
	})
	if sendErr != nil {
		if err := s.finalise(ctx, e, OutcomeSend, ReasonError, "", s.describeError(sendErr)); err != nil {
			return err
		}
		return sendErr
	}
	return s.finalise(ctx, e, OutcomeSend, reason, eventID, nil)
}

func (s *Service) finalise(ctx context.Context, e storage.DelayedEvent, outcome, reason, eventID string, errBody json.RawMessage) error {
	e.FinalisedAtMS = s.now().UnixMilli()
	e.Outcome = outcome
	e.Reason = reason
	e.EventID = eventID
	e.Error = errBody
	return s.store.FinaliseDelayedEvent(ctx, e)
}

// EventPersisted implements roomserver.Notifier.
//
// When someone else sets a piece of state, the delayed state events that
// would overwrite it are cancelled: a delayed event stands for "put this
// back if I stop answering", and must not clobber a change another user
// made in the meantime. The owner's own updates leave them alone, which is
// what lets a call member refresh its membership while its delayed leave
// keeps running.
//
// The emit stage must not block, so the lookup runs in the background.
func (s *Service) EventPersisted(_ context.Context, n roomserver.Notification) {
	if !s.Enabled() || n.Event == nil || !n.StateChanged {
		return
	}
	key, isState := n.Event.StateKey()
	if !isState {
		return
	}
	roomID, eventType, sender := n.RoomID, n.Event.Type(), n.Event.Sender()
	select {
	case <-s.stop:
		return
	default:
	}
	s.cancels.Add(1)
	go func() {
		defer s.cancels.Done()
		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		s.cancelOverwritten(ctx, roomID, eventType, key, sender)
	}()
}

func (s *Service) cancelOverwritten(ctx context.Context, roomID, eventType, stateKey, sender string) {
	pending, err := s.store.ScheduledDelayedStateEvents(ctx, roomID, eventType, stateKey)
	if err != nil {
		s.log.Warn("could not look up delayed state events", "room_id", roomID, "error", err)
		return
	}
	for _, e := range pending {
		if e.UserID == sender {
			continue
		}
		s.mu.Lock()
		err := s.finalise(ctx, e, OutcomeCancel, ReasonAction, "", nil)
		s.mu.Unlock()
		if err != nil && !errors.Is(err, storage.ErrNoSuchDelayedEvent) {
			s.log.Warn("could not cancel an overwritten delayed event", "delay_id", e.DelayID, "error", err)
		}
	}
}

// Start runs the scheduler until ctx is cancelled or Close is called.
func (s *Service) Start(ctx context.Context) {
	if !s.Enabled() || s.started {
		return
	}
	s.started = true
	go s.run(ctx)
}

// Close stops the scheduler and waits for it.
func (s *Service) Close() {
	s.closing.Do(func() { close(s.stop) })
	if s.started {
		<-s.done
	}
	s.cancels.Wait()
}

// poke wakes the scheduler to look at the schedule again.
func (s *Service) poke() {
	select {
	case s.wake <- struct{}{}:
	default:
	}
}

func (s *Service) run(ctx context.Context) {
	defer close(s.done)
	timer := time.NewTimer(0)
	defer timer.Stop()
	lastPurge := time.Time{}

	for {
		select {
		case <-ctx.Done():
			return
		case <-s.stop:
			return
		case <-timer.C:
		case <-s.wake:
		}

		s.SendDue(ctx)
		if now := s.now(); now.Sub(lastPurge) >= idleCheck {
			lastPurge = now
			if err := s.store.PurgeFinalisedDelayedEvents(ctx, now.Add(-finalisedRetention).UnixMilli()); err != nil {
				s.log.Warn("could not purge finalised delayed events", "error", err)
			}
		}

		wait := idleCheck
		next, ok, err := s.store.NextDelayedEventDue(ctx)
		switch {
		case err != nil:
			if ctx.Err() != nil {
				return
			}
			s.log.Warn("could not read the delayed event schedule", "error", err)
			wait = time.Second
		case ok:
			wait = min(max(time.Duration(next-s.now().UnixMilli())*time.Millisecond, 0), idleCheck)
		}
		if !timer.Stop() {
			select {
			case <-timer.C:
			default:
			}
		}
		timer.Reset(wait)
	}
}

// SendDue sends every event whose delay has run out. The scheduler calls
// it whenever the next event comes due; tests call it after moving the
// clock.
func (s *Service) SendDue(ctx context.Context) {
	for {
		due, err := s.store.DueDelayedEvents(ctx, s.now().UnixMilli(), dueBatch)
		if err != nil {
			if ctx.Err() == nil {
				s.log.Warn("could not read due delayed events", "error", err)
			}
			return
		}
		for _, e := range due {
			s.sendIfStillDue(ctx, e.DelayID)
		}
		if len(due) < dueBatch {
			return
		}
	}
}

// sendIfStillDue re-reads an event under the lock, because its owner may
// have restarted or cancelled it since the batch was read.
func (s *Service) sendIfStillDue(ctx context.Context, delayID string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	e, err := s.store.ScheduledDelayedEvent(ctx, delayID)
	if err != nil {
		return
	}
	if e.SendAtMS > s.now().UnixMilli() {
		return
	}
	if err := s.sendLocked(ctx, e, ReasonDelay); err != nil {
		s.log.Info("a delayed event could not be sent",
			"delay_id", e.DelayID, "room_id", e.RoomID, "type", e.Type, "error", err)
	}
}

// newDelayID makes a delay ID. It is a capability — anyone holding it may
// restart, cancel or send the event — so it is as long as an access token's
// random part.
func newDelayID() (string, error) {
	raw := make([]byte, 24)
	if _, err := rand.Read(raw); err != nil {
		return "", fmt.Errorf("generate delay ID: %w", err)
	}
	return "warp_de_" + base64.RawURLEncoding.EncodeToString(raw), nil
}