package storage

import (
	"context"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"sync"
)

// ── PVC Store ──────────────────────────────

// PVCStore implements ObjectStore using the local filesystem.
// Used with Kubernetes PersistentVolumeClaims (CephFS, NFS, local PV)
// or plain Docker volume mounts.
type PVCStore struct {
	basePath string

	// Content-type cache: key → content type.
	// Populated on Put, used by Get. PVC doesn't store metadata
	// alongside files, so we keep MIME types in memory. On restart,
	// Get falls back to "application/octet-stream" — the DB has the
	// authoritative content_type for serving to clients.
	mu       sync.RWMutex
	mimeMap  map[string]string
}

// NewPVC creates a PVCStore rooted at basePath.
// Returns an error if the path does not exist or is not writable.
func NewPVC(basePath string) (*PVCStore, error) {
	// Ensure base path exists
	if err := os.MkdirAll(basePath, 0o755); err != nil {
		return nil, fmt.Errorf("storage/pvc: cannot create base path %q: %w", basePath, err)
	}

	// Create well-known subdirectories
	for _, sub := range []string{"files", "processing"} {
		if err := os.MkdirAll(filepath.Join(basePath, sub), 0o755); err != nil {
			return nil, fmt.Errorf("storage/pvc: cannot create %s dir: %w", sub, err)
		}
	}

	s := &PVCStore{
		basePath: basePath,
		mimeMap:  make(map[string]string),
	}

	// Validate writability
	if err := s.Healthy(context.Background()); err != nil {
		return nil, err
	}

	return s, nil
}

func (s *PVCStore) Backend() string { return "pvc" }

// Put writes data to the filesystem at the resolved key path.
func (s *PVCStore) Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) error {
	if err := validateKey(key); err != nil {
		return err
	}

	absPath := s.resolve(key)

	// Ensure parent directory exists
	if err := os.MkdirAll(filepath.Dir(absPath), 0o755); err != nil {
		return fmt.Errorf("storage/pvc: mkdir for %q: %w", key, err)
	}

	// Write to temp file first, then rename (atomic on same filesystem)
	tmp, err := os.CreateTemp(filepath.Dir(absPath), ".upload-*")
	if err != nil {
		return fmt.Errorf("storage/pvc: create temp for %q: %w", key, err)
	}
	tmpName := tmp.Name()
	defer func() {
		tmp.Close()
		// Clean up temp file on failure
		if err != nil {
			os.Remove(tmpName)
		}
	}()

	written, err := io.Copy(tmp, r)
	if err != nil {
		return fmt.Errorf("storage/pvc: write %q: %w", key, err)
	}
	if err = tmp.Close(); err != nil {
		return fmt.Errorf("storage/pvc: close %q: %w", key, err)
	}

	// Verify size if provided (0 means unknown)
	if size > 0 && written != size {
		os.Remove(tmpName)
		return fmt.Errorf("storage/pvc: size mismatch for %q: expected %d, got %d", key, size, written)
	}

	// Atomic rename
	if err = os.Rename(tmpName, absPath); err != nil {
		return fmt.Errorf("storage/pvc: rename %q: %w", key, err)
	}

	// Cache content type
	if contentType != "" {
		s.mu.Lock()
		s.mimeMap[key] = contentType
		s.mu.Unlock()
	}

	return nil
}

// Get returns a reader for the file at key.
func (s *PVCStore) Get(ctx context.Context, key string) (io.ReadCloser, int64, string, error) {
	if err := validateKey(key); err != nil {
		return nil, 0, "", err
	}

	absPath := s.resolve(key)

	info, err := os.Stat(absPath)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, 0, "", ErrNotFound
		}
		return nil, 0, "", fmt.Errorf("storage/pvc: stat %q: %w", key, err)
	}

	f, err := os.Open(absPath)
	if err != nil {
		return nil, 0, "", fmt.Errorf("storage/pvc: open %q: %w", key, err)
	}

	// Look up cached content type
	s.mu.RLock()
	ct := s.mimeMap[key]
	s.mu.RUnlock()
	if ct == "" {
		ct = "application/octet-stream"
	}

	return f, info.Size(), ct, nil
}

