save

Author: belopash

typeorm/typeorm-store/src/database.ts

@@ -4,7 +4,9 @@ import assert from 'assert'
 import {DataSource, EntityManager} from 'typeorm'
 import {ChangeWriter, rollbackBlock} from './utils/changeWriter'
 import {DatabaseState, FinalTxInfo, HashAndHeight, HotTxInfo} from './interfaces'
-import {Store} from './store'
+import {CacheMode, FlushMode, ResetMode, Store} from './store'
+import {createLogger} from '@subsquid/logger'
+import {StateManager} from './utils/stateManager'
 import {sortMetadatasInCommitOrder} from './utils/commitOrder'
 
 
@@ -14,14 +16,21 @@ export type IsolationLevel = 'SERIALIZABLE' | 'READ COMMITTED' | 'REPEATABLE REA
 export interface TypeormDatabaseOptions {
     supportHotBlocks?: boolean
     isolationLevel?: IsolationLevel
+    flushMode?: FlushMode
+    resetMode?: ResetMode
+    cacheMode?: CacheMode
     stateSchema?: string
     projectDir?: string
 }
 
+const STATE_MANAGERS: WeakMap<DataSource, StateManager> = new WeakMap()
 
 export class TypeormDatabase {
     private statusSchema: string
     private isolationLevel: IsolationLevel
+    private flushMode: FlushMode
+    private resetMode: ResetMode
+    private cacheMode: CacheMode
     private con?: DataSource
     private projectDir: string
 
@@ -30,6 +39,9 @@ export class TypeormDatabase {
     constructor(options?: TypeormDatabaseOptions) {
         this.statusSchema = options?.stateSchema || 'squid_processor'
         this.isolationLevel = options?.isolationLevel || 'SERIALIZABLE'
+        this.resetMode = options?.resetMode || 'BATCH'
+        this.flushMode = options?.flushMode || 'AUTO'
+        this.cacheMode = options?.cacheMode || 'ALL'
         this.supportsHotBlocks = options?.supportHotBlocks !== false
         this.projectDir = options?.projectDir || process.cwd()
     }
@@ -43,48 +55,46 @@ export class TypeormDatabase {
         await this.con.initialize()
 
         try {
-            return await this.con.transaction('SERIALIZABLE', em => this.initTransaction(em))
-        } catch(e: any) {
+            return await this.con.transaction('SERIALIZABLE', (em) => this.initTransaction(em))
+        } catch (e: any) {
             await this.con.destroy().catch(() => {}) // ignore error
             this.con = undefined
             throw e
         }
     }
 
     async disconnect(): Promise<void> {
-        await this.con?.destroy().finally(() => this.con = undefined)
+        await this.con?.destroy().finally(() => (this.con = undefined))
     }
 
     private async initTransaction(em: EntityManager): Promise<DatabaseState> {
         let schema = this.escapedSchema()
 
-        await em.query(
-            `CREATE SCHEMA IF NOT EXISTS ${schema}`
-        )
+        await em.query(`CREATE SCHEMA IF NOT EXISTS ${schema}`)
         await em.query(
             `CREATE TABLE IF NOT EXISTS ${schema}.status (` +
-            `id int4 primary key, ` +
-            `height int4 not null, ` +
-            `hash text DEFAULT '0x', ` +
-            `nonce int4 DEFAULT 0`+
-            `)`
+                `id int4 primary key, ` +
+                `height int4 not null, ` +
+                `hash text DEFAULT '0x', ` +
+                `nonce int4 DEFAULT 0` +
+                `)`
         )
-        await em.query( // for databases created by prev version of typeorm store
+        await em.query(
+            // for databases created by prev version of typeorm store
             `ALTER TABLE ${schema}.status ADD COLUMN IF NOT EXISTS hash text DEFAULT '0x'`
         )
-        await em.query( // for databases created by prev version of typeorm store
-            `ALTER TABLE ${schema}.status ADD COLUMN IF NOT EXISTS nonce int DEFAULT 0`
-        )
         await em.query(
-            `CREATE TABLE IF NOT EXISTS ${schema}.hot_block (height int4 primary key, hash text not null)`
+            // for databases created by prev version of typeorm store
+            `ALTER TABLE ${schema}.status ADD COLUMN IF NOT EXISTS nonce int DEFAULT 0`
         )
+        await em.query(`CREATE TABLE IF NOT EXISTS ${schema}.hot_block (height int4 primary key, hash text not null)`)
         await em.query(
             `CREATE TABLE IF NOT EXISTS ${schema}.hot_change_log (` +
-            `block_height int4 not null references ${schema}.hot_block on delete cascade, ` +
-            `index int4 not null, ` +
-            `change jsonb not null, ` +
-            `PRIMARY KEY (block_height, index)` +
-            `)`
+                `block_height int4 not null references ${schema}.hot_block on delete cascade, ` +
+                `index int4 not null, ` +
+                `change jsonb not null, ` +
+                `PRIMARY KEY (block_height, index)` +
+                `)`
         )
 
         let status: (HashAndHeight & {nonce: number})[] = await em.query(
@@ -95,9 +105,7 @@ export class TypeormDatabase {
             status.push({height: -1, hash: '0x', nonce: 0})
         }
 
-        let top: HashAndHeight[] = await em.query(
-            `SELECT height, hash FROM ${schema}.hot_block ORDER BY height`
-        )
+        let top: HashAndHeight[] = await em.query(`SELECT height, hash FROM ${schema}.hot_block ORDER BY height`)
 
         return assertStateInvariants({...status[0], top})
     }
@@ -111,15 +119,13 @@ export class TypeormDatabase {
 
         assert(status.length == 1)
 
-        let top: HashAndHeight[] = await em.query(
-            `SELECT hash, height FROM ${schema}.hot_block ORDER BY height`
-        )
+        let top: HashAndHeight[] = await em.query(`SELECT hash, height FROM ${schema}.hot_block ORDER BY height`)
 
         return assertStateInvariants({...status[0], top})
     }
 
     transact(info: FinalTxInfo, cb: (store: Store) => Promise<void>): Promise<void> {
-        return this.submit(async em => {
+        return this.submit(async (em) => {
             let state = await this.getState(em)
             let {prevHead: prev, nextHead: next} = info
 
@@ -147,15 +153,21 @@ export class TypeormDatabase {
         })
     }
 
-    transactHot2(info: HotTxInfo, cb: (store: Store, sliceBeg: number, sliceEnd: number) => Promise<void>): Promise<void> {
-        return this.submit(async em => {
+    transactHot2(
+        info: HotTxInfo,
+        cb: (store: Store, sliceBeg: number, sliceEnd: number) => Promise<void>
+    ): Promise<void> {
+        return this.submit(async (em) => {
             let state = await this.getState(em)
             let chain = [state, ...state.top]
 
             assertChainContinuity(info.baseHead, info.newBlocks)
             assert(info.finalizedHead.height <= (maybeLast(info.newBlocks) ?? info.baseHead).height)
 
-            assert(chain.find(b => b.hash === info.baseHead.hash), RACE_MSG)
+            assert(
+                chain.find((b) => b.hash === info.baseHead.hash),
+                RACE_MSG
+            )
             if (info.newBlocks.length == 0) {
                 assert(last(chain).hash === info.baseHead.hash, RACE_MSG)
             }
@@ -170,18 +182,14 @@ export class TypeormDatabase {
             if (info.newBlocks.length) {
                 let finalizedEnd = info.finalizedHead.height - info.newBlocks[0].height + 1
                 if (finalizedEnd > 0) {
-                    await this.performUpdates(store => cb(store, 0, finalizedEnd), em)
+                    await this.performUpdates((store) => cb(store, 0, finalizedEnd), em)
                 } else {
                     finalizedEnd = 0
                 }
                 for (let i = finalizedEnd; i < info.newBlocks.length; i++) {
                     let b = info.newBlocks[i]
                     await this.insertHotBlock(em, b)
-                    await this.performUpdates(
-                        store => cb(store, i, i + 1),
-                        em,
-                        new ChangeWriter(em, this.statusSchema, b.height)
-                    )
+                    await this.performUpdates((store) => cb(store, i, i + 1), em, new ChangeWriter(em, this.statusSchema, b.height))
                 }
             }
 
@@ -196,17 +204,14 @@ export class TypeormDatabase {
     }
 
     private deleteHotBlocks(em: EntityManager, finalizedHeight: number): Promise<void> {
-        return em.query(
-            `DELETE FROM ${this.escapedSchema()}.hot_block WHERE height <= $1`,
-            [finalizedHeight]
-        )
+        return em.query(`DELETE FROM ${this.escapedSchema()}.hot_block WHERE height <= $1`, [finalizedHeight])
     }
 
     private insertHotBlock(em: EntityManager, block: HashAndHeight): Promise<void> {
-        return em.query(
-            `INSERT INTO ${this.escapedSchema()}.hot_block (height, hash) VALUES ($1, $2)`,
-            [block.height, block.hash]
-        )
+        return em.query(`INSERT INTO ${this.escapedSchema()}.hot_block (height, hash) VALUES ($1, $2)`, [
+            block.height,
+            block.hash,
+        ])
     }
 
     private async updateStatus(em: EntityManager, nonce: number, next: HashAndHeight): Promise<void> {
@@ -221,36 +226,30 @@ export class TypeormDatabase {
 
         // Will never happen if isolation level is SERIALIZABLE or REPEATABLE_READ,
         // but occasionally people use multiprocessor setups and READ_COMMITTED.
-        assert.strictEqual(
-            rowsChanged,
-            1,
-            RACE_MSG
-        )
+        assert.strictEqual(rowsChanged, 1, RACE_MSG)
     }
 
     private async performUpdates(
         cb: (store: Store) => Promise<void>,
         em: EntityManager,
-        changeTracker?: ChangeWriter
+        changeWriter?: ChangeWriter
     ): Promise<void> {
-        let running = true
-
-        let store = new Store(
-            () => {
-                assert(running, `too late to perform db updates, make sure you haven't forgot to await on db query`)
-                return em
-            },
-            {
-                tracker: changeTracker,
-                commitOrder: this.getCommitOrder()
-            }
-        )
+        let store = new Store({
+            em,
+            state: this.getStateManager(),
+            logger: this.getLogger(),
+            changes: changeWriter,
+            cacheMode: this.cacheMode,
+            flushMode: this.flushMode,
+            resetMode: this.resetMode,
+        })
 
         try {
             await cb(store)
             await store.flush()
+            if (this.resetMode === 'BATCH') store.reset()
         } finally {
-            running = false
+            store._close()
         }
     }
 
@@ -261,7 +260,7 @@ export class TypeormDatabase {
                 let con = this.con
                 assert(con != null, 'not connected')
                 return await con.transaction(this.isolationLevel, tx)
-            } catch(e: any) {
+            } catch (e: any) {
                 if (e.code == '40001' && retries) {
                     retries -= 1
                 } else {
@@ -277,10 +276,22 @@ export class TypeormDatabase {
     }
 
     @def
-    private getCommitOrder() {
-        let con = this.con
-        assert(con != null, 'not connected')
-        return sortMetadatasInCommitOrder(con.entityMetadatas)
+    private getLogger() {
+        return createLogger('sqd:typeorm-db')
+    }
+
+    private getStateManager() {
+        let con = assertNotNull(this.con)
+        let stateManager = STATE_MANAGERS.get(con)
+        if (stateManager != null) return stateManager
+
+        stateManager = new StateManager({
+            commitOrder: sortMetadatasInCommitOrder(con),
+            logger: this.getLogger(),
+        })
+        STATE_MANAGERS.set(con, stateManager)
+
+        return stateManager
     }
 }
 

typeorm/typeorm-store/src/index.ts

@@ -1,4 +1,4 @@
 export * from './database'
-export {FindManyOptions, FindOneOptions, Store, EntityLiteral} from './store'
+export {FindManyOptions, FindOneOptions, Store} from './store'
 export * from './decorators'
 export * from './utils/transformers'