feat: Add Microsoft Dataverse source and bootstrap provider by ruokun-niu · Pull Request #304 · drasi-project/drasi-core

ruokun-niu · 2026-03-03T22:46:18Z

Description

Adds a new Dataverse source plugin and bootstrap provider that enable Drasi to continuously monitor Microsoft Dataverse (Power Platform) tables for data changes using the OData Web API change tracking mechanism — the REST equivalent of the platform's RetrieveEntityChangesRequest.

New Crates

drasi-source-dataverse (components/sources/dataverse/) — Polling-based change detection source using OData delta links
drasi-bootstrap-dataverse (components/bootstrappers/dataverse/) — Initial data loader that fetches all existing records from configured entities

Key Features

Change Tracking

Uses Prefer: odata.track-changes headers and @odata.deltaLink tokens (Web API equivalent of RetrieveEntityChangesRequest with DataVersion)
Per-entity worker tasks with independent delta tokens
Delta tokens persisted to StateStore for crash recovery

Adaptive Backoff

Matches the platform's SyncWorker adaptive polling pattern
500ms minimum → 30s maximum interval
1.2× multiplier under 5s, 1.5× above 5s
Resets to minimum on detected changes

Authentication

Client credentials (OAuth2) — tenant_id + client_id + client_secret
Azure CLI — az account get-access-token for local development without secrets

Platform YAML Compatibility

Serde aliases for camelCase field names (endpoint, clientId, clientSecret, tenantId, useAzureCli)
entities field accepts both a JSON array and a comma-separated string

Configuration Example

apiVersion: v1
kind: Source
name: leads
spec:
  kind: Dataverse
  properties:
    endpoint: https://orgXXXXX.crm.dynamics.com/
    entities: lead
    clientId: <app-client-id>
    clientSecret: <app-client-secret>
    tenantId: <azure-ad-tenant-id>

Or with Azure CLI auth (no secrets):

apiVersion: v1
kind: Source
name: leads
spec:
  kind: Dataverse
  properties:
    endpoint: https://orgXXXXX.crm.dynamics.com/
    entities: lead
    useAzureCli: true

Files Added/Modified

New Files

Path	Purpose
`components/sources/dataverse/`	Source crate (config, auth, types, client, lib)
`components/bootstrappers/dataverse/`	Bootstrap crate (config, lib)
`components/sources/dataverse/AUTHENTICATION.md`	Auth method documentation
`examples/lib/dataverse-getting-started/`	Standalone example with auto-detected auth

Modified Files

Path	Change
`Cargo.toml`	Added both crates to workspace members

Testing

43 unit tests across both crates covering config validation, serde aliases, entity set naming, change parsing, auth modes, builder patterns, and value conversion
1 integration test (ignored, requires live Dataverse environment or wiremock)
All tests pass, build clean

Prerequisites for Dataverse Environment

Change tracking must be enabled on target entities (Settings → Customization → Entity → Enable Change Tracking)
For client credentials: Azure AD app registration with Dataverse API permissions + application user in the environment
For Azure CLI: az login with a user that has Dataverse access

Type of change

This pull request fixes a bug in Drasi and has an approved issue (issue link required).
This pull request adds or changes features of Drasi and has an approved issue (issue link required).
This pull request is a minor refactor, code cleanup, test improvement, or other maintenance task and doesn't change the functionality of Drasi (issue link optional).

Fixes: #issue_number

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

components/sources/dataverse/src/auth.rs

+            "c",
+            "s",
+            "https://myorg.crm.dynamics.com",
+            "http://localhost:8080/token",


components/sources/dataverse/src/auth.rs

+            "https://myorg.crm.dynamics.com",
+            "http://localhost:8080/token",
+        );
+        assert_eq!(tm.token_url, "http://localhost:8080/token");


Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

Copilot

Pull request overview

This PR adds a new Microsoft Dataverse source plugin and bootstrap provider for Drasi, enabling continuous monitoring of Dataverse tables for data changes using OData Web API change tracking (delta links). It also extends AzureIdentityProvider with a with_client_secret constructor and updates the source planner agent docs.

Changes:

New drasi-source-dataverse crate with polling-based change detection using OData delta links, adaptive backoff, per-entity workers, and OAuth2/Azure CLI authentication
New drasi-bootstrap-dataverse crate for initial data loading from Dataverse entities
Added AzureIdentityProvider::with_client_secret() to the shared identity library

Reviewed changes

Copilot reviewed 19 out of 19 changed files in this pull request and generated 10 comments.

