1package oshost23import (4 "fmt"5 "io"6 "log/slog"7 "os"8 "sort"9 "strings"10 "sync"11 "sync/atomic"12 "time"1314 "github.com/FelineStateMachine/atlas/format/semconv"15 "github.com/FelineStateMachine/atlas/format/vnext"16 "github.com/FelineStateMachine/atlas/internal/app/hostenv"17 "github.com/FelineStateMachine/atlas/internal/logging"18)1920// closeGrace is how long a retired bundle stays open after a rescan replaces21// it. A request served out of the previous snapshot finishes in milliseconds;22// the grace period outlives any of them without keeping descriptors around.23const closeGrace = 5 * time.Second2425// Volumes is a library kept as a directory of .atlas files.26//27// It scans at construction and after installation, and never watches (issue #528// §2, decision 15): a file copied directly into the library outside Atlas29// appears at the next launch. Desktop open and drop events install immediately.30// The fold that decides which build serves is format/vnext's and is pure; this31// type is the walk, the open files, and the swap.32//33// Every rescan builds a whole snapshot and swaps it in at once, so a request34// sees one consistent library however the directory is churning.35type Volumes struct {36 dir string37 snap atomic.Pointer[snapshot]3839 // scanning runs alone: two concurrent scans of one directory would race40 // to swap snapshots and could close each other's bundles.41 scanning sync.Mutex42}4344// snapshot is the library at one moment: the winning build per slug, the open45// readers behind them, and the order the handler lists them in.46type snapshot struct {47 serving map[string]vnext.Descriptor48 open map[string]*volume49 order []string50}5152// volume is one serving build with its archive held open, so answering for an53// entry is a seek and a read rather than a fresh open of a file that may be54// hundreds of megabytes.55type volume struct {56 reader *vnext.File57 locator string58 info hostenv.VolumeInfo59 outline vnext.Volume60}6162func (v *volume) Info() hostenv.VolumeInfo { return v.info }63func (v *volume) Outline() vnext.Volume { return v.outline }64func (v *volume) FeaturePage(request vnext.FeaturePageRequest) (vnext.FeaturePageResult, error) {65 return v.reader.FeaturePage(request)66}67func (v *volume) FeatureSetSummaries() ([]vnext.FeatureSetSummary, error) {68 return v.reader.FeatureSetSummaries()69}70func (v *volume) DemandAddressedFeatures() bool { return v.reader.DemandAddressedFeatures() }71func (v *volume) RasterTile(entry string) (vnext.RasterTile, error) {72 return v.reader.ReadRasterTile(entry)73}74func (v *volume) Blob(entry string) ([]byte, error) {75 return v.reader.Blob(entry)76}77func (v *volume) Schema() ([]byte, error) { return v.reader.FileSchema.Canonical() }78func (v *volume) TableBlock(entry string) ([]byte, error) { return v.reader.TableBlock(entry) }7980// NewVolumes answers for dir, which need not exist yet: a directory that is81// missing or empty is a library with no volumes, not an error. The first scan82// happens here, so a caller that gets a store gets a library.83func NewVolumes(dir string) (*Volumes, error) {84 v := &Volumes{dir: dir}85 v.snap.Store(&snapshot{serving: map[string]vnext.Descriptor{}, open: map[string]*volume{}})86 if _, err := v.Rescan(); err != nil {87 return nil, err88 }89 return v, nil90}9192// Location is the directory the library is kept in.93func (v *Volumes) Location() string { return v.dir }9495// Volumes lists the serving build of every installed volume, sorted by slug.96func (v *Volumes) Volumes() []hostenv.Volume {97 held := v.snap.Load()98 out := make([]hostenv.Volume, 0, len(held.order))99 for _, slug := range held.order {100 out = append(out, held.open[slug])101 }102 return out103}104105// Close closes every reader the current snapshot holds and leaves the store106// empty. The desktop host never calls it — a process done with its library is107// a process that is exiting — but a caller that opened a library it must108// later delete has to let the files go first, which on Windows is a rule109// rather than a courtesy.110func (v *Volumes) Close() error {111 v.scanning.Lock()112 defer v.scanning.Unlock()113 previous := v.snap.Load()114 v.snap.Store(&snapshot{serving: map[string]vnext.Descriptor{}, open: map[string]*volume{}})115 var first error116 for _, held := range previous.open {117 if err := held.reader.Close(); err != nil && first == nil {118 first = err119 }120 }121 return first122}123124// Rescan reads the directory whole and swaps in what it finds, reporting the125// volumes whose serving build moved. A bundle that will not open is logged126// and passed over rather than failing the scan: one bad download should not127// take a library down with it.128func (v *Volumes) Rescan() ([]string, error) {129 v.scanning.Lock()130 defer v.scanning.Unlock()131132 descriptors, skipped, err := vnext.Scan(v.dir, vnext.StandardSchema())133 if err != nil {134 return nil, fmt.Errorf("scan library: %w", err)135 }136 for _, passed := range skipped {137 slog.Warn("bundle skipped", logging.Op("scan"),138 logging.Path(passed.Locator), slog.Any("error", passed.Err))139 }140141 previous := v.snap.Load()142 winners := vnext.Fold(descriptors)143 next := &snapshot{144 serving: make(map[string]vnext.Descriptor, len(winners)),145 open: make(map[string]*volume, len(winners)),146 order: make([]string, 0, len(winners)),147 }148 for slug, winner := range winners {149 opened, err := carryOrOpen(previous, winner)150 if err != nil {151 slog.Warn("volume skipped", logging.Op("scan"),152 logging.Volume(slug), logging.Path(winner.Locator), slog.Any("error", err))153 continue154 }155 next.serving[slug] = winner156 next.open[slug] = opened157 next.order = append(next.order, slug)158 }159 sort.Strings(next.order)160 v.snap.Store(next)161162 retire(previous, next)163 changed := vnext.Changed(previous.serving, next.serving)164 for _, slug := range changed {165 if winner, still := next.serving[slug]; still {166 slog.Info("volume serving", logging.Op("scan"),167 logging.Volume(slug), logging.Stamp(vnext.ShortStamp(winner.Stamp)))168 } else {169 slog.Info("volume gone", logging.Op("scan"), logging.Volume(slug))170 }171 }172 return changed, nil173}174175// Install copies bundle content into the library under its versioned name and176// rescans. The content is staged to a temporary file first because the format177// validates a bundle before letting it in, and validation wants a file it can178// seek; the stage is removed either way.179func (v *Volumes) Install(name string, content io.Reader) (hostenv.Installed, error) {180 if err := os.MkdirAll(v.dir, 0o755); err != nil {181 return hostenv.Installed{}, fmt.Errorf("open library: %w", err)182 }183 // Staged inside the library so the install below is a rename within one184 // filesystem, and under a name that does not end in .atlas so a scan185 // running beside it passes the half-written file over.186 staged, err := os.CreateTemp(v.dir, ".importing-*")187 if err != nil {188 return hostenv.Installed{}, fmt.Errorf("stage %s: %w", name, err)189 }190 stage := staged.Name()191 defer os.Remove(stage)192 if _, err := io.Copy(staged, content); err != nil {193 staged.Close()194 return hostenv.Installed{}, fmt.Errorf("stage %s: %w", name, err)195 }196 if err := staged.Close(); err != nil {197 return hostenv.Installed{}, fmt.Errorf("stage %s: %w", name, err)198 }199200 standing := v.snap.Load().serving201 descriptor, err := vnext.Install(v.dir, stage, vnext.StandardSchema())202 if err != nil {203 return hostenv.Installed{}, fmt.Errorf("%s: %w", name, err)204 }205 installed := hostenv.Installed{206 Slug: descriptor.Slug,207 Title: descriptor.Title,208 Stamp: descriptor.Stamp,209 Bytes: descriptor.Size,210 Already: standing[descriptor.Slug].Stamp == descriptor.Stamp,211 }212 slog.Info("bundle installed", logging.Op("install"),213 logging.Volume(descriptor.Slug), logging.Stamp(vnext.ShortStamp(descriptor.Stamp)),214 logging.Path(descriptor.Locator))215216 changed, err := v.Rescan()217 if err != nil {218 return installed, err219 }220 installed.Changed = changed221 return installed, nil222}223224// carryOrOpen reuses the reader already open for a build whose file has not225// moved, and opens the file fresh otherwise.226func carryOrOpen(previous *snapshot, winner vnext.Descriptor) (*volume, error) {227 if held, ok := previous.open[winner.Slug]; ok && held.locator == winner.Locator && held.info.Stamp == winner.Stamp {228 return held, nil229 }230 reader, err := vnext.OpenFile(winner.Locator, vnext.StandardSchema())231 if err != nil {232 return nil, err233 }234 outline, err := reader.Outline()235 if err != nil {236 reader.Close()237 return nil, err238 }239 info, err := volumeInfo(winner, outline, reader.Reader)240 if err != nil {241 reader.Close()242 return nil, err243 }244 return &volume{reader: reader, locator: winner.Locator, outline: outline, info: info}, nil245}246247func volumeInfo(descriptor vnext.Descriptor, outline vnext.Volume, reader *vnext.Reader) (hostenv.VolumeInfo, error) {248 info := hostenv.VolumeInfo{249 Slug: descriptor.Slug, Title: descriptor.Title, Stamp: descriptor.Stamp,250 Release: descriptorRelease(descriptor), Worlds: make([]hostenv.WorldInfo, 0, len(outline.Worlds)),251 }252 summaries, err := reader.FeatureSetSummaries()253 if err != nil {254 return hostenv.VolumeInfo{}, fmt.Errorf("read feature summary: %w", err)255 }256 counts := make(map[string]int, len(summaries))257 for _, summary := range summaries {258 counts[summary.FeatureSet] = summary.Rows259 }260 for index, world := range outline.Worlds {261 listed := hostenv.WorldInfo{Slug: world.ID, Title: world.Title, UpdatedAt: descriptor.CreatedAt}262 for _, set := range world.FeatureSets {263 switch featureSetKind(set) {264 case semconv.GeometryPath:265 listed.Paths += counts[set.ID]266 case semconv.GeometryArea:267 listed.Areas += counts[set.ID]268 default:269 listed.Points += counts[set.ID]270 }271 }272 info.Worlds = append(info.Worlds, listed)273 if index == 0 {274 info.TileGrid = hostenv.TileGrid{275 SourceZoom: int(world.CoordinateSpace.SourceZoom),276 OriginX: int(world.CoordinateSpace.OriginX), OriginY: int(world.CoordinateSpace.OriginY),277 FirstTile: int(world.CoordinateSpace.FirstTile),278 TileSize: int(world.CoordinateSpace.TileSize), Size: int(world.CoordinateSpace.Size),279 }280 }281 }282 return info, nil283}284285func featureSetKind(set vnext.FeatureSet) string {286 for _, claim := range set.Claims {287 if claim.Field.Name == semconv.KeyGeometryKind {288 return claim.Value.String289 }290 }291 return strings.TrimPrefix(set.SemanticType, "geometry.")292}293294func descriptorRelease(descriptor vnext.Descriptor) vnext.Release {295 return vnext.Release{296 Title: descriptor.Title, CreatedAt: descriptor.CreatedAt, Revision: descriptor.Revision,297 Stamp: descriptor.Stamp, Worlds: descriptor.Worlds,298 }299}300301// retire closes the readers the previous snapshot held that the new one did302// not carry across, after a grace period long enough for any request already303// streaming out of one to finish.304func retire(previous, next *snapshot) {305 var closing []*volume306 for slug, held := range previous.open {307 if carried, still := next.open[slug]; still && carried == held {308 continue309 }310 closing = append(closing, held)311 }312 if len(closing) == 0 {313 return314 }315 time.AfterFunc(closeGrace, func() {316 for _, old := range closing {317 if err := old.reader.Close(); err != nil {318 slog.Warn("closing a retired bundle", logging.Op("scan"),319 logging.Path(old.locator), slog.Any("error", err))320 }321 }322 })323}