Migration Guide: Trigger.dev to Dagster ERP Pipeline

May contain stale references

This document was audited on 2026-06-22 and contains references that need updating: schedules replaced by event-driven sensors. Cross-check critical details against the current codebase before relying on them.

Overview

This guide documents the migration of ERP data synchronization pipelines from Trigger.dev to Dagster. This migration provides better observability, scheduling, and multi-tenant support for ERP data pipelines.

What Changed

Before (Trigger.dev)

• Location: apps/trigger/src/python/etl/
• Execution: Trigger.dev Python extension
• Scheduling: Manual triggers or TypeScript task wrappers
• Observability: Limited to Trigger.dev dashboard
• Multi-tenancy: Manual connection ID passing
• Orchestration: Basic sequential execution

After (Dagster)

• Location: apps/dagster/erp_pipeline/
• Execution: Dagster webserver with native Python support
• Scheduling: Built-in cron schedules and sensors
• Observability: Full Dagster UI with asset lineage, run history, and logs
• Multi-tenancy: Automatic discovery via sensors, dynamic partitions
• Orchestration: Declarative asset-based pipeline with dependencies

Key Improvements

1. Multi-Tenant Architecture

Before: Manual connection ID management

# Old: Manual connection passing
def sync_products(connection_id: str):
    # ... sync logic

After: Automatic discovery and partitioning

# New: Automatic partition discovery
@sensor(name="erp_connection_discovery_sensor")
def discover_connections(context):
    # Automatically discovers new tenants from database
    # Creates partitions for each connection

2. Better Observability

• 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

3. Improved Scheduling

Before: Manual triggers or basic cron

// Old: TypeScript wrapper
await tasks.trigger('sync-erp-entity', {
  connectionId: 'uuid',
  entityType: 'products',
});

After: Declarative schedules

# New: Built-in cron schedules
@schedule(cron_schedule="0 */2 * * *", job=bronze_sync)
def bronze_sync_schedule(context):
    return RunRequest()

4. Incremental Sync Support

The new pipeline includes built-in incremental sync capabilities:

• Automatically detects first sync vs. incremental sync
• Tracks last sync date from Iceberg tables
• Only fetches changed records for efficiency

Migration Steps

Step 1: Review Old Code (Optional)

The old Trigger.dev pipeline code has been moved to apps/dagster/erp_pipeline/ with improvements:

