package dbreads

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"net"
	"os"
	"strings"
	"sync/atomic"
	"testing"
	"time"

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

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
	"github.com/carlosframework/platform/internal/dbquery"
)

// workerPipe starts ServeWorker on one end of a pipe and returns the other.
// Not acceptance: the units are Task 15's.
func workerPipe(t *testing.T, opt WorkerOptions) (net.Conn, <-chan error) {
	t.Helper()
	agent, worker := net.Pipe()
	if opt.Scratch == "" {
		opt.Scratch = t.TempDir()
	}
	if opt.Exit == nil {
		opt.Exit = func(int) { t.Error("the worker asked to exit the process") }
	}
	done := make(chan error, 1)
	go func() {
		err := ServeWorker(worker, opt)
		worker.Close()
		done <- err
	}()
	t.Cleanup(func() { agent.Close() })
	return agent, done
}

// sendPlan sends a header and the planned files of dir's replica.
func sendPlan(t *testing.T, conn net.Conn, dir string) dbdoor.RestorePlan {
	t.Helper()
	rd := lsfile.NewReplicaClient(dir)
	plan, err := dbdoor.PlanFrom(context.Background(), rd, time.Time{})
	if err != nil {
		t.Fatal(err)
	}
	if err := writeJSONFrame(conn, frameHeader, planHeader{V: ProtocolVersion, Op: OpQuery, Files: plan.Planned, Target: plan.TXID, MaxRestored: DefaultMaxRestored}); err != nil {
		t.Fatal(err)
	}
	for _, f := range plan.Planned {
		rc, err := rd.OpenLTXFile(context.Background(), f.Level, ltx.TXID(f.MinTXID), ltx.TXID(f.MaxTXID), 0, 0)
		if err != nil {
			t.Fatal(err)
		}
		body, _ := io.ReadAll(rc)
		rc.Close()
		if err := writeFrame(conn, frameFile, body); err != nil {
			t.Fatal(err)
		}
	}
	return plan
}

func readTyped(t *testing.T, conn net.Conn, want byte, v any) {
	t.Helper()
	typ, p, err := readFrame(conn)
	if err != nil || typ != want {
		t.Fatalf("frame %q (%s), want %q: %v", typ, p, want, err)
	}
	if v != nil {
		if err := json.Unmarshal(p, v); err != nil {
			t.Fatal(err)
		}
	}
}

func TestWorkerAnswersHelloWithItsSelfView(t *testing.T) {
	t.Parallel()
	conn, done := workerPipe(t, WorkerOptions{Self: func() HelloReply {
		return HelloReply{V: ProtocolVersion, Interfaces: []string{"lo"}}
	}})
	if err := writeJSONFrame(conn, frameHello, hello{V: ProtocolVersion}); err != nil {
		t.Fatal(err)
	}
	var h HelloReply
	readTyped(t, conn, frameHelloReply, &h)
	if h.V != ProtocolVersion || len(h.Interfaces) != 1 {
		t.Fatalf("hello = %+v", h)
	}
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

// The default self-view reads /proc/self/net/dev, which needs no socket
// (plan review finding 1). Mutation: SelfView back on net.Interfaces makes
// TestRootDBReadReadiness's baseline red inside the installed unit.
func TestParseNetDev(t *testing.T) {
	t.Parallel()
	only := "Inter-|   Receive\n face |bytes\n    lo: 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0\n"
	if got := parseNetDev(only); len(got) != 1 || got[0] != "lo" {
		t.Fatalf("%q", got)
	}
	two := only + "  eth0: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16\n"
	if got := parseNetDev(two); len(got) != 2 || got[1] != "eth0" {
		t.Fatalf("%q", got)
	}
	if got := SelfView(); len(got.Errors) != 0 && !strings.Contains(strings.Join(got.Errors, " "), "/proc") {
		t.Fatalf("SelfView on this machine: %+v", got)
	}
}

func TestWorkerRestoresThenRunsOnlyWhenTold(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 3)
	conn, done := workerPipe(t, WorkerOptions{})
	plan := sendPlan(t, conn, dir)
	var ready readyMsg
	readTyped(t, conn, frameReady, &ready)
	if ready.TXID != plan.TXID {
		t.Fatalf("ready at %d, planned %d", ready.TXID, plan.TXID)
	}
	if err := writeJSONFrame(conn, frameRun, runMsg{SQL: "SELECT count(*) AS n FROM t"}); err != nil {
		t.Fatal(err)
	}
	var out bytes.Buffer
	for {
		typ, p, err := readFrame(conn)
		if err != nil {
			t.Fatal(err)
		}
		if typ == frameOutput {
			out.Write(p)
			continue
		}
		var end endMsg
		if typ != frameEnd || json.Unmarshal(p, &end) != nil || end.Code != "" {
			t.Fatalf("end %q %s", typ, p)
		}
		break
	}
	if out.String() != "n\n30\n" {
		t.Fatalf("output %q", out.String())
	}
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

func TestWorkerStopRunsNothing(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 1)
	conn, done := workerPipe(t, WorkerOptions{})
	sendPlan(t, conn, dir)
	readTyped(t, conn, frameReady, nil)
	if err := writeFrame(conn, frameStop, nil); err != nil {
		t.Fatal(err)
	}
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

// The worker refuses a file larger than its header said (spec §8.3).
func TestWorkerRefusesAFileLargerThanItsHeader(t *testing.T) {
	t.Parallel()
	conn, done := workerPipe(t, WorkerOptions{})
	files := []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 4}}
	if err := writeJSONFrame(conn, frameHeader, planHeader{V: ProtocolVersion, Op: OpQuery, Files: files, Target: 1, MaxRestored: DefaultMaxRestored}); err != nil {
		t.Fatal(err)
	}
	go writeFrame(conn, frameFile, []byte("12345"))
	var end endMsg
	readTyped(t, conn, frameEnd, &end)
	if end.Code != CodeCheckFailed {
		t.Fatalf("end = %+v", end)
	}
	<-done
}

