lewis.moe
forked from
tangled.org/core
Monorepo for Tangled
knotserver: add member to knot
anirudh.fi
+3
-2
knotserver/config/config.go
+3
-2
knotserver/config/config.go
···
22
22
}
23
23
24
24
type Config struct {
25
-
Repo Repo `env:",prefix=KNOT_REPO_"`
26
-
Server Server `env:",prefix=KNOT_SERVER_"`
25
+
Repo Repo `env:",prefix=KNOT_REPO_"`
26
+
Server Server `env:",prefix=KNOT_SERVER_"`
27
+
AppViewEndpoint string `env:"APPVIEW_ENDPOINT, default=https://tangled.sh"`
27
28
}
28
29
29
30
func Load(ctx context.Context) (*Config, error) {
+5
-86
knotserver/handler.go
+5
-86
knotserver/handler.go
···
2
2
3
3
import (
4
4
"context"
5
-
"encoding/json"
6
5
"fmt"
7
-
"log"
8
6
"net/http"
9
-
"time"
10
7
11
8
"github.com/go-chi/chi/v5"
12
-
tangled "github.com/sotangled/tangled/api/tangled"
13
9
"github.com/sotangled/tangled/knotserver/config"
14
10
"github.com/sotangled/tangled/knotserver/db"
15
11
"github.com/sotangled/tangled/knotserver/jsclient"
···
94
90
r.Put("/new", h.NewRepo)
95
91
})
96
92
93
+
r.Route("/member", func(r chi.Router) {
94
+
r.Use(h.VerifySignature)
95
+
r.Put("/add", h.NewRepo)
96
+
})
97
+
97
98
// Initialize the knot with an owner and public key.
98
99
r.With(h.VerifySignature).Post("/init", h.Init)
99
100
···
105
106
106
107
return r, nil
107
108
}
108
-
109
-
func (h *Handle) StartJetstream(ctx context.Context) error {
110
-
collections := []string{tangled.PublicKeyNSID, tangled.KnotMemberNSID}
111
-
dids := []string{}
112
-
113
-
var lastTimeUs int64
114
-
var err error
115
-
lastTimeUs, err = h.db.GetLastTimeUs()
116
-
if err != nil {
117
-
log.Println("couldn't get last time us, starting from now")
118
-
lastTimeUs = time.Now().UnixMicro()
119
-
}
120
-
// If last time is older than a week, start from now
121
-
if time.Now().UnixMicro()-lastTimeUs > 7*24*60*60*1000*1000 {
122
-
lastTimeUs = time.Now().UnixMicro()
123
-
log.Printf("last time us is older than a week. discarding that and starting from now.")
124
-
err = h.db.SaveLastTimeUs(lastTimeUs)
125
-
if err != nil {
126
-
log.Println("failed to save last time us")
127
-
}
128
-
}
129
-
130
-
log.Printf("found last time_us %d", lastTimeUs)
131
-
132
-
h.js = jsclient.NewJetstreamClient(collections, dids)
133
-
messages, err := h.js.ReadJetstream(ctx, lastTimeUs)
134
-
if err != nil {
135
-
return fmt.Errorf("failed to read from jetstream: %w", err)
136
-
}
137
-
138
-
go func() {
139
-
log.Println("waiting for knot to be initialized")
140
-
<-h.init
141
-
log.Println("initalized jetstream watcher")
142
-
143
-
for msg := range messages {
144
-
var data map[string]interface{}
145
-
if err := json.Unmarshal(msg, &data); err != nil {
146
-
log.Printf("error unmarshaling message: %v", err)
147
-
continue
148
-
}
149
-
150
-
if kind, ok := data["kind"].(string); ok && kind == "commit" {
151
-
commit := data["commit"].(map[string]interface{})
152
-
153
-
switch commit["collection"].(string) {
154
-
case tangled.PublicKeyNSID:
155
-
did := data["did"].(string)
156
-
record := commit["record"].(map[string]interface{})
157
-
if err := h.db.AddPublicKeyFromRecord(did, record); err != nil {
158
-
log.Printf("failed to add public key: %v", err)
159
-
} else {
160
-
log.Printf("added public key from firehose: %s", data["did"])
161
-
}
162
-
case tangled.KnotMemberNSID:
163
-
did := data["did"].(string)
164
-
record := commit["record"].(map[string]interface{})
165
-
ok, err := h.e.E.Enforce(did, ThisServer, ThisServer, "server:invite")
166
-
if err != nil || !ok {
167
-
log.Printf("failed to add member from did %s", did)
168
-
} else {
169
-
log.Printf("adding member")
170
-
if err := h.e.AddMember(ThisServer, record["member"].(string)); err != nil {
171
-
log.Printf("failed to add member: %v", err)
172
-
} else {
173
-
log.Printf("added member from firehose: %s", record["member"])
174
-
}
175
-
}
176
-
default:
177
-
}
178
-
179
-
lastTimeUs := int64(data["time_us"].(float64))
180
-
if err := h.db.SaveLastTimeUs(lastTimeUs); err != nil {
181
-
log.Printf("failed to save last time us: %v", err)
182
-
}
183
-
}
184
-
185
-
}
186
-
}()
187
-
188
-
return nil
189
-
}
+144
knotserver/jetstream.go
+144
knotserver/jetstream.go
···
1
+
package knotserver
2
+
3
+
import (
4
+
"context"
5
+
"encoding/json"
6
+
"fmt"
7
+
"io"
8
+
"log"
9
+
"net/http"
10
+
"path"
11
+
"strings"
12
+
"time"
13
+
14
+
"github.com/sotangled/tangled/api/tangled"
15
+
"github.com/sotangled/tangled/knotserver/db"
16
+
"github.com/sotangled/tangled/knotserver/jsclient"
17
+
)
18
+
19
+
func (h *Handle) StartJetstream(ctx context.Context) error {
20
+
collections := []string{tangled.PublicKeyNSID, tangled.KnotMemberNSID}
21
+
dids := []string{}
22
+
23
+
lastTimeUs, err := h.getLastTimeUs()
24
+
if err != nil {
25
+
return err
26
+
}
27
+
28
+
h.js = jsclient.NewJetstreamClient(collections, dids)
29
+
messages, err := h.js.ReadJetstream(ctx, lastTimeUs)
30
+
if err != nil {
31
+
return fmt.Errorf("failed to read from jetstream: %w", err)
32
+
}
33
+
34
+
go h.processMessages(messages)
35
+
36
+
return nil
37
+
}
38
+
39
+
func (h *Handle) getLastTimeUs() (int64, error) {
40
+
lastTimeUs, err := h.db.GetLastTimeUs()
41
+
if err != nil {
42
+
log.Println("couldn't get last time us, starting from now")
43
+
lastTimeUs = time.Now().UnixMicro()
44
+
}
45
+
46
+
// If last time is older than a week, start from now
47
+
if time.Now().UnixMicro()-lastTimeUs > 7*24*60*60*1000*1000 {
48
+
lastTimeUs = time.Now().UnixMicro()
49
+
log.Printf("last time us is older than a week. discarding that and starting from now.")
50
+
err = h.db.SaveLastTimeUs(lastTimeUs)
51
+
if err != nil {
52
+
log.Println("failed to save last time us")
53
+
}
54
+
}
55
+
56
+
log.Printf("found last time_us %d", lastTimeUs)
57
+
return lastTimeUs, nil
58
+
}
59
+
60
+
func (h *Handle) processPublicKey(did string, record map[string]interface{}) {
61
+
if err := h.db.AddPublicKeyFromRecord(did, record); err != nil {
62
+
log.Printf("failed to add public key: %v", err)
63
+
} else {
64
+
log.Printf("added public key from firehose: %s", did)
65
+
}
66
+
}
67
+
68
+
func (h *Handle) fetchAndAddKeys(did string) {
69
+
resp, err := http.Get(path.Join(h.c.AppViewEndpoint, did))
70
+
if err != nil {
71
+
log.Printf("error getting keys for %s: %v", did, err)
72
+
return
73
+
}
74
+
defer resp.Body.Close()
75
+
76
+
plaintext, err := io.ReadAll(resp.Body)
77
+
if err != nil {
78
+
log.Printf("error reading response body: %v", err)
79
+
return
80
+
}
81
+
82
+
for _, key := range strings.Split(string(plaintext), "\n") {
83
+
if key == "" {
84
+
continue
85
+
}
86
+
pk := db.PublicKey{
87
+
Did: did,
88
+
}
89
+
pk.Key = key
90
+
if err := h.db.AddPublicKey(pk); err != nil {
91
+
log.Printf("failed to add public key: %v", err)
92
+
}
93
+
}
94
+
}
95
+
96
+
func (h *Handle) processKnotMember(did string, record map[string]interface{}) {
97
+
ok, err := h.e.E.Enforce(did, ThisServer, ThisServer, "server:invite")
98
+
if err != nil || !ok {
99
+
log.Printf("failed to add member from did %s", did)
100
+
return
101
+
}
102
+
103
+
log.Printf("adding member")
104
+
if err := h.e.AddMember(ThisServer, record["member"].(string)); err != nil {
105
+
log.Printf("failed to add member: %v", err)
106
+
} else {
107
+
log.Printf("added member from firehose: %s", record["member"])
108
+
}
109
+
110
+
h.fetchAndAddKeys(did)
111
+
h.js.UpdateDids([]string{did})
112
+
}
113
+
114
+
func (h *Handle) processMessages(messages <-chan []byte) {
115
+
log.Println("waiting for knot to be initialized")
116
+
<-h.init
117
+
log.Println("initalized jetstream watcher")
118
+
119
+
for msg := range messages {
120
+
var data map[string]interface{}
121
+
if err := json.Unmarshal(msg, &data); err != nil {
122
+
log.Printf("error unmarshaling message: %v", err)
123
+
continue
124
+
}
125
+
126
+
if kind, ok := data["kind"].(string); ok && kind == "commit" {
127
+
commit := data["commit"].(map[string]interface{})
128
+
did := data["did"].(string)
129
+
record := commit["record"].(map[string]interface{})
130
+
131
+
switch commit["collection"].(string) {
132
+
case tangled.PublicKeyNSID:
133
+
h.processPublicKey(did, record)
134
+
case tangled.KnotMemberNSID:
135
+
h.processKnotMember(did, record)
136
+
}
137
+
138
+
lastTimeUs := int64(data["time_us"].(float64))
139
+
if err := h.db.SaveLastTimeUs(lastTimeUs); err != nil {
140
+
log.Printf("failed to save last time us: %v", err)
141
+
}
142
+
}
143
+
}
144
+
}
+39
-2
knotserver/routes.go
+39
-2
knotserver/routes.go
···
401
401
w.WriteHeader(http.StatusNoContent)
402
402
}
403
403
404
-
// TODO: make this set the initial user as the owner
404
+
func (h *Handle) AddMember(w http.ResponseWriter, r *http.Request) {
405
+
data := struct {
406
+
Did string `json:"did"`
407
+
PublicKeys []string `json:"keys"`
408
+
}{}
409
+
410
+
if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
411
+
writeError(w, "invalid request body", http.StatusBadRequest)
412
+
return
413
+
}
414
+
415
+
did := data.Did
416
+
for _, k := range data.PublicKeys {
417
+
pk := db.PublicKey{
418
+
Did: did,
419
+
}
420
+
pk.Key = k
421
+
err := h.db.AddPublicKey(pk)
422
+
if err != nil {
423
+
writeError(w, err.Error(), http.StatusInternalServerError)
424
+
return
425
+
}
426
+
}
427
+
428
+
h.js.UpdateDids([]string{did})
429
+
if err := h.e.AddMember(ThisServer, did); err != nil {
430
+
log.Println(err)
431
+
writeError(w, err.Error(), http.StatusInternalServerError)
432
+
return
433
+
}
434
+
435
+
w.WriteHeader(http.StatusNoContent)
436
+
}
437
+
405
438
func (h *Handle) Init(w http.ResponseWriter, r *http.Request) {
406
439
if h.knotInitialized {
407
440
writeError(w, "knot already initialized", http.StatusConflict)
···
441
474
}
442
475
443
476
h.js.UpdateDids([]string{data.Did})
444
-
h.e.AddOwner(ThisServer, data.Did)
477
+
if err := h.e.AddOwner(ThisServer, data.Did); err != nil {
478
+
log.Println(err)
479
+
writeError(w, err.Error(), http.StatusInternalServerError)
480
+
return
481
+
}
445
482
// Signal that the knot is ready
446
483
close(h.init)
447
484