package dbdoor

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"

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

// Doer sends one GET to the door for one instance. tail is relative to the
// instance's door root (ListTail or GetTail); the caller owns the console
// URL, the bearer and the account segment, which is why the client never
// sees them. The caller's response body is the client's to close.
type Doer func(ctx context.Context, tail string, header http.Header) (*http.Response, error)

// Client is the CLI's litestream.ReplicaClient over the db door (spec
// 2026-09-02 §6, "The read-only replica client"), the way internal/fleetapi
// implements release.Store over the edge store door: the member has no AWS
// credentials, by design.
//
// It supplies the five methods a restore needs (Type, Init, LTXFiles,
// OpenLTXFile, SetLogger). The three that mutate a replica return
// errReadOnly and have no other body. It must never implement
// litestream.ReplicaClientV3, or Restore would take the v0.3 path.
type Client struct {
	handles dbHandleState
	do      Doer
	logger  *slog.Logger
}

// NewClient returns a read-only replica client that reads through do.
func NewClient(do Doer) *Client {
	return &Client{do: do, logger: slog.Default()}
}

// errReadOnly is not a guard that can be switched off: these three methods
// have no other body. The two-replicators hazard internal/caps and
// internal/lease both warn about is unreachable from this client by
// construction, and a future edit that wanted to reach it would have to add
// the capability first.
var errReadOnly = errors.New("dbdoor: this client is read-only; it restores replicas and never writes them")

// DoorError is a refusal from the door, carrying the console's own sentence
// (its {"error"} body) so the CLI can show it as written.
type DoorError struct {
	Status  int
	Message string
	// RetryAfter is the door's Retry-After, in whole seconds, when it sent
	// one (a 429 from a full stream bound does); zero otherwise.
	RetryAfter time.Duration
	// Forward is the list door's forward marker on a 409 (db region
	// forward spec §5.5): the instance is readable through the forwarded
	// read in Forward.Region. nil for every other refusal.
	Forward *Forward
}

func (e *DoorError) Error() string {
	if e.Message == "" {
		return fmt.Sprintf("the console answered %d", e.Status)
	}
	return e.Message
}

func (c *Client) Type() string { return "carlos-db" }

// Init is a no-op: the transport needs no setup, and litestream requires
// it to be idempotent.
func (c *Client) Init(context.Context) error { return nil }

func (c *Client) SetLogger(l *slog.Logger) { c.logger = l }

// LTXFiles lists one level. seek is filtered here (wire specific 3), with
// litestream's own rule: files whose MinTXID is below seek are skipped.
// useMetadata is ignored; nothing in a non-timestamp restore reads
// CreatedAt except debug logs.
func (c *Client) LTXFiles(ctx context.Context, level int, seek ltx.TXID, _ bool) (ltx.FileIterator, error) {
	c.handles.listMu.Lock()
	defer c.handles.listMu.Unlock()
	files, ok := c.planFiles(level)
	if !ok {
		var err error
		files, err = c.listFiles(ctx, level)
		if err != nil {
			return nil, err
		}
	}
	return fileIterator(files, level, seek), nil
}

func (c *Client) listFiles(ctx context.Context, level int) ([]LTXFile, error) {
	res, err := c.do(ctx, ListTail(level), http.Header{})
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		return nil, doorError(res)
	}
	var list ListResponse
	if err := json.NewDecoder(io.LimitReader(res.Body, 16<<20)).Decode(&list); err != nil {
		return nil, fmt.Errorf("dbdoor: unreadable listing for level %d: %w", level, err)
	}
	return c.rememberFiles(level, list)
}

func fileIterator(files []LTXFile, level int, seek ltx.TXID) ltx.FileIterator {
	infos := make([]*ltx.FileInfo, 0, len(files))
	for _, f := range files {
		if ltx.TXID(f.MinTXID) < seek {
			continue
		}
		infos = append(infos, &ltx.FileInfo{
			Level: level, MinTXID: ltx.TXID(f.MinTXID), MaxTXID: ltx.TXID(f.MaxTXID),
			Size: f.Size, CreatedAt: f.CreatedAt,
		})
	}
	return ltx.NewFileInfoSliceIterator(infos)
}

// OpenLTXFile opens one file by coordinate, sending the S3 client's exact
// Range form. A 404 is os.ErrNotExist (wire specific 5): that is what lets
// RestoreTo tell a file that vanished under the plan from a refusal.
func (c *Client) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	h := http.Header{}
	handle, err := c.fileHandle(level, uint64(minTXID), uint64(maxTXID))
	if err != nil {
		return nil, err
	}
	if handle != "" {
		h.Set(HandleHeader, handle)
	}
	h.Set("Range", RangeHeader(offset, size))
	res, err := c.do(ctx, GetTail(level, uint64(minTXID), uint64(maxTXID)), h)
	if err != nil {
		return nil, err
	}
	switch res.StatusCode {
	case http.StatusOK, http.StatusPartialContent:
		if handle != "" && res.Header.Get(HandleHeader) != handle {
			res.Body.Close()
			return nil, errors.New("dbdoor: console did not enforce the pinned handle")
		}
		return res.Body, nil
	case http.StatusNotFound:
		res.Body.Close()
		return nil, fmt.Errorf("dbdoor: level %d %s: %w", level, Span(uint64(minTXID), uint64(maxTXID)), os.ErrNotExist)
	default:
		defer res.Body.Close()
		return nil, doorError(res)
	}
}

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

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

func (c *Client) DeleteAll(context.Context) error { return errReadOnly }

// doorError reads a refusal's {"error"} sentence, and the db region
// forward marker beside it on a 409. A body that is not that
// shape (the stdlib's 404 text, a proxy's page) keeps only the status.
func doorError(res *http.Response) error {
	raw, _ := io.ReadAll(io.LimitReader(res.Body, 64<<10))
	var body struct {
		Error      string   `json:"error"`
		Forward    int      `json:"forward"`
		Region     string   `json:"region"`
		RegionName string   `json:"region_name"`
		ForwardOps []string `json:"forward_ops"`
	}
	e := &DoorError{Status: res.StatusCode}
	if json.Unmarshal(raw, &body) == nil {
		e.Message = strings.TrimSpace(body.Error)
		if res.StatusCode == http.StatusConflict && body.Forward == 1 && forwardRegion.MatchString(body.Region) {
			e.Forward = &Forward{Region: body.Region, RegionName: body.RegionName, Ops: body.ForwardOps}
		}
	}
	// Seconds only (the door never sends a date), capped so a hostile or
	// broken value cannot park a download for hours.
	if secs, err := strconv.Atoi(strings.TrimSpace(res.Header.Get("Retry-After"))); err == nil && secs > 0 {
		// Clamp the count before multiplying: a large one would wrap.
		e.RetryAfter = time.Duration(min(secs, 60)) * time.Second
	}
	return e
}

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