@agstack/storage-elasticsearch

Enterprise-grade Elasticsearch/OpenSearch storage plugin for AGStack

The official search backend for the AGStack ecosystem. Indexes transactions asynchronously through the AGStack Worker Pool using the Elasticsearch Bulk API. Compatible with OpenSearch 2+.

Features

• Async Bulk Indexing — Never index one document at a time. Configurable batch size, flush interval, concurrency
• Index Lifecycle Management — Hot/Warm/Cold/Delete phases, automatic rollover, retention
• Index Templates — Composable templates, component templates, automatic creation
• Index Aliases — Write aliases, read aliases, multi-tenant support
• Connection Management — Connection pooling, node discovery, sniffing, TLS, compression
• Authentication — API keys, Basic auth, Bearer tokens, Elastic Cloud
• Failure Recovery — Retry queue, dead letter queue, partial failure handling, automatic backoff
• Health & Metrics — Cluster health, bulk throughput, indexing latency, failed documents
• Search Optimized — Keyword, text, date, IP, geo_point, nested objects, flattened fields

Installation

npm install @agstack/storage-elasticsearch @elastic/elasticsearch@8

Quick Start

import { StorageElasticsearchPlugin, DEFAULTS, validateConfig } from "@agstack/storage-elasticsearch";
import { createRuntime } from "@agstack/logger";

const runtime = createRuntime({
  plugins: [
    () => {
      const plugin = new StorageElasticsearchPlugin();
      runtime.registerPlugin("storage-elasticsearch", plugin, {
        options: validateConfig({
          nodes: [{ url: "http://localhost:9200" }],
          auth: { type: "api-key", apiKey: process.env.ES_API_KEY },
          index: {
            prefix: "transactions",
            strategy: "daily",
            shards: 3,
            replicas: 1,
          },
          bulk: {
            batchSize: 500,
            flushIntervalMs: 5000,
          },
        }),
      });
      return plugin;
    },
  ],
});

await runtime.start();

Configuration

| Option | Type | Default | Description | |--------|------|---------|-------------| | nodes | EsNodeConfig[] | [] | Elasticsearch node URLs | | cloudId | string | "" | Elastic Cloud ID | | auth.type | AuthType | "none" | Authentication type | | index.prefix | string | "agstack" | Index name prefix | | index.strategy | IndexStrategy | "daily" | Index naming strategy | | index.shards | number | 1 | Number of primary shards | | index.replicas | number | 0 | Number of replica shards | | bulk.batchSize | number | 500 | Documents per bulk request | | bulk.flushIntervalMs | number | 5000 | Auto-flush interval | | bulk.concurrency | number | 2 | Concurrent bulk workers | | workerCount | number | 2 | Background workers |

Architecture

Runtime → Plugin → BulkBuilder → Bulk API → Elasticsearch/OpenSearch
                         ↓                    ↓
                    Retry Queue           Index Manager
                         ↓              (templates, ILM,
                    Dead Letter Queue     aliases, mappings)

Index Strategies

• daily — {prefix}-YYYY.MM.DD
• monthly — {prefix}-YYYY.MM
• app-based — {prefix}-{app}-YYYY.MM
• env-based — {prefix}-{env}-YYYY.MM.DD
• custom — Template-based pattern

ILM Policy

When index.ilmEnabled is true, the plugin creates an ILM policy with configurable:

• Hot phase rollover (max size, age, docs)
• Warm phase (shrink, forcemerge)
• Cold phase (freeze)
• Delete phase (retention)

License

MIT