Show a summary per file

File	Description
`Cargo.toml`	Added both new crates to workspace members
`lib/src/identity/azure.rs`	Added `with_client_secret` constructor and tests
`components/sources/dataverse/Cargo.toml`	New crate manifest
`components/sources/dataverse/src/config.rs`	Source configuration with serde aliases and validation
`components/sources/dataverse/src/auth.rs`	Token manager with client credentials and Azure CLI support
`components/sources/dataverse/src/types.rs`	OData delta response types and change parsing
`components/sources/dataverse/src/client.rs`	HTTP client for Dataverse OData Web API
`components/sources/dataverse/src/lib.rs`	Main source plugin with polling loop and builder
`components/sources/dataverse/src/descriptor.rs`	Plugin descriptor for dynamic loading
`components/sources/dataverse/tests/integration_test.rs`	Wiremock-based integration test (ignored)
`components/sources/dataverse/README.md`	Source documentation
`components/bootstrappers/dataverse/Cargo.toml`	New crate manifest
`components/bootstrappers/dataverse/src/config.rs`	Bootstrap configuration with validation
`components/bootstrappers/dataverse/src/lib.rs`	Bootstrap provider with pagination
`components/bootstrappers/dataverse/src/descriptor.rs`	Plugin descriptor for dynamic loading
`components/bootstrappers/dataverse/README.md`	Bootstrap documentation
`.github/agents/source-planner.md`	Added authentication guidance for source plugins

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

components/sources/dataverse/Cargo.toml

@@ -0,0 +1,38 @@
+[package]


components/bootstrappers/dataverse/Cargo.toml

@@ -0,0 +1,37 @@
+[package]


components/bootstrappers/dataverse/src/lib.rs

+fn convert_json_value(value: &serde_json::Value) -> ElementValue {
+    match value {
+        serde_json::Value::Null => ElementValue::Null,
+        serde_json::Value::Bool(b) => ElementValue::Bool(*b),
+        serde_json::Value::Number(n) => {
+            if let Some(i) = n.as_i64() {
+                ElementValue::Integer(i)
+            } else if let Some(f) = n.as_f64() {
+                ElementValue::Float(ordered_float::OrderedFloat(f))
+            } else {
+                ElementValue::Null
+            }
+        }
+        serde_json::Value::String(s) => ElementValue::String(Arc::from(s.as_str())),
+        serde_json::Value::Array(arr) => {
+            if !arr.is_empty() {
+                if let Some(first_obj) = arr[0].as_object() {
+                    if first_obj.contains_key("Value") {
+                        let values: Vec<ElementValue> = arr
+                            .iter()
+                            .filter_map(|item| {
+                                item.as_object()
+                                    .and_then(|obj| obj.get("Value"))
+                                    .map(convert_json_value)
+                            })
+                            .collect();
+                        return ElementValue::List(values);
+                    }
+                }
+            }
+            ElementValue::List(arr.iter().map(convert_json_value).collect())
+        }
+        serde_json::Value::Object(obj) => {
+            if obj.contains_key("Value") && obj.len() <= 2 {
+                if let Some(val) = obj.get("Value") {
+                    return convert_json_value(val);
+                }
+            }
+            let mut map = ElementPropertyMap::new();
+            for (k, v) in obj {
+                map.insert(k, convert_json_value(v));
+            }
+            ElementValue::Object(map)
+        }
+    }
+}


components/sources/dataverse/tests/integration_test.rs

+        // Track how many times the delta endpoint is called to sequence responses
+        let delta_call_count = Arc::new(AtomicUsize::new(0));
+        let delta_count_clone = delta_call_count.clone();
+


components/sources/dataverse/src/lib.rs

+//! | `entities`             | Vec\<String\>             | required  | Entity logical names to monitor          |
+//! | `entity_set_overrides` | HashMap\<String, String\> | `{}`      | Override entity set name mapping         |
+//! | `entity_columns`       | HashMap\<String, Vec...\> | `{}`      | Per-entity column selection              |
+//! | `polling_interval_ms`  | u64                       | `5000`    | Base polling interval                    |


components/bootstrappers/dataverse/src/lib.rs

components/sources/dataverse/src/lib.rs

