feat(connectors): add SurrealDB source connector

### Description

Add a first-class SurrealDB source connector that reads durable SurrealDB changefeeds and publishes them to Apache Iggy as JSON messages.

SurrealDB changefeeds replay table changes through `SHOW CHANGES FOR TABLE ... SINCE ... LIMIT ...`. Ordered versionstamps map naturally to Iggy's source polling and persisted `ConnectorState` model.

This enables:

- Database change capture from SurrealDB without application-side dual writes.
- Fan-out of SurrealDB changes to Iggy consumers and downstream sinks.
- Replay after connector restarts, bounded by SurrealDB changefeed retention.
- Use of Iggy transforms, routing, metrics, and delivery tooling for SurrealDB changes.

The SurrealDB sink was implemented in #3453. The source should match its endpoint, authentication, TLS, retry, logging, and Docker-fixture conventions where applicable.

### Affected area / component

Connectors

### Proposed solution

Create `core/connectors/sources/surrealdb_source` as a `cdylib` plus `lib` crate implementing `iggy_connector_sdk::Source`.

The first implementation should be narrow and replayable: one SurrealDB table per connector instance, read through a durable changefeed over the HTTP `/sql` API. Operators can configure multiple source instances for multiple tables or Iggy destinations.

#### Phase 1: durable changefeed mode

Poll with:

```surql
SHOW CHANGES FOR TABLE
SINCE
LIMIT ;
```

A changefeed must already be enabled by the operator. The connector must not change table definitions or retention automatically.

Example for a new table:

```surql
DEFINE TABLE events CHANGEFEED 7d;
```

For an existing table, the README should document the appropriate `ALTER TABLE ... CHANGEFEED ...` command for the supported SurrealDB version. It must explain that retained history is limited by the configured changefeed duration and the initial cursor must be within that window.

#### Transport and dependency decision

Use `reqwest` with SurrealDB's HTTP `/sql` endpoint, following `surrealdb_sink`, unless implementation work proves that a required capability needs the official Rust SDK.

- `SHOW CHANGES` is available through SurrealQL over HTTP.
- The sink already provides patterns for endpoint normalization, scoped sign-in, namespace/database headers, response parsing, retry classification, and secret-safe logging.
- This avoids adding an embedded engine or a large new dependency tree.
- Live queries require WebSocket state and are outside Phase 1.

Do not refactor the sink and source into a shared crate in this issue. A shared SurrealDB transport crate can be proposed separately if both plugins later need one.

#### Draft configuration

```toml
type = "source"
key = "surrealdb"
enabled = true
version = 0
name = "SurrealDB source"
path = "../../target/release/libiggy_connector_surrealdb_source"
plugin_config_format = "toml"
verbose = false
benchmark = false

[[streams]]
stream = "surrealdb_events"
topic = "changes"
schema = "json"
batch_length = 1000
linger_time = "5ms"

[plugin_config]
endpoint = "127.0.0.1:8000"
namespace = "iggy"
database = "connectors"
table = "events"
username = "root"
password = "root"
auth_scope = "root"
use_tls = false

poll_interval = "500ms"
batch_size = 1000
start_from = "now"
# initial_versionstamp = "116395873313161217"
# initial_timestamp = "2026-07-17T00:00:00Z"

query_timeout = "30s"
max_retries = 3
retry_delay = "100ms"
max_retry_delay = "5s"
verbose_logging = false
```

#### Configuration contract

| Field | Default | Required behavior |
| --- | --- | --- |
| `endpoint` | required | Match `surrealdb_sink` validation. Accept host/port or an explicit HTTP(S) URL. Reject embedded credentials, paths, queries, and fragments. |
| `namespace` | required | Namespace sent with every query. Validate as a safe identifier. |
| `database` | required | Database sent with every query. Validate as a safe identifier. |
| `table` | required | One table per connector instance. Validate or quote safely before building SurrealQL. Never interpolate unchecked input. |
| `username` / `password` | none | Optional system-user credentials. Store `password` as `SecretString` and redact it from serialization and logs. |
| `auth_scope` | `root` | Match the sink: `root`, `namespace`, `database`, or `none`. |
| `use_tls` | `false` | Use HTTPS when no explicit scheme is present. |
| `poll_interval` | `1s` | Sleep before every poll, including the first, to avoid a busy loop. |
| `batch_size` | `1000` | Passed to `SHOW CHANGES ... LIMIT`. One returned versionstamp entry becomes one Iggy message. |
| `start_from` | `now` | Initial position used only without persisted state: `now`, `beginning`, `versionstamp`, or `timestamp`. |
| `initial_versionstamp` | none | Required for `start_from = "versionstamp"`. Preserve losslessly and never parse through floating point. |
| `initial_timestamp` | none | Required for `start_from = "timestamp"`. Validate as RFC 3339 UTC. |
| `query_timeout` | `30s` | Per-request timeout. |
| `max_retries` | `3` | Total attempts for transient failures. Minimum is 1. |
| `retry_delay` | `100ms` | Initial retry delay. |
| `max_retry_delay` | `5s` | Maximum exponential backoff delay. |
| `verbose_logging` | `false` | Emit per-batch success details at `info` instead of `debug`. |

