SAP Integration Design for Ordermatic

Status: Design Document (Ready for PR Review) Created: 2026-03-09 Target: Production implementation by end of Q1 2026

Executive Summary

This document defines the architecture for integrating SAP systems (Business One, S/4HANA Cloud, S/4HANA On-Prem, ECC) into Ordermatic using our existing Dagster → Iceberg DWH → Typesense data pipeline and TypeScript IERPService layer for real-time operations.

Key Design Principles:

✅ Three separate implementations (not a monolithic adapter)
✅ Follows existing Prophet21 patterns exactly
✅ Dagster orchestrates continuous syncs (hourly/daily)
✅ TypeScript IERPService handles real-time actions
✅ Iceberg DWH with partitioned layout (connection_id isolation)
✅ Typesense for search/matching (products, customers, cross-refs, addresses)
✅ Connection-based multi-tenancy (connection_id for isolation)

Architecture Overview

Data Flow

┌──────────────────────────────────────────────────────────────┐
│                      WEBAPP (Astro/React)                    │
│  - Order extraction from PDFs/Emails/Excel/Images            │
│  - Matching via Typesense (customer/product/cross-ref)       │
│  - Real-time order creation via IERPService                  │
└────────────────┬─────────────────────────────────────────────┘
                 │
        ┌────────┴─────────────────────────┐
        │                                   │
        v                                   v
┌──────────────────┐           ┌───────────────────────┐
│ TypeScript       │           │    Dagster Pipeline   │
│ IERPService      │           │  (Scheduled + Sensor) │
│  (Real-time)     │           │   (Continuous Sync)   │
│                  │           │                       │
│ - searchCustomer │           │  Bronze Layer:        │
│ - searchItems    │           │   ├─ fetch_sap_*      │
│ - getAddresses   │           │   │  (DLT sources)    │
│ - createOrder    │           │   │                   │
│ - submitPayment  │           │  Silver/Gold:         │
└────────┬─────────┘           │   ├─ dbt transforms  │
         │                     │   │ (normalize data)  │
         │                     │   │                   │
         │                     │  Typesense Layer:     │
         │                     │   ├─ sync assets     │
         │                     │   │ (products,etc)   │
         └─────────────────────┴────────────┬──────────┘
                                            │
                    ┌───────────────────────┴────────────────┐
                    │                                        │
                    v                                        v
            ┌──────────────┐                       ┌─────────────────┐
            │  SAP Systems │                       │  Iceberg DWH    │
            │              │◄──────────────────────│  (R2 Partitioned)
            │ • Business   │    OData APIs         │                 │
            │   One        │    (with auth)        │  Tables:        │
            │ • S/4HANA    │                       │  - products     │
            │   Cloud      │                       │  - customers    │
            │ • S/4HANA    │                       │  - xref         │
            │   On-Prem    │                       │  - addresses    │
            │ • ECC        │                       │                 │
            └──────────────┘                       └─────────────────┘
                                                           │
                                                           v
                                                  ┌──────────────────┐
                                                  │   Typesense      │
                                                  │   (Full-text)    │
                                                  │                  │
                                                  │  Collections:    │
                                                  │  - products      │
                                                  │  - customers     │
                                                  │  - xref          │
                                                  │  - addresses     │
                                                  └──────────────────┘

1. SAP Implementations (3 Separate Classes)

Each SAP system has a dedicated Dagster Orchestrator and DLT Source.

1.1 Business One Sync Orchestrator

File: apps/dagster/erp_pipeline/implementations/sap_b1_sync_orchestrator.py