+        let shutdown_tx_clone = shutdown_tx.clone();
+        let source_id = self.base.id.clone();
+        let combined_handle = tokio::spawn(async move {
+            for (i, handle) in task_handles.into_iter().enumerate() {
+                if let Err(e) = handle.await {
+                    log::error!("[{source_id}] Entity worker {i} terminated with error: {e}");
+                }
+            }
+            log::info!("[{source_id}] All entity workers stopped");
+        });
+
+        *self.base.task_handle.write().await = Some(combined_handle);
+
+        // Store shutdown_tx in a custom slot
+        // We abuse the oneshot field by creating a bridge
+        let (oneshot_tx, _oneshot_rx) = tokio::sync::oneshot::channel::<()>();
+        // When oneshot_tx is dropped (in stop()), we trigger shutdown
+        // Actually, let's use a different approach: store shutdown_tx in context
+        // For now, store it directly - we need the watch::Sender
+        {
+            // We'll trigger shutdown via the watch channel
+            // Store the sender so stop() can use it
+            let mut lock = self.base.shutdown_tx.write().await;
+            // Create a bridge: when the oneshot is sent, signal the watch
+            let shutdown_tx_for_stop = shutdown_tx.clone();
+            let (bridge_tx, bridge_rx) = tokio::sync::oneshot::channel::<()>();
+            tokio::spawn(async move {
+                let _ = bridge_rx.await;
+                let _ = shutdown_tx_for_stop.send(true);
+            });
+            *lock = Some(bridge_tx);
+        }


components/sources/dataverse/src/lib.rs

+
+/// Dynamic plugin entry point.
+///
+/// Dynamic plugin entry point.


components/bootstrappers/dataverse/src/lib.rs

+///
+/// Dynamic plugin entry point.


components/sources/dataverse/src/descriptor.rs

+        let source = crate::DataverseSourceBuilder::new(id)
+            .with_environment_url(config.environment_url.clone())
+            .with_tenant_id(config.tenant_id.clone())
+            .with_client_id(config.client_id.clone())
+            .with_client_secret(config.client_secret.clone())
+            .with_entities(config.entities.clone())
+            .with_polling_interval_ms(config.polling_interval_ms)
+            .with_min_interval_ms(config.min_interval_ms)
+            .with_max_interval_seconds(config.max_interval_seconds)
+            .with_api_version(config.api_version.clone())
+            .with_auto_start(auto_start)
+            .build()?;


danielgerlag · 2026-03-13T15:25:13Z

components/bootstrappers/dataverse/src/config.rs

+    /// Azure AD client secret.
+    /// Required for client credentials flow, ignored when `use_azure_cli` is true.
+    #[serde(default, alias = "clientSecret")]
+    pub client_secret: String,


We should wrap these in ConfigValue so that they can be mapped to env variables or secrets.

These fields are already wrapped in ConfigValue in descriptor.rs, which is the platform/dynamic plugin entry point. DtoMapper resolves env vars and secrets into plain strings, which are then stored here in config.rs. This matches the pattern used by the postgres source and bootstrap components.

danielgerlag · 2026-03-13T15:27:11Z

components/sources/dataverse/src/config.rs

+/// - `min_interval_ms`: Minimum adaptive polling interval (default: 500)
+/// - `max_interval_seconds`: Maximum adaptive polling interval (default: 30)
+/// - `api_version`: Dataverse Web API version (default: `v9.2`)
+#[derive(Debug, Clone, Serialize, Deserialize)]


These are internal domain objects, when do they need to be serialized / deserialized?

Good catch. they dont need ot be serialized

danielgerlag · 2026-03-13T15:33:07Z

components/sources/dataverse/src/descriptor.rs

+            api_version,
+        };
+
+        let source = crate::DataverseSourceBuilder::new(id)


entity_set_overrides and entity_columns are being dropped here

Good catch. I have added them back

danielgerlag · 2026-03-13T15:34:01Z

components/sources/dataverse/src/descriptor.rs

+            entities: dto.entities,
+            entity_set_overrides: dto.entity_set_overrides,
+            entity_columns: dto.entity_columns,
+            polling_interval_ms,


polling interval is never actually used in the polling loop?

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

Copilot

Pull request overview

This PR adds a new Microsoft Dataverse source plugin and bootstrap provider for Drasi, enabling continuous monitoring of Dataverse tables via OData Web API change tracking (the REST equivalent of RetrieveEntityChangesRequest). It also extends AzureIdentityProvider with a with_client_secret constructor and updates the source-planner agent documentation with authentication guidance.

Changes:

Two new crates: drasi-source-dataverse (polling-based CDC with adaptive backoff and per-entity workers) and drasi-bootstrap-dataverse (initial data loader with pagination)
New AzureIdentityProvider::with_client_secret() method in lib/src/identity/azure.rs for service-to-service OAuth2 authentication
Updated workspace Cargo.toml and source-planner agent docs

