// seedreplica feeds missing archive revisions to a replica without overwhelming its pull queue.
package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math/rand"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"sync"
	"text/template"
	"time"

	"github.com/rcowham/kingpin"
	"github.com/sirupsen/logrus"
)

const description = "Seed a replica by verifying missing archive revisions in batches while throttling its pull queue."

const VersionChange = "$Change: 34019 $"

var GoVersion = runtime.Version()
var Version = "dev"

var versionInfoTmpl = `
{{.program}}, version {{.version}} (change: {{.change}})
  go version:       {{.goVersion}}
`

func VersionPrint(program, version string) string {
	change := strings.TrimPrefix(VersionChange, "$Change: ")
	change = strings.TrimSuffix(change, " $")
	values := map[string]string{
		"program":   program,
		"version":   version,
		"change":    change,
		"goVersion": GoVersion,
	}
	template := template.Must(template.New("version").Parse(versionInfoTmpl))
	var output bytes.Buffer
	if err := template.ExecuteTemplate(&output, "version", values); err != nil {
		panic(err)
	}
	return strings.TrimSpace(output.String())
}

type config struct {
	input         string
	p4bin         string
	p4port        string
	p4user        string
	batchSize     int
	maxQueue      int
	pollInterval  time.Duration
	maxBatches    int
	checkpoint    string
	resume        bool
	follow        bool
	dryRun        bool
	discoverInput string
	discoverOnly  bool
	discoverers   int
	replaceInput  bool
}

type checkpoint struct {
	Input       string    `json:"input"`
	Offset      int64     `json:"offset"`
	LinesRead   int64     `json:"lines_read"`
	Processed   int64     `json:"processed"`
	Batches     int       `json:"batches"`
	UpdatedAt   time.Time `json:"updated_at"`
	CompletedAt time.Time `json:"completed_at,omitempty"`
}

type queueStatus struct {
	ActiveTransfers int
	TotalTransfers  int
	ActiveBytes     string
	TotalBytes      string
}

type p4Client struct {
	bin      string
	port     string
	user     string
	logger   *logrus.Logger
	tracking *trackCollector
}

type depot struct {
	name        string
	typeName    string
	streamDepth int
}

var (
	pullQueuePattern       = regexp.MustCompile(`(?m)^File transfers:\s*(\d+)\s+active/(\d+)\s+total,\s*bytes:\s*(\S+)\s+active/(\S+)\s+total\.`)
	missingRevisionPattern = regexp.MustCompile(`(?i)(//[^#\s]+#\d+)\b.*\b(?:BAD|MISSING)\b`)
	sizesFileCountPattern  = regexp.MustCompile(`(?i)(\d[\d,]*)\s+files?\b`)
	trackLapsePattern      = regexp.MustCompile(`^--- lapse ([0-9.]+s)$`)
	trackUsagePattern      = regexp.MustCompile(`^--- usage ([0-9.]+)\+([0-9.]+)us(?:\s|$)`)
	trackTablePattern      = regexp.MustCompile(`^--- (db\.\S+)$`)
	trackLockPattern       = regexp.MustCompile(`^---\s+total lock wait\+held read/write ([0-9.]+)ms\+([0-9.]+)ms/([0-9.]+)ms\+([0-9.]+)ms$`)
	trackLibraryPattern    = regexp.MustCompile(`^--- lbr (.+)$`)
	trackLibraryUsePattern = regexp.MustCompile(`^---\s+opens\+closes\+checkins\+exists (\d+)\+(\d+)\+(\d+)\+(\d+)$`)
)

type tableTrack struct {
	ReadWait  time.Duration
	ReadHeld  time.Duration
	WriteWait time.Duration
	WriteHeld time.Duration
}

type libraryTrack struct {
	Opens    int64
	Closes   int64
	Checkins int64
	Exists   int64
}

type trackOutput struct {
	Lapse     time.Duration
	UserCPU   time.Duration
	SystemCPU time.Duration
	Tables    map[string]tableTrack
	Libraries map[string]libraryTrack
	tracked   bool
}

type trackCollector struct {
	mu       sync.Mutex
	commands map[string]trackOutput
}

func parseTrackDuration(value string, unit string) time.Duration {
	duration, err := time.ParseDuration(value + unit)
	if err != nil {
		return 0
	}
	return duration
}

func parseTrackOutput(output string) trackOutput {
	track := trackOutput{Tables: make(map[string]tableTrack), Libraries: make(map[string]libraryTrack)}
	var table string
	var library string
	for _, line := range strings.Split(output, "\n") {
		line = strings.TrimSpace(line)
		if matches := trackLapsePattern.FindStringSubmatch(line); matches != nil {
			track.Lapse = parseTrackDuration(matches[1], "")
			track.tracked = true
			continue
		}
		if matches := trackUsagePattern.FindStringSubmatch(line); matches != nil {
			// P4 labels these fields "us", but reports their values in milliseconds.
			track.UserCPU = parseTrackDuration(matches[1], "ms")
			track.SystemCPU = parseTrackDuration(matches[2], "ms")
			track.tracked = true
			continue
		}
		if matches := trackTablePattern.FindStringSubmatch(line); matches != nil {
			table = matches[1]
			library = ""
			continue
		}
		if matches := trackLibraryPattern.FindStringSubmatch(line); matches != nil {
			library = matches[1]
			table = ""
			continue
		}
		if matches := trackLockPattern.FindStringSubmatch(line); matches != nil && table != "" {
			track.Tables[table] = tableTrack{
				ReadWait:  parseTrackDuration(matches[1], "ms"),
				ReadHeld:  parseTrackDuration(matches[2], "ms"),
				WriteWait: parseTrackDuration(matches[3], "ms"),
				WriteHeld: parseTrackDuration(matches[4], "ms"),
			}
			track.tracked = true
			continue
		}
		if matches := trackLibraryUsePattern.FindStringSubmatch(line); matches != nil && library != "" {
			track.Libraries[library] = libraryTrack{
				Opens:    parseTrackCount(matches[1]),
				Closes:   parseTrackCount(matches[2]),
				Checkins: parseTrackCount(matches[3]),
				Exists:   parseTrackCount(matches[4]),
			}
			track.tracked = true
		}
	}
	return track
}

