package console

import (
	"bufio"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/credentials"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/benbjohnson/litestream"
	lsfile "github.com/benbjohnson/litestream/file"
	lss3 "github.com/benbjohnson/litestream/s3"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/dbreads"
	"github.com/carlosframework/platform/internal/dbreads/dbreadstest"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// uploadReplica writes a file replica's LTX files into an S3 replica the way
// a box's litestream does (WriteLTXFile), level by level.
func uploadReplica(t *testing.T, from *lsfile.ReplicaClient, to *lss3.ReplicaClient) {
	t.Helper()
	ctx := context.Background()
	for level := 0; level <= litestream.SnapshotLevel; level++ {
		itr, err := from.LTXFiles(ctx, level, 0, false)
		if err != nil {
			t.Fatal(err)
		}
		for itr.Next() {
			f := itr.Item()
			rc, err := from.OpenLTXFile(ctx, level, f.MinTXID, f.MaxTXID, 0, 0)
			if err != nil {
				t.Fatal(err)
			}
			if _, err := to.WriteLTXFile(ctx, level, f.MinTXID, f.MaxTXID, rc); err != nil {
				t.Fatal(err)
			}
			rc.Close()
		}
		itr.Close()
	}
}

// dbfGateStore holds the box's chunk and result creates, once armed, until
// release closes, and records when each landed: T16 needs a box that
// publishes only after the console has ended the request (plan review
// finding 19).
type dbfGateStore struct {
	release.Store
	mu      sync.Mutex
	open    chan struct{}
	reached chan struct{}
	once    *sync.Once
	landed  []time.Time
}

func (g *dbfGateStore) arm() {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.open, g.reached, g.once, g.landed = make(chan struct{}), make(chan struct{}), &sync.Once{}, nil
}

func (g *dbfGateStore) release() {
	g.mu.Lock()
	defer g.mu.Unlock()
	close(g.open)
}

func (g *dbfGateStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	g.mu.Lock()
	open, reached, once := g.open, g.reached, g.once
	g.mu.Unlock()
	if open == nil || !(strings.Contains(key, "/out/") || strings.HasSuffix(key, "/result.json")) {
		return g.Store.Put(ctx, key, body, ifMatch)
	}
	once.Do(func() { close(reached) })
	<-open
	etag, err := g.Store.Put(ctx, key, body, ifMatch)
	if err == nil {
		g.mu.Lock()
		g.landed = append(g.landed, time.Now())
		g.mu.Unlock()
	}
	return etag, err
}

func (g *dbfGateStore) landedAfter(t time.Time) (n int, early bool) {
	g.mu.Lock()
	defer g.mu.Unlock()
	for _, at := range g.landed {
		if !at.After(t) {
			early = true
		}
	}
	return len(g.landed), early
}

// TestDBForwardMinio is T1 and T16 against a real S3 server: a console in ie
// and a box in uk sharing one deployment bucket (versioned), each with its
// own data bucket and its own credentials, on one endpoint (spec §12). The
// console's credentials cannot read uk's bucket (a MinIO policy), so a
// console that read it would fail. Two tenants in uk each get their own
// rows. Neither the console nor the box ever deletes or replaces anything
// under db-reads/, and a box's writes after the request ended change
// nothing. It runs only where .amadan/ci.d/30-minio-contract stands MinIO
// up; the box is plan 1's real code joined in process (dbreadstest).
func TestDBForwardMinio(t *testing.T) {
	if os.Getenv("CARLOS_MINIO_TEST") != "1" {
		t.Skip("set CARLOS_MINIO_TEST=1 (with a running MinIO and the DBF_* variables) to run; see .amadan/ci.d/30-minio-contract")
	}
	need := func(name string) string {
		v := os.Getenv(name)
		if v == "" {
			t.Fatalf("%s is not set", name)
		}
		return v
	}
	endpoint := need("MINIO_ENDPOINT")
	rootKey, rootSecret := need("DBF_ROOT_KEY"), need("DBF_ROOT_SECRET")
	consoleKey, consoleSecret := need("DBF_CONSOLE_KEY"), need("DBF_CONSOLE_SECRET")
	boxKey, boxSecret := need("DBF_BOX_KEY"), need("DBF_BOX_SECRET")
	store := func(key, secret, bucket string) *release.S3Store {
		return &release.S3Store{Bucket: bucket, Client: awss3.New(awss3.Options{
			Region: "us-east-1", Credentials: credentials.NewStaticCredentialsProvider(key, secret, ""),
			BaseEndpoint: aws.String(endpoint), UsePathStyle: true,
			RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
			ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
		})}
	}
	replica := func(key, secret, bucket, prefix string) *lss3.ReplicaClient {
		c := lss3.NewReplicaClient()
		c.AccessKeyID, c.SecretAccessKey, c.Region, c.Bucket, c.Path, c.Endpoint, c.ForcePathStyle =
			key, secret, "us-east-1", bucket, prefix, endpoint, true
		return c
	}
	run := strconv.FormatInt(time.Now().UnixNano(), 36)
	shop, other := "shop-"+run+".example.com", "other-"+run+".example.com"
	for host, batches := range map[string]int{shop: 3, other: 1} {
		dir := t.TempDir()
		dbdoortest.Replicate(t, dir, batches)
		uploadReplica(t, lsfile.NewReplicaClient(dir), replica(rootKey, rootSecret, "dbf-data-uk", "instances/"+host))
	}

	// The policy, shown: the console's credentials cannot list uk's bucket.
	itr, err := replica(consoleKey, consoleSecret, "dbf-data-uk", "instances/"+shop).LTXFiles(context.Background(), 0, 0, false)
	if err == nil {
		for itr.Next() {
		}
		err = itr.Err()
		itr.Close()
	}
	if err == nil {
		t.Fatal("the console's credentials read uk's data bucket: the test proves nothing")
	}

	deploy := store(consoleKey, consoleSecret, "dbf-deploy")
	fx := newConsoleFixtureOn(t, true, strictStore{Store: deploy, t: t})
	fx.srv.cfg.DBForward, fx.srv.cfg.Region, fx.srv.cfg.HibernateBucket = true, "ie", "dbf-data-ie"
	fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
		return replica(consoleKey, consoleSecret, "dbf-data-ie", prefix), nil // the only data bucket this console has
	}
	if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", time.Now()); err != nil {
		t.Fatal(err)
	}
	if err := fx.st.AddMember(OpsAccountID, "member@example.com", RoleMember, time.Now()); err != nil {
		t.Fatal(err)
	}
	member := cliToken(t, fx, "member@example.com", "code-minio")
	for _, host := range []string{shop, other} {
		putShopRecord(t, fx, provision.Record{Host: host, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)})
	}
	putEdgesFile(t, fx.rel, "uk-"+run, "uk", []string{dbreads.Capability}, false)
	putBoxStatus(t, fx.rel, "uk-"+run, time.Now())
	gate := &dbfGateStore{Store: store(boxKey, boxSecret, "dbf-deploy")}
	box := dbreadstest.Start(t, dbreadstest.Options{
		Store: strictStore{Store: gate, t: t}, Region: "uk", Box: "uk-" + run,
		Pub: fx.pub, HibernatePrefix: "instances",
		OpenReplica: func(_ context.Context, prefix string) (dbdoor.ReplicaReader, error) {
			return replica(boxKey, boxSecret, "dbf-data-uk", prefix), nil
		},
	})
	runBox(t, box)
	dfx := &dbfFixture{consoleFixture: fx, member: member}

	// T1: a forwarded query end to end, for each tenant, each its own rows.
	for host, want := range map[string]string{shop: "n\n30\n", other: "n\n10\n"} {
		body, key := queryBody(t, "SELECT count(*) AS n FROM t")
		res := dfx.openRead(t, host, member, body)
		lines := readLines(t, res.Body)
		res.Body.Close()
		if rows, _ := openAnswer(t, key, lines); string(rows) != want {
			t.Fatalf("%s: %q, want %q", host, rows, want)
		}
	}

	// A pull through MinIO (plan 3): the planned files, sealed as the ltx
	// stream, adding up to the summary, whose file list ends at its copy.
	pkey, _ := dbreads.NewReplyKey()
	pres := dfx.openRead(t, shop, member, dbreads.ReadBody{Op: dbreads.OpPull, ReplyKey: dbreads.EncodeReplyKey(pkey.PublicKey())})
	plines := readLines(t, pres.Body)
	pres.Body.Close()
	presult := plines[len(plines)-1].Result
	if presult == nil || presult.State != dbreads.StateDone {
		t.Fatalf("pull: %+v", plines[len(plines)-1])
	}
	psum, err := dbreads.OpenSummary(pkey, plines[0].Admitted, presult.Sealed)
	if err != nil || len(psum.Files) == 0 || psum.TXID != psum.Files[len(psum.Files)-1].MaxTXID {
		t.Fatalf("pull summary %+v %v", psum, err)
	}
	ph := sha256.New()
	var pn int64
	for _, l := range plines {
		if l.Chunk == 0 {
			continue
		}
		plain, err := dbreads.OpenChunk(pkey, plines[0].Admitted, l.Chunk, dbreads.StreamLTX, l.Chunk == presult.LastSeq, l.Data)
		if err != nil {
			t.Fatalf("pull chunk %d: %v", l.Chunk, err)
		}
		ph.Write(plain)
		pn += int64(len(plain))
	}
	if hex.EncodeToString(ph.Sum(nil)) != psum.SHA256 || pn != psum.Bytes {
		t.Fatal("the pulled bytes do not add up to the summary")
	}

	// T16 (plan review finding 19): the real box reaches publication and is
	// held there; the console's request ends, by disconnect and then by its
	// bound; only then does the box publish. Its late chunks and result land
	// after the console ended, change nothing the console sends, and the
	// strict stores see no delete and no replacing write from either side.
	late := func(t *testing.T, end func(res *http.Response, br *bufio.Reader)) {
		gate.arm()
		body, _ := queryBody(t, "SELECT count(*) AS n FROM t")
		res := dfx.openRead(t, shop, member, body)
		br := bufio.NewReader(res.Body)
		line, _ := br.ReadBytes('\n')
		var adm dbreads.Line
		json.Unmarshal(line, &adm)
		end(res, br)
		// The handler's last act is the cancel: once it is there, the
		// console has ended the request.
		waitKey := func(key string) {
			deadline := time.Now().Add(60 * time.Second)
			for {
				if _, _, err := deploy.Get(context.Background(), key); err == nil {
					return
				}
				if time.Now().After(deadline) {
					t.Fatalf("no %s", key)
				}
				time.Sleep(100 * time.Millisecond)
			}
		}
		waitKey(dbreads.CancelKey(adm.Admitted))
		consoleEnded := time.Now()
		select {
		case <-gate.reached:
		case <-time.After(60 * time.Second):
			t.Fatal("the box never reached publication")
		}
		if keys, _ := deploy.List(context.Background(), dbreads.OutPrefix(adm.Admitted)); len(keys) != 0 {
			t.Fatalf("output published before the console ended: %v", keys)
		}
		gate.release()
		waitKey(dbreads.ResultKey(adm.Admitted))
		// At least the result lands late; a held chunk lands too unless the
		// owner's cancel watch ended its create first (plan 1's watchCancel).
		if n, early := gate.landedAfter(consoleEnded); n < 1 || early {
			t.Fatalf("%d late writes, any before the console ended: %v", n, early)
		}
	}
	t.Run("by disconnect", func(t *testing.T) {
		late(t, func(res *http.Response, _ *bufio.Reader) {
			select {
			case <-gate.reached:
			case <-time.After(60 * time.Second):
				t.Fatal("the box never reached publication")
			}
			res.Body.Close()
		})
	})
	t.Run("by timeout", func(t *testing.T) {
		// The request's own bound, well past the box's restore and query of
		// three batches, well short of the box's 20-minute owner deadline.
		fx.srv.dbf.bound, fx.srv.dbf.grace = 20*time.Second, 0
		late(t, func(res *http.Response, br *bufio.Reader) {
			defer res.Body.Close()
			lines := readLines(t, br)
			if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult {
				t.Fatalf("by timeout: %+v", last)
			}
			for _, l := range lines {
				if l.Chunk != 0 || l.Result != nil {
					t.Fatalf("a held box's output reached the console: %+v", l)
				}
			}
		})
	})
}