Reviewed changes

Copilot reviewed 19 out of 19 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
`Cargo.toml`	Added both new crates to workspace members
`lib/src/identity/azure.rs`	Added `with_client_secret()` constructor and tests
`components/sources/dataverse/Cargo.toml`	New source crate definition
`components/sources/dataverse/src/config.rs`	Source configuration with validation
`components/sources/dataverse/src/types.rs`	OData delta response types and parsing
`components/sources/dataverse/src/auth.rs`	TokenManager with client credentials and Azure CLI auth
`components/sources/dataverse/src/client.rs`	HTTP client for OData change tracking API
`components/sources/dataverse/src/lib.rs`	Main source impl with per-entity polling workers
`components/sources/dataverse/src/descriptor.rs`	Dynamic plugin descriptor for the source
`components/sources/dataverse/tests/integration_test.rs`	Wiremock-based integration test (ignored)
`components/sources/dataverse/README.md`	Source documentation
`components/sources/dataverse/Makefile`	Build/test/lint targets
`components/bootstrappers/dataverse/Cargo.toml`	New bootstrap crate definition
`components/bootstrappers/dataverse/src/config.rs`	Bootstrap configuration with validation
`components/bootstrappers/dataverse/src/lib.rs`	Bootstrap provider with pagination and label filtering
`components/bootstrappers/dataverse/src/descriptor.rs`	Dynamic plugin descriptor for bootstrap
`components/bootstrappers/dataverse/README.md`	Bootstrap documentation
`components/bootstrappers/dataverse/Makefile`	Build/test/lint targets
`.github/agents/source-planner.md`	Auth guidance for source development

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

components/sources/dataverse/tests/integration_test.rs

+        let delta_call_count = Arc::new(AtomicUsize::new(0));
+        let delta_count_clone = delta_call_count.clone();
+
+        // 2. Mount OAuth2 token mock (always available)
+        Mock::given(method("POST"))
+            .and(path_regex(r".*/oauth2/v2.0/token"))
+            .respond_with(token_response())
+            .expect(1..)
+            .mount(&mock_server)
+            .await;
+
+        // 3. Mount initial change tracking request (GET /api/data/v9.2/accounts with Prefer header)
+        let mock_uri_for_initial = mock_uri.clone();
+        Mock::given(method("GET"))
+            .and(path("/api/data/v9.2/accounts"))
+            .and(header("Prefer", "odata.track-changes"))
+            .respond_with(empty_delta_response(&mock_uri_for_initial))
+            .expect(1)
+            .mount(&mock_server)
+            .await;
+
+        // 4. Mount delta link endpoint - returns sequenced responses
+        // We use a custom responder that returns different responses based on call count
+        let mock_uri_for_delta = mock_uri.clone();
+        let mock_uri_for_delta2 = mock_uri.clone();
+        let mock_uri_for_delta3 = mock_uri.clone();
+        let mock_uri_for_delta4 = mock_uri.clone();


components/sources/dataverse/tests/integration_test.rs

+        let mock_uri_for_initial = mock_uri.clone();
+        Mock::given(method("GET"))
+            .and(path("/api/data/v9.2/accounts"))
+            .and(header("Prefer", "odata.track-changes"))


components/sources/dataverse/src/lib.rs

+        let shutdown_tx_clone = shutdown_tx.clone();
+        let source_id = self.base.id.clone();
+        let combined_handle = tokio::spawn(async move {
+            for (i, handle) in task_handles.into_iter().enumerate() {
+                if let Err(e) = handle.await {
+                    log::error!("[{source_id}] Entity worker {i} terminated with error: {e}");
+                }
+            }
+            log::info!("[{source_id}] All entity workers stopped");
+        });
+
+        *self.base.task_handle.write().await = Some(combined_handle);
+
+        // Store shutdown_tx in a custom slot
+        // We abuse the oneshot field by creating a bridge
+        let (oneshot_tx, _oneshot_rx) = tokio::sync::oneshot::channel::<()>();
+        // When oneshot_tx is dropped (in stop()), we trigger shutdown
+        // Actually, let's use a different approach: store shutdown_tx in context
+        // For now, store it directly - we need the watch::Sender


components/bootstrappers/dataverse/Cargo.toml

@@ -0,0 +1,37 @@
+[package]


components/sources/dataverse/src/lib.rs

