MCPcopy Index your code
hub / github.com/cloudquery/cloudquery / injectPlatformDestination

Function injectPlatformDestination

cli/internal/platform/inject.go:387–440  ·  view source on GitHub ↗

injectPlatformDestination appends the reserved `platform` destination carrying the cqpd_ token. The caller guarantees a source already targets it (the opt-in) and that no user-defined `platform` destination exists. recommendedVersion, when set and not overridden by the env, pins the plugin version;

(logger zerolog.Logger, destinations []*specs.Destination, sources []*specs.Source, token, recommendedVersion, tenantID string)

Source from the content-addressed store, hash-verified

385// when set and not overridden by the env, pins the plugin version; an unknown
386// registry skips injection (returns the spec unchanged).
387func injectPlatformDestination(logger zerolog.Logger, destinations []*specs.Destination, sources []*specs.Source, token, recommendedVersion, tenantID string) []*specs.Destination {
388 plugin := pluginCoords()
389 // Version precedence: env override > platform-pinned > CLI default.
390 // pluginCoords() already applied the env override (or the default), so only
391 // let the platform's pin win when the env override isn't set.
392 if os.Getenv(envPluginVersion) == "" && recommendedVersion != "" {
393 plugin.Version = recommendedVersion
394 }
395 parsedRegistry, err := specs.RegistryFromString(plugin.Registry)
396 if err != nil {
397 logger.Warn().Err(err).Str("registry", plugin.Registry).Msg("platform destination: unknown plugin registry; skipping auto-injection")
398 return destinations
399 }
400
401 // Report the path+version of the sources that target platform so it can
402 // reject (before any upload) versions the asset view can't process.
403 sourceVersions := make([]sourceVersion, 0, len(sources))
404 for _, s := range sources {
405 if slices.Contains(s.Destinations, destinationName) {
406 sourceVersions = append(sourceVersions, sourceVersion{Name: s.Name, Path: s.Path, Version: s.Version})
407 }
408 }
409 dest := &specs.Destination{
410 Metadata: specs.Metadata{
411 Name: destinationName,
412 Path: plugin.Path,
413 Registry: parsedRegistry,
414 Version: plugin.Version,
415 },
416 SyncSummary: true,
417 // sync_group_id is rejected with the default overwrite-delete-stale mode.
418 WriteMode: specs.WriteModeAppend,
419 // Unique per invocation so concurrent runs don't wipe each other's rows.
420 SyncGroupId: strconv.FormatUint(allocateSyncGroupID(time.Now()), 10),
421 Spec: map[string]any{
422 // api_url is omitted: the cqpd_ token carries the tenant's API URL,
423 // and the platform destination derives it from the token.
424 "token": token,
425 "source_versions": sourceVersions,
426 },
427 }
428 dest.SetDefaults()
429 destinations = append(destinations, dest)
430
431 evt := logger.Info().
432 Str("registry", plugin.Registry).
433 Str("path", plugin.Path).
434 Str("version", plugin.Version)
435 if tenantID != "" {
436 evt = evt.Str("tenant_id", tenantID)
437 }
438 evt.Msg("auto-injected platform destination")
439 return destinations
440}
441
442// resolveCredentials fetches a token and team for best-effort injection when
443// the sync command didn't authenticate. Overridable in tests.

Callers 1

MaybeInjectDestinationFunction · 0.85

Calls 3

SetDefaultsMethod · 0.95
pluginCoordsFunction · 0.85
allocateSyncGroupIDFunction · 0.85

Tested by

no test coverage detected