forked from
tangled.org/core
Monorepo for Tangled
knotserver: switch to using official jetstream pkg
+161
-278
+13
-8
go.mod
+13
-8
go.mod
···
8
8
github.com/Blank-Xu/sql-adapter v1.1.1
9
9
github.com/bluekeyes/go-gitdiff v0.8.0
10
10
github.com/bluesky-social/indigo v0.0.0-20250123072624-9e3b84fdbb20
11
+
github.com/bluesky-social/jetstream v0.0.0-20241210005130-ea96859b93d1
11
12
github.com/casbin/casbin/v2 v2.103.0
12
13
github.com/gliderlabs/ssh v0.3.5
13
14
github.com/go-chi/chi/v5 v5.2.0
···
34
35
github.com/bmatcuk/doublestar/v4 v4.7.1 // indirect
35
36
github.com/carlmjohnson/versioninfo v0.22.5 // indirect
36
37
github.com/casbin/govaluate v1.3.0 // indirect
37
-
github.com/cespare/xxhash/v2 v2.2.0 // indirect
38
+
github.com/cespare/xxhash/v2 v2.3.0 // indirect
38
39
github.com/cloudflare/circl v1.4.0 // indirect
39
40
github.com/cyphar/filepath-securejoin v0.3.3 // indirect
40
41
github.com/davecgh/go-spew v1.1.1 // indirect
···
44
45
github.com/go-git/go-billy/v5 v5.5.0 // indirect
45
46
github.com/go-logr/logr v1.4.1 // indirect
46
47
github.com/go-logr/stdr v1.2.2 // indirect
48
+
github.com/goccy/go-json v0.10.2 // indirect
47
49
github.com/gogo/protobuf v1.3.2 // indirect
48
50
github.com/gorilla/css v1.0.1 // indirect
49
51
github.com/gorilla/securecookie v1.1.2 // indirect
···
66
68
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
67
69
github.com/jbenet/goprocess v0.1.4 // indirect
68
70
github.com/kevinburke/ssh_config v1.2.0 // indirect
71
+
github.com/klauspost/compress v1.17.9 // indirect
69
72
github.com/klauspost/cpuid/v2 v2.2.7 // indirect
70
73
github.com/mattn/go-isatty v0.0.20 // indirect
71
-
github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
72
74
github.com/minio/sha256-simd v1.0.1 // indirect
73
75
github.com/mr-tron/base58 v1.2.0 // indirect
74
76
github.com/multiformats/go-base32 v0.1.0 // indirect
···
80
82
github.com/pjbgf/sha1cd v0.3.0 // indirect
81
83
github.com/pmezard/go-difflib v1.0.0 // indirect
82
84
github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f // indirect
83
-
github.com/prometheus/client_golang v1.17.0 // indirect
84
-
github.com/prometheus/client_model v0.5.0 // indirect
85
-
github.com/prometheus/common v0.45.0 // indirect
86
-
github.com/prometheus/procfs v0.12.0 // indirect
85
+
github.com/prometheus/client_golang v1.19.1 // indirect
86
+
github.com/prometheus/client_model v0.6.1 // indirect
87
+
github.com/prometheus/common v0.54.0 // indirect
88
+
github.com/prometheus/procfs v0.15.1 // indirect
87
89
github.com/sergi/go-diff v1.3.1 // indirect
88
90
github.com/skeema/knownhosts v1.3.0 // indirect
89
91
github.com/spaolacci/murmur3 v1.1.0 // indirect
···
101
103
golang.org/x/crypto v0.32.0 // indirect
102
104
golang.org/x/net v0.33.0 // indirect
103
105
golang.org/x/sys v0.29.0 // indirect
104
-
golang.org/x/time v0.3.0 // indirect
105
-
google.golang.org/protobuf v1.33.0 // indirect
106
+
golang.org/x/time v0.5.0 // indirect
107
+
google.golang.org/protobuf v1.34.2 // indirect
106
108
gopkg.in/warnings.v0 v0.1.2 // indirect
107
109
gopkg.in/yaml.v3 v3.0.1 // indirect
108
110
lukechampine.com/blake3 v1.2.1 // indirect
···
111
113
replace github.com/sergi/go-diff => github.com/sergi/go-diff v1.1.0
112
114
113
115
replace github.com/go-git/go-git/v5 => github.com/go-git/go-git/v5 v5.6.1
116
+
117
+
// from bluesky-social/indigo
118
+
replace github.com/gocql/gocql => github.com/scylladb/gocql v1.14.4
+22
-18
go.sum
+22
-18
go.sum
···
22
22
github.com/bluekeyes/go-gitdiff v0.8.0/go.mod h1:WWAk1Mc6EgWarCrPFO+xeYlujPu98VuLW3Tu+B/85AE=
23
23
github.com/bluesky-social/indigo v0.0.0-20250123072624-9e3b84fdbb20 h1:yHusfYYi8odoCcsI6AurU+dRWb7itHAQNwt3/Rl9Vfs=
24
24
github.com/bluesky-social/indigo v0.0.0-20250123072624-9e3b84fdbb20/go.mod h1:Qp4YqWf+AQ3TwQCxV5Ls8O2tXE55zVTGVs3zTmn7BOg=
25
+
github.com/bluesky-social/jetstream v0.0.0-20241210005130-ea96859b93d1 h1:CFvRtYNSnWRAi/98M3O466t9dYuwtesNbu6FVPymRrA=
26
+
github.com/bluesky-social/jetstream v0.0.0-20241210005130-ea96859b93d1/go.mod h1:WiYEeyJSdUwqoaZ71KJSpTblemUCpwJfh5oVXplK6T4=
25
27
github.com/bmatcuk/doublestar/v4 v4.6.1/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
26
28
github.com/bmatcuk/doublestar/v4 v4.7.1 h1:fdDeAqgT47acgwd9bd9HxJRDmc9UAmPpc+2m0CXv75Q=
27
29
github.com/bmatcuk/doublestar/v4 v4.7.1/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
···
35
37
github.com/casbin/govaluate v1.2.0/go.mod h1:G/UnbIjZk/0uMNaLwZZmFQrR72tYRZWQkO70si/iR7A=
36
38
github.com/casbin/govaluate v1.3.0 h1:VA0eSY0M2lA86dYd5kPPuNZMUD9QkWnOCnavGrw9myc=
37
39
github.com/casbin/govaluate v1.3.0/go.mod h1:G/UnbIjZk/0uMNaLwZZmFQrR72tYRZWQkO70si/iR7A=
38
-
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
39
-
github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
40
+
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
41
+
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
40
42
github.com/cloudflare/circl v1.1.0/go.mod h1:prBCrKB9DV4poKZY1l9zBXg2QJY7mvgRvtMxxK7fi4I=
41
43
github.com/cloudflare/circl v1.3.3/go.mod h1:5XYMA4rFBvNIrhs50XuiBJ15vF2pZn4nnUKZrLbUZFA=
42
44
github.com/cloudflare/circl v1.4.0 h1:BV7h5MgrktNzytKmWjpOtdYrf0lkkbF8YMlBGPhJQrY=
···
73
75
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
74
76
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
75
77
github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0=
78
+
github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
79
+
github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
76
80
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
77
81
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
78
82
github.com/golang/mock v1.4.4 h1:l75CXGRSwbaYNpl/Z2X1XIIAMSCquvXgpVZDhwEIJsc=
···
147
151
github.com/kevinburke/ssh_config v1.2.0/go.mod h1:CT57kijsi8u/K/BOFA39wgDQJ9CxiF4nAY/ojJ6r6mM=
148
152
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
149
153
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
154
+
github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
155
+
github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
150
156
github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM=
151
157
github.com/klauspost/cpuid/v2 v2.2.7/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
152
158
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
···
164
170
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
165
171
github.com/mattn/go-sqlite3 v1.14.24 h1:tpSp2G2KyMnnQu99ngJ47EIkWVmliIizyZBfPrBWDRM=
166
172
github.com/mattn/go-sqlite3 v1.14.24/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
167
-
github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 h1:jWpvCLoY8Z/e3VKvlsiIGKtc+UG6U5vzxaoagmhXfyg=
168
-
github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0/go.mod h1:QUyp042oQthUoa9bqDv0ER0wrtXnBruoNd7aNjkbP+k=
169
173
github.com/microcosm-cc/bluemonday v1.0.27 h1:MpEUotklkwCSLeH+Qdx1VJgNqLlpY2KXwXFM08ygZfk=
170
174
github.com/microcosm-cc/bluemonday v1.0.27/go.mod h1:jFi9vgW+H7c3V0lb6nR74Ib/DIB5OBs92Dimizgw2cA=
171
175
github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM=
···
197
201
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
198
202
github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f h1:VXTQfuJj9vKR4TCkEuWIckKvdHFeJH/huIFJ9/cXOB0=
199
203
github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f/go.mod h1:/zvteZs/GwLtCgZ4BL6CBsk9IKIlexP43ObX9AxTqTw=
200
-
github.com/prometheus/client_golang v1.17.0 h1:rl2sfwZMtSthVU752MqfjQozy7blglC+1SOtjMAMh+Q=
201
-
github.com/prometheus/client_golang v1.17.0/go.mod h1:VeL+gMmOAxkS2IqfCq0ZmHSL+LjWfWDUmp1mBz9JgUY=
202
-
github.com/prometheus/client_model v0.5.0 h1:VQw1hfvPvk3Uv6Qf29VrPF32JB6rtbgI6cYPYQjL0Qw=
203
-
github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI=
204
-
github.com/prometheus/common v0.45.0 h1:2BGz0eBc2hdMDLnO/8n0jeB3oPrt2D08CekT0lneoxM=
205
-
github.com/prometheus/common v0.45.0/go.mod h1:YJmSTw9BoKxJplESWWxlbyttQR4uaEcGyv9MZjVOJsY=
206
-
github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
207
-
github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
204
+
github.com/prometheus/client_golang v1.19.1 h1:wZWJDwK+NameRJuPGDhlnFgx8e8HN3XHQeLaYJFJBOE=
205
+
github.com/prometheus/client_golang v1.19.1/go.mod h1:mP78NwGzrVks5S2H6ab8+ZZGJLZUq1hoULYBAYBw1Ho=
206
+
github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
207
+
github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
208
+
github.com/prometheus/common v0.54.0 h1:ZlZy0BgJhTwVZUn7dLOkwCZHUkrAqd3WYtcFCWnM1D8=
209
+
github.com/prometheus/common v0.54.0/go.mod h1:/TQgMJP5CuVYveyT7n/0Ix8yLNNXy9yRSkhnLTHPDIQ=
210
+
github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
211
+
github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
208
212
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
209
-
github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
210
-
github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
213
+
github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
214
+
github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
211
215
github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
212
216
github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
213
217
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
···
362
366
golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
363
367
golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
364
368
golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
365
-
golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
366
-
golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
369
+
golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
370
+
golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
367
371
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
368
372
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
369
373
golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
···
384
388
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
385
389
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 h1:+cNy6SZtPcJQH3LJVLOSmiC7MMxXNOb3PU/VUEz+EhU=
386
390
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
387
-
google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
388
-
google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
391
+
google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
392
+
google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
389
393
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
390
394
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
391
395
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+2
-3
knotserver/handler.go
+2
-3
knotserver/handler.go
···
9
9
"github.com/go-chi/chi/v5"
10
10
"github.com/sotangled/tangled/knotserver/config"
11
11
"github.com/sotangled/tangled/knotserver/db"
12
-
"github.com/sotangled/tangled/knotserver/jsclient"
13
12
"github.com/sotangled/tangled/rbac"
14
13
)
15
14
···
20
19
type Handle struct {
21
20
c *config.Config
22
21
db *db.DB
23
-
js *jsclient.JetstreamClient
22
+
jc *JetstreamClient
24
23
e *rbac.Enforcer
25
24
l *slog.Logger
26
25
···
61
60
if len(dids) > 0 {
62
61
h.knotInitialized = true
63
62
close(h.init)
64
-
h.js.UpdateDids(dids)
63
+
h.jc.UpdateDids(dids)
65
64
}
66
65
67
66
r.Get("/", h.Index)
+121
-76
knotserver/jetstream.go
+121
-76
knotserver/jetstream.go
···
8
8
"net/http"
9
9
"net/url"
10
10
"strings"
11
+
"sync"
11
12
"time"
12
13
14
+
"github.com/bluesky-social/jetstream/pkg/client"
15
+
"github.com/bluesky-social/jetstream/pkg/client/schedulers/sequential"
16
+
"github.com/bluesky-social/jetstream/pkg/models"
13
17
"github.com/sotangled/tangled/api/tangled"
14
18
"github.com/sotangled/tangled/knotserver/db"
15
-
"github.com/sotangled/tangled/knotserver/jsclient"
16
19
"github.com/sotangled/tangled/log"
17
20
)
18
21
22
+
type JetstreamClient struct {
23
+
cfg *client.ClientConfig
24
+
client *client.Client
25
+
reconnectCh chan struct{}
26
+
mu sync.RWMutex
27
+
}
28
+
19
29
func (h *Handle) StartJetstream(ctx context.Context) error {
20
30
l := h.l.With("component", "jetstream")
21
31
ctx = log.IntoContext(ctx, l)
···
27
37
return err
28
38
}
29
39
30
-
h.js = jsclient.NewJetstreamClient(collections, dids)
31
-
messages, err := h.js.ReadJetstream(ctx, lastTimeUs)
40
+
cfg := client.DefaultClientConfig()
41
+
cfg.WebsocketURL = "wss://jetstream1.us-west.bsky.network/subscribe"
42
+
cfg.WantedCollections = collections
43
+
cfg.WantedDids = dids
44
+
45
+
sched := sequential.NewScheduler("knotserver", l, h.processMessages)
46
+
47
+
client, err := client.NewClient(cfg, l, sched)
32
48
if err != nil {
33
-
return fmt.Errorf("failed to read from jetstream: %w", err)
49
+
l.Error("failed to create jetstream client", "error", err)
34
50
}
35
51
36
-
go h.processMessages(ctx, messages)
52
+
jc := &JetstreamClient{
53
+
cfg: cfg,
54
+
client: client,
55
+
reconnectCh: make(chan struct{}),
56
+
}
37
57
58
+
h.jc = jc
59
+
60
+
go func() {
61
+
for len(h.jc.cfg.WantedDids) == 0 {
62
+
time.Sleep(time.Second)
63
+
}
64
+
h.connectAndRead(ctx, &lastTimeUs)
65
+
}()
38
66
return nil
39
67
}
40
68
69
+
func (h *Handle) connectAndRead(ctx context.Context, cursor *int64) {
70
+
l := log.FromContext(ctx)
71
+
for {
72
+
select {
73
+
case <-h.jc.reconnectCh:
74
+
l.Info("reconnecting jetstream client")
75
+
h.jc.client.Scheduler.Shutdown()
76
+
if err := h.jc.client.ConnectAndRead(ctx, cursor); err != nil {
77
+
l.Error("error reading jetstream", "error", err)
78
+
}
79
+
default:
80
+
if err := h.jc.client.ConnectAndRead(ctx, cursor); err != nil {
81
+
l.Error("error reading jetstream", "error", err)
82
+
}
83
+
}
84
+
}
85
+
}
86
+
87
+
func (j *JetstreamClient) UpdateDids(dids []string) {
88
+
j.mu.Lock()
89
+
j.cfg.WantedDids = dids
90
+
j.mu.Unlock()
91
+
j.reconnectCh <- struct{}{}
92
+
}
93
+
41
94
func (h *Handle) getLastTimeUs(ctx context.Context) (int64, error) {
42
95
l := log.FromContext(ctx)
43
96
lastTimeUs, err := h.db.GetLastTimeUs()
···
60
113
return lastTimeUs, nil
61
114
}
62
115
63
-
func (h *Handle) processPublicKey(ctx context.Context, did string, record map[string]interface{}) error {
116
+
func (h *Handle) processPublicKey(ctx context.Context, did string, record tangled.PublicKey) error {
64
117
l := log.FromContext(ctx)
65
-
if err := h.db.AddPublicKeyFromRecord(did, record); err != nil {
118
+
pk := db.PublicKey{
119
+
Did: did,
120
+
PublicKey: record,
121
+
}
122
+
if err := h.db.AddPublicKey(pk); err != nil {
66
123
l.Error("failed to add public key", "error", err)
67
124
return fmt.Errorf("failed to add public key: %w", err)
68
125
}
···
70
127
return nil
71
128
}
72
129
130
+
func (h *Handle) processKnotMember(ctx context.Context, did string, record tangled.KnotMember) error {
131
+
l := log.FromContext(ctx)
132
+
133
+
if record.Domain != h.c.Server.Hostname {
134
+
l.Error("domain mismatch", "domain", record.Domain, "expected", h.c.Server.Hostname)
135
+
return fmt.Errorf("domain mismatch: %s != %s", record.Domain, h.c.Server.Hostname)
136
+
}
137
+
138
+
ok, err := h.e.E.Enforce(did, ThisServer, ThisServer, "server:invite")
139
+
if err != nil || !ok {
140
+
l.Error("failed to add member", "did", did)
141
+
return fmt.Errorf("failed to enforce permissions: %w", err)
142
+
}
143
+
144
+
l.Info("adding member")
145
+
if err := h.e.AddMember(ThisServer, record.Member); err != nil {
146
+
l.Error("failed to add member", "error", err)
147
+
return fmt.Errorf("failed to add member: %w", err)
148
+
}
149
+
l.Info("added member from firehose", "member", record.Member)
150
+
151
+
if err := h.db.AddDid(did); err != nil {
152
+
l.Error("failed to add did", "error", err)
153
+
return fmt.Errorf("failed to add did: %w", err)
154
+
}
155
+
156
+
if err := h.fetchAndAddKeys(ctx, did); err != nil {
157
+
return fmt.Errorf("failed to fetch and add keys: %w", err)
158
+
}
159
+
160
+
h.jc.UpdateDids([]string{did})
161
+
return nil
162
+
}
163
+
73
164
func (h *Handle) fetchAndAddKeys(ctx context.Context, did string) error {
74
165
l := log.FromContext(ctx)
75
166
···
87
178
defer resp.Body.Close()
88
179
89
180
if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/plain") {
90
-
l.Error("unexpected content type", "content-type", ct)
91
181
return fmt.Errorf("unexpected content type: %s", ct)
92
182
}
93
183
···
113
203
return nil
114
204
}
115
205
116
-
func (h *Handle) processKnotMember(ctx context.Context, did string, record map[string]interface{}) error {
117
-
l := log.FromContext(ctx)
206
+
func (h *Handle) processMessages(ctx context.Context, event *models.Event) error {
207
+
did := event.Did
118
208
119
-
if record["domain"] != h.c.Server.Hostname {
120
-
l.Error("domain mismatch", "domain", record["domain"], "expected", h.c.Server.Hostname)
121
-
return fmt.Errorf("domain mismatch: %s != %s", record["domain"], h.c.Server.Hostname)
122
-
}
209
+
raw := json.RawMessage(event.Commit.Record)
123
210
124
-
ok, err := h.e.E.Enforce(did, ThisServer, ThisServer, "server:invite")
125
-
if err != nil || !ok {
126
-
l.Error("failed to add member", "did", did)
127
-
return fmt.Errorf("failed to enforce permissions: %w", err)
128
-
}
129
-
130
-
l.Info("adding member")
131
-
if err := h.e.AddMember(ThisServer, record["member"].(string)); err != nil {
132
-
l.Error("failed to add member", "error", err)
133
-
return fmt.Errorf("failed to add member: %w", err)
134
-
}
135
-
l.Info("added member from firehose", "member", record["member"])
211
+
switch event.Commit.Collection {
212
+
case tangled.PublicKeyNSID:
213
+
var record tangled.PublicKey
214
+
if err := json.Unmarshal(raw, &record); err != nil {
215
+
return fmt.Errorf("failed to unmarshal record: %w", err)
216
+
}
217
+
if err := h.processPublicKey(ctx, did, record); err != nil {
218
+
return fmt.Errorf("failed to process public key: %w", err)
219
+
}
136
220
137
-
if err := h.db.AddDid(did); err != nil {
138
-
l.Error("failed to add did", "error", err)
139
-
return fmt.Errorf("failed to add did: %w", err)
221
+
case tangled.KnotMemberNSID:
222
+
var record tangled.KnotMember
223
+
if err := json.Unmarshal(raw, &record); err != nil {
224
+
return fmt.Errorf("failed to unmarshal record: %w", err)
225
+
}
226
+
if err := h.processKnotMember(ctx, did, record); err != nil {
227
+
return fmt.Errorf("failed to process knot member: %w", err)
228
+
}
140
229
}
141
230
142
-
if err := h.fetchAndAddKeys(ctx, did); err != nil {
143
-
return fmt.Errorf("failed to fetch and add keys: %w", err)
231
+
lastTimeUs := event.TimeUS
232
+
if err := h.db.SaveLastTimeUs(lastTimeUs); err != nil {
233
+
return fmt.Errorf("failed to save last time us: %w", err)
144
234
}
145
235
146
-
h.js.UpdateDids([]string{did})
147
236
return nil
148
237
}
149
-
150
-
func (h *Handle) processMessages(ctx context.Context, messages <-chan []byte) {
151
-
l := log.FromContext(ctx)
152
-
l.Info("waiting for knot to be initialized")
153
-
<-h.init
154
-
l.Info("initialized jetstream watcher")
155
-
156
-
for msg := range messages {
157
-
var data map[string]interface{}
158
-
if err := json.Unmarshal(msg, &data); err != nil {
159
-
l.Error("error unmarshaling message", "error", err)
160
-
continue
161
-
}
162
-
163
-
if kind, ok := data["kind"].(string); ok && kind == "commit" {
164
-
commit := data["commit"].(map[string]interface{})
165
-
did := data["did"].(string)
166
-
record := commit["record"].(map[string]interface{})
167
-
168
-
var processErr error
169
-
switch commit["collection"].(string) {
170
-
case tangled.PublicKeyNSID:
171
-
if err := h.processPublicKey(ctx, did, record); err != nil {
172
-
processErr = fmt.Errorf("failed to process public key: %w", err)
173
-
}
174
-
case tangled.KnotMemberNSID:
175
-
if err := h.processKnotMember(ctx, did, record); err != nil {
176
-
processErr = fmt.Errorf("failed to process knot member: %w", err)
177
-
}
178
-
}
179
-
180
-
if processErr != nil {
181
-
l.Error("error processing message", "error", processErr)
182
-
continue
183
-
}
184
-
185
-
lastTimeUs := int64(data["time_us"].(float64))
186
-
if err := h.db.SaveLastTimeUs(lastTimeUs); err != nil {
187
-
l.Error("failed to save last time us", "error", err)
188
-
continue
189
-
}
190
-
}
191
-
}
192
-
}
-170
knotserver/jsclient/jetstream.go
-170
knotserver/jsclient/jetstream.go
···
1
-
package jsclient
2
-
3
-
import (
4
-
"context"
5
-
"fmt"
6
-
"log"
7
-
"net/url"
8
-
"sync"
9
-
"time"
10
-
11
-
"github.com/gorilla/websocket"
12
-
)
13
-
14
-
type JetstreamClient struct {
15
-
collections []string
16
-
dids []string
17
-
conn *websocket.Conn
18
-
mu sync.RWMutex
19
-
reconnectCh chan struct{}
20
-
}
21
-
22
-
func NewJetstreamClient(collections, dids []string) *JetstreamClient {
23
-
return &JetstreamClient{
24
-
collections: collections,
25
-
dids: dids,
26
-
reconnectCh: make(chan struct{}, 1),
27
-
}
28
-
}
29
-
30
-
func (j *JetstreamClient) buildWebsocketURL(queryParams string) url.URL {
31
-
32
-
u := url.URL{
33
-
Scheme: "wss",
34
-
Host: "jetstream1.us-west.bsky.network",
35
-
Path: "/subscribe",
36
-
RawQuery: queryParams,
37
-
}
38
-
39
-
return u
40
-
}
41
-
42
-
// UpdateCollections updates the collections list and triggers a reconnection
43
-
func (j *JetstreamClient) UpdateCollections(collections []string) {
44
-
j.mu.Lock()
45
-
j.collections = collections
46
-
j.mu.Unlock()
47
-
j.triggerReconnect()
48
-
}
49
-
50
-
// UpdateDids updates the Dids list and triggers a reconnection
51
-
func (j *JetstreamClient) UpdateDids(dids []string) {
52
-
j.mu.Lock()
53
-
j.dids = dids
54
-
j.mu.Unlock()
55
-
j.triggerReconnect()
56
-
}
57
-
58
-
// Adds one did to the did list
59
-
func (j *JetstreamClient) AddDid(did string) {
60
-
j.mu.Lock()
61
-
j.dids = append(j.dids, did)
62
-
j.mu.Unlock()
63
-
j.triggerReconnect()
64
-
}
65
-
66
-
func (j *JetstreamClient) triggerReconnect() {
67
-
select {
68
-
case j.reconnectCh <- struct{}{}:
69
-
default:
70
-
// Channel already has a pending reconnect
71
-
}
72
-
}
73
-
74
-
func (j *JetstreamClient) buildQueryParams(cursor int64) string {
75
-
j.mu.RLock()
76
-
defer j.mu.RUnlock()
77
-
78
-
var collections, dids string
79
-
if len(j.collections) > 0 {
80
-
collections = fmt.Sprintf("wantedCollections=%s&cursor=%d", j.collections[0], cursor)
81
-
for _, collection := range j.collections[1:] {
82
-
collections += fmt.Sprintf("&wantedCollections=%s", collection)
83
-
}
84
-
}
85
-
if len(j.dids) > 0 {
86
-
for i, did := range j.dids {
87
-
if i == 0 {
88
-
dids = fmt.Sprintf("wantedDids=%s", did)
89
-
} else {
90
-
dids += fmt.Sprintf("&wantedDids=%s", did)
91
-
}
92
-
}
93
-
}
94
-
95
-
var queryStr string
96
-
if collections != "" && dids != "" {
97
-
queryStr = collections + "&" + dids
98
-
} else if collections != "" {
99
-
queryStr = collections
100
-
} else if dids != "" {
101
-
queryStr = dids
102
-
}
103
-
104
-
return queryStr
105
-
}
106
-
107
-
func (j *JetstreamClient) connect(cursor int64) error {
108
-
queryParams := j.buildQueryParams(cursor)
109
-
u := j.buildWebsocketURL(queryParams)
110
-
111
-
dialer := websocket.Dialer{
112
-
HandshakeTimeout: 10 * time.Second,
113
-
}
114
-
115
-
conn, _, err := dialer.Dial(u.String(), nil)
116
-
if err != nil {
117
-
return err
118
-
}
119
-
120
-
if j.conn != nil {
121
-
j.conn.Close()
122
-
}
123
-
j.conn = conn
124
-
return nil
125
-
}
126
-
127
-
func (j *JetstreamClient) readMessages(ctx context.Context, messages chan []byte) {
128
-
defer close(messages)
129
-
defer j.conn.Close()
130
-
131
-
ticker := time.NewTicker(1 * time.Second)
132
-
defer ticker.Stop()
133
-
134
-
for {
135
-
select {
136
-
case <-ctx.Done():
137
-
return
138
-
case <-j.reconnectCh:
139
-
// Reconnect with new parameters
140
-
cursor := time.Now().Add(-5 * time.Second).UnixMicro()
141
-
if err := j.connect(cursor); err != nil {
142
-
log.Printf("error reconnecting to jetstream: %v", err)
143
-
return
144
-
}
145
-
case <-ticker.C:
146
-
_, message, err := j.conn.ReadMessage()
147
-
if err != nil {
148
-
log.Printf("error reading from websocket: %v", err)
149
-
return
150
-
}
151
-
messages <- message
152
-
}
153
-
}
154
-
}
155
-
156
-
func (j *JetstreamClient) ReadJetstream(ctx context.Context, lastTimestamp int64) (chan []byte, error) {
157
-
if lastTimestamp == 0 {
158
-
lastTimestamp = time.Now().Add(-5 * time.Second).UnixMicro()
159
-
}
160
-
161
-
if err := j.connect(lastTimestamp); err != nil {
162
-
log.Printf("error connecting to jetstream: %v", err)
163
-
return nil, err
164
-
}
165
-
166
-
messages := make(chan []byte)
167
-
go j.readMessages(ctx, messages)
168
-
169
-
return messages, nil
170
-
}
+3
-3
knotserver/routes.go
+3
-3
knotserver/routes.go
···
436
436
return
437
437
}
438
438
439
-
h.js.UpdateDids([]string{did})
439
+
h.jc.UpdateDids([]string{did})
440
440
if err := h.e.AddMember(ThisServer, did); err != nil {
441
441
l.Error("adding member", "error", err.Error())
442
442
writeError(w, err.Error(), http.StatusInternalServerError)
···
472
472
writeError(w, err.Error(), http.StatusInternalServerError)
473
473
return
474
474
}
475
-
h.js.UpdateDids([]string{data.Did})
475
+
h.jc.UpdateDids([]string{data.Did})
476
476
477
477
repoName := filepath.Join(ownerDid, repo)
478
478
if err := h.e.AddRepo(data.Did, ThisServer, repoName); err != nil {
···
520
520
return
521
521
}
522
522
523
-
h.js.UpdateDids([]string{data.Did})
523
+
h.jc.UpdateDids([]string{data.Did})
524
524
if err := h.e.AddOwner(ThisServer, data.Did); err != nil {
525
525
l.Error("adding owner", "error", err.Error())
526
526
writeError(w, err.Error(), http.StatusInternalServerError)