class SAPBusinessOneSyncOrchestratorV1:
    """
    SAP Business One sync orchestrator.

    Features:
    - Version-aware B1 REST API routing
    - Credential management via Infisical
    - DLT pipeline for data extraction
    """

    def __init__(self, connection_id: str, connection_config: dict[str, Any] | None = None):
        self.connection_id = connection_id
        self.dataset_name = "erp_data_native_v2"
        self._r2_client: Any | None = None

        # Load from database if not provided
        if connection_config is None:
            connection_config = self._load_connection_config()

        self.connection_config = connection_config

        # B1 specific config
        extra_config = connection_config.get("extra_config") or {}
        self.sap_url = connection_config.get("api_url")  # e.g., https://sap-b1.example.com:50000
        self.use_https = extra_config.get("use_https", True)
        self.sap_version = extra_config.get("sap_version")  # e.g., "9.3"

    def _load_connection_config(self) -> dict[str, Any]:
        """Load connection config from database."""
        from pdf_shared.crud import get_db_connection

        with get_db_connection() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT config FROM erp_connections WHERE id = %s AND erp_type = 'sap_b1'",
                    (self.connection_id,)
                )
                row = cur.fetchone()
                if not row:
                    raise ValueError(f"SAP Business One connection {self.connection_id} not found")
                return row[0]

    def sync_products(self, full_sync: bool = False) -> dict[str, Any]:
        """Sync products from SAP Business One."""
        logger.info(f"Syncing SAP B1 products for {self.connection_id}, full_sync={full_sync}")

        # Use DLT source to fetch and load
        dlt_result = dlt.run(
            sap_b1_source(
                connection_id=self.connection_id,
                connection_config=self.connection_config,
                entity="products",
                full_sync=full_sync,
            ),
            table_name="products",
            dataset_name=self.dataset_name,
            destination="iceberg",
        )

        return {"records_loaded": dlt_result.load_info.records_written}

    # Similar methods for customers, addresses, cross-references

1.2 S/4HANA Cloud Sync Orchestrator

File: apps/dagster/erp_pipeline/implementations/sap_s4_cloud_sync_orchestrator.py

class SAPS4HANACloudSyncOrchestratorV1:
    """
    SAP S/4HANA Cloud sync orchestrator.

    Features:
    - OAuth2 authentication
    - Cloud-specific OData v4 endpoints
    - Credential caching via extra_config
    """

    def __init__(self, connection_id: str, connection_config: dict[str, Any] | None = None):
        self.connection_id = connection_id
        self.dataset_name = "erp_data_native_v2"

        if connection_config is None:
            connection_config = self._load_connection_config()

        self.connection_config = connection_config

        # S/4HANA Cloud specific
        extra_config = connection_config.get("extra_config") or {}
        self.auth_type = extra_config.get("auth_type", "oauth2")  # oauth2 or basic
        self.sap_client = extra_config.get("sap_client")  # e.g., "100"
        self.odata_version = extra_config.get("odata_version", "v4")  # v4 for cloud

    def sync_customers(self, full_sync: bool = False) -> dict[str, Any]:
        """Sync customers from SAP S/4HANA Cloud."""
        logger.info(f"Syncing S/4HANA Cloud customers for {self.connection_id}")

        dlt_result = dlt.run(
            sap_s4_cloud_source(
                connection_id=self.connection_id,
                connection_config=self.connection_config,
                entity="customers",
                full_sync=full_sync,
            ),
            table_name="customers",
            dataset_name=self.dataset_name,
            destination="iceberg",
        )

        return {"records_loaded": dlt_result.load_info.records_written}

1.3 S/4HANA On-Prem Sync Orchestrator

File: apps/dagster/erp_pipeline/implementations/sap_s4_onprem_sync_orchestrator.py

class SAPS4HANAOnPremSyncOrchestratorV1:
    """
    SAP S/4HANA On-Premises sync orchestrator.

    Features:
    - Basic auth or Kerberos
    - On-prem-specific connectivity (direct IP, VPN)
    - RFC gateway support (optional)
    """

    def __init__(self, connection_id: str, connection_config: dict[str, Any] | None = None):
        self.connection_id = connection_id
        self.dataset_name = "erp_data_native_v2"

        if connection_config is None:
            connection_config = self._load_connection_config()

        self.connection_config = connection_config
        extra_config = connection_config.get("extra_config") or {}

        self.auth_type = extra_config.get("auth_type", "basic")
        self.sap_client = extra_config.get("sap_client")
        self.use_rfc = extra_config.get("use_rfc", False)

2. DLT Sources (Data Extraction Layer)

2.1 SAP Business One DLT Source

File: apps/dagster/erp_pipeline/sources/sap_b1/dlt_source.py

