feat: esOptions, track_total_hits, external versioning and $noFetch for documents ES v2

CHANGELOG.md

@@ -1,3 +1,10 @@
+## [1.0.2-documents-esx.0](https://github.com/refrens/feathers-esx/compare/1.0.1...1.0.2-documents-esx.0) (2026-08-17)
+
+
+### Features
+
+* adds es params and versioning ([cb44a5b](https://github.com/refrens/feathers-esx/commit/cb44a5be3bad2749fed58c40a1a9db1e1d43d315))
+
 ## 1.0.1 (2025-06-25)
 
 

lib/core/create-bulk/6.0.js

@@ -6,6 +6,36 @@ const createBulkCore = require('./core');
 // https://www.elastic.co/guide/en/elasticsearch/client/javascript-api/current/api-reference.html#api-bulk
 // reduce() here is acting as a map() mapping one element into two.
 
+/**
+ * External versioning, when the service declares `elasticsearch.versionField`.
+ *
+ * Stamps each bulk item with `version` derived from that field (typically `updatedAt`) and
+ * `version_type: external_gte`. Elasticsearch then refuses any write carrying a version older
+ * than what is already stored, which makes bulk loads idempotent and order-independent: a row
+ * derived from an older snapshot can never overwrite a newer live write, and re-running a
+ * backfill is safe. Rejected items come back as 409 version_conflict_engine_exception, which
+ * the caller should count and ignore rather than treat as an error.
+ */
+function getVersioning(service, doc) {
+  const field = service.esOptions && service.esOptions.versionField;
+
+  if (!field || doc[field] === undefined || doc[field] === null) {
+    return {};
+  }
+
+  // `new Date(v).getTime()`, not `Date.parse(v)`. A Mongoose `updatedAt` is a Date object, and
+  // Date.parse coerces it via toString(), whose output carries no milliseconds - so two writes in
+  // the same second would get equal versions, which external_gte accepts, letting the older one
+  // win. This handles Date, ISO string and epoch number alike, and still yields NaN for garbage.
+  const version = typeof doc[field] === 'number' ? doc[field] : new Date(doc[field]).getTime();
+
+  if (!Number.isFinite(version)) {
+    return {};
+  }
+
+  return { version, version_type: 'external_gte' };
+}
+
 function getBulkCreateParams(service, data, params) {
   return {
     body: data.reduce((result, item) => {
@@ -19,7 +49,10 @@ function getBulkCreateParams(service, data, params) {
         };
       }
 
-      result.push({ [method]: { _id: id, routing } });
+      // `create` fails outright if the document exists, so versioning only applies to `index`.
+      const versioning = method === 'index' ? getVersioning(service, doc) : {};
+
+      result.push({ [method]: { _id: id, routing, ...versioning } });
       result.push(doc);
 
       return result;

lib/core/create-bulk/core.js

@@ -6,6 +6,14 @@ function createBulkCore(service, data, params, { getBulkCreateParams }) {
 
   return service.Model.bulk(bulkCreateParams).then((results) => {
     const created = mapBulk(results.items, service.id, service.meta, service.join);
+
+    // Callers that discard the return value (e.g. a background sync writing to a secondary
+    // store) can skip the follow-up mget entirely. Without this a 500-document bulk costs
+    // two round trips instead of one, for data nobody reads.
+    if (params.$noFetch) {
+      return created;
+    }
+
     // We are fetching only items which have been correctly created.
     const docs = created
       .map((item, index) => ({

lib/core/find.js

@@ -6,10 +6,18 @@ function find(service, params) {
   const findParams = {
     _source: filters.$select,
     from: filters.$skip,
-    size: filters.$limit,
+    // `paginate: false` with no `$limit` leaves size undefined, so ES applies its default of 10
+    // and a full export silently returns 10 rows.
+    size: paginate === false ? filters.$limit || 10000 : filters.$limit,
     sort: filters.$sort,
     body: {
       query: esQuery ? { bool: esQuery } : undefined,
+      // Elasticsearch stops counting matches at 10,000 by default and reports
+      // { value: 10000, relation: 'gte' }, so any paginated result set larger than that
+      // reports a wrong `total`. Opt in per service via `elasticsearch.trackTotalHits`.
+      ...(service.esOptions && service.esOptions.trackTotalHits !== undefined
+        ? { track_total_hits: service.esOptions.trackTotalHits }
+        : {}),
     },
     ...service.esParams,
   };

lib/index.js

@@ -17,12 +17,20 @@ class Service extends AdapterService {
       throw new Error('Elasticsearch `Model` (client) needs to be provided');
     }
 
+    // `esParams` is spread verbatim into EVERY Elasticsearch request (search, bulk, index,
+    // get, ...). The legacy client passes any key it does not recognise straight through as
+    // a query-string param, and Elasticsearch 8 then rejects the request with
+    // `unrecognized parameter`. So behaviour options must be separated out here and kept
+    // out of esParams - they are exposed as `esOptions` instead.
+    const { trackTotalHits, versionField, ...esRequestParams } = options.elasticsearch || {};
+
     super({
       id: '_id',
       parent: '_parent',
       routing: '_routing',
       meta: '_meta',
-      esParams: { refresh: false, ...options.elasticsearch },
+      esParams: { refresh: false, ...esRequestParams },
+      esOptions: { trackTotalHits, versionField },
       whitelist: [
         '$prefix',
         '$wildcard',
@@ -48,12 +56,13 @@ class Service extends AdapterService {
     });
 
     // Alias getters for options
-    ['Model', 'parent', 'routing', 'meta', 'join', 'esVersion', 'esParams'].forEach((name) =>
-      Object.defineProperty(this, name, {
-        get() {
-          return this.options[name];
-        },
-      }),
+    ['Model', 'parent', 'routing', 'meta', 'join', 'esVersion', 'esParams', 'esOptions'].forEach(
+      (name) =>
+        Object.defineProperty(this, name, {
+          get() {
+            return this.options[name];
+          },
+        }),
     );
 
     this.core = core(options.esVersion);

lib/utils/parse-query.js

@@ -218,7 +218,23 @@ function parseQuery(query, idProp) {
       return specialQueryHandlers[key](value, result, idProp);
     }
 
-    validateType(value, key, ['number', 'string', 'boolean', 'undefined', 'object', 'array']);
+    validateType(value, key, [
+      'number',
+      'string',
+      'boolean',
+      'undefined',
+      'object',
+      'array',
+      'null',
+    ]);
+
+    // Handle null values - convert to "must_not exists" query to search for null fields
+    if (value === null) {
+      result.must_not = result.must_not || [];
+      result.must_not.push({ exists: { field: key } });
+      return result;
+    }
+
     // The value is not an object, which means it's supposed to be a primitive or an array.
     // We need add simple filter[{term: {}}] query.
     if (type !== 'object') {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@refrens/feathers-esx",
-  "version": "1.0.1",
+  "version": "1.0.2-documents-esx.0",
   "description": "Refrens fork of feathers-elasticsearch with fixes and extensions",
   "main": "lib/",
   "types": "types",

test/service-options.js

@@ -0,0 +1,202 @@
+/* eslint-disable no-underscore-dangle */
+const { expect } = require('chai');
+const { Service } = require('../lib');
+
+/**
+ * These run without an Elasticsearch server: the constructor makes no requests, and the bulk
+ * tests below drive a stub client. They cover the `esParams` / `esOptions` split and the
+ * external versioning stamped onto bulk actions.
+ */
+describe('Elasticsearch service options', () => {
+  const makeService = (elasticsearch) =>
+    new Service({ Model: {}, esVersion: '6.0', elasticsearch });
+
+  describe('esParams / esOptions split', () => {
+    // The regression test for the staging + qa01 outage: a behaviour option that leaks into
+    // esParams is spread into every request, and Elasticsearch 8 rejects the whole search with
+    // `unrecognized parameter: [versionField]`.
+    it('keeps behaviour options out of esParams', () => {
+      const service = makeService({
+        index: 'refrens_documents',
+        trackTotalHits: true,
+        versionField: 'updatedAt',
+      });
+
+      expect(service.esParams).to.not.have.property('trackTotalHits');
+      expect(service.esParams).to.not.have.property('versionField');
+    });
+
+    it('exposes behaviour options on esOptions', () => {
+      const service = makeService({ trackTotalHits: true, versionField: 'updatedAt' });
+
+      expect(service.esOptions.trackTotalHits).to.equal(true);
+      expect(service.esOptions.versionField).to.equal('updatedAt');
+    });
+
+    it('passes every other elasticsearch option through to esParams unchanged', () => {
+      const service = makeService({ index: 'people', type: 'doc', refresh: true });
+
+      expect(service.esParams).to.deep.equal({ index: 'people', type: 'doc', refresh: true });
+    });
+
+    it('defaults esParams.refresh to false and esOptions to undefined when not configured', () => {
+      const service = makeService({ index: 'people' });
+
+      expect(service.esParams).to.deep.equal({ index: 'people', refresh: false });
+      expect(service.esOptions.trackTotalHits).to.equal(undefined);
+      expect(service.esOptions.versionField).to.equal(undefined);
+    });
+  });
+
+  describe('external versioning on bulk create', () => {
+    // Returns the action descriptors (every other element) of the bulk body the service sent.
+    const bulkActionsFor = async (elasticsearch, docs, params) => {
+      let body;
+      const service = new Service({
+        Model: {
+          bulk: (sent) => {
+            body = sent.body;
+            return Promise.resolve({ items: [] });
+          },
+        },
+        esVersion: '6.0',
+        elasticsearch,
+      });
+
+      await service._create(docs, params);
+
+      return body.filter((_, index) => index % 2 === 0);
+    };
+
+    it('stamps external_gte from the configured versionField', async () => {
+      const actions = await bulkActionsFor(
+        { index: 'people', versionField: 'updatedAt' },
+        [{ _id: '1', updatedAt: '2026-08-24T10:00:00.123Z' }],
+        { upsert: true },
+      );
+
+      expect(actions[0].index.version).to.equal(Date.parse('2026-08-24T10:00:00.123Z'));
+      expect(actions[0].index.version_type).to.equal('external_gte');
+    });
+
+    // Date.parse() on a Date object goes through toString(), which drops milliseconds. Two
+    // writes in the same second would then tie, and external_gte accepts a tie.
+    it('preserves milliseconds when the version field is a Date object', async () => {
+      const updatedAt = new Date('2026-08-24T10:00:00.123Z');
+      const actions = await bulkActionsFor(
+        { index: 'people', versionField: 'updatedAt' },
+        [{ _id: '1', updatedAt }],
+        { upsert: true },
+      );
+
+      expect(actions[0].index.version).to.equal(updatedAt.getTime());
+    });
+
+    it('accepts an epoch number as the version', async () => {
+      const actions = await bulkActionsFor(
+        { index: 'people', versionField: 'updatedAt' },
+        [{ _id: '1', updatedAt: 1787565600123 }],
+        { upsert: true },
+      );
+
+      expect(actions[0].index.version).to.equal(1787565600123);
+    });
+
+    it('writes unversioned when the field is missing or unparsable', async () => {
+      const actions = await bulkActionsFor(
+        { index: 'people', versionField: 'updatedAt' },
+        [{ _id: '1' }, { _id: '2', updatedAt: 'not a date' }],
+        { upsert: true },
+      );
+
+      actions.forEach((action) => {
+        expect(action.index).to.not.have.property('version');
+        expect(action.index).to.not.have.property('version_type');
+      });
+    });
+
+    it('does not version the create action, which cannot overwrite anyway', async () => {
+      const actions = await bulkActionsFor(
+        { index: 'people', versionField: 'updatedAt' },
+        [{ _id: '1', updatedAt: '2026-08-24T10:00:00.123Z' }],
+        {},
+      );
+
+      expect(actions[0]).to.have.property('create');
+      expect(actions[0].create).to.not.have.property('version');
+    });
+
+    it('writes unversioned when no versionField is configured', async () => {
+      const actions = await bulkActionsFor(
+        { index: 'people' },
+        [{ _id: '1', updatedAt: '2026-08-24T10:00:00.123Z' }],
+        { upsert: true },
+      );
+
+      expect(actions[0].index).to.not.have.property('version');
+    });
+  });
+
+  describe('$noFetch', () => {
+    it('returns the bulk metadata without a follow-up mget', async () => {
+      let mgetCalled = false;
+      const service = new Service({
+        Model: {
+          bulk: () => Promise.resolve({ items: [{ index: { _id: '1', status: 201 } }] }),
+          mget: () => {
+            mgetCalled = true;
+            return Promise.resolve({ docs: [] });
+          },
+        },
+        esVersion: '6.0',
+        elasticsearch: { index: 'people' },
+      });
+
+      const result = await service._create([{ _id: '1' }], { upsert: true, $noFetch: true });
+
+      expect(mgetCalled).to.equal(false);
+      expect(result[0]._meta).to.deep.equal({ _id: '1', status: 201 });
+    });
+  });
+
+  /**
+   * `paginate: false` with no `$limit` leaves `size` undefined, so Elasticsearch applies its
+   * own default of 10 and a full export silently returns 10 rows. master fixed this in
+   * `653ff91` (REF-19943); this branch forked before that commit and dropped it, which is how
+   * it came back. Driven through a stub `Model.search` so no server is needed.
+   */
+  describe('find: size when paginate is false', () => {
+    const captureSearch = async (params) => {
+      let sent;
+      const service = new Service({
+        Model: {
+          search: (findParams) => {
+            sent = findParams;
+            return Promise.resolve({ hits: { hits: [], total: 0 } });
+          },
+        },
+        esVersion: '6.0',
+        elasticsearch: { index: 'people' },
+        paginate: { default: 10, max: 50 },
+      });
+      await service.find(params);
+      return sent;
+    };
+
+    it('falls back to 10000 when paginate is false and no $limit is given', async () => {
+      const sent = await captureSearch({ paginate: false, query: {} });
+      expect(sent.size).to.equal(10000);
+    });
+
+    it('still honours an explicit $limit when paginate is false', async () => {
+      const sent = await captureSearch({ paginate: false, query: { $limit: 25 } });
+      expect(sent.size).to.equal(25);
+    });
+
+    it('leaves the paginated path alone', async () => {
+      const sent = await captureSearch({ query: { $limit: 5 } });
+      expect(sent.size).to.equal(5);
+    });
+  });
+});
+/* eslint-enable no-underscore-dangle */