Implemented shared S3 audio caching for prepare and restore --include-audio
This commit is contained in:
223
internal/audio/s3_audio.go
Normal file
223
internal/audio/s3_audio.go
Normal file
@@ -0,0 +1,223 @@
|
||||
package audio
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||
)
|
||||
|
||||
// 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 := 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 := 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 := 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 := os.Chmod(tmpPath, 0o644); err != nil {
|
||||
return fmt.Errorf("set temp file permissions: %w", err)
|
||||
}
|
||||
if err := os.Rename(tmpPath, destPath); err != nil {
|
||||
return fmt.Errorf("install downloaded file: %w", err)
|
||||
}
|
||||
removeTmp = false
|
||||
return nil
|
||||
}
|
||||
|
||||
func copyFileAtomicWithChecksum(src, dst string, perm os.FileMode) (string, error) {
|
||||
if strings.TrimSpace(src) == "" || strings.TrimSpace(dst) == "" {
|
||||
return "", fmt.Errorf("source and destination paths are required")
|
||||
}
|
||||
in, err := os.Open(src)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer func() { _ = in.Close() }()
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil {
|
||||
return "", fmt.Errorf("create destination directory: %w", err)
|
||||
}
|
||||
base := filepath.Base(dst)
|
||||
tmp, err := os.CreateTemp(filepath.Dir(dst), "."+base+".tmp-*")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("create temp file: %w", err)
|
||||
}
|
||||
tmpPath := tmp.Name()
|
||||
removeTmp := true
|
||||
defer func() {
|
||||
if removeTmp {
|
||||
_ = os.Remove(tmpPath)
|
||||
}
|
||||
}()
|
||||
|
||||
digest := sha256.New()
|
||||
if _, err := io.Copy(io.MultiWriter(tmp, digest), in); err != nil {
|
||||
_ = tmp.Close()
|
||||
return "", fmt.Errorf("copy file: %w", err)
|
||||
}
|
||||
if err := tmp.Sync(); err != nil {
|
||||
_ = tmp.Close()
|
||||
return "", fmt.Errorf("sync temp file: %w", err)
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
return "", fmt.Errorf("close temp file: %w", err)
|
||||
}
|
||||
if err := os.Chmod(tmpPath, perm); err != nil {
|
||||
return "", fmt.Errorf("chmod temp file: %w", err)
|
||||
}
|
||||
if err := os.Rename(tmpPath, dst); err != nil {
|
||||
return "", fmt.Errorf("install temp file: %w", err)
|
||||
}
|
||||
removeTmp = false
|
||||
return hex.EncodeToString(digest.Sum(nil)), nil
|
||||
}
|
||||
Reference in New Issue
Block a user