package db_test

import (
	"testing"

	"github.com/jmoiron/sqlx"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"git.offline-twitter.com/offline-labs/gas-stack/pkg/db"
	"git.offline-twitter.com/offline-labs/gas-stack/pkg/must"
)

// migrationTestSchema has a foreign key (`children.parent_id` => `parents.rowid`) along with one
// valid parent/child pair, so that migrations can be written which either preserve or break
// referential integrity.
var migrationTestSchema = `
	create table db_version (version integer primary key) strict, without rowid;
	insert into db_version values (0);

	create table parents (
		rowid integer primary key,
		name text not null
	);
	insert into parents (rowid, name) values (1, 'p1');

	create table children (
		rowid integer primary key,
		parent_id integer not null references parents(rowid)
	);
	insert into children (rowid, parent_id) values (1, 1);
`

func newMigrationTestDB() *sqlx.DB {
	return must.Get(db.Init(&migrationTestSchema, &[]db.Migration{}).Create(":memory:"))
}

func assertForeignKeysAreGood(t *testing.T, conn *sqlx.DB) {
	t.Helper()

	// FKs should be enabled and intact at all times
	var fkEnabled bool
	require.NoError(t, conn.Get(&fkEnabled, `pragma foreign_keys`))
	assert.True(t, fkEnabled)
	assert.NoError(t, db.PragmaForeignKeyCheck(conn))

	// ...and enforcement should be in effect again, not merely reported as on
	_, err := conn.Exec(`insert into children (rowid, parent_id) values (99, 40235235234)`)
	assert.True(t, db.IsSqliteFkError(err), "expected a foreign key error, got: %v", err)
}

func TestMigrationApply(t *testing.T) {
	conn := newMigrationTestDB()

	testcases := []struct {
		Migration        db.Migration
		ExpectedParents  int
		ExpectedChildren int
	}{
		{
			Migration: db.Migration{
				ID:    1,
				Title: "add a second parent",
				SQL:   `insert into parents (rowid, name) values (2, 'p2')`,
			},
			ExpectedParents:  2,
			ExpectedChildren: 1,
		},
		{
			Migration: db.Migration{ID: 2, SQL: `
				insert into children (rowid, parent_id) values (2, 2);
				insert into parents (rowid, name) values (3, 'p3');
			`},
			ExpectedParents:  3,
			ExpectedChildren: 2,
		},
	}

	for _, tc := range testcases {
		t.Run(tc.Migration.Title, func(t *testing.T) {
			require.NoError(t, tc.Migration.Apply(conn))

			// Check that it applied
			var parents int
			require.NoError(t, conn.Get(&parents, `select count(*) from parents`))
			assert.Equal(t, tc.ExpectedParents, parents)
			var children int
			require.NoError(t, conn.Get(&children, `select count(*) from children`))
			assert.Equal(t, tc.ExpectedChildren, children)

			// Should bump the DB version
			var version int
			require.NoError(t, conn.Get(&version, `select version from db_version`))
			assert.Equal(t, tc.Migration.ID, version, "db version should be bumped to the migration's ID")

			assertForeignKeysAreGood(t, conn)
		})
	}
}

