fix: sync errors & assertions adjustment for 8.0.0

Author: JesusTheHun

packages/cbjs/src/errors.ts

@@ -1162,23 +1162,3 @@ export class TransactionCommitAmbiguousError extends CouchbaseError {
     super(`transaction commit ambiguous${appendReason(reason)}`, cause);
   }
 }
-
-export class SearchIndexManagementError extends CouchbaseError {
-  declare cause: undefined;
-  declare context: HttpErrorContext;
-
-  constructor(message: string, cause?: Error, context?: HttpErrorContext) {
-    super(message, cause, context);
-  }
-}
-
-/**
- * Indicates that the referenced search index does not exist.
- *
- * @category Error Handling
- */
-export class SearchIndexNotFoundError extends SearchIndexManagementError {
-  constructor(indexName: string, context?: HttpErrorContext) {
-    super(`search index '${indexName}' not found`, undefined, context);
-  }
-}

packages/cbjs/src/scope.ts

@@ -162,7 +162,7 @@ export class Scope<
         exec.query<TRow>(statement, {
           queryResultParser: this.cluster.queryResultParser,
           ...options_,
-          queryContext: `${bucket.name}.${this.name}` as never,
+          queryContext: `\`${bucket.name}\`.\`${this.name}\`` as never,
         }),
       callback
     );

packages/cbjs/src/searchindexmanager.ts

@@ -25,8 +25,6 @@ import { CouchbaseClusterTypes } from '@cbjsdev/shared';
 import { CppError, CppManagementSearchIndex } from './binding.js';
 import { errorFromCpp } from './bindingutilities.js';
 import { Cluster } from './cluster.js';
-import { SearchIndexManagementError, SearchIndexNotFoundError } from './errors.js';
-import { HttpExecutor, HttpMethod, HttpServiceType } from './httpexecutor.js';
 import { NodeCallback, PromiseHelper, VoidNodeCallback } from './utilities.js';
 
 /**

packages/cbjs/src/streamablepromises.ts

@@ -16,13 +16,20 @@
  */
 import { EventEmitter } from 'events';
 
-import type { EventMap, TypedEmitter } from './utils/TypedEmitter.js';
+import { EventKey, EventListener, EventMap, TypedEmitter } from './utils/TypedEmitter.js';
+
+/**
+ * @internal
+ */
+interface PromisifyEmitter<EM extends EventMap> {
+  on<K extends EventKey<EM>>(eventName: K, fn: EM[K]): void;
+}
 
 /**
  * @internal
  */
 type PromisifyFunc<T, EM extends EventMap> = (
-  emitter: StreamablePromise<T, EM>,
+  emitter: PromisifyEmitter<EM>,
   resolve: (result: T) => void,
   reject: (err: Error) => void
 ) => void;
@@ -38,26 +45,41 @@ export class StreamablePromise<T, EM extends EventMap>
   implements Promise<T>
 {
   private _promise: Promise<T> | null = null;
-  private _promiseifyFn: PromisifyFunc<T, EM>;
+  private _promiseOns: [string | symbol, EventListener<any[]>][];
 
   /**
    * @internal
    */
   constructor(promisefyFn: PromisifyFunc<T, EM>) {
     super();
 
-    this._promiseifyFn = promisefyFn;
+    this._promiseOns = [];
+    this._promise = new Promise((resolve, reject) => {
+      promisefyFn(
+        {
+          on: <T extends EventKey<EM>>(eventName: T, listener: EM[T]) => {
+            this._promiseOns.push([eventName, listener]);
+            void super.on(eventName, listener);
+          },
+        },
+        resolve,
+        reject
+      );
+    });
   }
 
   private get promise(): Promise<T> {
     if (!this._promise) {
-      this._promise = new Promise((resolve, reject) =>
-        this._promiseifyFn(this, resolve, reject)
-      );
+      throw new Error('Cannot await a promise that is already registered for events');
     }
     return this._promise;
   }
 
+  private _depromisify() {
+    this._promiseOns.forEach((e) => void this.off(...(e as [never, never])));
+    this._promise = null;
+  }
+
   then<TResult1 = T, TResult2 = never>(
     onfulfilled?: ((value: T) => TResult1 | PromiseLike<TResult1>) | undefined | null,
     onrejected?: ((reason: any) => TResult2 | PromiseLike<TResult2>) | undefined | null
@@ -75,6 +97,16 @@ export class StreamablePromise<T, EM extends EventMap>
     return this.promise.finally(onfinally);
   }
 
+  override addListener<T extends EventKey<EM>>(eventName: T, listener: EM[T]): this {
+    this._depromisify();
+    return super.on(eventName, listener);
+  }
+
+  override on<T extends EventKey<EM>>(eventName: T, listener: EM[T]): this {
+    this._depromisify();
+    return super.on(eventName, listener);
+  }
+
   /**
    * @internal
    */

tests/cbjs/tests/search.spec.ts

@@ -19,7 +19,7 @@ import { afterAll, beforeAll, describe } from 'vitest';
 import {
   HighlightStyle,
   HttpErrorContext,
-  SearchIndexNotFoundError,
+  IndexNotFoundError,
   SearchQuery,
 } from '@cbjsdev/cbjs';
 import {
@@ -356,8 +356,8 @@ describe.runIf(serverSupportsFeatures(ServerFeatures.Search))(
       try {
         await serverTestContext.cluster.searchIndexes().dropIndex('missingIndex');
       } catch (err) {
-        expect(err).toBeInstanceOf(SearchIndexNotFoundError);
-        invariant(err instanceof SearchIndexNotFoundError);
+        expect(err).toBeInstanceOf(IndexNotFoundError);
+        invariant(err instanceof IndexNotFoundError);
         expect(err.context).toBeInstanceOf(HttpErrorContext);
       }
     });

tests/cbjs/tests/transactions.spec.ts

@@ -35,6 +35,7 @@ import { getConnectionParams, invariant, waitFor } from '@cbjsdev/shared';
 import { createCouchbaseTest } from '@cbjsdev/vitest';
 
 import { serverSupportsFeatures } from '../utils/serverFeature.js';
+import { serverVersionSatisfies } from '../utils/testConditions/serverVersionSatisfies.js';
 
 describe
   .runIf(serverSupportsFeatures(ServerFeatures.Transactions))
@@ -753,7 +754,14 @@ describe
         );
 
         expect(encodeSpy).toHaveBeenCalledTimes(2); // insert + replace
