package agent

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"path/filepath"

	"github.com/pingcap/errors"
	"github.com/pingcap/tiflow/engine/enginepb"
	dmopenapi "github.com/pingcap/tiflow/engine/jobmaster/dm/openapi"
	"github.com/pingcap/tiflow/engine/pkg/openapi"
	"github.com/tidbcloud/dataflow-service/internal/constants"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/emptypb"
)

type EngineClient struct {
	AgentClient
	engineAddr string
}

func (c *EngineClient) generateEngineUrl(base, tenantID, projectID string) string {
	return fmt.Sprintf("%s?tenant_id=%s&project_id=%s", base, tenantID, projectID)
}

func engineErrorUnwrapper(logger *zap.Logger, bodyBytes []byte) error {
	var errResp openapi.HTTPError
	if len(bodyBytes) == 0 {
		return errors.Errorf("code: %s", errResp.Code)
	} else if err2 := json.Unmarshal(bodyBytes, &errResp); err2 == nil {
		return errors.Errorf("code: %s, msg: %s", errResp.Code, errResp.Message)
	} else {
		logger.Error("failed to unmarshal response body as HTTPError", zap.Error(err2))
	}
	return nil
}

func (c *EngineClient) requestEngine(ctx context.Context, method, path string, body any) ([]byte, error) {

	path = filepath.Join("proxy/engine", path)
	extraHeaders := map[string]string{
		constants.EngineAddrHeader: c.engineAddr,
	}
	return c.FetchResponseWithHeader(ctx, method, path, body, extraHeaders, engineErrorUnwrapper)
}

func (c *EngineClient) CreateEngineJob(ctx context.Context, req *enginepb.CreateJobRequest) (*enginepb.Job, error) {
	if c == nil {
		return nil, errors.New("DM cluster encountered compatibility issues, please contact technical support")
	}
	path := c.generateEngineUrl("api/v1/jobs", req.TenantId, req.ProjectId)
	respBytes, err := c.requestEngine(ctx, http.MethodPost, path, req.Job)
	if err != nil {
		return nil, err
	}

	var resp enginepb.Job
	if err = protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (c *EngineClient) CancelEngineJob(ctx context.Context, req *enginepb.CancelJobRequest) (*enginepb.Job, error) {
	if c == nil {
		return nil, errors.New("DM cluster encountered compatibility issues, please contact technical support")
	}
	path := c.generateEngineUrl(fmt.Sprintf("api/v1/jobs/%s/cancel", req.Id), req.TenantId, req.ProjectId)
	respBytes, err := c.requestEngine(ctx, http.MethodPost, path, nil)
	if err != nil {
		return nil, err
	}

	var resp enginepb.Job
	if err = protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (c *EngineClient) DeleteEngineJob(ctx context.Context, req *enginepb.DeleteJobRequest) (*emptypb.Empty, error) {
	if c == nil {
		return nil, errors.New("DM cluster encountered compatibility issues, please contact technical support")
	}
	path := c.generateEngineUrl(fmt.Sprintf("api/v1/jobs/%s", req.Id), req.TenantId, req.ProjectId)
	_, err := c.requestEngine(ctx, http.MethodDelete, path, nil)
	if err != nil {
		return nil, err
	}

	return &emptypb.Empty{}, nil
}

type OperateJobReq struct {
	Id        string
	TenantId  string
	ProjectId string
	Operation dmopenapi.OperateJobRequestOp
}

func (c *EngineClient) OperateEngineJob(ctx context.Context, req *OperateJobReq) error {
	if c == nil {
		return errors.New("DM cluster encountered compatibility issues, please contact technical support")
	}
	path := c.generateEngineUrl(fmt.Sprintf("api/v1/jobs/%s/status", req.Id), req.TenantId, req.ProjectId)
	// this api returns Empty on success
	_, err := c.requestEngine(ctx, http.MethodPut, path, &dmopenapi.OperateJobRequest{
		Op: req.Operation,
	})
	return err
}

func (c *EngineClient) GetEngineJob(ctx context.Context, req *enginepb.GetJobRequest) (*enginepb.Job, error) {
	if c == nil {
		return nil, errors.New("DM cluster encountered compatibility issues, please contact technical support")
	}
	path := c.generateEngineUrl(fmt.Sprintf("api/v1/jobs/%s", req.Id), req.TenantId, req.ProjectId)
	respBytes, err := c.requestEngine(ctx, http.MethodGet, path, nil)
	if err != nil {
		return nil, err
	}

	var resp enginepb.Job
	if err = protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}