package pipeline import ( "context" "fmt" "net/http" "strings" "sync" "evobgp/internal/httpclient" "evobgp/internal/store" ) // RefreshTenantModules ingests all listed modules in parallel and updates per-module snapshots. func RefreshTenantModules(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, moduleIDs []string) error { if hc == nil { hc = httpclient.New(httpclient.DefaultTimeout) } var ids []string seen := make(map[string]struct{}) for _, id := range moduleIDs { id = strings.TrimSpace(id) if id == "" { continue } if _, ok := seen[id]; ok { continue } seen[id] = struct{}{} ids = append(ids, id) } if len(ids) == 0 { return nil } if len(ids) == 1 { return RefreshModuleIngest(ctx, st, hc, tenantID, ids[0]) } sem := make(chan struct{}, collectConcurrency()) errs := make([]error, len(ids)) var wg sync.WaitGroup for i, mid := range ids { wg.Add(1) go func(idx int, moduleID string) { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() errs[idx] = RefreshModuleIngest(ctx, st, hc, tenantID, moduleID) }(i, mid) } wg.Wait() for _, err := range errs { if err != nil { return err } } return nil } // PickTenantRefreshTriggerModule returns a module id for revision metadata (first enabled due module). func PickTenantRefreshTriggerModule(st store.Backend, tenantID string, moduleIDs []string) (string, error) { for _, id := range moduleIDs { id = strings.TrimSpace(id) if id == "" { continue } mod, err := st.GetModule(tenantID, id) if err != nil { return "", err } if mod.Enabled { return mod.ID, nil } } for _, mod := range st.ListModules(tenantID) { if mod != nil && mod.Enabled { return mod.ID, nil } } return "", fmt.Errorf("pipeline: no enabled module for tenant refresh") }