// Delete removes the file at key.
func (s *PVCStore) Delete(ctx context.Context, key string) error {
	if err := validateKey(key); err != nil {
		return err
	}

	absPath := s.resolve(key)

	err := os.Remove(absPath)
	if err != nil && !os.IsNotExist(err) {
		return fmt.Errorf("storage/pvc: delete %q: %w", key, err)
	}

	// Clean up cache
	s.mu.Lock()
	delete(s.mimeMap, key)
	s.mu.Unlock()

	return nil
}

// DeletePrefix removes all files and directories under the given prefix.
func (s *PVCStore) DeletePrefix(ctx context.Context, prefix string) error {
	if err := validateKey(prefix); err != nil {
		return err
	}

	absPath := s.resolve(prefix)

	// Check if the path exists
	if _, err := os.Stat(absPath); os.IsNotExist(err) {
		return nil // Nothing to delete
	}

	if err := os.RemoveAll(absPath); err != nil {
		return fmt.Errorf("storage/pvc: delete prefix %q: %w", prefix, err)
	}

	// Clean up cache entries with this prefix
	s.mu.Lock()
	for k := range s.mimeMap {
		if strings.HasPrefix(k, prefix) {
			delete(s.mimeMap, k)
		}
	}
	s.mu.Unlock()

	return nil
}

// Exists checks if a file exists at key.
func (s *PVCStore) Exists(ctx context.Context, key string) (bool, error) {
	if err := validateKey(key); err != nil {
		return false, err
	}

	_, err := os.Stat(s.resolve(key))
	if err == nil {
		return true, nil
	}
	if os.IsNotExist(err) {
		return false, nil
	}
	return false, fmt.Errorf("storage/pvc: exists %q: %w", key, err)
}

// Healthy verifies the storage path is writable.
func (s *PVCStore) Healthy(ctx context.Context) error {
	probe := filepath.Join(s.basePath, ".health-probe")
	if err := os.WriteFile(probe, []byte("ok"), 0o644); err != nil {
		return fmt.Errorf("storage/pvc: not writable: %w", err)
	}
	os.Remove(probe)
	return nil
}

// Stats returns file count and total size under the base path.
func (s *PVCStore) Stats(ctx context.Context) (*StorageStats, error) {
	stats := &StorageStats{
		Backend:    "pvc",
		Path:       s.basePath,
		Configured: true,
	}

	// Check health
	if err := s.Healthy(ctx); err != nil {
		stats.Healthy = false
		return stats, nil
	}
	stats.Healthy = true

	// Walk files directory for counts
	fileDir := filepath.Join(s.basePath, "files")
	err := filepath.Walk(fileDir, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return nil // skip errors (e.g. permission denied)
		}
		if !info.IsDir() {
			stats.TotalFiles++
			stats.TotalBytes += info.Size()
		}
		return nil
	})
	if err != nil && !os.IsNotExist(err) {
		return stats, fmt.Errorf("storage/pvc: stats walk: %w", err)
	}

	return stats, nil
}

// ── Helpers ────────────────────────────────

// resolve converts a storage key to an absolute filesystem path.
func (s *PVCStore) resolve(key string) string {
	return filepath.Join(s.basePath, filepath.FromSlash(key))
}

// validateKey ensures the key is safe (no path traversal, no absolute paths).
func validateKey(key string) error {
	if key == "" {
		return fmt.Errorf("storage: empty key")
	}
	if strings.HasPrefix(key, "/") || strings.HasPrefix(key, "\\") {
		return fmt.Errorf("storage: absolute key not allowed: %q", key)
	}
	cleaned := filepath.Clean(key)
	if strings.Contains(cleaned, "..") {
		return fmt.Errorf("storage: path traversal not allowed: %q", key)
	}
	return nil
}
