DXTA Pipeline

TypeScript monorepo for extract/transform pipelines using Temporal for workflow orchestration.

Prerequisites

Option 1: Nix (recommended)

nix develop

Provides: Node.js 24, pnpm, Biome, Temporal CLI, git, jq

Option 2: Manual

• Node.js >= 18.17.1
• pnpm 9.12.2
• Temporal server

Quick Start

1. Install dependencies

pnpm install

2. Start Temporal server

temporal server start-dev

Temporal UI available at http://localhost:8233

3. Configure environment

cp .env.example .env
# Edit .env with your credentials

Required variables:

Variable	Used By	Description
TENANT_DATABASE_AUTH_TOKEN	extract, transform	Auth token for tenant databases
SUPER_DATABASE_URL	extract, transform	Super database connection URL
SUPER_DATABASE_AUTH_TOKEN	extract, transform	Super database auth token
GITHUB_APP_ID	extract	GitHub App ID
GITHUB_APP_PRIVATE_KEY	extract	GitHub App private key (multiline)

Optional variables:

Variable	Default	Description
TEMPORAL_ADDRESS	localhost:7233	Temporal server address
TEMPORAL_NAMESPACE	default	Temporal namespace
PORT	3000	Orchestrator HTTP port
PER_PAGE	30	GitHub API pagination size
FETCH_TIMELINE_EVENTS_PER_PAGE	1000	Timeline events per page
EXTRACT_SCHEDULE_ENABLED	true	Enable extract schedule
TRANSFORM_SCHEDULE_ENABLED	true	Enable transform schedule

4. Start workers

Run in separate terminals:

# Extract worker (polls 'extract' queue)
pnpm --filter @dxta/worker-extract dev
# Transform worker (polls 'transform' queue)
pnpm --filter @dxta/worker-transform dev
# Orchestrator (creates schedules, exposes health endpoint)
pnpm --filter @dxta/orchestrator dev

5. Trigger workflows manually

# Start extract workflow
pnpm --filter @dxta/orchestrator start:extract
# Start transform workflow
pnpm --filter @dxta/orchestrator start:transform

Architecture

┌─────────────────┐ ┌──────────────────────────────────────┐
│ Orchestrator │────▶│ Temporal Server │
│ (schedules + │ │ localhost:7233 │
│ CLI triggers) │ └──────────────┬───────────────────────┘
└─────────────────┘ │
▼
┌──────────────────┴──────────────────┐
│ │
┌───────▼───────┐ ┌─────────▼─────────┐
│ worker-extract│ │ worker-transform │
│ (queue:extract)│ │ (queue:transform) │
└───────┬───────┘ └─────────┬─────────┘
│ │
▼ ▼
┌───────────────┐ ┌─────────────────┐
│ GitHub API │ │ Databases │
│ + Databases │ │ │
└───────────────┘ └─────────────────┘

Project Structure

apps/
orchestrator/ # Schedule management and CLI commands
worker-extract/ # Extract pipeline worker (17 activities)
worker-transform/ # Transform pipeline worker (2 activities)
workflows/ # Temporal workflow definitions
stack/ # Legacy SST/AWS infra (deprecated)
packages/
config/ # Shared configuration
functions/ # Business logic (extract, transform, crawl)
integrations/ # External service clients (source-control)
schemas/ # Database schemas (Drizzle)

Commands

Command	Description
pnpm install	Install all dependencies
pnpm run build	Build all packages
pnpm run dev	Start all apps in dev mode (parallel)
pnpm run type-check	Type-check all packages
pnpm run lint	Lint with Biome
pnpm run lint:fix	Auto-fix lint issues
pnpm run format	Format with Biome
pnpm run test	Run all tests
pnpm run db:generate	Generate database migrations

Single package commands

pnpm --filter @dxta/worker-extract dev
pnpm --filter @dxta/transform-functions test
pnpm --filter ./apps/orchestrator type-check

Run a single test

pnpm --filter @dxta/transform-functions test -- --runTestsByPath src/parse-hunks.test.ts
pnpm --filter @dxta/transform-functions test -- -t "parse hunks"

Workflows

Extract pipeline:

• extractTenantsWorkflow - Entry point, fans out to repositories
• extractRepositoryWorkflow - Per-repo extraction
• extractMergeRequestWorkflow - Per-MR extraction

Transform pipeline:

• transformTenantsWorkflow - Entry point, fans out to repositories
• transformRepositoryWorkflow - Per-repo transformation

Scheduling

When the orchestrator starts, it creates Temporal schedules:

• Extract: Every 15 minutes at minute 8 (:08, :23, :38, :53)
• Transform: Every 15 minutes on the quarter-hour (:00, :15, :30, :45)

Database Migrations

The project uses Drizzle Kit to generate SQL migrations. There are two migration targets:

Schema	Purpose	Output
@dxta/combined-schema	All per-tenant tables	migrations/combined/
@dxta/super-schema	Central tenant registry (deprecated)	migrations/super/

Generate migrations

# Generate all migrations
pnpm run db:generate
# Generate combined migrations only
pnpm --filter @dxta/combined-schema db:generate
# Generate super migrations only (deprecated)
pnpm --filter @dxta/super-schema db:generate

The combined-schema aggregates tables from extract-schema, transform-schema, tenant-schema, and crawl-schema. When you modify any of these base schemas, regenerate combined migrations.

Scripts

Seed Schema

Seeds dimension tables (dates, times, forge users) in a tenant database.

pnpm tsx scripts/seed-schema.mts -u "<DATABASE_URL>" -t "<AUTH_TOKEN>"

Arguments:

• -u, --url (required): LibSQL database URL
• -t, --authToken (optional): Database auth token

The script seeds data for a 10-year range (5 years before and after the current date).