package dbreads

import (
	"bytes"
	"context"
	"crypto/ecdh"
	"crypto/ed25519"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"net"
	"os"
	"sort"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	lsfile "github.com/benbjohnson/litestream/file"

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

const fxAccount, fxApp, fxHost = "bab", "shop", "shop.example.com"

// inProcWorkers runs the real ServeWorker in process over net.Pipe: not
// acceptance (Task 15 drives the installed units). units counts the live
// workers, the way the box counts carlos-dbread@ units.
type inProcWorkers struct {
	t     *testing.T
	opt   WorkerOptions
	live  atomic.Int64
	dials atomic.Int64
	// linger keeps a finished worker counted as a live unit for a while,
	// as systemd's deactivation does; liveAtDial records the live count
	// each dial saw.
	linger     time.Duration
	mu         sync.Mutex
	liveAtDial []int64
	// failStart makes the worker end at once without a word, as a unit
	// whose service fails to start after the socket accepted does.
	failStart atomic.Bool
	// sent is the type of every frame the agent wrote, in order.
	sent []byte
}

// frameTap records the first byte of each Write the agent makes: one frame
// type per frame (writeFrame writes a frame in one Write).
type frameTap struct {
	net.Conn
	w *inProcWorkers
}

func (f frameTap) Write(p []byte) (int, error) {
	if len(p) > 0 {
		f.w.mu.Lock()
		f.w.sent = append(f.w.sent, p[0])
		f.w.mu.Unlock()
	}
	return f.Conn.Write(p)
}

func (w *inProcWorkers) sentFrames() []byte {
	w.mu.Lock()
	defer w.mu.Unlock()
	return append([]byte(nil), w.sent...)
}

func (w *inProcWorkers) dial(context.Context) (net.Conn, error) {
	agentEnd, workerEnd := net.Pipe()
	w.mu.Lock()
	w.liveAtDial = append(w.liveAtDial, w.live.Load())
	w.mu.Unlock()
	w.dials.Add(1)
	w.live.Add(1)
	if w.failStart.Load() {
		// The socket accepted and the service died: the plan's header is
		// read by nobody useful, and the connection drops.
		go func() {
			readFrame(workerEnd)
			workerEnd.Close()
			time.Sleep(w.linger)
			w.live.Add(-1)
		}()
		return frameTap{Conn: agentEnd, w: w}, nil
	}
	opt := w.opt
	if opt.Scratch == "" {
		opt.Scratch = w.t.TempDir()
	}
	if opt.Exit == nil {
		opt.Exit = func(int) {}
	}
	if opt.Self == nil {
		opt.Self = func() HelloReply { return HelloReply{V: ProtocolVersion, Interfaces: []string{"lo"}} }
	}
	go func() {
		ServeWorker(workerEnd, opt)
		workerEnd.Close()
		time.Sleep(w.linger)
		w.live.Add(-1)
	}()
	return frameTap{Conn: agentEnd, w: w}, nil
}

func (w *inProcWorkers) units(context.Context) (int, error) { return int(w.live.Load()), nil }

func plentyFree() (uint64, uint64, error) { return 100 << 30, 200 << 30, nil }

type fixture struct {
	t        *testing.T
	store    release.Store
	rootPub  ed25519.PublicKey
	rootPriv ed25519.PrivateKey
	reply    *ecdh.PrivateKey
	replica  string
	now      time.Time
}

// newFixture is a deployment store holding shop.example.com's record in uk,
// and a file replica of its database with batches*10 rows.
func newFixture(t *testing.T, batches int) *fixture {
	t.Helper()
	pub, priv, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	reply, err := NewReplyKey()
	if err != nil {
		t.Fatal(err)
	}
	fx := &fixture{t: t, store: release.NewDirStore(t.TempDir()), rootPub: pub, rootPriv: priv, reply: reply,
		replica: t.TempDir(), now: time.Now().UTC().Truncate(time.Second)}
	if batches > 0 {
		dbdoortest.Replicate(t, fx.replica, batches)
	}
	fx.putRecord(provision.Record{Host: fxHost, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)})
	return fx
}

func (fx *fixture) putRecordUnder(acct string, rec provision.Record) {
	fx.t.Helper()
	rec.Account, rec.App = acct, fxApp
	if rec.Host == "" {
		rec.Host = fxHost
	}
	if rec.Hibernate == nil {
		yes := true
		rec.Hibernate = &yes
	}
	body, err := json.Marshal(rec)
	if err != nil {
		fx.t.Fatal(err)
	}
	key := provision.RecordKey(acct, fxApp, rec.Host)
	if err := fx.store.Delete(context.Background(), key); err != nil {
		fx.t.Fatal(err)
	}
	if _, err := fx.store.Put(context.Background(), key, body, ""); err != nil {
		fx.t.Fatal(err)
	}
}

