repo-sync/mirror/sync.go

package mirror

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"syscall"
	"time"

	"github.com/grmrgecko/repo-sync/fetch"
	log "github.com/sirupsen/logrus"
)

// lockDestination takes an exclusive lock under the destination directory
// so overlapping runs, such as from cron, cannot race on the same tree.
// The caller closes the returned file to release the lock.
func lockDestination(dest string) (*os.File, error) {
	if err := os.MkdirAll(dest, 0755); err != nil {
		return nil, err
	}
	f, err := os.OpenFile(filepath.Join(dest, fetch.LockFileName), os.O_CREATE|os.O_RDWR, 0644)
	if err != nil {
		return nil, err
	}
	if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
		f.Close()
		return nil, fmt.Errorf("destination %s is locked by another repo-sync run", dest)
	}
	return f, nil
}

// Sync synchronizes every configured repository URL into the destination,
// continuing after per-repository failures so one bad repository does not
// block the rest.
func Sync(ctx context.Context, opts *Options) error {
	fetch.Reload()

	// Hold the destination for the whole run; the lock file itself stays
	// behind, as removing it would race a waiting run.
	lock, err := lockDestination(opts.Destination)
	if err != nil {
		return err
	}
	defer lock.Close()

	// Expand discovery URLs into concrete repository URLs.
	repoURLs := opts.URLs
	if opts.Discover {
		repoURLs = nil
		for _, base := range opts.URLs {
			found, err := discoverRepos(ctx, base, opts.Type, opts.DiscoverDepth)
			if err != nil {
				return fmt.Errorf("discover repositories under %s: %w", base, err)
			}
			log.WithFields(log.Fields{"url": base, "repositories": len(found)}).Info("Discovered repositories.")
			repoURLs = append(repoURLs, found...)
		}
	}
	if len(repoURLs) == 0 {
		return errors.New("no repositories to synchronize")
	}

	// Synchronize each repository in turn.
	var failures int
	for _, repoURL := range repoURLs {
		if err := ctx.Err(); err != nil {
			return err
		}
		log.WithField("url", repoURL).Info("Synchronizing repository.")
		fetch.Stats.Reset()
		start := time.Now()
		if err := syncOne(ctx, repoURL, opts); err != nil {
			if errors.Is(err, context.Canceled) {
				return err
			}
			log.WithError(err).WithField("url", repoURL).Error("Repository synchronization failed.")
			failures++
			continue
		}
		logSummary(repoURL, opts, time.Since(start))
	}
	if failures > 0 {
		return fmt.Errorf("%d of %d repositories failed to synchronize", failures, len(repoURLs))
	}
	return nil
}

// logSummary reports one repository's transfer activity after a successful
// synchronization.
func logSummary(repoURL string, opts *Options, elapsed time.Duration) {
	fields := log.Fields{
		"url":         repoURL,
		"fetched":     fetch.Stats.Fetched.Load(),
		"transferred": fetch.FormatBytes(fetch.Stats.FetchedBytes.Load()),
		"unchanged":   fetch.Stats.Unchanged.Load(),
		"duration":    elapsed.Round(time.Millisecond).String(),
	}
	if opts.DryRun {
		fields["would_fetch"] = fetch.Stats.Planned.Load()
		fields["would_transfer"] = fetch.FormatBytes(fetch.Stats.PlannedBytes.Load())
		if opts.Prune {
			fields["would_prune"] = fetch.Stats.Pruned.Load()
		}
		log.WithFields(fields).Info("Dry run complete.")
		return
	}
	if opts.Prune {
		fields["pruned"] = fetch.Stats.Pruned.Load()
	}
	log.WithFields(fields).Info("Repository synchronized.")
}

// syncOne resolves the source for a single repository URL and runs the
// format-specific synchronization. Destination paths derive from the layout
// URL, so a mirror list maps to the first mirror's path rather than the
// list's own URL.
func syncOne(ctx context.Context, repoURL string, opts *Options) error {
	src, layoutURL, err := resolveSource(ctx, repoURL, opts.Type)
	if err != nil {
		return err
	}
	switch opts.Type {
	case RepoRPM:
		destDir, err := opts.repoDest(layoutURL)
		if err != nil {
			return err
		}
		log.WithField("destination", destDir).Debug("Resolved repository destination.")
		return syncRPM(ctx, src, destDir, opts)
	case RepoDeb:
		return syncDeb(ctx, src, layoutURL, opts)
	case RepoArch:
		destDir, err := opts.repoDest(layoutURL)
		if err != nil {
			return err
		}
		log.WithField("destination", destDir).Debug("Resolved repository destination.")
		return syncArch(ctx, src, layoutURL, destDir, opts)
	case RepoApk:
		destDir, err := opts.repoDest(layoutURL)
		if err != nil {
			return err
		}
		log.WithField("destination", destDir).Debug("Resolved repository destination.")
		return syncApk(ctx, src, destDir, opts)
	default:
		return fmt.Errorf("unsupported repository type %q", opts.Type)
	}
}

// SyncInto synchronizes one repository into an explicit destination
// directory, bypassing URL-derived destination mapping. It exists for the
// mirror server, whose destinations follow request paths rather than
// upstream URL paths. For deb the destination is the archive root
// directory; for other types it is the repository directory itself.
// Concurrent callers interleave into fetch.Stats, so the counters carry no
// per-repository meaning on this path and are neither reset nor reported.
// A trace enabled here still reports correct totals: it accumulates its own
// counters for the duration of the crawl rather than reading the shared
// collector.
func SyncInto(ctx context.Context, typ RepoType, src *fetch.Source, repoURL, dest string, opts *Options) error {
	// Copy before the deb path pins destBase so the caller's Options is
	// never mutated, keeping the struct safe to reuse or share.
	serverOpts := *opts
	opts = &serverOpts
	switch typ {
	case RepoRPM:
		return syncRPM(ctx, src, dest, opts)
	case RepoDeb:
		opts.destBase = dest
		return syncDeb(ctx, src, repoURL, opts)
	case RepoArch:
		return syncArch(ctx, src, repoURL, dest, opts)
	case RepoApk:
		return syncApk(ctx, src, dest, opts)
	default:
		return fmt.Errorf("unsupported repository type %q", typ)
	}
}