Elasticsearch NodeJS Sdk does not work with bun, so wrote my favorite APIs using the REST API, in an API compatible manner.

elasticsearch-rest.js

import logger from "#lib/utils/logger.js";

const host = process.env.ELASTIC_SERVER_HOST;
const username = process.env.ELASTIC_USER;
const password = process.env.ELASTIC_PASSWORD;

function getAuthHeader() {
  if (username && password) {
    return btoa(`${username}:${password}`);
  }
  return null;
}

async function makeRequest(path, options = {}) {
  if (!host) {
    throw new Error("ELASTIC_SERVER_HOST not set");
  }

  const authHeader = getAuthHeader();
  const headers = {
    "Content-Type": "application/json",
    ...options.headers,
  };

  if (authHeader) {
    headers["Authorization"] = `Basic ${authHeader}`;
  }

  const url = `${host}${path}`;
  const fetchOptions = {
    ...options,
    headers,
  };

  const response = await fetch(url, fetchOptions);

  return response;
}

function handleResponse(response, returnNullOn404 = false) {
  if (!response.ok) {
    const error = new Error(
      `Elasticsearch request failed: ${response.status} ${response.statusText}`,
    );
    error.meta = { statusCode: response.status };
    if (returnNullOn404 && response.status === 404) {
      return null;
    }
    throw error;
  }
  return response;
}

function normalizeError(error) {
  if (error.name === "AbortError") {
    return new Error("Request timeout");
  }
  return error;
}

export async function indicesExists(index) {
  try {
    const response = await makeRequest(`/${index}`, {
      method: "HEAD",
    });
    return response.ok;
  } catch (error) {
    logger.error(`Error checking if index ${index} exists:`, error);
    throw normalizeError(error);
  }
}

