test: zswap event subscription tests

qa/tests/tests/e2e/shielded-transactions.test.ts

@@ -21,8 +21,16 @@ import { ToolkitWrapper, type ToolkitTransactionResult } from '@utils/toolkit/to
 import type { Transaction } from '@utils/indexer/indexer-types';
 import { getBlockByHashWithRetry, getTransactionByHashWithRetry } from './test-utils';
 import { TestContext } from 'vitest';
+import { collectValidZswapEvents } from 'tests/shared/zswap-events-utils';
+import { RegularTransactionSchema, ZswapLedgerEventSchema } from '@utils/indexer/graphql/schema';
+import { IndexerWsClient } from '@utils/indexer/websocket-client';
+import { EventCoordinator } from '@utils/event-coordinator';
+import { collectValidDustLedgerEvents } from 'tests/shared/dust-ledger-utils';
 
 describe('shielded transactions', () => {
+  let indexerWsClient: IndexerWsClient;
+  let indexerEventCoordinator: EventCoordinator;
+  let previousMaxZswapId: number;
   let toolkit: ToolkitWrapper;
   let transactionResult: ToolkitTransactionResult;
 
@@ -33,6 +41,9 @@ describe('shielded transactions', () => {
   let destinationAddress: string;
 
   beforeAll(async () => {
+    indexerWsClient = new IndexerWsClient();
+    indexerEventCoordinator = new EventCoordinator();
+    await indexerWsClient.connectionInit();
     // Start a one-off toolkit container
     toolkit = new ToolkitWrapper({});
 
@@ -41,6 +52,14 @@ describe('shielded transactions', () => {
     // Derive shielded addresses from seeds
     destinationAddress = (await toolkit.showAddress(destinationSeed)).shielded;
 
+    const beforeZswapEvents = await collectValidZswapEvents(
+      indexerWsClient,
+      indexerEventCoordinator,
+      1,
+    );
+    previousMaxZswapId = beforeZswapEvents[0].data!.zswapLedgerEvents.maxId;
+    log.debug(`Previous max zswap ledger ID before tx = ${previousMaxZswapId}`);
+
     // Submit one shielded->shielded transfer (1 STAR)
     transactionResult = await toolkit.generateSingleTx(
       sourceSeed,
@@ -59,7 +78,7 @@ describe('shielded transactions', () => {
   }, 200_000);
 
   afterAll(async () => {
-    await Promise.all([toolkit.stop()]);
+    await Promise.all([toolkit.stop(), indexerWsClient.connectionClose()]);
   });
 
   describe('a successful shielded transaction transferring 1 Shielded Token between two wallets', async () => {
@@ -115,13 +134,79 @@ describe('shielded transactions', () => {
       );
       // The expected transaction might take a bit more to show up by indexer, so we retry a few times
       const transactionResponse = await getTransactionByHashWithRetry(transactionResult.txHash!);
-
       expect(transactionResponse).toBeSuccess();
       expect(transactionResponse?.data?.transactions).toBeDefined();
       expect(transactionResponse?.data?.transactions?.length).toBeGreaterThan(0);
-      expect(
-        transactionResponse?.data?.transactions?.map((tx: Transaction) => `${tx.hash}`),
-      ).toContain(transactionResult.txHash);
+
+      const tx = transactionResponse!.data!.transactions!.find(
+        (t: Transaction) => t.hash === transactionResult.txHash,
+      );
+
+      expect(tx).toBeDefined();
+
+      // Validate transaction shape and narrow type using schema
+      const parsed = RegularTransactionSchema.safeParse(tx);
+      expect(parsed.success, JSON.stringify(parsed.error?.format(), null, 2)).toBe(true);
+
+      const regularTx = parsed.data!;
+
+      // Shielded transactions do NOT expose unshielded details
+      expect(regularTx.unshieldedCreatedOutputs).toEqual([]);
+      expect(regularTx.unshieldedSpentOutputs).toEqual([]);
+
+      // Shielded transactions do NOT expose fees
+      expect(regularTx.fees.paidFees).toBe('0');
+      expect(regularTx.fees.estimatedFees).toBe('0');
+    });
+
+    /**
+     * After a shielded transaction is confirmed, the indexer streams the Zswap
+     * events in sequence, followed by a DustSpendProcessed event.
+     *
+     * @given a confirmed shielded transaction
+     * @when we subscribe to Zswap events starting from (previousMaxId + 1)
+     * @then the Zswap events are delivered in order
+     * @and the following event is DustSpendProcessed
+     */
+    test('should stream Zswap events followed by DustSpendProcessed after a shielded transaction', async () => {
+      const received = await collectValidZswapEvents(
+        indexerWsClient,
+        indexerEventCoordinator,
+        3,
+        previousMaxZswapId + 1,
+      );
+      expect(received).toHaveLength(3);
+
+      received.forEach((msg) => {
+        const event = msg.data!.zswapLedgerEvents;
+        const parsed = ZswapLedgerEventSchema.safeParse(event);
+        expect(
+          parsed.success,
+          `Schema error: ${JSON.stringify(parsed.error?.format(), null, 2)}`,
+        ).toBe(true);
+      });
+
+      // Validate Zswap event grouping and ordering
+      const events = received.map((m) => m.data!.zswapLedgerEvents);
+      expect(new Set(events.map((e) => e.maxId)).size).toBe(1);
+
+      events.slice(1).forEach((e, i) => {
+        expect(e.id).toBe(events[i].id + 1);
+      });
+
+      const lastZswapMaxId = received.at(-1)!.data!.zswapLedgerEvents.maxId;
+
+      // verify the Dust event directly follows the Zswap events
+      const dustEvents = await collectValidDustLedgerEvents(
+        indexerWsClient,
+        indexerEventCoordinator,
+        1,
+        lastZswapMaxId + 1,
+      );
+      expect(dustEvents).toHaveLength(1);
+      const dust = dustEvents[0].data!.dustLedgerEvents;
+      expect(dust.__typename).toBe('DustSpendProcessed');
+      expect(dust.id).toBe(lastZswapMaxId + 1);
     });
 
     /**

qa/tests/tests/e2e/unshielded-transactions.test.ts

@@ -29,10 +29,43 @@ import {
   UnshieldedUtxo,
 } from '@utils/indexer/indexer-types';
 import { IndexerWsClient, UnshieldedTxSubscriptionResponse } from '@utils/indexer/websocket-client';
-import { collectValidDustEvents } from 'tests/shared/dust-utils';
+import { collectValidDustLedgerEvents } from 'tests/shared/dust-ledger-utils';
 import { EventCoordinator } from '@utils/event-coordinator';
 import { DustLedgerEventsUnionSchema } from '@utils/indexer/graphql/schema';
 
+/**
+ * Helper function to find a progress update event with an incremented transaction ID.
+ * This is the logic used inside the retry function for both source and destination address tests.
+ *
+ * @param events - The events array to search
+ * @param baselineTransactionId - The transaction ID to compare against
+ * @param addressLabel - Label for error messages (e.g., 'source' or 'destination')
+ * @returns The found event
+ * @throws Error if no matching event is found
+ */
+function findProgressUpdateEvent(
+  events: UnshieldedTxSubscriptionResponse[],
+  baselineTransactionId: number,
+  addressLabel: string,
+): UnshieldedTxSubscriptionResponse {
+  const event = events.find((event) => {
+    const txEvent = event.data?.unshieldedTransactions as UnshieldedTransactionEvent;
+
+    log.debug(`waiting for UnshieldedTransactionsProgress event`);
+    if (txEvent.__typename === 'UnshieldedTransactionsProgress') {
+      const progressUpdate = txEvent;
+      log.debug(`progressUpdate received: ${JSON.stringify(progressUpdate, null, 2)}`);
+      if (progressUpdate.highestTransactionId > baselineTransactionId) {
+        return true;
+      }
+    }
+  });
+  if (!event) {
+    throw new Error(`${addressLabel} address progress update event not found yet`);
+  }
+  return event;
+}
+
 describe('unshielded transactions', { timeout: 200_000 }, () => {
   let indexerWsClient: IndexerWsClient;
   let indexerHttpClient: IndexerHttpClient;
@@ -74,7 +107,11 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
 
     destinationAddress = walletFixture.destinations[0].destinationAddress;
 
-    const beforeEvents = await collectValidDustEvents(indexerWsClient, indexerEventCoordinator, 1);
+    const beforeEvents = await collectValidDustLedgerEvents(
+      indexerWsClient,
+      indexerEventCoordinator,
+      1,
+    );
     previousMaxDustId = beforeEvents[0].data!.dustLedgerEvents.maxId;
     log.debug(`Previous max dust ID before tx = ${previousMaxDustId}`);
 
@@ -96,39 +133,6 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
     await Promise.all([toolkit.stop(), indexerWsClient.connectionClose()]);
   });
 
-  /**
-   * Helper function to find a progress update event with an incremented transaction ID.
-   * This is the logic used inside the retry function for both source and destination address tests.
-   *
-   * @param events - The events array to search
-   * @param baselineTransactionId - The transaction ID to compare against
-   * @param addressLabel - Label for error messages (e.g., 'source' or 'destination')
-   * @returns The found event
-   * @throws Error if no matching event is found
-   */
-  function findProgressUpdateEvent(
-    events: UnshieldedTxSubscriptionResponse[],
-    baselineTransactionId: number,
-    addressLabel: string,
-  ): UnshieldedTxSubscriptionResponse {
-    const event = events.find((event) => {
-      const txEvent = event.data?.unshieldedTransactions as UnshieldedTransactionEvent;
-
-      log.debug(`waiting for UnshieldedTransactionsProgress event`);
-      if (txEvent.__typename === 'UnshieldedTransactionsProgress') {
-        const progressUpdate = txEvent as UnshieldedTransactionsProgress;
-        log.debug(`progressUpdate received: ${JSON.stringify(progressUpdate, null, 2)}`);
-        if (progressUpdate.highestTransactionId > baselineTransactionId) {
-          return true;
-        }
-      }
-    });
-    if (!event) {
-      throw new Error(`${addressLabel} address progress update event not found yet`);
-    }
-    return event;
-  }
-
   describe('a successful unshielded transaction transferring 1 STAR between two addresses', async () => {
     /**
      * Once an unshielded transaction has been submitted to node and confirmed, the indexer should report
@@ -150,7 +154,7 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
       );
 
       // The expected block might take a bit more to show up by indexer, so we retry a few times
-      const blockResponse = await getBlockByHashWithRetry(transactionResult.blockHash!);
+      const blockResponse = await getBlockByHashWithRetry(transactionResult.blockHash);
 
       // Verify the transaction appears in the block
       expect(blockResponse?.data?.block?.transactions).toBeDefined();
@@ -339,7 +343,7 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
       );
 
       // The expected block might take a bit more to show up by indexer, so we retry a few times
-      const blockResponse = await getBlockByHashWithRetry(transactionResult.blockHash!);
+      const blockResponse = await getBlockByHashWithRetry(transactionResult.blockHash);
 
       // Find the transaction with unshielded outputs
       const unshieldedTx = blockResponse.data?.block?.transactions?.find((tx: Transaction) => {
@@ -390,12 +394,12 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
       );
 
       log.debug('Progress updates before transaction:');
-      progressUpdatesBeforeTransaction!.forEach((update) => {
+      progressUpdatesBeforeTransaction.forEach((update) => {
         log.debug(`${JSON.stringify(update, null, 2)}`);
       });
 
       const highestTransactionIdBeforeTransaction = (
-        progressUpdatesBeforeTransaction![progressUpdatesBeforeTransaction!.length - 1].data
+        progressUpdatesBeforeTransaction.at(-1)?.data
           ?.unshieldedTransactions as UnshieldedTransactionsProgress
       ).highestTransactionId;
       log.info(
@@ -407,7 +411,7 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
       });
 
       log.debug('Progress updates after transaction:');
-      progressUpdatesAfterTransaction!.forEach((update) => {
+      progressUpdatesAfterTransaction.forEach((update) => {
         log.debug(`${JSON.stringify(update, null, 2)}`);
       });
 
@@ -454,12 +458,12 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
         });
 
       log.debug('Progress updates before transaction:');
-      progressUpdatesBeforeTransaction!.forEach((update) => {
+      progressUpdatesBeforeTransaction.forEach((update) => {
         log.debug(`${JSON.stringify(update, null, 2)}`);
       });
 
       const highestTransactionIdBeforeTransaction = (
-        progressUpdatesBeforeTransaction![progressUpdatesBeforeTransaction!.length - 1].data
+        progressUpdatesBeforeTransaction.at(-1)?.data
           ?.unshieldedTransactions as UnshieldedTransactionsProgress
       ).highestTransactionId;
       log.info(
@@ -473,7 +477,7 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
       );
 
       log.debug('Progress updates after transaction:');
-      progressUpdatesAfterTransaction!.forEach((update) => {
+      progressUpdatesAfterTransaction.forEach((update) => {
         log.debug(`${JSON.stringify(update, null, 2)}`);
       });
 
@@ -513,7 +517,7 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
      * DustGenerationDtimeUpdate, DustInitialUtxo, DustSpendProcessed
      */
     test('should deliver dust events in correct sequence after unshielded transaction', async () => {
-      const received = await collectValidDustEvents(
+      const received = await collectValidDustLedgerEvents(
         indexerWsClient,
         indexerEventCoordinator,
         3,
@@ -572,7 +576,7 @@ describe('unshielded transactions', { timeout: 200_000 }, () => {
           break;
         }
 
-        // A single tx hash can have multiple records because failed txs don't reserve hashes. 
+        // A single tx hash can have multiple records because failed txs don't reserve hashes.
         // This means the same hash might appear as:
         //   - a failed attempt (no created/spent UTXOs), or
         //   - a later successful attempt (with created/spent UTXOs).

qa/tests/tests/integration/basic/queries/dust-queries.test.ts

@@ -586,11 +586,6 @@ describe('dust generation status queries', () => {
         testKey: 'PM-18911',
       };
 
-      ctx.skip?.(
-        true,
-        'Skipping this test for when this has been delivered by developers https://shielded.atlassian.net/browse/PM-20789',
-      );
-
       let rewardAddress: string;
       const connectedCardanoNetworkType = env.getCardanoNetworkType();
 

qa/tests/tests/integration/basic/subscriptions/block-subscriptions.test.ts

@@ -62,10 +62,6 @@ describe('block subscriptions', () => {
         if (receivedBlocks.length === expectedCount) {
           eventCoordinator.notify(eventName);
           log.debug(`${expectedCount} blocks received`);
-          indexerWsClient.send<GraphQLCompleteMessage>({
-            id: '1',
-            type: 'complete',
-          });
         }
       },
     };