package operator

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	sharedseal "amadan.net/oficina/home/handoff/seal"
	"amadan.net/oficina/home/internal/home/directorydata"
	family "amadan.net/rastrillo/rastrillo/crypto"
	rdb "amadan.net/rastrillo/rastrillo/db"
	"amadan.net/rastrillo/rastrillo/migrate"
	_ "modernc.org/sqlite"
)

type engineFixture struct {
	engine *Engine
	db     *sql.DB
	raw    []byte
	now    *time.Time
	path   string
	key    *family.Keypair
	auth   Authorization
}

func newEngineFixture(t *testing.T) engineFixture {
	t.Helper()
	k, v, a := fixture(t)
	path := filepath.Join(t.TempDir(), "operator.db")
	handle, e := rdb.Open(path, nil)
	if e != nil {
		t.Fatal(e)
	}
	t.Cleanup(func() { handle.Close() })
	db := handle.Writer()
	db.SetMaxOpenConns(4)
	if _, e = db.Exec("PRAGMA journal_mode=WAL"); e != nil {
		t.Fatal(e)
	}
	if _, e = migrate.Apply(context.Background(), handle, migrate.Merge(directorydata.Schema, directorydata.RosterSchema, directorydata.OperatorSchema, DirectorySchema)); e != nil {
		t.Fatal(e)
	}
	if _, e = db.Exec("CREATE TABLE test_items(id INTEGER PRIMARY KEY)"); e != nil {
		t.Fatal(e)
	}
	sealer, e := sharedseal.New(bytes.Repeat([]byte{6}, 32), "oficina/operator")
	if e != nil {
		t.Fatal(e)
	}
	now := time.Unix(a.Issued, 0)
	engine, e := NewEngine(db, v, sealer, bytes.Repeat([]byte{7}, 32), func() time.Time { return now })
	if e != nil {
		t.Fatal(e)
	}
	return engineFixture{engine, db, signed(t, k, a), &now, path, k, a}
}

var privateResult = []byte(`{"bearer":"synthetic-private-bearer-keep-out-of-db","receipt":"synthetic-private-proof"}`)

func insertOne(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
	_, e := c.ExecContext(ctx, "INSERT INTO test_items VALUES(1)")
	if e != nil {
		return nil, e
	}
	return privateResult, nil
}
func countRows(t *testing.T, db *sql.DB, table string) int {
	t.Helper()
	var n int
	if e := db.QueryRow("SELECT count(*) FROM " + table).Scan(&n); e != nil {
		t.Fatal(e)
	}
	return n
}
func TestEngineCommitsOnceAndRecoversExpiredResultPrivately(t *testing.T) {
	f := newEngineFixture(t)
	out, e := f.engine.Execute(context.Background(), f.raw, insertOne)
	if e != nil || !bytes.Equal(out, privateResult) {
		t.Fatal(e)
	}
	*f.now = f.now.Add(301 * time.Second)
	out, e = f.engine.Execute(context.Background(), f.raw, func(context.Context, *sql.Conn, Request) ([]byte, error) {
		t.Fatal("replay executed mutation")
		return nil, nil
	})
	if e != nil || !bytes.Equal(out, privateResult) {
		t.Fatal(e)
	}
	if countRows(t, f.db, "test_items") != 1 || countRows(t, f.db, "directory_operator_results") != 1 {
		t.Fatal("non-atomic replay")
	}
	for _, suffix := range []string{"", "-wal"} {
		b, e := os.ReadFile(f.path + suffix)
		if e != nil && !errors.Is(e, os.ErrNotExist) {
			t.Fatal(e)
		}
		for _, secret := range [][]byte{privateResult, []byte("synthetic-private-bearer"), []byte("synthetic-private-proof")} {
			if bytes.Contains(b, secret) {
				t.Fatal("private result at rest")
			}
		}
	}
	if _, e = f.db.Exec("DELETE FROM directory_operator_state"); e != nil {
		t.Fatal(e)
	}
	if _, e = f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrRefused {
		t.Fatal("missing authority accepted", e)
	}
}
func TestEngineFailureRollsBackMutationReceiptAndSequence(t *testing.T) {
	for _, mode := range []string{"mutation", "expiry", "seal"} {
		t.Run(mode, func(t *testing.T) {
			f := newEngineFixture(t)
			if mode == "seal" {
				f.engine.sealer = failingSeal{}
			}
			_, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
				if _, e := insertOne(ctx, c, r); e != nil {
					return nil, e
				}
				switch mode {
				case "mutation":
					return nil, errors.New("private@example.test")
				case "expiry":
					*f.now = f.now.Add(301 * time.Second)
				}
				return privateResult, nil
			})
			if e != ErrRefused && e != ErrUnavailable {
				t.Fatal("unexpected error", e)
			}
			for _, table := range []string{"test_items", "directory_operator_results", "directory_operator_state"} {
				if countRows(t, f.db, table) != 0 {
					t.Fatal("partial mutation", table)
				}
			}
		})
	}
}

