# SkillPatch skill: azure-eventhub-ts

This skill guides agents in building event streaming applications using the Azure Event Hubs SDK for TypeScript/JavaScript (@azure/event-hubs). It covers installation, authentication via Azure Identity, sending events in batches, targeting specific partitions, and consuming events with and without checkpointing using Azure Blob Storage. Ideal for high-throughput ingestion, IoT telemetry, and event-driven architectures.

You (the agent) were given this URL and asked to use this skill. This is a **public** skill — no SkillPatch account, API key, or prior setup is required. Two ways to use it:

**1. Use it right now** — the skill's complete file tree (SKILL.md + all reference files) is inlined below; read `SKILL.md` first, then follow it, consulting the other files as it directs.

**2. Install the exact package onto disk** (recommended if you can run a shell — this reproduces the skill byte-for-byte, including any binary assets that can't be inlined):

```bash
mkdir -p .claude/skills/azure-eventhub-ts
curl -sSL https://skillpatch.dev/install_skill/azure-eventhub-ts | tar -xz -C .claude/skills/
```

(`.claude/skills/` is Claude Code's convention; use whatever directory your agent loads skills from.)


---

## Skill files (3)

- `SKILL.md`
- `references/checkpointing.md`
- `references/event-processing.md`


### `SKILL.md`

````markdown
---
name: azure-eventhub-ts
description: Build event streaming applications using Azure Event Hubs SDK for JavaScript (@azure/event-hubs). Use when implementing high-throughput event ingestion, real-time analytics, IoT telemetry, or event-driven architectures with partitioned consumers.
license: MIT
metadata:
  author: Microsoft
  version: "1.0.0"
  package: '@azure/event-hubs'
---

# Azure Event Hubs SDK for TypeScript

High-throughput event streaming and real-time data ingestion.

## Installation

```bash
npm install @azure/event-hubs @azure/identity
```

For checkpointing with consumer groups:
```bash
npm install @azure/eventhubs-checkpointstore-blob @azure/storage-blob
```

## Environment Variables

```bash
EVENTHUB_NAMESPACE=<namespace>.servicebus.windows.net
EVENTHUB_NAME=my-eventhub
STORAGE_ACCOUNT_NAME=<storage-account>
STORAGE_CONTAINER_NAME=checkpoints
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
```

## Authentication

```typescript
import { EventHubProducerClient, EventHubConsumerClient } from "@azure/event-hubs";
import { DefaultAzureCredential, ManagedIdentityCredential } from "@azure/identity";

const fullyQualifiedNamespace = process.env.EVENTHUB_NAMESPACE!;
const eventHubName = process.env.EVENTHUB_NAME!;
// Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
const credential = new DefaultAzureCredential({requiredEnvVars: ["AZURE_TOKEN_CREDENTIALS"]});
// Or use a specific credential directly in production:
// See https://learn.microsoft.com/javascript/api/overview/azure/identity-readme?view=azure-node-latest#credential-classes
// const credential = new ManagedIdentityCredential();

// Producer
const producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential);

// Consumer
const consumer = new EventHubConsumerClient(
  "$Default", // Consumer group
  fullyQualifiedNamespace,
  eventHubName,
  credential
);
```

## Core Workflow

### Send Events

```typescript
const producer = new EventHubProducerClient(namespace, eventHubName, credential);

// Create batch and add events
const batch = await producer.createBatch();
batch.tryAdd({ body: { temperature: 72.5, deviceId: "sensor-1" } });
batch.tryAdd({ body: { temperature: 68.2, deviceId: "sensor-2" } });

await producer.sendBatch(batch);
await producer.close();
```

### Send to Specific Partition

```typescript
// By partition ID
const batch = await producer.createBatch({ partitionId: "0" });

// By partition key (consistent hashing)
const batch = await producer.createBatch({ partitionKey: "device-123" });
```

### Receive Events (Simple)

```typescript
const consumer = new EventHubConsumerClient("$Default", namespace, eventHubName, credential);

const subscription = consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Partition: ${context.partitionId}, Body: ${JSON.stringify(event.body)}`);
    }
  },
  processError: async (err, context) => {
    console.error(`Error on partition ${context.partitionId}: ${err.message}`);
  },
});

