ERP-Unlocked Architecture Evolution: Brownbag Presentation

Date: January 2025
Duration: 45 minutes
Audience: Engineering Team

Slide 1: Title

ERP-Unlocked Architecture Evolution

From Trigger.dev to Dagster: Building a Modern Multi-ERP ETL Pipeline

Presented by: Engineering Team

Slide 2: Agenda

The Problem: Why we needed to change
The Solution: New architecture overview
Key Components: Deep dive into Dagster, dlt, dbt, Typesense
Multi-ERP Support: How we handle different ERP systems
Infrastructure as Code: Terraform + Ansible automation
Migration Path: How we got here
What’s Next: Future enhancements

Slide 3: The Problem

Old Architecture Limitations

Trigger.dev + TypeScript + PostgreSQL

❌ Limited Observability

Basic Trigger.dev dashboard
No asset lineage
Difficult to debug failures

❌ Manual Multi-Tenancy

Connection IDs passed manually
No automatic discovery
Difficult to scale

❌ Tight Coupling

ERP-specific logic in TypeScript
Hard to add new ERP systems
No standardized transformation layer

❌ Storage Limitations

PostgreSQL for large datasets
No time travel or schema evolution
Expensive scaling

Slide 4: The Solution

New Architecture: Modern Data Stack

Dagster + dlt + dbt + Typesense + Iceberg

✅ Full Observability

Dagster UI with asset lineage
Complete run history
Error tracking and debugging

✅ Automatic Multi-Tenancy

Sensors discover connections
Dynamic partitions per tenant
Parallel processing

✅ Extensible Architecture

Factory pattern for ERPs
Standardized contracts
Easy to add new systems

✅ Scalable Storage

Iceberg tables on R2
Schema evolution
Time travel queries

Slide 5: Architecture Diagram

┌─────────────────────────────────────────────────────────┐
│              ERP Systems (Prophet21, NetSuite, SAP)      │
└───────────────────────┬─────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────┐
│         Dagster Orchestration (Discovery, Scheduling)   │
└───────────────────────┬─────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────┐
│         Extract (dlt) → Transform (Python) → Bronze     │
│         (Iceberg on R2 - Partitioned by connection_id)  │
└───────────────────────┬─────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────┐
│         Transform (dbt) → Silver → Gold                │
│         (DuckDB engine, Parquet on R2)                 │
└───────────────────────┬─────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────┐
│         Load (Typesense) → Search Collections          │
│         (Sub-millisecond search with typo tolerance)    │
└─────────────────────────────────────────────────────────┘

Slide 6: Dagster Orchestration

What is Dagster?

Modern data orchestration platform

Asset-Based: Models data as first-class citizens
Observability: Built-in UI with lineage and run history
Partitions: Native support for multi-tenant workloads
Sensors: Event-driven orchestration

Key Features

Automatic Discovery

@sensor(name="erp_connection_discovery_sensor")
def discover_connections(context):
    # Runs every 5 minutes
    # Queries erp_connections table
    # Creates partitions for new connections

Dynamic Partitions

One partition per connection_id
Automatic creation/deletion
Parallel processing per tenant

Schedules

Bronze sync: Every 2 hours (incremental)
dbt transform: Hourly
Typesense sync: Hourly

Slide 7: Data Extraction (dlt)

What is dlt?

Data Load Tool - Python library for building data pipelines

Features

✅ Declarative Sources

@dlt.source(name="prophet21")
def prophet21_source(api_url, username, password):
    # Returns resources for products, customers, etc.

✅ Automatic Schema Inference

Detects field types
Handles schema evolution

✅ Incremental Loading

Watermark-based syncs
Only fetches changed records

✅ Built-in Retry Logic

Exponential backoff
Rate limiting
Error handling

Output: Iceberg Tables on R2

Partitioned by connection_id
ACID transactions
Schema evolution
Time travel queries

Slide 8: Transformation Layer

Two-Stage Transformation

