iotlcm/server/internal/service/fullsyncretry/client.go

package fullsyncretry

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/gogf/gf/v2/errors/gcode"
	"github.com/gogf/gf/v2/errors/gerror"
)

const maxResponse = 512 << 10

var ErrConfig = errors.New("full-sync worker is not configured")

type Client struct {
	BaseURL string
	HTTP    *http.Client
}

type Receipt struct {
	ReplayRunID string `json:"replayRunId"`
	Status      string `json:"status"`
}

type Status struct {
	ReplayRunID string `json:"replayRunId"`
	SourceRunID string `json:"sourceRunId"`
	Status      string `json:"status"`
	Remaining   int64  `json:"remaining"`
	Enqueued    int64  `json:"enqueued"`
	RunStatus   string `json:"runStatus,omitempty"`
	Error       string `json:"error,omitempty"`
}

type RunMeta struct {
	RunID                string     `json:"runId"`
	Operation            string     `json:"operation"`
	QueryMonth           string     `json:"queryMonth"`
	Status               string     `json:"status"`
	ProducerDone         bool       `json:"producerDone"`
	CurrentRound         int        `json:"currentRound"`
	ActiveCount          int64      `json:"activeCount"`
	InflightCount        int64      `json:"inflightCount"`
	NextRoundCount       int64      `json:"nextRoundCount"`
	SuccessCount         int64      `json:"successCount"`
	FailedCount          int64      `json:"failedCount"`
	CreatedAt            time.Time  `json:"createdAt"`
	ProducerCompletedAt  time.Time  `json:"producerCompletedAt"`
	CompletedAt          time.Time  `json:"completedAt"`
	SyncType             string     `json:"syncType,omitempty"`
	SourceRunID          string     `json:"sourceRunId,omitempty"`
	ReplayCycle          int        `json:"replayCycle,omitempty"`
	TriggerSource        string     `json:"triggerSource,omitempty"`
	ScannedCount         int64      `json:"scannedCount,omitempty"`
	EnqueuedCount        int64      `json:"enqueuedCount,omitempty"`
	UpstreamAttemptCount int64      `json:"upstreamAttemptCount,omitempty"`
	BatchCount           int64      `json:"batchCount,omitempty"`
	MaxRound             int        `json:"maxRound,omitempty"`
	PendingFailureCount  int64      `json:"pendingFailureCount,omitempty"`
	StartedAt            *time.Time `json:"startedAt,omitempty"`
	ScanCompletedAt      *time.Time `json:"scanCompletedAt,omitempty"`
	ArchiveStatus        string     `json:"archiveStatus,omitempty"`
	ArchivedAt           *time.Time `json:"archivedAt,omitempty"`
}

type RoundStats struct {
	RunID                string     `json:"runId"`
	Round                int        `json:"round"`
	EnqueuedCount        int64      `json:"enqueuedCount"`
	ActiveCount          int64      `json:"activeCount"`
	NextRoundCount       int64      `json:"nextRoundCount"`
	SuccessCount         int64      `json:"successCount"`
	FailedCount          int64      `json:"failedCount"`
	UpstreamAttemptCount int64      `json:"upstreamAttemptCount"`
	BatchCount           int64      `json:"batchCount"`
	StartedAt            *time.Time `json:"startedAt,omitempty"`
	CompletedAt          *time.Time `json:"completedAt,omitempty"`
}

type RunView struct {
	Source string       `json:"source"`
	Run    RunMeta      `json:"run"`
	Rounds []RoundStats `json:"rounds,omitempty"`
}

type RunList struct {
	Items []RunView `json:"items"`
	Total int       `json:"total"`
}

type Failure struct {
	TaskID        string    `json:"taskId"`
	CardID        int64     `json:"cardId"`
	ErrorCategory string    `json:"errorCategory"`
	ErrorMessage  string    `json:"errorMessage"`
	FailedAt      time.Time `json:"failedAt"`
	Round         int       `json:"round"`
}