// Stop after some time
setTimeout(async () => {
  await subscription.close();
  await consumer.close();
}, 60000);
```

### Receive with Checkpointing (Production)

```typescript
import { EventHubConsumerClient } from "@azure/event-hubs";
import { ContainerClient } from "@azure/storage-blob";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

const containerClient = new ContainerClient(
  `https://${storageAccount}.blob.core.windows.net/${containerName}`,
  credential
);

const checkpointStore = new BlobCheckpointStore(containerClient);

const consumer = new EventHubConsumerClient(
  "$Default",
  namespace,
  eventHubName,
  credential,
  checkpointStore
);

const subscription = consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Processing: ${JSON.stringify(event.body)}`);
    }
    // Checkpoint after processing batch
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(`Error: ${err.message}`);
  },
});
```

### Receive from Specific Position

```typescript
const subscription = consumer.subscribe({
  processEvents: async (events, context) => { /* ... */ },
  processError: async (err, context) => { /* ... */ },
}, {
  startPosition: {
    // Start from beginning
    "0": { offset: "@earliest" },
    // Start from end (new events only)
    "1": { offset: "@latest" },
    // Start from specific offset
    "2": { offset: "12345" },
    // Start from specific time
    "3": { enqueuedOn: new Date("2024-01-01") },
  },
});
```

## Event Hub Properties

```typescript
// Get hub info
const hubProperties = await producer.getEventHubProperties();
console.log(`Partitions: ${hubProperties.partitionIds}`);

// Get partition info
const partitionProperties = await producer.getPartitionProperties("0");
console.log(`Last sequence: ${partitionProperties.lastEnqueuedSequenceNumber}`);
```

## Batch Processing Options

```typescript
const subscription = consumer.subscribe(
  {
    processEvents: async (events, context) => { /* ... */ },
    processError: async (err, context) => { /* ... */ },
  },
  {
    maxBatchSize: 100,           // Max events per batch
    maxWaitTimeInSeconds: 30,    // Max wait for batch
  }
);
```

## Key Types

```typescript
import {
  EventHubProducerClient,
  EventHubConsumerClient,
  EventData,
  ReceivedEventData,
  PartitionContext,
  Subscription,
  SubscriptionEventHandlers,
  CreateBatchOptions,
  EventPosition,
} from "@azure/event-hubs";

import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";
```

## Event Properties

```typescript
// Send with properties
const batch = await producer.createBatch();
batch.tryAdd({
  body: { data: "payload" },
  properties: {
    eventType: "telemetry",
    deviceId: "sensor-1",
  },
  contentType: "application/json",
  correlationId: "request-123",
});

// Access in receiver
consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Type: ${event.properties?.eventType}`);
      console.log(`Sequence: ${event.sequenceNumber}`);
      console.log(`Enqueued: ${event.enqueuedTimeUtc}`);
      console.log(`Offset: ${event.offset}`);
    }
  },
});
```

## Error Handling

```typescript
consumer.subscribe({
  processEvents: async (events, context) => {
    try {
      for (const event of events) {
        await processEvent(event);
      }
      await context.updateCheckpoint(events[events.length - 1]);
    } catch (error) {
      // Don't checkpoint on error - events will be reprocessed
      console.error("Processing failed:", error);
    }
  },
  processError: async (err, context) => {
    if (err.name === "MessagingError") {
      // Transient error - SDK will retry
      console.warn("Transient error:", err.message);
    } else {
      // Fatal error
      console.error("Fatal error:", err);
    }
  },
});
```

## Best Practices

1. **Use checkpointing** - Always checkpoint in production for exactly-once processing
2. **Batch sends** - Use `createBatch()` for efficient sending
3. **Partition keys** - Use partition keys to ensure ordering for related events
4. **Consumer groups** - Use separate consumer groups for different processing pipelines
5. **Handle errors gracefully** - Don't checkpoint on processing failures
6. **Close clients** - Always close producer/consumer when done
7. **Monitor lag** - Track `lastEnqueuedSequenceNumber` vs processed sequence

````


### `references/checkpointing.md`

````markdown
# Checkpointing Reference

Persistent checkpointing with BlobCheckpointStore for Azure Event Hubs consumer applications.

## Overview

Checkpointing tracks the last successfully processed event position per partition. This enables:
- **Resumption** — Continue from where you left off after restart
- **Load balancing** — Distribute partitions across multiple consumers
- **Exactly-once processing** — When combined with idempotent downstream operations

## Installation

```bash
npm install @azure/event-hubs @azure/eventhubs-checkpointstore-blob @azure/storage-blob @azure/identity
```

## Key Interfaces

```typescript
import { CheckpointStore, Checkpoint, PartitionOwnership } from "@azure/event-hubs";

