// Package dbreadstest runs the box's half of a forwarded read inside a test
// process: the real dbreads.Agent and the real dbreads.ServeWorker, joined by
// net.Pipe instead of the installed socket. Test support only. It is not
// acceptance for the units (internal/host's root-runner tests are), and
// nothing outside a _test.go file may import it.
package dbreadstest

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

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

type Options struct {
	Store           release.Store
	Region          string
	Box             string
	Pub             ed25519.PublicKey
	HibernatePrefix string
	OpenReplica     func(ctx context.Context, prefix string) (dbdoor.ReplicaReader, error)
	MaxRestored     int64
	// Worker's Scratch, Exit and Self are filled in when empty.
	Worker dbreads.WorkerOptions
}

type Box struct {
	Agent  *dbreads.Agent
	Dials  atomic.Int64
	live   atomic.Int64
	worker dbreads.WorkerOptions
}

// Start builds a capable in-process box: its probe passes against an
// in-process worker reporting a sandboxed view.
func Start(t testing.TB, o Options) *Box {
	t.Helper()
	b := &Box{worker: o.Worker}
	if b.worker.Scratch == "" {
		b.worker.Scratch = t.TempDir()
	}
	if b.worker.Exit == nil {
		b.worker.Exit = func(int) {}
	}
	if b.worker.Self == nil {
		b.worker.Self = func() dbreads.HelloReply {
			return dbreads.HelloReply{V: dbreads.ProtocolVersion, Interfaces: []string{"lo"}}
		}
	}
	b.Agent = dbreads.NewAgent(dbreads.AgentConfig{
		Store: o.Store, Region: o.Region, Box: o.Box, Boot: "boot-" + o.Box, Pub: o.Pub,
		HibernatePrefix: o.HibernatePrefix, OpenReplica: o.OpenReplica,
		DialWorker: b.dial, Units: b.units,
		Free:        func() (uint64, uint64, error) { return 100 << 30, 200 << 30, nil },
		MaxRestored: o.MaxRestored, Jitter: func() time.Duration { return 0 }, Logf: t.Logf,
	})
	b.Agent.Probe(context.Background())
	if !b.Agent.Capable() {
		t.Fatal("dbreadstest: the in-process box failed its probe")
	}
	return b
}

func (b *Box) dial(context.Context) (net.Conn, error) {
	agentEnd, workerEnd := net.Pipe()
	b.Dials.Add(1)
	b.live.Add(1)
	go func() {
		defer b.live.Add(-1)
		dbreads.ServeWorker(workerEnd, b.worker)
		workerEnd.Close()
	}()
	return agentEnd, nil
}

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

// Serve passes until the read's result exists.
func (b *Box) Serve(t testing.TB, store release.Store, id string, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		b.Agent.Pass(context.Background())
		b.Agent.Wait()
		if _, _, err := store.Get(context.Background(), dbreads.ResultKey(id)); err == nil {
			return
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("dbreadstest: no result for %s within %s", id, timeout)
}

// Submit is the console's two creates for req, signed with root: for tests
// that have no console.
func Submit(t testing.TB, store release.Store, root ed25519.PrivateKey, req dbreads.Request) {
	t.Helper()
	wire, err := dbreads.Sign(release.RootSigner(root), req)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.Put(context.Background(), dbreads.RequestKey(req.ID), wire, ""); err != nil {
		t.Fatal(err)
	}
	issued, err := req.Issued()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.Put(context.Background(), dbreads.QueueKey(req.Region, issued, req.ID), nil, ""); err != nil {
		t.Fatal(err)
	}
}

// Answer reads a read's result and opens its chunks with the CLI's key,
// checking the count and hash a done read's summary carries.
func Answer(t testing.TB, store release.Store, id string, priv *ecdh.PrivateKey, stream string) ([]byte, dbreads.Summary, dbreads.Result) {
	t.Helper()
	raw, _, err := store.Get(context.Background(), dbreads.ResultKey(id))
	if err != nil {
		t.Fatalf("dbreadstest: no result for %s: %v", id, err)
	}
	var r dbreads.Result
	if err := json.Unmarshal(raw, &r); err != nil || r.Valid() != nil {
		t.Fatalf("dbreadstest: result %s: %v", raw, err)
	}
	s, err := dbreads.OpenSummary(priv, id, r.Sealed)
	if err != nil {
		t.Fatal(err)
	}
	var out bytes.Buffer
	for seq := int64(1); seq <= r.LastSeq; seq++ {
		sealed, _, err := store.Get(context.Background(), dbreads.ChunkKey(id, seq))
		if err != nil {
			t.Fatalf("dbreadstest: chunk %d: %v", seq, err)
		}
		plain, err := dbreads.OpenChunk(priv, id, seq, stream, seq == r.LastSeq, sealed)
		if err != nil {
			t.Fatalf("dbreadstest: chunk %d: %v", seq, err)
		}
		out.Write(plain)
	}
	if r.State == dbreads.StateDone {
		sum := sha256.Sum256(out.Bytes())
		if int64(out.Len()) != s.Bytes || hex.EncodeToString(sum[:]) != s.SHA256 {
			t.Fatalf("dbreadstest: the answer does not add up: %d bytes, summary %+v", out.Len(), s)
		}
	}
	return out.Bytes(), s, r
}