ERP-Unlocked Dagster ETL Architecture

Version: 2.0
Last Updated: January 2025
Status: Production

Executive Summary

ERP-Unlocked has migrated from a TypeScript + Trigger.dev ERP synchronization engine to a modern Python-based ETL pipeline using Dagster, dlt (Data Load Tool), and dbt (Data Build Tool). This architectural shift provides:

Better Observability: Full asset lineage, run history, and error tracking via Dagster UI
Multi-Tenant Support: Automatic discovery and partitioning of ERP connections
Multi-ERP Architecture: Extensible factory pattern supporting Prophet21, NetSuite, SAP B1, and more
Modern Data Stack: Iceberg tables on Cloudflare R2, dbt transformations, Typesense search
Infrastructure as Code: Complete Terraform + Ansible automation for deployment

Architecture Overview

┌─────────────────────────────────────────────────────────────────────────────┐
│                    ERP-Unlocked ETL Pipeline Architecture                   │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                    ERP Systems (External)                            │  │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐               │  │
│  │  │  Prophet21   │  │   NetSuite   │  │     SAP      │  ...          │  │
│  │  │    OData     │  │  SuiteTalk   │  │  Business 1  │               │  │
│  │  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘               │  │
│  └─────────┼──────────────────┼──────────────────┼──────────────────────┘  │
│            │                  │                  │                           │
│            └──────────────────┼──────────────────┘                           │
│                               ▼                                               │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                    Dagster Orchestration Layer                        │  │
│  │  ┌────────────────────────────────────────────────────────────────┐  │  │
│  │  │  Discovery Sensor: Auto-discovers ERP connections from DB       │  │  │
│  │  │  Dynamic Partitions: One partition per connection_id            │  │  │
│  │  │  Schedules: Hourly incremental syncs, daily full syncs         │  │  │
│  │  └────────────────────────────────────────────────────────────────┘  │  │
│  │                               │                                       │  │
│  │                               ▼                                       │  │
│  │  ┌────────────────────────────────────────────────────────────────┐  │  │
│  │  │  ERPSourceFactory: Creates ERP-specific orchestrators          │  │  │
│  │  │  - Prophet21SyncOrchestrator                                    │  │  │
│  │  │  - NetSuiteSyncOrchestrator (planned)                           │  │  │
│  │  │  - SAPB1SyncOrchestrator (planned)                             │  │  │
│  │  └────────────────────────────────────────────────────────────────┘  │  │
│  └───────────────────────────────┬───────────────────────────────────────┘  │
│                                  │                                           │
│                                  ▼                                           │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                    Extract Layer (dlt)                               │  │
│  │  ┌────────────────────────────────────────────────────────────────┐  │  │
│  │  │  DLT Sources: ERP-specific data extraction                      │  │  │
│  │  │  - p21_odata.py: Prophet21 OData API                           │  │  │
│  │  │  - netsuite_source.py: NetSuite REST API (planned)             │  │  │
│  │  │  - sap_b1_source.py: SAP Business One API (planned)            │  │  │
│  │  │                                                                 │  │  │
│  │  │  Features:                                                      │  │  │
│  │  │  - Automatic pagination                                         │  │  │
│  │  │  - Incremental sync via date_last_modified                      │  │  │
│  │  │  - Rate limiting and retry logic                                │  │  │
│  │  │  - Schema inference                                             │  │  │
│  │  └────────────────────────────────────────────────────────────────┘  │  │
│  └───────────────────────────────┬───────────────────────────────────────┘  │
│                                  │                                           │
│                                  ▼                                           │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                    Transform Layer (Python)                           │  │
│  │  ┌────────────────────────────────────────────────────────────────┐  │  │
│  │  │  TransformerRegistry: Maps ERP-specific → Canonical Schema      │  │  │
│  │  │  - transform_p21_product()                                     │  │  │
│  │  │  - transform_ns_product() (planned)                            │  │  │
│  │  │  - transform_sap_b1_product() (planned)                        │  │  │
│  │  │                                                                 │  │  │
│  │  │  Entity Contracts: Validates canonical schema compliance        │  │  │
│  │  │  - ProductContract                                             │  │  │
│  │  │  - CustomerContract                                            │  │  │
│  │  │  - ShippingAddressContract                                     │  │  │
│  │  │  - CrossReferenceContract                                      │  │  │
│  │  └────────────────────────────────────────────────────────────────┘  │  │
│  └───────────────────────────────┬───────────────────────────────────────┘  │
│                                  │                                           │
│                                  ▼                                           │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                    Bronze Layer (Iceberg on R2)                       │  │
│  │  ┌────────────────────────────────────────────────────────────────┐  │  │
│  │  │  Apache Iceberg Tables (Cloudflare R2)                          │  │  │
│  │  │  - products (partitioned by connection_id)                       │  │  │
│  │  │  - customers (partitioned by connection_id)                     │  │  │
│  │  │  - cross_references (partitioned by connection_id)              │  │  │
│  │  │  - shipping_addresses (partitioned by connection_id)            │  │  │
│  │  │  - ship_to_links (partitioned by connection_id)                │  │  │
│  │  │                                                                 │  │  │
│  │  │  Features:                                                      │  │  │
│  │  │  - Schema evolution without rewriting data                      │  │  │
│  │  │  - Time travel queries                                          │  │  │
│  │  │  - ACID transactions                                            │  │  │
│  │  │  - Partition pruning for efficient queries                      │  │  │
│  │  └────────────────────────────────────────────────────────────────┘  │  │
│  └───────────────────────────────┬───────────────────────────────────────┘  │
│                                  │                                           │
│                                  ▼                                           │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                    Transform Layer (dbt)                              │  │
│  │  ┌────────────────────────────────────────────────────────────────┐  │  │
│  │  │  dbt Models (DuckDB engine)                                     │  │  │
│  │  │                                                                 │  │  │
│  │  │  Silver Layer (Cleaning):                                       │  │  │
│  │  │  - stg_products: Deduplication, type coercion                   │  │  │
│  │  │  - stg_customers: Deduplication, standardization              │  │  │
│  │  │  - stg_cross_references: Deduplication                         │  │  │
│  │  │  - stg_shipping_addresses: Deduplication                       │  │  │
│  │  │                                                                 │  │  │
│  │  │  Gold Layer (Business Logic):                                   │  │  │
│  │  │  - dim_products: OCR variants, format variants, search text   │  │  │
│  │  │  - dim_customers: Search text, normalized fields              │  │  │
│  │  │  - dim_shipping_addresses: Normalized addresses               │  │  │
│  │  │  - xref_products: Customer part number → product mappings     │  │  │
│  │  └────────────────────────────────────────────────────────────────┘  │  │
│  └───────────────────────────────┬───────────────────────────────────────┘  │
│                                  │                                           │
│                                  ▼                                           │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                    Load Layer (Typesense)                             │  │
│  │  ┌────────────────────────────────────────────────────────────────┐  │  │
│  │  │  Typesense Collections (Search Engine)                          │  │  │
│  │  │  - products: Full-text search with OCR/format variants          │  │  │
│  │  │  - customers: Customer search with fuzzy matching               │  │  │
│  │  │  - cross_references: Part number cross-reference search         │  │  │
│  │  │  - shipping_addresses: Address search                            │  │  │
│  │  │                                                                 │  │  │
│  │  │  Features:                                                      │  │  │
│  │  │  - Typo tolerance (2-3 typos)                                   │  │  │
│  │  │  - Prefix matching                                              │  │  │
│  │  │  - Faceted filtering by connection_id                           │  │  │
│  │  │  - Sub-millisecond search performance                          │  │  │
│  │  └────────────────────────────────────────────────────────────────┘  │  │
│  └──────────────────────────────────────────────────────────────────────┘  │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

