140 lines
4.5 KiB
Go
140 lines
4.5 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
)
|
|
|
|
// RemoteCurrentState captures discovered committed remote archive state for one session.
|
|
type RemoteCurrentState struct {
|
|
Bucket string
|
|
SessionPrefix string
|
|
CurrentRunIDKey string
|
|
CurrentManifestKey string
|
|
RunID string
|
|
SessionID string
|
|
Campaign string
|
|
Manifest *manifest.Manifest
|
|
}
|
|
|
|
func discoverRemoteCurrentState(ctx context.Context, cfg *config.Config, store storage.ObjectStore) (*RemoteCurrentState, error) {
|
|
if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil {
|
|
return nil, fmt.Errorf("resolved config with pipeline/session is required")
|
|
}
|
|
if store == nil {
|
|
return nil, fmt.Errorf("remote object store is required")
|
|
}
|
|
|
|
bucket := artifacts.ResolveArchiveBucket(cfg, nil)
|
|
if strings.TrimSpace(bucket) == "" {
|
|
return nil, fmt.Errorf("archive bucket is required")
|
|
}
|
|
sessionPrefix, err := artifacts.ResolveArchiveSessionPrefix(cfg, nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("resolve archive session prefix: %w", err)
|
|
}
|
|
currentManifestKey, currentRunIDKey := artifacts.ResolveArchiveCurrentStateKeys(sessionPrefix)
|
|
|
|
exists, err := store.Exists(ctx, currentRunIDKey)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("check remote current run pointer %q: %w", currentRunIDKey, err)
|
|
}
|
|
if !exists {
|
|
return nil, fmt.Errorf("remote current run pointer missing: %q", currentRunIDKey)
|
|
}
|
|
|
|
runIDPath, err := downloadObjectToTemp(ctx, store, currentRunIDKey, "narratio-restore-current-run-id-*.txt")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("download remote current run pointer %q: %w", currentRunIDKey, err)
|
|
}
|
|
defer func() { _ = os.Remove(runIDPath) }()
|
|
|
|
runIDData, err := os.ReadFile(runIDPath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("read downloaded run pointer %q: %w", currentRunIDKey, err)
|
|
}
|
|
runID := strings.TrimSpace(string(runIDData))
|
|
if runID == "" {
|
|
return nil, fmt.Errorf("remote current run pointer %q is empty", currentRunIDKey)
|
|
}
|
|
|
|
exists, err = store.Exists(ctx, currentManifestKey)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("check remote current manifest %q: %w", currentManifestKey, err)
|
|
}
|
|
if !exists {
|
|
return nil, fmt.Errorf("remote current manifest missing: %q", currentManifestKey)
|
|
}
|
|
|
|
manifestPath, err := downloadObjectToTemp(ctx, store, currentManifestKey, "narratio-restore-current-manifest-*.json")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("download remote current manifest %q: %w", currentManifestKey, err)
|
|
}
|
|
defer func() { _ = os.Remove(manifestPath) }()
|
|
|
|
manifestStore := &manifest.LocalStore{}
|
|
remoteManifest, err := manifestStore.Load(ctx, manifestPath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("remote current manifest decode failed: %w", err)
|
|
}
|
|
|
|
requestedSession := strings.TrimSpace(cfg.Session.SessionID)
|
|
requestedCampaign := strings.TrimSpace(cfg.Session.Campaign)
|
|
manifestSession := strings.TrimSpace(remoteManifest.SessionID)
|
|
manifestCampaign := strings.TrimSpace(remoteManifest.Campaign)
|
|
|
|
if manifestSession != requestedSession {
|
|
return nil, fmt.Errorf(
|
|
"remote current manifest session_id %q does not match requested session_id %q",
|
|
manifestSession,
|
|
requestedSession,
|
|
)
|
|
}
|
|
if manifestCampaign == "" {
|
|
return nil, fmt.Errorf("remote current manifest campaign is required")
|
|
}
|
|
if manifestCampaign != requestedCampaign {
|
|
return nil, fmt.Errorf(
|
|
"remote current manifest campaign %q does not match requested campaign %q",
|
|
manifestCampaign,
|
|
requestedCampaign,
|
|
)
|
|
}
|
|
|
|
return &RemoteCurrentState{
|
|
Bucket: bucket,
|
|
SessionPrefix: sessionPrefix,
|
|
CurrentRunIDKey: currentRunIDKey,
|
|
CurrentManifestKey: currentManifestKey,
|
|
RunID: runID,
|
|
SessionID: manifestSession,
|
|
Campaign: manifestCampaign,
|
|
Manifest: remoteManifest,
|
|
}, nil
|
|
}
|
|
|
|
func downloadObjectToTemp(ctx context.Context, store storage.ObjectStore, key, pattern string) (string, error) {
|
|
tmp, err := os.CreateTemp("", pattern)
|
|
if err != nil {
|
|
return "", fmt.Errorf("create temp file: %w", err)
|
|
}
|
|
path := tmp.Name()
|
|
if err := tmp.Close(); err != nil {
|
|
_ = os.Remove(path)
|
|
return "", fmt.Errorf("close temp file: %w", err)
|
|
}
|
|
|
|
if err := store.Download(ctx, key, path); err != nil {
|
|
_ = os.Remove(path)
|
|
return "", err
|
|
}
|
|
return path, nil
|
|
}
|