Stage 1: Python Transformers

Map ERP-specific → Canonical schema
Validate against contracts
Handle version differences

Stage 2: dbt Models

Silver: Deduplication, cleaning
Gold: Business logic, search optimization

Example: Product Transformation

Prophet21 Product          →  Canonical Product
─────────────────────────────────────────────────
item_id                    →  erp_product_id
item_desc                  →  name
price1, price2, ...         →  pricing.price1, pricing.price2
delete_flag = "Y"          →  delete_flag = true
date_last_modified         →  updated_at

dbt Gold Layer:

OCR variants: “WIDGET-100” → [“WIDGET100”, “WIDGET_100”, …]
Format variants: Normalized part numbers
Search text: Concatenated searchable fields

Slide 9: Storage: Apache Iceberg

Why Iceberg?

Open table format for large analytics datasets

✅ Schema Evolution

Add columns without rewriting data
Backward compatible changes

✅ Time Travel

Query historical snapshots
Audit trail built-in

✅ Partition Pruning

Efficient queries by connection_id
Only scans relevant partitions

✅ ACID Transactions

Safe concurrent writes
No data corruption

Storage Location

Cloudflare R2 (S3-compatible)

Cost-effective object storage
Global CDN integration
No egress fees

Table Structure:

s3://erp-data-catalog/
├── products/          (Iceberg table)
├── customers/         (Iceberg table)
├── cross_references/  (Iceberg table)
└── shipping_addresses/ (Iceberg table)

Slide 10: Search: Typesense

Why Typesense?

Fast, typo-tolerant search engine

Features

✅ Sub-Millisecond Performance

< 1ms search latency
Handles millions of documents

✅ Typo Tolerance

2-3 character typos
OCR error handling

✅ Prefix Matching

Fast partial matches
Great for part numbers

✅ Faceted Filtering

Filter by connection_id
Multi-tenant isolation

Collections

products: Full-text search with variants
customers: Fuzzy customer matching
cross_references: Part number lookup
shipping_addresses: Address search

Integration

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

Slide 11: Multi-ERP Support

Factory Pattern

ERP-Agnostic Architecture

def create_orchestrator(connection_id: str):
    connection = fetch_connection(connection_id)

    if connection.type == "prophet21":
        return Prophet21SyncOrchestrator(...)
    elif connection.type == "netsuite":
        return NetSuiteSyncOrchestrator(...)
    elif connection.type == "sap_b1":
        return SAPB1SyncOrchestrator(...)

Supported ERPs

ERP	Status	API Type	Key Features
Prophet21	✅ Production	OData	Version detection, input sanitization
Demo	✅ Testing	Mock	Test data generation
NetSuite	🔜 Planned	REST	OAuth 2.0, SuiteQL
SAP B1	🔜 Planned	REST	Session management, nested data

Slide 12: Multi-Version Support

The Challenge

ERP systems have version fragmentation

Example: Prophet21

API v1 vs v2
OData v3 vs v4
Data Services present/absent
Field casing: PascalCase vs snake_case

The Solution

Version Detection & Capability Flags

# Auto-detect on first connection
version_info = P21VersionDetector.detect()
# Returns: {api_version: "v2", odata_version: "v4", use_data_services: true}

# Store in erp_connections.extra_config
connection.extra_config = version_info.to_dict()

# Use in client
client = Prophet21Client(auth, version_info)
# Routes to correct endpoints based on version

Capability Flags

Per-ERP default capabilities
Per-connection overrides
Handles missing features gracefully

Slide 13: Infrastructure as Code

Terraform + Ansible

Complete automation for infrastructure

Terraform

Provisioning

Hetzner Cloud servers
Private networks
Firewall rules
Load balancers
DNS (Cloudflare)

Features

1Password integration for secrets
Remote state in R2
Multi-environment support

Ansible

Configuration

Server hardening
Docker installation
Coolify setup
Security policies

Features

Dynamic inventory (Hetzner API)
Idempotent playbooks
Role-based organization

