go-sip/internal/contract/contract.go

package contract

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"
	"unicode/utf8"

	"git.ipao.vip/rogee/go-sip/contracts"
	"github.com/santhosh-tekuri/jsonschema/v6"
)

type CommandEnvelope struct {
	DispatcherID  string          `json:"dispatcher_id,omitempty"`
	SchemaVersion string          `json:"schema_version"`
	CommandType   string          `json:"command_type"`
	CommandID     string          `json:"command_id"`
	TenantID      string          `json:"tenant_id"`
	TenantKey     string          `json:"tenant_key"`
	TraceID       string          `json:"trace_id"`
	IssuedAt      string          `json:"issued_at"`
	NotAfter      string          `json:"not_after"`
	Payload       json.RawMessage `json:"payload"`
}

type ExecutePayload struct {
	ExecutionID       string         `json:"execution_id"`
	TaskID            string         `json:"task_id"`
	TaskItemID        string         `json:"task_item_id"`
	TaskRevision      int64          `json:"task_revision"`
	Callee            string         `json:"callee"`
	RoutePolicyID     string         `json:"route_policy_id"`
	CallerProfileID   string         `json:"caller_profile_id"`
	AgentVersionID    string         `json:"agent_version_id"`
	Variables         map[string]any `json:"variables"`
	RingTimeoutMS     int64          `json:"ring_timeout_ms"`
	MaxCallDurationMS int64          `json:"max_call_duration_ms"`
}

type EventEnvelope struct {
	SchemaVersion    string         `json:"schema_version"`
	EventID          string         `json:"event_id"`
	EventType        string         `json:"event_type"`
	TenantID         string         `json:"tenant_id"`
	TenantKey        string         `json:"tenant_key"`
	TraceID          string         `json:"trace_id"`
	OccurredAt       string         `json:"occurred_at"`
	AggregateType    string         `json:"aggregate_type"`
	AggregateID      string         `json:"aggregate_id"`
	AggregateVersion int64          `json:"aggregate_version"`
	Payload          map[string]any `json:"payload"`
}

var ErrInvalidTenantKey = errors.New("invalid tenant_key")

// ValidateJSON applies the imported JSON Schema. It intentionally validates
// the source contract instead of maintaining a second hand-written schema.
func ValidateJSON(raw []byte) error {
	return ValidateSourceSchema("mq.schema.json", raw)
}

// ValidateEvent applies the event-specific payload contract in addition to the
// generic MQ envelope contract.
func ValidateEvent(raw []byte) error {
	return ValidateSourceSchema("event-payloads.schema.json", raw)
}

func ValidateLocalConfigRead(raw []byte) error {
	return validateLocalSchema("config-read-v0.1.schema.json", raw)
}

func ValidateLocalTaskDiscovery(raw []byte) error {
	return validateLocalSchema("task-discovery-v0.2-proposal.schema.json", raw)
}

func ValidateLocalCommandNext(raw []byte) error {
	return validateLocalSchema("command-next-v0.1-proposal.schema.json", raw)
}

func ValidateLocalCallResult(raw []byte) error {
	return validateLocalSchema("call-result-v0.1-proposal.schema.json", raw)
}

func validateLocalSchema(schemaName string, raw []byte) error {
	value, err := jsonschema.UnmarshalJSON(bytes.NewReader(raw))
	if err != nil {
		return fmt.Errorf("decode %s response: %w", schemaName, err)
	}
	schema, err := localSchemaFromBundle(schemaName)
	if err != nil {
		return err
	}
	if err := schema.Validate(value); err != nil {
		var validation *jsonschema.ValidationError
		if errors.As(err, &validation) {
			return fmt.Errorf("%s validation failed at /%s", schemaName, strings.Join(validation.InstanceLocation, "/"))
		}
		return fmt.Errorf("%s validation failed", schemaName)
	}
	return nil
}

func ValidateSourceSchema(schemaName string, raw []byte) error {
	value, err := jsonschema.UnmarshalJSON(bytes.NewReader(raw))
	if err != nil {
		return fmt.Errorf("decode json: %w", err)
	}
	schema, err := schemaFromBundle(contracts.SourceCommit, schemaName)
	if err != nil {
		return err
	}
	if err := schema.Validate(value); err != nil {
		return fmt.Errorf("%s validation: %w", schemaName, err)
	}
	return nil
}

func DecodeExecute(raw []byte) (CommandEnvelope, ExecutePayload, error) {
	envelope, err := DecodeMQCommand(raw)
	if err != nil {
		return CommandEnvelope{}, ExecutePayload{}, err
	}
	if envelope.CommandType != "call.execute" {
		return CommandEnvelope{}, ExecutePayload{}, fmt.Errorf("unsupported command_type %q", envelope.CommandType)
	}
	if err := ValidateTenantKey(envelope.TenantKey); err != nil {
		return CommandEnvelope{}, ExecutePayload{}, err
	}
	var payload ExecutePayload
	if err := json.Unmarshal(envelope.Payload, &payload); err != nil {
		return CommandEnvelope{}, ExecutePayload{}, fmt.Errorf("decode call.execute payload: %w", err)
	}
	return envelope, payload, nil
}

func ValidateTenantKey(key string) error {
	if key == "" || !utf8.ValidString(key) {
		return fmt.Errorf("%w: must be non-empty valid UTF-8", ErrInvalidTenantKey)
	}
	if len([]byte(key)) > 224 {
		return fmt.Errorf("%w: %d UTF-8 bytes exceeds 224-byte routing budget", ErrInvalidTenantKey, len([]byte(key)))
	}
	return nil
}

func NotAfterExpired(raw string, now time.Time) (bool, error) {
	deadline, err := time.Parse(time.RFC3339Nano, raw)
	if err != nil {
		return false, fmt.Errorf("parse not_after: %w", err)
	}
	return !now.Before(deadline), nil
}

func CloneJSON(raw []byte) json.RawMessage {
	return bytes.Clone(raw)
}

type EventBuilder struct {
	DispatcherID string
	TenantID     string
	TenantKey    string
	TraceID      string
	EventType    string
	Aggregate    string
	AggregateID  string
	Version      int64
	Payload      map[string]any
}

func (b EventBuilder) Marshal(now time.Time, eventID string) ([]byte, error) {
	return b.MarshalMQ(b.DispatcherID, now, eventID)
}