func TestWorkerSQLErrorIsSQLitesText(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 1)
	conn, done := workerPipe(t, WorkerOptions{})
	sendPlan(t, conn, dir)
	readTyped(t, conn, frameReady, nil)
	writeJSONFrame(conn, frameRun, runMsg{SQL: "SELECT * FROM nope"})
	var end endMsg
	readTyped(t, conn, frameEnd, &end)
	if end.Code != CodeSQLError || end.Error == "" {
		t.Fatalf("end = %+v", end)
	}
	if err := dbquery.CheckSQL("SELECT * FROM nope"); err != nil {
		t.Fatal(err)
	}
	<-done
}

func TestWorkerQueryTimeout(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 1)
	conn, done := workerPipe(t, WorkerOptions{QueryTimeout: 50 * time.Millisecond})
	sendPlan(t, conn, dir)
	readTyped(t, conn, frameReady, nil)
	writeJSONFrame(conn, frameRun, runMsg{SQL: "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c) SELECT count(*) FROM c"})
	for {
		typ, p, err := readFrame(conn)
		if err != nil {
			t.Fatal(err)
		}
		if typ == frameEnd {
			var end endMsg
			json.Unmarshal(p, &end)
			if end.Code != CodeTimeout {
				t.Fatalf("end = %+v", end)
			}
			break
		}
	}
	<-done
}

// The worker stops when its agent goes (spec §8.3, codex 8): EOF cancels the
// context, and a step that ignores the context is ended by the exit after
// the grace.
func TestWorkerExitsWhenTheAgentGoesMidRestore(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	dbdoortest.Replicate(t, dir, 1)
	var exited atomic.Int32
	hang := make(chan struct{})
	t.Cleanup(func() { close(hang) })
	conn, _ := workerPipe(t, WorkerOptions{
		Grace: 50 * time.Millisecond,
		Exit:  func(code int) { exited.Store(int32(code)) },
		Restore: func(context.Context, litestream.ReplicaClient, string, uint64) error {
			<-hang // a compaction that ignores its context
			return nil
		},
	})
	sendPlan(t, conn, dir)
	conn.Close()
	deadline := time.Now().Add(5 * time.Second)
	for exited.Load() == 0 && time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
	}
	if exited.Load() == 0 {
		t.Fatal("the worker outlived its agent past the grace")
	}
}

func TestWorkerCheckFailsAMissingFile(t *testing.T) {
	t.Parallel()
	conn, done := workerPipe(t, WorkerOptions{})
	if err := writeJSONFrame(conn, frameHeader, planHeader{V: ProtocolVersion, Op: OpQuery, Files: nil, Target: 9, MaxRestored: DefaultMaxRestored}); err != nil {
		t.Fatal(err)
	}
	var end endMsg
	readTyped(t, conn, frameEnd, &end)
	if end.Code != CodeCheckFailed {
		t.Fatalf("an empty plan = %+v", end)
	}
	<-done
}

// The worker bounds the bytes it will write across ALL files, not just each
// file (Astra round 1, finding 1): a header whose files sum past
// PlanBytesFactor times its restored-size cap, or whose cap is above the
// worker's own, is refused before anything is written. LimitFSIZE bounds
// one file only.
func TestWorkerRefusesAPlanOverItsTotal(t *testing.T) {
	t.Parallel()
	for name, h := range map[string]planHeader{
		"files past twice the cap": {V: ProtocolVersion, Op: OpQuery, Target: 3, MaxRestored: 1000,
			Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 800}, {Level: 0, MinTXID: 2, MaxTXID: 2, Size: 800}, {Level: 0, MinTXID: 3, MaxTXID: 3, Size: 800}}},
		"a cap above the worker's": {V: ProtocolVersion, Op: OpQuery, Target: 1, MaxRestored: 5000,
			Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: 10}}},
		"a negative size": {V: ProtocolVersion, Op: OpQuery, Target: 2, MaxRestored: 1000,
			Files: []dbdoor.LTXFile{{Level: 0, MinTXID: 1, MaxTXID: 1, Size: -5}, {Level: 0, MinTXID: 2, MaxTXID: 2, Size: 10}}},
	} {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			scratch := t.TempDir()
			conn, done := workerPipe(t, WorkerOptions{Scratch: scratch, MaxRestored: 1000})
			if err := writeJSONFrame(conn, frameHeader, h); err != nil {
				t.Fatal(err)
			}
			var end endMsg
			readTyped(t, conn, frameEnd, &end)
			if end.Code != CodeCheckFailed {
				t.Fatalf("end = %+v", end)
			}
			<-done
			if ents, _ := os.ReadDir(scratch); len(ents) != 0 {
				t.Fatalf("the worker wrote %d entries for a refused plan", len(ents))
			}
		})
	}
}