New optional fields must use `Option` or `#[serde(default)]`.

Valid persisted state always wins over the initial-position settings. A configuration change must not silently reset an existing source.

Initial-position behavior:

- `now`: store the current RFC 3339 UTC time for a fresh source, query from it until the first versionstamp is returned, then persist versionstamps.
- `beginning`: start from versionstamp `0`, limited by retained history.
- `versionstamp`: require `initial_versionstamp`.
- `timestamp`: require `initial_timestamp`.

Invalid or conflicting fields must fail during initialization with an actionable error.

#### Message contract

Emit `Schema::Json`. Create one Iggy message per SurrealDB versionstamp entry, preserving the complete ordered `changes` array:

```json
{
"source": "surrealdb",
"namespace": "iggy",
"database": "connectors",
"table": "events",
"versionstamp": "116395873313161217",
"changes": [
{
"update": {
"id": "events:01J...",
"type": "order.created"
}
}
]
}
```

Requirements:

- Preserve the native mutation shape. Do not invent a normalized CRUD format. Responses differ when `INCLUDE ORIGINAL` is enabled, and updates can contain reverse JSON Patch operations.
- Represent versionstamps losslessly, preferably as decimal strings in payloads and state.
- Set deterministic string headers for namespace, database, table, and versionstamp.
- Derive a stable `ProducedMessage.id: u128` from namespace, database, table, and versionstamp. A replay must produce the same ID.
- Leave `origin_timestamp` unset unless SurrealDB supplies a real event timestamp. A versionstamp is an ordering token, not wall-clock time.
- Leave Iggy-owned timestamp and checksum fields unset.
- Reject malformed responses without advancing state.
- Support both normal changefeeds and `INCLUDE ORIGINAL` response shapes.

One message per versionstamp preserves the SurrealDB change group and makes `LIMIT` and cursor advancement unambiguous.

#### State and delivery semantics

Persist bounded MessagePack state through `ConnectorState::serialize` and `ConnectorState::deserialize`:

```rust
struct State {
cursor: Cursor,
messages_produced: u64,
}
```

The cursor stores either the initial timestamp or the latest fully emitted versionstamp.

Rules:

- `poll(&self)` sleeps first.
- Read the cursor under a brief `tokio::sync::Mutex` lock, drop the guard, perform HTTP I/O, build the full batch, then lock briefly to update state.
- Never hold a lock across network I/O.
- Return serialized state with every `ProducedMessages`, including empty polls.
- Advance only after validating the full response and building every output message.
- Do not advance on transport, auth, query, decoding, or serialization errors.
- Never modify or delete source records.
- Treat restart delivery as at-least-once. Stable message IDs make replay identifiable. Do not claim exactly-once, especially while the broader source acknowledgment contract in #3635 remains unresolved.
- If a stored cursor is older than retained history, return a clear permanent error and preserve it. Never jump silently to the newest entry.

Integration tests must establish whether `SINCE ` is inclusive or exclusive for the supported SurrealDB version. Consecutive polls must neither loop on the last entry nor skip the next one.

#### Lifecycle and error handling

`open(&mut self)` should:

1. Validate configuration and cursor combinations.
2. Build the HTTP client with the configured timeout.
3. Authenticate at the chosen scope when credentials are supplied.
4. Select namespace and database using the same protocol conventions as `surrealdb_sink`.
5. Verify connectivity and issue a non-destructive query proving that the table/changefeed is readable.
6. Fail clearly when the table is missing or has no changefeed.

`poll(&self)` should:

- Retry transport errors, timeouts, HTTP 408, HTTP 429, and HTTP 5xx responses with capped backoff.
- Treat HTTP 401/403, bad credentials, bad identifiers, missing changefeed, malformed SurrealQL, and incompatible responses as permanent.
- Avoid logging response bodies that can contain sensitive record data. Truncate or summarize backend details.
- Use `tracing` and include connector name and ID in lifecycle/error logs.
- Never log passwords, tokens, or endpoints with credentials.

`close(&mut self)` releases the client and logs the total produced count. The plugin must not spawn long-running Tokio tasks because the runtime owns lifecycle and shutdown.

#### Expected files and registration

```text
core/connectors/sources/surrealdb_source/
|-- Cargo.toml
|-- README.md
|-- config.toml
`-- src/lib.rs

