How-To: Common Team Questions

How do I add a new ERP integration?

1. Implement the TypeScript ERP service interface in apps/webapp/src/services/erp/implementations/
2. Register it in ERPServiceFactory (apps/webapp/src/services/erp/erp-service-factory.ts and apps/ts-temporal-worker/src/services/erp-service-factory.ts)
3. Add a corresponding type-declaration stub in apps/ts-temporal-worker/src/webapp-types/
4. If the ERP needs Python-side sync (for the Dagster pipeline), add a source in apps/dagster/erp_pipeline/sources/ and register its orchestrator in Python’s ERPSourceFactory (apps/dagster/erp_pipeline/sources/erp_source_factory.py)
5. Add it to the ERPType enum and registry (packages/integrations/src/registry/registry.ts)

Do not add ERP-specific branching anywhere outside the factory pattern.

How do I force a manual ERP sync?

Use the canonical admin endpoints (not the legacy Trigger.dev ones):

• POST /api/erp/force-sync — force a full ERP catalog sync
• POST /api/erp/force-image-sync — force image sync only

The legacy /api/trigger/force-erp-sync still works as a compat shim but will be removed. Use the new endpoints.

How do I add a new feature flag?

1. Create the flag in Flagsmith (staging first, then production) and add it to infrastructure/flagsmith/configs/flags.json. Any change affecting ERP writes, extraction, validation, or email triage must follow the Feature Flag Policy for ERP Writes.
2. TypeScript (webapp, ts-temporal-worker): Evaluate via @repo/feature-flags (getFeatureFlag on the server, useFeatureFlag in React).
3. Python (apps/temporal-worker): Evaluate via packages/feature-flags-py (from feature_flags_py import get_feature_flag).
4. Dagster (apps/dagster): In apps/dagster/erp_pipeline/config.py, add a get_<flag_name>() helper using get_iceberg_catalog_enabled() or get_p21_custom_source_enabled() as a template (falls back to env vars for local dev).
5. Gate your code on these helpers rather than raw Flagsmith SDK calls.

How do I add a new API route?

Extend BaseAPIHandler in apps/webapp/src/lib/api/base-handler.ts.

• Use requireAuth for user-facing routes
• Use requireAdmin for admin-only routes
• All routes automatically get rate limiting, bot blocking, and default-deny auth

How do I trigger a Temporal workflow from the webapp?

Use the helpers in apps/webapp/src/lib/temporal.ts — don’t create new Temporal client instances. For PDF extraction:

import { startPDFExtractionWorkflow } from '@/lib/temporal';
const { workflowId } = await startPDFExtractionWorkflow(documentId, { ... });

For ERP order submission, use the starters in apps/webapp/src/temporal/starters/.

How do I run the Dagster pipeline locally?

cd apps/dagster
uv run dagster dev

The feature flag mediator (config.py) reads from env vars in local dev, so set ICEBERG_CATALOG_ENABLED=true etc. to test flag-gated paths.

How do I add a column to the Iceberg schema?

1. Add to FIELD_ORDER in apps/dagster/erp_pipeline/utils/schema_registry.py first
2. Add to the transformer(s) that produce that column
3. The canonical field order must be maintained — schema_registry.py is the single source of truth

Where do background jobs live?

• Temporal (apps/temporal-worker, apps/ts-temporal-worker): durable workflows (PDF extraction, ERP submission, billing, email routing)
• Dagster (apps/dagster): scheduled syncs, dbt runs, catalog jobs
• Do not use Trigger.dev — it’s retired

How do I debug a stuck Temporal workflow?

Check the Temporal UI (ask David for the URL). Workflows that are stuck in QUOTE_PENDING or processing usually mean an activity failed after retries exhausted. Look at the workflow history for the specific error.

For PDF extraction specifically, the workflow ID is pdf-extraction-{documentId}.