Maven/internal/media/store.go

package media

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"
)

// DefaultMaxBytes — the per-blob cap when a store is built without one. 64 MiB
// is about an hour of 16 kHz mono PCM, which is also the hearing capture's own
// ceiling; a single item bigger than that is a mistake, not a meeting.
const DefaultMaxBytes int64 = 64 << 20

// DefaultMaxAudioBytes — the per-blob cap for audio. Separate from
// DefaultMaxBytes because the two kinds are not the same size of thing: an
// image over 64 MiB is a mistake, and a two-hour meeting at 16 kHz mono is
// about 230 MB of PCM by design. With one shared cap, capture's own
// DefaultMaxDuration of two hours and this store's 64 MiB contradicted each
// other, and the meeting that hit the limit was the one that failed to store.
const DefaultMaxAudioBytes int64 = 512 << 20

// DefaultRetention — how long a blob is kept when no retention is configured.
// Seven days is long enough to re-run a transcription that came out wrong and
// short enough that "she has a month of my meetings on disk" is never true.
const DefaultRetention = 7 * 24 * time.Hour

// DefaultMaxTotalBytes — the whole-store budget when one is not configured. The
// per-blob cap bounds one call and nothing bounded the sum of them: 64 MiB per
// call, an unlimited number of calls, and a seven-day window fills the disk
// mavend's database lives on. Content addressing does not help, because one
// flipped pixel is a different digest. 4 GiB is roughly sixty meetings or a few
// thousand photos inside the window.
const DefaultMaxTotalBytes int64 = 4 << 30

const (
	// dirPerm and filePerm: these are recordings of people, so the daemon's
	// user is the only reader.
	dirPerm  fs.FileMode = 0o700
	filePerm fs.FileMode = 0o600
	// metaExt is the sidecar suffix. Anything else under a bucket is blob
	// bytes, which is how the walkers tell the two apart.
	metaExt = ".json"
	// spoolName is the incremental-write directory, held outside the kind
	// directories so no walker mistakes a half-written file for a blob.
	spoolName = "spool"
	// bucketPrefix is how many leading id characters name the subdirectory, so
	// one kind is spread over 256 directories rather than one flat listing.
	bucketPrefix = 2
	// idLen is the length of a hex sha256, which is the only id shape a path
	// is ever built from.
	idLen = 64
)

// allKinds is every kind a walker has to visit. A store-wide operation covers
// all of them, and this is the one list to extend when a third kind lands.
var allKinds = []Kind{KindImage, KindAudio}

// ErrStoreFull — the store is at its total-bytes budget. Distinct from
// ErrTooLarge: the payload is a reasonable size and there is no room for it, so
// the answer is to prune or raise the budget, not to send something smaller.
var ErrStoreFull = errors.New("media: store is full")

// Store — a content-addressed blob directory. Zero value is not usable; build
// one with Open, which creates the directory 0700. The store holds no lock and
// no cache: every operation is a filesystem call, and two writers of the same
// bytes produce the same file, so concurrent Puts do not need coordinating.
type Store struct {
	dir       string
	maxBytes  int64
	maxAudio  int64
	maxTotal  int64
	retention time.Duration
	now       func() time.Time

	// total is the running sum of stored blob bytes, seeded by Open with a
	// directory walk and kept up to date by Put, Delete and Prune. It is a
	// cache of something the filesystem already knows: re-walking on every Put
	// would be correct too and would make an image intake O(store size).
	totalMu sync.Mutex
	total   int64
}

// Open prepares a blob store rooted at dir. maxBytes ≤ 0 ⇒ DefaultMaxBytes;
// retention ≤ 0 ⇒ DefaultRetention. The directory (and every kind subdirectory
// created later) is 0700: these are recordings of people, and the daemon's user
// is the only reader.
func Open(dir string, maxBytes int64, retention time.Duration) (*Store, error) {
	return OpenWithBudget(dir, maxBytes, 0, retention)
}

