feat(natsstore): serve all tenants from a single muxed K/V bucket

.gitignore

@@ -0,0 +1,5 @@
+# build artifacts (compiled binaries)
+# `go build -o opa ./cmd/opa-nats` (Dockerfile, README) -> opa
+# bare `go build ./cmd/opa-nats`                          -> opa-nats
+/opa
+/opa-nats

README.md

@@ -41,12 +41,14 @@ The plugin can be configured through OPA's configuration system. Here's an examp
 plugins:
   nats:
     server_url: "nats://localhost:4222"
+    bucket: "POLICY_DATA"   # REQUIRED: the single muxed K/V bucket holding every tenant
+                     # as keys "<tenant>.<key...>"
     ttl: "10m"
     refresh_interval: "30s"
     max_reconnect_attempts: 10
     reconnect_wait: "2s"
-    max_bucket_watchers: 10
-    root_bucket: ""  # Optional - leave empty for multi-bucket mode
+    max_bucket_watchers: 10   # LRU cap on concurrent per-tenant watchers
+    root_tenant: ""  # Optional - a tenant whose subtree mounts at the OPA data root
 
     # Authentication (choose one)
     credentials: "/path/to/nats.creds"
@@ -72,12 +74,13 @@ services:
 | Option | Type | Default | Description |
 |--------|------|---------|-------------|
 | `server_url` | string | `"nats://localhost:4222"` | NATS server URL |
+| `bucket` | string | **(required)** | The single muxed K/V bucket name (keys `<tenant>.<key>`). Mandatory — no default; must match the producer's bucket. |
 | `ttl` | duration | `"10m"` | TTL for cache entries |
 | `refresh_interval` | duration | `"30s"` | How often to refresh cache from NATS |
 | `max_reconnect_attempts` | int | `10` | Maximum reconnection attempts |
 | `reconnect_wait` | duration | `"2s"` | Wait time between reconnection attempts |
 | `max_bucket_watchers` | int | `10` | Maximum number of bucket watchers in LRU cache |
-| `root_bucket` | string | `""` | Root bucket name for default data (optional) |
+| `root_tenant` | string | `""` | A tenant whose subtree mounts at the OPA data root instead of under `data.nats.kv.<tenant>` (optional) |
 | `credentials` | string | `""` | Path to NATS credentials file |
 | `token` | string | `""` | NATS token for authentication |
 | `username` | string | `""` | NATS username |
@@ -89,28 +92,35 @@ services:
 
 ## Built-in Functions
 
-The plugin provides custom Rego built-in functions for interacting with NATS K/V:
+The plugin provides custom Rego built-in functions for interacting with NATS K/V.
 
-### `nats.kv.watch_bucket(bucket_name)`
+> **Single muxed bucket model.** All tenants live in one K/V bucket (`bucket` in
+> config), keyed as `<tenant>.<key...>`. The first argument to both builtins is
+> the **tenant id** — the plugin reads only that tenant's slice (a prefix-filtered
+> watch), never the whole bucket. The data lands at `data.nats.kv.<tenant>.<...>`,
+> the same place as before, so policies are unchanged.
 
-Watches a NATS K/V bucket and returns all its data. This function automatically manages bucket watchers with LRU caching.
+### `nats.kv.watch_bucket(tenant_id)`
+
+Watches a single tenant's slice of the muxed bucket and returns its data. Watchers
+are managed with LRU caching (one ordered consumer per watched tenant).
 
 ```rego
-# Watch a specific bucket
+# Watch one tenant
 group_data := nats.kv.watch_bucket("550e8400-e29b-41d4-a716-446655440000")
 
 # Access nested data
 members := group_data.members
 permissions := group_data.permissions
 ```
 
-### `nats.kv.get_data(bucket_name, key)`
+### `nats.kv.get_data(tenant_id, key)`
 
