diff --git a/go.mod b/go.mod index 1cb0709..7f61ab1 100644 --- a/go.mod +++ b/go.mod @@ -13,6 +13,8 @@ require ( github.com/aws/aws-sdk-go-v2 v1.42.0 github.com/aws/aws-sdk-go-v2/credentials v1.19.25 github.com/aws/aws-sdk-go-v2/service/s3 v1.104.1 + github.com/aws/aws-sdk-go-v2/service/sns v1.40.2 + github.com/aws/aws-sdk-go-v2/service/sqs v1.44.1 github.com/aws/smithy-go v1.27.1 github.com/compose-spec/compose-go/v2 v2.12.1 github.com/go-playground/validator/v10 v10.30.3 @@ -21,7 +23,10 @@ require ( github.com/jackc/pgx/v5 v5.10.0 github.com/moby/moby/api v1.54.2 github.com/moby/moby/client v0.4.1 + github.com/nats-io/nats.go v1.52.0 github.com/spf13/cobra v1.10.2 + github.com/twmb/franz-go v1.21.4 + github.com/twmb/franz-go/pkg/kadm v1.18.0 github.com/zalando/go-keyring v0.2.8 golang.org/x/mod v0.37.0 golang.org/x/sync v0.20.0 @@ -74,6 +79,7 @@ require ( github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/klauspost/compress v1.18.6 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/lucasb-eyer/go-colorful v1.3.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect @@ -86,14 +92,18 @@ require ( github.com/muesli/mango-cobra v1.2.0 // indirect github.com/muesli/mango-pflag v0.1.0 // indirect github.com/muesli/roff v0.1.0 // indirect + github.com/nats-io/nkeys v0.4.15 // indirect + github.com/nats-io/nuid v1.0.1 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/pierrec/lz4/v4 v4.1.26 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/rivo/uniseg v0.4.7 // indirect github.com/santhosh-tekuri/jsonschema/v6 v6.0.1 // indirect github.com/sirupsen/logrus v1.9.3 // indirect github.com/spf13/pflag v1.0.9 // indirect + github.com/twmb/franz-go/pkg/kmsg v1.13.1 // indirect github.com/xhit/go-str2duration/v2 v2.1.0 // indirect github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect go.opentelemetry.io/auto/sdk v1.1.0 // indirect diff --git a/go.sum b/go.sum index 6821687..e0f5360 100644 --- a/go.sum +++ b/go.sum @@ -44,6 +44,10 @@ github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.30 h1:4HbXxyipSYxex github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.30/go.mod h1:G7RP+uhagpKtKhd1BM9N6JQqjCcGEU47K5lBVZQyRQw= github.com/aws/aws-sdk-go-v2/service/s3 v1.104.1 h1:yb03KevaOAG5e8suo79Af74vjIQvoeKmjl79WQchLrs= github.com/aws/aws-sdk-go-v2/service/s3 v1.104.1/go.mod h1:mreYODw0Y4yv7xeczvqC6vciwFao8lPE9k1l1ulfY6E= +github.com/aws/aws-sdk-go-v2/service/sns v1.40.2 h1:00dZG/qsR/Uwn5SSF6DKnV2uazRaI8JA7kS+nV4sg30= +github.com/aws/aws-sdk-go-v2/service/sns v1.40.2/go.mod h1:V9szvM64GdG5VJUeDRstvLmt/ozgWiSNg3gYnp3mSkk= +github.com/aws/aws-sdk-go-v2/service/sqs v1.44.1 h1:Iu9jGoXkvMGZJ7kJLruiPkNQ964pvuqYdoClTMIkj8M= +github.com/aws/aws-sdk-go-v2/service/sqs v1.44.1/go.mod h1:d7eKytKiwDFJeNAP4VWo47VCiNM9z539tkoCGJ6PjXw= github.com/aws/smithy-go v1.27.1 h1:4T340VFndXtADGF52gYa1POyL7s9E4Z1OeZ1hCscIw8= github.com/aws/smithy-go v1.27.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/aymanbagabas/go-udiff v0.4.1 h1:OEIrQ8maEeDBXQDoGCbbTTXYJMYRCRO1fnodZ12Gv5o= @@ -147,6 +151,8 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= github.com/lucasb-eyer/go-colorful v1.3.0 h1:2/yBRLdWBZKrf7gB40FoiKfAWYQ0lqNcbuQwVHXptag= @@ -175,12 +181,20 @@ github.com/muesli/mango-pflag v0.1.0 h1:UADqbYgpUyRoBja3g6LUL+3LErjpsOwaC9ywvBWe github.com/muesli/mango-pflag v0.1.0/go.mod h1:YEQomTxaCUp8PrbhFh10UfbhbQrM/xJ4i2PB8VTLLW0= github.com/muesli/roff v0.1.0 h1:YD0lalCotmYuF5HhZliKWlIx7IEhiXeSfq7hNjFqGF8= github.com/muesli/roff v0.1.0/go.mod h1:pjAHQM9hdUUwm/krAfrLGgJkXJ+YuhtsfZ42kieB2Ig= +github.com/nats-io/nats.go v1.52.0 h1:n3avV4VBsCgsdwh71TppsTwtv+QdPs7ntSKM8qJLGsc= +github.com/nats-io/nats.go v1.52.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= +github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= +github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M= +github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY= +github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= @@ -203,6 +217,12 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/twmb/franz-go v1.21.4 h1:skglTjGHOHHKxVdUG3A563gynBDhvSWFBBHXKOOMS8M= +github.com/twmb/franz-go v1.21.4/go.mod h1:rfoMTnVk7107fhTGxfEKIHP/e7tPe6oyij/ywzO0czk= +github.com/twmb/franz-go/pkg/kadm v1.18.0 h1:WRf/LZmDdcDXwX7WMbtDU++v+b3NzYh2bCGoPMmzirw= +github.com/twmb/franz-go/pkg/kadm v1.18.0/go.mod h1:XeLhGoLXLFzK8/ryv5FfpxPxGwj4oFEGpPJMB/x6KDE= +github.com/twmb/franz-go/pkg/kmsg v1.13.1 h1:fG5kItwysTk5UXqVwb64EpQEy3TydF3vYYK21nUQ+bI= +github.com/twmb/franz-go/pkg/kmsg v1.13.1/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY= github.com/xhit/go-str2duration/v2 v2.1.0 h1:lxklc02Drh6ynqX+DdPyp5pCKLUQpRT8bp8Ydu2Bstc= github.com/xhit/go-str2duration/v2 v2.1.0/go.mod h1:ohY8p+0f07DiV6Em5LKB0s2YpLtXVyJfNt1+BlmyAsU= github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no= diff --git a/internal/cli/messaging.go b/internal/cli/messaging.go new file mode 100644 index 0000000..63ce580 --- /dev/null +++ b/internal/cli/messaging.go @@ -0,0 +1,129 @@ +package cli + +import ( + "fmt" + + "github.com/spf13/cobra" + + "github.com/open-source-cloud/devstack/internal/config" + "github.com/open-source-cloud/devstack/internal/orchestrate" + "github.com/open-source-cloud/devstack/internal/state" +) + +// This file holds the shared plumbing for the messaging resource groups (spec 29 +// §messaging): queue / topic / stream. Each verb group mirrors the db/s3 flow — +// create/rm go through the up-saga's lock → overlay → provisioner → ledger → event +// path via internal/orchestrate; list is a lock-free ledger read. The DECISION is +// NATIVE-by-default: NATS backs queues+streams, Redpanda/Kafka backs topics+kafka +// streams, and LocalStack (SQS/SNS) is opt-in via --engine sqs|sns. A bare verb +// with no --engine infers from the ACTIVE shared engines (never auto-starting one) +// and errors clearly when unsatisfiable. + +// msgPrefixed computes the DNS-safe tenant-scoped resource name - +// (spec 29 §tenant naming) — hyphen-separated, valid for NATS streams, Kafka +// topics, SQS/SNS names and Redis keys alike — unless --no-prefix keeps the literal. +func msgPrefixed(project, name string, noPrefix bool) string { + if noPrefix { + return name + } + return project + "-" + name +} + +// engineForFlag maps a user-facing --engine value to the registry engine key. The +// LocalStack provisioner is keyed by the template name "localstack" (its +// `provides: aws` endpoint), so sqs/sns/aws all resolve there. +func engineForFlag(flag string) string { + switch flag { + case "sqs", "sns", "aws", "localstack": + return "localstack" + case "nats": + return "nats" + case "kafka": + return "kafka" + case "redis": + return "redis" + default: + return flag + } +} + +// inferEngine picks the first engine in the preference order that has a live shared +// instance in the workspace (never auto-starting one). Returns "" if none match. +func inferEngine(d orchestrate.UpDeps, order []string) string { + for _, e := range order { + if _, ok := orchestrate.ResolveInstance(d.Model, e); ok { + return e + } + } + return "" +} + +// resolveMsgEngine turns the --engine flag (validated against allowed) plus the +// native-default inference order into a concrete registry engine key, or a clear +// error when the flag is unknown or nothing is inferable. +func resolveMsgEngine(d orchestrate.UpDeps, flag string, allowed []string, order []string, domain string) (string, error) { + if flag != "" { + if !contains(allowed, flag) { + return "", fmt.Errorf("invalid --engine %q for %s (want one of %v)", flag, domain, allowed) + } + engine := engineForFlag(flag) + if _, ok := orchestrate.ResolveInstance(d.Model, engine); !ok { + return "", fmt.Errorf("no shared %s engine (for --engine %s) in this workspace; declare one under workspace.shared and run `devstack up` (devstack never auto-starts an engine)", engine, flag) + } + return engine, nil + } + engine := inferEngine(d, order) + if engine == "" { + return "", fmt.Errorf("no active engine to back a %s in this workspace (tried %v); add one under workspace.shared and run `devstack up`, or pass --engine", domain, order) + } + return engine, nil +} + +// listMessagingRows returns the project's provisioned rows of the given kind +// (queue|topic|stream) — a lock-free ledger read, prefix-scoped to the project +// unless --all. +func listMessagingRows(cmd *cobra.Command, kind, project string, all bool) ([]state.Provisioned, string, error) { + mgr, closeFn, err := buildManager(cmd) + if err != nil { + return nil, "", err + } + defer closeFn() + proj := project + if proj == "" { + proj = defaultProjectFromModel(mgr.Model) + } + var rows []state.Provisioned + if proj != "" && !all { + rows, err = mgr.DB.ProvisionedFor(proj) + } else { + rows, err = mgr.DB.AllProvisioned() + } + if err != nil { + return nil, proj, err + } + var out []state.Provisioned + for _, r := range rows { + if r.Kind == kind { + out = append(out, r) + } + } + return out, proj, nil +} + +// defaultProjectFromModel returns the workspace's single/first project by name. +func defaultProjectFromModel(m *config.Model) string { + names := sortedProjectNames(m) + if len(names) > 0 { + return names[0] + } + return "" +} + +func contains(s []string, v string) bool { + for _, x := range s { + if x == v { + return true + } + } + return false +} diff --git a/internal/cli/messaging_test.go b/internal/cli/messaging_test.go new file mode 100644 index 0000000..b06e1ce --- /dev/null +++ b/internal/cli/messaging_test.go @@ -0,0 +1,164 @@ +package cli + +import ( + "testing" + + "github.com/open-source-cloud/devstack/internal/config" + "github.com/open-source-cloud/devstack/internal/orchestrate" +) + +func TestMessagingGroupsRegistered(t *testing.T) { + root := NewRootCmd(Options{}) + for _, path := range [][]string{ + {"queue", "create"}, {"queue", "list"}, {"queue", "rm"}, + {"topic", "create"}, {"topic", "list"}, {"topic", "rm"}, + {"stream", "create"}, {"stream", "list"}, {"stream", "rm"}, + } { + c, _, err := root.Find(path) + if err != nil || c.RunE == nil { + t.Fatalf("%v not registered as a real command: %v", path, err) + } + } +} + +func TestMessagingCreateFlags(t *testing.T) { + root := NewRootCmd(Options{}) + cases := map[string][]string{ + "queue": {"engine", "fifo", "dlq", "max-receive", "no-prefix", "project"}, + "topic": {"engine", "subscribe", "no-prefix", "project"}, + "stream": {"engine", "partitions", "replicas", "retention", "no-prefix", "project"}, + } + for group, flags := range cases { + c, _, err := root.Find([]string{group, "create"}) + if err != nil { + t.Fatalf("%s create: %v", group, err) + } + for _, f := range flags { + if c.Flags().Lookup(f) == nil { + t.Errorf("%s create missing --%s", group, f) + } + } + } +} + +// loadModelWS builds a config.Model from a temp workspace with the given shared +// block, so engine-inference can be exercised without a live daemon. +func loadModelWS(t *testing.T, sharedYAML string) *config.Model { + t.Helper() + root := writeWS(t, + "apiVersion: devstack/v1\nkind: Workspace\nname: demo\n"+sharedYAML+"projects:\n - { name: web, path: web }\n", + map[string]string{"web": "apiVersion: devstack/v1\nkind: Project\nname: web\nservices:\n app: { template: node.vite }\n"}, + ) + m, err := config.LoadAt(root) + if err != nil { + t.Fatalf("load: %v", err) + } + return m +} + +func TestMessagingEngineInference(t *testing.T) { + tests := []struct { + name string + sharedYAML string + allowed []string + order []string + flag string + wantEngine string + wantErr bool + }{ + { + name: "queue infers nats when present (native default)", + sharedYAML: "shared:\n nats: { template: nats }\n redis: { template: redis }\n", + allowed: queueEngines, order: queueOrder, flag: "", wantEngine: "nats", + }, + { + name: "queue falls back to redis when no nats", + sharedYAML: "shared:\n redis: { template: redis }\n", + allowed: queueEngines, order: queueOrder, flag: "", wantEngine: "redis", + }, + { + name: "explicit --engine sqs maps to localstack instance", + sharedYAML: "shared:\n localstack: { template: localstack }\n", + allowed: queueEngines, order: queueOrder, flag: "sqs", wantEngine: "localstack", + }, + { + name: "stream infers nats over kafka", + sharedYAML: "shared:\n kafka: { template: kafka }\n nats: { template: nats }\n", + allowed: streamEngines, order: streamOrder, flag: "", wantEngine: "nats", + }, + { + name: "topic infers kafka (native default)", + sharedYAML: "shared:\n kafka: { template: kafka }\n", + allowed: topicEngines, order: topicOrder, flag: "", wantEngine: "kafka", + }, + { + name: "unsatisfiable inference errors (no engine, never auto-starts)", + sharedYAML: "shared:\n postgres: { template: postgres }\n", + allowed: queueEngines, order: queueOrder, flag: "", wantErr: true, + }, + { + name: "invalid --engine value errors", + sharedYAML: "shared:\n nats: { template: nats }\n", + allowed: queueEngines, order: queueOrder, flag: "pulsar", wantErr: true, + }, + { + name: "explicit engine not in workspace errors (never auto-starts)", + sharedYAML: "shared:\n nats: { template: nats }\n", + allowed: queueEngines, order: queueOrder, flag: "sqs", wantErr: true, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + d := orchestrate.UpDeps{Model: loadModelWS(t, tc.sharedYAML)} + eng, err := resolveMsgEngine(d, tc.flag, tc.allowed, tc.order, "queue") + if tc.wantErr { + if err == nil { + t.Fatalf("want error, got engine %q", eng) + } + return + } + if err != nil { + t.Fatalf("resolveMsgEngine: %v", err) + } + if eng != tc.wantEngine { + t.Errorf("engine = %q, want %q", eng, tc.wantEngine) + } + }) + } +} + +func TestEngineForFlagAndPrefix(t *testing.T) { + for flag, want := range map[string]string{ + "sqs": "localstack", "sns": "localstack", "nats": "nats", "kafka": "kafka", "redis": "redis", + } { + if got := engineForFlag(flag); got != want { + t.Errorf("engineForFlag(%q) = %q, want %q", flag, got, want) + } + } + if got := msgPrefixed("web", "jobs", false); got != "web-jobs" { + t.Errorf("msgPrefixed = %q, want web-jobs", got) + } + if got := msgPrefixed("web", "external", true); got != "external" { + t.Errorf("msgPrefixed --no-prefix = %q, want external", got) + } +} + +// TestValidateStreamFlags asserts the parse-time guard: --partitions/--replicas are +// Kafka-only (rejected for NATS) and --retention must parse (spec 29). +func TestValidateStreamFlags(t *testing.T) { + if err := validateStreamFlags("nats", true, false, ""); err == nil { + t.Error("--partitions with --engine nats must error") + } + if err := validateStreamFlags("nats", false, true, ""); err == nil { + t.Error("--replicas with --engine nats must error") + } + if err := validateStreamFlags("kafka", true, true, "168h"); err != nil { + t.Errorf("kafka with partitions/replicas/retention must be valid: %v", err) + } + if err := validateStreamFlags("nats", false, false, "168h"); err != nil { + t.Errorf("nats with only --retention must be valid: %v", err) + } + if err := validateStreamFlags("nats", false, false, "garbage"); err == nil { + t.Error("invalid --retention must error") + } +} diff --git a/internal/cli/queue.go b/internal/cli/queue.go new file mode 100644 index 0000000..a55f8c9 --- /dev/null +++ b/internal/cli/queue.go @@ -0,0 +1,198 @@ +package cli + +import ( + "fmt" + + "github.com/spf13/cobra" + + "github.com/open-source-cloud/devstack/internal/orchestrate" + "github.com/open-source-cloud/devstack/internal/resource" +) + +// newQueueCmd wires the `devstack queue` group (spec 29 §messaging — queues): +// tenant-scoped queues on the NATIVE default NATS (durable work-queue), a Redis +// list namespace, or opt-in LocalStack SQS. create/rm go through the flock via +// internal/orchestrate; list is a lock-free ledger read. Names are project-PREFIXED +// unless --no-prefix. The ledger kind is `queue` (free-text; no migration). +func newQueueCmd(g *GlobalOpts) *cobra.Command { + cmd := &cobra.Command{ + Use: "queue", + Short: "Tenant-scoped queues on the shared NATS / Redis / SQS engines", + } + cmd.AddCommand(newQueueCreateCmd(g), newQueueListCmd(g), newQueueRmCmd(g)) + return cmd +} + +// queueEngines are the accepted --engine values; queueOrder is the native-default +// inference order (prefer NATS, then Redis, then LocalStack SQS). +var ( + queueEngines = []string{"nats", "sqs", "redis"} + queueOrder = []string{"nats", "redis", "localstack"} +) + +func newQueueCreateCmd(g *GlobalOpts) *cobra.Command { + var project, engine, dlq string + var fifo, noPrefix bool + var maxReceive int + cmd := &cobra.Command{ + Use: "create ", + Short: "Create a tenant queue (idempotent). Engine inferred from active engines unless --engine.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + d, closeFn, err := buildUpDeps(cmd) + if err != nil { + return err + } + defer closeFn() + proj := project + if proj == "" { + proj = defaultProject(d) + } + eng, err := resolveMsgEngine(d, engine, queueEngines, queueOrder, "queue") + if err != nil { + return err + } + name := msgPrefixed(proj, args[0], noPrefix) + params := map[string]any{} + if fifo { + params["fifo"] = true + } + // A DLQ (SQS only) is created first as its own recorded resource, then the + // main queue references it for a redrive policy (spec 29 §SQS FIFO/DLQ). + var dlqPhysical string + if dlq != "" { + if eng != "localstack" { + return fmt.Errorf("--dlq is only supported for --engine sqs") + } + dlqPhysical = msgPrefixed(proj, dlq, noPrefix) + if _, err := orchestrate.CreateResource(cmd.Context(), d, resource.Resource{ + Engine: eng, Kind: "queue", Name: dlqPhysical, Owner: proj, + Params: map[string]any{"fifo": fifo}, CredKind: resource.CredPredictable, + }); err != nil { + return fmt.Errorf("create DLQ %q: %w", dlqPhysical, err) + } + params["dlq"] = dlqPhysical + if maxReceive > 0 { + params["max_receive"] = maxReceive + } + } + r := resource.Resource{ + Engine: eng, Kind: "queue", Name: name, Owner: proj, + Params: params, CredKind: resource.CredPredictable, + } + attrs, err := orchestrate.CreateResource(cmd.Context(), d, r) + if err != nil { + return err + } + if g.JSON { + out := map[string]any{"kind": "queue", "name": name, "project": proj, "engine": eng, "endpoint": attrs["endpoint"]} + if dlqPhysical != "" { + out["dlq"] = dlqPhysical + } + if attrs["queueUrl"] != "" { + out["queueUrl"] = attrs["queueUrl"] + } + if attrs["hint"] != "" { + out["hint"] = attrs["hint"] + } + return writeJSON(cmd, out) + } + w := cmd.OutOrStdout() + fmt.Fprintf(w, "created queue %q on %s\n", name, eng) + if attrs["queueUrl"] != "" { + fmt.Fprintf(w, "url: %s\n", attrs["queueUrl"]) + } + if attrs["hint"] != "" { + fmt.Fprintf(w, "hint: %s\n", attrs["hint"]) + } + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().StringVar(&engine, "engine", "", "queue backend: nats|sqs|redis (default: inferred from active engines)") + cmd.Flags().BoolVar(&fifo, "fifo", false, "FIFO queue (SQS: appends .fifo + sets FifoQueue)") + cmd.Flags().StringVar(&dlq, "dlq", "", "dead-letter queue name (SQS only; created first)") + cmd.Flags().IntVar(&maxReceive, "max-receive", 0, "redrive maxReceiveCount for the DLQ (SQS only)") + cmd.Flags().BoolVar(&noPrefix, "no-prefix", false, "use the literal name (skip the - prefix)") + return cmd +} + +func newQueueListCmd(g *GlobalOpts) *cobra.Command { + var project string + var all bool + cmd := &cobra.Command{ + Use: "list", + Short: "List the project's provisioned queues (lock-free)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + rows, _, err := listMessagingRows(cmd, "queue", project, all) + if err != nil { + return err + } + if g.JSON { + return writeJSON(cmd, map[string]any{"queues": rows}) + } + w := cmd.OutOrStdout() + if len(rows) == 0 { + fmt.Fprintln(w, "no queues") + return nil + } + for _, r := range rows { + fmt.Fprintf(w, "%-10s %-24s %s\n", r.Project, r.Name, r.CreatedAt) + } + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().BoolVar(&all, "all", false, "list every project's queues") + return cmd +} + +func newQueueRmCmd(g *GlobalOpts) *cobra.Command { + var project, engine string + var yes, noPrefix bool + cmd := &cobra.Command{ + Use: "rm ", + Short: "Remove a tenant queue (destructive; confirm required)", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if g.JSON && !yes { + return fmt.Errorf("refusing to remove a queue without --yes for --json/non-interactive use") + } + d, closeFn, err := buildUpDeps(cmd) + if err != nil { + return err + } + defer closeFn() + proj := project + if proj == "" { + proj = defaultProject(d) + } + eng, err := resolveMsgEngine(d, engine, queueEngines, queueOrder, "queue") + if err != nil { + return err + } + name := msgPrefixed(proj, args[0], noPrefix) + if !yes { + if !confirm(cmd, fmt.Sprintf("This REMOVES queue %q on %s. Type 'yes' to continue: ", name, eng)) { + fmt.Fprintln(cmd.OutOrStdout(), "aborted") + return nil + } + } + r := resource.Resource{Engine: eng, Kind: "queue", Name: name, Owner: proj} + if err := orchestrate.DropResource(cmd.Context(), d, r, true); err != nil { + return err + } + if g.JSON { + return writeJSON(cmd, map[string]any{"removed": map[string]string{"kind": "queue", "name": name, "project": proj, "engine": eng}}) + } + fmt.Fprintf(cmd.OutOrStdout(), "removed queue %q\n", name) + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().StringVar(&engine, "engine", "", "queue backend: nats|sqs|redis (default: inferred)") + cmd.Flags().BoolVar(&yes, "yes", false, "skip the confirmation prompt") + cmd.Flags().BoolVar(&noPrefix, "no-prefix", false, "use the literal name (skip the - prefix)") + return cmd +} diff --git a/internal/cli/root.go b/internal/cli/root.go index f7de9ed..0c404e0 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -94,6 +94,9 @@ func NewRootCmd(opts Options) *cobra.Command { newResourceCmd(g), newDbCmd(g), newS3Cmd(g), + newQueueCmd(g), + newTopicCmd(g), + newStreamCmd(g), newAwsCmd(g), newWsCmd(g), newWorkspaceCmd(g), diff --git a/internal/cli/stream.go b/internal/cli/stream.go new file mode 100644 index 0000000..dd617d4 --- /dev/null +++ b/internal/cli/stream.go @@ -0,0 +1,214 @@ +package cli + +import ( + "fmt" + "time" + + "github.com/spf13/cobra" + + "github.com/open-source-cloud/devstack/internal/orchestrate" + "github.com/open-source-cloud/devstack/internal/resource" +) + +// validateStreamFlags enforces the per-engine flag rules at parse time: --partitions +// and --replicas are KAFKA-only (a NATS JetStream stream has no partition model), and +// --retention (when set) must be a valid Go duration. +func validateStreamFlags(engine string, partitionsChanged, replicasChanged bool, retention string) error { + if engine == "nats" { + if partitionsChanged { + return fmt.Errorf("--partitions is only valid for --engine kafka (a NATS stream has no partitions)") + } + if replicasChanged { + return fmt.Errorf("--replicas is only valid for --engine kafka") + } + } + if retention != "" { + if _, err := time.ParseDuration(retention); err != nil { + return fmt.Errorf("invalid --retention %q (want a Go duration like 168h): %w", retention, err) + } + } + return nil +} + +// newStreamCmd wires the `devstack stream` group (spec 29 §messaging — durable +// streams): tenant-scoped streams on the NATIVE default NATS (JetStream) or a +// Kafka (Redpanda) partitioned topic. create/rm go through the flock via internal/ +// orchestrate; list is a lock-free ledger read. Names are project-PREFIXED unless +// --no-prefix. --partitions/--replicas are KAFKA-ONLY (rejected for NATS at parse +// time); --retention maps to NATS --max-age or Kafka retention (spec 29 §NATS vs +// Kafka retention models). The ledger kind is `stream` (free-text; no migration). +func newStreamCmd(g *GlobalOpts) *cobra.Command { + cmd := &cobra.Command{ + Use: "stream", + Short: "Tenant-scoped durable streams on the shared NATS / Kafka engines", + } + cmd.AddCommand(newStreamCreateCmd(g), newStreamListCmd(g), newStreamRmCmd(g)) + return cmd +} + +// streamEngines are the accepted --engine values; streamOrder is the native-default +// inference order (prefer NATS JetStream, then Kafka). +var ( + streamEngines = []string{"nats", "kafka"} + streamOrder = []string{"nats", "kafka"} +) + +func newStreamCreateCmd(g *GlobalOpts) *cobra.Command { + var project, engine, retention string + var partitions, replicas int + var noPrefix bool + cmd := &cobra.Command{ + Use: "create ", + Short: "Create a tenant stream (idempotent). Engine inferred from active engines unless --engine.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + d, closeFn, err := buildUpDeps(cmd) + if err != nil { + return err + } + defer closeFn() + proj := project + if proj == "" { + proj = defaultProject(d) + } + eng, err := resolveMsgEngine(d, engine, streamEngines, streamOrder, "stream") + if err != nil { + return err + } + // --partitions/--replicas are a Kafka partition-model concept; a NATS + // JetStream stream has neither. Reject them for NATS at parse time + // (spec 29 §NATS work-queue vs Kafka partitions). --retention must parse. + if err := validateStreamFlags(eng, cmd.Flags().Changed("partitions"), cmd.Flags().Changed("replicas"), retention); err != nil { + return err + } + name := msgPrefixed(proj, args[0], noPrefix) + params := map[string]any{} + if retention != "" { + params["retention"] = retention + } + if partitions > 0 { + params["partitions"] = partitions + } + if replicas > 0 { + params["replicas"] = replicas + } + r := resource.Resource{ + Engine: eng, Kind: "stream", Name: name, Owner: proj, + Params: params, CredKind: resource.CredPredictable, + } + attrs, err := orchestrate.CreateResource(cmd.Context(), d, r) + if err != nil { + return err + } + if g.JSON { + out := map[string]any{"kind": "stream", "name": name, "project": proj, "engine": eng, "endpoint": attrs["endpoint"]} + if retention != "" { + out["retention"] = retention + } + if attrs["subjects"] != "" { + out["subjects"] = attrs["subjects"] + } + if attrs["partitions"] != "" { + out["partitions"] = attrs["partitions"] + } + return writeJSON(cmd, out) + } + w := cmd.OutOrStdout() + fmt.Fprintf(w, "created stream %q on %s\n", name, eng) + if attrs["subjects"] != "" { + fmt.Fprintf(w, "subjects: %s\n", attrs["subjects"]) + } + if attrs["partitions"] != "" { + fmt.Fprintf(w, "partitions: %s\n", attrs["partitions"]) + } + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().StringVar(&engine, "engine", "", "stream backend: nats|kafka (default: inferred from active engines)") + cmd.Flags().IntVar(&partitions, "partitions", 0, "partition count (KAFKA only)") + cmd.Flags().IntVar(&replicas, "replicas", 0, "replication factor (KAFKA only)") + cmd.Flags().StringVar(&retention, "retention", "", "retention window as a Go duration (e.g. 168h)") + cmd.Flags().BoolVar(&noPrefix, "no-prefix", false, "use the literal name (skip the - prefix)") + return cmd +} + +func newStreamListCmd(g *GlobalOpts) *cobra.Command { + var project string + var all bool + cmd := &cobra.Command{ + Use: "list", + Short: "List the project's provisioned streams (lock-free)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + rows, _, err := listMessagingRows(cmd, "stream", project, all) + if err != nil { + return err + } + if g.JSON { + return writeJSON(cmd, map[string]any{"streams": rows}) + } + w := cmd.OutOrStdout() + if len(rows) == 0 { + fmt.Fprintln(w, "no streams") + return nil + } + for _, r := range rows { + fmt.Fprintf(w, "%-10s %-24s %s\n", r.Project, r.Name, r.CreatedAt) + } + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().BoolVar(&all, "all", false, "list every project's streams") + return cmd +} + +func newStreamRmCmd(g *GlobalOpts) *cobra.Command { + var project, engine string + var yes, noPrefix bool + cmd := &cobra.Command{ + Use: "rm ", + Short: "Remove a tenant stream (destructive; confirm required)", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if g.JSON && !yes { + return fmt.Errorf("refusing to remove a stream without --yes for --json/non-interactive use") + } + d, closeFn, err := buildUpDeps(cmd) + if err != nil { + return err + } + defer closeFn() + proj := project + if proj == "" { + proj = defaultProject(d) + } + eng, err := resolveMsgEngine(d, engine, streamEngines, streamOrder, "stream") + if err != nil { + return err + } + name := msgPrefixed(proj, args[0], noPrefix) + if !yes { + if !confirm(cmd, fmt.Sprintf("This REMOVES stream %q on %s. Type 'yes' to continue: ", name, eng)) { + fmt.Fprintln(cmd.OutOrStdout(), "aborted") + return nil + } + } + r := resource.Resource{Engine: eng, Kind: "stream", Name: name, Owner: proj} + if err := orchestrate.DropResource(cmd.Context(), d, r, true); err != nil { + return err + } + if g.JSON { + return writeJSON(cmd, map[string]any{"removed": map[string]string{"kind": "stream", "name": name, "project": proj, "engine": eng}}) + } + fmt.Fprintf(cmd.OutOrStdout(), "removed stream %q\n", name) + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().StringVar(&engine, "engine", "", "stream backend: nats|kafka (default: inferred)") + cmd.Flags().BoolVar(&yes, "yes", false, "skip the confirmation prompt") + cmd.Flags().BoolVar(&noPrefix, "no-prefix", false, "use the literal name (skip the - prefix)") + return cmd +} diff --git a/internal/cli/topic.go b/internal/cli/topic.go new file mode 100644 index 0000000..2bdb973 --- /dev/null +++ b/internal/cli/topic.go @@ -0,0 +1,176 @@ +package cli + +import ( + "fmt" + + "github.com/spf13/cobra" + + "github.com/open-source-cloud/devstack/internal/orchestrate" + "github.com/open-source-cloud/devstack/internal/resource" +) + +// newTopicCmd wires the `devstack topic` group (spec 29 §messaging — pub/sub +// topics): tenant-scoped topics on the NATIVE default Kafka (Redpanda), a NATS +// subject convention, or opt-in LocalStack SNS. create/rm go through the flock via +// internal/orchestrate; list is a lock-free ledger read. Names are project-PREFIXED +// unless --no-prefix. The ledger kind is `topic` (free-text; no migration). +func newTopicCmd(g *GlobalOpts) *cobra.Command { + cmd := &cobra.Command{ + Use: "topic", + Short: "Tenant-scoped pub/sub topics on the shared Kafka / NATS / SNS engines", + } + cmd.AddCommand(newTopicCreateCmd(g), newTopicListCmd(g), newTopicRmCmd(g)) + return cmd +} + +// topicEngines are the accepted --engine values; topicOrder is the native-default +// inference order (prefer Kafka, then NATS subjects, then LocalStack SNS). +var ( + topicEngines = []string{"sns", "kafka", "nats"} + topicOrder = []string{"kafka", "nats", "localstack"} +) + +func newTopicCreateCmd(g *GlobalOpts) *cobra.Command { + var project, engine, subscribe string + var noPrefix bool + cmd := &cobra.Command{ + Use: "create ", + Short: "Create a tenant topic (idempotent). Engine inferred from active engines unless --engine.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + d, closeFn, err := buildUpDeps(cmd) + if err != nil { + return err + } + defer closeFn() + proj := project + if proj == "" { + proj = defaultProject(d) + } + eng, err := resolveMsgEngine(d, engine, topicEngines, topicOrder, "topic") + if err != nil { + return err + } + name := msgPrefixed(proj, args[0], noPrefix) + params := map[string]any{} + if subscribe != "" { + if eng != "localstack" { + return fmt.Errorf("--subscribe (SNS→SQS fan-out) is only supported for --engine sns") + } + params["subscribe"] = msgPrefixed(proj, subscribe, noPrefix) + } + r := resource.Resource{ + Engine: eng, Kind: "topic", Name: name, Owner: proj, + Params: params, CredKind: resource.CredPredictable, + } + attrs, err := orchestrate.CreateResource(cmd.Context(), d, r) + if err != nil { + return err + } + if g.JSON { + out := map[string]any{"kind": "topic", "name": name, "project": proj, "engine": eng, "endpoint": attrs["endpoint"]} + if attrs["topicArn"] != "" { + out["topicArn"] = attrs["topicArn"] + } + if attrs["subscribed"] != "" { + out["subscribed"] = attrs["subscribed"] + } + return writeJSON(cmd, out) + } + w := cmd.OutOrStdout() + fmt.Fprintf(w, "created topic %q on %s\n", name, eng) + if attrs["topicArn"] != "" { + fmt.Fprintf(w, "arn: %s\n", attrs["topicArn"]) + } + if attrs["subscribed"] != "" { + fmt.Fprintf(w, "subscribed: %s\n", attrs["subscribed"]) + } + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().StringVar(&engine, "engine", "", "topic backend: sns|kafka|nats (default: inferred from active engines)") + cmd.Flags().StringVar(&subscribe, "subscribe", "", "SQS queue to subscribe (SNS only; the classic SNS→SQS fan-out)") + cmd.Flags().BoolVar(&noPrefix, "no-prefix", false, "use the literal name (skip the - prefix)") + return cmd +} + +func newTopicListCmd(g *GlobalOpts) *cobra.Command { + var project string + var all bool + cmd := &cobra.Command{ + Use: "list", + Short: "List the project's provisioned topics (lock-free)", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + rows, _, err := listMessagingRows(cmd, "topic", project, all) + if err != nil { + return err + } + if g.JSON { + return writeJSON(cmd, map[string]any{"topics": rows}) + } + w := cmd.OutOrStdout() + if len(rows) == 0 { + fmt.Fprintln(w, "no topics") + return nil + } + for _, r := range rows { + fmt.Fprintf(w, "%-10s %-24s %s\n", r.Project, r.Name, r.CreatedAt) + } + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().BoolVar(&all, "all", false, "list every project's topics") + return cmd +} + +func newTopicRmCmd(g *GlobalOpts) *cobra.Command { + var project, engine string + var yes, noPrefix bool + cmd := &cobra.Command{ + Use: "rm ", + Short: "Remove a tenant topic (destructive; confirm required)", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if g.JSON && !yes { + return fmt.Errorf("refusing to remove a topic without --yes for --json/non-interactive use") + } + d, closeFn, err := buildUpDeps(cmd) + if err != nil { + return err + } + defer closeFn() + proj := project + if proj == "" { + proj = defaultProject(d) + } + eng, err := resolveMsgEngine(d, engine, topicEngines, topicOrder, "topic") + if err != nil { + return err + } + name := msgPrefixed(proj, args[0], noPrefix) + if !yes { + if !confirm(cmd, fmt.Sprintf("This REMOVES topic %q on %s. Type 'yes' to continue: ", name, eng)) { + fmt.Fprintln(cmd.OutOrStdout(), "aborted") + return nil + } + } + r := resource.Resource{Engine: eng, Kind: "topic", Name: name, Owner: proj} + if err := orchestrate.DropResource(cmd.Context(), d, r, true); err != nil { + return err + } + if g.JSON { + return writeJSON(cmd, map[string]any{"removed": map[string]string{"kind": "topic", "name": name, "project": proj, "engine": eng}}) + } + fmt.Fprintf(cmd.OutOrStdout(), "removed topic %q\n", name) + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "owner project") + cmd.Flags().StringVar(&engine, "engine", "", "topic backend: sns|kafka|nats (default: inferred)") + cmd.Flags().BoolVar(&yes, "yes", false, "skip the confirmation prompt") + cmd.Flags().BoolVar(&noPrefix, "no-prefix", false, "use the literal name (skip the - prefix)") + return cmd +} diff --git a/internal/orchestrate/messaging_ops_test.go b/internal/orchestrate/messaging_ops_test.go new file mode 100644 index 0000000..f353102 --- /dev/null +++ b/internal/orchestrate/messaging_ops_test.go @@ -0,0 +1,180 @@ +package orchestrate + +import ( + "context" + "os" + "path/filepath" + "slices" + "testing" + + "github.com/open-source-cloud/devstack/internal/config" + "github.com/open-source-cloud/devstack/internal/docker" + "github.com/open-source-cloud/devstack/internal/generate" + "github.com/open-source-cloud/devstack/internal/resource" + "github.com/open-source-cloud/devstack/internal/state" + "github.com/open-source-cloud/devstack/internal/template" + "github.com/open-source-cloud/devstack/internal/workspace" + "github.com/open-source-cloud/devstack/templates" +) + +// --- minimal in-memory messaging admins (no live broker/endpoint) ------------- + +type fakeNatsAdmin struct{ streams map[string]bool } + +func (f *fakeNatsAdmin) EnsureStream(_ context.Context, s resource.StreamSpec) error { + f.streams[s.Name] = true + return nil +} +func (f *fakeNatsAdmin) DeleteStream(_ context.Context, name string) error { + delete(f.streams, name) + return nil +} +func (f *fakeNatsAdmin) ListStreams(context.Context) ([]string, error) { + var out []string + for n := range f.streams { + out = append(out, n) + } + return out, nil +} +func (f *fakeNatsAdmin) EnsureConsumer(context.Context, string, resource.ConsumerSpec) error { + return nil +} +func (f *fakeNatsAdmin) DeleteConsumer(context.Context, string, string) error { return nil } +func (f *fakeNatsAdmin) Close() error { return nil } + +type fakeKafkaAdmin struct{ topics map[string]bool } + +func (f *fakeKafkaAdmin) CreateTopic(_ context.Context, name string, _, _ int) error { + f.topics[name] = true + return nil +} +func (f *fakeKafkaAdmin) DeleteTopic(_ context.Context, name string) error { + delete(f.topics, name) + return nil +} +func (f *fakeKafkaAdmin) ListTopics(context.Context) ([]string, error) { + var out []string + for n := range f.topics { + out = append(out, n) + } + return out, nil +} +func (f *fakeKafkaAdmin) Close() error { return nil } + +// msgFixture builds a workspace with nats + kafka shared instances and injects the +// fake admin factories so CreateResource runs daemon-free. +func msgFixture(t *testing.T) (UpDeps, *state.DB, *fakeNatsAdmin, *fakeKafkaAdmin) { + t.Helper() + root := t.TempDir() + write := func(rel, body string) { + p := filepath.Join(root, rel) + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + } + write("workspace.yaml", "apiVersion: devstack/v1\nkind: Workspace\nname: demo\nshared:\n nats: { template: nats }\n kafka: { template: kafka }\nprojects:\n - { name: web, path: web }\n") + write("web/devstack.yaml", "apiVersion: devstack/v1\nkind: Project\nname: web\nservices:\n app:\n template: node.vite\n uses: [workspace.shared.nats]\n") + + m, err := config.LoadAt(root) + if err != nil { + t.Fatalf("load: %v", err) + } + db, err := state.Open(context.Background(), filepath.Join(root, "state"), "ctx") + if err != nil { + t.Fatalf("state: %v", err) + } + t.Cleanup(func() { db.Close() }) + + mc := &docker.MockClient{ + Containers: []docker.Container{ + {ID: "n1", Name: "devstack-shared-nats-1", State: "running", Labels: map[string]string{generate.LabelManaged: "true", generate.LabelShared: "nats"}}, + {ID: "k1", Name: "devstack-shared-kafka-1", State: "running", Labels: map[string]string{generate.LabelManaged: "true", generate.LabelShared: "kafka"}}, + }, + Details: map[string]docker.ContainerDetails{ + "n1": {ID: "n1", State: "running", Running: true, Health: docker.HealthHealthy}, + "k1": {ID: "k1", State: "running", Running: true, Health: docker.HealthHealthy}, + }, + } + src := template.NewFSSource(templates.FS) + lockPath := filepath.Join(root, "lock") + mgr := &workspace.Manager{Model: m, DB: db, Docker: mc, Source: src, LockPath: lockPath} + fn := &fakeNatsAdmin{streams: map[string]bool{}} + fk := &fakeKafkaAdmin{topics: map[string]bool{}} + d := UpDeps{ + Model: m, DB: db, Docker: mc, Manager: mgr, Source: src, + LockPath: lockPath, Runner: &fakeRunner{}, Env: map[string]string{}, + NatsFactory: func(context.Context, resource.Target) (resource.NatsAdmin, error) { return fn, nil }, + KafkaFactory: func(context.Context, resource.Target) (resource.KafkaAdmin, error) { return fk, nil }, + } + return d, db, fn, fk +} + +func TestCreateStreamNatsRecordsLedgerKind(t *testing.T) { + d, db, fn, _ := msgFixture(t) + _, err := CreateResource(context.Background(), d, resource.Resource{ + Engine: "nats", Kind: "stream", Name: "web-orders", Owner: "web", + Params: map[string]any{"retention": "168h"}, CredKind: resource.CredPredictable, + }) + if err != nil { + t.Fatalf("CreateResource stream: %v", err) + } + if !fn.streams["web-orders"] { + t.Errorf("nats stream not created: %v", fn.streams) + } + rows, _ := db.ProvisionedFor("web") + var kinds []string + for _, r := range rows { + kinds = append(kinds, r.Kind+":"+r.Name) + } + if !slices.Contains(kinds, "stream:web-orders") { + t.Errorf("ledger kind=stream not recorded: %v", kinds) + } +} + +func TestCreateTopicKafkaRecordsLedgerKind(t *testing.T) { + d, db, _, fk := msgFixture(t) + _, err := CreateResource(context.Background(), d, resource.Resource{ + Engine: "kafka", Kind: "topic", Name: "web-events", Owner: "web", + Params: map[string]any{"partitions": 6}, CredKind: resource.CredPredictable, + }) + if err != nil { + t.Fatalf("CreateResource topic: %v", err) + } + if !fk.topics["web-events"] { + t.Errorf("kafka topic not created: %v", fk.topics) + } + rows, _ := db.ProvisionedFor("web") + var kinds []string + for _, r := range rows { + kinds = append(kinds, r.Kind+":"+r.Name) + } + if !slices.Contains(kinds, "topic:web-events") { + t.Errorf("ledger kind=topic not recorded: %v", kinds) + } +} + +func TestMessagingDropUntracksLedgerRow(t *testing.T) { + d, db, fn, _ := msgFixture(t) + ctx := context.Background() + _, err := CreateResource(ctx, d, resource.Resource{ + Engine: "nats", Kind: "queue", Name: "web-jobs", Owner: "web", CredKind: resource.CredPredictable, + }) + if err != nil { + t.Fatalf("create: %v", err) + } + if err := DropResource(ctx, d, resource.Resource{Engine: "nats", Kind: "queue", Name: "web-jobs", Owner: "web"}, true); err != nil { + t.Fatalf("drop: %v", err) + } + if fn.streams["web-jobs"] { + t.Error("nats stream not dropped on purge") + } + rows, _ := db.ProvisionedFor("web") + for _, r := range rows { + if r.Kind == "queue" && r.Name == "web-jobs" { + t.Errorf("queue ledger row not reclaimed: %v", rows) + } + } +} diff --git a/internal/orchestrate/resource_ops.go b/internal/orchestrate/resource_ops.go index 5873bc1..5e0762a 100644 --- a/internal/orchestrate/resource_ops.go +++ b/internal/orchestrate/resource_ops.go @@ -41,10 +41,15 @@ func ResolveInstance(m *config.Model, engine string) (string, bool) { // instance's params don't set rootUser/rootPassword (matching the engine // templates: postgres → devstack, minio → devstackadmin). func engineDefaultAdmin(engine string) string { - if engine == "minio" { + switch engine { + case "minio": return "devstackadmin" + case "localstack": + // LocalStack accepts any credentials; the community convention is "test". + return "test" + default: + return "devstack" } - return "devstack" } // engineTarget resolves the host-reachable admin endpoint for an instance: it diff --git a/internal/orchestrate/resources.go b/internal/orchestrate/resources.go index 270f13b..44d0217 100644 --- a/internal/orchestrate/resources.go +++ b/internal/orchestrate/resources.go @@ -41,6 +41,16 @@ var engineOverlays = map[string]perEngineOverlay{ "postgres": {provisionPurpose, provisionPortBase, 5432}, "redis": {"redis-provision", 46379, 6379}, "minio": {"minio-provision", 49000, 9000}, + "nats": {"nats-provision", 44222, 4222}, + // Kafka (Redpanda) advertises its EXTERNAL listener at a fixed 127.0.0.1:49092 + // (template), so host clients must reach the broker on exactly that port — the + // overlay publishes the in-container external listener (19092) there. The port + // base is 49092 to match the advertised address (a mismatch breaks the Kafka + // bootstrap→redirect handshake, the #1 local-Kafka footgun). + "kafka": {"kafka-provision", 49092, 19092}, + // LocalStack's edge port (4566) serves every AWS service (SQS/SNS/S3/…); keyed by + // the template name "localstack" (its `provides: aws` is reached on this port). + "localstack": {"localstack-provision", 44566, 4566}, } // declaredKind reports whether a ledger kind is one the declarative resources @@ -127,6 +137,10 @@ func buildRegistry(d UpDeps) *resource.Registry { return resource.NewRegistry( resource.Postgres{Connect: toResourceConnector(d.PgConnect)}, resource.MinIO{Factory: d.S3Factory}, + resource.NATS{Factory: d.NatsFactory}, + resource.Kafka{Factory: d.KafkaFactory}, + resource.LocalStack{SQSFactory: d.SQSFactory, SNSFactory: d.SNSFactory}, + resource.Redis{}, ) } diff --git a/internal/orchestrate/up.go b/internal/orchestrate/up.go index 866d99f..23d953b 100644 --- a/internal/orchestrate/up.go +++ b/internal/orchestrate/up.go @@ -64,6 +64,14 @@ type UpDeps struct { // nil → the real pure-Go aws-sdk-go-v2 path-style client. Injected for tests so // bucket ops run without a live endpoint. S3Factory resource.S3Factory + // NatsFactory / KafkaFactory / SQSFactory / SNSFactory build the admin clients + // for the messaging provisioners (spec 29); nil → the real pure-Go clients + // (nats.go / franz-go / aws-sdk-go-v2). Injected for tests so queue/topic/stream + // ops run without a live broker/endpoint. + NatsFactory resource.NatsFactory + KafkaFactory resource.KafkaFactory + SQSFactory resource.SQSFactory + SNSFactory resource.SNSFactory Build bool // compose up --build (honors the generate ledger's selective-rebuild hash) Rebuild bool // force `compose build --no-cache` before up (spec 26 --rebuild) diff --git a/internal/resource/kafka.go b/internal/resource/kafka.go new file mode 100644 index 0000000..3404838 --- /dev/null +++ b/internal/resource/kafka.go @@ -0,0 +1,225 @@ +package resource + +import ( + "context" + "fmt" + "sort" + "strings" + + "github.com/twmb/franz-go/pkg/kadm" + "github.com/twmb/franz-go/pkg/kgo" +) + +// This file is the Kafka (Redpanda) Provisioner (spec 29 §messaging — the NATIVE +// default for topics + Kafka streams). It creates/inspects/deletes TOPICS +// (kind=topic and kind=stream both map to a partitioned Kafka topic) using the +// PURE-GO franz-go admin client (kadm) in-process (CGO-free, so it stays inside the +// single static binary — never an external `rpk`/`kafka-topics` tool). The client +// sits behind the small KafkaAdmin seam so unit/race tests run without a live +// broker: inject Kafka.Factory with a fake. Topic names are transparently +// PROJECT-PREFIXED for tenant isolation; callers escape with --no-prefix. Idempotent +// (an already-existing topic is a no-op, not an error). + +// KafkaAdmin is the subset of Kafka admin ops the provisioner uses, behind a +// mockable seam. The default impl wraps franz-go/kadm; tests inject a fake. +type KafkaAdmin interface { + CreateTopic(ctx context.Context, name string, partitions int, replicas int) error + DeleteTopic(ctx context.Context, name string) error + ListTopics(ctx context.Context) ([]string, error) + Close() error +} + +// KafkaFactory builds a KafkaAdmin for a resolved Target (the external +// 127.0.0.1 listener the kafka template advertises). Injectable so the provisioner +// is broker-free in tests; nil selects the real franz-go client. +type KafkaFactory func(ctx context.Context, t Target) (KafkaAdmin, error) + +// Kafka is the kafka/Redpanda Provisioner. Factory nil → the real client. +type Kafka struct { + Factory KafkaFactory +} + +var _ Provisioner = Kafka{} + +// Engine reports the shared-template capability this provisioner serves (the kafka +// template's `provides: kafka`). +func (Kafka) Engine() string { return "kafka" } + +// Kinds are the resource kinds this provisioner can create. +func (Kafka) Kinds() []string { return []string{"topic", "stream", "acl"} } + +func (k Kafka) admin(ctx context.Context, t Target) (KafkaAdmin, error) { + if k.Factory != nil { + return k.Factory(ctx, t) + } + return defaultKafkaAdmin(ctx, t) +} + +// kafkaAddr is the host-reachable broker address (the advertised external listener). +func kafkaAddr(t Target) string { return fmt.Sprintf("%s:%d", t.Host, t.Port) } + +// Ensure idempotently creates a Kafka topic (kind=topic or kind=stream). partitions +// (default 1) and replicas (default 1) come from Params. An already-existing topic +// is a no-op. Returns the connection facts. +func (k Kafka) Ensure(ctx context.Context, t Target, r Resource) (Attrs, error) { + name := r.Name + if name == "" { + name = r.Owner + } + a, err := k.admin(ctx, t) + if err != nil { + return nil, err + } + defer func() { _ = a.Close() }() + + partitions := intParam(r.Params, "partitions") + if partitions <= 0 { + partitions = 1 + } + replicas := intParam(r.Params, "replicas") + if replicas <= 0 { + replicas = 1 + } + if err := a.CreateTopic(ctx, name, partitions, replicas); err != nil { + if !isTopicExists(err) { + return nil, fmt.Errorf("create kafka topic %q: %w", name, err) + } + } + return k.attrs(t, name, partitions), nil +} + +// Drop deletes the topic. Idempotent: a missing topic is not an error. Never +// touches the shared broker container. +func (k Kafka) Drop(ctx context.Context, t Target, r Resource) error { + name := r.Name + if name == "" { + name = r.Owner + } + a, err := k.admin(ctx, t) + if err != nil { + return err + } + defer func() { _ = a.Close() }() + if err := a.DeleteTopic(ctx, name); err != nil { + if isTopicNotFound(err) { + return nil + } + return fmt.Errorf("delete kafka topic %q: %w", name, err) + } + return nil +} + +// Preflight verifies the broker is reachable (list topics). Absence degrades only +// the topic/stream verbs, never `up`. +func (k Kafka) Preflight(ctx context.Context, t Target) error { + a, err := k.admin(ctx, t) + if err != nil { + return err + } + defer func() { _ = a.Close() }() + _, err = a.ListTopics(ctx) + return err +} + +// ListTopics returns the tenant's topic names (prefix-filtered). A lock-free read. +func (k Kafka) ListTopics(ctx context.Context, t Target, prefix string) ([]string, error) { + a, err := k.admin(ctx, t) + if err != nil { + return nil, err + } + defer func() { _ = a.Close() }() + all, err := a.ListTopics(ctx) + if err != nil { + return nil, err + } + var out []string + for _, name := range all { + if strings.HasPrefix(name, "_") || strings.HasPrefix(name, "__") { + continue // skip internal topics + } + if prefix != "" && !strings.HasPrefix(name, prefix) { + continue + } + out = append(out, name) + } + sort.Strings(out) + return out, nil +} + +func (Kafka) attrs(t Target, name string, partitions int) Attrs { + return Attrs{ + "endpoint": kafkaAddr(t), + "host": sharedHost(t.Instance), + "port": "9092", + "topic": name, + "partitions": fmt.Sprintf("%d", partitions), + "broker": kafkaAddr(t), + } +} + +// isTopicExists reports the idempotent "topic already exists" case. +func isTopicExists(err error) bool { + if err == nil { + return false + } + return strings.Contains(strings.ToLower(err.Error()), "already exists") +} + +// isTopicNotFound reports the "unknown topic" family so Drop stays idempotent. +func isTopicNotFound(err error) bool { + if err == nil { + return false + } + e := strings.ToLower(err.Error()) + return strings.Contains(e, "unknown topic") || strings.Contains(e, "does not exist") +} + +// --- default franz-go-backed KafkaAdmin --------------------------------------- + +type kafkaAdmin struct { + cl *kgo.Client + adm *kadm.Client +} + +func defaultKafkaAdmin(_ context.Context, t Target) (KafkaAdmin, error) { + cl, err := kgo.NewClient(kgo.SeedBrokers(kafkaAddr(t))) + if err != nil { + return nil, fmt.Errorf("connect kafka %s: %w", kafkaAddr(t), err) + } + return &kafkaAdmin{cl: cl, adm: kadm.NewClient(cl)}, nil +} + +func (a *kafkaAdmin) CreateTopic(ctx context.Context, name string, partitions int, replicas int) error { + _, err := a.adm.CreateTopic(ctx, int32(partitions), int16(replicas), nil, name) + return err +} + +func (a *kafkaAdmin) DeleteTopic(ctx context.Context, name string) error { + resp, err := a.adm.DeleteTopics(ctx, name) + if err != nil { + return err + } + if r, ok := resp[name]; ok && r.Err != nil { + return r.Err + } + return nil +} + +func (a *kafkaAdmin) ListTopics(ctx context.Context) ([]string, error) { + td, err := a.adm.ListTopics(ctx) + if err != nil { + return nil, err + } + var out []string + for name := range td { + out = append(out, name) + } + return out, nil +} + +func (a *kafkaAdmin) Close() error { + if a.cl != nil { + a.cl.Close() + } + return nil +} diff --git a/internal/resource/kafka_test.go b/internal/resource/kafka_test.go new file mode 100644 index 0000000..c32f0fb --- /dev/null +++ b/internal/resource/kafka_test.go @@ -0,0 +1,126 @@ +package resource + +import ( + "context" + "errors" + "slices" + "testing" +) + +// fakeKafka is an in-memory KafkaAdmin recording topics + their partition counts. +type fakeKafka struct { + topics map[string]int // name → partitions + createErr error + closed bool + createCall int +} + +func newFakeKafka() *fakeKafka { return &fakeKafka{topics: map[string]int{}} } + +func (f *fakeKafka) CreateTopic(_ context.Context, name string, partitions, _ int) error { + f.createCall++ + if f.createErr != nil { + return f.createErr + } + if _, ok := f.topics[name]; ok { + return errors.New("TOPIC_ALREADY_EXISTS: topic already exists") + } + f.topics[name] = partitions + return nil +} +func (f *fakeKafka) DeleteTopic(_ context.Context, name string) error { + if _, ok := f.topics[name]; !ok { + return errors.New("UNKNOWN_TOPIC_OR_PARTITION: unknown topic") + } + delete(f.topics, name) + return nil +} +func (f *fakeKafka) ListTopics(_ context.Context) ([]string, error) { + out := []string{"__consumer_offsets"} // internal topic must be filtered out + for n := range f.topics { + out = append(out, n) + } + return out, nil +} +func (f *fakeKafka) Close() error { f.closed = true; return nil } + +func kafkaTarget() Target { return Target{Instance: "kafka", Host: "127.0.0.1", Port: 49092} } + +func kafkaWith(f *fakeKafka) Kafka { + return Kafka{Factory: func(context.Context, Target) (KafkaAdmin, error) { return f, nil }} +} + +func TestKafkaEngineAndKinds(t *testing.T) { + k := Kafka{} + if k.Engine() != "kafka" { + t.Errorf("Engine() = %q, want kafka", k.Engine()) + } + if !slices.Contains(k.Kinds(), "topic") || !slices.Contains(k.Kinds(), "stream") { + t.Errorf("Kinds() = %v", k.Kinds()) + } +} + +func TestKafkaEnsureTopicPartitions(t *testing.T) { + tests := []struct { + name string + res Resource + wantPart int + }{ + {"default single partition", Resource{Engine: "kafka", Kind: "topic", Name: "web-events", Owner: "web"}, 1}, + {"explicit partitions via stream", Resource{Engine: "kafka", Kind: "stream", Name: "web-orders", Owner: "web", Params: map[string]any{"partitions": 6}}, 6}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + f := newFakeKafka() + attrs, err := kafkaWith(f).Ensure(context.Background(), kafkaTarget(), tc.res) + if err != nil { + t.Fatalf("Ensure: %v", err) + } + if f.topics[tc.res.Name] != tc.wantPart { + t.Errorf("partitions = %d, want %d", f.topics[tc.res.Name], tc.wantPart) + } + if attrs["topic"] != tc.res.Name { + t.Errorf("attrs[topic] = %q", attrs["topic"]) + } + }) + } +} + +func TestKafkaEnsureIdempotent(t *testing.T) { + f := newFakeKafka() + k := kafkaWith(f) + ctx := context.Background() + r := Resource{Engine: "kafka", Kind: "topic", Name: "web-events", Owner: "web"} + for i := 0; i < 3; i++ { + if _, err := k.Ensure(ctx, kafkaTarget(), r); err != nil { + t.Fatalf("Ensure #%d: an existing topic must be a no-op, got %v", i, err) + } + } + if len(f.topics) != 1 { + t.Errorf("want a single topic after 3 ensures, got %v", f.topics) + } +} + +func TestKafkaDropIdempotentAndListFilters(t *testing.T) { + f := newFakeKafka() + k := kafkaWith(f) + ctx := context.Background() + for _, name := range []string{"web-a", "api-b"} { + _, _ = k.Ensure(ctx, kafkaTarget(), Resource{Engine: "kafka", Kind: "topic", Name: name, Owner: "x"}) + } + // List filters internal topics + applies the tenant prefix. + got, err := k.ListTopics(ctx, kafkaTarget(), "web-") + if err != nil { + t.Fatalf("ListTopics: %v", err) + } + if !slices.Equal(got, []string{"web-a"}) { + t.Errorf("ListTopics(web-) = %v, want [web-a]", got) + } + // Drop is idempotent. + if err := k.Drop(ctx, kafkaTarget(), Resource{Engine: "kafka", Kind: "topic", Name: "web-a", Owner: "web"}); err != nil { + t.Fatalf("Drop: %v", err) + } + if err := k.Drop(ctx, kafkaTarget(), Resource{Engine: "kafka", Kind: "topic", Name: "web-a", Owner: "web"}); err != nil { + t.Errorf("Drop of missing topic must be idempotent: %v", err) + } +} diff --git a/internal/resource/localstack.go b/internal/resource/localstack.go new file mode 100644 index 0000000..1598351 --- /dev/null +++ b/internal/resource/localstack.go @@ -0,0 +1,376 @@ +package resource + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/sns" + "github.com/aws/aws-sdk-go-v2/service/sqs" + sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" +) + +// This file is the LocalStack (SQS/SNS) Provisioner (spec 29 §messaging — the +// OPT-IN cloud backend selected via --engine sqs|sns). It creates/inspects/deletes +// SQS QUEUES (kind=queue) and SNS TOPICS (kind=topic) using the PURE-GO +// aws-sdk-go-v2 SQS + SNS clients in-process (CGO-free, so they stay inside the +// single static binary — like the MinIO S3 provisioner), pointed at the LocalStack +// endpoint (the localstack template's `provides: aws`). The SDK sits behind the +// small SQSAPI/SNSAPI seams so unit/race tests run without a live LocalStack: +// inject the factories with fakes. Queue/topic names are transparently +// PROJECT-PREFIXED for tenant isolation; callers escape with --no-prefix. Idempotent. + +// SQSAPI is the subset of the aws-sdk-go-v2 SQS client the provisioner uses. +type SQSAPI interface { + CreateQueue(context.Context, *sqs.CreateQueueInput, ...func(*sqs.Options)) (*sqs.CreateQueueOutput, error) + GetQueueUrl(context.Context, *sqs.GetQueueUrlInput, ...func(*sqs.Options)) (*sqs.GetQueueUrlOutput, error) + GetQueueAttributes(context.Context, *sqs.GetQueueAttributesInput, ...func(*sqs.Options)) (*sqs.GetQueueAttributesOutput, error) + DeleteQueue(context.Context, *sqs.DeleteQueueInput, ...func(*sqs.Options)) (*sqs.DeleteQueueOutput, error) + ListQueues(context.Context, *sqs.ListQueuesInput, ...func(*sqs.Options)) (*sqs.ListQueuesOutput, error) +} + +// SNSAPI is the subset of the aws-sdk-go-v2 SNS client the provisioner uses. +type SNSAPI interface { + CreateTopic(context.Context, *sns.CreateTopicInput, ...func(*sns.Options)) (*sns.CreateTopicOutput, error) + DeleteTopic(context.Context, *sns.DeleteTopicInput, ...func(*sns.Options)) (*sns.DeleteTopicOutput, error) + ListTopics(context.Context, *sns.ListTopicsInput, ...func(*sns.Options)) (*sns.ListTopicsOutput, error) + Subscribe(context.Context, *sns.SubscribeInput, ...func(*sns.Options)) (*sns.SubscribeOutput, error) +} + +// SQSFactory / SNSFactory build the admin clients for a resolved Target (the +// 127.0.0.1 overlay endpoint + dev creds). Injectable so the provisioner is +// endpoint-free in tests; nil selects the real aws-sdk-go-v2 client. +type ( + SQSFactory func(ctx context.Context, t Target) (SQSAPI, error) + SNSFactory func(ctx context.Context, t Target) (SNSAPI, error) +) + +// LocalStack is the LocalStack (SQS/SNS) Provisioner. It is keyed by the shared +// TEMPLATE name "localstack" (what ResolveInstance matches on) while the SQS/SNS +// clients target the template's `provides: aws` endpoint on :4566. Factories nil → +// the real clients. +type LocalStack struct { + SQSFactory SQSFactory + SNSFactory SNSFactory +} + +var _ Provisioner = LocalStack{} + +// Engine reports the shared-template name this provisioner serves ("localstack"; +// the template's `provides: aws` is reached over the same instance's endpoint). +func (LocalStack) Engine() string { return "localstack" } + +// Kinds are the resource kinds this provisioner can create. +func (LocalStack) Kinds() []string { return []string{"queue", "topic"} } + +func awsEndpoint(t Target) string { return fmt.Sprintf("http://%s:%d", t.Host, t.Port) } + +func (l LocalStack) sqsClient(ctx context.Context, t Target) (SQSAPI, error) { + if l.SQSFactory != nil { + return l.SQSFactory(ctx, t) + } + cfg := awsDevConfig(t) + return sqs.NewFromConfig(cfg, func(o *sqs.Options) { o.BaseEndpoint = aws.String(awsEndpoint(t)) }), nil +} + +func (l LocalStack) snsClient(ctx context.Context, t Target) (SNSAPI, error) { + if l.SNSFactory != nil { + return l.SNSFactory(ctx, t) + } + cfg := awsDevConfig(t) + return sns.NewFromConfig(cfg, func(o *sns.Options) { o.BaseEndpoint = aws.String(awsEndpoint(t)) }), nil +} + +// awsDevConfig builds the LocalStack dev config (the "test"/"test" cred convention, +// or the instance's params). Mirrors the MinIO defaultS3Client pattern. +func awsDevConfig(t Target) aws.Config { + access := t.AdminEnv["user"] + if access == "" { + access = "test" + } + secret := t.AdminEnv["password"] + if secret == "" { + secret = "test" + } + return aws.Config{ + Region: "us-east-1", + Credentials: credentials.NewStaticCredentialsProvider(access, secret, ""), + } +} + +// Ensure idempotently provisions an SQS queue (kind=queue) or SNS topic +// (kind=topic). For queues: --fifo appends .fifo + sets FifoQueue, and a +// Params["dlq"] (already-provisioned DLQ physical name) wires a redrive policy with +// Params["max_receive"]. For topics: an optional Params["subscribe"] (an SQS queue +// physical name) wires the classic SNS→SQS fan-out. Returns the connection facts. +func (l LocalStack) Ensure(ctx context.Context, t Target, r Resource) (Attrs, error) { + switch r.Kind { + case "queue": + return l.ensureQueue(ctx, t, r) + case "topic": + return l.ensureTopic(ctx, t, r) + default: + return nil, fmt.Errorf("aws engine does not support kind %q", r.Kind) + } +} + +func (l LocalStack) ensureQueue(ctx context.Context, t Target, r Resource) (Attrs, error) { + c, err := l.sqsClient(ctx, t) + if err != nil { + return nil, err + } + name := r.Name + if name == "" { + name = r.Owner + } + attrsMap := map[string]string{} + if boolParam(r.Params, "fifo") { + if !strings.HasSuffix(name, ".fifo") { + name += ".fifo" + } + attrsMap["FifoQueue"] = "true" + } + // A DLQ is created first (by the caller, as its own recorded resource); here we + // resolve its ARN and set the redrive policy on the main queue. + if dlq := paramStr(r.Params, "dlq"); dlq != "" { + arn, err := l.queueArn(ctx, c, dlq) + if err != nil { + return nil, fmt.Errorf("resolve DLQ %q arn: %w", dlq, err) + } + maxRecv := intParam(r.Params, "max_receive") + if maxRecv <= 0 { + maxRecv = 5 + } + attrsMap["RedrivePolicy"] = fmt.Sprintf(`{"deadLetterTargetArn":%q,"maxReceiveCount":"%d"}`, arn, maxRecv) + } + out, err := c.CreateQueue(ctx, &sqs.CreateQueueInput{ + QueueName: aws.String(name), + Attributes: attrsMap, + }) + if err != nil { + return nil, fmt.Errorf("create sqs queue %q: %w", name, err) + } + return Attrs{ + "endpoint": awsEndpoint(t), + "host": sharedHost(t.Instance), + "port": "4566", + "queue": name, + "queueUrl": aws.ToString(out.QueueUrl), + }, nil +} + +func (l LocalStack) ensureTopic(ctx context.Context, t Target, r Resource) (Attrs, error) { + c, err := l.snsClient(ctx, t) + if err != nil { + return nil, err + } + name := r.Name + if name == "" { + name = r.Owner + } + out, err := c.CreateTopic(ctx, &sns.CreateTopicInput{Name: aws.String(name)}) + if err != nil { + return nil, fmt.Errorf("create sns topic %q: %w", name, err) + } + topicArn := aws.ToString(out.TopicArn) + attrs := Attrs{ + "endpoint": awsEndpoint(t), + "host": sharedHost(t.Instance), + "port": "4566", + "topic": name, + "topicArn": topicArn, + } + if sub := paramStr(r.Params, "subscribe"); sub != "" { + sqsC, err := l.sqsClient(ctx, t) + if err != nil { + return nil, err + } + arn, err := l.queueArn(ctx, sqsC, sub) + if err != nil { + return nil, fmt.Errorf("resolve subscribe queue %q arn: %w", sub, err) + } + if _, err := c.Subscribe(ctx, &sns.SubscribeInput{ + TopicArn: aws.String(topicArn), + Protocol: aws.String("sqs"), + Endpoint: aws.String(arn), + }); err != nil { + return nil, fmt.Errorf("subscribe %q to topic %q: %w", sub, name, err) + } + attrs["subscribed"] = sub + } + return attrs, nil +} + +// queueArn resolves the ARN of an existing SQS queue by physical name. +func (LocalStack) queueArn(ctx context.Context, c SQSAPI, name string) (string, error) { + u, err := c.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{QueueName: aws.String(name)}) + if err != nil { + return "", err + } + attr, err := c.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: u.QueueUrl, + AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameQueueArn}, + }) + if err != nil { + return "", err + } + return attr.Attributes[string(sqstypes.QueueAttributeNameQueueArn)], nil +} + +// Drop removes the SQS queue or SNS topic. Idempotent: a missing object is not an +// error. Never touches the shared LocalStack container. +func (l LocalStack) Drop(ctx context.Context, t Target, r Resource) error { + name := r.Name + if name == "" { + name = r.Owner + } + switch r.Kind { + case "queue": + c, err := l.sqsClient(ctx, t) + if err != nil { + return err + } + u, err := c.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{QueueName: aws.String(name)}) + if err != nil { + if isAwsNotFound(err) { + return nil + } + return fmt.Errorf("resolve sqs queue %q: %w", name, err) + } + if _, err := c.DeleteQueue(ctx, &sqs.DeleteQueueInput{QueueUrl: u.QueueUrl}); err != nil { + if isAwsNotFound(err) { + return nil + } + return fmt.Errorf("delete sqs queue %q: %w", name, err) + } + return nil + case "topic": + c, err := l.snsClient(ctx, t) + if err != nil { + return err + } + arn, err := l.topicArn(ctx, c, name) + if err != nil { + return fmt.Errorf("resolve sns topic %q: %w", name, err) + } + if arn == "" { + return nil // already gone + } + if _, err := c.DeleteTopic(ctx, &sns.DeleteTopicInput{TopicArn: aws.String(arn)}); err != nil { + if isAwsNotFound(err) { + return nil + } + return fmt.Errorf("delete sns topic %q: %w", name, err) + } + return nil + } + return fmt.Errorf("aws engine cannot drop kind %q", r.Kind) +} + +// Preflight verifies the endpoint is reachable (a ListQueues round-trip). +func (l LocalStack) Preflight(ctx context.Context, t Target) error { + c, err := l.sqsClient(ctx, t) + if err != nil { + return err + } + _, err = c.ListQueues(ctx, &sqs.ListQueuesInput{}) + return err +} + +// ListQueues returns the tenant's SQS queue names (prefix-filtered). Lock-free read. +func (l LocalStack) ListQueues(ctx context.Context, t Target, prefix string) ([]string, error) { + c, err := l.sqsClient(ctx, t) + if err != nil { + return nil, err + } + in := &sqs.ListQueuesInput{} + if prefix != "" { + in.QueueNamePrefix = aws.String(prefix) + } + out, err := c.ListQueues(ctx, in) + if err != nil { + return nil, err + } + var names []string + for _, u := range out.QueueUrls { + names = append(names, queueNameFromURL(u)) + } + sort.Strings(names) + return names, nil +} + +// ListTopics returns the tenant's SNS topic names (prefix-filtered). Lock-free read. +func (l LocalStack) ListTopics(ctx context.Context, t Target, prefix string) ([]string, error) { + c, err := l.snsClient(ctx, t) + if err != nil { + return nil, err + } + out, err := c.ListTopics(ctx, &sns.ListTopicsInput{}) + if err != nil { + return nil, err + } + var names []string + for _, tp := range out.Topics { + name := topicNameFromARN(aws.ToString(tp.TopicArn)) + if prefix != "" && !strings.HasPrefix(name, prefix) { + continue + } + names = append(names, name) + } + sort.Strings(names) + return names, nil +} + +// topicArn resolves an SNS topic ARN by physical name via ListTopics (LocalStack +// exposes no GetTopicByName). +func (LocalStack) topicArn(ctx context.Context, c SNSAPI, name string) (string, error) { + out, err := c.ListTopics(ctx, &sns.ListTopicsInput{}) + if err != nil { + return "", err + } + for _, tp := range out.Topics { + arn := aws.ToString(tp.TopicArn) + if topicNameFromARN(arn) == name { + return arn, nil + } + } + return "", nil +} + +// queueNameFromURL extracts the queue name (the last path segment) from a queue URL. +func queueNameFromURL(u string) string { + if i := strings.LastIndexByte(u, '/'); i >= 0 { + return u[i+1:] + } + return u +} + +// topicNameFromARN extracts the topic name (the last colon segment) from a topic ARN. +func topicNameFromARN(arn string) string { + if i := strings.LastIndexByte(arn, ':'); i >= 0 { + return arn[i+1:] + } + return arn +} + +// isAwsNotFound reports the SQS/SNS "does not exist" family so Drop stays idempotent. +func isAwsNotFound(err error) bool { + if err == nil { + return false + } + var qne *sqstypes.QueueDoesNotExist + if errors.As(err, &qne) { + return true + } + code := apiCode(err) + switch code { + case "AWS.SimpleQueueService.NonExistentQueue", "NotFound", "ResourceNotFoundException", "QueueDoesNotExist": + return true + } + e := strings.ToLower(err.Error()) + return strings.Contains(e, "nonexistent") || strings.Contains(e, "does not exist") || strings.Contains(e, "not found") +} diff --git a/internal/resource/localstack_test.go b/internal/resource/localstack_test.go new file mode 100644 index 0000000..30a71b2 --- /dev/null +++ b/internal/resource/localstack_test.go @@ -0,0 +1,186 @@ +package resource + +import ( + "context" + "slices" + "strings" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sns" + snstypes "github.com/aws/aws-sdk-go-v2/service/sns/types" + "github.com/aws/aws-sdk-go-v2/service/sqs" + sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" +) + +// fakeSQS embeds SQSAPI (nil) so it satisfies the interface; only the used methods +// are overridden. It records queues + their attributes. +type fakeSQS struct { + SQSAPI + queues map[string]map[string]string // name → attributes (incl. RedrivePolicy) + created []string +} + +func newFakeSQS() *fakeSQS { return &fakeSQS{queues: map[string]map[string]string{}} } + +func (f *fakeSQS) CreateQueue(_ context.Context, in *sqs.CreateQueueInput, _ ...func(*sqs.Options)) (*sqs.CreateQueueOutput, error) { + name := aws.ToString(in.QueueName) + f.created = append(f.created, name) + f.queues[name] = in.Attributes + return &sqs.CreateQueueOutput{QueueUrl: aws.String("http://localhost:4566/000000000000/" + name)}, nil +} +func (f *fakeSQS) GetQueueUrl(_ context.Context, in *sqs.GetQueueUrlInput, _ ...func(*sqs.Options)) (*sqs.GetQueueUrlOutput, error) { + name := aws.ToString(in.QueueName) + if _, ok := f.queues[name]; !ok { + return nil, &sqstypes.QueueDoesNotExist{} + } + return &sqs.GetQueueUrlOutput{QueueUrl: aws.String("http://localhost:4566/000000000000/" + name)}, nil +} +func (f *fakeSQS) GetQueueAttributes(_ context.Context, in *sqs.GetQueueAttributesInput, _ ...func(*sqs.Options)) (*sqs.GetQueueAttributesOutput, error) { + name := queueNameFromURL(aws.ToString(in.QueueUrl)) + return &sqs.GetQueueAttributesOutput{Attributes: map[string]string{ + string(sqstypes.QueueAttributeNameQueueArn): "arn:aws:sqs:us-east-1:000000000000:" + name, + }}, nil +} +func (f *fakeSQS) DeleteQueue(_ context.Context, in *sqs.DeleteQueueInput, _ ...func(*sqs.Options)) (*sqs.DeleteQueueOutput, error) { + delete(f.queues, queueNameFromURL(aws.ToString(in.QueueUrl))) + return &sqs.DeleteQueueOutput{}, nil +} +func (f *fakeSQS) ListQueues(_ context.Context, in *sqs.ListQueuesInput, _ ...func(*sqs.Options)) (*sqs.ListQueuesOutput, error) { + var urls []string + for n := range f.queues { + if in.QueueNamePrefix != nil && !strings.HasPrefix(n, aws.ToString(in.QueueNamePrefix)) { + continue + } + urls = append(urls, "http://localhost:4566/000000000000/"+n) + } + return &sqs.ListQueuesOutput{QueueUrls: urls}, nil +} + +// fakeSNS embeds SNSAPI (nil); records topics + subscriptions. +type fakeSNS struct { + SNSAPI + topics []string + subs []string +} + +func (f *fakeSNS) CreateTopic(_ context.Context, in *sns.CreateTopicInput, _ ...func(*sns.Options)) (*sns.CreateTopicOutput, error) { + name := aws.ToString(in.Name) + if !slices.Contains(f.topics, name) { + f.topics = append(f.topics, name) + } + return &sns.CreateTopicOutput{TopicArn: aws.String("arn:aws:sns:us-east-1:000000000000:" + name)}, nil +} +func (f *fakeSNS) ListTopics(_ context.Context, _ *sns.ListTopicsInput, _ ...func(*sns.Options)) (*sns.ListTopicsOutput, error) { + var out []snstypes.Topic + for _, n := range f.topics { + out = append(out, snstypes.Topic{TopicArn: aws.String("arn:aws:sns:us-east-1:000000000000:" + n)}) + } + return &sns.ListTopicsOutput{Topics: out}, nil +} +func (f *fakeSNS) DeleteTopic(_ context.Context, in *sns.DeleteTopicInput, _ ...func(*sns.Options)) (*sns.DeleteTopicOutput, error) { + name := topicNameFromARN(aws.ToString(in.TopicArn)) + f.topics = slices.DeleteFunc(f.topics, func(s string) bool { return s == name }) + return &sns.DeleteTopicOutput{}, nil +} +func (f *fakeSNS) Subscribe(_ context.Context, in *sns.SubscribeInput, _ ...func(*sns.Options)) (*sns.SubscribeOutput, error) { + f.subs = append(f.subs, aws.ToString(in.Endpoint)) + return &sns.SubscribeOutput{SubscriptionArn: aws.String("sub-arn")}, nil +} + +func awsTarget() Target { + return Target{Instance: "localstack", Host: "127.0.0.1", Port: 44566, AdminEnv: map[string]string{"user": "test", "password": "test"}} +} + +func localstackWith(sq *fakeSQS, sn *fakeSNS) LocalStack { + return LocalStack{ + SQSFactory: func(context.Context, Target) (SQSAPI, error) { return sq, nil }, + SNSFactory: func(context.Context, Target) (SNSAPI, error) { return sn, nil }, + } +} + +func TestLocalStackEngineAndKinds(t *testing.T) { + l := LocalStack{} + if l.Engine() != "localstack" { + t.Errorf("Engine() = %q, want localstack (the template name; provides: aws)", l.Engine()) + } + if !slices.Contains(l.Kinds(), "queue") || !slices.Contains(l.Kinds(), "topic") { + t.Errorf("Kinds() = %v", l.Kinds()) + } +} + +func TestLocalStackSQSCreateFifoAndDLQRedrive(t *testing.T) { + sq := newFakeSQS() + l := localstackWith(sq, &fakeSNS{}) + ctx := context.Background() + // DLQ first (as the CLI does), then the FIFO main queue with a redrive policy. + if _, err := l.Ensure(ctx, awsTarget(), Resource{Engine: "aws", Kind: "queue", Name: "web-jobs-dead", Owner: "web"}); err != nil { + t.Fatalf("create DLQ: %v", err) + } + attrs, err := l.Ensure(ctx, awsTarget(), Resource{Engine: "aws", Kind: "queue", Name: "web-jobs", Owner: "web", + Params: map[string]any{"fifo": true, "dlq": "web-jobs-dead", "max_receive": 5}}) + if err != nil { + t.Fatalf("create main queue: %v", err) + } + // FIFO appends .fifo and sets FifoQueue=true. + if !slices.Contains(sq.created, "web-jobs.fifo") { + t.Errorf("FIFO queue not created with .fifo suffix: %v", sq.created) + } + main := sq.queues["web-jobs.fifo"] + if main["FifoQueue"] != "true" { + t.Errorf("FifoQueue attr = %q, want true", main["FifoQueue"]) + } + if rp := main["RedrivePolicy"]; !strings.Contains(rp, "web-jobs-dead") || !strings.Contains(rp, `"maxReceiveCount":"5"`) { + t.Errorf("redrive policy wrong: %q", rp) + } + if attrs["queue"] != "web-jobs.fifo" { + t.Errorf("attrs[queue] = %q", attrs["queue"]) + } +} + +func TestLocalStackSNSCreateAndSubscribe(t *testing.T) { + sq := newFakeSQS() + sn := &fakeSNS{} + l := localstackWith(sq, sn) + ctx := context.Background() + // The SQS queue to subscribe must exist first. + _, _ = l.Ensure(ctx, awsTarget(), Resource{Engine: "aws", Kind: "queue", Name: "web-jobs", Owner: "web"}) + attrs, err := l.Ensure(ctx, awsTarget(), Resource{Engine: "aws", Kind: "topic", Name: "web-events", Owner: "web", + Params: map[string]any{"subscribe": "web-jobs"}}) + if err != nil { + t.Fatalf("create topic: %v", err) + } + if !slices.Contains(sn.topics, "web-events") { + t.Errorf("topic not created: %v", sn.topics) + } + if len(sn.subs) != 1 || !strings.Contains(sn.subs[0], "web-jobs") { + t.Errorf("SNS→SQS subscription not wired: %v", sn.subs) + } + if attrs["subscribed"] != "web-jobs" { + t.Errorf("attrs[subscribed] = %q", attrs["subscribed"]) + } +} + +func TestLocalStackDropAndListPrefix(t *testing.T) { + sq := newFakeSQS() + sn := &fakeSNS{} + l := localstackWith(sq, sn) + ctx := context.Background() + for _, n := range []string{"web-a", "api-b"} { + _, _ = l.Ensure(ctx, awsTarget(), Resource{Engine: "aws", Kind: "queue", Name: n, Owner: "x"}) + } + got, err := l.ListQueues(ctx, awsTarget(), "web-") + if err != nil { + t.Fatalf("ListQueues: %v", err) + } + if !slices.Equal(got, []string{"web-a"}) { + t.Errorf("ListQueues(web-) = %v, want [web-a]", got) + } + // Drop is idempotent (missing queue → no error). + if err := l.Drop(ctx, awsTarget(), Resource{Engine: "aws", Kind: "queue", Name: "web-a", Owner: "web"}); err != nil { + t.Fatalf("Drop: %v", err) + } + if err := l.Drop(ctx, awsTarget(), Resource{Engine: "aws", Kind: "queue", Name: "web-a", Owner: "web"}); err != nil { + t.Errorf("Drop of missing queue must be idempotent: %v", err) + } +} diff --git a/internal/resource/nats.go b/internal/resource/nats.go new file mode 100644 index 0000000..f02a6d9 --- /dev/null +++ b/internal/resource/nats.go @@ -0,0 +1,287 @@ +package resource + +import ( + "context" + "fmt" + "sort" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// This file is the NATS (JetStream) Provisioner (spec 29 §messaging — the NATIVE +// default for queues + streams). It creates/inspects/deletes JetStream STREAMS +// (kind=stream) and durable CONSUMERS acting as work-QUEUES (kind=queue) using the +// PURE-GO nats.go client in-process (CGO-free, so it stays inside the single static +// binary — never an external `nats` tool). The client sits behind the small +// NatsAdmin seam so unit/race tests run without a live NATS: inject NATS.Factory +// with a fake. Stream/consumer names are transparently PROJECT-PREFIXED for tenant +// isolation on the shared broker; callers escape with --no-prefix. Idempotent +// (CreateOrUpdateStream/Consumer are upserts). + +// StreamSpec is the engine-agnostic stream intent the seam receives (a NATS stream +// with limits/work-queue retention). It never leaks a nats type so tests are trivial. +type StreamSpec struct { + Name string // the tenant-scoped stream name + Subjects []string // subjects the stream binds (default ".>") + WorkType bool // work-queue retention (a queue) vs limits (a stream) + MaxAge time.Duration // retention window (0 → unlimited) + Replicas int // replica count (0/1 → single) +} + +// ConsumerSpec is the durable-consumer intent (a NATS "queue" binds a durable +// consumer to a work-queue stream). +type ConsumerSpec struct { + Durable string // the durable consumer name +} + +// NatsAdmin is the subset of JetStream admin ops the provisioner uses, behind a +// mockable seam. The default impl wraps nats.go/jetstream; tests inject a fake. +type NatsAdmin interface { + EnsureStream(ctx context.Context, s StreamSpec) error + DeleteStream(ctx context.Context, name string) error + ListStreams(ctx context.Context) ([]string, error) + EnsureConsumer(ctx context.Context, stream string, c ConsumerSpec) error + DeleteConsumer(ctx context.Context, stream, durable string) error + Close() error +} + +// NatsFactory builds a NatsAdmin for a resolved Target (the 127.0.0.1 overlay +// endpoint). Injectable so the provisioner is endpoint-free in tests; nil selects +// the real nats.go client. +type NatsFactory func(ctx context.Context, t Target) (NatsAdmin, error) + +// NATS is the nats JetStream Provisioner. Factory nil → the real client. +type NATS struct { + Factory NatsFactory +} + +var _ Provisioner = NATS{} + +// Engine reports the shared-template capability this provisioner serves (the nats +// template's `provides: nats`). +func (NATS) Engine() string { return "nats" } + +// Kinds are the resource kinds this provisioner can create. +func (NATS) Kinds() []string { return []string{"stream", "queue", "consumer"} } + +// natsURL is the loopback admin endpoint for the overlay-published instance. +func natsURL(t Target) string { return fmt.Sprintf("nats://%s:%d", t.Host, t.Port) } + +func (n NATS) admin(ctx context.Context, t Target) (NatsAdmin, error) { + if n.Factory != nil { + return n.Factory(ctx, t) + } + return defaultNatsAdmin(ctx, t) +} + +// natsSubjects derives the stream subjects from Params["subjects"] (comma list) +// or the default ".>" wildcard. +func natsSubjects(name string, params map[string]any) []string { + if s := paramStr(params, "subjects"); s != "" { + var out []string + for _, p := range strings.Split(s, ",") { + if p = strings.TrimSpace(p); p != "" { + out = append(out, p) + } + } + if len(out) > 0 { + return out + } + } + return []string{name + ".>"} +} + +// Ensure idempotently provisions a JetStream stream (kind=stream) or a work-queue +// stream + durable consumer (kind=queue). Returns the connection facts. +func (n NATS) Ensure(ctx context.Context, t Target, r Resource) (Attrs, error) { + name := r.Name + if name == "" { + name = r.Owner + } + a, err := n.admin(ctx, t) + if err != nil { + return nil, err + } + defer func() { _ = a.Close() }() + + spec := StreamSpec{ + Name: name, + Subjects: natsSubjects(name, r.Params), + WorkType: r.Kind == "queue", + MaxAge: durationParam(r.Params, "retention"), + Replicas: intParam(r.Params, "replicas"), + } + if err := a.EnsureStream(ctx, spec); err != nil { + return nil, fmt.Errorf("ensure nats stream %q: %w", name, err) + } + attrs := n.attrs(t, name, spec.Subjects) + if r.Kind == "queue" { + if err := a.EnsureConsumer(ctx, name, ConsumerSpec{Durable: name}); err != nil { + return nil, fmt.Errorf("ensure nats consumer %q: %w", name, err) + } + attrs["consumer"] = name + } + return attrs, nil +} + +// Drop removes the stream (which also removes its consumers). Idempotent: a missing +// stream is not an error. Never touches the shared broker container. +func (n NATS) Drop(ctx context.Context, t Target, r Resource) error { + name := r.Name + if name == "" { + name = r.Owner + } + a, err := n.admin(ctx, t) + if err != nil { + return err + } + defer func() { _ = a.Close() }() + if err := a.DeleteStream(ctx, name); err != nil { + if isNatsNotFound(err) { + return nil + } + return fmt.Errorf("delete nats stream %q: %w", name, err) + } + return nil +} + +// Preflight verifies the endpoint is reachable (connect + close). Absence degrades +// only the queue/stream verbs, never `up`. +func (n NATS) Preflight(ctx context.Context, t Target) error { + a, err := n.admin(ctx, t) + if err != nil { + return err + } + return a.Close() +} + +// ListStreams returns the tenant's stream names (prefix-filtered). A lock-free read. +func (n NATS) ListStreams(ctx context.Context, t Target, prefix string) ([]string, error) { + a, err := n.admin(ctx, t) + if err != nil { + return nil, err + } + defer func() { _ = a.Close() }() + all, err := a.ListStreams(ctx) + if err != nil { + return nil, err + } + var out []string + for _, s := range all { + if prefix != "" && !strings.HasPrefix(s, prefix) { + continue + } + out = append(out, s) + } + sort.Strings(out) + return out, nil +} + +func (NATS) attrs(t Target, name string, subjects []string) Attrs { + return Attrs{ + "endpoint": natsURL(t), + "host": sharedHost(t.Instance), + "port": "4222", + "stream": name, + "subjects": strings.Join(subjects, ","), + } +} + +// isNatsNotFound reports the "stream not found" family so Drop stays idempotent. +func isNatsNotFound(err error) bool { + if err == nil { + return false + } + if err == jetstream.ErrStreamNotFound || err == jetstream.ErrConsumerNotFound { + return true + } + return strings.Contains(strings.ToLower(err.Error()), "not found") +} + +// --- default nats.go-backed NatsAdmin ----------------------------------------- + +type natsAdmin struct { + nc *nats.Conn + js jetstream.JetStream +} + +func defaultNatsAdmin(_ context.Context, t Target) (NatsAdmin, error) { + nc, err := nats.Connect(natsURL(t), nats.Timeout(5*time.Second)) + if err != nil { + return nil, fmt.Errorf("connect nats %s: %w", natsURL(t), err) + } + js, err := jetstream.New(nc) + if err != nil { + nc.Close() + return nil, fmt.Errorf("open jetstream: %w", err) + } + return &natsAdmin{nc: nc, js: js}, nil +} + +func (a *natsAdmin) EnsureStream(ctx context.Context, s StreamSpec) error { + cfg := jetstream.StreamConfig{ + Name: s.Name, + Subjects: s.Subjects, + Retention: jetstream.LimitsPolicy, + MaxAge: s.MaxAge, + } + if s.WorkType { + cfg.Retention = jetstream.WorkQueuePolicy + } + if s.Replicas > 1 { + cfg.Replicas = s.Replicas + } + _, err := a.js.CreateOrUpdateStream(ctx, cfg) + return err +} + +func (a *natsAdmin) DeleteStream(ctx context.Context, name string) error { + return a.js.DeleteStream(ctx, name) +} + +func (a *natsAdmin) ListStreams(ctx context.Context) ([]string, error) { + lister := a.js.StreamNames(ctx) + var out []string + for name := range lister.Name() { + out = append(out, name) + } + if err := lister.Err(); err != nil { + return nil, err + } + return out, nil +} + +func (a *natsAdmin) EnsureConsumer(ctx context.Context, stream string, c ConsumerSpec) error { + _, err := a.js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{ + Durable: c.Durable, + AckPolicy: jetstream.AckExplicitPolicy, + }) + return err +} + +func (a *natsAdmin) DeleteConsumer(ctx context.Context, stream, durable string) error { + return a.js.DeleteConsumer(ctx, stream, durable) +} + +func (a *natsAdmin) Close() error { + if a.nc != nil { + a.nc.Close() + } + return nil +} + +// durationParam reads a Go duration param (e.g. "168h"), 0 when absent/invalid. +func durationParam(p map[string]any, key string) time.Duration { + s := paramStr(p, key) + if s == "" { + return 0 + } + d, err := time.ParseDuration(s) + if err != nil { + return 0 + } + return d +} diff --git a/internal/resource/nats_test.go b/internal/resource/nats_test.go new file mode 100644 index 0000000..411aa3b --- /dev/null +++ b/internal/resource/nats_test.go @@ -0,0 +1,170 @@ +package resource + +import ( + "context" + "slices" + "testing" + "time" +) + +// fakeNats is an in-memory NatsAdmin: it records streams + consumers so tests +// assert create/list/delete + idempotency without a live NATS. +type fakeNats struct { + streams map[string]StreamSpec + consumers map[string][]string // stream → durable names + closed bool +} + +func newFakeNats() *fakeNats { + return &fakeNats{streams: map[string]StreamSpec{}, consumers: map[string][]string{}} +} + +func (f *fakeNats) EnsureStream(_ context.Context, s StreamSpec) error { + f.streams[s.Name] = s // upsert → idempotent + return nil +} +func (f *fakeNats) DeleteStream(_ context.Context, name string) error { + delete(f.streams, name) + delete(f.consumers, name) + return nil +} +func (f *fakeNats) ListStreams(_ context.Context) ([]string, error) { + var out []string + for n := range f.streams { + out = append(out, n) + } + return out, nil +} +func (f *fakeNats) EnsureConsumer(_ context.Context, stream string, c ConsumerSpec) error { + for _, d := range f.consumers[stream] { + if d == c.Durable { + return nil // idempotent + } + } + f.consumers[stream] = append(f.consumers[stream], c.Durable) + return nil +} +func (f *fakeNats) DeleteConsumer(_ context.Context, stream, durable string) error { return nil } +func (f *fakeNats) Close() error { f.closed = true; return nil } + +func natsTarget() Target { + return Target{Instance: "nats", Host: "127.0.0.1", Port: 44222} +} + +func natsWith(f *fakeNats) NATS { + return NATS{Factory: func(context.Context, Target) (NatsAdmin, error) { return f, nil }} +} + +func TestNatsEngineAndKinds(t *testing.T) { + n := NATS{} + if n.Engine() != "nats" { + t.Errorf("Engine() = %q, want nats", n.Engine()) + } + if !slices.Contains(n.Kinds(), "stream") || !slices.Contains(n.Kinds(), "queue") { + t.Errorf("Kinds() = %v, want stream+queue", n.Kinds()) + } +} + +func TestNatsEnsureStreamAndQueue(t *testing.T) { + tests := []struct { + name string + res Resource + wantWorkType bool + wantConsumer bool + wantMaxAge time.Duration + }{ + { + name: "stream limits retention with max-age", + res: Resource{Engine: "nats", Kind: "stream", Name: "web-orders", Owner: "web", Params: map[string]any{"retention": "168h"}}, + wantMaxAge: 168 * time.Hour, + }, + { + name: "queue is a work-queue stream + durable consumer", + res: Resource{Engine: "nats", Kind: "queue", Name: "web-jobs", Owner: "web"}, + wantWorkType: true, + wantConsumer: true, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + f := newFakeNats() + attrs, err := natsWith(f).Ensure(context.Background(), natsTarget(), tc.res) + if err != nil { + t.Fatalf("Ensure: %v", err) + } + s, ok := f.streams[tc.res.Name] + if !ok { + t.Fatalf("stream %q not created: %v", tc.res.Name, f.streams) + } + if s.WorkType != tc.wantWorkType { + t.Errorf("WorkType = %v, want %v", s.WorkType, tc.wantWorkType) + } + if s.MaxAge != tc.wantMaxAge { + t.Errorf("MaxAge = %v, want %v", s.MaxAge, tc.wantMaxAge) + } + if def := s.Subjects; len(def) == 0 || def[0] != tc.res.Name+".>" { + t.Errorf("subjects = %v, want default %q", def, tc.res.Name+".>") + } + if tc.wantConsumer { + if !slices.Contains(f.consumers[tc.res.Name], tc.res.Name) { + t.Errorf("durable consumer not created: %v", f.consumers) + } + if attrs["consumer"] != tc.res.Name { + t.Errorf("attrs[consumer] = %q, want %q", attrs["consumer"], tc.res.Name) + } + } + if attrs["host"] != "shared-nats" { + t.Errorf("host attr = %q, want shared-nats", attrs["host"]) + } + }) + } +} + +func TestNatsIdempotentAndPrefixIsolation(t *testing.T) { + f := newFakeNats() + n := natsWith(f) + ctx := context.Background() + // Two projects create a same-logical "events" stream; the tenant prefix keeps + // them distinct so project A can never clobber B. + for _, name := range []string{"web-events", "api-events", "web-events"} { // web-events twice → idempotent + if _, err := n.Ensure(ctx, natsTarget(), Resource{Engine: "nats", Kind: "stream", Name: name, Owner: "x"}); err != nil { + t.Fatalf("Ensure %q: %v", name, err) + } + } + if len(f.streams) != 2 { + t.Errorf("want 2 distinct streams (web-events, api-events), got %v", f.streams) + } +} + +func TestNatsDropIdempotent(t *testing.T) { + f := newFakeNats() + n := natsWith(f) + ctx := context.Background() + _, _ = n.Ensure(ctx, natsTarget(), Resource{Engine: "nats", Kind: "stream", Name: "web-orders", Owner: "web"}) + if err := n.Drop(ctx, natsTarget(), Resource{Engine: "nats", Kind: "stream", Name: "web-orders", Owner: "web"}); err != nil { + t.Fatalf("Drop: %v", err) + } + if _, ok := f.streams["web-orders"]; ok { + t.Error("stream not deleted") + } + // Dropping a missing stream is not an error. + if err := n.Drop(ctx, natsTarget(), Resource{Engine: "nats", Kind: "stream", Name: "gone", Owner: "web"}); err != nil { + t.Errorf("Drop of missing stream must be idempotent: %v", err) + } +} + +func TestNatsListStreamsPrefixFiltered(t *testing.T) { + f := newFakeNats() + n := natsWith(f) + ctx := context.Background() + for _, name := range []string{"web-a", "web-b", "api-c"} { + _, _ = n.Ensure(ctx, natsTarget(), Resource{Engine: "nats", Kind: "stream", Name: name, Owner: "x"}) + } + got, err := n.ListStreams(ctx, natsTarget(), "web-") + if err != nil { + t.Fatalf("ListStreams: %v", err) + } + if !slices.Equal(got, []string{"web-a", "web-b"}) { + t.Errorf("ListStreams(web-) = %v, want [web-a web-b]", got) + } +} diff --git a/internal/resource/redis.go b/internal/resource/redis.go new file mode 100644 index 0000000..ad070e6 --- /dev/null +++ b/internal/resource/redis.go @@ -0,0 +1,60 @@ +package resource + +import ( + "context" + "fmt" +) + +// This file is the Redis Provisioner (spec 29 §messaging — a lightweight, +// server-object-free queue/topic backend). A Redis "queue" is a LIST/STREAM key +// namespace the app honours (created on first LPUSH/XADD), and a "topic" is a +// Pub/Sub channel namespace — neither needs a server-side create call, so Ensure +// records nothing on the engine and only surfaces the tenant-scoped key + a +// connection hint. Tenant isolation comes from the transparently PROJECT-PREFIXED +// key. There is no external tool and no in-process client: this stays trivially +// inside the single static binary. + +// Redis is the redis Provisioner (queue/topic key namespaces). It holds no client. +type Redis struct{} + +var _ Provisioner = Redis{} + +// Engine reports the shared-template capability this provisioner serves. +func (Redis) Engine() string { return "redis" } + +// Kinds are the resource kinds this provisioner can create. +func (Redis) Kinds() []string { return []string{"queue", "topic"} } + +// Ensure is a no-op on the engine: a Redis list/stream/channel is created lazily on +// first use. It returns the tenant-scoped key + a connection hint so the caller can +// print how the app reaches it. Idempotent by construction. +func (Redis) Ensure(_ context.Context, t Target, r Resource) (Attrs, error) { + name := r.Name + if name == "" { + name = r.Owner + } + kind := r.Kind + if kind == "" { + kind = "queue" + } + // Logical key convention: is already tenant-prefixed by the CLI. + key := name + hint := fmt.Sprintf("LPUSH %s … / BRPOP %s 0", key, key) + if kind == "topic" { + hint = fmt.Sprintf("PUBLISH %s … / SUBSCRIBE %s", key, key) + } + return Attrs{ + "host": sharedHost(t.Instance), + "port": "6379", + "key": key, + "type": kind, + "hint": hint, + }, nil +} + +// Drop is a no-op on the engine (a Redis list/channel has no durable schema object +// to delete beyond its data, which the app owns). Idempotent. +func (Redis) Drop(_ context.Context, _ Target, _ Resource) error { return nil } + +// Preflight always succeeds: the Redis queue/topic path needs no admin client. +func (Redis) Preflight(_ context.Context, _ Target) error { return nil } diff --git a/internal/resource/redis_test.go b/internal/resource/redis_test.go new file mode 100644 index 0000000..d5ee472 --- /dev/null +++ b/internal/resource/redis_test.go @@ -0,0 +1,57 @@ +package resource + +import ( + "context" + "slices" + "strings" + "testing" +) + +func TestRedisEngineAndKinds(t *testing.T) { + r := Redis{} + if r.Engine() != "redis" { + t.Errorf("Engine() = %q, want redis", r.Engine()) + } + if !slices.Contains(r.Kinds(), "queue") || !slices.Contains(r.Kinds(), "topic") { + t.Errorf("Kinds() = %v", r.Kinds()) + } +} + +func TestRedisEnsureIsHintOnly(t *testing.T) { + tgt := Target{Instance: "redis", Host: "127.0.0.1", Port: 46379} + tests := []struct { + kind string + wantHint string + }{ + {"queue", "LPUSH"}, + {"topic", "PUBLISH"}, + } + for _, tc := range tests { + t.Run(tc.kind, func(t *testing.T) { + attrs, err := Redis{}.Ensure(context.Background(), tgt, + Resource{Engine: "redis", Kind: tc.kind, Name: "web-jobs", Owner: "web"}) + if err != nil { + t.Fatalf("Ensure: %v", err) + } + if attrs["key"] != "web-jobs" { + t.Errorf("attrs[key] = %q, want web-jobs (tenant-prefixed)", attrs["key"]) + } + if attrs["host"] != "shared-redis" { + t.Errorf("attrs[host] = %q, want shared-redis", attrs["host"]) + } + if !strings.Contains(attrs["hint"], tc.wantHint) { + t.Errorf("hint = %q, want it to mention %q", attrs["hint"], tc.wantHint) + } + }) + } +} + +func TestRedisDropAndPreflightNoop(t *testing.T) { + tgt := Target{Instance: "redis", Host: "127.0.0.1", Port: 46379} + if err := (Redis{}).Drop(context.Background(), tgt, Resource{Engine: "redis", Kind: "queue", Name: "web-jobs"}); err != nil { + t.Errorf("Drop must be a no-op: %v", err) + } + if err := (Redis{}).Preflight(context.Background(), tgt); err != nil { + t.Errorf("Preflight must succeed without a client: %v", err) + } +} diff --git a/internal/resource/resource.go b/internal/resource/resource.go index 234e795..b1adf76 100644 --- a/internal/resource/resource.go +++ b/internal/resource/resource.go @@ -72,12 +72,14 @@ type Provisioner interface { // without a live connection. Engines absent here are forward-tolerant (unknown → // no kind check; the config resolver only enforces membership for known engines). var supportedKinds = map[string][]string{ - "postgres": {"database", "role", "user"}, - "redis": {"redis_index", "acl_user"}, - "minio": {"bucket", "lifecycle", "access_key"}, + "postgres": {"database", "role", "user"}, + "redis": {"redis_index", "acl_user", "queue", "topic"}, + "minio": {"bucket", "lifecycle", "access_key"}, + // The localstack template (`provides: aws`) backs SQS queues + SNS topics; the + // provisioner is keyed by the template name "localstack". "localstack": {"bucket", "queue", "topic", "stream", "table"}, - "nats": {"stream", "consumer", "kv"}, - "kafka": {"topic", "acl"}, + "nats": {"stream", "consumer", "kv", "queue", "topic"}, + "kafka": {"topic", "acl", "stream"}, } // Kinds returns the kinds a given engine's provisioner can create, or nil if the