// A migration whose SQL fails must leave no trace, including the statements which had already
// succeeded before the failing one.  Every migration below therefore starts by creating
// `new_table`, which must never survive.
func TestMigrationFailureRollsBack(t *testing.T) {
	conn := newMigrationTestDB()

	nothingHappens := func(t *testing.T) {
		t.Helper()
		// Everything should be rolled back, including the statements which ran before the failing one
		var newTables int
		require.NoError(t, conn.Get(&newTables, `select count(*) from sqlite_master where type = 'table' and name = 'new_table'`))
		assert.Equal(t, 0, newTables, "the migration's earlier statements should be rolled back too")

		var version int
		require.NoError(t, conn.Get(&version, `select version from db_version`))
		assert.Equal(t, 0, version)

		var parents int
		require.NoError(t, conn.Get(&parents, `select count(*) from parents`))
		assert.Equal(t, 1, parents)

		var children int
		require.NoError(t, conn.Get(&children, `select count(*) from children`))
		assert.Equal(t, 1, children)
	}

	testcases2 := []db.Migration{
		{
			ID:    1,
			Title: "syntax error",
			SQL: `
				create table new_table (rowid integer primary key);
				this is not valid sql;
			`,
		},
		{
			ID:    1,
			Title: "unknown table",
			SQL: `
				create table new_table (rowid integer primary key);
				insert into no_such_table (rowid) values (1);
			`,
		},
		{
			ID:    1,
			Title: "constraint violation",
			SQL: `
				create table new_table (rowid integer primary key);
				insert into parents (rowid, name) values (1, 'duplicate rowid');
			`,
		},
		{
			ID:    1,
			Title: "failure after a successful write",
			SQL: `
				create table new_table (rowid integer primary key);
				insert into parents (rowid, name) values (2, 'p2');
				insert into parents (rowid, name) values (1, 'duplicate rowid');
			`,
		},
	}
	for _, tc := range testcases2 {
		t.Run(tc.Title, func(t *testing.T) {
			assert.Panics(t, func() { _ = tc.Apply(conn) })
			nothingHappens(t)
			assertForeignKeysAreGood(t, conn)
		})
	}

	// ---------
	// Foreign key failures should trigger a rollback too
	// ---------

	testcases := []db.Migration{
		{
			ID:    1,
			Title: "child pointing at a nonexistent parent",
			SQL: `
				create table new_table (rowid integer primary key);
				insert into children (rowid, parent_id) values (2, 404);
			`,
		},
		{
			ID:    1,
			Title: "parent deleted out from under an existing child",
			SQL: `
				create table new_table (rowid integer primary key);
				delete from parents where rowid = 1;
			`,
		},
	}
	for _, tc := range testcases {
		t.Run(tc.Title, func(t *testing.T) {
			// Should notify with the correct detail about the error
			func() {
				defer func() {
					err := recover()
					require.NotNil(t, err)
					switch err := err.(type) {
					case error:
						var violations db.PragmaFKViolations
						require.ErrorAs(t, err, &violations)
						require.Len(t, violations, 1)
						assert.Equal(t, "children", violations[0].SrcTable)
						assert.Equal(t, "parents", violations[0].DstTable)
					default:
						panic(err) // Something weird happened
					}
				}()
				_ = tc.Apply(conn)
			}()

			// ...as well as the usual
			nothingHappens(t)
			assertForeignKeysAreGood(t, conn)
		})
	}
}

// Migrations are sequential: a migration may only be applied to a DB sitting at the immediately
// preceding version.
func TestMigrationApplyRejectsOutOfOrderMigrations(t *testing.T) {
	const sql = `insert into parents (name) values ('p2')`

	t.Run("migration from the future", func(t *testing.T) {
		conn := newMigrationTestDB()

		err := db.Migration{ID: 2, SQL: sql}.Apply(conn)
		require.ErrorIs(t, err, db.ErrMigrationOrder)

		var version int
		require.NoError(t, conn.Get(&version, `select version from db_version`))
		assert.Equal(t, 0, version)

		var parents int
		require.NoError(t, conn.Get(&parents, `select count(*) from parents`))
		assert.Equal(t, 1, parents, "the migration's SQL should not have run")

		assertForeignKeysAreGood(t, conn)
	})

	t.Run("migration applied twice", func(t *testing.T) {
		conn := newMigrationTestDB()
		m := db.Migration{ID: 1, SQL: sql}
		require.NoError(t, m.Apply(conn))

		err := m.Apply(conn)
		require.ErrorIs(t, err, db.ErrMigrationOrder)

		var version int
		require.NoError(t, conn.Get(&version, `select version from db_version`))
		assert.Equal(t, 1, version)

		var parents int
		require.NoError(t, conn.Get(&parents, `select count(*) from parents`))
		assert.Equal(t, 2, parents, "the migration's SQL should not have run again")

		assertForeignKeysAreGood(t, conn)
	})
}