-Retrieves a specific key from a NATS K/V bucket.
+Retrieves a specific key from a tenant's slice (key is tenant-relative).
 
 ```rego
-# Get specific data from a bucket
-user_data := nats.kv.get_data("users", "123e4567-e89b-12d3-a456-426614174000")
+# Get specific data for a tenant
+user_data := nats.kv.get_data("550e8400-e29b-41d4-a716-446655440000", "members")
 ```
 
 ## Examples
@@ -276,8 +286,19 @@ accounts: {
     jetstream: enabled
     users: [
       {user: "opa", pass: "secret", permissions: {
-        subscribe: ["$JS.API.>", "$KV.>"]
-        publish: ["$JS.API.>", "$KV.>"]
+        # Scope to the single muxed bucket (POLICY_DATA). The plugin is a multi-tenant
+        # cloud reader living in the shared account, so it sees all tenants;
+        # narrow further per-deployment if desired.
+        #
+        # Caveats:
+        #  - `$JS.API.>` is account-wide JetStream API access (manage/observe every
+        #    stream and KV bucket in this account). For genuine least privilege,
+        #    scope it to this bucket's API subjects instead of granting `>`.
+        #  - The bucket name must stay in lockstep with the plugin's `bucket`
+        #    config: if you change `bucket`, change `$KV.<bucket>.>` here too, or
+        #    the watch is silently denied.
+        subscribe: ["$JS.API.>", "$KV.POLICY_DATA.>"]
+        publish: ["$JS.API.>", "$KV.POLICY_DATA.>"]
       }}
     ]
   }

examples/opa-nats/config-compose.yaml

@@ -4,12 +4,13 @@
 plugins:
   nats:
     server_url: "nats://nats:4222"
+    bucket: "POLICY_DATA"
     ttl: "10m"
     refresh_interval: "30s"
     max_reconnect_attempts: 10
     reconnect_wait: "2s"
     max_bucket_watchers: 1
-    root_bucket: "permit-groups-default"
+    root_tenant: "permit-groups-default"
 
 services:
   authz:

examples/opa-nats/config.yaml

@@ -6,12 +6,13 @@ plugins:
     # OPA runs inside the compose network and needs to reach NATS via the
     # docker service name `nats`.
     server_url: "nats://localhost:4222"
+    bucket: "POLICY_DATA"
     ttl: "10m"
     refresh_interval: "30s"
     max_reconnect_attempts: 10
     reconnect_wait: "2s"
     max_bucket_watchers: 1
-    root_bucket: "permit-groups-default"
+    root_tenant: "permit-groups-default"
 
 services:
   authz:

examples/opa-nats/docker-compose.yaml

@@ -59,67 +59,50 @@ services:
       - NATS_URL=nats://nats:4222
     command: >
       /bin/sh -c "
