feat: handle projection

- add Mappers.withHandles
- add Mappers.filterProducedUtxoByAssetPolicyId
- add Mappers.filterMintByPolicyIds
- add new test projection data source "with-handle.json"

packages/projection-typeorm/src/entity/Handle.entity.ts

@@ -0,0 +1,18 @@
+import { AssetEntity } from './Asset.entity';
+import { Cardano } from '@cardano-sdk/core';
+import { Column, Entity, JoinColumn, OneToOne, PrimaryColumn } from 'typeorm';
+
+@Entity()
+export class HandleEntity {
+  @PrimaryColumn()
+  handle?: string;
+  @Column({ nullable: true })
+  cardanoAddress?: Cardano.PaymentAddress;
+  @OneToOne(() => AssetEntity, { onDelete: 'CASCADE' })
+  @JoinColumn()
+  asset?: AssetEntity;
+  @Column()
+  policyId?: Cardano.PolicyId;
+  @Column()
+  hasDatum?: boolean;
+}

packages/projection-typeorm/src/entity/index.ts

@@ -9,3 +9,4 @@ export * from './Asset.entity';
 export * from './Tokens.entity';
 export * from './Output.entity';
 export * from './CurrentPoolMetrics.entity';
+export * from './Handle.entity';

packages/projection-typeorm/src/operators/index.ts

@@ -5,5 +5,6 @@ export * from './withTypeormTransaction';
 export * from './storeBlock';
 export * from './storeAssets';
 export * from './storeUtxo';
+export * from './storeHandles';
 export * from './util';
 export * from './storePoolMetricsUpdateJob';

packages/projection-typeorm/src/operators/storeAssets.ts

@@ -1,49 +1,65 @@
-import { AssetEntity } from '../entity/Asset.entity';
+import { AssetEntity } from '../entity';
 import { Cardano, ChainSyncEventType } from '@cardano-sdk/core';
 import { Mappers } from '@cardano-sdk/projection';
+import { QueryRunner } from 'typeorm';
 import { typeormOperator } from './util';
 
 type MintedAssetSupplies = Partial<Record<Cardano.AssetId, bigint>>;
+type StoreAssetEventParams = {
+  mint: Mappers.Mint[];
+  queryRunner: QueryRunner;
+  header: Cardano.PartialBlockHeader;
+};
+
 export type WithMintedAssetSupplies = {
   mintedAssetTotalSupplies: MintedAssetSupplies;
 };
 
