serverops

Documentation

Overview

Package serverops provides core infrastructure for server operations including data persistence, state management, error handling, and security utilities and other primitives or wiring for libraries.

Subpackages are prohibited from cross-importing. Shared utilities in other words: subpackages of serverops are NEVER allowed to use other subpackages of serverops.

Index

• Constants
• Variables
• func AssignModelToPool(ctx context.Context, _ *Config, tx libdb.Exec, model *store.Model, ...) error
• func CheckPassword(password, encodedHash, salt, signingKey string) (bool, error)
• func CheckResourceAuthorization(ctx context.Context, storeInstance store.Store, args ResourceArgs) error
• func CheckServiceAuthorization[T ServiceMeta](ctx context.Context, storeInstance store.Store, s T, ...) error
• func CreateAuthToken(subject string, permissions store.AccessList) (string, time.Time, error)
• func Decode[T any](r *http.Request) (T, error)
• func Encode[T any](w http.ResponseWriter, _ *http.Request, status int, v T) error
• func Error(w http.ResponseWriter, r *http.Request, err error, op Operation) error
• func GetIdentity(ctx context.Context) (string, error)
• func InitCredentials(ctx context.Context, config *Config, tx libdb.Exec) error
• func InitEmbedModel(ctx context.Context, config *Config, tx libdb.Exec, created bool) (*store.Model, error)
• func InitEmbedPool(ctx context.Context, config *Config, tx libdb.Exec, created bool) (*store.Pool, error)
• func InitTasksModel(ctx context.Context, config *Config, tx libdb.Exec, created bool) (*store.Model, error)
• func InitTasksPool(ctx context.Context, config *Config, tx libdb.Exec, created bool) (*store.Pool, error)
• func LoadConfig[T any](cfg *T) error
• func NewPasswordHash(password, signingKey string) (encodedHash, encodedSalt string, err error)
• func NewServiceManager(config *Config) error
• func RefreshPlainToken(ctx context.Context, token string, withGracePeriod *time.Duration) (string, bool, time.Time, error)
• func RefreshToken(ctx context.Context) (string, bool, time.Time, error)
• func ValidateConfig(cfg *Config) error
• type ActivityTracker
• type Config
• type ConfigTokenizerService
• type LLMChatClient
• type LLMEmbedClient
• type LLMPromptExecClient
• type LLMStreamClient
• type Message
• type NoopTracker
  • func (NoopTracker) Start(ctx context.Context, operation string, subject string, kvArgs ...any) (func(error), func(string, any), func())
• type Operation
• type ResourceArgs
• type ServiceManager
  • func GetManagerInstance() ServiceManager
• type ServiceMeta

Constants

const DefaultDefaultServiceGroup = "admin_panel"

const DefaultServerGroup = "server"

const EmbedPoolID = "internal_embed_pool"

const EmbedPoolName = "Embedder"

const TasksPoolID = "internal_tasks_pool"

const TasksPoolName = "Tasks"

const TenantID = "6ba7b810-9dad-11d1-80b4-00c04fd430c8"

Variables

var (
	ErrEncodeInvalidJSON      = errors.New("serverops: encoding failing, invalid json")
	ErrDecodeInvalidJSON      = errors.New("serverops: decoding failing, invalid json")
	ErrDecodeInvalidYAML      = errors.New("serverops: decoding failing, invalid yaml")
	ErrDecodeBase64           = errors.New("serverops: decoding failing, invalid base64 data")
	ErrUnsupportedContentType = errors.New("serverops: unsupported content type for decoding")
	ErrReadingRequestBody     = errors.New("serverops: failed to read request body")
	ErrMalformedContentType   = errors.New("serverops: malformed Content-Type header")
)

var CoreVersion = "CORE-UNSET-dev"

var DefaultAdminUser string = "admin@admin.com"

var ErrBadPathValue = errors.New("serverops: bad path value")

var ErrFileEmpty = errors.New("serverops: file cannot be empty")

ErrFileEmpty indicates an attempt to upload an empty file.

var ErrFileSizeLimitExceeded = errors.New("serverops: file size limit exceeded")

ErrFileSizeLimitExceeded indicates the specific file exceeded its allowed size limit.

var ErrImmutableModel = errors.New("serverops: immutable model")

var ErrImmutablePool = errors.New("serverops: immutable pool")

var ErrInvalidParameterValue = errors.New("serverops: invalid parameter value type")

var ErrMissingParameter = errors.New("serverops: missing parameter")

Functions

func AssignModelToPool

func AssignModelToPool(ctx context.Context, _ *Config, tx libdb.Exec, model *store.Model, pool *store.Pool) error

