package console

import (
	"context"
	"database/sql"
	"errors"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/release"
	"path/filepath"
	"testing"
	"time"
)

func receiptFixture(t *testing.T, s *Store) hostops.SQLReceiptScope {
	t.Helper()
	id, err := s.sqlDatabaseID()
	if err != nil {
		t.Fatal(err)
	}
	return hostops.SQLReceiptScope{Database: id, Account: "bab", App: "blog", AppID: gateScope.AppID, Operation: "ingest", Effect: "ingest", Epoch: 1, Owner: hostops.RegistrationOwner{Console: "c", Enrollment: "e", Controller: "i"}}
}
func preparedBatch(t *testing.T, s *Store, r *hostops.SQLReceiptScope, fn func(*Store) error) error {
	t.Helper()
	return s.withPreparedClaimSQL(t.Context(), []claimSQLScope{gateScope}, allowAllClaimSQL, allowAllClaimSQL, []hostops.SQLReceiptScope{*r}, true, fn)
}
func TestPlan9SQLRecoveryBeforeBeginRejectsDelayedWriter(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	r := receiptFixture(t, s)
	if err := s.FenceSQL(t.Context(), r); err != nil {
		t.Fatal(err)
	}
	called := false
	err := preparedBatch(t, s, &r, func(v *Store) error {
		called = true
		_, err := v.db.Exec(`UPDATE audit SET action='late' WHERE account_id=7 AND app='blog'`)
		return err
	})
	if err == nil || called {
		t.Fatalf("fenced transaction ran: called=%v err=%v", called, err)
	}
	other := r
	other.Database = "other"
	if s.FenceSQL(t.Context(), other) == nil {
		t.Fatal("wrong database accepted")
	}
	if _, err = s.db.Exec(`DELETE FROM claim_sql_receipts`); !errors.Is(err, ErrClaimSQLForbidden) {
		t.Fatalf("public receipt write: %v", err)
	}
}
func TestPlan9SQLRecoveryWaitsForTransactionAndFencesReplay(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	r := receiptFixture(t, s)
	entered, release := make(chan struct{}), make(chan struct{})
	done := make(chan error, 1)
	go func() {
		done <- preparedBatch(t, s, &r, func(v *Store) error {
			_, err := v.db.Exec(`UPDATE audit SET action='joined' WHERE account_id=7 AND app='blog'`)
			close(entered)
			<-release
			return err
		})
	}()
	<-entered
	recovery := make(chan error, 1)
	go func() { recovery <- s.FenceSQL(t.Context(), r) }()
	select {
	case err := <-recovery:
		t.Fatalf("recovery crossed running transaction: %v", err)
	case <-time.After(30 * time.Millisecond):
	}
	close(release)
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	if err := <-recovery; err != nil {
		t.Fatal(err)
	}
	if countWhere(t, s, "audit", "action='joined'") == 0 {
		t.Fatal("committed SQL disappeared")
	}
	if err := preparedBatch(t, s, &r, func(*Store) error { return nil }); err == nil {
		t.Fatal("receipt replay accepted")
	}
}
func TestPlan9SQLBatchRollbackAndCommitCancellation(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	r := receiptFixture(t, s)
	failure := errors.New("last write failed")
	err := preparedBatch(t, s, &r, func(v *Store) error {
		tx, err := v.db.Begin()
		if err != nil {
			return err
		}
		defer tx.Rollback()
		if _, err = tx.Exec(`UPDATE audit SET action='rolled-back' WHERE account_id=7 AND app='blog'`); err != nil {
			return err
		}
		if err = tx.Commit(); err != nil {
			return err
		}
		return failure
	})
	if !errors.Is(err, failure) {
		t.Fatal(err)
	}
	if countWhere(t, s, "audit", "action='rolled-back'") != 0 {
		t.Fatal("nested commit escaped batch rollback")
	}
	// Simulate deadline passing immediately after the last statement.
	err = preparedBatch(t, s, &r, func(v *Store) error {
		_, err := v.db.Exec(`UPDATE audit SET action='expired' WHERE account_id=7 AND app='blog'`)
		v.bound.deadline = time.Now().Add(-time.Second)
		return err
	})
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("commit deadline: %v", err)
	}
	if countWhere(t, s, "audit", "action='expired'") != 0 {
		t.Fatal("expired commit landed")
	}
}
func TestPlan9SQLReceiptCommitsWithRows(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	r := receiptFixture(t, s)
	if err := preparedBatch(t, s, &r, func(v *Store) error {
		_, err := v.db.Exec(`UPDATE audit SET action='receipt' WHERE account_id=7 AND app='blog'`)
		return err
	}); err != nil {
		t.Fatal(err)
	}
	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=%s err=%v", state, err)
	}
	// Driver-only receipt door does not expose raw mutation to a Store callback.
}