// CheckpointStore interface (implemented by BlobCheckpointStore)
interface CheckpointStore {
  listCheckpoints(
    fullyQualifiedNamespace: string,
    eventHubName: string,
    consumerGroup: string
  ): Promise<Checkpoint[]>;

  updateCheckpoint(checkpoint: Checkpoint): Promise<void>;

  listOwnership(
    fullyQualifiedNamespace: string,
    eventHubName: string,
    consumerGroup: string
  ): Promise<PartitionOwnership[]>;

  claimOwnership(
    partitionOwnership: PartitionOwnership[]
  ): Promise<PartitionOwnership[]>;
}

// Checkpoint structure
interface Checkpoint {
  fullyQualifiedNamespace: string;
  eventHubName: string;
  consumerGroup: string;
  partitionId: string;
  sequenceNumber: number;
  offset: string;
}

// Partition ownership (for load balancing)
interface PartitionOwnership {
  fullyQualifiedNamespace: string;
  eventHubName: string;
  consumerGroup: string;
  partitionId: string;
  ownerId: string;
  lastModifiedTimeInMs?: number;
  etag?: string;
}
```

## Basic Checkpointing Setup

```typescript
import { EventHubConsumerClient } from "@azure/event-hubs";
import { ContainerClient } from "@azure/storage-blob";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";
import { DefaultAzureCredential } from "@azure/identity";

const credential = new DefaultAzureCredential();

// Storage for checkpoints
const storageAccountName = process.env.STORAGE_ACCOUNT_NAME!;
const containerName = "eventhub-checkpoints";

const containerClient = new ContainerClient(
  `https://${storageAccountName}.blob.core.windows.net/${containerName}`,
  credential
);

// Ensure container exists
await containerClient.createIfNotExists();

// Create checkpoint store
const checkpointStore = new BlobCheckpointStore(containerClient);

// Create consumer with checkpoint store
const consumerClient = new EventHubConsumerClient(
  "$Default",
  "my-namespace.servicebus.windows.net",
  "my-event-hub",
  credential,
  checkpointStore  // Pass checkpoint store here
);

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
    }
    
    // Checkpoint after successful processing
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(`Error: ${err.message}`);
  }
});
```

## Connection String Authentication

```typescript
import { EventHubConsumerClient } from "@azure/event-hubs";
import { ContainerClient } from "@azure/storage-blob";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

// Storage connection string
const storageConnectionString = process.env.STORAGE_CONNECTION_STRING!;
const containerName = "checkpoints";

const containerClient = new ContainerClient(
  storageConnectionString,
  containerName
);

await containerClient.createIfNotExists();

const checkpointStore = new BlobCheckpointStore(containerClient);

// Event Hub connection string
const eventHubConnectionString = process.env.EVENTHUB_CONNECTION_STRING!;
const eventHubName = process.env.EVENTHUB_NAME!;