func CheckPassword

func CheckPassword(password, encodedHash, salt, signingKey string) (bool, error)

func CheckResourceAuthorization

func CheckResourceAuthorization(ctx context.Context, storeInstance store.Store, args ResourceArgs) error

CheckResourceAuthorization checks if the user has the required permission for a given resource.

func CheckServiceAuthorization

func CheckServiceAuthorization[T ServiceMeta](ctx context.Context, storeInstance store.Store, s T, permission store.Permission) error

func CreateAuthToken

func CreateAuthToken(subject string, permissions store.AccessList) (string, time.Time, error)

func Decode

func Decode[T any](r *http.Request) (T, error)

func Encode

func Encode[T any](w http.ResponseWriter, _ *http.Request, status int, v T) error

func Error

func Error(w http.ResponseWriter, r *http.Request, err error, op Operation) error

Error sends a JSON-encoded error response with an appropriate status code

func GetIdentity

func GetIdentity(ctx context.Context) (string, error)

GetIdentity extracts the identity from the context using the JWT secret from the ServiceManager.

func InitCredentials

func InitCredentials(ctx context.Context, config *Config, tx libdb.Exec) error

func InitEmbedModel

func InitEmbedModel(ctx context.Context, config *Config, tx libdb.Exec, created bool) (*store.Model, error)

func InitEmbedPool

func InitEmbedPool(ctx context.Context, config *Config, tx libdb.Exec, created bool) (*store.Pool, error)

func InitTasksModel

func InitTasksModel(ctx context.Context, config *Config, tx libdb.Exec, created bool) (*store.Model, error)

func InitTasksPool

func InitTasksPool(ctx context.Context, config *Config, tx libdb.Exec, created bool) (*store.Pool, error)

func LoadConfig

func LoadConfig[T any](cfg *T) error

func NewPasswordHash

func NewPasswordHash(password, signingKey string) (encodedHash, encodedSalt string, err error)

func NewServiceManager

func NewServiceManager(config *Config) error

NewServiceManager creates a new instance of server.

func RefreshPlainToken

func RefreshPlainToken(ctx context.Context, token string, withGracePeriod *time.Duration) (string, bool, time.Time, error)

func RefreshToken

func RefreshToken(ctx context.Context) (string, bool, time.Time, error)

func ValidateConfig

func ValidateConfig(cfg *Config) error

Types

type ActivityTracker

type ActivityTracker interface {
	// Start initiates the tracking of an operation.
	// It records the start time and context for the operation.
	//
	// Parameters:
	//   - ctx: The context for the operation, used for cancellation, deadlines,
	//          and carrying request-scoped values like trace IDs.
	//   - operation: A verb describing the action being performed (e.g., "create", "read", "process").
	//   - subject: A noun identifying the primary type of entity being acted upon (e.g., "user", "file", "order").
	//   - kvArgs: Optional key-value pairs or other metadata providing additional context
	//           at the start of the operation (e.g., relevant IDs, tags).
	//
	// Returns:
	//   - reportErr: A function to call *only* if the operation fails. Pass the error encountered.
	//   - reportChange: A function to call *only* if the operation succeeds *and* causes
	//                   a reportable state change. Pass the ID of the affected entity
	//                   and optional data about the change.
	//   - end: A function to call when the operation completes, regardless of success or failure.
	//          It signals the end of the tracked duration. Must be called exactly once.
	//          Typically called via `defer`.
	Start(
		ctx context.Context,
		operation string,
		subject string,
		kvArgs ...any,
	) (
		reportErr func(err error),
		reportChange func(id string, data any),
		end func(),
	)
}

ActivityTracker defines a standard interface for instrumenting operations within an application. It acts as a hook mechanism to observe the lifecycle of an operation (start, potential error, potential state change, end) without tightly coupling the core logic to specific monitoring implementations.

Implementations of this interface are typically used for:

• Recording metrics (latency, error rates, operation counts).
• Emitting structured logs at various lifecycle stages.
• Distributed tracing (creating and managing spans).
• Generating audit trails or activity streams, especially via `reportChange`.
• Tracking side effects or specific state changes.

The core method is `Start`, which should be invoked at the beginning of the operation being tracked. It returns three functions (`reportErr`, `reportChange`, `end`) which *must* be used correctly to signal the outcome and completion of the operation.

Correct Usage Pattern:

1. Call `Start` at the beginning of the operation.
2. Immediately `defer` the returned `end` function to ensure it's called on function exit (signaling completion and allowing duration calculation).
3. Execute the core operation logic.
4. If the operation fails, call the returned `reportErr` function with the error.
5. If the operation succeeds *and* results in a reportable state change, call the returned `reportChange` function with the relevant ID and optional data.

