storage

package
v0.1.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 28, 2025 License: MIT Imports: 22 Imported by: 0

README

Storage Extension - Production-Ready Implementation

A robust, production-ready object storage extension for Forge with comprehensive resilience, security, and observability features.

Features

🔒 Security
  • Path Validation: Comprehensive path traversal protection
  • Input Sanitization: Automatic key sanitization
  • Secure Presigned URLs: Cryptographically signed URLs with expiration
  • Content Type Validation: Strict content type checking
  • Metadata Validation: Size and format constraints
🛡️ Resilience
  • Circuit Breaker Pattern: Prevent cascade failures with automatic recovery
  • Exponential Backoff Retry: Configurable retry logic with backoff
  • Rate Limiting: Token bucket algorithm for rate limiting
  • Timeout Management: Configurable operation timeouts
  • Non-Retryable Error Detection: Smart error classification
⚡ Performance
  • Buffer Pooling: sync.Pool for zero-allocation I/O
  • Concurrent Operations: Thread-safe file operations with fine-grained locking
  • ETag Caching: Cached MD5 checksums
  • Chunked Uploads: Large file support with multipart uploads
  • Connection Pooling: Optimized for cloud backends
📊 Observability
  • Comprehensive Metrics: Upload/download counts, durations, error rates
  • Health Checks: Multi-backend health monitoring
  • Structured Logging: Contextual logging with trace IDs
  • Circuit Breaker State Tracking: Real-time resilience monitoring
🗄️ Backend Support
  • Enhanced Local: Production-ready local filesystem with atomic operations
  • AWS S3: Full S3 support with presigned URLs
  • GCS: Google Cloud Storage (coming soon)
  • Azure Blob: Azure Blob Storage (coming soon)

Installation

Basic Installation
cd v2
go get -u github.com/xraph/forge/extensions/storage
With AWS S3 Support
go get -u github.com/aws/aws-sdk-go-v2/aws
go get -u github.com/aws/aws-sdk-go-v2/config
go get -u github.com/aws/aws-sdk-go-v2/service/s3
go get -u github.com/aws/aws-sdk-go-v2/feature/s3/manager

Configuration

YAML Configuration
extensions:
  storage:
    default: "local"
    use_enhanced_backend: true
    
    # Backends configuration
    backends:
      local:
        type: "local"
        config:
          root_dir: "./storage"
          base_url: "http://localhost:8080/files"
          secret: "your-secret-key-here"
          chunk_size: 5242880  # 5MB
          max_upload_size: 5368709120  # 5GB
      
      s3:
        type: "s3"
        config:
          region: "us-east-1"
          bucket: "my-bucket"
          prefix: "uploads"
          access_key_id: "${AWS_ACCESS_KEY_ID}"
          secret_access_key: "${AWS_SECRET_ACCESS_KEY}"
    
    # Resilience configuration
    resilience:
      max_retries: 3
      initial_backoff: "100ms"
      max_backoff: "10s"
      backoff_multiplier: 2.0
      
      circuit_breaker_enabled: true
      circuit_breaker_threshold: 5
      circuit_breaker_timeout: "60s"
      circuit_breaker_half_open_max: 3
      
      rate_limit_enabled: true
      rate_limit_per_sec: 100
      rate_limit_burst: 200
      
      operation_timeout: "30s"
    
    # Features
    enable_presigned_urls: true
    presign_expiry: "15m"
    max_upload_size: 5368709120  # 5GB
    chunk_size: 5242880          # 5MB
    
    # CDN configuration
    enable_cdn: false
    cdn_base_url: ""
Programmatic Configuration
package main

import (
    "github.com/xraph/forge"
    "github.com/xraph/forge/extensions/storage"
)

func main() {
    // Create app
    app := forge.New()

    // Configure storage
    config := storage.DefaultConfig()
    config.Default = "s3"
    config.UseEnhancedBackend = true
    config.Backends["s3"] = storage.BackendConfig{
        Type: "s3",
        Config: map[string]interface{}{
            "region": "us-east-1",
            "bucket": "my-bucket",
        },
    }
    
    // Customize resilience
    config.Resilience.MaxRetries = 5
    config.Resilience.CircuitBreakerThreshold = 10

    // Register extension
    app.Use(storage.NewExtension(config))

    app.Run()
}

Usage

Basic Operations
// Get storage manager from DI
storageManager := forge.Must[*storage.StorageManager](app.Container(), "storage")

ctx := context.Background()

// Upload file
file, _ := os.Open("document.pdf")
defer file.Close()

err := storageManager.Upload(ctx, "documents/report.pdf", file,
    storage.WithContentType("application/pdf"),
    storage.WithMetadata(map[string]string{
        "user_id": "12345",
        "uploaded_by": "john@example.com",
    }),
)

// Download file
reader, err := storageManager.Download(ctx, "documents/report.pdf")
if err != nil {
    log.Fatal(err)
}
defer reader.Close()

// Copy data
io.Copy(os.Stdout, reader)

