package pgstore

import (
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"strings"

	"net/url"
	"os"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"

	"textmachine/platform/internal/auth"
)

// The database-backed battery. It runs against TM_PLATFORM_TEST_DSN and skips loudly without it —
// `make check` names the skip, because an invisible skip reads as coverage.
//
// Each run gets its OWN database, created and dropped here: a test that leaves rows behind passes
// once and then lies.
func testDB(t testing.TB) (*Store, context.Context) {
	t.Helper()
	s, ctx, _ := testDBWithDSN(t)
	return s, ctx
}

func testDBWithDSN(t testing.TB) (*Store, context.Context, string) {
	t.Helper()
	admin := os.Getenv("TM_PLATFORM_TEST_DSN")
	if admin == "" {
		t.Skip("TM_PLATFORM_TEST_DSN not set: schema and session tests need a live Postgres")
	}
	ctx := t.Context()

	var suffix [6]byte
	if _, err := rand.Read(suffix[:]); err != nil {
		t.Fatal(err)
	}
	name := "tm_platform_test_" + hex.EncodeToString(suffix[:])

	adminConn, err := pgx.Connect(ctx, admin)
	if err != nil {
		t.Fatalf("connect: %v", err)
	}
	if _, err := adminConn.Exec(ctx, "create database "+pgx.Identifier{name}.Sanitize()); err != nil {
		adminConn.Close(ctx)
		t.Skipf("cannot create a scratch database (%v): grant CREATEDB or point the DSN at one", err)
	}
	t.Cleanup(func() {
		dropCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
		defer cancel()
		_, _ = adminConn.Exec(dropCtx, "drop database if exists "+pgx.Identifier{name}.Sanitize()+" with (force)")
		adminConn.Close(dropCtx)
	})

	dsn := swapDatabase(t, admin, name)
	if err := Migrate(ctx, dsn); err != nil {
		t.Fatalf("migrate: %v", err)
	}
	// Twice, because a rollout re-runs it on every replica.
	if err := Migrate(ctx, dsn); err != nil {
		t.Fatalf("second migrate must be a no-op: %v", err)
	}
	s, err := Open(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(s.Close)
	if err := s.Ping(ctx); err != nil {
		t.Fatal(err)
	}
	return s, ctx, dsn
}

func swapDatabase(t testing.TB, dsn, name string) string {
	t.Helper()
	u, err := url.Parse(dsn)
	if err != nil {
		t.Fatalf("TM_PLATFORM_TEST_DSN must be a URL: %v", err)
	}
	u.Path = "/" + name
	return u.String()
}

func TestSessionLifecycle(t *testing.T) {
	s, ctx := testDB(t)
	now := time.Now().UTC().Truncate(time.Millisecond)
	seedUser(t, s, ctx, "u1")

	token := auth.NewToken()
	digest := auth.Digest(token)
	if err := s.CreateSession(ctx, digest, "u1", now, time.Hour, 24*time.Hour); err != nil {
		t.Fatal(err)
	}

	got, err := s.Lookup(ctx, digest, now)
	if err != nil {
		t.Fatalf("lookup: %v", err)
	}
	if got.UserID != "u1" {
		t.Fatalf("session = %+v", got)
	}

	// Expiry is a clause of the query, not a caller's duty.
	if _, err := s.Lookup(ctx, digest, now.Add(2*time.Hour)); !errors.Is(err, auth.ErrNoSession) {
		t.Fatalf("expired idle window: %v", err)
	}
	// Sliding never outlives the absolute deadline.
	if err := s.Touch(ctx, digest, now.Add(30*time.Minute), 48*time.Hour); err != nil {
		t.Fatal(err)
	}
	slid, err := s.Lookup(ctx, digest, now.Add(30*time.Minute))
	if err != nil {
		t.Fatal(err)
	}
	if slid.IdleExpiresAt.After(slid.AbsoluteExpiresAt) {
		t.Fatalf("idle %s outlived absolute %s", slid.IdleExpiresAt, slid.AbsoluteExpiresAt)
	}

	if err := s.RevokeSession(ctx, digest, now); err != nil {
		t.Fatal(err)
	}
	if _, err := s.Lookup(ctx, digest, now); !errors.Is(err, auth.ErrNoSession) {
		t.Fatalf("revoked session still resolves: %v", err)
	}

	// The sweep also takes revoked rows: a revoked session is the one a compromised account most
	// wants gone, and it used to sit until its absolute expiry ninety days later.
	n, err := s.SweepSessions(ctx, now)
	if err != nil {
		t.Fatal(err)
	}
	if n != 1 {
		t.Fatalf("sweep removed %d rows, want the revoked one", n)
	}
}

// PD-1. What the database holds must be the HASH of the credential, and the property has to be
// checked by something other than the function that produces it: asserting through auth.Digest is
// self-consistent and survives a Digest that returns the plaintext. The oracle here is
// crypto/sha256 in the test, plus a search of the whole rendered row for the token itself.
// Mutation caught: `func Digest(t string) []byte { return []byte(t) }`.
func TestStoredCredentialIsAHashNotTheToken(t *testing.T) {
	s, ctx := testDB(t)
	seedUser(t, s, ctx, "u1")
	now := time.Now().UTC()

	token := auth.NewToken()
	if err := s.CreateSession(ctx, auth.Digest(token), "u1", now, time.Hour, 24*time.Hour); err != nil {
		t.Fatal(err)
	}

	var stored []byte
	if err := s.pool.QueryRow(ctx, `select token_sha256 from sessions`).Scan(&stored); err != nil {
		t.Fatal(err)
	}
	want := sha256.Sum256([]byte(token))
	if !bytes.Equal(stored, want[:]) {
		t.Fatalf("stored credential is not SHA-256 of the token: %x", stored)
	}

	// Broader than the column: any future column that copied the token in would fail this too.
	var row string
	if err := s.pool.QueryRow(ctx, `select sessions::text from sessions`).Scan(&row); err != nil {
		t.Fatal(err)
	}
	if strings.Contains(row, token) {
		t.Fatal("the plaintext token is present in the sessions row")
	}

	// And the credential still resolves, so the two assertions above are about a real session.
	if _, err := s.Lookup(ctx, auth.Digest(token), now); err != nil {
		t.Fatalf("lookup: %v", err)
	}
}

// PD-4. Touch is only reachable after a successful Lookup, so this is depth: a query able to
// resurrect an idle-expired session is not one to leave for the next caller.
// Mutation caught: dropping `idle_expires_at > $2` from Touch's WHERE.
func TestTouchCannotResurrectAnIdleExpiredSession(t *testing.T) {
	s, ctx := testDB(t)
	seedUser(t, s, ctx, "u1")
	now := time.Now().UTC()
	digest := auth.Digest(auth.NewToken())
	if err := s.CreateSession(ctx, digest, "u1", now, time.Hour, 24*time.Hour); err != nil {
		t.Fatal(err)
	}

	later := now.Add(2 * time.Hour) // past the idle window, inside the absolute one
	if err := s.Touch(ctx, digest, later, time.Hour); err != nil {
		t.Fatal(err)
	}
	if _, err := s.Lookup(ctx, digest, later); !errors.Is(err, auth.ErrNoSession) {
		t.Fatalf("an idle-expired session came back to life: %v", err)
	}
}

// The rollback exists and runs. A down path that has never been executed is a claim, not a path.
// ⚠ The rollback is exercised with DATA in the tables whose VOCABULARY the migration rewrites, and
// that is the whole test: an empty database proves the DDL is well-formed, not that a deployment can
// actually roll back. goose runs a migration in one transaction, so the first violated CHECK aborts
// the entire rollback — and the three vocabulary rewrites of 00016 had no data-fixing updates until
// a reviewer reproduced exactly that abort against a live Postgres.
func TestARollbackSurvivesTheDataTheNewVocabulariesWrote(t *testing.T) {
	s, ctx, dsn := testDBWithDSN(t)
	exec(t, s, ctx, `insert into users (id) values ('u1')`)
	exec(t, s, ctx, `insert into books (id, owner_id, title, source_lang, target_lang, status, workdir, engine_book_id)
	                 values ('bk1','u1','蛊真人','zh','ru','not_started','/srv/books/bk1','bk1')`)
	// One ordinary row per table, written exactly as the new code writes them.
	exec(t, s, ctx, `insert into bank_terms (id, book_id, src, dst, status, origin)
	                 values ('t1','bk1','方源','Фан Юань','proposed','found')`)
	exec(t, s, ctx, `insert into bank_decisions (book_id, term_id, action, dst, decided_by)
	                 values ('bk1','t1','approve','Фан Юань','u1')`)
	// 00018's own rows: a chapter (the deferred uniqueness) and an idempotency key (the primary key
	// that moved onto a digest). Without them the rollback was tested over a schema nobody had used.
	exec(t, s, ctx, `insert into chapters (id, book_id, number, units_total)
	                 values ('ch1','bk1',1,2)`)
	exec(t, s, ctx, `insert into idempotency_keys (user_id, method, path, path_sha256, key, request_sha256, claim_token)
	                 values ('u1','POST','/v0/books', sha256(convert_to('/v0/books','UTF8')), 'k1', '\x00'::bytea, 'clm_test')`)
	p, closeProvider, err := newProvider(dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer closeProvider()
	// Down to version 5 and no further. 00005's own down path restores `email NOT NULL` and the
	// unique index, and both are violated by rows production writes, so rolling to zero here would
	// test that known limit instead of this migration.
	//
	// ⚠ The DOCUMENTED floor of a real rollback is 15, not 5 (`deploy/README.md`, "Откат релиза: не
	// ниже версии 15"): 00015's down path narrows `runs_paused_reason_check` back to one value while
	// production writes three. That floor is about DATA, not schema — it bites only a base that has
	// held such a pause — so this base rolls past it honestly, and this test passing is not evidence
	// against the documented floor.
	if _, err := p.DownTo(ctx, 5); err != nil {
		t.Fatalf("rolling back a database that holds real rows: %v", err)
	}
	if _, err := p.Up(ctx); err != nil {
		t.Fatalf("re-applying after the rollback: %v", err)
	}
}

func TestMigrationsRollBackAndReapply(t *testing.T) {
	s, ctx, dsn := testDBWithDSN(t) // migrated up, twice, by the helper

	p, closeDB, err := newProvider(dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer closeDB()

	if _, err := p.DownTo(ctx, 0); err != nil {
		t.Fatalf("down: %v", err)
	}
	var exists bool
	if err := s.pool.QueryRow(ctx, `select to_regclass('public.sessions') is not null`).Scan(&exists); err != nil {
		t.Fatal(err)
	}
	if exists {
		t.Fatal("sessions survived a full rollback")
	}
	if _, err := p.Up(ctx); err != nil {
		t.Fatalf("re-apply: %v", err)
	}
	if err := s.pool.QueryRow(ctx, `select to_regclass('public.sessions') is not null`).Scan(&exists); err != nil {
		t.Fatal(err)
	}
	if !exists {
		t.Fatal("re-apply did not restore the schema")
	}
}

func TestUnknownTokenIsIndistinguishable(t *testing.T) {
	s, ctx := testDB(t)
	if _, err := s.Lookup(ctx, auth.Digest("never-issued"), time.Now()); !errors.Is(err, auth.ErrNoSession) {
		t.Fatalf("want ErrNoSession, got %v", err)
	}
}

// The read-model's derivation rules are DDL, so a materializer bug fails at the write instead of
// reaching a reader. These assert the constraints actually fire.
func TestReadModelConstraints(t *testing.T) {
	s, ctx := testDB(t)
	seedUser(t, s, ctx, "u1")
	exec(t, s, ctx, `insert into books (id, owner_id, title, source_lang, target_lang, status, workdir, engine_book_id)
	                 values ('bk1','u1','蛊真人','zh','ru','translating','/srv/books/bk1','gzr')`)
	exec(t, s, ctx, `insert into chapters (id, book_id, number) values ('ch1','bk1',1)`)

	t.Run("translated needs text", func(t *testing.T) {
		assertViolation(t, s, ctx, "units_translated_has_text",
			`insert into units (id, chapter_id, ordinal, source, target, state)
			 values ('un1','ch1',1,'第一节','','translated')`)
	})
	t.Run("withheld carries none", func(t *testing.T) {
		assertViolation(t, s, ctx, "units_unshipped_is_empty",
			`insert into units (id, chapter_id, ordinal, source, target, state)
			 values ('un2','ch1',2,'第二节','перевод','withheld')`)
	})
	// ⚠ The vocabularies below are the CONTRACT's, not the engine's: 0.3.0 renamed both axes so that
	// a wave name (`draft`), a stage name (`mined`) and a pair-specific word (`ruby`) stop crossing
	// this seam, and the mapping is done where the sidecar is read. The DDL is what makes a path that
	// forgot to translate fail here rather than on a client's screen.
	t.Run("an approval needs a rendering", func(t *testing.T) {
		exec(t, s, ctx, `insert into bank_terms (id, book_id, src, status, origin) values ('t1','bk1','方源','in_progress','found')`)
		assertViolation(t, s, ctx, "bank_decisions_approve_has_dst",
			`insert into bank_decisions (book_id, term_id, action, dst, decided_by) values ('bk1','t1','approve','','u1')`)
	})
	t.Run("the engine's own words are refused", func(t *testing.T) {
		assertViolation(t, s, ctx, "bank_terms_origin_check",
			`insert into bank_terms (id, book_id, src, status, origin) values ('t9','bk1','源','proposed','mined')`)
		assertViolation(t, s, ctx, "bank_terms_status_check",
			`insert into bank_terms (id, book_id, src, status, origin) values ('t8','bk1','源','draft','found')`)
	})
	t.Run("kind may be absent but not invented", func(t *testing.T) {
		// A candidate that is neither a name nor a place carries no kind, and the row is still
		// signable — the contract forbids inventing one for it.
		exec(t, s, ctx, `insert into bank_terms (id, book_id, src, kind, status, origin) values ('t2','bk1','李',null,'in_progress','annotated')`)
		assertViolation(t, s, ctx, "bank_terms_kind_check",
			`insert into bank_terms (id, book_id, src, kind, status, origin) values ('t3','bk1','王','org','in_progress','annotated')`)
	})
	t.Run("one live run per book", func(t *testing.T) {
		exec(t, s, ctx, `insert into runs (id, book_id, status, verify_bank) values ('r1','bk1','translating',true)`)
		assertViolation(t, s, ctx, "runs_one_live_per_book",
			`insert into runs (id, book_id, status, verify_bank) values ('r2','bk1','translating',false)`)
	})
	t.Run("the paused status exists", func(t *testing.T) {
		// D39.100: a ceiling stop is the eleventh book status, and it is not `failed`.
		exec(t, s, ctx, `update books set status='paused' where id='bk1'`)
		assertViolation(t, s, ctx, "books_status_check", `update books set status='exhausted' where id='bk1'`)
	})
}

func seedUser(t testing.TB, s *Store, ctx context.Context, id string) {
	t.Helper()
	exec(t, s, ctx, `insert into users (id, email) values ($1, $1 || '@example.org')`, id)
}

func exec(t testing.TB, s *Store, ctx context.Context, sql string, args ...any) {
	t.Helper()
	if _, err := s.pool.Exec(ctx, sql, args...); err != nil {
		t.Fatalf("exec %s: %v", sql, err)
	}
}

func assertViolation(t *testing.T, s *Store, ctx context.Context, constraint, sql string) {
	t.Helper()
	_, err := s.pool.Exec(ctx, sql)
	var pgErr *pgconn.PgError
	if !errors.As(err, &pgErr) {
		t.Fatalf("want a constraint violation, got %v", err)
	}
	if pgErr.ConstraintName != constraint {
		t.Fatalf("violated %q, want %q", pgErr.ConstraintName, constraint)
	}
}

// The upgrade path from an already-released schema. This is the shape of the defect that reusing a
// migration number produced: goose records only the NUMBER, so a database that stopped at version 3
// accepted "migrations applied" and got none of the tables the new release added.
func TestDatabaseAtAnOlderReleaseCatchesUp(t *testing.T) {
	_, ctx, dsn := testDBWithDSN(t)
	p, closeDB, err := newProvider(dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer closeDB()

	// Back to the previous release, then forward with the current set — a deployment, not a fresh
	// install.
	if _, err := p.DownTo(ctx, 3); err != nil {
		t.Fatalf("down to the previous release: %v", err)
	}
	if _, err := p.Up(ctx); err != nil {
		t.Fatalf("catch up: %v", err)
	}

	after, err := Open(ctx, dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer after.Close()
	for _, table := range []string{"identities", "auth_states", "login_events", "credit_ledger", "account_balances", "reservations"} {
		var exists bool
		if err := after.pool.QueryRow(ctx,
			`select to_regclass('public.' || $1) is not null`, table).Scan(&exists); err != nil {
			t.Fatal(err)
		}
		if !exists {
			t.Fatalf("%s is missing after catching up: the release reported success and did nothing", table)
		}
	}
	// And the draft that the credit model replaced is gone rather than orphaned.
	var stale bool
	if err := after.pool.QueryRow(ctx,
		`select to_regclass('public.usage_windows') is not null`).Scan(&stale); err != nil {
		t.Fatal(err)
	}
	if stale {
		t.Fatal("usage_windows survived the upgrade")
	}
}

// Readiness has to mean "this database is the one this build was made for", not "something answered
// on port 5432". Migrate is off by default and the deploy notes make migrating a separate step, so
// "process up, schema not applied" is the ordinary middle of a rollout — and an instance that calls
// itself ready there fails every query it then serves. Found by review.
// Mutation caught: readiness reduced to Ping; comparing the wrong way round.
func TestReadinessRefusesADatabaseWithoutTheSchema(t *testing.T) {
	s, ctx := testDB(t)

	if err := s.Ready(ctx); err != nil {
		t.Fatalf("a migrated database must be ready: %v", err)
	}

	want, err := latestMigration()
	if err != nil {
		t.Fatal(err)
	}
	// Wind the recorded version back one step without touching the tables: the shape of "the binary
	// carries a migration this database has not seen".
	exec(t, s, ctx, `delete from goose_db_version where version_id = $1`, want)
	err = s.Ready(ctx)
	if !errors.Is(err, ErrSchemaBehind) {
		t.Fatalf("a database behind this build reported ready: %v", err)
	}

	// And a database that has never been migrated at all — no version table.
	exec(t, s, ctx, `drop table goose_db_version`)
	if err := s.Ready(ctx); !errors.Is(err, ErrSchemaBehind) {
		t.Fatalf("an unmigrated database reported ready: %v", err)
	}
	// Reachability alone still says yes, which is exactly why it is not the readiness question.
	if err := s.Ping(ctx); err != nil {
		t.Fatalf("ping should still succeed: %v", err)
	}
}

// The pool sizes an operator writes into the DSN must survive, and "the operator said nothing" must
// be read from the DSN rather than inferred from the value pgx happened to pick. Found by review:
// the previous form compared against pgxpool's own default, which is indistinguishable from an
// operator choosing that same number, and pgx's min-conns default of 0 made an explicit 0 impossible.
// Mutation caught: going back to a value comparison or a substring search.
func TestExplicitPoolSizesInTheDSNSurvive(t *testing.T) {
	for name, tc := range map[string]struct {
		dsn              string
		wantMax, wantMin int32
	}{
		"nothing said, ours apply": {
			"postgres://u@h:5432/db?sslmode=disable", defaultMaxConns, defaultMinConns},
		"url form, both set": {
			"postgres://u@h:5432/db?pool_max_conns=8&pool_min_conns=0&sslmode=disable", 8, 0},
		"keyword form, both set": {
			"host=h user=u dbname=db pool_max_conns=8 pool_min_conns=0", 8, 0},
		// The case that broke the substring test it replaced.
		"a password that merely contains the key name": {
			"postgres://u:pool_max_conns%3D99@h:5432/db?sslmode=disable", defaultMaxConns, defaultMinConns},
		"keyword form with a quoted password containing the key name": {
			`host=h user=u password='pool_max_conns=99 x' dbname=db`, defaultMaxConns, defaultMinConns},
	} {
		t.Run(name, func(t *testing.T) {
			s, err := Open(t.Context(), tc.dsn)
			if err != nil {
				t.Fatalf("open: %v", err)
			}
			defer s.Close()
			if got := s.pool.Config().MaxConns; got != tc.wantMax {
				t.Errorf("MaxConns = %d, want %d", got, tc.wantMax)
			}
			if got := s.pool.Config().MinConns; got != tc.wantMin {
				t.Errorf("MinConns = %d, want %d", got, tc.wantMin)
			}
		})
	}
}