common/sdks/uploader/uploader.go

package uploadersdk

import (
	"fmt"
	"gitlink.org.cn/cloudream/common/pkgs/types"
	sch "gitlink.org.cn/cloudream/common/sdks/pcmscheduler"
	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
	"gitlink.org.cn/cloudream/common/sdks/storage/cdsapi"
	"gitlink.org.cn/cloudream/common/utils/http2"
	"gitlink.org.cn/cloudream/common/utils/serder"
	"net/url"
	"strings"
)

type DataScheduleReq struct {
	PackageID      cdssdk.PackageID `json:"packageID"`
	DataType       string           `json:"dataType"`
	ScheduleTarget ScheduleTarget   `json:"scheduleTarget"`
}

//type DataScheduleResp struct {
//	Results []sch.DataScheduleResult `json:"data"`
//}

type TmpDataScheduleResult struct {
	Cluster   sch.DataDetail   `json:"cluster"`
	PackageID cdssdk.PackageID `json:"packageID"`
	Status    bool             `json:"status"`
	Msg       string           `json:"msg"`
}

func (c *Client) DataSchedule(req DataScheduleReq) ([]sch.DataScheduleResult, error) {
	targetUrl, err := url.JoinPath(c.baseURL, "/dataSchedule")
	if err != nil {
		return nil, err
	}

	resp, err := http2.PostJSON(targetUrl, http2.RequestParam{
		Body: req,
	})
	if err != nil {
		return nil, err
	}
	println(resp.Body)

	contType := resp.Header.Get("Content-Type")
	if strings.Contains(contType, http2.ContentTypeJSON) {
		var codeResp response[[]TmpDataScheduleResult]
		if err := serder.JSONToObjectStream(resp.Body, &codeResp); err != nil {
			return nil, fmt.Errorf("parsing response: %w", err)
		}

		if codeResp.Code == ResponseCodeOK {
			var results []sch.DataScheduleResult
			for _, tmpResult := range codeResp.Data {
				result := sch.DataScheduleResult{
					PackageID: tmpResult.PackageID,
					Status:    tmpResult.Status,
					Msg:       tmpResult.Msg,
					Clusters: []sch.DataDetail{
						tmpResult.Cluster,
					},
				}
				results = append(results, result)
			}
			return results, nil
		}

		return nil, codeResp.ToError()
	}

	return nil, fmt.Errorf("unknow response content type: %s", contType)
}

type UploadReq struct {
	DataType string       `json:"dataType"`
	Source   UploadSource `json:"source"`
	Target   UploadTarget `json:"target"`
	//StorageIDs []cdssdk.StorageID `json:"storageIDs"`
}

type UploadSource interface {
	Noop()
}

var UploadSourceTypeUnion = types.NewTypeUnion[UploadSource](
	(*PackageSource)(nil),
	(*UrlSource)(nil),
)

var _ = serder.UseTypeUnionInternallyTagged(&UploadSourceTypeUnion, "type")

type PackageSource struct {
	serder.Metadata `union:"jcs"`
	UploadSourceBase
	Type      string           `json:"type"`
	PackageID cdssdk.PackageID `json:"packageID"`
}

type UrlSource struct {
	serder.Metadata `union:"url"`
	UploadSourceBase
	Type string `json:"type"`
	Url  string `json:"url"`
}

type UploadSourceBase struct{}

func (d *UploadSourceBase) Noop() {}

type UploadTarget interface {
	Noop()
}

var UploadTargetTypeUnion = types.NewTypeUnion[UploadTarget](
	(*UrlTarget)(nil),
	(*ApiTarget)(nil),
)

var _ = serder.UseTypeUnionInternallyTagged(&UploadTargetTypeUnion, "type")

type UrlTarget struct {
	serder.Metadata `union:"url"`
	UploadTargetBase
	Type          string                  `json:"type"`
	ClusterID     ClusterID               `json:"clusterId"`
	JCSUploadInfo cdsapi.ObjectUploadInfo `form:"JCSUploadInfo"`
}

type ApiTarget struct {
	serder.Metadata `union:"api"`
	UploadTargetBase
	Type     string      `json:"type"`
	Clusters []ClusterID `json:"clusters"`
}

type UploadTargetBase struct{}

func (d *UploadTargetBase) Noop() {}

type UploadResp struct {
	PackageID cdssdk.PackageID  `json:"packageID"`
	ObjectIDs []cdssdk.ObjectID `json:"objectIDs"`
	ClusterID ClusterID         `json:"clusterID"`
	JsonData  string            `json:"jsonData"`
	Status    bool              `json:"status"`
	Message   string            `json:"message"`
}

func (c *Client) Upload(req UploadReq) (*UploadResp, error) {
	targetUrl, err := url.JoinPath(c.baseURL, "/dataUpload")
	if err != nil {
		return nil, err
	}

	resp, err := http2.PostJSON(targetUrl, http2.RequestParam{
		Body: req,
	})
	if err != nil {
		return nil, err
	}

	contType := resp.Header.Get("Content-Type")
	if strings.Contains(contType, http2.ContentTypeJSON) {
		var codeResp response[UploadResp]
		if err := serder.JSONToObjectStream(resp.Body, &codeResp); err != nil {
			return nil, fmt.Errorf("parsing response: %w", err)
		}

		if codeResp.Code == ResponseCodeOK {
			return &codeResp.Data, nil
		}

		return nil, codeResp.ToError()
	}

	return nil, fmt.Errorf("unknow response content type: %s", contType)
}