• Sources: apps/trigger/src/python/etl/sources/ → apps/dagster/erp_pipeline/sources/
• Transformers: apps/trigger/src/python/etl/transformers/ → apps/dagster/erp_pipeline/transformers/
• Pipelines: apps/trigger/src/python/etl/*_pipeline.py → apps/dagster/erp_pipeline/implementations/

Step 2: Update Environment Variables

Add Dagster-specific environment variables to your .env:

# Dagster Configuration
DAGSTER_HOME=/path/to/dagster-home  # Optional: for local development

# R2 Storage (required for Iceberg tables)
R2_ENDPOINT_URL=https://<account-id>.r2.cloudflarestorage.com
R2_ACCESS_KEY_ID=...
R2_SECRET_ACCESS_KEY=...
R2_DATA_CATALOG_BUCKET=erp-data-catalog
R2_PRODUCT_IMAGES_BUCKET=erp-product-images

# Typesense (for search sync - use admin key for ETL operations)
TYPESENSE_HOST=localhost
TYPESENSE_PORT=8108
TYPESENSE_ADMIN_API_KEY=...
TYPESENSE_PROTOCOL=http

# Database (for ERP connection discovery)
DATABASE_URL=postgresql://...
ENCRYPTION_KEY=...  # For decrypting ERP passwords

Step 3: Start Dagster Service

Option A: Docker (Recommended)

# From project root
docker compose up -d dagster
docker compose logs -f dagster

Access Dagster UI at: http://localhost:3000

Option B: Native Python

cd apps/dagster
uv sync
uv run dagster dev

Step 4: Verify Pipeline

1. Check Asset Graph: Navigate to Assets → View the asset dependency graph
2. Test Discovery: Verify sensors are discovering ERP connections
3. Run Test Sync: Materialize a single asset for a test connection

Step 5: Disable Old Trigger.dev Tasks (Optional)

If you want to completely disable the old Trigger.dev pipeline:

# In apps/trigger/.env
ENABLE_ICEBERG_SYNC=false
ENABLE_DLT_NATIVE_SYNC=false

Note: The old code has been removed from apps/trigger/src/python/etl/, so these flags are no longer functional. They’re kept for backward compatibility only.

Architecture Comparison

Old Architecture (Trigger.dev)

┌─────────────────┐
│  Trigger.dev    │
│  TypeScript     │
│  Task Wrapper   │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Python DLT     │
│  Pipeline       │
│  (Sequential)   │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Iceberg (R2)   │
└─────────────────┘

New Architecture (Dagster)

┌─────────────────────────────────────┐
│  Dagster Webserver                  │
│  ┌───────────────────────────────┐ │
│  │  Discovery Sensor              │ │
│  │  (Auto-discovers connections)  │ │
│  └──────────────┬────────────────┘ │
│                 │                   │
│  ┌──────────────▼────────────────┐ │
│  │  Dynamic Partitions            │ │
│  │  (One per connection_id)       │ │
│  └──────────────┬────────────────┘ │
│                 │                   │
│  ┌──────────────▼────────────────┐ │
│  │  Asset Graph                   │ │
│  │  Bronze → dbt → Typesense     │ │
│  └──────────────┬────────────────┘ │
└─────────────────┼──────────────────┘
                  │
                  ▼
         ┌─────────────────┐
         │  Iceberg (R2)   │
         │  Typesense      │
         └─────────────────┘

Code Mapping

Pipeline Files

Old Location	New Location	Notes
apps/trigger/src/python/etl/base_pipeline.py	apps/dagster/erp_pipeline/implementations/p21_sync_orchestrator.py	Refactored as orchestrator
apps/trigger/src/python/etl/products_pipeline.py	apps/dagster/erp_pipeline/assets/bronze.py	Converted to Dagster asset
apps/trigger/src/python/etl/customers_pipeline.py	apps/dagster/erp_pipeline/assets/bronze.py	Converted to Dagster asset
apps/trigger/src/python/etl/sources/p21_odata.py	apps/dagster/erp_pipeline/sources/p21_odata.py	Moved, improved
apps/trigger/src/python/etl/transformers/	apps/dagster/erp_pipeline/transformers/	Moved, added registry

Key Changes

1. Factory Pattern: Added ERPSourceFactory for ERP-agnostic design
2. Transformer Registry: Centralized transformer mapping
3. Incremental Sync: Built-in sync state management
4. Multi-ERP Support: Protocol-based design for adding new ERPs

Running Pipelines

Old Way (Trigger.dev)

// From TypeScript code
await tasks.trigger('sync-erp-entity', {
  connectionId: 'uuid',
  entityType: 'products',
  options: { fullSync: true },
});

New Way (Dagster)

Via UI:

1. Navigate to Assets → Select asset (e.g., bronze/products)
2. Click Materialize → Select partition(s)
3. Monitor run in Runs tab

Via CLI:

# Sync specific connection
uv run dagster job execute -j bronze_sync --partition "connection-uuid"

# Sync all connections
uv run dagster job execute -j bronze_sync

Via GraphQL API:

const response = await fetch(DAGSTER_CLOUD_URL + '/graphql', {
  method: 'POST',
  headers: {
    'Dagster-Cloud-Api-Token': DAGSTER_CLOUD_API_TOKEN,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    query: `mutation LaunchRun($jobName: String!, $partitionKey: String) {
      launchRun(executionParams: {
        selector: { jobName: $jobName }
        runConfigData: {}
        tags: [{ key: "dagster/partition", value: $partitionKey }]
      }) {
        ... on LaunchRunSuccess { run { runId } }
      }
    }`,
    variables: {
      jobName: 'full_sync_pipeline',
      partitionKey: 'connection-uuid',
    },
  }),
});

Monitoring & Debugging

Old Way (Trigger.dev)

• Trigger.dev dashboard
• Limited run history
• Basic error messages

New Way (Dagster)

Dagster UI Features:

• Asset Lineage: Visual dependency graph
• Run History: Complete audit trail
• Logs: Structured logging with context
• Metrics: Performance and data volume metrics
• Retries: Automatic retry on failure

Debug Tools:

• apps/dagster/debug_auth.py - Test ERP authentication
• apps/dagster/test_image_sync.py - Test image downloads
• apps/dagster/list_runs.py - List and filter runs
• apps/dagster/cancel_hanging_runs.py - Cancel stuck runs

Scheduling

Old Way (Trigger.dev)

Manual triggers or basic cron via TypeScript tasks.

New Way (Dagster)

Built-in schedules defined in apps/dagster/erp_pipeline/schedules.py:

Schedule	Cron	Description
bronze_sync_schedule	0 */2 * * *	Incremental extract every 2 hours
dbt_transform_schedule	30 * * * *	Transform hourly
typesense_sync_schedule	45 * * * *	Search sync hourly