// OpenWithBudget is Open with the whole-store budget spelled out. maxTotal ≤ 0
// ⇒ DefaultMaxTotalBytes.
func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duration) (*Store, error) {
	if strings.TrimSpace(dir) == "" {
		return nil, errors.New("media: empty dir")
	}
	abs, err := filepath.Abs(dir)
	if err != nil {
		return nil, fmt.Errorf("media: resolve dir: %w", err)
	}
	if err := os.MkdirAll(abs, dirPerm); err != nil {
		return nil, fmt.Errorf("media: create dir: %w", err)
	}
	if maxBytes <= 0 {
		maxBytes = DefaultMaxBytes
	}
	maxAudio := DefaultMaxAudioBytes
	if maxBytes > maxAudio {
		maxAudio = maxBytes
	}
	if maxTotal <= 0 {
		maxTotal = DefaultMaxTotalBytes
	}
	if maxTotal < maxAudio {
		maxAudio = maxTotal
	}
	if maxTotal < maxBytes {
		return nil, fmt.Errorf("media: max_total_bytes %d is below the per-blob cap %d", maxTotal, maxBytes)
	}
	if retention <= 0 {
		retention = DefaultRetention
	}
	s := &Store{dir: abs, maxBytes: maxBytes, maxAudio: maxAudio, maxTotal: maxTotal,
		retention: retention, now: time.Now}
	s.total = s.measure()
	return s, nil
}

// measure sums what is already on disk, so a restart does not start the budget
// over at zero.
func (s *Store) measure() int64 {
	var total int64
	spool := filepath.Join(s.dir, spoolName)
	_ = filepath.WalkDir(s.dir, func(path string, d fs.DirEntry, err error) error {
		if err == nil && d.IsDir() && path == spool {
			// Spool files are not blobs yet and PutFile counts them when they
			// become one. Counting them here too would double them.
			return filepath.SkipDir
		}
		if err != nil || d.IsDir() || strings.HasSuffix(path, metaExt) {
			return nil //nolint:nilerr // an unreadable corner is not worth refusing to boot over
		}
		if info, err := d.Info(); err == nil {
			total += info.Size()
		}
		return nil
	})
	return total
}

// Total is the number of blob bytes currently stored, and Budget the cap Put
// checks it against. Both are exported so the daemon can log how close it is.
func (s *Store) Total() int64 {
	s.totalMu.Lock()
	defer s.totalMu.Unlock()
	return s.total
}

// Budget is the whole-store cap.
func (s *Store) Budget() int64 { return s.maxTotal }

// Dir is the store root. Exported for logs and for pointing a subprocess at a
// path under it.
func (s *Store) Dir() string { return s.dir }

// Retention is the configured age limit Prune enforces.
func (s *Store) Retention() time.Duration { return s.retention }

// reserve claims size against the whole-store budget and reports ErrStoreFull
// when there is no room. Claiming before the write means two concurrent Puts
// cannot both pass a check that only one of them fits through.
func (s *Store) reserve(size int64) error {
	s.totalMu.Lock()
	room := s.total+size <= s.maxTotal
	if room {
		s.total += size
	}
	stored := s.total
	s.totalMu.Unlock()
	if !room {
		return fmt.Errorf("%w: %d stored, %d budget, %d more asked for",
			ErrStoreFull, stored, s.maxTotal, size)
	}
	return nil
}

// release gives size back, for a reservation whose write failed and for bytes
// a delete removed. The floor at zero keeps a miscount from reading as a store
// that owes itself space.
func (s *Store) release(size int64) {
	s.totalMu.Lock()
	s.total -= size
	if s.total < 0 {
		s.total = 0
	}
	s.totalMu.Unlock()
}

// bucket is the directory an id lives in. Every path the store builds goes
// through here, so the traversal guard in validID has one place to sit.
func (s *Store) bucket(kind Kind, id string) string {
	return filepath.Join(s.dir, string(kind), id[:bucketPrefix])
}

// newBlob describes what is about to be stored. A blob already on disk keeps
// its first-seen time: re-sending the same photo every hour must not keep it
// alive past retention.
func (s *Store) newBlob(kind Kind, mime, source, id, blobPath, metaPath string, size int64) Blob {
	b := Blob{ID: id, Kind: kind, MIME: mime, Size: size, Source: source,
		Created: s.now().UTC(), Path: blobPath}
	if prev, err := readMeta(metaPath); err == nil && !prev.Created.IsZero() {
		b.Created = prev.Created
	}
	return b
}

