Zero-Downtime Typesense Collection Reindexing and Alias Cutover Gate

id: typesense-collection-alias-zero-downtime-reindex
namespace: company.team

description: |
  Orchestrate zero-downtime search collection rebuilds in Typesense using collection aliasing.
  Creates a versioned collection with the updated schema, bulk indexes catalog records,
  validates search recall and facet distributions with canary queries, and atomically swaps
  the production alias with zero dropped queries.

inputs:
  - id: typesense_host
    type: STRING
    displayName: Typesense Host
    description: Hostname or IP address of the Typesense node or load balancer.
    defaults: "localhost"

  - id: catalog_feed_url
    type: STRING
    displayName: Catalog Feed URL
    description: HTTP(S) URL returning the product catalog JSON to index. The
      default is a public demo endpoint returning real sample products; swap in
      your PIM or commerce platform export URL for production use.
    defaults: "https://dummyjson.com/products?limit=100"

  - id: typesense_port
    type: STRING
    displayName: Typesense Port
    description: HTTP API port of the Typesense instance (8108 by default).
    defaults: "8108"

  - id: alias_name
    type: STRING
    displayName: Production Alias Name
    description: Virtual alias queried by frontend applications and storefront search bars.
    defaults: "products"

  - id: min_expected_docs
    type: INT
    displayName: Minimum Expected Documents
    description: Minimum number of successfully indexed documents required to
      approve alias cutover.
    defaults: 5

  - id: canary_query
    type: STRING
    displayName: Canary Search Query
    description: High-value storefront search query used to validate full-text
      search recall on the staged collection.
    defaults: "wireless headphones"

  - id: canary_min_hits
    type: INT
    displayName: Canary Minimum Hits
    description: Minimum document matches expected from the canary full-text query.
    defaults: 1

  - id: canary_facet_field
    type: STRING
    displayName: Canary Facet Field
    description: Facet field to evaluate for valid category distribution and
      indexing integrity.
    defaults: "category"

  - id: retire_previous_collection
    type: BOOL
    displayName: Retire Previous Collection
    description: If true, permanently drops the superseded collection after cutover.
      If false, preserves it for instant rollback.
    defaults: false

triggers:
  - id: reindex_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook endpoint for CI/CD pipelines, PIM updates, or
      catalog ETL triggers.
    key: "{{ secret('TYPESENSE_REINDEX_WEBHOOK_KEY') }}"
    disabled: true

  - id: weekly_reindex_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled maintenance trigger for weekly zero-downtime catalog
      re-indexing. Disabled by default.
    cron: "0 2 * * 0"
    disabled: true

