// Package runartifact owns the bounded, identity-bound representation used to
// carry Workflow Run values that cannot fit one ordinary reducer payload.
// References are data, not capabilities: every store/API consumer must resolve
// them against the complete immutable identity and verify the stored digest.
package runartifact

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
)

const (
	// Kind is part of the immutable artifact identity. Keep the discriminator
	// meaningful: the same storage/read protocol is also used for reducer
	// values and oversized effect text, not only Join aggregates.
	KindJoinAggregate = "workflow_join_aggregate"
	KindEffectText    = "workflow_effect_text"
	KindReducerValue  = "workflow_reducer_value"
	Version           = "workflow.run.artifact/v1"
	MarkerKey         = "$workflowArtifact"

	// MaxArtifactBytes is deliberately larger than one ordinary Run payload,
	// while remaining below the configured TiDB LONG BLOB and page consumers'
	// practical limits. It is a hard storage/consumer budget, not a reducer
	// payload limit.
	MaxArtifactBytes = 64 << 20
	// MaxPageBytes bounds every materialized read response and every page carried
	// through a CLI/API envelope.
	MaxPageBytes        = 256 << 10
	DefaultPageBytes    = 64 << 10
	MaxPathSegments     = 32
	MaxPathSegmentBytes = 256
	MaxCursorBytes      = 16 << 10
)

var (
	ErrInvalidReference       = errors.New("workflow artifact reference is invalid")
	ErrUnauthorizedReference  = errors.New("workflow artifact reference is unauthorized")
	ErrStaleReference         = errors.New("workflow artifact reference is stale")
	ErrArtifactBudgetExceeded = errors.New("workflow artifact exceeds durable artifact budget")
	ErrInvalidCursor          = errors.New("workflow artifact cursor is invalid")
	ErrInvalidRead            = errors.New("workflow artifact read is invalid")
)

type Ref struct {
	Kind                 string `json:"kind"`
	WorkspaceID          string `json:"workspaceId"`
	RunID                string `json:"runId"`
	ProducerNodeID       string `json:"producerNodeId"`
	ProducerActivationID string `json:"producerActivationId"`
	ArtifactID           string `json:"artifactId"`
	Version              string `json:"version"`
	CanonicalByteLength  int64  `json:"canonicalByteLength"`
	Digest               string `json:"digest"`
}

// Draft is the one transaction-boundary write intent for an immutable artifact.
// ArtifactID is deterministic from the complete producer identity and digest;
// replay therefore performs an idempotent insert of the same row.
type Draft struct {
	Ref     Ref
	Content []byte
}

func (ref Ref) Validate() error {
	switch ref.Kind {
	case KindJoinAggregate, KindEffectText, KindReducerValue:
	default:
		return fmt.Errorf("%w: unsupported artifact kind", ErrInvalidReference)
	}
	if ref.Version != Version {
		return fmt.Errorf("%w: unsupported artifact version", ErrInvalidReference)
	}
	for name, value := range map[string]string{
		"workspaceId": ref.WorkspaceID, "runId": ref.RunID,
		"producerNodeId": ref.ProducerNodeID, "producerActivationId": ref.ProducerActivationID,
		"artifactId": ref.ArtifactID, "digest": ref.Digest,
	} {
		if strings.TrimSpace(value) == "" || value != strings.TrimSpace(value) {
			return fmt.Errorf("%w: %s is missing or non-canonical", ErrInvalidReference, name)
		}
	}
	if !strings.HasPrefix(ref.ArtifactID, "wfa_") || len(ref.ArtifactID) != 68 || ref.ArtifactID != artifactID(ref) {
		return fmt.Errorf("%w: artifactId does not match its immutable identity", ErrInvalidReference)
	}
	if !strings.HasPrefix(ref.Digest, "sha256:") || len(ref.Digest) != len("sha256:")+64 {
		return fmt.Errorf("%w: digest is invalid", ErrInvalidReference)
	}
	if _, err := hex.DecodeString(strings.TrimPrefix(ref.Digest, "sha256:")); err != nil {
		return fmt.Errorf("%w: digest is invalid", ErrInvalidReference)
	}
	if ref.CanonicalByteLength < 1 || ref.CanonicalByteLength > MaxArtifactBytes {
		return fmt.Errorf("%w: canonical byte length is outside the artifact budget", ErrInvalidReference)
	}
	return nil
}