// Put stores data and returns its Blob. The id is the sha256 of data, so
// storing the same bytes twice is idempotent: the second call rewrites the
// sidecar (keeping the ORIGINAL creation time, so a re-send cannot extend
// retention indefinitely) and returns the same id.
//
// mime is recorded as given and used only to pick a file extension; nothing
// dispatches on it. Callers that need the mime to be trustworthy sniff it
// first — see SniffImage.
func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) {
	if !kind.Valid() {
		return Blob{}, ErrBadKind
	}
	if len(data) == 0 {
		return Blob{}, ErrEmpty
	}
	if cap := s.capFor(kind); int64(len(data)) > cap {
		return Blob{}, fmt.Errorf("%w: %d > %d", ErrTooLarge, len(data), cap)
	}
	sum := sha256.Sum256(data)
	id := hex.EncodeToString(sum[:])

	blobPath, metaPath, err := s.paths(kind, id, mime)
	if err != nil {
		return Blob{}, err
	}
	if err := os.MkdirAll(filepath.Dir(blobPath), dirPerm); err != nil {
		return Blob{}, fmt.Errorf("media: create bucket: %w", err)
	}

	b := s.newBlob(kind, mime, source, id, blobPath, metaPath, int64(len(data)))

	// A blob already on disk costs nothing more, so dedupe is checked before
	// the budget rather than after it.
	_, statErr := os.Stat(blobPath)
	fresh := statErr != nil
	if fresh {
		if err := s.reserve(b.Size); err != nil {
			return Blob{}, err
		}
	}

	// Every return past the reservation has to give it back, so the defer owns
	// that rather than each error path: a path that forgot over-counted the
	// store until the next Open re-walked the directory.
	stored := false
	defer func() {
		if fresh && !stored {
			s.release(b.Size)
		}
	}()

	// The sidecar goes first. Written second, a full disk or a crash between
	// the two left the bytes on disk with no sidecar, and List only sees
	// sidecars, so Prune could never collect them: Put returned an error and an
	// image nobody knew about became permanent.
	if err := writeMeta(metaPath, b); err != nil {
		return Blob{}, err
	}
	if err := writeFile(blobPath, data); err != nil {
		_ = os.Remove(metaPath)
		return Blob{}, err
	}
	stored = true
	return b, nil
}

// capFor is the per-blob cap for a kind. Audio has its own, larger one.
func (s *Store) capFor(kind Kind) int64 {
	if kind == KindAudio {
		return s.maxAudio
	}
	return s.maxBytes
}

// PutFile stores a file that is already on disk, by moving it into place rather
// than reading it into memory. It exists for meeting audio: a two-hour capture
// is a couple of hundred megabytes, and Put's []byte means that much heap in
// the process that owns the database, twice over while the WAV is built.
//
// src is consumed: on success it has been renamed into the store, and on a
// duplicate it is removed. On failure it is left where it is, so a caller that
// still needs the bytes can fall back to reading them.
func (s *Store) PutFile(kind Kind, mime, source, src string) (Blob, error) {
	if !kind.Valid() {
		return Blob{}, ErrBadKind
	}
	info, err := os.Stat(src)
	if err != nil {
		return Blob{}, fmt.Errorf("media: stat spool: %w", err)
	}
	if info.Size() == 0 {
		return Blob{}, ErrEmpty
	}
	if cap := s.capFor(kind); info.Size() > cap {
		return Blob{}, fmt.Errorf("%w: %d > %d", ErrTooLarge, info.Size(), cap)
	}
	id, err := hashFile(src)
	if err != nil {
		return Blob{}, err
	}
	blobPath, metaPath, err := s.paths(kind, id, mime)
	if err != nil {
		return Blob{}, err
	}
	if err := os.MkdirAll(filepath.Dir(blobPath), dirPerm); err != nil {
		return Blob{}, fmt.Errorf("media: create bucket: %w", err)
	}
	b := s.newBlob(kind, mime, source, id, blobPath, metaPath, info.Size())
	_, statErr := os.Stat(blobPath)
	fresh := statErr != nil
	if fresh {
		if err := s.reserve(b.Size); err != nil {
			return Blob{}, err
		}
	}
	// Same reasoning as Put: the reservation is released by one defer, not by
	// whichever error path remembered to.
	stored := false
	defer func() {
		if fresh && !stored {
			s.release(b.Size)
		}
	}()

	if err := writeMeta(metaPath, b); err != nil {
		return Blob{}, err
	}
	if !fresh {
		// Same bytes already here. Drop the spool copy.
		_ = os.Remove(src)
		stored = true
		return b, nil
	}
	if err := os.Chmod(src, filePerm); err != nil {
		return Blob{}, fmt.Errorf("media: chmod spool: %w", err)
	}
	if err := os.Rename(src, blobPath); err != nil {
		_ = os.Remove(metaPath)
		return Blob{}, fmt.Errorf("media: move spool: %w", err)
	}
	stored = true
	return b, nil
}

