package hostops

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"io"
	"time"

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

type effectPermitKey struct{}
type effectPermit struct{ Resource string }

// conditionalRecordStore enforces D7's record-shape protocol beneath the
// mutation boundary: provisioning records are created only inside their own
// issued record effect; every other write replaces an existing exact ETag
// and preserves identity. Authority (which app, which permit) is the
// boundary's job, not this layer's. Named private fields; no embedding.
type conditionalRecordStore struct {
	homeRegion string
	raw        release.Store
	pub        ed25519.PublicKey
}

var (
	_ release.Store              = conditionalRecordStore{}
	_ release.ConditionalDeleter = conditionalRecordStore{}
	_ release.Opener             = conditionalRecordStore{}
	_ release.ModLister          = conditionalRecordStore{}
	_ release.SizeLister         = conditionalRecordStore{}
	_ release.PageLister         = conditionalRecordStore{}
	_ release.ConditionalGetter  = conditionalRecordStore{}
	_ release.ETagLister         = conditionalRecordStore{}
	_ release.BucketListDirs     = conditionalRecordStore{}
)

func NewConditionalRecordStore(raw release.Store, pub ed25519.PublicKey, homeRegion ...string) (release.Store, error) {
	if raw == nil || len(pub) != ed25519.PublicKeySize {
		return nil, ErrContract
	}
	home, err := consoleHomeRegion(homeRegion)
	if err != nil {
		return nil, err
	}
	return conditionalRecordStore{raw: raw, pub: pub, homeRegion: home}, nil
}

func provisioningKey(key string) bool {
	return provision.IsRecordKey(key)
}

func (s conditionalRecordStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	return s.raw.Get(ctx, key)
}
func (s conditionalRecordStore) List(ctx context.Context, prefix string) ([]string, error) {
	return s.raw.List(ctx, prefix)
}
func (s conditionalRecordStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	return s.raw.ListDirs(ctx, prefix)
}
func (s conditionalRecordStore) BucketShapedListDirs() bool { return release.HasBucketListDirs(s.raw) }
func (s conditionalRecordStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
	return release.OpenObject(ctx, s.raw, key)
}
func (s conditionalRecordStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
	if l, ok := s.raw.(release.ModLister); ok {
		return l.ListMod(ctx, prefix)
	}
	return nil, ErrContract
}
func (s conditionalRecordStore) ListSizes(ctx context.Context, prefix string) ([]release.KeySize, error) {
	if l, ok := s.raw.(release.SizeLister); ok {
		return l.ListSizes(ctx, prefix)
	}
	return nil, ErrContract
}
func (s conditionalRecordStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	if l, ok := s.raw.(release.PageLister); ok {
		return l.ListPage(ctx, prefix, startAfter, limit)
	}
	return nil, ErrContract
}

func (s conditionalRecordStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	if g, ok := s.raw.(release.ConditionalGetter); ok {
		return g.GetIfChanged(ctx, key, etag)
	}
	return nil, "", false, ErrContract
}
func (s conditionalRecordStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
	if l, ok := s.raw.(release.ETagLister); ok {
		return l.ListETags(ctx, prefix)
	}
	return nil, ErrContract
}

func (s conditionalRecordStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
	if !provisioningKey(key) {
		return s.raw.Put(ctx, key, body, etag)
	}
	if err := CheckCreateFreeze(ctx, s.raw, s.pub); err != nil {
		return "", err
	}
	// Rev 3 operation-bound lifecycle record writer (lifecycle_record.go).
	if permit, ok := ctx.Value(lifecycleRecordKey{}).(lifecycleRecordPermit); ok {
		return s.putLifecycle(ctx, key, body, etag, permit)
	}
	if permit, ok := ctx.Value(effectPermitKey{}).(effectPermit); ok && (permit.Resource == key || permit.Resource == "control-object:"+key) {
		return putSequencedRecord(ctx, s.raw, key, body, etag)
	}
	// Not-found is never turned into create outside the record effect.
	if etag == "" {
		return "", ErrContract
	}
	current, err := ReadRecord(ctx, s.raw, key)
	if err != nil {
		return "", err
	}
	if current.ETag != etag {
		return "", release.ErrCASConflict
	}
	// Move journal fields change only through the lifecycle writer above.
	if !sameMoveJournal(current.Body, body) {
		return "", ErrContract
	}
	var next struct {
		Account     string    `json:"account"`
		App         string    `json:"app"`
		Host        string    `json:"host"`
		Name        string    `json:"name"`
		Acquisition string    `json:"acq"`
		Region      string    `json:"region"`
		CreatedAt   time.Time `json:"created_at"`
	}
	if json.Unmarshal(body, &next) != nil {
		return "", ErrIdentity
	}
	n := current.Holder
	if next.Account != n.Account || next.App != n.App || next.Host != n.Host || next.Name != n.Name || next.Acquisition != n.Acquisition || next.Region != n.RecordRegion || !next.CreatedAt.Equal(n.CreatedAt) {
		return "", ErrIdentity
	}
	obs, err := hostclaim.Read(ctx, s.raw, n.Host)
	if err != nil {
		return "", err
	}
	if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindServing || !matches(current, obs.Claim.Holder) {
		return "", ErrPending
	}
	return putSequencedRecord(ctx, s.raw, key, body, etag)
}

func (s conditionalRecordStore) Delete(ctx context.Context, key string) error {
	if provisioningKey(key) {
		return ErrContract
	}
	return s.raw.Delete(ctx, key)
}

func (s conditionalRecordStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
	return release.DeleteIfMatch(ctx, s.raw, key, etag)
}