options.go

/*
	Copyright NetFoundry Inc.

	Licensed under the Apache License, Version 2.0 (the "License");
	you may not use this file except in compliance with the License.
	You may obtain a copy of the License at

	https://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing, software
	distributed under the License is distributed on an "AS IS" BASIS,
	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	See the License for the specific language governing permissions and
	limitations under the License.
*/

package channel

import (
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"math"
	"time"

	"github.com/pkg/errors"
)

// MessageStrategy provides pluggable message serialization and deserialization.
type MessageStrategy interface {
	GetMarshaller() MessageMarshaller
	GetStreamProducer() StreamMessageProducer
	GetPacketProducer() PacketMessageProducer
}

// MessageMarshaller serializes a Message into bytes for transmission.
type MessageMarshaller func(m *Message) ([]byte, error)

// StreamMessageProducer reads a Message from a stream-oriented reader.
type StreamMessageProducer func(r io.Reader) (*Message, error)

// PacketMessageProducer reads a Message from a packet buffer.
type PacketMessageProducer func(b []byte) (*Message, error)

// Options configures channel behavior including queue sizes, timeouts, and message strategy.
type Options struct {
	OutQueueSize int
	ConnectOptions
	WriteTimeout    time.Duration
	MessageStrategy MessageStrategy

	// Logger, when set, is the slog.Logger this channel uses for its lifecycle
	// events. It takes precedence over the resolver installed via SetLoggerFor,
	// letting an owner give each channel (or channel type) its own logger and level
	// control - e.g. an SDK context injecting its logger, or a router naming
	// its link vs ctrl channels distinctly. When nil, the channel falls back
	// to the SetLoggerFor resolver and then to the slog.Default()-backed default.
	Logger *slog.Logger `json:"-"`
}

// DefaultOptions returns Options with sensible defaults.
func DefaultOptions() *Options {
	return &Options{
		OutQueueSize:   DefaultOutQueueSize,
		ConnectOptions: DefaultConnectOptions(),
	}
}

// DefaultConnectOptions returns ConnectOptions with sensible defaults.
func DefaultConnectOptions() ConnectOptions {
	return ConnectOptions{
		MaxQueuedConnects:      DefaultQueuedConnects,
		MaxOutstandingConnects: DefaultOutstandingConnects,
		ConnectTimeout:         DefaultConnectTimeout,
	}
}

// LoadOptions parses channel Options from a configuration map, starting from DefaultOptions.
func LoadOptions(data map[interface{}]interface{}) (*Options, error) {
	options := DefaultOptions()
	if err := options.Load(data); err != nil {
		return nil, err
	}
	return options, nil
}

// Load applies configuration values from the given map onto the Options instance.
// This allows callers to adjust defaults before loading configuration.
func (o *Options) Load(data map[interface{}]interface{}) error {
	if value, found := data["outQueueSize"]; found {
		// Accept both int and float64 (YAML decodes integers as int, JSON as
		// float64), but reject values that are invalid as a channel buffer size:
		// a negative size panics in make(chan ...), and fractional/out-of-range
		// floats must be rejected rather than silently truncated.
		var size int
		switch v := value.(type) {
		case int:
			size = v
		case float64:
			if v != math.Trunc(v) {
				return errors.Errorf("invalid outQueueSize %v: must be a whole number", v)
			}
			if v < math.MinInt || v > math.MaxInt {
				return errors.Errorf("invalid outQueueSize %v: out of range", v)
			}
			size = int(v)
		default:
			return errors.Errorf("invalid outQueueSize (%T): expected a number", value)
		}
		if size < 0 {
			return errors.Errorf("invalid outQueueSize %d: must not be negative", size)
		}
		o.OutQueueSize = size
	}

	if value, found := data["maxQueuedConnects"]; found {
		if intVal, ok := value.(int); ok {
			o.MaxQueuedConnects = intVal
		}
	}

	if value, found := data["maxOutstandingConnects"]; found {
		if intVal, ok := value.(int); ok {
			o.MaxOutstandingConnects = intVal
		}
	}

	// Prefer connectTimeout as a duration string (e.g. "5s"); fall back to the
	// legacy connectTimeoutMs integer-millisecond form when it is not present.
	if value, found := data["connectTimeout"]; found {
		if strVal, ok := value.(string); ok {
			d, err := time.ParseDuration(strVal)
			if err != nil {
				return errors.Wrapf(err, "invalid value for connectTimeout: %v", value)
			}
			o.ConnectTimeout = d
		} else {
			return errors.Errorf("invalid (non-string) value for connectTimeout: %v", value)
		}
	} else if value, found := data["connectTimeoutMs"]; found {
		if intVal, ok := value.(int); ok {
			o.ConnectTimeout = time.Duration(intVal) * time.Millisecond
		}
	}

	if value, found := data["writeTimeout"]; found {
		if strVal, ok := value.(string); ok {
			if d, err := time.ParseDuration(strVal); err == nil {
				o.WriteTimeout = d
			} else {
				return errors.Wrapf(err, "invalid value for writeTimeout: %v", value)
			}
		} else {
			return errors.Errorf("invalid (non-string) value for writeTimeout: %v", value)
		}
	}

	return nil
}

func (o Options) String() string {
	data, err := json.MarshalIndent(o, "", "  ")
	if err != nil {
		return err.Error()
	}
	return string(data)
}

// ConnectOptions controls connection acceptance limits and timeouts.
type ConnectOptions struct {
	MaxQueuedConnects      int
	MaxOutstandingConnects int
	ConnectTimeout         time.Duration
}

// Validate checks that all connect options are within acceptable bounds.
func (co *ConnectOptions) Validate() error {
	if err := co.validateConnectTimeout(); err != nil {
		return err
	}

	if err := co.validateOutstandingConnects(); err != nil {
		return err
	}

	if err := co.validateQueueConnects(); err != nil {
		return err
	}

	return nil
}

func (co *ConnectOptions) validateQueueConnects() error {
	if co.MaxQueuedConnects < MinQueuedConnects {
		return fmt.Errorf("maxQueuedConnects must be at least %d", MinQueuedConnects)
	} else if co.MaxQueuedConnects > MaxQueuedConnects {
		return fmt.Errorf("maxQueuedConnects must be at most %d", MaxQueuedConnects)
	}
	return nil
}

func (co *ConnectOptions) validateOutstandingConnects() error {
	if co.MaxOutstandingConnects < MinOutstandingConnects {
		return fmt.Errorf("maxOutstandingConnects must be at least %d", MinOutstandingConnects)
	} else if co.MaxOutstandingConnects > MaxOutstandingConnects {
		return fmt.Errorf("maxOutstandingConnects must be at most %d", MaxOutstandingConnects)
	}

	return nil
}

func (co *ConnectOptions) validateConnectTimeout() error {
	if co.ConnectTimeout < MinConnectTimeout {
		return fmt.Errorf("connectTimeoutMs must be at least %d ms", MinConnectTimeout.Milliseconds())
	} else if co.ConnectTimeout > MaxConnectTimeout {
		return fmt.Errorf("connectTimeoutMs must be at most %d ms", MaxConnectTimeout.Milliseconds())
	}

	return nil
}

// MultiChannelOptions configures multi-underlay channels with connect, write, and message options.
type MultiChannelOptions struct {
	ConnectOptions
	WriteTimeout    time.Duration
	MessageStrategy MessageStrategy
}