Migration: From Trigger.dev to Dagster

Old Architecture (Trigger.dev)

Location: apps/trigger/src/python/etl/ (deprecated)

Characteristics:

TypeScript task wrappers calling Python DLT pipelines
Manual connection ID passing
Limited observability (Trigger.dev dashboard only)
Basic sequential execution
PostgreSQL as primary storage

Limitations:

No automatic tenant discovery
Manual orchestration required
Limited error tracking and lineage
Difficult to scale across multiple ERP systems
No standardized transformation layer

New Architecture (Dagster)

Location: apps/dagster/erp_pipeline/

Characteristics:

Native Python orchestration with Dagster
Automatic connection discovery via sensors
Full observability with asset lineage
Declarative asset-based pipeline
Iceberg tables on R2 for scalable storage

Benefits:

Automatic Multi-Tenancy: Sensors discover new connections and create partitions
Better Observability: Complete run history, asset dependencies, error tracking
Extensibility: Factory pattern makes adding new ERPs straightforward
Data Quality: Contract validation ensures schema compliance
Performance: Partition pruning, incremental syncs, efficient queries

Core Components

1. Dagster Orchestration

Location: apps/dagster/erp_pipeline/

Definitions (`definitions.py`)

Main entry point for Dagster
Registers all assets, resources, jobs, schedules, and sensors
Conditionally loads dbt assets if manifest exists

