package dbdoor

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"sort"
	"sync"
	"time"

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

// ReplicaReader is the read half of a litestream replica client: the two
// methods a door or a planner may hold. A *lss3.ReplicaClient from
// lease.ReadReplicaClient is one; so is a file replica client.
type ReplicaReader interface {
	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)
}

// ListLevel lists one level of rd in the door's wire shape, ordered by
// MinTXID, with an empty level as an empty list, and with S3's LastModified
// as CreatedAt (useMetadata false). The console's door and the box's planner
// both list through it, so a forwarded --at stops at the transaction an
// in-region one does (spec §6, T6).
func ListLevel(ctx context.Context, rd ReplicaReader, level int) (ListResponse, error) {
	itr, err := rd.LTXFiles(ctx, level, 0, false)
	if err != nil {
		return ListResponse{}, err
	}
	out := ListResponse{Files: []LTXFile{}}
	for itr.Next() {
		f := itr.Item()
		out.Files = append(out.Files, LTXFile{
			Level: level, MinTXID: uint64(f.MinTXID), MaxTXID: uint64(f.MaxTXID),
			Size: f.Size, CreatedAt: f.CreatedAt,
		})
	}
	err = itr.Err()
	if cerr := itr.Close(); err == nil {
		err = cerr
	}
	if err != nil {
		return ListResponse{}, err
	}
	sort.Slice(out.Files, func(i, j int) bool { return out.Files[i].MinTXID < out.Files[j].MinTXID })
	return out, nil
}

// listing is a read-only litestream.ReplicaClient over rd that lists through
// ListLevel. For an as-of plan it also does what the CLI's recorder does
// over the door's listings: refuse a listing with no times (ErrNoTimes) and
// remember the earliest file a restore could start from.
type listing struct {
	rd     ReplicaReader
	asOf   bool
	mu     sync.Mutex
	oldest time.Time
}

func (l *listing) Type() string               { return "carlos-db-listing" }
func (l *listing) Init(context.Context) error { return nil }
func (l *listing) SetLogger(*slog.Logger)     {}

func (l *listing) LTXFiles(ctx context.Context, level int, seek ltx.TXID, _ bool) (ltx.FileIterator, error) {
	out, err := ListLevel(ctx, l.rd, level)
	if err != nil {
		return nil, err
	}
	if l.asOf {
		l.mu.Lock()
		defer l.mu.Unlock()
		for _, f := range out.Files {
			if f.CreatedAt.IsZero() {
				return nil, ErrNoTimes
			}
			if level == litestream.SnapshotLevel || f.MinTXID == 1 {
				if l.oldest.IsZero() || f.CreatedAt.Before(l.oldest) {
					l.oldest = f.CreatedAt
				}
			}
		}
	}
	return fileIterator(out.Files, level, seek), nil
}

func (l *listing) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	return l.rd.OpenLTXFile(ctx, level, minTXID, maxTXID, offset, size)
}

func (l *listing) WriteLTXFile(context.Context, int, ltx.TXID, ltx.TXID, io.Reader) (*ltx.FileInfo, error) {
	return nil, errReadOnly
}

func (l *listing) DeleteLTXFiles(context.Context, []*ltx.FileInfo) error { return errReadOnly }

func (l *listing) DeleteAll(context.Context) error { return errReadOnly }

var _ litestream.ReplicaClient = (*listing)(nil)

// PlanFrom plans the restore RestoreAsOf would make through the door, from
// rd directly: the newest copy when at is zero, else the copy holding every
// file listed before at rounded down to its second. Its errors are
// RestoreAsOf's: ErrNothingReplicated, a *BeforeHistoryError naming the
// oldest point kept, ErrNoTimes.
func PlanFrom(ctx context.Context, rd ReplicaReader, at time.Time) (RestorePlan, error) {
	l := &listing{rd: rd, asOf: !at.IsZero()}
	var ts time.Time
	if !at.IsZero() {
		if at.Truncate(time.Second).Before(EarliestAsOf) {
			at = EarliestAsOf
		}
		ts = at.Truncate(time.Second)
	}
	infos, err := litestream.CalcRestorePlan(ctx, l, 0, ts, slog.Default())
	switch {
	case errors.Is(err, ErrNoTimes):
		return RestorePlan{}, ErrNoTimes
	case errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots):
		l.mu.Lock()
		oldest := l.oldest
		l.mu.Unlock()
		if !ts.IsZero() && !oldest.IsZero() {
			if !oldest.Before(ts) {
				return RestorePlan{}, &BeforeHistoryError{Oldest: oldest.Truncate(time.Second).Add(time.Second)}
			}
			return RestorePlan{}, err
		}
		return RestorePlan{}, ErrNothingReplicated
	case err != nil:
		return RestorePlan{}, err
	case len(infos) == 0:
		return RestorePlan{}, ErrNothingReplicated
	}
	return finishPlan(ctx, l, infos, 0)
}

// RestoreLocal restores the replica c holds, a litestream file replica in a
// private directory (the db-read worker's, or the CLI's for a forwarded
// pull), exactly to target, at out, and proves it with lease.CheckDBContext.
// On failure nothing is left at out.
func RestoreLocal(ctx context.Context, c litestream.ReplicaClient, out string, target uint64) error {
	// litestream's compaction checks no context, so every read of the local
	// files checks it here, as the door path's downloaded.Read does (Astra
	// round 1 on plan 3, finding 2).
	if err := restoreOnce(ctx, ctxReplica{ReplicaClient: c, ctx: ctx}, out, target, time.Time{}); err != nil {
		removeCopy(out)
		if cerr := ctx.Err(); cerr != nil {
			return cerr
		}
		return err
	}
	if err := checkDB(ctx, out); err != nil {
		removeCopy(out)
		if cerr := ctx.Err(); cerr != nil {
			return cerr
		}
		return fmt.Errorf("%w: %v", ErrCheckFailed, err)
	}
	if err := ctx.Err(); err != nil {
		removeCopy(out)
		return err
	}
	return nil
}

// ctxReplica is a local replica client whose file reads stop once ctx is
// done.
type ctxReplica struct {
	litestream.ReplicaClient
	ctx context.Context
}

func (c ctxReplica) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	if err := c.ctx.Err(); err != nil {
		return nil, err
	}
	rc, err := c.ReplicaClient.OpenLTXFile(ctx, level, minTXID, maxTXID, offset, size)
	if err != nil {
		return nil, err
	}
	return ctxReadCloser{ReadCloser: rc, ctx: c.ctx}, nil
}

type ctxReadCloser struct {
	io.ReadCloser
	ctx context.Context
}

func (r ctxReadCloser) Read(p []byte) (int, error) {
	if err := r.ctx.Err(); err != nil {
		return 0, err
	}
	return r.ReadCloser.Read(p)
}