# npm
npm install @async-fusion/data

# yarn
yarn add @async-fusion/data

# pnpm
pnpm add @async-fusion/data

# Kafka support (peer dep)
npm install kafkajs

# React hooks support
npm install react react-dom

import { PipelineBuilder } from '@async-fusion/data';

const pipeline = new PipelineBuilder({ name: 'user-activity' })
  .source('kafka', {
    topic: 'user-clicks',
    brokers: ['localhost:9092'],
  })
  .transform(data => ({
    ...data,
    processedAt: new Date().toISOString(),
  }))
  .transform(data => (data.value > 100 ? data : null))
  .sink('console', { format: 'pretty' });

await pipeline.run();

new PipelineBuilder(config: PipelineConfig, options?: PipelineOptions)

import { PipelineBuilder } from '@async-fusion/data';

const pipeline = new PipelineBuilder(
  {
    name: 'etl-pipeline',
    checkpointLocation: './checkpoints',
    parallelism: 4,
  },
  {
    retryConfig: { maxAttempts: 5, delayMs: 1000, backoffMultiplier: 2 },
    errorHandler: (error, ctx) => {
      console.error(`[${ctx.pipelineName}]`, error.message);
    },
  }
);

pipeline
  .source('kafka', { topic: 'orders', brokers: ['localhost:9092'] })
  .transform(validate)
  .transform(enrich)
  .sink('database', { table: 'processed_orders' })
  .sink('kafka', { topic: 'enriched-orders' });

const lineage = pipeline.lineage();
const metrics = pipeline.getMetrics();
await pipeline.run();

// Fan-in: merge two Kafka topics
pipeline
  .source('kafka', { topic: 'clicks',      brokers: ['localhost:9092'] })
  .source('kafka', { topic: 'impressions', brokers: ['localhost:9092'] })
  .transform(mergeEvents)
  .sink('spark', { job: 'funnel-analysis' });

import { Producer } from '@async-fusion/data';

const producer = new Producer({
  clientId: 'my-app',
  brokers: ['localhost:9092'],
});

await producer.connect();

await producer.send({
  topic: 'orders',
  messages: [{ key: 'order-1', value: JSON.stringify({ id: 1, total: 250 }) }],
});

await producer.sendBatch([
  { topic: 'orders',    messages: [{ value: '...' }] },
  { topic: 'audit-log', messages: [{ value: '...' }] },
]);

await producer.disconnect();

import { Consumer } from '@async-fusion/data';

const consumer = new Consumer({
  clientId: 'my-app',
  brokers: ['localhost:9092'],
  groupId: 'analytics-group',
});

await consumer.connect();
await consumer.subscribe({ topic: 'orders', fromBeginning: false });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const order = JSON.parse(message.value!.toString());
    await processOrder(order);
  },
});

import { KafkaStream } from '@async-fusion/data';

const stream = new KafkaStream('clickstream', {
  windowSize: 60_000,    // 1-min tumbling window
  slideInterval: 30_000, // slide every 30s
  watermarkDelay: 5_000, // tolerate 5s late data
});

const result = stream
  .filter(e => e.type === 'purchase')
  .window(60_000)
  .groupBy(e => e.userId)
  .avg(e => e.amount);

for await (const stat of result) {
  console.log('Avg purchase per user/min', stat);
}

import { SparkClient } from '@async-fusion/data';

const spark = new SparkClient({ host: 'spark-master', port: 7077, appName: 'my-job' });

const jobId = await spark.submitJob({
  mainClass: 'com.example.MyJob',
  jarPath: '/jobs/my-job.jar',
  args: ['--input', 'hdfs://data/input'],
});

const status = await spark.waitForCompletion(jobId);
console.log('Job finished with status:', status);

import { SparkSQL } from '@async-fusion/data';

const sql = new SparkSQL({ host: 'spark-thrift-server', port: 10000 });

const rows = await sql.query(`
  SELECT userId, SUM(amount) AS total
  FROM orders
  WHERE created_at > NOW() - INTERVAL 7 DAYS
  GROUP BY userId
  ORDER BY total DESC
  LIMIT 100
`);

