knotserver: add repo DID migration on startup

Signed-off-by: Lewis <[e-mail]>

appview/db/db.go

···
 		"_journal_mode=WAL",
 		"_synchronous=NORMAL",
 		"_auto_vacuum=incremental",
+		"_busy_timeout=5000",
 	}
 
 	logger := log.FromContext(ctx)

eventconsumer/consumer.go

···
 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"`
 }
 
···
 				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)

knotserver/db/db.go

···
 		"_journal_mode=WAL",
 		"_synchronous=NORMAL",
 		"_auto_vacuum=incremental",
+		"_busy_timeout=5000",
 	}
 
 	logger := log.FromContext(ctx)

knotserver/events.go

···
 		}
 	}()
 
-	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)

knotserver/migrate.go

···
+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
+}

knotserver/server.go

···
 
 	notifier := notifier.New()
 
+	go migrateReposOnStartup(ctx, c, db, e, &notifier, log.SubLogger(logger, "migrate"))
+
 	mux, err := Setup(ctx, c, db, e, jc, &notifier)
 	if err != nil {
 		return fmt.Errorf("failed to setup server: %w", err)

spindle/db/db.go

···
 		"_journal_mode=WAL",
 		"_synchronous=NORMAL",
 		"_auto_vacuum=incremental",
+		"_busy_timeout=5000",
 	}
 
 	db, err := sql.Open("sqlite3", dbPath+"?"+strings.Join(opts, "&"))

spindle/stream.go

···
 		}
 	}()
 
-	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
···
 		}
 
 		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)