Antigravity AI Data Pipeline × ETL Automation Production Guide

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
// Fetch data from REST API
async function fetchApiData(endpoint: string, apiKey: string) {
  const response = await fetch(endpoint, {
    headers: { Authorization: `Bearer ${apiKey}` },
  });
  const data = await response.json();
  return data;
}
 
// Query PostgreSQL for large datasets
async function fetchPostgresData(
  connectionString: string,
  query: string
): Promise<Record<string, any>[]> {
  const { Client } = await import("pg");
  const pgClient = new Client({ connectionString });
  await pgClient.connect();
  const result = await pgClient.query(query);
  await pgClient.end();
  return result.rows;
}
 
// Stream JSON Lines from S3
async function fetchS3Data(
  bucket: string,
  key: string,
  awsAccessKey: string,
  awsSecretKey: string
) {
  const { S3Client, GetObjectCommand } = await import("@aws-sdk/client-s3");
  const s3Client = new S3Client({
    region: "us-east-1",
    credentials: {
      accessKeyId: awsAccessKey,
      secretAccessKey: awsSecretKey,
    },
  });
 
  const command = new GetObjectCommand({ Bucket: bucket, Key: key });
  const response = await s3Client.send(command);
  const stream = response.Body;
  const lines: Record<string, any>[] = [];
 
  return new Promise((resolve, reject) => {
    stream.on("data", (chunk: Buffer) => {
      const text = chunk.toString();
      text.split("\n").forEach((line) => {
        if (line.trim()) {
          lines.push(JSON.parse(line));
        }
      });
    });
    stream.on("end", () => resolve(lines));
    stream.on("error", reject);
  });
}

async function inferSchema(
  sampleData: Record<string, any>[],
  dataSourceName: string
) {
  const stats = analyzeSampleStatistics(sampleData);
 
  const messages: Anthropic.Messages.MessageParam[] = [
    {
      role: "user",
      content: `You are a data engineer analyzing a new data source.
 
Source: ${dataSourceName}
Sample records count: ${sampleData.length}
Data statistics:
${JSON.stringify(stats, null, 2)}
 
First 5 records:
${JSON.stringify(sampleData.slice(0, 5), null, 2)}
 
Your task:
1. Infer the schema (column names, types, constraints)
2. Identify business keys (likely primary keys)
3. Detect data quality issues
4. Suggest normalization approach
 
Respond with JSON:
- schema (array of {name, type, nullable, constraints})
- businessKeys (array)
- dataQualityIssues (array)
- normalizationStrategy (string)`,
    },
  ];
 
  const response = await client.messages.create({
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 2048,
    messages,
  });
 
  const textContent = response.content.find((block) => block.type === "text");
  if (!textContent || textContent.type !== "text") {
    throw new Error("No text response from schema inference");
  }
 
  const jsonMatch = textContent.text.match(/\{[\s\S]*\}/);
  if (!jsonMatch) {
    throw new Error("Could not parse schema");
  }
 
  return JSON.parse(jsonMatch[0]);
}
 
function analyzeSampleStatistics(data: Record<string, any>[]) {
  const stats: Record<string, any> = {};
 
  if (data.length === 0) return stats;
 
  const firstRecord = data[0];
  for (const key of Object.keys(firstRecord)) {
    const values = data.map((r) => r[key]);
    const nullCount = values.filter((v) => v === null || v === undefined).length;
    const uniqueCount = new Set(values).size;
 
    stats[key] = {
      nullCount,
      nullPercentage: (nullCount / data.length) * 100,
      uniqueCount,
      uniquePercentage: (uniqueCount / data.length) * 100,
      sampleValues: values.slice(0, 3),
      types: [
        ...new Set(
          values.map((v) =>
            v === null ? "null" : typeof v === "object" ? "object" : typeof v
          )
        ),
      ],
    };
  }
 
  return stats;
}

interface CleaningRule {
  field: string;
  operation: string;
  parameters: Record<string, any>;
  description: string;
}
 
