steamcache2/steamcache/steamcache.go

package steamcache

import (
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"s1d3sw1ped/SteamCache2/steamcache/avgcachestate"
	"s1d3sw1ped/SteamCache2/steamcache/logger"
	"s1d3sw1ped/SteamCache2/vfs"
	"s1d3sw1ped/SteamCache2/vfs/cache"
	"s1d3sw1ped/SteamCache2/vfs/cachestate"
	"s1d3sw1ped/SteamCache2/vfs/disk"
	"s1d3sw1ped/SteamCache2/vfs/gc"
	"s1d3sw1ped/SteamCache2/vfs/memory"
	syncfs "s1d3sw1ped/SteamCache2/vfs/sync"
	"strings"
	"sync"
	"time"

	"github.com/docker/go-units"
)

type SteamCache struct {
	address string
	vfs     vfs.VFS

	memory *memory.MemoryFS
	disk   *disk.DiskFS

	hits *avgcachestate.AvgCacheState

	dirty bool
	mu    sync.Mutex
}

func New(address string, memorySize string, memoryMultiplier int, diskSize string, diskMultiplier int, diskPath string) *SteamCache {
	memorysize, err := units.FromHumanSize(memorySize)
	if err != nil {
		panic(err)
	}

	disksize, err := units.FromHumanSize(diskSize)
	if err != nil {
		panic(err)
	}

	m := memory.New(memorysize)
	d := disk.New(diskPath, disksize)

	sc := &SteamCache{
		address: address,
		vfs: syncfs.New(
			cache.New(
				gc.New(
					m,
					memoryMultiplier,
					memorygc,
				),
				gc.New(
					d,
					diskMultiplier,
					diskgc,
				),
				cachehandler,
			),
		),

		memory: m,
		disk:   d,

		hits: avgcachestate.New(10000),
	}

	if d.Size() > d.Capacity() {
		diskgc(d, int(d.Size()-d.Capacity()))
	}

	return sc
}

func (sc *SteamCache) Run() {
	logger.Logger.Info().Str("address", sc.address).Msg("listening")

	sc.mu.Lock()
	sc.dirty = true
	sc.mu.Unlock()

	sc.LogStats()
	t := time.NewTicker(10 * time.Second)
	go func() {
		for range t.C {
			sc.LogStats()
		}
	}()

	err := http.ListenAndServe(sc.address, sc)
	if err != nil {
		if err == http.ErrServerClosed {
			logger.Logger.Info().Msg("shutdown")
			return
		}
		logger.Logger.Error().Err(err).Msg("Failed to start SteamCache2")
		os.Exit(1)
	}
}

func (sc *SteamCache) LogStats() {
	sc.mu.Lock()
	defer sc.mu.Unlock()
	if sc.dirty {
		logger.Logger.Info().
			Str("memory", fmt.Sprintf("%s/%s", units.HumanSize(float64(sc.memory.Size())), units.HumanSize(float64(sc.memory.Capacity())))).Int("memory-files", len(sc.memory.StatAll())).
			Str("disk", fmt.Sprintf("%s/%s", units.HumanSize(float64(sc.disk.Size())), units.HumanSize(float64(sc.disk.Capacity())))).Int("disk-files", len(sc.disk.StatAll())).
			Str("hitrate", fmt.Sprintf("%.2f%%", sc.hits.Avg()*100)).
			Msg("stats")
		sc.dirty = false
	}
}

func (sc *SteamCache) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		http.Error(w, "Only GET method is supported", http.StatusMethodNotAllowed)
		return
	}

	if r.URL.String() == "/lancache-heartbeat" {
		w.Header().Add("X-LanCache-Processed-By", "SteamCache2")
		w.WriteHeader(http.StatusNoContent)
		w.Write(nil)
		return
	}

	if r.Header.Get("User-Agent") != "Valve/Steam HTTP Client 1.0" {
		http.Error(w, "Only Valve/Steam HTTP Client 1.0 is supported", http.StatusForbidden)
		return
	}

	if strings.Contains(r.URL.String(), "manifest") {
		w.Header().Add("X-LanCache-Processed-By", "SteamCache2")
		forward(w, r)
		return
	}

	// tstart := time.Now()
	// defer func() {
	// logger.Logger.Info().Str("method", r.Method).Str("url", r.URL.String()).Str("status", w.Header().Get("X-LanCache-Status")).Dur("duration", time.Since(tstart)).Msg("Request")
	// }()

	sc.mu.Lock()
	sc.dirty = true
	sc.mu.Unlock()

	w.Header().Add("X-LanCache-Processed-By", "SteamCache2") // SteamPrefill uses this header to determine if the request was processed by the cache maybe steam uses it too

	cacheKey := r.URL.String()

	// if vfs is also a vfs.GetSer, we can use it to get the cache state

	data, err := sc.vfs.Get(cacheKey)
	if err == nil {
		sc.hits.Add(cachestate.CacheStateHit)
		w.Header().Add("X-LanCache-Status", "HIT")
		w.Write(data)
		return
	}

	htt := "http://"
	if r.Header.Get("X-Sls-Https") == "enable" {
		htt = "https://"
	}

	base := htt + r.Host

	hosturl, err := url.JoinPath(base, cacheKey)
	if err != nil {
		http.Error(w, "Failed to join URL path", http.StatusInternalServerError)
		return
	}

	resp, err := http.Get(hosturl)
	if err != nil {
		http.Error(w, "Failed to fetch the requested URL", http.StatusInternalServerError)
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		http.Error(w, "Failed to fetch the requested URL", resp.StatusCode)
		return
	}

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		http.Error(w, "Failed to read response body", http.StatusInternalServerError)
		return
	}

	sc.vfs.Set(cacheKey, body)
	sc.hits.Add(cachestate.CacheStateMiss)
	w.Header().Add("X-LanCache-Status", "MISS")
	w.Write(body)
}

func forward(w http.ResponseWriter, r *http.Request) {
	htt := "http://"
	if r.Header.Get("X-Sls-Https") == "enable" {
		htt = "https://"
	}

	base := htt + r.Host

	cacheKey := r.URL.String()

	hosturl, err := url.JoinPath(base, cacheKey)
	if err != nil {
		http.Error(w, "Failed to join URL path", http.StatusInternalServerError)
		return
	}

	resp, err := http.Get(hosturl)
	if err != nil {
		http.Error(w, "Failed to fetch the requested URL", http.StatusInternalServerError)
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		http.Error(w, "Failed to fetch the requested URL", resp.StatusCode)
		return
	}

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		http.Error(w, "Failed to read response body", http.StatusInternalServerError)
		return
	}

	w.Write(body)
}