feat: sync cached data to network assets

Author: beeman

api-schema.graphql

@@ -491,6 +491,7 @@ type Mutation {
   adminAddCommunityMember(communityId: String!, input: AdminAddCommunityMemberInput!): CommunityMember
   adminCacheConfigSet(key: CacheConfigKey!, value: String!): Boolean
   adminCacheResolve(cacheId: String!, cluster: NetworkCluster!): JSON
+  adminCacheSync(cacheId: String!, cluster: NetworkCluster!): JSON
   adminCleanupNetworkAssets(cluster: NetworkCluster!): Boolean
   adminCreateBackup: Boolean!
   adminCreateBot(input: AdminCreateBotInput!): Bot

libs/api/cache/data-access/src/lib/api-cache.data-access.module.ts

@@ -1,6 +1,7 @@
 import { Module } from '@nestjs/common'
 import { ApiCacheConfigDataAccessModule } from '@pubkey-link/api-cache-config-data-access'
 import { ApiCoreDataAccessModule } from '@pubkey-link/api-core-data-access'
+import { ApiNetworkAssetDataAccessModule } from '@pubkey-link/api-network-asset-data-access'
 import { ApiNetworkDataAccessModule } from '@pubkey-link/api-network-data-access'
 import { ApiNetworkTokenDataAccessModule } from '@pubkey-link/api-network-token-data-access'
 import { ApiCacheService } from './api-cache.service'
@@ -10,6 +11,7 @@ import { ApiCacheService } from './api-cache.service'
     ApiCacheConfigDataAccessModule,
     ApiCoreDataAccessModule,
     ApiNetworkDataAccessModule,
+    ApiNetworkAssetDataAccessModule,
     ApiNetworkTokenDataAccessModule,
   ],
   providers: [ApiCacheService],

libs/api/cache/data-access/src/lib/api-cache.service.ts

@@ -1,6 +1,6 @@
 import { Injectable, Logger } from '@nestjs/common'
 import { OnEvent } from '@nestjs/event-emitter'