async function generateCleaningRules(
  schema: Record<string, any>,
  sampleData: Record<string, any>[],
  businessContext: string
): Promise<CleaningRule[]> {
  const dataIssues = detectDataQualityIssues(sampleData);
 
  const messages: Anthropic.Messages.MessageParam[] = [
    {
      role: "user",
      content: `You are a data quality expert designing cleaning rules.
 
Business Context: ${businessContext}
 
Detected Schema:
${JSON.stringify(schema, null, 2)}
 
Data Quality Issues:
${JSON.stringify(dataIssues, null, 2)}
 
Problem records:
${JSON.stringify(sampleData.slice(0, 3), null, 2)}
 
Generate cleaning rules as JSON array. For each field, include:
- field: column name
- operation: trim, normalize_email, remove_duplicates, fix_phone, parse_date, remove_outliers, etc.
- parameters: operation-specific configuration
- description: why this rule is needed
 
Return production-ready rules.`,
    },
  ];
 
  const response = await client.messages.create({
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 2048,
    messages,
  });
 
  const textContent = response.content.find((block) => block.type === "text");
  if (!textContent || textContent.type !== "text") {
    throw new Error("No text response");
  }
 
  const jsonMatch = textContent.text.match(/\[[\s\S]*\]/);
  if (!jsonMatch) {
    throw new Error("Could not parse rules");
  }
 
  return JSON.parse(jsonMatch[0]);
}
 
function detectDataQualityIssues(data: Record<string, any>[]) {
  const issues: Record<string, string[]> = {};
 
  data.forEach((record, idx) => {
    Object.entries(record).forEach(([key, value]) => {
      if (!issues[key]) issues[key] = [];
 
      if (value === null || value === undefined) {
        issues[key].push(`Row ${idx}: null/undefined`);
      } else if (typeof value === "string") {
        if (value.trim() === "") issues[key].push(`Row ${idx}: empty string`);
        if (value !== value.toLowerCase() && /\d/.test(value)) {
          issues[key].push(`Row ${idx}: mixed case with numbers`);
        }
      } else if (typeof value === "number") {
        if (value < 0 && key.includes("count")) {
          issues[key].push(`Row ${idx}: negative count`);
        }
      }
    });
  });
 
  return Object.fromEntries(
    Object.entries(issues).filter(([_, v]) => v.length > 0)
  );
}
 
// Apply cleaning operations
function applyCleaningRules(
  record: Record<string, any>,
  rules: CleaningRule[]
): Record<string, any> {
  const cleaned = { ...record };
 
  rules.forEach((rule) => {
    const value = cleaned[rule.field];
 
    switch (rule.operation) {
      case "trim":
        cleaned[rule.field] =
          typeof value === "string" ? value.trim() : value;
        break;
 
      case "normalize_email":
        cleaned[rule.field] =
          typeof value === "string" ? value.toLowerCase().trim() : value;
        break;
 
      case "parse_date":
        try {
          cleaned[rule.field] = new Date(value as string).toISOString();
        } catch {
          cleaned[rule.field] = null;
        }
        break;
 
      case "remove_outliers":
        const { threshold } = rule.parameters;
        if (typeof value === "number" && Math.abs(value) > threshold) {
          cleaned[rule.field] = null;
        }
        break;
 
      case "fix_phone":
        if (typeof value === "string") {
          cleaned[rule.field] = value.replace(/\D/g, "");
        }
        break;
    }
  });
 
  return cleaned;
}

async function enrichWithExternalData(
  record: Record<string, any>,
  enrichmentConfig: {
    geoLookup?: boolean;
    currencyConversion?: boolean;
    entityMatching?: boolean;
  }
): Promise<Record<string, any>> {
  const enriched = { ...record };
 
  if (enrichmentConfig.geoLookup && enriched.ip_address) {
    enriched.geo_country = "JP";
    enriched.geo_city = "Tokyo";
  }
 
  if (enrichmentConfig.currencyConversion && enriched.amount) {
    enriched.amount_jpy = (enriched.amount as number) * 130;
  }
 
  if (enrichmentConfig.entityMatching && enriched.company_name) {
    const messages: Anthropic.Messages.MessageParam[] = [
      {
        role: "user",
        content: `Find matching company for: "${enriched.company_name}"
 
Return JSON with:
- matched_name (exact name or null)
- confidence (0-1)
- entity_id (internal ID or null)`,
      },
    ];
 
    const response = await client.messages.create({
      model: "claude-3-5-sonnet-20241022",
      max_tokens: 256,
      messages,
    });
 
    const textContent = response.content.find((block) => block.type === "text");
    if (textContent && textContent.type === "text") {
      const jsonMatch = textContent.text.match(/\{[\s\S]*\}/);
      if (jsonMatch) {
        const match = JSON.parse(jsonMatch[0]);
        Object.assign(enriched, match);
      }
    }
  }
 
  return enriched;
}