// SpoolFile creates an empty file under the store, outside the kind
// directories, for a caller that is writing a blob incrementally. Prune never
// looks at it and List never reports it; PutFile is what turns it into a blob.
// The caller owns removing it if it never gets that far.
func (s *Store) SpoolFile(prefix string) (*os.File, error) {
	dir := filepath.Join(s.dir, spoolName)
	if err := os.MkdirAll(dir, dirPerm); err != nil {
		return nil, fmt.Errorf("media: create spool: %w", err)
	}
	f, err := os.CreateTemp(dir, prefix+"-*")
	if err != nil {
		return nil, fmt.Errorf("media: spool: %w", err)
	}
	if err := f.Chmod(filePerm); err != nil {
		f.Close()
		return nil, fmt.Errorf("media: chmod spool: %w", err)
	}
	return f, nil
}

// hashFile streams the digest so the id costs one buffer rather than the whole
// file.
func hashFile(path string) (string, error) {
	f, err := os.Open(path)
	if err != nil {
		return "", fmt.Errorf("media: open spool: %w", err)
	}
	defer f.Close()
	h := sha256.New()
	if _, err := io.Copy(h, f); err != nil {
		return "", fmt.Errorf("media: hash spool: %w", err)
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

// Get returns the blob's metadata without reading its bytes.
func (s *Store) Get(id string) (Blob, error) {
	if !validID(id) {
		return Blob{}, ErrBadID
	}
	for _, kind := range allKinds {
		b, err := readMeta(filepath.Join(s.bucket(kind, id), id+metaExt))
		if err != nil {
			continue
		}
		p, err := s.locate(kind, id)
		if err != nil {
			continue
		}
		b.Path = p
		return b, nil
	}
	return Blob{}, ErrNotFound
}

// Read returns the blob's bytes together with its metadata. This is the only
// way out of the store, and it is a local read: nothing in this package can
// send bytes anywhere.
func (s *Store) Read(id string) (Blob, []byte, error) {
	b, err := s.Get(id)
	if err != nil {
		return Blob{}, nil, err
	}
	data, err := os.ReadFile(b.Path)
	if err != nil {
		return Blob{}, nil, fmt.Errorf("media: read %s: %w", shortID(id), err)
	}
	return b, data, nil
}

// List returns every blob of the given kind, newest first. An empty kind lists
// both. It walks the directory; at personal volumes (tens to hundreds of items
// inside the retention window) that is cheap, and it means the sidecars are the
// single source of truth with no index to fall out of sync.
func (s *Store) List(kind Kind) ([]Blob, error) {
	kinds := allKinds
	if kind != "" {
		if !kind.Valid() {
			return nil, ErrBadKind
		}
		kinds = []Kind{kind}
	}
	var out []Blob
	for _, k := range kinds {
		err := s.walkSidecars(k, func(path string, _ fs.DirEntry) error {
			b, err := readMeta(path)
			if err != nil {
				return nil // a corrupt sidecar is skipped, not fatal
			}
			if p, err := s.locate(b.Kind, b.ID); err == nil {
				b.Path = p
			}
			out = append(out, b)
			return nil
		})
		if err != nil {
			return nil, fmt.Errorf("media: list %s: %w", k, err)
		}
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Created.Equal(out[j].Created) {
			return out[i].ID < out[j].ID
		}
		return out[i].Created.After(out[j].Created)
	})
	return out, nil
}

// Delete removes a blob and its sidecar. Missing is not an error: the caller
// asked for it gone and it is gone.
func (s *Store) Delete(id string) error {
	if !validID(id) {
		return ErrBadID
	}
	for _, kind := range allKinds {
		bucket := s.bucket(kind, id)
		entries, err := os.ReadDir(bucket)
		if err != nil {
			continue
		}
		for _, e := range entries {
			if !strings.HasPrefix(e.Name(), id) {
				continue
			}
			// Only the bytes count against the budget, so the sidecar's own
			// size is never given back.
			var size int64
			if info, err := e.Info(); err == nil && !strings.HasSuffix(e.Name(), metaExt) {
				size = info.Size()
			}
			if err := os.Remove(filepath.Join(bucket, e.Name())); err != nil && !errors.Is(err, fs.ErrNotExist) {
				return fmt.Errorf("media: delete %s: %w", shortID(id), err)
			}
			if size > 0 {
				s.release(size)
			}
		}
	}
	return nil
}

// Prune deletes every blob older than the store's retention and reports how
// many went. It is the enforcement half of the retention promise; a caller that
// never runs it has a store that grows without bound, which is why the daemon
// runs it on the digestion tick rather than leaving it to a cron the operator
// might not add.
func (s *Store) Prune() (int, error) {
	blobs, err := s.List("")
	if err != nil {
		return 0, err
	}
	now := s.now()
	deleted := 0
	known := map[string]bool{}
	for _, b := range blobs {
		known[b.ID] = true
		if b.Age(now) <= s.retention {
			continue
		}
		if err := s.Delete(b.ID); err != nil {
			return deleted, err
		}
		delete(known, b.ID)
		deleted++
	}
	n, err := s.pruneOrphans(known, now)
	return deleted + n, err
}

// pruneOrphans collects blob files with no readable sidecar. List walks
// sidecars, so those files were invisible to retention and stayed on disk
// forever: audio of people accumulating is the exact failure this package
// exists to prevent, and a half-finished Put from an older build is enough to
// produce one. They are only collected once they are older than retention, so a
// Put racing a Prune does not lose its bytes.
func (s *Store) pruneOrphans(known map[string]bool, now time.Time) (int, error) {
	deleted := 0
	for _, kind := range allKinds {
		err := s.walkBlobFiles(kind, func(path string, d fs.DirEntry) error {
			id, _, _ := strings.Cut(d.Name(), ".")
			if known[id] {
				return nil
			}
			info, err := d.Info()
			if err != nil {
				return nil //nolint:nilerr // gone underneath us is the outcome we wanted
			}
			if now.Sub(info.ModTime()) <= s.retention {
				return nil
			}
			if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) {
				return err
			}
			s.release(info.Size())
			deleted++
			return nil
		})
		if err != nil {
			return deleted, fmt.Errorf("media: prune %s: %w", kind, err)
		}
	}
	return deleted, nil
}