-        echo 'Setting up NATS K/V buckets for UUID-based groups (each group UUID = separate bucket)...';
-
-        # Create individual buckets for each group UUID
-        echo 'Creating group UUID buckets...';
-        nats kv add 550e8400-e29b-41d4-a716-446655440000 --replicas=1 --history=10 --ttl=12h;
-        nats kv add 660e8400-e29b-41d4-a716-446655440001 --replicas=1 --history=10 --ttl=12h;
-        nats kv add 770e8400-e29b-41d4-a716-446655440002 --replicas=1 --history=10 --ttl=12h;
-
-        # Create default bucket for non-group data (users)
-        nats kv add permit-groups-default --replicas=1 --history=10 --ttl=12h;
-
-        echo 'All buckets created successfully';
-
-        echo 'Populating Administrators group (bucket: 550e8400-e29b-41d4-a716-446655440000)...';
-        # In this bucket, store data without the groups.{uuid} prefix since the bucket IS the group
-        nats kv put 550e8400-e29b-41d4-a716-446655440000 metadata '{\"id\":\"550e8400-e29b-41d4-a716-446655440000\",\"name\":\"Administrators\",\"description\":\"System administrators group\",\"created_at\":\"2024-01-15T10:00:00Z\"}';
-        nats kv put 550e8400-e29b-41d4-a716-446655440000 members.123e4567-e89b-12d3-a456-426614174000 '{\"user_id\":\"123e4567-e89b-12d3-a456-426614174000\",\"joined_at\":\"2024-01-15T10:05:00Z\",\"role\":\"admin\"}';
-        nats kv put 550e8400-e29b-41d4-a716-446655440000 members.234e5678-e89b-12d3-a456-426614174001 '{\"user_id\":\"234e5678-e89b-12d3-a456-426614174001\",\"joined_at\":\"2024-01-15T10:10:00Z\",\"role\":\"member\"}';
-        nats kv put 550e8400-e29b-41d4-a716-446655440000 permissions.perm-admin-api '{\"id\":\"perm-admin-api\",\"resource\":\"api\",\"action\":\"admin\",\"effect\":\"allow\"}';
-        nats kv put 550e8400-e29b-41d4-a716-446655440000 permissions.perm-admin-users '{\"id\":\"perm-admin-users\",\"resource\":\"users\",\"action\":\"manage\",\"effect\":\"allow\"}';
-
-        echo 'Populating Developers group (bucket: 660e8400-e29b-41d4-a716-446655440001)...';
-        nats kv put 660e8400-e29b-41d4-a716-446655440001 metadata '{\"id\":\"660e8400-e29b-41d4-a716-446655440001\",\"name\":\"Developers\",\"description\":\"Development team group\",\"created_at\":\"2024-01-15T10:30:00Z\"}';
-        nats kv put 660e8400-e29b-41d4-a716-446655440001 members.345e6789-e89b-12d3-a456-426614174002 '{\"user_id\":\"345e6789-e89b-12d3-a456-426614174002\",\"joined_at\":\"2024-01-15T10:35:00Z\",\"role\":\"lead\"}';
-        nats kv put 660e8400-e29b-41d4-a716-446655440001 members.234e5678-e89b-12d3-a456-426614174001 '{\"user_id\":\"234e5678-e89b-12d3-a456-426614174001\",\"joined_at\":\"2024-01-15T10:40:00Z\",\"role\":\"member\"}';
-        nats kv put 660e8400-e29b-41d4-a716-446655440001 permissions.perm-dev-code '{\"id\":\"perm-dev-code\",\"resource\":\"code\",\"action\":\"write\",\"effect\":\"allow\"}';
-        nats kv put 660e8400-e29b-41d4-a716-446655440001 permissions.perm-dev-deploy '{\"id\":\"perm-dev-deploy\",\"resource\":\"deployment\",\"action\":\"deploy\",\"effect\":\"allow\"}';
-
-        echo 'Populating ReadOnly Users group (bucket: 770e8400-e29b-41d4-a716-446655440002)...';
-        nats kv put 770e8400-e29b-41d4-a716-446655440002 metadata '{\"id\":\"770e8400-e29b-41d4-a716-446655440002\",\"name\":\"ReadOnly Users\",\"description\":\"Read-only access group\",\"created_at\":\"2024-01-15T11:00:00Z\"}';
-        nats kv put 770e8400-e29b-41d4-a716-446655440002 members.456e7890-e89b-12d3-a456-426614174003 '{\"user_id\":\"456e7890-e89b-12d3-a456-426614174003\",\"joined_at\":\"2024-01-15T11:05:00Z\",\"role\":\"member\"}';
-        nats kv put 770e8400-e29b-41d4-a716-446655440002 permissions.perm-readonly-data '{\"id\":\"perm-readonly-data\",\"resource\":\"data\",\"action\":\"read\",\"effect\":\"allow\"}';
-
-        echo 'Populating default bucket with user group assignments...';
-        # User-group assignments go to the default bucket since they are not group-specific
-        nats kv put permit-groups-default metadata '{\"id\":\"permit-groups-default\",\"name\":\"Default Group\",\"description\":\"Default group for non-grouped data\",\"created_at\":\"2024-01-15T10:00:00Z\"}';
-
-        nats kv put permit-groups-default users '[\"550e8400-e29b-41d4-a716-446655440000\",\"660e8400-e29b-41d4-a716-446655440001\"]';
-        # nats kv put permit-groups-default users.234e5678-e89b-12d3-a456-426614174001.groups '[\"150e8400-e29b-41d4-a716-446655440000\",\"660e8400-e29b-41d4-a716-446655440001\"]';
-        # nats kv put permit-groups-default users.345e6789-e89b-12d3-a456-426614174002.groups '[\"160e8400-e29b-41d4-a716-446655440001\"]';
-        # nats kv put permit-groups-default users.456e7890-e89b-12d3-a456-426614174003.groups '[\"170e8400-e29b-41d4-a716-446655440002\"]';
-        # nats kv put permit-groups-default users.123e4567-e89b-12d3-a456-426614174000.groups '[\"150e8400-e29b-41d4-a716-446655440000\"]';
+        echo 'Setting up the single muxed NATS K/V bucket POLICY_DATA (keys are <tenant>.<key>)...';
+
+        # ONE bucket holds every tenant; the group UUID is the leading key token.
+        nats kv add POLICY_DATA --replicas=1 --history=10 --ttl=12h;
+        echo 'Bucket POLICY_DATA created successfully';
+
+        echo 'Populating Administrators group (tenant 550e8400-e29b-41d4-a716-446655440000)...';
+        nats kv put POLICY_DATA 550e8400-e29b-41d4-a716-446655440000.metadata '{\"id\":\"550e8400-e29b-41d4-a716-446655440000\",\"name\":\"Administrators\",\"description\":\"System administrators group\",\"created_at\":\"2024-01-15T10:00:00Z\"}';
+        nats kv put POLICY_DATA 550e8400-e29b-41d4-a716-446655440000.members.123e4567-e89b-12d3-a456-426614174000 '{\"user_id\":\"123e4567-e89b-12d3-a456-426614174000\",\"joined_at\":\"2024-01-15T10:05:00Z\",\"role\":\"admin\"}';
+        nats kv put POLICY_DATA 550e8400-e29b-41d4-a716-446655440000.members.234e5678-e89b-12d3-a456-426614174001 '{\"user_id\":\"234e5678-e89b-12d3-a456-426614174001\",\"joined_at\":\"2024-01-15T10:10:00Z\",\"role\":\"member\"}';
+        nats kv put POLICY_DATA 550e8400-e29b-41d4-a716-446655440000.permissions.perm-admin-api '{\"id\":\"perm-admin-api\",\"resource\":\"api\",\"action\":\"admin\",\"effect\":\"allow\"}';
+        nats kv put POLICY_DATA 550e8400-e29b-41d4-a716-446655440000.permissions.perm-admin-users '{\"id\":\"perm-admin-users\",\"resource\":\"users\",\"action\":\"manage\",\"effect\":\"allow\"}';
+
+        echo 'Populating Developers group (tenant 660e8400-e29b-41d4-a716-446655440001)...';
+        nats kv put POLICY_DATA 660e8400-e29b-41d4-a716-446655440001.metadata '{\"id\":\"660e8400-e29b-41d4-a716-446655440001\",\"name\":\"Developers\",\"description\":\"Development team group\",\"created_at\":\"2024-01-15T10:30:00Z\"}';
+        nats kv put POLICY_DATA 660e8400-e29b-41d4-a716-446655440001.members.345e6789-e89b-12d3-a456-426614174002 '{\"user_id\":\"345e6789-e89b-12d3-a456-426614174002\",\"joined_at\":\"2024-01-15T10:35:00Z\",\"role\":\"lead\"}';
+        nats kv put POLICY_DATA 660e8400-e29b-41d4-a716-446655440001.members.234e5678-e89b-12d3-a456-426614174001 '{\"user_id\":\"234e5678-e89b-12d3-a456-426614174001\",\"joined_at\":\"2024-01-15T10:40:00Z\",\"role\":\"member\"}';
+        nats kv put POLICY_DATA 660e8400-e29b-41d4-a716-446655440001.permissions.perm-dev-code '{\"id\":\"perm-dev-code\",\"resource\":\"code\",\"action\":\"write\",\"effect\":\"allow\"}';
+        nats kv put POLICY_DATA 660e8400-e29b-41d4-a716-446655440001.permissions.perm-dev-deploy '{\"id\":\"perm-dev-deploy\",\"resource\":\"deployment\",\"action\":\"deploy\",\"effect\":\"allow\"}';
+
+        echo 'Populating ReadOnly Users group (tenant 770e8400-e29b-41d4-a716-446655440002)...';
+        nats kv put POLICY_DATA 770e8400-e29b-41d4-a716-446655440002.metadata '{\"id\":\"770e8400-e29b-41d4-a716-446655440002\",\"name\":\"ReadOnly Users\",\"description\":\"Read-only access group\",\"created_at\":\"2024-01-15T11:00:00Z\"}';
+        nats kv put POLICY_DATA 770e8400-e29b-41d4-a716-446655440002.members.456e7890-e89b-12d3-a456-426614174003 '{\"user_id\":\"456e7890-e89b-12d3-a456-426614174003\",\"joined_at\":\"2024-01-15T11:05:00Z\",\"role\":\"member\"}';
+        nats kv put POLICY_DATA 770e8400-e29b-41d4-a716-446655440002.permissions.perm-readonly-data '{\"id\":\"perm-readonly-data\",\"resource\":\"data\",\"action\":\"read\",\"effect\":\"allow\"}';
+
+        echo 'Populating the root tenant (permit-groups-default mounts at the OPA data root)...';
+        nats kv put POLICY_DATA permit-groups-default.metadata '{\"id\":\"permit-groups-default\",\"name\":\"Default Group\",\"description\":\"Default group for non-grouped data\",\"created_at\":\"2024-01-15T10:00:00Z\"}';
+        nats kv put POLICY_DATA permit-groups-default.users '[\"550e8400-e29b-41d4-a716-446655440000\",\"660e8400-e29b-41d4-a716-446655440001\"]';
 
         echo 'Test data added successfully';
         echo '';
