From caf7b4ca00102045e90feeeb96fd3031cdbbccef Mon Sep 17 00:00:00 2001 From: Anirudh Oppiliappan Date: Sun, 2 Feb 2025 15:51:56 +0200 Subject: [PATCH] knotserver: /init and /user endpoints When the knot is initialized, the h.init channel is closed signalling the Jetstream watcher to begin, with the owner DID in filter. There's no null filter so starting it without any DIDs will give us events for all DIDs. /init also checks if h.knotInitialized is true, and will 409 Conflict if /init is called again. PUT /user is similar, but will only trigger a call to UpdateDids. --- flake.nix | 3 +- knotserver/db/init.go | 6 +++ knotserver/db/pubkeys.go | 4 +- knotserver/db/users.go | 11 +++++ knotserver/handler.go | 25 +++++++++-- knotserver/jsclient/jetstream.go | 2 - knotserver/routes.go | 77 ++++++++++++++++++++++++++++++++ 7 files changed, 118 insertions(+), 10 deletions(-) create mode 100644 knotserver/db/users.go diff --git a/flake.nix b/flake.nix index 3fef51ba..8c1bac35 100644 --- a/flake.nix +++ b/flake.nix @@ -45,7 +45,6 @@ nativeBuildInputs = [ pkgs.go pkgs.air - pkgs.templ pkgs.gopls pkgs.httpie pkgs.indigo-lexgen @@ -54,7 +53,7 @@ ]; }; }); - apps = forAllSystems (system: + apps = forAllSystems (system: let pkgs = nixpkgsFor."${system}"; air-watcher = name: pkgs.writeShellScriptBin "run" diff --git a/knotserver/db/init.go b/knotserver/db/init.go index 0e20c836..f99a1e31 100644 --- a/knotserver/db/init.go +++ b/knotserver/db/init.go @@ -25,6 +25,12 @@ func Setup(dbPath string) (*DB, error) { created timestamp default current_timestamp, unique(did, name, key) ); + create table if not exists users ( + id integer primary key autoincrement, + did text not null, + unique(did), + foreign key (did) references public_keys(did) on delete cascade + ); create table if not exists repos ( id integer primary key autoincrement, did text not null, diff --git a/knotserver/db/pubkeys.go b/knotserver/db/pubkeys.go index b9a1f1b9..455c5c05 100644 --- a/knotserver/db/pubkeys.go +++ b/knotserver/db/pubkeys.go @@ -11,7 +11,7 @@ type PublicKey struct { tangled.PublicKey } -func (d *DB) AddPublicKeyFromRecord(recordIface map[string]interface{}) error { +func (d *DB) AddPublicKeyFromRecord(did string, recordIface map[string]interface{}) error { record := make(map[string]string) for k, v := range recordIface { if str, ok := v.(string); ok { @@ -20,7 +20,7 @@ func (d *DB) AddPublicKeyFromRecord(recordIface map[string]interface{}) error { } pk := PublicKey{ - Did: record["did"], + Did: did, } pk.Name = record["name"] pk.Key = record["key"] diff --git a/knotserver/db/users.go b/knotserver/db/users.go new file mode 100644 index 00000000..a2f484e8 --- /dev/null +++ b/knotserver/db/users.go @@ -0,0 +1,11 @@ +package db + +func (d *DB) AddUser(did string) error { + _, err := d.db.Exec(`insert into users (did) values (?)`, did) + return err +} + +func (d *DB) RemoveUser(did string) error { + _, err := d.db.Exec(`delete from users where did = ?`, did) + return err +} diff --git a/knotserver/handler.go b/knotserver/handler.go index cd7cc564..ef00993f 100644 --- a/knotserver/handler.go +++ b/knotserver/handler.go @@ -18,14 +18,20 @@ type Handle struct { c *config.Config db *db.DB js *jsclient.JetstreamClient + + // init is a channel that is closed when the knot has been initailized + // i.e. when the first user (knot owner) has been added. + init chan struct{} + knotInitialized bool } func Setup(ctx context.Context, c *config.Config, db *db.DB) (http.Handler, error) { r := chi.NewRouter() h := Handle{ - c: c, - db: db, + c: c, + db: db, + init: make(chan struct{}), } err := h.StartJetstream(ctx) @@ -33,6 +39,11 @@ func Setup(ctx context.Context, c *config.Config, db *db.DB) (http.Handler, erro return nil, fmt.Errorf("failed to start jetstream: %w", err) } + // TODO: close this channel and set h.knotInitialized *only after* + // checking if we have an owner. + close(h.init) + h.knotInitialized = true + r.Get("/", h.Index) r.Route("/{did}", func(r chi.Router) { // Repo routes @@ -63,7 +74,8 @@ func Setup(ctx context.Context, c *config.Config, db *db.DB) (http.Handler, erro }) // Add a new user to the knot - // r.With(h.VerifySignature).Put("/user", h.AddUser) + r.With(h.VerifySignature).Put("/user", h.AddUser) + r.With(h.VerifySignature).Post("/init", h.Init) // Health check. Used for two-way verification with appview. r.With(h.VerifySignature).Get("/health", h.Health) @@ -85,6 +97,10 @@ func (h *Handle) StartJetstream(ctx context.Context) error { } go func() { + log.Println("waiting for knot to be initialized") + <-h.init + log.Println("initalized jetstream watcher") + for msg := range messages { var data map[string]interface{} if err := json.Unmarshal(msg, &data); err != nil { @@ -97,8 +113,9 @@ func (h *Handle) StartJetstream(ctx context.Context) error { switch commit["collection"].(string) { case tangled.PublicKeyNSID: + did := data["did"].(string) record := commit["record"].(map[string]interface{}) - if err := h.db.AddPublicKeyFromRecord(record); err != nil { + if err := h.db.AddPublicKeyFromRecord(did, record); err != nil { log.Printf("failed to add public key: %v", err) } log.Printf("added public key from firehose: %s", data["did"]) diff --git a/knotserver/jsclient/jetstream.go b/knotserver/jsclient/jetstream.go index 1017a319..f87920ed 100644 --- a/knotserver/jsclient/jetstream.go +++ b/knotserver/jsclient/jetstream.go @@ -100,8 +100,6 @@ func (j *JetstreamClient) connect(cursor int64) error { queryParams := j.buildQueryParams(cursor) u := j.buildWebsocketURL(queryParams) - log.Printf("connecting to jetstream at: %s", u.String()) - dialer := websocket.Dialer{ HandshakeTimeout: 10 * time.Second, } diff --git a/knotserver/routes.go b/knotserver/routes.go index 7c246e33..77ab23a0 100644 --- a/knotserver/routes.go +++ b/knotserver/routes.go @@ -391,6 +391,83 @@ func (h *Handle) NewRepo(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } +func (h *Handle) AddUser(w http.ResponseWriter, r *http.Request) { + data := struct { + DID string `json:"did"` + PublicKey string `json:"pubkey"` + }{} + + if err := json.NewDecoder(r.Body).Decode(&data); err != nil { + writeError(w, "invalid request body", http.StatusBadRequest) + return + } + + did := data.DID + key := data.PublicKey + + if err := h.db.AddUser(did); err == nil { + pk := db.PublicKey{ + Did: did, + } + pk.Key = key + pk.Name = "default" + err := h.db.AddPublicKey(pk) + if err != nil { + writeError(w, err.Error(), http.StatusInternalServerError) + return + } + } else { + writeError(w, err.Error(), http.StatusInternalServerError) + return + } + + h.js.UpdateDids([]string{did}) + + w.WriteHeader(http.StatusNoContent) +} + +// TODO: make this set the initial user as the owner +func (h *Handle) Init(w http.ResponseWriter, r *http.Request) { + if h.knotInitialized { + writeError(w, "knot already initialized", http.StatusConflict) + return + } + + data := struct { + DID string `json:"did"` + PublicKey string `json:"pubkey"` + }{} + + if err := json.NewDecoder(r.Body).Decode(&data); err != nil { + writeError(w, "invalid request body", http.StatusBadRequest) + return + } + + did := data.DID + key := data.PublicKey + + if err := h.db.AddUser(did); err == nil { + pk := db.PublicKey{ + Did: did, + } + pk.Key = key + pk.Name = "default" + err := h.db.AddPublicKey(pk) + if err != nil { + writeError(w, err.Error(), http.StatusInternalServerError) + return + } + } else { + writeError(w, err.Error(), http.StatusInternalServerError) + return + } + + h.js.UpdateDids([]string{did}) + // Signal that the knot is ready + close(h.init) + w.WriteHeader(http.StatusNoContent) +} + func (h *Handle) Health(w http.ResponseWriter, r *http.Request) { log.Println("got health check") mac := hmac.New(sha256.New, []byte(h.c.Server.Secret)) -- knotmirror 0.0.1