type FailurePage struct {
	Items      []Failure `json:"items"`
	NextCursor string    `json:"nextCursor"`
}

type StartReceipt struct {
	RunID    string `json:"runId"`
	SyncKind string `json:"syncKind"`
	Status   string `json:"status"`
}

func FromEnvironment() Client {
	return Client{BaseURL: "http://iotlcm-worker-svr:18081"}
}

func ValidateID(id string) bool {
	if len(id) == 0 || len(id) > 128 {
		return false
	}
	for _, c := range id {
		if c < '0' || c > '9' {
			if c < 'a' || c > 'z' {
				if c < 'A' || c > 'Z' {
					if c != '-' && c != '_' && c != ':' && c != '.' {
						return false
					}
				}
			}
		}
	}
	return true
}

// MaxReplayRounds mirrors the Worker's server-side ceiling so an oversized
// request is rejected at the edge instead of surfacing as an opaque 400.
const MaxReplayRounds = 100

// Replay requests a re-run of the source run's failures. maxRounds == 0 asks
// the Worker to apply its configured default; any other value is passed
// through and must not exceed MaxReplayRounds.
func (c Client) Replay(ctx context.Context, id, actor string, maxRounds int) (Receipt, error) {
	if !ValidateID(id) || actor == "" {
		return Receipt{}, gerror.NewCode(gcode.CodeInvalidParameter, "Invalid replay request")
	}
	if maxRounds < 0 || maxRounds > MaxReplayRounds {
		return Receipt{}, gerror.NewCode(gcode.CodeInvalidParameter, "Invalid replay maxRounds")
	}
	body, _ := json.Marshal(struct {
		RequestedBy string `json:"requestedBy"`
		MaxRounds   int    `json:"maxRounds,omitempty"`
	}{actor, maxRounds})
	var receipt Receipt
	if err := c.do(ctx, http.MethodPost, "/internal/v1/full-sync-runs/"+url.PathEscape(id)+"/replay", body, http.StatusAccepted, &receipt); err != nil {
		return Receipt{}, err
	}
	if !ValidateID(receipt.ReplayRunID) {
		return Receipt{}, gerror.NewCode(gcode.CodeInternalError, "Invalid Worker response")
	}
	return receipt, nil
}

// ValidSyncKind matches the Worker's full-sync kinds for both monitoring and starts.
func ValidSyncKind(kind string) bool {
	switch kind {
	case "traffic_usage", "sim_status", "imei", "traffic_usage_daily", "traffic_usage_monthly":
		return true
	default:
		return false
	}
}

func (c Client) ListRuns(ctx context.Context, kind string, limit int) (RunList, error) {
	if (kind != "" && !ValidSyncKind(kind)) || limit < 0 || limit > 100 {
		return RunList{}, gerror.NewCode(gcode.CodeInvalidParameter, "Invalid full-sync filter")
	}
	query := url.Values{}
	if kind != "" {
		query.Set("syncKind", kind)
	}
	if limit > 0 {
		query.Set("limit", fmt.Sprint(limit))
	}
	path := "/internal/v1/full-sync-runs"
	if len(query) > 0 {
		path += "?" + query.Encode()
	}
	var result RunList
	if err := c.do(ctx, http.MethodGet, path, nil, http.StatusOK, &result); err != nil {
		return RunList{}, err
	}
	if result.Items == nil {
		result.Items = []RunView{}
	}
	return result, nil
}

func (c Client) Run(ctx context.Context, id string) (RunView, error) {
	if !ValidateID(id) {
		return RunView{}, gerror.NewCode(gcode.CodeInvalidParameter, "Invalid run ID")
	}
	var result RunView
	if err := c.do(ctx, http.MethodGet, "/internal/v1/full-sync-runs/"+url.PathEscape(id), nil, http.StatusOK, &result); err != nil {
		return RunView{}, err
	}
	if result.Run.RunID != id {
		return RunView{}, gerror.NewCode(gcode.CodeInternalError, "Invalid Worker response")
	}
	return result, nil
}