func CanonicalJSON(value any) ([]byte, error) {
	payload, err := json.Marshal(value)
	if err != nil {
		return nil, fmt.Errorf("encode canonical workflow artifact: %w", err)
	}
	if len(payload) == 0 || len(payload) > MaxArtifactBytes {
		return nil, fmt.Errorf("%w: canonical bytes=%d maximum=%d", ErrArtifactBudgetExceeded, len(payload), MaxArtifactBytes)
	}
	return payload, nil
}

func Digest(content []byte) string {
	sum := sha256.Sum256(content)
	return "sha256:" + hex.EncodeToString(sum[:])
}

// NewDraftForKind creates an immutable artifact for one of the closed logical
// kinds. The kind is part of the derived identity and storage uniqueness key.
func NewDraftForKind(kind, workspaceID, runID, nodeID, activationID string, content []byte) (Draft, error) {
	content = append([]byte(nil), content...)
	if len(content) == 0 || len(content) > MaxArtifactBytes {
		return Draft{}, fmt.Errorf("%w: canonical bytes=%d maximum=%d", ErrArtifactBudgetExceeded, len(content), MaxArtifactBytes)
	}
	kind = strings.TrimSpace(kind)
	workspaceID, runID, nodeID, activationID = strings.TrimSpace(workspaceID), strings.TrimSpace(runID), strings.TrimSpace(nodeID), strings.TrimSpace(activationID)
	if workspaceID == "" || runID == "" || nodeID == "" || activationID == "" {
		return Draft{}, fmt.Errorf("%w: producer identity is incomplete", ErrInvalidReference)
	}
	digest := Digest(content)
	ref := Ref{
		Kind: kind, WorkspaceID: workspaceID, RunID: runID,
		ProducerNodeID: nodeID, ProducerActivationID: activationID,
		Version: Version, CanonicalByteLength: int64(len(content)), Digest: digest,
	}
	ref.ArtifactID = artifactID(ref)
	if err := ref.Validate(); err != nil {
		return Draft{}, err
	}
	return Draft{Ref: ref, Content: content}, nil
}

func artifactID(ref Ref) string {
	hash := sha256.New()
	_, _ = hash.Write([]byte("loop.workflow.run.artifact/v2\x00"))
	for _, value := range []string{ref.Kind, ref.WorkspaceID, ref.RunID, ref.ProducerNodeID, ref.ProducerActivationID, ref.Version, ref.Digest} {
		_, _ = hash.Write([]byte(fmt.Sprintf("%d:", len(value))))
		_, _ = hash.Write([]byte(value))
	}
	return "wfa_" + hex.EncodeToString(hash.Sum(nil))
}

func Marker(ref Ref) map[string]any {
	return map[string]any{MarkerKey: ref}
}

// ExternalizeObjectField replaces one oversized JSON object field with the
// marker returned by add. It keeps the JSON boundary in the artifact package
// so Store code persists references without owning a wire encoder.
func ExternalizeObjectField(raw []byte, field string, threshold int, add func([]byte) ([]byte, error)) ([]byte, error) {
	if add == nil || threshold < 1 {
		return nil, ErrUnauthorizedReference
	}
	var fields map[string]json.RawMessage
	if err := json.Unmarshal(raw, &fields); err != nil {
		return nil, err
	}
	value, ok := fields[field]
	if !ok || len(value) <= threshold {
		return raw, nil
	}
	marker, err := add(value)
	if err != nil {
		return nil, err
	}
	fields[field] = marker
	return json.Marshal(fields)
}