Coolify Automation

Application Deployment

API-based setup
Environment variables from 1Password
GitHub webhook integration
Automatic SSL/TLS

Slide 14: Production Infrastructure

Server Architecture

Server	Type	Role	Cost
erp-web	CPX21	Webapp (Astro/React)	€10/mo
erp-processing	CPX31	PDF API + Worker	€20/mo
erp-data	CPX41	Dagster web + daemon	€40/mo
erp-failover	CPX41	Staging + failover	€40/mo
Load Balancer	LB11	Traffic distribution	€5/mo
Total			~€115/mo

Network

Private network (10.0.0.0/16)
Firewall rules for security
Load balancer for high availability

Deployment

Coolify: Docker container orchestration
GitHub Actions: CI/CD automation
1Password: Secrets management

Slide 15: Data Flow Example

Complete Sync Flow

1. Discovery Sensor (every 5 min)
   └─> Finds new connection: "abc-123" (Prophet21)
       └─> Creates partition

2. Sync Trigger (every 2 hours)
   └─> Triggers bronze_sync for "abc-123"

3. Bronze Sync
   └─> ERPSourceFactory → Prophet21SyncOrchestrator
       └─> dlt extracts from P21 API
           └─> Transformers normalize data
               └─> Writes to Iceberg (partitioned by connection_id)

4. dbt Transform (hourly)
   └─> Reads from Iceberg
       └─> Silver: Deduplication
           └─> Gold: OCR variants, search text
               └─> Writes to R2 (Parquet)

5. Typesense Sync (hourly)
   └─> Reads from Gold
       └─> Upserts to Typesense collections
           └─> Search available immediately

Time: ~15-20 minutes end-to-end for incremental sync

Slide 16: Performance Characteristics

Sync Performance

Operation	Time	Notes
Incremental Sync	5-10 min	10K changed records
Full Sync	30-60 min	100K records
Parallel Processing	N/A	Multiple connections simultaneously
Partition Pruning	Fast	Only processes changed tenants

Search Performance

Metric	Value
Search Latency	< 1ms
Typo Tolerance	2-3 characters
Prefix Matching	Instant
Faceted Filtering	< 1ms

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

Slide 17: Migration Path

How We Got Here

Phase 1: Planning (Q4 2024)

Evaluated Dagster vs alternatives
Designed multi-ERP architecture
Created integration playbooks

Phase 2: Implementation (Q4 2024 - Q1 2025)

Built Dagster pipeline
Migrated Prophet21 integration
Set up dbt transformations
Integrated Typesense

Phase 3: Infrastructure (Q1 2025)

Terraform for provisioning
Ansible for configuration
Coolify automation
CI/CD integration

Phase 4: Production (Q1 2025)

Deployed to production
Migrated existing connections
Monitored performance
Optimized workflows

Slide 18: Key Benefits

What We Gained

✅ Better Observability

Full asset lineage
Complete run history
Easy debugging

✅ Automatic Scaling

Auto-discovers new tenants
Parallel processing
No manual configuration

✅ Extensibility

Easy to add new ERPs
Standardized patterns
Reusable components

✅ Data Quality

Contract validation
Schema enforcement
Error tracking

✅ Performance

Fast incremental syncs
Efficient queries
Sub-millisecond search

Slide 19: What’s Next

Planned Enhancements

Additional ERP Systems

✅ Prophet21 (Production)
🔜 NetSuite (In Progress)
🔜 SAP Business One (Planned)
🔜 Microsoft Dynamics (Planned)

Enhanced Observability

BetterStack integration
Custom metrics dashboards
Alerting for sync failures

Performance Optimization

Parallel entity syncs
Streaming transformations
Caching layer

Data Quality

Automated quality checks
Anomaly detection
Data lineage tracking

Slide 20: Resources & Documentation

Key Documents

Architecture Docs: docs/architecture/dagster-etl-architecture.md
Dagster README: apps/dagster/README.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