-        echo 'Created buckets:';
+        echo 'Bucket:';
         nats kv list;
         echo '';
         echo 'Sample verification:';
-        echo 'Administrators group metadata (bucket 550e8400-e29b-41d4-a716-446655440000):';
-        nats kv get 550e8400-e29b-41d4-a716-446655440000 metadata;
-        echo '';
-        echo 'Developers group members (bucket 660e8400-e29b-41d4-a716-446655440001):';
-        nats kv ls 660e8400-e29b-41d4-a716-446655440001 | grep members;
+        echo 'Administrators group metadata (tenant 550e8400-...):';
+        nats kv get POLICY_DATA 550e8400-e29b-41d4-a716-446655440000.metadata;
         echo '';
-        echo 'User group assignments (default bucket):';
-        nats kv get permit-groups-default users.234e5678-e89b-12d3-a456-426614174001.groups;
+        echo 'Developers group keys (tenant 660e8400-...):';
+        nats kv ls POLICY_DATA | grep 660e8400-e29b-41d4-a716-446655440001;
         echo '';
-        echo 'Setup complete - ready for bucket-per-group UUID testing';
-        echo 'Architecture: Each group UUID is its own NATS bucket containing that groups metadata, members, and permissions';
-        echo 'User group assignments stored in default bucket since they span multiple groups';
+        echo 'Setup complete - single muxed bucket POLICY_DATA, keys <tenant>.<key>';
+        echo 'Architecture: one bucket; each group UUID is the leading key token (the tenant);';
+        echo 'the plugin reads only a tenant slice via a prefix-filtered watch.';
       "
     networks:
       - nats-net