func parseTrackCount(value string) int64 {
	count, err := strconv.ParseInt(value, 10, 64)
	if err != nil {
		return 0
	}
	return count
}

func (track *trackOutput) add(other trackOutput) {
	track.Lapse += other.Lapse
	track.UserCPU += other.UserCPU
	track.SystemCPU += other.SystemCPU
	for name, locks := range other.Tables {
		total := track.Tables[name]
		total.ReadWait += locks.ReadWait
		total.ReadHeld += locks.ReadHeld
		total.WriteWait += locks.WriteWait
		total.WriteHeld += locks.WriteHeld
		track.Tables[name] = total
	}
	for name, usage := range other.Libraries {
		total := track.Libraries[name]
		total.Opens += usage.Opens
		total.Closes += usage.Closes
		total.Checkins += usage.Checkins
		total.Exists += usage.Exists
		track.Libraries[name] = total
	}
}

func (collector *trackCollector) add(command string, track trackOutput) {
	if collector == nil || !track.tracked {
		return
	}
	collector.mu.Lock()
	defer collector.mu.Unlock()
	if collector.commands == nil {
		collector.commands = make(map[string]trackOutput)
	}
	total, ok := collector.commands[command]
	if !ok {
		total = trackOutput{Tables: make(map[string]tableTrack), Libraries: make(map[string]libraryTrack)}
	}
	total.add(track)
	collector.commands[command] = total
}

func p4CommandName(args []string) string {
	for _, arg := range args {
		switch arg {
		case "depots", "dirs", "sizes", "verify", "pull":
			return arg
		}
	}
	return "p4"
}

func (collector *trackCollector) writeSummary(output io.Writer, phase string) {
	if collector == nil {
		return
	}
	collector.mu.Lock()
	defer collector.mu.Unlock()
	commands := make([]string, 0, len(collector.commands))
	for command := range collector.commands {
		commands = append(commands, command)
	}
	sort.Strings(commands)
	for _, command := range commands {
		track := collector.commands[command]
		fmt.Fprintf(output, "P4 tracking %s %s: lapse=%s user=%s system=%s\n", phase, command, track.Lapse, track.UserCPU, track.SystemCPU)
		tables := make([]string, 0, len(track.Tables))
		for table := range track.Tables {
			tables = append(tables, table)
		}
		sort.Strings(tables)
		for _, table := range tables {
			locks := track.Tables[table]
			fmt.Fprintf(output, "P4 tracking %s %s %s: read wait/held=%s/%s write wait/held=%s/%s\n", phase, command, table, locks.ReadWait, locks.ReadHeld, locks.WriteWait, locks.WriteHeld)
		}
		libraries := make([]string, 0, len(track.Libraries))
		for library := range track.Libraries {
			libraries = append(libraries, library)
		}
		sort.Strings(libraries)
		for _, library := range libraries {
			usage := track.Libraries[library]
			fmt.Fprintf(output, "P4 tracking %s %s lbr %s: opens/closes/checkins/exists=%d/%d/%d/%d\n", phase, command, library, usage.Opens, usage.Closes, usage.Checkins, usage.Exists)
		}
	}
}

func (p p4Client) commandArgs(args ...string) []string {
	commandArgs := make([]string, 0, len(args)+5)
	commandArgs = append(commandArgs, "-Ztrack")
	if p.port != "" {
		commandArgs = append(commandArgs, "-p", p.port)
	}
	if p.user != "" {
		commandArgs = append(commandArgs, "-u", p.user)
	}
	return append(commandArgs, args...)
}

func (p p4Client) run(ctx context.Context, args ...string) ([]byte, error) {
	commandArgs := p.commandArgs(args...)
	p.logger.Debugf("Running: %s %s", p.bin, strings.Join(commandArgs, " "))
	output, err := exec.CommandContext(ctx, p.bin, commandArgs...).CombinedOutput()
	p.recordTrack(args, string(output))
	if err != nil {
		return output, fmt.Errorf("p4 %s: %w: %s", strings.Join(args, " "), err, strings.TrimSpace(string(output)))
	}
	return output, nil
}

func (p p4Client) recordTrack(args []string, output string) {
	track := parseTrackOutput(output)
	if !track.tracked {
		return
	}
	command := p4CommandName(args)
	p.tracking.add(command, track)
	p.logger.Debugf("P4 tracking %s: lapse=%s user=%s system=%s", command, track.Lapse, track.UserCPU, track.SystemCPU)
	for table, locks := range track.Tables {
		p.logger.Debugf("P4 tracking %s %s: read wait/held=%s/%s write wait/held=%s/%s", command, table, locks.ReadWait, locks.ReadHeld, locks.WriteWait, locks.WriteHeld)
	}
	for library, usage := range track.Libraries {
		p.logger.Debugf("P4 tracking %s lbr %s: opens/closes/checkins/exists=%d/%d/%d/%d", command, library, usage.Opens, usage.Closes, usage.Checkins, usage.Exists)
	}
}

func missingRevision(line string) (string, bool) {
	matches := missingRevisionPattern.FindStringSubmatch(line)
	if matches == nil {
		return "", false
	}
	return matches[1], true
}

