Files
narratio/internal/audio/s3_audio.go

171 lines
4.9 KiB
Go

package audio
import (
"context"
"fmt"
"os"
"path/filepath"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/fileops"
)
// S3MaterializeRequest describes one S3-backed audio materialization.
type S3MaterializeRequest struct {
Store storage.ObjectStore
Object storage.ObjectInfo
Bucket string
CacheRoot string
CacheEnabled bool
SpoolPath string
DestPath string
}
// S3MaterializeResult captures local materialization provenance.
type S3MaterializeResult struct {
Checksum string
CachePath string
SpoolPath string
CacheHit bool
Downloaded bool
}
// MaterializeS3Audio installs one S3 audio object into the destination path,
// reusing and refreshing the durable local cache when enabled.
func MaterializeS3Audio(ctx context.Context, req S3MaterializeRequest) (S3MaterializeResult, error) {
if err := ctx.Err(); err != nil {
return S3MaterializeResult{}, err
}
if req.Store == nil {
return S3MaterializeResult{}, fmt.Errorf("object store is required")
}
key := strings.TrimSpace(req.Object.Key)
if key == "" {
return S3MaterializeResult{}, fmt.Errorf("s3 object key is required")
}
if strings.TrimSpace(req.DestPath) == "" {
return S3MaterializeResult{}, fmt.Errorf("destination path is required")
}
result := S3MaterializeResult{}
if req.CacheEnabled {
cachePath, err := artifacts.S3AudioCachePath(req.CacheRoot, req.Bucket, key)
if err != nil {
return S3MaterializeResult{}, fmt.Errorf("resolve audio cache path: %w", err)
}
result.CachePath = cachePath
if ok, err := validCachedAudio(cachePath, req.Object.Size); err != nil {
return S3MaterializeResult{}, err
} else if ok {
checksum, err := fileops.CopyFileAtomicWithChecksum(cachePath, req.DestPath, 0o644)
if err != nil {
return S3MaterializeResult{}, fmt.Errorf("materialize cached audio %q: %w", cachePath, err)
}
result.Checksum = checksum
result.CacheHit = true
return result, nil
}
}
spoolPath := strings.TrimSpace(req.SpoolPath)
if spoolPath == "" {
return S3MaterializeResult{}, fmt.Errorf("spool path is required for s3 audio download")
}
if err := downloadObjectAtomic(ctx, req.Store, key, spoolPath); err != nil {
return S3MaterializeResult{}, fmt.Errorf("download s3 audio object %q: %w", key, err)
}
if err := validateLocalAudio(spoolPath, req.Object.Size); err != nil {
return S3MaterializeResult{}, fmt.Errorf("validate downloaded audio %q: %w", spoolPath, err)
}
checksum, err := fileops.CopyFileAtomicWithChecksum(spoolPath, req.DestPath, 0o644)
if err != nil {
return S3MaterializeResult{}, fmt.Errorf("materialize downloaded audio %q: %w", filepath.Base(req.DestPath), err)
}
result.Checksum = checksum
result.SpoolPath = spoolPath
result.Downloaded = true
if result.CachePath != "" {
if _, err := fileops.CopyFileAtomicWithChecksum(spoolPath, result.CachePath, 0o644); err != nil {
return S3MaterializeResult{}, fmt.Errorf("populate audio cache %q: %w", result.CachePath, err)
}
}
return result, nil
}
func validCachedAudio(path string, expectedSize int64) (bool, error) {
info, err := os.Stat(path)
if err != nil {
if os.IsNotExist(err) {
return false, nil
}
return false, fmt.Errorf("stat cached audio %q: %w", path, err)
}
if info.IsDir() {
return false, fmt.Errorf("cached audio path is a directory: %q", path)
}
if info.Size() <= 0 {
return false, nil
}
if expectedSize > 0 && info.Size() != expectedSize {
return false, nil
}
return true, nil
}
func validateLocalAudio(path string, expectedSize int64) error {
info, err := os.Stat(path)
if err != nil {
return fmt.Errorf("stat: %w", err)
}
if info.IsDir() {
return fmt.Errorf("path is a directory")
}
if info.Size() <= 0 {
return fmt.Errorf("file is empty")
}
if expectedSize > 0 && info.Size() != expectedSize {
return fmt.Errorf("size %d does not match remote size %d", info.Size(), expectedSize)
}
return nil
}
func downloadObjectAtomic(ctx context.Context, store storage.ObjectStore, key, destPath string) error {
if strings.TrimSpace(destPath) == "" {
return fmt.Errorf("destination path is required")
}
dir := filepath.Dir(destPath)
if err := os.MkdirAll(dir, 0o755); err != nil {
return fmt.Errorf("create destination directory: %w", err)
}
base := filepath.Base(destPath)
tmp, err := os.CreateTemp(dir, "."+base+".download-*.tmp")
if err != nil {
return fmt.Errorf("create temp file: %w", err)
}
tmpPath := tmp.Name()
if err := tmp.Close(); err != nil {
_ = os.Remove(tmpPath)
return fmt.Errorf("close temp file: %w", err)
}
removeTmp := true
defer func() {
if removeTmp {
_ = os.Remove(tmpPath)
}
}()
if err := store.Download(ctx, key, tmpPath); err != nil {
return err
}
if err := fileops.InstallDownloadedTempFile(tmpPath, destPath, 0o644); err != nil {
return err
}
removeTmp = false
return nil
}