<commit_before>package cli

import (
	"context"
	"fmt"
	"sort"
	"sync"
	"time"

	"github.com/kopia/kopia/snapshot"
	"github.com/kopia/kopia/snapshot/snapshotfs"
	"github.com/kopia/repo"
)

var (
	migrateCommand      = snapshotCommands.Command("migrate", "Migrate snapshots from another repository")
	migrateSourceConfig = migrateCommand.Flag("source-config", "Configuration file for the source repository").Required().ExistingFile()
	migrateSources      = migrateCommand.Flag("sources", "List of sources to migrate").Strings()
	migrateAll          = migrateCommand.Flag("all", "Migrate all sources").Bool()
	migrateLatestOnly   = migrateCommand.Flag("latest-only", "Only migrate the latest snapshot").Bool()
	migrateIgnoreErrors = migrateCommand.Flag("ignore-errors", "Ignore errors when reading source backup").Bool()
	migrateParallelism  = migrateCommand.Flag("parallelism", "Number of sources to migrate in parallel").Default("1").Int()
)

func runMigrateCommand(ctx context.Context, destRepo *repo.Repository) error {
	sourceRepo, err := repo.Open(ctx, *migrateSourceConfig, mustGetPasswordFromFlags(false, false), applyOptionsFromFlags(nil))
	if err != nil {
		return fmt.Errorf("can't open source repository: %v", err)
	}

	sources, err := getSourcesToMigrate(ctx, sourceRepo)
	if err != nil {
		return fmt.Errorf("can't retrieve sources: %v", err)
	}

	semaphore := make(chan struct{}, *migrateParallelism)
	var wg sync.WaitGroup

	for _, s := range sources {
		uploader := snapshotfs.NewUploader(destRepo)
		uploader.Progress = cliProgress
		uploader.IgnoreFileErrors = *migrateIgnoreErrors
		onCtrlC(uploader.Cancel)

		wg.Add(1)
		semaphore <- struct{}{}
		go func(s snapshot.SourceInfo) {
			defer func() {
				<-semaphore
				wg.Done()
			}()

			if err := migrateSingleSource(ctx, uploader, sourceRepo, destRepo, s); err != nil {
				log.Warningf("unable to migrate source: %v", err)
			}
		}(s)
	}
	wg.Wait()

	return nil
}

func findPreviousSnapshotManifestWithStartTime(ctx context.Context, rep *repo.Repository, sourceInfo snapshot.SourceInfo, startTime time.Time) (*snapshot.Manifest, error) {
	previous, err := snapshot.ListSnapshots(ctx, rep, sourceInfo)
	if err != nil {
		return nil, fmt.Errorf("error listing previous backups: %v", err)
	}

	for _, p := range previous {
		if p.StartTime == startTime {
			return p, nil
		}
	}

	return nil, nil
}

