81 lines
1.7 KiB
Go
81 lines
1.7 KiB
Go
package pipeline
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
|
|
"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 = http.DefaultClient
|
|
}
|
|
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")
|
|
}
|