const consumerClient = new EventHubConsumerClient(
  "$Default",
  eventHubConnectionString,
  eventHubName,
  checkpointStore
);
```

## Checkpointing Strategies

### Checkpoint Every Batch (Default)

```typescript
const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
    }
    
    // Checkpoint after every batch
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
```

### Checkpoint Every N Events

```typescript
let processedCount = 0;
const checkpointInterval = 100;

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
      processedCount++;
      
      // Checkpoint every N events
      if (processedCount % checkpointInterval === 0) {
        await context.updateCheckpoint(event);
        console.log(`Checkpointed at ${event.sequenceNumber}`);
      }
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
```

### Checkpoint on Time Interval

```typescript
let lastCheckpointTime = Date.now();
let lastEvent: ReceivedEventData | null = null;
const checkpointIntervalMs = 30000; // 30 seconds

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
      lastEvent = event;
    }
    
    // Checkpoint based on time
    const now = Date.now();
    if (lastEvent && (now - lastCheckpointTime) >= checkpointIntervalMs) {
      await context.updateCheckpoint(lastEvent);
      lastCheckpointTime = now;
      console.log(`Time-based checkpoint at ${lastEvent.sequenceNumber}`);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
```

### Checkpoint Only on Success

```typescript
const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    const results = await Promise.allSettled(
      events.map(event => processEvent(event))
    );
    
    // Find last successfully processed event
    let lastSuccessIndex = -1;
    for (let i = results.length - 1; i >= 0; i--) {
      if (results[i].status === "fulfilled") {
        lastSuccessIndex = i;
        break;
      }
    }
    
    // Only checkpoint up to last success
    if (lastSuccessIndex >= 0) {
      await context.updateCheckpoint(events[lastSuccessIndex]);
    }
    
    // Log failures
    const failures = results.filter(r => r.status === "rejected");
    if (failures.length > 0) {
      console.error(`${failures.length} events failed processing`);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
```

## Load Balancing Across Consumers

Multiple consumers with the same consumer group automatically balance partitions:

```typescript
// Consumer Instance 1
const consumer1 = new EventHubConsumerClient(
  "my-consumer-group",
  namespace,
  eventHubName,
  credential,
  checkpointStore
);

// Consumer Instance 2 (separate process)
const consumer2 = new EventHubConsumerClient(
  "my-consumer-group",  // Same consumer group
  namespace,
  eventHubName,
  credential,
  checkpointStore       // Same checkpoint store
);

// Both subscribe - partitions automatically distributed
consumer1.subscribe({ /* handlers */ });
consumer2.subscribe({ /* handlers */ });
```

## Checkpoint Store Blob Structure

The BlobCheckpointStore creates blobs in this structure:

```
<container>/
├── <namespace>/<eventhub>/<consumergroup>/checkpoint/
│   ├── 0    (checkpoint for partition 0)
│   ├── 1    (checkpoint for partition 1)
│   └── ...
└── <namespace>/<eventhub>/<consumergroup>/ownership/
    ├── 0    (ownership for partition 0)
    ├── 1    (ownership for partition 1)
    └── ...
```

## Inspecting Checkpoints

```typescript
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

const checkpointStore = new BlobCheckpointStore(containerClient);

// List all checkpoints
const checkpoints = await checkpointStore.listCheckpoints(
  "my-namespace.servicebus.windows.net",
  "my-event-hub",
  "$Default"
);

for (const cp of checkpoints) {
  console.log(`Partition ${cp.partitionId}:`);
  console.log(`  Sequence: ${cp.sequenceNumber}`);
  console.log(`  Offset: ${cp.offset}`);
}

// List partition ownership
const ownerships = await checkpointStore.listOwnership(
  "my-namespace.servicebus.windows.net",
  "my-event-hub",
  "$Default"
);

for (const own of ownerships) {
  console.log(`Partition ${own.partitionId}: owned by ${own.ownerId}`);
}
```

## Error Handling

```typescript
const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    try {
      for (const event of events) {
        await processEvent(event);
      }
      
      if (events.length > 0) {
        await context.updateCheckpoint(events[events.length - 1]);
      }
    } catch (processingError) {
      // Don't checkpoint on error - events will be reprocessed
      console.error("Processing failed:", processingError);
      // Optionally: send to dead letter, alert, etc.
    }
  },
  
  processError: async (err, context) => {
    // SDK-level errors (connection, auth, etc.)
    console.error(`SDK error on partition ${context.partitionId}:`, err.message);
    
    // The SDK handles reconnection automatically
    // Consider alerting for persistent errors
  }
});
```

## Graceful Shutdown with Final Checkpoint

```typescript
let lastProcessedEvent: Map<string, ReceivedEventData> = new Map();

