knotserver,knotmirror/xrpc: `sync.requestCrawl` support

knotmirror ingests repo creation event from atproto relay through tap,
not from knotstream. Therefore, knotserver will request
`sync.requestCrawl` with optional new repo information to notify the
repo creation event.

Knot will call `sync.requestCrawl` on following cases:
- on startup
- when `/event` stream has failed
- on repo creation (mandatory)

Signed-off-by: Seongmin Lee <[e-mail]>

knotmirror/hostutil/hostutil.go

···
+package hostutil
+
+import (
+	"fmt"
+	"net/url"
+	"strings"
+
+	"github.com/bluesky-social/indigo/atproto/syntax"
+)
+
+func ParseHostname(raw string) (hsotname string, noSSL bool, err error) {
+	// handle case of bare hostname
+	if !strings.Contains(raw, "://") {
+		if strings.HasPrefix(raw, "localhost:") {
+			raw = "http://" + raw
+		} else {
+			raw = "https://" + raw
+		}
+	}
+
+	u, err := url.Parse(raw)
+	if err != nil {
+		return "", false, fmt.Errorf("not a valid host URL: %w", err)
+	}
+
+	switch u.Scheme {
+	case "https", "wss":
+		noSSL = false
+	case "http", "ws":
+		noSSL = true
+	default:
+		return "", false, fmt.Errorf("unsupported URL scheme: %s", u.Scheme)
+	}
+
+	// 'localhost' (exact string) is allowed *with* a required port number; SSL is optional
+	if u.Hostname() == "localhost" {
+		if u.Port() == "" || !strings.HasPrefix(u.Host, "localhost:") {
+			return "", false, fmt.Errorf("port number is required for localhost")
+		}
+		return u.Host, noSSL, nil
+	}
+
+	// port numbers not allowed otherwise
+	if u.Port() != "" {
+		return "", false, fmt.Errorf("port number not allowed for non-local names")
+	}
+
+	// check it is a real hostname (eg, not IP address or single-word alias)
+	h, err := syntax.ParseHandle(u.Host)
+	if err != nil {
+		return "", false, fmt.Errorf("not a public hostname")
+	}
+
+	// lower-case in response
+	return h.Normalize().String(), noSSL, nil
+}
+

knotmirror/knotmirror.go

···
 	}
 	logger.Info(fmt.Sprintf("clearing resyning states: %d records updated", rows))
 
-	xrpc := xrpc.New(logger, cfg, db, resolver)
 	knotstream := knotstream.NewKnotStream(logger, db, cfg)
 	crawler := NewCrawler(logger, db)
 	resyncer := NewResyncer(logger, db, gitm, cfg)
 	adminpage := NewAdminServer(logger, db, resyncer)
+	xrpc := xrpc.New(logger, cfg, db, resolver, knotstream)
 
 	// maintain repository list with tap
 	// NOTE: this can be removed once we introduce did-for-repo because then we can just listen to KnotStream for #identity events.

knotmirror/resyncer.go

···
 
 	repoUrl += "/info/refs?service=git-upload-pack"
 
+	r.logger.Debug("checking knot reachability", "url", repoUrl)
+
 	client := http.Client{
 		Timeout: 30 * time.Second,
 	}

knotmirror/xrpc/sync_requestCrawl.go

