From e2f1c490f0d4d021c0e7b0a65a0e9c17edc1904b Mon Sep 17 00:00:00 2001 From: "Ismael B." Date: Mon, 27 Apr 2026 11:15:15 +0200 Subject: [PATCH 1/2] [feat] chore: expose response headers in DefaultHttp DefaultHttp#request now returns {status:, body:, headers:} (previously {status:, body:}). Backward-compatible: existing adapters (HubSpot, Airtable) ignore the new key. Required by the upcoming Dynamics365Adapter to read OData-EntityId after upsert/create. --- lib/etlify/adapters/default_http.rb | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/etlify/adapters/default_http.rb b/lib/etlify/adapters/default_http.rb index d907400..aedaaf8 100644 --- a/lib/etlify/adapters/default_http.rb +++ b/lib/etlify/adapters/default_http.rb @@ -6,7 +6,7 @@ module Etlify module Adapters # Simple Net::HTTP client used by default (dependency-free). # Shared across adapters (Airtable, HubSpot, etc.). - # Signature: request(method, url, headers:, body:) → {status:, body:} + # Signature: request(method, url, headers:, body:) → {status:, body:, headers:} class DefaultHttp OPEN_TIMEOUT = 5 READ_TIMEOUT = 30 @@ -30,7 +30,11 @@ def request(method, url, headers: {}, body: nil) http_request.body = body if body response = http.request(http_request) - {status: response.code.to_i, body: response.body} + { + status: response.code.to_i, + body: response.body, + headers: response.to_hash, + } end end end From 1602112ea846de21d56bd12d0efec0d407bf6784 Mon Sep 17 00:00:00 2001 From: "Ismael B." Date: Mon, 27 Apr 2026 11:19:03 +0200 Subject: [PATCH 2/2] [feat] feat: add Dynamics365Adapter for Microsoft Dynamics 365 / Dataverse Web API v9.2 Adapter built on the standard Etlify contract: - upsert! supports PATCH by GUID, PATCH by Dataverse alternate key (with OData escaping), and POST create. Returns the GUID parsed from the OData-EntityId response header. - delete! is idempotent (404 -> false). - batch_upsert! / batch_delete! raise NotImplementedError; native $batch (multipart/mixed) support is planned for a follow-up. Auth uses OAuth 2.0 client_credentials with an in-memory bearer token cache (Mutex + Hash, 60s safety margin) and a transparent re-fetch on a single 401 retry. Architecture follows the Airtable two-layer pattern (Adapter facade + Client for HTTP/auth/error mapping). Zero external dependency (Net::HTTP + JSON stdlib). Specs: 75 examples, 100% line coverage on adapter, client and token cache. Includes thread-safety test on the cache. --- CHANGELOG.md | 3 + README.md | 62 ++ lib/etlify/adapters.rb | 1 + lib/etlify/adapters/dynamics_365/client.rb | 323 ++++++++++ .../adapters/dynamics_365/token_cache.rb | 61 ++ lib/etlify/adapters/dynamics_365_adapter.rb | 153 +++++ .../install/templates/initializer.rb.tt | 21 + spec/adapters/dynamics_365/client_spec.rb | 416 +++++++++++++ .../adapters/dynamics_365/token_cache_spec.rb | 96 +++ spec/adapters/dynamics_365_adapter_spec.rb | 554 ++++++++++++++++++ 10 files changed, 1690 insertions(+) create mode 100644 lib/etlify/adapters/dynamics_365/client.rb create mode 100644 lib/etlify/adapters/dynamics_365/token_cache.rb create mode 100644 lib/etlify/adapters/dynamics_365_adapter.rb create mode 100644 spec/adapters/dynamics_365/client_spec.rb create mode 100644 spec/adapters/dynamics_365/token_cache_spec.rb create mode 100644 spec/adapters/dynamics_365_adapter_spec.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index 9af7092..0f9a015 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,8 @@ # UNRELEASED +- Feat: Add `Dynamics365Adapter` for Microsoft Dynamics 365 / Dataverse Web API v9.2. Supports OAuth 2.0 `client_credentials` with an in-memory bearer token cache (auto-refresh on expiration and on 401 retry), single-record `upsert!` (PATCH by GUID, PATCH by Dataverse alternate key, or POST create depending on inputs) and `delete!` (DELETE by GUID with idempotent 404). Batch operations (`batch_upsert!` / `batch_delete!`) raise `NotImplementedError` in this version — use `Etlify::SyncJob` per record. Native `$batch` multipart support is planned for a follow-up release. Zero external dependency (Net::HTTP + JSON stdlib). +- Feat: `Etlify::Adapters::DefaultHttp#request` now returns `{status:, body:, headers:}` (previously `{status:, body:}`). Backward-compatible: existing adapters ignore the new key. Required by `Dynamics365Adapter` to read the `OData-EntityId` response header that carries the Dataverse GUID after upsert. + # V0.11.0 - Feat: Add `enabled:` flag to `Etlify::CRM.register` (default `true`). When a CRM is registered with `enabled: false`, all sync and delete calls become a no-op: `Model#crm_sync!` and `Model#crm_delete!` return `true` without enqueuing any job, `Etlify::Synchronizer.call` and `Etlify::Deleter.call` return `:disabled`, `Etlify::BatchSynchronizer.call` returns stats with `disabled: true`, and `Etlify::StaleRecords::BatchSync.call` silently skips disabled CRMs while still processing enabled ones. No adapter call, no write to `crm_synchronisations`. Useful to keep Etlify dormant in development or test environments. New public helper `Etlify::CRM.enabled?(name)` (returns `true` for unknown CRMs as a safe default). diff --git a/README.md b/README.md index 2175185..04ac6f1 100644 --- a/README.md +++ b/README.md @@ -629,6 +629,67 @@ adapter.batch_delete!( --- +## Microsoft Dynamics 365 adapter (Web API v9.2) + +Etlify ships with `Etlify::Adapters::Dynamics365Adapter` for Microsoft Dynamics 365 / Dataverse. It uses `Net::HTTP` (no external dependency) and authenticates via OAuth 2.0 `client_credentials`. The bearer token is cached in-memory per process, refreshed before expiration, and re-fetched transparently on a 401 response (one retry per request). + +### Configuration + +```ruby +Etlify.configure do |config| + Etlify::CRM.register( + :dynamics_365, + adapter: Etlify::Adapters::Dynamics365Adapter.new( + tenant_id: ENV["DYNAMICS_TENANT_ID"], + client_id: ENV["DYNAMICS_CLIENT_ID"], + client_secret: ENV["DYNAMICS_CLIENT_SECRET"], + resource_uri: ENV["DYNAMICS_RESOURCE_URI"] + ), + enabled: Rails.env.production? || Rails.env.staging?, + options: { + job_class: Etlify::SyncJob, + rate_limit: { max_requests: 100, period: 5 }, + } + ) +end +``` + +`resource_uri` is the Dataverse environment URL (e.g. `https://contoso.crm.dynamics.com`). `api_version` defaults to `"9.2"` and can be overridden. + +### Behaviour + +- `object_type`: the Dataverse entity set name (plural, lowercase — e.g. `"contacts"`, `"accounts"`, `"leads"`). +- `id_property`: the **alternate key** field name configured on the Dataverse entity (e.g. `"emailaddress1"` for contacts). Must be declared as an alternate key in Dataverse, otherwise upsert returns a 400. +- `crm_id`: if provided, the adapter PATCHes the record directly by GUID and skips alternate-key resolution. +- The Dataverse GUID is read back from the `OData-EntityId` response header after every upsert/create. + +### Example: Contact upsert + +```ruby +class User < ApplicationRecord + include Etlify::Model + + dynamics_365_etlified_with( + serializer: DynamicsContactSerializer, + crm_object_type: "contacts", + id_property: "emailaddress1", + sync_if: ->(user) { user.email.present? } + ) +end + +# Later +user.dynamics_365_sync! +``` + +### Limitations + +- **No batch support in v1.** `batch_upsert!` and `batch_delete!` raise `NotImplementedError`. `Etlify::BatchSyncJob` is therefore not usable with `:dynamics_365` — register the CRM with `options: { job_class: Etlify::SyncJob }` (per-record sync). Native Dataverse `$batch` (multipart/mixed changeset) support is planned for a follow-up release. +- **Alternate keys must exist in Dataverse.** The property passed as `id_property` has to be configured as an alternate key on the target entity beforehand. +- **Dataverse Service Protection limits.** Dataverse enforces ~6000 requests / 300s / application user. The recommended `rate_limit` above (`100/5s`) keeps a comfortable margin. +- **Token cache is per process.** Each Sidekiq worker maintains its own bearer cache (~1 OAuth round-trip per worker per hour, negligible). + +--- + ## Writing your own adapter Implement the following interface: @@ -737,6 +798,7 @@ expect(fake_adapter).to have_received(:upsert!).with( - `Etlify::Adapters::NullAdapter` (default; no-op) - `Etlify::Adapters::HubspotV3Adapter` (API v3, with batch support) - `Etlify::Adapters::AirtableV0Adapter` (API v0, with batch support) +- `Etlify::Adapters::Dynamics365Adapter` (Dataverse Web API v9.2, OAuth client_credentials, single-record only) --- diff --git a/lib/etlify/adapters.rb b/lib/etlify/adapters.rb index 14b0dca..234af6a 100644 --- a/lib/etlify/adapters.rb +++ b/lib/etlify/adapters.rb @@ -1,3 +1,4 @@ require_relative "adapters/null_adapter" require_relative "adapters/hubspot_v3_adapter" require_relative "adapters/airtable_v0_adapter" +require_relative "adapters/dynamics_365_adapter" diff --git a/lib/etlify/adapters/dynamics_365/client.rb b/lib/etlify/adapters/dynamics_365/client.rb new file mode 100644 index 0000000..9a4894a --- /dev/null +++ b/lib/etlify/adapters/dynamics_365/client.rb @@ -0,0 +1,323 @@ +require "json" +require "uri" + +module Etlify + module Adapters + module Dynamics365 + # Low-level HTTP client for the Microsoft Dynamics 365 + # Dataverse Web API. Handles OAuth 2.0 (client_credentials), + # OData URL building, request execution, JSON parsing, and + # error mapping. Wraps an injectable HTTP transport so the + # adapter remains free of network access in tests. + class Client + LOGIN_BASE = "https://login.microsoftonline.com" + DEFAULT_API_VERSION = "9.2" + + attr_accessor :rate_limiter + + def initialize( + tenant_id:, + client_id:, + client_secret:, + resource_uri:, + api_version:, + http:, + token_cache: + ) + @tenant_id = tenant_id + @client_id = client_id + @client_secret = client_secret + @resource_uri = resource_uri.sub(%r{/+\z}, "") + @api_version = api_version + @http = http + @token_cache = token_cache + end + + # --- Public API used by the adapter --- + + def patch_by_guid(entity_set, guid, payload) + path = guid_path(entity_set, guid) + response = request(:patch, path, body: payload) + raise_for_error!(response, path: path) + response + end + + # @return [Hash] response with :crm_id extracted from OData-EntityId + def patch_by_alternate_key(entity_set, key_name, key_value, payload) + path = alternate_key_path(entity_set, key_name, key_value) + response = request(:patch, path, body: payload) + raise_for_error!(response, path: path) + response[:crm_id] = extract_crm_id_from_entity_id(response, path) + response + end + + # @return [Hash] response with :crm_id extracted from OData-EntityId + def post_create(entity_set, payload) + path = entity_set_path(entity_set) + response = request(:post, path, body: payload) + raise_for_error!(response, path: path) + response[:crm_id] = extract_crm_id_from_entity_id(response, path) + response + end + + def delete_by_guid(entity_set, guid) + path = guid_path(entity_set, guid) + response = request(:delete, path) + return response if response[:status] == 404 + + raise_for_error!(response, path: path) + response + end + + def raise_for_error!(response, path:) + status = response[:status].to_i + return if status.between?(200, 299) + + error_detail = extract_error_detail(response) + message = error_detail["message"] || + "Dynamics 365 API request failed" + code = error_detail["code"] + correlation_id = extract_correlation_id(response) + + full_message = "#{message} (status=#{status}, path=#{path}" + full_message << ", code=#{code}" if code + full_message << ")" + + raise error_class_for(status).new( + full_message, + status: status, + code: code, + category: code, + correlation_id: correlation_id, + details: error_detail, + raw: response[:body] + ) + end + + private + + # --- URL building --- + + def entity_set_path(entity_set) + "/api/data/v#{@api_version}/#{encode_segment(entity_set)}" + end + + def guid_path(entity_set, guid) + "#{entity_set_path(entity_set)}(#{encode_segment(guid)})" + end + + def alternate_key_path(entity_set, key_name, key_value) + escaped = escape_odata_string(key_value) + "#{entity_set_path(entity_set)}(#{key_name}='#{escaped}')" + end + + def encode_segment(segment) + URI.encode_www_form_component(segment.to_s).gsub("+", "%20") + end + + def escape_odata_string(value) + value.to_s.gsub("'", "''") + end + + # --- HTTP execution --- + + def request(method, path, body: nil, query: {}, retried_on_401: false) + @rate_limiter&.throttle! + + url = build_url(path, query) + token = @token_cache.fetch { fetch_oauth_token } + headers = base_headers(token) + raw_body = body && JSON.dump(body) + + response = perform_http(method, url, headers, raw_body) + response[:json] = parse_json_safe(response[:body]) + + if response[:status] == 401 && !retried_on_401 + @token_cache.invalidate! + return request( + method, + path, + body: body, + query: query, + retried_on_401: true + ) + end + + response + end + + def perform_http(method, url, headers, raw_body) + @http.request(method, url, headers: headers, body: raw_body) + rescue => exception + raise Etlify::TransportError.new( + "HTTP transport error: #{exception.class}: #{exception.message}", + status: 0, + raw: nil + ) + end + + def build_url(path, query) + url = "#{@resource_uri}#{path}" + return url if query.nil? || query.empty? + + "#{url}?#{URI.encode_www_form(query)}" + end + + def base_headers(token) + { + "Authorization" => "Bearer #{token}", + "Accept" => "application/json", + "Content-Type" => "application/json; charset=utf-8", + "OData-MaxVersion" => "4.0", + "OData-Version" => "4.0", + "If-None-Match" => "null", + } + end + + # --- OAuth --- + + def fetch_oauth_token + url = "#{LOGIN_BASE}/#{@tenant_id}/oauth2/v2.0/token" + headers = { + "Content-Type" => "application/x-www-form-urlencoded", + "Accept" => "application/json", + } + body = URI.encode_www_form( + grant_type: "client_credentials", + client_id: @client_id, + client_secret: @client_secret, + scope: "#{@resource_uri}/.default" + ) + + response = perform_http(:post, url, headers, body) + response[:json] = parse_json_safe(response[:body]) + status = response[:status].to_i + + unless status.between?(200, 299) + raise_oauth_error!(response, url, status) + end + + parsed_token_entry(response) + end + + def parsed_token_entry(response) + data = response[:json] || {} + access_token = data["access_token"] + expires_in = data["expires_in"] + + unless access_token.is_a?(String) && !access_token.empty? + raise Etlify::ApiError.new( + "OAuth response did not include an access_token", + status: response[:status].to_i, + raw: response[:body] + ) + end + + ttl = expires_in.is_a?(Numeric) ? expires_in.to_i : 0 + {token: access_token, expires_at: Time.now + ttl} + end + + def raise_oauth_error!(response, url, status) + data = response[:json].is_a?(Hash) ? response[:json] : {} + error_code = data["error"] + error_description = data["error_description"] + full_message = [ + "OAuth token request failed", + "(status=#{status}, url=#{url}, error=#{error_code})", + ].join(" ") + full_message = "#{full_message}: #{error_description}" if error_description + + klass = + case status + when 400, 401, 403 then Etlify::Unauthorized + when 429 then Etlify::RateLimited + else Etlify::ApiError + end + + raise klass.new( + full_message, + status: status, + code: error_code, + category: error_code, + details: data, + raw: response[:body] + ) + end + + # --- Response parsing --- + + def parse_json_safe(raw_body) + return nil if raw_body.nil? || raw_body.empty? + + JSON.parse(raw_body) + rescue JSON::ParserError + nil + end + + def extract_error_detail(response) + body = response[:json].is_a?(Hash) ? response[:json] : {} + detail = body["error"] + detail.is_a?(Hash) ? detail : {} + end + + def extract_correlation_id(response) + headers = response[:headers] + return nil unless headers.is_a?(Hash) + + values = headers["x-ms-service-request-id"] || + headers["X-Ms-Service-Request-Id"] + Array(values).first + end + + def extract_crm_id_from_entity_id(response, path) + entity_id = entity_id_header(response) + + if entity_id.nil? || entity_id.empty? + raise Etlify::ApiError.new( + [ + "Dynamics response did not include OData-EntityId header", + "(path=#{path})", + ].join(" "), + status: response[:status].to_i, + raw: response[:body] + ) + end + + match = entity_id.match(/\(([^)]+)\)\z/) + unless match + raise Etlify::ApiError.new( + [ + "Could not extract crm_id from OData-EntityId header", + "(value=#{entity_id}, path=#{path})", + ].join(" "), + status: response[:status].to_i, + raw: response[:body] + ) + end + + match[1] + end + + def entity_id_header(response) + headers = response[:headers] + return nil unless headers.is_a?(Hash) + + values = headers["odata-entityid"] || + headers["OData-EntityId"] || + headers["Odata-Entityid"] + Array(values).first + end + + def error_class_for(status) + case status + when 401, 403 then Etlify::Unauthorized + when 404 then Etlify::NotFound + when 409, 422 then Etlify::ValidationFailed + when 429 then Etlify::RateLimited + else Etlify::ApiError + end + end + end + end + end +end diff --git a/lib/etlify/adapters/dynamics_365/token_cache.rb b/lib/etlify/adapters/dynamics_365/token_cache.rb new file mode 100644 index 0000000..4ddd66f --- /dev/null +++ b/lib/etlify/adapters/dynamics_365/token_cache.rb @@ -0,0 +1,61 @@ +module Etlify + module Adapters + module Dynamics365 + # Thread-safe in-memory cache for the Dataverse OAuth + # bearer token. The cached entry expires + # SAFETY_MARGIN_SECONDS before its declared TTL to + # avoid race conditions with the IdP clock. + class TokenCache + SAFETY_MARGIN_SECONDS = 60 + + # @param clock [#call] returns the current Time + def initialize(clock: Time.method(:now)) + @clock = clock + @mutex = Mutex.new + @entry = nil + end + + # @yieldreturn [Hash{Symbol => Object}] {token: String, expires_at: Time} + # @return [String] the cached or freshly-fetched token + def fetch + @mutex.synchronize do + return @entry[:token] if cached_entry_valid? + + fresh = yield + validate_entry!(fresh) + @entry = fresh + @entry[:token] + end + end + + def invalidate! + @mutex.synchronize { @entry = nil } + end + + private + + def cached_entry_valid? + return false unless @entry.is_a?(Hash) + return false unless @entry[:token].is_a?(String) + return false unless @entry[:expires_at].is_a?(Time) + + @entry[:expires_at] > @clock.call + SAFETY_MARGIN_SECONDS + end + + def validate_entry!(entry) + return if valid_entry?(entry) + + raise ArgumentError, + "TokenCache block must return {token: String, expires_at: Time}" + end + + def valid_entry?(entry) + entry.is_a?(Hash) && + entry[:token].is_a?(String) && + !entry[:token].empty? && + entry[:expires_at].is_a?(Time) + end + end + end + end +end diff --git a/lib/etlify/adapters/dynamics_365_adapter.rb b/lib/etlify/adapters/dynamics_365_adapter.rb new file mode 100644 index 0000000..520a70b --- /dev/null +++ b/lib/etlify/adapters/dynamics_365_adapter.rb @@ -0,0 +1,153 @@ +require_relative "default_http" +require_relative "dynamics_365/client" +require_relative "dynamics_365/token_cache" + +module Etlify + module Adapters + # Microsoft Dynamics 365 (Dataverse Web API v9.2) Adapter. + # + # Authentication is OAuth 2.0 client_credentials. The bearer + # token is cached in-memory per process and refreshed + # transparently before expiration (and on a 401 retry). + # + # Usage: + # adapter = Etlify::Adapters::Dynamics365Adapter.new( + # tenant_id: ENV["DYNAMICS_TENANT_ID"], + # client_id: ENV["DYNAMICS_CLIENT_ID"], + # client_secret: ENV["DYNAMICS_CLIENT_SECRET"], + # resource_uri: ENV["DYNAMICS_RESOURCE_URI"], + # ) + # adapter.upsert!( + # object_type: "contacts", + # payload: {firstname: "John", emailaddress1: "j@example.com"}, + # id_property: "emailaddress1", + # ) + # adapter.delete!(object_type: "contacts", crm_id: guid) + # + # Limitations: + # - batch_upsert! and batch_delete! raise NotImplementedError. + # Native Dataverse $batch (multipart/mixed) support is + # planned in a follow-up. Use Etlify::SyncJob (per-record) + # rather than Etlify::BatchSyncJob with this adapter for now. + # - id_property must be configured as a Dataverse alternate key + # on the target entity, otherwise upsert! will return a 400. + class Dynamics365Adapter + DEFAULT_API_VERSION = "9.2" + RESOURCE_URI_REGEX = %r{\Ahttps://}.freeze + API_VERSION_REGEX = /\A\d+\.\d+\z/.freeze + + def rate_limiter + @client.rate_limiter + end + + def rate_limiter=(limiter) + @client.rate_limiter = limiter + end + + def initialize( + tenant_id:, + client_id:, + client_secret:, + resource_uri:, + api_version: DEFAULT_API_VERSION, + http_client: nil, + token_cache: nil + ) + validate_string!(:tenant_id, tenant_id) + validate_string!(:client_id, client_id) + validate_string!(:client_secret, client_secret) + validate_string!(:resource_uri, resource_uri) + validate_string!(:api_version, api_version) + validate_resource_uri!(resource_uri) + validate_api_version!(api_version) + + @client = Dynamics365::Client.new( + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: resource_uri, + api_version: api_version, + http: http_client || DefaultHttp.new, + token_cache: token_cache || Dynamics365::TokenCache.new + ) + end + + # @return [String] Dataverse GUID of the upserted record + def upsert!(object_type:, payload:, id_property: nil, crm_id: nil) + validate_string!(:object_type, object_type) + raise ArgumentError, "payload must be a Hash" unless payload.is_a?(Hash) + + normalized_crm_id = crm_id.to_s.strip + unless normalized_crm_id.empty? + @client.patch_by_guid(object_type, normalized_crm_id, payload) + return normalized_crm_id + end + + if id_property + key_value = lookup_payload_value(payload, id_property) + unless key_value.nil? || key_value.to_s.empty? + response = @client.patch_by_alternate_key( + object_type, id_property.to_s, key_value, payload + ) + return response[:crm_id] + end + end + + response = @client.post_create(object_type, payload) + response[:crm_id] + end + + def delete!(object_type:, crm_id:) + validate_string!(:object_type, object_type) + validate_string!(:crm_id, crm_id) + + response = @client.delete_by_guid(object_type, crm_id) + return false if response[:status] == 404 + + true + end + + def batch_upsert!(object_type:, records:, id_property:) + raise NotImplementedError, batch_not_supported_message + end + + def batch_delete!(object_type:, crm_ids:) + raise NotImplementedError, batch_not_supported_message + end + + private + + def lookup_payload_value(payload, id_property) + key_str = id_property.to_s + payload[key_str] || payload[key_str.to_sym] + end + + def batch_not_supported_message + [ + "Batch operations are not yet supported by", + "Dynamics365Adapter. Use upsert!/delete! per record,", + "or wait for the multipart $batch implementation.", + ].join(" ") + end + + def validate_string!(name, value) + return if value.is_a?(String) && !value.empty? + + raise ArgumentError, "#{name} must be a non-empty String" + end + + def validate_resource_uri!(value) + return if value.match?(RESOURCE_URI_REGEX) + + raise ArgumentError, "resource_uri must start with https://" + end + + def validate_api_version!(value) + return if value.match?(API_VERSION_REGEX) + + raise ArgumentError, + "api_version must match \\A\\d+\\.\\d+\\z (e.g. \"9.2\")" + end + end + end +end diff --git a/lib/generators/etlify/install/templates/initializer.rb.tt b/lib/generators/etlify/install/templates/initializer.rb.tt index c3af0b0..33ac642 100644 --- a/lib/generators/etlify/install/templates/initializer.rb.tt +++ b/lib/generators/etlify/install/templates/initializer.rb.tt @@ -21,6 +21,27 @@ Etlify.configure do |config| # will provide DSL below for models # another_crm_etlified_with(...) + # Microsoft Dynamics 365 (Dataverse Web API v9.2) + # Note: Dynamics365Adapter does NOT support batch operations + # in v1 (batch_upsert! / batch_delete! raise NotImplementedError). + # Use Etlify::SyncJob (per-record) rather than BatchSyncJob. + # Etlify::CRM.register( + # :dynamics_365, + # adapter: Etlify::Adapters::Dynamics365Adapter.new( + # tenant_id: ENV["DYNAMICS_TENANT_ID"], + # client_id: ENV["DYNAMICS_CLIENT_ID"], + # client_secret: ENV["DYNAMICS_CLIENT_SECRET"], + # resource_uri: ENV["DYNAMICS_RESOURCE_URI"], + # ), + # enabled: Rails.env.production? || Rails.env.staging?, + # options: { + # job_class: Etlify::SyncJob, + # rate_limit: { max_requests: 100, period: 5 }, + # } + # ) + # will provide DSL below for models + # dynamics_365_etlified_with(...) + # overridable defaults config # @digest_strategy = Etlify::Digest.method(:stable_sha256) # @job_queue_name = "low" diff --git a/spec/adapters/dynamics_365/client_spec.rb b/spec/adapters/dynamics_365/client_spec.rb new file mode 100644 index 0000000..99ef73f --- /dev/null +++ b/spec/adapters/dynamics_365/client_spec.rb @@ -0,0 +1,416 @@ +require "rails_helper" +require "etlify/adapters/dynamics_365/client" +require "etlify/adapters/dynamics_365/token_cache" + +RSpec.describe Etlify::Adapters::Dynamics365::Client do + let(:tenant_id) { "tenant-uuid" } + let(:client_id) { "client-uuid" } + let(:client_secret) { "secret-value" } + let(:resource_uri) { "https://contoso.crm.dynamics.com" } + let(:api_version) { "9.2" } + let(:http) { instance_double("HttpClient") } + let(:token_cache) { instance_double("TokenCache") } + + subject(:client) do + described_class.new( + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: resource_uri, + api_version: api_version, + http: http, + token_cache: token_cache + ) + end + + let(:api_base) { "https://contoso.crm.dynamics.com/api/data/v9.2" } + + before do + allow(token_cache).to receive(:fetch).and_return("fake-token") + allow(token_cache).to receive(:invalidate!) + end + + def stub_http(status:, body: "", headers: {}) + {status: status, body: body, headers: headers} + end + + def entity_url(entity_set, key) + "#{api_base}/#{entity_set}(#{key})" + end + + def entity_id_header(entity_set, guid) + {"odata-entityid" => [entity_url(entity_set, guid)]} + end + + describe "URL building and required headers" do + it "PATCHes by GUID with all OData headers" do + expect(http).to receive(:request).with( + :patch, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts(abc-123)", + headers: hash_including( + "Authorization" => "Bearer fake-token", + "Accept" => "application/json", + "Content-Type" => "application/json; charset=utf-8", + "OData-MaxVersion" => "4.0", + "OData-Version" => "4.0", + "If-None-Match" => "null" + ), + body: '{"firstname":"John"}' + ).and_return(stub_http(status: 204)) + + response = client.patch_by_guid( + "contacts", "abc-123", {firstname: "John"} + ) + expect(response[:status]).to eq(204) + end + + it "PATCHes by alternate key with OData escaping for single quotes" do + expect(http).to receive(:request).with( + :patch, + entity_url("contacts", "emailaddress1='o''brian@example.com'"), + headers: hash_including("Authorization" => "Bearer fake-token"), + body: anything + ).and_return( + stub_http( + status: 204, + headers: entity_id_header( + "contacts", "11111111-2222-3333-4444-555555555555" + ) + ) + ) + + response = client.patch_by_alternate_key( + "contacts", "emailaddress1", "o'brian@example.com", + {firstname: "John"} + ) + expect(response[:crm_id]).to eq("11111111-2222-3333-4444-555555555555") + end + + it "POSTs to the entity set on create" do + expect(http).to receive(:request).with( + :post, + "#{api_base}/contacts", + headers: hash_including("Authorization" => "Bearer fake-token"), + body: '{"firstname":"John"}' + ).and_return( + stub_http( + status: 204, + headers: { + "OData-EntityId" => [ + entity_url("contacts", "99999999-9999-9999-9999-999999999999"), + ], + } + ) + ) + + response = client.post_create("contacts", {firstname: "John"}) + expect(response[:crm_id]).to eq("99999999-9999-9999-9999-999999999999") + end + + it "DELETEs by GUID" do + expect(http).to receive(:request).with( + :delete, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts(abc-123)", + headers: hash_including("Authorization" => "Bearer fake-token"), + body: nil + ).and_return(stub_http(status: 204)) + + response = client.delete_by_guid("contacts", "abc-123") + expect(response[:status]).to eq(204) + end + + it "returns the 404 response without raising on delete" do + allow(http).to receive(:request).and_return(stub_http(status: 404)) + + response = client.delete_by_guid("contacts", "missing") + expect(response[:status]).to eq(404) + end + + it "trims trailing slashes on resource_uri" do + trimmed = described_class.new( + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: "https://contoso.crm.dynamics.com//", + api_version: api_version, + http: http, + token_cache: token_cache + ) + + expect(http).to receive(:request).with( + :patch, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts(abc)", + anything + ).and_return(stub_http(status: 204)) + + trimmed.patch_by_guid("contacts", "abc", {}) + end + end + + describe "401 retry logic" do + it "invalidates the cache and retries exactly once on a single 401" do + call_count = 0 + allow(http).to receive(:request) do + call_count += 1 + if call_count == 1 + stub_http( + status: 401, + body: '{"error":{"code":"0x80048306","message":"Token expired"}}' + ) + else + stub_http(status: 204) + end + end + + response = client.patch_by_guid("contacts", "abc", {}) + + expect(response[:status]).to eq(204) + expect(call_count).to eq(2) + expect(token_cache).to have_received(:invalidate!).once + end + + it "raises Unauthorized when 401 persists after one retry" do + allow(http).to receive(:request).and_return( + stub_http( + status: 401, + body: '{"error":{"code":"0x80048306","message":"Bad token"}}' + ) + ) + + expect do + client.patch_by_guid("contacts", "abc", {}) + end.to raise_error(Etlify::Unauthorized, /status=401/) + end + end + + describe "OData-EntityId extraction" do + it "raises ApiError when the header is missing after upsert" do + allow(http).to receive(:request).and_return(stub_http(status: 204)) + + expect do + client.patch_by_alternate_key( + "contacts", "emailaddress1", "x@y.z", {} + ) + end.to raise_error(Etlify::ApiError, /OData-EntityId/) + end + + it "raises ApiError when the header is malformed" do + allow(http).to receive(:request).and_return( + stub_http( + status: 204, + headers: {"odata-entityid" => ["not-a-valid-entity-url"]} + ) + ) + + expect do + client.post_create("contacts", {}) + end.to raise_error(Etlify::ApiError, /Could not extract crm_id/) + end + end + + describe "#raise_for_error!" do + let(:dataverse_error) do + JSON.dump( + error: { + code: "0x80040217", + message: "An entity with the same key already exists.", + } + ) + end + + { + 401 => Etlify::Unauthorized, + 403 => Etlify::Unauthorized, + 404 => Etlify::NotFound, + 409 => Etlify::ValidationFailed, + 422 => Etlify::ValidationFailed, + 429 => Etlify::RateLimited, + 500 => Etlify::ApiError, + 503 => Etlify::ApiError, + }.each do |status, klass| + it "maps HTTP #{status} to #{klass}" do + error = capture_error do + client.raise_for_error!( + { + status: status, + body: dataverse_error, + json: JSON.parse(dataverse_error), + headers: {}, + }, + path: "/whatever" + ) + end + + expect(error).to be_a(klass) + expect(error.status).to eq(status) + expect(error.code).to eq("0x80040217") if status != 404 + end + end + + it "extracts correlation_id from x-ms-service-request-id header" do + error = capture_error do + client.raise_for_error!( + { + status: 422, + body: dataverse_error, + json: JSON.parse(dataverse_error), + headers: {"x-ms-service-request-id" => ["req-correlation-uuid"]}, + }, + path: "/x" + ) + end + + expect(error).to be_a(Etlify::ValidationFailed) + expect(error.correlation_id).to eq("req-correlation-uuid") + end + + def capture_error + yield + nil + rescue Etlify::Error => caught + caught + end + end + + describe "transport errors" do + it "wraps StandardError into TransportError" do + allow(http).to receive(:request).and_raise( + StandardError, "ECONNRESET" + ) + + expect do + client.patch_by_guid("contacts", "abc", {}) + end.to raise_error(Etlify::TransportError, /ECONNRESET/) + end + end + + describe "OAuth token fetching" do + let(:cache) { Etlify::Adapters::Dynamics365::TokenCache.new } + subject(:real_client) do + described_class.new( + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: resource_uri, + api_version: api_version, + http: http, + token_cache: cache + ) + end + + it "POSTs to the IdP with form-encoded credentials and parses the token" do + expect(http).to receive(:request).with( + :post, + "https://login.microsoftonline.com/#{tenant_id}/oauth2/v2.0/token", + headers: hash_including( + "Content-Type" => "application/x-www-form-urlencoded" + ), + body: satisfy do |body| + params = URI.decode_www_form(body).to_h + params["grant_type"] == "client_credentials" && + params["client_id"] == client_id && + params["client_secret"] == client_secret && + params["scope"] == "#{resource_uri}/.default" + end + ).and_return( + stub_http( + status: 200, + body: JSON.dump( + access_token: "real-bearer-xyz", + expires_in: 3599, + token_type: "Bearer" + ) + ) + ) + + expect(http).to receive(:request).with( + :patch, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts(abc)", + headers: hash_including("Authorization" => "Bearer real-bearer-xyz"), + body: anything + ).and_return(stub_http(status: 204)) + + real_client.patch_by_guid("contacts", "abc", {}) + end + + it "raises Unauthorized on invalid_client error" do + allow(http).to receive(:request).and_return( + stub_http( + status: 400, + body: JSON.dump( + error: "invalid_client", + error_description: "AADSTS7000215: Invalid client secret" + ) + ) + ) + + expect do + real_client.patch_by_guid("contacts", "abc", {}) + end.to raise_error(Etlify::Unauthorized, /invalid_client/) + end + + it "raises ApiError when the token endpoint returns 500" do + allow(http).to receive(:request).and_return( + stub_http(status: 500, body: "") + ) + + expect do + real_client.patch_by_guid("contacts", "abc", {}) + end.to raise_error(Etlify::ApiError, /OAuth token request failed/) + end + + it "raises ApiError when the token endpoint returns no access_token" do + allow(http).to receive(:request).and_return( + stub_http(status: 200, body: JSON.dump(token_type: "Bearer")) + ) + + expect do + real_client.patch_by_guid("contacts", "abc", {}) + end.to raise_error(Etlify::ApiError, /access_token/) + end + + it "raises RateLimited on a 429 from the token endpoint" do + allow(http).to receive(:request).and_return( + stub_http( + status: 429, + body: JSON.dump(error: "throttled") + ) + ) + + expect do + real_client.patch_by_guid("contacts", "abc", {}) + end.to raise_error(Etlify::RateLimited) + end + end + + describe "JSON parsing edge cases" do + it "returns a usable response when the body is not valid JSON" do + allow(http).to receive(:request).and_return( + stub_http(status: 204, body: "not json", + headers: entity_id_header("contacts", "abc-123")) + ) + + response = client.patch_by_alternate_key( + "contacts", "emailaddress1", "x@y.z", {} + ) + expect(response[:json]).to be_nil + expect(response[:crm_id]).to eq("abc-123") + end + end + + describe "request URL building with query strings" do + it "appends query parameters when provided" do + expect(http).to receive(:request).with( + :patch, + "#{entity_url('contacts', 'abc')}?%24select=fullname", + headers: anything, + body: anything + ).and_return(stub_http(status: 204)) + + client.send(:request, :patch, + "/api/data/v9.2/contacts(abc)", + body: {}, + query: {"$select" => "fullname"}) + end + end +end diff --git a/spec/adapters/dynamics_365/token_cache_spec.rb b/spec/adapters/dynamics_365/token_cache_spec.rb new file mode 100644 index 0000000..2c8daeb --- /dev/null +++ b/spec/adapters/dynamics_365/token_cache_spec.rb @@ -0,0 +1,96 @@ +require "rails_helper" +require "etlify/adapters/dynamics_365/token_cache" + +RSpec.describe Etlify::Adapters::Dynamics365::TokenCache do + let(:now) { Time.utc(2026, 4, 27, 12, 0, 0) } + let(:clock) { -> { now } } + + subject(:cache) { described_class.new(clock: clock) } + + describe "#fetch" do + it "calls the block on first call" do + calls = 0 + token = cache.fetch do + calls += 1 + {token: "abc", expires_at: now + 3600} + end + + expect(token).to eq("abc") + expect(calls).to eq(1) + end + + it "does not call the block when the entry is still fresh" do + cache.fetch { {token: "abc", expires_at: now + 3600} } + + calls = 0 + token = cache.fetch do + calls += 1 + raise "should not be called" + end + + expect(token).to eq("abc") + expect(calls).to eq(0) + end + + it "calls the block again when the entry is past its safety margin" do + cache.fetch { {token: "old", expires_at: now + 30} } + + token = cache.fetch { {token: "new", expires_at: now + 3600} } + + expect(token).to eq("new") + end + + it "calls the block again when the entry is exactly at the safety margin" do + margin = described_class::SAFETY_MARGIN_SECONDS + cache.fetch { {token: "old", expires_at: now + margin} } + + token = cache.fetch { {token: "new", expires_at: now + 3600} } + + expect(token).to eq("new") + end + + it "raises if the block returns an invalid entry" do + expect do + cache.fetch { {token: nil, expires_at: now + 3600} } + end.to raise_error(ArgumentError, /TokenCache block/) + end + + it "raises if the block returns a non-Hash" do + expect do + cache.fetch { "not a hash" } + end.to raise_error(ArgumentError, /TokenCache block/) + end + end + + describe "#invalidate!" do + it "forces the next fetch to call the block again" do + cache.fetch { {token: "abc", expires_at: now + 3600} } + cache.invalidate! + + token = cache.fetch { {token: "def", expires_at: now + 3600} } + + expect(token).to eq("def") + end + end + + describe "thread safety" do + it "calls the block exactly once when many threads hit a cold cache" do + mutex = Mutex.new + call_count = 0 + + block = proc do + mutex.synchronize { call_count += 1 } + sleep(0.01) + {token: "shared", expires_at: now + 3600} + end + + threads = Array.new(20) do + Thread.new { cache.fetch(&block) } + end + results = threads.map(&:value) + + expect(results.uniq).to eq(["shared"]) + expect(call_count).to eq(1) + end + end +end diff --git a/spec/adapters/dynamics_365_adapter_spec.rb b/spec/adapters/dynamics_365_adapter_spec.rb new file mode 100644 index 0000000..29a39ad --- /dev/null +++ b/spec/adapters/dynamics_365_adapter_spec.rb @@ -0,0 +1,554 @@ +require "rails_helper" +require "etlify/adapters/dynamics_365_adapter" + +RSpec.describe Etlify::Adapters::Dynamics365Adapter do + let(:tenant_id) { "tenant-uuid" } + let(:client_id) { "client-uuid" } + let(:client_secret) { "secret-value" } + let(:resource_uri) { "https://contoso.crm.dynamics.com" } + let(:http) { instance_double("HttpClient") } + let(:token_cache) { instance_double("TokenCache") } + + subject(:adapter) do + described_class.new( + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: resource_uri, + http_client: http, + token_cache: token_cache + ) + end + + before do + allow(token_cache).to receive(:fetch).and_return("fake-token") + allow(token_cache).to receive(:invalidate!) + end + + let(:api_base) { "https://contoso.crm.dynamics.com/api/data/v9.2" } + + def stub_http(status:, body: "", headers: {}) + {status: status, body: body, headers: headers} + end + + def entity_url(entity_set, key) + "#{api_base}/#{entity_set}(#{key})" + end + + def entity_id_header(entity_set, guid) + {"odata-entityid" => [entity_url(entity_set, guid)]} + end + + describe "#initialize" do + { + tenant_id: "tenant_id", + client_id: "client_id", + client_secret: "client_secret", + resource_uri: "resource_uri", + }.each do |arg, name| + it "raises on blank #{name}" do + params = { + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: resource_uri, + } + params[arg] = "" + + expect { described_class.new(**params) }.to raise_error( + ArgumentError, /#{Regexp.escape(name)}/ + ) + end + end + + it "raises when resource_uri is not https" do + expect do + described_class.new( + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: "http://contoso.crm.dynamics.com" + ) + end.to raise_error(ArgumentError, /https/) + end + + it "raises when api_version is malformed" do + expect do + described_class.new( + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: resource_uri, + api_version: "v9" + ) + end.to raise_error(ArgumentError, /api_version/) + end + + it "raises when api_version is blank" do + expect do + described_class.new( + tenant_id: tenant_id, + client_id: client_id, + client_secret: client_secret, + resource_uri: resource_uri, + api_version: "" + ) + end.to raise_error(ArgumentError, /api_version/) + end + end + + describe "#upsert!" do + context "with crm_id provided" do + it "PATCHes by GUID and returns the crm_id verbatim" do + expect(http).to receive(:request).with( + :patch, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts(abc-123)", + headers: hash_including("Authorization" => "Bearer fake-token"), + body: anything + ).and_return(stub_http(status: 204)) + + result = adapter.upsert!( + object_type: "contacts", + payload: {firstname: "John"}, + crm_id: "abc-123" + ) + expect(result).to eq("abc-123") + end + + it "ignores id_property when crm_id is also given" do + allow(http).to receive(:request).and_return(stub_http(status: 204)) + + result = adapter.upsert!( + object_type: "contacts", + payload: {emailaddress1: "x@y.z"}, + id_property: "emailaddress1", + crm_id: "abc-123" + ) + expect(result).to eq("abc-123") + end + + it "trims whitespace from crm_id" do + expect(http).to receive(:request).with( + :patch, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts(abc-123)", + anything + ).and_return(stub_http(status: 204)) + + result = adapter.upsert!( + object_type: "contacts", + payload: {}, + crm_id: " abc-123 " + ) + expect(result).to eq("abc-123") + end + + it "falls back to POST create when crm_id is whitespace-only" do + guid = "55555555-5555-5555-5555-555555555555" + expect(http).to receive(:request).with( + :post, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts", + anything + ).and_return( + stub_http(status: 204, headers: entity_id_header("contacts", guid)) + ) + + result = adapter.upsert!( + object_type: "contacts", + payload: {firstname: "John"}, + crm_id: " " + ) + expect(result).to eq(guid) + end + + it "raises NotFound when PATCH by GUID returns 404" do + allow(http).to receive(:request).and_return( + stub_http( + status: 404, + body: JSON.dump( + error: { + code: "0x80040217", + message: "Record with id GUID does not exist", + } + ) + ) + ) + + expect do + adapter.upsert!( + object_type: "contacts", + payload: {firstname: "John"}, + crm_id: "missing-guid" + ) + end.to raise_error(Etlify::NotFound) + end + end + + context "with id_property and a value present in the payload" do + let(:guid) { "11111111-2222-3333-4444-555555555555" } + + it "PATCHes by alternate key and returns the GUID from OData-EntityId" do + expect(http).to receive(:request).with( + :patch, + entity_url("contacts", "emailaddress1='john@example.com'"), + headers: hash_including("Authorization" => "Bearer fake-token"), + body: satisfy do |body| + JSON.parse(body)["firstname"] == "John" + end + ).and_return( + stub_http(status: 204, headers: entity_id_header("contacts", guid)) + ) + + result = adapter.upsert!( + object_type: "contacts", + payload: {firstname: "John", emailaddress1: "john@example.com"}, + id_property: "emailaddress1" + ) + expect(result).to eq(guid) + end + + it "accepts a Symbol id_property" do + expect(http).to receive(:request).with( + :patch, + entity_url("contacts", "emailaddress1='john@example.com'"), + anything + ).and_return( + stub_http(status: 204, headers: entity_id_header("contacts", guid)) + ) + + result = adapter.upsert!( + object_type: "contacts", + payload: {emailaddress1: "john@example.com"}, + id_property: :emailaddress1 + ) + expect(result).to eq(guid) + end + + it "OData-escapes single quotes in the alternate key value" do + expect(http).to receive(:request).with( + :patch, + entity_url("contacts", "emailaddress1='o''brian@example.com'"), + anything + ).and_return( + stub_http(status: 204, headers: entity_id_header("contacts", guid)) + ) + + adapter.upsert!( + object_type: "contacts", + payload: {emailaddress1: "o'brian@example.com"}, + id_property: "emailaddress1" + ) + end + end + + context "with id_property but value missing from payload" do + let(:guid) { "99999999-9999-9999-9999-999999999999" } + + it "falls back to POST create when the key is absent" do + expect(http).to receive(:request).with( + :post, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts", + anything + ).and_return( + stub_http(status: 204, headers: entity_id_header("contacts", guid)) + ) + + result = adapter.upsert!( + object_type: "contacts", + payload: {firstname: "John"}, + id_property: "emailaddress1" + ) + expect(result).to eq(guid) + end + + it "falls back to POST create when the key value is an empty string" do + expect(http).to receive(:request).with( + :post, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts", + anything + ).and_return( + stub_http(status: 204, headers: entity_id_header("contacts", guid)) + ) + + result = adapter.upsert!( + object_type: "contacts", + payload: {firstname: "John", emailaddress1: ""}, + id_property: "emailaddress1" + ) + expect(result).to eq(guid) + end + + it "falls back to POST create when the key value is nil" do + expect(http).to receive(:request).with( + :post, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts", + anything + ).and_return( + stub_http(status: 204, headers: entity_id_header("contacts", guid)) + ) + + result = adapter.upsert!( + object_type: "contacts", + payload: {firstname: "John", emailaddress1: nil}, + id_property: "emailaddress1" + ) + expect(result).to eq(guid) + end + end + + context "without crm_id and without id_property" do + let(:guid) { "99999999-9999-9999-9999-999999999999" } + + it "POSTs create and returns the GUID from OData-EntityId" do + expect(http).to receive(:request).with( + :post, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts", + anything + ).and_return( + stub_http(status: 204, headers: entity_id_header("contacts", guid)) + ) + + result = adapter.upsert!( + object_type: "contacts", + payload: {firstname: "John"} + ) + expect(result).to eq(guid) + end + end + + describe "validations" do + it "raises when object_type is blank" do + expect do + adapter.upsert!(object_type: "", payload: {}) + end.to raise_error(ArgumentError, /object_type/) + end + + it "raises when payload is not a Hash" do + expect do + adapter.upsert!(object_type: "contacts", payload: "nope") + end.to raise_error(ArgumentError, /payload/) + end + end + + describe "error mapping" do + it "succeeds after a single 401 by transparently re-fetching the token" do + call_count = 0 + allow(http).to receive(:request) do + call_count += 1 + if call_count == 1 + stub_http(status: 401, body: "{}") + else + stub_http( + status: 204, + headers: entity_id_header("contacts", "abc") + ) + end + end + + result = adapter.upsert!( + object_type: "contacts", + payload: {emailaddress1: "x@y.z"}, + id_property: "emailaddress1" + ) + expect(result).to eq("abc") + end + + it "raises Unauthorized when 401 persists after retry" do + allow(http).to receive(:request).and_return( + stub_http(status: 401, body: "{}") + ) + + expect do + adapter.upsert!( + object_type: "contacts", + payload: {emailaddress1: "x@y.z"}, + id_property: "emailaddress1" + ) + end.to raise_error(Etlify::Unauthorized) + end + + it "raises ValidationFailed on 422" do + allow(http).to receive(:request).and_return( + stub_http( + status: 422, + body: JSON.dump( + error: { + code: "0x80040217", + message: "Duplicate record detected", + } + ) + ) + ) + + expect do + adapter.upsert!( + object_type: "contacts", + payload: {emailaddress1: "x@y.z"}, + id_property: "emailaddress1" + ) + end.to raise_error(Etlify::ValidationFailed) + end + + it "raises RateLimited on 429" do + allow(http).to receive(:request).and_return( + stub_http(status: 429, body: "{}") + ) + + expect do + adapter.upsert!( + object_type: "contacts", + payload: {emailaddress1: "x@y.z"}, + id_property: "emailaddress1" + ) + end.to raise_error(Etlify::RateLimited) + end + + it "raises ApiError when OData-EntityId is missing after upsert" do + allow(http).to receive(:request).and_return(stub_http(status: 204)) + + expect do + adapter.upsert!( + object_type: "contacts", + payload: {emailaddress1: "x@y.z"}, + id_property: "emailaddress1" + ) + end.to raise_error(Etlify::ApiError, /OData-EntityId/) + end + + it "wraps transport errors into TransportError" do + allow(http).to receive(:request).and_raise( + StandardError, "ECONNRESET" + ) + + expect do + adapter.upsert!( + object_type: "contacts", + payload: {}, + crm_id: "abc" + ) + end.to raise_error(Etlify::TransportError) + end + end + end + + describe "#delete!" do + it "returns true on 2xx" do + expect(http).to receive(:request).with( + :delete, + "https://contoso.crm.dynamics.com/api/data/v9.2/contacts(abc-123)", + headers: hash_including("Authorization" => "Bearer fake-token"), + body: nil + ).and_return(stub_http(status: 204)) + + expect( + adapter.delete!(object_type: "contacts", crm_id: "abc-123") + ).to eq(true) + end + + it "returns false on 404 (idempotent)" do + allow(http).to receive(:request).and_return(stub_http(status: 404)) + + expect( + adapter.delete!(object_type: "contacts", crm_id: "missing") + ).to eq(false) + end + + it "raises on 401 after retry" do + allow(http).to receive(:request).and_return(stub_http(status: 401)) + + expect do + adapter.delete!(object_type: "contacts", crm_id: "abc") + end.to raise_error(Etlify::Unauthorized) + end + + it "raises ApiError on 500" do + allow(http).to receive(:request).and_return(stub_http(status: 500)) + + expect do + adapter.delete!(object_type: "contacts", crm_id: "abc") + end.to raise_error(Etlify::ApiError) + end + + it "raises ValidationFailed on 422" do + allow(http).to receive(:request).and_return( + stub_http( + status: 422, + body: JSON.dump( + error: { + code: "0x80048408", + message: "Cannot delete record because of FK constraint", + } + ) + ) + ) + + expect do + adapter.delete!(object_type: "contacts", crm_id: "abc") + end.to raise_error(Etlify::ValidationFailed) + end + + it "raises RateLimited on 429" do + allow(http).to receive(:request).and_return(stub_http(status: 429)) + + expect do + adapter.delete!(object_type: "contacts", crm_id: "abc") + end.to raise_error(Etlify::RateLimited) + end + + it "wraps transport errors into TransportError" do + allow(http).to receive(:request).and_raise( + StandardError, "ECONNRESET" + ) + + expect do + adapter.delete!(object_type: "contacts", crm_id: "abc") + end.to raise_error(Etlify::TransportError) + end + + it "raises on blank object_type" do + expect do + adapter.delete!(object_type: "", crm_id: "abc") + end.to raise_error(ArgumentError, /object_type/) + end + + it "raises on blank crm_id" do + expect do + adapter.delete!(object_type: "contacts", crm_id: "") + end.to raise_error(ArgumentError, /crm_id/) + end + end + + describe "#batch_upsert!" do + it "raises NotImplementedError with an explanatory message" do + expect do + adapter.batch_upsert!( + object_type: "contacts", + records: [{}], + id_property: "emailaddress1" + ) + end.to raise_error(NotImplementedError, /Batch operations/) + end + end + + describe "#batch_delete!" do + it "raises NotImplementedError with an explanatory message" do + expect do + adapter.batch_delete!( + object_type: "contacts", + crm_ids: ["abc"] + ) + end.to raise_error(NotImplementedError, /Batch operations/) + end + end + + describe "rate_limiter delegation" do + it "exposes a rate_limiter accessor that delegates to the client" do + limiter = double("limiter") + adapter.rate_limiter = limiter + expect(adapter.rate_limiter).to be(limiter) + end + + it "returns nil when no rate_limiter has been assigned" do + expect(adapter.rate_limiter).to be_nil + end + end +end