fetchAndDiscover downloads and scans jobs concurrently, returning a channel of results. Each job downloads its repo zip, discovers the individual resources, and emits them as Items. The DB is never touched here — callers serialize the upserts — so the workers need no locking.
(ctx context.Context, kind string, entityKind entities.Kind, cacheDir string, jobs []repoJob)
| 224 | // and emits them as Items. The DB is never touched here — callers serialize the |
| 225 | // upserts — so the workers need no locking. |
| 226 | func (svc *Service) fetchAndDiscover(ctx context.Context, kind string, entityKind entities.Kind, cacheDir string, jobs []repoJob) <-chan repoResult { |
| 227 | out := make(chan repoResult) |
| 228 | if len(jobs) == 0 { |
| 229 | close(out) |
| 230 | return out |
| 231 | } |
| 232 | |
| 233 | workers := min(refreshConcurrency, len(jobs)) |
| 234 | |
| 235 | jobCh := make(chan repoJob) |
| 236 | go func() { |
| 237 | defer close(jobCh) |
| 238 | for _, j := range jobs { |
| 239 | select { |
| 240 | case <-ctx.Done(): |
| 241 | return |
| 242 | case jobCh <- j: |
| 243 | } |
| 244 | } |
| 245 | }() |
| 246 | |
| 247 | var wg sync.WaitGroup |
| 248 | wg.Add(workers) |
| 249 | targetApps := strings.Join(defaultTargetApps(kind), ",") |
| 250 | for range workers { |
| 251 | go func() { |
| 252 | defer wg.Done() |
| 253 | for job := range jobCh { |
| 254 | dest := filepath.Join(cacheDir, kind, job.owner+"-"+job.repo) |
| 255 | _ = os.RemoveAll(dest) |
| 256 | root, err := svc.fetcher.Fetch(job.owner, job.repo, job.branch, dest) |
| 257 | if err != nil { |
| 258 | out <- repoResult{err: fmt.Sprintf("%s/%s: %v", job.owner, job.repo, err)} |
| 259 | _ = os.RemoveAll(dest) |
| 260 | continue |
| 261 | } |
| 262 | |
| 263 | resources := DiscoverResources(root, job.subPath, entityKind) |
| 264 | if job.catalogFile != "" { |
| 265 | if catalogResources := DiscoverCatalogResources(root, job.catalogFile, entityKind); len(catalogResources) > 0 { |
| 266 | resources = catalogResources |
| 267 | } |
| 268 | } else if len(resources) == 0 { |
| 269 | if catalogFile := inferCatalogFile(root, entityKind); catalogFile != "" { |
| 270 | resources = DiscoverCatalogResources(root, catalogFile, entityKind) |
| 271 | } |
| 272 | } |
| 273 | items := make([]Item, 0, len(resources)) |
| 274 | for _, res := range resources { |
| 275 | // Catalog rows that link to a real source repo are attributed to |
| 276 | // that repo (not the catalog/awesome-list repo). This is what stops |
| 277 | // awesome-list "pointer" catalogs from duplicating skills already |
| 278 | // indexed by a direct scan: both produce the same install key |
| 279 | // (sourceOwner/sourceRepo:name) and merge on the unique constraint. |
| 280 | owner, repo, branch, itemPath := job.owner, job.repo, job.branch, res.RelPath |
| 281 | if res.SourceOwner != "" && res.SourceRepo != "" { |
| 282 | owner, repo = res.SourceOwner, res.SourceRepo |
| 283 | if res.SourceBranch != "" { |