at-container-registry/pkg/appview/db/stats_migration.go

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
}