go-kit

import "github.com/anatolykoptev/go-kit/env"

port := env.Int("PORT", 8080)
debug := env.Bool("DEBUG", false)

// Docker secrets / Kubernetes volumes
dbPass := env.File("DB_PASSWORD_FILE", "")

// Variable expansion
dbURL := env.Expand("DATABASE_URL", "postgres://localhost:5432/mydb")

// Binary data
cert := env.Base64("TLS_CERT", nil)
key := env.Hex("API_KEY_HEX", nil)

// Testability — decouple from os.Getenv
env.DefaultSource = env.MapSource(map[string]string{
    "PORT": "9090",
})

headers := env.Map("HEADERS", "")              // "Content-Type:json,Accept:*/*" → map
endpoint := env.URL("API_URL", "http://localhost:8080")  // parsed *url.URL
maxConns := env.Uint("MAX_CONNS", 100)

// Error-returning variants — return ParseError on invalid values
port, err := env.IntE("PORT", 8080)        // err if PORT="abc"
debug, err := env.BoolE("DEBUG", false)     // err if DEBUG="maybe"
timeout, err := env.DurationE("TIMEOUT", 30*time.Second) // accepts "5s", "100ms", "2m30s"

// Required — must be set, returns NotSetError if missing
dbURL, err := env.Required("DATABASE_URL")

// Lookup — distinguish "not set" from "set to empty"
val, ok := env.Lookup("OPTIONAL_VAR")

// Must* — panic on invalid (for fail-fast main() init)
dbURL := env.MustRequired("DATABASE_URL")
port := env.MustInt("PORT", 8080)

import "github.com/anatolykoptev/go-kit/llm"

client := llm.NewClient(baseURL, apiKey, model,
    llm.WithFallbackKeys([]string{backupKey}),
    llm.WithMaxTokens(16384),
    llm.WithTemperature(0.1),
)

// Simple text completion (unchanged)
response, err := client.Complete(ctx, systemPrompt, userPrompt)

// Full chat with tool calling
resp, err := client.Chat(ctx, messages,
    llm.WithTools([]llm.Tool{
        llm.NewTool("get_weather", "Get weather for a city", params),
    }),
)
for _, call := range resp.ToolCalls { ... }
fmt.Printf("Tokens: %d\n", resp.Usage.TotalTokens)

// Structured output — auto-generates JSON Schema from struct
// Schema constraint tags enrich the JSON Schema for better LLM output:
type User struct {
    Name string `json:"name" jsonschema:"description=Full legal name"`
    Age  int    `json:"age"  jsonschema:"minimum=0,maximum=150"`
    Role string `json:"role" jsonschema:"enum=admin|user|guest"`
}
var recipe Recipe
err := client.ChatTyped(ctx, messages, &recipe)

// SSE streaming
stream, err := client.Stream(ctx, messages)
defer stream.Close()
for chunk, ok := stream.Next(); ok; chunk, ok = stream.Next() {
    fmt.Print(chunk.Delta)
}

// Structured extraction with validation retry (Instructor-style)
type Recipe struct {
    Name        string   `json:"name"`
    Ingredients []string `json:"ingredients"`
}
var recipe Recipe
err := client.Extract(ctx, messages, &recipe,
    llm.WithValidator(func(v any) error {
        r := v.(*Recipe)
        if len(r.Ingredients) == 0 {
            return errors.New("need at least one ingredient")
        }
        return nil
    }),
)

// Union types — LLM chooses between multiple response types
type SearchAction struct {
    Query string `json:"query"`
}
type AnswerAction struct {
    Answer string `json:"answer"`
}
result, err := client.ExtractOneOf(ctx, messages, []llm.VariantDef{
    llm.Variant("search", SearchAction{}),
    llm.Variant("answer", AnswerAction{}),
})
switch v := result.(type) {
case *SearchAction:
    fmt.Println("Search:", v.Query)
case *AnswerAction:
    fmt.Println("Answer:", v.Answer)
}