@dlt.source
def sap_b1_source(
    connection_id: str,
    connection_config: dict[str, Any],
    entity: str = "all",  # products, customers, addresses, cross_references
    full_sync: bool = False,
) -> List[dlt.Resource]:
    """
    DLT source for SAP Business One OData API.

    Supports:
    - Products (Item Master)
    - Customers (Business Partner)
    - Addresses (Address Master)
    - Cross-references (Customer-specific item aliases)
    """

    client = SAPB1Client(connection_config)

    resources = []

    if entity in ("products", "all"):
        resources.append(
            dlt.resource(
                fetch_sap_b1_products(client, connection_id, full_sync),
                name="sap_b1_products",
                write_disposition="merge",
                primary_key="erp_id",
            )
        )

    if entity in ("customers", "all"):
        resources.append(
            dlt.resource(
                fetch_sap_b1_customers(client, connection_id, full_sync),
                name="sap_b1_customers",
                write_disposition="merge",
                primary_key="erp_id",
            )
        )

    # Similar for addresses, cross-references

    return resources

async def fetch_sap_b1_products(
    client: SAPB1Client,
    connection_id: str,
    full_sync: bool,
) -> Generator[dict, None, None]:
    """Fetch products from SAP Business One."""

    # Build OData query
    query = "$filter=IsActive eq 'Y'"
    if not full_sync:
        last_sync = get_last_sync_date("sap_b1_products", connection_id)
        if last_sync:
            query += f" and UpdateDate gt '{last_sync}'"

    # Fetch from B1 REST API
    response = await client.get(
        "/b1s/v1/Items",
        params={"$filter": query, "$top": 1000}
    )

    for item in response.get("value", []):
        yield {
            "connection_id": connection_id,
            "erp_id": item["ItemCode"],
            "name": item["ItemName"],
            "description": item["ItemDescription"],
            "upc": item.get("Barcode"),
            "category": item.get("InventoryItem", {}).get("InventoryClassCode"),
            # ... more fields
            "_sap_raw": item,
        }

2.2 Similar sources for S/4HANA Cloud & On-Prem

apps/dagster/erp_pipeline/sources/sap_s4_cloud/dlt_source.py
apps/dagster/erp_pipeline/sources/sap_s4_onprem/dlt_source.py

3. Dagster Assets (Bronze → Silver → Gold → Typesense)

3.1 Bronze Assets (Raw DLT Data)

File: apps/dagster/erp_pipeline/assets/bronze.py (extend existing)

@asset(
    partitions_def=erp_connection_partitions,
    group_name="bronze",
    tags={"kind": "erp_sync"},
    config_schema={
        "full_sync": Field(bool, default_value=False),
    }
)
def sap_products(context: AssetExecutionContext) -> None:
    """Extract SAP products (Business One, S/4 Cloud, S/4 On-Prem)."""

    connection_id = get_connection_id_from_context(context)
    erp_type = get_erp_type_for_connection(connection_id)

    if erp_type not in ("sap_b1", "sap_s4_cloud", "sap_s4_onprem"):
        return  # Not a SAP connection

    # Get sync parameters
    full_sync, last_sync_date = get_sync_params_safe(
        context, "sap_products", connection_id
    )

    # Get appropriate orchestrator
    orchestrator = ERPSourceFactory.create_orchestrator(connection_id)

    # Execute sync
    result = orchestrator.sync_products(full_sync=full_sync)

    context.log.info(f"SAP products synced: {result}")

    yield MaterializeResult(
        metadata={
            "records_loaded": result.get("records_loaded", 0),
            "full_sync": full_sync,
            "connection_id": connection_id,
            "erp_type": erp_type,
        }
    )

# Similar assets: sap_customers, sap_addresses, sap_cross_references

3.2 Silver/Gold Assets (dbt Transforms)

File: apps/dagster/erp_pipeline/dbt/models/silver/stg_sap_products.sql

{{ config(
    materialized = 'iceberg',
    partitioned_by = ['connection_id'],
    sort_by = 'connection_id',
    tags = ['sap', 'products']
) }}