console.table(rows);

import { SparkStreaming } from '@async-fusion/data';

const streaming = new SparkStreaming({
  host: 'spark-master',
  checkpointDir: '/tmp/checkpoints',
});

await streaming.startQuery({
  name: 'order-aggregation',
  source: { type: 'kafka', topic: 'orders', brokers: ['localhost:9092'] },
  transform: 'SELECT userId, COUNT(*) AS cnt FROM stream GROUP BY userId',
  sink: { type: 'console', outputMode: 'complete' },
});

import {
  useKafkaTopic,
  useSparkQuery,
  useRealtimeData,
} from '@async-fusion/data/react';

// Live Kafka topic
function LiveOrders() {
  const { messages, isConnected, error } = useKafkaTopic('orders', {
    brokers: ['localhost:9092'],
    groupId: 'ui-consumer',
  });
  return (
    <div>
      <span>{isConnected ? 'Live' : 'Reconnecting…'}</span>
      {messages.map(m => <div key={m.offset}>{m.value}</div>)}
    </div>
  );
}

// Spark query result
function TopUsers() {
  const { data, loading } = useSparkQuery(
    'SELECT userId, SUM(amount) as total FROM orders GROUP BY userId',
    { refreshInterval: 30_000 }
  );
  if (loading) return <p>Loading…</p>;
  return <pre>{JSON.stringify(data, null, 2)}</pre>;
}

import { PipelineBuilder, dashboardManager } from '@async-fusion/data';

// Option A: attach to a pipeline
const pipeline = new PipelineBuilder({ name: 'clickstream' })
  .enableDashboard(3000)
  .source('kafka', { topic: 'clicks', brokers: ['localhost:9092'] })
  .transform(enrich)
  .sink('kafka', { topic: 'enriched-clicks' });

await pipeline.run();

// Option B: standalone manager
dashboardManager.start(3000);
dashboardManager.recordMetric('my-pipeline', 1000, 2);
dashboardManager.stop();

import { LLMPipelineGenerator } from '@async-fusion/data';

const gen = new LLMPipelineGenerator({
  provider: 'openai', // 'gemini' | 'anthropic' | 'groq' | 'openrouter' | 'local'
  apiKey: process.env.OPENAI_KEY!,
  model: 'gpt-4o',
});

const result = await gen.generateFromDescription(
  `Create a pipeline that reads from Kafka topic 'user-clicks',
   filters clicks where amount > 50, groups by user_id every 5min.`
);

console.log(result.code);
console.log(result.transformations);

await result.pipeline?.run();
await gen.saveToFile(result.code, 'generated-pipeline.ts');

import {
  withRetry, CircuitBreaker, RetryableError, FatalError, sleep,
} from '@async-fusion/data';

// Retry with exponential back-off
const result = await withRetry(
  () => fetchFromExternalAPI(),
  { maxAttempts: 5, delayMs: 500, backoffMultiplier: 2 }
);

// Circuit Breaker
const cb = new CircuitBreaker({
  threshold: 5,         // open after 5 failures
  resetTimeout: 30_000, // retry after 30s
});
const data = await cb.exec(() => riskyOperation());

// Error classification
throw new RetryableError('Transient DB timeout'); // triggers retry
throw new FatalError('Invalid schema');            // goes to DLQ

await sleep(2000);

import type {
  PipelineConfig, PipelineOptions, RetryConfig,
  GeneratorConfig, GeneratedPipeline,
  DashboardMetrics, PipelineMetrics,
} from '@async-fusion/data';

interface PipelineConfig {
  name: string;
  checkpointLocation?: string;
  parallelism?: number;
}

interface RetryConfig {
  maxAttempts: number;
  delayMs: number;
  backoffMultiplier?: number;
}

interface GeneratorConfig {
  provider: 'openai' | 'gemini' | 'anthropic' | 'groq' | 'openrouter' | 'local';
  apiKey: string;
  model?: string;
  baseURL?: string;
}