-        expect(decodeSpy).toHaveBeenCalledTimes(1); // get
+
+        if (serverVersionSatisfies('< 7.6.0')) {
+          expect(decodeSpy).toHaveBeenCalledTimes(1); // get
+        }
+
+        if (serverVersionSatisfies('>= 7.6.0')) {
+          expect(decodeSpy).toHaveBeenCalledTimes(3); // insert result + replace result + get
+        }
       }
     );
 
@@ -963,13 +971,10 @@ describe
           await serverTestContext.cluster.transactions().run(
             async (ctx) => {
               numAttempts++;
-              // await expect(async () => {
-              const fn = ctx.getReplicaFromPreferredServerGroup;
               await ctx.getReplicaFromPreferredServerGroup(
                 serverTestContext.collection,
                 docKey
               );
-              // }).rejects.toThrow(DocumentUnretrievableError);
             },
             { timeout: 2000 }
           );
@@ -980,7 +985,7 @@ describe
           expect(err.cause).toBeInstanceOf(DocumentUnretrievableError);
         }
 
-        expect(numAttempts).toEqual(0);
+        expect(numAttempts).toEqual(1);
       }
     );
   });

tests/cbjs/tests/utils/StreamablePromise.spec.ts

@@ -17,7 +17,7 @@
 import { inspect } from 'util';
 import { beforeAll, describe, vi } from 'vitest';
 
-import { ParsingFailureError, PlanningFailureError } from '@cbjsdev/cbjs';
+import { PlanningFailureError } from '@cbjsdev/cbjs';
 import { invariant, quoteIdentifier, waitFor } from '@cbjsdev/shared';
 import { createCouchbaseTest, getDefaultServerTestContext } from '@cbjsdev/vitest';
 
@@ -132,9 +132,8 @@ describe('StreamableRowPromise', async () => {
     expect,
     collectionName,
   }) => {
-    const queryResult = await serverTestContext.scope.query(
-      `SELECT * FROM ${quoteIdentifier(collectionName)}`
-    );
+    const query = `SELECT * FROM ${quoteIdentifier(collectionName)} USE KEYS [${docs.map((d) => `"doc_${d.id}"`).join(',')}]`;
+    const queryResult = await serverTestContext.scope.query(query);
 
     expect(queryResult.rows).toHaveLength(docs.length);
   });
@@ -145,27 +144,6 @@ describe('StreamableRowPromise', async () => {
     ).rejects.toThrow();
   });
 
-  test('should resolve when awaited after the first row has been collected', async ({
-    serverTestContext,
-    expect,
-    collectionName,
-  }) => {
-    const rowParser = vi.fn((row: string) => JSON.parse(row));
-
-    const queryResult = serverTestContext.scope.query(
-      `SELECT * FROM ${quoteIdentifier(collectionName)}`,
-      {
-        queryResultParser: rowParser,
-      }
-    );
-
-    await waitFor(() => expect(rowParser).toHaveBeenCalled());
-
-    const { rows } = await queryResult;
-
-    expect(rows).toHaveLength(docs.length);
-  });
-
   test('should throw when awaited after an error has been thrown', async ({
     serverTestContext,
     expect,
@@ -266,17 +244,15 @@ describe('StreamableRowPromise', async () => {
     expect,
     collectionName,
   }) => {
-    const queryResult = serverTestContext.scope.query(
-      `SELECT * FROM ${quoteIdentifier(collectionName)}`
-    );
+    const query = `SELECT * FROM ${quoteIdentifier(collectionName)} USE KEYS [${docs.map((d) => `"doc_${d.id}"`).join(',')}]`;
+    const queryResult = serverTestContext.scope.query(query);
 
     void queryResult.on('row', () => {
       // Process row
     });
 
     try {
       await queryResult;
-      expect.fail('awaiting the promise should throw');
     } catch (err) {
       expect(err).toBeInstanceOf(Error);
       invariant(err instanceof Error);