// walkSidecars visits every sidecar of one kind, and walkBlobFiles every file
// that is not one. The sidecars are the store's index and the rest are the
// bytes, so a walker always wants exactly one of the two.
func (s *Store) walkSidecars(kind Kind, fn func(path string, d fs.DirEntry) error) error {
	return s.walkKind(kind, true, fn)
}

func (s *Store) walkBlobFiles(kind Kind, fn func(path string, d fs.DirEntry) error) error {
	return s.walkKind(kind, false, fn)
}

// walkKind walks one kind's directory tree. A kind that was never used has no
// directory, which is silence rather than an error.
func (s *Store) walkKind(kind Kind, sidecars bool, fn func(path string, d fs.DirEntry) error) error {
	root := filepath.Join(s.dir, string(kind))
	return filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			if errors.Is(err, fs.ErrNotExist) {
				return nil
			}
			return err
		}
		if d.IsDir() || strings.HasSuffix(path, metaExt) != sidecars {
			return nil
		}
		return fn(path, d)
	})
}

// paths returns the blob and sidecar paths for an id.
func (s *Store) paths(kind Kind, id, mime string) (blobPath, metaPath string, err error) {
	if !validID(id) {
		return "", "", ErrBadID
	}
	bucket := s.bucket(kind, id)
	return filepath.Join(bucket, id+extFor(mime, kind)), filepath.Join(bucket, id+metaExt), nil
}