func (p p4Client) discoverMissing(ctx context.Context, filespec string, output io.Writer) (int64, error) {
	filespecs := strings.Fields(filespec)
	if len(filespecs) == 0 {
		return 0, errors.New("verify requires at least one depot path")
	}
	args := append([]string{"verify", "-q", "--only", "MISSING"}, filespecs...)
	commandArgs := p.commandArgs(args...)
	p.logger.Debugf("Running: %s %s", p.bin, strings.Join(commandArgs, " "))
	command := exec.CommandContext(ctx, p.bin, commandArgs...)
	stdout, err := command.StdoutPipe()
	if err != nil {
		return 0, err
	}
	stderr, err := command.StderrPipe()
	if err != nil {
		return 0, err
	}
	if err := command.Start(); err != nil {
		return 0, err
	}
	type scanResult struct {
		found       int64
		diagnostics string
		tracking    string
		err         error
	}
	results := make(chan scanResult, 2)
	var outputMu sync.Mutex
	scan := func(reader io.Reader) {
		result := scanResult{}
		var diagnostics strings.Builder
		var tracking strings.Builder
		scanner := bufio.NewScanner(reader)
		scanner.Buffer(make([]byte, 64*1024), 4*1024*1024)
		for scanner.Scan() {
			line := scanner.Text()
			if strings.HasPrefix(line, "--- ") {
				tracking.WriteString(line)
				tracking.WriteByte('\n')
				continue
			}
			if revision, ok := missingRevision(line); ok {
				if result.err == nil {
					outputMu.Lock()
					_, result.err = fmt.Fprintln(output, revision)
					outputMu.Unlock()
				}
				result.found++
				continue
			}
			if strings.TrimSpace(line) != "" {
				diagnostics.WriteString(line)
				diagnostics.WriteByte('\n')
			}
		}
		if err := scanner.Err(); err != nil && result.err == nil {
			result.err = err
		}
		result.diagnostics = diagnostics.String()
		result.tracking = tracking.String()
		results <- result
	}
	go scan(stdout)
	go scan(stderr)
	first, second := <-results, <-results
	found := first.found + second.found
	waitErr := command.Wait()
	p.recordTrack(args, first.tracking+second.tracking)
	if first.err != nil {
		return found, first.err
	}
	if second.err != nil {
		return found, second.err
	}
	if waitErr != nil {
		var exitError *exec.ExitError
		if errors.As(waitErr, &exitError) && exitError.ExitCode() == 1 && found > 0 {
			return found, nil
		}
		diagnostics := strings.TrimSpace(first.diagnostics + second.diagnostics)
		return found, fmt.Errorf("p4 %s: %w: %s", strings.Join(args, " "), waitErr, diagnostics)
	}
	return found, nil
}

func (p p4Client) localOrStreamDepots(ctx context.Context) ([]depot, error) {
	output, err := p.run(ctx, "-ztag", "depots")
	if err != nil {
		return nil, err
	}
	var depots []depot
	current := depot{}
	addDepot := func() {
		if (current.typeName == "local" || current.typeName == "stream") && current.name != "" {
			depots = append(depots, current)
		}
	}
	for _, line := range strings.Split(string(output), "\n") {
		fields := strings.Fields(line)
		if len(fields) < 3 || fields[0] != "..." {
			continue
		}
		key := strings.ToLower(fields[1])
		value := strings.Join(fields[2:], " ")
		switch key {
		case "depot", "name":
			addDepot()
			current = depot{name: value}
		case "type":
			current.typeName = strings.ToLower(value)
		case "streamdepth", "depth":
			depth, err := strconv.Atoi(value)
			if err != nil || depth < 1 {
				return nil, fmt.Errorf("invalid stream depth %q for depot %q", value, current.name)
			}
			current.streamDepth = depth
		}
	}
	addDepot()
	return depots, nil
}

func (p p4Client) directories(ctx context.Context, filespec string) ([]string, error) {
	output, err := p.run(ctx, "dirs", "-D", filespec)
	if err != nil {
		return nil, err
	}
	var directories []string
	for _, line := range strings.Split(string(output), "\n") {
		directory := strings.TrimSpace(line)
		if strings.Contains(directory, " - no such file(s).") {
			p.logger.Warnf("Ignoring p4 dirs diagnostic: %s", directory)
			continue
		}
		if strings.HasPrefix(directory, "//") {
			directories = append(directories, directory)
		}
	}
	return directories, nil
}

func (p p4Client) streamDirectories(ctx context.Context, name string, depth int) ([]string, error) {
	if depth < 1 {
		return nil, nil
	}
	pattern := "//" + name + "/" + strings.TrimSuffix(strings.Repeat("*/", depth), "/")
	output, err := p.run(ctx, "dirs", pattern)
	if err != nil {
		return nil, err
	}
	var directories []string
	for _, line := range strings.Split(string(output), "\n") {
		directory := strings.TrimSpace(line)
		if strings.Contains(directory, " - no such file(s).") {
			p.logger.Warnf("Ignoring p4 dirs diagnostic: %s", directory)
			continue
		}
		if strings.HasPrefix(directory, "//"+name+"/") {
			directories = append(directories, directory)
		}
	}
	return directories, nil
}

func (p p4Client) sizesWithinTimeout(ctx context.Context, filespec string, timeout time.Duration) (bool, error) {
	timedContext, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	_, err := p.run(timedContext, "sizes", "-sh", filespec)
	if err == nil {
		return true, nil
	}
	if errors.Is(timedContext.Err(), context.DeadlineExceeded) {
		return false, nil
	}
	return false, err
}

func (p p4Client) filesWithinTimeout(ctx context.Context, filespec string, timeout time.Duration) (int64, error) {
	timedContext, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	output, err := p.run(timedContext, "sizes", "-sh", filespec)
	if err != nil {
		if errors.Is(timedContext.Err(), context.DeadlineExceeded) {
			return 0, fmt.Errorf("p4 sizes timed out for %q", filespec)
		}
		return 0, err
	}
	matches := sizesFileCountPattern.FindStringSubmatch(string(output))
	if matches == nil {
		return 0, fmt.Errorf("could not parse file count from p4 sizes output for %q", filespec)
	}
	files, err := strconv.ParseInt(strings.ReplaceAll(matches[1], ",", ""), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("parse p4 sizes file count for %q: %w", filespec, err)
	}
	return files, nil
}