// Check if exists
exists, err := storageManager.Exists(ctx, "documents/report.pdf")

// Get metadata
metadata, err := storageManager.Metadata(ctx, "documents/report.pdf")
fmt.Printf("Size: %d, Last Modified: %v\n", metadata.Size, metadata.LastModified)

// List files
objects, err := storageManager.List(ctx, "documents/",
    storage.WithRecursive(true),
    storage.WithLimit(100),
)

// Delete file
err = storageManager.Delete(ctx, "documents/report.pdf")
Advanced Operations
// Copy file
err := storageManager.Copy(ctx, "source.pdf", "destination.pdf")

// Move file
err := storageManager.Move(ctx, "old-location.pdf", "new-location.pdf")

// Generate presigned upload URL
uploadURL, err := storageManager.PresignUpload(ctx, "uploads/file.pdf", 15*time.Minute)
// Share uploadURL with client for direct upload

// Generate presigned download URL
downloadURL, err := storageManager.PresignDownload(ctx, "downloads/file.pdf", 1*time.Hour)
// Share downloadURL with client for direct download
Health Checks
// Basic health check
err := storageManager.Health(ctx)
if err != nil {
    log.Printf("Storage unhealthy: %v", err)
}

// Detailed health check
health, err := storageManager.HealthDetailed(ctx, true) // Check all backends
fmt.Printf("Healthy: %v, %d/%d backends healthy\n",
    health.Healthy, health.HealthyCount, health.BackendCount)

for name, backend := range health.Backends {
    fmt.Printf("Backend %s: %v (response time: %v)\n",
        name, backend.Healthy, backend.ResponseTime)
}

// Check specific backend
backendHealth, err := storageManager.BackendHealth(ctx, "s3")
Using Specific Backend
// Get specific backend
s3Backend := storageManager.Backend("s3")
if s3Backend != nil {
    err := s3Backend.Upload(ctx, "key", data)
}

Architecture

Component Overview
┌─────────────────────────────────────────────────────────┐
│                    Storage Extension                     │
├─────────────────────────────────────────────────────────┤
│                    Storage Manager                       │
│  ┌────────────────────────────────────────────────────┐ │
│  │              Health Checker                        │ │
│  │  • Multi-backend health monitoring                 │ │
│  │  • Write/List-based checks                         │ │
│  │  • Concurrent health checks                        │ │
│  └────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────┐
│                  Resilience Layer                        │
│  ┌────────────────┬──────────────┬──────────────────┐  │
│  │ Circuit Breaker │ Rate Limiter │ Retry w/ Backoff │  │
│  │ • Open/Closed   │ • Token Bucket│ • Exponential   │  │
│  │ • Half-Open     │ • Burst Support│ • Timeout      │  │
│  └────────────────┴──────────────┴──────────────────┘  │
└─────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────┐
│                    Storage Backends                      │
│  ┌──────────────┬──────────────┬──────────────────────┐│
│  │ Local Enhanced│    AWS S3    │  GCS / Azure (Soon) ││
│  │ • Atomic ops  │ • Presigned  │  • Coming soon      ││
│  │ • File locking│ • Multipart  │                     ││
│  │ • Buffer pool │ • Pagination │                     ││
│  └──────────────┴──────────────┴──────────────────────┘│
└─────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────┐
│                Cross-Cutting Concerns                    │
│  ┌────────────────┬──────────────┬──────────────────┐  │
│  │ Path Validator │ Buffer Pool  │ Security         │  │
│  │ • Traversal    │ • sync.Pool  │ • Validation     │  │
│  │ • Sanitization │ • Zero-alloc │ • Encryption     │  │
│  └────────────────┴──────────────┴──────────────────┘  │
└─────────────────────────────────────────────────────────┘
Resilience Patterns
Circuit Breaker States
    ┌─────────┐
    │ Closed  │──────► Failures ≥ Threshold
    └────┬────┘                 │
         │                      ▼
    Success             ┌───────────┐
         │              │   Open    │
         │              └─────┬─────┘
         │                    │
    ┌────▼─────┐              │ Timeout
    │Half-Open │◄─────────────┘
    └──────────┘

Metrics

The extension exposes the following metrics:

Operation Metrics
  • storage_uploads - Total upload count
  • storage_downloads - Total download count
  • storage_deletes - Total delete count
  • storage_upload_bytes - Total bytes uploaded
  • storage_upload_duration - Upload duration histogram
  • storage_download_duration - Download duration histogram
Error Metrics
  • storage_upload_errors - Upload error count
  • storage_download_errors - Download error count
  • storage_delete_errors - Delete error count
Resilience Metrics
  • storage_retries - Retry attempt count
  • storage_max_retries_exceeded - Max retries exceeded count
  • storage_circuit_breaker_open - Circuit breaker open events
  • storage_circuit_breaker_state - Current circuit breaker state (0=closed, 1=open, 2=half-open)
  • storage_rate_limit_allowed - Rate limit allowed requests
  • storage_rate_limit_rejected - Rate limit rejected requests
Health Metrics
  • storage_backend_health - Backend health status (1=healthy, 0=unhealthy)
  • storage_health_check_duration - Health check duration histogram

Testing

Run All Tests
cd v2/extensions/storage
go test -v
Run Integration Tests
go test -v -run 'TestStorageManager_.*'
Run Short Tests (Skip Integration)
go test -v -short
Run Benchmarks
go test -v -bench=. -benchmem
Coverage
go test -cover -coverprofile=coverage.out
go tool cover -html=coverage.out

Production Considerations

Local Backend
  • Use fast SSD storage for best performance
  • Configure regular backups
  • Monitor disk space
  • Use proper file permissions (0750 for directories, 0640 for files)
  • Generate and securely store the secret key
S3 Backend
  • Use IAM roles instead of access keys when possible
  • Enable versioning for important buckets
  • Configure lifecycle policies for automatic archival
  • Use VPC endpoints for better performance
  • Enable server-side encryption
  • Set appropriate CORS policies for presigned URLs
Resilience Tuning
  • Circuit Breaker Threshold: Higher for stable backends, lower for unstable
  • Retry Count: 3-5 retries for most cases
  • Backoff: Start at 100ms, max 10s
  • Rate Limit: Based on backend capacity and costs
  • Timeouts: 30s for normal ops, 5min for large uploads
Monitoring
  • Track error rates and set up alerts
  • Monitor circuit breaker state changes
  • Watch for rate limit rejections
  • Track P95/P99 latencies
  • Set up health check alerts

Security Best Practices

  1. Never commit secrets: Use environment variables or secret managers
  2. Rotate secrets regularly: Especially presigned URL secrets
  3. Validate all inputs: The extension validates, but add application-level checks
  4. Use HTTPS: Always use HTTPS for presigned URLs
  5. Set short expiry: Use minimum necessary expiry for presigned URLs
  6. Implement access control: Add application-level authorization
  7. Audit file access: Log all storage operations
  8. Scan uploaded files: Implement virus scanning for user uploads

Performance Tips

  1. Use buffer pooling: Reuse buffers for multiple operations
  2. Batch operations: Group related operations
  3. Use streaming: For large files, stream instead of loading into memory
  4. Enable CDN: For public assets, use CDN with presigned URLs
  5. Optimize chunks: Tune chunk size based on file sizes
  6. Parallel uploads: Use goroutines for multiple file uploads
  7. Cache metadata: Cache frequently accessed metadata

Troubleshooting

Circuit Breaker Keeps Opening
  • Check backend health and connectivity
  • Review timeout configuration
  • Examine error logs for root cause
  • Consider increasing threshold or timeout
High Latency
  • Check network latency to backend
  • Review buffer pool configuration
  • Monitor disk I/O (local backend)
  • Check for lock contention
Rate Limit Exceeded
  • Increase rate limit configuration
  • Implement client-side queuing
  • Use multiple backends for load distribution
Path Validation Errors
  • Review path validation rules
  • Use SanitizeKey() to clean paths
  • Check for special characters

Contributing

See CONTRIBUTING.md for development guidelines.

License

See LICENSE for license information.

Documentation

Overview

Package storage provides unified object storage with support for multiple backends including local filesystem, S3, GCS, and Azure Blob Storage.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// Configuration errors
	ErrNoBackendsConfigured   = errors.New("no storage backends configured")
	ErrNoDefaultBackend       = errors.New("no default backend specified")
	ErrDefaultBackendNotFound = errors.New("default backend not found in configuration")
	ErrInvalidBackendType     = errors.New("invalid backend type")
	ErrBackendNotFound        = errors.New("backend not found")

	// Operation errors
	ErrObjectNotFound        = errors.New("object not found")
	ErrObjectAlreadyExists   = errors.New("object already exists")
	ErrInvalidKey            = errors.New("invalid object key")
	ErrUploadFailed          = errors.New("upload failed")
	ErrDownloadFailed        = errors.New("download failed")
	ErrDeleteFailed          = errors.New("delete failed")
	ErrPresignNotSupported   = errors.New("presigned URLs not supported for this backend")
	ErrMultipartNotSupported = errors.New("multipart upload not supported for this backend")

	// Resilience errors
	ErrCircuitBreakerOpen = errors.New("circuit breaker is open")
	ErrRateLimitExceeded  = errors.New("rate limit exceeded")

	// Validation errors
	ErrFileTooLarge       = errors.New("file size exceeds maximum allowed size")
	ErrInvalidContentType = errors.New("invalid content type")
	ErrInvalidPath        = errors.New("invalid path: potential path traversal detected")
)

Functions

func IsValidSize 

func IsValidSize(size int64, maxSize int64) bool

IsValidSize checks if a file size is within limits

func NewExtension 

func NewExtension(config Config) forge.Extension

NewExtension creates a new storage extension

func ValidateMetadata 

func ValidateMetadata(metadata map[string]string) error

ValidateMetadata validates metadata keys and values

Types

type BackendConfig 