Partitions (`partitions.py`)

Dynamic Partitions: erp_connection_partitions
Automatically discovers ERP connections from erp_connections table
One partition per connection_id for tenant isolation

Sensors (`sensors.py`)

erp_connection_discovery_sensor: Runs every 5 minutes
- Queries erp_connections for active connections
- Registers new connections as partitions
- Removes partitions for deactivated connections
erp_sync_trigger_sensor: Runs every 2 hours
- Creates run requests for all active connection partitions
- Enables parallel syncs across tenants

Schedules (`schedules.py`)

bronze_sync_schedule: 0 */2 * * * (every 2 hours)
- Incremental extract for all connections
dbt_transform_schedule: 30 * * * * (hourly at :30)
- Transform bronze → silver → gold
typesense_sync_schedule: 45 * * * * (hourly at :45)
- Incremental search sync for all connections

2. ERP Source Factory

Location: apps/dagster/erp_pipeline/sources/erp_source_factory.py

Purpose: Creates ERP-specific orchestrators based on connection type

def create_orchestrator(connection_id: str) -> ERPSourceProtocol:
    connection = fetch_connection_from_db(connection_id)
    erp_type = connection.type

    if erp_type == "prophet21":
        return Prophet21SyncOrchestrator(connection_id, connection.config)
    elif erp_type == "demo":
        return DemoSyncOrchestrator(connection_id, connection.config)
    elif erp_type == "netsuite":
        return NetSuiteSyncOrchestrator(connection_id, connection.config)  # Planned
    # ... other ERPs

Supported ERPs:

✅ Prophet21: OData API with version detection (v1/v2, OData v3/v4)
✅ Demo: Mock data for testing
🔜 NetSuite: SuiteTalk REST API (see netsuite-integration-playbook.md)
🔜 SAP Business One: Service Layer API (see sap-business-one-integration-playbook.md)

3. Data Extraction (dlt)

Location: apps/dagster/erp_pipeline/sources/

Technology: dlt (Data Load Tool)

Features:

Declarative source definitions
Automatic schema inference
Incremental loading via watermarks
Built-in pagination and retry logic
Native Iceberg output support

Example: Prophet21 Source (p21_odata.py)

@dlt.source(name="prophet21")
def prophet21_source(
    api_url: str,
    username: str,
    password: str,
    connection_id: str,
    last_sync_iso: str | None = None,
) -> DltSource:
    # Returns DLT resources for products, customers, etc.
    # Handles OData pagination, filtering, authentication

Incremental Sync:

Uses date_last_modified field for watermarking
Only fetches records modified since last sync
Supports full refresh when needed

4. Transformation Layer

Python Transformers

Location: apps/dagster/erp_pipeline/transformers/

Purpose: Map ERP-specific field names to canonical schema

Registry Pattern:

TRANSFORMER_REGISTRY = {
    "prophet21": {
        "products": transform_p21_product,
        "customers": transform_p21_customer,
        # ...
    },
    "netsuite": {
        "products": transform_ns_product,  # Planned
        # ...
    }
}

Entity Contracts:

Location: apps/dagster/erp_pipeline/contracts/
Validates transformed records against canonical schema
Ensures data quality and consistency
Contracts: ProductContract, CustomerContract, ShippingAddressContract, etc.

dbt Transformations

Location: apps/dagster/dbt/

Technology: dbt with DuckDB engine

Why DuckDB?

In-memory speed for analytical queries
Native iceberg_scan() function for reading Iceberg tables
No server required (runs as library)
S3/R2 compatible for direct object storage access

Data Layers:

Bronze (Raw): Views over Iceberg tables
- raw_products, raw_customers, etc.
- Minimal transformation, just exposes Iceberg data
Silver (Cleaned): Deduplication and standardization
- stg_products: Dedupe by (connection_id, erp_product_id)
- stg_customers: Dedupe by (connection_id, erp_customer_id)
- Type coercion, null handling, field standardization
Gold (Business Logic): Pre-computed dimensions
- dim_products: OCR variants, format variants, search text
- dim_customers: Search text, normalized fields
- xref_products: Customer part number → product ID mappings

Partitioning:

dbt models support connection_id filtering
Partitioned runs process one tenant at a time
Full refresh processes all tenants