tasks:
  - id: inspect_current_alias
    type: io.kestra.plugin.core.http.Request
    description: Read the current target collection of the production alias to
      preserve it for safe rollback tracking.
    uri: "http://{{ inputs.typesense_host }}:{{ inputs.typesense_port }}/aliases/{{
      inputs.alias_name }}"
    method: GET
    headers:
      X-TYPESENSE-API-KEY: "{{ secret('TYPESENSE_API_KEY') }}"
    options:
      allowFailed: true
    allowFailure: true

  - id: stage_new_collection
    type: io.kestra.plugin.core.http.Request
    description: Create a fresh versioned collection with explicit field definitions
      and facet indexing before ingesting documents.
    uri: "http://{{ inputs.typesense_host }}:{{ inputs.typesense_port
      }}/collections"
    method: POST
    contentType: application/json
    headers:
      X-TYPESENSE-API-KEY: "{{ secret('TYPESENSE_API_KEY') }}"
    body: |
      {
        "name": "{{ inputs.alias_name }}_v{{ execution.id | slice(0, 8) }}",
        "fields": [
          {"name": "id", "type": "string"},
          {"name": "name", "type": "string"},
          {"name": "description", "type": "string"},
          {"name": "brand", "type": "string", "facet": true},
          {"name": "category", "type": "string", "facet": true},
          {"name": "price", "type": "float", "facet": true},
          {"name": "popularity", "type": "float"},
          {"name": "in_stock", "type": "bool", "facet": true}
        ],
        "default_sorting_field": "popularity"
      }

  - id: download_catalog_feed
    type: io.kestra.plugin.core.http.Download
    description: Download the source product catalog JSON into Kestra internal
      storage. The demo endpoint returns real sample products; swap in your PIM
      or commerce platform export URL for production use.
    uri: "{{ inputs.catalog_feed_url }}"

  - id: prepare_catalog_payload
    type: io.kestra.plugin.scripts.python.Script
    description: Reshape the downloaded catalog into line-delimited JSON matching
      the Typesense collection schema.
    inputFiles:
      catalog.json: "{{ outputs.download_catalog_feed.uri }}"
    outputFiles:
      - catalog.jsonl
    script: |
      import json

      with open("catalog.json", encoding="utf-8") as f:
          data = json.load(f)

      count = 0
      with open("catalog.jsonl", "w", encoding="utf-8") as out:
          for p in data["products"]:
              doc = {
                  "id": str(p["id"]),
                  "name": p["title"],
                  "description": p["description"],
                  "brand": p.get("brand") or "unbranded",
                  "category": p["category"],
                  "price": float(p["price"]),
                  "popularity": float(p.get("rating", 0)),
                  "in_stock": int(p.get("stock", 0)) > 0,
              }
              out.write(json.dumps(doc) + "\n")
              count += 1

      print(f"Staged {count} catalog items into catalog.jsonl")

  - id: bulk_index_catalog
    type: io.kestra.plugin.typesense.BulkIndex
    description: Stream catalog documents into the staged versioned collection in
      chunked HTTP batches.
    host: "{{ inputs.typesense_host }}"
    port: "{{ inputs.typesense_port }}"
    apiKey: "{{ secret('TYPESENSE_API_KEY') }}"
    collection: "{{ inputs.alias_name }}_v{{ execution.id | slice(0, 8) }}"
    https: false
    from: "{{ outputs.prepare_catalog_payload.outputFiles['catalog.jsonl'] }}"
    chunk: 100

  - id: run_canary_search
    type: io.kestra.plugin.typesense.Search
    description: Execute verification search query against staged collection to
      confirm full-text recall and latency SLAs.
    host: "{{ inputs.typesense_host }}"
    port: "{{ inputs.typesense_port }}"
    apiKey: "{{ secret('TYPESENSE_API_KEY') }}"
    collection: "{{ inputs.alias_name }}_v{{ execution.id | slice(0, 8) }}"
    https: false
    query: "{{ inputs.canary_query }}"
    queryBy: "name,description,category"

  - id: run_canary_facet_search
    type: io.kestra.plugin.typesense.FacetSearch
    description: Query facet distributions on staged collection to ensure faceting
      structures are intact before cutover.
    host: "{{ inputs.typesense_host }}"
    port: "{{ inputs.typesense_port }}"
    apiKey: "{{ secret('TYPESENSE_API_KEY') }}"
    collection: "{{ inputs.alias_name }}_v{{ execution.id | slice(0, 8) }}"
    https: false
    query: "*"
    queryBy: "name"
    facetBy: "{{ inputs.canary_facet_field }}"

  - id: evaluate_cutover_gate
    type: io.kestra.plugin.core.flow.If
    description: Evaluate indexing thresholds; conditionally perform atomic alias
      swap or abort and roll back.
    condition: "{{ (outputs.bulk_index_catalog.size >= inputs.min_expected_docs) and
      (outputs.run_canary_search.totalHits >= inputs.canary_min_hits) }}"
    then:
      - id: atomic_alias_cutover
        type: io.kestra.plugin.core.http.Request
        description: Atomically update the production alias pointer in Typesense memory
          without interrupting search traffic.
        uri: "http://{{ inputs.typesense_host }}:{{ inputs.typesense_port }}/aliases/{{
          inputs.alias_name }}"
        method: PUT
        contentType: application/json
        headers:
          X-TYPESENSE-API-KEY: "{{ secret('TYPESENSE_API_KEY') }}"
        body: |
          {
            "collection_name": "{{ inputs.alias_name }}_v{{ execution.id | slice(0, 8) }}"
          }

      - id: evaluate_previous_collection_cleanup
        type: io.kestra.plugin.core.flow.If
        description: Check whether the superseded collection should be permanently
          dropped or preserved for rollback.
        condition: "{{ inputs.retire_previous_collection and
          (outputs.inspect_current_alias.code == 200) and
          (fromJson(outputs.inspect_current_alias.body).collection_name !=
          (inputs.alias_name ~ '_v' ~ (execution.id | slice(0, 8)))) }}"
        then:
          - id: delete_superseded_collection
            type: io.kestra.plugin.core.http.Request
            description: Drop previous physical collection to reclaim cluster RAM and
              storage capacity.
            uri: "http://{{ inputs.typesense_host }}:{{ inputs.typesense_port
              }}/collections/{{
              fromJson(outputs.inspect_current_alias.body).collection_name }}"
            method: DELETE
            headers:
              X-TYPESENSE-API-KEY: "{{ secret('TYPESENSE_API_KEY') }}"

        else:
          - id: log_retention_decision
            type: io.kestra.plugin.core.log.Log
            description: Log collection preservation status for operational auditability.
            message: >-
              Previous collection {{ outputs.inspect_current_alias.code == 200 ?
              fromJson(outputs.inspect_current_alias.body).collection_name :
              'none' }} retained for rollback safety.

      - id: log_cutover_success
        type: io.kestra.plugin.core.log.Log
        description: Record successful zero-downtime cutover metadata into Kestra
          execution logs.
        message: >-
          SUCCESS: Typesense alias '{{ inputs.alias_name }}' atomically cut over
          to '{{ inputs.alias_name }}_v{{ execution.id | slice(0, 8) }}'
          (Indexed: {{ outputs.bulk_index_catalog.size }} docs, Canary hits: {{
          outputs.run_canary_search.totalHits }}).

    else:
      - id: abort_and_delete_staged_collection
        type: io.kestra.plugin.core.http.Request
        description: Validation failed. Delete defective staged collection to avoid disk
          sprawl; live alias is untouched.
        uri: "http://{{ inputs.typesense_host }}:{{ inputs.typesense_port
          }}/collections/{{ inputs.alias_name }}_v{{ execution.id | slice(0, 8)
          }}"
        method: DELETE
        headers:
          X-TYPESENSE-API-KEY: "{{ secret('TYPESENSE_API_KEY') }}"

      - id: alert_canary_gate_failure
        type: io.kestra.plugin.core.log.Log
        description: Emit error audit log detailing why the staged collection failed
          validation criteria.
        level: ERROR
        message: >-
          REINDEX ABORTED: Staged collection '{{ inputs.alias_name }}_v{{
          execution.id | slice(0, 8) }}' failed canary validation (Indexed: {{
          outputs.bulk_index_catalog.size }}/{{ inputs.min_expected_docs }},
          Canary hits: {{ outputs.run_canary_search.totalHits }}/{{
          inputs.canary_min_hits }}). Live alias '{{ inputs.alias_name }}'
          remains active.

errors:
  - id: emergency_cleanup_staged_collection
    type: io.kestra.plugin.core.http.Request
    description: Safety task ensuring orphaned staged collection is deleted if an
      unhandled error terminates the flow.
    uri: "http://{{ inputs.typesense_host }}:{{ inputs.typesense_port
      }}/collections/{{ inputs.alias_name }}_v{{ execution.id | slice(0, 8) }}"
    method: DELETE
    options:
      allowFailed: true
    allowFailure: true
    headers:
      X-TYPESENSE-API-KEY: "{{ secret('TYPESENSE_API_KEY') }}"

  - id: log_fatal_failure
    type: io.kestra.plugin.core.log.Log
    description: Record execution failure details in workflow logs.
    level: ERROR
    message: >-
      FATAL: Zero-downtime reindexing pipeline failed unexpectedly during
      execution {{ execution.id }}. Production alias was protected.