import { Client } from "pg";
 
async function loadToPostgres(
  connectionString: string,
  tableName: string,
  records: Record<string, any>[],
  schema: Record<string, any>
): Promise<{ insertedRows: number; failedRows: number }> {
  const pgClient = new Client({ connectionString });
  await pgClient.connect();
 
  try {
    const createTableSQL = generateCreateTableSQL(tableName, schema);
    await pgClient.query(createTableSQL);
 
    let insertedRows = 0;
    let failedRows = 0;
 
    for (const record of records) {
      try {
        const values = schema.fields.map(
          (field: { name: string }) => record[field.name]
        );
        await pgClient.query(
          `INSERT INTO ${tableName} (${schema.fields.map((f: { name: string }) => f.name).join(", ")})
           VALUES (${schema.fields.map((_, i) => `$${i + 1}`).join(", ")})`,
          values
        );
        insertedRows++;
      } catch (err) {
        console.error(`Failed to insert:`, err);
        failedRows++;
      }
    }
 
    return { insertedRows, failedRows };
  } finally {
    await pgClient.end();
  }
}
 
function generateCreateTableSQL(
  tableName: string,
  schema: Record<string, any>
): string {
  const columns = schema.fields
    .map((field: { name: string; type: string; nullable?: boolean }) => {
      const typeMap: Record<string, string> = {
        string: "TEXT",
        number: "NUMERIC",
        integer: "INTEGER",
        boolean: "BOOLEAN",
        datetime: "TIMESTAMP",
        date: "DATE",
      };
 
      const sqlType = typeMap[field.type] || "TEXT";
      const nullable = field.nullable !== false ? "" : " NOT NULL";
 
      return `"${field.name}" ${sqlType}${nullable}`;
    })
    .join(", ");
 
  return `
    CREATE TABLE IF NOT EXISTS ${tableName} (
      id SERIAL PRIMARY KEY,
      ${columns},
      created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    );
  `;
}

import { BigQuery } from "@google-cloud/bigquery";
 
async function loadToBigQuery(
  projectId: string,
  datasetId: string,
  tableId: string,
  records: Record<string, any>[],
  schema: Record<string, any>,
  mode: "INSERT" | "OVERWRITE"
) {
  const bigquery = new BigQuery({ projectId });
  const dataset = bigquery.dataset(datasetId);
  const table = dataset.table(tableId);
 
  const bqSchema = schema.fields.map((field: { name: string; type: string }) => {
    const typeMap: Record<string, string> = {
      string: "STRING",
      number: "FLOAT64",
      integer: "INT64",
      boolean: "BOOL",
      datetime: "TIMESTAMP",
      date: "DATE",
      object: "JSON",
    };
 
    return {
      name: field.name,
      type: typeMap[field.type] || "STRING",
      mode: "NULLABLE",
    };
  });
 
  const options = {
    schema: bqSchema,
    writeDisposition:
      mode === "OVERWRITE" ? "WRITE_TRUNCATE" : "WRITE_APPEND",
    skipLeadingRows: 0,
  };
 
  try {
    const [job] = await table.insert(records, options);
    await job.waitForCompletion();
 
    console.log(`Loaded ${records.length} rows to BigQuery`);
    return { success: true, rowsLoaded: records.length };
  } catch (err) {
    console.error("BigQuery load failed:", err);
    throw err;
  }
}

import {
  S3Client,
  PutObjectCommand,
} from "@aws-sdk/client-s3";
 
async function loadToS3DataLake(
  bucket: string,
  records: Record<string, any>[],
  sourceSystem: string,
  partitionDate: string,
  awsAccessKey: string,
  awsSecretKey: string
) {
  const s3Client = new S3Client({
    region: "us-east-1",
    credentials: {
      accessKeyId: awsAccessKey,
      secretAccessKey: awsSecretKey,
    },
  });
 
  // Partition: s3://bucket/source/year/month/day/hour/data.jsonl
  const [year, month, day] = partitionDate.split("-");
  const hour = new Date().getHours().toString().padStart(2, "0");
 
  const s3Key = `data-lake/${sourceSystem}/year=${year}/month=${month}/day=${day}/hour=${hour}/data_${Date.now()}.jsonl`;
 
  const jsonlData = records
    .map((r) => JSON.stringify(r))
    .join("\n");
 
  const putCommand = new PutObjectCommand({
    Bucket: bucket,
    Key: s3Key,
    Body: jsonlData,
    ContentType: "application/x-ndjson",
  });
 
  try {
    await s3Client.send(putCommand);
    console.log(`Loaded ${records.length} records to S3`);
    return { success: true, s3Path: s3Key };
  } catch (err) {
    console.error("S3 load failed:", err);
    throw err;
  }
}

