// Package spool persists gzip-compressed batches that could not be delivered, under data_dir/spool/.
//
// Files are named <unixnano>.json.gz so lexical order is chronological; the oldest is re-sent first. A size cap
// (default 200 MiB) drops the oldest files when exceeded. Files hold the exact compressed bytes: the signature
// covers those bytes, so each retry only needs a fresh timestamp.
package spool

import (
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

// DefaultCap is the on-disk size limit.
const DefaultCap int64 = 200 << 20

// Spool is a directory of pending batches. Safe for concurrent use.
type Spool struct {
	dir string
	cap int64
	mu  sync.Mutex
}

// Open creates dir (0700) if needed. capBytes ≤ 0 selects DefaultCap.
func Open(dir string, capBytes int64) (*Spool, error) {
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return nil, fmt.Errorf("spool: %w", err)
	}
	if capBytes <= 0 {
		capBytes = DefaultCap
	}
	return &Spool{dir: dir, cap: capBytes}, nil
}

// Dir returns the spool directory.
func (s *Spool) Dir() string { return s.dir }

// Write stores a compressed batch and enforces the cap. Returns the file name.
func (s *Spool) Write(gz []byte) (string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	name := strconv.FormatInt(time.Now().UnixNano(), 10) + ".json.gz"
	final := filepath.Join(s.dir, name)
	tmp := final + ".tmp"
	if err := os.WriteFile(tmp, gz, 0o600); err != nil {
		return "", fmt.Errorf("spool: %w", err)
	}
	if err := os.Rename(tmp, final); err != nil {
		os.Remove(tmp)
		return "", fmt.Errorf("spool: %w", err)
	}
	s.enforceCapLocked()
	return name, nil
}

// entry is one spooled file.
type entry struct {
	name string
	size int64
}

func (s *Spool) listLocked() []entry {
	des, err := os.ReadDir(s.dir)
	if err != nil {
		return nil
	}
	out := make([]entry, 0, len(des))
	for _, de := range des {
		if de.IsDir() || !strings.HasSuffix(de.Name(), ".json.gz") {
			continue
		}
		info, err := de.Info()
		if err != nil {
			continue
		}
		out = append(out, entry{name: de.Name(), size: info.Size()})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].name < out[j].name })
	return out
}

func (s *Spool) enforceCapLocked() {
	entries := s.listLocked()
	var total int64
	for _, e := range entries {
		total += e.size
	}
	for i := 0; total > s.cap && i < len(entries); i++ {
		if err := os.Remove(filepath.Join(s.dir, entries[i].name)); err == nil {
			total -= entries[i].size
		}
	}
	// Stale temp files from a crash mid-write.
	des, _ := os.ReadDir(s.dir)
	for _, de := range des {
		if strings.HasSuffix(de.Name(), ".tmp") {
			os.Remove(filepath.Join(s.dir, de.Name()))
		}
	}
}

// ErrEmpty is returned by Oldest when nothing is spooled.
var ErrEmpty = errors.New("spool: empty")

// Oldest returns the oldest pending batch (name and compressed bytes).
func (s *Spool) Oldest() (string, []byte, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, e := range s.listLocked() {
		data, err := os.ReadFile(filepath.Join(s.dir, e.name))
		if err != nil {
			os.Remove(filepath.Join(s.dir, e.name)) // unreadable → discard
			continue
		}
		return e.name, data, nil
	}
	return "", nil, ErrEmpty
}

// Remove deletes a delivered (or rejected) batch.
func (s *Spool) Remove(name string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	err := os.Remove(filepath.Join(s.dir, filepath.Base(name)))
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	return err
}

// Stats returns (file count, total bytes).
func (s *Spool) Stats() (int, int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	entries := s.listLocked()
	var total int64
	for _, e := range entries {
		total += e.size
	}
	return len(entries), total
}

// Bytes returns the total spooled size.
func (s *Spool) Bytes() int64 {
	_, b := s.Stats()
	return b
}