diff --git a/pkg/appview/db/stats_migration.go b/pkg/appview/db/stats_migration.go deleted file mode 100644 index 8c6eb8b..0000000 --- a/pkg/appview/db/stats_migration.go +++ /dev/null @@ -1,215 +0,0 @@ -package db - -import ( - "bytes" - "context" - "database/sql" - "encoding/json" - "fmt" - "io" - "log/slog" - "net/http" - "time" - - "atcr.io/pkg/atproto" -) - -// MigrateStatsToHolds migrates existing repository_stats data to hold services. -// This is a one-time migration that runs on startup. -// -// The migration: -// 1. Checks if migration has already completed -// 2. Reads all repository_stats entries -// 3. For each entry, looks up the hold DID from manifests table -// 4. Calls the hold's setStats endpoint (no auth required - temporary migration endpoint) -// 5. Marks migration complete after all entries are processed -// -// If a hold is offline, the migration logs a warning and continues. -// The hold will receive real-time stats updates via Jetstream once online. -func MigrateStatsToHolds(ctx context.Context, db *sql.DB) error { - // Check if migration already done - var migrationDone bool - err := db.QueryRowContext(ctx, ` - SELECT EXISTS( - SELECT 1 FROM schema_migrations WHERE version = 1000 - ) - `).Scan(&migrationDone) - - // Table might not exist yet on fresh install - if err == sql.ErrNoRows { - migrationDone = false - } else if err != nil { - // Check if it's a "no such table" error (fresh install) - if err.Error() != "no such table: schema_migrations" { - return fmt.Errorf("failed to check migration status: %w", err) - } - migrationDone = false - } - - if migrationDone { - slog.Debug("Stats migration already complete, skipping", "component", "migration") - return nil - } - - slog.Info("Starting stats migration to holds", "component", "migration") - - // Get all repository_stats entries - rows, err := db.QueryContext(ctx, ` - SELECT did, repository, pull_count, last_pull, push_count, last_push - FROM repository_stats - WHERE pull_count > 0 OR push_count > 0 - `) - if err != nil { - // Table might not exist on fresh install - if err.Error() == "no such table: repository_stats" { - slog.Info("No repository_stats table found, skipping migration", "component", "migration") - return markMigrationComplete(db) - } - return fmt.Errorf("failed to query repository_stats: %w", err) - } - defer rows.Close() - - var stats []struct { - DID string - Repository string - PullCount int64 - LastPull sql.NullString - PushCount int64 - LastPush sql.NullString - } - - for rows.Next() { - var stat struct { - DID string - Repository string - PullCount int64 - LastPull sql.NullString - PushCount int64 - LastPush sql.NullString - } - if err := rows.Scan(&stat.DID, &stat.Repository, &stat.PullCount, &stat.LastPull, &stat.PushCount, &stat.LastPush); err != nil { - return fmt.Errorf("failed to scan stat: %w", err) - } - stats = append(stats, stat) - } - - if len(stats) == 0 { - slog.Info("No stats to migrate", "component", "migration") - return markMigrationComplete(db) - } - - slog.Info("Found stats entries to migrate", "component", "migration", "count", len(stats)) - - // Process each stat - successCount := 0 - skipCount := 0 - errorCount := 0 - - for _, stat := range stats { - // Look up hold DID from manifests table - holdDID, err := GetLatestHoldDIDForRepo(db, stat.DID, stat.Repository) - if err != nil || holdDID == "" { - slog.Debug("No hold DID found for repo, skipping", "component", "migration", - "did", stat.DID, "repository", stat.Repository) - skipCount++ - continue - } - - // Resolve hold DID to HTTP URL - holdURL := atproto.ResolveHoldURL(holdDID) - if holdURL == "" { - slog.Warn("Failed to resolve hold DID, skipping", "component", "migration", - "hold_did", holdDID) - errorCount++ - continue - } - - // Call hold's setStats endpoint (no auth required for migration) - err = callSetStats(ctx, holdURL, stat.DID, stat.Repository, - stat.PullCount, stat.PushCount, stat.LastPull.String, stat.LastPush.String) - if err != nil { - slog.Warn("Failed to migrate stats to hold, continuing", "component", "migration", - "did", stat.DID, "repository", stat.Repository, "hold", holdDID, "error", err) - errorCount++ - continue - } - - successCount++ - slog.Debug("Migrated stats", "component", "migration", - "did", stat.DID, "repository", stat.Repository, "hold", holdDID, - "pull_count", stat.PullCount, "push_count", stat.PushCount) - } - - slog.Info("Stats migration completed", "component", "migration", - "success", successCount, "skipped", skipCount, "errors", errorCount, "total", len(stats)) - - // Only mark complete if there were no errors - // Skipped repos (no hold DID) will never migrate - that's fine - // Errors are transient failures that should be retried - if errorCount == 0 { - return markMigrationComplete(db) - } - - slog.Warn("Stats migration had errors, will retry on next startup", "component", "migration", - "errors", errorCount) - return nil -} - -// markMigrationComplete records that the stats migration has been done -func markMigrationComplete(db *sql.DB) error { - _, err := db.Exec(` - INSERT INTO schema_migrations (version, applied_at) - VALUES (1000, datetime('now')) - ON CONFLICT(version) DO NOTHING - `) - if err != nil { - return fmt.Errorf("failed to mark migration complete: %w", err) - } - return nil -} - -// callSetStats calls the hold's io.atcr.hold.setStats endpoint -// No authentication required - this is a temporary migration endpoint -func callSetStats(ctx context.Context, holdURL, ownerDID, repository string, pullCount, pushCount int64, lastPull, lastPush string) error { - // Build request - reqBody := map[string]any{ - "ownerDid": ownerDID, - "repository": repository, - "pullCount": pullCount, - "pushCount": pushCount, - } - if lastPull != "" { - reqBody["lastPull"] = lastPull - } - if lastPush != "" { - reqBody["lastPush"] = lastPush - } - - body, err := json.Marshal(reqBody) - if err != nil { - return fmt.Errorf("failed to marshal request: %w", err) - } - - // Create HTTP request - req, err := http.NewRequestWithContext(ctx, "POST", holdURL+atproto.HoldSetStats, bytes.NewReader(body)) - if err != nil { - return fmt.Errorf("failed to create request: %w", err) - } - - req.Header.Set("Content-Type", "application/json") - - // Send request with timeout - client := &http.Client{Timeout: 10 * time.Second} - resp, err := client.Do(req) - if err != nil { - return fmt.Errorf("request failed: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - body, _ := io.ReadAll(resp.Body) - return fmt.Errorf("setStats failed: status %d, body: %s", resp.StatusCode, body) - } - - return nil -} diff --git a/pkg/hold/oci/xrpc.go b/pkg/hold/oci/xrpc.go index e3ccf9c..7cb0042 100644 --- a/pkg/hold/oci/xrpc.go +++ b/pkg/hold/oci/xrpc.go @@ -40,10 +40,6 @@ func NewXRPCHandler(holdPDS *pds.HoldPDS, s3Service s3.S3Service, driver storage // RegisterHandlers registers all OCI XRPC endpoints with the chi router func (h *XRPCHandler) RegisterHandlers(r chi.Router) { - // Temporary migration endpoint - no auth required - // TODO: Remove after stats migration is complete - r.Post(atproto.HoldSetStats, h.HandleSetStats) - // All multipart upload endpoints require blob:write permission r.Group(func(r chi.Router) { r.Use(h.requireBlobWriteAccess) @@ -385,48 +381,6 @@ func (h *XRPCHandler) HandleNotifyManifest(w http.ResponseWriter, r *http.Reques RespondJSON(w, http.StatusOK, resp) } -// HandleSetStats sets absolute stats values for a repository (used by migration) -// This is a temporary migration-only endpoint that allows AppView to sync existing stats to holds. -// No authentication required - this endpoint will be removed after migration is complete. -// TODO: Remove this endpoint after stats migration is complete -func (h *XRPCHandler) HandleSetStats(w http.ResponseWriter, r *http.Request) { - ctx := r.Context() - - // Parse request - var req struct { - OwnerDID string `json:"ownerDid"` - Repository string `json:"repository"` - PullCount int64 `json:"pullCount"` - PushCount int64 `json:"pushCount"` - LastPull string `json:"lastPull,omitempty"` - LastPush string `json:"lastPush,omitempty"` - } - - if err := DecodeJSON(r, &req); err != nil { - RespondError(w, http.StatusBadRequest, err.Error()) - return - } - - // Validate required fields - if req.OwnerDID == "" || req.Repository == "" { - RespondError(w, http.StatusBadRequest, "ownerDid and repository are required") - return - } - - // Set stats using the SetStats method - if err := h.pds.SetStats(ctx, req.OwnerDID, req.Repository, req.PullCount, req.PushCount, req.LastPull, req.LastPush); err != nil { - slog.Error("Failed to set stats", "error", err) - RespondError(w, http.StatusInternalServerError, fmt.Sprintf("failed to set stats: %v", err)) - return - } - - slog.Info("Stats set via migration", "owner_did", req.OwnerDID, "repository", req.Repository, "pull_count", req.PullCount, "push_count", req.PushCount) - - RespondJSON(w, http.StatusOK, map[string]any{ - "success": true, - }) -} - // requireBlobWriteAccess middleware - validates DPoP + OAuth and checks for blob:write permission func (h *XRPCHandler) requireBlobWriteAccess(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {