steamcache2/steamcache/steamcache.go

// steamcache/steamcache.go
package steamcache

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"sync"
	"time"

	"github.com/docker/go-units"
	"golang.org/x/sync/semaphore"

	"s1d3sw1ped/steamcache2/steamcache/logger"
	"s1d3sw1ped/steamcache2/steamcache/metrics"
	"s1d3sw1ped/steamcache2/vfs"
	"s1d3sw1ped/steamcache2/vfs/cache"
	"s1d3sw1ped/steamcache2/vfs/disk"
	"s1d3sw1ped/steamcache2/vfs/gc"
	"s1d3sw1ped/steamcache2/vfs/memory"
)

type SteamCache struct {
	address  string
	upstream string

	vfs vfs.VFS

	memory *memory.MemoryFS
	disk   *disk.DiskFS

	memorygc *gc.AsyncGCFS
	diskgc   *gc.AsyncGCFS

	server *http.Server
	client *http.Client
	cancel context.CancelFunc
	wg     sync.WaitGroup

	// Shutdown safety (Once hardening per existing patterns)
	shutdownOnce sync.Once

	// shutdownCh closed during Shutdown() to allow delayed disk-attach goroutines (the Size barrier waits) to skip late SetSlow and avoid post-shutdown side effects. Also enables wg tracking for hygiene (reaps on wg.Wait).
	shutdownCh chan struct{}

	// Request coalescing (owned by coalescer wrapper; see coalescing.go)
	coalescer *coalescer

	// Concurrency control
	maxConcurrentRequests int64
	requestSemaphore      *semaphore.Weighted

	// Per-client rate limiting (owned by clientRateLimiter wrapper; see ratelimit.go)
	clientRateLimiter    *clientRateLimiter
	maxRequestsPerClient int64

	// Hardening config fields (plumbed)
	maxObjectSize  int64
	trustedProxies []string

	// Service management
	serviceManager *ServiceManager

	// Metrics
	metrics *metrics.Metrics

	// processor (Phase 3 foundation) + 4 narrow interfaces + ctor injection in New.
	// Bulk logic preserved on SteamCache (intentionally; see handler.go deferral block).
	// SteamCache remains thin coordinator + lifecycle owner. Enables future fakes.
	processor *requestProcessor
}

// New creates a new SteamCache instance.
// Returns an error (instead of panicking) on invalid memorySize or diskSize strings.
// Also validates maxObjectSize and accepts trustedProxies for X-Forwarded-For handling.
// Empty maxObjectSize or nil trustedProxies are normalized to safe defaults before parsing.
// Callers must check the returned error.
// The two new positional parameters are a breaking change for direct importers of the simple constructor.
// Prefer NewWithOptions (or config file) for forward compatibility. See README migration notes.
func New(address string, memorySize string, diskSize string, diskPath, upstream, memoryGC, diskGC string, maxConcurrentRequests int64, maxRequestsPerClient int64, maxObjectSize string, trustedProxies []string) (*SteamCache, error) {
	memorysize, err := units.FromHumanSize(memorySize)
	if err != nil {
		return nil, fmt.Errorf("invalid memory size: %w", err)
	}

	disksize, err := units.FromHumanSize(diskSize)
	if err != nil {
		return nil, fmt.Errorf("invalid disk size: %w", err)
	}

	// Apply safe defaults before parsing user-provided sizes (handles zero-value Options)
	if maxObjectSize == "" {
		maxObjectSize = "0"
	}
	if trustedProxies == nil {
		trustedProxies = []string{}
	}

	maxObjBytes, err := units.FromHumanSize(maxObjectSize)
	if err != nil {
		return nil, fmt.Errorf("invalid max object size: %w", err)
	}

	c := cache.New()

	var m *memory.MemoryFS
	var mgc *gc.AsyncGCFS
	if memorysize > 0 {
		var err error
		m, err = memory.New(memorysize)
		if err != nil {
			return nil, err
		}
		memoryGCAlgo := gc.GCAlgorithm(memoryGC)
		if memoryGCAlgo == "" {
			memoryGCAlgo = gc.LRU // default to LRU
		}
		// Use hybrid async GC with thresholds: 80% async, 95% sync, 100% hard limit
		mgc = gc.NewAsync(m, memoryGCAlgo, true, 0.8, 0.95, 1.0)
	}

	var d *disk.DiskFS
	var dgc *gc.AsyncGCFS // for disk cases, created inside delayed attach goroutine (just before SetSlow) so preemptive ticker does not run during proxy/init window (delayed attach goroutine hygiene for shutdown races during disk Size() barrier)
	if disksize > 0 {
		diskGCAlgo := gc.GCAlgorithm(diskGC)
		if diskGCAlgo == "" {
			diskGCAlgo = gc.LRU // default to LRU
		}
		evictFn := gc.GetGCAlgorithm(diskGCAlgo)
		var err error
		d, err = disk.New(diskPath, disksize, evictFn)
		if err != nil {
			return nil, err
		}
		// Use hybrid async GC with thresholds: 80% async, 95% sync, 100% hard limit
		dgc = gc.NewAsync(d, diskGCAlgo, true, 0.8, 0.95, 1.0)
	}

	// NOTE: cache tier attach (SetFast/SetSlow + delayed disk attach gos) moved *after* sc construction (see below).
	// This enables wg tracking on sc.wg + shutdownCh signaling for attach goroutines (goroutine hygiene + robust Shutdown during pending init).
	// (Early return for invalid no-mem/no-disk still needed before heavy setup.)

	if disksize == 0 && memorysize == 0 {
		return nil, fmt.Errorf("no memory or disk cache configured")
	}

	transport := newHTTPTransport()
	client := newHTTPClient(transport)
	server := newHTTPServer(address)

	sc := &SteamCache{
		upstream: upstream,
		address:  address,
		vfs:      c,
		memory:   m,
		disk:     d,
		memorygc: mgc,
		diskgc:   dgc,
		client:   client,
		server:   server,

		// Initialize concurrency control fields (wrappers own maps + internal stop chans for cleanup goroutine)
		coalescer:             newCoalescer(),
		maxConcurrentRequests: maxConcurrentRequests,
		requestSemaphore:      semaphore.NewWeighted(maxConcurrentRequests),
		clientRateLimiter:     newClientRateLimiter(maxRequestsPerClient),
		maxRequestsPerClient:  maxRequestsPerClient,
		shutdownCh:            make(chan struct{}),

		// Hardening config plumbed
		maxObjectSize:  maxObjBytes,
		trustedProxies: trustedProxies,

		// Initialize service management
		serviceManager: NewServiceManager(),

		// Initialize metrics
		metrics: metrics.NewMetrics(),
	}

	// Wire metrics into TieredCache so promotions are counted.
	c.SetMetrics(sc.metrics)

	// Wire metrics into the concrete storage tiers (MemoryFS / DiskFS) so per-tier hits and evictions are counted.
	if m != nil {
		m.SetMetrics(sc.metrics)
	}
	if d != nil {
		d.SetMetrics(sc.metrics)
	}

	// Wire the request processor (constructor injection of interfaces for the owned wrappers + vfs + client + scalars).
	// Done after all fields including wrappers are set; before attach goroutines (no impact on lifecycle paths).
	sc.processor = newRequestProcessor(sc)

	// configure the cache to match the specified mode (memory only, disk only, or memory and disk) based on the provided sizes.
	// Disk tier SetSlow delayed until after disk init+eviction (evict func passed at disk.New for race-free guard; via Size barrier); mem immediate.
	// Attach goroutines now tracked on sc.wg (Add before go + Done), guarded by shutdownCh (skip SetSlow if Shutdown racing) for goroutine hygiene. (GC wrapper preemption addressed via analysis + wontfix on literal defer-creation.)

	if disksize == 0 && memorysize != 0 {
		// memory only mode - no disk
		c.SetSlow(mgc)
	} else if disksize != 0 && memorysize == 0 {
		// disk only mode: delay attach until disk ready (pure-proxy during scan; Create returns ErrNotFound until slow tier Set)
		sc.wg.Add(1)
		go func() {
			defer sc.wg.Done()
			_ = d.Size() // block on barrier per design (all Size callers during window do this; documented)
			select {
			case <-sc.shutdownCh:
				return // Shutdown raced; do not attach or SetSlow after stop
			default:
				c.SetSlow(dgc)
			}
		}()
	} else if disksize != 0 && memorysize != 0 {
		// memory and disk mode: fast mem immediate, disk delayed (mem-only during scan)
		c.SetFast(mgc)
		sc.wg.Add(1)
		go func() {
			defer sc.wg.Done()
			t0 := time.Now()
			_ = d.Size()
			select {
			case <-sc.shutdownCh:
				return
			default:
				c.SetSlow(dgc)
				logger.Logger.Info().
					Dur("attach_delay", time.Since(t0)).
					Msg("Disk slow tier attached (mixed mode); prior traffic was memory-only")
			}
		}()
	}

	// Log GC algorithm configuration
	if m != nil {
		logger.Logger.Info().Str("memory_gc", memoryGC).Msg("Memory cache GC algorithm configured")
	}
	if d != nil {
		logger.Logger.Info().Str("disk_gc", diskGC).Msg("Disk cache GC algorithm configured")
	}

	return sc, nil
}

func (sc *SteamCache) Run() error {
	if sc.upstream != "" {
		resp, err := sc.client.Get(sc.upstream)
		if err != nil {
			if resp != nil {
				_ = resp.Body.Close() // best-effort body close on connectivity error path; primary err returned
			}
			logger.Logger.Error().Err(err).Str("upstream", sc.upstream).Msg("Failed upstream connectivity check")
			return err
		}
		if resp.StatusCode != http.StatusOK {
			_ = resp.Body.Close() // best-effort body close on non-OK status; primary error returned
			logger.Logger.Error().Int("status_code", resp.StatusCode).Str("upstream", sc.upstream).Msg("Upstream connectivity check returned non-OK status")
			return fmt.Errorf("upstream returned status %d", resp.StatusCode)
		}
		_ = resp.Body.Close() // best-effort; success path close before ListenAndServe
	}

	sc.server.Handler = sc
	ctx, cancel := context.WithCancel(context.Background())
	sc.cancel = cancel

	// Start cleanup goroutine for old client limiters
	sc.wg.Add(1)
	go func() {
		defer sc.wg.Done()
		sc.cleanupOldClientLimiters()
	}()

	sc.wg.Add(1)
	go func() {
		defer sc.wg.Done()
		err := sc.server.ListenAndServe()
		if err != nil && err != http.ErrServerClosed {
			logger.Logger.Error().Err(err).Msg("Failed to start SteamCache2")
			return // goroutine cannot return err to caller; logged; shutdown path handles
		}
	}()

	<-ctx.Done()
	_ = sc.server.Shutdown(ctx) // shutdown error secondary (ListenAndServe already terminated or context done); logged elsewhere if fatal
	sc.wg.Wait()
	return nil
}

func (sc *SteamCache) Shutdown() {
	if sc == nil {
		return
	}
	sc.shutdownOnce.Do(func() {
		if sc.cancel != nil {
			sc.cancel()
		}
		// Stop all background managers started in New() (critical for test hygiene + no leaks when using httptest direct ServeHTTP paths that never call Run()).
		if sc.memorygc != nil {
			sc.memorygc.Stop()
		}
		if sc.diskgc != nil {
			sc.diskgc.Stop()
		}
		// Signal cleanup goroutine to exit so wg.Wait below does not hang indefinitely.
		// (stop chan owned by clientRateLimiter wrapper; signaling delegated for full lifecycle encapsulation)
		if sc.clientRateLimiter != nil {
			sc.clientRateLimiter.stop()
		}
		// Signal delayed attach goroutines (attach goroutine shutdown safety) to skip SetSlow if still waiting Size() or just finished.
		if sc.shutdownCh != nil {
			select {
			case <-sc.shutdownCh:
			default:
				close(sc.shutdownCh)
			}
		}
		sc.wg.Wait()
		// Brief reap window after stopping workers (helps goroutine delta checks see low counts quickly).
		time.Sleep(10 * time.Millisecond)
	})
}

