package moves

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"strconv"
	"strings"

	"github.com/carlosframework/platform/internal/lease"
	"github.com/superfly/ltx"
)

// SealedObject is one pinned object version of a generation build's
// source: an LTX file named by its full key, the exact version id the seal
// (or a genselect plan, or a current listing) pinned, and its size. Task 5's
// Seal lists the staging lane's replica/ objects in this shape (plan:688);
// region HA's relocation feeds the same shape from its own pins (L16).
type SealedObject struct {
	Key       string `json:"key"`
	VersionID string `json:"version_id"`
	Size      int64  `json:"size"`
}

// ErrSealedReadOnly is every write a SealedClient refuses: the pinned set
// is immutable after the flip, and the build never deletes (§7.6).
var ErrSealedReadOnly = errors.New("moves: the sealed replica is read-only")

// SealedClient is a litestream.ReplicaClient over exactly one pinned set of
// object versions. Litestream's planner and restore read through it and
// never see a version the seal did not name, whether a newer write to a
// sealed key or an unsealed sibling under the same prefix. A key's position
// (level, min TXID, max TXID) is its last two path elements,
// `<level>/<min>-<max>.ltx`, so the client names no lane or generation.
type SealedClient struct {
	store  lease.VersionedStore
	bucket string
	byPos  map[sealedPos]SealedObject
	infos  []ltx.FileInfo
	logger *slog.Logger
}

type sealedPos struct {
	level    int
	min, max ltx.TXID
}

var _ interface {
	Type() string
	Init(ctx context.Context) error
	LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)
	OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error)
	WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error)
	DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error
	DeleteAll(ctx context.Context) error
	SetLogger(logger *slog.Logger)
} = (*SealedClient)(nil)

// NewSealedClient pins source. It refuses, before any read, an object with
// no usable version id (G27: "" is no version and "null" is what a
// versioning-suspended bucket answers, so neither pins anything), a key
// that is not an LTX file under a level directory, a size below an LTX
// header (litestream's restore refuses such a file, and a size of zero
// would make its premature-EOF detection blind), and two objects at one
// position, which would make the plan ambiguous.
func NewSealedClient(store lease.VersionedStore, bucket string, source []SealedObject) (*SealedClient, error) {
	if store == nil || bucket == "" {
		return nil, errors.New("moves: sealed client needs a store and a bucket")
	}
	c := &SealedClient{store: store, bucket: bucket, byPos: map[sealedPos]SealedObject{}, logger: slog.Default()}
	for _, o := range source {
		if !lease.UsableVersionID(o.VersionID) {
			return nil, fmt.Errorf("moves: sealed object %s has no usable version id %q", o.Key, o.VersionID)
		}
		pos, err := sealedPosition(o.Key)
		if err != nil {
			return nil, err
		}
		if o.Size < ltx.HeaderSize {
			return nil, fmt.Errorf("moves: sealed object %s has size %d, below an LTX header", o.Key, o.Size)
		}
		if prev, dup := c.byPos[pos]; dup {
			return nil, fmt.Errorf("moves: sealed objects %s and %s share one position", prev.Key, o.Key)
		}
		c.byPos[pos] = o
		c.infos = append(c.infos, ltx.FileInfo{Level: pos.level, MinTXID: pos.min, MaxTXID: pos.max, Size: o.Size})
	}
	return c, nil
}

// sealedPosition reads `<level>/<min>-<max>.ltx` off the end of key, with
// the level directory's four lowercase hex digits and the canonical file
// name, the same shape ReplicaSource's keep admits.
func sealedPosition(key string) (sealedPos, error) {
	parts := strings.Split(key, "/")
	if len(parts) < 2 {
		return sealedPos{}, fmt.Errorf("moves: sealed object %s is not <level>/<min>-<max>.ltx", key)
	}
	dir, name := parts[len(parts)-2], parts[len(parts)-1]
	level, err := strconv.ParseUint(dir, 16, 16)
	if err != nil || len(dir) != 4 || fmt.Sprintf("%04x", level) != dir {
		return sealedPos{}, fmt.Errorf("moves: sealed object %s has no level directory", key)
	}
	min, max, err := ltx.ParseFilename(name)
	if err != nil || min == 0 || min > max || ltx.FormatFilename(min, max) != name {
		return sealedPos{}, fmt.Errorf("moves: sealed object %s is not an LTX file name", key)
	}
	return sealedPos{level: int(level), min: min, max: max}, nil
}

// Type names the client in litestream's logs.
func (c *SealedClient) Type() string { return "sealed" }

// Init has nothing to open: the store is already bound.
func (c *SealedClient) Init(ctx context.Context) error { return nil }

// SetLogger stores the logger litestream hands over.
func (c *SealedClient) SetLogger(logger *slog.Logger) {
	if logger != nil {
		c.logger = logger
	}
}

// LTXFiles lists the pinned files at level from seek on, in TXID order.
// useMetadata is ignored: a pinned set has no timestamps to fetch, and the
// build restores by TXID, never by time.
func (c *SealedClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
	var out []*ltx.FileInfo
	for i := range c.infos {
		info := c.infos[i]
		if info.Level != level || info.MinTXID < seek {
			continue
		}
		out = append(out, &info) // a copy: the iterator sorts in place
	}
	return ltx.NewFileInfoSliceIterator(out), nil
}

// OpenLTXFile streams exactly the pinned version of the file at the
// position, or os.ErrNotExist when the seal names none. offset and size
// follow litestream's convention (size 0 reads to the end) and go through
// the store's VersionRangeOpener when it has one, otherwise the whole
// version is opened and the range is cut locally.
func (c *SealedClient) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	o, ok := c.byPos[sealedPos{level: level, min: minTXID, max: maxTXID}]
	if !ok {
		return nil, fmt.Errorf("moves: no sealed object at level %d %s: %w", level, ltx.FormatFilename(minTXID, maxTXID), os.ErrNotExist)
	}
	if offset < 0 || size < 0 {
		return nil, fmt.Errorf("moves: invalid range %d+%d for %s", offset, size, o.Key)
	}
	if offset == 0 && size == 0 {
		return c.store.OpenVersion(ctx, c.bucket, o.Key, o.VersionID)
	}
	if ranged, ok := c.store.(lease.VersionRangeOpener); ok {
		length := int64(-1)
		if size > 0 {
			length = size
		}
		return ranged.OpenVersionRange(ctx, c.bucket, o.Key, o.VersionID, offset, length)
	}
	rc, err := c.store.OpenVersion(ctx, c.bucket, o.Key, o.VersionID)
	if err != nil {
		return nil, err
	}
	if offset > 0 {
		if _, err := io.CopyN(io.Discard, rc, offset); err != nil {
			rc.Close()
			return nil, fmt.Errorf("moves: skip %d bytes of %s: %w", offset, o.Key, err)
		}
	}
	if size > 0 {
		return struct {
			io.Reader
			io.Closer
		}{io.LimitReader(rc, size), rc}, nil
	}
	return rc, nil
}

// WriteLTXFile is refused: the pinned set is immutable.
func (c *SealedClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
	return nil, ErrSealedReadOnly
}

// DeleteLTXFiles is refused: the build never deletes.
func (c *SealedClient) DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error {
	return ErrSealedReadOnly
}

// DeleteAll is refused: the build never deletes.
func (c *SealedClient) DeleteAll(ctx context.Context) error { return ErrSealedReadOnly }