> ## Documentation Index
> Fetch the complete documentation index at: https://ihub.learnille.com/llms.txt
> Use this file to discover all available pages before exploring further.

# OpenSearch Product Migration System

> Migrated from the previous Learnille internal documentation hub

## Overview

The OpenSearch migration system provides a **zero-downtime** approach to reindexing products (courses and consultations) into new index versions. It uses an **alias-based architecture** where applications query aliases instead of direct indices, allowing seamless index swaps.

## Architecture Components

### 1. **OpenSearchController** (`opensearch.controller.ts`)

The main API entry point for triggering migrations and reindexing operations.

**Key Endpoints:**

#### One-Click Product Migration

```http theme={null}
POST /elastic/products/migrate
Content-Type: application/json

{
  "type": "all",           // "all" | "course" | "consultation"
  "version": 2,            // Optional: auto-increments if omitted
  "batchSize": 100,        // Optional: default 100
  "updatedSince": "2024-01-01T00:00:00Z"  // Optional: only migrate updated items
}
```

**Response:**

```json theme={null}
{
  "message": "Product migration started for all targeting version 2",
  "jobId": "12345"
}
```

#### Other Useful Endpoints

* **`GET /opensearch/index-status`** - View current index versions and alias mappings
* **`POST /opensearch/reindex-batch`** - Batch reindex without creating new versions
* **`POST /opensearch/reindex-document`** - Reindex a single document by ID
* **`POST /opensearch/migrate/:index/:version`** - Migrate a specific index to a version
* **`POST /opensearch/drop-all`** - Development only: drop all indices

***

## Migration Process Flow

### **Step-by-Step: What Happens When You Click "Migrate"**

```mermaid theme={null}
sequenceDiagram
    participant Client
    participant Controller
    participant MigrationService
    participant Worker
    participant ElasticService
    participant OpenSearch

    Client->>Controller: POST /elastic/products/migrate
    Controller->>MigrationService: enqueueMigration(payload)
    MigrationService->>Worker: Add job to queue
    Controller-->>Client: Return jobId

    Note over Worker: Background Processing Starts

    Worker->>ElasticService: createNewIndexVersion(index, version)
    ElasticService->>OpenSearch: Create index (e.g., "course_v2")
    OpenSearch-->>ElasticService: Index created

    Worker->>Worker: Calculate batches (totalCount / batchSize)

    loop For each batch
        Worker->>Worker: Enqueue child reindex job
        Worker->>OpenSearch: Bulk index documents
    end

    Worker->>ElasticService: verifyReindex(oldIndex, newIndex)
    ElasticService->>OpenSearch: Count documents in both indices
    OpenSearch-->>ElasticService: Counts match ✓

    Worker->>ElasticService: switchAliasToNew(alias, newIndex)
    ElasticService->>OpenSearch: Update alias (course → course_v2)
    OpenSearch-->>ElasticService: Alias switched

    Worker->>Worker: Record metadata (version, metrics)
    Worker-->>Client: Migration complete
```

***

## Detailed Component Breakdown

### 2. **ProductIndexMigrationService** (`product-index-migration.service.ts`)

**Purpose:** Enqueues migration jobs into BullMQ queue.

**Key Method:**

```typescript theme={null}
async enqueueMigration(payload: ProductMigrationJobData) {
  const job = await this.migrationQueue.add(
    JobEnum.PRODUCT_INDEX_MIGRATION,
    payload
  );
  return { jobId: job.id };
}
```

***

### 3. **ProductMigrationWorker** (`product-migration.worker.ts`)

**Purpose:** Background worker that orchestrates the entire migration process.

**Key Responsibilities:**

1. **Version Resolution**
   ```typescript theme={null}
   // Auto-increment if version not specified
   const currentVersion = await this.elasticIndexMetadataService
     .getCurrentVersion(config.elasticIndex, environment);
   targetVersion = (currentVersion || 0) + 1;
   ```

