client.go

package main

import (
	"bytes"
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"math"
	"net"
	"net/http"
	"net/url"
	"os"
	"strings"
	"text/template"
	"time"
)

const TempServerProblem = "Temporary server problem"

type userAgentRoundTripper struct {
	base http.RoundTripper
	ua   string
}

func (rt *userAgentRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
	if req.Header.Get("User-Agent") == "" {
		req.Header.Set("User-Agent", rt.ua)
	}

	return rt.base.RoundTrip(req)
}

// InitializeHTTPClient configures and initializes the HTTP client based on the provided configuration.
func InitializeHTTPClient(cfg *Config, observability ...*Observability) *http.Client {
	client := &http.Client{
		Timeout:   configuredHTTPClientTimeout(cfg),
		Transport: &userAgentRoundTripper{base: buildHTTPTransport(cfg), ua: fmt.Sprintf("PostfixToHTTP/%s", version)},
	}

	client.CheckRedirect = func(_ *http.Request, _ []*http.Request) error {
		return http.ErrUseLastResponse
	}

	if len(observability) > 0 && observability[0] != nil {
		observability[0].InstrumentHTTPClient(client)
	}

	return client
}

// configuredHTTPClientTimeout returns the configured timeout or the default client timeout.
func configuredHTTPClientTimeout(cfg *Config) time.Duration {
	if cfg.Server.HTTPClient.Timeout > 0 {
		return cfg.Server.HTTPClient.Timeout
	}

	return 60 * time.Second
}

// buildHTTPTransport creates the shared outbound HTTP transport.
func buildHTTPTransport(cfg *Config) *http.Transport {
	dialer := &net.Dialer{
		Timeout:   10 * time.Second,
		KeepAlive: 10 * time.Second,
	}

	transport := &http.Transport{
		DialContext:         dialer.DialContext,
		TLSHandshakeTimeout: 10 * time.Second,
		Proxy:               proxyFuncFromConfig(cfg.Server.HTTPClient.Proxy),
		MaxConnsPerHost:     cfg.Server.HTTPClient.MaxConnsPerHost,
		MaxIdleConns:        cfg.Server.HTTPClient.MaxIdleConns,
		MaxIdleConnsPerHost: cfg.Server.HTTPClient.MaxIdleConnsPerHost,
		IdleConnTimeout:     cfg.Server.HTTPClient.IdleConnTimeout,
		DisableCompression:  true,
	}

	if cfg.Server.TLS.Enabled {
		transport.TLSClientConfig = buildHTTPClientTLSConfig(cfg.Server.TLS)
	}

	return transport
}

// proxyFuncFromConfig returns the configured proxy or the environment proxy fallback.
func proxyFuncFromConfig(proxy string) func(*http.Request) (*url.URL, error) {
	if proxy == "" {
		return http.ProxyFromEnvironment
	}

	proxyURL, err := url.Parse(proxy)
	if err != nil {
		return http.ProxyFromEnvironment
	}

	return http.ProxyURL(proxyURL)
}

// buildHTTPClientTLSConfig converts the TLS config into a client-side TLS configuration.
func buildHTTPClientTLSConfig(cfg TLS) *tls.Config {
	tlsConfig := &tls.Config{
		MinVersion: tls.VersionTLS12,
	}

	if cfg.SkipVerify {
		tlsConfig.InsecureSkipVerify = true
	}

	if cfg.Cert != "" && cfg.Key != "" {
		cert, err := tls.LoadX509KeyPair(cfg.Cert, cfg.Key)
		if err != nil {
			panic(fmt.Sprintf("failed to load client certificate: %v", err))
		}

		tlsConfig.Certificates = []tls.Certificate{cert}
	}

	if cfg.RootCA != "" {
		caCert, err := os.ReadFile(cfg.RootCA)
		if err != nil {
			panic(fmt.Sprintf("failed to read root CA file: %v", err))
		}

		caCertPool := x509.NewCertPool()
		if !caCertPool.AppendCertsFromPEM(caCert) {
			panic("failed to append root CA certificate")
		}

		tlsConfig.RootCAs = caCertPool
	}

	return tlsConfig
}