// Model-level fallback chains
client = llm.NewClient("", "", "",
    llm.WithEndpoints([]llm.Endpoint{
        {URL: geminiURL, Key: key1, Model: "gemini-2.5-flash"},
        {URL: openaiURL, Key: key2, Model: "gpt-4o"},
    }),
)

// Health-aware fallback chains — skip models the proxy no longer serves.
// BuildModelChainEndpointsFiltered checks each chain entry against the live
// {baseURL}/v1/models set, so a model a provider silently removed is dropped
// instead of burning a 503 round-trip on every request. The /v1/models set is
// cached per baseURL (default 5m TTL). If /v1/models is unreachable/garbage, or
// filtering would empty the chain, the FULL unfiltered chain is returned —
// graceful degradation, never a new failure mode. Existing
// BuildModelChainEndpoints is unchanged; this is opt-in.
reg := llm.NewModelRegistry() // share one across a service; caches per baseURL
eps := llm.BuildModelChainEndpointsFiltered(ctx, reg, baseURL, apiKey,
    primary, fallbackChain,
    func(ev llm.ModelFilterEvent) {
        // observability: operator sees "N models skipped as absent from /v1/models"
        if ev.Degraded {
            llmChainDegraded.WithLabelValues(ev.Reason).Inc()
        }
        for _, dead := range ev.Dropped {
            llmModelDropped.WithLabelValues(dead).Inc()
        }
    },
)
client = llm.NewClient("", "", "", llm.WithEndpoints(eps), llm.WithMaxRetries(1))

// Request/response middleware
client = llm.NewClient(baseURL, apiKey, model,
    llm.WithMiddleware(func(ctx context.Context, req *llm.ChatRequest,
        next func(context.Context, *llm.ChatRequest) (*llm.ChatResponse, error)) (*llm.ChatResponse, error) {
        start := time.Now()
        resp, err := next(ctx, req)
        log.Printf("LLM call took %v", time.Since(start))
        return resp, err
    }),
)

import "github.com/anatolykoptev/go-kit/cache"

// L1-only (no Redis dependency at runtime)
c := cache.New(cache.Config{
    L1MaxItems: 1000,
    L1TTL:      30 * time.Minute,
})

// L1 + L2 Redis (read-through, write-through)
c := cache.New(cache.Config{
    L1MaxItems: 1000,
    L1TTL:      30 * time.Minute,
    RedisURL:   "redis://localhost:6379",
    RedisDB:    0,
    Prefix:     "myapp:",
    L2TTL:      24 * time.Hour,
})

// Custom L2 store (testing or alternative backends)
c := cache.New(cache.Config{L1MaxItems: 100, L1TTL: time.Minute})
c.WithL2(myCustomStore)

defer c.Close()

c.Set(ctx, "key", data)
data, ok := c.Get(ctx, "key")

// Cache-aside with singleflight (concurrent loads deduplicated)
data, err := c.GetOrLoad(ctx, "key", func(ctx context.Context) ([]byte, error) {
    return fetchFromDB(ctx, "key")
})

// Statistics
stats := c.Stats()
fmt.Printf("Hit ratio: %.1f%%, Evictions: %d\n", stats.HitRatio*100, stats.Evictions)

// Short TTL for fast-changing data (e.g. job listings)
c.SetWithTTL(ctx, "jobs:123", data, 15*time.Minute)

// Cache-aside with custom TTL
data, err := c.GetOrLoadWithTTL(ctx, "company:456", 24*time.Hour,
    func(ctx context.Context) ([]byte, error) {
        return fetchCompanyData(ctx, "456")
    },
)

c.SetWithTags(ctx, "user:1:profile", data, []string{"user:1", "profile"})
c.SetWithTags(ctx, "user:1:settings", data, []string{"user:1"})

n := c.InvalidateByTag(ctx, "user:1") // removes both entries, returns 2
tags := c.Tags("user:1:profile")      // []string{"user:1", "profile"}

cache.SetJSON(c, ctx, "user:1", User{Name: "Alice", Age: 30})

user, ok, err := cache.GetJSON[User](c, ctx, "user:1")