const subscription = consumerClient.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEvent(event);
      lastProcessedEvent.set(context.partitionId, event);
    }
    
    // Regular checkpointing
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});

// Graceful shutdown
process.on("SIGTERM", async () => {
  console.log("Shutting down gracefully...");
  
  // Close subscription (stops receiving)
  await subscription.close();
  
  // Close consumer client
  await consumerClient.close();
  
  console.log("Shutdown complete");
  process.exit(0);
});
```

## Best Practices

1. **Checkpoint after processing** — Never before, to avoid data loss
2. **Balance checkpoint frequency** — Too often = high storage costs; too rare = more reprocessing on restart
3. **Use same checkpoint store** — All consumers in a group must share the same store
4. **Create container before use** — Call `createIfNotExists()` on startup
5. **Handle checkpoint failures** — Log and alert if `updateCheckpoint` fails repeatedly
6. **Use consumer groups** — Separate groups for different processing pipelines
7. **Design for reprocessing** — Make downstream operations idempotent

## Checkpoint Frequency Trade-offs

| Frequency | Pros | Cons |
|-----------|------|------|
| Every event | Minimal reprocessing | High storage I/O |
| Every batch | Good balance | Some reprocessing possible |
| Time-based | Predictable I/O | Variable reprocessing |
| Every N events | Controlled I/O | May miss some on crash |

## See Also

- [Event Processing Reference](./event-processing.md)
- [Azure Blob Storage](https://learn.microsoft.com/azure/storage/blobs/)
- [Event Hubs Consumer Groups](https://learn.microsoft.com/azure/event-hubs/event-hubs-features#consumer-groups)

````


### `references/event-processing.md`

````markdown
# Event Processing Reference

Receiving and processing events with Azure Event Hubs using the @azure/event-hubs TypeScript SDK.

## Overview

Event Hubs provides high-throughput event streaming. This reference covers the EventHubConsumerClient, subscription patterns, event handlers, and processing strategies.

## Key Interfaces

```typescript
import {
  EventHubConsumerClient,
  ReceivedEventData,
  PartitionContext,
  SubscriptionEventHandlers,
  SubscribeOptions,
  Subscription
} from "@azure/event-hubs";

// ReceivedEventData - structure of received events
interface ReceivedEventData {
  body: any;
  contentType?: string;
  correlationId?: string | number | Buffer;
  enqueuedTimeUtc: Date;
  messageId?: string | number | Buffer;
  offset: string;
  partitionKey: string | null;
  properties?: Record<string, any>;
  sequenceNumber: number;
  systemProperties?: Record<string, any>;
}

// PartitionContext - context for event handlers
interface PartitionContext {
  readonly consumerGroup: string;
  readonly eventHubName: string;
  readonly fullyQualifiedNamespace: string;
  readonly partitionId: string;
  readonly lastEnqueuedEventProperties?: LastEnqueuedEventProperties;
  updateCheckpoint(eventData: ReceivedEventData): Promise<void>;
}

// Handler types
type ProcessEventsHandler = (
  events: ReceivedEventData[],
  context: PartitionContext
) => Promise<void>;

type ProcessErrorHandler = (
  error: Error,
  context: PartitionContext
) => Promise<void>;

type ProcessInitializeHandler = (
  context: PartitionContext
) => Promise<void>;

type ProcessCloseHandler = (
  reason: CloseReason,
  context: PartitionContext
) => Promise<void>;
```

## Basic Subscribe Pattern

```typescript
import { EventHubConsumerClient, earliestEventPosition } from "@azure/event-hubs";
import { DefaultAzureCredential } from "@azure/identity";

const fullyQualifiedNamespace = "my-namespace.servicebus.windows.net";
const eventHubName = "my-event-hub";
const consumerGroup = "$Default";

const client = new EventHubConsumerClient(
  consumerGroup,
  fullyQualifiedNamespace,
  eventHubName,
  new DefaultAzureCredential()
);

