Make snapshot query paginable

Author: Park Juhyung

src/models/logic/utxo.ts

@@ -1,5 +1,7 @@
 import { H160, H256, U64 } from "codechain-sdk/lib/core/classes";
+import * as _ from "lodash";
 import * as Sequelize from "sequelize";
+import sequelize = require("sequelize");
 import models from "..";
 import * as Exception from "../../exception";
 import { utxoPagination } from "../../routers/pagination";
@@ -447,48 +449,125 @@ export async function getByTxHashIndex(
     }
 }
 
-export async function getSnapshot(assetType: H256, blockNumber: number) {
+export async function getSnapshot(params: {
+    assetType: H256;
+    blockNumber: number;
+    lastEvaluatedKey?: number[] | null;
+}) {
+    const { assetType, blockNumber, lastEvaluatedKey } = params;
+    const transaction = await models.sequelize.transaction({
+        isolationLevel:
+            models.Sequelize.Transaction.ISOLATION_LEVELS.REPEATABLE_READ,
+        deferrable: models.Sequelize.Deferrable.SET_DEFERRED
+    });
     try {
-        return models.UTXO.findAll({
-            where: {
-                assetType: strip0xPrefix(assetType.value),
-                usedBlockNumber: {
-                    [Sequelize.Op.or]: [
-                        { [Sequelize.Op.gt]: blockNumber },
-                        { [Sequelize.Op.eq]: null }
-                    ]
+        const [
+            fromBlockNumber,
+            fromTransactionIndex,
+            fromTransactionOutputIndex
+        ] = lastEvaluatedKey || [Number.MAX_SAFE_INTEGER, 0, 0];
+        const itemsPerPage = 100;
+
+        const rows = await models.sequelize.query(
+            `SELECT SUM("UTXO"."quantity") AS "totalAssetQuantity", "UTXO"."address", "UTXO"."assetType",
+            COUNT("UTXO"."assetType") AS "utxoQuantity"
+            FROM (SELECT * FROM "UTXOs" WHERE ("blockNumber", "transactionIndex", "transactionOutputIndex")<(:fromBlockNumber, :fromTransactionIndex, :fromTransactionOutputIndex)
+              AND "assetType"=:assetType
+              ORDER BY "blockNumber" DESC, "transactionIndex" DESC, "transactionOutputIndex" DESC
+              LIMIT :itemsPerPage) "UTXO"
+            WHERE ("UTXO"."usedBlockNumber" > :blockNumber OR "UTXO"."usedBlockNumber" IS NULL)
+              AND "UTXO"."blockNumber" <= :blockNumber
+            GROUP BY "UTXO"."address", "UTXO"."assetType"
+        `,
+            {
+                replacements: {
+                    fromBlockNumber,
+                    fromTransactionIndex,
+                    fromTransactionOutputIndex,
+                    blockNumber,
+                    assetType: strip0xPrefix(assetType.toString()),
+                    itemsPerPage
                 },
-                blockNumber: {
-                    [Sequelize.Op.lte]: blockNumber
-                }
-            },
-            attributes: [
-                [
-                    Sequelize.fn("SUM", Sequelize.col("quantity")),
-                    "totalAssetQuantity"
-                ],
-                "address",
-                "assetType",
-                [
-                    Sequelize.fn("COUNT", Sequelize.col("UTXO.assetType")),
-                    "utxoQuantity"
-                ]
-            ],
-            order: Sequelize.literal(
-                `"totalAssetQuantity" DESC, "assetType" DESC`
-            ),
-            include: [
+                raw: true,
+                transaction,
+                type: sequelize.QueryTypes.SELECT
+            }
+        );
+
+        // The SQL query is copied from the query above.
+        const hasNextPage =
+            (await models.sequelize.query(
+                `SELECT COUNT(*) as count
+                FROM (SELECT id FROM "UTXOs"
+                WHERE ("blockNumber", "transactionIndex", "transactionOutputIndex")<(:fromBlockNumber, :fromTransactionIndex, :fromTransactionOutputIndex)
+                    AND "assetType"=:assetType
+                ORDER BY "blockNumber" DESC, "transactionIndex" DESC, "transactionOutputIndex" DESC
+                LIMIT :itemsPerPage) "UTXO"`,
                 {
-                    as: "assetScheme",
-                    model: models.AssetScheme
+                    replacements: {
+                        fromBlockNumber,
+                        fromTransactionIndex,
+                        fromTransactionOutputIndex,
+                        assetType: strip0xPrefix(assetType.toString()),
+                        itemsPerPage: itemsPerPage + 1
+                    },
+                    plain: true,
+                    raw: true,
+                    transaction,
+                    type: sequelize.QueryTypes.SELECT
                 }
-            ],
-            group: ["UTXO.address", "UTXO.assetType", "assetScheme.assetType"]
-        }).then(instances =>
-            instances.map(instance => instance.get({ plain: true }))
+            )).count ===
+            itemsPerPage + 1;
+
+        // The SQL query is copied from the query above.
+        const lastRow = await models.sequelize.query(
+            `SELECT *
+            FROM (SELECT * FROM "UTXOs" WHERE ("blockNumber", "transactionIndex", "transactionOutputIndex")<(:fromBlockNumber, :fromTransactionIndex, :fromTransactionOutputIndex)
+                  AND "assetType"=:assetType
+                  ORDER BY "blockNumber" DESC, "transactionIndex" DESC, "transactionOutputIndex" DESC
+                  LIMIT :itemsPerPage) "UTXO"
+            ORDER BY "blockNumber" ASC, "transactionIndex" ASC, "transactionOutputIndex" ASC
+            LIMIT 1
+            `,
+            {
+                replacements: {
+                    fromBlockNumber,
+                    fromTransactionIndex,
+                    fromTransactionOutputIndex,
+                    assetType: strip0xPrefix(assetType.toString()),
+                    itemsPerPage
+                },
+                plain: true,
+                raw: true,
+                transaction,
+                type: sequelize.QueryTypes.SELECT
+            }
         );
+
+        const assetScheme = await AssetSchemeModel.getByAssetType(assetType, {
+            transaction
+        });
+        transaction.commit();
+        return {
+            data: _.map(rows, (row: any) => {
+                row.assetScheme =
+                    assetScheme && assetScheme.get({ plain: true });
+                return row;
+            }),
+            hasNextPage,
+            hasPreviousPage: null,
+            firstEvaluatedKey: null,
+            lastEvaluatedKey: lastRow
+                ? JSON.stringify([
+                      lastRow.blockNumber,
+                      lastRow.transactionIndex,
+                      lastRow.transactionOutputIndex
+                  ])
+                : null
+        };
     } catch (err) {
         console.error(err);
+        transaction.rollback();
         throw Exception.DBError();
     }
 }

src/routers/asset.ts

@@ -16,6 +16,7 @@ import {
     aggsUTXOPaginationSchema,
     assetTypeSchema,
     paginationSchema,
+    snapshotPaginationSchema,
     snapshotSchema,
     utxoPaginationSchema,
     utxoSchema,
@@ -513,6 +514,11 @@ export function handle(context: IndexerContext, router: Router) {
      *         in: query
      *         required: true
      *         type: string
+     *       - name: lastEvaluatedKey
+     *         description: the evaulated key of the last item in the previous page. It will be used for the pagination
+     *         in: query
+     *         required: false
+     *         type: string
      *     responses:
      *       200:
      *         description: snapshot, return null if the block does not exist yet
@@ -523,21 +529,32 @@ export function handle(context: IndexerContext, router: Router) {
      *               type: integer
      *             blockHash:
      *               type: string
-     *             snapshot:
+     *             data:
      *               type: array
      *               items:
      *                 $ref: '#/definitions/UTXO'
+     *             hasNextPage:
+     *               type: string
+     *             hasPreviousPage:
+     *               type: string
+     *             firstEvaluatedKey:
+     *               type: string
+     *             lastEvaluatedKey:
+     *               type: string
      */
     router.get(
         "/snapshot",
+        parseEvaluatedKey,
         validate({
             query: {
-                ...snapshotSchema
+                ...snapshotSchema,
+                ...snapshotPaginationSchema
             }
         }),
         async (req, res, next) => {
             const assetTypeString = req.query.assetType;
             const date = req.query.date;
+            const lastEvaluatedKey = req.query.lastEvaluatedKey;
             try {
                 const assetType = new H160(assetTypeString);
                 const snapshotTime = moment(date);
@@ -553,14 +570,15 @@ export function handle(context: IndexerContext, router: Router) {
                     return;
                 }
 
-                const snapshot = await UTXOModel.getSnapshot(
+                const snapshot = await UTXOModel.getSnapshot({
                     assetType,
-                    block.get("number")
-                );
+                    blockNumber: block.get("number"),
+                    lastEvaluatedKey
+                });
                 res.json({
                     blockHash: block.get("hash"),
                     blockNumber: block.get("number"),
-                    snapshot
+                    ...snapshot
                 });
             } catch (e) {
                 next(e);

src/routers/validator.ts

@@ -103,6 +103,14 @@ export const pendingTxPaginationSchema = {
     lastEvaluatedKey: Joi.array().items(Joi.number())
 };
 
+export const snapshotPaginationSchema = {
+    lastEvaluatedKey: Joi.array().items(
+        Joi.number(),
+        Joi.number(),
+        Joi.number()
+    )
+};
+
 export const txSchema = {
     address,
     assetType: assetTypeSchema,