// GetMetrics returns current metrics
func (sc *SteamCache) GetMetrics() *metrics.Stats {
	// Update cache sizes
	if sc.memory != nil {
		sc.metrics.SetMemoryCacheSize(sc.memory.Size())
	}
	if sc.disk != nil {
		// Note: blocks on initDone (post-eviction state) for accurate post-attach size during long disk init window.
		sc.metrics.SetDiskCacheSize(sc.disk.Size())
	}

	return sc.metrics.GetStats()
}

// ResetMetrics resets all metrics to zero
func (sc *SteamCache) ResetMetrics() {
	sc.metrics.Reset()
}

// newHTTPTransport returns a tuned http.Transport for upstream fetches.
// Extracted to shrink New (Phase 3).
func newHTTPTransport() *http.Transport {
	return &http.Transport{
		// Connection pooling optimizations
		MaxIdleConns:        500,               // Increased for high concurrency
		MaxIdleConnsPerHost: 100,               // Increased for better connection reuse
		MaxConnsPerHost:     200,               // Limit connections per host
		IdleConnTimeout:     300 * time.Second, // Longer idle timeout for better reuse

		// Dial optimizations
		DialContext: (&net.Dialer{
			Timeout:   10 * time.Second, // Faster connection timeout
			KeepAlive: 60 * time.Second, // Longer keep-alive
			DualStack: true,             // Enable dual-stack (IPv4/IPv6)
		}).DialContext,

		// Timeout optimizations
		TLSHandshakeTimeout:   5 * time.Second,  // Faster TLS handshake
		ResponseHeaderTimeout: 15 * time.Second, // Faster header timeout
		ExpectContinueTimeout: 1 * time.Second,  // Faster expect-continue

		// Performance optimizations
		DisableCompression: true,  // Steam doesn't use compression
		ForceAttemptHTTP2:  true,  // Enable HTTP/2 if available
		DisableKeepAlives:  false, // Enable keep-alives

		// Buffer optimizations
		WriteBufferSize: 64 * 1024, // 64KB write buffer
		ReadBufferSize:  64 * 1024, // 64KB read buffer

		// Connection reuse optimizations
		MaxResponseHeaderBytes: 1 << 20, // 1MB max header size
	}
}

// newHTTPClient returns a tuned http.Client wrapping the provided transport.
// Extracted to shrink New (Phase 3).
func newHTTPClient(transport *http.Transport) *http.Client {
	return &http.Client{
		Transport: transport,
		Timeout:   60 * time.Second, // Optimized timeout for better responsiveness
		// Add redirect policy for better performance
		CheckRedirect: func(req *http.Request, via []*http.Request) error {
			// Limit redirects to prevent infinite loops
			if len(via) >= 10 {
				return http.ErrUseLastResponse
			}
			return nil
		},
	}
}

// newHTTPServer returns a tuned http.Server for the given address.
// Extracted to shrink New (Phase 3).
func newHTTPServer(address string) *http.Server {
	return &http.Server{
		Addr:              address,
		ReadTimeout:       15 * time.Second,  // Optimized for faster response
		WriteTimeout:      30 * time.Second,  // Optimized for faster response
		IdleTimeout:       300 * time.Second, // Longer idle timeout for better connection reuse
		ReadHeaderTimeout: 5 * time.Second,   // Faster header timeout
		MaxHeaderBytes:    1 << 20,           // 1MB max header size
		// Connection optimizations will be handled in ServeHTTP method
	}
}