Schedules automatically run for all active connections (discovered via sensors).

Multi-Tenant Support

Old Way

Manual connection ID passing:

# Old: Manual connection management
def sync_products(connection_id: str):
    # ... sync logic

New Way

Automatic discovery and isolation:

1. Discovery Sensor: Runs every 5 minutes, discovers active connections
2. Dynamic Partitions: Each connection becomes a partition
3. Data Isolation: All queries filtered by connection_id
4. Parallel Execution: Multiple tenants sync simultaneously

Rollback Plan

If you need to rollback to Trigger.dev:

1. Restore Old Code: Checkout commit before migration
2. Re-enable Tasks: Set ENABLE_ICEBERG_SYNC=true in Trigger.dev
3. Stop Dagster: docker compose stop dagster

Note: The old Trigger.dev pipeline code has been removed. Rollback requires restoring from git history.

Troubleshooting

Issue: Dagster not discovering connections

Solution: Check database connection and ensure erp_connections table has active records.

# Test database connection
cd apps/dagster
uv run python -c "from erp_pipeline.partitions import get_active_erp_connections; print(get_active_erp_connections())"

Issue: Assets not materializing

Solution: Check Dagster logs and ensure R2 credentials are correct.

# Check Dagster logs
docker compose logs dagster

# Verify R2 access
uv run python -c "from erp_pipeline.utils.sync_state import get_iceberg_table; print(get_iceberg_table('products'))"

Issue: Incremental sync not working

Solution: Verify sync state utilities can read from Iceberg tables.

# Test sync state
uv run python -c "from erp_pipeline.utils.sync_state import get_last_sync_date; print(get_last_sync_date('products', 'connection-uuid'))"

Next Steps

1. Review Dagster README: apps/dagster/README.md for detailed usage
2. Monitor First Runs: Watch initial syncs in Dagster UI
3. Adjust Schedules: Modify cron schedules in schedules.py if needed
4. Add Tests: Consider adding integration tests for critical paths

Support

• Dagster Documentation: https://docs.dagster.io/
• Project README: apps/dagster/README.md
• Architecture Docs: docs/architecture/erp-sync-architecture.md

Summary

✅ Migrated: ERP sync pipelines from Trigger.dev to Dagster
✅ Improved: Multi-tenant support, observability, scheduling
✅ Maintained: Same data flow (DLT → Iceberg → dbt → Typesense)
✅ Enhanced: Incremental sync, automatic discovery, better error handling

The migration maintains backward compatibility with existing data structures while providing a more robust, observable, and maintainable pipeline infrastructure.