From 5c25e50c84f7a3a99de9829840fc8d4c40404f6a Mon Sep 17 00:00:00 2001 From: Lewis Date: Fri, 06 Mar 2026 06:57:30 +0000 Subject: [PATCH] knotserver: add repo DID migration on startup Signed-off-by: Lewis --- eventconsumer/consumer.go | 13 ++++++++----- knotserver/events.go | 14 +++++++------- knotserver/migrate.go | 271 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ knotserver/server.go | 2 ++ spindle/stream.go | 21 +++++++++++---------- 5 file(s) changed, 299 insertion(s)(+), 22 deletion(s)(-) diff --git a/eventconsumer/consumer.go b/eventconsumer/consumer.go --- a/eventconsumer/consumer.go +++ b/eventconsumer/consumer.go @@ -19,9 +19,9 @@ type ProcessFunc func(ctx context.Context, source Source, message Message) error type Message struct { - Rkey string - Nsid string - // do not full deserialize this portion of the message, processFunc can do that + Rkey string + Nsid string + Created int64 `json:"created"` EventJson json.RawMessage `json:"event"` } @@ -159,8 +159,11 @@ c.logger.Error("error deserializing message", "source", j.source.Key(), "err", err) return } - // update cursor - c.cfg.CursorStore.Set(j.source.Key(), time.Now().UnixNano()) + cursorVal := msg.Created + if cursorVal == 0 { + cursorVal = time.Now().UnixNano() + } + c.cfg.CursorStore.Set(j.source.Key(), cursorVal) if err := c.cfg.ProcessFunc(ctx, j.source, msg); err != nil { c.logger.Error("error processing message", "source", j.source, "err", err) diff --git a/knotserver/events.go b/knotserver/events.go --- a/knotserver/events.go +++ b/knotserver/events.go @@ -46,14 +46,14 @@ } } }() - defaultCursor := time.Now().UnixNano() + var cursor int64 cursorStr := r.URL.Query().Get("cursor") - cursor, err := strconv.ParseInt(cursorStr, 10, 64) - if err != nil { - l.Error("empty or invalid cursor", "invalidCursor", cursorStr, "default", defaultCursor) - } - if cursor == 0 { - cursor = defaultCursor + if cursorStr != "" { + cursor, err = strconv.ParseInt(cursorStr, 10, 64) + if err != nil { + l.Error("invalid cursor, starting from beginning", "invalidCursor", cursorStr) + cursor = 0 + } } l.Debug("going through backfill", "cursor", cursor) diff --git a/knotserver/migrate.go b/knotserver/migrate.go new file mode 100644 --- /dev/null +++ b/knotserver/migrate.go @@ -0,0 +1,271 @@ +package knotserver + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log/slog" + "math/rand/v2" + "os" + "path/filepath" + "strings" + "time" + + "tangled.org/core/knotserver/config" + "tangled.org/core/knotserver/db" + "tangled.org/core/knotserver/repodid" + "tangled.org/core/notifier" + "tangled.org/core/rbac" +) + +type legacyRepo struct { + ownerDid string + repoName string + oldPath string +} + +func scanLegacyRepos(scanPath string, logger *slog.Logger) []legacyRepo { + topEntries, err := os.ReadDir(scanPath) + if err != nil { + logger.Error("reading scan path", "error", err) + return nil + } + + var repos []legacyRepo + for _, entry := range topEntries { + if !entry.IsDir() || !strings.HasPrefix(entry.Name(), "did:") { + continue + } + + ownerPath := filepath.Join(scanPath, entry.Name()) + + if _, headErr := os.Stat(filepath.Join(ownerPath, "HEAD")); headErr == nil { + continue + } + + subEntries, readErr := os.ReadDir(ownerPath) + if readErr != nil { + logger.Error("reading owner dir", "ownerDir", ownerPath, "error", readErr) + continue + } + + ownerDid := entry.Name() + for _, sub := range subEntries { + if !sub.IsDir() { + continue + } + subPath := filepath.Join(ownerPath, sub.Name()) + if _, statErr := os.Stat(filepath.Join(subPath, "HEAD")); statErr != nil { + logger.Warn("skipping non-repo directory", "path", subPath) + continue + } + repos = append(repos, legacyRepo{ + ownerDid: ownerDid, + repoName: sub.Name(), + oldPath: subPath, + }) + } + } + return repos +} + +func migrateReposOnStartup(ctx context.Context, c *config.Config, d *db.DB, e *rbac.Enforcer, n *notifier.Notifier, logger *slog.Logger) { + repos := scanLegacyRepos(c.Repo.ScanPath, logger) + if len(repos) == 0 { + logger.Info("no legacy repos found, migration complete") + return + } + + logger.Info("starting legacy repo migration", "count", len(repos)) + start := time.Now() + + knotServiceUrl := "https://" + c.Server.Hostname + if c.Server.Dev { + knotServiceUrl = "http://" + c.Server.Hostname + } + + migrated := 0 + for _, repo := range repos { + select { + case <-ctx.Done(): + logger.Info("migration interrupted by shutdown", "migrated", migrated, "remaining", len(repos)-migrated) + return + default: + } + + err := migrateOneRepo(ctx, c, d, e, n, logger, repo, knotServiceUrl) + if err != nil { + logger.Error("migration failed for repo", "owner", repo.ownerDid, "repo", repo.repoName, "error", err) + continue + } + migrated++ + } + + logger.Info("legacy repo migration complete", "migrated", migrated, "total", len(repos), "duration", time.Since(start)) +} + +func migrateOneRepo( + ctx context.Context, + c *config.Config, + d *db.DB, + e *rbac.Enforcer, + n *notifier.Notifier, + logger *slog.Logger, + repo legacyRepo, + knotServiceUrl string, +) error { + l := logger.With("owner", repo.ownerDid, "repo", repo.repoName) + + repoDid, err := d.GetRepoDid(repo.ownerDid, repo.repoName) + needsMint := errors.Is(err, sql.ErrNoRows) + if err != nil && !needsMint { + return fmt.Errorf("checking repo_keys: %w", err) + } + + if needsMint { + repoDid, err = mintAndStoreRepoDID(ctx, c, d, l, repo, knotServiceUrl) + if err != nil { + return err + } + } + + if err := rewriteRBACPolicies(e, repo.ownerDid, repo.repoName, repoDid, l); err != nil { + l.Error("RBAC rewrite failed (non-fatal)", "error", err) + } + + newPath := filepath.Join(c.Repo.ScanPath, repoDid) + + if err := moveRepoOnDisk(repo.oldPath, newPath, l); err != nil { + return err + } + + ownerDir := filepath.Dir(repo.oldPath) + if err := os.Remove(ownerDir); err != nil && !errors.Is(err, os.ErrNotExist) { + l.Warn("could not remove empty owner dir", "path", ownerDir, "error", err) + } + + if err := d.EmitDIDAssign(n, repo.ownerDid, repo.repoName, repoDid, ""); err != nil { + l.Error("emitting didAssign event failed (non-fatal)", "error", err) + } + + l.Info("migrated repo", "repoDid", repoDid) + return nil +} + +func mintAndStoreRepoDID( + ctx context.Context, + c *config.Config, + d *db.DB, + l *slog.Logger, + repo legacyRepo, + knotServiceUrl string, +) (string, error) { + prepared, err := repodid.PrepareRepoDID(c.Server.PlcUrl, knotServiceUrl) + if err != nil { + return "", fmt.Errorf("preparing DID: %w", err) + } + + if err := submitWithBackoff(ctx, prepared, l); err != nil { + return "", fmt.Errorf("PLC submission: %w", err) + } + + if err := d.StoreRepoKey(prepared.RepoDid, prepared.SigningKeyRaw, repo.ownerDid, repo.repoName, ""); err != nil { + return "", fmt.Errorf("storing repo key: %w", err) + } + + return prepared.RepoDid, nil +} + +func submitWithBackoff(ctx context.Context, prepared *repodid.PreparedDID, l *slog.Logger) error { + backoff := 2 * time.Second + const maxBackoff = 5 * time.Minute + const maxAttempts = 20 + + for attempt := range maxAttempts { + plcCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + err := prepared.Submit(plcCtx) + cancel() + + if err == nil { + return nil + } + + errMsg := err.Error() + retryable := strings.Contains(errMsg, "429") || + strings.Contains(errMsg, "500") || + strings.Contains(errMsg, "502") || + strings.Contains(errMsg, "503") || + strings.Contains(errMsg, "504") + + if !retryable || attempt == maxAttempts-1 { + return err + } + + jitter := time.Duration(rand.Int64N(int64(backoff / 4))) + delay := backoff + jitter + + l.Info("PLC rate limited, backing off", "delay", delay, "attempt", attempt+1) + + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(delay): + } + + backoff = min(backoff*2, maxBackoff) + } + + return fmt.Errorf("unreachable: exceeded max attempts") +} + +func moveRepoOnDisk(oldPath, newPath string, l *slog.Logger) error { + if _, err := os.Stat(newPath); err == nil { + l.Info("new path already exists, skipping rename", "newPath", newPath) + return nil + } + + if _, err := os.Stat(oldPath); errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("old path %s does not exist and new path %s not found", oldPath, newPath) + } + + if err := os.Rename(oldPath, newPath); err != nil { + return fmt.Errorf("renaming %s -> %s: %w", oldPath, newPath, err) + } + + return nil +} + +func rewriteRBACPolicies(e *rbac.Enforcer, ownerDid, repoName, repoDid string, l *slog.Logger) error { + oldResource := ownerDid + "/" + repoName + + policies, err := e.E.GetFilteredPolicy(1, rbac.ThisServer, oldResource) + if err != nil { + return fmt.Errorf("getting old policies: %w", err) + } + + var addPolicies [][]string + var removePolicies [][]string + for _, p := range policies { + removePolicies = append(removePolicies, p) + newPolicy := make([]string, len(p)) + copy(newPolicy, p) + newPolicy[2] = repoDid + addPolicies = append(addPolicies, newPolicy) + } + + if len(addPolicies) > 0 { + if _, addErr := e.E.AddPolicies(addPolicies); addErr != nil { + return fmt.Errorf("adding new policies: %w", addErr) + } + } + + if len(removePolicies) > 0 { + if _, rmErr := e.E.RemovePolicies(removePolicies); rmErr != nil { + return fmt.Errorf("removing old policies: %w", rmErr) + } + } + + l.Info("rewrote RBAC policies", "old", oldResource, "new", repoDid, "count", len(policies)) + return nil +} diff --git a/knotserver/server.go b/knotserver/server.go --- a/knotserver/server.go +++ b/knotserver/server.go @@ -89,6 +89,8 @@ } notifier := notifier.New() + go migrateReposOnStartup(ctx, c, db, e, ¬ifier, log.SubLogger(logger, "migrate")) + mux, err := Setup(ctx, c, db, e, jc, ¬ifier) if err != nil { return fmt.Errorf("failed to setup server: %w", err) diff --git a/spindle/stream.go b/spindle/stream.go --- a/spindle/stream.go +++ b/spindle/stream.go @@ -52,14 +52,14 @@ } } }() - defaultCursor := time.Now().UnixNano() + var cursor int64 cursorStr := r.URL.Query().Get("cursor") - cursor, err := strconv.ParseInt(cursorStr, 10, 64) - if err != nil { - l.Error("empty or invalid cursor", "invalidCursor", cursorStr, "default", defaultCursor) - } - if cursor == 0 { - cursor = defaultCursor + if cursorStr != "" { + cursor, err = strconv.ParseInt(cursorStr, 10, 64) + if err != nil { + l.Error("invalid cursor, starting from beginning", "invalidCursor", cursorStr) + cursor = 0 + } } // complete backfill first before going to live data @@ -239,9 +239,10 @@ return err } jsonMsg, err := json.Marshal(map[string]any{ - "rkey": event.Rkey, - "nsid": event.Nsid, - "event": eventJson, + "rkey": event.Rkey, + "nsid": event.Nsid, + "event": eventJson, + "created": event.Created, }) if err != nil { s.l.Error("failed to marshal record", "err", err) -- tangled.sh