package worker

import (
	"context"
	"database/sql"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"hitkeep/internal/database"
)

// BackupWorker periodically exports all DuckDB databases to Parquet snapshots.
type BackupWorker struct {
	tenantMgr      *database.TenantStoreManager
	dataPath       string
	backupPath     string
	intervalMin    int
	retentionCount int
	s3Config       *S3Config
	status         *database.BackupStatusTracker
}

// NewBackupWorker creates a BackupWorker. If backupPath is empty, Start is a no-op.
func NewBackupWorker(
	tenantMgr *database.TenantStoreManager,
	dataPath string,
	backupPath string,
	intervalMin int,
	retentionCount int,
	s3Config *S3Config,
	status *database.BackupStatusTracker,
) *BackupWorker {
	return &BackupWorker{
		tenantMgr:      tenantMgr,
		dataPath:       dataPath,
		backupPath:     strings.TrimSpace(backupPath),
		intervalMin:    intervalMin,
		retentionCount: retentionCount,
		s3Config:       s3Config,
		status:         status,
	}
}

// Start runs the backup worker on a ticker loop. It returns immediately if
// backupPath is empty (backups disabled).
func (w *BackupWorker) Start(ctx context.Context) {
	if w.backupPath == "" {
		return
	}

	if IsS3ArchivePath(w.backupPath) {
		slog.Info("S3 backup enabled", "path", w.backupPath, "interval_min", w.intervalMin, "retention", w.retentionCount)
	} else {
		slog.Info("Local backup enabled", "path", w.backupPath, "interval_min", w.intervalMin, "retention", w.retentionCount)
	}
	w.setNextBackup(time.Now().UTC().Add(30 * time.Second))

	// Initial run after a short delay to let DB settle.
	go func() {
		time.Sleep(30 * time.Second)
		if err := w.Run(ctx); err != nil {
			slog.Error("Initial backup run failed", "error", err)
		}
	}()

	interval := time.Duration(w.intervalMin) * time.Minute
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := w.Run(ctx); err != nil {
				slog.Error("Backup worker failed", "error", err)
			}
		}
	}
}

// Run executes a single backup cycle: export shared DB + all tenant DBs,
// then prune old snapshots beyond the retention count.
func (w *BackupWorker) Run(ctx context.Context) (err error) {
	defer func() {
		finishedAt := time.Now().UTC()
		if err != nil {
			w.recordFailure(finishedAt, err)
			return
		}
		w.recordSuccess(finishedAt)
	}()

	timestamp := time.Now().UTC().Format("2006-01-02T150405Z")
	slog.Info("Starting database backup", "timestamp", timestamp)

	isS3 := IsS3ArchivePath(w.backupPath)

	// Backup shared DB.
	sharedDest := joinArchivePath(w.backupPath, "shared", timestamp)
	if err := w.exportDatabase(ctx, w.tenantMgr.Shared().DB(), sharedDest, isS3); err != nil {
		return fmt.Errorf("backup shared database: %w", err)
	}
	slog.Info("Shared database backed up", "dest", sharedDest)

	// Discover non-default tenants (gracefully skip if tenants table or
	// is_default column doesn't exist yet — pre-migration state).
	tenantIDs, err := w.tenantMgr.Shared().ListNonDefaultTenantIDs(ctx)
	if err != nil {
		if isMissingRelationError(err, "tenants") || isBinderError(err) {
			slog.Debug("Tenants schema not ready, skipping tenant backups", "error", err)
			tenantIDs = nil
		} else {
			slog.Error("Failed to list tenant IDs for backup", "error", err)
			tenantIDs = nil
		}
	}

	// Backup each non-default tenant DB.
	for _, tenantID := range tenantIDs {
		tenantStore, err := w.tenantMgr.ForTenant(ctx, tenantID)
		if err != nil {
			slog.Error("Failed to open tenant store for backup", "tenant_id", tenantID, "error", err)
			continue
		}

		tenantDest := joinArchivePath(w.backupPath, "tenants", tenantID.String(), timestamp)
		if err := w.exportDatabase(ctx, tenantStore.DB(), tenantDest, isS3); err != nil {
			slog.Error("Failed to backup tenant database", "tenant_id", tenantID, "error", err)
			continue
		}
		slog.Info("Tenant database backed up", "tenant_id", tenantID, "dest", tenantDest)
	}

	// Prune old snapshots.
	if !isS3 {
		w.pruneLocalSnapshots(filepath.Join(w.backupPath, "shared"))
		for _, tenantID := range tenantIDs {
			w.pruneLocalSnapshots(filepath.Join(w.backupPath, "tenants", tenantID.String()))
		}
	} else {
		slog.Debug("S3 backup pruning: configure S3 lifecycle policies to manage snapshot retention")
	}

	slog.Info("Database backup completed", "timestamp", timestamp)
	return nil
}