@@ -138,10 +121,12 @@ services:
     ports:
       - "8181:8181"
     volumes:
-      - ./config.yaml:/config/config.yaml
+      # In-compose OPA must reach NATS via the service name, so mount the
+      # compose config (server_url=nats://nats:4222), NOT config.yaml's
+      # host-friendly localhost default — otherwise `localhost:4222` resolves to
+      # the OPA container itself and the plugin can't reach NATS.
+      - ./config-compose.yaml:/config/config.yaml
       - ./rego:/policies
-    environment:
-      - NATS_URL=nats://nats:4222
     networks:
       - nats-net
 

examples/opa-nats/rego/test.rego

@@ -2,6 +2,15 @@ package test
 
 import rego.v1
 
+# input.bucket_id identifies the tenant whose slice of the single muxed bucket
+# we want. The Rego interface is unchanged from the bucket-per-tenant model: the
+# value is now the tenant token.
+#
+# Placement: a normal (watched) tenant's data is injected at
+# data.nats.kv.<bucket_id>.<...>. The one exception is the configured
+# `root_tenant` (e.g. permit-groups-default), which is mounted at the OPA data
+# ROOT with its tenant token stripped — so its keys land at data.<...>, not
+# under data.nats.kv. This example addresses non-root tenants below.
 default bucket_watched := false
 
 bucket_watched := nats.kv.watch_bucket(input.bucket_id)