const subscription = client.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Partition: ${context.partitionId}`);
      console.log(`Body: ${JSON.stringify(event.body)}`);
      console.log(`Sequence: ${event.sequenceNumber}`);
      console.log(`Enqueued: ${event.enqueuedTimeUtc}`);
    }
  },
  processError: async (err, context) => {
    console.error(`Error on partition ${context.partitionId}: ${err.message}`);
  }
}, {
  startPosition: earliestEventPosition
});

// Stop after some time
setTimeout(async () => {
  await subscription.close();
  await client.close();
}, 60000);
```

## Full Subscription with All Handlers

```typescript
import {
  EventHubConsumerClient,
  ReceivedEventData,
  PartitionContext,
  CloseReason,
  earliestEventPosition,
  latestEventPosition
} from "@azure/event-hubs";

const client = new EventHubConsumerClient(
  "$Default",
  connectionString,
  eventHubName
);

const subscription = client.subscribe({
  processInitialize: async (context: PartitionContext) => {
    console.log(`Started receiving from partition ${context.partitionId}`);
  },

  processEvents: async (events: ReceivedEventData[], context: PartitionContext) => {
    if (events.length === 0) {
      console.log(`No events received. Waiting...`);
      return;
    }

    console.log(`Received ${events.length} events from partition ${context.partitionId}`);
    
    for (const event of events) {
      // Process each event
      await processEvent(event, context);
    }

    // Checkpoint after processing batch
    const lastEvent = events[events.length - 1];
    await context.updateCheckpoint(lastEvent);
  },

  processError: async (err: Error, context: PartitionContext) => {
    console.error(`Error on partition ${context.partitionId}:`);
    console.error(err.message);
  },

  processClose: async (reason: CloseReason, context: PartitionContext) => {
    console.log(`Stopped receiving from partition ${context.partitionId}`);
    console.log(`Reason: ${reason}`);
  }
}, {
  startPosition: earliestEventPosition,
  maxBatchSize: 100,
  maxWaitTimeInSeconds: 30
});
```

## Subscribe to Specific Partition

```typescript
const client = new EventHubConsumerClient(
  "$Default",
  connectionString,
  eventHubName
);

// Get partition IDs
const partitionIds = await client.getPartitionIds();
console.log(`Partitions: ${partitionIds.join(", ")}`);

// Subscribe to specific partition
const subscription = client.subscribe(
  partitionIds[0], // First partition only
  {
    processEvents: async (events, context) => {
      console.log(`Partition ${context.partitionId}: ${events.length} events`);
    },
    processError: async (err, context) => {
      console.error(err);
    }
  },
  { startPosition: earliestEventPosition }
);
```

## Start Position Options

```typescript
import { 
  earliestEventPosition, 
  latestEventPosition,
  EventPosition 
} from "@azure/event-hubs";

// Start from beginning (all historical events)
const fromBeginning = { startPosition: earliestEventPosition };

// Start from end (new events only)
const fromEnd = { startPosition: latestEventPosition };

// Start from specific offset
const fromOffset: SubscribeOptions = {
  startPosition: { offset: "12345" }
};

// Start from specific sequence number
const fromSequence: SubscribeOptions = {
  startPosition: { sequenceNumber: 1000 }
};

// Start from specific time
const fromTime: SubscribeOptions = {
  startPosition: { enqueuedOn: new Date("2024-01-01T00:00:00Z") }
};

// Different positions per partition
const perPartition: SubscribeOptions = {
  startPosition: {
    "0": earliestEventPosition,
    "1": latestEventPosition,
    "2": { offset: "5000" }
  }
};
```

## SubscribeOptions Reference

```typescript
interface SubscribeOptions {
  /** Starting position for reading events */
  startPosition?: EventPosition | Record<string, EventPosition>;
  
  /** Max events per batch (default: varies) */
  maxBatchSize?: number;
  
  /** Max wait time in seconds for a batch */
  maxWaitTimeInSeconds?: number;
  
  /** Track last enqueued event info (for lag monitoring) */
  trackLastEnqueuedEventProperties?: boolean;
  
  /** Owner level for exclusive access */
  ownerLevel?: number;
}
```