// splitHeader splits a header string into a key and value pair, separating by the first colon.
// Returns empty strings if the input is improperly formatted.
func splitHeader(header string) (headerKey string, headerValue string) {
	headerParts := strings.SplitN(header, ":", 2)

	if len(headerParts) != 2 {
		return "", ""
	}

	headerKey = strings.TrimSpace(headerParts[0])
	headerValue = strings.TrimSpace(headerParts[1])

	return headerKey, headerValue
}

// getNestedValue recursively searches for a key in a nested map and returns the value and a boolean indicating success.
// responseData is the map[string]any to search within.
// key is the string to search for.
// level is the current recursion depth, used to prevent infinite recursion.
// Returns the value associated with the key and a boolean indicating whether the key was found.
func getNestedValue(responseData map[string]any, key string, level uint) (any, bool) {
	if level == math.MaxUint16 {
		return nil, false
	}

	for k1, v1 := range responseData {
		if k1 == key {
			return v1, true
		}

		if v2, ok := v1.(map[string]any); ok {
			if value, ok := getNestedValue(v2, key, level+1); ok {
				return value, true
			}
		}
	}

	return nil, false
}

// convertResponse converts a rawValue of any type to a string based on its type, respecting a maximum recursion level.
func convertResponse(rawValue any, level uint16) string {
	if level == math.MaxUint16 {
		return ""
	}

	switch value := rawValue.(type) {
	case bool:
		return fmt.Sprintf("%t", value)
	case float64:
		return fmt.Sprintf("%f", value)
	case string:
		return value
	case []string:
		return strings.Join(value, ",")
	case []any:
		var values []string

		for _, v := range value {
			values = append(values, convertResponse(v, level+1))
		}

		return strings.Join(values, ",")
	default:
		return ""
	}
}

// parsePolicyResult splits a given string value into an action and text using a single-space separator.
// Returns the first part as the action and the second part as the text if it exists; otherwise, text is empty.
func parsePolicyResult(value string) (action string, text string) {
	parts := strings.SplitN(value, " ", 2)

	action = parts[0]
	if len(parts) > 1 {
		text = parts[1]
	}

	return action, text
}

// GenericClient is an interface for managing the lifecycle of communication processes between senders and receivers.
type GenericClient interface {
	// SetReceiver sets the Receiver that will handle incoming NetString data for the implementing GenericClient.
	SetReceiver(Receiver)

	// GetSender retrieves the Sender used for handling outgoing data in the implementing GenericClient.
	GetSender() Sender

	// RenderTemplate processes the provided template string,
	// substituting placeholders with context data, and returns the result.
	RenderTemplate(string) (string, error)

	// SendAndReceive initiates the request-response process
	// by sending data with the Sender and receiving a response with the Receiver.
	SendAndReceive() error

	// SendAndReceiveContext sends data while preserving the caller's request context.
	SendAndReceiveContext(context.Context) error
}

// MapClient represents a client capable of handling data flow between a receiver and sender within a configurable setup.
// It uses a configuration object to manage operations and leverages Receiver and Sender interfaces to facilitate communication.
// Receives data via a Receiver, processes it using defined logic, and sends results through a Sender.
type MapClient struct {
	config        *Config
	logger        *slog.Logger
	httpClient    *http.Client
	oidcManager   *OIDCManager
	respCache     ResponseCache
	observability *Observability
	receiver      Receiver
	sender        Sender
}

// SetReceiver assigns the specified Receiver to the MapClient for handling incoming data operations.
func (c *MapClient) SetReceiver(receiver Receiver) {
	c.receiver = receiver
}

// GetSender retrieves the Sender instance currently associated with the MapClient.
func (c *MapClient) GetSender() Sender {
	return c.sender
}