func groupStreamPaths(name string, directories []string, depth int, averageFiles, maxFiles int64) ([]string, error) {
	if len(directories) == 0 {
		return nil, nil
	}
	prefix := "//" + name + "/"
	partsByDirectory := make(map[string][]string, len(directories))
	for _, directory := range directories {
		relative := strings.TrimPrefix(directory, prefix)
		parts := strings.Split(strings.Trim(relative, "/"), "/")
		if relative == directory || len(parts) != depth {
			return nil, fmt.Errorf("stream directory %q does not match depth %d", directory, depth)
		}
		partsByDirectory[directory] = parts
	}
	type streamGroup struct {
		filespec string
		streams  int
	}
	var grouped []streamGroup
	var split func([]string, int)
	split = func(candidates []string, level int) {
		groups := make(map[string][]string)
		for _, candidate := range candidates {
			key := strings.Join(partsByDirectory[candidate][:level], "/")
			groups[key] = append(groups[key], candidate)
		}
		keys := make([]string, 0, len(groups))
		for key := range groups {
			keys = append(keys, key)
		}
		sort.Strings(keys)
		for _, key := range keys {
			candidates := groups[key]
			if int64(len(candidates))*averageFiles <= maxFiles || level == depth {
				grouped = append(grouped, streamGroup{filespec: prefix + key + "/...", streams: len(candidates)})
				continue
			}
			split(candidates, level+1)
		}
	}
	split(directories, 1)

	var batches []string
	var batch []string
	var batchFiles int64
	flushBatch := func() {
		if len(batch) == 0 {
			return
		}
		batches = append(batches, strings.Join(batch, " "))
		batch = nil
		batchFiles = 0
	}
	for _, group := range grouped {
		groupFiles := int64(group.streams) * averageFiles
		if len(batch) > 0 && batchFiles+groupFiles > maxFiles {
			flushBatch()
		}
		batch = append(batch, group.filespec)
		batchFiles += groupFiles
	}
	flushBatch()
	return batches, nil
}

func (p p4Client) generateStreamPaths(ctx context.Context, stream depot, timeout time.Duration, maxFiles int64, samples int) ([]string, error) {
	directories, err := p.streamDirectories(ctx, stream.name, stream.streamDepth)
	if err != nil {
		return nil, err
	}
	if len(directories) == 0 {
		return nil, nil
	}
	if samples > len(directories) {
		samples = len(directories)
	}
	random := rand.New(rand.NewSource(time.Now().UnixNano()))
	indices := random.Perm(len(directories))[:samples]
	var totalFiles int64
	for _, index := range indices {
		files, err := p.filesWithinTimeout(ctx, strings.TrimSuffix(directories[index], "/")+"/...", timeout)
		if err != nil {
			return nil, err
		}
		totalFiles += files
	}
	averageFiles := (totalFiles + int64(samples) - 1) / int64(samples)
	return groupStreamPaths(stream.name, directories, stream.streamDepth, averageFiles, maxFiles)
}

func generateDepotPaths(ctx context.Context, p4 p4Client, output string, timeout time.Duration, maxStreamFiles int64, streamSizeSamples int) (int, error) {
	return generateDepotPathsWithWorkers(ctx, p4, output, timeout, maxStreamFiles, streamSizeSamples, 10)
}

func (p p4Client) generateLocalDepotPaths(ctx context.Context, name string, timeout time.Duration) ([]string, error) {
	paths := []string{"//" + name + "/..."}
	var generated []string
	for len(paths) > 0 {
		path := paths[0]
		paths = paths[1:]
		complete, err := p.sizesWithinTimeout(ctx, path, timeout)
		if err != nil {
			return nil, err
		}
		if complete {
			generated = append(generated, path)
			continue
		}
		directories, err := p.directories(ctx, strings.TrimSuffix(path, "/...")+"/*")
		if err != nil {
			return nil, err
		}
		if len(directories) == 0 {
			return nil, fmt.Errorf("cannot subdivide timed-out depot path %q", path)
		}
		for _, directory := range directories {
			paths = append(paths, strings.TrimSuffix(directory, "/")+"/...")
		}
	}
	return generated, nil
}

func generateDepotPathsWithWorkers(ctx context.Context, p4 p4Client, output string, timeout time.Duration, maxStreamFiles int64, streamSizeSamples, workers int) (int, error) {
	if timeout <= 0 {
		return 0, errors.New("path timeout must be positive")
	}
	if maxStreamFiles < 1 || streamSizeSamples < 1 || workers < 1 {
		return 0, errors.New("maximum stream files, stream size samples, and path workers must be positive")
	}
	if _, err := os.Stat(output); err == nil {
		return 0, fmt.Errorf("depot path output %q already exists", output)
	} else if !errors.Is(err, os.ErrNotExist) {
		return 0, err
	}
	depots, err := p4.localOrStreamDepots(ctx)
	if err != nil {
		return 0, err
	}
	if workers > len(depots) {
		workers = len(depots)
	}
	generatedByDepot := make([][]string, len(depots))
	workerContext, cancel := context.WithCancel(ctx)
	defer cancel()
	jobs := make(chan int)
	errorsCh := make(chan error, 1)
	var workersDone sync.WaitGroup
	for range workers {
		workersDone.Add(1)
		go func() {
			defer workersDone.Done()
			for index := range jobs {
				depot := depots[index]
				var paths []string
				var err error
				if depot.typeName == "stream" {
					paths, err = p4.generateStreamPaths(workerContext, depot, timeout, maxStreamFiles, streamSizeSamples)
				} else {
					paths, err = p4.generateLocalDepotPaths(workerContext, depot.name, timeout)
				}
				if err != nil {
					select {
					case errorsCh <- err:
						cancel()
					default:
					}
					return
				}
				generatedByDepot[index] = paths
			}
		}()
	}
dispatch:
	for index := range depots {
		select {
		case jobs <- index:
		case <-workerContext.Done():
			break dispatch
		}
	}
	close(jobs)
	workersDone.Wait()
	select {
	case err := <-errorsCh:
		return 0, err
	default:
	}
	if ctx.Err() != nil {
		return 0, ctx.Err()
	}
	var generated []string
	for _, paths := range generatedByDepot {
		generated = append(generated, paths...)
	}
	outputDir := filepath.Dir(output)
	if err := os.MkdirAll(outputDir, 0755); err != nil && outputDir != "." {
		return 0, err
	}
	temporary, err := os.CreateTemp(outputDir, ".seedreplica-paths-*.txt")
	if err != nil {
		return 0, err
	}
	temporaryName := temporary.Name()
	defer os.Remove(temporaryName)
	for _, path := range generated {
		if _, err := fmt.Fprintln(temporary, path); err != nil {
			temporary.Close()
			return 0, err
		}
	}
	if err := temporary.Close(); err != nil {
		return 0, err
	}
	if err := os.Rename(temporaryName, output); err != nil {
		return 0, err
	}
	return len(generated), nil
}

