376 lines
12 KiB
Go
376 lines
12 KiB
Go
package app
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"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/audio"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/fileops"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
)
|
|
|
|
// RestoreExecutionResult captures concrete file-install results for one restore execution.
|
|
type RestoreExecutionResult struct {
|
|
DownloadedCount int
|
|
}
|
|
|
|
func executeRestorePlan(
|
|
ctx context.Context,
|
|
cfg *config.Config,
|
|
current *RemoteCurrentState,
|
|
plan *RestorePlan,
|
|
report *RestoreReport,
|
|
store storage.ObjectStore,
|
|
) (*RestoreExecutionResult, error) {
|
|
if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
|
|
return nil, fmt.Errorf("resolved config with pipeline/session is required")
|
|
}
|
|
if current == nil {
|
|
return nil, fmt.Errorf("remote current state is required")
|
|
}
|
|
if plan == nil {
|
|
return nil, fmt.Errorf("restore plan is required")
|
|
}
|
|
if store == nil {
|
|
return nil, fmt.Errorf("remote object store is required")
|
|
}
|
|
|
|
sessionRoot := artifacts.SessionWorkDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
|
|
if err := fileops.EnsureWorkspaceDirectory(sessionRoot); err != nil {
|
|
return nil, fmt.Errorf("create session root for restore: %w", err)
|
|
}
|
|
root, err := os.OpenRoot(sessionRoot)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open session root for restore: %w", err)
|
|
}
|
|
defer func() { _ = root.Close() }()
|
|
manifestActions := make([]RestoreAction, 0, 1)
|
|
actions := make([]RestoreAction, 0, len(plan.Actions))
|
|
for _, action := range plan.Actions {
|
|
if action.Kind != RestoreActionDownload {
|
|
continue
|
|
}
|
|
if action.LocalRelativePath == config.PathManifestFile {
|
|
manifestActions = append(manifestActions, action)
|
|
continue
|
|
}
|
|
actions = append(actions, action)
|
|
}
|
|
if len(manifestActions) > 1 {
|
|
return nil, fmt.Errorf("restore plan includes multiple manifest download actions")
|
|
}
|
|
if len(manifestActions) == 1 {
|
|
actions = append(actions, manifestActions[0])
|
|
}
|
|
|
|
result := &RestoreExecutionResult{}
|
|
for _, action := range actions {
|
|
if err := executeRestoreDownloadAction(ctx, cfg, sessionRoot, root, current, action, store); err != nil {
|
|
if report != nil {
|
|
report.markFailed(action, err)
|
|
}
|
|
return nil, fmt.Errorf("install %q from %q: %w", action.LocalRelativePath, action.RemoteKey, err)
|
|
}
|
|
if report != nil {
|
|
report.markDownloaded(action)
|
|
}
|
|
result.DownloadedCount++
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
func executeRestoreDownloadAction(
|
|
ctx context.Context,
|
|
cfg *config.Config,
|
|
sessionRoot string,
|
|
root *os.Root,
|
|
current *RemoteCurrentState,
|
|
action RestoreAction,
|
|
store storage.ObjectStore,
|
|
) error {
|
|
safeLocalPath, err := joinWithinSessionRoot(sessionRoot, action.LocalRelativePath)
|
|
if err != nil {
|
|
return fmt.Errorf("resolve safe local path: %w", err)
|
|
}
|
|
if strings.TrimSpace(action.LocalPath) != "" && filepath.Clean(action.LocalPath) != safeLocalPath {
|
|
return fmt.Errorf("restore plan local path mismatch for %q", action.LocalRelativePath)
|
|
}
|
|
parent, leaf, err := openRestoreDestinationParent(root, action.LocalRelativePath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = parent.Close() }()
|
|
if err := validateRestoreDestinationLeaf(parent, leaf); err != nil {
|
|
return err
|
|
}
|
|
|
|
if restoreActionIsAudio(action) {
|
|
return executeRestoreAudioAction(ctx, cfg, safeLocalPath, parent, leaf, action, store)
|
|
}
|
|
|
|
temporaryPath, err := downloadRestoreTemp(safeLocalPath, func(destination io.Writer) error {
|
|
if len(action.VerifiedContent) > 0 {
|
|
_, err := io.Copy(destination, bytes.NewReader(action.VerifiedContent))
|
|
return err
|
|
}
|
|
return storage.DownloadTo(ctx, store, action.RemoteKey, destination)
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("download to temp file: %w", err)
|
|
}
|
|
defer func() { _ = os.Remove(temporaryPath) }()
|
|
if err := verifyRestoredObject(ctx, store, action, temporaryPath); err != nil {
|
|
return err
|
|
}
|
|
|
|
if isRestoreManifest(action.LocalRelativePath) {
|
|
file, err := os.Open(temporaryPath)
|
|
if err != nil {
|
|
return fmt.Errorf("open restored manifest: %w", err)
|
|
}
|
|
destinationRoot := sessionRoot
|
|
requireCurrentIdentity := action.LocalRelativePath == config.PathManifestFile
|
|
if !requireCurrentIdentity {
|
|
destinationRoot = filepath.Dir(safeLocalPath)
|
|
}
|
|
restored, prepareErr := prepareRestoredManifest(ctx, cfg, current, file, destinationRoot, requireCurrentIdentity)
|
|
closeErr := file.Close()
|
|
if prepareErr != nil {
|
|
return prepareErr
|
|
}
|
|
if closeErr != nil {
|
|
return fmt.Errorf("close restored manifest: %w", closeErr)
|
|
}
|
|
if err := validateRestoreDestinationLeaf(parent, leaf); err != nil {
|
|
return err
|
|
}
|
|
if err := (&manifest.LocalStore{}).Save(ctx, safeLocalPath, restored); err != nil {
|
|
return fmt.Errorf("install rebased manifest atomically: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
if err := validateRestoreDestinationLeaf(parent, leaf); err != nil {
|
|
return err
|
|
}
|
|
if err := fileops.InstallSiblingTempFile(temporaryPath, safeLocalPath, fileops.WorkspaceFileMode); err != nil {
|
|
return fmt.Errorf("install file atomically: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func openRestoreDestinationParent(root *os.Root, relative string) (*os.Root, string, error) {
|
|
if root == nil {
|
|
return nil, "", fmt.Errorf("restore session root is required")
|
|
}
|
|
name := filepath.Clean(filepath.FromSlash(strings.TrimSpace(relative)))
|
|
if name == "." || name == ".." || filepath.IsAbs(name) || strings.HasPrefix(name, ".."+string(filepath.Separator)) {
|
|
return nil, "", fmt.Errorf("restore path %q is not a normalized relative path", relative)
|
|
}
|
|
parts := strings.Split(name, string(filepath.Separator))
|
|
current := root
|
|
for _, part := range parts[:len(parts)-1] {
|
|
info, err := current.Lstat(part)
|
|
if errors.Is(err, os.ErrNotExist) {
|
|
if err := current.Mkdir(part, fileops.WorkspaceDirectoryMode.Perm()); err != nil && !errors.Is(err, os.ErrExist) {
|
|
if current != root {
|
|
_ = current.Close()
|
|
}
|
|
return nil, "", fmt.Errorf("create restore destination directory %q: %w", part, err)
|
|
}
|
|
info, err = current.Lstat(part)
|
|
}
|
|
if err != nil {
|
|
if current != root {
|
|
_ = current.Close()
|
|
}
|
|
return nil, "", fmt.Errorf("inspect restore destination ancestor %q: %w", part, err)
|
|
}
|
|
if !info.IsDir() || info.Mode()&os.ModeSymlink != 0 {
|
|
if current != root {
|
|
_ = current.Close()
|
|
}
|
|
return nil, "", fmt.Errorf("restore destination ancestor %q is not a directory", part)
|
|
}
|
|
if err := current.Chmod(part, fileops.WorkspaceDirectoryMode); err != nil {
|
|
if current != root {
|
|
_ = current.Close()
|
|
}
|
|
return nil, "", fmt.Errorf("set restore destination directory permissions %q: %w", part, err)
|
|
}
|
|
next, err := current.OpenRoot(part)
|
|
if current != root {
|
|
_ = current.Close()
|
|
}
|
|
if err != nil {
|
|
return nil, "", fmt.Errorf("open restore destination ancestor %q: %w", part, err)
|
|
}
|
|
current = next
|
|
}
|
|
return current, parts[len(parts)-1], nil
|
|
}
|
|
|
|
func validateRestoreDestinationLeaf(parent *os.Root, leaf string) error {
|
|
info, err := parent.Lstat(leaf)
|
|
if errors.Is(err, os.ErrNotExist) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("inspect restore destination %q: %w", leaf, err)
|
|
}
|
|
if !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 {
|
|
return fmt.Errorf("restore destination %q is not an eligible regular file", leaf)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func isRestoreManifest(relativePath string) bool {
|
|
clean := filepath.ToSlash(filepath.Clean(strings.TrimSpace(relativePath)))
|
|
return clean == config.PathManifestFile || clean == config.PathPreviousDirSegment+"/"+config.PathManifestFile
|
|
}
|
|
|
|
func downloadRestoreTemp(destination string, download func(io.Writer) error) (string, error) {
|
|
if strings.TrimSpace(destination) == "" {
|
|
return "", fmt.Errorf("restore destination path is required")
|
|
}
|
|
if download == nil {
|
|
return "", fmt.Errorf("restore temporary writer is required")
|
|
}
|
|
directory := filepath.Dir(destination)
|
|
if err := fileops.EnsureWorkspaceDirectory(directory); err != nil {
|
|
return "", fmt.Errorf("create destination directory: %w", err)
|
|
}
|
|
temporary, err := os.CreateTemp(directory, "."+filepath.Base(destination)+".tmp-*")
|
|
if err != nil {
|
|
return "", fmt.Errorf("create sibling temporary file: %w", err)
|
|
}
|
|
temporaryPath := temporary.Name()
|
|
if err := download(temporary); err != nil {
|
|
_ = temporary.Close()
|
|
_ = os.Remove(temporaryPath)
|
|
return "", err
|
|
}
|
|
if err := temporary.Close(); err != nil {
|
|
_ = os.Remove(temporaryPath)
|
|
return "", fmt.Errorf("close sibling temporary file: %w", err)
|
|
}
|
|
return temporaryPath, nil
|
|
}
|
|
|
|
func verifyRestoredObject(ctx context.Context, store storage.ObjectStore, action RestoreAction, temporaryPath string) error {
|
|
if strings.TrimSpace(action.SHA256) == "" && strings.TrimSpace(action.Generation) == "" {
|
|
return nil
|
|
}
|
|
if strings.TrimSpace(action.SHA256) == "" || strings.TrimSpace(action.Generation) == "" {
|
|
return fmt.Errorf("committed object identity for %q is incomplete", action.RemoteKey)
|
|
}
|
|
file, err := os.Open(temporaryPath)
|
|
if err != nil {
|
|
return fmt.Errorf("open downloaded object for verification: %w", err)
|
|
}
|
|
digest := sha256.New()
|
|
count, copyErr := io.Copy(digest, file)
|
|
closeErr := file.Close()
|
|
if copyErr != nil {
|
|
return fmt.Errorf("checksum downloaded object: %w", copyErr)
|
|
}
|
|
if closeErr != nil {
|
|
return fmt.Errorf("close downloaded object: %w", closeErr)
|
|
}
|
|
if count != action.Size {
|
|
return fmt.Errorf("committed object size mismatch for %q: got %d, want %d", action.RemoteKey, count, action.Size)
|
|
}
|
|
if got := hex.EncodeToString(digest.Sum(nil)); got != action.SHA256 {
|
|
return fmt.Errorf("committed object checksum mismatch for %q: got %s, want %s", action.RemoteKey, got, action.SHA256)
|
|
}
|
|
objects, err := store.List(ctx, action.RemoteKey)
|
|
if err != nil {
|
|
return fmt.Errorf("read committed object identity for %q: %w", action.RemoteKey, err)
|
|
}
|
|
var found *storage.ObjectInfo
|
|
for _, object := range objects {
|
|
if normalizeRemoteKey(object.Key) != normalizeRemoteKey(action.RemoteKey) {
|
|
continue
|
|
}
|
|
if found != nil {
|
|
return fmt.Errorf("committed object %q is ambiguous", action.RemoteKey)
|
|
}
|
|
copy := object
|
|
found = ©
|
|
}
|
|
if found == nil {
|
|
return fmt.Errorf("committed object %q is missing", action.RemoteKey)
|
|
}
|
|
if found.Size != action.Size || strings.TrimSpace(found.ETag) != action.Generation {
|
|
return fmt.Errorf("committed object generation mismatch for %q", action.RemoteKey)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func executeRestoreAudioAction(
|
|
ctx context.Context,
|
|
cfg *config.Config,
|
|
safeLocalPath string,
|
|
parent *os.Root,
|
|
leaf string,
|
|
action RestoreAction,
|
|
store storage.ObjectStore,
|
|
) error {
|
|
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Storage.S3 == nil || cfg.Session == nil {
|
|
return fmt.Errorf("resolved s3 config and session are required")
|
|
}
|
|
spoolDir := artifacts.SessionSpoolRestoreAudioDir(cfg.Pipeline.Spool.Root, cfg.Session.Campaign, cfg.Session.SessionID)
|
|
spoolPath := filepath.Join(spoolDir, filepath.Base(safeLocalPath))
|
|
temporary, err := os.CreateTemp(filepath.Dir(safeLocalPath), "."+filepath.Base(safeLocalPath)+".restore-*")
|
|
if err != nil {
|
|
return fmt.Errorf("create sibling audio staging file: %w", err)
|
|
}
|
|
temporaryPath := temporary.Name()
|
|
if err := temporary.Close(); err != nil {
|
|
_ = os.Remove(temporaryPath)
|
|
return fmt.Errorf("close sibling audio staging file: %w", err)
|
|
}
|
|
defer func() { _ = os.Remove(temporaryPath) }()
|
|
|
|
cacheEnabled := cfg.Pipeline.Cache.S3Audio == nil || *cfg.Pipeline.Cache.S3Audio
|
|
_, err = audio.MaterializeS3Audio(ctx, audio.S3MaterializeRequest{
|
|
Store: store,
|
|
Object: storage.ObjectInfo{
|
|
Key: action.RemoteKey,
|
|
Size: action.Size,
|
|
ETag: action.ETag,
|
|
},
|
|
Bucket: strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket),
|
|
CacheRoot: strings.TrimSpace(cfg.Pipeline.Cache.Root),
|
|
CacheEnabled: cacheEnabled,
|
|
SpoolPath: spoolPath,
|
|
DestPath: temporaryPath,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("materialize audio: %w", err)
|
|
}
|
|
if err := verifyRestoredObject(ctx, store, action, temporaryPath); err != nil {
|
|
return err
|
|
}
|
|
if err := validateRestoreDestinationLeaf(parent, leaf); err != nil {
|
|
return err
|
|
}
|
|
if err := fileops.InstallSiblingTempFile(temporaryPath, safeLocalPath, fileops.WorkspaceFileMode); err != nil {
|
|
return fmt.Errorf("install verified audio atomically: %w", err)
|
|
}
|
|
return nil
|
|
}
|