member-console/internal/workflows/client.go

package workflows

import (
	"context"
	"fmt"
	"log/slog"
	"math/rand/v2"
	"time"

	"go.temporal.io/sdk/client"
)

// ClientConfig holds configuration for the Temporal client.
type ClientConfig struct {
	HostPort           string
	Namespace          string
	Logger             *slog.Logger
	OAuthTokenProvider *OAuthTokenProviderConfig
	// ConnectTimeout bounds how long NewClient retries a failed initial Dial
	// before giving up. Zero uses defaultConnectTimeout.
	ConnectTimeout time.Duration
}

const (
	// defaultConnectTimeout bounds the initial-Dial retry budget. It comfortably
	// exceeds Temporal's ~1-minute JWKS refresh window, so a cold-start JWT
	// rejection (server reachable, signing keys not yet loaded) self-heals
	// without the operator having to re-run `start`.
	defaultConnectTimeout = 90 * time.Second
	connectInitialBackoff = 500 * time.Millisecond
	connectMaxBackoff     = 5 * time.Second
)

// DefaultClientConfig returns a ClientConfig with sensible defaults.
func DefaultClientConfig() ClientConfig {
	return ClientConfig{
		HostPort:  "localhost:7233",
		Namespace: "default",
	}
}

// NewClient creates a new Temporal client.
func NewClient(ctx context.Context, cfg ClientConfig) (client.Client, error) {
	if cfg.HostPort == "" {
		cfg.HostPort = "localhost:7233"
	}
	if cfg.Namespace == "" {
		cfg.Namespace = "default"
	}

	opts := client.Options{
		HostPort:  cfg.HostPort,
		Namespace: cfg.Namespace,
	}

	if cfg.OAuthTokenProvider != nil {
		headersProvider, err := NewOAuthTokenProvider(*cfg.OAuthTokenProvider)
		if err != nil {
			return nil, fmt.Errorf("failed to configure Temporal auth: %w", err)
		}
		opts.HeadersProvider = headersProvider
	}

	if cfg.Logger != nil {
		opts.Logger = newSlogAdapter(cfg.Logger)
	}

	timeout := cfg.ConnectTimeout
	if timeout <= 0 {
		timeout = defaultConnectTimeout
	}

	c, err := dialWithRetry(ctx, func() (client.Client, error) { return client.Dial(opts) },
		timeout, connectInitialBackoff, connectMaxBackoff, cfg.Logger)
	if err != nil {
		return nil, fmt.Errorf("failed to create Temporal client: %w", err)
	}

	if cfg.Logger != nil {
		cfg.Logger.Info("connected to Temporal server",
			slog.String("host", cfg.HostPort),
			slog.String("namespace", cfg.Namespace))
	}

	return c, nil
}

// dialWithRetry dials Temporal, retrying transient boot-time failures with
// bounded exponential backoff + jitter. client.Dial eagerly issues an
// authenticated GetSystemInfo RPC, so a cold server — reachable but with a JWKS
// cache that has not yet loaded the identity provider's signing keys, briefly
// rejecting an otherwise-valid token — surfaces here as a Dial error. Rather
// than make that fatal (forcing the operator to re-run `start`), we retry within
// a bounded budget so the first boot after `docker compose up` succeeds on its
// own; the same loop rides out Temporal or IdP restarts in real deployments.
//
// dial is injected so the retry policy can be unit-tested without a live server.
func dialWithRetry(ctx context.Context, dial func() (client.Client, error), budget, initialBackoff, maxBackoff time.Duration, logger *slog.Logger) (client.Client, error) {
	deadline := time.Now().Add(budget)
	backoff := initialBackoff
	for attempt := 1; ; attempt++ {
		c, err := dial()
		if err == nil {
			if logger != nil && attempt > 1 {
				logger.Info("connected to Temporal after retry", slog.Int("attempts", attempt))
			}
			return c, nil
		}
		if !time.Now().Before(deadline) {
			return nil, fmt.Errorf("could not connect to Temporal after %d attempt(s) within %s: %w", attempt, budget, err)
		}
		sleep := jitter(backoff)
		if until := time.Until(deadline); sleep > until {
			sleep = until
		}
		if logger != nil {
			logger.Warn("Temporal not ready yet, retrying",
				slog.Int("attempt", attempt),
				slog.Duration("retry_in", sleep),
				slog.Any("error", err))
		}
		select {
		case <-ctx.Done():
			return nil, fmt.Errorf("Temporal connect canceled after %d attempt(s): %w", attempt, ctx.Err())
		case <-time.After(sleep):
		}
		if backoff *= 2; backoff > maxBackoff {
			backoff = maxBackoff
		}
	}
}

// jitter returns d plus a random fraction up to d/2, so simultaneous clients
// don't reconnect in lockstep.
func jitter(d time.Duration) time.Duration {
	if d <= 0 {
		return 0
	}
	half := d / 2
	if half <= 0 {
		return d
	}
	return d + rand.N(half)
}

// slogAdapter adapts slog.Logger to Temporal's log.Logger interface.
type slogAdapter struct {
	logger *slog.Logger
}

func newSlogAdapter(logger *slog.Logger) *slogAdapter {
	return &slogAdapter{logger: logger}
}

func (s *slogAdapter) Debug(msg string, keyvals ...interface{}) {
	s.logger.Debug(msg, keyvals...)
}

func (s *slogAdapter) Info(msg string, keyvals ...interface{}) {
	s.logger.Info(msg, keyvals...)
}

func (s *slogAdapter) Warn(msg string, keyvals ...interface{}) {
	s.logger.Warn(msg, keyvals...)
}

func (s *slogAdapter) Error(msg string, keyvals ...interface{}) {
	s.logger.Error(msg, keyvals...)
}