// ReadCallbackRecoveryReceipt extracts the two durable action fields without
// allowing arbitrary result metadata to become recovery authority.
func ReadCallbackRecoveryReceipt(raw []byte) (string, string, error) {
	if len(strings.TrimSpace(string(raw))) == 0 {
		return "", "", nil
	}
	var value struct {
		CallbackAction               string
		CallbackActionIdempotencyKey string
	}
	if err := json.Unmarshal(raw, &value); err != nil {
		return "", "", err
	}
	return strings.TrimSpace(value.CallbackAction), strings.TrimSpace(value.CallbackActionIdempotencyKey), nil
}

// WriteCallbackRecoveryReceipt records the closed non-terminal callback action
// receipt while preserving unrelated result fields for replay diagnostics.
func WriteCallbackRecoveryReceipt(raw []byte, action, idempotencyKey string) ([]byte, error) {
	var fields map[string]json.RawMessage
	if len(strings.TrimSpace(string(raw))) == 0 {
		fields = make(map[string]json.RawMessage)
	} else if err := json.Unmarshal(raw, &fields); err != nil || fields == nil {
		return nil, fmt.Errorf("workflow Run callback recovery receipt is invalid")
	}
	encodedAction, _ := json.Marshal(action)
	encodedKey, _ := json.Marshal(idempotencyKey)
	fields["callbackAction"] = encodedAction
	fields["callbackActionIdempotencyKey"] = encodedKey
	encoded, err := json.Marshal(fields)
	if err != nil {
		return nil, fmt.Errorf("encode workflow Run callback recovery receipt: %w", err)
	}
	return encoded, nil
}

// ParseMarker recognizes only the closed one-field marker shape. A marker-like
// object supplied by a caller is still untrusted and must fail at store
// resolution when its complete binding/digest is not present in the database.
func ParseMarker(value any) (Ref, bool, error) {
	object, ok := value.(map[string]any)
	if !ok {
		return Ref{}, false, nil
	}
	raw, present := object[MarkerKey]
	if !present {
		return Ref{}, false, nil
	}
	if len(object) != 1 {
		return Ref{}, true, fmt.Errorf("%w: marker has unknown fields", ErrInvalidReference)
	}
	payload, err := json.Marshal(raw)
	if err != nil {
		return Ref{}, true, fmt.Errorf("%w: marker cannot be encoded", ErrInvalidReference)
	}
	var ref Ref
	decoder := json.NewDecoder(strings.NewReader(string(payload)))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&ref); err != nil {
		return Ref{}, true, fmt.Errorf("%w: marker reference is malformed", ErrInvalidReference)
	}
	if err := ref.Validate(); err != nil {
		return Ref{}, true, err
	}
	return ref, true, nil
}

// FindMarkerAtPath finds a marker at the requested envelope path and returns
// the remaining path that must be evaluated inside the immutable artifact.
// Markers may be nested in dependency/business-value objects; callers still
// authorize the returned reference before materializing any bytes.
func FindMarkerAtPath(raw []byte, path []string) (Ref, []string, bool, error) {
	for prefix := 0; prefix <= len(path); prefix++ {
		selected, err := SelectPath(raw, path[:prefix])
		if err != nil {
			return Ref{}, nil, false, nil
		}
		var object map[string]any
		decoder := json.NewDecoder(strings.NewReader(string(selected)))
		decoder.UseNumber()
		if err := decoder.Decode(&object); err != nil {
			continue
		}
		ref, marker, err := ParseMarker(object)
		if marker || err != nil {
			return ref, append([]string(nil), path[prefix:]...), marker, err
		}
	}
	return Ref{}, nil, false, nil
}

