Skip to content
Learnille Documentation

Elasticsearch Product Migration System

Elasticsearch Product Migration System

Overview

The Elasticsearch 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. ElasticController (elastic.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 /elastic/index-status - View current index versions and alias mappings
  • POST /elastic/reindex-batch - Batch reindex without creating new versions
  • POST /elastic/reindex-document - Reindex a single document by ID
  • POST /elastic/migrate/:index/:version - Migrate a specific index to a version
  • POST /elastic/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 Elasticsearch


    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->>Elasticsearch: Create index (e.g., "course_v2")
    Elasticsearch-->>ElasticService: Index created


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


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


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


    Worker->>ElasticService: switchAliasToNew(alias, newIndex)
    ElasticService->>Elasticsearch: Update alias (course → course_v2)
    Elasticsearch-->>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:

  1. Version Resolution

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

  2. Create New Index

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

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

  4. Verification

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

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

  5. 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 IndexAliasPhysical Indices
coursecoursecourse_v1, course_v2, course_v3
consultationconsultationconsultation_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)

Terminal window
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)

Terminal window
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

Terminal window
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

Terminal window
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

Elasticsearch’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 Elasticsearch mapping for courses.

Solution:

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

Terminal window
# 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)

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

Environment Variables

Terminal window
ELASTICSEARCH_HOST=https://localhost:9200
ELASTICSEARCH_USERNAME=elastic
ELASTICSEARCH_PASSWORD=changeme
NODE_ENV=development

Best Practices

  1. Always Use Aliases in Application Code

    // ✅ 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:

    Terminal window
    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! 🎉