package cli import ( "context" "encoding/json" "fmt" "io" "os" "path/filepath" "sort" "strings" "text/tabwriter" "time" "git.eeqj.de/sneak/vaultik/internal/config" "git.eeqj.de/sneak/vaultik/internal/database" "git.eeqj.de/sneak/vaultik/internal/globals" "git.eeqj.de/sneak/vaultik/internal/log" "git.eeqj.de/sneak/vaultik/internal/s3" "git.eeqj.de/sneak/vaultik/internal/snapshot" "github.com/dustin/go-humanize" "github.com/spf13/cobra" "go.uber.org/fx" ) // SnapshotCreateOptions contains options for the snapshot create command type SnapshotCreateOptions struct { Daemon bool Cron bool Prune bool } // SnapshotCreateApp contains all dependencies needed for creating snapshots type SnapshotCreateApp struct { Globals *globals.Globals Config *config.Config Repositories *database.Repositories ScannerFactory snapshot.ScannerFactory SnapshotManager *snapshot.SnapshotManager S3Client *s3.Client DB *database.DB Lifecycle fx.Lifecycle Shutdowner fx.Shutdowner Stdout io.Writer Stderr io.Writer Stdin io.Reader } // SnapshotApp contains dependencies for snapshot commands type SnapshotApp struct { *SnapshotCreateApp // Reuse snapshot creation functionality S3Client *s3.Client } // SnapshotInfo represents snapshot information for listing type SnapshotInfo struct { ID string `json:"id"` Timestamp time.Time `json:"timestamp"` CompressedSize int64 `json:"compressed_size"` } // NewSnapshotCommand creates the snapshot command and subcommands func NewSnapshotCommand() *cobra.Command { cmd := &cobra.Command{ Use: "snapshot", Short: "Snapshot management commands", Long: "Commands for creating, listing, and managing snapshots", } // Add subcommands cmd.AddCommand(newSnapshotCreateCommand()) cmd.AddCommand(newSnapshotListCommand()) cmd.AddCommand(newSnapshotPurgeCommand()) cmd.AddCommand(newSnapshotVerifyCommand()) return cmd } // newSnapshotCreateCommand creates the 'snapshot create' subcommand func newSnapshotCreateCommand() *cobra.Command { opts := &SnapshotCreateOptions{} cmd := &cobra.Command{ Use: "create", Short: "Create a new snapshot", Long: `Creates a new snapshot of the configured directories. Config is located at /etc/vaultik/config.yml by default, but can be overridden by specifying a path using --config or by setting VAULTIK_CONFIG to a path.`, Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, args []string) error { // Use unified config resolution configPath, err := ResolveConfigPath() if err != nil { return err } // Use the backup functionality from cli package rootFlags := GetRootFlags() return RunWithApp(cmd.Context(), AppOptions{ ConfigPath: configPath, LogOptions: log.LogOptions{ Verbose: rootFlags.Verbose, Debug: rootFlags.Debug, Cron: opts.Cron, }, Modules: []fx.Option{ snapshot.Module, s3.Module, fx.Provide(fx.Annotate( func(g *globals.Globals, cfg *config.Config, repos *database.Repositories, scannerFactory snapshot.ScannerFactory, snapshotManager *snapshot.SnapshotManager, s3Client *s3.Client, db *database.DB, lc fx.Lifecycle, shutdowner fx.Shutdowner) *SnapshotCreateApp { return &SnapshotCreateApp{ Globals: g, Config: cfg, Repositories: repos, ScannerFactory: scannerFactory, SnapshotManager: snapshotManager, S3Client: s3Client, DB: db, Lifecycle: lc, Shutdowner: shutdowner, Stdout: os.Stdout, Stderr: os.Stderr, Stdin: os.Stdin, } }, )), }, Invokes: []fx.Option{ fx.Invoke(func(app *SnapshotCreateApp, lc fx.Lifecycle) { // Create a cancellable context for the snapshot snapshotCtx, snapshotCancel := context.WithCancel(context.Background()) lc.Append(fx.Hook{ OnStart: func(ctx context.Context) error { // Start the snapshot creation in a goroutine go func() { // Run the snapshot creation if err := app.runSnapshot(snapshotCtx, opts); err != nil { if err != context.Canceled { log.Error("Snapshot creation failed", "error", err) } } // Shutdown the app when snapshot completes if err := app.Shutdowner.Shutdown(); err != nil { log.Error("Failed to shutdown", "error", err) } }() return nil }, OnStop: func(ctx context.Context) error { log.Debug("Stopping snapshot creation") // Cancel the snapshot context snapshotCancel() return nil }, }) }), }, }) }, } cmd.Flags().BoolVar(&opts.Daemon, "daemon", false, "Run in daemon mode with inotify monitoring") cmd.Flags().BoolVar(&opts.Cron, "cron", false, "Run in cron mode (silent unless error)") cmd.Flags().BoolVar(&opts.Prune, "prune", false, "Delete all previous snapshots and unreferenced blobs after backup") return cmd } // runSnapshot executes the snapshot creation operation func (app *SnapshotCreateApp) runSnapshot(ctx context.Context, opts *SnapshotCreateOptions) error { snapshotStartTime := time.Now() log.Info("Starting snapshot creation", "version", app.Globals.Version, "commit", app.Globals.Commit, "index_path", app.Config.IndexPath, ) // Clean up incomplete snapshots FIRST, before any scanning // This is critical for data safety - see CleanupIncompleteSnapshots for details hostname := app.Config.Hostname if hostname == "" { hostname, _ = os.Hostname() } // CRITICAL: This MUST succeed. If we fail to clean up incomplete snapshots, // the deduplication logic will think files from the incomplete snapshot were // already backed up and skip them, resulting in data loss. if err := app.SnapshotManager.CleanupIncompleteSnapshots(ctx, hostname); err != nil { return fmt.Errorf("cleanup incomplete snapshots: %w", err) } if opts.Daemon { log.Info("Running in daemon mode") // TODO: Implement daemon mode with inotify return fmt.Errorf("daemon mode not yet implemented") } // Resolve source directories to absolute paths resolvedDirs := make([]string, 0, len(app.Config.SourceDirs)) for _, dir := range app.Config.SourceDirs { absPath, err := filepath.Abs(dir) if err != nil { return fmt.Errorf("failed to resolve absolute path for %s: %w", dir, err) } // Resolve symlinks resolvedPath, err := filepath.EvalSymlinks(absPath) if err != nil { // If the path doesn't exist yet, use the absolute path if os.IsNotExist(err) { resolvedPath = absPath } else { return fmt.Errorf("failed to resolve symlinks for %s: %w", absPath, err) } } resolvedDirs = append(resolvedDirs, resolvedPath) } // Create scanner with progress enabled (unless in cron mode) scanner := app.ScannerFactory(snapshot.ScannerParams{ EnableProgress: !opts.Cron, }) // Perform a single snapshot run log.Notice("Starting snapshot", "source_dirs", len(resolvedDirs)) _, _ = fmt.Fprintf(app.Stdout, "Starting snapshot with %d source directories\n", len(resolvedDirs)) for i, dir := range resolvedDirs { log.Info("Source directory", "index", i+1, "path", dir) _, _ = fmt.Fprintf(app.Stdout, "Source directory %d: %s\n", i+1, dir) } // Statistics tracking totalFiles := 0 totalBytes := int64(0) totalChunks := 0 totalBlobs := 0 totalBytesSkipped := int64(0) totalFilesSkipped := 0 totalBytesUploaded := int64(0) totalBlobsUploaded := 0 uploadDuration := time.Duration(0) // Create a new snapshot at the beginning snapshotID, err := app.SnapshotManager.CreateSnapshot(ctx, hostname, app.Globals.Version, app.Globals.Commit) if err != nil { return fmt.Errorf("creating snapshot: %w", err) } log.Info("Created snapshot", "snapshot_id", snapshotID) _, _ = fmt.Fprintf(app.Stdout, "\nCreated snapshot: %s\n", snapshotID) for _, dir := range resolvedDirs { // Check if context is cancelled select { case <-ctx.Done(): log.Info("Snapshot creation cancelled") return ctx.Err() default: } log.Info("Scanning directory", "path", dir) result, err := scanner.Scan(ctx, dir, snapshotID) if err != nil { return fmt.Errorf("failed to scan %s: %w", dir, err) } totalFiles += result.FilesScanned totalBytes += result.BytesScanned totalChunks += result.ChunksCreated totalBlobs += result.BlobsCreated totalFilesSkipped += result.FilesSkipped totalBytesSkipped += result.BytesSkipped log.Info("Directory scan complete", "path", dir, "files", result.FilesScanned, "files_skipped", result.FilesSkipped, "bytes", result.BytesScanned, "bytes_skipped", result.BytesSkipped, "chunks", result.ChunksCreated, "blobs", result.BlobsCreated, "duration", result.EndTime.Sub(result.StartTime)) // Human-friendly output _, _ = fmt.Fprintf(app.Stdout, "\nDirectory: %s\n", dir) _, _ = fmt.Fprintf(app.Stdout, " Scanned: %d files (%s)\n", result.FilesScanned, humanize.Bytes(uint64(result.BytesScanned))) _, _ = fmt.Fprintf(app.Stdout, " Skipped: %d files (%s) - already backed up\n", result.FilesSkipped, humanize.Bytes(uint64(result.BytesSkipped))) _, _ = fmt.Fprintf(app.Stdout, " Created: %d chunks, %d blobs\n", result.ChunksCreated, result.BlobsCreated) _, _ = fmt.Fprintf(app.Stdout, " Duration: %s\n", result.EndTime.Sub(result.StartTime).Round(time.Millisecond)) } // Get upload statistics from scanner progress if available if s := scanner.GetProgress(); s != nil { stats := s.GetStats() totalBytesUploaded = stats.BytesUploaded.Load() totalBlobsUploaded = int(stats.BlobsUploaded.Load()) uploadDuration = time.Duration(stats.UploadDurationMs.Load()) * time.Millisecond } // Update snapshot statistics with extended fields extStats := snapshot.ExtendedBackupStats{ BackupStats: snapshot.BackupStats{ FilesScanned: totalFiles, BytesScanned: totalBytes, ChunksCreated: totalChunks, BlobsCreated: totalBlobs, BytesUploaded: totalBytesUploaded, }, BlobUncompressedSize: 0, // Will be set from database query below CompressionLevel: app.Config.CompressionLevel, UploadDurationMs: uploadDuration.Milliseconds(), } if err := app.SnapshotManager.UpdateSnapshotStatsExtended(ctx, snapshotID, extStats); err != nil { return fmt.Errorf("updating snapshot stats: %w", err) } // Mark snapshot as complete if err := app.SnapshotManager.CompleteSnapshot(ctx, snapshotID); err != nil { return fmt.Errorf("completing snapshot: %w", err) } // Export snapshot metadata // Export snapshot metadata without closing the database // The export function should handle its own database connection if err := app.SnapshotManager.ExportSnapshotMetadata(ctx, app.Config.IndexPath, snapshotID); err != nil { return fmt.Errorf("exporting snapshot metadata: %w", err) } // Calculate final statistics snapshotDuration := time.Since(snapshotStartTime) totalFilesChanged := totalFiles - totalFilesSkipped totalBytesChanged := totalBytes totalBytesAll := totalBytes + totalBytesSkipped // Calculate upload speed var avgUploadSpeed string if totalBytesUploaded > 0 && uploadDuration > 0 { bytesPerSec := float64(totalBytesUploaded) / uploadDuration.Seconds() bitsPerSec := bytesPerSec * 8 if bitsPerSec >= 1e9 { avgUploadSpeed = fmt.Sprintf("%.1f Gbit/s", bitsPerSec/1e9) } else if bitsPerSec >= 1e6 { avgUploadSpeed = fmt.Sprintf("%.0f Mbit/s", bitsPerSec/1e6) } else if bitsPerSec >= 1e3 { avgUploadSpeed = fmt.Sprintf("%.0f Kbit/s", bitsPerSec/1e3) } else { avgUploadSpeed = fmt.Sprintf("%.0f bit/s", bitsPerSec) } } else { avgUploadSpeed = "N/A" } // Get total blob sizes from database totalBlobSizeCompressed := int64(0) totalBlobSizeUncompressed := int64(0) if blobHashes, err := app.Repositories.Snapshots.GetBlobHashes(ctx, snapshotID); err == nil { for _, hash := range blobHashes { if blob, err := app.Repositories.Blobs.GetByHash(ctx, hash); err == nil && blob != nil { totalBlobSizeCompressed += blob.CompressedSize totalBlobSizeUncompressed += blob.UncompressedSize } } } // Calculate compression ratio var compressionRatio float64 if totalBlobSizeUncompressed > 0 { compressionRatio = float64(totalBlobSizeCompressed) / float64(totalBlobSizeUncompressed) } else { compressionRatio = 1.0 } // Print comprehensive summary _, _ = fmt.Fprintln(app.Stdout, "\n=== Snapshot Summary ===") _, _ = fmt.Fprintf(app.Stdout, "Snapshot ID: %s\n", snapshotID) _, _ = fmt.Fprintf(app.Stdout, "Source files: %s files, %s total\n", formatNumber(totalFiles), humanize.Bytes(uint64(totalBytesAll))) _, _ = fmt.Fprintf(app.Stdout, "Changed files: %s files, %s\n", formatNumber(totalFilesChanged), humanize.Bytes(uint64(totalBytesChanged))) _, _ = fmt.Fprintf(app.Stdout, "Unchanged files: %s files, %s\n", formatNumber(totalFilesSkipped), humanize.Bytes(uint64(totalBytesSkipped))) _, _ = fmt.Fprintf(app.Stdout, "Blob storage: %s uncompressed, %s compressed (%.2fx ratio, level %d)\n", humanize.Bytes(uint64(totalBlobSizeUncompressed)), humanize.Bytes(uint64(totalBlobSizeCompressed)), compressionRatio, app.Config.CompressionLevel) _, _ = fmt.Fprintf(app.Stdout, "Upload activity: %s uploaded, %d blobs, %s duration, %s avg speed\n", humanize.Bytes(uint64(totalBytesUploaded)), totalBlobsUploaded, formatDuration(uploadDuration), avgUploadSpeed) _, _ = fmt.Fprintf(app.Stdout, "Total time: %s\n", formatDuration(snapshotDuration)) _, _ = fmt.Fprintln(app.Stdout, "==========================") if opts.Prune { log.Info("Pruning enabled - will delete old snapshots after snapshot") // TODO: Implement pruning } return nil } // newSnapshotListCommand creates the 'snapshot list' subcommand func newSnapshotListCommand() *cobra.Command { var jsonOutput bool cmd := &cobra.Command{ Use: "list", Short: "List all snapshots", Long: "Lists all snapshots with their ID, timestamp, and compressed size", RunE: func(cmd *cobra.Command, args []string) error { return runSnapshotCommand(cmd.Context(), func(app *SnapshotApp) error { return app.List(cmd.Context(), jsonOutput) }) }, } cmd.Flags().BoolVar(&jsonOutput, "json", false, "Output in JSON format") return cmd } // newSnapshotPurgeCommand creates the 'snapshot purge' subcommand func newSnapshotPurgeCommand() *cobra.Command { var keepLatest bool var olderThan string var force bool cmd := &cobra.Command{ Use: "purge", Short: "Purge old snapshots", Long: "Removes snapshots based on age or count criteria", RunE: func(cmd *cobra.Command, args []string) error { // Validate flags if !keepLatest && olderThan == "" { return fmt.Errorf("must specify either --keep-latest or --older-than") } if keepLatest && olderThan != "" { return fmt.Errorf("cannot specify both --keep-latest and --older-than") } return runSnapshotCommand(cmd.Context(), func(app *SnapshotApp) error { return app.Purge(cmd.Context(), keepLatest, olderThan, force) }) }, } cmd.Flags().BoolVar(&keepLatest, "keep-latest", false, "Keep only the latest snapshot") cmd.Flags().StringVar(&olderThan, "older-than", "", "Remove snapshots older than duration (e.g., 30d, 6m, 1y)") cmd.Flags().BoolVar(&force, "force", false, "Skip confirmation prompt") return cmd } // newSnapshotVerifyCommand creates the 'snapshot verify' subcommand func newSnapshotVerifyCommand() *cobra.Command { var deep bool cmd := &cobra.Command{ Use: "verify ", Short: "Verify snapshot integrity", Long: "Verifies that all blobs referenced in a snapshot exist", Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { return runSnapshotCommand(cmd.Context(), func(app *SnapshotApp) error { return app.Verify(cmd.Context(), args[0], deep) }) }, } cmd.Flags().BoolVar(&deep, "deep", false, "Download and verify blob hashes") return cmd } // List lists all snapshots func (app *SnapshotApp) List(ctx context.Context, jsonOutput bool) error { // Get all remote snapshots remoteSnapshots := make(map[string]bool) objectCh := app.S3Client.ListObjectsStream(ctx, "metadata/", false) for object := range objectCh { if object.Err != nil { return fmt.Errorf("listing remote snapshots: %w", object.Err) } // Extract snapshot ID from paths like metadata/hostname-20240115-143052Z/ parts := strings.Split(object.Key, "/") if len(parts) >= 2 && parts[0] == "metadata" && parts[1] != "" { remoteSnapshots[parts[1]] = true } } // Get all local snapshots localSnapshots, err := app.Repositories.Snapshots.ListRecent(ctx, 10000) if err != nil { return fmt.Errorf("listing local snapshots: %w", err) } // Build a map of local snapshots for quick lookup localSnapshotMap := make(map[string]*database.Snapshot) for _, s := range localSnapshots { localSnapshotMap[s.ID] = s } // Remove local snapshots that don't exist remotely for _, snapshot := range localSnapshots { if !remoteSnapshots[snapshot.ID] { log.Info("Removing local snapshot not found in remote", "snapshot_id", snapshot.ID) if err := app.Repositories.Snapshots.Delete(ctx, snapshot.ID); err != nil { log.Error("Failed to delete local snapshot", "snapshot_id", snapshot.ID, "error", err) } delete(localSnapshotMap, snapshot.ID) } } // Build final snapshot list snapshots := make([]SnapshotInfo, 0, len(remoteSnapshots)) for snapshotID := range remoteSnapshots { // Check if we have this snapshot locally if localSnap, exists := localSnapshotMap[snapshotID]; exists && localSnap.CompletedAt != nil { // Use local data snapshots = append(snapshots, SnapshotInfo{ ID: localSnap.ID, Timestamp: localSnap.StartedAt, CompressedSize: localSnap.BlobSize, }) } else { // Remote snapshot not in local DB - fetch manifest to get size timestamp, err := parseSnapshotTimestamp(snapshotID) if err != nil { log.Warn("Failed to parse snapshot timestamp", "id", snapshotID, "error", err) continue } // Try to download manifest to get size totalSize, err := app.getManifestSize(ctx, snapshotID) if err != nil { return fmt.Errorf("failed to get manifest size for %s: %w", snapshotID, err) } snapshots = append(snapshots, SnapshotInfo{ ID: snapshotID, Timestamp: timestamp, CompressedSize: totalSize, }) } } // Sort by timestamp (newest first) sort.Slice(snapshots, func(i, j int) bool { return snapshots[i].Timestamp.After(snapshots[j].Timestamp) }) if jsonOutput { // JSON output encoder := json.NewEncoder(os.Stdout) encoder.SetIndent("", " ") return encoder.Encode(snapshots) } // Table output w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) if _, err := fmt.Fprintln(w, "SNAPSHOT ID\tTIMESTAMP\tCOMPRESSED SIZE"); err != nil { return err } if _, err := fmt.Fprintln(w, "───────────\t─────────\t───────────────"); err != nil { return err } for _, snap := range snapshots { if _, err := fmt.Fprintf(w, "%s\t%s\t%s\n", snap.ID, snap.Timestamp.Format("2006-01-02 15:04:05"), formatBytes(snap.CompressedSize)); err != nil { return err } } return w.Flush() } // Purge removes old snapshots based on criteria func (app *SnapshotApp) Purge(ctx context.Context, keepLatest bool, olderThan string, force bool) error { // Sync with remote first if err := app.syncWithRemote(ctx); err != nil { return fmt.Errorf("syncing with remote: %w", err) } // Get snapshots from local database dbSnapshots, err := app.Repositories.Snapshots.ListRecent(ctx, 10000) if err != nil { return fmt.Errorf("listing snapshots: %w", err) } // Convert to SnapshotInfo format, only including completed snapshots snapshots := make([]SnapshotInfo, 0, len(dbSnapshots)) for _, s := range dbSnapshots { if s.CompletedAt != nil { snapshots = append(snapshots, SnapshotInfo{ ID: s.ID, Timestamp: s.StartedAt, CompressedSize: s.BlobSize, }) } } // Sort by timestamp (newest first) sort.Slice(snapshots, func(i, j int) bool { return snapshots[i].Timestamp.After(snapshots[j].Timestamp) }) var toDelete []SnapshotInfo if keepLatest { // Keep only the most recent snapshot if len(snapshots) > 1 { toDelete = snapshots[1:] } } else if olderThan != "" { // Parse duration duration, err := parseDuration(olderThan) if err != nil { return fmt.Errorf("invalid duration: %w", err) } cutoff := time.Now().UTC().Add(-duration) for _, snap := range snapshots { if snap.Timestamp.Before(cutoff) { toDelete = append(toDelete, snap) } } } if len(toDelete) == 0 { fmt.Println("No snapshots to delete") return nil } // Show what will be deleted fmt.Printf("The following snapshots will be deleted:\n\n") for _, snap := range toDelete { fmt.Printf(" %s (%s, %s)\n", snap.ID, snap.Timestamp.Format("2006-01-02 15:04:05"), formatBytes(snap.CompressedSize)) } // Confirm unless --force is used if !force { fmt.Printf("\nDelete %d snapshot(s)? [y/N] ", len(toDelete)) var confirm string if _, err := fmt.Scanln(&confirm); err != nil { // Treat EOF or error as "no" fmt.Println("Cancelled") return nil } if strings.ToLower(confirm) != "y" { fmt.Println("Cancelled") return nil } } else { fmt.Printf("\nDeleting %d snapshot(s) (--force specified)\n", len(toDelete)) } // Delete snapshots for _, snap := range toDelete { log.Info("Deleting snapshot", "id", snap.ID) if err := app.deleteSnapshot(ctx, snap.ID); err != nil { return fmt.Errorf("deleting snapshot %s: %w", snap.ID, err) } } fmt.Printf("Deleted %d snapshot(s)\n", len(toDelete)) // TODO: Run blob pruning to clean up unreferenced blobs return nil } // Verify checks snapshot integrity func (app *SnapshotApp) Verify(ctx context.Context, snapshotID string, deep bool) error { fmt.Printf("Verifying snapshot %s...\n", snapshotID) // Download and parse manifest manifest, err := app.downloadManifest(ctx, snapshotID) if err != nil { return fmt.Errorf("downloading manifest: %w", err) } fmt.Printf("Manifest contains %d blobs\n", len(manifest)) // Check each blob exists missing := 0 verified := 0 for _, blobHash := range manifest { blobPath := fmt.Sprintf("blobs/%s/%s/%s", blobHash[:2], blobHash[2:4], blobHash) if deep { // Download and verify hash // TODO: Implement deep verification fmt.Printf("Deep verification not yet implemented\n") return nil } else { // Just check existence _, err := app.S3Client.StatObject(ctx, blobPath) if err != nil { fmt.Printf(" Missing: %s\n", blobHash) missing++ } else { verified++ } } } fmt.Printf("\nVerification complete:\n") fmt.Printf(" Verified: %d\n", verified) fmt.Printf(" Missing: %d\n", missing) if missing > 0 { return fmt.Errorf("%d blobs are missing", missing) } return nil } // getManifestSize downloads a manifest and returns the total compressed size func (app *SnapshotApp) getManifestSize(ctx context.Context, snapshotID string) (int64, error) { manifestPath := fmt.Sprintf("metadata/%s/manifest.json.zst", snapshotID) reader, err := app.S3Client.GetObject(ctx, manifestPath) if err != nil { return 0, fmt.Errorf("downloading manifest: %w", err) } defer func() { _ = reader.Close() }() manifest, err := snapshot.DecodeManifest(reader) if err != nil { return 0, fmt.Errorf("decoding manifest: %w", err) } return manifest.TotalCompressedSize, nil } // downloadManifest downloads and parses a snapshot manifest (for verify command) func (app *SnapshotApp) downloadManifest(ctx context.Context, snapshotID string) ([]string, error) { manifestPath := fmt.Sprintf("metadata/%s/manifest.json.zst", snapshotID) reader, err := app.S3Client.GetObject(ctx, manifestPath) if err != nil { return nil, err } defer func() { _ = reader.Close() }() manifest, err := snapshot.DecodeManifest(reader) if err != nil { return nil, fmt.Errorf("decoding manifest: %w", err) } // Extract blob hashes hashes := make([]string, len(manifest.Blobs)) for i, blob := range manifest.Blobs { hashes[i] = blob.Hash } return hashes, nil } // deleteSnapshot removes a snapshot and its metadata func (app *SnapshotApp) deleteSnapshot(ctx context.Context, snapshotID string) error { // List all objects under metadata/{snapshotID}/ prefix := fmt.Sprintf("metadata/%s/", snapshotID) objectCh := app.S3Client.ListObjectsStream(ctx, prefix, true) var objectsToDelete []string for object := range objectCh { if object.Err != nil { return fmt.Errorf("listing objects: %w", object.Err) } objectsToDelete = append(objectsToDelete, object.Key) } // Delete all objects for _, key := range objectsToDelete { if err := app.S3Client.RemoveObject(ctx, key); err != nil { return fmt.Errorf("removing %s: %w", key, err) } } return nil } // syncWithRemote syncs local database with remote snapshots func (app *SnapshotApp) syncWithRemote(ctx context.Context) error { log.Info("Syncing with remote snapshots") // Get all remote snapshot IDs remoteSnapshots := make(map[string]bool) objectCh := app.S3Client.ListObjectsStream(ctx, "metadata/", false) for object := range objectCh { if object.Err != nil { return fmt.Errorf("listing remote snapshots: %w", object.Err) } // Extract snapshot ID from paths like metadata/hostname-20240115-143052Z/ parts := strings.Split(object.Key, "/") if len(parts) >= 2 && parts[0] == "metadata" && parts[1] != "" { remoteSnapshots[parts[1]] = true } } log.Debug("Found remote snapshots", "count", len(remoteSnapshots)) // Get all local snapshots (use a high limit to get all) localSnapshots, err := app.Repositories.Snapshots.ListRecent(ctx, 10000) if err != nil { return fmt.Errorf("listing local snapshots: %w", err) } // Remove local snapshots that don't exist remotely removedCount := 0 for _, snapshot := range localSnapshots { if !remoteSnapshots[snapshot.ID] { log.Info("Removing local snapshot not found in remote", "snapshot_id", snapshot.ID) if err := app.Repositories.Snapshots.Delete(ctx, snapshot.ID); err != nil { log.Error("Failed to delete local snapshot", "snapshot_id", snapshot.ID, "error", err) } else { removedCount++ } } } if removedCount > 0 { log.Info("Removed local snapshots not found in remote", "count", removedCount) } return nil } // parseSnapshotTimestamp extracts timestamp from snapshot ID // Format: hostname-20240115-143052Z func parseSnapshotTimestamp(snapshotID string) (time.Time, error) { // The snapshot ID format is: hostname-YYYYMMDD-HHMMSSZ // We need to find the timestamp part which starts after the hostname // Split by hyphen parts := strings.Split(snapshotID, "-") if len(parts) < 3 { return time.Time{}, fmt.Errorf("invalid snapshot ID format: expected hostname-YYYYMMDD-HHMMSSZ") } // The last two parts should be the date and time with Z suffix dateStr := parts[len(parts)-2] timeStr := parts[len(parts)-1] // Reconstruct the full timestamp fullTimestamp := dateStr + "-" + timeStr // Parse the timestamp with Z suffix return time.Parse("20060102-150405Z", fullTimestamp) } // parseDuration is now in duration.go // runSnapshotCommand creates the FX app and runs the given function func runSnapshotCommand(ctx context.Context, fn func(*SnapshotApp) error) error { var result error rootFlags := GetRootFlags() // Use unified config resolution configPath, err := ResolveConfigPath() if err != nil { return err } err = RunWithApp(ctx, AppOptions{ ConfigPath: configPath, LogOptions: log.LogOptions{ Verbose: rootFlags.Verbose, Debug: rootFlags.Debug, }, Modules: []fx.Option{ s3.Module, fx.Provide(func( g *globals.Globals, cfg *config.Config, db *database.DB, repos *database.Repositories, s3Client *s3.Client, lc fx.Lifecycle, shutdowner fx.Shutdowner, ) *SnapshotApp { snapshotCreateApp := &SnapshotCreateApp{ Globals: g, Config: cfg, Repositories: repos, ScannerFactory: nil, // Not needed for snapshot commands S3Client: s3Client, DB: db, Lifecycle: lc, Shutdowner: shutdowner, } return &SnapshotApp{ SnapshotCreateApp: snapshotCreateApp, S3Client: s3Client, } }), }, Invokes: []fx.Option{ fx.Invoke(func(app *SnapshotApp, shutdowner fx.Shutdowner) { result = fn(app) // Shutdown after command completes go func() { time.Sleep(100 * time.Millisecond) // Brief delay to ensure clean shutdown if err := shutdowner.Shutdown(); err != nil { log.Error("Failed to shutdown", "error", err) } }() }), }, }) if err != nil { return err } return result } // formatNumber formats a number with comma separators func formatNumber(n int) string { if n < 1000 { return fmt.Sprintf("%d", n) } return humanize.Comma(int64(n)) } // formatDuration formats a duration in a human-readable way func formatDuration(d time.Duration) string { if d < time.Second { return fmt.Sprintf("%dms", d.Milliseconds()) } if d < time.Minute { return fmt.Sprintf("%.1fs", d.Seconds()) } if d < time.Hour { mins := int(d.Minutes()) secs := int(d.Seconds()) % 60 if secs > 0 { return fmt.Sprintf("%dm%ds", mins, secs) } return fmt.Sprintf("%dm", mins) } hours := int(d.Hours()) mins := int(d.Minutes()) % 60 if mins > 0 { return fmt.Sprintf("%dh%dm", hours, mins) } return fmt.Sprintf("%dh", hours) }