OpenSearch Product Migration System

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

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:

{
  "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"

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:

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:

Version Resolution

// Auto-increment if version not specified
const currentVersion = await this.elasticIndexMetadataService
  .getCurrentVersion(config.elasticIndex, environment);
targetVersion = (currentVersion || 0) + 1;

Create New Index

const creationResult = await this.elasticService
  .createNewIndexVersion(config.elasticIndex, version);
// Creates: "course_v2", "consultation_v2", etc.

Batch Processing

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'
  });
}

Verification

const verification = await this.elasticService
  .verifyReindex(sourceIndex, targetIndex);

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

Alias Switch (Zero-Downtime)

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.

// 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.

// 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.

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)

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)

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

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

curl -X GET http://localhost:3000/elastic/index-status \
  -H "Authorization: Bearer YOUR_TOKEN"

Response:

{
  "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:

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:

# 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:

# 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:

Update mapping in code
Trigger migration
Old index remains until you manually delete it

Configuration

Alias Mapping (Config)

// config/default.ts
elasticAliasMap: {
  'course': 'course',
  'consultation': 'consultation',
  'instructor': 'instructor',
  'consultant': 'consultant'
}

Environment Variables

OPENSEARCH_HOST=https://your-aiven-opensearch-endpoint
OPENSEARCH_USERNAME=your-username
OPENSEARCH_PASSWORD=your-password
NODE_ENV=development

Best Practices

Always Use Aliases in Application Code

// ✅ Good
await elasticService.search('course', query);

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

Monitor Job Progress Use BullMQ dashboard or logs to track migration jobs.
Test in Staging First Always test migrations in staging before production.
Clean Up Old Indices After verifying new index works, manually delete old versions:
```
DELETE /course_v1
```
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

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

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

Overview​

Architecture Components​

1. OpenSearchController (opensearch.controller.ts)​

One-Click Product Migration​

Other Useful Endpoints​

Migration Process Flow​

Step-by-Step: What Happens When You Click "Migrate"​

Detailed Component Breakdown​

2. ProductIndexMigrationService (product-index-migration.service.ts)​

3. ProductMigrationWorker (product-migration.worker.ts)​

4. ElasticService (elastic.service.ts)​

createNewIndexVersion(index, version)​

switchAliasToNew(alias, newIndex, context)​

verifyReindex(sourceIndex, destIndex)​

Index Naming Convention​

Migration Strategies​

Strategy 1: Full Migration (All Products)​

Strategy 2: Incremental Migration (Updated Products Only)​

Strategy 3: Single Product Type​

Monitoring Migration Progress​

Check Index Status​

Safety Features​

1. Verification Before Alias Switch​

2. Atomic Alias Updates​

3. Job Tracking​

4. Metadata Tracking​

Common Use Cases​

Use Case 1: Schema Change​

Use Case 2: Data Corruption​

Use Case 3: Performance Optimization​

Configuration​

Alias Mapping (Config)​

Environment Variables​

Best Practices​

Troubleshooting​

Issue: "Reindex verification failed"​

Issue: "Index already exists"​

Issue: Migration stuck​

Summary: One-Click Migration Flow​

Overview

Architecture Components

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

One-Click Product Migration

Other Useful Endpoints

Migration Process Flow

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

Detailed Component Breakdown

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

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

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

`createNewIndexVersion(index, version)`

`switchAliasToNew(alias, newIndex, context)`

`verifyReindex(sourceIndex, destIndex)`

Index Naming Convention

Migration Strategies

Strategy 1: Full Migration (All Products)

Strategy 2: Incremental Migration (Updated Products Only)

Strategy 3: Single Product Type

Monitoring Migration Progress

Check Index Status

Safety Features

1. Verification Before Alias Switch

2. Atomic Alias Updates

3. Job Tracking

4. Metadata Tracking

Common Use Cases

Use Case 1: Schema Change

Use Case 2: Data Corruption

Use Case 3: Performance Optimization

Configuration

Alias Mapping (Config)

Environment Variables

Best Practices

Troubleshooting

Issue: "Reindex verification failed"

Issue: "Index already exists"

Issue: Migration stuck

Summary: One-Click Migration Flow