type sqlNetworkCheck struct {
	release.Store
	s *Store
}

func (w *sqlNetworkCheck) Get(ctx context.Context, key string) ([]byte, string, error) {
	if w.s.rawPool.Stats().InUse != 0 {
		return nil, "", errors.New("network read under SQL connection")
	}
	return w.Store.Get(ctx, key)
}
func (w *sqlNetworkCheck) Put(ctx context.Context, key string, b []byte, tag string) (string, error) {
	if w.s.rawPool.Stats().InUse != 0 {
		return "", errors.New("network write under SQL connection")
	}
	return w.Store.Put(ctx, key, b, tag)
}
func TestPlan9SQLProductionAdapterNeverCallsNetworkUnderConnection(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	if _, err := s.db.Exec(`UPDATE accounts SET sqid='bab' WHERE id=7`); err != nil {
		t.Fatal(err)
	}
	raw := &sqlNetworkCheck{Store: release.NewDirStore(t.TempDir()), s: s}
	w := claimTestWriter(t, raw)
	seedClaimAdmission(t, raw, "bab", "blog", hostops.AppAdmission{V: 1, AppID: gateScope.AppID, Revision: 1, Epoch: 1, State: "open", Registrations: map[string]hostops.Registration{}, Effects: map[string]hostops.Effect{}})
	bucket := claimTestBoundary(t, raw, w)
	if err := s.bindClaimSQLMutations(bucket); err != nil {
		t.Fatal(err)
	}
	err := s.withAutoAppSQL(t.Context(), gateAcct, "blog", "test", func(v *Store) error {
		tx, err := v.db.Begin()
		if err != nil {
			return err
		}
		defer tx.Rollback()
		for _, value := range []string{"one", "two"} {
			if _, err = tx.Exec(`UPDATE audit SET action=? WHERE account_id=7 AND app='blog'`, value); err != nil {
				return err
			}
		}
		return tx.Commit()
	})
	if err != nil {
		t.Fatal(err)
	}
	a, err := hostops.ReadCanonicalAdmission(t.Context(), raw, "bab", "blog")
	if err != nil {
		t.Fatal(err)
	}
	if a.Value.State != "open" || a.Value.Op != nil || len(a.Value.Registrations) != 0 {
		t.Fatalf("ingest did not settle: %+v", a.Value)
	}
	if len(a.Value.Effects) != 1 {
		t.Fatalf("effects=%d want one batch", len(a.Value.Effects))
	}
}

func TestPlan9SQLCopiedDatabaseCannotProveOriginalTransaction(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	r := receiptFixture(t, s)
	copied := filepath.Join(t.TempDir(), "copied.db")
	if _, err := s.rawBoot.Exec(`VACUUM INTO ?`, copied); err != nil {
		t.Fatal(err)
	}
	other, err := OpenGated(copied)
	if err != nil {
		t.Fatal(err)
	}
	defer other.Close()
	if err = other.FenceSQL(t.Context(), r); !errors.Is(err, hostops.ErrIdentity) {
		t.Fatalf("backup claimed original lock: %v", err)
	}
}

func TestPlan9SQLBatchClosesRetainedRowsBeforeCommit(t *testing.T) {
	s := openGatedStore(t)
	seedGateApp(t, s)
	r := receiptFixture(t, s)
	ctx, cancel := context.WithTimeout(t.Context(), time.Second)
	defer cancel()
	var retained *sql.Rows
	err := s.withPreparedClaimSQL(ctx, []claimSQLScope{gateScope}, allowAllClaimSQL, allowAllClaimSQL, []hostops.SQLReceiptScope{r}, true, func(v *Store) error {
		var err error
		retained, err = v.db.Query(`SELECT action FROM audit`)
		return err
	})
	if err != nil {
		t.Fatal(err)
	}
	if retained.Next() {
		t.Fatal("retained rows survived callback")
	}
}