func discoverRevisions(ctx context.Context, p4 p4Client, filespecs, output string, workers int, replace bool) (int64, error) {
	if workers < 1 {
		return 0, errors.New("discovery workers must be positive")
	}
	if _, err := os.Stat(output); err == nil && !replace {
		return 0, fmt.Errorf("discovery output %q already exists; use --replace-input to overwrite it", output)
	} else if err != nil && !errors.Is(err, os.ErrNotExist) {
		return 0, err
	}
	input, err := os.Open(filespecs)
	if err != nil {
		return 0, err
	}
	defer input.Close()
	outputDir := filepath.Dir(output)
	if err := os.MkdirAll(outputDir, 0755); err != nil && outputDir != "." {
		return 0, err
	}

	workerFiles := make([]*os.File, workers)
	for index := range workerFiles {
		workerFiles[index], err = os.CreateTemp(outputDir, ".seedreplica-discovery-*.txt")
		if err != nil {
			for _, file := range workerFiles {
				if file != nil {
					file.Close()
					os.Remove(file.Name())
				}
			}
			return 0, err
		}
		defer os.Remove(workerFiles[index].Name())
	}

	workerContext, cancel := context.WithCancel(ctx)
	defer cancel()
	jobs := make(chan string)
	errorsCh := make(chan error, 1)
	counts := make([]int64, workers)
	var workersDone sync.WaitGroup
	for index, file := range workerFiles {
		workersDone.Add(1)
		go func(index int, file *os.File) {
			defer workersDone.Done()
			writer := bufio.NewWriter(file)
			defer writer.Flush()
			for {
				select {
				case <-workerContext.Done():
					return
				case filespec, ok := <-jobs:
					if !ok {
						return
					}
					count, err := p4.discoverMissing(workerContext, filespec, writer)
					if err != nil {
						select {
						case errorsCh <- err:
							cancel()
						default:
						}
						return
					}
					counts[index] += count
				}
			}
		}(index, file)
	}

	scanner := bufio.NewScanner(input)
	scanner.Buffer(make([]byte, 64*1024), 4*1024*1024)
	var producerErr error
	for scanner.Scan() {
		filespecs := strings.Fields(scanner.Text())
		if len(filespecs) == 0 {
			continue
		}
		for _, filespec := range filespecs {
			if !strings.HasPrefix(filespec, "//") {
				producerErr = fmt.Errorf("discovery input contains an invalid depot path: %q", filespec)
				cancel()
				break
			}
		}
		if producerErr != nil {
			break
		}
		select {
		case jobs <- strings.Join(filespecs, " "):
		case <-workerContext.Done():
		}
		if workerContext.Err() != nil {
			break
		}
	}
	scanErr := scanner.Err()
	close(jobs)
	workersDone.Wait()
	for _, file := range workerFiles {
		if err := file.Close(); err != nil {
			return 0, err
		}
	}
	if scanErr != nil {
		return 0, scanErr
	}
	if producerErr != nil {
		return 0, producerErr
	}
	select {
	case err := <-errorsCh:
		return 0, err
	default:
	}
	if ctx.Err() != nil {
		return 0, ctx.Err()
	}

	temporary, err := os.CreateTemp(outputDir, ".seedreplica-input-*.txt")
	if err != nil {
		return 0, err
	}
	temporaryName := temporary.Name()
	defer os.Remove(temporaryName)
	var total int64
	for index, file := range workerFiles {
		contents, err := os.ReadFile(file.Name())
		if err != nil {
			temporary.Close()
			return 0, err
		}
		if _, err := temporary.Write(contents); err != nil {
			temporary.Close()
			return 0, err
		}
		total += counts[index]
	}
	if err := temporary.Close(); err != nil {
		return 0, err
	}
	if replace {
		if err := os.Remove(output); err != nil && !errors.Is(err, os.ErrNotExist) {
			return 0, err
		}
	}
	if err := os.Rename(temporaryName, output); err != nil {
		return 0, err
	}
	return total, nil
}

func (p p4Client) queueStatus(ctx context.Context) (queueStatus, error) {
	output, err := p.run(ctx, "pull", "-lsh")
	if err != nil {
		return queueStatus{}, err
	}
	return parseQueueStatus(string(output))
}

func parseQueueStatus(output string) (queueStatus, error) {
	matches := pullQueuePattern.FindStringSubmatch(output)
	if matches == nil {
		return queueStatus{}, fmt.Errorf("could not parse p4 pull -lsh output: %q", strings.TrimSpace(output))
	}
	active, activeErr := strconv.Atoi(matches[1])
	total, totalErr := strconv.Atoi(matches[2])
	if activeErr != nil || totalErr != nil {
		return queueStatus{}, errors.New("invalid transfer counts in p4 pull -lsh output")
	}
	return queueStatus{ActiveTransfers: active, TotalTransfers: total, ActiveBytes: matches[3], TotalBytes: matches[4]}, nil
}

func readBatch(filename string, offset, linesRead int64, batchSize int) ([]string, int64, int64, bool, error) {
	input, err := os.Open(filename)
	if err != nil {
		return nil, offset, linesRead, false, err
	}
	defer input.Close()
	if _, err := input.Seek(offset, io.SeekStart); err != nil {
		return nil, offset, linesRead, false, err
	}

	reader := bufio.NewReader(input)
	batch := make([]string, 0, batchSize)
	for len(batch) < batchSize {
		line, readErr := reader.ReadString('\n')
		if len(line) > 0 {
			offset += int64(len(line))
			linesRead++
			value := strings.TrimSpace(line)
			if value != "" {
				if !strings.HasPrefix(value, "//") || !strings.Contains(value, "#") {
					return nil, offset, linesRead, false, fmt.Errorf("input line %d is not a depot revision: %q", linesRead, value)
				}
				batch = append(batch, value)
			}
		}
		if readErr == io.EOF {
			return batch, offset, linesRead, true, nil
		}
		if readErr != nil {
			return nil, offset, linesRead, false, readErr
		}
	}
	return batch, offset, linesRead, false, nil
}