interface AnomalyDetectionConfig {
  zscore_threshold?: number;
  iqr_multiplier?: number;
  pattern_rules?: Array<{
    field: string;
    pattern: string;
    description: string;
  }>;
}
 
async function detectAnomalies(
  records: Record<string, any>[],
  schema: Record<string, any>,
  config: AnomalyDetectionConfig = {}
): Promise<
  Array<{
    recordIndex: number;
    anomalies: Array<{ field: string; reason: string; severity: string }>;
  }>
> {
  const zscoreThreshold = config.zscore_threshold || 3;
  const anomalousRecords = [];
 
  const numericStats = computeNumericStats(records, schema);
 
  for (let idx = 0; idx < records.length; idx++) {
    const record = records[idx];
    const anomalies: Array<{
      field: string;
      reason: string;
      severity: string;
    }> = [];
 
    // Z-score detection
    schema.fields.forEach((field: { name: string; type: string }) => {
      if (field.type === "number" && numericStats[field.name]) {
        const value = record[field.name];
        if (typeof value === "number") {
          const stats = numericStats[field.name];
          const zscore = Math.abs((value - stats.mean) / stats.stddev);
 
          if (zscore > zscoreThreshold) {
            anomalies.push({
              field: field.name,
              reason: `Z-score ${zscore.toFixed(2)} exceeds ${zscoreThreshold}`,
              severity: zscore > zscoreThreshold * 1.5 ? "CRITICAL" : "WARNING",
            });
          }
        }
      }
    });
 
    // Pattern-based rules
    if (config.pattern_rules) {
      for (const rule of config.pattern_rules) {
        const value = record[rule.field];
        const regex = new RegExp(rule.pattern);
 
        if (value && !regex.test(String(value))) {
          anomalies.push({
            field: rule.field,
            reason: `Does not match pattern: ${rule.description}`,
            severity: "WARNING",
          });
        }
      }
    }
 
    if (anomalies.length > 0) {
      anomalousRecords.push({ recordIndex: idx, anomalies });
    }
  }
 
  return anomalousRecords;
}
 
function computeNumericStats(
  records: Record<string, any>[],
  schema: Record<string, any>
): Record<
  string,
  { mean: number; stddev: number; min: number; max: number }
> {
  const stats: Record<
    string,
    { mean: number; stddev: number; min: number; max: number }
  > = {};
 
  schema.fields.forEach((field: { name: string; type: string }) => {
    if (field.type === "number") {
      const values = records
        .map((r) => r[field.name])
        .filter((v) => typeof v === "number");
 
      if (values.length > 0) {
        const mean = values.reduce((a, b) => a + b, 0) / values.length;
        const variance =
          values.reduce((sum, v) => sum + Math.pow(v - mean, 2), 0) /
          values.length;
        const stddev = Math.sqrt(variance);
 
        stats[field.name] = {
          mean,
          stddev: stddev || 1,
          min: Math.min(...values),
          max: Math.max(...values),
        };
      }
    }
  });
 
  return stats;
}

async function validateBusinessRules(
  records: Record<string, any>[],
  businessRulesDescription: string
): Promise<Array<{ recordIndex: number; violations: string[] }>> {
  const sampleRecords = records.slice(0, 10);
 
  const messages: Anthropic.Messages.MessageParam[] = [
    {
      role: "user",
      content: `You are a business rules validator. Check these records against the described rules.
 
Business Rules:
${businessRulesDescription}
 
Sample Records:
${JSON.stringify(sampleRecords, null, 2)}
 
Return JSON with violations:
[
  {
    recordIndex: number,
    violations: ["violation description 1", "violation description 2"]
  }
]
 
Only include records with violations.`,
    },
  ];
 
  const response = await client.messages.create({
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 1024,
    messages,
  });
 
  const textContent = response.content.find((block) => block.type === "text");
  if (!textContent || textContent.type !== "text") {
    return [];
  }
 
  const jsonMatch = textContent.text.match(/\[[\s\S]*\]/);
  if (!jsonMatch) {
    return [];
  }
 
  return JSON.parse(jsonMatch[0]);
}

