Warp/server/internal/appservice/appservice.go

// Package appservice is the homeserver's side of the Application Service
// API: the registry of application services (bridges and bots with a
// namespace of users, aliases and rooms of their own), and the sender that
// delivers them the events they are interested in.
//
// The Client-Server half — an application service acting as one of its
// users, registering them, the M_EXCLUSIVE refusals — is in clientapi,
// which asks the Registry here. What this package does on its own is
// talk to the services: transactions, the user and alias queries, and the
// ping.
//
// Delivery reads the roomserver's event stream from a cursor kept per
// service in the database, rather than queueing events as they are
// persisted. The stream is already durable and ordered, so a cursor is all
// a service that has been down for a day needs to be caught up, and the
// emit stage only has to say "there is more" rather than hand anything
// over.
package appservice

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

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

// Rooms is what the sender reads from the roomserver. *roomserver.API
// satisfies it.
type Rooms interface {
	StreamTop(ctx context.Context) (int64, error)
	StreamEvents(ctx context.Context, after, upTo int64, limit int) ([]roomserver.StreamEvent, error)
	JoinedUserIDs(ctx context.Context, roomID string) ([]string, error)
	LocalAliases(ctx context.Context, roomID string) ([]string, error)
}

// DefaultTimeout bounds one request to a service.
const DefaultTimeout = 30 * time.Second

// Options configure the component.
type Options struct {
	Registry *Registry
	Store    storage.AppServiceStore

	// Rooms is the roomserver. Without one there is no stream to deliver
	// and the sender does not start; the registry and the queries still
	// work.
	Rooms Rooms

	// Timeout bounds one request to a service; zero is DefaultTimeout.
	Timeout time.Duration

	Logger *slog.Logger

	// Now is the clock, for the status timestamps.
	Now func() time.Time
}

// API is the component.
type API struct {
	registry *Registry
	store    storage.AppServiceStore
	rooms    Rooms
	client   *Client
	log      *slog.Logger
	now      func() time.Time

	workers map[string]*worker

	mu      sync.Mutex
	cancel  context.CancelFunc
	wg      sync.WaitGroup
	started bool
}

// New builds the component. It does not start delivering; Start does.
func New(opts Options) (*API, error) {
	if opts.Registry == nil {
		var err error
		if opts.Registry, err = NewRegistry("", nil); err != nil {
			return nil, err
		}
	}
	if opts.Store == nil && opts.Rooms != nil {
		return nil, errors.New("appservice: Store must not be nil when there is a roomserver")
	}
	if opts.Timeout <= 0 {
		opts.Timeout = DefaultTimeout
	}
	if opts.Logger == nil {
		opts.Logger = slog.Default()
	}
	if opts.Now == nil {
		opts.Now = time.Now
	}
	a := &API{
		registry: opts.Registry,
		store:    opts.Store,
		rooms:    opts.Rooms,
		client:   NewClient(opts.Timeout),
		log:      opts.Logger,
		now:      opts.Now,
		workers:  map[string]*worker{},
	}
	for _, svc := range opts.Registry.Services() {
		// A service with no URL is never sent anything, so it has no
		// worker and no cursor.
		if svc.URL != "" {
			a.workers[svc.ID] = newWorker(svc)
		}
	}
	return a, nil
}

// Registry is every registered service.
func (a *API) Registry() *Registry { return a.registry }

// Start starts a delivery worker per service with a URL. Calling it twice
// is harmless.
func (a *API) Start(ctx context.Context) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.started || a.rooms == nil || len(a.workers) == 0 {
		return
	}
	a.started = true
	ctx, a.cancel = context.WithCancel(context.WithoutCancel(ctx))
	for _, w := range a.workers {
		a.wg.Add(1)
		go a.run(ctx, w)
	}
	a.log.Info("application service delivery started", "services", len(a.workers))
}

// Close stops the workers and waits for them. A transaction being sent is
// abandoned and sent again, with the same ID, on the next start.
func (a *API) Close() {
	a.mu.Lock()
	cancel := a.cancel
	a.mu.Unlock()
	if cancel != nil {
		cancel()
	}
	a.wg.Wait()
}

// EventPersisted implements roomserver.Notifier: it wakes every worker.
// Deciding whether the event concerns a service is the worker's job, done
// off the room's write lock.
func (a *API) EventPersisted(context.Context, roomserver.Notification) {
	for _, w := range a.workers {
		w.poke()
	}
}

// Statuses reports the delivery state of every service with a URL, in
// registration order.
func (a *API) Statuses() []Status {
	out := make([]Status, 0, len(a.workers))
	for _, svc := range a.registry.Services() {
		if w, ok := a.workers[svc.ID]; ok {
			out = append(out, w.snapshot())
		}
	}
	return out
}

// QueryUser asks the services whose namespaces match a user whether it
// exists, stopping at the first that says yes. A service that fails to
// answer counts as a no, and is logged: the request that wanted the user
// goes on without it rather than failing.
func (a *API) QueryUser(ctx context.Context, userID string) bool {
	for _, svc := range a.registry.UserOwners(userID) {
		ok, err := a.client.QueryUser(ctx, svc, userID)
		if err != nil {
			a.log.Warn("an application service failed a user query", "appservice", svc.ID, "user_id", userID, "error", err)
			continue
		}
		if ok {
			return true
		}
	}
	return false
}

// QueryAlias is QueryUser for a room alias.
func (a *API) QueryAlias(ctx context.Context, alias string) bool {
	for _, svc := range a.registry.AliasOwners(alias) {
		ok, err := a.client.QueryAlias(ctx, svc, alias)
		if err != nil {
			a.log.Warn("an application service failed an alias query", "appservice", svc.ID, "alias", alias, "error", err)
			continue
		}
		if ok {
			return true
		}
	}
	return false
}

// Ping makes the homeserver ping a service, for the service's own
// POST /_matrix/client/v1/appservice/{appserviceId}/ping. A service that
// answers has any transaction waiting out a backoff sent at once.
func (a *API) Ping(ctx context.Context, svc *Service, transactionID string) (time.Duration, error) {
	took, err := a.client.Ping(ctx, svc, transactionID)
	if err != nil {
		return 0, err
	}
	if w, ok := a.workers[svc.ID]; ok {
		w.retryNow()
	}
	return took, nil
}

// WaitIdle blocks until every worker has caught up with the top of the
// stream, or ctx ends. Tests use it to know a transaction has been sent.
func (a *API) WaitIdle(ctx context.Context) error {
	if a.rooms == nil {
		return nil
	}
	for {
		top, err := a.rooms.StreamTop(ctx)
		if err != nil {
			return err
		}
		idle := true
		for _, w := range a.workers {
			if w.snapshot().StreamPos < top {
				idle = false
				break
			}
		}
		if idle {
			return nil
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("application services are still delivering: %w", ctx.Err())
		case <-time.After(10 * time.Millisecond):
		}
	}
}