package common

import (
	"context"
	"database/sql"
	"os"
	"testing"

	"github.com/stretchr/testify/require"

	"github.com/pingcap/test-infra/common/model"

	wf "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	_ "github.com/go-sql-driver/mysql"
)

func newTestDB(t *testing.T) *sql.DB {
	dsn := os.Getenv("TEST_DSN")
	if len(dsn) == 0 {
		t.Skip("TEST_DSN is required for running this test")
	}
	db, err := sql.Open("mysql", dsn)
	require.NoError(t, err)
	return db
}

func initJobStore(t *testing.T, db *sql.DB) {
	_, err := db.Exec("drop table if exists test_job")
	require.NoError(t, err)
	_, err = db.Exec(`create table test_job (
    id         bigint,
    plan_exec  text not null,
    workflow   text not null,
    status     varchar(40),
    message    text,
    created_at int,
    updated_at int,
    primary key (id),
    key (status)
);`)
	require.NoError(t, err)
}

func TestSQLStore(t *testing.T) {
	ctx := context.TODO()
	db := newTestDB(t)
	initJobStore(t, db)
	store := sqlJobStore{db: db}

	jobs, err := store.ListJobs(ctx, ListJobConditions{})
	require.NoError(t, err)
	require.Empty(t, jobs)

	for _, job1 := range []*ArgoJob{
		{
			PlanExec: &model.TestPlanExecution{ID: 1},
			Workflow: &wf.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{"workflows.argoproj.io/phase": "Pending"},
				},
			},
		},
		{
			PlanExec: &model.TestPlanExecution{ID: 2},
			Workflow: &wf.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{"workflows.argoproj.io/phase": "Running"},
				},
			},
		},
		{
			PlanExec: &model.TestPlanExecution{ID: 3},
			Workflow: &wf.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{"workflows.argoproj.io/phase": "Running"},
				},
			},
			Status: ArgoTestJobDone,
		},
		{
			PlanExec: &model.TestPlanExecution{ID: 1},
			Workflow: &wf.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{"workflows.argoproj.io/phase": "Running"},
				},
			},
		},
		{
			PlanExec: &model.TestPlanExecution{ID: 4},
			Workflow: &wf.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{"workflows.argoproj.io/phase": "Succeeded"},
				},
			},
		},
		{
			PlanExec: &model.TestPlanExecution{ID: 5},
			Status:   ArgoTestJobPending,
		},
	} {
		if job1.Status == ArgoTestJobPending {
			context.TODO()
		}
		job1.UpdateStatus(nil)
		require.NoError(t, store.PutJob(ctx, job1, OnDupUpdate))
		job2, err := store.GetJob(ctx, job1.ID())
		require.NoError(t, err)
		require.Equal(t, job1, job2)
	}

	jobs, err = store.ListJobs(ctx, ListJobConditions{})
	require.NoError(t, err)
	require.Len(t, jobs, 5)
	jobs, err = store.ListJobs(ctx, ListJobConditions{Limit: 1})
	require.NoError(t, err)
	require.Len(t, jobs, 1)
	jobs, err = store.ListJobs(ctx, ListJobConditions{Status: ArgoTestJobDone})
	require.NoError(t, err)
	require.Len(t, jobs, 2)
	jobs, err = store.ListJobs(ctx, ListJobConditions{Status: ArgoTestJobRunning})
	require.NoError(t, err)
	require.Len(t, jobs, 2)
	jobs, err = store.ListJobs(ctx, ListJobConditions{Status: ArgoTestJobRunning, Limit: 1})
	require.NoError(t, err)
	require.Len(t, jobs, 1)
	jobs, err = store.ListJobs(ctx, ListJobConditions{Status: ArgoTestJobPending})
	require.NoError(t, err)
	require.Len(t, jobs, 1)
}