SELECT
    connection_id,
    erp_id,
    name,
    description,
    upc,
    category,
    -- Normalize pricing fields (SAP has various price lists)
    price1,
    price2,
    -- ... more prices
    unit_of_measure,
    weight,
    -- Metadata
    _sap_raw,
    _dlt_load_id,
    _dlt_timestamp
FROM {{ source('bronze', 'sap_b1_products') }}
WHERE connection_id = dbt_config.get('connection_id')

UNION ALL

SELECT
    connection_id,
    erp_id,
    name,
    description,
    upc,
    category,
    price1,
    -- ... (S/4 schema may differ)
FROM {{ source('bronze', 'sap_s4_cloud_products') }}
WHERE connection_id = dbt_config.get('connection_id')

UNION ALL

-- S/4 On-Prem products

3.3 Typesense Sync Assets

File: apps/dagster/erp_pipeline/assets/typesense.py (extend existing)

@asset(
    partitions_def=erp_connection_partitions,
    group_name="typesense",
    deps=[sap_gold_products],
)
def sap_products_typesense(context: AssetExecutionContext) -> None:
    """Sync SAP products to Typesense for search."""

    connection_id = get_connection_id_from_context(context)

    # Read from gold layer
    records = get_gold_parquet_records("dim_sap_products", connection_id)

    # Transform for Typesense
    typesense_docs = [
        {
            "id": f"{connection_id}_{rec['erp_id']}",
            "connection_id": connection_id,
            "erp_product_id": rec["erp_id"],
            "name": rec["name"],
            "description": rec["description"],
            "upc": rec["upc"],
            "category_code": rec["category"],
            "ocr_variants": [],  # Populated during extraction
            "format_variants": [],
            "search_text": f"{rec['name']} {rec['description']} {rec['upc']}",
            "price1": rec.get("price1"),
            # ... price2-price10, list_price
            "unit_of_measure": rec.get("unit_of_measure"),
            "weight": rec.get("weight"),
            "has_product_image": False,
            "delete_flag": False,
        }
        for rec in records
    ]

    # Upsert to Typesense
    typesense_resource = TypesenseResource.get()
    typesense_resource.upsert_collection(
        collection_name="products",
        documents=typesense_docs,
    )

    yield MaterializeResult(
        metadata={
            "documents_upserted": len(typesense_docs),
            "connection_id": connection_id,
        }
    )

4. TypeScript IERPService Implementations

4.1 Base SAP Service Class

File: apps/webapp/src/services/erp/implementations/sap-service-base.ts

import type {
  IERPService,
  ErpCustomerSearchParams,
  ErpCustomer,
} from '@repo/integrations/services';
import { logger } from '@/utils/logger';
import { ERPAuthService } from '@/services/erp-auth';
import httpx from 'httpx';

/**
 * Abstract base class for SAP ERP services.
 * Implements common auth, OData query building, and error handling.
 */
export abstract class SAPServiceBase implements IERPService {
  protected readonly connection: ERPConnection;
  protected readonly apiUrl: string;
  protected authService: ERPAuthService;
  protected cachedAccessToken: string | null = null;

  constructor(connection: ERPConnection) {
    this.connection = connection;
    this.apiUrl = connection.apiUrl;
    this.authService = new ERPAuthService(connection);
  }

  /**
   * Get access token with caching (valid for 1 hour typically).
   */
  protected async getAccessToken(): Promise<string> {
    if (this.cachedAccessToken) {
      return this.cachedAccessToken;
    }

    const token = await this.authService.getAccessToken();
    this.cachedAccessToken = token;

    // Reset cache after 50 minutes (SAP tokens valid 1 hour typically)
    setTimeout(
      () => {
        this.cachedAccessToken = null;
      },
      50 * 60 * 1000
    );

    return token;
  }

  /**
   * Build OData filter for customer search.
   * Override in subclasses for system-specific field names.
   */
  protected abstract buildCustomerFilterExpression(params: ErpCustomerSearchParams): string;