func migrateSingleSource(ctx context.Context, uploader *snapshotfs.Uploader, sourceRepo, destRepo *repo.Repository, s snapshot.SourceInfo) error {
	manifests, err := snapshot.ListSnapshotManifests(ctx, sourceRepo, &s)
	if err != nil {
		return err
	}
	snapshots, err := snapshot.LoadSnapshots(ctx, sourceRepo, manifests)
	if err != nil {
		return fmt.Errorf("unable to load snapshot manifests for %v: %v", s, err)
	}

	sort.Slice(snapshots, func(i, j int) bool {
		return snapshots[i].StartTime.Before(snapshots[j].StartTime)
	})

	for _, m := range filterSnapshotsToMigrate(snapshots) {
		if m.IncompleteReason != "" {
			log.Infof("ignoring incomplete %v at %v", s, formatTimestamp(m.StartTime))<commit_msg>cli: fixed 'repo migrate' cancellation<commit_after>package cli

import (
	"context"
	"fmt"
	"sort"
	"sync"
	"time"

	"github.com/kopia/kopia/snapshot"
	"github.com/kopia/kopia/snapshot/snapshotfs"
	"github.com/kopia/repo"
)

var (
	migrateCommand      = snapshotCommands.Command("migrate", "Migrate snapshots from another repository")
	migrateSourceConfig = migrateCommand.Flag("source-config", "Configuration file for the source repository").Required().ExistingFile()
	migrateSources      = migrateCommand.Flag("sources", "List of sources to migrate").Strings()
	migrateAll          = migrateCommand.Flag("all", "Migrate all sources").Bool()
	migrateLatestOnly   = migrateCommand.Flag("latest-only", "Only migrate the latest snapshot").Bool()
	migrateIgnoreErrors = migrateCommand.Flag("ignore-errors", "Ignore errors when reading source backup").Bool()
	migrateParallelism  = migrateCommand.Flag("parallelism", "Number of sources to migrate in parallel").Default("1").Int()
)

func runMigrateCommand(ctx context.Context, destRepo *repo.Repository) error {
	sourceRepo, err := repo.Open(ctx, *migrateSourceConfig, mustGetPasswordFromFlags(false, false), applyOptionsFromFlags(nil))
	if err != nil {
		return fmt.Errorf("can't open source repository: %v", err)
	}

	sources, err := getSourcesToMigrate(ctx, sourceRepo)
	if err != nil {
		return fmt.Errorf("can't retrieve sources: %v", err)
	}

	semaphore := make(chan struct{}, *migrateParallelism)
	var wg sync.WaitGroup

	var (
		mu              sync.Mutex
		cancelled       bool
		activeUploaders = map[snapshot.SourceInfo]*snapshotfs.Uploader{}
	)

	onCtrlC(func() {
		mu.Lock()
		mu.Unlock()

		if !cancelled {
			cancelled = true
			for s, u := range activeUploaders {
				log.Warningf("cancelling active uploader for %v", s)
				u.Cancel()
			}
		}
	})

	for _, s := range sources {
		// start a new uploader unless already cancelled
		mu.Lock()
		if cancelled {
			mu.Unlock()
			break
		}

		uploader := snapshotfs.NewUploader(destRepo)
		uploader.Progress = cliProgress
		uploader.IgnoreFileErrors = *migrateIgnoreErrors
		activeUploaders[s] = uploader
		mu.Unlock()

		wg.Add(1)
		semaphore <- struct{}{}
		go func(s snapshot.SourceInfo) {
			defer func() {
				mu.Lock()
				delete(activeUploaders, s)
				mu.Unlock()

				<-semaphore
				wg.Done()
			}()

			if err := migrateSingleSource(ctx, uploader, sourceRepo, destRepo, s); err != nil {
				log.Warningf("unable to migrate source: %v", err)
			}
		}(s)
	}
	wg.Wait()

	return nil
}

func findPreviousSnapshotManifestWithStartTime(ctx context.Context, rep *repo.Repository, sourceInfo snapshot.SourceInfo, startTime time.Time) (*snapshot.Manifest, error) {
	previous, err := snapshot.ListSnapshots(ctx, rep, sourceInfo)
	if err != nil {
		return nil, fmt.Errorf("error listing previous backups: %v", err)
	}

	for _, p := range previous {
		if p.StartTime == startTime {
			return p, nil
		}
	}

	return nil, nil
}

func migrateSingleSource(ctx context.Context, uploader *snapshotfs.Uploader, sourceRepo, destRepo *repo.Repository, s snapshot.SourceInfo) error {
	manifests, err := snapshot.ListSnapshotManifests(ctx, sourceRepo, &s)
	if err != nil {
		return err
	}
	snapshots, err := snapshot.LoadSnapshots(ctx, sourceRepo, manifests)
	if err != nil {
		return fmt.Errorf("unable to load snapshot manifests for %v: %v", s, err)
	}

	sort.Slice(snapshots, func(i, j int) bool {
		return snapshots[i].StartTime.Before(snapshots[j].StartTime)
	})

	for _, m := range filterSnapshotsToMigrate(snapshots) {
		if uploader.IsCancelled() {
			break
		}
		if m.IncompleteReason != "" {
			log.Infof("ignoring incomplete %v at %v", s, formatTimestamp(m.StartTime))
{'fraction_non_alphanumeric': 0.1290704558910598, 'fraction_numerical': 0.0005920663114268798, 'mean_word_length': 4.506927465362673, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1983567', 'n_tokens_mistral': 2369, 'n_tokens_neox': 2232, 'n_words': 542}