···
+package xrpc
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/atproto/atclient"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/bluesky-social/indigo/xrpc"
+	"tangled.org/core/api/tangled"
+	"tangled.org/core/knotmirror/db"
+	"tangled.org/core/knotmirror/hostutil"
+	"tangled.org/core/knotmirror/models"
+)
+
+func (x *Xrpc) RequestCrawl(w http.ResponseWriter, r *http.Request) {
+	var input tangled.SyncRequestCrawl_Input
+	if err := json.NewDecoder(r.Body).Decode(&input); err != nil {
+		panic("unimplemented")
+	}
+
+	ctx := r.Context()
+
+	l := x.logger.With("input", input)
+
+	hostname, noSSL, err := hostutil.ParseHostname(input.Hostname)
+	if err != nil {
+		l.Error("invalid hostname", "err", err)
+		writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("hostname field empty or invalid: %s", input.Hostname)})
+		return
+	}
+
+	// TODO: check if host is Knot with knot.describeServer
+
+	// store given repoAt to db
+	// this will allow knotmirror to ingest repo creation event bypassing tap.
+	// this step won't be needed once we introduce did-for-repo
+	// TODO(boltless): remove this section
+	if input.EnsureRepo != nil {
+		repoAt, err := syntax.ParseATURI(*input.EnsureRepo)
+		if err != nil {
+			l.Error("invalid repo at-uri", "err", err)
+			writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("repo parameter invalid: %s", *input.EnsureRepo)})
+			return
+		}
+		owner, err := x.resolver.ResolveIdent(ctx, repoAt.Authority().String())
+		if err != nil || owner.Handle.IsInvalidHandle() {
+			l.Error("failed to resolve ident", "err", err, "owner", repoAt.Authority().String())
+			writeErr(w, fmt.Errorf("failed to resolve repo owner"))
+			return
+		}
+		xrpcc := xrpc.Client{Host: owner.PDSEndpoint()}
+		out, err := atproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String())
+		if err != nil {
+			l.Error("failed to get repo record", "err", err, "repo", repoAt)
+			writeErr(w, fmt.Errorf("failed to get repo record"))
+			return
+		}
+		record := out.Value.Val.(*tangled.Repo)
+
+		knotUrl := record.Knot
+		if !strings.Contains(record.Knot, "://") {
+			if noSSL {
+				knotUrl = "http://" + knotUrl
+			} else {
+				knotUrl = "https://" + knotUrl
+			}
+		}
+
+		repo := &models.Repo{
+			Did:        owner.DID,
+			Rkey:       repoAt.RecordKey(),
+			Cid:        (*syntax.CID)(out.Cid),
+			Name:       record.Name,
+			KnotDomain: knotUrl,
+			State:      models.RepoStatePending,
+			ErrorMsg:   "",
+			RetryAfter: 0,
+			RetryCount: 0,
+		}
+
+		if err := db.UpsertRepo(ctx, x.db, repo); err != nil {
+			l.Error("failed to upsert repo", "err", err)
+			writeErr(w, err)
+			return
+		}
+	}
+
+	// subscribe to requested host
+	if !x.ks.CheckIfSubscribed(hostname) {
+		if err := x.ks.SubscribeHost(ctx, hostname, noSSL); err != nil {
+			// TODO(boltless): return HostBanned on banned hosts
+			l.Error("failed to subscribe host", "err", err)
+			writeErr(w, err)
+			return
+		}
+	}
+
+	w.WriteHeader(http.StatusOK)
+}

knotmirror/xrpc/xrpc.go

···
 	"tangled.org/core/api/tangled"
 	"tangled.org/core/idresolver"
 	"tangled.org/core/knotmirror/config"
+	"tangled.org/core/knotmirror/knotstream"
 	"tangled.org/core/log"
 )
 
···
 	cfg      *config.Config
 	db       *sql.DB
 	resolver *idresolver.Resolver
+	ks       *knotstream.KnotStream
 	logger   *slog.Logger
 }
 
-func New(logger *slog.Logger, cfg *config.Config, db *sql.DB, resolver *idresolver.Resolver) *Xrpc {
+func New(logger *slog.Logger, cfg *config.Config, db *sql.DB, resolver *idresolver.Resolver, ks *knotstream.KnotStream) *Xrpc {
 	return &Xrpc{
 		cfg,
 		db,
 		resolver,
+		ks,
 		log.SubLogger(logger, "xrpc"),
 	}
 }
···
 	r.Get("/"+tangled.GitTempListCommitsNSID, x.ListCommits)
 	r.Get("/"+tangled.GitTempListLanguagesNSID, x.ListLanguages)
 	r.Get("/"+tangled.GitTempListTagsNSID, x.ListTags)
+	r.Post("/"+tangled.SyncRequestCrawlNSID, x.RequestCrawl)
 
 	return r
 }

knotserver/config/config.go

···
 }
 
 type Config struct {
-	Repo            Repo   `env:",prefix=KNOT_REPO_"`
-	Server          Server `env:",prefix=KNOT_SERVER_"`
-	Git             Git    `env:",prefix=KNOT_GIT_"`
-	AppViewEndpoint string `env:"APPVIEW_ENDPOINT, default=https://tangled.org"`
+	Repo            Repo     `env:",prefix=KNOT_REPO_"`
+	Server          Server   `env:",prefix=KNOT_SERVER_"`
+	Git             Git      `env:",prefix=KNOT_GIT_"`
+	AppViewEndpoint string   `env:"APPVIEW_ENDPOINT, default=https://tangled.org"`
+	KnotMirrors     []string `env:"KNOT_MIRRORS, default=https://mirror.tangled.network"`
 }
 
 func Load(ctx context.Context) (*Config, error) {

knotserver/events.go

···
 	"strconv"
 	"time"
 
+	"github.com/bluesky-social/indigo/xrpc"
 	"github.com/gorilla/websocket"
+	"tangled.org/core/api/tangled"
 	"tangled.org/core/log"
 )
 
···
 		return
 	}
 