// ExpandJSON replaces every trusted marker by its canonical JSON content. The
// resolver is responsible for workspace/Run authorization and row/digest
// validation; this package never resolves by artifact ID alone.
func ExpandJSON(raw []byte, resolve func(Ref) ([]byte, error)) ([]byte, error) {
	if resolve == nil {
		return nil, ErrUnauthorizedReference
	}
	decoder := json.NewDecoder(strings.NewReader(string(raw)))
	decoder.UseNumber()
	var value any
	if err := decoder.Decode(&value); err != nil {
		return nil, fmt.Errorf("decode workflow artifact envelope: %w", err)
	}
	expanded, err := expandValue(value, resolve)
	if err != nil {
		return nil, err
	}
	return json.Marshal(expanded)
}

func expandValue(value any, resolve func(Ref) ([]byte, error)) (any, error) {
	if ref, marker, err := ParseMarker(value); marker {
		if err != nil {
			return nil, err
		}
		content, err := resolve(ref)
		if err != nil {
			return nil, err
		}
		if Digest(content) != ref.Digest || int64(len(content)) != ref.CanonicalByteLength {
			return nil, fmt.Errorf("%w: artifact %s content digest/length changed", ErrStaleReference, ref.ArtifactID)
		}
		decoder := json.NewDecoder(strings.NewReader(string(content)))
		decoder.UseNumber()
		var expanded any
		if err := decoder.Decode(&expanded); err != nil {
			return nil, fmt.Errorf("%w: artifact %s content is not JSON", ErrStaleReference, ref.ArtifactID)
		}
		return expanded, nil
	}
	switch typed := value.(type) {
	case map[string]any:
		for key, child := range typed {
			expanded, err := expandValue(child, resolve)
			if err != nil {
				return nil, err
			}
			typed[key] = expanded
		}
	case []any:
		for index, child := range typed {
			expanded, err := expandValue(child, resolve)
			if err != nil {
				return nil, err
			}
			typed[index] = expanded
		}
	}
	return value, nil
}

// Cursor is intentionally only a frontier and an authenticator. The complete
// reference and read selector are MAC input, not duplicated client data; the
// caller supplies them again and DecodeCursor re-authorizes that binding.
type Cursor struct {
	Offset int    `json:"offset"`
	MAC    string `json:"mac"`
}

type cursorBinding struct {
	Reference Ref    `json:"reference"`
	Section   string `json:"section"`
	Path      string `json:"path,omitempty"`
	Offset    int    `json:"offset"`
	Limit     int    `json:"limit"`
}

func EncodeCursor(ref Ref, section string, path []string, offset, limit int, secret []byte) (string, error) {
	if err := ref.Validate(); err != nil {
		return "", err
	}
	if err := validateRead(section, path, limit); err != nil {
		return "", err
	}
	if offset < 0 {
		return "", fmt.Errorf("%w: offset is negative", ErrInvalidCursor)
	}
	cursor := Cursor{Offset: offset}
	cursor.MAC = cursorMAC(ref, section, path, offset, limit, secret)
	payload, err := json.Marshal(cursor)
	if err != nil {
		return "", err
	}
	encoded := base64.RawURLEncoding.EncodeToString(payload)
	if len(encoded) > MaxCursorBytes {
		return "", fmt.Errorf("%w: cursor is too large", ErrInvalidCursor)
	}
	return encoded, nil
}

func DecodeCursor(raw string, ref Ref, section string, path []string, limit int, secret []byte) (Cursor, error) {
	if len(raw) == 0 || len(raw) > MaxCursorBytes {
		return Cursor{}, ErrInvalidCursor
	}
	if err := ref.Validate(); err != nil {
		return Cursor{}, err
	}
	if err := validateRead(section, path, limit); err != nil {
		return Cursor{}, err
	}
	payload, err := base64.RawURLEncoding.DecodeString(raw)
	if err != nil {
		return Cursor{}, ErrInvalidCursor
	}
	var cursor Cursor
	decoder := json.NewDecoder(strings.NewReader(string(payload)))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&cursor); err != nil || cursor.Offset < 0 || !hmac.Equal([]byte(cursor.MAC), []byte(cursorMAC(ref, section, path, cursor.Offset, limit, secret))) {
		return Cursor{}, ErrInvalidCursor
	}
	return cursor, nil
}