func (fx *fixture) putRecord(rec provision.Record) { fx.putRecordUnder(fxAccount, rec) }

func (fx *fixture) record() provision.Record {
	fx.t.Helper()
	rec, err := provision.GetRecord(context.Background(), fx.store, fxAccount, fxApp, fxHost)
	if err != nil {
		fx.t.Fatal(err)
	}
	return *rec
}

// request is the envelope the console would sign for the current record.
func (fx *fixture) request(sqlText, at string) Request {
	fx.t.Helper()
	rec := fx.record()
	inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
	if err != nil {
		fx.t.Fatal(err)
	}
	id, err := MintID()
	if err != nil {
		fx.t.Fatal(err)
	}
	return Request{ID: id, Account: fxAccount, App: fxApp, Host: fxHost, Incarnation: inc, Region: "uk",
		Op: OpQuery, SQL: sqlText, At: at, ReplyKey: EncodeReplyKey(fx.reply.PublicKey()),
		By: "member@example.com", IssuedAt: fx.now.Format(time.RFC3339)}
}

// submit is the console's two creates: the signed request, then the queue
// entry.
func (fx *fixture) submit(req Request) QueueEntry {
	fx.t.Helper()
	wire, err := Sign(release.RootSigner(fx.rootPriv), req)
	if err != nil {
		fx.t.Fatal(err)
	}
	if _, err := fx.store.Put(context.Background(), RequestKey(req.ID), wire, ""); err != nil {
		fx.t.Fatal(err)
	}
	issued, _ := req.Issued()
	key := QueueKey(req.Region, issued, req.ID)
	if _, err := fx.store.Put(context.Background(), key, nil, ""); err != nil {
		fx.t.Fatal(err)
	}
	e, _ := ParseQueueKey(key)
	return e
}

// agent is a capable agent in uk over the fixture's store and replica.
// Its probe is Task 12's: it starts capable.
func (fx *fixture) agent(box string, mut ...func(*AgentConfig, *inProcWorkers)) (*Agent, *inProcWorkers) {
	w := &inProcWorkers{t: fx.t}
	cfg := AgentConfig{
		Store: fx.store, Region: "uk", Box: box, Boot: "boot-" + box, Pub: fx.rootPub, HibernatePrefix: "instances",
		OpenReplica: func(_ context.Context, prefix string) (dbdoor.ReplicaReader, error) {
			if prefix != "instances/"+fxHost {
				fx.t.Errorf("the agent asked for prefix %q", prefix)
			}
			return lsfile.NewReplicaClient(fx.replica), nil
		},
		Units: w.units, Free: plentyFree, Logf: fx.t.Logf,
		Jitter: func() time.Duration { return 0 },
	}
	for _, m := range mut {
		m(&cfg, w)
	}
	cfg.DialWorker = w.dial
	a := NewAgent(cfg)
	a.probeOK.Store(true)
	return a, w
}

// chunkKeys lists a read's chunk objects.
func (fx *fixture) chunkKeys(id string) []string {
	keys, err := fx.store.List(context.Background(), OutPrefix(id))
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		fx.t.Fatal(err)
	}
	sort.Strings(keys)
	return keys
}

// answer reads a read's result and opens its chunks as the CLI would.
func (fx *fixture) answer(id string) ([]byte, Summary, Result) {
	fx.t.Helper()
	raw, _, err := fx.store.Get(context.Background(), ResultKey(id))
	if err != nil {
		fx.t.Fatalf("no result for %s: %v", id, err)
	}
	var r Result
	if err := json.Unmarshal(raw, &r); err != nil || r.Valid() != nil {
		fx.t.Fatalf("result %s: %v", raw, err)
	}
	s, err := OpenSummary(fx.reply, id, r.Sealed)
	if err != nil {
		fx.t.Fatal(err)
	}
	var out bytes.Buffer
	for seq := int64(1); seq <= r.LastSeq; seq++ {
		sealed, _, err := fx.store.Get(context.Background(), ChunkKey(id, seq))
		if err != nil {
			fx.t.Fatalf("chunk %d: %v", seq, err)
		}
		plain, err := OpenChunk(fx.reply, id, seq, StreamRows, seq == r.LastSeq, sealed)
		if err != nil {
			fx.t.Fatalf("chunk %d: %v", seq, err)
		}
		out.Write(plain)
	}
	sum := sha256.Sum256(out.Bytes())
	if r.State == StateDone && (int64(out.Len()) != s.Bytes || hex.EncodeToString(sum[:]) != s.SHA256) {
		fx.t.Fatalf("the answer does not add up: %d bytes, summary %+v", out.Len(), s)
	}
	return out.Bytes(), s, r
}