2. **Create New Index**
   ```typescript theme={null}
   const creationResult = await this.elasticService
     .createNewIndexVersion(config.elasticIndex, version);
   // Creates: "course_v2", "consultation_v2", etc.
   ```

3. **Batch Processing**
   ```typescript theme={null}
   const totalCount = await repository.count({ where: whereClause });
   const totalPages = Math.ceil(totalCount / batchSize);

   // Dispatch child jobs for parallel processing
   for (let page = 0; page < totalPages; page++) {
     await this.reindexQueue.add(JobEnum.REINDEX_PRODUCT_PAGE, {
       type: 'courses',
       page,
       batchSize,
       targetIndex: 'course_v2'
     });
   }
   ```

4. **Verification**
   ```typescript theme={null}
   const verification = await this.elasticService
     .verifyReindex(sourceIndex, targetIndex);

   if (!verification.matches) {
     throw new Error('Reindex verification failed');
   }
   ```

5. **Alias Switch (Zero-Downtime)**
   ```typescript theme={null}
   await this.elasticService.switchAliasToNew(aliasName, targetIndex, {
     logicalIndex: config.elasticIndex,
     version,
     metrics: {
       sourceCount: verification.sourceCount,
       destinationCount: verification.destinationCount
     }
   });
   ```

***

### 4. **ElasticService** (`elastic.service.ts`)

**Core Methods:**

#### `createNewIndexVersion(index, version)`

Creates a new versioned index with proper mappings.

```typescript theme={null}
// Example: Creates "course_v2" with course mappings
const newIndex = `${index}_v${version}`;
await this.esService.indices.create({
  index: newIndex,
  ...mapping
});
```

#### `switchAliasToNew(alias, newIndex, context)`

Atomically switches alias from old index to new index.

```typescript theme={null}
// Before: "course" alias → "course_v1"
// After:  "course" alias → "course_v2"

await this.esService.indices.updateAliases({
  body: {
    actions: [
      { remove: { index: currentIndex, alias: alias } },
      { add: { index: newIndex, alias: alias } }
    ]
  }
});
```

#### `verifyReindex(sourceIndex, destIndex)`

Ensures document counts match between old and new indices.

```typescript theme={null}
const sourceCount = await this.esService.count({ index: sourceIndex });
const destCount = await this.esService.count({ index: destIndex });
return {
  matches: sourceCount.count === destCount.count,
  sourceCount: sourceCount.count,
  destinationCount: destCount.count
};
```

***

## Index Naming Convention

| Logical Index | Alias | Physical Indices |
| - | - | - |
| `course` | `course` | `course_v1`, `course_v2`, `course_v3` |
| `consultation` | `consultation` | `consultation_v1`, `consultation_v2` |

**Application Code:** Always queries the **alias** (e.g., `course`), never the versioned index directly.

***

## Migration Strategies

### **Strategy 1: Full Migration (All Products)**

```bash theme={null}
curl -X POST http://localhost:3000/elastic/products/migrate \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "type": "all"
  }'
```

**What happens:**

* Auto-increments version for both courses and consultations
* Creates `course_v2` and `consultation_v2`
* Reindexes all documents
* Switches aliases atomically

***

### **Strategy 2: Incremental Migration (Updated Products Only)**

```bash theme={null}
curl -X POST http://localhost:3000/elastic/products/migrate \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "type": "all",
    "updatedSince": "2024-12-01T00:00:00Z"
  }'
```

**What happens:**

* Only migrates products updated after the specified date
* Useful for quick updates without full reindex

***

### **Strategy 3: Single Product Type**

```bash theme={null}
curl -X POST http://localhost:3000/elastic/products/migrate \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -d '{
    "type": "course",
    "version": 3
  }'
```

**What happens:**

* Only migrates courses to version 3
* Consultations remain unchanged

***

## Monitoring Migration Progress

### Check Index Status

```bash theme={null}
curl -X GET http://localhost:3000/elastic/index-status \
  -H "Authorization: Bearer YOUR_TOKEN"
```

**Response:**