  /**
   * Search customers in SAP.
   */
  async searchCustomers(params: ErpCustomerSearchParams): Promise<ErpCustomer[]> {
    logger.info('SAP: Searching customers', { params });

    const token = await this.getAccessToken();
    const filter = this.buildCustomerFilterExpression(params);

    const response = await httpx.get(
      `${this.apiUrl}/odata/v4/c_customertp?$filter=${encodeURIComponent(filter)}&$top=10`,
      {
        headers: {
          Authorization: `Bearer ${token}`,
          'Content-Type': 'application/json',
        },
        timeout: 30.0,
      }
    );

    const results = response.json();

    return (results.value || []).map((customer: any) => ({
      id: customer.customer_id,
      name: customer.customer_name,
      email: customer.email,
      customerId: customer.customer_id,
      companyId: this.connection.companyId,
    }));
  }

  /**
   * Get customer addresses.
   */
  async getAddresses(
    customerId: string,
    addressType?: 'shipping' | 'billing' | 'both'
  ): Promise<ErpAddress[]> {
    const token = await this.getAccessToken();

    const response = await httpx.get(
      `${this.apiUrl}/odata/v4/c_customertpaddresstp?$filter=customer_id eq '${customerId}'&$top=100`,
      {
        headers: {
          Authorization: `Bearer ${token}`,
        },
        timeout: 30.0,
      }
    );

    return (response.json().value || []).map((addr: any) => ({
      id: addr.address_id,
      customerId,
      name: addr.address_description,
      addressLine1: addr.street,
      addressLine2: addr.street2,
      city: addr.city,
      state: addr.state,
      country: addr.country,
      postalCode: addr.postal_code,
      types: [addressType || 'shipping'],
    }));
  }

  /**
   * Create order in SAP.
   */
  async createOrder(input: ErpOrderInput): Promise<ErpOrderResult> {
    logger.info('SAP: Creating order', { customerId: input.customerId });

    const token = await this.getAccessToken();

    const orderPayload = {
      customer_id: input.customerId,
      order_date: new Date().toISOString(),
      currency: input.currency || 'USD',
      items: input.items.map(item => ({
        item_id: item.itemId,
        quantity: item.quantity,
        unit_price: item.unitPrice,
      })),
    };

    const response = await httpx.post(`${this.apiUrl}/odata/v4/c_salesordertp`, {
      headers: {
        Authorization: `Bearer ${token}`,
        'Content-Type': 'application/json',
      },
      json: orderPayload,
      timeout: 60.0,
    });

    const result = response.json();

    return {
      success: true,
      orderId: result.order_id,
      orderNumber: result.order_number,
      status: 'created',
    };
  }

  // Abstract methods - implement in subclasses
  abstract validateConnection(): Promise<boolean>;
  abstract searchItems(params: ErpItemSearchParams): Promise<ErpItem[]>;
  abstract getERPCapabilities(): Promise<ERPCapabilities>;
}

4.2 SAP Business One Service

File: apps/webapp/src/services/erp/implementations/sap-b1-service.ts

export class SAPBusinessOneService extends SAPServiceBase {
  /**
   * Business One specific OData field mapping.
   */
  protected buildCustomerFilterExpression(params: ErpCustomerSearchParams): string {
    const filters = [];

    if (params.name) {
      filters.push(`substringof('${this.sanitizeInput(params.name)}', CardName)`);
    }

    if (params.customerId) {
      filters.push(`CardCode eq '${this.sanitizeInput(params.customerId)}'`);
    }

    return filters.join(' or ') || `CardType eq 'C'`;
  }

  async validateConnection(): Promise<boolean> {
    try {
      const token = await this.getAccessToken();
      const response = await httpx.get(`${this.apiUrl}/b1s/v1/$metadata`, {
        headers: { Authorization: `Bearer ${token}` },
        timeout: 10.0,
      });
      return response.status === 200;
    } catch (error) {
      logger.error('SAP B1 validation failed', error);
      return false;
    }
  }

