iotlcm/worker-svr/internal/fullsync/contract/replay.go

package contract

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"strings"
	"time"
	"unicode"
	"unicode/utf8"
)

const MaxIdempotencyKeyBytes = 256

// ReplayRequest combines the path, header and body. Only requestedBy is in the
// HTTP body. Idempotency keys are scoped to the source run; fingerprints cover
// the source and exact request body, not the key or a generated target run ID.
type ReplayRequest struct {
	SourceRunID    string `json:"-"`
	IdempotencyKey string `json:"-"`
	RequestedBy    string `json:"requestedBy"`
}

func ValidateIdempotencyKey(key string) error {
	if strings.TrimSpace(key) == "" || len(key) > MaxIdempotencyKeyBytes || !utf8.ValidString(key) || strings.ContainsFunc(key, unicode.IsControl) {
		return invalid("invalid Idempotency-Key")
	}
	return nil
}

func (r ReplayRequest) Validate() error {
	if err := ValidateID(r.SourceRunID); err != nil {
		return err
	}
	if err := ValidateIdempotencyKey(r.IdempotencyKey); err != nil {
		return err
	}
	if strings.TrimSpace(r.RequestedBy) == "" || !utf8.ValidString(r.RequestedBy) {
		return invalid("requestedBy is required")
	}
	return nil
}

func (r ReplayRequest) Fingerprint() (string, error) {
	if err := r.Validate(); err != nil {
		return "", err
	}
	// JSON avoids ambiguous delimiter concatenation and fixes field ordering.
	data, err := json.Marshal(struct {
		SourceRunID string `json:"sourceRunId"`
		RequestedBy string `json:"requestedBy"`
	}{r.SourceRunID, r.RequestedBy})
	if err != nil {
		return "", invalid("cannot encode replay request: %v", err)
	}
	digest := sha256.Sum256(data)
	return hex.EncodeToString(digest[:]), nil
}

// Counters describe distinct source failures, not scan visits or retry calls.
// ScanCursor alone is not an enqueue/cleanup checkpoint: repositories must
// retain sufficient durable per-batch evidence to resume partial writes.
type ReplayProgress struct {
	ScanCursor      Cursor `json:"scanCursor"`
	DiscoveredCount int64  `json:"discoveredCount"`
	EnqueuedCount   int64  `json:"enqueuedCount"`
	CleanedCount    int64  `json:"cleanedCount"`
	FailureCount    int64  `json:"failureCount"`
}

func (p ReplayProgress) Validate() error {
	if err := nonnegative(p.DiscoveredCount, p.EnqueuedCount, p.CleanedCount, p.FailureCount); err != nil {
		return err
	}
	if p.EnqueuedCount > p.DiscoveredCount || p.CleanedCount > p.EnqueuedCount {
		return invalid("replay progress must satisfy cleaned <= enqueued <= discovered")
	}
	return nil
}

type ReplayAction struct {
	ActionID       string          `json:"actionId"`
	SourceRunID    string          `json:"sourceRunId"`
	TargetRunID    string          `json:"targetRunId"`
	RequestedBy    string          `json:"requestedBy"`
	IdempotencyKey string          `json:"-"`
	Fingerprint    string          `json:"-"`
	Status         ActionStatus    `json:"status"`
	ReplayCycle    int             `json:"replayCycle"`
	Progress       ReplayProgress  `json:"progress"`
	LeaseOwner     string          `json:"-"`
	LeaseToken     string          `json:"-"`
	LeaseUntil     time.Time       `json:"-"`
	Revision       uint64          `json:"-"`
	Error          string          `json:"error,omitempty"`
	CreatedAt      time.Time       `json:"createdAt"`
	UpdatedAt      time.Time       `json:"updatedAt"`
	CompletedAt    *time.Time      `json:"completedAt,omitempty"`
	Receipt        *ArchiveReceipt `json:"receipt,omitempty"`
}

func (a ReplayAction) Validate() error {
	for _, id := range []string{a.ActionID, a.SourceRunID, a.TargetRunID} {
		if err := ValidateID(id); err != nil {
			return err
		}
	}
	if a.SourceRunID == a.TargetRunID || a.ReplayCycle < 1 || !a.Status.Valid() || strings.TrimSpace(a.RequestedBy) == "" {
		return invalid("invalid replay action identity or status")
	}
	if len(a.Error) > MaxErrorMessageBytes || !utf8.ValidString(a.Error) {
		return invalid("action error exceeds message bounds")
	}
	if err := a.Progress.Validate(); err != nil {
		return err
	}
	if a.Receipt != nil {
		if err := a.Receipt.Validate(); err != nil {
			return err
		}
	}
	if a.Status == ActionCompleted && (a.Receipt == nil || a.CompletedAt == nil || a.CompletedAt.IsZero()) {
		return invalid("completed action requires a durable receipt and completedAt")
	}
	return nil
}

type ReplayActionPage struct {
	Items      []ReplayAction `json:"items"`
	NextCursor Cursor         `json:"nextCursor"`
}

type ReplayActionCreation struct {
	Action  ReplayAction
	Created bool
}

// Lease token fences a particular acquisition, even if the same owner name is
// reused. Repository methods check expiry against their authoritative clock.
type ReplayLease struct {
	ActionID string
	Owner    string
	Token    string
	Until    time.Time
}

// ReplayActionUpdate is a compare-and-swap of mutable action state. Identity,
// provenance, request fingerprint, and target run must never be changed.
type ReplayActionUpdate struct {
	Lease            ReplayLease
	ExpectedRevision uint64
	Status           ActionStatus
	Progress         ReplayProgress
	Error            string
	Receipt          *ArchiveReceipt
	CompletedAt      *time.Time
}

// Cleanup contains the exact source versions durably accepted by the target.
// The engine must not request cleanup until target enqueue is confirmed.
type ReplayCleanup struct {
	Lease       ReplayLease
	SourceRunID string
	TargetRunID string
	Entries     []FailureEntry
}

type ReplayCleanupResult struct {
	CleanedCount        int64
	AlreadyCleanedCount int64
	ConflictCount       int64
}