Merge pull request #302 from gitcoinco/fix/block-from-timestamp

Fix getBlockFromTimestamp

package.json

@@ -29,6 +29,7 @@
   "license": "ISC",
   "dependencies": {
     "@sentry/node": "^7.51.0",
+    "better-sqlite3": "^8.6.0",
     "chainsauce": "github:gitcoinco/chainsauce#main",
     "cors": "^2.8.5",
     "csv-parser": "^3.0.0",
@@ -69,6 +70,7 @@
     "@types/supertest": "^2.0.12",
     "@types/throttle-debounce": "^5.0.0",
     "@types/tmp": "^0.2.3",
+    "@types/better-sqlite3": "^7.6.5",
     "@types/write-file-atomic": "^4.0.0",
     "@typescript-eslint/eslint-plugin": "^5.55.0",
     "@typescript-eslint/parser": "^5.55.0",

src/blockCache.ts

@@ -0,0 +1,22 @@
+export type Block = {
+  chainId: number;
+  blockNumber: bigint;
+  timestampInSecs: number;
+};
+
+export interface BlockCache {
+  init(): Promise<void>;
+  getTimestampByBlockNumber(
+    chainId: number,
+    blockNumber: bigint
+  ): Promise<number | null>;
+  getBlockNumberByTimestamp(
+    chainId: number,
+    timestampInSecs: number
+  ): Promise<bigint | null>;
+  saveBlock(block: Block): Promise<void>;
+  getClosestBoundsForTimestamp(
+    chainId: number,
+    timestampInSecs: number
+  ): Promise<{ before: Block | null; after: Block | null }>;
+}

src/blockCache/sqlite.test.ts

@@ -0,0 +1,97 @@
+import { createSqliteBlockCache } from "./sqlite.js";
+import { BlockCache } from "../blockCache.js";
+import { it, describe, expect, beforeEach, afterEach } from "vitest";
+import Sqlite from "better-sqlite3";
+import fs from "fs/promises";
+import os from "os";
+import path from "path";
+
+describe("createSqliteBlockCache", () => {
+  let db: Sqlite.Database;
+  let blockCache: BlockCache;
+
+  beforeEach(() => {
+    db = new Sqlite(":memory:");
+    blockCache = createSqliteBlockCache({ db });
+  });
+
+  afterEach(() => {
+    db.close();
+  });
+
+  it("should initialize without errors", async () => {
+    await expect(blockCache.init()).resolves.not.toThrow();
+  });
+
+  it("should initialize if using invalid table name", () => {
+    expect(() => {
+      createSqliteBlockCache({
+        db,
+        tableName: "invalid table name",
+      });
+    }).toThrow();
+
+    expect(() => {
+      createSqliteBlockCache({
+        db,
+        tableName: "table/",
+      });
+    }).toThrow();
+  });
+
+  it("should throw if already initialized", async () => {
+    await blockCache.init();
+    await expect(blockCache.init()).rejects.toThrow("Already initialized");
+  });
+
+  it("should save and retrieve a block by number", async () => {
+    await blockCache.init();
+    const block = {
+      chainId: 1,
+      blockNumber: BigInt(1),
+      timestampInSecs: 12345,
+    };
+    await blockCache.saveBlock(block);
+
+    const timestampInSecs = await blockCache.getTimestampByBlockNumber(
+      1,
+      BigInt(1)
+    );
+    expect(timestampInSecs).toEqual(block.timestampInSecs);
+  });
+
+  it("should save and retrieve a block by timestamp", async () => {
+    await blockCache.init();
+    const block = {
+      chainId: 1,
+      blockNumber: BigInt(1),
+      timestampInSecs: 12345,
+    };
+    await blockCache.saveBlock(block);
+
+    const blockNumber = await blockCache.getBlockNumberByTimestamp(1, 12345);
+    expect(blockNumber).toEqual(block.blockNumber);
+  });
+
+  it("should get closest bounds for timestamp", async () => {
+    await blockCache.init();
+    const block1 = { chainId: 1, blockNumber: BigInt(1), timestampInSecs: 10 };
+    const block2 = { chainId: 1, blockNumber: BigInt(2), timestampInSecs: 20 };
+
+    await blockCache.saveBlock(block1);
+    await blockCache.saveBlock(block2);
+
+    const bounds = await blockCache.getClosestBoundsForTimestamp(1, 15);
+    expect(bounds.before).toEqual(block1);
+    expect(bounds.after).toEqual(block2);
+  });
+});
+
+describe("createSqliteBlockCache with dbPath", () => {
+  it("should initialize without errors using dbPath", async () => {
+    const tmpFilePath = path.join(os.tmpdir(), `tmpdb-${Date.now()}.db`);
+    const diskBlockCache = createSqliteBlockCache({ dbPath: tmpFilePath });
+    await expect(diskBlockCache.init()).resolves.not.toThrow();
+    await fs.rm(tmpFilePath);
+  });
+});

src/blockCache/sqlite.ts

@@ -0,0 +1,168 @@
+import Sqlite from "better-sqlite3";
+import { BlockCache, Block } from "../blockCache.js";
+
+const defaultTableName = "blocks";
+
+export type Options =
+  | {
+      dbPath: string;
+      tableName?: string;
+    }
+  | { db: Sqlite.Database; tableName?: string };
+
+interface Row {
+  chainId: number;
+  blockNumber: string;
+  timestamp: number;
+}
+
+type UninitializedState = { state: "uninitialized" };
+type InitializedState = {
+  state: "initialized";
+  db: Sqlite.Database;
+  getTimestampByBlockNumberStmt: Sqlite.Statement;
+  getBlockNumberByTimestampStmt: Sqlite.Statement;
+  saveBlockStmt: Sqlite.Statement;
+  getBeforeStmt: Sqlite.Statement;
+  getAfterStmt: Sqlite.Statement;
+};
+type State = UninitializedState | InitializedState;
+
+export function createSqliteBlockCache(opts: Options): BlockCache {
+  let dbState: State = { state: "uninitialized" };
+
+  if (opts.tableName !== undefined && /[^a-zA-Z0-9_]/.test(opts.tableName)) {
+    throw new Error(`Table name ${opts.tableName} has invalid characters.`);
+  }
+
+  const tableName = opts.tableName ?? defaultTableName;
+
+  return {
+    async init(): Promise<void> {
+      if (dbState.state === "initialized") {
+        throw new Error("Already initialized");
+      }
+
+      const db = "db" in opts ? opts.db : new Sqlite(opts.dbPath);
+
+      db.exec("PRAGMA journal_mode = WAL;");
+
+      // TODO: Add proper migrations, with Kysely?
+      db.exec(
+        `CREATE TABLE IF NOT EXISTS ${tableName} (
+          chainId INTEGER,
+          blockNumber TEXT,
+          timestamp INTEGER,
+          PRIMARY KEY (chainId, blockNumber)
+        )`
+      );
+
+      db.exec(
+        `CREATE INDEX IF NOT EXISTS idx_chainId_timestamp_blockNumber 
+         ON ${tableName} (chainId, timestamp, blockNumber DESC);`
+      );
+
+      dbState = {
+        state: "initialized",
+        db,
+        getTimestampByBlockNumberStmt: db.prepare(
+          `SELECT * FROM ${tableName} WHERE chainId = ? AND blockNumber = ?`
+        ),
+        getBlockNumberByTimestampStmt: db.prepare(
+          `SELECT * FROM ${tableName} WHERE chainId = ? AND timestamp = ?`
+        ),
+        saveBlockStmt: db.prepare(
+          `INSERT OR REPLACE INTO ${tableName} (chainId, blockNumber, timestamp) VALUES (?, ?, ?)`
+        ),
+        getBeforeStmt: db.prepare(
+          `SELECT * FROM ${tableName} WHERE chainId = ? AND timestamp < ? ORDER BY timestamp DESC, blockNumber DESC LIMIT 1`
+        ),
+        getAfterStmt: db.prepare(
+          `SELECT * FROM ${tableName} WHERE chainId = ? AND timestamp >= ? ORDER BY timestamp ASC, blockNumber ASC LIMIT 1`
+        ),
+      };
+
+      return Promise.resolve();
+    },
+
+    async getTimestampByBlockNumber(
+      chainId,
+      blockNumber
+    ): Promise<number | null> {
+      if (dbState.state === "uninitialized") {
+        throw new Error("SQLite database not initialized");
+      }
+
+      const row = dbState.getTimestampByBlockNumberStmt.get(
+        chainId,
+        blockNumber.toString()
+      ) as Row | undefined;
+
+      return Promise.resolve(row ? row.timestamp : null);
+    },
+
+    async getBlockNumberByTimestamp(
+      chainId,
+      timestamp
+    ): Promise<bigint | null> {
+      if (dbState.state === "uninitialized") {
+        throw new Error("SQLite database not initialized");
+      }
+
+      const row = dbState.getBlockNumberByTimestampStmt.get(
+        chainId,
+        timestamp
+      ) as Row | undefined;
+
+      return Promise.resolve(row ? BigInt(row.blockNumber) : null);
+    },
+
+    async saveBlock(block: Block): Promise<void> {
+      if (dbState.state === "uninitialized") {
+        throw new Error("SQLite database not initialized");
+      }
+
+      dbState.saveBlockStmt.run(
+        block.chainId,
+        block.blockNumber.toString(),
+        block.timestampInSecs
+      );
+
+      return Promise.resolve();
+    },
+
+    async getClosestBoundsForTimestamp(
+      chainId,
+      timestamp
+    ): Promise<{ before: Block | null; after: Block | null }> {
+      if (dbState.state === "uninitialized") {
+        throw new Error("SQLite database not initialized");
+      }
+
+      const before = dbState.getBeforeStmt.get(chainId, timestamp) as
+        | Row
+        | undefined;
+
+      const after = dbState.getAfterStmt.get(chainId, timestamp) as
+        | Row
+        | undefined;
+
+      return Promise.resolve({
+        before: before
+          ? {
+              chainId: before.chainId,
+              timestampInSecs: before.timestamp,
+              blockNumber: BigInt(before.blockNumber),
+            }
+          : null,
+        after: after
+          ? {
+              chainId: after.chainId,
+              timestampInSecs: after.timestamp,
+              blockNumber: BigInt(after.blockNumber),
+            }
+          : null,
+      });
+    },
+  };
+}

src/index.ts

@@ -223,6 +223,7 @@ async function catchupAndWatchChain(
       rpcProvider,
       chain: config.chain,
       logger: chainLogger.child({ subsystem: "PriceUpdater" }),
+      blockCachePath: path.join(config.storageDir, "..", "blockCache.db"),
       withCacheFn:
         pricesCache === null
           ? undefined