type BackendConfig struct {
	Type   string                 `yaml:"type" json:"type"` // local, s3, gcs, azure
	Config map[string]interface{} `yaml:"config" json:"config"`
}

BackendConfig is the configuration for a storage backend

type BackendHealth 

type BackendHealth struct {
	Name         string        `json:"name"`
	Healthy      bool          `json:"healthy"`
	ResponseTime time.Duration `json:"response_time"`
	Error        string        `json:"error,omitempty"`
	LastChecked  time.Time     `json:"last_checked"`
	CheckType    string        `json:"check_type"`
}

BackendHealth represents health status of a backend

type BufferPool 

type BufferPool struct {
	// contains filtered or unexported fields
}

BufferPool manages a pool of reusable buffers for efficient I/O operations

func NewBufferPool 

func NewBufferPool(bufferSize int) *BufferPool

NewBufferPool creates a new buffer pool with the specified buffer size

func (*BufferPool) Get 

func (bp *BufferPool) Get() []byte

Get retrieves a buffer from the pool

func (*BufferPool) Put 

func (bp *BufferPool) Put(buf []byte)

Put returns a buffer to the pool

type BytesBufferPool 

type BytesBufferPool struct {
	// contains filtered or unexported fields
}

BytesBufferPool manages a pool of bytes.Buffer for efficient operations

func NewBytesBufferPool 

func NewBytesBufferPool() *BytesBufferPool

NewBytesBufferPool creates a new bytes buffer pool

func (*BytesBufferPool) Get 

func (bbp *BytesBufferPool) Get() *bytes.Buffer

Get retrieves a buffer from the pool

func (*BytesBufferPool) Put 

func (bbp *BytesBufferPool) Put(buf *bytes.Buffer)

Put returns a buffer to the pool

type CircuitBreaker 

type CircuitBreaker struct {
	// contains filtered or unexported fields
}

CircuitBreaker implements circuit breaker pattern

func NewCircuitBreaker 

func NewCircuitBreaker(config ResilienceConfig, logger forge.Logger, metrics forge.Metrics) *CircuitBreaker

NewCircuitBreaker creates a new circuit breaker

func (*CircuitBreaker) CanAttempt 

func (cb *CircuitBreaker) CanAttempt() bool

CanAttempt checks if a request can be attempted

func (*CircuitBreaker) Execute 

func (cb *CircuitBreaker) Execute(ctx context.Context, name string, fn func() error) error

Execute executes a function with circuit breaker protection

func (*CircuitBreaker) GetState 

func (cb *CircuitBreaker) GetState() CircuitState

GetState returns current circuit state

func (*CircuitBreaker) RecordFailure 

func (cb *CircuitBreaker) RecordFailure()

RecordFailure records a failed execution

func (*CircuitBreaker) RecordSuccess 

func (cb *CircuitBreaker) RecordSuccess()

RecordSuccess records a successful execution

func (*CircuitBreaker) Reset 

func (cb *CircuitBreaker) Reset()

Reset resets the circuit breaker

type CircuitState 

type CircuitState int

CircuitState represents circuit breaker state 

const (
	CircuitClosed CircuitState = iota
	CircuitOpen
	CircuitHalfOpen
)

func (CircuitState) String 

func (s CircuitState) String() string

type Config 

type Config struct {
	// Default backend name
	Default string `yaml:"default" json:"default" default:"local"`

	// Backend configurations
	Backends map[string]BackendConfig `yaml:"backends" json:"backends"`

	// Features
	EnablePresignedURLs bool          `yaml:"enable_presigned_urls" json:"enable_presigned_urls" default:"true"`
	PresignExpiry       time.Duration `yaml:"presign_expiry" json:"presign_expiry" default:"15m"`
	MaxUploadSize       int64         `yaml:"max_upload_size" json:"max_upload_size" default:"5368709120"` // 5GB
	ChunkSize           int           `yaml:"chunk_size" json:"chunk_size" default:"5242880"`              // 5MB

	// CDN
	EnableCDN  bool   `yaml:"enable_cdn" json:"enable_cdn" default:"false"`
	CDNBaseURL string `yaml:"cdn_base_url" json:"cdn_base_url"`

	// Resilience configuration
	Resilience ResilienceConfig `yaml:"resilience" json:"resilience"`

	// Use enhanced backend (with locking, pooling, etc.)
	UseEnhancedBackend bool `yaml:"use_enhanced_backend" json:"use_enhanced_backend" default:"true"`
}

Config is the storage extension configuration

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns the default storage configuration

func (*Config) Validate 

func (c *Config) Validate() error

Validate validates the configuration

type EnhancedLocalBackend 

type EnhancedLocalBackend struct {
	// contains filtered or unexported fields
}

EnhancedLocalBackend implements enhanced local filesystem storage with proper locking and pooling

func NewEnhancedLocalBackend 

func NewEnhancedLocalBackend(config map[string]interface{}, logger forge.Logger, metrics forge.Metrics) (*EnhancedLocalBackend, error)

