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/domain"
12	"bnema.dev/ermite/internal/ports"
13)
14
15// eventService implements ermiteconnect.EventServiceHandler.
16type eventService struct{ events ports.EventService }
17
18// Watch streams events until the client goes away or the server stops. The
19// stream ends with success in both cases: the contract has no error for
20// "server is shutting down". It ends with unauthenticated when the token of
21// the caller is revoked or expires, and a caller that has too many streams
22// open gets resource_exhausted.
23func (s *eventService) Watch(ctx context.Context, req *connect.Request[pb.WatchRequest], stream *connect.ServerStream[pb.WatchResponse]) error {
24	params := ports.WatchParams{
25		Topics: req.Msg.GetTopics(), ResumeAfter: req.Msg.GetResumeAfter(),
26		Client: clientFromContext(ctx),
27		// A client's call returns when the response begins: send the headers
28		// now, not with the first event, which may be far away. Connect flushes
29		// them for a nil message. A failure shows on the first real send.
30		Ready: func() { _ = stream.Send(nil) },
31	}
32	err := s.events.Watch(ctx, caller(ctx), params, func(we ports.WatchEvent) error {
33		ev := watchEventToPB(we)
34		if ev == nil {
35			return nil
36		}
37		return stream.Send(&pb.WatchResponse{Event: ev})
38	})
39	return connectError(ctx, err)
40}
41
42func releaseEventToPB(we ports.WatchEvent) *pb.Event {
43	out := &pb.Event{Id: we.ID}
44	switch ev := we.Event.(type) {
45	case domain.ReleasePublished:
46		r := ev.Release
47		out.Time = timestamp(ev.At)
48		out.Kind = &pb.Event_ReleasePublished{ReleasePublished: &pb.ReleasePublished{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}}
49	case domain.ReleaseUpdated:
50		r := ev.Release
51		out.Time = timestamp(ev.At)
52		out.Kind = &pb.Event_ReleaseUpdated{ReleaseUpdated: &pb.ReleaseUpdated{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}}
53	case domain.ReleaseDeleted:
54		r := ev.Release
55		out.Time = timestamp(ev.At)
56		out.Kind = &pb.Event_ReleaseDeleted{ReleaseDeleted: &pb.ReleaseDeleted{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag}}
57	case domain.ReleaseUnpublished:
58		r := ev.Release
59		out.Time = timestamp(ev.At)
60		out.Kind = &pb.Event_ReleaseUnpublished{ReleaseUnpublished: &pb.ReleaseUnpublished{Repo: r.Repo.String(), Id: r.ID, Tag: r.Tag, Draft: r.Draft, Prerelease: r.Prerelease}}
61	default:
62		return nil
63	}
64	return out
65}