+	// try request crawl when connection closed
+	defer func() {
+		go func() {
+			retryCtx, retryCancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer retryCancel()
+			if err := h.requestCrawl(retryCtx); err != nil {
+				l.Error("error requesting crawls", "err", err)
+			}
+		}()
+	}()
+
 	for {
 		// wait for new data or timeout
 		select {
···
 
 	return nil
 }
+
+func (h *Knot) requestCrawl(ctx context.Context) error {
+	h.l.Info("requesting crawl", "mirrors", h.c.KnotMirrors)
+	input := &tangled.SyncRequestCrawl_Input{
+		Hostname: h.c.Server.Hostname,
+	}
+	for _, knotmirror := range h.c.KnotMirrors {
+		xrpcc := xrpc.Client{Host: knotmirror}
+		if err := tangled.SyncRequestCrawl(ctx, &xrpcc, input); err != nil {
+			h.l.Error("error requesting crawl", "err", err)
+		} else {
+			h.l.Info("crawl requested successfully")
+		}
+	}
+	return nil
+}

knotserver/server.go

···
 	"fmt"
 	"net/http"
 
+	"github.com/bluesky-social/indigo/xrpc"
 	"github.com/urfave/cli/v3"
 	"tangled.org/core/api/tangled"
 	"tangled.org/core/hook"
···
 
 	logger.Info("starting internal server", "address", c.Server.InternalListenAddr)
 	go http.ListenAndServe(c.Server.InternalListenAddr, imux)
+
+	// TODO(boltless): too lazy here. should clear this up
+	go func() {
+		input := &tangled.SyncRequestCrawl_Input{
+			Hostname: c.Server.Hostname,
+		}
+		for _, knotmirror := range c.KnotMirrors {
+			xrpcc := xrpc.Client{Host: knotmirror}
+			if err := tangled.SyncRequestCrawl(ctx, &xrpcc, input); err != nil {
+				logger.Error("error requesting crawl", "err", err)
+			} else {
+				logger.Info("crawl requested successfully")
+			}
+		}
+	}()
 
 	logger.Info("starting main server", "address", c.Server.ListenAddr)
 	logger.Error("server error", "error", http.ListenAndServe(c.Server.ListenAddr, mux))

knotserver/xrpc/create_repo.go

···
 package xrpc
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
 	"path/filepath"
 	"strings"
+	"time"
 
 	comatproto "github.com/bluesky-social/indigo/api/atproto"
 	"github.com/bluesky-social/indigo/atproto/syntax"
···
 		repoPath,
 	)
 
+	// request crawl for this repository
+	repoAt := fmt.Sprintf("at://%s/%s/%s", actorDid, tangled.RepoNSID, rkey)
+	go func() {
+		rCtx, rCancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer rCancel()
+		h.requestCrawl(rCtx, &tangled.SyncRequestCrawl_Input{
+			Hostname:   h.Config.Server.Hostname,
+			EnsureRepo: &repoAt,
+		})
+	}()
+
 	w.WriteHeader(http.StatusOK)
+}
+
+func (h *Xrpc) requestCrawl(ctx context.Context, input *tangled.SyncRequestCrawl_Input) error {
+	h.Logger.Info("requesting crawl", "mirrors", h.Config.KnotMirrors)
+	for _, knotmirror := range h.Config.KnotMirrors {
+		xrpcc := xrpc.Client{Host: knotmirror}
+		if err := tangled.SyncRequestCrawl(ctx, &xrpcc, input); err != nil {
+			h.Logger.Error("error requesting crawl", "err", err)
+		} else {
+			h.Logger.Info("crawl requested successfully")
+		}
+	}
+	return nil
 }
 
 func validateRepoName(name string) error {

nix/modules/knot.nix

···
           '';
         };
 
+        knotmirrors = mkOption {
+          type = types.listOf types.str;
+          default = [
+            "https://mirror.tangled.network"
+          ];
+          description = "List of knotmirror hosts to request crawl";
+        };
+
         server = {
           listenAddr = mkOption {
             type = types.str;
···
             "KNOT_SERVER_PLC_URL=${cfg.server.plcUrl}"
             "KNOT_SERVER_JETSTREAM_ENDPOINT=${cfg.server.jetstreamEndpoint}"
             "KNOT_SERVER_OWNER=${cfg.server.owner}"
+            "KNOT_MIRRORS=${concatStringsSep "," cfg.knotmirrors}"
             "KNOT_SERVER_LOG_DIDS=${
               if cfg.server.logDids
               then "true"

nix/vm.nix

···
             jetstreamEndpoint = jetstream;
             listenAddr = "0.0.0.0:6444";
           };
+          knotmirrors = [
+            "http://localhost:7000"
+          ];
         };
         services.tangled.spindle = {
           enable = true;