package console

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"github.com/carlosframework/platform/internal/hostops"
	"runtime"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

// plan9SealRetainedStore is a retained *Store from a batch callback writing
// after the callback returned.
func plan9SealRetainedStore(v *Store, out chan<- error) {
	_, err := v.db.Exec(`UPDATE audit SET action='late-store' WHERE account_id=7 AND app='blog'`)
	out <- err
}

// plan9SealQueuedDriver is an operation that already passed the view's public
// check and is queued on the held driver connection.
func plan9SealQueuedDriver(d *claimBoundDB, out chan<- error) {
	_, err := d.conn.ExecContext(context.Background(), `UPDATE audit SET action='late-driver' WHERE account_id=7 AND app='blog'`)
	out <- err
}

// plan9SealQueuedStmtExec and plan9SealQueuedStmtQuery run an escaped prepared
// statement: one that was never registered with the view, as if the statement
// had been prepared before the callback returned.
func plan9SealQueuedStmtExec(st *sql.Stmt, out chan<- error) {
	_, err := st.ExecContext(context.Background())
	out <- err
}

func plan9SealQueuedStmtQuery(st *sql.Stmt, out chan<- error) {
	rows, err := st.QueryContext(context.Background())
	if err == nil {
		for rows.Next() {
		}
		err = rows.Err()
		rows.Close()
	}
	out <- err
}

// waitPlan9SealQueued returns once the goroutine running fn has finished
// (done has a value) or is parked on the database/sql driver-connection lock,
// which the batch commit holds.
func waitPlan9SealQueued(t *testing.T, fn string, done chan error) {
	t.Helper()
	deadline := time.Now().Add(10 * time.Second)
	buf := make([]byte, 1<<20)
	for time.Now().Before(deadline) {
		if len(done) > 0 {
			return
		}
		n := runtime.Stack(buf, true)
		for _, g := range strings.Split(string(buf[:n]), "\n\n") {
			onDriver := strings.Contains(g, "database/sql.withLock(") || strings.Contains(g, "database/sql.resultFromStatement(") || strings.Contains(g, "database/sql.rowsiFromStatement(")
			if strings.Contains(g, "console."+fn+"(") && onDriver && strings.Contains(g, "sync.(*Mutex).lockSlow(") {
				return
			}
		}
		runtime.Gosched()
	}
	t.Fatalf("%s neither finished nor queued on the driver connection", fn)
}

// TestPlan9SQLBatchSealsRetainedWriteQueuedAcrossCommit checks the view fast
// path and queues driver and prepared-statement writes across COMMIT. None
// may run under the transaction permit once its commit begins.
func TestPlan9SQLBatchSealsRetainedWriteQueuedAcrossCommit(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	r := receiptFixture(t, s)
	var view *Store
	var returned atomic.Bool
	store, driver := make(chan error, 1), make(chan error, 1)
	stmtExec, stmtQuery := make(chan error, 1), make(chan error, 1)
	var escapedExec, escapedQuery *sql.Stmt
	local := func(ctx context.Context, _ claimSQLScope) error {
		if !returned.CompareAndSwap(true, false) {
			return ctx.Err()
		}
		// The commit authorization: endBatch holds the driver connection.
		go plan9SealRetainedStore(view, store)
		waitPlan9SealQueued(t, "plan9SealRetainedStore", store)
		go plan9SealQueuedDriver(view.db.(*claimBoundDB), driver)
		waitPlan9SealQueued(t, "plan9SealQueuedDriver", driver)
		go plan9SealQueuedStmtExec(escapedExec, stmtExec)
		waitPlan9SealQueued(t, "plan9SealQueuedStmtExec", stmtExec)
		go plan9SealQueuedStmtQuery(escapedQuery, stmtQuery)
		waitPlan9SealQueued(t, "plan9SealQueuedStmtQuery", stmtQuery)
		return nil
	}
	err := s.withPreparedClaimSQL(gateCtx(t), []claimSQLScope{gateScope}, allowAllClaimSQL, local, []hostops.SQLReceiptScope{r}, true, func(v *Store) error {
		view = v
		// Prepared on the raw sql.Conn, bypassing the view's registration, so
		// these statements are past it and only the driver gate can stop them.
		d := v.db.(*claimBoundDB)
		var perr error
		if escapedExec, perr = d.conn.PrepareContext(context.Background(), `UPDATE audit SET action='late-stmt-exec' WHERE account_id=7 AND app='blog'`); perr != nil {
			return perr
		}
		if escapedQuery, perr = d.conn.PrepareContext(context.Background(), `UPDATE audit SET action='late-stmt-query' WHERE account_id=7 AND app='blog' RETURNING id`); perr != nil {
			return perr
		}
		_, err := v.db.Exec(`UPDATE audit SET action='batch' WHERE account_id=7 AND app='blog'`)
		returned.Store(true)
		return err
	})
	if err != nil {
		t.Fatal(err)
	}
	if escapedExec != nil {
		defer escapedExec.Close()
	}
	if escapedQuery != nil {
		defer escapedQuery.Close()
	}
	if returned.Load() {
		t.Fatal("commit authorization never ran the local check")
	}
	for name, ch := range map[string]chan error{
		"retained store (view fast path)": store,
		"queued driver":                   driver,
		"escaped stmt Exec":               stmtExec,
		"escaped stmt QueryContext":       stmtQuery,
	} {
		select {
		case werr := <-ch:
			if !errors.Is(werr, ErrClaimSQLRevoked) {
				t.Errorf("%s write across COMMIT: err=%v, want ErrClaimSQLRevoked", name, werr)
			}
		case <-time.After(10 * time.Second):
			t.Fatalf("%s write never finished", name)
		}
	}
	if n := countWhere(t, s, "audit", "action LIKE 'late-%'"); n != 0 {
		t.Errorf("%d retained writes landed outside the receipted transaction", n)
	}
	if countWhere(t, s, "audit", "action='batch'") != 1 {
		t.Error("batch write did not commit")
	}
	var state string
	if err := s.rawBoot.QueryRow(`SELECT state FROM claim_sql_receipts WHERE effect=?`, r.Effect).Scan(&state); err != nil || state != "committed" {
		t.Fatalf("receipt=%q err=%v", state, err)
	}
}