pkg/natsstore/additional_test.go

@@ -17,9 +17,8 @@ import (
 // Test more complex scenarios and edge cases
 
 func TestDataTransformer_ensureParentPaths_Comprehensive(t *testing.T) {
-	config := DefaultConfig()
 	logger := logging.Get()
-	dt, err := NewDataTransformer(config, logger)
+	dt, err := NewDataTransformer(logger)
 	require.NoError(t, err)
 
 	store := NewMockStore()
@@ -61,9 +60,8 @@ func TestDataTransformer_ensureParentPaths_Comprehensive(t *testing.T) {
 }
 
 func TestDataTransformer_ensureParentPathsRecursive_Comprehensive(t *testing.T) {
-	config := DefaultConfig()
 	logger := logging.Get()
-	dt, err := NewDataTransformer(config, logger)
+	dt, err := NewDataTransformer(logger)
 	require.NoError(t, err)
 
 	store := NewMockStore()
@@ -105,9 +103,8 @@ func TestDataTransformer_ensureParentPathsRecursive_Comprehensive(t *testing.T)
 }
 
 func TestDataTransformer_createNestedStructure_Comprehensive(t *testing.T) {
-	config := DefaultConfig()
 	logger := logging.Get()
-	dt, err := NewDataTransformer(config, logger)
+	dt, err := NewDataTransformer(logger)
 	require.NoError(t, err)
 
 	store := NewMockStore()
@@ -164,6 +161,7 @@ func TestPluginFactory_Validate_EdgeCases(t *testing.T) {
 			name: "valid minimal config",
 			configData: map[string]interface{}{
 				"server_url": "nats://localhost:4222",
+				"bucket":     "DATA",
 			},
 			expectError: false,
 		},
@@ -189,7 +187,8 @@ func TestPluginFactory_Validate_EdgeCases(t *testing.T) {
 				"max_reconnect_attempts": 5,
 				"reconnect_wait":         "1s",
 				"max_bucket_watchers":    20,
-				"root_bucket":            "test-root",
+				"bucket":                 "DATA",
+				"root_tenant":            "test-root",
 				"username":               "testuser",
 				"password":               "testpass",
 			},