## Event Processing Patterns

### Sequential Processing

```typescript
const subscription = client.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      await processEventSequentially(event);
    }
    await context.updateCheckpoint(events[events.length - 1]);
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
```

### Parallel Processing within Batch

```typescript
const subscription = client.subscribe({
  processEvents: async (events, context) => {
    // Process events in parallel
    await Promise.all(
      events.map(event => processEventAsync(event))
    );
    
    // Checkpoint after all complete
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
```

### With Error Handling per Event

```typescript
const subscription = client.subscribe({
  processEvents: async (events, context) => {
    const results = await Promise.allSettled(
      events.map(event => processEvent(event))
    );
    
    const failures = results.filter(r => r.status === "rejected");
    if (failures.length > 0) {
      console.error(`${failures.length} events failed processing`);
      // Decide: checkpoint anyway or skip?
    }
    
    // Checkpoint even with some failures (at-least-once)
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
```

## Monitoring Consumer Lag

```typescript
const subscription = client.subscribe({
  processEvents: async (events, context) => {
    // Check lag if tracking enabled
    if (context.lastEnqueuedEventProperties) {
      const lastEnqueued = context.lastEnqueuedEventProperties.sequenceNumber;
      const lastProcessed = events.length > 0 
        ? events[events.length - 1].sequenceNumber 
        : 0;
      
      const lag = lastEnqueued - lastProcessed;
      console.log(`Partition ${context.partitionId} lag: ${lag} events`);
    }
    
    // Process events...
  },
  processError: async (err, context) => {
    console.error(err);
  }
}, {
  trackLastEnqueuedEventProperties: true
});
```

## Event Properties Access

```typescript
const subscription = client.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      // Standard properties
      console.log(`Body: ${event.body}`);
      console.log(`Partition Key: ${event.partitionKey}`);
      console.log(`Sequence Number: ${event.sequenceNumber}`);
      console.log(`Offset: ${event.offset}`);
      console.log(`Enqueued Time: ${event.enqueuedTimeUtc}`);
      
      // Custom properties (set by producer)
      if (event.properties) {
        console.log(`Event Type: ${event.properties.eventType}`);
        console.log(`Device ID: ${event.properties.deviceId}`);
      }
      
      // System properties
      if (event.systemProperties) {
        console.log(`System Props: ${JSON.stringify(event.systemProperties)}`);
      }
      
      // Content type
      if (event.contentType) {
        console.log(`Content Type: ${event.contentType}`);
      }
    }
  },
  processError: async (err, context) => {
    console.error(err);
  }
});
```

## Graceful Shutdown

```typescript
let subscription: Subscription;

async function start() {
  const client = new EventHubConsumerClient(
    "$Default",
    connectionString,
    eventHubName
  );

  subscription = client.subscribe({
    processEvents: async (events, context) => {
      // Process events...
    },
    processError: async (err, context) => {
      console.error(err);
    }
  });

  // Handle shutdown signals
  process.on("SIGINT", async () => {
    console.log("Shutting down...");
    await subscription.close();
    await client.close();
    process.exit(0);
  });

  process.on("SIGTERM", async () => {
    console.log("Shutting down...");
    await subscription.close();
    await client.close();
    process.exit(0);
  });
}

start().catch(console.error);
```

## Best Practices

1. **Always handle empty batches** — `processEvents` may receive empty arrays
2. **Checkpoint after processing** — Not before, to avoid data loss
3. **Use consumer groups** — Separate groups for different processing pipelines
4. **Monitor lag** — Enable `trackLastEnqueuedEventProperties`
5. **Handle errors gracefully** — Don't crash on individual event failures
6. **Close resources** — Always close subscription and client on shutdown
7. **Set appropriate batch size** — Balance throughput vs latency

## See Also

- [Checkpointing Reference](./checkpointing.md)
- [Azure Event Hubs Documentation](https://learn.microsoft.com/azure/event-hubs/)
- [Event Hubs Quotas](https://learn.microsoft.com/azure/event-hubs/event-hubs-quotas)

````