  async searchItems(params: ErpItemSearchParams): Promise<ErpItem[]> {
    const token = await this.getAccessToken();
    const nameFilter = `substringof('${this.sanitizeInput(params.itemName)}', ItemName)`;

    const response = await httpx.get(
      `${this.apiUrl}/b1s/v1/Items?$filter=${encodeURIComponent(nameFilter)}&$top=10`,
      {
        headers: { Authorization: `Bearer ${token}` },
        timeout: 30.0,
      }
    );

    return (response.json().value || []).map((item: any) => ({
      id: item.ItemCode,
      itemId: item.ItemCode,
      name: item.ItemName,
      description: item.ItemDescription,
      unitOfMeasure: item.BaseUnit,
      price: item.LastPurchasePrice,
    }));
  }

  async getERPCapabilities(): Promise<ERPCapabilities> {
    return {
      canSearchCustomers: true,
      canSearchItems: true,
      canGetAddresses: true,
      canCreateOrders: true,
      canSubmitOrders: false, // Requires manual approval in B1
      canGetInventory: true,
      canGetPricing: true,
    };
  }

  private sanitizeInput(value: string): string {
    return value.replace(/'/g, "''"); // Escape single quotes for OData
  }
}

4.3 SAP S/4HANA Cloud Service

File: apps/webapp/src/services/erp/implementations/sap-s4-cloud-service.ts

export class SAPS4HANACloudService extends SAPServiceBase {
  protected buildCustomerFilterExpression(params: ErpCustomerSearchParams): string {
    const filters = [];

    if (params.name) {
      filters.push(`contains(CustomerName,'${this.sanitizeInput(params.name)}')`);
    }

    if (params.customerId) {
      filters.push(`Customer eq '${this.sanitizeInput(params.customerId)}'`);
    }

    return filters.join(' or ') || `IsBlocked eq false`;
  }

  // S/4 Cloud uses OAuth2 + modern OData v4 endpoints
  async validateConnection(): Promise<boolean> {
    try {
      const token = await this.getAccessToken();
      const response = await httpx.get(`${this.apiUrl}/sap/opu/odata/IWFND/CATALOGSERVICE_SRVD`, {
        headers: { Authorization: `Bearer ${token}` },
        timeout: 10.0,
      });
      return response.status === 200;
    } catch (error) {
      logger.error('SAP S/4 Cloud validation failed', error);
      return false;
    }
  }

  async searchItems(params: ErpItemSearchParams): Promise<ErpItem[]> {
    const token = await this.getAccessToken();
    const filter = `contains(Product,'${this.sanitizeInput(params.itemName)}')`;

    const response = await httpx.get(
      `${this.apiUrl}/sap/opu/odata/sap/C_PRODUCT_SRV/C_ProductTP?$filter=${encodeURIComponent(filter)}&$top=10`,
      {
        headers: { Authorization: `Bearer ${token}` },
        timeout: 30.0,
      }
    );

    return (response.json().value || []).map((item: any) => ({
      id: item.Product,
      itemId: item.Product,
      name: item.ProductName,
      description: item.ProductDescription,
      unitOfMeasure: item.BaseUnit,
      price: item.StandardPrice,
    }));
  }

  async getERPCapabilities(): Promise<ERPCapabilities> {
    return {
      canSearchCustomers: true,
      canSearchItems: true,
      canGetAddresses: true,
      canCreateOrders: true,
      canSubmitOrders: true, // Cloud supports direct submission
      canGetInventory: true,
      canGetPricing: true,
    };
  }