NewEnhancedLocalBackend creates a new enhanced local filesystem backend

func (*EnhancedLocalBackend) Copy 

func (b *EnhancedLocalBackend) Copy(ctx context.Context, srcKey, dstKey string) error

Copy copies a file with proper locking

func (*EnhancedLocalBackend) Delete 

func (b *EnhancedLocalBackend) Delete(ctx context.Context, key string) error

Delete deletes a file with proper locking

func (*EnhancedLocalBackend) Download 

func (b *EnhancedLocalBackend) Download(ctx context.Context, key string) (io.ReadCloser, error)

Download downloads a file with proper locking

func (*EnhancedLocalBackend) Exists 

func (b *EnhancedLocalBackend) Exists(ctx context.Context, key string) (bool, error)

Exists checks if an object exists

func (*EnhancedLocalBackend) List 

func (b *EnhancedLocalBackend) List(ctx context.Context, prefix string, opts ...ListOption) ([]Object, error)

List lists files with a prefix

func (*EnhancedLocalBackend) Metadata 

func (b *EnhancedLocalBackend) Metadata(ctx context.Context, key string) (*ObjectMetadata, error)

Metadata retrieves object metadata

func (*EnhancedLocalBackend) Move 

func (b *EnhancedLocalBackend) Move(ctx context.Context, srcKey, dstKey string) error

Move moves a file with proper locking

func (*EnhancedLocalBackend) PresignDownload 

func (b *EnhancedLocalBackend) PresignDownload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignDownload generates a presigned URL for download

func (*EnhancedLocalBackend) PresignUpload 

func (b *EnhancedLocalBackend) PresignUpload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignUpload generates a presigned URL for upload

func (*EnhancedLocalBackend) Upload 

func (b *EnhancedLocalBackend) Upload(ctx context.Context, key string, data io.Reader, opts ...UploadOption) error

Upload uploads a file with proper locking and validation

type EnhancedLocalConfig 

type EnhancedLocalConfig struct {
	RootDir       string
	BaseURL       string
	Secret        string
	ChunkSize     int64
	MaxUploadSize int64
}

EnhancedLocalConfig contains configuration for enhanced local backend

type Extension 

type Extension struct {
	// contains filtered or unexported fields
}

Extension implements the storage extension

func (*Extension) Dependencies 

func (e *Extension) Dependencies() []string

Dependencies returns the list of extension dependencies

func (*Extension) Description 

func (e *Extension) Description() string

Description returns the extension description

func (*Extension) Health 

func (e *Extension) Health(ctx context.Context) error

Health checks the extension health

func (*Extension) Name 

func (e *Extension) Name() string

Name returns the extension name

func (*Extension) Register 

func (e *Extension) Register(app forge.App) error

Register registers the extension with the application

func (*Extension) Start 

func (e *Extension) Start(ctx context.Context) error

Start starts the extension

func (*Extension) Stop 

func (e *Extension) Stop(ctx context.Context) error

Stop stops the extension

func (*Extension) Version 

func (e *Extension) Version() string

Version returns the extension version

type HealthCheckConfig 

type HealthCheckConfig struct {
	Timeout       time.Duration `yaml:"timeout" json:"timeout" default:"5s"`
	WriteTestFile bool          `yaml:"write_test_file" json:"write_test_file" default:"true"`
	TestKey       string        `yaml:"test_key" json:"test_key" default:".health_check"`
	CheckAll      bool          `yaml:"check_all" json:"check_all" default:"false"`
	EnableMetrics bool          `yaml:"enable_metrics" json:"enable_metrics" default:"true"`
}

HealthCheckConfig configures health check behavior

func DefaultHealthCheckConfig 

func DefaultHealthCheckConfig() HealthCheckConfig

DefaultHealthCheckConfig returns default health check configuration

type HealthChecker 

type HealthChecker struct {
	// contains filtered or unexported fields
}

HealthChecker performs comprehensive health checks on storage backends

func NewHealthChecker 

func NewHealthChecker(backends map[string]Storage, logger forge.Logger, metrics forge.Metrics, config HealthCheckConfig) *HealthChecker

NewHealthChecker creates a new health checker

func (*HealthChecker) CheckHealth 

func (hc *HealthChecker) CheckHealth(ctx context.Context, defaultBackend string, checkAll bool) (*OverallHealth, error)

CheckHealth performs health check on default backend or all backends

func (*HealthChecker) GetBackendHealth 

func (hc *HealthChecker) GetBackendHealth(ctx context.Context, name string) (*BackendHealth, error)

GetBackendHealth gets health status of a specific backend

type ListOption 

type ListOption func(*ListOptions)

ListOption is a functional option for listing

func WithLimit 

func WithLimit(limit int) ListOption

WithLimit sets the limit

func WithMarker 

func WithMarker(marker string) ListOption

WithMarker sets the marker

func WithRecursive 

func WithRecursive(recursive bool) ListOption

WithRecursive sets recursive listing

type ListOptions 

type ListOptions struct {
	Limit     int
	Marker    string
	Recursive bool
}

