diff --git a/internal/audio/pcmwav.go b/internal/audio/pcmwav.go index 4889fa3..b9da0f8 100644 --- a/internal/audio/pcmwav.go +++ b/internal/audio/pcmwav.go @@ -94,14 +94,33 @@ func PCMFromWAV(wav []byte) (Format, []byte, error) { // header so the result can be written to disk and played with `aplay`. // Used by the reference client to write the TTS reply; not on the wire. func WAVFromPCM(format Format, pcm []byte) ([]byte, error) { + hdr, err := WAVHeader(format, len(pcm)) + if err != nil { + return nil, err + } + out := make([]byte, wavHeaderSize+len(pcm)) + copy(out, hdr) + copy(out[wavHeaderSize:], pcm) + return out, nil +} + +// WAVHeaderSize is the fixed size of the header WAVHeader writes. A caller +// spooling audio to a file reserves this many bytes up front and rewrites them +// once it knows the length. +const WAVHeaderSize = wavHeaderSize + +// WAVHeader builds just the 44-byte canonical header for n bytes of PCM. It +// exists so a long recording can be written straight to a file: holding the +// whole meeting in memory to prepend 44 bytes is what the streaming path is +// avoiding. +func WAVHeader(format Format, n int) ([]byte, error) { if !format.IsValid() { return nil, fmt.Errorf("audio: WAVFromPCM: %w: %+v", ErrNotCanonicalPCM, format) } - out := make([]byte, wavHeaderSize+len(pcm)) - copy(out[wavHeaderSize:], pcm) + out := make([]byte, wavHeaderSize) // RIFF header copy(out[0:4], []byte("RIFF")) - binary.LittleEndian.PutUint32(out[4:8], uint32(36+len(pcm))) + binary.LittleEndian.PutUint32(out[4:8], uint32(36+n)) copy(out[8:12], []byte("WAVE")) // fmt chunk copy(out[12:16], []byte("fmt ")) @@ -116,6 +135,6 @@ func WAVFromPCM(format Format, pcm []byte) ([]byte, error) { binary.LittleEndian.PutUint16(out[34:36], uint16(format.SampleBits)) // data chunk copy(out[36:40], []byte("data")) - binary.LittleEndian.PutUint32(out[40:44], uint32(len(pcm))) + binary.LittleEndian.PutUint32(out[40:44], uint32(n)) return out, nil } diff --git a/internal/media/store.go b/internal/media/store.go index f3d4ef5..32d3fc4 100644 --- a/internal/media/store.go +++ b/internal/media/store.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "io/fs" "os" "path/filepath" @@ -20,6 +21,14 @@ import ( // 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. @@ -45,6 +54,7 @@ var ErrStoreFull = errors.New("media: store is full") type Store struct { dir string maxBytes int64 + maxAudio int64 maxTotal int64 retention time.Duration now func() time.Time @@ -81,16 +91,24 @@ func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duratio 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, maxTotal: maxTotal, retention: retention, now: time.Now} + s := &Store{dir: abs, maxBytes: maxBytes, maxAudio: maxAudio, maxTotal: maxTotal, + retention: retention, now: time.Now} s.total = s.measure() return s, nil } @@ -99,7 +117,13 @@ func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duratio // over at zero. func (s *Store) measure() int64 { var total int64 + spool := filepath.Join(s.dir, "spool") _ = 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, ".json") { return nil //nolint:nilerr // an unreadable corner is not worth refusing to boot over } @@ -144,8 +168,8 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) { if len(data) == 0 { return Blob{}, ErrEmpty } - if int64(len(data)) > s.maxBytes { - return Blob{}, fmt.Errorf("%w: %d > %d", ErrTooLarge, len(data), s.maxBytes) + 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[:]) @@ -202,6 +226,121 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) { 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), 0o700); err != nil { + return Blob{}, fmt.Errorf("media: create bucket: %w", err) + } + b := Blob{ID: id, Kind: kind, MIME: mime, Size: info.Size(), Source: source, + Created: s.now().UTC(), Path: blobPath} + if prev, err := readMeta(metaPath); err == nil && !prev.Created.IsZero() { + b.Created = prev.Created + } + _, already := os.Stat(blobPath) + if already != nil { + s.totalMu.Lock() + room := s.total+b.Size <= s.maxTotal + if room { + s.total += b.Size + } + s.totalMu.Unlock() + if !room { + return Blob{}, fmt.Errorf("%w: %d stored, %d budget, %d more asked for", + ErrStoreFull, s.Total(), s.maxTotal, b.Size) + } + } + if err := writeMeta(metaPath, b); err != nil { + return Blob{}, err + } + if already == nil { + // Same bytes already here. Drop the spool copy. + _ = os.Remove(src) + return b, nil + } + if err := os.Chmod(src, 0o600); err != nil { + return Blob{}, fmt.Errorf("media: chmod spool: %w", err) + } + if err := os.Rename(src, blobPath); err != nil { + _ = os.Remove(metaPath) + s.totalMu.Lock() + s.total -= b.Size + s.totalMu.Unlock() + return Blob{}, fmt.Errorf("media: move spool: %w", err) + } + 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, "spool") + if err := os.MkdirAll(dir, 0o700); 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(0o600); 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) {