SK-3223, SK-3233 implementation of new flowvault interfaces in go sdk

common/config/config.go

@@ -7,15 +7,25 @@
 // its own, because the vault APIs differ between products.
 package config
 
+import "time"
+
+// Env is the Skyflow environment a vault runs in. The zero value means "not set": it
+// resolves to PROD when the vault URL is built, and UpdateVaultConfig leaves the current
+// Env unchanged when the update doesn't set one.
 type Env int
 
 const (
-	PROD Env = iota
+	PROD Env = iota + 1
 	STAGE
 	SANDBOX
 	DEV
 )
 
+// IsSet reports whether the Env was set explicitly.
+func (e Env) IsSet() bool {
+	return e != 0
+}
+
 type Credentials struct {
 	Path              string
 	Roles             []string
@@ -33,6 +43,32 @@ type VaultConfig struct {
 	BaseVaultURL string
 	Env          Env
 	Credentials  Credentials
+
+	// Optional HTTP settings for this vault. Zero values (and a nil MaxRetries)
+	// mean "not set": the client-level HttpConfig applies, then the SDK default.
+	Timeout           time.Duration
+	MaxRetries        *int
+	InitialRetryDelay time.Duration
+	MaxRetryDelay     time.Duration
+}
+
+// HttpConfig holds client-level call timeout and retry settings. A value set on
+// VaultConfig wins over the one here.
+type HttpConfig struct {
+	Timeout           time.Duration
+	MaxRetries        *int
+	InitialRetryDelay time.Duration
+	MaxRetryDelay     time.Duration
+}
+
+// HttpConfig returns the vault's own HTTP settings.
+func (c VaultConfig) HttpConfig() HttpConfig {
+	return HttpConfig{
+		Timeout:           c.Timeout,
+		MaxRetries:        c.MaxRetries,
+		InitialRetryDelay: c.InitialRetryDelay,
+		MaxRetryDelay:     c.MaxRetryDelay,
+	}
 }
 
 type ConnectionConfig struct {

common/config/messages.go

@@ -12,6 +12,7 @@ const (
 	errInvalidApiKey                      = " Initialization failed. Invalid credentials. Specify valid api key."
 	errEmptyRoles                         = " Initialization failed. Invalid roles. Specify at least one role."
 	errEmptyRoleInRoles                   = " Initialization failed. Invalid role. Specify a valid role."
+	errNegativeHttpConfig                 = " Initialization failed. Invalid HTTP config. Timeout, retry count and retry delays must not be negative."
 )
 
 const (
@@ -24,4 +25,5 @@ const (
 	logInvalidApiKey                      = "Invalid credentials. Api key is invalid."
 	logEmptyRoles                         = "Invalid credentials. Roles can not be empty."
 	logEmptyOrNullRoleInRoles             = "Invalid credentials. Role can not be null or empty in roles at index %v."
+	logNegativeHttpConfig                 = "Invalid HTTP config. Timeout, retry count and retry delays must not be negative."
 )

common/config/validate.go

@@ -55,6 +55,15 @@ func (v Validator) ValidateVaultConfig(vaultConfig VaultConfig) *skyflowError.Sk
 	} else if !isValidHTTPURL(baseVaultUrl) {
 		return v.invalid(logVaultUrlIsInvalid, errInvalidVaultUrl)
 	}
+	return v.ValidateHttpConfig(vaultConfig.HttpConfig())
+}
+
+// ValidateHttpConfig checks that timeout, retry count and retry delays are not negative.
+func (v Validator) ValidateHttpConfig(httpConfig HttpConfig) *skyflowError.SkyflowError {
+	if httpConfig.Timeout < 0 || httpConfig.InitialRetryDelay < 0 || httpConfig.MaxRetryDelay < 0 ||
+		(httpConfig.MaxRetries != nil && *httpConfig.MaxRetries < 0) {
+		return v.invalid(logNegativeHttpConfig, errNegativeHttpConfig)
+	}
 	return nil
 }
 

common/vault/operations.go

@@ -0,0 +1,63 @@
+// Package vault defines the vault operations shared by the Skyflow Go SDKs.
+//
+// The SDKs (v2 for SkyVault, flowvault for FlowVault) expose the same methods but with
+// their own request, options and response types, because the vault APIs differ. Each
+// interface here is generic over those types, so both SDKs' vault services satisfy the
+// same contract:
+//
+//	var _ vault.Inserter[common.InsertRequest, common.InsertOptions, common.InsertResponse] = (*vaultService)(nil)
+//
+// Code that works with either SDK can depend on these interfaces instead of a concrete
+// vault service.
+package vault
+
+import (
+	"context"
+
+	skyflowError "github.com/skyflowapi/skyflow-go/common/errors"
+)
+
+// Inserter inserts records into a vault.
+type Inserter[Req, Opts, Resp any] interface {
+	Insert(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}
+
+// Updater updates existing records in a vault.
+type Updater[Req, Opts, Resp any] interface {
+	Update(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}
+
+// Getter retrieves records from a vault.
+type Getter[Req, Opts, Resp any] interface {
+	Get(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}
+
+// Deleter deletes records from a vault.
+type Deleter[Req, Opts, Resp any] interface {
+	Delete(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}
+
+// Querier runs a query against a vault.
+type Querier[Req, Opts, Resp any] interface {
+	Query(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}
+
+// Detokenizer returns the values behind tokens.
+type Detokenizer[Req, Opts, Resp any] interface {
+	Detokenize(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}
+
+// TokenGetter returns existing tokens for values.
+type TokenGetter[Req, Opts, Resp any] interface {
+	GetTokens(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}
+
+// FilesUploader uploads files into file columns.
+type FilesUploader[Req, Opts, Resp any] interface {
+	UploadFiles(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}
+
+// FilesDeleter deletes files from file columns.
+type FilesDeleter[Req, Opts, Resp any] interface {
+	DeleteFiles(ctx context.Context, request Req, options Opts) (*Resp, *skyflowError.SkyflowError)
+}

flowvault/client/client.go

@@ -2,20 +2,24 @@ package client
 
 import (
 	"fmt"
+	"sync"
 
 	fvhelpers "github.com/skyflowapi/skyflow-go/flowvault/internal/helpers"
+	"github.com/skyflowapi/skyflow-go/flowvault/internal/vault/controller"
 	fvcommon "github.com/skyflowapi/skyflow-go/flowvault/utils/common"
 	fverrors "github.com/skyflowapi/skyflow-go/flowvault/utils/error"
 	fvlogger "github.com/skyflowapi/skyflow-go/flowvault/utils/logger"
 	logs "github.com/skyflowapi/skyflow-go/flowvault/utils/messages"
 )
 
-// Skyflow is the flowvault client entry point.
+// Skyflow is the flowvault client entry point. It is safe for concurrent use: vaults and
+// client settings can be changed while calls are running.
 type Skyflow struct {
+	mu            sync.RWMutex
 	vaultServices map[string]*VaultService
 	credentials   *fvcommon.Credentials
 	logLevel      fvlogger.LogLevel
-	customHeaders map[fvcommon.CustomHeaderKey]string
+	httpConfig    fvcommon.HttpConfig
 }
 
 type Option func(*Skyflow) *fverrors.SkyflowError
@@ -60,22 +64,55 @@ func WithCredentials(credentials fvcommon.Credentials) Option {
 // WithLogLevel sets the logging level.
 func WithLogLevel(logLevel fvlogger.LogLevel) Option {
 	return func(s *Skyflow) *fverrors.SkyflowError {
+		s.mu.Lock()
+		defer s.mu.Unlock()
 		s.logLevel = logLevel
 		fvlogger.SetLogLevel(logLevel)
 		return nil
 	}
 }
 
-// WithCustomHeaders sets custom headers sent with every request from this client.
-func WithCustomHeaders(headers map[fvcommon.CustomHeaderKey]string) Option {
+// WithHttpConfig sets client-level timeout and retry defaults. A value set on a
+// VaultConfig wins over the one set here.
+func WithHttpConfig(httpConfig fvcommon.HttpConfig) Option {
 	return func(s *Skyflow) *fverrors.SkyflowError {
-		s.customHeaders = headers
+		if err := fvhelpers.ConfigValidator().ValidateHttpConfig(httpConfig); err != nil {
+			return err
+		}
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		s.httpConfig = copyHttpConfig(httpConfig)
+		s.syncVaults()
 		return nil
 	}
 }
 
+// clientSettings returns the settings vaults fall back to. The caller holds s.mu.
+func (s *Skyflow) clientSettings() controller.ClientSettings {
+	return controller.ClientSettings{Credentials: s.credentials, HttpConfig: s.httpConfig}
+}
+
+// syncVaults pushes client-level settings to every registered vault. The caller holds s.mu.
+func (s *Skyflow) syncVaults() {
+	settings := s.clientSettings()
+	for _, svc := range s.vaultServices {
+		svc.controller.SetClientSettings(settings)
+	}
+}
+
+// copyHttpConfig copies MaxRetries so later changes to the caller's value don't leak in.
+func copyHttpConfig(config fvcommon.HttpConfig) fvcommon.HttpConfig {
+	if config.MaxRetries != nil {
+		retries := *config.MaxRetries
+		config.MaxRetries = &retries
+	}
+	return config
+}
+
 // Vault returns the vault service registered under vaultId.
 func (s *Skyflow) Vault(vaultId string) (*VaultService, *fverrors.SkyflowError) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	svc, ok := s.vaultServices[vaultId]
 	if !ok {
 		return nil, fverrors.NewSkyflowError(fverrors.INVALID_INPUT_CODE, fverrors.VAULT_ID_NOT_IN_CONFIG_LIST)
@@ -94,11 +131,17 @@ func (s *Skyflow) AddVaultConfig(cfg fvcommon.VaultConfig) *fverrors.SkyflowErro
 	if err := fvhelpers.ConfigValidator().ValidateVaultConfig(cfg); err != nil {
 		return err
 	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	if _, exists := s.vaultServices[cfg.VaultId]; exists {
 		fvlogger.Error(fmt.Sprintf(logs.VAULT_CONFIG_EXISTS, cfg.VaultId))
 		return fverrors.NewSkyflowError(fverrors.INVALID_INPUT_CODE, fmt.Sprintf(fverrors.VAULT_ID_EXISTS_IN_CONFIG_LIST, cfg.VaultId))
 	}
-	s.vaultServices[cfg.VaultId] = newVaultService(s, &cfg)
+	if cfg.MaxRetries != nil {
+		retries := *cfg.MaxRetries
+		cfg.MaxRetries = &retries
+	}
+	s.vaultServices[cfg.VaultId] = newVaultService(controller.NewVaultController(cfg, s.clientSettings()))
 	fvlogger.Info(fmt.Sprintf(logs.VAULT_CONTROLLER_INITIALIZED, cfg.VaultId))
 	return nil
 }
@@ -110,30 +153,54 @@ func (s *Skyflow) UpdateVaultConfig(updatedConfig fvcommon.VaultConfig) *fverror
 	if err := fvhelpers.ConfigValidator().ValidateUpdateVaultConfig(updatedConfig); err != nil {
 		return err
 	}
+	if err := fvhelpers.ConfigValidator().ValidateHttpConfig(updatedConfig.HttpConfig()); err != nil {
+		return err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	svc, exists := s.vaultServices[updatedConfig.VaultId]
 	if !exists {
 		fvlogger.Error(fmt.Sprintf(logs.VAULT_CONFIG_DOES_NOT_EXIST, updatedConfig.VaultId))
 		return fverrors.NewSkyflowError(fverrors.INVALID_INPUT_CODE, fverrors.VAULT_ID_NOT_IN_CONFIG_LIST)
 	}
-	// The service holds a pointer to its config, so in-place updates reach it
-	// and any controller it has built; cached tokens are dropped on change.
-	if !fvcommon.IsCredentialsEmpty(updatedConfig.Credentials) {
-		svc.config.Credentials = updatedConfig.Credentials
-		svc.resetAuth()
+	// Only the fields set on updatedConfig change. The merged config replaces the old one in
+	// a single step, so a call that is running keeps the config it started with.
+	config := svc.controller.Config()
+	credentialsChanged := !fvcommon.IsCredentialsEmpty(updatedConfig.Credentials)
+	if credentialsChanged {
+		config.Credentials = updatedConfig.Credentials
 	}
 	if updatedConfig.ClusterId != "" {
-		svc.config.ClusterId = updatedConfig.ClusterId
+		config.ClusterId = updatedConfig.ClusterId
 	}
 	if updatedConfig.BaseVaultUrl != "" {
-		svc.config.BaseVaultUrl = updatedConfig.BaseVaultUrl
+		config.BaseVaultUrl = updatedConfig.BaseVaultUrl
+	}
+	if updatedConfig.Env.IsSet() {
+		config.Env = updatedConfig.Env
+	}
+	if updatedConfig.Timeout != 0 {
+		config.Timeout = updatedConfig.Timeout
+	}
+	if updatedConfig.MaxRetries != nil {
+		retries := *updatedConfig.MaxRetries
+		config.MaxRetries = &retries
+	}
+	if updatedConfig.InitialRetryDelay != 0 {
+		config.InitialRetryDelay = updatedConfig.InitialRetryDelay
 	}
-	svc.config.Env = updatedConfig.Env
+	if updatedConfig.MaxRetryDelay != 0 {
+		config.MaxRetryDelay = updatedConfig.MaxRetryDelay
+	}
+	svc.controller.SetConfig(config, credentialsChanged)
 	return nil
 }
 
 // RemoveVaultConfig removes the vault config registered under vaultId.
 // Returns an error if the vault ID is not registered.
 func (s *Skyflow) RemoveVaultConfig(vaultId string) *fverrors.SkyflowError {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	if _, exists := s.vaultServices[vaultId]; !exists {
 		fvlogger.Error(fmt.Sprintf(logs.VAULT_CONFIG_DOES_NOT_EXIST, vaultId))
 		return fverrors.NewSkyflowError(fverrors.INVALID_INPUT_CODE, fverrors.VAULT_ID_NOT_IN_CONFIG_LIST)
@@ -142,13 +209,17 @@ func (s *Skyflow) RemoveVaultConfig(vaultId string) *fverrors.SkyflowError {
 	return nil
 }
 
-// GetVaultConfig returns the VaultConfig registered under vaultId, or an error if it is not found.
+// GetVaultConfig returns a copy of the VaultConfig registered under vaultId, or an error if
+// it is not found. Use UpdateVaultConfig to change it.
 func (s *Skyflow) GetVaultConfig(vaultId string) (*fvcommon.VaultConfig, *fverrors.SkyflowError) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	svc, exists := s.vaultServices[vaultId]
 	if !exists {
 		return nil, fverrors.NewSkyflowError(fverrors.INVALID_INPUT_CODE, fverrors.VAULT_ID_NOT_IN_CONFIG_LIST)
 	}
-	return svc.config, nil
+	config := svc.controller.Config()
+	return &config, nil
 }
 
 // ---------------------------------------------------------------------------
@@ -161,10 +232,11 @@ func (s *Skyflow) AddSkyflowCredentials(credentials fvcommon.Credentials) *fverr
 	if err := fvhelpers.ConfigValidator().ValidateCredentials(credentials); err != nil {
 		return err
 	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	// A new pointer, so every vault drops its cached token.
 	s.credentials = &credentials
-	for _, svc := range s.vaultServices {
-		svc.resetAuth()
-	}
+	s.syncVaults()
 	return nil
 }
 
@@ -175,7 +247,13 @@ func (s *Skyflow) UpdateSkyflowCredentials(credentials fvcommon.Credentials) *fv
 
 // GetSkyflowCredentials returns the client-level credentials, or nil if none are set.
 func (s *Skyflow) GetSkyflowCredentials() *fvcommon.Credentials {
-	return s.credentials
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if s.credentials == nil {
+		return nil
+	}
+	credentials := *s.credentials
+	return &credentials
 }
 
 // ---------------------------------------------------------------------------
@@ -184,13 +262,17 @@ func (s *Skyflow) GetSkyflowCredentials() *fvcommon.Credentials {
 
 // UpdateLogLevel changes the log level for the client.
 func (s *Skyflow) UpdateLogLevel(logLevel fvlogger.LogLevel) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	fvlogger.Info(fmt.Sprintf(logs.CURRENT_LOG_LEVEL, s.logLevel))
 	s.logLevel = logLevel
 	fvlogger.SetLogLevel(logLevel)
 }
 
 // GetLoglevel returns the current log level of the client.
 func (s *Skyflow) GetLoglevel() *fvlogger.LogLevel {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	logLevel := s.logLevel
 	return &logLevel
 }