feat: use prisma for queries

apps/server/package.json

@@ -11,13 +11,15 @@
   },
   "dependencies": {
     "@dotenvx/dotenvx": "1.43.0",
+    "@prisma/client": "6.7.0",
     "@sentry/node": "9.17.0",
     "@stacks/blockchain-api-client": "8.10.0",
     "@stacks/stacks-blockchain-api-types": "7.14.1",
     "@stackspulse/protocols": "workspace:*",
     "@t3-oss/env-core": "0.13.4",
     "better-sqlite3": "11.10.0",
-    "db0": "0.2.4",
+    "consola": "3.4.2",
+    "db0": "0.3.2",
     "drizzle-orm": "0.39.3",
     "h3": "1.15.3",
     "nitro-cors": "0.7.1",
@@ -29,6 +31,6 @@
   },
   "devDependencies": {
     "nitropack": "2.11.11",
-    "prisma": "6.3.1"
+    "prisma": "6.7.0"
   }
 }

apps/server/prisma/migrations/2_optimisations/migration.sql

@@ -0,0 +1,8 @@
+   CREATE INDEX IF NOT EXISTS idx_txs_contract_call_contract_id ON txs (contract_call_contract_id);
+   CREATE INDEX IF NOT EXISTS idx_principal_stx_txs_principal ON principal_stx_txs (principal);
+   CREATE INDEX IF NOT EXISTS idx_stx_events_sender ON stx_events (sender);
+   CREATE INDEX IF NOT EXISTS idx_stx_events_recipient ON stx_events (recipient);
+   CREATE INDEX IF NOT EXISTS idx_ft_events_sender ON ft_events (sender);
+   CREATE INDEX IF NOT EXISTS idx_ft_events_recipient ON ft_events (recipient);
+   CREATE INDEX IF NOT EXISTS idx_nft_events_sender ON nft_events (sender);
+   CREATE INDEX IF NOT EXISTS idx_nft_events_recipient ON nft_events (recipient);

apps/server/src/api/protocols/stackingdao/index.get.ts

@@ -1,5 +1,5 @@
-import { sql } from "~/db/db";
 import { apiCacheConfig } from "~/lib/api";
+import { prisma } from "~/lib/prisma";
 
 type StackingDAOProtocolStatsResponse = {
   month: string;
@@ -8,7 +8,13 @@ type StackingDAOProtocolStatsResponse = {
 }[];
 
 export default defineCachedEventHandler(async (event) => {
-  const result = await sql`
+  const result = await prisma.$queryRaw<
+    {
+      month: string;
+      deposits: bigint;
+      withdrawals: bigint;
+    }[]
+  >`
 WITH monthly_blocks AS (
     SELECT
         DATE_TRUNC('month', TO_TIMESTAMP(burn_block_time)) AS month,
@@ -26,16 +32,16 @@ deposits AS (
     SELECT
         mb.month,
         SUM(se.amount) AS deposits
-    FROM 
+    FROM
         monthly_blocks mb
-    JOIN 
+    JOIN
         stx_events se ON se.block_height BETWEEN mb.min_block_height AND mb.max_block_height
     WHERE
         se.recipient = 'SP4SZE494VC2YC5JYG7AYFQ44F5Q4PYV7DVMDPBG.reserve-v1'
         AND se.sender NOT LIKE 'SP4SZE494VC2YC5JYG7AYFQ44F5Q4PYV7DVMDPBG%'
         AND canonical = TRUE
         AND microblock_canonical = TRUE
-    GROUP BY 
+    GROUP BY
         mb.month
 ),
 
@@ -74,8 +80,8 @@ ORDER BY
     // format of the month is "2021-08-01 00:00:00+00"
     // we want to output "2021-08"
     month: row.month.slice(0, 7),
-    deposits: Number.parseInt(row.deposits),
-    withdrawals: Number.parseInt(row.withdrawals),
+    deposits: Number.parseInt(row.deposits.toString()),
+    withdrawals: Number.parseInt(row.withdrawals.toString()),
   }));
 
   return stats;

apps/server/src/api/protocols/users/index.ts

@@ -1,9 +1,9 @@
+import { Prisma } from "@prisma/client";
 import type { Protocol } from "@stackspulse/protocols";
-import type postgres from "postgres";
 import { z } from "zod";
-import { sql } from "~/db/db";
 import { apiCacheConfig } from "~/lib/api";
 import { getValidatedQueryZod } from "~/lib/nitro";
+import { prisma } from "~/lib/prisma";
 
 const protocolUsersRouteSchema = z.object({
   mode: z.enum(["direct", "nested"]).optional(),
@@ -27,14 +27,14 @@ export default defineCachedEventHandler(async (event) => {
   };
   const daysToSubtract = daysToSubtractMap[query.date];
 
-  let result: postgres.Row[];
+  let result: { protocol_name: Protocol; unique_senders: number }[];
   if (mode === "direct") {
     result = await getProtocolUsersDirect({
       limit,
       daysToSubtract,
     });
   } else {
-    result = await getProtocolUsersNested({
+    result = await getProtocolUsersNestedPrisma({
       limit,
       daysToSubtract,
     });
@@ -57,95 +57,89 @@ const getProtocolUsersDirect = async ({
   limit,
   daysToSubtract,
 }: QueryParams) => {
-  let dateCondition = "";
-  if (daysToSubtract) {
-    dateCondition = `AND txs.block_time >= EXTRACT(EPOCH FROM (NOW() - INTERVAL '${daysToSubtract} days'))`;
-  }
+  const dateCondition = daysToSubtract
+    ? Prisma.sql`AND txs.block_time >= EXTRACT(EPOCH FROM (NOW() - INTERVAL '${daysToSubtract} days'))`
+    : Prisma.sql``;
 
-  const result = await sql`
- SELECT
-     dapps.id as protocol_name,
-     COUNT(DISTINCT txs.sender_address) AS unique_senders
- FROM
-     txs
- JOIN
-     dapps ON txs.contract_call_contract_id = ANY (dapps.contracts)
- WHERE
-   txs.type_id = 2
-   ${sql.unsafe(dateCondition)}
- GROUP BY
-     dapps.id
- ORDER BY
-     unique_senders DESC
- LIMIT ${limit};
-   `;
+  const result = await prisma.$queryRaw<
+    {
+      contract_call_contract_id: string;
+      unique_senders: bigint;
+    }[]
+  >`
+    SELECT
+      dapps.id as protocol_name,
+      COUNT(DISTINCT txs.sender_address) AS unique_senders
+    FROM
+      txs
+    JOIN
+      dapps ON txs.contract_call_contract_id = ANY (dapps.contracts)
+    WHERE
+      type_id = 2
+      ${dateCondition}
+    GROUP BY
+      dapps.id
+    ORDER BY
+      unique_senders DESC
+    LIMIT ${limit}
+  `;
 
-  return result;
+  return result.map((r) => ({
+    protocol_name: r.contract_call_contract_id,
+    unique_senders: Number(r.unique_senders),
+  }));
 };
 
-const getProtocolUsersNested = async ({
+const getProtocolUsersNestedPrisma = async ({
   limit,
   daysToSubtract,
 }: QueryParams) => {
-  let dateCondition = "";
-  if (daysToSubtract) {
-    dateCondition = `AND txs.block_time >= EXTRACT(EPOCH FROM (NOW() - INTERVAL '${daysToSubtract} days'))`;
-  }
+  const dateCondition = daysToSubtract
+    ? Prisma.sql`AND txs.block_time >= EXTRACT(EPOCH FROM (NOW() - INTERVAL '${daysToSubtract} days'))`
+    : Prisma.sql``;
 
-  const result = await sql`
-WITH protocol_contracts AS (
-    SELECT id, UNNEST(contracts) AS contract_address
-    FROM dapps
-),
-
-address_txs AS (
-    SELECT DISTINCT tx_id, index_block_hash, microblock_hash, protocol_contracts.id AS protocol_name
-    FROM (
-        SELECT tx_id, index_block_hash, microblock_hash, contract_call_contract_id AS address
-        FROM txs
-        UNION ALL
-        SELECT tx_id, index_block_hash, microblock_hash, principal
-        FROM principal_stx_txs
-        UNION ALL
-        SELECT tx_id, index_block_hash, microblock_hash, sender
-        FROM stx_events
-        UNION ALL
-        SELECT tx_id, index_block_hash, microblock_hash, recipient
-        FROM stx_events
-        UNION ALL
-        SELECT tx_id, index_block_hash, microblock_hash, sender
-        FROM ft_events
-        UNION ALL
-        SELECT tx_id, index_block_hash, microblock_hash, recipient
-        FROM ft_events
-        UNION ALL
-        SELECT tx_id, index_block_hash, microblock_hash, sender
-        FROM nft_events
-        UNION ALL
-        SELECT tx_id, index_block_hash, microblock_hash, recipient
-        FROM nft_events
-    ) sub
-    JOIN protocol_contracts ON sub.address = protocol_contracts.contract_address
-)
-
-SELECT
-    atxs.protocol_name,
-    COUNT(DISTINCT txs.sender_address) AS unique_senders
-FROM
-    address_txs atxs
-JOIN
-    txs ON atxs.tx_id = txs.tx_id
-JOIN
-    blocks ON txs.block_height = blocks.block_height
-WHERE
-    1=1
-    ${sql.unsafe(dateCondition)}
-GROUP BY
-    atxs.protocol_name
-ORDER BY
-    unique_senders DESC
-LIMIT ${limit};
+  const result = await prisma.$queryRaw<
+    {
+      contract_call_contract_id: string;
+      unique_senders: bigint;
+    }[]
+  >`
+    WITH protocol_contracts AS (
+      SELECT
+        id,
+        UNNEST(contracts) AS contract_address
+      FROM
+        dapps
+    ),
+    address_txs AS (
+      SELECT DISTINCT
+        tx_id,
+        index_block_hash,
+        microblock_hash
+      FROM
+        txs
+      WHERE
+        contract_call_contract_id IN (SELECT contract_address FROM protocol_contracts)
+        ${dateCondition}
+        AND canonical = TRUE
+        AND microblock_canonical = TRUE
+    )
+    SELECT
+      contract_call_contract_id,
+      COUNT(DISTINCT sender_address) AS unique_senders
+    FROM
+      txs
+    WHERE
+      tx_id IN (SELECT tx_id FROM address_txs)
+    GROUP BY
+      contract_call_contract_id
+    ORDER BY
+      unique_senders DESC
+    LIMIT ${limit}
   `;
 
-  return result;
+  return result.map((r) => ({
+    protocol_name: r.contract_call_contract_id,
+    unique_senders: Number(r.unique_senders),
+  }));
 };

apps/server/src/api/tokens/transaction-volume.get.ts

@@ -1,7 +1,7 @@
 import { z } from "zod";
-import { sql } from "~/db/db";
 import { apiCacheConfig } from "~/lib/api";
 import { getValidatedQueryZod } from "~/lib/nitro";
+import { prisma } from "~/lib/prisma";
 
 const tokensTransactionVolumeRouteSchema = z.object({
   token: z.string(),
@@ -18,21 +18,26 @@ export default defineCachedEventHandler(async (event) => {
     tokensTransactionVolumeRouteSchema,
   );
 
-  const result = await sql`
-SELECT
-    DATE_TRUNC('day', TO_TIMESTAMP(blocks.burn_block_time)) AS date,
-    SUM(ft_events.amount::numeric) AS daily_volume
-FROM
-    ft_events
-JOIN
-    blocks ON ft_events.index_block_hash = blocks.index_block_hash
-WHERE
-    ft_events.asset_identifier = ${query.token}
-    AND ft_events.canonical = true
-GROUP BY
-    DATE_TRUNC('day', TO_TIMESTAMP(blocks.burn_block_time))
-ORDER BY
-    date;
+  const result = await prisma.$queryRaw<
+    {
+      date: string;
+      daily_volume: string;
+    }[]
+  >`
+    SELECT
+        DATE_TRUNC('day', TO_TIMESTAMP(blocks.burn_block_time)) AS date,
+        SUM(ft_events.amount::numeric) AS daily_volume
+    FROM
+        ft_events
+    JOIN
+        blocks ON ft_events.index_block_hash = blocks.index_block_hash
+    WHERE
+        ft_events.asset_identifier = ${query.token}
+        AND ft_events.canonical = true
+    GROUP BY
+        DATE_TRUNC('day', TO_TIMESTAMP(blocks.burn_block_time))
+    ORDER BY
+        date;
   `;
 
   const stats: TokensTransactionVolumeRouteResponse = result.map((row) => ({