// CanonicalPath is the shared continuation representation: both durable
// artifact cursors and live payload cursors bind JSON Pointer text, while the
// durable resolver may still use segments internally for marker traversal.
func CanonicalPath(path []string) string {
	if len(path) == 0 {
		return ""
	}
	segments := make([]string, len(path))
	for index, segment := range path {
		segments[index] = strings.ReplaceAll(strings.ReplaceAll(segment, "~", "~0"), "/", "~1")
	}
	return "/" + strings.Join(segments, "/")
}

// PathSegments accepts the same JSON-pointer/CLI convenience forms used by
// the public Workflow reads and returns the canonical artifact path segments.
func PathSegments(path string) ([]string, error) {
	path = strings.TrimSpace(path)
	if path == "" || path == "$" || path == "/" {
		return nil, nil
	}
	if strings.HasPrefix(path, "$/") {
		path = path[1:]
	}
	var segments []string
	if strings.HasPrefix(path, "$.") {
		segments = strings.Split(path[2:], ".")
	} else if strings.HasPrefix(path, "/") {
		segments = strings.Split(strings.TrimPrefix(path, "/"), "/")
		for index := range segments {
			segments[index] = strings.ReplaceAll(strings.ReplaceAll(segments[index], "~1", "/"), "~0", "~")
		}
	} else {
		segments = strings.Split(path, ".")
	}
	if err := validateRead("root", segments, DefaultPageBytes); err != nil {
		return nil, err
	}
	return segments, nil
}

func validateRead(section string, path []string, limit int) error {
	if section == "" {
		return fmt.Errorf("%w: section is required", ErrInvalidRead)
	}
	switch section {
	case "root", "input", "output":
	default:
		return fmt.Errorf("%w: unknown section", ErrInvalidRead)
	}
	if len(path) > MaxPathSegments {
		return fmt.Errorf("%w: path is too deep", ErrInvalidRead)
	}
	for _, segment := range path {
		if segment == "" || len(segment) > MaxPathSegmentBytes || strings.ContainsAny(segment, "\x00\r\n") {
			return fmt.Errorf("%w: path segment is invalid", ErrInvalidRead)
		}
	}
	if limit < 1 || limit > MaxPageBytes {
		return fmt.Errorf("%w: limit must be between 1 and %d", ErrInvalidRead, MaxPageBytes)
	}
	return nil
}

func cursorMAC(ref Ref, section string, path []string, offset, limit int, secret []byte) string {
	payload, _ := json.Marshal(cursorBinding{
		Reference: ref, Section: section, Path: CanonicalPath(path), Offset: offset, Limit: limit,
	})
	mac := hmac.New(sha256.New, secret)
	_, _ = mac.Write(payload)
	return hex.EncodeToString(mac.Sum(nil))
}

// SelectPath returns the exact canonical JSON bytes at a selected object/array
// path without re-encoding the selected raw value.
func SelectPath(content []byte, path []string) ([]byte, error) {
	selected := append([]byte(nil), content...)
	for _, segment := range path {
		selected = bytesAtPath(selected, segment)
		if selected == nil {
			return nil, fmt.Errorf("%w: path segment %q was not found", ErrInvalidRead, segment)
		}
	}
	return selected, nil
}

func bytesAtPath(raw []byte, segment string) []byte {
	var object map[string]json.RawMessage
	if len(raw) > 0 && raw[0] == '{' && json.Unmarshal(raw, &object) == nil {
		return append([]byte(nil), object[segment]...)
	}
	var array []json.RawMessage
	if len(raw) > 0 && raw[0] == '[' && json.Unmarshal(raw, &array) == nil {
		index, err := strconv.Atoi(segment)
		if err == nil && strconv.Itoa(index) == segment && index >= 0 && index < len(array) {
			return append([]byte(nil), array[index]...)
		}
	}
	return nil
}