type failingSeal struct{}

func (failingSeal) Seal(string, []byte) ([]byte, error) {
	return nil, errors.New("private seal failure")
}
func (failingSeal) Open(string, []byte) ([]byte, error) {
	return nil, errors.New("private seal failure")
}
func TestEngineConcurrentRetryHasOneMutation(t *testing.T) {
	f := newEngineFixture(t)
	var called atomic.Int32
	var wg sync.WaitGroup
	errs := make(chan error, 8)
	for i := 0; i < 8; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
				called.Add(1)
				return insertOne(ctx, c, r)
			})
			errs <- e
		}()
	}
	wg.Wait()
	close(errs)
	for e := range errs {
		if e != nil {
			t.Fatal(e)
		}
	}
	if called.Load() != 1 {
		t.Fatal("duplicate side effects", called.Load())
	}
}
func TestEngineForeignKeyEnforcementUsesMutationConnection(t *testing.T) {
	f := newEngineFixture(t)
	if _, e := f.db.Exec("CREATE TABLE parent(id INTEGER PRIMARY KEY); CREATE TABLE child(parent INTEGER REFERENCES parent(id))"); e != nil {
		t.Fatal(e)
	}
	_, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
		var fk int
		if e := c.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&fk); e != nil || fk != 1 {
			t.Fatal("foreign keys not on actual connection")
		}
		_, e := c.ExecContext(ctx, "INSERT INTO child VALUES(999)")
		return privateResult, e
	})
	if e != ErrUnavailable || countRows(t, f.db, "child") != 0 || countRows(t, f.db, "directory_operator_results") != 0 {
		t.Fatal("foreign key violation escaped", e)
	}
}
func TestEngineRefusesExpiredNewAndCorruptExistingResult(t *testing.T) {
	f := newEngineFixture(t)
	*f.now = f.now.Add(300 * time.Second)
	if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrRefused {
		t.Fatal("expired new")
	}
	if countRows(t, f.db, "test_items") != 0 {
		t.Fatal("expired side effect")
	}
	*f.now = f.now.Add(-300 * time.Second)
	if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != nil {
		t.Fatal(e)
	}
	if _, e := f.db.Exec("UPDATE directory_operator_results SET sealed_result=zeroblob(29)"); e != nil {
		t.Fatal(e)
	}
	if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrUnavailable {
		t.Fatal("corrupt result", e)
	}
}

func TestEngineCommandSchemaCheckRunsInsideTransactionBeforeReplayOrMutation(t *testing.T) {
	f := newEngineFixture(t)
	ctx := t.Context()
	var checked *sql.Conn
	f.engine.checkSchema = func(ctx context.Context, conn *sql.Conn) error {
		checked = conn
		if _, err := conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err == nil {
			t.Fatal("schema check outside transaction")
		}
		return nil
	}
	mutate := func(ctx context.Context, conn *sql.Conn, r Request) ([]byte, error) {
		if conn != checked {
			t.Fatal("schema checked a different connection")
		}
		return insertOne(ctx, conn, r)
	}
	if _, err := f.engine.Execute(ctx, f.raw, mutate); err != nil {
		t.Fatal(err)
	}
	f.engine.checkSchema = func(context.Context, *sql.Conn) error { return ErrRefused }
	if out, err := f.engine.Execute(ctx, f.raw, mutate); err != ErrRefused || len(out) != 0 {
		t.Fatal("replay bypassed schema refusal", err)
	}
	if countRows(t, f.db, "test_items") != 1 || countRows(t, f.db, "directory_operator_results") != 1 {
		t.Fatal("schema refusal changed committed data")
	}
	fresh := newEngineFixture(t)
	fresh.engine.checkSchema = func(ctx context.Context, conn *sql.Conn) error {
		if _, err := conn.ExecContext(ctx, "INSERT INTO test_items VALUES(2)"); err != nil {
			t.Fatal(err)
		}
		return ErrUnavailable
	}
	if out, err := fresh.engine.Execute(ctx, fresh.raw, func(context.Context, *sql.Conn, Request) ([]byte, error) {
		t.Fatal("mutation ran after failed schema check")
		return nil, nil
	}); err != ErrUnavailable || len(out) != 0 {
		t.Fatal("schema failure accepted", err)
	}
	if countRows(t, fresh.db, "test_items") != 0 || countRows(t, fresh.db, "directory_operator_state") != 0 || countRows(t, fresh.db, "directory_operator_results") != 0 {
		t.Fatal("failed preflight transaction survived")
	}
}