event.go RawHistoryFolder

1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package rpc
4
5import (
6	"context"
7
8	"connectrpc.com/connect"
9
10	pb "bnema.dev/ermite-api/gen/ermite"
11	"bnema.dev/ermite/internal/ports"
12)
13
14// eventService implements ermiteconnect.EventServiceHandler.
15type eventService struct{ events ports.EventService }
16
17// Watch streams events until the client goes away or the server stops. The
18// stream ends with success in both cases: the contract has no error for
19// "server is shutting down". It ends with unauthenticated when the token of
20// the caller is revoked or expires, and a caller that has too many streams
21// open gets resource_exhausted.
22func (s *eventService) Watch(ctx context.Context, req *connect.Request[pb.WatchRequest], stream *connect.ServerStream[pb.WatchResponse]) error {
23	params := ports.WatchParams{
24		Topics: req.Msg.GetTopics(), ResumeAfter: req.Msg.GetResumeAfter(),
25		Client: clientFromContext(ctx),
26		// A client's call returns when the response begins: send the headers
27		// now, not with the first event, which may be far away. Connect flushes
28		// them for a nil message. A failure shows on the first real send.
29		Ready: func() { _ = stream.Send(nil) },
30	}
31	err := s.events.Watch(ctx, caller(ctx), params, func(we ports.WatchEvent) error {
32		ev := watchEventToPB(we)
33		if ev == nil {
34			return nil
35		}
36		return stream.Send(&pb.WatchResponse{Event: ev})
37	})
38	return connectError(ctx, err)
39}