//go:generate ${GOBIN}/mockgen -destination ops_client_mock.go -package=dataflowcli github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow OpsInternalServiceClient
//go:generate ${GOBIN}/mockgen -destination connector_client_mock.go -package=dataflowcli github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow ConnectorServiceClient
//go:generate ${GOBIN}/mockgen -destination ng_changefeed_client_mock.go -package=dataflowcli github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi ChangefeedServiceClient
package dataflowcli

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"github.com/go-resty/resty/v2"

	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	importintong "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/import_into/v1beta2"
	premiumdm "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/premium_dm/v1beta2"
	dataflow "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	dm "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/replay"
	"github.com/tidbcloud/devops/pkg/infra/utils/grpcutil"
	"google.golang.org/grpc"
)

type DataflowCli struct {
	Ops            dataflow.OpsInternalServiceClient
	Connector      dataflow.ConnectorServiceClient
	DM             dm.ServiceClient
	Replay         replay.ReplayServiceClient
	NgOps          dataflowng.OpsInternalServiceClient
	NgChangefeed   dataflowng.ChangefeedServiceClient
	Backup         dataflow.BackupConsoleServiceClient
	BackupSvc      dataflow.BackupServiceClient
	BackupInternal dataflow.BackupInternalServiceClient
	Import         dataflow.ImportServiceClient
	PremiumDM      premiumdm.PremiumMigrationServiceClient
	ImportInto     importintong.ImportIntoServiceClient

	httpCli *resty.Client
}

type ListDMJobsParam struct {
	ProjectId uint64
	ClusterId uint64
	Page      *int32
	PageSize  *int32
	OrgId     uint64
}

type ResetDMJobParam struct {
	ClusterID string `json:"cluster_id"`
	JobId     uint64 `json:"job_id"`
}

type GetClusterDMConfigResp struct {
	ClusterID        string   `json:"cluster_id"`
	Namespace        string   `json:"namespace"`
	ExecutorPodNames []string `json:"executor_pod_names"`
}

func NewDataflowCli(dataflowAddr string, dataflowNgAddr string) (*DataflowCli, error) {
	c := &DataflowCli{}

	opts := grpcutil.PresetDailOption().Default()

	// Single shared connection for all dataflow services at dataflowAddr.
	conn, err := grpc.NewClient(dataflowAddr, opts...)
	if err != nil {
		return nil, fmt.Errorf("dial dataflow svc %s failed: %s", dataflowAddr, err)
	}
	c.Ops = dataflow.NewOpsInternalServiceClient(conn)
	c.Connector = dataflow.NewConnectorServiceClient(conn)
	c.DM = dm.NewServiceClient(conn)
	c.Replay = replay.NewReplayServiceClient(conn)
	c.Backup = dataflow.NewBackupConsoleServiceClient(conn)
	c.BackupSvc = dataflow.NewBackupServiceClient(conn)
	c.BackupInternal = dataflow.NewBackupInternalServiceClient(conn)
	c.Import = dataflow.NewImportServiceClient(conn)

	// Single shared connection for all dataflow-ng services at dataflowNgAddr.
	ngConn, err := grpc.NewClient(dataflowNgAddr, opts...)
	if err != nil {
		return nil, fmt.Errorf("dial dataflow-ng svc %s failed: %s", dataflowNgAddr, err)
	}
	c.NgOps = dataflowng.NewOpsInternalServiceClient(ngConn)
	c.NgChangefeed = dataflowng.NewChangefeedServiceClient(ngConn)
	c.PremiumDM = premiumdm.NewPremiumMigrationServiceClient(ngConn)
	c.ImportInto = importintong.NewImportIntoServiceClient(ngConn)

	httpCli := resty.New().
		SetTransport(http.DefaultTransport).
		SetTimeout(60 * time.Second).
		OnAfterResponse(func(c *resty.Client, r *resty.Response) error {
			if !r.IsSuccess() {
				return fmt.Errorf("call dataflow service failed, url: %s, method: %s, status: %s, body: %s",
					r.Request.URL, r.Request.Method, r.Status(), string(r.Body()))
			}
			return nil
		})
	c.httpCli = httpCli

	return c, nil
}

func (cli *DataflowCli) ListDMJobs(ctx context.Context, params ListDMJobsParam) ([]*dm.Job, int, error) {
	jobs := make([]*dm.Job, 0)

	req := &dm.ListJobsReq{
		ProjectId: params.ProjectId,
		ClusterId: params.ClusterId,
		Page:      *params.Page,
		PageSize:  *params.PageSize,
		OrgId:     params.OrgId,
	}
	resp, err := cli.DM.ListJobs(ctx, req)
	if err != nil {
		return jobs, 0, err
	}

	return resp.Jobs, len(resp.Jobs), nil
}

func (cli *DataflowCli) ResetDMJob(ctx context.Context, agentURL string, params ResetDMJobParam) error {
	url := fmt.Sprintf("%s/api/internal/v1beta1/dm/reset-error", agentURL)
	_, err := cli.httpCli.R().
		SetContext(ctx).
		SetBody(&params).
		Post(url)
	return err
}

func (cli *DataflowCli) GetClusterDMConfig(ctx context.Context, agentURL string, clusterID string) (*GetClusterDMConfigResp, error) {
	url := fmt.Sprintf("%s/api/internal/v1beta1/dm/config", agentURL)

	var config GetClusterDMConfigResp
	_, err := cli.httpCli.R().
		SetContext(ctx).
		SetQueryParam("cluster_id", clusterID).
		SetResult(&config).
		Get(url)
	if err != nil {
		return nil, err
	}
	return &config, nil
}