ListOptions contains list options

type LocalBackend 

type LocalBackend struct {
	// contains filtered or unexported fields
}

LocalBackend implements storage using local filesystem

func NewLocalBackend 

func NewLocalBackend(config map[string]interface{}, logger forge.Logger, metrics forge.Metrics) (*LocalBackend, error)

NewLocalBackend creates a new local filesystem backend

func (*LocalBackend) Copy 

func (b *LocalBackend) Copy(ctx context.Context, srcKey, dstKey string) error

Copy copies a file

func (*LocalBackend) Delete 

func (b *LocalBackend) Delete(ctx context.Context, key string) error

Delete deletes a file

func (*LocalBackend) Download 

func (b *LocalBackend) Download(ctx context.Context, key string) (io.ReadCloser, error)

Download downloads a file

func (*LocalBackend) Exists 

func (b *LocalBackend) Exists(ctx context.Context, key string) (bool, error)

Exists checks if an object exists

func (*LocalBackend) List 

func (b *LocalBackend) List(ctx context.Context, prefix string, opts ...ListOption) ([]Object, error)

List lists files with a prefix

func (*LocalBackend) Metadata 

func (b *LocalBackend) Metadata(ctx context.Context, key string) (*ObjectMetadata, error)

Metadata retrieves object metadata

func (*LocalBackend) Move 

func (b *LocalBackend) Move(ctx context.Context, srcKey, dstKey string) error

Move moves a file

func (*LocalBackend) PresignDownload 

func (b *LocalBackend) PresignDownload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignDownload generates a presigned URL for download

func (*LocalBackend) PresignUpload 

func (b *LocalBackend) PresignUpload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignUpload generates a presigned URL for upload

func (*LocalBackend) Upload 

func (b *LocalBackend) Upload(ctx context.Context, key string, data io.Reader, opts ...UploadOption) error

Upload uploads a file

type Object 

type Object struct {
	Key          string            `json:"key"`
	Size         int64             `json:"size"`
	LastModified time.Time         `json:"last_modified"`
	ETag         string            `json:"etag"`
	ContentType  string            `json:"content_type"`
	Metadata     map[string]string `json:"metadata"`
}

Object represents a storage object

type ObjectMetadata 

type ObjectMetadata struct {
	Key          string            `json:"key"`
	Size         int64             `json:"size"`
	LastModified time.Time         `json:"last_modified"`
	ETag         string            `json:"etag"`
	ContentType  string            `json:"content_type"`
	Metadata     map[string]string `json:"metadata"`
}

ObjectMetadata represents object metadata

type OverallHealth 

type OverallHealth struct {
	Healthy        bool                     `json:"healthy"`
	BackendCount   int                      `json:"backend_count"`
	HealthyCount   int                      `json:"healthy_count"`
	UnhealthyCount int                      `json:"unhealthy_count"`
	Backends       map[string]BackendHealth `json:"backends"`
	CheckedAt      time.Time                `json:"checked_at"`
}

OverallHealth represents overall storage health

type PathValidator 

type PathValidator struct {
	// contains filtered or unexported fields
}

PathValidator validates storage paths for security

func NewPathValidator 

func NewPathValidator() *PathValidator

NewPathValidator creates a new path validator with default rules

func (*PathValidator) SanitizeKey 

func (pv *PathValidator) SanitizeKey(key string) string

SanitizeKey sanitizes a key to make it safe

func (*PathValidator) ValidateContentType 

func (pv *PathValidator) ValidateContentType(contentType string) error

ValidateContentType validates content type

func (*PathValidator) ValidateKey 

func (pv *PathValidator) ValidateKey(key string) error

ValidateKey validates a storage key for security issues

type RateLimiter 

type RateLimiter struct {
	// contains filtered or unexported fields
}

RateLimiter implements token bucket rate limiting

func NewRateLimiter 

func NewRateLimiter(config ResilienceConfig, metrics forge.Metrics) *RateLimiter

NewRateLimiter creates a new rate limiter

func (*RateLimiter) Allow 

func (rl *RateLimiter) Allow() bool

Allow checks if a request is allowed

type ResilienceConfig 

type ResilienceConfig struct {
	// Retry configuration
	MaxRetries        int           `yaml:"max_retries" json:"max_retries" default:"3"`
	InitialBackoff    time.Duration `yaml:"initial_backoff" json:"initial_backoff" default:"100ms"`
	MaxBackoff        time.Duration `yaml:"max_backoff" json:"max_backoff" default:"10s"`
	BackoffMultiplier float64       `yaml:"backoff_multiplier" json:"backoff_multiplier" default:"2.0"`

	// Circuit breaker configuration
	CircuitBreakerEnabled     bool          `yaml:"circuit_breaker_enabled" json:"circuit_breaker_enabled" default:"true"`
	CircuitBreakerThreshold   uint32        `yaml:"circuit_breaker_threshold" json:"circuit_breaker_threshold" default:"5"`
	CircuitBreakerTimeout     time.Duration `yaml:"circuit_breaker_timeout" json:"circuit_breaker_timeout" default:"60s"`
	CircuitBreakerHalfOpenMax uint32        `yaml:"circuit_breaker_half_open_max" json:"circuit_breaker_half_open_max" default:"3"`

	// Rate limiting
	RateLimitEnabled bool `yaml:"rate_limit_enabled" json:"rate_limit_enabled" default:"true"`
	RateLimitPerSec  int  `yaml:"rate_limit_per_sec" json:"rate_limit_per_sec" default:"100"`
	RateLimitBurst   int  `yaml:"rate_limit_burst" json:"rate_limit_burst" default:"200"`

	// Timeout configuration
	OperationTimeout time.Duration `yaml:"operation_timeout" json:"operation_timeout" default:"30s"`
}