func (w *BackupWorker) recordSuccess(at time.Time) {
	if w.status == nil {
		return
	}
	w.status.SetLastBackup(at)
	if next, ok := w.nextBackupTime(at); ok {
		w.status.SetNextBackup(next)
	}
}

func (w *BackupWorker) recordFailure(at time.Time, err error) {
	if w.status == nil {
		return
	}
	w.status.SetFailed(at, err.Error())
	if next, ok := w.nextBackupTime(at); ok {
		w.status.SetNextBackup(next)
	}
}

func (w *BackupWorker) setNextBackup(at time.Time) {
	if w.status == nil {
		return
	}
	w.status.SetNextBackup(at)
}

func (w *BackupWorker) nextBackupTime(after time.Time) (time.Time, bool) {
	if w.intervalMin <= 0 {
		return time.Time{}, false
	}
	return after.Add(time.Duration(w.intervalMin) * time.Minute), true
}

// exportDatabase checkpoints and exports a DuckDB database to the given destination.
func (w *BackupWorker) exportDatabase(ctx context.Context, db *sql.DB, dest string, isS3 bool) error {
	// Ensure local directory exists.
	if !isS3 {
		if err := os.MkdirAll(dest, 0755); err != nil {
			return fmt.Errorf("create backup directory %s: %w", dest, err)
		}
	}

	safeDest := strings.ReplaceAll(dest, "'", "''")
	query := fmt.Sprintf("EXPORT DATABASE '%s' (FORMAT PARQUET);", safeDest)
	return database.WithDuckDBSession(ctx, db, database.DuckDBSessionOptions{
		S3: s3ConfigForSession(isS3, w.s3Config),
	}, func(conn *sql.Conn) error {
		if _, err := conn.ExecContext(ctx, "CHECKPOINT;"); err != nil {
			slog.Warn("Checkpoint before export failed (continuing)", "error", err)
		}
		if _, err := conn.ExecContext(ctx, query); err != nil {
			return fmt.Errorf("export database to %s: %w", dest, err)
		}
		return nil
	})
}

// pruneLocalSnapshots removes the oldest snapshot directories in dir,
// keeping at most retentionCount. Snapshot dirs are ISO timestamp names
// that sort lexicographically.
func (w *BackupWorker) pruneLocalSnapshots(dir string) {
	entries, err := os.ReadDir(dir)
	if err != nil {
		if !os.IsNotExist(err) {
			slog.Warn("Could not read backup directory for pruning", "dir", dir, "error", err)
		}
		return
	}

	// Collect only directories (snapshot timestamps).
	dirs := make([]string, 0, len(entries))
	for _, e := range entries {
		if e.IsDir() {
			dirs = append(dirs, e.Name())
		}
	}

	if len(dirs) <= w.retentionCount {
		return
	}

	sort.Strings(dirs)

	toRemove := dirs[:len(dirs)-w.retentionCount]
	for _, name := range toRemove {
		path := filepath.Join(dir, name)
		if err := os.RemoveAll(path); err != nil {
			slog.Error("Failed to prune old backup snapshot", "path", path, "error", err)
		} else {
			slog.Info("Pruned old backup snapshot", "path", path)
		}
	}
}