Tools & Technologies

Dagster: https://dagster.io/
dlt: https://dlthub.com/
dbt: https://docs.getdbt.com/
Typesense: https://typesense.org/
Apache Iceberg: https://iceberg.apache.org/

Slide 21: Q&A

Questions?

Contact:

Engineering Team
Slack: #engineering
GitHub: erp-unlocked

Next Steps:

Review documentation
Explore Dagster UI
Check out integration playbooks

Appendix: Technical Deep Dives

A. Adding a New ERP

5 Steps to Add NetSuite:

Create DLT Source (sources/netsuite_source.py)
- Define @dlt.source with resources
- Handle OAuth 2.0, SuiteQL queries
Create Transformers (transformers/netsuite_products.py)
- Map NetSuite fields → Canonical schema
- Register in TransformerRegistry
Create Orchestrator (implementations/netsuite_sync_orchestrator.py)
- Implement ERPSourceProtocol
- Wire DLT → Transformers → Iceberg
Update Factory (sources/erp_source_factory.py)
- Add elif erp_type == "netsuite" case
Test & Deploy
- Integration tests with sandbox
- Contract validation
- Production rollout

See: debt/backlog/netsuite-integration-playbook.md for full details

B. Version Detection Example

Prophet21 Version Detection:

class P21VersionDetector:
    def detect(self) -> P21VersionInfo:
        # Probe OData v4 support
        odata_version = self._detect_odata_version()
        # Probe Data Services
        use_data_services = self._detect_data_services()

        return P21VersionInfo(
            api_version="v2" if use_data_services else "v1",
            odata_version=odata_version,
            use_data_services=use_data_services,
            detected_at=datetime.now().isoformat()
        )

Stored in: erp_connections.extra_config JSONB column

C. Typesense Search Example

Product Search:

results = typesense.collections['products'].documents.search({
    'q': 'widget-100',
    'filter_by': 'connection_id:=abc-123 && delete_flag:=false',
    'query_by': 'erp_product_id,ocr_variants,format_variants,name',
    'num_typos': 2,
    'prefix': True
})

Features:

Handles typos: “widgt-100” → finds “WIDGET-100”
OCR variants: “WIDGET100” → finds “WIDGET-100”
Format variants: “widget_100” → finds “WIDGET-100”
Sub-millisecond response

D. Infrastructure Commands

Terraform:

cd infrastructure/terraform/environments/production
terraform init
terraform plan
terraform apply

Ansible:

cd infrastructure/ansible
ansible-playbook playbooks/site.yml -i inventories/production/hcloud.yml

Coolify:

cd infrastructure/coolify
./scripts/full-coolify-setup.sh

See: infrastructure/README.md for full workflow

Presentation Notes

Slide Timing

Slides 1-5: 10 minutes (Problem & Solution)
Slides 6-10: 15 minutes (Components Deep Dive)
Slides 11-15: 10 minutes (Multi-ERP & Infrastructure)
Slides 16-21: 10 minutes (Performance & Q&A)

Key Points to Emphasize

Automatic Multi-Tenancy: No manual configuration needed
Extensibility: Easy to add new ERP systems
Observability: Full visibility into data pipeline
Performance: Fast incremental syncs and search
Infrastructure as Code: Complete automation

Demo Suggestions

Dagster UI: Show asset lineage graph
Run History: Show successful sync run
Typesense Search: Demo typo-tolerant search
Infrastructure: Show Terraform plan output

Questions to Anticipate

Q: Why Dagster instead of Airflow? A: Better asset-based model, native Python support, modern UI

Q: Why Iceberg instead of Parquet? A: Schema evolution, time travel, ACID transactions

Q: Why Typesense instead of Elasticsearch? A: Simpler setup, better typo tolerance, faster for our use case

Q: How do we handle ERP API changes? A: Version detection, capability flags, fallback chains

Q: What about data quality? A: Contract validation, cursor store tracking, error monitoring