Fixes

buildcore-io · Apr 21, 2024 · 821ffd0 · 821ffd0
1 parent 1326528
commit 821ffd0
Show file tree

Hide file tree

Showing 19 changed files with 150 additions and 114 deletions.
diff --git a/packages/database/src/pg/impl/pubsub.ts b/packages/database/src/pg/impl/pubsub.ts
@@ -8,14 +8,15 @@ const topic = pubSub.topic('onupsert');
 export const subscriptions: { [key: string]: Promise<Subscription> } = {};
 
 export const getSubscription = (table: string) => {
-  const subsName = Math.random().toString().replace('0.', 'subs');
   if (!subscriptions[table]) {
+    const subsName = Math.random().toString().replace('0.', 'subs-onupsert-');
     subscriptions[table] = new Promise<Subscription>(async (res) => {
       const [subscription] = await topic.createSubscription(subsName, {
         filter: `attributes.table="${table}"`,
         ackDeadlineSeconds: 600,
         expirationPolicy: { ttl: { seconds: 86400 } },
       });
+      subscription.setMaxListeners(0);
       res(subscription);
     });
   }

diff --git a/packages/database/src/pg/impl/tables/milestone.ts b/packages/database/src/pg/impl/tables/milestone.ts
@@ -9,10 +9,10 @@ export class MilestoneConverter implements Converter<Milestone, PgMilestone> {
     uid: get(milestone, 'uid', ''),
     createdOn: milestone.createdOn?.toDate(),
     completed: milestone.completed,
-    completedOn: milestone.completedOn.toDate(),
+    completedOn: milestone.completedOn?.toDate(),
     listenerNodeId: milestone.listenerNodeId,
     milestone: milestone.milestone,
-    milestoneTimestamp: milestone.milestoneTimestamp.toDate(),
+    milestoneTimestamp: milestone.milestoneTimestamp?.toDate(),
     trxConflictCount: milestone.trxConflictCount,
     trxFailedCount: milestone.trxFailedCount,
     trxValidCount: milestone.trxValidCount,

diff --git a/packages/database/src/pg/impl/tables/space.ts b/packages/database/src/pg/impl/tables/space.ts
@@ -40,7 +40,7 @@ export class SpaceConverter implements Converter<Space, PgSpace> {
     alias_address: space.alias?.address,
     alias_aliasId: space.alias?.aliasId,
     alias_blockId: space.alias?.blockId,
-    alias_mintedOn: space.alias?.mintedOn.toDate(),
+    alias_mintedOn: space.alias?.mintedOn?.toDate(),
     alias_mintedBy: space.alias?.mintedBy,
   });
 

diff --git a/packages/database/src/pg/impl/tables/stake_reward.ts b/packages/database/src/pg/impl/tables/stake_reward.ts
@@ -12,9 +12,9 @@ export class StakeRewardConverter implements Converter<StakeReward, PgStakeRewar
     updatedOn: sr.updatedOn?.toDate(),
     createdBy: sr.createdBy,
 
-    startDate: sr.startDate.toDate(),
-    endDate: sr.endDate.toDate(),
-    tokenVestingDate: sr.tokenVestingDate.toDate(),
+    startDate: sr.startDate?.toDate(),
+    endDate: sr.endDate?.toDate(),
+    tokenVestingDate: sr.tokenVestingDate?.toDate(),
     tokensToDistribute: sr.tokensToDistribute,
     token: sr.token,
     totalStaked: sr.totalStaked,

diff --git a/packages/database/src/pg/interfaces/document.ts b/packages/database/src/pg/interfaces/document.ts
@@ -1,5 +1,5 @@
 import { COL, SUB_COL } from '@build-5/interfaces';
-import { Message, Subscription } from '@google-cloud/pubsub';
+import { Message } from '@google-cloud/pubsub';
 import { Knex } from 'knex';
 import { get, isArray, isDate, isObject, isUndefined, unset } from 'lodash';
 import { getPgData, getSubscription } from '../impl/pubsub';
@@ -42,7 +42,7 @@ export class IDocument<C, B extends BaseRecord, U extends Update> {
 
   onSnapshot = (
     callback: (data: C | undefined) => Promise<void> | void,
-    _error?: (err: any) => void,
+    onError?: (err: any) => void,
   ) => {
     const onMemssage = (message: Message) => {
       message.ack();
@@ -51,16 +51,20 @@ export class IDocument<C, B extends BaseRecord, U extends Update> {
         callback(this.converter.fromPg(update));
       }
     };
-    let subscription: Subscription | undefined = undefined;
 
-    this.getQuery().then(async (raw) => {
-      callback(raw ? this.converter.fromPg(raw) : undefined);
-      subscription = await getSubscription(this.table);
-      subscription.on('message', onMemssage);
-    });
+    const subsPromise = getSubscription(this.table);
+
+    this.getQuery()
+      .then(async (raw) => {
+        callback(raw ? this.converter.fromPg(raw) : undefined);
+        (await subsPromise).on('message', onMemssage);
+      })
+      .catch(onError);
 
     return () => {
-      subscription?.removeListener('message', onMemssage);
+      subsPromise.then((subs) => {
+        subs.removeListener('message', onMemssage);
+      });
     };
   };
 

diff --git a/packages/database/src/pg/interfaces/query.ts b/packages/database/src/pg/interfaces/query.ts
@@ -1,5 +1,5 @@
 import { COL, SUB_COL } from '@build-5/interfaces';
-import { Message, Subscription } from '@google-cloud/pubsub';
+import { Message } from '@google-cloud/pubsub';
 import { Knex } from 'knex';
 import { get, gt, gte, isEqual, lt, lte, orderBy, take } from 'lodash';
 import { getPgData, getSubscription } from '../impl/pubsub';
@@ -24,7 +24,7 @@ export class IQuery<C, Q extends BaseRecord> {
     this.table = getTableName(col);
   }
 
-  private getQuery = () => {
+  private getQuery = async () => {
     let query = this.con(this.table).select('*');
 
     for (const ins of this.whereIns) {
@@ -78,7 +78,14 @@ export class IQuery<C, Q extends BaseRecord> {
       query = query.orderBy(orderBy.key, orderBy.dir);
     }
 
-    return query;
+    const sql = query.toQuery();
+    const queryPlan = await this.con.raw(`EXPLAIN ${sql}`);
+    const usesIndex = queryPlan.rows.some((row: any) => row['QUERY PLAN'].includes('Index'));
+    if (!usesIndex) {
+      throw new Error('Query requires and index.\n' + sql);
+    }
+
+    return await query;
   };
 
   get = async (): Promise<C[]> => {
@@ -147,7 +154,7 @@ export class IQuery<C, Q extends BaseRecord> {
     return act;
   };
 
-  onSnapshot = (callback: (data: C[]) => Promise<void> | void, error?: (err: any) => void) => {
+  onSnapshot = (callback: (data: C[]) => Promise<void> | void, onError?: (err: any) => void) => {
     let pgData: Q[] = [];
 
     const onMemssage = (message: Message) => {
@@ -161,29 +168,24 @@ export class IQuery<C, Q extends BaseRecord> {
           callback(callbackData);
         }
       } catch (err) {
-        console.error('onSnapshot-onMemssage', this.table, this.getQuery().toSQL(), err);
-        error?.({ code: 500, message: 'Internal server error' });
+        onError?.(err);
       }
     };
 
-    let subscription: Subscription | undefined = undefined;
+    const subsPromise = getSubscription(this.table);
 
     this.getQuery()
       .then(async (raw: Q[]) => {
         pgData = raw;
-
         callback(raw.map(this.converter.fromPg));
-
-        subscription = await getSubscription(this.table);
-        subscription.on('message', onMemssage);
+        (await subsPromise).on('message', onMemssage);
       })
-      .catch((err) => {
-        console.error('onSnapshot-getQuery', this.table, this.getQuery().toSQL(), err);
-        error?.({ code: 500, message: 'Internal server error' });
-      });
+      .catch(onError);
 
     return () => {
-      subscription?.removeListener('message', onMemssage);
+      subsPromise.then((subs) => {
+        subs.removeListener('message', onMemssage);
+      });
     };
   };
 
@@ -206,7 +208,7 @@ export class IQuery<C, Q extends BaseRecord> {
     return this;
   };
 
-  startAfter = (data: C) => {
+  startAfter = (data: C | undefined) => {
     this.startAfterData = data;
     return this;
   };

diff --git a/packages/functions/src/controls/collection/collection-mint.control.ts b/packages/functions/src/controls/collection/collection-mint.control.ts
@@ -6,6 +6,7 @@ import {
   CollectionStatus,
   CollectionType,
   Network,
+  Nft,
   TRANSACTION_AUTO_EXPIRY_MS,
   Transaction,
   TransactionPayloadType,
@@ -137,17 +138,16 @@ const getNftsTotalStorageDeposit = async (
 ) => {
   let storageDeposit = 0;
   let nftsToMint = 0;
-  let lastUid = '';
+  let lastDoc: Nft | undefined = undefined;
   do {
-    const nfts = await build5Db()
+    const nfts: Nft[] = await build5Db()
       .collection(COL.NFT)
       .where('collection', '==', collection.uid)
       .where('placeholderNft', '==', false)
-      .where('uid', '>', lastUid)
-      .orderBy('uid')
+      .startAfter(lastDoc)
       .limit(500)
       .get();
-    lastUid = last(nfts)?.uid || '';
+    lastDoc = last(nfts);
 
     const promises = nfts.map(async (nft) => {
       if (unsoldMintingOptions === UnsoldMintingOptions.BURN_UNSOLD && !nft.sold) {
@@ -162,7 +162,7 @@ const getNftsTotalStorageDeposit = async (
     const amounts = await Promise.all(promises);
     storageDeposit += amounts.reduce((acc, act) => acc + act, 0);
     nftsToMint += amounts.filter((a) => a !== 0).length;
-  } while (lastUid);
+  } while (lastDoc);
 
   return { storageDeposit, nftsToMint };
 };

diff --git a/packages/search/src/common.ts b/packages/search/src/common.ts
@@ -28,40 +28,61 @@ const queryStrToParams = (url: string) => {
 export const isHiddenNft = (dataset: Dataset, data?: any) =>
   dataset === Dataset.NFT && data?.hidden === true;
 
-export const queryToObservable = <C>(query: IQuery<C, BaseRecord>) =>
-  new Observable<C[]>((observer) => {
-    const unsubscribe = query.onSnapshot(
-      (data) => {
-        observer.next(data);
-      },
-      (error) => {
-        observer.error(error);
-      },
-    );
-    return () => {
-      unsubscribe();
-    };
+export const queryToObservable = <C>(query: IQuery<C, BaseRecord>, isLive: boolean) => {
+  if (isLive) {
+    return new Observable<C[]>((obs) => {
+      const unsubscribe = query.onSnapshot(
+        (data) => {
+          obs.next(data);
+        },
+        (error) => {
+          obs.error(error);
+        },
+      );
+      return () => {
+        unsubscribe();
+      };
+    });
+  }
+
+  return new Observable<C[]>((obs) => {
+    query
+      .get()
+      .then((r) => obs.next(r))
+      .catch((err) => obs.error(err));
   });
+};
 
 export const documentToObservable = <C, B extends BaseRecord, U extends Update>(
   doc: IDocument<C, B, U>,
-) =>
-  new Observable<C>((observer) => {
-    const unsubscribe = doc.onSnapshot(
-      (data) => {
-        observer.next(data);
-      },
-      (error) => {
-        observer.error(error);
-      },
-    );
-    return () => {
-      unsubscribe();
-    };
+  isLive: boolean,
+): Observable<C> => {
+  if (isLive) {
+    return new Observable<C>((observer) => {
+      const unsubscribe = doc.onSnapshot(
+        (data) => {
+          observer.next(data);
+        },
+        (error) => {
+          observer.error(error);
+        },
+      );
+      return () => {
+        unsubscribe();
+      };
+    });
+  }
+
+  return new Observable<C>((obs) => {
+    doc
+      .get()
+      .then((r) => obs.next(r))
+      .catch((err) => obs.error(err));
   });
+};
 
-export const getHeadPriceObs = (query: IQuery<TokenPurchase, PgTokenPurchase>) =>
-  queryToObservable(query).pipe(map((r) => head(r)?.price || 0));
+export const getHeadPriceObs = (query: IQuery<TokenPurchase, PgTokenPurchase>, isLive: boolean) =>
+  queryToObservable(query, isLive).pipe(map((r) => head(r)?.price || 0));
 
 // Used to be 42, changed to 5 to support milestone get and transactions subset
 export const minAddressLength = 5;

diff --git a/packages/search/src/getAddresses.ts b/packages/search/src/getAddresses.ts
@@ -13,7 +13,7 @@ const getAddressesSchema = Joi.object({
   createdAfter: Joi.number().min(0).max(MAX_MILLISECONDS).integer().required(),
 });
 
-export const getAddresses = async (url: string) => {
+export const getAddresses = async (url: string, isLive: boolean) => {
   const body = getQueryParams<GetAddressesRequest>(url, getAddressesSchema);
 
   const query = build5Db()
@@ -23,7 +23,7 @@ export const getAddresses = async (url: string) => {
     .orderBy('createdOn')
     .limit(1000);
 
-  return queryToObservable(query).pipe(map((mnemonics) => mnemonics.map(sanitizeMnemonic)));
+  return queryToObservable(query, isLive).pipe(map((mnemonics) => mnemonics.map(sanitizeMnemonic)));
 };
 
 const sanitizeMnemonic = (mnemonic: Mnemonic) => ({

diff --git a/packages/search/src/getAvgPrice.ts b/packages/search/src/getAvgPrice.ts
@@ -13,20 +13,20 @@ const getAvgPriceSchema = Joi.object({
     .required(),
 });
 
-export const getAvgPrice = async (url: string) => {
+export const getAvgPrice = async (url: string, isLive: boolean) => {
   const body = getQueryParams<GetAvgPriceRequest>(url, getAvgPriceSchema);
 
   const tokens = Array.isArray(body.token) ? body.token : [body.token];
-  const changes = tokens.map(getAvgLive);
+  const changes = tokens.map((t) => getAvgLive(t, isLive));
   const observable = combineLatest(changes).pipe(map((r) => (r.length === 1 ? r[0] : r)));
 
   return observable;
 };
 
-const getAvgLive = (token: string) => {
-  const lowestPurchaseObs = getHeadPriceObs(purchaseQuery(token, true));
-  const highestPurchaseObs = getHeadPriceObs(purchaseQuery(token, false));
-  const lastPurchaseObs = getHeadPriceObs(purchaseQuery(token));
+const getAvgLive = (token: string, isLive: boolean) => {
+  const lowestPurchaseObs = getHeadPriceObs(purchaseQuery(token, true), isLive);
+  const highestPurchaseObs = getHeadPriceObs(purchaseQuery(token, false), isLive);
+  const lastPurchaseObs = getHeadPriceObs(purchaseQuery(token), isLive);
   return combineLatest([lowestPurchaseObs, highestPurchaseObs, lastPurchaseObs]).pipe(
     map(([lowest, highest, last]) => (highest + lowest + last) / 3),
     map((avg) => ({ id: token, avg })),

diff --git a/packages/search/src/getById.ts b/packages/search/src/getById.ts
@@ -21,7 +21,7 @@ const getByIdSchema = Joi.object({
   subsetId: CommonJoi.uid(false, 7),
 });
 
-export const getById = async (url: string) => {
+export const getById = async (url: string, isLive: boolean) => {
   const body = getQueryParams<GetByIdRequest>(url, getByIdSchema);
 
   const docRef = build5Db().doc(
@@ -31,7 +31,7 @@ export const getById = async (url: string) => {
     body.subsetId,
   )! as unknown as IDocument<any, BaseRecord, Update>;
 
-  const observable = documentToObservable(docRef).pipe(
+  const observable = documentToObservable(docRef, isLive).pipe(
     map((data) => {
       if (!data || isHiddenNft(body.dataset, data)) {
         return {};

diff --git a/packages/search/src/getMany.ts b/packages/search/src/getMany.ts
@@ -48,7 +48,7 @@ const getManySchema = Joi.object({
   startAfter: CommonJoi.uid(false),
 });
 
-export const getMany = async (project: string, url: string) => {
+export const getMany = async (project: string, url: string, isLive: boolean) => {
   const body = getQueryParams<GetManyRequest>(url, getManySchema);
 
   let query = build5Db()
@@ -96,7 +96,7 @@ export const getMany = async (project: string, url: string) => {
     }
   }
 
-  const observable = queryToObservable(query).pipe(
+  const observable = queryToObservable(query, isLive).pipe(
     map((snap) => snap.filter((d) => !isEmpty(d)).map((d) => ({ id: d.uid, ...d }))),
   );
   return observable;