  private sanitizeInput(value: string): string {
    return value.replace(/'/g, "''");
  }
}

4.4 SAP Service Factory (Updated)

File: apps/webapp/src/services/erp/erp-service-factory.ts (extend)

export class ERPServiceFactory {
  static createService(connection: ERPConnection): IERPService {
    switch (connection.erpType) {
      case 'prophet21':
        return new Prophet21Service(connection);

      case 'netsuite':
        return new NetSuiteService(connection);

      case 'sap_b1':
        return new SAPBusinessOneService(connection);

      case 'sap_s4_cloud':
        return new SAPS4HANACloudService(connection);

      case 'sap_s4_onprem':
        return new SAPS4HANAOnPremService(connection);

      case 'demo':
        return new DemoService(connection);

      default:
        throw new Error(`Unsupported ERP type: ${connection.erpType}`);
    }
  }
}

5. Database Schema Updates

5.1 Update erp_connections table

The existing erp_connections table in PostgreSQL should support SAP configurations:

-- erp_connections already has these fields:
-- - id (UUID)
-- - organization_id (UUID)
-- - erp_type (VARCHAR) ← 'sap_b1', 'sap_s4_cloud', 'sap_s4_onprem'
-- - api_url (VARCHAR)
-- - config (JSONB) ← store auth credentials, settings
-- - extra_config (JSONB) ← version, auth_type, etc.

-- Example config for Business One:
{
  "username": "sap_user@company.com",
  "api_url": "https://sap-b1.example.com:50000",
  "auth_type": "basic",
  "extra_config": {
    "sap_version": "9.3",
    "use_https": true,
    "access_method": "api"
  }
}

-- Example config for S/4 Cloud:
{
  "client_id": "from-sap-iam",
  "auth_type": "oauth2",
  "api_url": "https://api.s4hana.ondemand.com/sap/opu/odata",
  "extra_config": {
    "sap_client": "100",
    "odata_version": "v4"
  }
}

5.2 Credential Storage (Infisical)

SAP secrets stored in Infisical vault:

op://Ordermatic/SAP-{connection_id}/credential
  ├─ password (for basic auth)
  ├─ client_secret (for OAuth2)
  └─ mTLS cert (if using certificate auth)

6. Dagster Schedule & Sensor

6.1 Continuous Sync Schedule

File: apps/dagster/erp_pipeline/schedules.py (extend)

@schedule(
    job=full_sync_pipeline,
    cron_schedule="0 2 * * *",  # 2 AM daily
    name="sap_full_sync_schedule",
    description="Nightly full sync for SAP products/customers",
)
def sap_full_sync_schedule(context: ScheduleEvaluationContext):
    # Trigger for all SAP connections
    return RunRequest(
        tags={
            "kind": "full_sync",
            "erp_type": "sap",
        }
    )

6.2 Bronze Completion Sensor

Existing bronze_completion_sensor in sensors.py automatically triggers dbt transform when bronze data lands.

7. Authentication Strategies

7.1 Business One (Basic Auth)

GET https://sap-b1.example.com:50000/b1s/v1/Items
Authorization: Basic base64(username:password)

Credentials stored in Infisical, retrieved at sync time.

7.2 S/4HANA Cloud (OAuth2)

POST https://auth-server.com/oauth/token
  client_id: from-iam
  client_secret: from-infisical
  grant_type: client_credentials

→ Returns: access_token (valid 1 hour)

7.3 S/4HANA On-Prem (Basic or Kerberos)

GET https://sap-onprem.company.local:50000/sap/opu/odata/...
Authorization: Basic base64(username:password)
  OR
Authorization: Negotiate [kerberos-token]

8. Implementation Timeline

Phase 1: Foundation (Week 1-2)

✅ Create SAP Business One orchestrator + DLT source
✅ Create SAP S/4HANA Cloud orchestrator + DLT source
✅ Bronze/Silver/Gold/Typesense assets
✅ Basic TypeScript IERPService for Business One

Phase 2: Complete Implementations (Week 2-3)

✅ S/4HANA On-Prem implementation
✅ S/4HANA Cloud service completion
✅ Dagster schedules + sensors
✅ Integration testing with sandbox systems

Phase 3: Production Hardening (Week 3-4)

✅ Error handling + retry logic
✅ Performance optimization
✅ Monitoring + alerting
✅ Documentation + runbooks

9. Testing Strategy

9.1 Unit Tests

def test_sync_products_full():
    """Test full product sync from Business One."""
    ...

def test_sync_products_incremental():
    """Test incremental product sync."""
    ...

9.2 Integration Tests

test('searchCustomers returns matching customers', async () => {
  const service = new SAPBusinessOneService(mockConnection);
  const results = await service.searchCustomers({ name: 'Acme' });
  expect(results).toHaveLength(1);
  expect(results[0].name).toBe('Acme Corp');
});

9.3 Sandbox Testing

Use SAP sandbox systems for each version
Mock OData responses for unit tests
Full end-to-end flow in staging environment

10. Error Handling & Resilience

10.1 Timeout Configuration

# All SAP HTTP calls have explicit timeouts
httpx.get(url, timeout=30.0)  # OData queries
httpx.post(url, timeout=60.0)  # Order creation

10.2 Retry Policy (Dagster)

@asset
def sap_products(...):
    """Auto-retry on transient failures."""
    retry_policy = RetryPolicy(
        max_retries=3,
        delay=30,
        backoff=exponential_backoff
    )

10.3 Duplicate Prevention (Idempotency)

DLT’s write_disposition="merge" prevents duplicate products
Order submission uses Temporal for durability (future)

11. Monitoring & Observability

11.1 Dagster Asset Metrics

yield MaterializeResult(
    metadata={
        "records_loaded": 1523,
        "sync_duration_seconds": 45,
        "connection_id": "...",
        "erp_type": "sap_b1",
    }
)

11.2 OTEL Instrumentation

from opentelemetry import trace

with trace.get_tracer(__name__).start_as_current_span("sap.sync_products") as span:
    span.set_attribute("sap.connection_id", connection_id)
    span.set_attribute("sap.erp_type", "b1")
    # ... sync logic

12. File Structure Summary

apps/dagster/erp_pipeline/
├── implementations/
│   ├── sap_b1_sync_orchestrator.py          [NEW]
│   ├── sap_s4_cloud_sync_orchestrator.py    [NEW]
│   └── sap_s4_onprem_sync_orchestrator.py   [NEW]
├── sources/
│   ├── sap_b1/                              [NEW]
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── client.py
│   │   └── dlt_source.py
│   ├── sap_s4_cloud/                        [NEW]
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── client.py
│   │   └── dlt_source.py
│   └── sap_s4_onprem/                       [NEW]
│       ├── __init__.py
│       ├── auth.py
│       ├── client.py
│       └── dlt_source.py
├── assets/
│   ├── bronze.py                            [EXTEND with sap_* assets]
│   ├── typesense.py                         [EXTEND with sap_* assets]
│   └── dbt/
│       └── models/
│           ├── silver/stg_sap_products.sql  [NEW]
│           ├── silver/stg_sap_customers.sql [NEW]
│           └── gold/dim_sap_products.sql    [NEW]
└── tests/
    ├── test_sap_b1_orchestrator.py          [NEW]
    ├── test_sap_s4_orchestrators.py         [NEW]
    └── integration/
        └── test_sap_sync_e2e.py             [NEW]

apps/webapp/src/services/erp/
├── implementations/
│   ├── sap-service-base.ts                  [NEW]
│   ├── sap-b1-service.ts                    [NEW]
│   ├── sap-s4-cloud-service.ts              [NEW]
│   └── sap-s4-onprem-service.ts             [NEW]
└── erp-service-factory.ts                   [EXTEND]

apps/webapp/src/services/
└── erp-auth/
    ├── sap-auth.ts                          [NEW]
    └── providers/
        ├── sap-b1-auth.ts                   [NEW]
        ├── sap-oauth2-auth.ts               [NEW]
        └── sap-kerberos-auth.ts             [NEW]

13. Success Criteria

✅ Functional Requirements:

SAP Business One products/customers syncing to Iceberg
SAP S/4HANA Cloud OAuth2 auth working
SAP S/4HANA On-Prem basic auth working
All data appearing in Typesense within 5 minutes of sync
PDF extraction can match against SAP products via Typesense
Orders can be created in SAP via IERPService

✅ Non-Functional Requirements:

Sync completes in <5 minutes for 10k products
OData timeouts handled gracefully (retry + alert)
Connection validation works for all 3 SAP systems
Monitoring shows asset metrics in Dagster UI
OTEL traces visible in Better Stack for each sync

14. Future Enhancements

Temporal Workflow for order creation (durability)
Bidirectional sync (Ordermatic → SAP inventory updates)
Custom field mappings per customer
Real-time webhooks (SAP changes push to Ordermatic)
Multi-language support for product names
Batch order submission API

Document History

Date	Version	Author	Notes
2026-03-09	1.0	Jimmy	Initial design document for SAP integration