ResilienceConfig configures resilience features

func DefaultResilienceConfig 

func DefaultResilienceConfig() ResilienceConfig

DefaultResilienceConfig returns default resilience configuration

type ResilientStorage 

type ResilientStorage struct {
	// contains filtered or unexported fields
}

ResilientStorage wraps a storage backend with resilience features

func NewResilientStorage 

func NewResilientStorage(backend Storage, config ResilienceConfig, logger forge.Logger, metrics forge.Metrics) *ResilientStorage

NewResilientStorage creates a resilient storage wrapper

func (*ResilientStorage) Copy 

func (rs *ResilientStorage) Copy(ctx context.Context, srcKey, dstKey string) error

Copy copies with resilience

func (*ResilientStorage) Delete 

func (rs *ResilientStorage) Delete(ctx context.Context, key string) error

Delete deletes with resilience

func (*ResilientStorage) Download 

func (rs *ResilientStorage) Download(ctx context.Context, key string) (io.ReadCloser, error)

Download downloads with resilience

func (*ResilientStorage) Exists 

func (rs *ResilientStorage) Exists(ctx context.Context, key string) (bool, error)

Exists checks existence with resilience

func (*ResilientStorage) GetCircuitBreakerState 

func (rs *ResilientStorage) GetCircuitBreakerState() CircuitState

GetCircuitBreakerState returns circuit breaker state

func (*ResilientStorage) List 

func (rs *ResilientStorage) List(ctx context.Context, prefix string, opts ...ListOption) ([]Object, error)

List lists with resilience

func (*ResilientStorage) Metadata 

func (rs *ResilientStorage) Metadata(ctx context.Context, key string) (*ObjectMetadata, error)

Metadata gets metadata with resilience

func (*ResilientStorage) Move 

func (rs *ResilientStorage) Move(ctx context.Context, srcKey, dstKey string) error

Move moves with resilience

func (*ResilientStorage) PresignDownload 

func (rs *ResilientStorage) PresignDownload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignDownload presigns download URL with resilience

func (*ResilientStorage) PresignUpload 

func (rs *ResilientStorage) PresignUpload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignUpload presigns upload URL with resilience

func (*ResilientStorage) ResetCircuitBreaker 

func (rs *ResilientStorage) ResetCircuitBreaker()

ResetCircuitBreaker resets the circuit breaker

func (*ResilientStorage) Upload 

func (rs *ResilientStorage) Upload(ctx context.Context, key string, data io.Reader, opts ...UploadOption) error

Upload uploads with resilience

type S3Backend 

type S3Backend struct {
	// contains filtered or unexported fields
}

S3Backend implements storage using AWS S3

func NewS3Backend 

func NewS3Backend(configMap map[string]interface{}, logger forge.Logger, metrics forge.Metrics) (*S3Backend, error)

NewS3Backend creates a new S3 storage backend

func (*S3Backend) Copy 

func (b *S3Backend) Copy(ctx context.Context, srcKey, dstKey string) error

Copy copies a file within S3

func (*S3Backend) Delete 

func (b *S3Backend) Delete(ctx context.Context, key string) error

Delete deletes a file from S3

func (*S3Backend) Download 

func (b *S3Backend) Download(ctx context.Context, key string) (io.ReadCloser, error)

Download downloads a file from S3

func (*S3Backend) Exists 

func (b *S3Backend) Exists(ctx context.Context, key string) (bool, error)

Exists checks if an object exists

func (*S3Backend) List 

func (b *S3Backend) List(ctx context.Context, prefix string, opts ...ListOption) ([]Object, error)

List lists files with a prefix

func (*S3Backend) Metadata 

func (b *S3Backend) Metadata(ctx context.Context, key string) (*ObjectMetadata, error)

Metadata retrieves object metadata

func (*S3Backend) Move 

func (b *S3Backend) Move(ctx context.Context, srcKey, dstKey string) error

Move moves a file within S3 (copy + delete)

func (*S3Backend) PresignDownload 

func (b *S3Backend) PresignDownload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignDownload generates a presigned URL for download

func (*S3Backend) PresignUpload 

func (b *S3Backend) PresignUpload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignUpload generates a presigned URL for upload

func (*S3Backend) Upload 