interface PipelineStep {
  name: string;
  fn: (context: PipelineContext) => Promise<void>;
  retryConfig?: { maxRetries: number; backoffMs: number };
  timeout?: number;
  dependencies?: string[];
}
 
interface PipelineContext {
  sourceData: Record<string, any>[];
  schema: Record<string, any>;
  cleaningRules?: any[];
  cleanedData?: Record<string, any>[];
  validationResults?: any[];
  loadResults?: Record<string, any>;
  metadata: {
    startTime: Date;
    pipelineId: string;
    executionMode: "full" | "incremental";
  };
}
 
async function executePipelineWithRetry(
  step: PipelineStep,
  context: PipelineContext,
  attempt: number = 1
): Promise<void> {
  const maxRetries = step.retryConfig?.maxRetries || 3;
  const backoffMs = step.retryConfig?.backoffMs || 5000;
 
  try {
    const timeoutPromise = new Promise<void>((_, reject) => {
      setTimeout(() => reject(new Error("Step timeout")), step.timeout || 300000);
    });
 
    const executionPromise = step.fn(context);
 
    await Promise.race([executionPromise, timeoutPromise]);
  } catch (err) {
    if (attempt < maxRetries) {
      console.log(
        `Step "${step.name}" failed (attempt ${attempt}/${maxRetries}). Retrying...`
      );
      await new Promise((resolve) => setTimeout(resolve, backoffMs));
      await executePipelineWithRetry(step, context, attempt + 1);
    } else {
      console.error(`Step "${step.name}" failed after ${maxRetries} attempts:`, err);
      throw err;
    }
  }
}
 
async function runFullPipeline(
  steps: PipelineStep[],
  initialContext: PipelineContext
) {
  const context = initialContext;
  const executedSteps = new Set<string>();
 
  while (executedSteps.size < steps.length) {
    let stepExecuted = false;
 
    for (const step of steps) {
      if (executedSteps.has(step.name)) continue;
 
      const depsReady =
        !step.dependencies ||
        step.dependencies.every((dep) => executedSteps.has(dep));
 
      if (depsReady) {
        console.log(`Executing: ${step.name}`);
        const startTime = Date.now();
 
        try {
          await executePipelineWithRetry(step, context);
          executedSteps.add(step.name);
 
          const duration = ((Date.now() - startTime) / 1000).toFixed(2);
          console.log(`✓ ${step.name} completed in ${duration}s`);
 
          stepExecuted = true;
        } catch (err) {
          console.error(`✗ ${step.name} failed:`, err);
          throw err;
        }
      }
    }
 
    if (!stepExecuted) {
      throw new Error("Circular dependency detected");
    }
  }
 
  console.log("Pipeline completed successfully");
  return context;
}

import cron from "node-cron";
 
interface ScheduleConfig {
  cronExpression: string;
  pipelineName: string;
  steps: PipelineStep[];
  alertOnFailure: {
    enabled: boolean;
    webhookUrl?: string;
  };
}
 
class PipelineOrchestrator {
  private schedules = new Map<string, any>();
 
  scheduleRecurringPipeline(config: ScheduleConfig) {
    const job = cron.schedule(config.cronExpression, async () => {
      console.log(`Starting: ${config.pipelineName}`);
 
      const context: PipelineContext = {
        sourceData: [],
        schema: {},
        metadata: {
          startTime: new Date(),
          pipelineId: `${config.pipelineName}_${Date.now()}`,
          executionMode: "full",
        },
      };
 
      try {
        await runFullPipeline(config.steps, context);
        this.logExecution(config.pipelineName, "SUCCESS", context);
      } catch (err) {
        console.error(`Pipeline failed: ${err}`);
 
        if (config.alertOnFailure.enabled) {
          await this.sendAlert(config, err as Error);
        }
 
        this.logExecution(config.pipelineName, "FAILED", context, err as Error);
      }
    });
 
    this.schedules.set(config.pipelineName, job);
    console.log(
      `Scheduled pipeline "${config.pipelineName}" with cron: ${config.cronExpression}`
    );
  }
 