-export const storeAssets = typeormOperator<Mappers.WithMint, WithMintedAssetSupplies>(
-  // TODO: refactor
-  // eslint-disable-next-line sonarjs/cognitive-complexity
-  async ({ mint, block: { header }, eventType, queryRunner }) => {
-    const repository = queryRunner.manager.getRepository(AssetEntity);
-    const mintedAssetTotalSupplies: MintedAssetSupplies = {};
-    if (eventType === ChainSyncEventType.RollForward) {
-      for (const { assetId, quantity } of mint) {
-        const storedAsset = await repository.findOne({ select: { supply: true }, where: { id: assetId } });
-        if (storedAsset) {
-          const newSupply = storedAsset.supply! + quantity;
-          await repository.update({ id: assetId }, { supply: newSupply });
-          mintedAssetTotalSupplies[assetId] = newSupply;
-        } else {
-          await repository.insert({
-            firstMintBlock: { slot: header.slot },
-            id: assetId,
-            supply: quantity
-          });
-          mintedAssetTotalSupplies[assetId] = quantity;
-        }
-      }
+const rollForward = async ({ mint, queryRunner, header }: StoreAssetEventParams): Promise<MintedAssetSupplies> => {
+  const mintedAssetTotalSupplies: MintedAssetSupplies = {};
+  const repository = queryRunner.manager.getRepository(AssetEntity);
+  for (const { assetId, quantity } of mint) {
+    const storedAsset = await repository.findOne({ select: { supply: true }, where: { id: assetId } });
+    if (storedAsset) {
+      const newSupply = storedAsset.supply! + quantity;
+      await repository.update({ id: assetId }, { supply: newSupply });
+      mintedAssetTotalSupplies[assetId] = newSupply;
     } else {
-      for (const { assetId, quantity } of mint) {
-        const isPositiveQuantity = quantity > 0n;
-        const absQuantity = isPositiveQuantity ? quantity : -1n * quantity;
-        const queryResponse = await queryRunner.manager
-          .createQueryBuilder(AssetEntity, 'asset')
-          .update()
-          .set({ supply: () => `supply ${isPositiveQuantity ? '-' : '+'} ${absQuantity}` })
-          .where({ id: assetId })
-          .returning(['supply'])
-          .execute();
-        mintedAssetTotalSupplies[assetId] = queryResponse.affected === 0 ? 0n : BigInt(queryResponse.raw[0].supply);
-      }
+      await repository.insert({
+        firstMintBlock: { slot: header.slot },
+        id: assetId,
+        supply: quantity
+      });
+      mintedAssetTotalSupplies[assetId] = quantity;
     }
+  }
+  return mintedAssetTotalSupplies;
+};
+
+const rollBackward = async ({ mint, queryRunner }: StoreAssetEventParams): Promise<MintedAssetSupplies> => {
+  const mintedAssetTotalSupplies: MintedAssetSupplies = {};
+  for (const { assetId, quantity } of mint) {
+    const isPositiveQuantity = quantity > 0n;
+    const absQuantity = isPositiveQuantity ? quantity : -1n * quantity;
+    const queryResponse = await queryRunner.manager
+      .createQueryBuilder(AssetEntity, 'asset')
+      .update()
+      .set({ supply: () => `supply ${isPositiveQuantity ? '-' : '+'} ${absQuantity}` })
+      .where({ id: assetId })
+      .returning(['supply'])
+      .execute();
+    mintedAssetTotalSupplies[assetId] = queryResponse.affected === 0 ? 0n : BigInt(queryResponse.raw[0].supply);
+  }
+  return mintedAssetTotalSupplies;
+};
+
+export const storeAssets = typeormOperator<Mappers.WithMint, WithMintedAssetSupplies>(
+  async ({ mint, block: { header }, eventType, queryRunner }) => {
+    const storeAssetEventParams: StoreAssetEventParams = { header, mint, queryRunner };
+    const mintedAssetTotalSupplies: MintedAssetSupplies =
+      eventType === ChainSyncEventType.RollForward
+        ? await rollForward(storeAssetEventParams)
+        : await rollBackward(storeAssetEventParams);
     return { mintedAssetTotalSupplies };
   }
 );

packages/projection-typeorm/src/operators/storeHandles.ts

@@ -0,0 +1,119 @@
+import { AssetEntity, HandleEntity } from '../entity';
+import { Cardano, ChainSyncEventType } from '@cardano-sdk/core';
+import { Mappers } from '@cardano-sdk/projection';
+import { QueryRunner } from 'typeorm';
+import { WithMintedAssetSupplies } from './storeAssets';
+import { typeormOperator } from './util';
+
+type HandleWithTotalSupply = Mappers.Handle & { totalSupply: bigint };
+
+type HandleEventParams = {
+  handles: Array<HandleWithTotalSupply>;
+  mint: Mappers.Mint[];
+  queryRunner: QueryRunner;
+  block: Cardano.Block;
+};
+
+const getOwner = async (
+  queryRunner: QueryRunner,
+  assetId: string
+): Promise<{ cardanoAddress: Cardano.Address | null; hasDatum: boolean }> => {
+  const rows = await queryRunner.manager
+    .createQueryBuilder('tokens', 't')
+    .innerJoinAndSelect('output', 'o', 'o.id = t.output_id')
+    .select('address, o.datum')
+    .distinct()
+    .where('o.consumed_at_slot IS NULL')
+    .andWhere('t.asset_id = :assetId', { assetId })
+    .getRawMany();
+  if (rows.length !== 1)
+    return {
+      cardanoAddress: null,
+      hasDatum: false
+    };
+  return {
+    cardanoAddress: rows[0].address,
+    hasDatum: !!rows[0].datum
+  };
+};
+
+const getSupply = async (queryRunner: QueryRunner, assetId: Cardano.AssetId) => {
+  const asset = await queryRunner.manager
+    .getRepository(AssetEntity)
+    .findOne({ select: { supply: true }, where: { id: assetId } });
+  if (!asset) return 0n;
+  return asset.supply!;
+};
+
+const rollForward = async ({ handles, queryRunner }: HandleEventParams) => {
+  const handleRepository = queryRunner.manager.getRepository(HandleEntity);
+
+  for (const { assetId, handle, policyId, address, datum, totalSupply } of handles) {
+    if (totalSupply === 1n) {
+      // if !address then it's burning it, otherwise transferring
+      const { cardanoAddress, hasDatum } = address
+        ? { cardanoAddress: address, hasDatum: !!datum }
+        : await getOwner(queryRunner, assetId);
+      await handleRepository.upsert(
+        {
+          asset: assetId,
+          cardanoAddress,
+          handle,
+          hasDatum,
+          policyId
+        },
+        {
+          conflictPaths: {
+            handle: true
+          }
+        }
+      );
+    } else {
+      // multiple handles that leads to invalid tx
+      await handleRepository.update({ handle }, { cardanoAddress: null });
+    }
+  }
+};
+
+const rollBackward = async ({ handles, queryRunner }: HandleEventParams) => {
+  const handleRepository = queryRunner.manager.getRepository(HandleEntity);
+  for (const { assetId, handle, totalSupply } of handles) {
+    const newOwnerAddressAndDatum =
+      totalSupply === 1n ? await getOwner(queryRunner, assetId) : { cardanoAddress: null, hasDatum: false };
+    await handleRepository.update({ handle }, newOwnerAddressAndDatum);
+  }
+};
+
+const withTotalSupplies = (
+  queryRunner: QueryRunner,
+  handles: Mappers.Handle[],
+  mintedAssetTotalSupplies: WithMintedAssetSupplies['mintedAssetTotalSupplies']
+): Promise<HandleWithTotalSupply[]> =>
+  Promise.all(
+    handles.map(
+      async (handle): Promise<HandleWithTotalSupply> => ({
+        ...handle,
+        totalSupply: mintedAssetTotalSupplies[handle.assetId] || (await getSupply(queryRunner, handle.assetId))
+      })
+    )
+  );
+// const supply = totalSupplies[assetId] || ;
+
+export const storeHandles = typeormOperator<Mappers.WithHandles & Mappers.WithMint & WithMintedAssetSupplies>(
+  async ({ mint, handles, queryRunner, eventType, block, mintedAssetTotalSupplies }) => {
+    const handleEventParams: HandleEventParams = {
+      block,
+      handles: await withTotalSupplies(queryRunner, handles, mintedAssetTotalSupplies),
+      mint,
+      queryRunner
+    };
+
+    try {
+      eventType === ChainSyncEventType.RollForward
+        ? await rollForward(handleEventParams)
+        : await rollBackward(handleEventParams);
+    } catch (error) {
+      throw new Error((error as Error).message);
+    }
+  }
+);