func (b *S3Backend) Upload(ctx context.Context, key string, data io.Reader, opts ...UploadOption) error

Upload uploads a file to S3

type S3Config 

type S3Config struct {
	Region          string
	Bucket          string
	Prefix          string
	AccessKeyID     string
	SecretAccessKey string
	SessionToken    string
	Endpoint        string // For S3-compatible services
	UsePathStyle    bool   // For S3-compatible services
}

S3Config contains S3 configuration

type Storage 

type Storage interface {
	// Upload uploads an object
	Upload(ctx context.Context, key string, data io.Reader, opts ...UploadOption) error

	// Download downloads an object
	Download(ctx context.Context, key string) (io.ReadCloser, error)

	// Delete deletes an object
	Delete(ctx context.Context, key string) error

	// List lists objects with a prefix
	List(ctx context.Context, prefix string, opts ...ListOption) ([]Object, error)

	// Metadata retrieves object metadata
	Metadata(ctx context.Context, key string) (*ObjectMetadata, error)

	// Exists checks if an object exists
	Exists(ctx context.Context, key string) (bool, error)

	// Copy copies an object
	Copy(ctx context.Context, srcKey, dstKey string) error

	// Move moves an object
	Move(ctx context.Context, srcKey, dstKey string) error

	// PresignUpload generates a presigned URL for upload
	PresignUpload(ctx context.Context, key string, expiry time.Duration) (string, error)

	// PresignDownload generates a presigned URL for download
	PresignDownload(ctx context.Context, key string, expiry time.Duration) (string, error)
}

Storage defines the unified storage interface

type StorageManager 

type StorageManager struct {
	// contains filtered or unexported fields
}

StorageManager manages multiple storage backends

func NewStorageManager 

func NewStorageManager(config Config, logger forge.Logger, metrics forge.Metrics) *StorageManager

NewStorageManager creates a new storage manager

func (*StorageManager) Backend 

func (m *StorageManager) Backend(name string) Storage

Backend returns a specific backend

func (*StorageManager) BackendHealth 

func (m *StorageManager) BackendHealth(ctx context.Context, name string) (*BackendHealth, error)

BackendHealth returns health of a specific backend

func (*StorageManager) Copy 

func (m *StorageManager) Copy(ctx context.Context, srcKey, dstKey string) error

Copy copies in the default backend

func (*StorageManager) Delete 

func (m *StorageManager) Delete(ctx context.Context, key string) error

Delete deletes from the default backend

func (*StorageManager) Download 

func (m *StorageManager) Download(ctx context.Context, key string) (io.ReadCloser, error)

Download downloads from the default backend

func (*StorageManager) Exists 

func (m *StorageManager) Exists(ctx context.Context, key string) (bool, error)

Exists checks existence in the default backend

func (*StorageManager) GetURL 

func (m *StorageManager) GetURL(ctx context.Context, key string) string

GetURL returns the URL for an object (CDN or direct)

func (*StorageManager) Health 

func (m *StorageManager) Health(ctx context.Context) error

Health checks the health of all backends

func (*StorageManager) HealthDetailed 

func (m *StorageManager) HealthDetailed(ctx context.Context, checkAll bool) (*OverallHealth, error)

HealthDetailed returns detailed health information

func (*StorageManager) List 

func (m *StorageManager) List(ctx context.Context, prefix string, opts ...ListOption) ([]Object, error)

List lists from the default backend

func (*StorageManager) Metadata 

func (m *StorageManager) Metadata(ctx context.Context, key string) (*ObjectMetadata, error)

Metadata gets metadata from the default backend

func (*StorageManager) Move 

func (m *StorageManager) Move(ctx context.Context, srcKey, dstKey string) error

Move moves in the default backend

func (*StorageManager) PresignDownload 

func (m *StorageManager) PresignDownload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignDownload generates a presigned download URL for the default backend

func (*StorageManager) PresignUpload 

func (m *StorageManager) PresignUpload(ctx context.Context, key string, expiry time.Duration) (string, error)

PresignUpload generates a presigned upload URL for the default backend

func (*StorageManager) Start 

func (m *StorageManager) Start(ctx context.Context) error

Start initializes all storage backends

func (*StorageManager) Stop 

func (m *StorageManager) Stop(ctx context.Context) error

Stop closes all storage backends

func (*StorageManager) Upload 

func (m *StorageManager) Upload(ctx context.Context, key string, data io.Reader, opts ...UploadOption) error

Upload uploads to the default backend

type UploadOption 

type UploadOption func(*UploadOptions)

UploadOption is a functional option for uploads

func WithACL 

func WithACL(acl string) UploadOption

WithACL sets ACL

func WithContentType 

func WithContentType(contentType string) UploadOption

WithContentType sets the content type

func WithMetadata 

func WithMetadata(metadata map[string]string) UploadOption

WithMetadata sets metadata

type UploadOptions 

type UploadOptions struct {
	ContentType string
	Metadata    map[string]string
	ACL         string
}

UploadOptions contains upload options

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.