user, err := cache.GetOrLoadJSON[User](c, ctx, "user:1", func(ctx context.Context) (User, error) {
    return fetchUser(ctx, 1)
})

c := cache.New(cache.Config{
    L1MaxItems: 1000,
    L1TTL:      30 * time.Minute,
    OnEvict: func(key string, data []byte, reason cache.EvictReason) {
        switch reason {
        case cache.EvictCapacity:
            metrics.Incr("cache.evict.capacity")
        case cache.EvictExpired:
            metrics.Incr("cache.evict.expired")
        case cache.EvictExplicit:
            metrics.Incr("cache.evict.explicit")
        }
    },
})

import "github.com/anatolykoptev/go-kit/hedge"

// Start fn; if no response after 1s, launch a second call in parallel.
// First success wins, loser is cancelled automatically.
result, err := hedge.Do(ctx, time.Second, func(ctx context.Context) (string, error) {
    return callLLM(ctx)
})

// Zero/negative delay: run fn once, no goroutines.
result, err := hedge.Do(ctx, 0, fn)

import "github.com/anatolykoptev/go-kit/ratelimit"

// Single rate limiter: 10 requests/sec, burst of 5
lim := ratelimit.New(10, 5)
if lim.Allow() {
    // proceed
}

// Blocking wait (respects context cancellation)
err := lim.Wait(ctx)

// Per-key rate limiting (per-domain, per-API-key)
kl := ratelimit.NewKeyLimiter(5, 3) // 5/sec per key, burst 3
defer kl.Close()

kl.Allow("api.linkedin.com")
kl.Wait(ctx, "api.twitter.com")

// Background cleanup of idle limiters
kl.StartCleanup(time.Minute, 10*time.Minute)

// Limit to 5 concurrent operations
cl := ratelimit.NewConcurrencyLimiter(5)
release, err := cl.Acquire(ctx) // blocking; respects context
if err != nil { return err }
defer release()

// Non-blocking variant
release, ok := cl.TryAcquire()
cl.Available() // free slots
cl.Size()      // max slots

import "github.com/anatolykoptev/go-kit/retry"

result, err := retry.Do(ctx, retry.Options{
    MaxAttempts:    5,
    InitialDelay:  500 * time.Millisecond,
    MaxDelay:      10 * time.Second,
    MaxElapsedTime: 30 * time.Second,  // total budget
    Jitter:        true,               // ±25% random jitter
}, func() (string, error) {
    return callAPI()
})

// HTTP-specific: retries on 429/5xx, auto-parses Retry-After header
resp, err := retry.HTTP(ctx, retry.Options{Jitter: true}, doRequest)

// Override backoff from fn:
return "", retry.RetryAfter(5*time.Second, err)

// Abort on specific errors (never retry)
retry.Do(ctx, retry.Options{
    AbortOn: []error{context.DeadlineExceeded},
}, fn)

// Opt-in retry: only marked errors are retried
retry.Do(ctx, retry.Options{RetryableOnly: true}, func() (T, error) {
    return result, retry.MarkRetryable(err) // will retry
})

// Permanent error — stop retrying immediately
retry.Do(ctx, retry.Options{MaxAttempts: 5}, func() (T, error) {
    if isFatal(err) {
        return zero, retry.Permanent(err) // unwrapped and returned
    }
    return zero, err
})

// OnRetry callback — log each failed attempt
retry.Do(ctx, retry.Options{
    MaxAttempts: 5,
    OnRetry: func(attempt int, err error) {
        log.Printf("attempt %d failed: %v", attempt, err)
    },
}, fn)

// RetryIf — custom predicate (overrides AbortOn + RetryableOnly)
retry.Do(ctx, retry.Options{
    MaxAttempts: 5,
    RetryIf: func(err error) bool {
        var netErr net.Error
        return errors.As(err, &netErr) && netErr.Temporary()
    },
}, fn)

import "github.com/anatolykoptev/go-kit/metrics"

reg := metrics.NewRegistry()

// Counters
reg.Incr("requests")
reg.Add("bytes", 1024)