5. Storage Layer

Apache Iceberg on Cloudflare R2

Why Iceberg?

Schema Evolution: Add columns without rewriting data
Time Travel: Query historical snapshots
Partition Pruning: Efficient queries by connection_id
ACID Transactions: Safe concurrent writes
Open Format: Vendor-neutral, S3-compatible

Table Structure:

-- Example: products table
CREATE TABLE products (
    connection_id STRING NOT NULL,  -- Partition key
    erp_product_id STRING NOT NULL, -- Primary key component
    name STRING,
    description STRING,
    list_price DECIMAL,
    delete_flag BOOLEAN,
    updated_at TIMESTAMP,
    -- ... other fields
) USING ICEBERG
PARTITIONED BY (connection_id)
LOCATION 's3://erp-data-catalog/products/'

Partitioning Strategy:

All tables partitioned by connection_id
Enables efficient tenant isolation
Supports parallel processing per tenant

Typesense Collections

Location: apps/dagster/erp_pipeline/assets/typesense.py

Purpose: Full-text search engine for fast product/customer lookup

Collections:

products: Product search with OCR/format variants
customers: Customer search with fuzzy matching
cross_references: Part number cross-reference search
shipping_addresses: Address search

Document Structure:

{
  "id": "{connection_id}_{erp_id}",
  "connection_id": "abc-123",
  "erp_product_id": "WIDGET-100",
  "name": "Widget Standard",
  "ocr_variants": ["WIDGET100", "WIDGET-100", "WIDGET_100"],
  "format_variants": ["WIDGET-100", "WIDGET100"],
  "search_text": "widget standard widget-100 widget100"
  // ... other fields
}

Features:

Typo tolerance (2-3 typos)
Prefix matching
Faceted filtering by connection_id
Sub-millisecond search performance

Sync Strategy:

Reads from Gold layer (dbt dim_* tables)
Incremental upserts based on updated_at
Handles deletes via delete_flag

6. Multi-ERP Support

Version Detection & Capability Flags

Location: apps/dagster/erp_pipeline/capabilities.py

Purpose: Handle ERP version fragmentation and feature differences

Example: Prophet21

Detects API version (v1/v2)
Detects OData version (v3/v4)
Detects Data Services availability
Stores in erp_connections.extra_config JSONB column

Capability Flags:

DEFAULT_CAPABILITIES["prophet21"] = ERPCapabilities(
    erp_type="prophet21",
    products=True,
    customers=True,
    cross_references=True,
    shipping_addresses=True,
    ship_to_links=True,
    product_images=True,
)

Per-Connection Overrides:

Some ERP instances may have custom configurations
Use capability_overrides JSONB column
Example: Disable product images for specific P21 instance

ERP-Specific Patterns

Prophet21:

Version Detection: Probes OData v4 support, Data Services availability
Input Sanitization: Prevents SQL injection in OData queries
Fallback Chains: Multiple query strategies for robustness
Multi-Method Support: API access + direct database access for on-premise deployments
Order Creation Tiers: Transaction API → File Import → Direct Database (with sign-off)
See: debt/projects/multi-erp-foundation/prophet21-refactor-playbook.md

NetSuite (Planned):

OAuth 2.0: Refresh token or client credentials JWT flow
SuiteQL: SQL-like queries for flexible data extraction
Address Book: Flattening nested addressBook.items structure
See: debt/backlog/netsuite-integration-playbook.md

SAP Business One (Planned):

Session Management: B1SESSION cookie with auto-renewal
Nested Data: Flattening BPAddresses and BPCatalogNumbers
Delete Semantics: Using Valid/Frozen flags instead of hard deletes
See: debt/backlog/sap-business-one-integration-playbook.md

Data Flow Example

Complete Sync Flow

1. Discovery Sensor (every 5 min)
   └─> Queries erp_connections table
       └─> Finds new connection: id="abc-123", type="prophet21"
           └─> Creates partition: "abc-123"

2. Sync Trigger Sensor (every 2 hours)
   └─> Creates run request for partition "abc-123"
       └─> Triggers bronze_sync job

3. Bronze Sync Job
   └─> ERPSourceFactory.create_orchestrator("abc-123")
       └─> Returns Prophet21SyncOrchestrator
           └─> sync_products(full_sync=False, last_sync_iso="2024-01-01T00:00:00Z")
               └─> DLT extracts from P21 OData API
                   └─> TransformerRegistry.transform_p21_product()
                       └─> Writes to Iceberg: products (connection_id="abc-123")