// RenderTemplate parses and renders the provided template string using data from the receiver's key.
func (c *MapClient) RenderTemplate(tmpl string) (string, error) {
	var rendered bytes.Buffer

	parsedTemplate, err := template.New("template").Parse(tmpl)
	if err != nil {
		return "", err
	}

	templateData := struct {
		Key string
	}{
		Key: c.receiver.GetKey(),
	}

	if err = parsedTemplate.Execute(&rendered, templateData); err != nil {
		return "", err
	}

	return rendered.String(), nil
}

// buildOutboundJSONRequest creates a context-aware JSON POST request for a configured backend.
func buildOutboundJSONRequest(ctx context.Context, component, name string, settings Request, renderedPayload string) (*http.Request, error) {
	bodyReader, err := backendRequestBody(settings, renderedPayload)
	if err != nil {
		return nil, err
	}

	ctx = ContextWithBackendOperation(ctx, component, name)

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, settings.Target, bodyReader)
	if err != nil {
		return nil, err
	}

	applyBackendRequestHeaders(req, settings)

	return req, nil
}

// backendRequestBody returns the optionally compressed backend request body.
func backendRequestBody(settings Request, renderedPayload string) (io.Reader, error) {
	if !settings.HTTPRequestCompression {
		return bytes.NewBuffer([]byte(renderedPayload)), nil
	}

	compressed, err := gzipCompressor.Compress([]byte(renderedPayload))
	if err != nil {
		return nil, err
	}

	return bytes.NewBuffer(compressed), nil
}

// applyBackendRequestHeaders applies common JSON, compression, and custom headers.
func applyBackendRequestHeaders(req *http.Request, settings Request) {
	req.Header.Set("Content-Type", "application/json")

	if settings.HTTPRequestCompression {
		req.Header.Set("Content-Encoding", gzipCompressor.Name())
	}

	if settings.HTTPResponseCompression {
		req.Header.Set("Accept-Encoding", gzipCompressor.Name())
	}

	for _, header := range settings.CustomHeaders {
		headerKey, headerValue := splitHeader(header)
		if headerKey != "" && headerValue != "" {
			req.Header.Set(headerKey, headerValue)
		}
	}
}

// SendAndReceive handles the data flow by sending a POST request with rendered payload and processes the response.
// It retrieves the target and payload template from the configuration, renders the template, and sends the request.
// Custom headers can be included if specified in the settings. Timeout and other errors update the sender's status.
// Validates the response against expected status and value fields, updating the sender's status accordingly.
// Returns an error if request creation, template rendering, or response handling fails.
func (c *MapClient) SendAndReceive() error {
	return c.SendAndReceiveContext(context.Background())
}

// SendAndReceiveContext handles the data flow using the supplied request context.
func (c *MapClient) SendAndReceiveContext(ctx context.Context) error {
	c.sender = NewPostfixSender()
	ctx = ContextWithObservability(ctx, c.observability)

	settings, ok := c.config.SocketMaps[c.receiver.GetName()]
	if !ok {
		return errors.New("receiver settings not found in socket maps")
	}

	if settings.Target == "" {
		return errors.New("target URL is not specified in the settings")
	}

	renderedPayload, err := c.RenderTemplate(settings.Payload)
	if err != nil {
		return err
	}

	req, err := buildOutboundJSONRequest(ctx, componentSocketMap, c.receiver.GetName(), settings, renderedPayload)
	if err != nil {
		return err
	}

	// Add OIDC token if enabled
	_, _, err = addOIDCAuth(req, c.receiver.GetName(), settings.BackendOIDCAuth, c.oidcManager, c.logger)
	if err != nil {
		c.sender.SetStatus("TEMP")
		c.sender.SetData("authentication service unavailable")

		return nil
	}

	resp, err := c.httpClient.Do(req)
	if resp != nil && resp.Body != nil {
		defer func(Body io.ReadCloser) {
			_ = Body.Close()
		}(resp.Body)
	}

	if err != nil {
		c.handleBackendError(err)

		return nil
	}

	if resp == nil {
		c.sender.SetStatus("TEMP")
		c.sender.SetData("no response received")

		return nil
	}

	err = c.handleResponse(resp, settings)
	c.cacheSuccessfulResponse(err)

	return err
}

