package provision

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"strconv"
	"strings"

	"github.com/carlosframework/platform/internal/release"
)

// IsRecordKey recognizes both instance and sidecar control records.
func IsRecordKey(key string) bool {
	p := strings.Split(key, "/")
	return len(p) == 5 && p[0] == "apps" && (p[3] == "instances" || p[3] == "sidecars") && strings.HasSuffix(p[4], ".json")
}

var ErrRecordSequence = errors.New("provision: invalid record sequence")
var ErrRecordSequenceExhausted = errors.New("provision: record sequence exhausted")

type recordSequenceKey struct{}
type preparedRecordKey struct{}

// WithRecordSequence introduces the counter on a first claim-mode write.
// A populated counter advances even without this context. This is encoding,
// not mutation authority; callers retain their existing claim and app gates.
func WithRecordSequence(ctx context.Context) context.Context {
	return context.WithValue(ctx, recordSequenceKey{}, true)
}

// RecordSequence preserves unknown fields and parses the counter without a
// float64 round trip. Missing means legacy zero; null, aliases, duplicates,
// fractions, negatives and overflow refuse at the claim-on write boundary.
func RecordSequence(body []byte) (map[string]json.RawMessage, uint64, error) {
	d := json.NewDecoder(bytes.NewReader(body))
	if first, err := d.Token(); err != nil || first != json.Delim('{') {
		return nil, 0, ErrRecordSequence
	}
	fields := map[string]json.RawMessage{}
	var seq uint64
	for d.More() {
		token, err := d.Token()
		if err != nil {
			return nil, 0, ErrRecordSequence
		}
		name, ok := token.(string)
		if !ok {
			return nil, 0, ErrRecordSequence
		}
		if _, exists := fields[name]; exists {
			return nil, 0, ErrRecordSequence
		}
		var value json.RawMessage
		if d.Decode(&value) != nil {
			return nil, 0, ErrRecordSequence
		}
		if strings.EqualFold(name, "record_seq") {
			if name != "record_seq" {
				return nil, 0, ErrRecordSequence
			}
			seq, err = strconv.ParseUint(string(value), 10, 64)
			if err != nil {
				return nil, 0, ErrRecordSequence
			}
		}
		fields[name] = value
	}
	if end, err := d.Token(); err != nil || end != json.Delim('}') {
		return nil, 0, ErrRecordSequence
	}
	if _, err := d.Token(); err != io.EOF {
		return nil, 0, ErrRecordSequence
	}
	return fields, seq, nil
}

// AdvanceRecordSequence alone owns the counter. Callers must preserve the
// read value, including on retries; they cannot reset it or choose a future
// revision. Exhaustion refuses instead of wrapping to a reusable body.
func AdvanceRecordSequence(before, body []byte) ([]byte, error) {
	_, base, err := RecordSequence(before)
	if err != nil {
		return nil, err
	}
	fields, candidate, err := RecordSequence(body)
	if err != nil {
		return nil, err
	}
	if candidate != base {
		return nil, ErrRecordSequence
	}
	if base == ^uint64(0) {
		return nil, ErrRecordSequenceExhausted
	}
	fields["record_seq"] = json.RawMessage(strconv.FormatUint(base+1, 10))
	return json.Marshal(fields)
}

// preparedRecord binds one synchronous Put to its bytes and CAS base. Store
// wrappers may call WriteRecord again without advancing twice. An inner
// claim boundary can introduce the counter and report the written value back
// to the typed outer writer. This private context grants no write authority.
type preparedRecord struct {
	key, etag string
	body      []byte
	base, seq uint64
	sequenced bool
}

// WriteRecord is the only record-key Put primitive. before is the exact
// If-Match read ({} for create). body preserves that read's counter. Never
// call it with a synthesized before to reset a live record's sequence.
// Legacy absent counters and no claim context retain the exact body bytes.
func WriteRecord(ctx context.Context, s release.Store, key string, before, body []byte, etag string) (string, uint64, error) {
	if !IsRecordKey(key) {
		return "", 0, ErrRecordSequence
	}
	fields, base, err := RecordSequence(before)
	if err != nil {
		return "", 0, err
	}
	_, candidate, err := RecordSequence(body)
	if err != nil {
		return "", 0, err
	}
	enabled, _ := ctx.Value(recordSequenceKey{}).(bool)
	_, present := fields["record_seq"]
	p, nested := ctx.Value(preparedRecordKey{}).(*preparedRecord)
	if nested {
		if p.key != key || p.etag != etag || p.base != base || p.seq != candidate || !bytes.Equal(p.body, body) {
			return "", 0, ErrRecordSequence
		}
	} else {
		if candidate != base {
			return "", 0, ErrRecordSequence
		}
		p = &preparedRecord{key: key, etag: etag, body: append([]byte(nil), body...), base: base, seq: base}
		ctx = context.WithValue(ctx, preparedRecordKey{}, p)
	}
	if !p.sequenced && (enabled || present) {
		body, err = AdvanceRecordSequence(before, body)
		if err != nil {
			return "", 0, err
		}
		p.body, p.seq, p.sequenced = append([]byte(nil), body...), base+1, true
	}
	tag, err := s.Put(ctx, key, body, etag)
	return tag, p.seq, err
}