package impl

import (
	"context"
	"database/sql"
	"errors"
	"regexp"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	fastflow_entity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/stretchr/testify/require"
	gormMysql "gorm.io/driver/mysql"
	"gorm.io/gorm"
)

// newTestImpl wires a workflowImpl whose db handle is backed by go-sqlmock so
// that we can assert the exact SQL the new selective-first path generates.
// We deliberately do not initialize fastflow's keeper/store/commander because
// FindDagInstancesByBusinessID does not touch them.
func newTestImpl(t *testing.T) (*workflowImpl, sqlmock.Sqlmock, *sql.DB) {
	t.Helper()
	rawDB, dbMock, err := sqlmock.New(
		sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp),
	)
	require.NoError(t, err)

	gormDB, err := gorm.Open(gormMysql.New(gormMysql.Config{
		Conn:                      rawDB,
		SkipInitializeWithVersion: true,
	}), &gorm.Config{})
	require.NoError(t, err)

	return &workflowImpl{db: gormDB}, dbMock, rawDB
}

func TestFindDagInstancesByBusinessID_RejectsEmptyBusinessID(t *testing.T) {
	w, _, rawDB := newTestImpl(t)
	defer rawDB.Close()

	_, err := w.FindDagInstancesByBusinessID(context.Background(), "", nil, nil, 0)
	require.Error(t, err)
}

func TestFindDagInstancesByBusinessID_NilDB(t *testing.T) {
	w := &workflowImpl{db: nil}
	_, err := w.FindDagInstancesByBusinessID(context.Background(), "1", nil, nil, 0)
	require.Error(t, err)
}

func TestFindDagInstancesByBusinessID_TagLookupEmpty_NoSecondQuery(t *testing.T) {
	w, dbMock, rawDB := newTestImpl(t)
	defer rawDB.Close()

	// Step 1: tag lookup returns zero rows. We must short-circuit and not
	// fire the dag_instances query at all — that's the whole point of
	// avoiding a wasted PK-IN query on an empty set.
	dbMock.ExpectQuery(regexp.QuoteMeta(
		"SELECT `dag_ins_id` FROM `dag_instance_tags` WHERE `key` = ? AND `value` = ?",
	)).
		WithArgs("business_id", "2986158").
		WillReturnRows(sqlmock.NewRows([]string{"dag_ins_id"}))

	out, err := w.FindDagInstancesByBusinessID(
		context.Background(),
		"2986158",
		[]string{"create_snapshot_backup_v1.0"},
		[]fastflow_entity.DagInstanceStatus{fastflow_entity.DagInstanceStatusFailed},
		1,
	)
	require.NoError(t, err)
	require.Empty(t, out)
	require.NoError(t, dbMock.ExpectationsWereMet())
}

func TestFindDagInstancesByBusinessID_TagLookupError_PropagatesError(t *testing.T) {
	w, dbMock, rawDB := newTestImpl(t)
	defer rawDB.Close()

	dbMock.ExpectQuery(regexp.QuoteMeta(
		"SELECT `dag_ins_id` FROM `dag_instance_tags` WHERE `key` = ? AND `value` = ?",
	)).
		WithArgs("business_id", "2986158").
		WillReturnError(errors.New("connection refused"))

	_, err := w.FindDagInstancesByBusinessID(
		context.Background(), "2986158", nil, nil, 0,
	)
	require.Error(t, err)
	require.Contains(t, err.Error(), "dag_instance_tags")
	require.NoError(t, dbMock.ExpectationsWereMet())
}

func TestFindDagInstancesByBusinessID_AppliesAllFilters(t *testing.T) {
	w, dbMock, rawDB := newTestImpl(t)
	defer rawDB.Close()

	// Step 1: tag lookup returns one candidate dag_ins_id.
	dbMock.ExpectQuery(regexp.QuoteMeta(
		"SELECT `dag_ins_id` FROM `dag_instance_tags` WHERE `key` = ? AND `value` = ?",
	)).
		WithArgs("business_id", "2986158").
		WillReturnRows(sqlmock.NewRows([]string{"dag_ins_id"}).AddRow("dag-ins-42"))

	// Step 2: PK-IN with status and dag_id filters + limit. We assert by
	// regex rather than exact match because gorm reorders the IN clauses
	// based on slice iteration order, and we don't want a brittle test.
	dbMock.ExpectQuery(
		`SELECT \* FROM .dag_instances. WHERE id IN .\?. AND dag_id IN .\?,\?. AND status IN .\?,\?. LIMIT \?`,
	).
		WithArgs(
			"dag-ins-42",
			"create_snapshot_backup_v1.0", "create_weekly_snapshot_backup_v1.0",
			fastflow_entity.DagInstanceStatusFailed, fastflow_entity.DagInstanceStatusCanceled,
			1,
		).
		WillReturnRows(sqlmock.NewRows([]string{"id", "dag_id", "status"}).
			AddRow("dag-ins-42", "create_snapshot_backup_v1.0", string(fastflow_entity.DagInstanceStatusFailed)))

	out, err := w.FindDagInstancesByBusinessID(
		context.Background(),
		"2986158",
		[]string{"create_snapshot_backup_v1.0", "create_weekly_snapshot_backup_v1.0"},
		[]fastflow_entity.DagInstanceStatus{
			fastflow_entity.DagInstanceStatusFailed,
			fastflow_entity.DagInstanceStatusCanceled,
		},
		1,
	)
	require.NoError(t, err)
	require.Len(t, out, 1)
	require.Equal(t, "dag-ins-42", out[0].ID)
	require.NoError(t, dbMock.ExpectationsWereMet())
}

func TestFindDagInstancesByBusinessID_OmitsOptionalFilters(t *testing.T) {
	w, dbMock, rawDB := newTestImpl(t)
	defer rawDB.Close()

	dbMock.ExpectQuery(regexp.QuoteMeta(
		"SELECT `dag_ins_id` FROM `dag_instance_tags` WHERE `key` = ? AND `value` = ?",
	)).
		WithArgs("business_id", "99").
		WillReturnRows(sqlmock.NewRows([]string{"dag_ins_id"}).AddRow("a").AddRow("b"))

	// No dag_id, no status, no limit — only the PK-IN clause should appear.
	dbMock.ExpectQuery(
		`SELECT \* FROM .dag_instances. WHERE id IN .\?,\?.$`,
	).
		WithArgs("a", "b").
		WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow("a"))

	out, err := w.FindDagInstancesByBusinessID(
		context.Background(), "99", nil, nil, 0,
	)
	require.NoError(t, err)
	require.Len(t, out, 1)
	require.NoError(t, dbMock.ExpectationsWereMet())
}