feat(connectors): add SurrealDB sink connector

.github/workflows/_build_rust_artifacts.yml

@@ -46,7 +46,7 @@ on:
       connector_plugins:
         type: string
         required: false
-        default: "iggy_connector_elasticsearch_sink,iggy_connector_elasticsearch_source,iggy_connector_iceberg_sink,iggy_connector_postgres_sink,iggy_connector_postgres_source,iggy_connector_quickwit_sink,iggy_connector_random_source,iggy_connector_stdout_sink"
+        default: "iggy_connector_elasticsearch_sink,iggy_connector_elasticsearch_source,iggy_connector_iceberg_sink,iggy_connector_postgres_sink,iggy_connector_postgres_source,iggy_connector_quickwit_sink,iggy_connector_random_source,iggy_connector_stdout_sink,iggy_connector_surrealdb_sink"
         description: "Comma-separated list of connector plugin crates to build as shared libraries"
     outputs:
       artifact_name:

.github/workflows/edge-release.yml

@@ -109,6 +109,7 @@ jobs:
             - `iggy_connector_quickwit_sink`
             - `iggy_connector_random_source`
             - `iggy_connector_stdout_sink`
+            - `iggy_connector_surrealdb_sink`
 
             ## Downloads
 

Cargo.toml

@@ -43,6 +43,7 @@ members = [
     "core/connectors/sinks/postgres_sink",
     "core/connectors/sinks/quickwit_sink",
     "core/connectors/sinks/stdout_sink",
+    "core/connectors/sinks/surrealdb_sink",
     "core/connectors/sources/elasticsearch_source",
     "core/connectors/sources/influxdb_source",
     "core/connectors/sources/postgres_source",

core/connectors/README.md

@@ -86,6 +86,7 @@ Each sink should have its own, custom configuration, which is passed along with
 - **PostgreSQL Sink** - stores messages in PostgreSQL database tables
 - **Quickwit Sink** - indexes messages in Quickwit search engine
 - **Stdout Sink** - prints messages to standard output (useful for debugging/development)
+- **SurrealDB Sink** - writes messages into SurrealDB with deterministic record IDs for idempotent replay
 
 ## Source
 

core/connectors/sinks/README.md

@@ -15,6 +15,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s
 | **postgres_sink** | Stores messages in PostgreSQL database tables with configurable schemas |
 | **quickwit_sink** | Indexes messages in Quickwit search engine for log analytics |
 | **stdout_sink** | Prints messages to standard output (useful for debugging and development) |
+| **surrealdb_sink** | Writes messages into SurrealDB with deterministic record IDs for idempotent replay |
 
 The sink is represented by the single `Sink` trait, which defines the basic interface for all sink connectors. It provides methods for initializing the sink, writing data to external destination, and closing the sink.
 

core/connectors/sinks/surrealdb_sink/Cargo.toml

@@ -0,0 +1,49 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_surrealdb_sink"
+version = "0.4.1-edge.1"
+description = "Iggy SurrealDB sink connector for writing stream messages into SurrealDB"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "surrealdb", "sink"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+publish = false
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+base64 = { workspace = true }
+bytes = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+reqwest = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+simd-json = { workspace = true }

core/connectors/sinks/surrealdb_sink/README.md

@@ -0,0 +1,99 @@
+# SurrealDB Sink Connector
+
+Writes Apache Iggy stream messages into SurrealDB over the HTTP API.
+
+The sink writes one SurrealQL bulk `INSERT IGNORE` per connector batch. Each
+record uses a deterministic SurrealDB record id derived from stream, topic,
+partition, offset and Iggy message id, so replayed batches are idempotent and
+existing records are left untouched.
+
+Persistent sink failures are at-most-once from the runtime's perspective:
+messages may already be committed in Iggy before this connector exhausts its
+write attempts, so failed writes are logged but not redelivered.
+
+## Configuration
+
+```toml
+type = "sink"
+key = "surrealdb"
+enabled = true
+version = 0
+name = "SurrealDB sink"
+path = "target/release/libiggy_connector_surrealdb_sink"
+plugin_config_format = "toml"
+
+[[streams]]
+stream = "example_stream"
+topics = ["example_topic"]
+schema = "json"
+batch_length = 1000
+poll_interval = "5ms"
+consumer_group = "surrealdb_sink_connector"
+
+[plugin_config]
+endpoint = "127.0.0.1:8000"
+namespace = "iggy"
+database = "connectors"
+table = "iggy_messages"
+username = "root"
+password = "root"
+auth_scope = "root"
+use_tls = false
+auto_define_table = true
+define_indexes = true
+batch_size = 1000
+payload_format = "auto"
+include_metadata = true
+include_headers = true
+include_checksum = true
+include_origin_timestamp = true
+query_timeout = "30s"
+max_retries = 3
+retry_delay = "100ms"
+max_retry_delay = "5s"
+verbose_logging = false
+```
+
+### Plugin Fields
+
+| Field | Default | Description |
+| --- | --- | --- |
+| `endpoint` | required | SurrealDB HTTP host and port without scheme, for example `127.0.0.1:8000`. Full `http://` or `https://` URLs are also accepted. |
+| `namespace` | required | SurrealDB namespace selected during `open()`. |
+| `database` | required | SurrealDB database selected during `open()`. |
+| `table` | required | Target table. Must be a safe SurrealQL identifier. |
+| `username` / `password` | none | Optional credentials. |
+| `auth_scope` | `root` | `root`, `namespace`, `database`, or `none`. |
+| `use_tls` | `false` | Uses `https://` when true and `endpoint` has no scheme, `http://` otherwise. |
+| `auto_define_table` | `false` | Runs `DEFINE TABLE IF NOT EXISTS <table> SCHEMALESS`. |
+| `define_indexes` | `false` | Defines an offset index on stream/topic/partition/offset. Requires `auto_define_table`. |
+| `batch_size` | `1000` | Maximum number of records per SurrealDB request. |
+| `payload_format` | `auto` | `auto`, `json`, `text`, `base64`, or `binary` (`binary` is an alias for `base64`). |
+| `include_metadata` | `true` | Stores stream/topic/partition/offset/timestamps/schema fields. |
+| `include_headers` | `true` | Stores Iggy headers as a deterministic object. Raw headers are base64 encoded. |
+| `include_checksum` | `true` | Stores `iggy_checksum`. |
+| `include_origin_timestamp` | `true` | Stores `iggy_origin_timestamp`. |
+| `query_timeout` | `30s` | SurrealDB HTTP request timeout. |
+| `max_retries` | `3` | Total attempts for transient write failures. Values below `1` are raised to `1`. |
+| `retry_delay` | `100ms` | Base retry delay. |
+| `max_retry_delay` | `5s` | Capped exponential retry delay. |
+| `verbose_logging` | `false` | Emits per-batch success logs at `info`. |
+
+## Stored Shape
+
+With metadata enabled, records contain:
+
+- `id`: deterministic SurrealDB record id key
+- `iggy_message_id`: original Iggy message id as a string
+- `iggy_stream`, `iggy_topic`, `iggy_partition_id`, `iggy_offset`
+- `iggy_timestamp`, `iggy_origin_timestamp`, `iggy_checksum`, `iggy_schema`
+- `iggy_headers`
+- `payload`
+- `payload_encoding`
+
+`payload_format = "auto"` stores decoded JSON payloads as queryable SurrealDB
+values, text payloads as strings, and binary payloads as base64 strings.
+
+The `messages_processed` counter reports valid records submitted to SurrealDB.
+With `INSERT IGNORE`, duplicates can be ignored by SurrealDB while still being
+counted as submitted.

core/connectors/sinks/surrealdb_sink/config.toml

@@ -0,0 +1,58 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+type = "sink"
+key = "surrealdb"
+enabled = true
+version = 0
+name = "SurrealDB sink"
+path = "target/release/libiggy_connector_surrealdb_sink"
+plugin_config_format = "toml"
+verbose = false
+benchmark = false
+
+[[streams]]
+stream = "example_stream"
+topics = ["example_topic"]
+schema = "json"
+batch_length = 1000
+poll_interval = "5ms"
+consumer_group = "surrealdb_sink_connector"
+
+[plugin_config]
+endpoint = "127.0.0.1:8000"
+namespace = "iggy"
+database = "connectors"
+table = "iggy_messages"
+username = "root"
+password = "root"
+auth_scope = "root"
+use_tls = false
+auto_define_table = true
+define_indexes = true
+batch_size = 1000
+payload_format = "auto"
+include_metadata = true
+include_headers = true
+include_checksum = true
+include_origin_timestamp = true
+query_timeout = "30s"
+# Total write attempts for transient failures. Minimum is 1. A value of 0 is raised to 1.
+max_retries = 3
+retry_delay = "100ms"
+max_retry_delay = "5s"
+verbose_logging = false