package grpcli

import (
	"context"
	"strconv"

	"github.com/tidbcloud/pkgv2/bizerr"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
)

// ExportGrpcClient is the narrow dataflow-service-ng boundary to the existing
// management Export service. Requests may contain database and object-storage
// credentials, so implementations must not log request or response payloads.
type ExportGrpcClient interface {
	GetExport(ctx context.Context, caller ExportCaller, req *globalngv1.GetExportRequest) (*globalngv1.Export, bizerr.BizError)
	CreateExport(ctx context.Context, caller ExportCaller, req *globalngv1.CreateExportRequest, snapshotTSO uint64) (*globalngv1.Export, bizerr.BizError)
	DeleteExport(ctx context.Context, caller ExportCaller, req *globalngv1.DeleteExportRequest) bizerr.BizError
	CancelExport(ctx context.Context, caller ExportCaller, req *globalngv1.CancelExportRequest) bizerr.BizError
}

// Metadata keys the management service reads the acting identity from. They
// are defined by that service; see its utilities/grpc/metadata package.
const (
	exportOrgIDMetadataKey  = "x-org-id"
	exportUserIDMetadataKey = "x-user-id"
	// Data Pipeline is scoped to a cluster, not to a project. Without this
	// flag the management service refuses to forward the identity to the
	// service that actually performs the write, because its propagation rule
	// demands a default project and a valid project list.
	exportNoProjectMetadataKey = "x-no-project-required"
)

// ExportCaller is the identity a management Export call acts for.
//
// It is a parameter rather than something read off the context because these
// calls originate in workflow actions, which run detached from the request
// that created the Pipeline. The management service stamps the identity onto
// the workflow it starts and rejects a write that arrives without it, so a
// caller that forgets it fails at runtime; making it an argument moves that
// failure to compile time.
type ExportCaller struct {
	OrganizationID uint64
	UserID         uint64
}

func (c ExportCaller) validate() bizerr.BizError {
	if c.OrganizationID == 0 || c.UserID == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "Export caller organization and user are required")
	}
	return nil
}

// attach puts the identity on the outgoing context. The management service
// looks in the incoming metadata of the call it receives, which is exactly
// what outgoing metadata here becomes.
func (c ExportCaller) attach(ctx context.Context) context.Context {
	return metadata.AppendToOutgoingContext(ctx,
		exportOrgIDMetadataKey, strconv.FormatUint(c.OrganizationID, 10),
		exportUserIDMetadataKey, strconv.FormatUint(c.UserID, 10),
		exportNoProjectMetadataKey, "true",
	)
}

// exportServiceClient mirrors the generated
// tidb_mgmt_service.v1.ExportServiceClient so tests can substitute a fake.
type exportServiceClient interface {
	GetExport(ctx context.Context, in *globalngv1.GetExportRequest, opts ...grpc.CallOption) (*globalngv1.Export, error)
	CreateExport(ctx context.Context, in *globalngv1.CreateExportRequest, opts ...grpc.CallOption) (*globalngv1.Export, error)
	DeleteExport(ctx context.Context, in *globalngv1.DeleteExportRequest, opts ...grpc.CallOption) (*globalngv1.Export, error)
	CancelExport(ctx context.Context, in *globalngv1.CancelExportRequest, opts ...grpc.CallOption) (*globalngv1.Export, error)
}

// The generated client must keep satisfying the local boundary; a signature
// drift in the API dependency then fails the build instead of at runtime.
var _ exportServiceClient = (mgmtv1.ExportServiceClient)(nil)

func newManagementExportServiceClient(conn grpc.ClientConnInterface) exportServiceClient {
	return mgmtv1.NewExportServiceClient(conn)
}

type exportGrpcClientImpl struct {
	client exportServiceClient
}

func newExportGrpcClient(client exportServiceClient) ExportGrpcClient {
	return &exportGrpcClientImpl{client: client}
}

func (c *exportGrpcClientImpl) GetExport(
	ctx context.Context,
	caller ExportCaller,
	req *globalngv1.GetExportRequest,
) (*globalngv1.Export, bizerr.BizError) {
	if err := caller.validate(); err != nil {
		return nil, err
	}
	resp, err := c.client.GetExport(caller.attach(ctx), req)
	if err != nil {
		return nil, exportGrpcError(err)
	}
	return resp, nil
}

func (c *exportGrpcClientImpl) CreateExport(
	ctx context.Context,
	caller ExportCaller,
	req *globalngv1.CreateExportRequest,
	snapshotTSO uint64,
) (*globalngv1.Export, bizerr.BizError) {
	if err := caller.validate(); err != nil {
		return nil, err
	}
	if req == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "create Export request is required")
	}
	// The snapshot boundary is the whole reason this client exists. Omitting it
	// would let the regional service pick its own timestamp and silently break
	// the alignment between the Export and the incremental changefeed.
	if snapshotTSO == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "snapshot TSO is required to create an Export")
	}
	if req.GetExport() == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "create Export request needs an Export")
	}
	// Clone so a workflow retry cannot observe a request the caller still owns
	// being mutated between attempts.
	cloned, ok := proto.Clone(req).(*globalngv1.CreateExportRequest)
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "clone create Export request")
	}
	// The boundary travels on Export.snapshot_tso: that is the only field the
	// merged management contract has for it.
	//
	// TODO: if the management API grows a request-level snapshot_tso, move it
	// there - a field that is also populated on read is easy to misread as
	// output-only.
	cloned.Export.SnapshotTso = snapshotTSO
	req = cloned

	resp, err := c.client.CreateExport(caller.attach(ctx), req)
	if err != nil {
		return nil, exportGrpcError(err)
	}
	return resp, nil
}

func (c *exportGrpcClientImpl) DeleteExport(
	ctx context.Context,
	caller ExportCaller,
	req *globalngv1.DeleteExportRequest,
) bizerr.BizError {
	if err := caller.validate(); err != nil {
		return err
	}
	if _, err := c.client.DeleteExport(caller.attach(ctx), req); err != nil {
		return exportGrpcError(err)
	}
	return nil
}

func (c *exportGrpcClientImpl) CancelExport(
	ctx context.Context,
	caller ExportCaller,
	req *globalngv1.CancelExportRequest,
) bizerr.BizError {
	if err := caller.validate(); err != nil {
		return err
	}
	if _, err := c.client.CancelExport(caller.attach(ctx), req); err != nil {
		return exportGrpcError(err)
	}
	return nil
}

func exportGrpcError(err error) bizerr.BizError {
	if grpcStatus, ok := status.FromError(err); ok {
		return GrpcStatus2BizError(grpcStatus)
	}
	return bizerr.WrapError(err, "call management Export service")
}