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...) }