  private async sendAlert(config: ScheduleConfig, error: Error) {
    if (config.alertOnFailure.webhookUrl) {
      await fetch(config.alertOnFailure.webhookUrl, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          pipeline: config.pipelineName,
          status: "FAILED",
          error: error.message,
          timestamp: new Date().toISOString(),
        }),
      });
    }
  }
 
  private logExecution(
    pipelineName: string,
    status: string,
    context: PipelineContext,
    error?: Error
  ) {
    const duration = Date.now() - context.metadata.startTime.getTime();
    console.log(
      `[${new Date().toISOString()}] ${pipelineName} - ${status} (${(duration / 1000).toFixed(2)}s)`
    );
 
    if (error) {
      console.error(`Error: ${error.message}`);
    }
  }
 
  stopPipeline(pipelineName: string) {
    const job = this.schedules.get(pipelineName);
    if (job) {
      job.stop();
      this.schedules.delete(pipelineName);
      console.log(`Stopped: ${pipelineName}`);
    }
  }
}

// Don't use inference as-is; lock the critical columns' types
const lockedSchema = {
  ...inferredSchema,
  revenue: { type: "number", nullable: true, default: 0 },
  date: { type: "date", nullable: false },
  appId: { type: "string", nullable: false },
};

interface DataLineageRecord {
  recordId: string;
  sourceSystem: string;
  sourceTimestamp: Date;
  transformations: Array<{
    step: string;
    ruleId: string;
    timestamp: Date;
  }>;
  destinationSystems: Array<{
    system: string;
    loadTimestamp: Date;
  }>;
  qualityScore: number;
}
 
async function trackDataLineage(
  record: Record<string, any>,
  sourceSystem: string,
  transformationHistory: Array<{
    step: string;
    ruleId: string;
  }>
): Promise<DataLineageRecord> {
  return {
    recordId: `${sourceSystem}_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`,
    sourceSystem,
    sourceTimestamp: new Date(),
    transformations: transformationHistory.map((t) => ({
      step: t.step,
      ruleId: t.ruleId,
      timestamp: new Date(),
    })),
    destinationSystems: [],
    qualityScore: 0.95,
  };
}
 
interface PipelineMetrics {
  totalRecordsProcessed: number;
  successRate: number;
  averageProcessingTime: number;
  anomaliesDetected: number;
  dataQualityScore: number;
}
 
function computePipelineMetrics(
  context: PipelineContext
): PipelineMetrics {
  const duration =
    Date.now() - context.metadata.startTime.getTime();
  const totalRecords = context.sourceData.length;
  const anomalyCount =
    context.validationResults?.filter(
      (r) => r.anomalies?.length > 0
    ).length || 0;
 
  return {
    totalRecordsProcessed: totalRecords,
    successRate:
      context.loadResults?.insertedRows /
        (context.loadResults?.insertedRows + context.loadResults?.failedRows || 1) || 0,
    averageProcessingTime: duration / totalRecords,
    anomaliesDetected: anomalyCount,
    dataQualityScore:
      1 - (anomalyCount) / totalRecords,
  };
}

async function processBatchesInParallel(
  records: Record<string, any>[],
  batchSize: number = 10000,
  processFunction: (batch: Record<string, any>[]) => Promise<Record<string, any>[]>,
  concurrency: number = 4
) {
  const batches = [];
  for (let i = 0; i < records.length; i += batchSize) {
    batches.push(records.slice(i, i + batchSize));
  }
 
  const results: Record<string, any>[] = [];
 
  for (let i = 0; i < batches.length; i += concurrency) {
    const batchGroup = batches.slice(i, i + concurrency);
    const groupResults = await Promise.all(
      batchGroup.map((batch) => processFunction(batch))
    );
 
    results.push(...groupResults.flat());
 
    console.log(
      `Processed ${Math.min((i + concurrency) * batchSize, records.length)} / ${records.length}`
    );
  }
 
  return results;
}

import { Transform, pipeline } from "stream";
import { createReadStream, createWriteStream } from "fs";
 
async function processLargeFileWithStream(
  inputPath: string,
  outputPath: string,
  transformFn: (record: Record<string, any>) => Promise<Record<string, any>>
) {
  const inputStream = createReadStream(inputPath, { encoding: "utf8" });
 
  const transformStream = new Transform({
    objectMode: true,
    async transform(chunk: string, _, callback) {
      try {
        const record = JSON.parse(chunk.trim());
        const transformed = await transformFn(record);
        callback(null, JSON.stringify(transformed) + "\n");
      } catch (err) {
        callback(err);
      }
    },
  });
 
  const outputStream = createWriteStream(outputPath, { encoding: "utf8" });
 
  return new Promise((resolve, reject) => {
    pipeline(inputStream, transformStream, outputStream, (err) => {
      if (err) reject(err);
      else resolve(null);
    });
  });
}