package sample_data

import (
	"database/sql"
	"fmt"
	"sync"
	"time"

	"github.com/tidbcloud/dataflow-service-e2e/internal/log"
	"github.com/tidbcloud/dataflow-service-e2e/internal/misc"

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

/*
*
example:
writer := NewSampleDataWriter(tidbCluster, dbName, 3*time.Second)
writer.StartWriting()
...
writer.StopWriting()
*/
const SAMPLE_TABLE = "sample_table"

// SampleDataWriter encapsulates continuous data writing
type SampleDataWriter struct {
	lgr         log.SimpleLogger
	tidbCluster *MysqlParameter
	dbName      string
	interval    time.Duration

	db     *sql.DB
	stopCh chan struct{}
	wg     sync.WaitGroup
}

type MysqlParameter struct {
	User     string
	Password string
	Host     string
	Port     string
	CACert   string
}

// NewSampleDataWriter creates a new instance
func NewSampleDataWriter(lgr log.SimpleLogger, cluster *MysqlParameter, dbName string, interval time.Duration) *SampleDataWriter {
	return &SampleDataWriter{
		lgr:         lgr,
		tidbCluster: cluster,
		dbName:      dbName,
		interval:    interval,
		stopCh:      make(chan struct{}),
	}
}

func (w *SampleDataWriter) InitDBTable() error {
	db, err := Connect2TiDB(
		w.tidbCluster.User,
		w.tidbCluster.Password,
		w.tidbCluster.Host,
		w.tidbCluster.Port,
		w.tidbCluster.CACert,
		"")
	if err != nil {
		w.lgr.Error("connect to tidbCluster failed, error: %s, tidbCluster: %s", err.Error(), misc.JSON(w.tidbCluster))
		return fmt.Errorf("connect to tidbCluster error")
	}

	// Create database if not exists
	_, err = db.Exec(fmt.Sprintf("CREATE DATABASE IF NOT EXISTS %s", w.dbName))
	if err != nil {
		w.lgr.Error("create database failed, dbName: %s, error: %s, tidbCluster: %s", w.dbName, err.Error(), misc.JSON(w.tidbCluster))
		return fmt.Errorf("create database error")
	}

	w.lgr.Info("database '%s' is ready!", w.dbName)

	err = db.Close()
	if err != nil {
		w.lgr.Warn("close db connection failed after database created, error: %s, tidbCluster: %s", err.Error(), misc.JSON(w.tidbCluster))
	}

	// Reconnect with the newly created database
	w.db, err = Connect2TiDB(
		w.tidbCluster.User,
		w.tidbCluster.Password,
		w.tidbCluster.Host,
		w.tidbCluster.Port,
		w.tidbCluster.CACert,
		w.dbName)
	if err != nil {
		w.lgr.Error("connect to tidbCluster with database(%s) failed, error: %s, tidbCluster: %s", w.dbName, err.Error(), misc.JSON(w.tidbCluster))
		return fmt.Errorf("connect to tidbCluster with database failed")
	}

	// Ensure table exists
	_, err = w.db.Exec(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (
		id INT AUTO_INCREMENT PRIMARY KEY,
		created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
	);`, SAMPLE_TABLE))
	if err != nil {
		w.lgr.Error("create table '%s' failed, error: %s, tidbCluster: %s", SAMPLE_TABLE, err.Error(), misc.JSON(w.tidbCluster))
		return fmt.Errorf("create table '%s' failed", SAMPLE_TABLE)
	}
	return nil
}

// StartWriting starts inserting sample data at regular intervals
func (w *SampleDataWriter) StartWriting() error {
	w.wg.Add(1)
	go func() {
		defer w.wg.Done()
		w.lgr.Info("Starting data writing...")

		ticker := time.NewTicker(w.interval)
		defer ticker.Stop()

		for {
			select {
			case <-ticker.C:
				if err := w.insertSampleData(); err != nil {
					w.lgr.Warn("Error inserting data: %s", err.Error())
				}
			case <-w.stopCh:
				w.lgr.Info("Stopping data writing...")
				return
			}
		}
	}()
	return nil
}

// StopWriting stops the continuous data writing
func (w *SampleDataWriter) StopWriting() {
	close(w.stopCh)
	w.wg.Wait() // Ensure the goroutine finishes before returning
	if w.db != nil {
		w.db.Close()
	}
}

// insertSampleData inserts a sample row into the database
func (w *SampleDataWriter) insertSampleData() error {
	_, err := w.db.Exec(fmt.Sprintf("INSERT INTO %s (created_at) VALUES (?)", SAMPLE_TABLE), time.Now())
	return err
}