// locate finds the stored bytes for an id whose extension we do not know,
// because the extension came from the mime at Put time.
func (s *Store) locate(kind Kind, id string) (string, error) {
	if !validID(id) {
		return "", ErrBadID
	}
	bucket := s.bucket(kind, id)
	entries, err := os.ReadDir(bucket)
	if err != nil {
		return "", ErrNotFound
	}
	for _, e := range entries {
		name := e.Name()
		if strings.HasPrefix(name, id) && !strings.HasSuffix(name, metaExt) {
			return filepath.Join(bucket, name), nil
		}
	}
	return "", ErrNotFound
}

// validID guards every path built from an id. Without it a caller-supplied id
// is a path traversal: Get("../../etc/passwd") would read outside the store.
func validID(id string) bool {
	if len(id) != idLen {
		return false
	}
	for i := 0; i < len(id); i++ {
		c := id[i]
		if (c < '0' || c > '9') && (c < 'a' || c > 'f') {
			return false
		}
	}
	return true
}

// extFor maps a mime to a file extension, defaulting per kind. The extension is
// cosmetic — the id is the key — but it is what makes the store browsable and
// lets a subprocess that sniffs by name (piper, some image tools) cope.
func extFor(mime string, kind Kind) string {
	switch strings.ToLower(strings.TrimSpace(mime)) {
	case "image/jpeg", "image/jpg":
		return ".jpg"
	case "image/png":
		return ".png"
	case "image/gif":
		return ".gif"
	case "image/webp":
		// Unreachable for images today: SniffImage refuses webp before
		// anything reaches Put, because this build has no webp decoder. Kept
		// so the mapping is right on the day one arrives.
		return ".webp"
	case "audio/wav", "audio/x-wav", "audio/wave":
		return ".wav"
	case "audio/l16", "audio/pcm":
		return ".pcm"
	}
	if kind == KindImage {
		return ".bin"
	}
	return ".pcm"
}

// writeFile writes data 0600 via a temp file in the same directory, so a
// crash mid-write cannot leave a truncated blob under a digest that claims
// to describe the whole thing.
func writeFile(path string, data []byte) error {
	tmp, err := os.CreateTemp(filepath.Dir(path), ".tmp-*")
	if err != nil {
		return fmt.Errorf("media: temp: %w", err)
	}
	defer os.Remove(tmp.Name())
	if err := tmp.Chmod(filePerm); err != nil {
		tmp.Close()
		return fmt.Errorf("media: chmod: %w", err)
	}
	if _, err := tmp.Write(data); err != nil {
		tmp.Close()
		return fmt.Errorf("media: write: %w", err)
	}
	if err := tmp.Close(); err != nil {
		return fmt.Errorf("media: close: %w", err)
	}
	if err := os.Rename(tmp.Name(), path); err != nil {
		return fmt.Errorf("media: rename: %w", err)
	}
	return nil
}

func writeMeta(path string, b Blob) error {
	data, err := json.Marshal(b)
	if err != nil {
		return fmt.Errorf("media: marshal meta: %w", err)
	}
	return writeFile(path, data)
}

func readMeta(path string) (Blob, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return Blob{}, err
	}
	var b Blob
	if err := json.Unmarshal(data, &b); err != nil {
		return Blob{}, err
	}
	if !validID(b.ID) || !b.Kind.Valid() {
		return Blob{}, errors.New("media: corrupt sidecar")
	}
	b.Created = b.Created.UTC()
	return b, nil
}