4. dbt Transform (hourly at :30)
   └─> Reads from Iceberg: products (connection_id="abc-123")
       └─> Silver: stg_products (deduplication)
           └─> Gold: dim_products (OCR variants, search text)
               └─> Writes to R2: gold/dim_products.parquet

5. Typesense Sync (hourly at :45)
   └─> Reads from Gold: dim_products (connection_id="abc-123")
       └─> Upserts to Typesense: products collection
           └─> Search available immediately

Incremental Sync Flow

1. Bronze Sync (every 2 hours)
   └─> Reads cursor from cursor_store: last_sync="2024-01-15T10:00:00Z"
       └─> DLT filters: date_last_modified > "2024-01-15T10:00:00Z"
           └─> Only fetches changed records
               └─> Updates cursor_store: last_sync="2024-01-15T12:00:00Z"

2. dbt Transform
   └─> Processes only changed partitions
       └─> Incremental materialization where possible

3. Typesense Sync
   └─> Reads only changed records from Gold
       └─> Incremental upsert (no full rebuild)

Infrastructure as Code

Location: infrastructure/

Terraform

Purpose: Provision cloud infrastructure

Structure:

infrastructure/terraform/
├── modules/
│   ├── hetzner-server/      # Server provisioning
│   ├── hetzner-network/     # Private network
│   ├── hetzner-firewall/    # Firewall rules
│   ├── hetzner-load-balancer/ # Load balancer
│   └── cloudflare-dns/     # DNS management
└── environments/
    ├── production/          # Production config
    └── staging/            # Staging config

Features:

1Password Integration: Secrets fetched via Terraform provider
Remote State: Stored in Cloudflare R2
Dynamic Inventory: Ansible uses Hetzner Cloud API
Multi-Environment: Production and staging support

Production Servers:

erp-web: Webapp (CPX21, €10/mo)
erp-processing: PDF API + Worker (CPX31, €20/mo)
erp-data: Dagster web + daemon (CPX41, €40/mo)
erp-failover: Staging + failover (CPX41, €40/mo)
Load Balancer (LB11, €5/mo)

Ansible

Purpose: Configure servers and deploy applications

Structure:

infrastructure/ansible/
├── playbooks/
│   ├── site.yml            # Master playbook
│   ├── provision.yml       # Initial server setup
│   ├── security.yml        # Security hardening
│   └── coolify.yml         # Coolify installation
├── roles/
│   ├── base-server/        # Common server setup
│   ├── docker/             # Docker installation
│   ├── security/           # Security hardening
│   └── coolify-worker/     # Coolify worker setup
└── inventories/
    └── production/
        └── hcloud.yml      # Dynamic inventory config

Features:

Dynamic Inventory: Uses Hetzner Cloud API (no static files)
Idempotent: Safe to run multiple times
Security Hardening: fail2ban, UFW, SSH hardening
Docker Setup: Installs Docker and Docker Compose
Coolify Integration: Sets up Coolify workers

Coolify Automation

Location: infrastructure/coolify/

Purpose: Automated application deployment via Coolify API

Scripts:

create-project.sh: Creates Coolify project
add-servers.sh: Adds servers to Coolify
create-applications.sh: Creates applications (webapp, pdf-api, dagster, etc.)
set-environment.sh: Sets environment variables from 1Password
configure-webhooks.sh: Configures GitHub webhooks for CI/CD

Integration:

GitHub Actions triggers Coolify deployments
Environment variables synced from 1Password
Automatic SSL/TLS via Coolify reverse proxy

Monitoring & Observability

Dagster UI

Access: http://dagster.ordermatic.co (production)

Features:

Asset Lineage: Visual graph of data dependencies
Run History: Complete audit trail of all syncs
Error Tracking: Detailed error messages and stack traces
Performance Metrics: Execution time, data volume, success rates
Partition Management: View and manage connection partitions

Cursor Store

Location: apps/dagster/erp_pipeline/utils/cursor_store.py

Purpose: Track sync state per entity per connection

Schema:

CREATE TABLE sync_cursors (
    connection_id UUID NOT NULL,
    entity_type VARCHAR(50) NOT NULL,
    timestamp TIMESTAMP,
    record_count INTEGER,
    load_id VARCHAR(255),
    status VARCHAR(20),  -- 'running', 'success', 'failed'
    error_message TEXT,
    PRIMARY KEY (connection_id, entity_type)
);

Usage:

Tracks last successful sync timestamp
Enables incremental syncs
Monitors sync health
Provides audit trail

Logging

Structured Logging:

Python: logging module with structured context
Dagster: Built-in logging with asset context
BetterStack: Application performance monitoring (planned)

Development Workflow

Local Development

# Start Dagster locally
cd apps/dagster
uv sync
uv run dagster dev

# Access UI at http://localhost:3000

Adding a New ERP

Create DLT Source (sources/netsuite_source.py)
- Define @dlt.source with resources
- Handle authentication, pagination, incremental sync
Create Transformers (transformers/netsuite_products.py)
- Map ERP fields to canonical schema
- Register in TransformerRegistry
Create Orchestrator (implementations/netsuite_sync_orchestrator.py)
- Implement ERPSourceProtocol interface
- Wire DLT source → transformers → Iceberg
Update Factory (sources/erp_source_factory.py)
- Add case for new ERP type
- Return appropriate orchestrator
Add Capabilities (capabilities.py)
- Define default capabilities
- Document per-connection overrides
Test & Deploy
- Integration tests with sandbox data
- Contract validation
- Production rollout

Performance Characteristics

Sync Performance

Incremental Sync: ~5-10 minutes for 10K changed records
Full Sync: ~30-60 minutes for 100K records
Parallel Processing: Multiple connections sync simultaneously
Partition Pruning: Only processes changed tenants

Search Performance

Typesense: Sub-millisecond search (< 1ms)
Typo Tolerance: Handles 2-3 character typos
Prefix Matching: Fast partial matches
Faceted Filtering: Efficient tenant isolation

Storage Efficiency

Iceberg: Columnar format, efficient compression
Partitioning: Only queries relevant tenant data
Incremental Updates: Only writes changed records
Time Travel: Historical data without duplication

Security Considerations

Authentication

ERP Connections: Encrypted credentials in erp_connections table
OAuth 2.0: Secure token management for NetSuite, SAP
Session Management: Auto-renewal for session-based auth (SAP B1)

Data Isolation

Partitioning: All data partitioned by connection_id
Query Filtering: Always filter by connection_id
Typesense: Faceted filtering prevents cross-tenant access

Secrets Management

1Password: Centralized secrets via Terraform provider
Environment Variables: No secrets in code or config files
Encryption: Database credentials encrypted at rest

Future Enhancements

Planned Features

Additional ERP Systems
- NetSuite (in progress)
- SAP Business One (planned)
- Microsoft Dynamics (planned)
Enhanced Observability
- BetterStack integration
- Custom metrics and dashboards
- Alerting for sync failures
Performance Optimization
- Parallel entity syncs within connection
- Streaming transformations
- Caching layer for frequently accessed data
Data Quality
- Automated data quality checks
- Anomaly detection
- Data lineage tracking

References

Dagster Docs: apps/dagster/docs/architecture.md
Prophet21 Playbook: debt/backlog/prophet21-refactor-playbook.md
NetSuite Playbook: debt/backlog/netsuite-integration-playbook.md
SAP B1 Playbook: debt/backlog/sap-business-one-integration-playbook.md
Infrastructure Docs: infrastructure/README.md
Coolify Setup: infrastructure/coolify/README.md

ERP-Unlocked Dagster ETL Architecture

Executive Summary

Architecture Overview

Migration: From Trigger.dev to Dagster

Old Architecture (Trigger.dev)

New Architecture (Dagster)

Core Components

1. Dagster Orchestration

Definitions (definitions.py)

Partitions (partitions.py)

Sensors (sensors.py)

Schedules (schedules.py)

2. ERP Source Factory

3. Data Extraction (dlt)

4. Transformation Layer

Python Transformers

dbt Transformations

5. Storage Layer

Apache Iceberg on Cloudflare R2

Typesense Collections

6. Multi-ERP Support

Version Detection & Capability Flags

ERP-Specific Patterns

Data Flow Example

Complete Sync Flow

Incremental Sync Flow

Infrastructure as Code

Terraform

Ansible

Coolify Automation

Monitoring & Observability

Dagster UI

Cursor Store

Logging

Development Workflow

Local Development

Adding a New ERP

Performance Characteristics

Sync Performance

Search Performance

Storage Efficiency

Security Considerations

Authentication

Data Isolation

Secrets Management

Future Enhancements

Planned Features

References

Definitions (`definitions.py`)

Partitions (`partitions.py`)

Sensors (`sensors.py`)

Schedules (`schedules.py`)