fix: tx kysely typing and add more verbose log

packages/data-flow/src/data-loader/dataLoader.ts

@@ -63,19 +63,27 @@ export class DataLoader implements IDataLoader {
         }
 
         await this.transactionManager.runInTransaction(async (tx) => {
-            this.logger.debug("Starting transaction...");
+            this.logger.debug(`Starting transaction on ${changesets.length} changesets...`, {
+                className: DataLoader.name,
+            });
             for (const changeset of changesets) {
                 try {
                     //TODO: inside each handler, we should add zod validation that the args match the expected type
                     await this.handlers[changeset.type](changeset as never, tx);
                 } catch (error) {
                     this.logger.debug(
                         `Error applying changeset ${changeset.type}. Rolling back transaction with ${changesets.length} changesets`,
+                        {
+                            className: DataLoader.name,
+                        },
                     );
 
                     throw error;
                 }
             }
         });
+        this.logger.debug(`Successfully applied ${changesets.length} changesets`, {
+            className: DataLoader.name,
+        });
     }
 }

packages/repository/src/repositories/kysely/application.repository.ts

@@ -7,11 +7,12 @@ import {
     ApplicationNotFound,
     Database,
     IApplicationRepository,
+    KyselyTransaction,
     NewApplication,
     PartialApplication,
 } from "../../internal.js";
 
