package host

import (
	"context"
	"os"
	"strings"
	"testing"

	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/release"
)

// bucketStore is a release.Store over one bucket of a versioned bucket,
// as production's S3 store is over the data bucket its versioned client
// lists: current reads with the current version as the token, create-only
// and compare-and-swap writes, deletes as delete markers.
type bucketStore struct {
	b      *leasetest.VersionedBucket
	bucket string
}

func (s bucketStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	return casBucketStore(s).Get(ctx, key)
}

func (s bucketStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	return casBucketStore(s).Put(ctx, key, body, ifMatch)
}

func (s bucketStore) List(ctx context.Context, prefix string) ([]string, error) {
	return casBucketStore(s).List(ctx, prefix)
}

func (s bucketStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	keys, err := s.List(ctx, prefix)
	if err != nil {
		return nil, err
	}
	seen := map[string]bool{}
	var out []string
	for _, k := range keys {
		dir, _, ok := strings.Cut(strings.TrimPrefix(k, prefix), "/")
		if ok && !seen[dir] {
			seen[dir] = true
			out = append(out, dir)
		}
	}
	return out, nil
}

func (s bucketStore) Delete(_ context.Context, key string) error {
	if _, ok := s.b.Current(s.bucket, key); !ok {
		return os.ErrNotExist
	}
	s.b.BareDelete(s.bucket, key)
	return nil
}

var (
	_ release.Store        = bucketStore{}
	_ lease.VersionedStore = (*leasetest.VersionedBucket)(nil)
)

// Merge round 6, finding 1: relocation names a generation's files by the
// one files prefix rule (moves.GenerationFilesPrefix), so what its build
// writes is what main's provider and the park view read. P's prepared
// manifest is at movesFilesPrefix(location, P); every files object under
// the location belongs to a generation by the park view's classifier
// (landingGenerationOf), where the old files/<generation>/ layout
// belonged to none and refused every park of the location; and the files
// live set pins every object P's manifest names, which it never read
// under the old layout.
func TestRelocationFilesUseTheParkViewsPrefix(t *testing.T) {
	f := newBuildFixtureWith(t, buildOptions{Files: map[string]string{"public/a.txt": "a", "uploads/b.bin": "b"}})
	if err := f.build(); err != nil {
		t.Fatal(err)
	}
	ctx := t.Context()
	location := f.obs.Claim.Holder.Location
	m, found, err := hostfiles.Load(ctx, f.files, movesFilesPrefix(location, "p1"))
	if err != nil || !found || m.Seq != 0 || len(m.Entries) != 2 {
		t.Fatalf("P's prepared manifest at the files prefix rule: found=%v %+v %v", found, m, err)
	}
	listing, err := f.bucket.ListVersions(ctx, "data", location+"/files/")
	if err != nil || len(listing) == 0 {
		t.Fatalf("listing: %v", err)
	}
	for _, x := range listing {
		if g, ok := landingGenerationOf(location, x.Key); !ok || (g != "g1" && g != "p1") {
			t.Fatalf("%s belongs to %q (%v), not a generation", x.Key, g, ok)
		}
	}
	if _, ok := landingGenerationOf(location, location+"/files/p1/manifest.json"); ok {
		t.Fatal("the old files/<generation>/ layout is classified")
	}
	v := park.View{GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}
	if err := pinFilesLiveSet(ctx, f.bucket, "data", location, listing, &v, "p1"); err != nil {
		t.Fatal(err)
	}
	for _, e := range m.Entries {
		if !v.Pinned[park.Ref{Key: e.Object, VersionID: f.bucket.LatestID("data", e.Object)}] {
			t.Fatalf("%s, which P's manifest names, is not pinned: %v", e.Object, v.Pinned)
		}
	}
}