// handleBackendError maps socket-map backend errors to cache hits or Postfix statuses.
func (c *MapClient) handleBackendError(err error) {
	if c.respCache != nil {
		if entry, ok := c.respCache.Get(c.receiver.GetName(), c.receiver.GetKey()); ok {
			c.logBackendError("Backend request failed, serving cached response")
			c.sender.SetStatus(entry.Status)
			c.sender.SetData(entry.Data)

			return
		}
	}

	c.logBackendError("Backend request failed, no cache entry")

	if errors.Is(err, http.ErrHandlerTimeout) {
		c.sender.SetStatus("TIMEOUT")
		c.sender.SetData("request timed out")

		return
	}

	c.sender.SetStatus("TEMP")
	c.sender.SetData("service temporarily unavailable")
}

// logBackendError writes a socket-map backend error with stable log fields.
func (c *MapClient) logBackendError(message string) {
	if c.logger == nil {
		return
	}

	c.logger.Error(message,
		"component", componentSocketMap,
		"name", c.receiver.GetName(),
		"result", resultError)
}

// cacheSuccessfulResponse stores definitive socket-map successes for later backend outages.
func (c *MapClient) cacheSuccessfulResponse(err error) {
	if err != nil || c.respCache == nil {
		return
	}

	ps, ok := c.sender.(*PostfixSender)
	if !ok || ps.status != string(DovecotCmdOK) {
		return
	}

	c.respCache.Set(c.receiver.GetName(), c.receiver.GetKey(), CachedResponse{Status: ps.status, Data: ps.data})
}

var _ GenericClient = (*MapClient)(nil)

// handleResponse processes the HTTP response and updates the sender's status and data based on the response content.
func (c *MapClient) handleResponse(resp *http.Response, request Request) error {
	var responseData map[string]any

	defer func(Body io.ReadCloser) {
		_ = Body.Close()
	}(resp.Body)

	var reader io.Reader = resp.Body

	if request.HTTPResponseCompression && strings.EqualFold(resp.Header.Get("Content-Encoding"), gzipCompressor.Name()) {
		zr, err := gzipCompressor.Decompress(resp.Body)
		if err != nil {
			return fmt.Errorf("failed to init gzip reader: %w", err)
		}

		defer func(zr io.ReadCloser) {
			_ = zr.Close()
		}(zr)

		reader = zr
	}

	bodyBytes, err := io.ReadAll(reader)
	if err != nil {
		return fmt.Errorf("failed to read response body: %w", err)
	}

	if err = json.Unmarshal(bodyBytes, &responseData); err != nil {
		return fmt.Errorf("failed to unmarshal JSON: %w", err)
	}

	if request.ErrorField != "" {
		if rawValue, found := getNestedValue(responseData, request.ErrorField, 0); found {
			value := convertResponse(rawValue, 0)
			if value != "" && request.NoErrorValue != "" && value != request.NoErrorValue {
				c.sender.SetStatus("PERM")
				c.sender.SetData(fmt.Sprintf("unexpected error received: %s", value))

				return nil
			}
		}
	}

	if resp.StatusCode != request.StatusCode {
		c.sender.SetStatus("PERM")
		c.sender.SetData("unexpected status code received: " + resp.Status)

		return nil
	}

	if rawValue, found := getNestedValue(responseData, request.ValueField, 0); !found {
		c.sender.SetStatus("NOTFOUND")
		c.sender.SetData("")
	} else {
		value := convertResponse(rawValue, 0)
		if value == "" {
			c.sender.SetStatus("PERM")
			c.sender.SetData(fmt.Sprintf("unexpected value received: type=%T value=%v", rawValue, rawValue))

			return nil
		}

		maxReplySize := c.config.Server.SockmapMaxReplySize
		if maxReplySize == 0 {
			maxReplySize = defaultSockmapMaxReplySize
		}

		if len(value) > maxReplySize {
			c.sender.SetStatus("PERM")
			c.sender.SetData(fmt.Sprintf("value too long: %d", len(value)))

			return nil
		}

		c.sender.SetStatus("OK")
		c.sender.SetData(value)
	}

	return nil
}