-export class KyselyApplicationRepository implements IApplicationRepository<Kysely<Database>> {
+export class KyselyApplicationRepository implements IApplicationRepository<KyselyTransaction> {
     constructor(
         private readonly db: Kysely<Database>,
         private readonly schemaName: string,
@@ -96,7 +97,7 @@ export class KyselyApplicationRepository implements IApplicationRepository<Kysel
     }
 
     /* @inheritdoc */
-    async insertApplication(application: NewApplication, tx?: Kysely<Database>): Promise<void> {
+    async insertApplication(application: NewApplication, tx?: KyselyTransaction): Promise<void> {
         const _application = this.formatApplication(application);
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
 
@@ -107,7 +108,7 @@ export class KyselyApplicationRepository implements IApplicationRepository<Kysel
     async updateApplication(
         where: { id: string; chainId: ChainId; roundId: string },
         application: PartialApplication,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const _application = this.formatApplication(application);
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);

packages/repository/src/repositories/kysely/applicationPayout.repository.ts

@@ -1,9 +1,14 @@
 import { Kysely } from "kysely";
 
-import { Database, IApplicationPayoutRepository, NewApplicationPayout } from "../../internal.js";
+import {
+    Database,
+    IApplicationPayoutRepository,
+    KyselyTransaction,
+    NewApplicationPayout,
+} from "../../internal.js";
 
 export class KyselyApplicationPayoutRepository
-    implements IApplicationPayoutRepository<Kysely<Database>>
+    implements IApplicationPayoutRepository<KyselyTransaction>
 {
     constructor(
         private readonly db: Kysely<Database>,
@@ -13,7 +18,7 @@ export class KyselyApplicationPayoutRepository
     /** @inheritdoc */
     async insertApplicationPayout(
         applicationPayout: NewApplicationPayout,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.insertInto("applicationsPayouts").values(applicationPayout).execute();

packages/repository/src/repositories/kysely/donation.repository.ts

@@ -1,15 +1,15 @@
 import { Kysely } from "kysely";
 
-import { Database, IDonationRepository, NewDonation } from "../../internal.js";
+import { Database, IDonationRepository, KyselyTransaction, NewDonation } from "../../internal.js";
 
-export class KyselyDonationRepository implements IDonationRepository<Kysely<Database>> {
+export class KyselyDonationRepository implements IDonationRepository<KyselyTransaction> {
     constructor(
         private readonly db: Kysely<Database>,
         private readonly schemaName: string,
     ) {}
 
     /** @inheritdoc */
-    async insertDonation(donation: NewDonation, tx?: Kysely<Database>): Promise<void> {
+    async insertDonation(donation: NewDonation, tx?: KyselyTransaction): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
 
         await queryBuilder
@@ -22,7 +22,7 @@ export class KyselyDonationRepository implements IDonationRepository<Kysely<Data
     }
 
     /** @inheritdoc */
-    async insertManyDonations(donations: NewDonation[], tx?: Kysely<Database>): Promise<void> {
+    async insertManyDonations(donations: NewDonation[], tx?: KyselyTransaction): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
 
         await queryBuilder

packages/repository/src/repositories/kysely/project.repository.ts

@@ -5,6 +5,7 @@ import { Address, ChainId } from "@grants-stack-indexer/shared";
 import { IProjectRepository } from "../../interfaces/projectRepository.interface.js";
 import {
     Database,
+    KyselyTransaction,
     NewPendingProjectRole,
     NewProject,
     NewProjectRole,
@@ -15,7 +16,7 @@ import {
     ProjectRoleNames,
 } from "../../internal.js";
 
-export class KyselyProjectRepository implements IProjectRepository<Kysely<Database>> {
+export class KyselyProjectRepository implements IProjectRepository<KyselyTransaction> {
     constructor(
         private readonly db: Kysely<Database>,
         private readonly schemaName: string,
@@ -73,7 +74,7 @@ export class KyselyProjectRepository implements IProjectRepository<Kysely<Databa
     }
 
     /* @inheritdoc */
-    async insertProject(project: NewProject, tx?: Kysely<Database>): Promise<void> {
+    async insertProject(project: NewProject, tx?: KyselyTransaction): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.insertInto("projects").values(project).execute();
     }
@@ -82,7 +83,7 @@ export class KyselyProjectRepository implements IProjectRepository<Kysely<Databa
     async updateProject(
         where: { id: string; chainId: ChainId },
         project: PartialProject,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder
@@ -96,7 +97,7 @@ export class KyselyProjectRepository implements IProjectRepository<Kysely<Databa
     // ============================ PROJECT ROLES ============================
 
     /* @inheritdoc */
-    async insertProjectRole(projectRole: NewProjectRole, tx?: Kysely<Database>): Promise<void> {
+    async insertProjectRole(projectRole: NewProjectRole, tx?: KyselyTransaction): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.insertInto("projectRoles").values(projectRole).execute();
     }
@@ -107,7 +108,7 @@ export class KyselyProjectRepository implements IProjectRepository<Kysely<Databa
         projectId: string,
         role: ProjectRoleNames,
         address?: Address,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         const query = queryBuilder
@@ -151,14 +152,14 @@ export class KyselyProjectRepository implements IProjectRepository<Kysely<Databa
     /* @inheritdoc */
     async insertPendingProjectRole(
         pendingProjectRole: NewPendingProjectRole,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.insertInto("pendingProjectRoles").values(pendingProjectRole).execute();
     }
 
     /* @inheritdoc */
-    async deleteManyPendingProjectRoles(ids: number[], tx?: Kysely<Database>): Promise<void> {
+    async deleteManyPendingProjectRoles(ids: number[], tx?: KyselyTransaction): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.deleteFrom("pendingProjectRoles").where("id", "in", ids).execute();
     }

packages/repository/src/repositories/kysely/round.repository.ts

@@ -5,6 +5,7 @@ import { Address, ChainId, stringify } from "@grants-stack-indexer/shared";
 import {
     Database,
     IRoundRepository,
+    KyselyTransaction,
     NewPendingRoundRole,
     NewRound,
     NewRoundRole,
@@ -17,7 +18,7 @@ import {
     RoundRoleNames,
 } from "../../internal.js";
 
-export class KyselyRoundRepository implements IRoundRepository<Kysely<Database>> {
+export class KyselyRoundRepository implements IRoundRepository<KyselyTransaction> {
     constructor(
         private readonly db: Kysely<Database>,
         private readonly schemaName: string,
@@ -114,7 +115,7 @@ export class KyselyRoundRepository implements IRoundRepository<Kysely<Database>>
     }
 
     /* @inheritdoc */
-    async insertRound(round: NewRound, tx?: Kysely<Database>): Promise<void> {
+    async insertRound(round: NewRound, tx?: KyselyTransaction): Promise<void> {
         const _round = this.formatRound(round);
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.insertInto("rounds").values(_round).execute();
@@ -124,7 +125,7 @@ export class KyselyRoundRepository implements IRoundRepository<Kysely<Database>>
     async updateRound(
         where: { id: string; chainId: ChainId } | { chainId: ChainId; strategyAddress: Address },
         round: PartialRound,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const _round = this.formatRound(round);
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
@@ -145,7 +146,7 @@ export class KyselyRoundRepository implements IRoundRepository<Kysely<Database>>
         where: { chainId: ChainId; roundId: string },
         amount: bigint,
         amountInUsd: string,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder
@@ -163,7 +164,7 @@ export class KyselyRoundRepository implements IRoundRepository<Kysely<Database>>
     async incrementRoundTotalDistributed(
         where: { chainId: ChainId; roundId: string },
         amount: bigint,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder
@@ -184,7 +185,7 @@ export class KyselyRoundRepository implements IRoundRepository<Kysely<Database>>
     }
 
     /* @inheritdoc */
-    async insertRoundRole(roundRole: NewRoundRole, tx?: Kysely<Database>): Promise<void> {
+    async insertRoundRole(roundRole: NewRoundRole, tx?: KyselyTransaction): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.insertInto("roundRoles").values(roundRole).execute();
     }
@@ -195,7 +196,7 @@ export class KyselyRoundRepository implements IRoundRepository<Kysely<Database>>
         roundId: string,
         role: RoundRoleNames,
         address: Address,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder
@@ -226,14 +227,14 @@ export class KyselyRoundRepository implements IRoundRepository<Kysely<Database>>
     /* @inheritdoc */
     async insertPendingRoundRole(
         pendingRoundRole: NewPendingRoundRole,
-        tx?: Kysely<Database>,
+        tx?: KyselyTransaction,
     ): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.insertInto("pendingRoundRoles").values(pendingRoundRole).execute();
     }
 
     /* @inheritdoc */
-    async deleteManyPendingRoundRoles(ids: number[], tx?: Kysely<Database>): Promise<void> {
+    async deleteManyPendingRoundRoles(ids: number[], tx?: KyselyTransaction): Promise<void> {
         const queryBuilder = (tx || this.db).withSchema(this.schemaName);
         await queryBuilder.deleteFrom("pendingRoundRoles").where("id", "in", ids).execute();
     }

packages/repository/src/repositories/kysely/transactionManager.ts

@@ -1,12 +1,12 @@
 import { Kysely } from "kysely";
 
-import { Database, ITransactionManager } from "../../internal.js";
+import { Database, ITransactionManager, KyselyTransaction } from "../../internal.js";
 
-export class KyselyTransactionManager implements ITransactionManager<Kysely<Database>> {
+export class KyselyTransactionManager implements ITransactionManager<KyselyTransaction> {
     constructor(private readonly db: Kysely<Database>) {}
 
     /** @inheritdoc */
-    async runInTransaction<T>(fn: (tx: Kysely<Database>) => Promise<T>): Promise<T> {
+    async runInTransaction<T>(fn: (tx: KyselyTransaction) => Promise<T>): Promise<T> {
         return this.db.transaction().execute(fn);
     }
 }

packages/repository/src/types/transaction.types.ts

@@ -1,8 +1,7 @@
-// packages/repository/src/types/transaction.types.ts
-import { Kysely } from "kysely";
+import { Transaction } from "kysely";
 
 import { Database } from "../internal.js";
 
-export type KyselyTransaction = Kysely<Database>;
+export type KyselyTransaction = Transaction<Database>;
 
 export type TransactionConnection = KyselyTransaction;