package lease

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"path"
	"strings"
	"time"

	"github.com/benbjohnson/litestream"
	"github.com/superfly/ltx"
)

var ErrGenerationRetention = errors.New("lease: generation retention requires exact versions through park")

var ErrGenerationBinding = errors.New("lease: invalid or changed generation binding")

const generationRequestLimit = 120 * time.Second

// BindGeneration fixes all replica data requests below the held lease's
// location. The authority callback is captured once and checked per request.
func (h *Held) BindGeneration(prefix string, check func(context.Context) error) error {
	if h == nil || h.leaser == nil || check == nil {
		return ErrGenerationBinding
	}
	base := strings.TrimSuffix(path.Clean(h.leaser.Path), "/")
	clean := path.Clean(prefix)
	rel, ok := strings.CutPrefix(clean, base+"/")
	if (prefix != clean && prefix != clean+"/") || strings.Contains(prefix, "\\") || base == "." || base == "" || !ok || strings.Contains(rel, "/") || !strings.HasPrefix(rel, "g-") || len(rel) == 2 {
		return ErrGenerationBinding
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.busy || h.spent || h.generationPrefix != "" {
		return ErrGenerationBinding
	}
	client := h.replicaClient
	if client == nil {
		var err error
		client, err = newReplicaClient(h.leaser.Bucket, clean, h.region)
		if err != nil {
			return err
		}
	}
	h.replicaClient = &generationClient{client: client, check: check}
	h.generationPrefix = clean
	return nil
}

// BindRoot is BindGeneration for the root layout (hostclaim.RootGeneration):
// the replica data sits directly below the held lease's location, so no
// prefix changes, but every replica request still first passes check, the
// authority read before each one (moves §5.3 step 5, ruling G3; Task 5 fix
// round 1, ruling 2). Without it a root-generation checkpoint would check
// its authority once and then run every request unchecked. As immutable as
// BindGeneration, and exclusive with it. Retention stays off under it, as
// under a bound generation: the checking client refuses DeleteLTXFiles.
func (h *Held) BindRoot(check func(context.Context) error) error {
	if h == nil || h.leaser == nil || check == nil {
		return ErrGenerationBinding
	}
	base := strings.TrimSuffix(path.Clean(h.leaser.Path), "/")
	if base == "." || base == "" {
		return ErrGenerationBinding
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.busy || h.spent || h.generationPrefix != "" {
		return ErrGenerationBinding
	}
	client := h.replicaClient
	if client == nil {
		var err error
		client, err = newReplicaClient(h.leaser.Bucket, base, h.region)
		if err != nil {
			return err
		}
	}
	h.replicaClient = &generationClient{client: client, check: check}
	// The lease's own path marks the binding, so neither BindGeneration nor
	// a second BindRoot can change the client under a running checkpoint,
	// and newStore keeps retention off.
	h.generationPrefix = base
	return nil
}

type generationClient struct {
	client litestream.ReplicaClient
	check  func(context.Context) error
}

func (c *generationClient) Type() string             { return c.client.Type() }
func (c *generationClient) SetLogger(l *slog.Logger) { c.client.SetLogger(l) }
func generationLevel(level int) error {
	if level < 0 || level > litestream.SnapshotLevel {
		return ErrGenerationBinding
	}
	return nil
}
func generationCoords(level int, min, max ltx.TXID) error {
	if generationLevel(level) != nil || min == 0 || max == 0 || min > max {
		return ErrGenerationBinding
	}
	return nil
}
func (c *generationClient) bounded(ctx context.Context) (context.Context, context.CancelFunc, error) {
	if ctx == nil {
		return nil, nil, ErrGenerationBinding
	}
	ctx, cancel := context.WithTimeout(ctx, generationRequestLimit)
	if err := c.check(ctx); err != nil {
		cancel()
		return nil, nil, err
	}
	return ctx, cancel, nil
}
func (c *generationClient) Init(ctx context.Context) error {
	ctx, cancel, err := c.bounded(ctx)
	if err != nil {
		return err
	}
	defer cancel()
	return c.client.Init(ctx)
}
func (c *generationClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, metadata bool) (ltx.FileIterator, error) {
	if err := generationLevel(level); err != nil {
		return nil, err
	}
	ctx, cancel, err := c.bounded(ctx)
	if err != nil {
		return nil, err
	}
	it, err := c.client.LTXFiles(ctx, level, seek, metadata)
	if err != nil {
		cancel()
		return nil, err
	}
	return &generationIterator{FileIterator: it, cancel: cancel}, nil
}
func (c *generationClient) OpenLTXFile(ctx context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	if err := generationCoords(level, min, max); err != nil || offset < 0 || size < 0 {
		return nil, ErrGenerationBinding
	}
	ctx, cancel, err := c.bounded(ctx)
	if err != nil {
		return nil, err
	}
	rc, err := c.client.OpenLTXFile(ctx, level, min, max, offset, size)
	if err != nil {
		cancel()
		return nil, err
	}
	return &generationReader{ReadCloser: rc, cancel: cancel}, nil
}
func (c *generationClient) WriteLTXFile(ctx context.Context, level int, min, max ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
	if pipe, ok := r.(*io.PipeReader); ok {
		defer pipe.Close()
	}
	if err := generationCoords(level, min, max); err != nil || r == nil {
		return nil, ErrGenerationBinding
	}
	ctx, cancel, err := c.bounded(ctx)
	if err != nil {
		return nil, err
	}
	defer cancel()
	return c.client.WriteLTXFile(ctx, level, min, max, r)
}
func (c *generationClient) DeleteLTXFiles(ctx context.Context, files []*ltx.FileInfo) error {
	if len(files) == 0 {
		return ErrGenerationBinding
	}
	for _, f := range files {
		if f == nil || generationCoords(f.Level, f.MinTXID, f.MaxTXID) != nil {
			return ErrGenerationBinding
		}
	}
	// FileInfo has no VersionID. Only the park engine may retire exact versions.
	return ErrGenerationRetention
}
func (c *generationClient) DeleteAll(context.Context) error {
	return fmt.Errorf("%w: DeleteAll refused", ErrGenerationBinding)
}

type generationReader struct {
	io.ReadCloser
	cancel context.CancelFunc
}

func (r *generationReader) Close() error { err := r.ReadCloser.Close(); r.cancel(); return err }

type generationIterator struct {
	ltx.FileIterator
	cancel context.CancelFunc
}

func (i *generationIterator) Close() error { err := i.FileIterator.Close(); i.cancel(); return err }