// NewMapClient creates and returns a new instance of MapClient configured using the provided dependencies.
func NewMapClient(deps *Deps, logger *slog.Logger) GenericClient {
	return &MapClient{
		config:        deps.GetConfig(),
		logger:        logger,
		httpClient:    deps.GetHTTPClient(),
		oidcManager:   deps.GetOIDCManager(),
		respCache:     deps.GetRespCache(),
		observability: deps.GetObservability(),
	}
}

// PolicyClient is a client structure for managing communication processes between senders and receivers.
type PolicyClient struct {
	config        *Config
	logger        *slog.Logger
	httpClient    *http.Client
	oidcManager   *OIDCManager
	respCache     ResponseCache
	observability *Observability
	receiver      Receiver
	sender        Sender
}

var _ GenericClient = (*PolicyClient)(nil)

// SetReceiver assigns the given Receiver to the PolicyClient, updating the client to use the specified receiver for its operations.
func (p *PolicyClient) SetReceiver(receiver Receiver) {
	p.receiver = receiver
}

// GetSender retrieves the Sender instance associated with the PolicyClient.
func (p *PolicyClient) GetSender() Sender {
	return p.sender
}

// RenderTemplate parses and executes a template string using the receiver's key and returns the rendered output or an error.
func (p *PolicyClient) RenderTemplate(tmpl string) (string, error) {
	var rendered bytes.Buffer

	parsedTemplate, err := template.New("template").Parse(tmpl)
	if err != nil {
		return "", err
	}

	templateData := struct {
		Key string
	}{
		Key: p.receiver.GetKey(),
	}

	if err = parsedTemplate.Execute(&rendered, templateData); err != nil {
		return "", err
	}

	return rendered.String(), nil
}

// SendAndReceive sends an HTTP request based on receiver settings and handles the response, updating the sender's status and data.
func (p *PolicyClient) SendAndReceive() error {
	return p.SendAndReceiveContext(context.Background())
}

// SendAndReceiveContext sends an HTTP request while preserving the caller's request context.
func (p *PolicyClient) SendAndReceiveContext(ctx context.Context) error {
	p.sender = NewPostfixSender()
	ctx = ContextWithObservability(ctx, p.observability)

	settings, ok := p.config.PolicyServices[p.receiver.GetName()]
	if !ok {
		return errors.New("receiver settings not found in socket maps")
	}

	if settings.Target == "" {
		return errors.New("target URL is not specified in the settings")
	}

	renderedPayload, err := p.RenderTemplate(settings.Payload)
	if err != nil {
		return err
	}

	req, err := buildOutboundJSONRequest(ctx, componentPolicyService, p.receiver.GetName(), settings, renderedPayload)
	if err != nil {
		return err
	}

	// Add OIDC token if enabled
	_, _, err = addOIDCAuth(req, p.receiver.GetName(), settings.BackendOIDCAuth, p.oidcManager, p.logger)
	if err != nil {
		p.sender.SetStatus("DEFER")
		p.sender.SetData(TempServerProblem)

		return nil
	}

	resp, err := p.httpClient.Do(req)
	if resp != nil && resp.Body != nil {
		defer func(Body io.ReadCloser) {
			_ = Body.Close()
		}(resp.Body)
	}

	if err != nil {
		p.handleBackendError(err)

		return nil
	}

	if resp == nil {
		p.sender.SetStatus("DEFER")
		p.sender.SetData(TempServerProblem)

		return nil
	}

	err = p.handleResponse(resp, settings)
	p.cacheSuccessfulResponse(err)

	return err
}

