import { Asset, LoggerInstance } from '@oceanprotocol/lib' import { AssetSelectionAsset } from '@shared/FormInput/InputElement/AssetSelection' import axios, { CancelToken, AxiosResponse } from 'axios' import { OrdersData_orders as OrdersData } from '../../@types/subgraph/OrdersData' import { metadataCacheUri } from '../../../app.config' import { SortDirectionOptions, SortTermOptions } from '../../@types/aquarius/SearchQuery' import { transformAssetToAssetSelection } from '../assetConvertor' export interface UserSales { id: string totalSales: number } export const MAXIMUM_NUMBER_OF_PAGES_WITH_RESULTS = 476 export function escapeEsReservedCharacters(value: string): string { // eslint-disable-next-line no-useless-escape const pattern = /([\!\*\+\-\=\<\>\&\|\(\)\[\]\{\}\^\~\?\:\\/"])/g return value?.replace(pattern, '\\$1') } /** * @param filterField the name of the actual field from the ddo schema e.g. 'id','service.attributes.main.type' * @param value the value of the filter * @returns json structure of the es filter */ export function getFilterTerm( filterField: string, value: string | number | boolean | number[] | string[] ): FilterTerm { const isArray = Array.isArray(value) return { [isArray ? 'terms' : 'term']: { [filterField]: value } } } // hardcoded did export function generateBaseQuery( baseQueryParams: BaseQueryParams ): SearchQuery { const filters: unknown[] = [getFilterTerm('_index', 'aquarius')] baseQueryParams.filters && filters.push(...baseQueryParams.filters) baseQueryParams.chainIds && filters.push(getFilterTerm('chainId', baseQueryParams.chainIds)) !baseQueryParams.ignorePurgatory && filters.push(getFilterTerm('purgatory.state', false)) !baseQueryParams.ignoreState && filters.push({ bool: { must_not: [ { term: { 'nft.state': 5 } } ], should: [ { terms: { 'metadata.tags.keyword': ['RegenRangersODC'] } }, { terms: { 'id.keyword': [ 'did:op:92067fe74892465ad77bb989abe48f67b76dfbb72bf547caa1805dfccb195f88', 'did:op:447ed4c1808d31f8aef9591fffea5f78a19a8518291174a9f1c159ff3c499669', 'did:op:7e984edf15c1b97d1ce4da9a01bc28d775d8c378305337da89868c859b5f5708', 'did:op:3ea3f29bba2b8db2a578603812a9750b74dd95f085a9840305bf0e910dcad240', 'did:op:a501d6eb47978d929ac2a9d4304a7997bcba9c2f11eaf9ebd8019ea4108645af', 'did:op:bd9991b34d925b3a7e3c8da7239ba9d6830d11794b76616201d5d3c16eb5064d', 'did:op:3a5f6705369b2a461753d292519fcf76d17bb5360cef9ef4b414f1397ebe2332', 'did:op:7df1cc9918f35163938ca219510a774c9ed80f8331c2743886fc54623acff218', 'did:op:8b133e8d3d2b37e94e25485e7c98503e505461b9095e78af6960666b5cdd4cf6', 'did:op:1d319077f7879e48b01aad52e4a69fc0ea06594c908575df4bd5cd015338b8cf', 'did:op:eac43d546ba84e5b82ddf4d2fbf4db9290711e8d2c2a167bce148b7209d41623', 'did:op:b2d554a5e09989bd81f1c54295fafab2a2784fb87ede3390a7103946f65df331', 'did:op:f5373f2561910d13d442a2598e5b52b6e831020fef0157319b963b9bed0a615b', 'did:op:8f69894fee4cbbeb5c8a5df9a2af58d3683bd8c989974b4a998dea2fb4d3e5cb', 'did:op:d0714e46ae0bf3f7c3488109d0bdd406f7f026ef5e99e9d25c4ce7ae61d7b572', 'did:op:1875f69b9ffe36d394f2aabd76e9e1f7a2b3f69c238fd9ddec177877c42f4117', 'did:op:80bdd287b517eebaed0fc7010d7909b3708d2d0ecb741c58168f89b3ebd012cc' ] } } ] } }) const generatedQuery = { from: baseQueryParams.esPaginationOptions?.from || 0, size: baseQueryParams.esPaginationOptions?.size >= 0 ? baseQueryParams.esPaginationOptions?.size : 1000, query: { bool: { ...baseQueryParams.nestedQuery, filter: filters } } } as SearchQuery if (baseQueryParams.aggs !== undefined) { generatedQuery.aggs = baseQueryParams.aggs } if (baseQueryParams.sortOptions !== undefined) generatedQuery.sort = { [baseQueryParams.sortOptions.sortBy]: baseQueryParams.sortOptions.sortDirection || SortDirectionOptions.Descending } return generatedQuery } export function transformQueryResult( queryResult: SearchResponse, from = 0, size = 21 ): PagedAssets { const result: PagedAssets = { results: [], page: 0, totalPages: 0, totalResults: 0, aggregations: [] } result.results = (queryResult.hits.hits || []).map( (hit) => hit._source as Asset ) result.aggregations = queryResult.aggregations result.totalResults = queryResult.hits.total.value result.totalPages = result.totalResults / size < 1 ? Math.floor(result.totalResults / size) : Math.ceil(result.totalResults / size) result.page = from ? from / size + 1 : 1 return result } export async function queryMetadata( query: SearchQuery, cancelToken: CancelToken ): Promise { try { const response: AxiosResponse = await axios.post( `${metadataCacheUri}/api/aquarius/assets/query`, { ...query }, { cancelToken } ) if (!response || response.status !== 200 || !response.data) return return transformQueryResult(response.data, query.from, query.size) } catch (error) { if (axios.isCancel(error)) { LoggerInstance.log(error.message) } else { LoggerInstance.error(error.message) } } } export async function getAsset( did: string, cancelToken: CancelToken ): Promise { try { const response: AxiosResponse = await axios.get( `${metadataCacheUri}/api/aquarius/assets/ddo/${did}`, { cancelToken } ) if (!response || response.status !== 200 || !response.data) return const data = { ...response.data } return data } catch (error) { if (axios.isCancel(error)) { LoggerInstance.log(error.message) } else { LoggerInstance.error(error.message) } } } export async function getAssetsNames( didList: string[], cancelToken: CancelToken ): Promise> { try { const response: AxiosResponse> = await axios.post( `${metadataCacheUri}/api/aquarius/assets/names`, { didList }, { cancelToken } ) if (!response || response.status !== 200 || !response.data) return return response.data } catch (error) { if (axios.isCancel(error)) { LoggerInstance.log(error.message) } else { LoggerInstance.error(error.message) } } } export async function getAssetsFromDids( didList: string[], chainIds: number[], cancelToken: CancelToken ): Promise { if (didList?.length === 0 || chainIds?.length === 0) return [] try { const orderedDDOListByDIDList: Asset[] = [] const baseQueryparams = { chainIds, filters: [getFilterTerm('_id', didList)], ignorePurgatory: true } as BaseQueryParams const query = generateBaseQuery(baseQueryparams) const result = await queryMetadata(query, cancelToken) didList.forEach((did: string) => { const ddo = result.results.find((ddo: Asset) => ddo.id === did) if (ddo) orderedDDOListByDIDList.push(ddo) }) return orderedDDOListByDIDList } catch (error) { LoggerInstance.error(error.message) } } export async function getAlgorithmDatasetsForCompute( algorithmId: string, datasetProviderUri: string, datasetChainId?: number, cancelToken?: CancelToken ): Promise { const baseQueryParams = { chainIds: [datasetChainId], nestedQuery: { must: { match: { 'services.compute.publisherTrustedAlgorithms.did': { query: algorithmId } } } }, sortOptions: { sortBy: SortTermOptions.Created, sortDirection: SortDirectionOptions.Descending } } as BaseQueryParams const query = generateBaseQuery(baseQueryParams) const computeDatasets = await queryMetadata(query, cancelToken) if (computeDatasets?.totalResults === 0) return [] const datasets = await transformAssetToAssetSelection( datasetProviderUri, computeDatasets.results, [] ) return datasets } export async function getPublishedAssets( accountId: string, chainIds: number[], cancelToken: CancelToken, ignoreState = false, page?: number, type?: string, accesType?: string ): Promise { if (!accountId) return const filters: FilterTerm[] = [] filters.push(getFilterTerm('nft.state', [0, 4, 5])) filters.push(getFilterTerm('nft.owner', accountId.toLowerCase())) accesType !== undefined && filters.push(getFilterTerm('services.type', accesType)) type !== undefined && filters.push(getFilterTerm('metadata.type', type)) const baseQueryParams = { chainIds, filters, sortOptions: { sortBy: SortTermOptions.Created, sortDirection: SortDirectionOptions.Descending }, aggs: { totalOrders: { sum: { field: SortTermOptions.Orders } } }, ignorePurgatory: true, ignoreState, esPaginationOptions: { from: (Number(page) - 1 || 0) * 9, size: 9 } } as BaseQueryParams const query = generateBaseQuery(baseQueryParams) try { const result = await queryMetadata(query, cancelToken) return result } catch (error) { if (axios.isCancel(error)) { LoggerInstance.log(error.message) } else { LoggerInstance.error(error.message) } } } async function getTopPublishers( chainIds: number[], cancelToken: CancelToken, page?: number, type?: string, accesType?: string ): Promise { const filters: FilterTerm[] = [] accesType !== undefined && filters.push(getFilterTerm('services.type', accesType)) type !== undefined && filters.push(getFilterTerm('metadata.type', type)) const baseQueryParams = { chainIds, filters, sortOptions: { sortBy: SortTermOptions.Created, sortDirection: SortDirectionOptions.Descending }, aggs: { topPublishers: { terms: { field: 'nft.owner.keyword', order: { totalSales: 'desc' } }, aggs: { totalSales: { sum: { field: SortTermOptions.Orders } } } } }, esPaginationOptions: { from: (Number(page) - 1 || 0) * 9, size: 9 } } as BaseQueryParams const query = generateBaseQuery(baseQueryParams) try { const result = await queryMetadata(query, cancelToken) return result } catch (error) { if (axios.isCancel(error)) { LoggerInstance.log(error.message) } else { LoggerInstance.error(error.message) } } } export async function getTopAssetsPublishers( chainIds: number[], nrItems = 9 ): Promise { const publishers: UserSales[] = [] const result = await getTopPublishers(chainIds, null) const { topPublishers } = result.aggregations for (let i = 0; i < topPublishers.buckets.length; i++) { publishers.push({ id: topPublishers.buckets[i].key, totalSales: parseInt(topPublishers.buckets[i].totalSales.value) }) } publishers.sort((a, b) => b.totalSales - a.totalSales) return publishers.slice(0, nrItems) } export async function getUserSales( accountId: string, chainIds: number[] ): Promise { try { const result = await getPublishedAssets(accountId, chainIds, null) const { totalOrders } = result.aggregations return totalOrders.value } catch (error) { LoggerInstance.error('Error getUserSales', error.message) } } export async function getDownloadAssets( dtList: string[], tokenOrders: OrdersData[], chainIds: number[], cancelToken: CancelToken, ignoreState = false ): Promise { const baseQueryparams = { chainIds, filters: [ getFilterTerm('services.datatokenAddress', dtList), getFilterTerm('services.type', 'access') ], ignorePurgatory: true, ignoreState } as BaseQueryParams const query = generateBaseQuery(baseQueryparams) try { const result = await queryMetadata(query, cancelToken) const downloadedAssets: DownloadedAsset[] = result.results .map((asset) => { const order = tokenOrders.find( ({ datatoken }) => datatoken?.address.toLowerCase() === asset.services[0].datatokenAddress.toLowerCase() ) return { asset, networkId: asset.chainId, dtSymbol: order?.datatoken?.symbol, timestamp: order?.createdTimestamp } }) .sort((a, b) => b.timestamp - a.timestamp) return downloadedAssets } catch (error) { if (axios.isCancel(error)) { LoggerInstance.log(error.message) } else { LoggerInstance.error(error.message) } } } export async function getTagsList( chainIds: number[], cancelToken: CancelToken ): Promise { const baseQueryParams = { chainIds, esPaginationOptions: { from: 0, size: 0 } } as BaseQueryParams const query = { ...generateBaseQuery(baseQueryParams), aggs: { tags: { terms: { field: 'metadata.tags.keyword', size: 1000 } } } } try { const response: AxiosResponse = await axios.post( `${metadataCacheUri}/api/aquarius/assets/query`, { ...query }, { cancelToken } ) if (response?.status !== 200 || !response?.data) return const { buckets }: { buckets: AggregatedTag[] } = response.data.aggregations.tags const tagsList = buckets .filter((tag) => tag.key !== '') .map((tag) => tag.key) return tagsList.sort() } catch (error) { if (axios.isCancel(error)) { LoggerInstance.log(error.message) } else { LoggerInstance.error(error.message) } } }