+
+/// Dynamic plugin entry point.
+///
+/// Dynamic plugin entry point.


components/bootstrappers/dataverse/src/lib.rs

+///
+/// Dynamic plugin entry point.


components/sources/dataverse/src/descriptor.rs

+        let source = crate::DataverseSourceBuilder::new(id)
+            .with_environment_url(config.environment_url.clone())
+            .with_tenant_id(config.tenant_id.clone())
+            .with_client_id(config.client_id.clone())
+            .with_client_secret(config.client_secret.clone())
+            .with_entities(config.entities.clone())
+            .with_min_interval_ms(config.min_interval_ms)
+            .with_max_interval_seconds(config.max_interval_seconds)
+            .with_api_version(config.api_version.clone())
+            .with_auto_start(auto_start);


components/sources/dataverse/Cargo.toml

@@ -0,0 +1,38 @@
+[package]


components/sources/dataverse/README.md

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

ruokun-niu added 2 commits February 27, 2026 14:22

agent work

f4189d3

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

dataverse

46e1a32

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

ruokun-niu requested a review from a team as a code owner March 3, 2026 22:46

ruokun-niu marked this pull request as draft March 3, 2026 22:46

github-advanced-security bot found potential problems Mar 3, 2026

View reviewed changes

ruokun-niu added 2 commits March 3, 2026 16:26

updated code to use identity provider

66ce250

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

updated source planner to reference identity provider

f263d07

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

ruokun-niu marked this pull request as ready for review March 4, 2026 00:29

ruokun-niu requested review from agentofreality, amansinghoriginal and danielgerlag March 4, 2026 00:29

ruokun-niu and others added 6 commits March 3, 2026 16:29

cargo fmt

88ecda8

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

Merge branch 'main' into dataverse-source

a238094

configure max_interval using square root scaling

87924b0

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

Merge branch 'main' into dataverse-source

8443baf

dataverse plugin test

ac4ef39

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

cargo fmt

07d4cd4

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

ruokun-niu self-assigned this Mar 11, 2026

danielgerlag requested a review from Copilot March 13, 2026 14:53

Copilot started reviewing on behalf of danielgerlag March 13, 2026 14:53 View session

Copilot AI reviewed Mar 13, 2026

View reviewed changes

danielgerlag reviewed Mar 13, 2026

View reviewed changes

ruokun-niu and others added 5 commits March 13, 2026 09:23

Potential fix for pull request finding

d4db266

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

removed serialization in config.rs

95a8ebb

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

removed polling-interval

2993757

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

added dropped variables

748f855

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

cargo fmt

b5234af

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

ruokun-niu requested a review from Copilot March 13, 2026 17:54

Copilot started reviewing on behalf of ruokun-niu March 13, 2026 17:55 View session

Merge branch 'main' into dataverse-source

234732d

Copilot AI reviewed Mar 13, 2026

View reviewed changes

ruokun-niu and others added 5 commits March 13, 2026 11:22

Potential fix for pull request finding

540489d

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

bug fix

9f62c46

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

nit fixes

0a6aaa5

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

copilot fix

0c5a60f

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

Merge branch 'main' into dataverse-source

1e08a97

ruokun-niu requested a review from danielgerlag March 17, 2026 16:48

cargo fmt

79d1866

Signed-off-by: ruokun-niu <ruokunniu@gmail.com>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Add Microsoft Dataverse source and bootstrap provider#304

feat: Add Microsoft Dataverse source and bootstrap provider#304
ruokun-niu wants to merge 22 commits intodrasi-project:mainfrom
ruokun-niu:dataverse-source

ruokun-niu commented Mar 3, 2026

Uh oh!

Check notice

Check notice

Copilot AI left a comment

Uh oh!

Uh oh!

danielgerlag Mar 13, 2026

Uh oh!

ruokun-niu Mar 13, 2026

Uh oh!

danielgerlag Mar 13, 2026

Uh oh!

ruokun-niu Mar 13, 2026

Uh oh!

danielgerlag Mar 13, 2026

Uh oh!

ruokun-niu Mar 13, 2026

Uh oh!

danielgerlag Mar 13, 2026

Uh oh!

ruokun-niu Mar 13, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

ruokun-niu commented Mar 3, 2026

Description

New Crates

Key Features

Change Tracking

Adaptive Backoff

Authentication

Platform YAML Compatibility

Configuration Example

Files Added/Modified

New Files

Modified Files

Testing

Prerequisites for Dataverse Environment

Type of change

Uh oh!

Check notice

Check notice

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants