Merge pull request #309 from metaDAOproject/fix/missing-trades

fix: not having user account....

packages/indexer/src/builders/swaps.ts

@@ -66,6 +66,25 @@ export class SwapPersistable {
           )}`
         );
       }
+      // Insert user if they aren't already in the database
+      const insertUsersResult = (await usingDb((db) =>
+          db
+            .insert(schema.users)
+            .values({ userAcct: this.ordersRecord.actorAcct })
+            .onConflictDoNothing()
+            .returning({ userAcct: schema.users.userAcct })
+        )) ?? [];
+      if (
+        insertUsersResult.length !== 1 ||
+        insertUsersResult[0].userAcct !== this.ordersRecord.actorAcct
+      ) {
+        logger.warn(
+          `Failed to upsert user ${this.ordersRecord.actorAcct}. ${JSON.stringify(
+            this.ordersRecord
+          )}`
+        );
+      }
+
       // const priceInsertRes =
       //   (await usingDb((db) =>
       //     db
@@ -164,11 +183,14 @@ export class SwapBuilder {
         const mintIx = tx.instructions?.find(
           (i) => i.name === "mintConditionalTokens"
         );
+        // What if there's more than one?
         const mergeIx = tx.instructions?.find((i) => i.name === "mergeConditionalTokensForUnderlyingTokens");
+
         if (mergeIx && mintIx) {
           console.error("ARB TRANSACTION DETECTED")
           return Err({ type: SwapPersistableError.ArbTransactionError });
         }
+
         const result = await this.buildOrderFromSwapIx(swapIx, tx, mintIx);
         if (!result.success) {
           return Err(result.error);
@@ -344,16 +366,23 @@ export class SwapBuilder {
       return Err({ type: AmmInstructionIndexerError.MissingMarket });
     }
 
-    const ammPrice =
-      quoteAmount.toString() && baseAmount.toString()
-        ? quoteAmount.mul(new BN(10).pow(new BN(12))).div(baseAmount)
-        : new BN(0);
+    let price: number | null = null;
+
+    if (quoteAmount.toString() && baseAmount.toString()) {
+      console.log(quoteAmount.toString(), baseAmount.toString());
+      try{
+        const ammPrice = quoteAmount.mul(new BN(10).pow(new BN(12))).div(baseAmount)
 
-    const price = getHumanPrice(
-      ammPrice,
-      baseToken[0].decimals,
-      quoteToken[0].decimals
-    );
+        price = getHumanPrice(
+          ammPrice,
+          baseToken[0].decimals,
+          quoteToken[0].decimals
+        );
+      } catch (e) {
+        logger.error("error getting price", e);
+        return Err({ type: SwapPersistableError.GeneralError });
+      }
+    }
     // TODO: Need to likely handle rounding.....
     // index a swap here
 
@@ -365,7 +394,7 @@ export class SwapBuilder {
       orderBlock: tx.slot.toString(),
       orderTime: now,
       orderTxSig: signature,
-      quotePrice: price.toString(),
+      quotePrice: price?.toString() ?? "0",
       actorAcct: userAcct.pubkey,
       // TODO: If and only if the transaction is SUCCESSFUL does this value equal this..
       filledBaseAmount: baseAmount.toString(),
@@ -384,7 +413,7 @@ export class SwapBuilder {
       orderBlock: tx.slot.toString(),
       orderTime: now,
       orderTxSig: signature,
-      quotePrice: price.toString(),
+      quotePrice: price?.toString() ?? "0",
       // TODO: this is coded into the market, in the case of our AMM, it's 1%
       // this fee is based on the INPUT value (so if we're buying its USDC, selling its TOKEN)
       takerBaseFee: BigInt(0),

packages/indexer/src/indexers/amm-market/amm-market-logs-subscribe-indexer.ts

@@ -20,7 +20,9 @@ export const AmmMarketLogsSubscribeIndexer: AccountLogsIndexer = {
         buildRes.error.type === SwapPersistableError.AlreadyPersistedSwap ||
         (buildRes.error.type === SwapPersistableError.TransactionParseError &&
           buildRes.error.value?.type ===
-            GetTransactionErrorType.NullGetTransactionResponse)
+            GetTransactionErrorType.NullGetTransactionResponse) ||
+        buildRes.error.type === SwapPersistableError.PriceError ||
+        buildRes.error.type === SwapPersistableError.ArbTransactionError
       ) {
         logger.error(
           `error with indexing amm logs, signature: ${logs.signature}`,

packages/indexer/src/indexers/amm-market/utils.ts

@@ -18,6 +18,7 @@ export enum AmmMarketAccountIndexingErrors {
   MarketMissingError = "MarketMissingError",
   AmmV4TwapIndexError = "AmmV4TwapIndexError",
   AmmTwapPriceError = "AmmTwapPriceError",
+  AmmTwapNoInsertError = "AmmTwapNoInsertError",
 }
 
 export async function indexAmmMarketAccountWithContext(
@@ -77,24 +78,29 @@ export async function indexAmmMarketAccountWithContext(
       lastPrice: ammMarketAccount.oracle.lastPrice.toString(),
     };
 
+    try{
     // TODO batch commits across inserts - maybe with event queue
-    const twapUpsertResult = await usingDb((db) =>
-      db
-        .insert(schema.twaps)
-        .values(newTwap)
-        .onConflictDoNothing()
-        .returning({ marketAcct: schema.twaps.marketAcct })
-    );
+      const twapUpsertResult = await usingDb((db) =>
+        db
+          .insert(schema.twaps)
+          .values(newTwap)
+          .onConflictDoNothing()
+          .returning({ marketAcct: schema.twaps.marketAcct })
+      );
 
-    if (twapUpsertResult === undefined || twapUpsertResult.length === 0) {
-      logger.error("failed to upsert twap");
-      return Err({ type: AmmMarketAccountIndexingErrors.AmmTwapIndexError });
+      if (twapUpsertResult === undefined || twapUpsertResult.length === 0) {
+        logger.error("failed to upsert twap", newTwap);
+        // return Err({ type: AmmMarketAccountIndexingErrors.AmmTwapNoInsertError });
+      }
+    } catch (e) {
+      logger.error("failed to upsert twap", e);
+      return Err({ type: AmmMarketAccountIndexingErrors.AmmTwapNoInsertError });
     }
   }
 
   let priceFromReserves: BN;
 
-  if (ammMarketAccount.baseAmount.toString() === "0" || ammMarketAccount.baseAmount.toString() === "0") {
+  if (ammMarketAccount.baseAmount.toNumber() === 0 || ammMarketAccount.baseAmount.toNumber() === 0) {
     logger.error("NO RESERVES", ammMarketAccount);
     return Ok("no price from reserves");
   }

packages/indexer/src/indexers/autocrat/autocrat-proposal-indexer.ts

@@ -76,7 +76,8 @@ export const AutocratProposalIndexer: IntervalFetchIndexer = {
           )
         )?.[0] ?? {};
 
-      if (!currentSlot || !currentTime) return;
+      console.log("currentSlot", currentSlot);
+      if (!currentSlot || !currentTime) return Err({ type: AutocratDaoIndexerError.MissingParamError });
 
       logger.log("Autocrat proposal indexer");
       const dbProposals: ProposalRecord[] =
@@ -94,7 +95,8 @@ export const AutocratProposalIndexer: IntervalFetchIndexer = {
       for (const proposal of onChainProposals) {
         if (
           !dbProposals.find((dbProposal) =>
-            new PublicKey(dbProposal.proposalAcct).equals(proposal.publicKey)
+            new PublicKey(dbProposal.proposalAcct).equals(proposal.publicKey) &&
+            dbProposal.endedAt === null
           )
         ) {
           proposalsToInsert.push(proposal);

packages/indexer/src/indexers/jupiter/jupiter-quotes-indexer.ts

@@ -110,6 +110,11 @@ export const fetchQuoteFromJupe = async (
           .where(eq(schema.tokens.mintAcct, quoteMint))
           .execute()
       )) ?? [];
+
+    if (baseToken.length === 0 || quoteToken.length === 0) {
+      console.log("quote or base token not found in db for jupiter quotes indexer", acct);
+      return null;
+    }
 
     const amountVal = 1 * 10 ** baseToken[0].decimals;
 

packages/indexer/src/types/errors.ts

@@ -10,4 +10,5 @@ export enum SwapPersistableError {
   NonSwapTransaction = "NonSwapTransaction",
   TransactionParseError = "TransactionParseError",
   ArbTransactionError = "ArbTransactionError",
+  PriceError = "PriceError",
 }