From 3d276bb62351546d0005521f80d3e36c11ec5572 Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Fri, 9 Oct 2026 18:43:24 +0200 Subject: [PATCH 1/2] feat(plugins)!: load product-owned Connectivity executables --- cmd/connectivity_cancel_test.go | 3 +- cmd/connectivity_external_test.go | 33 ++++ cmd/connectivity_test.go | 32 ++- cmd/plugins_install.go | 2 +- cmd/plugins_show.go | 2 +- cmd/plugins_sync.go | 2 +- cmd/plugins_target.go | 5 + cmd/root.go | 15 +- cmd/service_loader_test.go | 48 ++++- internal/plugin/external.go | 9 - internal/plugin/service_loader_test.go | 21 +- internal/plugin/validate_interaction_test.go | 7 +- internal/pluginmanager/auth_catalogue_test.go | 18 +- plugins/connectivity/connectivity.go | 132 ------------- plugins/connectivity/connectivity_test.go | 183 ------------------ plugins/connectivity/interaction.go | 47 ----- plugins/connectivity/manifest.go | 74 ------- plugins/connectivity/testdata/operations.json | 75 ------- plugins/connectivity/validation.go | 171 ---------------- 19 files changed, 155 insertions(+), 724 deletions(-) create mode 100644 cmd/connectivity_external_test.go delete mode 100644 plugins/connectivity/connectivity.go delete mode 100644 plugins/connectivity/connectivity_test.go delete mode 100644 plugins/connectivity/interaction.go delete mode 100644 plugins/connectivity/manifest.go delete mode 100644 plugins/connectivity/testdata/operations.json delete mode 100644 plugins/connectivity/validation.go diff --git a/cmd/connectivity_cancel_test.go b/cmd/connectivity_cancel_test.go index d5c73fe9..752c8b16 100644 --- a/cmd/connectivity_cancel_test.go +++ b/cmd/connectivity_cancel_test.go @@ -10,7 +10,6 @@ import ( "github.com/spf13/cobra" - "github.com/formancehq/fctl/v4/cmd" "github.com/formancehq/fctl/v4/internal/interactive" ) @@ -32,11 +31,11 @@ func TestConnectivityCancelAndDeclineNeverWrite(t *testing.T) { } return []string{"false"}, nil }) - root := cmd.NewRootCommand() args := []string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", action} if action == "delete" { args = append(args, "ingestion") } + root := connectivityRoot(t, args) root.SetArgs(args) root.SetOut(io.Discard) root.SetErr(io.Discard) diff --git a/cmd/connectivity_external_test.go b/cmd/connectivity_external_test.go new file mode 100644 index 00000000..9db1848b --- /dev/null +++ b/cmd/connectivity_external_test.go @@ -0,0 +1,33 @@ +package cmd_test + +import ( + "os" + "strings" + "testing" + + "github.com/spf13/cobra" + + "github.com/formancehq/fctl/v4/cmd" +) + +// Product integration is opt-in: the host must never import or build the +// Connectivity module. Supply a product-built executable at exact version 1.2.3. +func connectivityRoot(t *testing.T, args []string) *cobra.Command { + t.Helper() + binary := os.Getenv("FCTL_CONNECTIVITY_TEST_BINARY") + if binary == "" { + t.Skip("set FCTL_CONNECTIVITY_TEST_BINARY to product executable built for 1.2.3") + } + prefix := []string{} + for i := 0; i < len(args) && args[i] != "connectivity"; i++ { + prefix = append(prefix, args[i]) + } + install := append(append([]string{}, prefix...), "plugins", "install", "--service", "connectivity", "--binary", binary, "--service-version", "1.2.3") + out, trace, err := executeExternalCLI(t, install) + if err != nil || !strings.Contains(out, "connectivity") { + t.Fatalf("install: %s %s %v", out, trace, err) + } + // Cached metadata attaches forms without discovery; the actual execution + // still checks the live service version and uses the host HTTP broker. + return cmd.NewRootCommandWithArgs(t.Context(), append(prefix, "--help")) +} diff --git a/cmd/connectivity_test.go b/cmd/connectivity_test.go index 01b7005b..e79660ef 100644 --- a/cmd/connectivity_test.go +++ b/cmd/connectivity_test.go @@ -14,7 +14,6 @@ import ( "github.com/spf13/cobra" - "github.com/formancehq/fctl/v4/cmd" "github.com/formancehq/fctl/v4/internal/interactive" ) @@ -30,6 +29,10 @@ func TestConnectivityCreationFormPaginatesChoices(t *testing.T) { t.Setenv("FCTL_NO_INPUT", "") pages, writes := 0, 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/_info" { + writeConnectivityResponse(t, w, `{"version":"1.2.3"}`) + return + } if r.Method == http.MethodGet && r.URL.Path == "/connectors" { pages++ response := `{"cursor":{"pageSize":15,"hasMore":true,"next":"opaque +/=","data":[{"metadata":{"name":"adyen"},"spec":{"displayName":"Adyen"}}]}}` @@ -81,8 +84,9 @@ func TestConnectivityCreationFormPaginatesChoices(t *testing.T) { } return values, nil }) - root := cmd.NewRootCommand() - root.SetArgs([]string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "create"}) + args := []string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "create"} + root := connectivityRoot(t, args) + root.SetArgs(args) var output bytes.Buffer root.SetOut(&output) root.SetErr(io.Discard) @@ -97,8 +101,12 @@ func TestConnectivityCreationFormPaginatesChoices(t *testing.T) { func TestConnectivityExplicitBodiesBypassForms(t *testing.T) { body := `{"suspend":false,"version":null,"startSequence":9007199254740993}` server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/_info" { + writeConnectivityResponse(t, w, `{"version":"1.2.3"}`) + return + } data, err := io.ReadAll(r.Body) - if err != nil || string(data) != body || r.Method != http.MethodPatch || r.Header.Get("Content-Type") != "application/merge-patch+json" { + if err != nil || string(data) != `{"spec":`+body+`}` || r.Method != http.MethodPatch || r.Header.Get("Content-Type") != "application/merge-patch+json" { t.Errorf("method=%s type=%s body=%s err=%v", r.Method, r.Header.Get("Content-Type"), data, err) } writeConnectivityResponse(t, w, `{"metadata":{"name":"ingestion"}}`) @@ -110,8 +118,9 @@ func TestConnectivityExplicitBodiesBypassForms(t *testing.T) { } for _, input := range []string{body, "@" + file, "-"} { t.Run(input[:1], func(t *testing.T) { - root := cmd.NewRootCommand() - root.SetArgs([]string{"--config-dir", t.TempDir(), "--no-input", "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "patch", "ingestion", "--data", input}) + args := []string{"--config-dir", t.TempDir(), "--no-input", "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "patch", "ingestion", "--data", input} + root := connectivityRoot(t, args) + root.SetArgs(args) root.SetIn(strings.NewReader(body)) root.SetOut(io.Discard) root.SetErr(io.Discard) @@ -129,13 +138,17 @@ func TestConnectivityPatchFormAndInstanceSelection(t *testing.T) { patch := `{"suspend":true,"version":null}` writes := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/_info" { + writeConnectivityResponse(t, w, `{"version":"1.2.3"}`) + return + } if r.Method == http.MethodGet && r.URL.Path == "/connectorinstances" { writeConnectivityResponse(t, w, `{"cursor":{"hasMore":false,"data":[{"metadata":{"name":"ingestion"},"spec":{"connector":"stripe","ledger":"books"},"status":{"phase":"Running"}}]}}`) return } writes++ data, err := io.ReadAll(r.Body) - if err != nil || r.Method != http.MethodPatch || r.URL.Path != "/connectorinstances/ingestion" || string(data) != patch { + if err != nil || r.Method != http.MethodPatch || r.URL.Path != "/connectorinstances/ingestion" || string(data) != `{"spec":`+patch+`}` { t.Errorf("patch body=%s err=%v", data, err) } writeConnectivityResponse(t, w, `{}`) @@ -155,8 +168,9 @@ func TestConnectivityPatchFormAndInstanceSelection(t *testing.T) { } return values, nil }) - root := cmd.NewRootCommand() - root.SetArgs([]string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "patch"}) + args := []string{"--config-dir", t.TempDir(), "--auth-mode", "none", "--connectivity-url", server.URL, "connectivity", "instances", "patch"} + root := connectivityRoot(t, args) + root.SetArgs(args) root.SetOut(io.Discard) root.SetErr(io.Discard) if err := root.ExecuteContext(interactive.WithRunner(t.Context(), runner)); err != nil { diff --git a/cmd/plugins_install.go b/cmd/plugins_install.go index 8d20c3da..934d6f8e 100644 --- a/cmd/plugins_install.go +++ b/cmd/plugins_install.go @@ -21,7 +21,7 @@ func newPluginsInstallCommand(settings *connection.Settings) *cobra.Command { Annotations: map[string]string{"fctl.target": "stack"}, Example: " fctl plugins install --binary ./build/fctl-plugin-ledger --profile local", } - install.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger or auth") + install.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger, auth or connectivity") install.Flags().StringVar(&binary, "binary", "", "Explicit path to the trusted executable") install.Flags().StringVar(&localVersion, "service-version", "", "Exact service version; defaults to selected service /_info discovery") install.RunE = func(cmd *cobra.Command, _ []string) error { diff --git a/cmd/plugins_show.go b/cmd/plugins_show.go index 196d4ba2..d7a6ef41 100644 --- a/cmd/plugins_show.go +++ b/cmd/plugins_show.go @@ -9,7 +9,7 @@ import ( func newPluginsShowCommand(settings *connection.Settings) *cobra.Command { var service string show := &cobra.Command{Use: "show", Short: "Show the service plugin locked for the selected target", Args: cobra.NoArgs} - show.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger or auth") + show.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger, auth or connectivity") show.RunE = func(cmd *cobra.Command, _ []string) error { if err := distributionService(service); err != nil { return err diff --git a/cmd/plugins_sync.go b/cmd/plugins_sync.go index ff53012e..877efebe 100644 --- a/cmd/plugins_sync.go +++ b/cmd/plugins_sync.go @@ -17,7 +17,7 @@ func newPluginsSyncCommand(settings *connection.Settings) *cobra.Command { Annotations: map[string]string{"fctl.target": "stack"}, Example: " fctl plugins sync --organization ORG --stack STACK\n fctl plugins sync --catalogue registry.yaml --service-version 3.0.0 --profile local", } - sync.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger or auth") + sync.Flags().StringVar(&service, "service", "ledger", "Service plugin: ledger, auth or connectivity") sync.Flags().StringVar(&catalogue, "catalogue", "", "YAML or JSON catalogue file or HTTPS URL; defaults to the official registry (FCTL_PLUGIN_CATALOGUE)") sync.Flags().StringVar(&serviceVersion, "service-version", "", "Exact service version; defaults to selected service /_info discovery") sync.Flags().IntVar(&revision, "revision", 0, "Plugin revision; defaults to the highest for this exact service version") diff --git a/cmd/plugins_target.go b/cmd/plugins_target.go index 6e3abc71..134d76a0 100644 --- a/cmd/plugins_target.go +++ b/cmd/plugins_target.go @@ -25,6 +25,8 @@ func supportedPluginService(service string) (serviceDescriptor, error) { switch service { case "auth": return serviceDescriptor{name: "auth", title: "Auth"}, nil + case "connectivity": + return serviceDescriptor{name: "connectivity", title: "Connectivity"}, nil case "ledger": return serviceDescriptor{name: "ledger", title: "Ledger"}, nil default: @@ -49,6 +51,9 @@ func pluginServiceTarget(settings *connection.Settings, cmd *cobra.Command, serv if descriptor.name == "auth" { endpoint = options.AuthURL } + if descriptor.name == "connectivity" { + endpoint = options.ConnectivityURL + } if options.AuthMode == "cloud" { endpoint = cmp.Or(options.Issuer, cloud.DefaultIssuer) } else if endpoint == "" && options.StackURL != "" { diff --git a/cmd/root.go b/cmd/root.go index d05f000b..9276626a 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -2,6 +2,7 @@ package cmd import ( "context" + "fmt" "os" "os/signal" @@ -19,7 +20,6 @@ import ( "github.com/formancehq/fctl/v4/internal/plugin" "github.com/formancehq/fctl/v4/plugins/auth" cloudplugin "github.com/formancehq/fctl/v4/plugins/cloud" - "github.com/formancehq/fctl/v4/plugins/connectivity" "github.com/formancehq/fctl/v4/plugins/ledger" ) @@ -51,7 +51,8 @@ func newRootCommand(ctx context.Context, args []string) *cobra.Command { root.PersistentFlags().Bool("no-input", false, "Disable interactive forms and selections (FCTL_NO_INPUT, CI)") authFactory, authPreparationErr := prepareServicePlugin(ctx, root, settings, args, "auth") ledgerFactory, ledgerPreparationErr := prepareServicePlugin(ctx, root, settings, args, "ledger") - preparationErrors := map[string]error{"auth": authPreparationErr, "ledger": ledgerPreparationErr} + connectivityFactory, connectivityPreparationErr := prepareServicePlugin(ctx, root, settings, args, "connectivity") + preparationErrors := map[string]error{"auth": authPreparationErr, "ledger": ledgerPreparationErr, "connectivity": connectivityPreparationErr} root.PersistentPreRunE = func(cmd *cobra.Command, _ []string) error { if err := preparationErrors[serviceCommand(cmd)]; err != nil { return err @@ -70,7 +71,10 @@ func newRootCommand(ctx context.Context, args []string) *cobra.Command { if ledgerFactory == nil { ledgerFactory = ledger.New } - for _, factory := range []plugin.Factory{cloudplugin.New, authFactory, ledgerFactory, connectivity.New} { + for _, factory := range []plugin.Factory{cloudplugin.New, authFactory, ledgerFactory, connectivityFactory} { + if factory == nil { + continue + } if err := registry.Register(context.WithoutCancel(ctx), factory(nil), factory); err != nil { panic(err) // Embedded metadata is a build-time invariant, never user input. } @@ -78,6 +82,11 @@ func newRootCommand(ctx context.Context, args []string) *cobra.Command { if err := plugin.NewCommandWithRequest(registry, resolve).AddTo(root); err != nil { panic(err) } + if connectivityFactory == nil { + root.AddCommand(&cobra.Command{Use: "connectivity", Short: "Manage Connectivity with an installed product plugin", RunE: func(*cobra.Command, []string) error { + return fmt.Errorf("install the Connectivity plugin with fctl plugins install --service connectivity --binary PATH, or configure FCTL_PLUGIN_CATALOGUE") + }}) + } if err := cloudtools.AddTo(root, settings); err != nil { panic(err) } diff --git a/cmd/service_loader_test.go b/cmd/service_loader_test.go index 4c000d1a..266a1576 100644 --- a/cmd/service_loader_test.go +++ b/cmd/service_loader_test.go @@ -44,7 +44,7 @@ func loaderFlag(t *testing.T, root *cobra.Command, name, value string) { } func TestServiceLoaderTargets(t *testing.T) { - for _, service := range []string{"auth", "ledger"} { + for _, service := range []string{"auth", "ledger", "connectivity"} { t.Run(service, func(t *testing.T) { testServiceLoaderTarget(t, service) }) } root, settings := loaderSettings(t) @@ -72,6 +72,7 @@ func testServiceLoaderTarget(t *testing.T, service string) { } loaderFlag(t, root, "auth-url", "https://auth.example") loaderFlag(t, root, "ledger-url", "https://ledger.example") + loaderFlag(t, root, "connectivity-url", "https://connectivity.example") target, err = pluginServiceTarget(settings, root, service) if err != nil || target.Endpoint != "https://"+service+".example" { t.Fatalf("direct target: %+v %v", target, err) @@ -90,7 +91,7 @@ func testServiceLoaderTarget(t *testing.T, service string) { func TestServiceLoaderManifestBoundary(t *testing.T) { root, _ := loaderSettings(t) - for _, service := range []string{"auth", "ledger"} { + for _, service := range []string{"auth", "ledger", "connectivity"} { manifest := loaderManifest(service) if err := validateServiceManifest(t.Context(), root, service, manifest); err != nil { t.Fatal(err) @@ -138,10 +139,10 @@ func TestServiceLoaderOfflineRootsAndCacheIsolation(t *testing.T) { t.Fatal(err) } target := pluginmanager.Target{Profile: "", Organization: "org", Stack: "stack", Endpoint: settings.Options.Issuer} - for _, service := range []string{"auth", "ledger"} { + for _, service := range []string{"auth", "ledger", "connectivity"} { installLoaderFixture(t, manager, target, service) } - for _, service := range []string{"auth", "ledger"} { + for _, service := range []string{"auth", "ledger", "connectivity"} { lock, err := cachedServiceLock(settings, root, manager, service) if err != nil || lock.Service != service { t.Fatalf("cached %s: %+v %v", service, lock, err) @@ -259,6 +260,7 @@ func TestAuthLoaderExactVersionAndStrictTargets(t *testing.T) { func TestLedgerLoaderCompatibility(t *testing.T) { root, settings := loaderSettings(t) loaderFlag(t, root, "ledger-url", "https://ledger.example") + loaderFlag(t, root, "connectivity-url", "https://connectivity.example") target, err := ledgerTarget(settings, root) if err != nil { t.Fatal(err) @@ -296,3 +298,41 @@ func TestLedgerLoaderCompatibility(t *testing.T) { t.Fatalf("legacy version: %s %v", version, err) } } + +func TestConnectivityLoaderExactVersionAndStrictTargets(t *testing.T) { + t.Setenv("FCTL_PLUGIN_CATALOGUE", "") + base := http.DefaultTransport + http.DefaultTransport = discoveryTransport(func(request *http.Request) (*http.Response, error) { + if request.URL.String() != "http://127.0.0.1:1/_info" { + t.Fatalf("wrong Connectivity version endpoint: %s", request.URL) + } + return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"version":"v1.0.1"}`))}, nil + }) + t.Cleanup(func() { http.DefaultTransport = base }) + prep := discoveryPreparation(t, discoveryTransport(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusServiceUnavailable, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(""))}, nil + })) + prep.service = serviceDescriptor{name: "connectivity", title: "Connectivity"} + loaderFlag(t, prep.root, "auth-mode", "none") + loaderFlag(t, prep.root, "connectivity-url", "http://127.0.0.1:1") + target, version, err := pluginServiceSyncTarget(t.Context(), prep.settings, prep.root, "connectivity", "") + if err != nil || version != "1.0.1" { + t.Fatalf("exact Connectivity discovery: %+v %s %v", target, version, err) + } + empty := &pluginmanager.Catalogue{SchemaVersion: pluginmanager.SchemaVersion} + if _, err := prep.matchVersion(t.Context(), pluginmanager.DefaultCatalogue, empty); !errors.Is(err, pluginmanager.ErrNotInstalled) { + t.Fatalf("official unprepared missing release: %v", err) + } + t.Setenv("FCTL_PLUGIN_CATALOGUE", "https://custom.example/catalogue") + if _, err := prep.resolve(t.Context(), pluginBootstrap{commands: []string{"connectivity", "probe"}}); !errors.Is(err, pluginmanager.ErrCatalogueUnavailable) { + t.Fatalf("custom unavailable target fell back: %v", err) + } + t.Setenv("FCTL_PLUGIN_CATALOGUE", "") + installLoaderFixture(t, prep.manager, target, "connectivity") + if _, err := prep.matchVersion(t.Context(), pluginmanager.DefaultCatalogue, empty); !errors.Is(err, pluginmanager.ErrNoRelease) { + t.Fatalf("prepared target accepted missing exact release: %v", err) + } + if _, err := prep.matchVersion(t.Context(), pluginmanager.DefaultCatalogue, nil); !errors.Is(err, pluginmanager.ErrCatalogueUnavailable) { + t.Fatalf("prepared unavailable target fell back: %v", err) + } +} diff --git a/internal/plugin/external.go b/internal/plugin/external.go index 625e249a..a8302503 100644 --- a/internal/plugin/external.go +++ b/internal/plugin/external.go @@ -8,7 +8,6 @@ import ( "net/http" "reflect" "strings" - "sync" "github.com/formancehq/fctl/pkg/pluginsdk" "github.com/formancehq/fctl/pkg/pluginsdk/httpclient" @@ -36,8 +35,6 @@ type external struct { verify func(context.Context) (string, error) manifest pluginsdk.Manifest http *http.Client - mu sync.Mutex - checked string } func (p *external) GetManifest(ctx context.Context) (pluginsdk.Manifest, error) { @@ -71,11 +68,6 @@ func (p *external) Execute(ctx context.Context, request pluginsdk.ExecuteRequest } func (p *external) checkVersion(ctx context.Context, endpoint string) error { - p.mu.Lock() - defer p.mu.Unlock() - if p.checked != "" && p.checked == endpoint { - return nil - } client, err := httpclient.New(endpoint, p.http) if err != nil { return err @@ -94,6 +86,5 @@ func (p *external) checkVersion(ctx context.Context, endpoint string) error { if version != p.manifest.Version { return fmt.Errorf("installed plugin targets %s %s, service reports %s; run fctl plugins sync or install the matching binary", p.manifest.Service, p.manifest.Version, version) } - p.checked = endpoint return nil } diff --git a/internal/plugin/service_loader_test.go b/internal/plugin/service_loader_test.go index a31edda5..b1ff5f5d 100644 --- a/internal/plugin/service_loader_test.go +++ b/internal/plugin/service_loader_test.go @@ -15,9 +15,9 @@ type loaderTransport func(*http.Request) (*http.Response, error) func (f loaderTransport) RoundTrip(request *http.Request) (*http.Response, error) { return f(request) } -//nolint:gocognit // The same version-cache and exact-mismatch lifecycle applies to both services. +//nolint:gocognit // The same per-call version-check and exact-mismatch lifecycle applies to both services. func TestExternalServiceVersionBeforeProcess(t *testing.T) { - for _, service := range []string{"auth", "ledger"} { + for _, service := range []string{"auth", "ledger", "connectivity"} { t.Run(service, func(t *testing.T) { var requests atomic.Int32 p := &external{manifest: pluginsdk.Manifest{Service: service, Version: "1.2.3"}, http: &http.Client{Transport: loaderTransport(func(request *http.Request) (*http.Response, error) { @@ -33,11 +33,10 @@ func TestExternalServiceVersionBeforeProcess(t *testing.T) { t.Fatal(err) } } - if requests.Load() != 1 { + if requests.Load() != 2 { t.Fatalf("version checks: %d", requests.Load()) } p.manifest.Version = "1.2.4" - p.checked = "" err := p.checkVersion(t.Context(), endpoint) if err == nil || !strings.Contains(err.Error(), service+" 1.2.4") { t.Fatalf("service mismatch: %v", err) @@ -55,3 +54,17 @@ func TestExternalAuthVersionFailureDoesNotStartExecutable(t *testing.T) { t.Fatalf("version check did not precede startup: %v", err) } } + +func TestExternalRejectsVersionDriftAfterSuccessfulCheck(t *testing.T) { + version := "1.2.3" + p := &external{manifest: pluginsdk.Manifest{Service: "connectivity", Version: version}, http: &http.Client{Transport: loaderTransport(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"version":"` + version + `"}`))}, nil + })}} + if err := p.checkVersion(t.Context(), "https://connectivity.example"); err != nil { + t.Fatal(err) + } + version = "1.2.4" + if err := p.checkVersion(t.Context(), "https://connectivity.example"); err == nil { + t.Fatal("cached version allowed drift") + } +} diff --git a/internal/plugin/validate_interaction_test.go b/internal/plugin/validate_interaction_test.go index 2dad8d1f..191dda14 100644 --- a/internal/plugin/validate_interaction_test.go +++ b/internal/plugin/validate_interaction_test.go @@ -11,7 +11,6 @@ import ( "github.com/formancehq/fctl/v4/internal/plugin" "github.com/formancehq/fctl/v4/plugins/auth" cloudplugin "github.com/formancehq/fctl/v4/plugins/cloud" - "github.com/formancehq/fctl/v4/plugins/connectivity" "github.com/formancehq/fctl/v4/plugins/ledger" ) @@ -341,16 +340,16 @@ func sharedAlternativeInputs(manifest *pluginsdk.Manifest) { func TestInteractionAllEmbeddedManifestsRegisterAndRootStarts(t *testing.T) { t.Parallel() registry := &plugin.Registry{} - for _, factory := range []plugin.Factory{auth.New, cloudplugin.New, ledger.New, connectivity.New} { + for _, factory := range []plugin.Factory{auth.New, cloudplugin.New, ledger.New} { if err := registry.Register(t.Context(), factory(nil), factory); err != nil { t.Fatalf("embedded manifest registration failed: %v", err) } } - if len(registry.List()) != 4 { + if len(registry.List()) != 3 { t.Fatal("an embedded manifest was not registered") } root := cmd.NewRootCommand() - for _, name := range []string{"auth", "cloud", "ledger", "connectivity"} { + for _, name := range []string{"auth", "cloud", "ledger"} { leaf, _, err := root.Find([]string{name}) if err != nil || leaf == root { t.Fatalf("root command did not attach %s: %v", name, err) diff --git a/internal/pluginmanager/auth_catalogue_test.go b/internal/pluginmanager/auth_catalogue_test.go index 8b4fd79e..878a8498 100644 --- a/internal/pluginmanager/auth_catalogue_test.go +++ b/internal/pluginmanager/auth_catalogue_test.go @@ -20,8 +20,18 @@ func TestAuthCatalogueExactVersionAndIndependentLocks(t *testing.T) { if err != nil { t.Fatal(err) } - catalogue := catalogueFile(t, ledger, auth) - for service, version := range map[string]string{"ledger": "3.0.0", "auth": "1.0.0"} { + connectivity, err := m.Publish(t.Context(), registry.server.URL+"/formance/ledger", Release{Service: "connectivity", ServiceVersion: "2.0.0", Revision: 1, Platform: CurrentPlatform(), Manifest: func() pluginsdk.Manifest { + m := authManifest("2.0.0") + m.Name = "connectivity" + m.Service = "connectivity" + m.Root.Use = "connectivity" + return m + }()}, localBinary(t, []byte("connectivity fixture"))) + if err != nil { + t.Fatal(err) + } + catalogue := catalogueFile(t, ledger, auth, connectivity) + for service, version := range map[string]string{"ledger": "3.0.0", "auth": "1.0.0", "connectivity": "2.0.0"} { release, err := m.Resolve(t.Context(), catalogue, service, version, 0) if err != nil { t.Fatal(err) @@ -35,14 +45,14 @@ func TestAuthCatalogueExactVersionAndIndependentLocks(t *testing.T) { t.Fatalf("Auth version fallback: %v", err) } registry.server.Close() - for _, service := range []string{"ledger", "auth"} { + for _, service := range []string{"ledger", "auth", "connectivity"} { lock, err := m.Load(target("shared"), service) if err != nil || lock.Service != service || lock.Manifest.Service != service { t.Fatalf("independent offline lock %s: %+v %v", service, lock, err) } } locks, err := m.List() - if err != nil || len(locks) != 2 { + if err != nil || len(locks) != 3 { t.Fatalf("multi-service locks: %+v %v", locks, err) } } diff --git a/plugins/connectivity/connectivity.go b/plugins/connectivity/connectivity.go deleted file mode 100644 index 3c33e0c8..00000000 --- a/plugins/connectivity/connectivity.go +++ /dev/null @@ -1,132 +0,0 @@ -// Package connectivity implements the Connectivity API through the public plugin SDK. -package connectivity - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "net/http" - "net/url" - "strings" - - "github.com/formancehq/fctl/pkg/pluginsdk" - "github.com/formancehq/fctl/pkg/pluginsdk/httpclient" -) - -type plugin struct{ client *http.Client } - -// New creates a plugin with the host's authenticated client. Metadata needs no client. -func New(client *http.Client) pluginsdk.Plugin { return &plugin{client: client} } - -func (*plugin) GetManifest(ctx context.Context) (pluginsdk.Manifest, error) { - if err := ctx.Err(); err != nil { - return pluginsdk.Manifest{}, err - } - return manifest(), nil -} - -type operation struct { - id, command, method string - segments []string - list, query bool -} - -// Contract: connectivity-api/openapi.yaml at ce2324887f5b5ec4e3c2ec934ac874656d4c5348. -var operations = []operation{ - {"getQueryCapabilities", "query capabilities", http.MethodGet, []string{"_query", "capabilities"}, false, false}, - {"healthcheck", "health", http.MethodGet, []string{"_healthcheck"}, false, false}, - {"info", "info", http.MethodGet, []string{"_info"}, false, false}, - {"listConnectors", "connectors list", http.MethodGet, []string{"connectors"}, true, true}, - {"getConnectorFacets", "connectors facets", http.MethodGet, []string{"connectors", "_facets"}, false, true}, - {"getConnector", "connectors show", http.MethodGet, []string{"connectors", "$0"}, false, false}, - {"listConnectorVersions", "connectors versions list", http.MethodGet, []string{"connectors", "$0", "versions"}, true, true}, - {"getConnectorVersion", "connectors versions show", http.MethodGet, []string{"connectors", "$0", "versions", "$1"}, false, false}, - {"listConnectorInstances", "instances list", http.MethodGet, []string{"connectorinstances"}, true, true}, - {"createConnectorInstance", "instances create", http.MethodPost, []string{"connectorinstances"}, false, false}, - {"getConnectorInstance", "instances show", http.MethodGet, []string{"connectorinstances", "$0"}, false, false}, - {"replaceConnectorInstance", "instances replace", http.MethodPut, []string{"connectorinstances", "$0"}, false, false}, - {"patchConnectorInstance", "instances patch", http.MethodPatch, []string{"connectorinstances", "$0"}, false, false}, - {"deleteConnectorInstance", "instances delete", http.MethodDelete, []string{"connectorinstances", "$0"}, false, false}, -} - -func (p *plugin) Execute(ctx context.Context, req pluginsdk.ExecuteRequest) (pluginsdk.ExecuteResponse, error) { - req, err := pluginsdk.NormalizeRequest(manifest(), req) - if err != nil { - return pluginsdk.ExecuteResponse{}, err - } - command := strings.Join(req.CommandPath[1:], " ") - for _, op := range operations { - if command != op.command { - continue - } - return p.execute(ctx, op, req) - } - return pluginsdk.ExecuteResponse{}, fmt.Errorf("unsupported connectivity command %q", command) -} - -func (p *plugin) execute(ctx context.Context, op operation, req pluginsdk.ExecuteRequest) (pluginsdk.ExecuteResponse, error) { - if err := validateRequest(op, req); err != nil { - return pluginsdk.ExecuteResponse{}, err - } - if p.client == nil { - return pluginsdk.ExecuteResponse{}, fmt.Errorf("connectivity plugin requires an HTTP client") - } - client, err := httpclient.New(req.Endpoint, p.client) - if err != nil { - return pluginsdk.ExecuteResponse{}, err - } - headers := http.Header{} - if op.method == http.MethodPatch { - headers.Set("Content-Type", "application/merge-patch+json") - } - data, err := client.Do(ctx, op.method, operationPath(op, req.Args), operationQuery(op, req.Flags), req.Body, headers) - if err != nil { - return pluginsdk.ExecuteResponse{}, serviceError(err) - } - return pluginsdk.ExecuteResponse{Data: data}, nil -} - -func operationPath(op operation, args []string) string { - segments := make([]string, len(op.segments)) - for i, segment := range op.segments { - switch segment { - case "$0": - segments[i] = args[0] - case "$1": - segments[i] = args[1] - default: - segments[i] = segment - } - } - return httpclient.Path(segments...) -} - -func operationQuery(op operation, flags map[string]string) url.Values { - query := url.Values{} - if op.query && flags["query"] != "" { - query.Set("query", flags["query"]) - } - if op.list { - query.Set("pageSize", flags["page-size"]) - if flags["cursor"] != "" { - query.Set("cursor", flags["cursor"]) - } - } - return query -} - -func serviceError(err error) error { - // Connectivity uses code/message rather than Ledger's errorCode/errorMessage. - var failure *httpclient.Error - if errors.As(err, &failure) { - var details struct { - Code string `json:"code"` - Message string `json:"message"` - } - if json.Unmarshal(failure.Body, &details) == nil && details.Code != "" { - failure.Code, failure.Message = details.Code, details.Message - } - } - return err -} diff --git a/plugins/connectivity/connectivity_test.go b/plugins/connectivity/connectivity_test.go deleted file mode 100644 index 2258944f..00000000 --- a/plugins/connectivity/connectivity_test.go +++ /dev/null @@ -1,183 +0,0 @@ -package connectivity - -import ( - "encoding/json" - "errors" - "io" - "net/http" - "os" - "slices" - "strings" - "testing" - - "github.com/formancehq/fctl/pkg/pluginsdk" - "github.com/formancehq/fctl/pkg/pluginsdk/httpclient" -) - -type transportFunc func(*http.Request) (*http.Response, error) - -func (f transportFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } - -//nolint:gocognit // Each route checks method, path, body, media type, defaults and response against the pinned OpenAPI contract. -func TestOpenAPIContract(t *testing.T) { - t.Parallel() - data, err := os.ReadFile("testdata/operations.json") - if err != nil { - t.Fatal(err) - } - var contract struct { - Operations []struct{ ID, Method, Path string } - } - if err := json.Unmarshal(data, &contract); err != nil { - t.Fatal(err) - } - if len(contract.Operations) != len(operations) { - t.Fatal("OpenAPI operation coverage changed") - } - for _, expected := range contract.Operations { - t.Run(expected.ID, func(t *testing.T) { - i := slices.IndexFunc(operations, func(op operation) bool { return op.id == expected.ID }) - if i < 0 { - t.Fatal("missing OpenAPI operation") - } - op := operations[i] - req := pluginsdk.ExecuteRequest{CommandPath: strings.Fields("connectivity " + op.command), Endpoint: "https://api.example/prefix"} - if expected.Method == "DELETE" || expected.Method == "PUT" { - req.Flags = map[string]string{"confirm": "true"} - } - if strings.Contains(expected.Path, "{name}") { - req.Args = []string{"stripe"} - } - if strings.Contains(expected.Path, "{version}") { - req.Args = append(req.Args, "v1.2.3+build") - } - switch expected.Method { - case "POST": - req.Body = json.RawMessage(`{"name":"stripe","spec":{"connector":"stripe","ledger":"books","startSequence":9007199254740993}}`) - case "PUT": - req.Body = json.RawMessage(`{"spec":{"connector":"stripe","ledger":"books"}}`) - case "PATCH": - req.Body = json.RawMessage(`{"suspend":false,"version":null,"config":{"env":{"COUNT":{"value":"9007199254740993"}}}}`) - } - calls := 0 - p := New(&http.Client{Transport: transportFunc(func(r *http.Request) (*http.Response, error) { - calls++ - path := strings.NewReplacer("{name}", "stripe", "{version}", "v1.2.3+build").Replace(expected.Path) - if r.Method != expected.Method || r.URL.Path != "/prefix"+path { - t.Errorf("request %s %s, want %s %s", r.Method, r.URL.Path, expected.Method, path) - } - body, err := io.ReadAll(r.Body) - if err != nil || string(body) != string(req.Body) { - t.Errorf("body=%s err=%v", body, err) - } - if req.Body != nil { - contentType := "application/json" - if expected.Method == "PATCH" { - contentType = "application/merge-patch+json" - } - if r.Header.Get("Content-Type") != contentType { - t.Error("incorrect write content type") - } - } - if op.list && r.URL.Query().Get("pageSize") != "15" { - t.Error("page-size default missing") - } - status, response := http.StatusOK, `{"data":{"sequence":9007199254740993}}` - if expected.Method == "POST" { - status = http.StatusCreated - } - if expected.Method == "DELETE" { - status, response = http.StatusNoContent, "" - } - return &http.Response{StatusCode: status, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(response))}, nil - })}) - result, err := p.Execute(t.Context(), req) - want := `{"data":{"sequence":9007199254740993}}` - if expected.Method == "DELETE" { - want = "null" - } - if err != nil || string(result.Data) != want || calls != 1 { - t.Fatalf("result=%s calls=%d err=%v", result.Data, calls, err) - } - }) - } -} - -func TestRejectsInvalidInputBeforeHTTP(t *testing.T) { - t.Parallel() - for _, tc := range []struct { - name, command, body string - args []string - flags map[string]string - }{ - {"name", "instances show", "", []string{"../other"}, nil}, - {"version", "connectors versions show", "", []string{"stripe", ".."}, nil}, - {"page zero", "instances list", "", nil, map[string]string{"page-size": "0"}}, - {"page too large", "connectors list", "", nil, map[string]string{"page-size": "101"}}, - {"bad query", "connectors facets", "[]", nil, map[string]string{"query": "[]"}}, - {"missing data", "instances create", "", nil, nil}, - {"null data", "instances patch", "null", []string{"stripe"}, nil}, - {"missing spec", "instances create", `{"name":"stripe"}`, nil, nil}, - {"missing connector", "instances create", `{"name":"stripe","spec":{"ledger":"books"}}`, nil, nil}, - {"missing ledger", "instances create", `{"name":"stripe","spec":{"connector":"stripe"}}`, nil, nil}, - {"invalid labels", "instances create", `{"name":"stripe","labels":{"n":3},"spec":{"connector":"stripe","ledger":"books"}}`, nil, nil}, - {"negative sequence", "instances patch", `{"startSequence":-1}`, []string{"stripe"}, nil}, - {"fraction", "instances patch", `{"startSequence":1.5}`, []string{"stripe"}, nil}, - {"overflow", "instances patch", `{"startSequence":9223372036854775808}`, []string{"stripe"}, nil}, - {"wrong suspend", "instances patch", `{"suspend":"false"}`, []string{"stripe"}, nil}, - {"oversized", "instances patch", `{"note":"` + strings.Repeat("x", 1<<20) + `"}`, []string{"stripe"}, nil}, - {"missing confirm", "instances delete", "", []string{"stripe"}, nil}, - {"replace confirm", "instances replace", `{"spec":{}}`, []string{"stripe"}, nil}, - {"unexpected flag", "info", "", nil, map[string]string{"oops": "true"}}, - {"body on GET", "info", `{}`, nil, nil}, - } { - t.Run(tc.name, func(t *testing.T) { - p := New(&http.Client{Transport: transportFunc(func(*http.Request) (*http.Response, error) { t.Fatal("invalid input reached HTTP"); return nil, nil })}) - _, err := p.Execute(t.Context(), pluginsdk.ExecuteRequest{CommandPath: strings.Fields("connectivity " + tc.command), Args: tc.args, Flags: tc.flags, Body: json.RawMessage(tc.body), Endpoint: "https://api.example"}) - if err == nil { - t.Fatal("accepted invalid input") - } - }) - } -} - -func TestCursorQueryAndVersionEscaping(t *testing.T) { - t.Parallel() - for _, args := range [][]string{{"connectivity", "connectors", "list"}, {"connectivity", "connectors", "versions", "show"}} { - request := pluginsdk.ExecuteRequest{CommandPath: args, Endpoint: "https://api.example/prefix"} - if args[len(args)-1] == "list" { - request.Flags = map[string]string{"page-size": "100", "cursor": "a+/=?%&", "query": `{"$match":{"spec.category":"bank"}}`} - } else { - request.Args = []string{"stripe", "v1/x%2Fy"} - } - p := New(&http.Client{Transport: transportFunc(func(r *http.Request) (*http.Response, error) { - if request.Flags != nil { - if r.URL.Query().Get("cursor") != request.Flags["cursor"] || r.URL.Query().Get("query") != request.Flags["query"] { - t.Error("query or opaque cursor changed") - } - } else if r.URL.EscapedPath() != "/prefix/connectors/stripe/versions/v1%2Fx%252Fy" { - t.Error("version must be encoded exactly once") - } - return &http.Response{StatusCode: 200, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{}`))}, nil - })}) - if _, err := p.Execute(t.Context(), request); err != nil { - t.Fatal(err) - } - } -} - -func TestStructuredServiceErrors(t *testing.T) { - t.Parallel() - for _, status := range []int{400, 401, 403, 404, 409, 422, 500, 503} { - calls := 0 - p := New(&http.Client{Transport: transportFunc(func(*http.Request) (*http.Response, error) { - calls++ - return &http.Response{StatusCode: status, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(`{"code":"connector_immutable","message":"create a new instance","details":{"current":"stripe"}}`))}, nil - })}) - response, err := p.Execute(t.Context(), pluginsdk.ExecuteRequest{CommandPath: []string{"connectivity", "instances", "patch"}, Args: []string{"stripe"}, Body: json.RawMessage(`{"suspend":true}`), Endpoint: "https://api.example"}) - var failure *httpclient.Error - if !errors.As(err, &failure) || failure.StatusCode != status || failure.Code != "connector_immutable" || !strings.Contains(err.Error(), "create a new instance") || response.Data != nil || calls != 1 { - t.Fatalf("error contract: data=%s err=%v calls=%d", response.Data, err, calls) - } - } -} diff --git a/plugins/connectivity/interaction.go b/plugins/connectivity/interaction.go deleted file mode 100644 index 563f6201..00000000 --- a/plugins/connectivity/interaction.go +++ /dev/null @@ -1,47 +0,0 @@ -package connectivity - -import "github.com/formancehq/fctl/pkg/pluginsdk" - -func connectorInput() pluginsdk.InputSpec { - return pluginsdk.InputSpec{Title: "Connector", Kind: "select", Argument: new(0), ValueType: "string", Required: true, - Source: &pluginsdk.ChoiceSource{CommandPath: []string{"connectivity", "connectors", "list"}, ValueField: "metadata.name", LabelFields: []string{"spec.displayName"}, EmptyMessage: "No published connectors are available"}} -} - -func instanceInput() pluginsdk.InputSpec { - return pluginsdk.InputSpec{Title: "Connector instance", Kind: "select", Argument: new(0), ValueType: "string", Required: true, - Source: &pluginsdk.ChoiceSource{CommandPath: []string{"connectivity", "instances", "list"}, ValueField: "metadata.name", LabelFields: []string{"spec.connector", "spec.ledger", "status.phase"}, EmptyMessage: "No connector instances are available; create one first"}} -} - -func versionInput() pluginsdk.InputSpec { - return pluginsdk.InputSpec{Title: "Connector version", Kind: "select", Argument: new(1), ValueType: "string", Required: true, - Source: &pluginsdk.ChoiceSource{CommandPath: []string{"connectivity", "connectors", "versions", "list"}, Args: []string{"$arg0"}, ValueField: "version", LabelFields: []string{"phase"}, EmptyMessage: "This connector has no published versions; specify a channel alias explicitly if needed"}} -} - -func createInputs() []pluginsdk.InputSpec { - connector := connectorInput() - connector.Argument, connector.BodyPointer = nil, "/spec/connector" - return []pluginsdk.InputSpec{ - {Title: "Instance name", Description: "Lowercase letters, digits and hyphens", Kind: "input", BodyPointer: "/name", ValueType: "string", Required: true}, - connector, - {Title: "Ledger", Description: "Destination ledger name", Kind: "input", BodyPointer: "/spec/ledger", ValueType: "string", Required: true}, - {Title: "Version pin", Description: "Optional exact version; leave blank to select a channel", Kind: "input", BodyPointer: "/spec/version", ValueType: "string"}, - {Title: "Channel", Description: "Used when no version is pinned", Kind: "select", BodyPointer: "/spec/channel", ValueType: "string", Default: "stable", Options: channels()}, - {Title: "Configuration", Description: "JSON object with env and files matching the connector version's configSchema; use secretRef for sensitive values", Kind: "text", BodyPointer: "/spec/config", ValueType: "json"}, - {Title: "Start sequence", Description: "Optional nonnegative int64 ingestion sequence", Kind: "input", BodyPointer: "/spec/startSequence", ValueType: "number"}, - {Title: "Poll interval", Description: "Optional Go duration, e.g. 30s", Kind: "input", BodyPointer: "/spec/pollInterval", ValueType: "string"}, - {Title: "Suspend ingestion", Kind: "confirm", BodyPointer: "/spec/suspend", ValueType: "bool", Default: "false"}, - } -} - -func channels() []pluginsdk.InputOption { - return []pluginsdk.InputOption{{Label: "Stable", Value: "stable"}, {Label: "Release candidate", Value: "rc"}, {Label: "Beta", Value: "beta"}, {Label: "Alpha", Value: "alpha"}} -} - -func patchInputs() []pluginsdk.InputSpec { - // No defaults: omitted fields must preserve the existing spec. Explicit - // --data also supports null deletion and every API configuration field. - return []pluginsdk.InputSpec{ - instanceInput(), - {Title: "Spec patch", Description: "JSON object applied directly to spec, e.g. {\"suspend\":true}; use null to remove a field", Kind: "text", Flag: "data", ValueType: "string", Required: true}, - } -} diff --git a/plugins/connectivity/manifest.go b/plugins/connectivity/manifest.go deleted file mode 100644 index 0923261e..00000000 --- a/plugins/connectivity/manifest.go +++ /dev/null @@ -1,74 +0,0 @@ -package connectivity - -import "github.com/formancehq/fctl/pkg/pluginsdk" - -func manifest() pluginsdk.Manifest { - return pluginsdk.Manifest{Name: "connectivity", Version: "0.1.0", Service: "connectivity", ProtocolVersion: pluginsdk.ProtocolVersion, - Root: pluginsdk.CommandSpec{Use: "connectivity", Target: "stack", Short: "Manage connectors and ingestion instances", Subcommands: []pluginsdk.CommandSpec{ - leaf("info", "Show Connectivity server information", 0, nil), - leaf("health", "Check Connectivity health", 0, nil), - {Use: "query", Short: "Inspect supported query fields and operators", Subcommands: []pluginsdk.CommandSpec{leaf("capabilities", "Show the query contract", 0, nil)}}, - {Use: "connectors", Short: "Browse the connector catalogue", Subcommands: []pluginsdk.CommandSpec{ - list("list", "List published connectors", 0, nil), - facets(), - leaf("show CONNECTOR", "Show a connector", 1, []pluginsdk.InputSpec{connectorInput()}), - {Use: "versions", Short: "Browse connector versions and configuration schemas", Subcommands: []pluginsdk.CommandSpec{ - list("list CONNECTOR", "List connector versions", 1, []pluginsdk.InputSpec{connectorInput()}), - leaf("show CONNECTOR VERSION", "Show a version or channel alias and its configuration schema", 2, []pluginsdk.InputSpec{connectorInput(), versionInput()}), - }}, - }}, - {Use: "instances", Short: "Manage connector instances in this Connectivity namespace", Subcommands: []pluginsdk.CommandSpec{ - list("list", "List connector instances", 0, nil), - leaf("show INSTANCE", "Show instance configuration and reconciliation status", 1, []pluginsdk.InputSpec{instanceInput()}), - body("create", "Create a connector instance", 0, false, createInputs()), - body("replace INSTANCE", "Replace the full desired spec, preserving metadata and status", 1, true, []pluginsdk.InputSpec{instanceInput(), {Title: "Complete instance spec", Description: "JSON object replacing the entire spec; include connector, ledger and configuration", Kind: "text", BodyPointer: "/spec", ValueType: "json", Required: true}}), - body("patch INSTANCE", "Apply a JSON merge patch directly to spec; null removes a field", 1, false, patchInputs()), - deletion(), - }}, - }}, - } -} - -func leaf(use, short string, args int, inputs []pluginsdk.InputSpec) pluginsdk.CommandSpec { - return pluginsdk.CommandSpec{Use: use, Short: short, Runnable: true, Args: pluginsdk.ArgsSpec{Min: args, Max: args}, Inputs: inputs} -} - -func list(use, short string, args int, inputs []pluginsdk.InputSpec) pluginsdk.CommandSpec { - c := leaf(use, short, args, inputs) - c.Flags = []pluginsdk.FlagSpec{ - {Name: "page-size", Type: "uint32", Default: "15", Usage: "Records per page (1-100)"}, - {Name: "cursor", Type: "string", Usage: "Opaque next cursor from the previous page"}, - queryFlag(), - } - return c -} - -func queryFlag() pluginsdk.FlagSpec { - return pluginsdk.FlagSpec{Name: "query", Type: "string", Usage: "JSON query expression; see connectivity query capabilities"} -} - -func facets() pluginsdk.CommandSpec { - c := leaf("facets", "Count catalogue facets, optionally filtered by query", 0, nil) - c.Flags = []pluginsdk.FlagSpec{queryFlag()} - return c -} - -func body(use, short string, args int, confirm bool, inputs []pluginsdk.InputSpec) pluginsdk.CommandSpec { - c := leaf(use, short, args, inputs) - c.Flags = []pluginsdk.FlagSpec{{Name: "data", Type: "string", Required: true, Body: true, Usage: "JSON object (max 1 MiB); inline, @file, or - for stdin"}} - c.Confirm = confirm - if confirm { - c.Flags = append(c.Flags, confirmation()) - } - return c -} - -func confirmation() pluginsdk.FlagSpec { - return pluginsdk.FlagSpec{Name: "confirm", Type: "bool", Default: "false", Usage: "Confirm this destructive operation"} -} - -func deletion() pluginsdk.CommandSpec { - c := leaf("delete INSTANCE", "Delete a connector instance", 1, []pluginsdk.InputSpec{instanceInput()}) - c.Confirm, c.Flags = true, []pluginsdk.FlagSpec{confirmation()} - return c -} diff --git a/plugins/connectivity/testdata/operations.json b/plugins/connectivity/testdata/operations.json deleted file mode 100644 index 979f34da..00000000 --- a/plugins/connectivity/testdata/operations.json +++ /dev/null @@ -1,75 +0,0 @@ -{ - "source": "https://github.com/formancehq/connectivity/blob/ce2324887f5b5ec4e3c2ec934ac874656d4c5348/misc/connectivity-api/openapi.yaml", - "operations": [ - { - "id": "getQueryCapabilities", - "method": "GET", - "path": "/_query/capabilities" - }, - { - "id": "healthcheck", - "method": "GET", - "path": "/_healthcheck" - }, - { - "id": "info", - "method": "GET", - "path": "/_info" - }, - { - "id": "listConnectors", - "method": "GET", - "path": "/connectors" - }, - { - "id": "getConnectorFacets", - "method": "GET", - "path": "/connectors/_facets" - }, - { - "id": "getConnector", - "method": "GET", - "path": "/connectors/{name}" - }, - { - "id": "listConnectorVersions", - "method": "GET", - "path": "/connectors/{name}/versions" - }, - { - "id": "getConnectorVersion", - "method": "GET", - "path": "/connectors/{name}/versions/{version}" - }, - { - "id": "listConnectorInstances", - "method": "GET", - "path": "/connectorinstances" - }, - { - "id": "createConnectorInstance", - "method": "POST", - "path": "/connectorinstances" - }, - { - "id": "getConnectorInstance", - "method": "GET", - "path": "/connectorinstances/{name}" - }, - { - "id": "replaceConnectorInstance", - "method": "PUT", - "path": "/connectorinstances/{name}" - }, - { - "id": "patchConnectorInstance", - "method": "PATCH", - "path": "/connectorinstances/{name}" - }, - { - "id": "deleteConnectorInstance", - "method": "DELETE", - "path": "/connectorinstances/{name}" - } - ] -} diff --git a/plugins/connectivity/validation.go b/plugins/connectivity/validation.go deleted file mode 100644 index 1403e682..00000000 --- a/plugins/connectivity/validation.go +++ /dev/null @@ -1,171 +0,0 @@ -package connectivity - -import ( - "bytes" - "encoding/json" - "fmt" - "math" - "net/http" - "regexp" - "strconv" - "strings" - - "github.com/formancehq/fctl/pkg/pluginsdk" -) - -var dnsLabel = regexp.MustCompile(`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$`) - -func validateRequest(op operation, req pluginsdk.ExecuteRequest) error { - if err := validateIDs(req.Args); err != nil { - return err - } - if op.list { - n, err := strconv.ParseUint(req.Flags["page-size"], 10, 32) - if err != nil || n < 1 || n > 100 { - return fmt.Errorf("page-size must be between 1 and 100") - } - } - if op.query && req.Flags["query"] != "" { - if _, err := object([]byte(req.Flags["query"]), "query"); err != nil { - return err - } - } - switch op.method { - case http.MethodPost, http.MethodPut, http.MethodPatch: - return validateBody(op.method, req.Body) - default: - if req.Body != nil { - return fmt.Errorf("this read/delete command does not accept a JSON body") - } - } - return nil -} - -func validateIDs(args []string) error { - for i, arg := range args { - if i == 0 && !dnsLabel.MatchString(arg) { - return fmt.Errorf("resource name must contain lowercase letters, digits or hyphens and start and end with a letter or digit") - } - if strings.TrimSpace(arg) == "" || arg == "." || arg == ".." { - return fmt.Errorf("resource identifier must be non-empty and cannot be a dot or double dot") - } - } - return nil -} - -func validateBody(method string, data json.RawMessage) error { - if len(data) > 1<<20 { - return fmt.Errorf("connectivity request body exceeds 1 MiB") - } - payload, err := object(data, "data") - if err != nil { - return err - } - if method == http.MethodPatch { - return validateSpec(payload, true) - } - if err := validateMetadata(payload); err != nil { - return err - } - spec, err := object(payload["spec"], "data.spec") - if err != nil { - return err - } - if err := validateSpec(spec, false); err != nil { - return err - } - if method == http.MethodPost { - return validateCreate(payload, spec) - } - return nil -} - -func validateCreate(payload, spec map[string]json.RawMessage) error { - var name string - if json.Unmarshal(payload["name"], &name) != nil || !dnsLabel.MatchString(name) { - return fmt.Errorf("data.name must be a lowercase DNS label") - } - for _, key := range []string{"connector", "ledger"} { - var value string - if json.Unmarshal(spec[key], &value) != nil || strings.TrimSpace(value) == "" { - return fmt.Errorf("data.spec.%s is required for creation", key) - } - } - return nil -} - -func validateMetadata(payload map[string]json.RawMessage) error { - for _, key := range []string{"labels", "annotations"} { - if value, exists := payload[key]; exists { - values, err := object(value, "data."+key) - if err != nil { - return err - } - for _, v := range values { - if !isString(v) { - return fmt.Errorf("data.%s values must be strings", key) - } - } - } - } - return nil -} - -func object(data []byte, field string) (map[string]json.RawMessage, error) { - var result map[string]json.RawMessage - if json.Unmarshal(data, &result) != nil || result == nil { - return nil, fmt.Errorf("%s must be a JSON object", field) - } - return result, nil -} - -func isString(value json.RawMessage) bool { - return len(bytes.TrimSpace(value)) != 0 && bytes.TrimSpace(value)[0] == '"' -} - -func validateSpec(spec map[string]json.RawMessage, patch bool) error { - for key, value := range spec { - value = bytes.TrimSpace(value) - if patch && bytes.Equal(value, []byte("null")) { - continue - } - if err := validateSpecField(key, value); err != nil { - return err - } - } - return nil -} - -func validateSpecField(key string, value json.RawMessage) error { - switch key { - case "connector", "version", "channel", "ledger", "pollInterval", "connectivityRef": - if !isString(value) { - return fmt.Errorf("spec.%s must be a string", key) - } - case "suspend": - if string(value) != "true" && string(value) != "false" { - return fmt.Errorf("spec.suspend must be a boolean") - } - case "startSequence": - return validateInteger(value, key, math.MaxInt64) - case "replicas": - return validateInteger(value, key, 1) - case "config", "additionalEnv": - _, err := object(value, "spec."+key) - return err - case "additionalFiles": - var files []json.RawMessage - if json.Unmarshal(value, &files) != nil || files == nil { - return fmt.Errorf("spec.additionalFiles must be an array") - } - } - return nil -} - -func validateInteger(value json.RawMessage, key string, maximum int64) error { - var n int64 - if json.Unmarshal(value, &n) != nil || string(value) == "null" || n < 0 || n > maximum { - return fmt.Errorf("spec.%s must be an integer between 0 and %d", key, maximum) - } - return nil -} From 69983368f0ed7a693b0f887f9383461101f7b6b2 Mon Sep 17 00:00:00 2001 From: Maxence Maireaux Date: Fri, 9 Oct 2026 18:43:30 +0200 Subject: [PATCH 2/2] docs(plugins): document Connectivity installation and qualification --- docs/connectivity.md | 32 ++++++++++++++++++++++---------- docs/modules.md | 2 +- docs/plugin-distribution.md | 9 ++++++--- docs/plugins.md | 4 ++-- docs/v4-foundation.md | 4 ++-- docs/validation.md | 19 +++++++++++++++++++ 6 files changed, 52 insertions(+), 18 deletions(-) diff --git a/docs/connectivity.md b/docs/connectivity.md index 4619a818..1e4c8c52 100644 --- a/docs/connectivity.md +++ b/docs/connectivity.md @@ -1,14 +1,11 @@ # Connectivity plugin -The embedded `plugins/connectivity` module implements all 14 operations in -[connectivity-api/openapi.yaml](https://github.com/formancehq/connectivity/blob/ce2324887f5b5ec4e3c2ec934ac874656d4c5348/misc/connectivity-api/openapi.yaml), -API version `0.1.0`, pinned to commit `ce2324887f5b5ec4e3c2ec934ac874656d4c5348`. -The repository also provides a generated client under `pkg/client/http`. -This plugin uses the public `pluginsdk/httpclient` adapter to preserve raw -request/response JSON, including merge-patch nulls, unknown fields and exact -numbers, without introducing typed-model defaults such as `replicas`. -It imports no server, Cobra, UI or core packages. -The host owns authentication, endpoints, storage, forms and presentation. +The external product-owned module +[`github.com/formancehq/connectivity/misc/fctl-plugin`](https://github.com/formancehq/connectivity/tree/main/misc/fctl-plugin) +implements all 14 Connectivity API operations. It is installed through the +shared loader; fctl has no Go dependency on the product module and no embedded +Connectivity fallback. The product owns forms, payload validation and HTTP +mapping; fctl owns credentials, terminal interaction and cached manifests. ## Profiles @@ -102,9 +99,24 @@ Deletion maps the empty 204 response to JSON null. Local fixtures cover all routes/methods, media types, errors, cursor encoding, int64 precision, size limits, forms, selection and authentication boundaries. -The pinned operation snapshot is `plugins/connectivity/testdata/operations.json`. +The operation snapshot and current API contract tests live in the product module. An independent consumer verifies the public SDK boundary without core or UI dependencies. A real PTY check covers creation, resource selection, default-No deletion confirmation, styled forms and plain tables. Creation produces one write and exact JSON on stdout even with forced terminal colors. Deployment availability and ingestion require separate live runtime checks. + +## Install the exact product executable + +```sh +fctl plugins install --service connectivity --binary ./fctl-plugin-connectivity --profile connectivity-local +fctl plugins sync --service connectivity --profile connectivity-local +fctl plugins show --service connectivity --profile connectivity-local +``` + +Without an installed matching executable, Connectivity commands require plugin +installation. Help/completion use cached metadata without launching a process or +contacting the service. Each execution checks `/_info` again and rejects version +drift before starting the plugin. Patch input remains a spec-only JSON object; +the product wraps it as `{"spec":...}` for the API. Read-only replicas and +connectivityRef are rejected locally; use suspend. diff --git a/docs/modules.md b/docs/modules.md index 4c99b3a0..8958d2a7 100644 --- a/docs/modules.md +++ b/docs/modules.md @@ -1,6 +1,6 @@ # Auth and Ledger modules -The root registers Cloud, Auth, Ledger and Connectivity plugin factories. This page describes +The root registers Cloud, Auth and Ledger defaults plus installed external service plugins, including Connectivity. This page describes the service modules; [Cloud management](cloud.md) describes control-plane commands. Each plugin implements the public `pluginsdk.Plugin` manifest/execution contract. The core builds Cobra diff --git a/docs/plugin-distribution.md b/docs/plugin-distribution.md index 9a77aa27..8b4d7625 100644 --- a/docs/plugin-distribution.md +++ b/docs/plugin-distribution.md @@ -1,12 +1,13 @@ # External service plugin distribution -Ledger and Auth plugins can run as independent native executables. Their product +Ledger, Auth and Connectivity plugins can run as independent native executables. Their product modules own their commands and executable entry points: | Service | Product module | Executable | | --- | --- | --- | | Ledger | `github.com/formancehq/ledger/misc/fctl-plugin` | `fctl-plugin-ledger` | | Auth | `github.com/formancehq/auth/misc/fctl-plugin` | `fctl-plugin-auth` | +| Connectivity | `github.com/formancehq/connectivity/misc/fctl-plugin` | `fctl-plugin-connectivity` | Auth product sources remain in the Auth repository. Public executable distribution is a separate publication step and does not require publishing private sources. @@ -68,7 +69,7 @@ fctl plugins sync --service auth --catalogue ./registry.yaml --profile local ``` `plugins install`, `plugins sync` and `plugins show` accept `--service ledger` -or `--service auth`; the default remains `ledger`. Unknown services are rejected +or `--service auth` or `--service connectivity`; the default remains `ledger`. Unknown services are rejected before installation. The selected service must match the manifest service, name and command root. The manifest version must equal the exact service version. Without `--service-version`, installation and sync discover that version through @@ -135,7 +136,7 @@ access. Only explicitly trusted local binaries should be installed. ## Locks and updates Locks are identified by service plus profile, organization, stack and stable -connection endpoint. Auth and Ledger keep independent locks for the same target. +connection endpoint. Auth, Ledger and Connectivity keep independent locks for the same target. Two stacks in the same profile can use different service versions. Executables are cached by digest and platform under the private v4 configuration directory's `plugins` folder. @@ -175,3 +176,5 @@ endpoint restrictions, corrupt downloads and concurrent cache installation. The SDK and product module can also be built and tested independently. Production OCI publication and updating the official catalogue are separate release steps; snapshot packaging does not perform either action. + +Connectivity has no embedded provider or product Go dependency. Use the exact API version, independent of Core and connector versions. Cached help/completion remain available offline. Version checks now run on every execution, including interactive discovery; there is an extra info request per operation. diff --git a/docs/plugins.md b/docs/plugins.md index 2b2e1445..5c7f09f9 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -78,7 +78,7 @@ and `--out` effects; they are not connection configuration. The shared public HTTP client performs requests without automatic mutation retries and preserves JSON numbers. -Cloud, Auth, Ledger and Connectivity remain available as embedded defaults. Auth and Ledger providers can replace their embedded command trees with cached +Cloud, Auth and Ledger remain available as embedded defaults. Auth, Ledger and Connectivity providers can replace their embedded command trees with cached external manifests. Both providers use the same two-method SDK contract, forms, body validation, authentication and output renderer. Exactly one provider owns each service command root. @@ -109,6 +109,6 @@ commands, API validation and declarative forms in the Auth repository. fctl's depends on the public plugin SDK and generated Auth client, without fctl core, Cobra, profile storage or terminal libraries. -Auth and Ledger external providers use the same generic host loader. The host +Auth, Ledger and Connectivity external providers use the same generic host loader. The host accepts only the selected service's command root and connection boundary, including descendant service overrides. Locks are keyed by target and service. diff --git a/docs/v4-foundation.md b/docs/v4-foundation.md index d2f3a201..e7b5ad40 100644 --- a/docs/v4-foundation.md +++ b/docs/v4-foundation.md @@ -12,7 +12,7 @@ and exits with status 1. Signal cancellation is propagated through the command context. The root registers `version`, `profiles` management and login/logout. Cloud, -Auth, Ledger and Connectivity register through the embedded plugin registry and public SDK; the core +Auth and Ledger register through the embedded plugin registry; Connectivity uses the external loader and public SDK; the core builds their Cobra commands from manifests. Cobra provides help, `--version` and shell completion commands. No profile, authentication or network access is required to inspect help or @@ -60,7 +60,7 @@ confirmation flags and request bodies. See [migration examples](cloud.md#migrati Existing automation must adapt to the [v4 module commands](modules.md) and [profile settings](profiles.md), or keep using a compatible v3 binary. -Connectivity is embedded; its API contract and commands are documented in [Connectivity](connectivity.md). +Connectivity is loaded from its installed product executable; its API contract and commands are documented in [Connectivity](connectivity.md). ## Homebrew release migration diff --git a/docs/validation.md b/docs/validation.md index 03b48f2a..60cd56e7 100644 --- a/docs/validation.md +++ b/docs/validation.md @@ -361,3 +361,22 @@ gRPC adapter in ledgerctl. A local single-node Ledger was used to check fctl listing and `after` pagination over HTTP, alongside ledgerctl's 38-case matrix covering all 31 shared operations. The isolated node advertised `3.0.0-beta.10`; these checks do not establish a deployed service release. + +## Independent Connectivity executable qualification + +The host does not import or build the Connectivity Go module. Cross-repository +form tests run with a separately built `FCTL_CONNECTIVITY_TEST_BINARY`, version +1.2.3 revision 1: + +```sh +# In Connectivity, build the fixture from the product source. +nix develop --command just --justfile misc/fctl-plugin/justfile --working-directory misc/fctl-plugin build 1.2.3 1 +# In fctl, supply the absolute output path. +FCTL_CONNECTIVITY_TEST_BINARY=/path/to/connectivity/build/fctl-plugin-connectivity nix develop --command just pre-commit tests +``` + +Without the variable, product form tests explicitly skip; generic Connectivity +loader, identity, offline cache, exact discovery and lock tests always run. +This keeps the host's dependency closure independent of the product source. +The baseline is GitHub #193 at c07b2b5531564baad541d851a14f88787185d536; +the user-specified local aff263ec0 was unavailable on GitHub and this machine.