factory-data-pipeline-engineer

Apply the data-pipeline-engineer specialist workflow. Design data flow that fits the factory's pipeline conventions, not bespoke ETL plumbing. Load factory-data-pipelines and factory-stack through the host's skill capability when needed.

How to think (in order)

1. What's the data shape? Pick one:

   • One-shot or scheduled CSV → TS script in scripts/data_processing/
   • Event stream / time-series → JSONB envelope on a structured parent table
   • External API ingestion (slow operation) → submit/poll/fetch async pattern
   • Reference data (slowly changing) → YAML config (Python side)
   • Compute job (sim, optimization, ML) → Python service with three entry points

   If it doesn't match one, that's the finding — name it.

2. TS or Python? Default: Next.js side (TS). Move to Python when:

   • Numeric / scientific libraries are non-trivial (geopandas, shapely, numpy/scipy)
   • Existing Python expertise / models
   • Compute runtime > Vercel function timeout (~10s on hobby, 60s on pro)

3. Storage shape? Drizzle table with structured columns for what drives queries + JSONB for what doesn't. Rule: if you need to filter or sort by it at app speed, it earns a column.

4. Deployment shape?

   • TS script → run locally or in GitHub Action; commits the data to DB
   • Cloud Run API → HTTP endpoint, FastAPI, API key dependency
   • Cloud Run Pub/Sub handler → async job processor
   • Long-running compute → Cloud Run with extended timeout, or Cloud Run Jobs

5. Migrations? Run in CI, not at runtime. Drizzle's generate + push, or dbmate for raw SQL projects.

6. Idempotency? Default to upsert-on-conflict for CSV imports. Wrap in a transaction. Don't assume "imported once."

7. Converter vs service split? Pure transforms in *-converter.ts (client-safe). I/O in *-service.ts (server-only). Don't blur.

Reference: canonical TS import script

// scripts/data_processing/import-foo.ts
import { readFileSync } from 'fs';
import Papa from 'papaparse';
import { db } from '@/db';
import { foo } from '@/db/schema';

const csvText = readFileSync(process.argv[2], 'utf8');
const { data, errors } = Papa.parse<FooRow>(csvText, {
  header: true,
  skipEmptyLines: true,
  dynamicTyping: true,
});

if (errors.length) {
  console.error('Parse errors:', errors);
  process.exit(1);
}

await db.transaction(async (tx) => {
  for (const row of data) {
    await tx.insert(foo).values({
      externalId: row.external_id,
      name: row.name,
      // ... map every column
    }).onConflictDoUpdate({
      target: foo.externalId,
      set: { name: row.name, updatedAt: new Date() },
    });
  }
});

console.log(`Imported ${data.length} rows`);