package acp import ( "context" "fmt" "os" "path" "path/filepath" "sort" "strings" ) // UnexpectFileMode is the file-mode mask a walk refuses to descend into. const UnexpectFileMode = os.ModeType &^ os.ModeDir // Copyer is the af05f05c compatibility shell: the same job options, events and JSON report as // the commit that predates the streaming API, built on the push engine. // // New enumerates the requested jobs, submits every selected file to a StreamCopyer, and // translates each result into the terminal report row event the CLI, Report and // NewReportGetter machinery reads. Wait/WaitErr close that stream and report how the run ended. type Copyer struct { *option stream *StreamCopyer // feedErr is written by the feed goroutine before it closes feedDone. feedErr error feedDone chan struct{} } // New builds a shell run. Creation and validation errors, such as a target that is not a // directory or a source that does not exist, come back from New; an enumeration error, such as // two sources mapping to the same relative path, ends the run and comes back from WaitErr. func New(ctx context.Context, opts ...Option) (*Copyer, error) { opt, err := buildOption(opts...) if err != nil { return nil, err } c := &Copyer{option: opt, feedDone: make(chan struct{})} stream, err := newStream(ctx, c.reportResults, opt) if err != nil { return nil, err } c.stream = stream // The walk and the feed run beside the pipeline, exactly like the af05f05c index stage. go c.feed(ctx) return c, nil } // Wait blocks until the run finished. Use WaitErr to learn how it ended. func (c *Copyer) Wait() { _ = c.WaitErr() } // WaitErr closes the run, drains the pipeline and returns its terminal error. An item-level // failure is not a run error: it travels in that item's report row. func (c *Copyer) WaitErr() error { <-c.feedDone c.stream.setError(c.feedErr) _ = c.stream.Close() return c.stream.Wait() } // feed enumerates the job options and submits everything as one batch. One batch matters: the // engine keeps every accepted item, so a run stopped midway still reports every selected file // instead of losing the ones the feed had not reached. func (c *Copyer) feed(ctx context.Context) { defer close(c.feedDone) // End the feed even when the caller never waits, so the run finishes on its own. defer func() { _ = c.stream.Close() }() items, err := c.selectItems(ctx) if err != nil { c.feedErr = err return } if len(items) == 0 { return } // The engine records why a submission stopped, so Wait reports it; ErrTargetNoSpace is an // item outcome, not a run error, and the items carry it. _ = c.stream.Submit(items...) } // selectItems enumerates every job option into the items of one run. Enumeration errors are run // errors: a duplicate relative path would write one target twice, so it ends the run instead of // being logged and skipped. func (c *Copyer) selectItems(ctx context.Context) ([]Item, error) { items := make([]Item, 0, 64) for _, job := range c.wildcardJobs { if err := ctx.Err(); err != nil { return nil, err } entries, err := c.walkWildcard(job) if err != nil { return nil, err } for _, entry := range entries { items = append(items, entry) } } for _, job := range c.accurateJobs { if err := ctx.Err(); err != nil { return nil, err } item := c.accurateItem(job) if item == nil { continue } items = append(items, item) } return items, nil } // walkWildcard walks every source of one wildcard job, keeps regular files, maps each // source-relative path onto every target directory, and returns the entries in physical order. // A source that cannot be read is reported and skipped, which is what af05f05c did. func (c *Copyer) walkWildcard(job *wildcardJob) ([]*compatItem, error) { entries := make([]*compatItem, 0, 64) var walk func(src *source) walk = func(src *source) { path := src.src() info := src.info var err error if info == nil { info, err = os.Stat(path) } if err != nil { c.stream.reportError(path, "", fmt.Errorf("walk get stat, %w", err)) return } mode := info.Mode() if mode.IsRegular() { targets := make([]string, 0, len(job.dst)) for _, dst := range job.dst { targets = append(targets, src.dst(dst)) } entries = append(entries, compatItemOf(src, path, targets, info)) return } if mode&UnexpectFileMode != 0 { return } files, err := os.ReadDir(path) if err != nil { c.stream.reportError(path, "", fmt.Errorf("walk read dir, %w", err)) return } for _, file := range files { walk(src.append(file.Name())) } } for _, src := range job.src { walk(src) } // One physical order drives a linear target, so the request order is the platform's path // order rather than the order the file system returned. sort.Slice(entries, func(i, j int) bool { return comparePath(entries[i].relative(), entries[j].relative()) < 0 }) // A repeated relative path would write one target twice, so the run fails before it copies // anything instead of keeping the first and skipping the rest. for index := 1; index < len(entries); index++ { if entries[index].relative() != entries[index-1].relative() { continue } return nil, fmt.Errorf( "same relative path, sources= '%s' and '%s'", entries[index-1].source, entries[index].source, ) } return entries, nil } // accurateItem resolves one exact source and target pair, which copies a file to the path it was // given instead of mapping a relative path onto a target directory. func (c *Copyer) accurateItem(job *accurateJob) *compatItem { // Resolve the source once so Base plus Path and FullPath always describe the same file. path, err := filepath.Abs(job.src) if err != nil { c.stream.reportError(job.src, "", fmt.Errorf("resolve accurate source failed, %w", err)) return nil } info, err := os.Stat(path) if err != nil { c.stream.reportError(path, "", fmt.Errorf("accurate job get stat, %w", err)) return nil } if !info.Mode().IsRegular() { return nil } targets := make([]string, 0, len(job.dsts)) for _, dst := range job.dsts { targets = append(targets, filepath.Clean(dst)) } // af05f05c reported an exact source against the file-system root, with the source split into // path segments, so the old row keeps that base and those segments; FullPath names the source // directly for a consumer that needs one path. base := filepath.VolumeName(path) + string(filepath.Separator) return compatItemOf(&source{base: base, path: strings.TrimPrefix(path, base)}, path, targets, info) } // compatItemOf builds the report coordinates of one enumerated file. The shell's walk knows the // source base and the source-relative path, which are exactly the af05f05c report row fields. func compatItemOf(src *source, sourcePath string, targets []string, info os.FileInfo) *compatItem { return &compatItem{ source: sourcePath, info: info, targets: targets, base: src.base, path: pathSegments(src.path), } } // pathSegments splits a path into the slash-delimited segments a report row carries. Empty // segments are dropped, which is how af05f05c turned a path into a report row. func pathSegments(name string) []string { segments := strings.Split(filepath.ToSlash(name), "/") filtered := make([]string, 0, len(segments)) for _, segment := range segments { if segment == "" { continue } filtered = append(filtered, segment) } return filtered } // compatItem carries the report coordinates of one enumerated file, so a result can be // translated back into the row this surface promises: the whole source path plus the base and // the source-relative path segments the af05f05c report row is made of. type compatItem struct { info os.FileInfo source string targets []string base string path []string } func (i *compatItem) Source() string { return i.source } func (i *compatItem) Targets() []string { return i.targets } // relative is the source-relative path the row's segments join to, which is what orders one // enumeration and keys its report row. func (i *compatItem) relative() string { return path.Join(i.path...) } // reportResults translates every result into the terminal report row event the report machinery // consumes. An item-level failure travels under the empty target key, which is the slot no // requested target can occupy, so a failed item still has a row. func (c *Copyer) reportResults(results []Result) error { for _, result := range results { c.stream.submit(&EventUpdateJob{Job: reportRow(result)}) } return nil } // reportRow renders one result as a report row. func reportRow(result Result) *Job { job := &Job{ Status: JobStatusFinished, Size: result.Size, Mode: result.Mode, ModTime: result.ModTime, WriteTime: result.WriteTime, SHA256: sha256Hex(result.SHA256), SignatureCacheHit: result.SignatureCacheHit, } if item, ok := result.Job.(*compatItem); ok { job.FullPath, job.Base, job.Path = item.source, item.base, item.path } // An item ACP could not process has no target outcome to report: the failure travels under // the empty key, which is the shape the report has always used for it. if result.Err != nil { job.FailTargets = map[string]error{"": result.Err} return job } for _, target := range result.Targets { if target.Err == nil { job.SuccessTargets = append(job.SuccessTargets, target.Path) continue } if job.FailTargets == nil { job.FailTargets = make(map[string]error, len(result.Targets)) } job.FailTargets[target.Path] = target.Err } return job } type accurateJob struct { src string dsts []string } // AccurateJob copies one exact source to the given destination paths, without mapping a // relative path onto a target directory. func AccurateJob(src string, dsts []string) Option { return func(o *option) *option { o.accurateJobs = append(o.accurateJobs, &accurateJob{src: src, dsts: dsts}) return o } } type wildcardJob struct { src []*source dst []string } // check validates the job before a run starts: every target must be an existing directory, at // least one source must be given, and every source must exist. func (job *wildcardJob) check() error { filteredDst := make([]string, 0, len(job.dst)) for _, p := range job.dst { if p == "" { continue } p = filepath.Clean(p) dstStat, err := os.Stat(p) if err != nil { return fmt.Errorf("check dst path '%s', %w", p, err) } if !dstStat.IsDir() { return fmt.Errorf("dst path is not a dir, path= '%s'", p) } filteredDst = append(filteredDst, p) } job.dst = filteredDst if len(job.src) == 0 { return fmt.Errorf("source path not found") } sort.Slice(job.src, func(i, j int) bool { return comparePath(job.src[i].path, job.src[j].path) < 0 }) for _, s := range job.src { // Relative source coordinates resolve against one absolute base for both I/O and reports. base, err := filepath.Abs(s.base) if err != nil { return fmt.Errorf("resolve source base failed, %w", err) } if strings.HasSuffix(s.base, string(filepath.Separator)) && !strings.HasSuffix(base, string(filepath.Separator)) { base += string(filepath.Separator) } s.base = base src := s.src() info, err := os.Stat(src) if err != nil { return fmt.Errorf("check src path '%s', %w", src, err) } s.info = info } return nil } // WildcardJob copies every regular file below the given sources into every target directory, // keeping each file's source-relative path. func WildcardJob(opts ...WildcardJobOption) Option { return func(o *option) *option { j := new(wildcardJob) for _, opt := range opts { j = opt(j) } if len(j.src) == 0 { return o } o.wildcardJobs = append(o.wildcardJobs, j) return o } } // WildcardJobOption configures one WildcardJob. type WildcardJobOption func(*wildcardJob) *wildcardJob // Source adds source paths to a WildcardJob. A source may be a directory or a regular file. func Source(paths ...string) WildcardJobOption { return func(j *wildcardJob) *wildcardJob { for _, p := range paths { if p == "" { continue } p = filepath.Clean(p) base, name := filepath.Split(p) j.src = append(j.src, &source{base: base, path: name}) } return j } } // AccurateSource adds sources as base plus relative path segments to a WildcardJob. func AccurateSource(base string, paths ...[]string) WildcardJobOption { return func(j *wildcardJob) *wildcardJob { for _, path := range paths { j.src = append(j.src, &source{base: base, path: filepath.Join(path...)}) } return j } } // Target adds target directories to a WildcardJob. Every one of them must exist. func Target(paths ...string) WildcardJobOption { return func(j *wildcardJob) *wildcardJob { j.dst = append(j.dst, paths...) return j } } // WithHash selects whether the run computes a content hash and refreshes the stored hash cache. // It is the af05f05c form of WithHashPolicy: true maps to HashReadRefresh and false to HashOff, // and a repeated hash option is last-wins. func WithHash(b bool) Option { if !b { return WithHashPolicy(HashOff) } return WithHashPolicy(HashReadRefresh) }