core/integration/tests/connectors/
|-- fixtures/surrealdb/source.rs
`-- surrealdb/
|-- source.toml
|-- surrealdb_source.rs
`-- restart.rs
```

Also update:

- Workspace members and `Cargo.lock`.
- `core/connectors/README.md` and `core/connectors/sources/README.md`.
- `core/connectors/runtime/example_config/connectors/surrealdb_source.toml`.
- Release/version metadata wherever source packages are explicitly listed, including `scripts/bump-version.sh`.
- SurrealDB fixture and integration-test module registration.

Reuse the existing Docker-backed SurrealDB fixture. Do not create a second container implementation.

#### Required unit tests

Use connector BDD naming and include:

- The four canonical source state tests: persisted restore, no state, invalid state fallback, and serialization round trip.
- Defaults and validation for every config field.
- Invalid `start_from` combinations.
- Endpoint normalization and embedded-credential rejection.
- Namespace/database/table identifier validation.
- Secret redaction in `Debug` and serialized config.
- Empty, single-entry, and multi-entry changefeed parsing.
- Multiple mutations in one `changes` array.
- Update/delete and `INCLUDE ORIGINAL` reverse-diff parsing.
- Lossless large-versionstamp handling.
- Deterministic message IDs.
- Cursor advancement only for a fully valid batch.
- Stable state on empty polls and parse failures.
- Transient/permanent error classification.
- Retry-delay validation.

#### Required Docker-backed integration tests

Extend `core/integration/tests/connectors/fixtures/surrealdb` and use `#[iggy_harness]` with a real SurrealDB container.

1. Create, update, and delete records, then verify expected versionstamps and raw change shapes arrive in Iggy.
2. Produce more entries than `batch_size`; verify pagination, ordering, and no cursor-boundary duplicate loop.
3. Restart the connector runtime, add records, and verify restored state resumes after the last persisted versionstamp.
4. Verify `start_from = "now"` does not emit older retained entries.
5. Verify `beginning` or an explicit versionstamp replays retained entries.
6. Verify empty polls return cleanly and do not spin.
7. Verify missing changefeed and invalid auth fail clearly without advancing state.

Poll for conditions. Do not use fixed long sleeps and do not mock SurrealDB.

#### Acceptance criteria

- [ ] `iggy_connector_surrealdb_source` builds as `cdylib` plus `lib` and loads in the runtime.
- [ ] `SHOW CHANGES` entries are emitted as `Schema::Json` with complete ordered change arrays.
- [ ] Versionstamps remain lossless and message IDs are deterministic across replay.
- [ ] State survives connector/runtime restarts.
- [ ] Consecutive polls and restart recovery have no gaps or infinite duplicate loops.
- [ ] Empty polls return state and respect `poll_interval`.
- [ ] Credentials use `SecretString` and do not appear in logs or serialized config/stats.
- [ ] Transient failures retry with capped backoff; permanent failures are actionable.
- [ ] Real integration coverage reuses the existing SurrealDB testcontainer fixture.
- [ ] README and runnable config document setup, auth, cursors, retention, payload shape, replay semantics, and troubleshooting.
- [ ] Workspace, connector indexes, examples, release metadata, and license manifests are updated where required.
- [ ] `cargo fmt --all` passes.
- [ ] `cargo sort --no-format --workspace` passes.
- [ ] `cargo clippy -p iggy_connector_surrealdb_source --all-targets -- -D warnings` passes.
- [ ] `cargo test -p iggy_connector_surrealdb_source` passes.
- [ ] Targeted SurrealDB integration tests pass with Docker.

#### Out of scope for Phase 1

- `LIVE SELECT` / WebSocket mode. SurrealDB documents current live queries as single-node, best-effort ordered, and not durably replayable after disconnect.
- Generic snapshot polling or arbitrary SurrealQL.
- Multiple tables per source instance.
- Embedded SurrealDB engines.
- Automatic table/changefeed DDL.
- A shared sink/source client crate.
- A normalized CDC format that hides SurrealDB's native change representation.
- Exactly-once delivery.

### Alternatives considered

1. **Live queries first.** Lower latency, but WebSocket-only with no durable reconnect cursor and possible disconnect gaps.
2. **Periodic SELECT polling.** Needs an operator-defined cursor and cannot naturally capture deletes.
3. **Official Rust SDK.** Still possible if HTTP lacks a required feature, but Phase 1 only needs `SHOW CHANGES` and should match the existing sink.
4. **Generic HTTP source.** It would not provide SurrealDB auth, cursor persistence, deterministic replay IDs, changefeed mapping, or retention diagnostics.

### References

- [SurrealDB changefeeds](https://surrealdb.com/docs/learn/querying/real-time/changefeeds)
- [SHOW CHANGES reference](https://surrealdb.com/docs/reference/query-language/statements/show)
- [DEFINE TABLE CHANGEFEED reference](https://surrealdb.com/docs/reference/query-language/statements/define/table)
- [SurrealDB HTTP protocol](https://surrealdb.com/docs/reference/rest-api/http-protocol)
- [SurrealDB live-query limitations](https://surrealdb.com/docs/reference/query-language/statements/live-select)
- Related sink implementation: #3453
- Related source acknowledgment issue: #3635

### Contribution

- [x] I'm willing to submit a pull request to implement this feature

### Good first issue

- [ ] I think this could be a good first issue for a new contributor