Example:

// tracker is an instance of ActivityTracker
reportErr, reportChange, end := tracker.Start(ctx, "update", "user", userID, requestID)
defer end() // Ensures end() is called when the surrounding function returns

updatedUser, err := service.UpdateUser(ctx, userID, userData)
if err != nil {
    reportErr(err) // Report the error
    // return or handle error...
} else {
    // Optionally report the change, e.g., if auditing is needed
    reportChange(updatedUser.ID, updatedUser) // Report success and the resulting state
}

type Config

type Config struct {
	DatabaseURL         string `json:"database_url"`
	Port                string `json:"port"`
	Addr                string `json:"addr"`
	AllowedAPIOrigins   string `json:"allowed_api_origins"`
	AllowedMethods      string `json:"allowed_methods"`
	AllowedHeaders      string `json:"allowed_headers"`
	SigningKey          string `json:"signing_key"`
	EncryptionKey       string `json:"encryption_key"`
	JWTSecret           string `json:"jwt_secret"`
	JWTExpiry           string `json:"jwt_expiry"`
	TiKVPDEndpoint      string `json:"tikv_pd_endpoint"`
	NATSURL             string `json:"nats_url"`
	NATSUser            string `json:"nats_user"`
	NATSPassword        string `json:"nats_password"`
	SecurityEnabled     string `json:"security_enabled"`
	OpensearchURL       string `json:"opensearch_url"`
	ProxyOrigin         string `json:"proxy_origin"`
	UIBaseURL           string `json:"ui_base_url"`
	TokenizerServiceURL string `json:"tokenizer_service_url"`
	EmbedModel          string `json:"embed_model"`
	TasksModel          string `json:"tasks_model"`
	VectorStoreURL      string `json:"vector_store_url"`
	WorkerUserAccountID string `json:"worker_user_account_id"`
	WorkerUserPassword  string `json:"worker_user_password"`
	WorkerUserEmail     string `json:"worker_user_email"`
}

type ConfigTokenizerService

type ConfigTokenizerService struct {
	Addr                 string `json:"addr"`
	FallbackModel        string `json:"fallback_model"`
	ModelSourceAuthToken string `json:"model_source_auth_token"`
	PreloadModels        string `json:"preload_models"`
	UseDefaultURLs       string `json:"use_default_urls"`
}

type LLMChatClient

type LLMChatClient interface {
	Chat(ctx context.Context, Messages []Message) (Message, error)
}

Client interfaces for different capabilities

type LLMEmbedClient

type LLMEmbedClient interface {
	Embed(ctx context.Context, prompt string) ([]float64, error)
}

type LLMPromptExecClient

type LLMPromptExecClient interface {
	Prompt(ctx context.Context, prompt string) (string, error)
}

type LLMStreamClient

type LLMStreamClient interface {
	Stream(ctx context.Context, prompt string) (<-chan string, error)
}

type Message

type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

type NoopTracker

type NoopTracker struct{}

NoopTracker provides a no-operation implementation of the ActivityTracker interface. It adheres to the "Null Object Pattern".

This implementation is useful when:

• Tracking needs to be disabled (e.g., in tests, specific environments, or via configuration) without requiring conditional checks (`if tracker != nil`) in the calling code.
• Providing a safe default implementation when no specific tracker is configured.

Using NoopTracker allows instrumentation calls (`Start`, `reportErr`, etc.) to remain in the code but incur minimal runtime overhead when tracking is inactive.

func (NoopTracker) Start

func (NoopTracker) Start(
	ctx context.Context,
	operation string,
	subject string,
	kvArgs ...any,
) (
	func(error),
	func(string, any),
	func(),
)

Start returns three no-op functions that do nothing when called.

type Operation

type Operation uint16

const (
	CreateOperation Operation = iota
	GetOperation
	UpdateOperation
	DeleteOperation
	ListOperation
	AuthorizeOperation
	ServerOperation
	ExecuteOperation
)

type ResourceArgs

type ResourceArgs struct {
	ResourceType       string
	Resource           string
	RequiredPermission store.Permission
}

type ServiceManager

type ServiceManager interface {
	RegisterServices(s ...ServiceMeta) error
	GetServices() ([]ServiceMeta, error)
	IsSecurityEnabled(serviceName string) bool
	HasValidLicenseFor(serviceName string) bool
	GetSecret() string
	GetTokenExpiry() time.Duration
}

func GetManagerInstance

func GetManagerInstance() ServiceManager

type ServiceMeta

type ServiceMeta interface {
	GetServiceName() string
	GetServiceGroup() string
}