chore: move sql read/write type queries into correct classes

hirosystems · Jan 9, 2025 · d4b6fd3 · d4b6fd3
1 parent 01bb6fe
commit d4b6fd3
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 8 deletions.
diff --git a/src/event-stream/event-stream.ts b/src/event-stream/event-stream.ts
@@ -92,7 +92,7 @@ export class EventStreamHandler {
         const result = await this.db.ingestion.applyStackerDbChunk(sql, timestamp, chunk);
         appliedSignerMessageResults.push(...result);
       }
-      await this.db.updateLastIngestedRedisMsgId(sql, messageId);
+      await this.db.ingestion.updateLastIngestedRedisMsgId(sql, messageId);
     });
     this.logger.info(`Apply StackerDB chunks finished in ${time.getElapsedSeconds()}s`);
 
@@ -113,15 +113,15 @@ export class EventStreamHandler {
     // TODO: wrap in sql transaction
     const time = stopwatch();
     await this.db.sqlWriteTransaction(async sql => {
-      const lastIngestedBlockHeight = await this.db.ingestion.getLastIngestedBlockHeight(sql);
+      const lastIngestedBlockHeight = await this.db.getLastIngestedBlockHeight(sql);
       if (block.blockHeight <= lastIngestedBlockHeight) {
         this.logger.info(`Skipping previously ingested block ${block.blockHeight}`);
         return;
       }
       this.logger.info(`Apply block ${block.blockHeight}`);
       await this.db.ingestion.applyBlock(sql, block);
       await this.db.ingestion.updateChainTipBlockHeight(sql, block.blockHeight);
-      await this.db.updateLastIngestedRedisMsgId(sql, messageId);
+      await this.db.ingestion.updateLastIngestedRedisMsgId(sql, messageId);
     });
     this.logger.info(`Apply block ${block.blockHeight} finished in ${time.getElapsedSeconds()}s`);
   }

diff --git a/src/pg/ingestion/pg-write-store.ts b/src/pg/ingestion/pg-write-store.ts
@@ -42,9 +42,8 @@ export class PgWriteStore extends BasePgStoreModule {
     await sql`UPDATE chain_tip SET block_height = ${blockHeight}`;
   }
 
-  async getLastIngestedBlockHeight(sql: PgSqlClient): Promise<number> {
-    const result = await sql<{ block_height: number }[]>`SELECT block_height FROM chain_tip`;
-    return result[0].block_height;
+  async updateLastIngestedRedisMsgId(sql: PgSqlClient, msgId: string): Promise<void> {
+    await sql`UPDATE chain_tip SET last_redis_msg_id = ${msgId}`;
   }
 
   async applyStackerDbChunk(

diff --git a/src/pg/pg-store.ts b/src/pg/pg-store.ts
@@ -73,8 +73,9 @@ export class PgStore extends BasePgStore {
     this.notifications = new NotificationPgStore(this, sql, this.ingestion.events);
   }
 
-  async updateLastIngestedRedisMsgId(sql: PgSqlClient, msgId: string): Promise<void> {
-    await sql`UPDATE chain_tip SET last_redis_msg_id = ${msgId}`;
+  async getLastIngestedBlockHeight(sql: PgSqlClient): Promise<number> {
+    const result = await sql<{ block_height: number }[]>`SELECT block_height FROM chain_tip`;
+    return result[0].block_height;
   }
 
   public async getLastIngestedRedisMsgId(): Promise<string> {