// Gauges — track current values
reg.Gauge("connections").Inc()
reg.Gauge("cpu").Set(45.2)
reg.Gauge("queue").Dec()

// Timer — one-liner duration tracking
defer reg.StartTimer("api.latency").Stop()

// Labels — dimensional metrics
reg.Incr(metrics.Label("requests", "method", "GET"))
reg.Incr(metrics.Label("requests", "method", "POST"))

// Rate tracking (EWMA)
rate := reg.Rate("events")
rate.Update(1) // record event
rate.M1()      // events/sec, 1-minute window

// Histogram (percentiles via reservoir sampling)
h := reg.Histogram("latency")
h.Update(12.5) // record observation
snap := h.Snapshot()
// snap.P50, snap.P95, snap.P99, snap.Min, snap.Max, snap.Mean

// TTL for dynamic metrics
reg.IncrWithTTL(metrics.Label("api.calls", "path", "/users"), 10*time.Minute)
reg.CleanupExpired() // remove stale metrics

// Snapshot and reset (for periodic reporting)
all := reg.SnapshotAndReset() // reads + zeros atomically

// Output formatting
reg.WriteTo(os.Stdout, metrics.TextSink{})  // key=value lines
reg.WriteTo(os.Stdout, metrics.JSONSink{})  // JSON object

import "github.com/anatolykoptev/go-kit/strutil"

s := strutil.Truncate("Hello, world!", 5)       // "Hello..."
s = strutil.TruncateAtWord("Hello, world!", 8)  // "Hello,..."
s = strutil.TruncateMiddle("path/to/file.go", 10) // "path/...e.go"

// Custom placeholder
s = strutil.TruncateWith("Hello, world!", 5, "[...]")  // "Hello[...]"

// Case conversions
s = strutil.ToSnakeCase("myVariableName")  // "my_variable_name"
s = strutil.ToCamelCase("my_variable")     // "myVariable"
s = strutil.ToKebabCase("myVariableName")  // "my-variable-name"
s = strutil.ToPascalCase("my_variable")    // "MyVariable"

// Word wrap
wrapped := strutil.WordWrap("long text here...", 80)

// Clean invalid UTF-8
clean := strutil.Scrub(untrustedInput)

// Check all substrings present
strutil.ContainsAll(s, []string{"foo", "bar"})

ok := strutil.Contains([]string{"a", "b"}, "a")    // true
ok = strutil.ContainsAny("hello world", []string{"world"}) // true

import "github.com/anatolykoptev/go-kit/fileopt"

// Dispatch by extension
opt, err := fileopt.OptimizeBytes(ctx, data,
    fileopt.KindFromExt(filepath.Ext(filename)),
    fileopt.LevelEbook, 80)

// Or call specific optimizer
opt, err := fileopt.OptimizePNG(ctx, data)
opt, err := fileopt.OptimizePDF(ctx, data, fileopt.LevelEbook)
opt, err := fileopt.OptimizeWebP(ctx, data, 80)

// Expose Prometheus metrics
mux.Handle("/metrics", fileopt.MetricsHandler())

import "github.com/anatolykoptev/go-kit/rerank"

c := rerank.New(rerank.Config{
    URL:     "http://embed-server:8082",
    Model:   "gte-multi-rerank",
    Timeout: 4 * time.Second,
    MaxDocs: 20,
}, nil)

scored := c.Rerank(ctx, query, []rerank.Doc{
    {ID: "u1", Text: "..."},
    {ID: "u2", Text: "..."},
})
// scored sorted by .Score desc; .OrigRank preserved; zero Score for docs the server didn't rank.

import "github.com/anatolykoptev/go-kit/sparse"

c, _ := sparse.NewClient("http://embed-server:8082",
    sparse.WithModel("splade-v3-distilbert"),
    sparse.WithTimeout(30*time.Second),
)
defer c.Close()

vecs, err := c.EmbedSparse(ctx, []string{"first text", "second text"})
// each vec.Indices: BERT vocab token ids (uint32);
// each vec.Values:  log(1+ReLU(logit)) weights, sorted by weight desc.