1package bundle23import (4 "context"5 "encoding/json"6 "fmt"7 "io"8 "log/slog"9 "os"10 "path/filepath"11 "sort"12 "strings"13 "sync"14 "sync/atomic"15 "time"1617 "github.com/fsnotify/fsnotify"18)1920// BasePath is the URL prefix volume content is served under. It appears in the21// composed catalog so the frontend never assembles a content URL from parts22// it has to guess.23const BasePath = "/data/v"2425// closeGrace is how long a retired bundle stays open after a rescan replaces26// it. A request served from the old snapshot finishes in milliseconds; the27// grace period outlives any of them without keeping descriptors around.28const closeGrace = 5 * time.Second2930// Registry holds the bundles of one directory and answers for them as a31// single catalog. Every rescan builds a complete immutable snapshot and swaps32// it in whole, so a request sees one consistent world however the directory33// is churning.34type Registry struct {35 dir string36 snap atomic.Pointer[Snapshot]3738 // rescan runs alone: two concurrent scans of one directory would race to39 // swap snapshots and could close each other's bundles.40 rescan sync.Mutex4142 // onChange, when set, is told which games a rescan added, replaced, or43 // removed. Set it before watching begins; it is read under the rescan44 // lock.45 onChange func(changed []string)46}4748// Snapshot is the registry's world at one moment: the winning bundle per49// game and the catalog those winners compose.50type Snapshot struct {51 // Volumes maps each volume slug to its bundle.52 Volumes map[string]*Bundle53 // Catalog is the merged /data/catalog.json body, marshaled once at scan54 // time because every launch fetches it and it only changes here.55 Catalog []byte56}5758// catalogVolume is one volume as the composed catalog lists it. The base URL59// carries the version stamp, so everything beneath it may be cached as60// immutable and a new version of the volume arrives at new URLs.61type catalogVolume struct {62 Slug string `json:"slug"`63 Title string `json:"title"`64 Stamp string `json:"stamp"`65 Base string `json:"base"`66 TileGrid TileGrid `json:"tileGrid"`67 Worlds []WorldEntry `json:"worlds"`68}6970// NewRegistry answers for dir, which need not exist yet: a directory that is71// missing or empty is a catalog with no volumes, not an error.72func NewRegistry(dir string) *Registry {73 registry := &Registry{dir: dir}74 registry.snap.Store(&Snapshot{Volumes: map[string]*Bundle{}, Catalog: composeCatalog(dir, nil)})75 return registry76}7778// Dir is the directory the registry scans.79func (r *Registry) Dir() string { return r.dir }8081// SetOnChange registers the listener a rescan tells about arrivals, updates,82// and departures. Call it before Watch.83func (r *Registry) SetOnChange(fn func(changed []string)) {84 r.rescan.Lock()85 defer r.rescan.Unlock()86 r.onChange = fn87}8889// Snapshot returns the current world. The result is immutable; callers keep90// it for the length of one request and load afresh for the next.91func (r *Registry) Snapshot() *Snapshot {92 return r.snap.Load()93}9495// Rescan reads the directory whole and swaps in what it finds. A bundle that96// fails to open is logged and skipped rather than failing the scan: one bad97// download should not take the library down with it. Bundles retired by the98// swap close after a grace period, so requests already streaming from them99// finish undisturbed.100func (r *Registry) Rescan() error {101 r.rescan.Lock()102 defer r.rescan.Unlock()103104 paths, err := filepath.Glob(filepath.Join(r.dir, "*.atlas"))105 if err != nil {106 return fmt.Errorf("scan %s: %w", r.dir, err)107 }108 sort.Strings(paths)109110 previous := r.snap.Load()111 next := &Snapshot{Volumes: make(map[string]*Bundle, len(paths))}112 opened := make([]*Bundle, 0, len(paths))113 for _, path := range paths {114 // A bundle already serving is not reopened: the file it was opened115 // from still holds its content even if the name has since been116 // rewritten, and its manifest still describes it. A changed file117 // arrives as a fresh open below and wins or loses on its manifest.118 loaded, err := openOrCarry(previous, path)119 if err != nil {120 slog.Warn("atlas: skipping bundle", "path", path, "error", err)121 continue122 }123 opened = append(opened, loaded)124 if standing, ok := next.Volumes[loaded.Manifest.Volume.Slug]; !ok || MoreRecent(loaded, standing) {125 if ok {126 slog.Info("atlas: bundle shadowed",127 "volume", loaded.Manifest.Volume.Slug, "shadowed", standing.Path)128 }129 next.Volumes[loaded.Manifest.Volume.Slug] = loaded130 }131 }132 next.Catalog = composeCatalog(r.dir, next.Volumes)133 r.snap.Store(next)134135 // What the rescan changed, by volume: an arrival, a departure, or a winner136 // whose version stamp moved.137 var changed []string138 for slug, winner := range next.Volumes {139 before, had := previous.Volumes[slug]140 if !had || before.Manifest.Version.Stamp != winner.Manifest.Version.Stamp {141 changed = append(changed, slug)142 }143 }144 for slug := range previous.Volumes {145 if _, still := next.Volumes[slug]; !still {146 changed = append(changed, slug)147 }148 }149 sort.Strings(changed)150151 // Everything opened by this scan or held by the last snapshot that did152 // not win a slug is retired together.153 serving := make(map[*Bundle]bool, len(next.Volumes))154 for _, winner := range next.Volumes {155 serving[winner] = true156 }157 var retired []*Bundle158 for _, candidate := range opened {159 if !serving[candidate] {160 retired = append(retired, candidate)161 }162 }163 for _, held := range previous.Volumes {164 if !serving[held] {165 retired = append(retired, held)166 }167 }168 if len(retired) > 0 {169 time.AfterFunc(closeGrace, func() {170 for _, old := range retired {171 old.Close()172 }173 })174 }175 if len(changed) > 0 && r.onChange != nil {176 r.onChange(changed)177 }178 return nil179}180181// Watch rescans whenever the directory changes, until ctx ends. Events are182// debounced: a download that arrives as a burst of writes costs one rescan,183// and the rescan reads the directory whole, so no event is ever missed for184// having been coalesced.185func (r *Registry) Watch(ctx context.Context) error {186 watcher, err := fsnotify.NewWatcher()187 if err != nil {188 return fmt.Errorf("watch %s: %w", r.dir, err)189 }190 if err := watcher.Add(r.dir); err != nil {191 watcher.Close()192 return fmt.Errorf("watch %s: %w", r.dir, err)193 }194 go func() {195 defer watcher.Close()196 debounce := time.NewTimer(0)197 if !debounce.Stop() {198 <-debounce.C199 }200 for {201 select {202 case <-ctx.Done():203 return204 case event, ok := <-watcher.Events:205 if !ok {206 return207 }208 // Only finished bundles matter: the importer and the bundler209 // both assemble under other names and rename into place.210 if strings.HasSuffix(event.Name, ".atlas") {211 debounce.Reset(500 * time.Millisecond)212 }213 case err, ok := <-watcher.Errors:214 if !ok {215 return216 }217 slog.Warn("atlas: watching bundles", "error", err)218 case <-debounce.C:219 if err := r.Rescan(); err != nil {220 slog.Warn("atlas: rescanning bundles", "error", err)221 }222 }223 }224 }()225 return nil226}227228// Release closes the open archive behind one volume so its file can be renamed229// over on platforms that lock open files. The snapshot still lists the volume230// until the next rescan; a request in that window is answered with a 404,231// which the frontend already treats as its cue to refetch the catalog.232func (r *Registry) Release(slug string) {233 r.rescan.Lock()234 defer r.rescan.Unlock()235 if held := r.snap.Load().Volumes[slug]; held != nil {236 held.Close()237 }238}239240// Install copies bundle files into the directory under their versioned241// names -- <volume>-<capture-day>-<stamp>.atlas -- validating each before it242// is let in, and rescans once at the end. Versions sit side by side and the243// newest-wins fold serves the right one, so importing an old file can never244// overwrite a newer build; it just arrives already shadowed. It reports the245// volumes installed and, separately, what was wrong with anything refused:246// one bad file does not turn the rest of a multi-selection away.247func (r *Registry) Install(paths []string) (installed []string, refused []string) {248 for _, path := range paths {249 slug, err := r.installOne(path)250 if err != nil {251 refused = append(refused, err.Error())252 continue253 }254 installed = append(installed, slug)255 }256 if len(installed) > 0 {257 if err := r.Rescan(); err != nil {258 refused = append(refused, err.Error())259 }260 }261 return installed, refused262}263264func (r *Registry) installOne(path string) (string, error) {265 opened, err := Open(path)266 if err != nil {267 return "", err268 }269 defer opened.Close()270 if err := opened.Validate(); err != nil {271 return "", fmt.Errorf("%s: %w", filepath.Base(path), err)272 }273 slug := opened.Manifest.Volume.Slug274275 target := filepath.Join(r.dir, VersionedFileName(opened.Manifest))276 if same, err := filepath.Abs(path); err == nil {277 if resolved, err := filepath.Abs(target); err == nil && same == resolved {278 return slug, nil279 }280 }281 // This exact build already installed under its own name: nothing to copy.282 if _, err := os.Stat(target); err == nil {283 return slug, nil284 }285 source, err := os.Open(path)286 if err != nil {287 return "", err288 }289 defer source.Close()290 staged, err := os.CreateTemp(r.dir, ".importing-*")291 if err != nil {292 return "", err293 }294 defer os.Remove(staged.Name())295 if _, err := io.Copy(staged, source); err != nil {296 staged.Close()297 return "", fmt.Errorf("copy %s: %w", filepath.Base(path), err)298 }299 if err := staged.Close(); err != nil {300 return "", err301 }302 if err := os.Chmod(staged.Name(), 0o644); err != nil {303 return "", err304 }305 if err := os.Rename(staged.Name(), target); err != nil {306 // Windows refuses to rename over a file that is open. Closing the307 // volume's archive first is what makes replacing an installed volume308 // possible there; the moment of unreadability ends at the rescan309 // Install finishes with.310 r.Release(slug)311 if err := os.Rename(staged.Name(), target); err != nil {312 return "", fmt.Errorf("install %s: %w", filepath.Base(path), err)313 }314 }315 return slug, nil316}317318// openOrCarry reuses the already-open bundle for a path whose size and319// modification time have not moved, and opens the file fresh otherwise.320func openOrCarry(previous *Snapshot, path string) (*Bundle, error) {321 for _, held := range previous.Volumes {322 if held.Path != path {323 continue324 }325 info, err := os.Stat(path)326 if err == nil && info.Size() == held.size && info.ModTime().Equal(held.modTime) {327 return held, nil328 }329 break330 }331 return Open(path)332}333334func composeCatalog(dir string, games map[string]*Bundle) []byte {335 listed := make([]catalogVolume, 0, len(games))336 for slug, held := range games {337 manifest := held.Manifest338 listed = append(listed, catalogVolume{339 Slug: slug,340 Title: manifest.Volume.Title,341 Stamp: manifest.Version.Stamp,342 Base: strings.Join([]string{BasePath, slug, ShortStamp(manifest.Version.Stamp)}, "/"),343 TileGrid: manifest.TileGrid,344 Worlds: manifest.Worlds,345 })346 }347 sort.Slice(listed, func(i, j int) bool { return listed[i].Title < listed[j].Title })348 // The bundles directory rides along so an empty library can tell the349 // reader where a bundle goes, in the words of their own machine.350 composed, err := json.Marshal(struct {351 Volumes []catalogVolume `json:"volumes"`352 BundlesDir string `json:"bundlesDir,omitempty"`353 }{Volumes: listed, BundlesDir: dir})354 if err != nil {355 // Nothing in a validated manifest can fail to marshal; if something356 // does, an empty catalog is a saner face than a panic.357 slog.Error("atlas: compose catalog", "error", err)358 return []byte(`{"games":[]}`)359 }360 return composed361}