Monorepo for Tangled
api/tangled,lexutil: add missing types for DRISL-CBOR event streaming
indigo's `lexutil.Client` doesn't support subscription xrpc methods. New
`extlexutil.Client` extends the `LexDo` method to support subscription.
It won't perform redialing since we don't know which param is cursor and
which property is event sequence.
since lexgen doesn't generate code subscription xrpc methods, we hand
write it. `api/tangled/*_ext.go` files will be treated as non-generated
code.
Signed-off-by: Seongmin Lee <git@boltless.me>
authored by
Seongmin Lee
and committed by
Tangled
(Jul 4, 2026, 8:12 AM +0300)
5d8a2489
8b9f5a70
+398
-2
+1
.gitattributes
+1
.gitattributes
+98
api/tangled/pipelinesubscribeLogs_ext.go
+98
api/tangled/pipelinesubscribeLogs_ext.go
···
1
+
// extending code generated from sh.tangled.ci.pipeline.subscribeLogs
2
+
3
+
package tangled
4
+
5
+
import (
6
+
"context"
7
+
"fmt"
8
+
"io"
9
+
10
+
"github.com/bluesky-social/indigo/events"
11
+
lexutil "github.com/bluesky-social/indigo/lex/util"
12
+
cbg "github.com/whyrusleeping/cbor-gen"
13
+
extlexutil "tangled.org/core/lexutil"
14
+
)
15
+
16
+
// TODO: generate codes below from lexicon
17
+
type CiPipelineSubscribeLogs_Event struct {
18
+
Error *events.ErrorFrame
19
+
Control *CiPipelineSubscribeLogs_Control
20
+
Data *CiPipelineSubscribeLogs_Data
21
+
22
+
// some private fields for internal routing perf
23
+
Preserialized []byte `json:"-" cborgen:"-"`
24
+
}
25
+
26
+
func (xevt *CiPipelineSubscribeLogs_Event) Serialize(wc io.Writer) error {
27
+
header := events.EventHeader{Op: events.EvtKindMessage}
28
+
var obj lexutil.CBOR
29
+
30
+
switch {
31
+
case xevt.Error != nil:
32
+
header.Op = events.EvtKindErrorFrame
33
+
obj = xevt.Error
34
+
case xevt.Control != nil:
35
+
header.MsgType = "#control"
36
+
obj = xevt.Control
37
+
case xevt.Data != nil:
38
+
header.MsgType = "#data"
39
+
obj = xevt.Data
40
+
default:
41
+
return fmt.Errorf("unrecognized event kind")
42
+
}
43
+
44
+
cborWriter := cbg.NewCborWriter(wc)
45
+
if err := header.MarshalCBOR(cborWriter); err != nil {
46
+
return fmt.Errorf("failed to write header: %w", err)
47
+
}
48
+
return obj.MarshalCBOR(cborWriter)
49
+
}
50
+
51
+
func (xevt *CiPipelineSubscribeLogs_Event) Deserialize(r io.Reader) error {
52
+
var header events.EventHeader
53
+
if err := header.UnmarshalCBOR(r); err != nil {
54
+
return fmt.Errorf("reading header: %w", err)
55
+
}
56
+
switch header.Op {
57
+
case events.EvtKindMessage:
58
+
switch header.MsgType {
59
+
case "#control":
60
+
var evt CiPipelineSubscribeLogs_Control
61
+
if err := evt.UnmarshalCBOR(r); err != nil {
62
+
return fmt.Errorf("reading repoCommit event: %w", err)
63
+
}
64
+
xevt.Control = &evt
65
+
case "#data":
66
+
var evt CiPipelineSubscribeLogs_Data
67
+
if err := evt.UnmarshalCBOR(r); err != nil {
68
+
return fmt.Errorf("reading repoSync event: %w", err)
69
+
}
70
+
xevt.Data = &evt
71
+
}
72
+
case events.EvtKindErrorFrame:
73
+
var errframe events.ErrorFrame
74
+
if err := errframe.UnmarshalCBOR(r); err != nil {
75
+
return err
76
+
}
77
+
xevt.Error = &errframe
78
+
default:
79
+
return fmt.Errorf("unrecognized event stream type: %d", header.Op)
80
+
}
81
+
return nil
82
+
}
83
+
84
+
func CiPipelineSubscribeLogs(ctx context.Context, c extlexutil.LexClient, pipeline string, workflows []string, sched extlexutil.Scheduler[CiPipelineSubscribeLogs_Event]) error {
85
+
defer sched.Shutdown()
86
+
87
+
params := map[string]any{}
88
+
params["pipeline"] = pipeline
89
+
params["workflows"] = workflows
90
+
91
+
return c.LexDo(ctx, extlexutil.Subscription, "", CiPipelineSubscribeLogsNSID, params, nil, func(ctx context.Context, cr *cbg.CborReader) error {
92
+
var evt CiPipelineSubscribeLogs_Event
93
+
if err := evt.Deserialize(cr); err != nil {
94
+
return err
95
+
}
96
+
return sched.AddWork(ctx, "", &evt)
97
+
})
98
+
}
+9
-2
flake.nix
+9
-2
flake.nix
···
515
515
rootDir=$(jj --ignore-working-copy root || git rev-parse --show-toplevel) || (echo "error: can't find repo root?"; exit 1)
516
516
cd "$rootDir"
517
517
518
-
rm -f api/tangled/*
518
+
# *_ext.go are hand-written extensions; never remove or mutate them
519
+
find api/tangled -maxdepth 1 -type f -not -name '*_ext.go' -delete
519
520
lexgen --build-file lexicon-build-config.json lexicons
520
-
sed -i.bak 's/\tutil/\/\/\tutil/' api/tangled/*
521
+
522
+
# disable type registration temporarily while running cborgen
523
+
find api/tangled -maxdepth 1 -name '*.go' -not -name '*_ext.go' -exec \
524
+
sed -i.bak 's/\tutil/\/\/\tutil/' {} +
521
525
# lexgen generates incomplete Marshaler/Unmarshaler for union types
522
526
find api/tangled/*.go -not -name "cbor_gen.go" -exec \
523
527
sed -i '/^func.*\(MarshalCBOR\|UnmarshalCBOR\)/,/^}/ s/^/\/\/ /' {} +
528
+
for f in api/tangled/*_ext.go; do [ -e "''$f" ] && mv "''$f" "''$f.bak"; done
524
529
${pkgs.gotools}/bin/goimports -w api/tangled/*
525
530
CGO_ENABLED=0 go run ./cmd/cborgen/
531
+
for f in api/tangled/*_ext.go.bak; do [ -e "''$f" ] && mv "''$f" "''${f%.bak}"; done
532
+
526
533
lexgen --build-file lexicon-build-config.json lexicons
527
534
rm api/tangled/*.bak
528
535
'';
+241
lexutil/client.go
+241
lexutil/client.go
···
1
+
package lexutil
2
+
3
+
import (
4
+
"cmp"
5
+
"context"
6
+
"fmt"
7
+
"log/slog"
8
+
"net/http"
9
+
"net/url"
10
+
"time"
11
+
12
+
indigoxrpc "github.com/bluesky-social/indigo/xrpc"
13
+
"github.com/carlmjohnson/versioninfo"
14
+
"github.com/gorilla/websocket"
15
+
cbg "github.com/whyrusleeping/cbor-gen"
16
+
)
17
+
18
+
const minHealthyConn = 30 * time.Second
19
+
20
+
type Client struct {
21
+
indigoxrpc.Client
22
+
Dialer websocket.Dialer
23
+
Logger *slog.Logger
24
+
}
25
+
26
+
var _ LexClient = (*Client)(nil)
27
+
28
+
func makeParams(p map[string]any) url.Values {
29
+
params := url.Values{}
30
+
for k, v := range p {
31
+
if s, ok := v.([]string); ok {
32
+
for _, v := range s {
33
+
params.Add(k, v)
34
+
}
35
+
} else {
36
+
params.Add(k, fmt.Sprint(v))
37
+
}
38
+
}
39
+
return params
40
+
}
41
+
42
+
type processFn func(ctx context.Context, cr *cbg.CborReader) error
43
+
44
+
func (c *Client) LexDo(ctx context.Context, method string, inputEncoding string, endpoint string, params map[string]any, bodyData any, out any) error {
45
+
switch method {
46
+
case Subscription:
47
+
if process, ok := out.(processFn); ok {
48
+
return c.LexSubscribe(ctx, endpoint, params, process)
49
+
} else if redialer, ok := out.(Redialer); ok {
50
+
return c.LexSubscribeWithRedialer(ctx, endpoint, params, redialer)
51
+
} else {
52
+
return fmt.Errorf("unknown output type: %T", out)
53
+
}
54
+
default:
55
+
return c.Client.LexDo(ctx, method, inputEncoding, endpoint, params, bodyData, out)
56
+
}
57
+
}
58
+
59
+
func (c *Client) getHeader() http.Header {
60
+
header := http.Header{}
61
+
if c.UserAgent != nil {
62
+
header.Set("User-Agent", *c.UserAgent)
63
+
} else {
64
+
header.Set("User-Agent", "extlexutil/"+versioninfo.Short())
65
+
}
66
+
if c.Headers != nil {
67
+
for k, v := range c.Headers {
68
+
header.Set(k, v)
69
+
}
70
+
}
71
+
return header
72
+
}
73
+
74
+
func (c *Client) LexSubscribe(ctx context.Context, endpoint string, params map[string]any, process func(ctx context.Context, cr *cbg.CborReader) error) error {
75
+
logger := cmp.Or(c.Logger, slog.Default().With("system", "events"))
76
+
rurl, err := url.Parse(c.Host)
77
+
if err != nil {
78
+
return err
79
+
}
80
+
if rurl.Scheme == "http" {
81
+
rurl.Scheme = "ws"
82
+
} else {
83
+
rurl.Scheme = "wss"
84
+
}
85
+
surl := rurl.JoinPath("/xrpc", endpoint)
86
+
surl.RawQuery = makeParams(params).Encode()
87
+
88
+
header := c.getHeader()
89
+
90
+
u := surl.String()
91
+
conn, resp, err := c.Dialer.DialContext(ctx, u, header)
92
+
if err != nil {
93
+
return fmt.Errorf("%w: %w", ErrDialFailure, err)
94
+
}
95
+
96
+
logger.Debug("event subscription response", "code", resp.StatusCode, "url", u)
97
+
98
+
return c.handleConn(ctx, conn, process)
99
+
}
100
+
101
+
func (c *Client) LexSubscribeWithRedialer(ctx context.Context, endpoint string, params map[string]any, redialer Redialer) error {
102
+
logger := cmp.Or(c.Logger, slog.Default().With("system", "events"))
103
+
rurl, err := url.Parse(c.Host)
104
+
if err != nil {
105
+
return err
106
+
}
107
+
if rurl.Scheme == "http" {
108
+
rurl.Scheme = "ws"
109
+
} else {
110
+
rurl.Scheme = "wss"
111
+
}
112
+
surl := rurl.JoinPath("/xrpc", endpoint)
113
+
114
+
header := c.getHeader()
115
+
116
+
var backoff int
117
+
// returns false if the retry budget is exhausted
118
+
sleepBackoff := func() bool {
119
+
select {
120
+
case <-ctx.Done():
121
+
case <-time.After(time.Duration(5+backoff) * time.Second):
122
+
}
123
+
backoff++
124
+
return backoff <= 15
125
+
}
126
+
127
+
for {
128
+
select {
129
+
case <-ctx.Done():
130
+
return ctx.Err()
131
+
default:
132
+
}
133
+
134
+
surl.RawQuery = makeParams(params).Encode()
135
+
136
+
u := surl.String()
137
+
conn, resp, err := c.Dialer.DialContext(ctx, u, header)
138
+
if err != nil {
139
+
logger.Warn("dialing failed", "err", err, "backoff", backoff)
140
+
if !sleepBackoff() {
141
+
return fmt.Errorf("%w: %w", ErrDialFailure, err)
142
+
}
143
+
continue
144
+
}
145
+
146
+
logger.Debug("event subscription response", "code", resp.StatusCode, "url", u)
147
+
148
+
connectedAt := time.Now()
149
+
connErr := c.handleConn(ctx, conn, redialer.Process)
150
+
if connErr != nil {
151
+
logger.Warn("host connection failed", "err", connErr, "backoff", backoff)
152
+
}
153
+
154
+
// updates cursor
155
+
updated := redialer.UpdateParams(ctx, params)
156
+
157
+
// a connection that drops immediately shouldnt reset backoff
158
+
// this to avoid reconnect storms
159
+
if updated || time.Since(connectedAt) >= minHealthyConn {
160
+
backoff = 0
161
+
continue
162
+
}
163
+
if !sleepBackoff() {
164
+
return fmt.Errorf("%w: %w", ErrConnFailure, connErr)
165
+
}
166
+
}
167
+
}
168
+
169
+
func (c *Client) handleConn(ctx context.Context, conn *websocket.Conn, process func(ctx context.Context, cr *cbg.CborReader) error) error {
170
+
logger := cmp.Or(c.Logger, slog.Default().With("system", "events"))
171
+
ctx, cancel := context.WithCancel(ctx)
172
+
defer cancel()
173
+
174
+
go func() {
175
+
t := time.NewTicker(time.Second * 30)
176
+
defer t.Stop()
177
+
failcount := 0
178
+
179
+
for {
180
+
181
+
select {
182
+
case <-t.C:
183
+
if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second*10)); err != nil {
184
+
logger.Warn("failed to ping", "err", err)
185
+
failcount++
186
+
if failcount >= 4 {
187
+
logger.Error("too many ping fails", "count", failcount)
188
+
conn.Close()
189
+
return
190
+
}
191
+
} else {
192
+
failcount = 0 // ok ping
193
+
}
194
+
case <-ctx.Done():
195
+
conn.Close()
196
+
return
197
+
}
198
+
}
199
+
}()
200
+
201
+
conn.SetPingHandler(func(message string) error {
202
+
err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60))
203
+
if err == websocket.ErrCloseSent {
204
+
return nil
205
+
}
206
+
return err
207
+
})
208
+
209
+
conn.SetPongHandler(func(_ string) error {
210
+
if err := conn.SetReadDeadline(time.Now().Add(time.Minute)); err != nil {
211
+
logger.Error("failed to set read deadline", "err", err)
212
+
}
213
+
214
+
return nil
215
+
})
216
+
217
+
cr := new(cbg.CborReader)
218
+
219
+
for {
220
+
select {
221
+
case <-ctx.Done():
222
+
return ctx.Err()
223
+
default:
224
+
}
225
+
226
+
mt, rawReader, err := conn.NextReader()
227
+
if err != nil {
228
+
return fmt.Errorf("conn err at read: %w", err)
229
+
}
230
+
231
+
if mt != websocket.BinaryMessage {
232
+
return fmt.Errorf("expected binary message from subscription endpoint")
233
+
}
234
+
235
+
cr.SetReader(rawReader)
236
+
237
+
if err := process(ctx, cr); err != nil {
238
+
return err
239
+
}
240
+
}
241
+
}
+49
lexutil/lexutil.go
+49
lexutil/lexutil.go
···
1
+
// extended version of indigo/lex/util package before upstreaming it
2
+
package lexutil
3
+
4
+
import (
5
+
"context"
6
+
"errors"
7
+
"io"
8
+
9
+
lexutil "github.com/bluesky-social/indigo/lex/util"
10
+
cbg "github.com/whyrusleeping/cbor-gen"
11
+
)
12
+
13
+
type LexClient interface {
14
+
lexutil.LexClient
15
+
// LexSubscribe is basic event subscriber without redialing logic
16
+
//
17
+
// golang doesnt allow generics in method so we have to pass raw processFn here instead of Scheduler[T]
18
+
LexSubscribe(ctx context.Context, endpoint string, params map[string]any, process func(ctx context.Context, cr *cbg.CborReader) error) error
19
+
}
20
+
21
+
const Subscription = "subscription"
22
+
23
+
var (
24
+
ErrDialFailure = errors.New("dialing failed")
25
+
ErrConnFailure = errors.New("connection failed")
26
+
)
27
+
28
+
type EventStreamMessage interface {
29
+
Serialize(wc io.Writer) error
30
+
Deserialize(r io.Reader) error
31
+
}
32
+
33
+
type Scheduler[T any] interface {
34
+
AddWork(ctx context.Context, namespace string, val *T) error
35
+
Shutdown()
36
+
}
37
+
38
+
type SeqScheduler[T any] interface {
39
+
Scheduler[T]
40
+
LastSeq() int64
41
+
}
42
+
43
+
type Redialer interface {
44
+
// Process decodes the raw message and schedule it
45
+
Process(ctx context.Context, cr *cbg.CborReader) error
46
+
47
+
// UpdateParams increments the cursor parameter based on LastSeq stored in internal scheduler
48
+
UpdateParams(ctx context.Context, params map[string]any) (updated bool)
49
+
}