detached 46d89a8

feat(client): add Connect client adapter for the remote CLI

commit
46d89a83c500b0a70f6b664b697f1b7a457684d9
author
brice · · 2026-10-10 08:50 UTC
committer
brice · · 2026-10-10 09:42 UTC
parent
6a0af58
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
12 files · +1262 -0
Unified
UnifiedSplit
A internal/adapters/client/client.go +124 -0
@@ -0,0 +1,124 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import (
6+ "context"
7+ "fmt"
8+ "net"
9+ "net/http"
10+ "time"
11+
12+ "connectrpc.com/connect"
13+
14+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
15+)
16+
17+// Client gives access to the services of one ermite server. Every call carries
18+// the bearer token and the user agent; unary calls are bounded by the timeout.
19+// It is safe for concurrent use.
20+type Client struct {
21+ cfg Config
22+ base string
23+ http *http.Client
24+ opts []connect.ClientOption
25+}
26+
27+// New builds a Client from a validated configuration.
28+func New(cfg Config) (*Client, error) {
29+ u, err := ParseURL(cfg.URL)
30+ if err != nil {
31+ return nil, fmt.Errorf("%s: %w", EnvURL, err)
32+ }
33+ if cfg.Timeout <= 0 {
34+ cfg.Timeout = DefaultTimeout
35+ }
36+ cfg.URL = u.String()
37+ hc := &http.Client{
38+ Transport: transport(),
39+ // A redirect would replay a request elsewhere: report it instead.
40+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
41+ }
42+ return &Client{
43+ cfg: cfg,
44+ base: u.JoinPath(APIPath).String(),
45+ http: hc,
46+ opts: []connect.ClientOption{connect.WithInterceptors(headers{token: cfg.Token, agent: cfg.UserAgent, timeout: cfg.Timeout})},
47+ }, nil
48+}
49+
50+// transport bounds connection setup but not the exchange, which may be a
51+// long-lived stream. HTTP/2 pings detect a connection that died silently.
52+func transport() *http.Transport {
53+ t := http.DefaultTransport.(*http.Transport).Clone()
54+ t.DialContext = (&net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}).DialContext
55+ t.TLSHandshakeTimeout = 10 * time.Second
56+ t.HTTP2 = &http.HTTP2Config{SendPingTimeout: 30 * time.Second, PingTimeout: 15 * time.Second}
57+ return t
58+}
59+
60+// URL is the server address the client talks to.
61+func (c *Client) URL() string { return c.cfg.URL }
62+
63+// Repos returns the RepoService client.
64+func (c *Client) Repos() ermiteconnect.RepoServiceClient {
65+ return ermiteconnect.NewRepoServiceClient(c.http, c.base, c.opts...)
66+}
67+
68+// Tokens returns the TokenService client.
69+func (c *Client) Tokens() ermiteconnect.TokenServiceClient {
70+ return ermiteconnect.NewTokenServiceClient(c.http, c.base, c.opts...)
71+}
72+
73+// Events returns the EventService client.
74+func (c *Client) Events() ermiteconnect.EventServiceClient {
75+ return ermiteconnect.NewEventServiceClient(c.http, c.base, c.opts...)
76+}
77+
78+// Server returns the ServerService client.
79+func (c *Client) Server() ermiteconnect.ServerServiceClient {
80+ return ermiteconnect.NewServerServiceClient(c.http, c.base, c.opts...)
81+}
82+
83+// headers sets the credentials and the user agent on every call and bounds the
84+// unary ones.
85+type headers struct {
86+ token string
87+ agent string
88+ timeout time.Duration
89+}
90+
91+func (h headers) set(hd http.Header) {
92+ if h.token != "" {
93+ hd.Set("Authorization", "Bearer "+h.token)
94+ }
95+ if h.agent != "" {
96+ hd.Set("User-Agent", h.agent)
97+ }
98+}
99+
100+func (h headers) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
101+ return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
102+ if req.Spec().IsClient {
103+ h.set(req.Header())
104+ if h.timeout > 0 {
105+ var cancel context.CancelFunc
106+ ctx, cancel = context.WithTimeout(ctx, h.timeout)
107+ defer cancel()
108+ }
109+ }
110+ return next(ctx, req)
111+ }
112+}
113+
114+func (h headers) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
115+ return func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
116+ conn := next(ctx, spec)
117+ h.set(conn.RequestHeader())
118+ return conn
119+ }
120+}
121+
122+func (headers) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
123+ return next
124+}
@@ -0,0 +1,124 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import (
6+ "context"
7+ "fmt"
8+ "net"
9+ "net/http"
10+ "time"
11+
12+ "connectrpc.com/connect"
13+
14+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
15+)
16+
17+// Client gives access to the services of one ermite server. Every call carries
18+// the bearer token and the user agent; unary calls are bounded by the timeout.
19+// It is safe for concurrent use.
20+type Client struct {
21+ cfg Config
22+ base string
23+ http *http.Client
24+ opts []connect.ClientOption
25+}
26+
27+// New builds a Client from a validated configuration.
28+func New(cfg Config) (*Client, error) {
29+ u, err := ParseURL(cfg.URL)
30+ if err != nil {
31+ return nil, fmt.Errorf("%s: %w", EnvURL, err)
32+ }
33+ if cfg.Timeout <= 0 {
34+ cfg.Timeout = DefaultTimeout
35+ }
36+ cfg.URL = u.String()
37+ hc := &http.Client{
38+ Transport: transport(),
39+ // A redirect would replay a request elsewhere: report it instead.
40+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
41+ }
42+ return &Client{
43+ cfg: cfg,
44+ base: u.JoinPath(APIPath).String(),
45+ http: hc,
46+ opts: []connect.ClientOption{connect.WithInterceptors(headers{token: cfg.Token, agent: cfg.UserAgent, timeout: cfg.Timeout})},
47+ }, nil
48+}
49+
50+// transport bounds connection setup but not the exchange, which may be a
51+// long-lived stream. HTTP/2 pings detect a connection that died silently.
52+func transport() *http.Transport {
53+ t := http.DefaultTransport.(*http.Transport).Clone()
54+ t.DialContext = (&net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}).DialContext
55+ t.TLSHandshakeTimeout = 10 * time.Second
56+ t.HTTP2 = &http.HTTP2Config{SendPingTimeout: 30 * time.Second, PingTimeout: 15 * time.Second}
57+ return t
58+}
59+
60+// URL is the server address the client talks to.
61+func (c *Client) URL() string { return c.cfg.URL }
62+
63+// Repos returns the RepoService client.
64+func (c *Client) Repos() ermiteconnect.RepoServiceClient {
65+ return ermiteconnect.NewRepoServiceClient(c.http, c.base, c.opts...)
66+}
67+
68+// Tokens returns the TokenService client.
69+func (c *Client) Tokens() ermiteconnect.TokenServiceClient {
70+ return ermiteconnect.NewTokenServiceClient(c.http, c.base, c.opts...)
71+}
72+
73+// Events returns the EventService client.
74+func (c *Client) Events() ermiteconnect.EventServiceClient {
75+ return ermiteconnect.NewEventServiceClient(c.http, c.base, c.opts...)
76+}
77+
78+// Server returns the ServerService client.
79+func (c *Client) Server() ermiteconnect.ServerServiceClient {
80+ return ermiteconnect.NewServerServiceClient(c.http, c.base, c.opts...)
81+}
82+
83+// headers sets the credentials and the user agent on every call and bounds the
84+// unary ones.
85+type headers struct {
86+ token string
87+ agent string
88+ timeout time.Duration
89+}
90+
91+func (h headers) set(hd http.Header) {
92+ if h.token != "" {
93+ hd.Set("Authorization", "Bearer "+h.token)
94+ }
95+ if h.agent != "" {
96+ hd.Set("User-Agent", h.agent)
97+ }
98+}
99+
100+func (h headers) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
101+ return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) {
102+ if req.Spec().IsClient {
103+ h.set(req.Header())
104+ if h.timeout > 0 {
105+ var cancel context.CancelFunc
106+ ctx, cancel = context.WithTimeout(ctx, h.timeout)
107+ defer cancel()
108+ }
109+ }
110+ return next(ctx, req)
111+ }
112+}
113+
114+func (h headers) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc {
115+ return func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn {
116+ conn := next(ctx, spec)
117+ h.set(conn.RequestHeader())
118+ return conn
119+ }
120+}
121+
122+func (headers) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
123+ return next
124+}
A internal/adapters/client/client_test.go +141 -0
@@ -0,0 +1,141 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "context"
7+ "errors"
8+ "testing"
9+ "time"
10+
11+ "connectrpc.com/connect"
12+ "github.com/stretchr/testify/assert"
13+ "github.com/stretchr/testify/require"
14+
15+ "bnema.dev/ermite-api/gen/ermite"
16+ "bnema.dev/ermite/internal/adapters/client"
17+)
18+
19+func TestCallsCarryTokenAndUserAgent(t *testing.T) {
20+ f := &fake{version: func(context.Context, *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error) {
21+ return connect.NewResponse(&ermite.GetVersionResponse{Version: "v1"}), nil
22+ }}
23+ c := start(t, f, client.Config{Token: "ermite_secret", UserAgent: "ermite/v9 (cli)"})
24+ res, err := c.Server().GetVersion(t.Context(), connect.NewRequest(&ermite.GetVersionRequest{}))
25+ require.NoError(t, err)
26+ assert.Equal(t, "v1", res.Msg.GetVersion())
27+ require.Len(t, f.headers, 1)
28+ assert.Equal(t, "Bearer ermite_secret", f.headers[0].Get("Authorization"))
29+ assert.Equal(t, "ermite/v9 (cli)", f.headers[0].Get("User-Agent"))
30+}
31+
32+func TestAnonymousCallsSendNoAuthorization(t *testing.T) {
33+ f := &fake{version: func(context.Context, *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error) {
34+ return connect.NewResponse(&ermite.GetVersionResponse{}), nil
35+ }}
36+ c := start(t, f, client.Config{})
37+ _, err := c.Server().GetVersion(t.Context(), connect.NewRequest(&ermite.GetVersionRequest{}))
38+ require.NoError(t, err)
39+ assert.Empty(t, f.headers[0].Values("Authorization"))
40+}
41+
42+func TestUnaryCallsTimeOut(t *testing.T) {
43+ f := &fake{version: func(ctx context.Context, _ *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error) {
44+ <-ctx.Done()
45+ return nil, ctx.Err()
46+ }}
47+ c := start(t, f, client.Config{Timeout: 50 * time.Millisecond})
48+ start := time.Now()
49+ _, err := c.Server().GetVersion(t.Context(), connect.NewRequest(&ermite.GetVersionRequest{}))
50+ require.Error(t, err)
51+ assert.Less(t, time.Since(start), 5*time.Second)
52+ assert.Equal(t, connect.CodeDeadlineExceeded, connect.CodeOf(err))
53+ assert.Contains(t, c.Describe(err).Error(), "timed out")
54+}
55+
56+func TestRedirectIsNotFollowed(t *testing.T) {
57+ f := &fake{}
58+ c := start(t, f, client.Config{Token: "t"})
59+ _ = c // the redirect target below is a plain server
60+ redirect := redirectingServer(t)
61+ rc, err := client.New(client.Config{URL: redirect, Token: "t"})
62+ require.NoError(t, err)
63+ _, err = rc.Server().GetVersion(t.Context(), connect.NewRequest(&ermite.GetVersionRequest{}))
64+ require.Error(t, err)
65+ assert.Contains(t, rc.Describe(err).Error(), "redirected")
66+}
67+
68+func TestNewRejectsBadURL(t *testing.T) {
69+ _, err := client.New(client.Config{URL: "ftp://x"})
70+ require.Error(t, err)
71+}
72+
73+func TestDescribe(t *testing.T) {
74+ anon, err := client.New(client.Config{URL: "http://srv.test"})
75+ require.NoError(t, err)
76+ authed, err := client.New(client.Config{URL: "http://srv.test", Token: "t"})
77+ require.NoError(t, err)
78+
79+ tests := []struct {
80+ name string
81+ c *client.Client
82+ err error
83+ want []string
84+ }{
85+ {"unauthenticated anonymous", anon, connect.NewError(connect.CodeUnauthenticated, errors.New("no token")),
86+ []string{"authentication required", "ERMITE_TOKEN"}},
87+ {"unauthenticated bad token", authed, connect.NewError(connect.CodeUnauthenticated, errors.New("bad")),
88+ []string{"not authenticated", "invalid, expired or revoked"}},
89+ {"not found", authed, connect.NewError(connect.CodeNotFound, errors.New("repository \"x\" not found")),
90+ []string{"not found: repository \"x\" not found"}},
91+ {"permission denied", authed, connect.NewError(connect.CodePermissionDenied, errors.New("missing scope repo:admin")),
92+ []string{"permission denied: missing scope repo:admin", "token list"}},
93+ {"already exists", authed, connect.NewError(connect.CodeAlreadyExists, errors.New("name taken")),
94+ []string{"already exists: name taken"}},
95+ {"invalid", authed, connect.NewError(connect.CodeInvalidArgument, errors.New("bad name")),
96+ []string{"invalid request: bad name"}},
97+ {"precondition", authed, connect.NewError(connect.CodeFailedPrecondition, errors.New("no such branch")),
98+ []string{"precondition failed: no such branch"}},
99+ {"unavailable", authed, connect.NewError(connect.CodeUnavailable, errors.New("connection refused")),
100+ []string{"cannot reach the server at http://srv.test", "connection refused"}},
101+ {"unimplemented", authed, connect.NewError(connect.CodeUnimplemented, errors.New("x")),
102+ []string{"does not implement", "ermite version"}},
103+ {"multi-line message", authed, connect.NewError(connect.CodeInternal, errors.New("line1\nline2\n\tline3")),
104+ []string{"server error: line1 line2 line3"}},
105+ {"empty message", authed, connect.NewError(connect.CodeNotFound, errors.New("")),
106+ []string{"not found"}},
107+ }
108+ for _, tt := range tests {
109+ t.Run(tt.name, func(t *testing.T) {
110+ got := tt.c.Describe(tt.err)
111+ require.Error(t, got)
112+ for _, w := range tt.want {
113+ assert.Contains(t, got.Error(), w)
114+ }
115+ assert.NotContains(t, got.Error(), "\n")
116+ assert.NotContains(t, got.Error(), "[", "no connect code prefix")
117+ assert.ErrorIs(t, got, tt.err, "the original error stays reachable")
118+ })
119+ }
120+
121+ plain := errors.New("boom")
122+ assert.Same(t, plain, anon.Describe(plain))
123+ assert.NoError(t, anon.Describe(nil))
124+}
125+
126+func TestDescribeBoundsLongMessages(t *testing.T) {
127+ c, err := client.New(client.Config{URL: "http://srv.test"})
128+ require.NoError(t, err)
129+ long := make([]byte, 5000)
130+ for i := range long {
131+ long[i] = 'a'
132+ }
133+ got := c.Describe(connect.NewError(connect.CodeInternal, errors.New(string(long))))
134+ assert.Less(t, len(got.Error()), 400)
135+}
136+
137+func TestMissingFeatures(t *testing.T) {
138+ assert.Empty(t, client.MissingFeatures(client.UsedFeatures()))
139+ assert.Equal(t, []string{client.FeatureEventsResume}, client.MissingFeatures([]string{client.FeatureRepoUpdate, "other"}))
140+ assert.Equal(t, client.UsedFeatures(), client.MissingFeatures(nil))
141+}
@@ -0,0 +1,141 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "context"
7+ "errors"
8+ "testing"
9+ "time"
10+
11+ "connectrpc.com/connect"
12+ "github.com/stretchr/testify/assert"
13+ "github.com/stretchr/testify/require"
14+
15+ "bnema.dev/ermite-api/gen/ermite"
16+ "bnema.dev/ermite/internal/adapters/client"
17+)
18+
19+func TestCallsCarryTokenAndUserAgent(t *testing.T) {
20+ f := &fake{version: func(context.Context, *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error) {
21+ return connect.NewResponse(&ermite.GetVersionResponse{Version: "v1"}), nil
22+ }}
23+ c := start(t, f, client.Config{Token: "ermite_secret", UserAgent: "ermite/v9 (cli)"})
24+ res, err := c.Server().GetVersion(t.Context(), connect.NewRequest(&ermite.GetVersionRequest{}))
25+ require.NoError(t, err)
26+ assert.Equal(t, "v1", res.Msg.GetVersion())
27+ require.Len(t, f.headers, 1)
28+ assert.Equal(t, "Bearer ermite_secret", f.headers[0].Get("Authorization"))
29+ assert.Equal(t, "ermite/v9 (cli)", f.headers[0].Get("User-Agent"))
30+}
31+
32+func TestAnonymousCallsSendNoAuthorization(t *testing.T) {
33+ f := &fake{version: func(context.Context, *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error) {
34+ return connect.NewResponse(&ermite.GetVersionResponse{}), nil
35+ }}
36+ c := start(t, f, client.Config{})
37+ _, err := c.Server().GetVersion(t.Context(), connect.NewRequest(&ermite.GetVersionRequest{}))
38+ require.NoError(t, err)
39+ assert.Empty(t, f.headers[0].Values("Authorization"))
40+}
41+
42+func TestUnaryCallsTimeOut(t *testing.T) {
43+ f := &fake{version: func(ctx context.Context, _ *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error) {
44+ <-ctx.Done()
45+ return nil, ctx.Err()
46+ }}
47+ c := start(t, f, client.Config{Timeout: 50 * time.Millisecond})
48+ start := time.Now()
49+ _, err := c.Server().GetVersion(t.Context(), connect.NewRequest(&ermite.GetVersionRequest{}))
50+ require.Error(t, err)
51+ assert.Less(t, time.Since(start), 5*time.Second)
52+ assert.Equal(t, connect.CodeDeadlineExceeded, connect.CodeOf(err))
53+ assert.Contains(t, c.Describe(err).Error(), "timed out")
54+}
55+
56+func TestRedirectIsNotFollowed(t *testing.T) {
57+ f := &fake{}
58+ c := start(t, f, client.Config{Token: "t"})
59+ _ = c // the redirect target below is a plain server
60+ redirect := redirectingServer(t)
61+ rc, err := client.New(client.Config{URL: redirect, Token: "t"})
62+ require.NoError(t, err)
63+ _, err = rc.Server().GetVersion(t.Context(), connect.NewRequest(&ermite.GetVersionRequest{}))
64+ require.Error(t, err)
65+ assert.Contains(t, rc.Describe(err).Error(), "redirected")
66+}
67+
68+func TestNewRejectsBadURL(t *testing.T) {
69+ _, err := client.New(client.Config{URL: "ftp://x"})
70+ require.Error(t, err)
71+}
72+
73+func TestDescribe(t *testing.T) {
74+ anon, err := client.New(client.Config{URL: "http://srv.test"})
75+ require.NoError(t, err)
76+ authed, err := client.New(client.Config{URL: "http://srv.test", Token: "t"})
77+ require.NoError(t, err)
78+
79+ tests := []struct {
80+ name string
81+ c *client.Client
82+ err error
83+ want []string
84+ }{
85+ {"unauthenticated anonymous", anon, connect.NewError(connect.CodeUnauthenticated, errors.New("no token")),
86+ []string{"authentication required", "ERMITE_TOKEN"}},
87+ {"unauthenticated bad token", authed, connect.NewError(connect.CodeUnauthenticated, errors.New("bad")),
88+ []string{"not authenticated", "invalid, expired or revoked"}},
89+ {"not found", authed, connect.NewError(connect.CodeNotFound, errors.New("repository \"x\" not found")),
90+ []string{"not found: repository \"x\" not found"}},
91+ {"permission denied", authed, connect.NewError(connect.CodePermissionDenied, errors.New("missing scope repo:admin")),
92+ []string{"permission denied: missing scope repo:admin", "token list"}},
93+ {"already exists", authed, connect.NewError(connect.CodeAlreadyExists, errors.New("name taken")),
94+ []string{"already exists: name taken"}},
95+ {"invalid", authed, connect.NewError(connect.CodeInvalidArgument, errors.New("bad name")),
96+ []string{"invalid request: bad name"}},
97+ {"precondition", authed, connect.NewError(connect.CodeFailedPrecondition, errors.New("no such branch")),
98+ []string{"precondition failed: no such branch"}},
99+ {"unavailable", authed, connect.NewError(connect.CodeUnavailable, errors.New("connection refused")),
100+ []string{"cannot reach the server at http://srv.test", "connection refused"}},
101+ {"unimplemented", authed, connect.NewError(connect.CodeUnimplemented, errors.New("x")),
102+ []string{"does not implement", "ermite version"}},
103+ {"multi-line message", authed, connect.NewError(connect.CodeInternal, errors.New("line1\nline2\n\tline3")),
104+ []string{"server error: line1 line2 line3"}},
105+ {"empty message", authed, connect.NewError(connect.CodeNotFound, errors.New("")),
106+ []string{"not found"}},
107+ }
108+ for _, tt := range tests {
109+ t.Run(tt.name, func(t *testing.T) {
110+ got := tt.c.Describe(tt.err)
111+ require.Error(t, got)
112+ for _, w := range tt.want {
113+ assert.Contains(t, got.Error(), w)
114+ }
115+ assert.NotContains(t, got.Error(), "\n")
116+ assert.NotContains(t, got.Error(), "[", "no connect code prefix")
117+ assert.ErrorIs(t, got, tt.err, "the original error stays reachable")
118+ })
119+ }
120+
121+ plain := errors.New("boom")
122+ assert.Same(t, plain, anon.Describe(plain))
123+ assert.NoError(t, anon.Describe(nil))
124+}
125+
126+func TestDescribeBoundsLongMessages(t *testing.T) {
127+ c, err := client.New(client.Config{URL: "http://srv.test"})
128+ require.NoError(t, err)
129+ long := make([]byte, 5000)
130+ for i := range long {
131+ long[i] = 'a'
132+ }
133+ got := c.Describe(connect.NewError(connect.CodeInternal, errors.New(string(long))))
134+ assert.Less(t, len(got.Error()), 400)
135+}
136+
137+func TestMissingFeatures(t *testing.T) {
138+ assert.Empty(t, client.MissingFeatures(client.UsedFeatures()))
139+ assert.Equal(t, []string{client.FeatureEventsResume}, client.MissingFeatures([]string{client.FeatureRepoUpdate, "other"}))
140+ assert.Equal(t, client.UsedFeatures(), client.MissingFeatures(nil))
141+}
A internal/adapters/client/config.go +113 -0
@@ -0,0 +1,113 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+// Package client is the Connect client of the remote CLI: it turns the
4+// ERMITE_URL and ERMITE_TOKEN settings into authenticated service clients for
5+// the ermite-api contract, maps Connect errors to one-line messages and keeps
6+// a Watch stream alive across reconnections.
7+//
8+// It is a driving adapter: it never imports another adapter and never reads
9+// the process environment itself, the caller passes a lookup function.
10+package client
11+
12+import (
13+ "errors"
14+ "fmt"
15+ "net"
16+ "net/url"
17+ "strings"
18+ "time"
19+)
20+
21+// Variable names.
22+const (
23+ // EnvURL is the base URL of the ermite server.
24+ EnvURL = "ERMITE_URL"
25+ // EnvToken is the bearer token sent with every request. Optional: without
26+ // it the calls are anonymous.
27+ EnvToken = "ERMITE_TOKEN"
28+)
29+
30+const (
31+ // APIPath is where the server mounts the Connect handlers.
32+ APIPath = "/-/api"
33+ // DefaultTimeout bounds one unary call.
34+ DefaultTimeout = 30 * time.Second
35+)
36+
37+// Config configures a Client.
38+type Config struct {
39+ // URL is the server address, http or https, without credentials. A path
40+ // prefix is kept, for a server behind a sub-path proxy.
41+ URL string
42+ // Token is the bearer token. Empty means anonymous.
43+ Token string
44+ // Timeout bounds each unary call. Streams are not bounded. Zero means
45+ // DefaultTimeout.
46+ Timeout time.Duration
47+ // UserAgent identifies the CLI, for example "ermite/v0.1.0 (cli)".
48+ UserAgent string
49+}
50+
51+// ConfigFromEnv reads ERMITE_URL and ERMITE_TOKEN through getenv and validates
52+// them. Timeout and UserAgent are left for the caller to set.
53+func ConfigFromEnv(getenv func(string) string) (Config, error) {
54+ raw := strings.TrimSpace(getenv(EnvURL))
55+ if raw == "" {
56+ return Config{}, fmt.Errorf("%s is not set: set it to the address of your ermite server, for example https://git.example.com", EnvURL)
57+ }
58+ base, err := ParseURL(raw)
59+ if err != nil {
60+ return Config{}, fmt.Errorf("%s: %w", EnvURL, err)
61+ }
62+ return Config{URL: base.String(), Token: strings.TrimSpace(getenv(EnvToken))}, nil
63+}
64+
65+// ParseURL validates a server address: an absolute http(s) URL with a host,
66+// without credentials, query or fragment. The result has no trailing slash.
67+// Errors never echo a URL that may carry credentials.
68+func ParseURL(raw string) (*url.URL, error) {
69+ u, err := url.Parse(raw)
70+ if err != nil {
71+ return nil, errors.New("not a valid URL")
72+ }
73+ switch {
74+ case u.User != nil:
75+ return nil, errors.New("must not contain credentials: use ERMITE_TOKEN for the token")
76+ case u.Scheme != "http" && u.Scheme != "https":
77+ return nil, fmt.Errorf("%q is not an http(s) URL: it must start with http:// or https://", raw)
78+ case u.Hostname() == "":
79+ return nil, fmt.Errorf("%q has no host", raw)
80+ case u.RawQuery != "" || u.ForceQuery || u.Fragment != "":
81+ return nil, fmt.Errorf("%q must not contain a query or a fragment", raw)
82+ }
83+ u.Scheme = strings.ToLower(u.Scheme)
84+ u.Path = strings.TrimRight(u.Path, "/")
85+ u.RawPath = ""
86+ return u, nil
87+}
88+
89+// InsecureTokenWarning returns a warning when the token would travel over
90+// plain http to a host that is not on the loopback interface, and "" otherwise.
91+//
92+// It warns instead of refusing: ermite does not terminate TLS itself and is
93+// commonly reached over a VPN or a private network, where http is a legitimate
94+// deployment; a refusal would make those setups unusable. The warning keeps
95+// the risk visible on every call.
96+func (c Config) InsecureTokenWarning() string {
97+ if c.Token == "" {
98+ return ""
99+ }
100+ u, err := url.Parse(c.URL)
101+ if err != nil || u.Scheme != "http" || isLoopback(u.Hostname()) {
102+ return ""
103+ }
104+ return fmt.Sprintf("%s is sent over plain http to %s: anyone on the network path can read it; use an https:// URL unless the network is trusted", EnvToken, u.Hostname())
105+}
106+
107+func isLoopback(host string) bool {
108+ if host == "localhost" || strings.HasSuffix(host, ".localhost") {
109+ return true
110+ }
111+ ip := net.ParseIP(host)
112+ return ip != nil && ip.IsLoopback()
113+}
@@ -0,0 +1,113 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+// Package client is the Connect client of the remote CLI: it turns the
4+// ERMITE_URL and ERMITE_TOKEN settings into authenticated service clients for
5+// the ermite-api contract, maps Connect errors to one-line messages and keeps
6+// a Watch stream alive across reconnections.
7+//
8+// It is a driving adapter: it never imports another adapter and never reads
9+// the process environment itself, the caller passes a lookup function.
10+package client
11+
12+import (
13+ "errors"
14+ "fmt"
15+ "net"
16+ "net/url"
17+ "strings"
18+ "time"
19+)
20+
21+// Variable names.
22+const (
23+ // EnvURL is the base URL of the ermite server.
24+ EnvURL = "ERMITE_URL"
25+ // EnvToken is the bearer token sent with every request. Optional: without
26+ // it the calls are anonymous.
27+ EnvToken = "ERMITE_TOKEN"
28+)
29+
30+const (
31+ // APIPath is where the server mounts the Connect handlers.
32+ APIPath = "/-/api"
33+ // DefaultTimeout bounds one unary call.
34+ DefaultTimeout = 30 * time.Second
35+)
36+
37+// Config configures a Client.
38+type Config struct {
39+ // URL is the server address, http or https, without credentials. A path
40+ // prefix is kept, for a server behind a sub-path proxy.
41+ URL string
42+ // Token is the bearer token. Empty means anonymous.
43+ Token string
44+ // Timeout bounds each unary call. Streams are not bounded. Zero means
45+ // DefaultTimeout.
46+ Timeout time.Duration
47+ // UserAgent identifies the CLI, for example "ermite/v0.1.0 (cli)".
48+ UserAgent string
49+}
50+
51+// ConfigFromEnv reads ERMITE_URL and ERMITE_TOKEN through getenv and validates
52+// them. Timeout and UserAgent are left for the caller to set.
53+func ConfigFromEnv(getenv func(string) string) (Config, error) {
54+ raw := strings.TrimSpace(getenv(EnvURL))
55+ if raw == "" {
56+ return Config{}, fmt.Errorf("%s is not set: set it to the address of your ermite server, for example https://git.example.com", EnvURL)
57+ }
58+ base, err := ParseURL(raw)
59+ if err != nil {
60+ return Config{}, fmt.Errorf("%s: %w", EnvURL, err)
61+ }
62+ return Config{URL: base.String(), Token: strings.TrimSpace(getenv(EnvToken))}, nil
63+}
64+
65+// ParseURL validates a server address: an absolute http(s) URL with a host,
66+// without credentials, query or fragment. The result has no trailing slash.
67+// Errors never echo a URL that may carry credentials.
68+func ParseURL(raw string) (*url.URL, error) {
69+ u, err := url.Parse(raw)
70+ if err != nil {
71+ return nil, errors.New("not a valid URL")
72+ }
73+ switch {
74+ case u.User != nil:
75+ return nil, errors.New("must not contain credentials: use ERMITE_TOKEN for the token")
76+ case u.Scheme != "http" && u.Scheme != "https":
77+ return nil, fmt.Errorf("%q is not an http(s) URL: it must start with http:// or https://", raw)
78+ case u.Hostname() == "":
79+ return nil, fmt.Errorf("%q has no host", raw)
80+ case u.RawQuery != "" || u.ForceQuery || u.Fragment != "":
81+ return nil, fmt.Errorf("%q must not contain a query or a fragment", raw)
82+ }
83+ u.Scheme = strings.ToLower(u.Scheme)
84+ u.Path = strings.TrimRight(u.Path, "/")
85+ u.RawPath = ""
86+ return u, nil
87+}
88+
89+// InsecureTokenWarning returns a warning when the token would travel over
90+// plain http to a host that is not on the loopback interface, and "" otherwise.
91+//
92+// It warns instead of refusing: ermite does not terminate TLS itself and is
93+// commonly reached over a VPN or a private network, where http is a legitimate
94+// deployment; a refusal would make those setups unusable. The warning keeps
95+// the risk visible on every call.
96+func (c Config) InsecureTokenWarning() string {
97+ if c.Token == "" {
98+ return ""
99+ }
100+ u, err := url.Parse(c.URL)
101+ if err != nil || u.Scheme != "http" || isLoopback(u.Hostname()) {
102+ return ""
103+ }
104+ return fmt.Sprintf("%s is sent over plain http to %s: anyone on the network path can read it; use an https:// URL unless the network is trusted", EnvToken, u.Hostname())
105+}
106+
107+func isLoopback(host string) bool {
108+ if host == "localhost" || strings.HasSuffix(host, ".localhost") {
109+ return true
110+ }
111+ ip := net.ParseIP(host)
112+ return ip != nil && ip.IsLoopback()
113+}
A internal/adapters/client/config_test.go +69 -0
@@ -0,0 +1,69 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "testing"
7+
8+ "github.com/stretchr/testify/assert"
9+ "github.com/stretchr/testify/require"
10+
11+ "bnema.dev/ermite/internal/adapters/client"
12+)
13+
14+func env(kv map[string]string) func(string) string {
15+ return func(k string) string { return kv[k] }
16+}
17+
18+func TestConfigFromEnv(t *testing.T) {
19+ cfg, err := client.ConfigFromEnv(env(map[string]string{
20+ client.EnvURL: " HTTPS://Git.Example.com/base/ ", client.EnvToken: " ermite_abc\n",
21+ }))
22+ require.NoError(t, err)
23+ assert.Equal(t, "https://Git.Example.com/base", cfg.URL)
24+ assert.Equal(t, "ermite_abc", cfg.Token)
25+}
26+
27+func TestConfigFromEnvRejects(t *testing.T) {
28+ tests := []struct {
29+ name, url, want string
30+ }{
31+ {"unset", "", "ERMITE_URL is not set"},
32+ {"no scheme", "git.example.com", "not an http(s) URL"},
33+ {"ftp", "ftp://git.example.com", "not an http(s) URL"},
34+ {"no host", "https://", "has no host"},
35+ {"credentials", "https://user:s3cret@git.example.com", "must not contain credentials"},
36+ {"query", "https://git.example.com/?a=b", "query or a fragment"},
37+ {"fragment", "https://git.example.com/#x", "query or a fragment"},
38+ {"garbage", "http://[::1", "not a valid URL"},
39+ }
40+ for _, tt := range tests {
41+ t.Run(tt.name, func(t *testing.T) {
42+ _, err := client.ConfigFromEnv(env(map[string]string{client.EnvURL: tt.url}))
43+ require.Error(t, err)
44+ assert.Contains(t, err.Error(), tt.want)
45+ assert.NotContains(t, err.Error(), "s3cret", "credentials must never be echoed")
46+ })
47+ }
48+}
49+
50+func TestInsecureTokenWarning(t *testing.T) {
51+ tests := []struct {
52+ url, token string
53+ warn bool
54+ }{
55+ {"http://git.example.com", "t", true},
56+ {"http://10.0.0.5:7341", "t", true},
57+ {"http://git.example.com", "", false}, // nothing secret to leak
58+ {"https://git.example.com", "t", false},
59+ {"http://localhost:7341", "t", false},
60+ {"http://127.0.0.1:7341", "t", false},
61+ {"http://[::1]:7341", "t", false},
62+ {"http://dev.localhost", "t", false},
63+ }
64+ for _, tt := range tests {
65+ w := client.Config{URL: tt.url, Token: tt.token}.InsecureTokenWarning()
66+ assert.Equal(t, tt.warn, w != "", "%s token=%q: %q", tt.url, tt.token, w)
67+ assert.NotContains(t, w, tt.token+"_", "never echo the token")
68+ }
69+}
@@ -0,0 +1,69 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "testing"
7+
8+ "github.com/stretchr/testify/assert"
9+ "github.com/stretchr/testify/require"
10+
11+ "bnema.dev/ermite/internal/adapters/client"
12+)
13+
14+func env(kv map[string]string) func(string) string {
15+ return func(k string) string { return kv[k] }
16+}
17+
18+func TestConfigFromEnv(t *testing.T) {
19+ cfg, err := client.ConfigFromEnv(env(map[string]string{
20+ client.EnvURL: " HTTPS://Git.Example.com/base/ ", client.EnvToken: " ermite_abc\n",
21+ }))
22+ require.NoError(t, err)
23+ assert.Equal(t, "https://Git.Example.com/base", cfg.URL)
24+ assert.Equal(t, "ermite_abc", cfg.Token)
25+}
26+
27+func TestConfigFromEnvRejects(t *testing.T) {
28+ tests := []struct {
29+ name, url, want string
30+ }{
31+ {"unset", "", "ERMITE_URL is not set"},
32+ {"no scheme", "git.example.com", "not an http(s) URL"},
33+ {"ftp", "ftp://git.example.com", "not an http(s) URL"},
34+ {"no host", "https://", "has no host"},
35+ {"credentials", "https://user:s3cret@git.example.com", "must not contain credentials"},
36+ {"query", "https://git.example.com/?a=b", "query or a fragment"},
37+ {"fragment", "https://git.example.com/#x", "query or a fragment"},
38+ {"garbage", "http://[::1", "not a valid URL"},
39+ }
40+ for _, tt := range tests {
41+ t.Run(tt.name, func(t *testing.T) {
42+ _, err := client.ConfigFromEnv(env(map[string]string{client.EnvURL: tt.url}))
43+ require.Error(t, err)
44+ assert.Contains(t, err.Error(), tt.want)
45+ assert.NotContains(t, err.Error(), "s3cret", "credentials must never be echoed")
46+ })
47+ }
48+}
49+
50+func TestInsecureTokenWarning(t *testing.T) {
51+ tests := []struct {
52+ url, token string
53+ warn bool
54+ }{
55+ {"http://git.example.com", "t", true},
56+ {"http://10.0.0.5:7341", "t", true},
57+ {"http://git.example.com", "", false}, // nothing secret to leak
58+ {"https://git.example.com", "t", false},
59+ {"http://localhost:7341", "t", false},
60+ {"http://127.0.0.1:7341", "t", false},
61+ {"http://[::1]:7341", "t", false},
62+ {"http://dev.localhost", "t", false},
63+ }
64+ for _, tt := range tests {
65+ w := client.Config{URL: tt.url, Token: tt.token}.InsecureTokenWarning()
66+ assert.Equal(t, tt.warn, w != "", "%s token=%q: %q", tt.url, tt.token, w)
67+ assert.NotContains(t, w, tt.token+"_", "never echo the token")
68+ }
69+}
A internal/adapters/client/errors.go +95 -0
@@ -0,0 +1,95 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import (
6+ "errors"
7+ "regexp"
8+ "strings"
9+
10+ "connectrpc.com/connect"
11+)
12+
13+// redirected matches the message Connect gives to a 3xx answer.
14+var redirected = regexp.MustCompile(`\b30[1-8]\b`)
15+
16+// maxMessage caps a server message echoed to the terminal.
17+const maxMessage = 300
18+
19+// Error is a failed call described for a person: one line, with a hint when
20+// the code points to a likely cause. It wraps the original error.
21+type Error struct {
22+ msg string
23+ err error
24+}
25+
26+// Error implements error.
27+func (e *Error) Error() string { return e.msg }
28+
29+// Unwrap returns the original error, a *connect.Error for server answers.
30+func (e *Error) Unwrap() error { return e.err }
31+
32+// Describe turns an error of a call into a friendly one-line *Error. Errors
33+// that are not Connect errors are returned unchanged, nil stays nil.
34+func (c *Client) Describe(err error) error {
35+ var ce *connect.Error
36+ if err == nil || !errors.As(err, &ce) {
37+ return err
38+ }
39+ msg := oneLine(ce.Message())
40+ with := func(prefix string) string {
41+ if msg == "" {
42+ return prefix
43+ }
44+ return prefix + ": " + msg
45+ }
46+ var text string
47+ switch ce.Code() {
48+ case connect.CodeUnauthenticated:
49+ if c.cfg.Token == "" {
50+ text = "authentication required: set " + EnvToken + " to a valid token (create one with `ermite token create`)"
51+ } else {
52+ text = "not authenticated: " + EnvToken + " is invalid, expired or revoked (create a new one with `ermite token create`)"
53+ }
54+ case connect.CodePermissionDenied:
55+ text = with("permission denied") + " (the token lacks a required scope, see `ermite token list`)"
56+ case connect.CodeNotFound:
57+ text = with("not found")
58+ case connect.CodeAlreadyExists:
59+ text = with("already exists")
60+ case connect.CodeInvalidArgument:
61+ text = with("invalid request")
62+ case connect.CodeFailedPrecondition:
63+ text = with("precondition failed")
64+ case connect.CodeUnavailable:
65+ text = "cannot reach the server at " + c.cfg.URL
66+ if msg != "" {
67+ text += ": " + msg
68+ }
69+ case connect.CodeDeadlineExceeded:
70+ text = "the request to " + c.cfg.URL + " timed out"
71+ case connect.CodeCanceled:
72+ text = "canceled"
73+ case connect.CodeUnimplemented:
74+ text = "the server does not implement this call: check that " + EnvURL + " points to an ermite server, and its version with `ermite version`"
75+ case connect.CodeResourceExhausted:
76+ text = with("too many requests")
77+ case connect.CodeInternal:
78+ text = with("server error")
79+ default:
80+ text = with("unexpected server response")
81+ if redirected.MatchString(msg) {
82+ text += " (the server redirected the request: set " + EnvURL + " to the final address, probably https://)"
83+ }
84+ }
85+ return &Error{msg: text, err: err}
86+}
87+
88+// oneLine collapses whitespace and bounds the length of a server message.
89+func oneLine(s string) string {
90+ s = strings.Join(strings.Fields(s), " ")
91+ if len(s) > maxMessage {
92+ s = strings.ToValidUTF8(s[:maxMessage], "") + "…"
93+ }
94+ return s
95+}
@@ -0,0 +1,95 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import (
6+ "errors"
7+ "regexp"
8+ "strings"
9+
10+ "connectrpc.com/connect"
11+)
12+
13+// redirected matches the message Connect gives to a 3xx answer.
14+var redirected = regexp.MustCompile(`\b30[1-8]\b`)
15+
16+// maxMessage caps a server message echoed to the terminal.
17+const maxMessage = 300
18+
19+// Error is a failed call described for a person: one line, with a hint when
20+// the code points to a likely cause. It wraps the original error.
21+type Error struct {
22+ msg string
23+ err error
24+}
25+
26+// Error implements error.
27+func (e *Error) Error() string { return e.msg }
28+
29+// Unwrap returns the original error, a *connect.Error for server answers.
30+func (e *Error) Unwrap() error { return e.err }
31+
32+// Describe turns an error of a call into a friendly one-line *Error. Errors
33+// that are not Connect errors are returned unchanged, nil stays nil.
34+func (c *Client) Describe(err error) error {
35+ var ce *connect.Error
36+ if err == nil || !errors.As(err, &ce) {
37+ return err
38+ }
39+ msg := oneLine(ce.Message())
40+ with := func(prefix string) string {
41+ if msg == "" {
42+ return prefix
43+ }
44+ return prefix + ": " + msg
45+ }
46+ var text string
47+ switch ce.Code() {
48+ case connect.CodeUnauthenticated:
49+ if c.cfg.Token == "" {
50+ text = "authentication required: set " + EnvToken + " to a valid token (create one with `ermite token create`)"
51+ } else {
52+ text = "not authenticated: " + EnvToken + " is invalid, expired or revoked (create a new one with `ermite token create`)"
53+ }
54+ case connect.CodePermissionDenied:
55+ text = with("permission denied") + " (the token lacks a required scope, see `ermite token list`)"
56+ case connect.CodeNotFound:
57+ text = with("not found")
58+ case connect.CodeAlreadyExists:
59+ text = with("already exists")
60+ case connect.CodeInvalidArgument:
61+ text = with("invalid request")
62+ case connect.CodeFailedPrecondition:
63+ text = with("precondition failed")
64+ case connect.CodeUnavailable:
65+ text = "cannot reach the server at " + c.cfg.URL
66+ if msg != "" {
67+ text += ": " + msg
68+ }
69+ case connect.CodeDeadlineExceeded:
70+ text = "the request to " + c.cfg.URL + " timed out"
71+ case connect.CodeCanceled:
72+ text = "canceled"
73+ case connect.CodeUnimplemented:
74+ text = "the server does not implement this call: check that " + EnvURL + " points to an ermite server, and its version with `ermite version`"
75+ case connect.CodeResourceExhausted:
76+ text = with("too many requests")
77+ case connect.CodeInternal:
78+ text = with("server error")
79+ default:
80+ text = with("unexpected server response")
81+ if redirected.MatchString(msg) {
82+ text += " (the server redirected the request: set " + EnvURL + " to the final address, probably https://)"
83+ }
84+ }
85+ return &Error{msg: text, err: err}
86+}
87+
88+// oneLine collapses whitespace and bounds the length of a server message.
89+func oneLine(s string) string {
90+ s = strings.Join(strings.Fields(s), " ")
91+ if len(s) > maxMessage {
92+ s = strings.ToValidUTF8(s[:maxMessage], "") + "…"
93+ }
94+ return s
95+}
A internal/adapters/client/fake_test.go +62 -0
@@ -0,0 +1,62 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "context"
7+ "net/http"
8+ "net/http/httptest"
9+ "sync"
10+ "testing"
11+
12+ "connectrpc.com/connect"
13+
14+ "bnema.dev/ermite-api/gen/ermite"
15+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
16+ "bnema.dev/ermite/internal/adapters/client"
17+)
18+
19+// fake is a Connect server with stub services.
20+type fake struct {
21+ ermiteconnect.UnimplementedServerServiceHandler
22+ ermiteconnect.UnimplementedEventServiceHandler
23+
24+ mu sync.Mutex
25+ headers []http.Header
26+ version func(context.Context, *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error)
27+ watch func(context.Context, *connect.Request[ermite.WatchRequest], *connect.ServerStream[ermite.WatchResponse]) error
28+}
29+
30+func (f *fake) record(h http.Header) {
31+ f.mu.Lock()
32+ defer f.mu.Unlock()
33+ f.headers = append(f.headers, h.Clone())
34+}
35+
36+func (f *fake) GetVersion(ctx context.Context, req *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error) {
37+ f.record(req.Header())
38+ return f.version(ctx, req)
39+}
40+
41+func (f *fake) Watch(ctx context.Context, req *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
42+ f.record(req.Header())
43+ return f.watch(ctx, req, st)
44+}
45+
46+// start serves f under /-/api/ and returns a client for it.
47+func start(t *testing.T, f *fake, cfg client.Config) *client.Client {
48+ t.Helper()
49+ mux := http.NewServeMux()
50+ p, h := ermiteconnect.NewServerServiceHandler(f)
51+ mux.Handle("/-/api"+p, http.StripPrefix("/-/api", h))
52+ p, h = ermiteconnect.NewEventServiceHandler(f)
53+ mux.Handle("/-/api"+p, http.StripPrefix("/-/api", h))
54+ srv := httptest.NewServer(mux)
55+ t.Cleanup(srv.Close)
56+ cfg.URL = srv.URL
57+ c, err := client.New(cfg)
58+ if err != nil {
59+ t.Fatal(err)
60+ }
61+ return c
62+}
@@ -0,0 +1,62 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "context"
7+ "net/http"
8+ "net/http/httptest"
9+ "sync"
10+ "testing"
11+
12+ "connectrpc.com/connect"
13+
14+ "bnema.dev/ermite-api/gen/ermite"
15+ "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
16+ "bnema.dev/ermite/internal/adapters/client"
17+)
18+
19+// fake is a Connect server with stub services.
20+type fake struct {
21+ ermiteconnect.UnimplementedServerServiceHandler
22+ ermiteconnect.UnimplementedEventServiceHandler
23+
24+ mu sync.Mutex
25+ headers []http.Header
26+ version func(context.Context, *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error)
27+ watch func(context.Context, *connect.Request[ermite.WatchRequest], *connect.ServerStream[ermite.WatchResponse]) error
28+}
29+
30+func (f *fake) record(h http.Header) {
31+ f.mu.Lock()
32+ defer f.mu.Unlock()
33+ f.headers = append(f.headers, h.Clone())
34+}
35+
36+func (f *fake) GetVersion(ctx context.Context, req *connect.Request[ermite.GetVersionRequest]) (*connect.Response[ermite.GetVersionResponse], error) {
37+ f.record(req.Header())
38+ return f.version(ctx, req)
39+}
40+
41+func (f *fake) Watch(ctx context.Context, req *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
42+ f.record(req.Header())
43+ return f.watch(ctx, req, st)
44+}
45+
46+// start serves f under /-/api/ and returns a client for it.
47+func start(t *testing.T, f *fake, cfg client.Config) *client.Client {
48+ t.Helper()
49+ mux := http.NewServeMux()
50+ p, h := ermiteconnect.NewServerServiceHandler(f)
51+ mux.Handle("/-/api"+p, http.StripPrefix("/-/api", h))
52+ p, h = ermiteconnect.NewEventServiceHandler(f)
53+ mux.Handle("/-/api"+p, http.StripPrefix("/-/api", h))
54+ srv := httptest.NewServer(mux)
55+ t.Cleanup(srv.Close)
56+ cfg.URL = srv.URL
57+ c, err := client.New(cfg)
58+ if err != nil {
59+ t.Fatal(err)
60+ }
61+ return c
62+}
A internal/adapters/client/features.go +32 -0
@@ -0,0 +1,32 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import "slices"
6+
7+// Contract features this client relies on, as listed by
8+// ServerService.GetVersion. A server without one still works for everything
9+// else: the CLI warns instead of failing.
10+const (
11+ // FeatureRepoUpdate is RepoService.UpdateRepo (`ermite repo edit`).
12+ FeatureRepoUpdate = "repo.update"
13+ // FeatureEventsResume is EventService.Watch with resume_after, which
14+ // `ermite watch` uses to continue after a reconnection.
15+ FeatureEventsResume = "events.resume"
16+)
17+
18+// UsedFeatures lists the features the CLI uses, in the order they are reported.
19+func UsedFeatures() []string {
20+ return []string{FeatureRepoUpdate, FeatureEventsResume}
21+}
22+
23+// MissingFeatures returns the features the CLI uses that have is lacking.
24+func MissingFeatures(have []string) []string {
25+ var missing []string
26+ for _, f := range UsedFeatures() {
27+ if !slices.Contains(have, f) {
28+ missing = append(missing, f)
29+ }
30+ }
31+ return missing
32+}
@@ -0,0 +1,32 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import "slices"
6+
7+// Contract features this client relies on, as listed by
8+// ServerService.GetVersion. A server without one still works for everything
9+// else: the CLI warns instead of failing.
10+const (
11+ // FeatureRepoUpdate is RepoService.UpdateRepo (`ermite repo edit`).
12+ FeatureRepoUpdate = "repo.update"
13+ // FeatureEventsResume is EventService.Watch with resume_after, which
14+ // `ermite watch` uses to continue after a reconnection.
15+ FeatureEventsResume = "events.resume"
16+)
17+
18+// UsedFeatures lists the features the CLI uses, in the order they are reported.
19+func UsedFeatures() []string {
20+ return []string{FeatureRepoUpdate, FeatureEventsResume}
21+}
22+
23+// MissingFeatures returns the features the CLI uses that have is lacking.
24+func MissingFeatures(have []string) []string {
25+ var missing []string
26+ for _, f := range UsedFeatures() {
27+ if !slices.Contains(have, f) {
28+ missing = append(missing, f)
29+ }
30+ }
31+ return missing
32+}
A internal/adapters/client/redirect_test.go +20 -0
@@ -0,0 +1,20 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "net/http"
7+ "net/http/httptest"
8+ "testing"
9+)
10+
11+// redirectingServer answers every request with a redirect, like an http
12+// listener that forces https.
13+func redirectingServer(t *testing.T) string {
14+ t.Helper()
15+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
16+ http.Redirect(w, r, "https://example.invalid"+r.URL.Path, http.StatusMovedPermanently)
17+ }))
18+ t.Cleanup(srv.Close)
19+ return srv.URL
20+}
@@ -0,0 +1,20 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "net/http"
7+ "net/http/httptest"
8+ "testing"
9+)
10+
11+// redirectingServer answers every request with a redirect, like an http
12+// listener that forces https.
13+func redirectingServer(t *testing.T) string {
14+ t.Helper()
15+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
16+ http.Redirect(w, r, "https://example.invalid"+r.URL.Path, http.StatusMovedPermanently)
17+ }))
18+ t.Cleanup(srv.Close)
19+ return srv.URL
20+}
A internal/adapters/client/scopes.go +146 -0
@@ -0,0 +1,146 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import (
6+ "fmt"
7+ "slices"
8+ "strconv"
9+ "strings"
10+ "time"
11+
12+ "bnema.dev/ermite-api/gen/ermite"
13+)
14+
15+// Token presets (ADR 0006). The API knows scopes only; a preset is a client
16+// shortcut that expands to scopes.
17+const (
18+ presetCLI = "cli"
19+ presetAgent = "agent"
20+ presetCI = "ci"
21+)
22+
23+// PresetNames lists the token presets, see PresetScopes.
24+var PresetNames = []string{presetCLI, presetAgent, presetCI}
25+
26+// ScopeName is the name of a scope as users type it and as the docs write
27+// it: SCOPE_REPO_READ is "repo:read".
28+func ScopeName(s ermite.Scope) string {
29+ name, ok := ermite.Scope_name[int32(s)]
30+ if !ok {
31+ return strconv.Itoa(int(s))
32+ }
33+ return strings.Replace(strings.ToLower(strings.TrimPrefix(name, "SCOPE_")), "_", ":", 1)
34+}
35+
36+// AllScopes lists every scope the contract defines, in enum order. Deriving it
37+// keeps the `cli` preset ("every scope") true when the contract grows.
38+func AllScopes() []ermite.Scope {
39+ var scopes []ermite.Scope
40+ for n := range ermite.Scope_name {
41+ if s := ermite.Scope(n); s != ermite.Scope_SCOPE_UNSPECIFIED {
42+ scopes = append(scopes, s)
43+ }
44+ }
45+ slices.Sort(scopes)
46+ return scopes
47+}
48+
49+// ParseScope reads a scope as users type it, for example "repo:read".
50+func ParseScope(name string) (ermite.Scope, error) {
51+ want := strings.ToLower(strings.TrimSpace(name))
52+ for _, s := range AllScopes() {
53+ if ScopeName(s) == want {
54+ return s, nil
55+ }
56+ }
57+ return 0, fmt.Errorf("unknown scope %q: use one of %s", name, strings.Join(ScopeNames(AllScopes()), ", "))
58+}
59+
60+// ScopeNames maps ScopeName over scopes.
61+func ScopeNames(scopes []ermite.Scope) []string {
62+ names := make([]string, len(scopes))
63+ for i, s := range scopes {
64+ names[i] = ScopeName(s)
65+ }
66+ return names
67+}
68+
69+// PresetScopes expands a preset to its scopes.
70+func PresetScopes(name string) ([]ermite.Scope, error) {
71+ switch strings.ToLower(name) {
72+ case presetCLI:
73+ return AllScopes(), nil
74+ case presetAgent:
75+ return []ermite.Scope{ermite.Scope_SCOPE_REPO_WRITE, ermite.Scope_SCOPE_MR_WRITE}, nil
76+ case presetCI:
77+ return []ermite.Scope{ermite.Scope_SCOPE_REPO_READ}, nil
78+ default:
79+ return nil, fmt.Errorf("unknown preset %q: use one of %s", name, strings.Join(PresetNames, ", "))
80+ }
81+}
82+
83+// MergeScopes merges a preset and explicit scopes into a sorted, unique list.
84+// Neither given is an error: a token's power is never chosen by default.
85+func MergeScopes(preset string, scopeFlags []string) ([]ermite.Scope, error) {
86+ var scopes []ermite.Scope
87+ if preset != "" {
88+ s, err := PresetScopes(preset)
89+ if err != nil {
90+ return nil, err
91+ }
92+ scopes = append(scopes, s...)
93+ }
94+ for _, f := range scopeFlags {
95+ s, err := ParseScope(f)
96+ if err != nil {
97+ return nil, err
98+ }
99+ scopes = append(scopes, s)
100+ }
101+ if len(scopes) == 0 {
102+ return nil, fmt.Errorf("choose what the token may do: --preset %s, or one or more --scope", strings.Join(PresetNames, "|"))
103+ }
104+ slices.Sort(scopes)
105+ return slices.Compact(scopes), nil
106+}
107+
108+// ParseExpiry reads a lifetime such as "30d", "2w", "12h", "90m" or "1h30m"
109+// (days and weeks are 24 h and 7 d), or an absolute date, "2027-01-31" (end of
110+// that day, UTC) or an RFC 3339 instant. The result must lie in the future.
111+func ParseExpiry(s string, now time.Time) (time.Time, error) {
112+ s = strings.TrimSpace(s)
113+ bad := fmt.Errorf("invalid --expires %q: use a duration like 30d, 12h or 2w, or a date like 2027-01-31", s)
114+ var at time.Time
115+ switch {
116+ case s == "":
117+ return time.Time{}, bad
118+ case strings.HasSuffix(s, "d") || strings.HasSuffix(s, "w"):
119+ n, err := strconv.Atoi(s[:len(s)-1])
120+ if err != nil || n <= 0 {
121+ return time.Time{}, bad
122+ }
123+ unit := 24 * time.Hour
124+ if strings.HasSuffix(s, "w") {
125+ unit *= 7
126+ }
127+ if n > 100*365 {
128+ return time.Time{}, bad
129+ }
130+ at = now.Add(time.Duration(n) * unit)
131+ default:
132+ if d, err := time.ParseDuration(s); err == nil {
133+ at = now.Add(d)
134+ } else if t, err := time.Parse(time.RFC3339, s); err == nil {
135+ at = t
136+ } else if t, err := time.Parse(time.DateOnly, s); err == nil {
137+ at = t.Add(24*time.Hour - time.Second)
138+ } else {
139+ return time.Time{}, bad
140+ }
141+ }
142+ if !at.After(now) {
143+ return time.Time{}, fmt.Errorf("invalid --expires %q: it is in the past", s)
144+ }
145+ return at.UTC().Truncate(time.Second), nil
146+}
@@ -0,0 +1,146 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import (
6+ "fmt"
7+ "slices"
8+ "strconv"
9+ "strings"
10+ "time"
11+
12+ "bnema.dev/ermite-api/gen/ermite"
13+)
14+
15+// Token presets (ADR 0006). The API knows scopes only; a preset is a client
16+// shortcut that expands to scopes.
17+const (
18+ presetCLI = "cli"
19+ presetAgent = "agent"
20+ presetCI = "ci"
21+)
22+
23+// PresetNames lists the token presets, see PresetScopes.
24+var PresetNames = []string{presetCLI, presetAgent, presetCI}
25+
26+// ScopeName is the name of a scope as users type it and as the docs write
27+// it: SCOPE_REPO_READ is "repo:read".
28+func ScopeName(s ermite.Scope) string {
29+ name, ok := ermite.Scope_name[int32(s)]
30+ if !ok {
31+ return strconv.Itoa(int(s))
32+ }
33+ return strings.Replace(strings.ToLower(strings.TrimPrefix(name, "SCOPE_")), "_", ":", 1)
34+}
35+
36+// AllScopes lists every scope the contract defines, in enum order. Deriving it
37+// keeps the `cli` preset ("every scope") true when the contract grows.
38+func AllScopes() []ermite.Scope {
39+ var scopes []ermite.Scope
40+ for n := range ermite.Scope_name {
41+ if s := ermite.Scope(n); s != ermite.Scope_SCOPE_UNSPECIFIED {
42+ scopes = append(scopes, s)
43+ }
44+ }
45+ slices.Sort(scopes)
46+ return scopes
47+}
48+
49+// ParseScope reads a scope as users type it, for example "repo:read".
50+func ParseScope(name string) (ermite.Scope, error) {
51+ want := strings.ToLower(strings.TrimSpace(name))
52+ for _, s := range AllScopes() {
53+ if ScopeName(s) == want {
54+ return s, nil
55+ }
56+ }
57+ return 0, fmt.Errorf("unknown scope %q: use one of %s", name, strings.Join(ScopeNames(AllScopes()), ", "))
58+}
59+
60+// ScopeNames maps ScopeName over scopes.
61+func ScopeNames(scopes []ermite.Scope) []string {
62+ names := make([]string, len(scopes))
63+ for i, s := range scopes {
64+ names[i] = ScopeName(s)
65+ }
66+ return names
67+}
68+
69+// PresetScopes expands a preset to its scopes.
70+func PresetScopes(name string) ([]ermite.Scope, error) {
71+ switch strings.ToLower(name) {
72+ case presetCLI:
73+ return AllScopes(), nil
74+ case presetAgent:
75+ return []ermite.Scope{ermite.Scope_SCOPE_REPO_WRITE, ermite.Scope_SCOPE_MR_WRITE}, nil
76+ case presetCI:
77+ return []ermite.Scope{ermite.Scope_SCOPE_REPO_READ}, nil
78+ default:
79+ return nil, fmt.Errorf("unknown preset %q: use one of %s", name, strings.Join(PresetNames, ", "))
80+ }
81+}
82+
83+// MergeScopes merges a preset and explicit scopes into a sorted, unique list.
84+// Neither given is an error: a token's power is never chosen by default.
85+func MergeScopes(preset string, scopeFlags []string) ([]ermite.Scope, error) {
86+ var scopes []ermite.Scope
87+ if preset != "" {
88+ s, err := PresetScopes(preset)
89+ if err != nil {
90+ return nil, err
91+ }
92+ scopes = append(scopes, s...)
93+ }
94+ for _, f := range scopeFlags {
95+ s, err := ParseScope(f)
96+ if err != nil {
97+ return nil, err
98+ }
99+ scopes = append(scopes, s)
100+ }
101+ if len(scopes) == 0 {
102+ return nil, fmt.Errorf("choose what the token may do: --preset %s, or one or more --scope", strings.Join(PresetNames, "|"))
103+ }
104+ slices.Sort(scopes)
105+ return slices.Compact(scopes), nil
106+}
107+
108+// ParseExpiry reads a lifetime such as "30d", "2w", "12h", "90m" or "1h30m"
109+// (days and weeks are 24 h and 7 d), or an absolute date, "2027-01-31" (end of
110+// that day, UTC) or an RFC 3339 instant. The result must lie in the future.
111+func ParseExpiry(s string, now time.Time) (time.Time, error) {
112+ s = strings.TrimSpace(s)
113+ bad := fmt.Errorf("invalid --expires %q: use a duration like 30d, 12h or 2w, or a date like 2027-01-31", s)
114+ var at time.Time
115+ switch {
116+ case s == "":
117+ return time.Time{}, bad
118+ case strings.HasSuffix(s, "d") || strings.HasSuffix(s, "w"):
119+ n, err := strconv.Atoi(s[:len(s)-1])
120+ if err != nil || n <= 0 {
121+ return time.Time{}, bad
122+ }
123+ unit := 24 * time.Hour
124+ if strings.HasSuffix(s, "w") {
125+ unit *= 7
126+ }
127+ if n > 100*365 {
128+ return time.Time{}, bad
129+ }
130+ at = now.Add(time.Duration(n) * unit)
131+ default:
132+ if d, err := time.ParseDuration(s); err == nil {
133+ at = now.Add(d)
134+ } else if t, err := time.Parse(time.RFC3339, s); err == nil {
135+ at = t
136+ } else if t, err := time.Parse(time.DateOnly, s); err == nil {
137+ at = t.Add(24*time.Hour - time.Second)
138+ } else {
139+ return time.Time{}, bad
140+ }
141+ }
142+ if !at.After(now) {
143+ return time.Time{}, fmt.Errorf("invalid --expires %q: it is in the past", s)
144+ }
145+ return at.UTC().Truncate(time.Second), nil
146+}
A internal/adapters/client/scopes_test.go +94 -0
@@ -0,0 +1,94 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "strings"
7+ "testing"
8+ "time"
9+
10+ "github.com/stretchr/testify/assert"
11+ "github.com/stretchr/testify/require"
12+
13+ "bnema.dev/ermite-api/gen/ermite"
14+ "bnema.dev/ermite/internal/adapters/client"
15+)
16+
17+func TestScopeWireNames(t *testing.T) {
18+ want := map[ermite.Scope]string{
19+ ermite.Scope_SCOPE_REPO_READ: "repo:read",
20+ ermite.Scope_SCOPE_REPO_WRITE: "repo:write",
21+ ermite.Scope_SCOPE_REPO_ADMIN: "repo:admin",
22+ ermite.Scope_SCOPE_MR_WRITE: "mr:write",
23+ ermite.Scope_SCOPE_MR_MERGE: "mr:merge",
24+ ermite.Scope_SCOPE_CI_WRITE: "ci:write",
25+ ermite.Scope_SCOPE_SECRETS_WRITE: "secrets:write",
26+ ermite.Scope_SCOPE_RUNNER_REGISTER: "runner:register",
27+ ermite.Scope_SCOPE_TOKEN_ADMIN: "token:admin",
28+ }
29+ for s, name := range want {
30+ assert.Equal(t, name, client.ScopeName(s))
31+ got, err := client.ParseScope(strings.ToUpper(name))
32+ require.NoError(t, err)
33+ assert.Equal(t, s, got)
34+ }
35+ assert.Len(t, client.AllScopes(), len(want), "the cli preset covers every scope of the contract")
36+}
37+
38+func TestPresetExpansion(t *testing.T) {
39+ cli, err := client.PresetScopes("cli")
40+ require.NoError(t, err)
41+ assert.Equal(t, client.AllScopes(), cli)
42+ assert.NotContains(t, cli, ermite.Scope_SCOPE_UNSPECIFIED)
43+
44+ agent, err := client.PresetScopes("agent")
45+ require.NoError(t, err)
46+ assert.Equal(t, []ermite.Scope{ermite.Scope_SCOPE_REPO_WRITE, ermite.Scope_SCOPE_MR_WRITE}, agent)
47+
48+ ci, err := client.PresetScopes("CI")
49+ require.NoError(t, err)
50+ assert.Equal(t, []ermite.Scope{ermite.Scope_SCOPE_REPO_READ}, ci)
51+
52+ _, err = client.PresetScopes("root")
53+ require.ErrorContains(t, err, "unknown preset")
54+}
55+
56+func TestMergeScopes(t *testing.T) {
57+ got, err := client.MergeScopes("ci", []string{"repo:write", "repo:read"})
58+ require.NoError(t, err)
59+ assert.Equal(t, []ermite.Scope{ermite.Scope_SCOPE_REPO_READ, ermite.Scope_SCOPE_REPO_WRITE}, got, "preset and scopes add up, sorted, no duplicates")
60+
61+ got, err = client.MergeScopes("", []string{"token:admin"})
62+ require.NoError(t, err)
63+ assert.Equal(t, []ermite.Scope{ermite.Scope_SCOPE_TOKEN_ADMIN}, got)
64+
65+ _, err = client.MergeScopes("", nil)
66+ require.ErrorContains(t, err, "--preset")
67+ _, err = client.MergeScopes("", []string{"repo:everything"})
68+ require.ErrorContains(t, err, "unknown scope")
69+}
70+
71+func TestParseExpiry(t *testing.T) {
72+ now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
73+ tests := []struct {
74+ in string
75+ want time.Time
76+ }{
77+ {"30d", now.Add(30 * 24 * time.Hour)},
78+ {"2w", now.Add(14 * 24 * time.Hour)},
79+ {"12h", now.Add(12 * time.Hour)},
80+ {"1h30m", now.Add(90 * time.Minute)},
81+ {"2027-01-31", time.Date(2027, 1, 31, 23, 59, 59, 0, time.UTC)},
82+ {"2027-01-31T10:00:00+02:00", time.Date(2027, 1, 31, 8, 0, 0, 0, time.UTC)},
83+ }
84+ for _, tt := range tests {
85+ got, err := client.ParseExpiry(tt.in, now)
86+ require.NoError(t, err, tt.in)
87+ assert.True(t, tt.want.Equal(got), "%s: %s != %s", tt.in, tt.want, got)
88+ }
89+ for _, in := range []string{"", "soon", "0d", "-3d", "d", "1.5d", "2026-10-09", "-1h", "99999999d"} {
90+ _, err := client.ParseExpiry(in, now)
91+ require.Error(t, err, in)
92+ assert.Contains(t, err.Error(), "--expires")
93+ }
94+}
@@ -0,0 +1,94 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "strings"
7+ "testing"
8+ "time"
9+
10+ "github.com/stretchr/testify/assert"
11+ "github.com/stretchr/testify/require"
12+
13+ "bnema.dev/ermite-api/gen/ermite"
14+ "bnema.dev/ermite/internal/adapters/client"
15+)
16+
17+func TestScopeWireNames(t *testing.T) {
18+ want := map[ermite.Scope]string{
19+ ermite.Scope_SCOPE_REPO_READ: "repo:read",
20+ ermite.Scope_SCOPE_REPO_WRITE: "repo:write",
21+ ermite.Scope_SCOPE_REPO_ADMIN: "repo:admin",
22+ ermite.Scope_SCOPE_MR_WRITE: "mr:write",
23+ ermite.Scope_SCOPE_MR_MERGE: "mr:merge",
24+ ermite.Scope_SCOPE_CI_WRITE: "ci:write",
25+ ermite.Scope_SCOPE_SECRETS_WRITE: "secrets:write",
26+ ermite.Scope_SCOPE_RUNNER_REGISTER: "runner:register",
27+ ermite.Scope_SCOPE_TOKEN_ADMIN: "token:admin",
28+ }
29+ for s, name := range want {
30+ assert.Equal(t, name, client.ScopeName(s))
31+ got, err := client.ParseScope(strings.ToUpper(name))
32+ require.NoError(t, err)
33+ assert.Equal(t, s, got)
34+ }
35+ assert.Len(t, client.AllScopes(), len(want), "the cli preset covers every scope of the contract")
36+}
37+
38+func TestPresetExpansion(t *testing.T) {
39+ cli, err := client.PresetScopes("cli")
40+ require.NoError(t, err)
41+ assert.Equal(t, client.AllScopes(), cli)
42+ assert.NotContains(t, cli, ermite.Scope_SCOPE_UNSPECIFIED)
43+
44+ agent, err := client.PresetScopes("agent")
45+ require.NoError(t, err)
46+ assert.Equal(t, []ermite.Scope{ermite.Scope_SCOPE_REPO_WRITE, ermite.Scope_SCOPE_MR_WRITE}, agent)
47+
48+ ci, err := client.PresetScopes("CI")
49+ require.NoError(t, err)
50+ assert.Equal(t, []ermite.Scope{ermite.Scope_SCOPE_REPO_READ}, ci)
51+
52+ _, err = client.PresetScopes("root")
53+ require.ErrorContains(t, err, "unknown preset")
54+}
55+
56+func TestMergeScopes(t *testing.T) {
57+ got, err := client.MergeScopes("ci", []string{"repo:write", "repo:read"})
58+ require.NoError(t, err)
59+ assert.Equal(t, []ermite.Scope{ermite.Scope_SCOPE_REPO_READ, ermite.Scope_SCOPE_REPO_WRITE}, got, "preset and scopes add up, sorted, no duplicates")
60+
61+ got, err = client.MergeScopes("", []string{"token:admin"})
62+ require.NoError(t, err)
63+ assert.Equal(t, []ermite.Scope{ermite.Scope_SCOPE_TOKEN_ADMIN}, got)
64+
65+ _, err = client.MergeScopes("", nil)
66+ require.ErrorContains(t, err, "--preset")
67+ _, err = client.MergeScopes("", []string{"repo:everything"})
68+ require.ErrorContains(t, err, "unknown scope")
69+}
70+
71+func TestParseExpiry(t *testing.T) {
72+ now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
73+ tests := []struct {
74+ in string
75+ want time.Time
76+ }{
77+ {"30d", now.Add(30 * 24 * time.Hour)},
78+ {"2w", now.Add(14 * 24 * time.Hour)},
79+ {"12h", now.Add(12 * time.Hour)},
80+ {"1h30m", now.Add(90 * time.Minute)},
81+ {"2027-01-31", time.Date(2027, 1, 31, 23, 59, 59, 0, time.UTC)},
82+ {"2027-01-31T10:00:00+02:00", time.Date(2027, 1, 31, 8, 0, 0, 0, time.UTC)},
83+ }
84+ for _, tt := range tests {
85+ got, err := client.ParseExpiry(tt.in, now)
86+ require.NoError(t, err, tt.in)
87+ assert.True(t, tt.want.Equal(got), "%s: %s != %s", tt.in, tt.want, got)
88+ }
89+ for _, in := range []string{"", "soon", "0d", "-3d", "d", "1.5d", "2026-10-09", "-1h", "99999999d"} {
90+ _, err := client.ParseExpiry(in, now)
91+ require.Error(t, err, in)
92+ assert.Contains(t, err.Error(), "--expires")
93+ }
94+}
A internal/adapters/client/watch.go +159 -0
@@ -0,0 +1,159 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "math/rand/v2"
10+ "time"
11+
12+ "connectrpc.com/connect"
13+ "github.com/bnema/zerowrap"
14+
15+ "bnema.dev/ermite-api/gen/ermite"
16+)
17+
18+// Backoff is the delay between reconnections: it starts at Min, doubles after
19+// each failed attempt up to Max and resets once a connection has proved
20+// healthy. Delays are jittered by ±20 %.
21+type Backoff struct {
22+ Min time.Duration
23+ Max time.Duration
24+}
25+
26+// DefaultBackoff is 500 ms growing to 30 s.
27+var DefaultBackoff = Backoff{Min: 500 * time.Millisecond, Max: 30 * time.Second}
28+
29+// healthyAfter is how long a stream must stay up to count as healthy when it
30+// delivered no event, for example on a quiet server.
31+const healthyAfter = 10 * time.Second
32+
33+// WatchOptions configures Watch.
34+type WatchOptions struct {
35+ // Topics are the subjects to follow, see WatchRequest.topics.
36+ Topics []string
37+ // ResumeAfter is the id of the last event already seen, or "".
38+ ResumeAfter string
39+ // Backoff overrides DefaultBackoff when Min is set.
40+ Backoff Backoff
41+ // Sleep waits for d or until ctx ends; nil sleeps for real. For tests.
42+ Sleep func(ctx context.Context, d time.Duration) error
43+ // OnNotice receives connection notices (lost, retrying, back); optional.
44+ OnNotice func(msg string)
45+}
46+
47+// Watch follows EventService.Watch and calls onEvent for each event, in
48+// order, including Gap events (the caller decides how to tell the user). When
49+// the connection drops or the server closes the stream it reconnects with
50+// backoff, resuming after the last event id it saw. It returns nil when ctx
51+// ends, onEvent's error if it fails, and a described error when the server
52+// refuses the call for a reason a retry cannot fix (authentication, access,
53+// invalid topic, unsupported call).
54+func (c *Client) Watch(ctx context.Context, opts WatchOptions, onEvent func(*ermite.Event) error) error {
55+ log := zerowrap.FromCtx(ctx)
56+ bo := opts.Backoff
57+ if bo.Min <= 0 {
58+ bo = DefaultBackoff
59+ }
60+ bo.Max = max(bo.Max, bo.Min)
61+ sleep := opts.Sleep
62+ if sleep == nil {
63+ sleep = sleepCtx
64+ }
65+ notice := func(msg string) {
66+ if opts.OnNotice != nil {
67+ opts.OnNotice(msg)
68+ }
69+ }
70+
71+ events := c.Events()
72+ last := opts.ResumeAfter
73+ delay := bo.Min
74+ retrying := false
75+ for {
76+ started := time.Now()
77+ received := false
78+ stream, err := events.Watch(ctx, connect.NewRequest(&ermite.WatchRequest{Topics: opts.Topics, ResumeAfter: last}))
79+ if err == nil {
80+ for stream.Receive() {
81+ if !received && retrying {
82+ notice("reconnected")
83+ }
84+ received = true
85+ ev := stream.Msg().GetEvent()
86+ if ev == nil {
87+ continue
88+ }
89+ if err := onEvent(ev); err != nil {
90+ _ = stream.Close()
91+ return err
92+ }
93+ if ev.GetId() != "" {
94+ last = ev.GetId()
95+ }
96+ }
97+ err = stream.Err()
98+ _ = stream.Close()
99+ }
100+ if ctx.Err() != nil {
101+ return nil
102+ }
103+ if err != nil && !retryable(err) {
104+ return c.Describe(err)
105+ }
106+
107+ if received || time.Since(started) >= healthyAfter {
108+ delay = bo.Min
109+ }
110+ wait := jitter(delay)
111+ reason := "the server closed the stream"
112+ if err != nil {
113+ reason = c.Describe(err).Error()
114+ }
115+ msg := fmt.Sprintf("connection lost (%s); reconnecting in %s", reason, wait.Round(10*time.Millisecond))
116+ if last == "" {
117+ msg += " (no resume point yet: events during the outage will be missed)"
118+ }
119+ log.Debug().Str("reason", reason).Dur("wait", wait).Str("resume_after", last).Msg("watch reconnecting")
120+ notice(msg)
121+ if err := sleep(ctx, wait); err != nil {
122+ return nil
123+ }
124+ delay = min(delay*2, bo.Max)
125+ retrying = true
126+ }
127+}
128+
129+// retryable reports whether a reconnection can fix the error. The caller has
130+// checked that its own context is alive, so a cancellation here comes from the
131+// network (a proxy resetting the stream).
132+func retryable(err error) bool {
133+ var ce *connect.Error
134+ if !errors.As(err, &ce) {
135+ return true
136+ }
137+ switch ce.Code() {
138+ case connect.CodeUnavailable, connect.CodeUnknown, connect.CodeInternal, connect.CodeDeadlineExceeded,
139+ connect.CodeAborted, connect.CodeResourceExhausted, connect.CodeCanceled:
140+ return true
141+ default:
142+ return false
143+ }
144+}
145+
146+func jitter(d time.Duration) time.Duration {
147+ return time.Duration(float64(d) * (0.8 + 0.4*rand.Float64()))
148+}
149+
150+func sleepCtx(ctx context.Context, d time.Duration) error {
151+ t := time.NewTimer(d)
152+ defer t.Stop()
153+ select {
154+ case <-t.C:
155+ return nil
156+ case <-ctx.Done():
157+ return ctx.Err()
158+ }
159+}
@@ -0,0 +1,159 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "math/rand/v2"
10+ "time"
11+
12+ "connectrpc.com/connect"
13+ "github.com/bnema/zerowrap"
14+
15+ "bnema.dev/ermite-api/gen/ermite"
16+)
17+
18+// Backoff is the delay between reconnections: it starts at Min, doubles after
19+// each failed attempt up to Max and resets once a connection has proved
20+// healthy. Delays are jittered by ±20 %.
21+type Backoff struct {
22+ Min time.Duration
23+ Max time.Duration
24+}
25+
26+// DefaultBackoff is 500 ms growing to 30 s.
27+var DefaultBackoff = Backoff{Min: 500 * time.Millisecond, Max: 30 * time.Second}
28+
29+// healthyAfter is how long a stream must stay up to count as healthy when it
30+// delivered no event, for example on a quiet server.
31+const healthyAfter = 10 * time.Second
32+
33+// WatchOptions configures Watch.
34+type WatchOptions struct {
35+ // Topics are the subjects to follow, see WatchRequest.topics.
36+ Topics []string
37+ // ResumeAfter is the id of the last event already seen, or "".
38+ ResumeAfter string
39+ // Backoff overrides DefaultBackoff when Min is set.
40+ Backoff Backoff
41+ // Sleep waits for d or until ctx ends; nil sleeps for real. For tests.
42+ Sleep func(ctx context.Context, d time.Duration) error
43+ // OnNotice receives connection notices (lost, retrying, back); optional.
44+ OnNotice func(msg string)
45+}
46+
47+// Watch follows EventService.Watch and calls onEvent for each event, in
48+// order, including Gap events (the caller decides how to tell the user). When
49+// the connection drops or the server closes the stream it reconnects with
50+// backoff, resuming after the last event id it saw. It returns nil when ctx
51+// ends, onEvent's error if it fails, and a described error when the server
52+// refuses the call for a reason a retry cannot fix (authentication, access,
53+// invalid topic, unsupported call).
54+func (c *Client) Watch(ctx context.Context, opts WatchOptions, onEvent func(*ermite.Event) error) error {
55+ log := zerowrap.FromCtx(ctx)
56+ bo := opts.Backoff
57+ if bo.Min <= 0 {
58+ bo = DefaultBackoff
59+ }
60+ bo.Max = max(bo.Max, bo.Min)
61+ sleep := opts.Sleep
62+ if sleep == nil {
63+ sleep = sleepCtx
64+ }
65+ notice := func(msg string) {
66+ if opts.OnNotice != nil {
67+ opts.OnNotice(msg)
68+ }
69+ }
70+
71+ events := c.Events()
72+ last := opts.ResumeAfter
73+ delay := bo.Min
74+ retrying := false
75+ for {
76+ started := time.Now()
77+ received := false
78+ stream, err := events.Watch(ctx, connect.NewRequest(&ermite.WatchRequest{Topics: opts.Topics, ResumeAfter: last}))
79+ if err == nil {
80+ for stream.Receive() {
81+ if !received && retrying {
82+ notice("reconnected")
83+ }
84+ received = true
85+ ev := stream.Msg().GetEvent()
86+ if ev == nil {
87+ continue
88+ }
89+ if err := onEvent(ev); err != nil {
90+ _ = stream.Close()
91+ return err
92+ }
93+ if ev.GetId() != "" {
94+ last = ev.GetId()
95+ }
96+ }
97+ err = stream.Err()
98+ _ = stream.Close()
99+ }
100+ if ctx.Err() != nil {
101+ return nil
102+ }
103+ if err != nil && !retryable(err) {
104+ return c.Describe(err)
105+ }
106+
107+ if received || time.Since(started) >= healthyAfter {
108+ delay = bo.Min
109+ }
110+ wait := jitter(delay)
111+ reason := "the server closed the stream"
112+ if err != nil {
113+ reason = c.Describe(err).Error()
114+ }
115+ msg := fmt.Sprintf("connection lost (%s); reconnecting in %s", reason, wait.Round(10*time.Millisecond))
116+ if last == "" {
117+ msg += " (no resume point yet: events during the outage will be missed)"
118+ }
119+ log.Debug().Str("reason", reason).Dur("wait", wait).Str("resume_after", last).Msg("watch reconnecting")
120+ notice(msg)
121+ if err := sleep(ctx, wait); err != nil {
122+ return nil
123+ }
124+ delay = min(delay*2, bo.Max)
125+ retrying = true
126+ }
127+}
128+
129+// retryable reports whether a reconnection can fix the error. The caller has
130+// checked that its own context is alive, so a cancellation here comes from the
131+// network (a proxy resetting the stream).
132+func retryable(err error) bool {
133+ var ce *connect.Error
134+ if !errors.As(err, &ce) {
135+ return true
136+ }
137+ switch ce.Code() {
138+ case connect.CodeUnavailable, connect.CodeUnknown, connect.CodeInternal, connect.CodeDeadlineExceeded,
139+ connect.CodeAborted, connect.CodeResourceExhausted, connect.CodeCanceled:
140+ return true
141+ default:
142+ return false
143+ }
144+}
145+
146+func jitter(d time.Duration) time.Duration {
147+ return time.Duration(float64(d) * (0.8 + 0.4*rand.Float64()))
148+}
149+
150+func sleepCtx(ctx context.Context, d time.Duration) error {
151+ t := time.NewTimer(d)
152+ defer t.Stop()
153+ select {
154+ case <-t.C:
155+ return nil
156+ case <-ctx.Done():
157+ return ctx.Err()
158+ }
159+}
A internal/adapters/client/watch_test.go +207 -0
@@ -0,0 +1,207 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "context"
7+ "errors"
8+ "sync"
9+ "testing"
10+ "time"
11+
12+ "connectrpc.com/connect"
13+ "github.com/stretchr/testify/assert"
14+ "github.com/stretchr/testify/require"
15+
16+ "bnema.dev/ermite-api/gen/ermite"
17+ "bnema.dev/ermite/internal/adapters/client"
18+)
19+
20+func pushed(id string) *ermite.WatchResponse {
21+ return &ermite.WatchResponse{Event: &ermite.Event{Id: id, Kind: &ermite.Event_Pushed{Pushed: &ermite.Pushed{Repo: "r"}}}}
22+}
23+
24+// sleeps records the waits instead of sleeping.
25+type sleeps struct {
26+ mu sync.Mutex
27+ d []time.Duration
28+}
29+
30+func (s *sleeps) sleep(ctx context.Context, d time.Duration) error {
31+ s.mu.Lock()
32+ s.d = append(s.d, d)
33+ s.mu.Unlock()
34+ return ctx.Err()
35+}
36+
37+func TestWatchResumesAfterLastEventAndBacksOff(t *testing.T) {
38+ ctx, cancel := context.WithCancel(t.Context())
39+ defer cancel()
40+ var mu sync.Mutex
41+ var requests []*ermite.WatchRequest
42+ f := &fake{}
43+ f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
44+ mu.Lock()
45+ requests = append(requests, req.Msg)
46+ n := len(requests)
47+ mu.Unlock()
48+ switch n {
49+ case 1: // two events, then the connection ends
50+ _ = st.Send(pushed("e1"))
51+ _ = st.Send(pushed("e2"))
52+ return connect.NewError(connect.CodeUnavailable, errors.New("shutting down"))
53+ case 2: // fails at once, twice
54+ return connect.NewError(connect.CodeUnavailable, errors.New("still down"))
55+ case 3:
56+ return connect.NewError(connect.CodeUnavailable, errors.New("still down"))
57+ default: // back: one more event, then the user stops
58+ _ = st.Send(pushed("e3"))
59+ <-ctx.Done()
60+ return nil
61+ }
62+ }
63+ c := start(t, f, client.Config{})
64+
65+ var got []string
66+ var notices []string
67+ sl := &sleeps{}
68+ err := c.Watch(ctx, client.WatchOptions{
69+ Topics: []string{"home"},
70+ Backoff: client.Backoff{Min: time.Second, Max: 3 * time.Second},
71+ Sleep: sl.sleep,
72+ OnNotice: func(m string) { notices = append(notices, m) },
73+ }, func(ev *ermite.Event) error {
74+ got = append(got, ev.GetId())
75+ if ev.GetId() == "e3" {
76+ cancel()
77+ }
78+ return nil
79+ })
80+ require.NoError(t, err, "stopping with the context is not an error")
81+ assert.Equal(t, []string{"e1", "e2", "e3"}, got)
82+
83+ require.Len(t, requests, 4)
84+ assert.Equal(t, "", requests[0].GetResumeAfter())
85+ for _, r := range requests[1:3] {
86+ assert.Equal(t, "e2", r.GetResumeAfter())
87+ }
88+ assert.Equal(t, "e2", requests[3].GetResumeAfter())
89+ assert.Equal(t, []string{"home"}, requests[3].GetTopics())
90+
91+ // First wait: backoff was reset by the healthy first stream (Min), then it
92+ // doubles and is capped at Max. Delays are jittered by ±20 %.
93+ require.Len(t, sl.d, 3)
94+ assertNear(t, time.Second, sl.d[0])
95+ assertNear(t, 2*time.Second, sl.d[1])
96+ assertNear(t, 3*time.Second, sl.d[2])
97+
98+ require.NotEmpty(t, notices)
99+ assert.Contains(t, notices[0], "connection lost")
100+ assert.Contains(t, notices[0], "shutting down")
101+ assert.Contains(t, notices, "reconnected")
102+}
103+
104+func assertNear(t *testing.T, want, got time.Duration) {
105+ t.Helper()
106+ assert.InDelta(t, float64(want), float64(got), float64(want)*0.21)
107+}
108+
109+func TestWatchWarnsWhenThereIsNoResumePoint(t *testing.T) {
110+ ctx, cancel := context.WithCancel(t.Context())
111+ defer cancel()
112+ calls := 0
113+ f := &fake{}
114+ f.watch = func(context.Context, *connect.Request[ermite.WatchRequest], *connect.ServerStream[ermite.WatchResponse]) error {
115+ calls++
116+ if calls == 2 {
117+ cancel()
118+ }
119+ return connect.NewError(connect.CodeUnavailable, errors.New("down"))
120+ }
121+ c := start(t, f, client.Config{})
122+ var notices []string
123+ require.NoError(t, c.Watch(ctx, client.WatchOptions{
124+ Sleep: (&sleeps{}).sleep,
125+ OnNotice: func(m string) { notices = append(notices, m) },
126+ }, func(*ermite.Event) error { return nil }))
127+ require.NotEmpty(t, notices)
128+ assert.Contains(t, notices[0], "no resume point")
129+}
130+
131+func TestWatchStopsOnPermanentErrors(t *testing.T) {
132+ for _, code := range []connect.Code{connect.CodeUnauthenticated, connect.CodePermissionDenied, connect.CodeNotFound, connect.CodeInvalidArgument, connect.CodeUnimplemented} {
133+ t.Run(code.String(), func(t *testing.T) {
134+ calls := 0
135+ f := &fake{}
136+ f.watch = func(context.Context, *connect.Request[ermite.WatchRequest], *connect.ServerStream[ermite.WatchResponse]) error {
137+ calls++
138+ return connect.NewError(code, errors.New("nope"))
139+ }
140+ c := start(t, f, client.Config{})
141+ err := c.Watch(t.Context(), client.WatchOptions{Sleep: (&sleeps{}).sleep}, func(*ermite.Event) error { return nil })
142+ require.Error(t, err)
143+ assert.Equal(t, 1, calls, "no retry")
144+ var ce *connect.Error
145+ require.ErrorAs(t, err, &ce)
146+ assert.Equal(t, code, ce.Code())
147+ })
148+ }
149+}
150+
151+func TestWatchReturnsHandlerError(t *testing.T) {
152+ f := &fake{}
153+ f.watch = func(_ context.Context, _ *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
154+ _ = st.Send(pushed("e1"))
155+ <-time.After(time.Second)
156+ return nil
157+ }
158+ c := start(t, f, client.Config{})
159+ boom := errors.New("write failed")
160+ err := c.Watch(t.Context(), client.WatchOptions{}, func(*ermite.Event) error { return boom })
161+ assert.ErrorIs(t, err, boom)
162+}
163+
164+func TestWatchPassesGapAndIgnoresItAsResumePoint(t *testing.T) {
165+ ctx, cancel := context.WithCancel(t.Context())
166+ defer cancel()
167+ var resumes []string
168+ f := &fake{}
169+ f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
170+ resumes = append(resumes, req.Msg.GetResumeAfter())
171+ if len(resumes) == 1 {
172+ _ = st.Send(pushed("e1"))
173+ _ = st.Send(&ermite.WatchResponse{Event: &ermite.Event{Kind: &ermite.Event_Gap{Gap: &ermite.Gap{}}}})
174+ return connect.NewError(connect.CodeUnavailable, errors.New("x"))
175+ }
176+ _ = st.Send(pushed("e2"))
177+ <-ctx.Done()
178+ return nil
179+ }
180+ c := start(t, f, client.Config{})
181+ var kinds []string
182+ require.NoError(t, c.Watch(ctx, client.WatchOptions{Sleep: (&sleeps{}).sleep}, func(ev *ermite.Event) error {
183+ if ev.GetGap() != nil {
184+ kinds = append(kinds, "gap")
185+ } else {
186+ kinds = append(kinds, ev.GetId())
187+ }
188+ if ev.GetId() == "e2" {
189+ cancel()
190+ }
191+ return nil
192+ }))
193+ assert.Equal(t, []string{"e1", "gap", "e2"}, kinds)
194+ assert.Equal(t, []string{"", "e1"}, resumes, "a gap has no id and must not erase the resume point")
195+}
196+
197+func TestWatchResumesFromGivenID(t *testing.T) {
198+ var first string
199+ f := &fake{}
200+ f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], _ *connect.ServerStream[ermite.WatchResponse]) error {
201+ first = req.Msg.GetResumeAfter()
202+ return connect.NewError(connect.CodePermissionDenied, errors.New("stop"))
203+ }
204+ c := start(t, f, client.Config{})
205+ _ = c.Watch(t.Context(), client.WatchOptions{ResumeAfter: "abc"}, func(*ermite.Event) error { return nil })
206+ assert.Equal(t, "abc", first)
207+}
@@ -0,0 +1,207 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+package client_test
4+
5+import (
6+ "context"
7+ "errors"
8+ "sync"
9+ "testing"
10+ "time"
11+
12+ "connectrpc.com/connect"
13+ "github.com/stretchr/testify/assert"
14+ "github.com/stretchr/testify/require"
15+
16+ "bnema.dev/ermite-api/gen/ermite"
17+ "bnema.dev/ermite/internal/adapters/client"
18+)
19+
20+func pushed(id string) *ermite.WatchResponse {
21+ return &ermite.WatchResponse{Event: &ermite.Event{Id: id, Kind: &ermite.Event_Pushed{Pushed: &ermite.Pushed{Repo: "r"}}}}
22+}
23+
24+// sleeps records the waits instead of sleeping.
25+type sleeps struct {
26+ mu sync.Mutex
27+ d []time.Duration
28+}
29+
30+func (s *sleeps) sleep(ctx context.Context, d time.Duration) error {
31+ s.mu.Lock()
32+ s.d = append(s.d, d)
33+ s.mu.Unlock()
34+ return ctx.Err()
35+}
36+
37+func TestWatchResumesAfterLastEventAndBacksOff(t *testing.T) {
38+ ctx, cancel := context.WithCancel(t.Context())
39+ defer cancel()
40+ var mu sync.Mutex
41+ var requests []*ermite.WatchRequest
42+ f := &fake{}
43+ f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
44+ mu.Lock()
45+ requests = append(requests, req.Msg)
46+ n := len(requests)
47+ mu.Unlock()
48+ switch n {
49+ case 1: // two events, then the connection ends
50+ _ = st.Send(pushed("e1"))
51+ _ = st.Send(pushed("e2"))
52+ return connect.NewError(connect.CodeUnavailable, errors.New("shutting down"))
53+ case 2: // fails at once, twice
54+ return connect.NewError(connect.CodeUnavailable, errors.New("still down"))
55+ case 3:
56+ return connect.NewError(connect.CodeUnavailable, errors.New("still down"))
57+ default: // back: one more event, then the user stops
58+ _ = st.Send(pushed("e3"))
59+ <-ctx.Done()
60+ return nil
61+ }
62+ }
63+ c := start(t, f, client.Config{})
64+
65+ var got []string
66+ var notices []string
67+ sl := &sleeps{}
68+ err := c.Watch(ctx, client.WatchOptions{
69+ Topics: []string{"home"},
70+ Backoff: client.Backoff{Min: time.Second, Max: 3 * time.Second},
71+ Sleep: sl.sleep,
72+ OnNotice: func(m string) { notices = append(notices, m) },
73+ }, func(ev *ermite.Event) error {
74+ got = append(got, ev.GetId())
75+ if ev.GetId() == "e3" {
76+ cancel()
77+ }
78+ return nil
79+ })
80+ require.NoError(t, err, "stopping with the context is not an error")
81+ assert.Equal(t, []string{"e1", "e2", "e3"}, got)
82+
83+ require.Len(t, requests, 4)
84+ assert.Equal(t, "", requests[0].GetResumeAfter())
85+ for _, r := range requests[1:3] {
86+ assert.Equal(t, "e2", r.GetResumeAfter())
87+ }
88+ assert.Equal(t, "e2", requests[3].GetResumeAfter())
89+ assert.Equal(t, []string{"home"}, requests[3].GetTopics())
90+
91+ // First wait: backoff was reset by the healthy first stream (Min), then it
92+ // doubles and is capped at Max. Delays are jittered by ±20 %.
93+ require.Len(t, sl.d, 3)
94+ assertNear(t, time.Second, sl.d[0])
95+ assertNear(t, 2*time.Second, sl.d[1])
96+ assertNear(t, 3*time.Second, sl.d[2])
97+
98+ require.NotEmpty(t, notices)
99+ assert.Contains(t, notices[0], "connection lost")
100+ assert.Contains(t, notices[0], "shutting down")
101+ assert.Contains(t, notices, "reconnected")
102+}
103+
104+func assertNear(t *testing.T, want, got time.Duration) {
105+ t.Helper()
106+ assert.InDelta(t, float64(want), float64(got), float64(want)*0.21)
107+}
108+
109+func TestWatchWarnsWhenThereIsNoResumePoint(t *testing.T) {
110+ ctx, cancel := context.WithCancel(t.Context())
111+ defer cancel()
112+ calls := 0
113+ f := &fake{}
114+ f.watch = func(context.Context, *connect.Request[ermite.WatchRequest], *connect.ServerStream[ermite.WatchResponse]) error {
115+ calls++
116+ if calls == 2 {
117+ cancel()
118+ }
119+ return connect.NewError(connect.CodeUnavailable, errors.New("down"))
120+ }
121+ c := start(t, f, client.Config{})
122+ var notices []string
123+ require.NoError(t, c.Watch(ctx, client.WatchOptions{
124+ Sleep: (&sleeps{}).sleep,
125+ OnNotice: func(m string) { notices = append(notices, m) },
126+ }, func(*ermite.Event) error { return nil }))
127+ require.NotEmpty(t, notices)
128+ assert.Contains(t, notices[0], "no resume point")
129+}
130+
131+func TestWatchStopsOnPermanentErrors(t *testing.T) {
132+ for _, code := range []connect.Code{connect.CodeUnauthenticated, connect.CodePermissionDenied, connect.CodeNotFound, connect.CodeInvalidArgument, connect.CodeUnimplemented} {
133+ t.Run(code.String(), func(t *testing.T) {
134+ calls := 0
135+ f := &fake{}
136+ f.watch = func(context.Context, *connect.Request[ermite.WatchRequest], *connect.ServerStream[ermite.WatchResponse]) error {
137+ calls++
138+ return connect.NewError(code, errors.New("nope"))
139+ }
140+ c := start(t, f, client.Config{})
141+ err := c.Watch(t.Context(), client.WatchOptions{Sleep: (&sleeps{}).sleep}, func(*ermite.Event) error { return nil })
142+ require.Error(t, err)
143+ assert.Equal(t, 1, calls, "no retry")
144+ var ce *connect.Error
145+ require.ErrorAs(t, err, &ce)
146+ assert.Equal(t, code, ce.Code())
147+ })
148+ }
149+}
150+
151+func TestWatchReturnsHandlerError(t *testing.T) {
152+ f := &fake{}
153+ f.watch = func(_ context.Context, _ *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
154+ _ = st.Send(pushed("e1"))
155+ <-time.After(time.Second)
156+ return nil
157+ }
158+ c := start(t, f, client.Config{})
159+ boom := errors.New("write failed")
160+ err := c.Watch(t.Context(), client.WatchOptions{}, func(*ermite.Event) error { return boom })
161+ assert.ErrorIs(t, err, boom)
162+}
163+
164+func TestWatchPassesGapAndIgnoresItAsResumePoint(t *testing.T) {
165+ ctx, cancel := context.WithCancel(t.Context())
166+ defer cancel()
167+ var resumes []string
168+ f := &fake{}
169+ f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], st *connect.ServerStream[ermite.WatchResponse]) error {
170+ resumes = append(resumes, req.Msg.GetResumeAfter())
171+ if len(resumes) == 1 {
172+ _ = st.Send(pushed("e1"))
173+ _ = st.Send(&ermite.WatchResponse{Event: &ermite.Event{Kind: &ermite.Event_Gap{Gap: &ermite.Gap{}}}})
174+ return connect.NewError(connect.CodeUnavailable, errors.New("x"))
175+ }
176+ _ = st.Send(pushed("e2"))
177+ <-ctx.Done()
178+ return nil
179+ }
180+ c := start(t, f, client.Config{})
181+ var kinds []string
182+ require.NoError(t, c.Watch(ctx, client.WatchOptions{Sleep: (&sleeps{}).sleep}, func(ev *ermite.Event) error {
183+ if ev.GetGap() != nil {
184+ kinds = append(kinds, "gap")
185+ } else {
186+ kinds = append(kinds, ev.GetId())
187+ }
188+ if ev.GetId() == "e2" {
189+ cancel()
190+ }
191+ return nil
192+ }))
193+ assert.Equal(t, []string{"e1", "gap", "e2"}, kinds)
194+ assert.Equal(t, []string{"", "e1"}, resumes, "a gap has no id and must not erase the resume point")
195+}
196+
197+func TestWatchResumesFromGivenID(t *testing.T) {
198+ var first string
199+ f := &fake{}
200+ f.watch = func(_ context.Context, req *connect.Request[ermite.WatchRequest], _ *connect.ServerStream[ermite.WatchResponse]) error {
201+ first = req.Msg.GetResumeAfter()
202+ return connect.NewError(connect.CodePermissionDenied, errors.New("stop"))
203+ }
204+ c := start(t, f, client.Config{})
205+ _ = c.Watch(t.Context(), client.WatchOptions{ResumeAfter: "abc"}, func(*ermite.Event) error { return nil })
206+ assert.Equal(t, "abc", first)
207+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 2

Tags 45

↑↓ moveenter openesc close