```json theme={null}
{
  "course": {
    "currentVersion": 2,
    "activeIndex": "course_v2",
    "alias": "course",
    "lastMigration": "2024-12-10T15:30:00Z"
  },
  "consultation": {
    "currentVersion": 1,
    "activeIndex": "consultation_v1",
    "alias": "consultation"
  }
}
```

***

## Safety Features

### 1. **Verification Before Alias Switch**

The system verifies document counts match before switching aliases:

```typescript theme={null}
if (!verification.matches) {
  throw new Error('Reindex verification failed');
}
```

### 2. **Atomic Alias Updates**

OpenSearch's `updateAliases` API ensures atomic operations—no downtime.

### 3. **Job Tracking**

All migrations are tracked via BullMQ jobs with progress updates and logs.

### 4. **Metadata Tracking**

The `ElasticIndexMetadataService` records:

* Current active version per environment
* Migration timestamps
* Document counts
* Job IDs

***

## Common Use Cases

### **Use Case 1: Schema Change**

You updated the OpenSearch mapping for courses.

**Solution:**

```bash theme={null}
# Trigger migration to new version with updated mapping
POST /elastic/products/migrate
{
  "type": "course"
}
```

### **Use Case 2: Data Corruption**

Some documents are corrupted in the current index.

**Solution:**

```bash theme={null}
# Rebuild entire index from database
POST /elastic/products/migrate
{
  "type": "all"
}
```

### **Use Case 3: Performance Optimization**

You want to reindex with better settings (shards, replicas).

**Solution:**

1. Update mapping in code
2. Trigger migration
3. Old index remains until you manually delete it

***

## Configuration

### Alias Mapping (Config)

```typescript theme={null}
// config/default.ts
elasticAliasMap: {
  'course': 'course',
  'consultation': 'consultation',
  'instructor': 'instructor',
  'consultant': 'consultant'
}
```

### Environment Variables

```bash theme={null}
OPENSEARCH_HOST=https://your-aiven-opensearch-endpoint
OPENSEARCH_USERNAME=your-username
OPENSEARCH_PASSWORD=your-password
NODE_ENV=development
```

***

## Best Practices

1. **Always Use Aliases in Application Code**
   ```typescript theme={null}
   // ✅ Good
   await elasticService.search('course', query);

   // ❌ Bad
   await elasticService.search('course_v2', query);
   ```

2. **Monitor Job Progress**
   Use BullMQ dashboard or logs to track migration jobs.

3. **Test in Staging First**
   Always test migrations in staging before production.

4. **Clean Up Old Indices**
   After verifying new index works, manually delete old versions:
   ```bash theme={null}
   DELETE /course_v1
   ```

5. **Use Incremental Updates for Large Datasets**
   For millions of documents, use `updatedSince` to avoid full reindex.

***

## Troubleshooting

### Issue: "Reindex verification failed"

**Cause:** Document counts don't match between old and new indices.

**Solution:**

* Check for errors in child reindex jobs
* Verify database queries aren't filtered incorrectly
* Retry migration

### Issue: "Index already exists"

**Cause:** Trying to create a version that already exists.

**Solution:**

* Increment version number manually
* Or delete the existing versioned index first

### Issue: Migration stuck

**Cause:** Child jobs failing or queue issues.

**Solution:**

* Check BullMQ queue health
* Review worker logs
* Restart workers if needed

***

## Summary: One-Click Migration Flow

1. **Client** sends `POST /elastic/products/migrate { "type": "all" }`
2. **Controller** enqueues job and returns `jobId`
3. **Worker** creates new versioned indices (`course_v2`, `consultation_v2`)
4. **Worker** dispatches batch reindex jobs (parallel processing)
5. **Worker** waits for all batches to complete
6. **Worker** verifies document counts match
7. **Worker** atomically switches aliases to new indices
8. **Worker** records metadata and completes job
9. **Application** continues querying aliases with zero downtime

**Result:** Products are now in new indices with updated mappings/data, and users experienced no downtime! 🎉


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.