func writeCheckpoint(filename string, state checkpoint) error {
	state.UpdatedAt = time.Now().UTC()
	encoded, err := json.MarshalIndent(state, "", "  ")
	if err != nil {
		return err
	}
	if err := os.MkdirAll(filepath.Dir(filename), 0755); err != nil && filepath.Dir(filename) != "." {
		return err
	}
	temporary, err := os.CreateTemp(filepath.Dir(filename), ".seedreplica-*.json")
	if err != nil {
		return err
	}
	temporaryName := temporary.Name()
	defer os.Remove(temporaryName)
	if _, err := temporary.Write(append(encoded, '\n')); err != nil {
		temporary.Close()
		return err
	}
	if err := temporary.Close(); err != nil {
		return err
	}
	return os.Rename(temporaryName, filename)
}

func loadCheckpoint(filename, input string) (checkpoint, error) {
	encoded, err := os.ReadFile(filename)
	if err != nil {
		return checkpoint{}, err
	}
	var state checkpoint
	if err := json.Unmarshal(encoded, &state); err != nil {
		return checkpoint{}, fmt.Errorf("decode checkpoint: %w", err)
	}
	if state.Input != input {
		return checkpoint{}, fmt.Errorf("checkpoint input %q does not match %q", state.Input, input)
	}
	return state, nil
}

func verifyBatch(ctx context.Context, p4 p4Client, batch []string) (int, error) {
	temporary, err := os.CreateTemp("", "seedreplica-batch-*.txt")
	if err != nil {
		return 0, err
	}
	temporaryName := temporary.Name()
	defer os.Remove(temporaryName)
	if _, err := temporary.WriteString(strings.Join(batch, "\n") + "\n"); err != nil {
		temporary.Close()
		return 0, err
	}
	if err := temporary.Close(); err != nil {
		return 0, err
	}
	output, err := p4.run(ctx, "-x", temporaryName, "verify", "-qt", "--only", "MISSING")
	if err != nil {
		return 0, err
	}
	transfers := 0
	for _, line := range strings.Split(string(output), "\n") {
		if strings.Contains(line, "BAD") || strings.Contains(line, "MISSING") {
			transfers++
		}
	}
	return transfers, nil
}

