mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-08-29 04:06:58 +00:00
clean up temporary migration code
This commit is contained in:
Evan Jarrett
@@ -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
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user