func (c Client) Failures(ctx context.Context, id string, limit int, cursor string) (FailurePage, error) {
	if !ValidateID(id) || limit < 0 || limit > 500 || len(cursor) > 1024 {
		return FailurePage{}, gerror.NewCode(gcode.CodeInvalidParameter, "Invalid failure query")
	}
	query := url.Values{}
	if limit > 0 {
		query.Set("limit", fmt.Sprint(limit))
	}
	if cursor != "" {
		query.Set("cursor", cursor)
	}
	path := "/internal/v1/full-sync-runs/" + url.PathEscape(id) + "/failures"
	if len(query) > 0 {
		path += "?" + query.Encode()
	}
	var result FailurePage
	if err := c.do(ctx, http.MethodGet, path, nil, http.StatusOK, &result); err != nil {
		return FailurePage{}, err
	}
	if result.Items == nil {
		result.Items = []Failure{}
	}
	return result, nil
}

func (c Client) Get(ctx context.Context, id string) (Status, error) {
	if !ValidateID(id) {
		return Status{}, gerror.NewCode(gcode.CodeInvalidParameter, "Invalid replay run ID")
	}
	var status Status
	if err := c.do(ctx, http.MethodGet, "/internal/v1/full-sync-replays/"+url.PathEscape(id), nil, http.StatusOK, &status); err != nil {
		return Status{}, err
	}
	if status.ReplayRunID != id {
		return Status{}, gerror.NewCode(gcode.CodeInternalError, "Invalid Worker response")
	}
	return status, nil
}

func (c Client) do(ctx context.Context, method, path string, body []byte, expected int, result any) error {
	base, err := url.Parse(c.BaseURL)
	if err != nil || base == nil || base.Scheme != "http" || base.Host == "" || base.User != nil || base.RawQuery != "" || base.Fragment != "" || base.Path != "" {
		return gerror.NewCode(gcode.CodeInternalError, ErrConfig.Error())
	}
	client := c.HTTP
	if client == nil {
		client = &http.Client{Timeout: 5 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
	}
	request, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(base.String(), "/")+path, bytes.NewReader(body))
	if err != nil {
		return gerror.NewCode(gcode.CodeInternalError, "Invalid Worker request")
	}
	if body != nil {
		request.Header.Set("Content-Type", "application/json")
	}
	response, err := client.Do(request)
	if err != nil {
		return gerror.NewCode(gcode.CodeInternalError, "Worker unavailable")
	}
	defer response.Body.Close()
	if response.StatusCode != expected {
		return workerError(response.StatusCode)
	}
	decoder := json.NewDecoder(io.LimitReader(response.Body, maxResponse))
	if err := decoder.Decode(result); err != nil {
		return gerror.NewCode(gcode.CodeInternalError, "Invalid Worker response")
	}
	return nil
}

func workerError(status int) error {
	switch status {
	case http.StatusBadRequest:
		return gerror.NewCode(gcode.CodeInvalidParameter, "Invalid replay request")
	case http.StatusNotFound:
		return gerror.NewCode(gcode.CodeNotFound, "Run not found")
	case http.StatusGone:
		return gerror.NewCode(gcode.CodeNotFound, "Failure details are no longer available")
	case http.StatusServiceUnavailable:
		return gerror.NewCode(gcode.CodeInternalError, "Worker is busy")
	case http.StatusConflict:
		return gerror.NewCode(gcode.CodeInvalidParameter, "Run is not ready for replay")
	case http.StatusUnprocessableEntity:
		return gerror.NewCode(gcode.CodeInvalidParameter, "Run has no failures")
	default:
		return gerror.NewCode(gcode.CodeInternalError, fmt.Sprintf("Worker failed (HTTP %d)", status))
	}
}