func waitForQueue(ctx context.Context, p4 p4Client, maxQueue int, interval time.Duration) error {
	for {
		status, err := p4.queueStatus(ctx)
		if err != nil {
			return err
		}
		p4.logger.Infof("Pull queue: %d active/%d total transfers, %s active/%s total bytes", status.ActiveTransfers, status.TotalTransfers, status.ActiveBytes, status.TotalBytes)
		if status.TotalTransfers <= maxQueue {
			return nil
		}
		p4.logger.Infof("Pull queue size %d exceeds threshold %d; waiting %s", status.TotalTransfers, maxQueue, interval)
		timer := time.NewTimer(interval)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
}

func runSeed(ctx context.Context, cfg config, logger *logrus.Logger, stdout io.Writer) error {
	tracking := &trackCollector{}
	defer tracking.writeSummary(stdout, "seed")
	input, err := filepath.Abs(cfg.input)
	if err != nil {
		return err
	}
	if cfg.discoverInput != "" {
		filespecs, err := filepath.Abs(cfg.discoverInput)
		if err != nil {
			return err
		}
		p4 := p4Client{bin: cfg.p4bin, port: cfg.p4port, user: cfg.p4user, logger: logger, tracking: tracking}
		logger.Infof("Discovering missing revisions from %s with %d workers", filespecs, cfg.discoverers)
		found, err := discoverRevisions(ctx, p4, filespecs, input, cfg.discoverers, cfg.replaceInput)
		if err != nil {
			return fmt.Errorf("discover missing revisions: %w", err)
		}
		logger.Infof("Discovery wrote %d missing revisions to %s", found, input)
		if cfg.discoverOnly {
			return nil
		}
	}
	state := checkpoint{Input: input}
	if cfg.resume {
		state, err = loadCheckpoint(cfg.checkpoint, input)
		if err != nil {
			return err
		}
		logger.Infof("Resuming after %d processed revisions in %d batches", state.Processed, state.Batches)
	}
	p4 := p4Client{bin: cfg.p4bin, port: cfg.p4port, user: cfg.p4user, logger: logger, tracking: tracking}

	for {
		batch, offset, linesRead, eof, err := readBatch(input, state.Offset, state.LinesRead, cfg.batchSize)
		if err != nil {
			return err
		}
		if len(batch) == 0 {
			if !cfg.follow || !eof {
				state.CompletedAt = time.Now().UTC()
				return writeCheckpoint(cfg.checkpoint, state)
			}
			logger.Infof("Waiting for additional input in %s", input)
			timer := time.NewTimer(cfg.pollInterval)
			select {
			case <-ctx.Done():
				timer.Stop()
				return ctx.Err()
			case <-timer.C:
			}
			continue
		}

		batchNumber := state.Batches + 1
		transfers := 0
		if cfg.dryRun {
			logger.Infof("Dry run: batch %d contains %d revisions", batchNumber, len(batch))
		} else {
			transfers, err = verifyBatch(ctx, p4, batch)
			if err != nil {
				return fmt.Errorf("batch %d: %w", batchNumber, err)
			}
		}
		state.Offset = offset
		state.LinesRead = linesRead
		state.Processed += int64(len(batch))
		state.Batches++
		if err := writeCheckpoint(cfg.checkpoint, state); err != nil {
			return err
		}
		logger.Infof("Batch %d queued %d transfers from %d revisions (%d total processed)", batchNumber, transfers, len(batch), state.Processed)

		if !cfg.dryRun {
			if err := waitForQueue(ctx, p4, cfg.maxQueue, cfg.pollInterval); err != nil {
				return err
			}
		}
		if cfg.maxBatches > 0 && state.Batches >= cfg.maxBatches {
			logger.Infof("Reached maximum batch limit %d", cfg.maxBatches)
			return nil
		}
	}
}

func commandApp(name string, stderr io.Writer) *kingpin.Application {
	app := kingpin.New(name, description)
	app.ErrorWriter(stderr)
	app.UsageWriter(stderr)
	app.HelpFlag.Short('h')
	app.Version(VersionPrint(name, "v"+Version))
	app.VersionFlag.Short('V')
	return app
}

func commandLogger(stderr io.Writer, debug bool) *logrus.Logger {
	logger := logrus.New()
	logger.SetOutput(stderr)
	if debug {
		logger.SetLevel(logrus.DebugLevel)
	}
	return logger
}

func runPaths(args []string, stdout, stderr io.Writer) error {
	app := commandApp("p4seedreplica discover", stderr)
	output := app.Flag("output", "File to receive generated depot paths").Short('o').Required().String()
	p4bin := app.Flag("p4bin", "Path to p4 executable").Default("p4").String()
	p4port := app.Flag("p4port", "P4PORT override").Short('p').String()
	p4user := app.Flag("p4user", "P4USER override").Short('u').String()
	pathTimeout := app.Flag("path-timeout", "Maximum duration for each p4 sizes command").Default("5m").Duration()
	maxStreamFiles := app.Flag("max-stream-files", "Estimated maximum files in each generated stream path").Default("1000000").Int64()
	streamSizeSamples := app.Flag("stream-size-samples", "Random stream paths sampled to estimate file counts").Default("10").Int()
	pathWorkers := app.Flag("path-workers", "Parallel depot path generation workers").Default("10").Int()
	debug := app.Flag("debug", "Print p4 commands").Bool()
	if _, err := app.Parse(args); err != nil {
		return err
	}
	if *pathTimeout <= 0 || *maxStreamFiles < 1 || *streamSizeSamples < 1 || *pathWorkers < 1 {
		return errors.New("path timeout, maximum stream files, stream size samples, and path workers must be positive")
	}
	logger := commandLogger(stderr, *debug)
	tracking := &trackCollector{}
	defer tracking.writeSummary(stdout, "discover")
	count, err := generateDepotPathsWithWorkers(context.Background(), p4Client{bin: *p4bin, port: *p4port, user: *p4user, logger: logger, tracking: tracking}, *output, *pathTimeout, *maxStreamFiles, *streamSizeSamples, *pathWorkers)
	if err != nil {
		return fmt.Errorf("generate depot paths: %w", err)
	}
	fmt.Fprintf(stdout, "Wrote %d depot paths to %s\n", count, *output)
	return nil
}

func runVerify(args []string, stdout, stderr io.Writer) error {
	app := commandApp("p4seedreplica verify", stderr)
	paths := app.Flag("paths", "File of depot path specs to scan").Required().String()
	output := app.Flag("output", "File to receive missing depot revisions").Short('o').Required().String()
	p4bin := app.Flag("p4bin", "Path to p4 executable").Default("p4").String()
	p4port := app.Flag("p4port", "P4PORT override").Short('p').String()
	p4user := app.Flag("p4user", "P4USER override").Short('u').String()
	workers := app.Flag("workers", "Parallel missing-file discovery workers").Default("10").Int()
	replace := app.Flag("replace", "Allow replacing an existing output file").Bool()
	debug := app.Flag("debug", "Print p4 commands").Bool()
	if _, err := app.Parse(args); err != nil {
		return err
	}
	if *workers < 1 {
		return errors.New("discovery workers must be positive")
	}
	logger := commandLogger(stderr, *debug)
	tracking := &trackCollector{}
	defer tracking.writeSummary(stdout, "verify")
	pathsFile, err := filepath.Abs(*paths)
	if err != nil {
		return err
	}
	outputFile, err := filepath.Abs(*output)
	if err != nil {
		return err
	}
	logger.Infof("Verifying depot paths from %s with %d workers", pathsFile, *workers)
	found, err := discoverRevisions(context.Background(), p4Client{bin: *p4bin, port: *p4port, user: *p4user, logger: logger, tracking: tracking}, pathsFile, outputFile, *workers, *replace)
	if err != nil {
		return fmt.Errorf("verify missing revisions: %w", err)
	}
	fmt.Fprintf(stdout, "Wrote %d missing revisions to %s\n", found, outputFile)
	return nil
}

func runSeedCommand(args []string, stdout, stderr io.Writer) error {
	app := commandApp("p4seedreplica seed", stderr)
	input := app.Flag("input", "File of depot revisions, one //depot/file#rev per line").Required().String()
	p4bin := app.Flag("p4bin", "Path to p4 executable").Default("p4").String()
	p4port := app.Flag("p4port", "P4PORT override").Short('p').String()
	p4user := app.Flag("p4user", "P4USER override").Short('u').String()
	batchSize := app.Flag("batch-size", "Revisions submitted per verify command").Default("10000").Int()
	maxQueue := app.Flag("max-pull-queue", "Maximum queued archive transfers before waiting").Default("100000").Int()
	pollInterval := app.Flag("poll-interval", "Queue polling interval").Default("30s").Duration()
	maxBatches := app.Flag("max-batches", "Stop after this many batches; zero means unlimited").Default("0").Int()
	checkpointPath := app.Flag("checkpoint", "Checkpoint file (default: input.seedreplica.json)").String()
	resume := app.Flag("resume", "Resume from checkpoint").Bool()
	follow := app.Flag("follow", "Wait for appended input after reaching end of file").Bool()
	dryRun := app.Flag("dry-run", "Validate and checkpoint batches without running p4").Bool()
	debug := app.Flag("debug", "Print p4 commands").Bool()
	if _, err := app.Parse(args); err != nil {
		return err
	}
	if *batchSize < 1 || *maxQueue < 1 || *pollInterval <= 0 || *maxBatches < 0 {
		return errors.New("batch size, maximum queue, and polling interval must be positive; maximum batches cannot be negative")
	}
	checkpoint := *checkpointPath
	if checkpoint == "" {
		checkpoint = *input + ".seedreplica.json"
	}
	return runSeed(context.Background(), config{
		input: *input, p4bin: *p4bin, p4port: *p4port, p4user: *p4user, batchSize: *batchSize,
		maxQueue: *maxQueue, pollInterval: *pollInterval, maxBatches: *maxBatches, checkpoint: checkpoint,
		resume: *resume, follow: *follow, dryRun: *dryRun,
	}, commandLogger(stderr, *debug), stdout)
}

func workflowUsage(stderr io.Writer) {
	fmt.Fprintln(stderr, `usage: p4seedreplica <command> [<flags>]

Commands reflect the 3 phases of the process:
    discover   Generate bounded depot paths for verification.
    verify     Find missing archive revisions from depot paths using verify command.
    seed       Submit missing revisions to the replica pull queue.

Run "p4seedreplica <command> --help" for command options.`)
}

func run(args []string, stdout, stderr io.Writer) error {
	if len(args) == 0 || args[0] == "help" || args[0] == "--help" || args[0] == "-h" {
		workflowUsage(stderr)
		return nil
	}
	switch args[0] {
	case "discover":
		return runPaths(args[1:], stdout, stderr)
	case "verify":
		return runVerify(args[1:], stdout, stderr)
	case "seed":
		return runSeedCommand(args[1:], stdout, stderr)
	default:
		return runLegacy(args, stdout, stderr)
	}
}

func runLegacy(args []string, stdout, stderr io.Writer) error {
	app := commandApp("p4seedreplica", stderr)
	input := app.Flag("input", "File of depot revisions, one //depot/file#rev per line").String()
	p4bin := app.Flag("p4bin", "Path to p4 executable").Default("p4").String()
	p4port := app.Flag("p4port", "P4PORT override").Short('p').String()
	p4user := app.Flag("p4user", "P4USER override").Short('u').String()
	batchSize := app.Flag("batch-size", "Revisions submitted per verify command").Default("10000").Int()
	maxQueue := app.Flag("max-pull-queue", "Maximum queued archive transfers before waiting").Default("100000").Int()
	pollInterval := app.Flag("poll-interval", "Queue and follow polling interval").Default("30s").Duration()
	maxBatches := app.Flag("max-batches", "Stop after this many batches; zero means unlimited").Default("0").Int()
	checkpointPath := app.Flag("checkpoint", "Checkpoint file (default: input.seedreplica.json)").String()
	resume := app.Flag("resume", "Resume from checkpoint").Bool()
	follow := app.Flag("follow", "Wait for appended input after reaching end of file").Bool()
	dryRun := app.Flag("dry-run", "Validate and checkpoint batches without running p4").Bool()
	discoverInput := app.Flag("discover-input", "File of depot path specs to scan in parallel for missing revisions").String()
	discoverOnly := app.Flag("discover-only", "Stop after writing revisions discovered from --discover-input").Bool()
	discoverWorkers := app.Flag("discover-workers", "Parallel missing-file discovery workers").Default("10").Int()
	replaceInput := app.Flag("replace-input", "Allow discovery to replace an existing --input file").Bool()
	depotPaths := app.Flag("generate-depot-paths", "Write recursively sized local and stream depot paths to this file").String()
	pathTimeout := app.Flag("path-timeout", "Maximum duration for each p4 sizes command while generating depot paths").Default("5m").Duration()
	maxStreamFiles := app.Flag("max-stream-files", "Estimated maximum files in each generated stream path").Default("1000000").Int64()
	streamSizeSamples := app.Flag("stream-size-samples", "Random stream paths sampled to estimate file counts").Default("10").Int()
	pathWorkers := app.Flag("path-workers", "Parallel depot path generation workers").Default("10").Int()
	debug := app.Flag("debug", "Print p4 commands").Bool()
	if _, err := app.Parse(args); err != nil {
		return err
	}
	if *batchSize < 1 || *maxQueue < 1 || *pollInterval <= 0 || *maxBatches < 0 || *discoverWorkers < 1 || *pathTimeout <= 0 || *maxStreamFiles < 1 || *streamSizeSamples < 1 || *pathWorkers < 1 {
		return errors.New("batch size, maximum queue, and polling interval must be positive; maximum batches cannot be negative")
	}
	if *input == "" && *depotPaths == "" {
		return errors.New("--input is required unless --generate-depot-paths is used")
	}
	if *depotPaths != "" {
		if *discoverInput != "" || *discoverOnly || *resume || *follow || *dryRun {
			return errors.New("--generate-depot-paths cannot be combined with discovery or seeding options")
		}
		logger := logrus.New()
		logger.SetOutput(stderr)
		if *debug {
			logger.SetLevel(logrus.DebugLevel)
		}
		tracking := &trackCollector{}
		defer tracking.writeSummary(stdout, "discover")
		count, err := generateDepotPathsWithWorkers(context.Background(), p4Client{bin: *p4bin, port: *p4port, user: *p4user, logger: logger, tracking: tracking}, *depotPaths, *pathTimeout, *maxStreamFiles, *streamSizeSamples, *pathWorkers)
		if err != nil {
			return fmt.Errorf("generate depot paths: %w", err)
		}
		fmt.Fprintf(stdout, "Wrote %d depot paths to %s\n", count, *depotPaths)
		return nil
	}
	if *discoverOnly && *discoverInput == "" {
		return errors.New("--discover-only requires --discover-input")
	}
	if *dryRun && *discoverInput != "" {
		return errors.New("--dry-run cannot be combined with --discover-input")
	}
	if *resume && *discoverInput != "" {
		return errors.New("--resume cannot be combined with --discover-input")
	}
	checkpoint := *checkpointPath
	if checkpoint == "" {
		checkpoint = *input + ".seedreplica.json"
	}
	logger := logrus.New()
	logger.SetOutput(stderr)
	if *debug {
		logger.SetLevel(logrus.DebugLevel)
	}
	return runSeed(context.Background(), config{
		input: *input, p4bin: *p4bin, p4port: *p4port, p4user: *p4user, batchSize: *batchSize,
		maxQueue: *maxQueue, pollInterval: *pollInterval, maxBatches: *maxBatches, checkpoint: checkpoint,
		resume: *resume, follow: *follow, dryRun: *dryRun, discoverInput: *discoverInput,
		discoverOnly: *discoverOnly, discoverers: *discoverWorkers, replaceInput: *replaceInput,
	}, logger, stdout)
}

func main() {
	if err := run(os.Args[1:], os.Stdout, os.Stderr); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}