Reindex Migration

OpenSearch mappings are immutable once created — you can’t change a field’s type, add a normalizer, or modify dynamic templates on an existing index. Any schema change requires creating a new index and reindexing all data.

This page documents the zero-downtime reindex migration harness that automates this process as a Trigger.dev task.

Overview

When You Need This

• Changing a field type (e.g., keyword to text)
• Adding or removing a normalizer (e.g., lowercase_norm)
• Modifying dynamic templates
• Adding new mapped fields to the schema
• Any change to buildListingSchema() in listing-schema.ts

Migration Flow

Products vs Listings

Products are system-controlled (imported via CLI, not user-generated). During a migration, the full product stream in step 4 is sufficient — no timestamp-based catch-up is needed.

Listings are user-generated and continuously changing via the event-driven sync pipeline. Two catch-up passes ensure zero data loss:

Idempotency

If the task fails mid-way, re-triggering with the same targetVersion is safe:

• The index already exists — creation is skipped
• Bulk index operations are upserts — re-streaming overwrites existing documents
• The alias swap is atomic — it either succeeds or doesn’t

Error Handling

The service aborts immediately on the first error — any transform failure, batch index failure, or user fetch failure causes the task to stop. This is intentional: a migration must be all-or-nothing to avoid silently losing data. Since the migration is idempotent, you can fix the root cause and re-trigger safely.

Operator Recipe

export const SCHEMA_VERSION = 2 // Was 1

OpenSearch alias points to a different version than expected.
Run the reindex migration task to update.

{
  "targetVersion": 2
}

[creating_index] 0 indexed (0s)
[syncing_products] 45000 indexed (120s)
[syncing_listings] 12000 indexed (180s)
[catchup_listings] 150 indexed (182s)
[cleanup_deactivated] 3 indexed (182s)
[swapping_alias] 0 indexed (183s)
[post_swap_catchup] 8 indexed (184s)
[post_swap_cleanup_deactivated] 0 indexed (184s)
[cleanup] 0 indexed (184s)

curl http://localhost:9200/_aliases | jq
# Should show: product_listing_v2 → product_listing_current

Task Payload

import { z } from "zod"
 
export const reindexOpenSearchSchema = z.object({
  targetVersion: z.number().int().positive(),
  batchSize: z.number().int().positive().prefault(500),
  skipCleanup: z.boolean().prefault(false),
})

Schema Versioning

The schema version is a simple integer constant in listing-schema.ts:

export const SCHEMA_VERSION = 1
export const buildIndexName = (version: number) => `product_listing_v${version}`
export const CURRENT_INDEX_NAME = buildIndexName(SCHEMA_VERSION)

• SCHEMA_VERSION — bump when schema changes
• buildIndexName() — produces predictable index names (product_listing_v1, product_listing_v2, …)
• CURRENT_INDEX_NAME — used by SearchService.init() for the create-if-not-exists check

Code Locations