// TestPlan9SQLEndBatchLeftoverTxRefusesCommit drives the driver directly: a
// transaction that registered after the handles were joined forces the
// requested commit to roll back, and endBatch must say so instead of
// reporting success with nothing committed.
func TestPlan9SQLEndBatchLeftoverTxRefusesCommit(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	r := receiptFixture(t, s)
	ctx := gateCtx(t)
	conn, err := s.rawPool.Conn(ctx)
	if err != nil {
		t.Fatal(err)
	}
	defer conn.Close()
	sqids := make([]string, 1)
	if s.gate.needSqid() {
		if err := conn.QueryRowContext(ctx, `SELECT sqid FROM accounts WHERE id = ?`, gateScope.AccountID).Scan(&sqids[0]); err != nil {
			t.Fatal(err)
		}
	}
	scopes := []claimSQLScope{gateScope}
	b := &claimBinding{ctx: ctx, scopes: scopes, check: func(ctx context.Context, _ claimSQLScope) error { return ctx.Err() }, deadline: time.Now().Add(hostops.SQLTransactionLifetime), batch: true, receipts: []hostops.SQLReceiptScope{r}, dead: make([]bool, 1)}
	var endErr error
	err = conn.Raw(func(dc any) error {
		c := dc.(*claimConn)
		if err := c.bindScopes(ctx, b, sqids); err != nil {
			return err
		}
		defer func() { _ = c.unbind(b) }()
		if err := c.beginBatch(ctx, b); err != nil {
			return err
		}
		if _, err := c.ExecContext(ctx, `UPDATE audit SET action='batch' WHERE account_id=7 AND app='blog'`, nil); err != nil {
			return err
		}
		if _, err := c.BeginTx(ctx, driver.TxOptions{}); err != nil {
			return err
		}
		endErr = c.endBatch(b, true)
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := conn.Close(); err != nil {
		t.Fatal(err)
	}
	if endErr == nil {
		t.Fatal("endBatch reported success although a leftover transaction forced the commit to roll back")
	}
	if !errors.Is(endErr, ErrClaimSQLScope) {
		t.Errorf("endBatch err=%v, want ErrClaimSQLScope", endErr)
	}
	var n int
	if err := s.rawBoot.QueryRow(`SELECT count(*) FROM claim_sql_receipts WHERE effect=?`, r.Effect).Scan(&n); err != nil || n != 0 {
		t.Errorf("receipt rows=%d err=%v, want none", n, err)
	}
	if got := countWhere(t, s, "audit", "action='batch'"); got != 0 {
		t.Errorf("batch write present (%d rows) after forced rollback", got)
	}
}

// Seal while a handle has left the driver but has not entered the view's
// registry. It must be closed by the registering goroutine, not escape the
// closeHandles snapshot. Holding mu makes this boundary deterministic.
func TestPlan9SQLSealDuringHandleRegistration(t *testing.T) {
	for _, method := range []string{"PrepareContext", "BeginTx"} {
		t.Run(method, func(t *testing.T) {
			s := openGatedStore(t)
			seedGateApp(t, s)
			r := receiptFixture(t, s)
			err := s.withPreparedClaimSQL(gateCtx(t), []claimSQLScope{gateScope}, allowAllClaimSQL, allowAllClaimSQL, []hostops.SQLReceiptScope{r}, true, func(v *Store) error {
				d := v.db.(*claimBoundDB)
				d.mu.Lock()
				locked := true
				defer func() {
					if locked {
						d.mu.Unlock()
					}
				}()
				done := make(chan error, 1)
				go func() {
					if method == "PrepareContext" {
						st, err := d.PrepareContext(context.Background(), "SELECT 1")
						if st != nil {
							_ = st.Close()
						}
						done <- err
					} else {
						tx, err := d.BeginTx(context.Background(), nil)
						if tx != nil {
							_ = tx.Rollback()
						}
						done <- err
					}
				}()
				deadline := time.Now().Add(10 * time.Second)
				buf := make([]byte, 1<<20)
				waiting := false
				for time.Now().Before(deadline) && !waiting {
					n := runtime.Stack(buf, true)
					for _, g := range strings.Split(string(buf[:n]), "\n\n") {
						if strings.Contains(g, "console.(*claimBoundDB)."+method+"(") && strings.Contains(g, "sync.(*Mutex).lockSlow(") {
							waiting = true
							break
						}
					}
					runtime.Gosched()
				}
				if !waiting {
					t.Fatal("handle never reached registration")
				}
				d.b.sealed.Store(true)
				d.mu.Unlock()
				locked = false
				select {
				case err := <-done:
					if !errors.Is(err, ErrClaimSQLRevoked) {
						t.Errorf("late registration = %v, want revoked", err)
					}
				case <-time.After(10 * time.Second):
					t.Fatal("late registration did not join")
				}
				return nil
			})
			if err != nil {
				t.Fatal(err)
			}
		})
	}
}