diff --git a/DESIGN.md b/DESIGN.md new file mode 100644 index 0000000..0d04e15 --- /dev/null +++ b/DESIGN.md @@ -0,0 +1,360 @@ +# stackyrd — Design & Architecture + +## High-level architecture + +```mermaid +flowchart TB + subgraph cmd["cmd/app/"] + main["main.go
Parse flags"] + app["Application
Run()"] + end + + subgraph config["config/"] + cfg["Config (viper)
YAML + env overrides"] + end + + subgraph infra["pkg/infrastructure/"] + compReg["ComponentRegistry
singleton"] + infraInit["InfraInitManager
async init"] + end + + subgraph deps["pkg/registry/"] + depBag["Dependencies
name → interface{}"] + end + + subgraph plugin["pkg/plugin/"] + plugInit["Init()
scan builtins → instantiate"] + plugBridge["PluginBridge
infra component"] + end + + subgraph mw["internal/middleware/"] + mwReg["MiddlewareRegistry"] + end + + subgraph svc["internal/services/modules/"] + svcReg["ServiceRegistry"] + end + + subgraph server["internal/server/"] + srv["Echo server
routes / health / metrics"] + end + + main --> app + app --> cfg + cfg --> infraInit + infraInit --> compReg + compReg --> depBag + depBag --> plugInit + plugInit --> plugBridge + plugBridge --> compReg + compReg --> mwReg + mwReg --> srv + mwReg --> svcReg + svcReg --> srv +``` + +## Boot sequence + +```mermaid +sequenceDiagram + participant main as main.go + participant cfg as ConfigManager + participant app as Application + participant srv as Server.Start() + participant infraInit as InfraInitManager + participant compReg as ComponentRegistry + participant deps as Dependencies + participant plugin as plugin.Init() + participant mw as MiddlewareRegistry + participant svc as ServiceRegistry + + main->>cfg: LoadConfig() / ValidateConfig() + main->>app: Run() + app->>srv: New(cfg, logger) + + srv->>infraInit: StartAsyncInitialization() + infraInit->>compReg: Initialize(cfg, logger) + Note over compReg: Call every registered factory
(redis, postgres, mongo, kafka, ...) + compReg-->>infraInit: registry + + srv->>deps: NewDependencies() + srv->>deps: Set(name, component) for each infra + srv->>srv: setConnectionDefaults() + + srv->>plugin: Init(cfg, logger, pluginGroup) + plugin->>compReg: scanBuiltinPlugins() + plugin->>plugin: instantiatePlugin() per meta + plugin->>compReg: SetComponent("plugins", bridge) + + srv->>mw: ApplyConfig(cfg) + srv->>mw: AutoDiscoverMiddlewares() + mw-->>srv: []echo.MiddlewareFunc + srv->>srv: e.Use(mw...) + + srv->>compReg: RouteRegistrar routes + + srv->>svc: AutoDiscoverServices(cfg, logger, deps) + svc-->>srv: []interfaces.Service + srv->>svc: Boot(echo) + svc->>srv: RegisterRoutes on /api/v1 + + srv->>srv: e.Start(":" + port) +``` + +## Extension points + +```mermaid +flowchart LR + subgraph Service["Service (internal/services/modules/)"] + S1["interface
Name / WireName / Enabled
Endpoints / RegisterRoutes / Get"] + S2["init() → RegisterService(name, factory)"] + S3["factory: cfg, logger, deps → Service"] + S4["config.yaml: services.{wire_name}: true/false"] + end + + subgraph Middleware["Middleware (internal/middleware/)"] + M1["echo.MiddlewareFunc"] + M2["init() → RegisterMiddleware(name, factory)"] + M3["factory: cfg, logger → echo.MiddlewareFunc"] + M4["config.yaml: middleware.{name}: true/false"] + end + + subgraph Infra["Infrastructure (pkg/infrastructure/)"] + I1["interface
Name / Close / GetStatus"] + I2["init() → RegisterComponent(name, factory)"] + I3["factory: cfg, logger → InfrastructureComponent"] + I4["Optional: RouteRegistrar
RouteHandlers() []RouteHandler"] + I5["config.yaml: {redis|postgres|mongo|...}.enabled"] + end + + subgraph Plugin["Plugin (pkg/plugin/builtin/{name}/)"] + P1["Plugin interface
Execute(ctx, args) → Result"] + P2["PluginMeta (plugin.yaml)
name / version / entrypoint / limits"] + P3["Runtime registry
ts: / goja: / lua: / grpc: / wasm:"] + P4["PluginBridge (infra component)"] + P5["config.yaml: plugins.enabled / overrides / allowlist"] + end + + S1 --- S2 --- S3 --- S4 + M1 --- M2 --- M3 --- M4 + I1 --- I2 --- I3 --- I4 --- I5 + P1 --- P2 --- P3 --- P4 --- P5 +``` + +## Request lifecycle + +```mermaid +flowchart TD + req["HTTP Request"] + recover["echo.Recover()"] + mws["Auto-discovered Middleware
request_id → logger → permission_check → jwt → security"] + notfound["404 / 405 handler"] + infraRoutes["Infrastructure Routes
(RouteRegistrar)"] + apiGroup["/api/v1 group"] + services["Service Routes
users / products / tasks / ..."] + handler["Handler func"] + bind["request.Bind(c, &target)
Validate via go-playground/validator"] + deps["deps.Get(name) for infra"] + response["response.Success / Created / Error"] + + req --> recover + recover --> mws + mws --> notfound + notfound --> infraRoutes + infraRoutes --> apiGroup + apiGroup --> services + services --> handler + handler --> bind + handler --> deps + handler --> response +``` + +## Plugin system architecture + +```mermaid +flowchart TB + subgraph BuiltinFS["embed.FS (builtin/)"] + p1["plugin.yaml"] + ts["ts:scripts/main.ts"] + goja["goja:scripts/main.js"] + lua["lua:main.lua"] + grpc["grpc:./bin/worker"] + wasm["wasm:main.wasm"] + end + + subgraph PluginInit["plugin.Init()"] + scan["scanBuiltinPlugins()
read plugin.yaml → meta"] + instantiate["instantiatePlugin()
match entrypoint prefix → Runtime"] + store["reg.Store(name, plugin)"] + bridge["PluginBridge
SetComponent('plugins', bridge)"] + end + + subgraph RuntimeRegistry["runtime_registry.go"] + rr["RegisterRuntime(rt)
prefix-based dispatch"] + end + + subgraph Runtimes["Runtimes"] + tsrt["TSRuntime
transpile → goja VM"] + gojart["GojaRuntime
goja VM + program cache"] + luart["LuaRuntime
gopher-lua VM"] + grpcrt["gRPCRuntime
subprocess stdin/stdout"] + wasmrt["WasmRuntime
goja/VM sandbox"] + end + + subgraph Execution["PluginBridge.Execute()"] + ctx["Context{Logger, Registry, Limits}"] + exec["p.Execute(ctx, args)"] + metrics["CollectMetrics()
active_execs / goroutines / memory"] + end + + BuiltinFS --> scan + scan --> instantiate + instantiate --> rr + rr --> tsrt + rr --> gojart + rr --> luart + rr --> grpcrt + rr --> wasmrt + instantiate --> store + store --> bridge + bridge --> Execution + ctx --> exec + exec --> metrics +``` + +## Infrastructure component lifecycle + +```mermaid +flowchart LR + subgraph Registration["init() phase"] + reg["RegisterComponent(name, factory)"] + factories["factories map[string]ComponentFactory"] + end + + subgraph Initialization["Server.Start()"] + init["ComponentRegistry.Initialize(cfg, logger)"] + create["factory(cfg, logger) → component"] + store["components[name] = component"] + end + + subgraph AsyncHealth["InfraInitManager"] + health["GetStatus() goroutine per component
with 10s timeout"] + status["status map[string]*InfraInitStatus
progress 0.0 → 1.0"] + end + + subgraph Runtime["running server"] + get["registry.Get(name)"] + getAll["registry.GetAll()
TTL-cached snapshot (2s)"] + routes["RouteRegistrar routes mounted"] + depsSet["dependencies.Set(name, component)"] + end + + subgraph Shutdown["graceful shutdown"] + close["component.Close()
10s timeout per component"] + drain["WorkerPool.Stop()
drain queue → close stopChan"] + end + + reg --> factories + factories --> init + init --> create + create --> store + store --> asyncHealth + asyncHealth --> runtime + runtime --> shutdown +``` + +## Dependency injection flow + +```mermaid +flowchart LR + subgraph InfraComponents["Infrastructure Components"] + redis["redis.RedisManager"] + pg["postgres.PostgresConnectionManager"] + mongo["mongo.MongoConnectionManager"] + kafka["kafka.KafkaManager"] + grafana["grafana.GrafanaClient"] + cron["cron.CronManager"] + plugins["plugin.PluginBridge"] + end + + subgraph DepsBag["Dependencies (map[string]interface{})"] + d1["'redis' → *RedisManager"] + d2["'postgres' → *PostgresConnectionManager"] + d3["'postgres.default' → *PostgresManager"] + d4["'mongo' → *MongoConnectionManager"] + d5["'mongo.default' → *MongoClient"] + d6["'plugins' → *PluginBridge"] + d7["..."] + end + + subgraph ServiceFactories["Service Factory (init)"] + f1["registry.RegisterService('users', factory)"] + f2["registry.RegisterService('products', factory)"] + end + + subgraph AutoDiscover["AutoDiscoverServices()"] + ad["for each factory:
svc = factory(cfg, logger, deps)
serviceDiscovered.Store(name, svc.Get())"] + end + + InfraComponents --> DepsBag + DepsBag --> ServiceFactories + ServiceFactories --> AutoDiscover +``` + +## Config loading & validation + +```mermaid +flowchart TD + flags["parseFlags()
-c / -port / -verbose / -env"] + cm["ConfigManager"] + url{{"configURL != ''?"}} + file["LoadConfigFromFile()
viper.ReadInConfig()
./config.yaml or ./config/config.yaml"] + remote["LoadConfigFromURL()
utils.LoadConfigFromURL()"] + unmarshal["viper.Unmarshal(&cfg)
Config struct"] + validate["ValidateConfig(cfg)
port availability"] + defaults["viper defaults:
app.name / server.port /
redis.enabled=false / ..."] + env["viper.AutomaticEnv()
APP_NAME / SERVER_PORT / ..."] + + flags --> cm + cm --> url + url -->|yes| remote + url -->|no| file + remote --> unmarshal + file --> defaults + defaults --> env + env --> unmarshal + unmarshal --> validate +``` + +## Health & observability + +```mermaid +flowchart TB + subgraph HealthEndpoints["Health Endpoints"] + h1["GET /health"] + h2["GET /health/infrastructure"] + h3["GET /health/dependencies"] + h4["GET /health/resources"] + end + + subgraph Metrics["Observability"] + m1["Prometheus metrics
GET /metrics"] + m2["Metrics struct
HTTPRequestsTotal / Duration / Size
DB / Cache / CircuitBreaker
Webhook / WebSocket / Batch"] + end + + subgraph Sources["Data Sources"] + s1["infraInitManager.IsReady()"] + s2["infraInitManager.GetStatus()
per-component progress + error"] + s3["dependencies.GetAll()
TTL-cached map"] + s4["registry.GetServiceFactories()"] + s5["utils.GetMemSelf() / GetRoutine()"] + end + + h1 --> s1 + h1 --> s2 + h2 --> s2 + h3 --> s3 + h3 --> s4 + h4 --> s5 + m1 --> m2 +``` diff --git a/config.yaml b/config.yaml index 9873a7c..c3b5fa1 100644 --- a/config.yaml +++ b/config.yaml @@ -122,3 +122,8 @@ plugins: max_timeout_ms: 30000 max_memory_bytes: 104857600 # allowlist: [] + +tracing: + enabled: false + sample_rate: 1.0 + otlp_endpoint: "localhost:4318" diff --git a/config/config.go b/config/config.go index d28afb3..6a8a4ef 100644 --- a/config/config.go +++ b/config/config.go @@ -33,29 +33,39 @@ func setupViperDefaults() { viper.SetDefault("swagger.base_path", "/swagger") viper.SetDefault("metrics.enabled", false) viper.SetDefault("metrics.path", "/metrics") + viper.SetDefault("metrics.sample_rate", "") viper.SetDefault("webhook.enabled", false) viper.SetDefault("webhook.timeout_seconds", 30) viper.SetDefault("webhook.max_retries", 3) viper.SetDefault("webhook.endpoint", "/api/v1/webhook") + viper.SetDefault("pagination.type", "offset") + viper.SetDefault("pagination.max_per_page", 100) + viper.SetDefault("infrastructure.init_timeout", 30) + viper.SetDefault("tracing.enabled", false) + viper.SetDefault("tracing.sample_rate", 1.0) + viper.SetDefault("tracing.otlp_endpoint", "localhost:4318") } type Config struct { - App AppConfig `mapstructure:"app"` - Server ServerConfig `mapstructure:"server"` - Services ServicesConfig `mapstructure:"services"` - Middleware MiddlewareConfig `mapstructure:"middleware"` - Auth AuthConfig `mapstructure:"auth"` - Swagger SwaggerConfig `mapstructure:"swagger"` - Redis RedisConfig `mapstructure:"redis"` - Kafka KafkaConfig `mapstructure:"kafka"` - Postgres PostgresConfig `mapstructure:"postgres"` - Mongo MongoConfig `mapstructure:"mongo"` - Webhook WebhookConfig `mapstructure:"webhook"` - Metrics MetricsConfig `mapstructure:"metrics"` - Grafana GrafanaConfig `mapstructure:"grafana"` - Cron CronConfig `mapstructure:"cron"` - MinIO MinIOConfig `mapstructure:"minio"` - Encryption EncryptionConfig `mapstructure:"encryption"` + App AppConfig `mapstructure:"app"` + Server ServerConfig `mapstructure:"server"` + Services ServicesConfig `mapstructure:"services"` + Middleware MiddlewareConfig `mapstructure:"middleware"` + Auth AuthConfig `mapstructure:"auth"` + Swagger SwaggerConfig `mapstructure:"swagger"` + Redis RedisConfig `mapstructure:"redis"` + Kafka KafkaConfig `mapstructure:"kafka"` + Postgres PostgresConfig `mapstructure:"postgres"` + Mongo MongoConfig `mapstructure:"mongo"` + Webhook WebhookConfig `mapstructure:"webhook"` + Metrics MetricsConfig `mapstructure:"metrics"` + Grafana GrafanaConfig `mapstructure:"grafana"` + Cron CronConfig `mapstructure:"cron"` + MinIO MinIOConfig `mapstructure:"minio"` + Encryption EncryptionConfig `mapstructure:"encryption"` + Pagination PaginationConfig `mapstructure:"pagination"` + Infrastructure InfrastructureConfig `mapstructure:"infrastructure"` + Tracing TracingConfig `mapstructure:"tracing"` } // MiddlewareConfig is a dynamic map of middleware names to their enabled status. @@ -190,8 +200,24 @@ type MongoConfig struct { } type MetricsConfig struct { - Enabled bool `mapstructure:"enabled"` - Path string `mapstructure:"path"` + Enabled bool `mapstructure:"enabled"` + Path string `mapstructure:"path"` + SampleRate string `mapstructure:"sample_rate"` // fractional rate (e.g. "0.1") or empty for all +} + +type PaginationConfig struct { + Type string `mapstructure:"type"` // "offset" or "cursor" + MaxPerPage int `mapstructure:"max_per_page"` // maximum per_page value (default 100) +} + +type InfrastructureConfig struct { + InitTimeout int `mapstructure:"init_timeout"` // component init timeout (seconds) +} + +type TracingConfig struct { + Enabled bool `mapstructure:"enabled"` + SampleRate float64 `mapstructure:"sample_rate"` // 0.0–1.0 + OTLPEndpoint string `mapstructure:"otlp_endpoint"` // e.g. "otel-collector:4318" } type GrafanaConfig struct { @@ -235,4 +261,4 @@ func LoadConfigWithURL(configURL string) (*Config, error) { } return &cfg, nil -} +} \ No newline at end of file diff --git a/internal/middleware/prometheus.go b/internal/middleware/prometheus.go new file mode 100644 index 0000000..144b091 --- /dev/null +++ b/internal/middleware/prometheus.go @@ -0,0 +1,51 @@ +package middleware + +import ( + "net/http" + "time" + + "stackyrd/config" + "stackyrd/pkg/logger" + "stackyrd/pkg/metrics" + + "github.com/labstack/echo/v4" +) + +func init() { + RegisterMiddleware("prometheus", func(cfg *config.Config, logger *logger.Logger) (echo.MiddlewareFunc, error) { + return Prometheus(), nil + }) +} + +// Prometheus records HTTP request metrics (count, duration, sizes) with a +// bounded label set. The route template from c.Path() (e.g. "/users/:id") +// is used instead of the raw URL path, so cardinality stays bounded no matter +// how many distinct IDs flow through. +func Prometheus() echo.MiddlewareFunc { + m := metrics.GetMetrics() + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + start := time.Now() + err := next(c) + latency := time.Since(start) + + status := c.Response().Status + if status == 0 { + status = http.StatusOK + } + + path := c.Path() + if path == "" { + path = c.Request().URL.Path + } + + var reqSize int64 + if c.Request().ContentLength > 0 { + reqSize = c.Request().ContentLength + } + + m.RecordHTTPRequest(c.Request().Method, path, status, latency, reqSize, c.Response().Size) + return err + } + } +} diff --git a/internal/middleware/versioning.go b/internal/middleware/versioning.go new file mode 100644 index 0000000..63ac283 --- /dev/null +++ b/internal/middleware/versioning.go @@ -0,0 +1,192 @@ +package middleware + +import ( + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "stackyrd/config" + "stackyrd/pkg/logger" + "stackyrd/pkg/response" + + "github.com/labstack/echo/v4" +) + +func init() { + RegisterMiddleware("versioning", func(cfg *config.Config, logger *logger.Logger) (echo.MiddlewareFunc, error) { + return VersioningMiddleware(logger), nil + }) +} + +// VersionConfig holds version handler mapping +type VersionConfig struct { + Versions map[string]VersionHandler + DefaultVersion string + SunsetWarnings map[string]SunsetInfo +} + +// VersionHandler is a function that registers routes for a version +type VersionHandler func(group *echo.Group) + +// SunsetInfo describes deprecation info for a version +type SunsetInfo struct { + SunsetDate time.Time + Deprecation time.Time + Link string +} + +var versionConfig VersionConfig + +// RegisterVersionHandler registers routes for an API version +func RegisterVersionHandler(version string, handler VersionHandler) { + if versionConfig.Versions == nil { + versionConfig.Versions = make(map[string]VersionHandler) + versionConfig.SunsetWarnings = make(map[string]SunsetInfo) + versionConfig.DefaultVersion = "1" + } + versionConfig.Versions[version] = handler +} + +// RegisterSunsetWarning registers deprecation info for a version +func RegisterSunsetWarning(version string, info SunsetInfo) { + if versionConfig.SunsetWarnings == nil { + versionConfig.SunsetWarnings = make(map[string]SunsetInfo) + } + versionConfig.SunsetWarnings[version] = info +} + +// SetDefaultVersion sets the default API version +func SetDefaultVersion(v string) { + versionConfig.DefaultVersion = v +} + +// GetVersionConfig returns the version configuration +func GetVersionConfig() VersionConfig { + return versionConfig +} + +// VersionContext contains resolved API version for a request +type VersionContext struct { + Version string + MajorVersion int + MinorVersion int + AcceptHeader string +} + +// GetVersionFromContext retrieves version info from echo context +func GetVersionFromContext(c echo.Context) *VersionContext { + if v, ok := c.Get("api_version").(*VersionContext); ok { + return v + } + return nil +} + +// VersioningMiddleware reads Accept-Version header and sets version context +func VersioningMiddleware(l *logger.Logger) echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + version := c.Request().Header.Get("Accept-Version") + if version == "" { + version = c.QueryParam("version") + } + if version == "" { + version = versionConfig.DefaultVersion + } + + vc := &VersionContext{ + Version: version, + AcceptHeader: c.Request().Header.Get("Accept"), + } + + if maj, err := strconv.Atoi(version); err == nil { + vc.MajorVersion = maj + } else { + parts := strings.SplitN(version, ".", 2) + if len(parts) > 0 { + if maj, err := strconv.Atoi(parts[0]); err == nil { + vc.MajorVersion = maj + } + } + if len(parts) > 1 { + if min, err := strconv.Atoi(parts[1]); err == nil { + vc.MinorVersion = min + } + } + } + + c.Set("api_version", vc) + c.Response().Header().Set("X-API-Version", version) + + if info, ok := versionConfig.SunsetWarnings[version]; ok { + now := time.Now() + if !info.Deprecation.IsZero() && now.After(info.Deprecation) { + c.Response().Header().Set("Warning", fmt.Sprintf("299 - \"Version %s is deprecated. Upgrade by %s\"", version, info.SunsetDate.Format("2006-01-02"))) + c.Response().Header().Set("Deprecation", info.Deprecation.Format(http.TimeFormat)) + } + if !info.SunsetDate.IsZero() && now.Before(info.SunsetDate) { + c.Response().Header().Set("Sunset", info.SunsetDate.Format(http.TimeFormat)) + if info.Link != "" { + c.Response().Header().Set("Link", fmt.Sprintf("<%s>; rel=\"sunset\"", info.Link)) + } + } + } + + l.Debug("API version resolved", + "version", version, + "major", vc.MajorVersion, + "path", c.Request().URL.Path, + ) + + return next(c) + } + } +} + +// VersionMiddleware validates the requested version is supported +func VersionMiddleware(l *logger.Logger) echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + vc := GetVersionFromContext(c) + if vc == nil { + return next(c) + } + + if _, supported := versionConfig.Versions[vc.Version]; !supported { + return response.Error(c, + http.StatusNotAcceptable, + "UNSUPPORTED_API_VERSION", + fmt.Sprintf("API version '%s' is not supported", vc.Version), + map[string]interface{}{ + "supported_versions": getVersionKeys(versionConfig.Versions), + "current_version": vc.Version, + }, + ) + } + + return next(c) + } + } +} + +// SetupVersionRoutes configures versioned route groups +func SetupVersionRoutes(e *echo.Echo, basePath string, l *logger.Logger) { + for version, handler := range versionConfig.Versions { + v := version + h := handler + path := strings.TrimSuffix(basePath, "/") + versionPath := fmt.Sprintf("%s/v%s", path, v) + group := e.Group(versionPath) + h(group) + l.Info("Registered versioned routes", "version", v, "path", versionPath) + } +} + +func getVersionKeys(m map[string]VersionHandler) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + return keys +} \ No newline at end of file diff --git a/internal/server/server.go b/internal/server/server.go index 803c913..2fc77e5 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -18,6 +18,7 @@ import ( "stackyrd/pkg/plugin" "stackyrd/pkg/registry" "stackyrd/pkg/response" + "stackyrd/pkg/tracer" "stackyrd/pkg/utils" "github.com/labstack/echo/v4" @@ -131,6 +132,9 @@ func (s *Server) Start() error { s.e.GET(s.config.Metrics.Path, echo.WrapHandler(metrics.GetMetrics().Handler())) } + // Initialize OpenTelemetry tracing + _ = s.initTracing() + if s.config.Swagger.Enabled { s.logger.Info("Registering Swagger UI documentation...") middleware.RegisterSwaggerRoutes(s.e, middleware.SwaggerConfig{ @@ -259,3 +263,35 @@ func (s *Server) Shutdown(ctx context.Context, logger *logger.Logger) error { logger.Info("Graceful shutdown completed successfully") return nil } + +// initTracing initializes OpenTelemetry tracing +func (s *Server) initTracing() error { + cfg := s.config.Tracing + if !cfg.Enabled { + s.logger.Debug("OpenTelemetry tracing disabled") + return nil + } + + tracer, err := tracer.New(tracer.Config{ + ServiceName: s.config.App.Name, + ServiceVersion: s.config.App.Version, + Environment: s.config.App.Env, + OTLPEndpoint: cfg.OTLPEndpoint, + Enabled: cfg.Enabled, + SampleRate: cfg.SampleRate, + }) + if err != nil { + s.logger.Error("Failed to initialize OpenTelemetry tracer", err) + return err + } + + tracer.Set(tracer) + s.logger.Info("OpenTelemetry tracing initialized", + "endpoint", cfg.OTLPEndpoint, + "sample_rate", cfg.SampleRate, + ) + + // Wrap existing server with trace middleware + s.e.Use(tracer.TraceMiddleware(s.logger)) + return nil +} diff --git a/pkg/batch/operations.go b/pkg/batch/operations.go deleted file mode 100644 index 207ca98..0000000 --- a/pkg/batch/operations.go +++ /dev/null @@ -1,248 +0,0 @@ -package batch - -import ( - "context" - "sync" - "time" -) - -// BatchConfig holds batch operation configuration -type BatchConfig struct { - BatchSize int - Workers int - Timeout time.Duration - RetryAttempts int - RetryDelay time.Duration -} - -// DefaultBatchConfig returns default batch configuration -func DefaultBatchConfig() BatchConfig { - return BatchConfig{ - BatchSize: 100, - Workers: 4, - Timeout: 30 * time.Second, - RetryAttempts: 3, - RetryDelay: 100 * time.Millisecond, - } -} - -// BatchResult represents the result of a batch operation -type BatchResult struct { - Successful int - Failed int - Errors []error - Duration time.Duration -} - -// BatchProcessor processes items in batches -type BatchProcessor[T any] struct { - config BatchConfig - handler func(ctx context.Context, item T) error -} - -// NewBatchProcessor creates a new batch processor -func NewBatchProcessor[T any](config BatchConfig, handler func(ctx context.Context, item T) error) *BatchProcessor[T] { - return &BatchProcessor[T]{ - config: config, - handler: handler, - } -} - -// Process processes items in batches -func (bp *BatchProcessor[T]) Process(ctx context.Context, items []T) (*BatchResult, error) { - start := time.Now() - result := &BatchResult{} - - if len(items) == 0 { - return result, nil - } - - // Create batches - batches := bp.createBatches(items) - - // Process batches with workers - resultChan := make(chan *BatchResult, len(batches)) - errorChan := make(chan error, len(batches)) - - // Create worker pool - workerCtx, cancel := context.WithTimeout(ctx, bp.config.Timeout) - defer cancel() - - var wg sync.WaitGroup - semaphore := make(chan struct{}, bp.config.Workers) - - for i, batch := range batches { - wg.Add(1) - go func(batchIndex int, batchItems []T) { - defer wg.Done() - - semaphore <- struct{}{} - defer func() { <-semaphore }() - - batchResult := bp.processBatch(workerCtx, batchItems) - resultChan <- batchResult - }(i, batch) - } - - // Wait for all batches to complete - go func() { - wg.Wait() - close(resultChan) - close(errorChan) - }() - - // Collect results - for batchResult := range resultChan { - result.Successful += batchResult.Successful - result.Failed += batchResult.Failed - result.Errors = append(result.Errors, batchResult.Errors...) - } - - result.Duration = time.Since(start) - - return result, nil -} - -// createBatches creates batches from items -func (bp *BatchProcessor[T]) createBatches(items []T) [][]T { - var batches [][]T - - for i := 0; i < len(items); i += bp.config.BatchSize { - end := i + bp.config.BatchSize - if end > len(items) { - end = len(items) - } - batches = append(batches, items[i:end]) - } - - return batches -} - -// processBatch processes a single batch -func (bp *BatchProcessor[T]) processBatch(ctx context.Context, items []T) *BatchResult { - result := &BatchResult{} - - for _, item := range items { - err := bp.processItemWithRetry(ctx, item) - if err != nil { - result.Failed++ - result.Errors = append(result.Errors, err) - } else { - result.Successful++ - } - } - - return result -} - -// processItemWithRetry processes a single item with retry -func (bp *BatchProcessor[T]) processItemWithRetry(ctx context.Context, item T) error { - var lastErr error - - for attempt := 0; attempt <= bp.config.RetryAttempts; attempt++ { - if attempt > 0 { - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(bp.config.RetryDelay): - } - } - - err := bp.handler(ctx, item) - if err == nil { - return nil - } - - lastErr = err - } - - return lastErr -} - -// BatchWriter writes items in batches -type BatchWriter[T any] struct { - config BatchConfig - writer func(ctx context.Context, items []T) error - items []T - mu sync.Mutex -} - -// NewBatchWriter creates a new batch writer -func NewBatchWriter[T any](config BatchConfig, writer func(ctx context.Context, items []T) error) *BatchWriter[T] { - return &BatchWriter[T]{ - config: config, - writer: writer, - } -} - -// Add adds an item to the batch -func (bw *BatchWriter[T]) Add(ctx context.Context, item T) error { - bw.mu.Lock() - bw.items = append(bw.items, item) - shouldFlush := len(bw.items) >= bw.config.BatchSize - bw.mu.Unlock() - - if shouldFlush { - return bw.Flush(ctx) - } - - return nil -} - -// Flush flushes the batch -func (bw *BatchWriter[T]) Flush(ctx context.Context) error { - bw.mu.Lock() - if len(bw.items) == 0 { - bw.mu.Unlock() - return nil - } - - items := make([]T, len(bw.items)) - copy(items, bw.items) - bw.items = bw.items[:0] - bw.mu.Unlock() - - return bw.writer(ctx, items) -} - -// BatchReader reads items in batches -type BatchReader[T any] struct { - config BatchConfig - reader func(ctx context.Context, offset, limit int) ([]T, error) -} - -// NewBatchReader creates a new batch reader -func NewBatchReader[T any](config BatchConfig, reader func(ctx context.Context, offset, limit int) ([]T, error)) *BatchReader[T] { - return &BatchReader[T]{ - config: config, - reader: reader, - } -} - -// ReadAll reads all items in batches -func (br *BatchReader[T]) ReadAll(ctx context.Context) ([]T, error) { - var allItems []T - offset := 0 - - for { - items, err := br.reader(ctx, offset, br.config.BatchSize) - if err != nil { - return nil, err - } - - allItems = append(allItems, items...) - - if len(items) < br.config.BatchSize { - break - } - - offset += len(items) - } - - return allItems, nil -} - -// ReadBatch reads a single batch -func (br *BatchReader[T]) ReadBatch(ctx context.Context, offset int) ([]T, error) { - return br.reader(ctx, offset, br.config.BatchSize) -} diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index 43b2e01..08d557a 100644 --- a/pkg/cache/cache.go +++ b/pkg/cache/cache.go @@ -90,14 +90,27 @@ func (c *Cache[T]) Delete(key string) { } // Cleanup removes expired items. Run this in a goroutine for periodic cleanup. +// The scan runs under a read lock so concurrent Get calls are never blocked by +// the full sweep; only the actual deletions take the write lock briefly. func (c *Cache[T]) Cleanup() { - c.mu.Lock() - defer c.mu.Unlock() - now := time.Now().UnixNano() + + var expired []string + c.mu.RLock() for k, v := range c.items { if v.Expiration > 0 && now > v.Expiration { - delete(c.items, k) + expired = append(expired, k) } } + c.mu.RUnlock() + + if len(expired) == 0 { + return + } + + c.mu.Lock() + for _, k := range expired { + delete(c.items, k) + } + c.mu.Unlock() } diff --git a/pkg/infrastructure/async.go b/pkg/infrastructure/async.go index cd6ed5d..3774eb6 100644 --- a/pkg/infrastructure/async.go +++ b/pkg/infrastructure/async.go @@ -110,13 +110,10 @@ func NewBatchAsyncResult[T any](count int, batchSize int) *BatchAsyncResult[T] { } } -// Complete is removed: CompleteResult is the sole completer for BatchAsyncResult. -// Kept for callers that still reference it but no-ops to preserve backward compat. -func (br *BatchAsyncResult[T]) Complete() {} -// CompleteResult marks one individual operation result as done and, when all -// operations in the batch have completed, closes the batch Done channel. -// This is the sole completer for BatchAsyncResult; Close() delegates here. +// CompleteResult marks one operation done and, when all operations in the +// batch have completed, closes the batch Done channel. It is the sole +// completer for BatchAsyncResult; Close() delegates here. func (br *BatchAsyncResult[T]) CompleteResult(index int) { br.Results[index].Complete(br.Results[index].Value, br.Results[index].Error) if atomic.AddInt32(&br.pending, -1) == 0 { diff --git a/pkg/infrastructure/async_init.go b/pkg/infrastructure/async_init.go index e8eed60..0c4f60d 100644 --- a/pkg/infrastructure/async_init.go +++ b/pkg/infrastructure/async_init.go @@ -27,8 +27,16 @@ type InfraInitManager struct { doneChan chan struct{} wg sync.WaitGroup ready atomic.Bool + + // cacheMu guards GetStatus TTL-snapshot. updateStatus invalidates it. + cacheMu sync.Mutex + cacheExpiry time.Time + cachedStatus map[string]*InfraInitStatus } +// statusCacheTTL controls how stale a GetStatus snapshot may be. +const statusCacheTTL = 500 * time.Millisecond + // NewInfraInitManager creates a new infrastructure initialization manager func NewInfraInitManager(logger *logger.Logger) *InfraInitManager { return &InfraInitManager{ @@ -100,31 +108,51 @@ func (im *InfraInitManager) StartAsyncInitialization(cfg *config.Config, logger // updateStatus updates the initialization status of a component func (im *InfraInitManager) updateStatus(name string, status *InfraInitStatus) { im.mu.Lock() - defer im.mu.Unlock() im.status[name] = status + im.mu.Unlock() + im.invalidateStatusCache() +} + +func (im *InfraInitManager) invalidateStatusCache() { + im.cacheMu.Lock() + im.cacheExpiry = time.Time{} + im.cacheMu.Unlock() } // updateStatusProgress updates only the progress of a component func (im *InfraInitManager) updateStatusProgress(name string, progress float64) { im.mu.Lock() - defer im.mu.Unlock() if status, exists := im.status[name]; exists { status.Progress = progress } + im.mu.Unlock() + im.invalidateStatusCache() } -// GetStatus returns the current initialization status of all components +// GetStatus returns the current initialization status of all components. +// updateStatus replaces pointers in the source map (never mutates in place), +// so a shallow snapshot copy is safe to hand back while cached. func (im *InfraInitManager) GetStatus() map[string]*InfraInitStatus { - im.mu.RLock() - defer im.mu.RUnlock() + im.cacheMu.Lock() + if time.Now().Before(im.cacheExpiry) && im.cachedStatus != nil { + result := im.cachedStatus + im.cacheMu.Unlock() + return result + } + im.cacheMu.Unlock() - // Create a copy to avoid race conditions - status := make(map[string]*InfraInitStatus) + im.mu.RLock() + result := make(map[string]*InfraInitStatus, len(im.status)) for k, v := range im.status { - status[k] = v + result[k] = v } + im.mu.RUnlock() - return status + im.cacheMu.Lock() + im.cachedStatus = result + im.cacheExpiry = time.Now().Add(statusCacheTTL) + im.cacheMu.Unlock() + return result } // IsInitialized checks if a specific component is initialized diff --git a/pkg/infrastructure/mongo.go b/pkg/infrastructure/mongo.go index 947c218..aa41d4e 100644 --- a/pkg/infrastructure/mongo.go +++ b/pkg/infrastructure/mongo.go @@ -24,7 +24,7 @@ type MongoManager struct { statusTTL time.Duration statusExpiry time.Time statusCache map[string]interface{} - statusMu sync.Mutex + statusMu sync.RWMutex } // Name returns the display name of the component @@ -207,13 +207,13 @@ func (m *MongoManager) GetStatus() map[string]interface{} { } // Fast path: return cached result when still within TTL. - m.statusMu.Lock() + m.statusMu.RLock() if time.Now().Before(m.statusExpiry) && m.statusCache != nil { cached := m.statusCache - m.statusMu.Unlock() + m.statusMu.RUnlock() return cached } - m.statusMu.Unlock() + m.statusMu.RUnlock() // Slow path: actually ping the server and collect stats. err := m.Client.Ping(context.Background(), nil) diff --git a/pkg/infrastructure/postgres.go b/pkg/infrastructure/postgres.go index c628f1b..9d307d2 100644 --- a/pkg/infrastructure/postgres.go +++ b/pkg/infrastructure/postgres.go @@ -23,7 +23,7 @@ type PostgresManager struct { statusTTL time.Duration statusExpiry time.Time statusCache map[string]interface{} - statusMu sync.Mutex + statusMu sync.RWMutex } type PostgresConnectionManager struct { @@ -181,13 +181,13 @@ func (p *PostgresManager) GetStatus() map[string]interface{} { } // Fast path: return cached result when still within TTL. - p.statusMu.Lock() + p.statusMu.RLock() if time.Now().Before(p.statusExpiry) && p.statusCache != nil { cached := p.statusCache - p.statusMu.Unlock() + p.statusMu.RUnlock() return cached } - p.statusMu.Unlock() + p.statusMu.RUnlock() // Slow path: actually ping and collect DB stats. err := p.DB.Ping() @@ -550,7 +550,6 @@ func (p *PostgresManager) ExecuteBatchAsync(ctx context.Context, queries []strin for i := range result.Results { result.Results[i].Complete(nil, fmt.Errorf("queries and args length mismatch")) } - result.Complete() return result } diff --git a/pkg/infrastructure/redis.go b/pkg/infrastructure/redis.go index 34f406e..202d3a4 100644 --- a/pkg/infrastructure/redis.go +++ b/pkg/infrastructure/redis.go @@ -19,7 +19,7 @@ type RedisManager struct { // statusCache avoids re-running Ping + PoolStats on every /health call. statusCache map[string]interface{} statusExpiry time.Time - statusMu sync.Mutex + statusMu sync.RWMutex } // Name returns the display name of the component @@ -90,13 +90,13 @@ func (r *RedisManager) GetStatus() map[string]interface{} { } // Fast path: return cached result when still within TTL. - r.statusMu.Lock() + r.statusMu.RLock() if time.Now().Before(r.statusExpiry) && r.statusCache != nil { cached := r.statusCache - r.statusMu.Unlock() + r.statusMu.RUnlock() return cached } - r.statusMu.Unlock() + r.statusMu.RUnlock() // Slow path: actually ping the server. addr := r.Client.Options().Addr diff --git a/pkg/pagination/cursor.go b/pkg/pagination/cursor.go deleted file mode 100644 index ca5d1fa..0000000 --- a/pkg/pagination/cursor.go +++ /dev/null @@ -1,227 +0,0 @@ -package pagination - -import ( - "encoding/base64" - "encoding/json" - "fmt" - "strconv" - "time" -) - -// Cursor represents a pagination cursor -type Cursor struct { - ID string `json:"id"` - Timestamp time.Time `json:"timestamp"` - Value string `json:"value,omitempty"` -} - -// CursorPagination represents cursor-based pagination parameters -type CursorPagination struct { - First int `json:"first,omitempty"` - After *Cursor `json:"after,omitempty"` - Last int `json:"last,omitempty"` - Before *Cursor `json:"before,omitempty"` -} - -// CursorPage represents a page of results with cursor information -type CursorPage struct { - Edges []Edge `json:"edges"` - PageInfo PageInfo `json:"page_info"` - TotalCount int `json:"total_count"` -} - -// Edge represents an edge in a cursor-based pagination -type Edge struct { - Node interface{} `json:"node"` - Cursor string `json:"cursor"` -} - -// PageInfo represents pagination metadata -type PageInfo struct { - HasNextPage bool `json:"has_next_page"` - HasPreviousPage bool `json:"has_previous_page"` - StartCursor *string `json:"start_cursor,omitempty"` - EndCursor *string `json:"end_cursor,omitempty"` -} - -// NewCursorPagination creates a new cursor pagination from query parameters -func NewCursorPagination(first, last int, after, before string) (*CursorPagination, error) { - pagination := &CursorPagination{ - First: first, - Last: last, - } - - if after != "" { - cursor, err := DecodeCursor(after) - if err != nil { - return nil, fmt.Errorf("invalid after cursor: %w", err) - } - pagination.After = cursor - } - - if before != "" { - cursor, err := DecodeCursor(before) - if err != nil { - return nil, fmt.Errorf("invalid before cursor: %w", err) - } - pagination.Before = cursor - } - - if pagination.First < 0 { - pagination.First = 0 - } - if pagination.Last < 0 { - pagination.Last = 0 - } - - if pagination.First == 0 && pagination.Last == 0 { - pagination.First = 10 - } - - return pagination, nil -} - -// EncodeCursor encodes a cursor to a base64 string -func EncodeCursor(cursor *Cursor) (string, error) { - data, err := json.Marshal(cursor) - if err != nil { - return "", err - } - return base64.StdEncoding.EncodeToString(data), nil -} - -// DecodeCursor decodes a base64 cursor string -func DecodeCursor(cursorStr string) (*Cursor, error) { - data, err := base64.StdEncoding.DecodeString(cursorStr) - if err != nil { - return nil, err - } - - var cursor Cursor - if err := json.Unmarshal(data, &cursor); err != nil { - return nil, err - } - - return &cursor, nil -} - -// CreateCursor creates a cursor from an item -func CreateCursor(id string, timestamp time.Time, value string) (*Cursor, error) { - return &Cursor{ - ID: id, - Timestamp: timestamp, - Value: value, - }, nil -} - -// CreateEdge creates an edge from a node and cursor -func CreateEdge(node interface{}, cursor *Cursor) (*Edge, error) { - cursorStr, err := EncodeCursor(cursor) - if err != nil { - return nil, err - } - - return &Edge{ - Node: node, - Cursor: cursorStr, - }, nil -} - -// CreatePage creates a cursor page from edges -func CreatePage(edges []Edge, hasNextPage, hasPreviousPage bool, totalCount int) *CursorPage { - page := &CursorPage{ - Edges: edges, - TotalCount: totalCount, - PageInfo: PageInfo{ - HasNextPage: hasNextPage, - HasPreviousPage: hasPreviousPage, - }, - } - - if len(edges) > 0 { - startCursor := edges[0].Cursor - endCursor := edges[len(edges)-1].Cursor - page.PageInfo.StartCursor = &startCursor - page.PageInfo.EndCursor = &endCursor - } - - return page -} - -// GetLimit returns the limit for the query -func (p *CursorPagination) GetLimit() int { - if p.First > 0 { - return p.First + 1 - } - if p.Last > 0 { - return p.Last + 1 - } - return 11 -} - -// GetOffset returns the offset for the query (not used in cursor pagination) -func (p *CursorPagination) GetOffset() int { - return 0 -} - -// HasAfterCursor returns true if there's an after cursor -func (p *CursorPagination) HasAfterCursor() bool { - return p.After != nil -} - -// HasBeforeCursor returns true if there's a before cursor -func (p *CursorPagination) HasBeforeCursor() bool { - return p.Before != nil -} - -// IsForwardPagination returns true if forward pagination is requested -func (p *CursorPagination) IsForwardPagination() bool { - return p.First > 0 || (p.First == 0 && p.Last == 0) -} - -// IsBackwardPagination returns true if backward pagination is requested -func (p *CursorPagination) IsBackwardPagination() bool { - return p.Last > 0 -} - -// GetAfterID returns the after cursor ID -func (p *CursorPagination) GetAfterID() string { - if p.After != nil { - return p.After.ID - } - return "" -} - -// GetBeforeID returns the before cursor ID -func (p *CursorPagination) GetBeforeID() string { - if p.Before != nil { - return p.Before.ID - } - return "" -} - -// GetAfterTimestamp returns the after cursor timestamp -func (p *CursorPagination) GetAfterTimestamp() time.Time { - if p.After != nil { - return p.After.Timestamp - } - return time.Time{} -} - -// GetBeforeTimestamp returns the before cursor timestamp -func (p *CursorPagination) GetBeforeTimestamp() time.Time { - if p.Before != nil { - return p.Before.Timestamp - } - return time.Time{} -} - -// StringToInt converts a string to int -func StringToInt(s string) (int, error) { - return strconv.Atoi(s) -} - -// IntToString converts an int to string -func IntToString(i int) string { - return strconv.Itoa(i) -} diff --git a/pkg/plugin/plugin.go b/pkg/plugin/plugin.go index 03f1095..11b10e4 100644 --- a/pkg/plugin/plugin.go +++ b/pkg/plugin/plugin.go @@ -62,7 +62,7 @@ type PluginStats struct { ExecuteCount int64 `json:"execute_count"` LastExecuteMs float64 `json:"last_execution_ms"` TotalExecuteMs float64 `json:"total_execution_ms"` - MemoryUsageBytes int64 `json:"memory_usage_bytes"` + MemoryUsageBytes int64 `json:"memory_usage_bytes"` } type PluginManagerMetrics struct { diff --git a/pkg/plugin/registry.go b/pkg/plugin/registry.go index 4db24b1..d45c6f7 100644 --- a/pkg/plugin/registry.go +++ b/pkg/plugin/registry.go @@ -4,10 +4,16 @@ import ( "fmt" "sync" "sync/atomic" + "time" "github.com/spf13/afero" ) +// cacheTTL bounds how stale a read-only snapshot of the registry may be. The +// Prometheus/admin endpoints copy the full maps on every call; caching them +// for a short window avoids that allocation during bursts of health polls. +const cacheTTL = 2 * time.Second + type PluginFactory func(meta PluginMeta, fs afero.Fs) (Plugin, error) type PluginRegistry struct { @@ -18,6 +24,40 @@ type PluginRegistry struct { stats map[string]*PluginStats activeExecutions atomic.Int32 mu sync.RWMutex + + // statsMu holds a per-plugin lock guarding the mutable counters inside + // PluginStats (ExecuteCount, LastExecuteMs, TotalExecuteMs). This keeps + // concurrent IncrementExecuteCount calls off the global registry lock and + // lets GetAllStats deep-copy without racing. PluginStats itself stays a + // plain value type so it can be safely copied/appended/serialised. + statsMu sync.Map // map[string]*sync.Mutex + + // cacheMu guards the read-only snapshots below. cacheExpiry is set to the + // past by any mutating call (Store/Remove/SetMeta/SetStats/Increment*) + // so the next read rebuilds the snapshot. + cacheMu sync.Mutex + cacheExpiry time.Time + cachedPlugins map[string]Plugin + cachedMetas map[string]PluginMeta + cachedStats map[string]*PluginStats +} + +// statsLock returns the per-plugin mutex for the named stats entry, creating +// it on first use. +func (r *PluginRegistry) statsLock(name string) *sync.Mutex { + if v, ok := r.statsMu.Load(name); ok { + return v.(*sync.Mutex) + } + m := &sync.Mutex{} + actual, _ := r.statsMu.LoadOrStore(name, m) + return actual.(*sync.Mutex) +} + +// invalidateCache forces the next read of any cached snapshot to rebuild. +func (r *PluginRegistry) invalidateCache() { + r.cacheMu.Lock() + r.cacheExpiry = time.Time{} + r.cacheMu.Unlock() } var ( @@ -52,12 +92,14 @@ func (r *PluginRegistry) SetMeta(name string, meta PluginMeta) { r.mu.Lock() defer r.mu.Unlock() r.metas[name] = meta + r.invalidateCache() } func (r *PluginRegistry) SetFilesystem(name string, fs afero.Fs) { r.mu.Lock() defer r.mu.Unlock() r.fsystems[name] = fs + r.invalidateCache() } func (r *PluginRegistry) Get(name string) (Plugin, bool) { @@ -82,22 +124,48 @@ func (r *PluginRegistry) GetFilesystem(name string) (afero.Fs, bool) { } func (r *PluginRegistry) GetAll() map[string]Plugin { + r.cacheMu.Lock() + if time.Now().Before(r.cacheExpiry) && r.cachedPlugins != nil { + result := r.cachedPlugins + r.cacheMu.Unlock() + return result + } + r.cacheMu.Unlock() + r.mu.RLock() - defer r.mu.RUnlock() result := make(map[string]Plugin, len(r.plugins)) for k, v := range r.plugins { result[k] = v } + r.mu.RUnlock() + + r.cacheMu.Lock() + r.cachedPlugins = result + r.cacheExpiry = time.Now().Add(cacheTTL) + r.cacheMu.Unlock() return result } func (r *PluginRegistry) GetAllMetas() map[string]PluginMeta { + r.cacheMu.Lock() + if time.Now().Before(r.cacheExpiry) && r.cachedMetas != nil { + result := r.cachedMetas + r.cacheMu.Unlock() + return result + } + r.cacheMu.Unlock() + r.mu.RLock() - defer r.mu.RUnlock() result := make(map[string]PluginMeta, len(r.metas)) for k, v := range r.metas { result[k] = v } + r.mu.RUnlock() + + r.cacheMu.Lock() + r.cachedMetas = result + r.cacheExpiry = time.Now().Add(cacheTTL) + r.cacheMu.Unlock() return result } @@ -105,6 +173,7 @@ func (r *PluginRegistry) Store(name string, p Plugin) { r.mu.Lock() defer r.mu.Unlock() r.plugins[name] = p + r.invalidateCache() } func (r *PluginRegistry) Remove(name string) { @@ -115,6 +184,7 @@ func (r *PluginRegistry) Remove(name string) { delete(r.metas, name) delete(r.fsystems, name) delete(r.stats, name) + r.invalidateCache() } func (r *PluginRegistry) HasFactory(name string) bool { @@ -140,6 +210,7 @@ func (r *PluginRegistry) SetStats(name string, s *PluginStats) { r.mu.Lock() defer r.mu.Unlock() r.stats[name] = s + r.invalidateCache() } func (r *PluginRegistry) GetStats(name string) (*PluginStats, bool) { @@ -150,23 +221,46 @@ func (r *PluginRegistry) GetStats(name string) (*PluginStats, bool) { } func (r *PluginRegistry) GetAllStats() map[string]*PluginStats { + r.cacheMu.Lock() + if time.Now().Before(r.cacheExpiry) && r.cachedStats != nil { + result := r.cachedStats + r.cacheMu.Unlock() + return result + } + r.cacheMu.Unlock() + r.mu.RLock() - defer r.mu.RUnlock() result := make(map[string]*PluginStats, len(r.stats)) for k, v := range r.stats { - result[k] = v + // Deep-copy the stats entry under its own lock so the cached snapshot + // is never mutated concurrently by IncrementExecuteCount. + r.statsLock(k).Lock() + cp := *v + r.statsLock(k).Unlock() + result[k] = &cp } + r.mu.RUnlock() + + r.cacheMu.Lock() + r.cachedStats = result + r.cacheExpiry = time.Now().Add(cacheTTL) + r.cacheMu.Unlock() return result } func (r *PluginRegistry) IncrementExecuteCount(name string, durationMs float64) { - r.mu.Lock() - defer r.mu.Unlock() - if s, ok := r.stats[name]; ok { + // Fast path: grab the stats pointer under RLock then mutate under per-entry + // lock. This avoids serialising all goroutines on the global write lock. + r.mu.RLock() + s, ok := r.stats[name] + r.mu.RUnlock() + if ok { + r.statsLock(name).Lock() s.ExecuteCount++ s.LastExecuteMs = durationMs s.TotalExecuteMs += durationMs - _ = ok + r.statsLock(name).Unlock() + r.invalidateCache() } } diff --git a/pkg/response/response.go b/pkg/response/response.go index a786441..4f9c3e2 100644 --- a/pkg/response/response.go +++ b/pkg/response/response.go @@ -4,9 +4,7 @@ import ( "net/http" "time" - "fmt" "github.com/labstack/echo/v4" - "sync/atomic" ) type Response struct { @@ -18,7 +16,7 @@ type Response struct { Meta *Meta `json:"meta,omitempty"` Timestamp int64 `json:"timestamp"` Datetime string `json:"datetime"` - CorrelationID string `json:"correlation_id"` + CorrelationID string `json:"correlation_id,omitempty"` } type ErrorDetail struct { @@ -207,24 +205,15 @@ func Error(c echo.Context, statusCode int, errorCode string, message string, det } func getCorrelationID(c echo.Context) string { - id := c.Request().Header.Get("X-Request-ID") - if id == "" { - id = c.Request().Header.Get("X-Correlation-ID") + if id := c.Request().Header.Get("X-Request-ID"); id != "" { + return id } - - if id == "" { - id = genUUID() + if id := c.Request().Header.Get("X-Correlation-ID"); id != "" { + return id } - return id -} - -var uuidCounter uint64 - -func genUUID() string { - hi := atomic.AddUint64(&uuidCounter, 1) - lo := atomic.AddUint64(&uuidCounter, 1) - return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x", - uint32(hi>>32), uint16(hi), uint16(lo>>48), uint16(lo>>32), uint32(lo)) + // No upstream ID: leave it empty and let the omitempty tag drop the field + // rather than paying 2 atomic adds + formatting for every response. + return "" } func CalculateMeta(page, perPage int, total int64, extra ...map[string]interface{}) *Meta { diff --git a/pkg/tracer/middleware.go b/pkg/tracer/middleware.go new file mode 100644 index 0000000..f054ce6 --- /dev/null +++ b/pkg/tracer/middleware.go @@ -0,0 +1,149 @@ +package tracer + +import ( + "context" + "net/http" + "strings" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/trace" + "stackyrd/pkg/logger" + + "github.com/labstack/echo/v4" +) + +// TraceMiddleware returns a middleware that creates a span for each request +func TraceMiddleware(l *logger.Logger) echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + if tracer := GetTracer(); tracer == nil || !tracer.Enabled() { + return next(c) + } + + // Extract trace context from headers (for distributed tracing) + var ctx context.Context + if traceparent := c.Request().Header.Get("traceparent"); traceparent != "" { + prop := otel.GetTextMapPropagator() + ctx = prop.Extract(context.Background(), propagation.HeaderCarrier{ + "traceparent": []string{traceparent}, + }) + } else { + ctx = context.Background() + } + + // Add request attributes to context + ctx = trace.ContextWithSpan(ctx, trace.SpanFromContext(ctx)) + + // Start server span + spanName := c.Request().Method + " " + c.Request().URL.Path + ctx, span := tracer.Start(ctx, spanName, + trace.WithSpanKind(trace.SpanKindServer), + ) + defer span.End() + + // Add attributes + span.SetAttributes( + attribute.String("http.method", c.Request().Method), + attribute.String("http.url", c.Request().URL.String()), + attribute.String("http.path", c.Request().URL.Path), + attribute.String("http.host", c.Request().Host), + attribute.String("http.scheme", c.Request().URL.Scheme), + attribute.Int("http.status_code", 0), // Will be updated after request + attribute.String("net.peer.ip", c.RealIP()), + attribute.String("user_agent", c.Request().UserAgent()), + ) + + // Store span in context for handlers to use + c.Set("trace_span", span) + c.SetRequest(c.Request().WithContext(ctx)) + + // Execute handler + err := next(c) + + // Update status code + span.SetAttributes(attribute.Int("http.status_code", c.Response().Status)) + + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) + } + + return err + } + } +} + +// TraceHandler wraps an http.Handler with tracing +func TraceHandler(handler http.Handler, spanName string) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tracer := GetTracer() + if tracer == nil || !tracer.Enabled() { + handler.ServeHTTP(w, r) + return + } + + ctx, span := tracer.Start(r.Context(), spanName, trace.WithSpanKind(trace.SpanKindServer)) + defer span.End() + + span.SetAttributes( + attribute.String("http.method", r.Method), + attribute.String("http.url", r.URL.String()), + attribute.String("http.path", r.URL.Path), + ) + + handler.ServeHTTP(w, r.WithContext(ctx)) + }) +} + +// AddTraceHeaders adds trace context to response for client-side propagation +func AddTraceHeaders(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + err := next(c) + + // Propagate traceparent to client + if span := trace.SpanFromContext(c.Request().Context()); span != nil && span.SpanContext().IsValid() { + c.Response().Header().Set("traceparent", span.SpanContext().String()) + } + + return err + } +} + +// SpanFromContext retrieves the span from context +func SpanFromContext(ctx context.Context) trace.Span { + return trace.SpanFromContext(ctx) +} + +// Attribute helper +var attribute = struct { + String func(string, string) attribute.KeyValue + Int func(string, int) attribute.KeyValue + Float64 func(string, float64) attribute.KeyValue + Bool func(string, bool) attribute.KeyValue +}{ + String: attribute.String, + Int: attribute.Int, + Float64: attribute.Float64, + Bool: attribute.Bool, +} + +// Codes for span status +var codes = struct { + Ok int + Error int +}{ + Ok: int(codes.Ok), + Error: int(codes.Error), +} + +// SpanContextFromHeader parses a traceparent header +func SpanContextFromHeader(traceparent string) (string, error) { + // Parse W3C traceparent format: version-traceid-parentid-flags + parts := strings.Split(traceparent, "-") + if len(parts) < 3 { + return "", fmt.Errorf("invalid traceparent format") + } + // Return the full traceparent for propagation + return traceparent, nil +} \ No newline at end of file diff --git a/pkg/tracer/tracer.go b/pkg/tracer/tracer.go new file mode 100644 index 0000000..113054f --- /dev/null +++ b/pkg/tracer/tracer.go @@ -0,0 +1,129 @@ +package tracer + +import ( + "context" + "fmt" + "runtime" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + semconv "go.opentelemetry.io/otel/semconv/v1.26.0" + "go.opentelemetry.io/otel/trace" +) + +// Config holds OpenTelemetry configuration +type Config struct { + ServiceName string + ServiceVersion string + Environment string + OTLPEndpoint string // e.g. "otel-collector:4318" + Enabled bool + SampleRate float64 // 0.0–1.0 +} + +// Tracer wraps OpenTelemetry TracerProvider +type Tracer struct { + provider *sdktrace.TracerProvider + tracer trace.Tracer + config Config + enabled bool +} + +// New creates a tracer with OTLP HTTP exporter +func New(cfg Config) (*Tracer, error) { + if !cfg.Enabled { + return &Tracer{enabled: false, config: cfg}, nil + } + + sr := cfg.SampleRate + if sr <= 0 || sr > 1 { + sr = 1.0 + } + ep := cfg.OTLPEndpoint + if ep == "" { + ep = "localhost:4318" + } + + res := resource.NewWithAttributes( + semconv.SchemaURL, + semconv.ServiceName(cfg.ServiceName), + semconv.ServiceVersion(cfg.ServiceVersion), + attribute.String("deployment.environment", cfg.Environment), + attribute.String("go.version", runtime.Version()), + ) + + exporter, err := otlptracehttp.New(context.Background(), + otlptracehttp.WithEndpoint(ep), + otlptracehttp.WithInsecure(), + ) + if err != nil { + return nil, fmt.Errorf("create OTLP exporter: %w", err) + } + + var sampler sdktrace.Sampler + if sr < 1.0 { + sampler = sdktrace.ParentBased(sdktrace.TraceIDRatioBased(sr)) + } else { + sampler = sdktrace.AlwaysSample() + } + + provider := sdktrace.NewTracerProvider( + sdktrace.WithBatcher(exporter, + sdktrace.WithBatchTimeout(5*time.Second), + sdktrace.WithMaxExportBatchSize(512), + ), + sdktrace.WithResource(res), + sdktrace.WithSampler(sampler), + ) + + otel.SetTracerProvider(provider) + otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( + propagation.TraceContext{}, + propagation.Baggage{}, + )) + + return &Tracer{ + provider: provider, + tracer: provider.Tracer(cfg.ServiceName), + config: cfg, + enabled: true, + }, nil +} + +// Shutdown flushes and shuts down the tracer +func (t *Tracer) Shutdown(ctx context.Context) error { + if !t.enabled || t.provider == nil { + return nil + } + c, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + return t.provider.Shutdown(c) +} + +// Start creates a new span +func (t *Tracer) Start(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span) { + if !t.enabled { + return ctx, trace.SpanFromContext(ctx) + } + return t.tracer.Start(ctx, name, opts...) +} + +// Enabled returns whether tracing is active +func (t *Tracer) Enabled() bool { return t.enabled } + +// Config returns the tracer config +func (t *Tracer) Config() Config { return t.config } + +var global *Tracer + +// Get returns the global tracer +func Get() *Tracer { return global } + +// Set sets the global tracer +func Set(tr *Tracer) { global = tr } diff --git a/pkg/tui/dashboard.go b/pkg/tui/dashboard.go index d0e548c..0d1140e 100644 --- a/pkg/tui/dashboard.go +++ b/pkg/tui/dashboard.go @@ -64,9 +64,9 @@ var ( Padding(0, 2) dashBoxStyle = lipgloss.NewStyle(). - Border(lipgloss.RoundedBorder()). - BorderForeground(lipgloss.Color("#6272A4")). - Padding(0, 1) + Border(lipgloss.RoundedBorder()). + BorderForeground(lipgloss.Color("#6272A4")). + Padding(0, 2) dashHeaderStyle = lipgloss.NewStyle(). Bold(true). diff --git a/pkg/tui/live.go b/pkg/tui/live.go index 3cad337..0b95834 100644 --- a/pkg/tui/live.go +++ b/pkg/tui/live.go @@ -479,7 +479,7 @@ func (m *LiveModel) View() string { } // Wrap entire content with minimal padding - containerStyle := lipgloss.NewStyle().Padding(1) + containerStyle := lipgloss.NewStyle().Padding(7, 7, 0, 7) return containerStyle.Render(b.String()) } @@ -676,9 +676,9 @@ func parseLogLine(line string) (level, message string) { // updateFilteredLogs filters the logs based on filterText func (m *LiveModel) updateFilteredLogs() { if m.filterText == "" { - // No filter, show all logs - m.filteredLogs = make([]LogEntry, len(m.allLogs)) - copy(m.filteredLogs, m.allLogs) + // No filter active: the full slice is already the filtered view, so + // alias it directly instead of copying every time AddLog is called. + m.filteredLogs = m.allLogs return } diff --git a/pkg/tui/simple.go b/pkg/tui/simple.go index b1b2a36..38af1cd 100644 --- a/pkg/tui/simple.go +++ b/pkg/tui/simple.go @@ -193,7 +193,7 @@ func (r *SimpleRenderer) PrintBox(title, content string) { boxStyle := lipgloss.NewStyle(). Border(lipgloss.RoundedBorder()). BorderForeground(lipgloss.Color("#8daea5")). - Padding(0, 1) + Padding(0, 2) titleStyle := lipgloss.NewStyle(). Bold(true). diff --git a/pkg/tui/startup.go b/pkg/tui/startup.go index 5ab0c6f..12e984a 100644 --- a/pkg/tui/startup.go +++ b/pkg/tui/startup.go @@ -44,9 +44,9 @@ type StartupModel struct { var ( // Title styles titleStyle = lipgloss.NewStyle(). - Bold(true). - Foreground(lipgloss.Color("#8daea5")). - MarginBottom(1) + Bold(true). + Foreground(lipgloss.Color("#8daea5")). + MarginBottom(2) subtitleStyle = lipgloss.NewStyle(). Foreground(lipgloss.Color("#8BE9FD")). @@ -54,16 +54,16 @@ var ( // Banner style with gradient effect bannerStyle = lipgloss.NewStyle(). - Foreground(lipgloss.Color("#BD93F9")). - Bold(true). - MarginBottom(1) + Foreground(lipgloss.Color("#BD93F9")). + Bold(true). + MarginBottom(2) // Box styles boxStyle = lipgloss.NewStyle(). - Border(lipgloss.RoundedBorder()). - BorderForeground(lipgloss.Color("#6272A4")). - Padding(1, 2). - MarginTop(1) + Border(lipgloss.RoundedBorder()). + BorderForeground(lipgloss.Color("#6272A4")). + Padding(1, 3). + MarginTop(1) // Service status styles pendingStyle = lipgloss.NewStyle(). @@ -255,7 +255,7 @@ func (m StartupModel) View() string { b.WriteString(footerStyle.Render(footer)) // Wrap entire content with padding - containerStyle := lipgloss.NewStyle().Padding(2) + containerStyle := lipgloss.NewStyle().Padding(3) return containerStyle.Render(b.String()) } @@ -264,7 +264,7 @@ func (m StartupModel) renderServices() string { header := labelStyle.Render("◆ Services Initialization") lines = append(lines, header) - lines = append(lines, strings.Repeat("─", 40)) + lines = append(lines, strings.Repeat("─", 50)) for i, s := range m.services { var icon, status string @@ -301,7 +301,7 @@ func (m StartupModel) renderServices() string { line := fmt.Sprintf(" %s %s %s %s", icon, - lipgloss.NewStyle().Width(20).Render(name), + lipgloss.NewStyle().Width(25).Render(name), iconArrow, style.Render(status), ) diff --git a/pkg/tui/styles.go b/pkg/tui/styles.go index bd1e070..3ecc497 100644 --- a/pkg/tui/styles.go +++ b/pkg/tui/styles.go @@ -35,31 +35,31 @@ func TextEffect(text string, colors []string) string { var ( // Success box SuccessBoxStyle = lipgloss.NewStyle(). - Border(lipgloss.RoundedBorder()). - BorderForeground(lipgloss.Color("#50FA7B")). - Foreground(lipgloss.Color("#50FA7B")). - Padding(0, 1) + Border(lipgloss.RoundedBorder()). + BorderForeground(lipgloss.Color("#50FA7B")). + Foreground(lipgloss.Color("#50FA7B")). + Padding(0, 2) // Warning box WarningBoxStyle = lipgloss.NewStyle(). - Border(lipgloss.RoundedBorder()). - BorderForeground(lipgloss.Color("#F1FA8C")). - Foreground(lipgloss.Color("#F1FA8C")). - Padding(0, 1) + Border(lipgloss.RoundedBorder()). + BorderForeground(lipgloss.Color("#F1FA8C")). + Foreground(lipgloss.Color("#F1FA8C")). + Padding(0, 2) // Error box ErrorBoxStyle = lipgloss.NewStyle(). - Border(lipgloss.RoundedBorder()). - BorderForeground(lipgloss.Color("#FF5555")). - Foreground(lipgloss.Color("#FF5555")). - Padding(0, 1) + Border(lipgloss.RoundedBorder()). + BorderForeground(lipgloss.Color("#FF5555")). + Foreground(lipgloss.Color("#FF5555")). + Padding(0, 2) // Info box InfoBoxStyle = lipgloss.NewStyle(). - Border(lipgloss.RoundedBorder()). - BorderForeground(lipgloss.Color("#8BE9FD")). - Foreground(lipgloss.Color("#8BE9FD")). - Padding(0, 1) + Border(lipgloss.RoundedBorder()). + BorderForeground(lipgloss.Color("#8BE9FD")). + Foreground(lipgloss.Color("#8BE9FD")). + Padding(0, 2) // Primary box with double border PrimaryBoxStyle = lipgloss.NewStyle(). diff --git a/pkg/utils/system.go b/pkg/utils/system.go index a265e0d..317de23 100644 --- a/pkg/utils/system.go +++ b/pkg/utils/system.go @@ -214,16 +214,23 @@ func GetNetworkInfo() (map[string]string, error) { }, nil } +// ResetTerminal restores terminal to main screen buffer and resets attributes +// Useful after bubbletea alt-screen exits abnormally +func ResetTerminal() { + os.Stdout.WriteString("\033[?1049l\033[0m\033[H") +} + // ClearScreen clears the terminal screen (cross-platform) +// Also resets terminal state: restores main screen buffer, cursor, and text attributes. func ClearScreen() { - var cmd *exec.Cmd + ResetTerminal() + os.Stdout.WriteString("\033[2J\033[3J") + var cmd *exec.Cmd switch runtime.GOOS { case "windows": - // Windows: use cmd /c cls cmd = exec.Command("cmd", "/c", "cls") default: - // Linux, macOS, and others: use clear command cmd = exec.Command("clear") } diff --git a/scripts/build/build.go b/scripts/build/build.go index 8eb808b..708a4f6 100644 --- a/scripts/build/build.go +++ b/scripts/build/build.go @@ -135,16 +135,16 @@ func (ctx *BuildContext) checkPath(logger *Logger) error { return ctx.ensureProjectRoot(logger) } -// clear console screen +// clear console screen and reset terminal state func ClearScreen() { + fmt.Print("\033[?1049l\033[0m\033[H\033[2J\033[3J") + var cmd *exec.Cmd switch runtime.GOOS { case "windows": - // Windows: use cmd /c cls cmd = exec.Command("cmd", "/c", "cls") default: - // Linux, macOS, and others: use clear command cmd = exec.Command("clear") } @@ -1060,7 +1060,7 @@ func isTerminal() bool { // runTUIBuild runs the build with the bubbletea TUI func runTUIBuild(ctx *BuildContext, logger *Logger) { defer func() { - fmt.Print("\033[?25h\033[0m") // ensure cursor visible and attributes reset + fmt.Print("\033[?1049l\033[?25h\033[0m") // restore main buffer, show cursor, reset attrs }() _, err := RunBuildTUI(ctx, logger) diff --git a/scripts/docker/docker_build.go b/scripts/docker/docker_build.go index 226db30..245d560 100644 --- a/scripts/docker/docker_build.go +++ b/scripts/docker/docker_build.go @@ -500,8 +500,7 @@ func askVerbose(logger *DockerLogger) bool { // main function func main() { - // Clear the terminal screen - fmt.Print("\033[H\033[2J") + fmt.Print("\033[?1049l\033[0m\033[H\033[2J\033[3J") // Initialize a temporary logger for interactive prompts tempLogger := NewDockerLogger(false) diff --git a/scripts/pkg/pkg.go b/scripts/pkg/pkg.go index 68dc861..c233fac 100644 --- a/scripts/pkg/pkg.go +++ b/scripts/pkg/pkg.go @@ -255,7 +255,7 @@ func parseIndexLines(body string, logger *Logger) []*PackageInfo { } func fetchIndex(logger *Logger) ([]*PackageInfo, error) { - logger.Info("Fetching index from %s", INDEX_URL) + logger.Info("Fetching package index...") resp, err := http.Get(INDEX_URL) if err != nil { return nil, fmt.Errorf("failed to download index: %w", err) @@ -368,28 +368,45 @@ func promptUserByName(packages []*PackageInfo, logger *Logger) (*PackageInfo, er logger.Info("Selected package: %s", matches[0].Name) return matches[0], nil } - fmt.Printf("\n%sMatching packages:%s\n", B_PURPLE, RESET) - for _, pkg := range matches { - fmt.Printf(" %s%s%s\n", B_WHITE, pkg.Name, RESET) - } + const pageSize = 30 + totalPages := (len(matches) + pageSize - 1) / pageSize + currentPage := 1 for { - fmt.Printf("\n%sEnter the exact package name from the list (or 'search' to search again, 'cancel' to exit):%s ", B_YELLOW, RESET) + fmt.Printf("\n%sMatching packages (page %d/%d):%s\n", B_PURPLE, currentPage, totalPages, RESET) + start := (currentPage - 1) * pageSize + end := start + pageSize + if end > len(matches) { + end = len(matches) + } + for i, pkg := range matches[start:end] { + fmt.Printf(" %s%d.%s %s%s%s\n", P_CYAN, start+i+1, RESET, B_WHITE, pkg.Name, RESET) + } + nextHint := "" + if currentPage < totalPages { + nextHint = ", enter 'n' for next page" + } + fmt.Printf("\n%sSelect package by number (or 0 to search again%s, 'cancel' to exit):%s ", B_YELLOW, nextHint, RESET) choice, _ := reader.ReadString('\n') choice = strings.TrimSpace(choice) if strings.EqualFold(choice, "cancel") { fmt.Println("Installation cancelled.") os.Exit(0) } - if strings.EqualFold(choice, "search") { - break + if strings.EqualFold(choice, "n") && currentPage < totalPages { + currentPage++ + continue } - for _, pkg := range matches { - if pkg.Name == choice { - logger.Info("Selected package: %s", pkg.Name) - return pkg, nil + idx := 0 + if _, err := fmt.Sscanf(choice, "%d", &idx); err == nil { + if idx == 0 { + break + } + if idx >= 1 && idx <= len(matches) { + logger.Info("Selected package: %s", matches[idx-1].Name) + return matches[idx-1], nil } } - fmt.Printf("%sInvalid package name. Please choose exactly from the list above.%s\n", P_RED, RESET) + fmt.Printf("%sInvalid choice. Enter a number between 1 and %d (or 0 to search again%s):%s ", P_RED, len(matches), nextHint, RESET) } } } @@ -1073,7 +1090,7 @@ func cmdUpdate(logger *Logger) { if err != nil { manifest = &Manifest{Meta: ManifestMeta{IndexURL: INDEX_URL}, Packages: make(map[string]InstalledPackage)} } - logger.Info("Fetching index from %s", INDEX_URL) + logger.Info("Fetching package index...") resp, err := http.Get(INDEX_URL) if err != nil { logger.Error("Failed to fetch index: %v", err) diff --git a/scripts/plugin/pkg.go b/scripts/plugin/pkg.go index 8055227..4a10066 100644 --- a/scripts/plugin/pkg.go +++ b/scripts/plugin/pkg.go @@ -569,6 +569,8 @@ Run '%scommand -h%s' for subcommand-specific flags. } func main() { + fmt.Print("\033[?1049l\033[0m\033[H\033[2J\033[3J") + flag.Usage = func() { printUsage() }