// handleBackendError maps policy backend errors to cache hits or temporary Postfix failures.
func (p *PolicyClient) handleBackendError(err error) {
	if p.respCache != nil {
		if entry, ok := p.respCache.Get(p.receiver.GetName(), p.receiver.GetKey()); ok {
			p.logBackendError("Backend request failed, serving cached response")
			p.sender.SetStatus(entry.Status)
			p.sender.SetData(entry.Data)

			return
		}
	}

	p.logBackendError("Backend request failed, no cache entry")
	p.sender.SetStatus("DEFER")
	p.sender.SetData(TempServerProblem)
}

// logBackendError writes a policy backend error with stable log fields.
func (p *PolicyClient) logBackendError(message string) {
	if p.logger == nil {
		return
	}

	p.logger.Error(message,
		"component", componentPolicyService,
		"name", p.receiver.GetName(),
		"result", resultError)
}

// cacheSuccessfulResponse stores definitive policy actions for later backend outages.
func (p *PolicyClient) cacheSuccessfulResponse(err error) {
	if err != nil || p.respCache == nil {
		return
	}

	ps, ok := p.sender.(*PostfixSender)
	if !ok || ps.status == "" {
		return
	}

	p.respCache.Set(p.receiver.GetName(), p.receiver.GetKey(), CachedResponse{Status: ps.status, Data: ps.data})
}

// handleResponse processes an HTTP response, evaluates its contents, and sets the sender's status and data accordingly.
// It validates the response status code, error field, and value field based on the provided request parameters.
// Returns an error if the response body cannot be read or unmarshalled into JSON.
func (p *PolicyClient) handleResponse(resp *http.Response, request Request) error {
	var responseData map[string]any

	defer func(Body io.ReadCloser) {
		_ = Body.Close()
	}(resp.Body)

	var reader io.Reader = resp.Body

	if request.HTTPResponseCompression && strings.EqualFold(resp.Header.Get("Content-Encoding"), gzipCompressor.Name()) {
		zr, err := gzipCompressor.Decompress(resp.Body)
		if err != nil {
			return fmt.Errorf("failed to init gzip reader: %w", err)
		}

		defer func(zr io.ReadCloser) {
			_ = zr.Close()
		}(zr)

		reader = zr
	}

	bodyBytes, err := io.ReadAll(reader)
	if err != nil {
		return fmt.Errorf("failed to read response body: %w", err)
	}

	if err = json.Unmarshal(bodyBytes, &responseData); err != nil {
		return fmt.Errorf("failed to unmarshal JSON: %w", err)
	}

	if request.ErrorField != "" {
		if rawValue, found := getNestedValue(responseData, request.ErrorField, 0); found {
			value := convertResponse(rawValue, 0)
			if value != "" && request.NoErrorValue != "" && value != request.NoErrorValue {
				p.sender.SetStatus("DEFER")
				p.sender.SetData(TempServerProblem)

				return nil
			}
		}
	}

	if resp.StatusCode != request.StatusCode {
		p.sender.SetStatus("DEFER")
		p.sender.SetData(TempServerProblem)

		return nil
	}

	if rawValue, found := getNestedValue(responseData, request.ValueField, 0); !found {
		p.sender.SetStatus("DUNNO")
		p.sender.SetData("")
	} else {
		value := convertResponse(rawValue, 0)
		if value == "" {
			p.sender.SetStatus("DUNNO")
			p.sender.SetData("")

			return nil
		}

		action, text := parsePolicyResult(value)

		p.sender.SetStatus(action)
		p.sender.SetData(text)
	}

	return nil
}

// NewPolicyClient creates and returns a new instance of PolicyClient initialized with the provided dependencies.
func NewPolicyClient(deps *Deps, logger *slog.Logger) GenericClient {
	return &PolicyClient{
		config:        deps.GetConfig(),
		logger:        logger,
		httpClient:    deps.GetHTTPClient(),
		oidcManager:   deps.GetOIDCManager(),
		respCache:     deps.GetRespCache(),
		observability: deps.GetObservability(),
	}
}