export async function indicesCreate(index, mapping) {
  try {
    const response = await makeRequest(`/${index}`, {
      method: "PUT",
      body: JSON.stringify(mapping),
    });
    handleResponse(response);
    return { acknowledged: true, index };
  } catch (error) {
    logger.error(`Error creating index ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function indicesDelete(index) {
  try {
    const response = await makeRequest(`/${index}`, {
      method: "DELETE",
    });
    handleResponse(response);
    return { acknowledged: true };
  } catch (error) {
    logger.error(`Error deleting index ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function indexDocument(index, id, document, options = {}) {
  try {
    const path = id ? `/${index}/_doc/${id}` : `/${index}/_doc`;
    const method = "POST";

    const url = new URL(path, host);
    if (options.refresh) {
      url.searchParams.set("refresh", options.refresh);
    }
    if (options.op_type) {
      url.searchParams.set("op_type", options.op_type);
    }

    const response = await fetch(url.toString(), {
      method,
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
      body: JSON.stringify(document),
    });

    handleResponse(response);
    const data = await response.json();
    return { _id: data._id || id, ...data };
  } catch (error) {
    logger.error(`Error indexing document to ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function getDocument(index, id, options = {}) {
  try {
    const url = new URL(`/${index}/_doc/${id}`, host);
    if (options._source) {
      url.searchParams.set(
        "_source",
        Array.isArray(options._source)
          ? options._source.join(",")
          : options._source,
      );
    }

    const response = await fetch(url.toString(), {
      method: "GET",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
    });

    const result = await handleResponse(response, true);
    const data = await response.json();
    return data;
  } catch (error) {
    if (error.meta?.statusCode === 404) {
      return null;
    }
    logger.error(`Error getting document ${id} from ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function updateDocument(index, id, docOrScript, options = {}) {
  try {
    const url = new URL(`/${index}/_update/${id}`, host);

    if (options.refresh) {
      url.searchParams.set("refresh", options.refresh);
    }

    if (options.retry_on_conflict) {
      url.searchParams.set("retry_on_conflict", options.retry_on_conflict);
    }

    let body;
    if (options.script) {
      body = { script: options.script };
      if (options.upsert) {
        body.upsert = options.upsert;
      }
    } else if (docOrScript?.source) {
      body = { script: docOrScript };
      if (options.upsert) {
        body.upsert = options.upsert;
      }
    } else {
      body = { doc: docOrScript };
      if (options.upsert) {
        body.upsert = options.upsert;
      }
    }

    const response = await fetch(url.toString(), {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
      body: JSON.stringify(body),
    });

    handleResponse(response);
    const data = await response.json();
    return { _id: id, result: data.result || "updated", ...data };
  } catch (error) {
    logger.error(`Error updating document ${id} in ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function deleteDocument(index, id, options = {}) {
  try {
    const url = new URL(`/${index}/_doc/${id}`, host);

    if (options.refresh) {
      url.searchParams.set("refresh", options.refresh);
    }

    const response = await fetch(url.toString(), {
      method: "DELETE",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
    });

    const result = await handleResponse(response, true);
    const data = await response.json();
    return { result: data.result || "deleted", ...data };
  } catch (error) {
    if (error.meta?.statusCode === 404) {
      return { result: "not_found" };
    }
    logger.error(`Error deleting document ${id} from ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function deleteByQuery(index, query, options = {}) {
  try {
    const url = new URL(`/${index}/_delete_by_query`, host);

    if (options.refresh) {
      url.searchParams.set("refresh", options.refresh);
    }

    const response = await fetch(url.toString(), {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
      body: JSON.stringify({ query }),
    });

    handleResponse(response);
    const data = await response.json();
    return { deleted: data.deleted || 0, ...data };
  } catch (error) {
    logger.error(`Error deleting by query in ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function count(index, options = {}) {
  try {
    const url = new URL(`/${index}/_search`, host);

    let body;

    if (options.body) {
      body = { ...options.body };
    } else {
      body = {};
    }

    if (options.query) {
      body.query = options.query;
    }

    if (!body.track_total_hits) {
      body.track_total_hits = true;
    }

    if (body.size === undefined) {
      body.size = 0;
    }

    const response = await fetch(url.toString(), {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
      body: JSON.stringify(body),
    });

    handleResponse(response);
    const data = await response.json();

    return {
      count: data.hits?.total?.value || data.hits?.total || 0,
      _shards: data._shards,
      timed_out: data.timed_out,
    };
  } catch (error) {
    logger.error(`Error counting documents in ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function search(index, options = {}) {
  try {
    const url = new URL(`/${index}/_search`, host);

    let body;

    if (options.body) {
      body = { ...options.body };
    } else {
      body = {};
    }

    if (options.query) {
      body.query = options.query;
    }

    if (options.size !== undefined) {
      body.size = options.size;
    }

    if (options.from !== undefined) {
      body.from = options.from;
    }

    if (options.sort) {
      body.sort = options.sort;
    }

    if (options.aggs) {
      body.aggs = options.aggs;
    }

    if (options._source) {
      body._source = options._source;
    }

    const response = await fetch(url.toString(), {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
      body: JSON.stringify(body),
    });

    handleResponse(response);
    const data = await response.json();

    return {
      hits: {
        total: {
          value: data.hits?.total?.value || data.hits?.total || 0,
        },
        hits: data.hits?.hits || [],
      },
      aggregations: data.aggregations,
      took: data.took,
      _shards: data._shards,
      timed_out: data.timed_out,
    };
  } catch (error) {
    logger.error(`Error searching in ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function catCount(index) {
  try {
    const url = new URL(`/_cat/count/${index}?format=json`, host);

    const response = await fetch(url.toString(), {
      method: "GET",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
    });

    handleResponse(response);
    const data = await response.json();
    return data;
  } catch (error) {
    logger.error(`Error getting count for ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function documentExists(index, id) {
  try {
    const url = new URL(`/${index}/_doc/${id}`, host);

    const response = await fetch(url.toString(), {
      method: "HEAD",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
    });

    return response.ok;
  } catch (error) {
    logger.error(`Error checking if document ${id} exists:`, error);
    return false;
  }
}

export async function indicesStats(index) {
  try {
    const response = await makeRequest(`/${index}/_stats`, {
      method: "GET",
    });
    handleResponse(response);
    const data = await response.json();
    return data;
  } catch (error) {
    logger.error(`Error getting stats for index ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function indicesGetMapping(index) {
  try {
    const response = await makeRequest(`/${index}/_mapping`, {
      method: "GET",
    });
    handleResponse(response);
    const data = await response.json();
    return data;
  } catch (error) {
    logger.error(`Error getting mapping for index ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function indicesGetSettings(index) {
  try {
    const response = await makeRequest(`/${index}/_settings`, {
      method: "GET",
    });
    handleResponse(response);
    const data = await response.json();
    return data;
  } catch (error) {
    logger.error(`Error getting settings for index ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function info() {
  try {
    const response = await makeRequest(`/`, {
      method: "GET",
    });
    handleResponse(response);
    const data = await response.json();
    return data;
  } catch (error) {
    logger.error("Error getting Elasticsearch info:", error);
    throw normalizeError(error);
  }
}

export async function scrollSearch(index, scrollTime, size) {
  try {
    const url = new URL(`/${index}/_search`, host);
    url.searchParams.set("scroll", scrollTime);
    url.searchParams.set("size", size.toString());

    const response = await fetch(url.toString(), {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
      body: JSON.stringify({ query: { match_all: {} } }),
    });

    handleResponse(response);
    const data = await response.json();
    return data;
  } catch (error) {
    logger.error(`Error scrolling search in ${index}:`, error);
    throw normalizeError(error);
  }
}

export async function scroll(scrollId, scrollTime) {
  try {
    const url = new URL(`/_search/scroll`, host);
    const response = await fetch(url.toString(), {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
      body: JSON.stringify({
        scroll_id: scrollId,
        scroll: scrollTime,
      }),
    });

    handleResponse(response);
    const data = await response.json();
    return data;
  } catch (error) {
    logger.error(`Error scrolling with ID ${scrollId}:`, error);
    throw normalizeError(error);
  }
}

export async function clearScroll(scrollId) {
  try {
    const url = new URL(`/_search/scroll`, host);
    const response = await fetch(url.toString(), {
      method: "DELETE",
      headers: {
        "Content-Type": "application/json",
        ...(getAuthHeader()
          ? { Authorization: `Basic ${getAuthHeader()}` }
          : {}),
      },
      body: JSON.stringify({
        scroll_id: scrollId,
      }),
    });

    handleResponse(response);
    return { acknowledged: true };
  } catch (error) {
    logger.error(`Error clearing scroll ${scrollId}:`, error);
    throw normalizeError(error);
  }
}

export const client = {
  indices: {
    exists: async (params) => {
      const exists = await indicesExists(params.index);
      return { body: { exists } };
    },
    create: async (params) => {
      await indicesCreate(params.index, params.body);
      return { body: { acknowledged: true } };
    },
    delete: async (params) => {
      await indicesDelete(params.index);
      return { body: { acknowledged: true } };
    },
    stats: async (params) => {
      const data = await indicesStats(params.index);
      return { body: data };
    },
    getMapping: async (params) => {
      const data = await indicesGetMapping(params.index);
      return data;
    },
    getSettings: async (params) => {
      const data = await indicesGetSettings(params.index);
      return data;
    },
  },
  count: async (params) => {
    return await count(params.index, {
      query: params.query,
      body: params.body,
    });
  },
  cat: {
    count: async (params) => {
      const data = await catCount(params.index);
      return { body: data };
    },
  },
  index: async (params) => {
    const result = await indexDocument(
      params.index,
      params.id,
      params.document || params.body,
      {
        refresh: params.refresh,
        op_type: params.op_type,
      },
    );
    return { body: result, _id: result._id };
  },
  get: async (params) => {
    const data = await getDocument(params.index, params.id, {
      _source: params._source,
    });
    return data;
  },
  update: async (params) => {
    const result = await updateDocument(
      params.index,
      params.id,
      params.doc || params.body?.doc,
      {
        refresh: params.refresh,
        upsert: params.body?.upsert,
        script: params.body?.script,
        retry_on_conflict: params.retry_on_conflict,
      },
    );
    return { body: result, _id: result._id };
  },
  delete: async (params) => {
    const result = await deleteDocument(params.index, params.id, {
      refresh: params.refresh,
    });
    return { body: result };
  },
  deleteByQuery: async (params) => {
    const result = await deleteByQuery(params.index, params.query, {
      refresh: params.refresh,
    });
    return { body: result };
  },
  search: async (params) => {
    const result = await search(params.index, {
      query: params.query,
      size: params.size,
      from: params.from,
      sort: params.sort,
      aggs: params.aggs,
      _source: params._source,
      body: params.body,
    });
    return result;
  },
  exists: async (params) => {
    const exists = await documentExists(params.index, params.id);
    return exists;
  },
  info: async () => {
    const data = await info();
    return data;
  },
  scroll: async (params) => {
    const data = await scroll(params.scroll_id, params.scroll);
    return data;
  },
  clearScroll: async (params) => {
    await clearScroll(params.scroll_id);
    return { body: { acknowledged: true } };
  },
};