feat: support creating resources from spec

.github/workflows/ci.yml

@@ -15,6 +15,7 @@ jobs:
     runs-on: ubuntu-latest
     outputs:
       sdk: ${{ steps.filter.outputs.sdk }}
+      apply_schema: ${{ steps.filter.outputs.apply_schema }}
     steps:
       - uses: actions/checkout@v4
       - uses: dorny/paths-filter@v3
@@ -23,6 +24,25 @@ jobs:
           filters: |
             sdk:
               - 'sdk/**'
+            apply_schema:
+              - 'cli/src/cli.rs'
+              - 'cli/src/main.rs'
+              - 'lite/src/init.rs'
+              - 'cli/schema.json'
+
+  cli-schema-drift:
+    name: CLI Schema Drift
+    needs: [changes]
+    if: needs.changes.outputs.apply_schema == 'true'
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+      - uses: Swatinem/rust-cache@v2
+      - name: Generate apply schema
+        run: cargo run -q -p s2-cli -- apply --schema > /tmp/apply.schema.json
+      - name: Check for schema drift
+        run: diff -u cli/schema.json /tmp/apply.schema.json
 
   fmt:
     name: Format

Cargo.toml

@@ -54,6 +54,7 @@ s2-api = { path = "api", version = "0.27" }
 s2-common = { path = "common", version = "0.28" }
 s2-lite = { path = "lite", version = "0.29" }
 s2-sdk = { path = "sdk", version = "0.24" }
+schemars = "0.8"
 serde = "1.0"
 serde_json = "1.0"
 slatedb = "0.10"

api/src/v1/basin.rs

@@ -4,7 +4,7 @@ use s2_common::types::{
 };
 use serde::{Deserialize, Serialize};
 
-use super::config::BasinConfig;
+use super::config::{BasinConfig, BasinReconfiguration};
 
 #[rustfmt::skip]
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -116,8 +116,8 @@ impl From<types::basin::BasinState> for BasinState {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
 pub struct CreateOrReconfigureBasinRequest {
-    /// Basin configuration.
-    pub config: Option<BasinConfig>,
+    /// Basin reconfiguration.
+    pub config: Option<BasinReconfiguration>,
     /// Basin scope.
     /// This cannot be reconfigured.
     pub scope: Option<BasinScope>,

cli/schema.json

@@ -0,0 +1,225 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "ResourcesSpec",
+  "type": "object",
+  "properties": {
+    "basins": {
+      "type": "array",
+      "items": {
+        "$ref": "#/definitions/BasinSpec"
+      }
+    }
+  },
+  "definitions": {
+    "BasinConfigSpec": {
+      "type": "object",
+      "properties": {
+        "create_stream_on_append": {
+          "description": "Create stream on append if it doesn't exist, using the default stream configuration.",
+          "default": null,
+          "type": [
+            "boolean",
+            "null"
+          ]
+        },
+        "create_stream_on_read": {
+          "description": "Create stream on read if it doesn't exist, using the default stream configuration.",
+          "default": null,
+          "type": [
+            "boolean",
+            "null"
+          ]
+        },
+        "default_stream_config": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/StreamConfigSpec"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        }
+      },
+      "additionalProperties": false
+    },
+    "BasinSpec": {
+      "type": "object",
+      "required": [
+        "name"
+      ],
+      "properties": {
+        "config": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/BasinConfigSpec"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        "name": {
+          "type": "string"
+        },
+        "streams": {
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/StreamSpec"
+          }
+        }
+      },
+      "additionalProperties": false
+    },
+    "DeleteOnEmptySpec": {
+      "type": "object",
+      "properties": {
+        "min_age": {
+          "description": "Minimum age before an empty stream can be deleted. Set to 0 (default) to disable delete-on-empty (don't delete automatically).",
+          "anyOf": [
+            {
+              "$ref": "#/definitions/HumanDuration"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        }
+      },
+      "additionalProperties": false
+    },
+    "HumanDuration": {
+      "description": "A duration string in humantime format, e.g. \"1day\", \"2h 30m\"",
+      "examples": [
+        "1day",
+        "2h 30m"
+      ],
+      "type": "string"
+    },
+    "RetentionPolicySpec": {
+      "description": "Retain records unless explicitly trimmed (\"infinite\"), or automatically trim records older than the given duration (e.g. \"7days\", \"1week\").",
+      "examples": [
+        "infinite",
+        "7days",
+        "1week"
+      ],
+      "type": "string"
+    },
+    "StorageClassSpec": {
+      "description": "Storage class for recent writes.",
+      "type": "string",
+      "enum": [
+        "standard",
+        "express"
+      ]
+    },
+    "StreamConfigSpec": {
+      "type": "object",
+      "properties": {
+        "delete_on_empty": {
+          "description": "Delete-on-empty configuration.",
+          "anyOf": [
+            {
+              "$ref": "#/definitions/DeleteOnEmptySpec"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        "retention_policy": {
+          "description": "Retention policy for the stream. If unspecified, the default is to retain records for 7 days.",
+          "anyOf": [
+            {
+              "$ref": "#/definitions/RetentionPolicySpec"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        "storage_class": {
+          "description": "Storage class for recent writes.",
+          "default": null,
+          "anyOf": [
+            {
+              "$ref": "#/definitions/StorageClassSpec"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        "timestamping": {
+          "description": "Timestamping behavior.",
+          "anyOf": [
+            {
+              "$ref": "#/definitions/TimestampingSpec"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        }
+      },
+      "additionalProperties": false
+    },
+    "StreamSpec": {
+      "type": "object",
+      "required": [
+        "name"
+      ],
+      "properties": {
+        "config": {
+          "anyOf": [
+            {
+              "$ref": "#/definitions/StreamConfigSpec"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        "name": {
+          "type": "string"
+        }
+      },
+      "additionalProperties": false
+    },
+    "TimestampingModeSpec": {
+      "description": "Timestamping mode for appends that influences how timestamps are handled.",
+      "type": "string",
+      "enum": [
+        "client-prefer",
+        "client-require",
+        "arrival"
+      ]
+    },
+    "TimestampingSpec": {
+      "type": "object",
+      "properties": {
+        "mode": {
+          "description": "Timestamping mode for appends that influences how timestamps are handled.",
+          "default": null,
+          "anyOf": [
+            {
+              "$ref": "#/definitions/TimestampingModeSpec"
+            },
+            {
+              "type": "null"
+            }
+          ]
+        },
+        "uncapped": {
+          "description": "Allow client-specified timestamps to exceed the arrival time. If this is `false` or not set, client timestamps will be capped at the arrival time.",
+          "default": null,
+          "type": [
+            "boolean",
+            "null"
+          ]
+        }
+      },
+      "additionalProperties": false
+    }
+  }
+}