-import { NetworkCluster, NetworkToken, NetworkTokenType } from '@prisma/client'
+import { NetworkCluster, NetworkToken, NetworkTokenType, Prisma } from '@prisma/client'
 import {
   createResolver,
   getResolverOwner,
@@ -11,13 +11,15 @@ import {
 } from '@pubkey-cache/resolver'
 import { ApiCacheConfigService, EVENT_CACHE_CONFIG_LOADED } from '@pubkey-link/api-cache-config-data-access'
 import { ApiCoreService } from '@pubkey-link/api-core-data-access'
+import { ApiNetworkAssetService } from '@pubkey-link/api-network-asset-data-access'
 import { ApiNetworkTokenService } from '@pubkey-link/api-network-token-data-access'
 import { DAS, Helius } from 'helius-sdk'
 import { Storage } from 'unstorage'
 import { ApiCache } from './api-cache'
 import { createTraitCountMap, sortAssetsByName, TraitCountMap } from './create-trait-count-map'
 import { CacheStatus } from './entity/cache-status'
 import { getStorage } from './get-storage'
+import { formatSnapshot } from './helpers/format-snapshot'
 
 export interface CachedResult {
   cluster: NetworkCluster
@@ -57,6 +59,7 @@ export class ApiCacheService {
   constructor(
     private readonly core: ApiCoreService,
     private readonly cacheConfig: ApiCacheConfigService,
+    private readonly networkAsset: ApiNetworkAssetService,
     private readonly networkToken: ApiNetworkTokenService,
   ) {
     this.storage = getStorage({ redisUrl: this.core.config.redisUrl })
@@ -188,9 +191,10 @@ export class ApiCacheService {
           result,
           storage: cache.storage,
         })
+        await this.syncCacheNetworkAssets({ cluster: cache.cluster, id: resolver.id })
+
         const endTimeResolver = new Date().getTime()
         const durationResolver = endTimeResolver - startTimeResolver
-
         results.push(
           `Synced resolver ${resolver.id}, wrote ${writeCount} items to storage (${durationResolver / 1000} seconds)`,
         )
@@ -219,22 +223,27 @@ export class ApiCacheService {
   }
 
   async assetsSnapshot(param: { cluster: NetworkCluster; id: string }) {
+    try {
+      return await this.getSnapshot(param)
+    } catch (e) {
+      this.logger.error(`Error getting snapshot for ${param.id}`, e)
+      throw e
+    }
+  }
+
+  private async getSnapshot(param: { cluster: NetworkCluster; id: string }): Promise<CachedResult> {
     const { cache, resolver } = await this.getResolver(param)
     const result = await getResult({ resolver, storage: cache.storage })
+    if (!result) {
+      throw new Error('No result found')
+    }
 
     const cachedAt = result?.cachedAt ? new Date(result.cachedAt) : new Date()
-    const cacheAge = new Date().getTime() - cachedAt.getTime()
-    return result
-      ? {
-          id: result.id,
-          cachedAt,
-          cacheAge,
-          type: result.type,
-          total: result.total,
-          traits: result.traits,
-          items: result.items,
-        }
-      : { error: 'No snapshot found' }
+    return {
+      ...result,
+      cluster: param.cluster,
+      cachedAt,
+    }
   }
 
   private async getResolver(param: { cluster: NetworkCluster; id: string }) {
@@ -283,4 +292,34 @@ export class ApiCacheService {
         throw new Error(`Unknown resolver type: ${resolver.type}`)
     }
   }
+
+  async syncCacheNetworkAssets(param: { cluster: NetworkCluster; id: string }) {
+    const snapshot = await this.getSnapshot({ cluster: param.cluster, id: param.id })
+    if (!snapshot) {
+      throw new Error(`Snapshot not found for ${param.id}`)
+    }
+    const networkToken = await this.getSnapshotNetworkToken(snapshot)
+    const assets: Prisma.NetworkAssetCreateInput[] = formatSnapshot({
+      items: snapshot.items,
+      networkToken,
+      type: snapshot.type,
+    })
+    if (assets.length) {
+      this.logger.verbose(`syncCache: Upserting ${assets.length} assets`)
+      await this.networkAsset.sync.upsertAssets({ cluster: snapshot.cluster, assets, linkIdentity: false })
+    }
+    return {
+      total: snapshot.total,
+      items: snapshot.items,
+    }
+  }
+
+  private async getSnapshotNetworkToken(snapshot: CachedResult): Promise<NetworkToken> {
+    const account = snapshot.id?.split(':')[1]
+    const found = await this.core.data.networkToken.findFirst({ where: { account, cluster: snapshot.cluster } })
+    if (!found) {
+      throw new Error(`getSnapshotToken: Token ${account} not found on cluster ${snapshot.cluster}`)
+    }
+    return found
+  }
 }

libs/api/cache/data-access/src/lib/helpers/format-snapshot.ts

@@ -0,0 +1,113 @@
+import { Logger } from '@nestjs/common'
+import { NetworkResolver, NetworkToken, NetworkTokenType, Prisma } from '@prisma/client'
+import { ResolverType } from '@pubkey-cache/resolver'
+import { NetworkAssetInput } from '@pubkey-link/api-network-util'
+import { DAS } from 'helius-sdk'
+
+export function formatSnapshot({
+  items,
+  networkToken,
+  type,
+}: {
+  items: (DAS.GetAssetResponse | DAS.TokenAccounts)[]
+  networkToken: NetworkToken
+  type: ResolverType
+}): Prisma.NetworkAssetCreateInput[] {
+  switch (type) {
+    case ResolverType['helius-collection-assets']:
+      return (items as DAS.GetAssetResponse[]).map((asset) => convertHeliusAssetResponse({ asset, networkToken }))
+    case ResolverType['helius-token-accounts']:
+      return (items as DAS.TokenAccounts[]).map((item) => convertHeliusTokenAccounts({ item, networkToken }))
+    default:
+      return []
+  }
+}
+
+function convertHeliusTokenAccounts({
+  item,
+  networkToken,
+}: {
+  item: DAS.TokenAccounts
+  networkToken: NetworkToken
+}): NetworkAssetInput {
+  return {
+    network: { connect: { cluster: networkToken.cluster } },
+    resolver: NetworkResolver.SolanaFungible,
+    type: NetworkTokenType.Fungible,
+    account: item.address?.toString() ?? '',
+    name: networkToken.name ?? '',
+    symbol: networkToken.symbol ?? '',
+    owner: item.owner ?? '',
+    imageUrl: networkToken.imageUrl ?? '',
+    balance: item.amount?.toString() ?? '0',
+    group: item.mint ?? '',
+    decimals: 0,
+    mint: item.mint ?? '',
+    program: networkToken.program ?? '',
+  }
+}
+
+function convertHeliusAssetResponse({
+  asset,
+  networkToken,
+}: {
+  asset: DAS.GetAssetResponse
+  networkToken: NetworkToken
+}): NetworkAssetInput {
+  return {
+    network: { connect: { cluster: networkToken.cluster } },
+    resolver: NetworkResolver.SolanaNonFungible,
+    type: NetworkTokenType.NonFungible,
+    account: asset.id,
+    name: asset.content?.metadata.name ?? '',
+    symbol: asset.content?.metadata.symbol,
+    owner: asset.ownership.owner,
+    group: networkToken.account,
+    decimals: 0,
+    balance: '1',
+    burnt: asset.burnt,
+    mint: asset.id,
+    program: asset.token_info?.token_program,
+    imageUrl: asset.content?.files?.length ? asset.content.files[0].uri : '',
+    metadata: (asset.content?.metadata ?? {}) as Prisma.InputJsonValue,
+    attributes: convertHeliusAssetResponseAttributes(asset),
+  }
+}
+
+export function convertHeliusAssetResponseAttributes(asset: DAS.GetAssetResponse): [string, string][] {
+  const attributes = asset.content?.metadata.attributes?.length ? asset.content?.metadata.attributes : []
+
+  if (typeof attributes?.filter !== 'function') {
+    Logger.error(
+      `Invalid attributes for asset ${asset.id}: ${JSON.stringify(asset.content?.metadata)}`,
+      'getDasApiAssetAttributes',
+    )
+    return []
+  }
+
+  const main = attributes.length
+    ? (attributes
+        .filter((s) => !!s)
+        .filter((s) => s.trait_type?.length && s.value?.length)
+        .map((s) => [s.trait_type?.toString(), s.value?.toString()]) as [string, string][])
+    : []
+
+  const attributesExt = getAdditionalMetadata(asset)
+  const ext = attributesExt.length ? attributesExt : []
+
+  return [...main, ...ext]
+}
+
+// Get the Token2022 additional metadata from the mint extensions
+function getAdditionalMetadata(asset: DAS.GetAssetResponse): [string, string][] {
+  if (
+    asset.mint_extensions &&
+    asset.mint_extensions['metadata'] &&
+    asset.mint_extensions['metadata']['additionalMetadata']
+  ) {
+    const pairs = asset.mint_extensions['metadata']['additionalMetadata'] as unknown as [string, string][]
+    // Only return any items if they have a value with a length
+    return pairs.filter((pair) => pair[1]?.length)
+  }
+  return []
+}

libs/api/cache/feature/src/lib/api-cache-admin.resolver.ts

@@ -17,10 +17,7 @@ export class ApiCacheAdminResolver {
 
   @Query(() => GraphQLJSON, { nullable: true })
   adminCacheDetail(
-    @Args({
-      name: 'cluster',
-      type: () => NetworkCluster,
-    })
+    @Args({ name: 'cluster', type: () => NetworkCluster })
     cluster: NetworkCluster,
     @Args('cacheId') cacheId: string,
   ) {
@@ -29,13 +26,19 @@ export class ApiCacheAdminResolver {
 
   @Mutation(() => GraphQLJSON, { nullable: true })
   adminCacheResolve(
-    @Args({
-      name: 'cluster',
-      type: () => NetworkCluster,
-    })
+    @Args({ name: 'cluster', type: () => NetworkCluster })
     cluster: NetworkCluster,
     @Args('cacheId') cacheId: string,
   ) {
     return this.service.resolveCache({ cluster, id: cacheId })
   }
+
+  @Mutation(() => GraphQLJSON, { nullable: true })
+  adminCacheSync(
+    @Args({ name: 'cluster', type: () => NetworkCluster })
+    cluster: NetworkCluster,
+    @Args('cacheId') cacheId: string,
+  ) {
+    return this.service.syncCacheNetworkAssets({ cluster, id: cacheId })
+  }
 }

libs/api/collection/data-access/src/lib/api-collection.service.ts

@@ -37,22 +37,27 @@ export class ApiCollectionService {
       throw new Error(`Resolver ${collection.token.account} not found`)
     }
 
-    const snapshot = await this.cache.assetsSnapshot({ cluster: collection.token.cluster, id: resolver.id })
-    const items: DAS.GetAssetResponse[] = (snapshot.items ?? []) as DAS.GetAssetResponse[]
+    try {
+      const snapshot = await this.cache.assetsSnapshot({ cluster: collection.token.cluster, id: resolver.id })
+      const items: DAS.GetAssetResponse[] = (snapshot.items ?? []) as DAS.GetAssetResponse[]
 
-    const assets = items.map((asset) => ({
-      id: asset.id,
-      name: asset.content?.metadata?.name ?? '',
-      description: asset.content?.metadata?.description ?? '',
-      imageUrl: asset.content?.files?.[0]?.uri ?? '',
-      owner: asset.ownership.owner,
-      attributes: renameAttributes(asset.content?.metadata?.attributes ?? []),
-    }))
+      const assets = items.map((asset) => ({
+        id: asset.id,
+        name: asset.content?.metadata?.name ?? '',
+        description: asset.content?.metadata?.description ?? '',
+        imageUrl: asset.content?.files?.[0]?.uri ?? '',
+        owner: asset.ownership.owner,
+        attributes: renameAttributes(asset.content?.metadata?.attributes ?? []),
+      }))
 
-    return {
-      ...collection,
-      attributes: accumulateAttributes(assets),
-      assets,
+      return {
+        ...collection,
+        attributes: accumulateAttributes(assets),
+        assets,
+      }
+    } catch (e) {
+      console.log('error', e)
+      throw e
     }
   }