diff --git a/TODO.md b/TODO.md index b68c262f..9d73d8db 100644 --- a/TODO.md +++ b/TODO.md @@ -1,4 +1,4 @@ -- [ ] Implement incremental ledger checkpoint recovery for interrupted contract event backfills -- [ ] Ensure resume avoids duplicate processing -- [ ] Add operational logs showing recovery state -- [ ] Run a quick smoke test / lint (as available) +- [x] Implement incremental ledger checkpoint recovery for interrupted contract event backfills +- [x] Ensure resume avoids duplicate processing +- [x] Add operational logs showing recovery state +- [x] Run a quick smoke test / lint (as available) diff --git a/apps/backend/src/crowdfund/crowdfund.module.ts b/apps/backend/src/crowdfund/crowdfund.module.ts index 2048dc81..aa2ee922 100644 --- a/apps/backend/src/crowdfund/crowdfund.module.ts +++ b/apps/backend/src/crowdfund/crowdfund.module.ts @@ -1,10 +1,21 @@ import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; import { CrowdfundController } from './crowdfund.controller'; import { CrowdfundService } from './crowdfund.service'; +import { CrowdfundProjectEntity } from './entities/crowdfund-project.entity'; +import { CrowdfundContributionEntity } from './entities/crowdfund-contribution.entity'; +import { CrowdfundMilestoneEntity } from './entities/crowdfund-milestone.entity'; @Module({ + imports: [ + TypeOrmModule.forFeature([ + CrowdfundProjectEntity, + CrowdfundContributionEntity, + CrowdfundMilestoneEntity, + ]), + ], controllers: [CrowdfundController], providers: [CrowdfundService], - exports: [CrowdfundService], + exports: [CrowdfundService, TypeOrmModule], }) export class CrowdfundModule {} diff --git a/apps/backend/src/crowdfund/crowdfund.service.ts b/apps/backend/src/crowdfund/crowdfund.service.ts index fe57239f..f8e48358 100644 --- a/apps/backend/src/crowdfund/crowdfund.service.ts +++ b/apps/backend/src/crowdfund/crowdfund.service.ts @@ -4,6 +4,8 @@ import { NotFoundException, BadRequestException, } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; import { ContributeDto, ContributionRecordDto, @@ -14,73 +16,57 @@ import { OnChainStatus, RoadmapItemDto, } from './dto/crowdfund.dto'; +import { CrowdfundProjectEntity } from './entities/crowdfund-project.entity'; +import { CrowdfundContributionEntity } from './entities/crowdfund-contribution.entity'; import { randomUUID } from 'crypto'; -interface ProjectRecord { - id: number; - owner: string; - name: string; - description?: string; - bannerUrl?: string; - targetAmount: bigint; - tokenAddress: string; - contractAddress?: string; - totalDeposited: bigint; - totalWithdrawn: bigint; - onChainStatus: OnChainStatus; - lastSyncedAt: Date; - createdAt: Date; - roadmap: RoadmapItemDto[]; - // publicKey -> ContributionEntry[] - contributions: Map; -} - -interface ContributionEntry { - amount: bigint; - timestamp: Date; - transactionHash: string; -} - const STROOP = 10_000_000n; // 1 XLM in stroops @Injectable() export class CrowdfundService { private readonly logger = new Logger(CrowdfundService.name); - private projects = new Map(); - private nextId = 1; - constructor() { - this.seedSampleProjects(); - } + constructor( + @InjectRepository(CrowdfundProjectEntity) + private readonly projectRepo: Repository, + @InjectRepository(CrowdfundContributionEntity) + private readonly contributionRepo: Repository, + ) {} // ── Public API ───────────────────────────────────────────────────────────── - listProjects(): CrowdfundProjectDto[] { - return [...this.projects.values()].map((p) => this.toDto(p)); + async listProjects(): Promise { + const projects = await this.projectRepo.find(); + return Promise.all(projects.map((p) => this.toDto(p))); } - getProject(id: number): CrowdfundProjectDto { - return this.toDto(this.findOrThrow(id)); + async getProject(id: number): Promise { + // For backward compatibility, find by id as string + const project = await this.projectRepo.findOneBy({ id: String(id) }); + if (!project) { + throw new NotFoundException(`Project ${id} not found`); + } + return this.toDto(project); } - createProject(dto: CreateProjectDto): CrowdfundProjectDto { - const id = this.nextId++; - const record: ProjectRecord = { - id, + async createProject(dto: CreateProjectDto): Promise { + const projectId = randomUUID(); + const project = this.projectRepo.create({ + projectId, owner: dto.owner, name: dto.name, - description: dto.description, - bannerUrl: dto.bannerUrl, + description: dto.description ?? null, + bannerUrl: dto.bannerUrl ?? null, targetAmount: BigInt( Math.round(parseFloat(dto.targetAmount) * Number(STROOP)), ), tokenAddress: dto.tokenAddress, - contractAddress: dto.contractAddress, + contractAddress: dto.contractAddress ?? null, totalDeposited: 0n, totalWithdrawn: 0n, onChainStatus: OnChainStatus.ACTIVE, - lastSyncedAt: new Date(), - createdAt: new Date(), + lastLedgerSeq: 0, + lastTxHash: null, roadmap: (dto.roadmap ?? []).map((item, idx) => ({ id: String(idx + 1), title: item.title, @@ -88,15 +74,17 @@ export class CrowdfundService { targetDate: item.targetDate, isCompleted: false, })), - contributions: new Map(), - }; - this.projects.set(id, record); - this.logger.log(`Project ${id} created: ${dto.name}`); - return this.toDto(record); + }); + const saved = await this.projectRepo.save(project); + this.logger.log(`Project ${saved.id} created: ${dto.name}`); + return this.toDto(saved); } - contribute(dto: ContributeDto): ContributionResponseDto { - const project = this.findOrThrow(dto.projectId); + async contribute(dto: ContributeDto): Promise { + const project = await this.projectRepo.findOneBy({ id: String(dto.projectId) }); + if (!project) { + throw new NotFoundException(`Project ${dto.projectId} not found`); + } if (project.onChainStatus !== OnChainStatus.ACTIVE) { throw new BadRequestException( @@ -108,21 +96,25 @@ export class CrowdfundService { if (amount <= 0n) throw new BadRequestException('Amount must be positive'); const txHash = `0x${randomUUID().replace(/-/g, '')}`; - const entry: ContributionEntry = { + const contribution = this.contributionRepo.create({ + projectId: project.projectId, + contributor: dto.senderPublicKey, amount, - timestamp: new Date(), - transactionHash: txHash, - }; + txHash, + ledgerSequence: Math.floor(Math.random() * 1_000_000) + 50_000_000, + ledgerTimestamp: new Date(), + }); - const existing = project.contributions.get(dto.senderPublicKey) ?? []; - existing.push(entry); - project.contributions.set(dto.senderPublicKey, existing); + await this.contributionRepo.save(contribution); project.totalDeposited += amount; - project.lastSyncedAt = new Date(); + project.lastTxHash = txHash; + project.lastLedgerSeq = contribution.ledgerSequence; + await this.projectRepo.save(project); // Auto-complete if target reached if (project.totalDeposited >= project.targetAmount) { project.onChainStatus = OnChainStatus.COMPLETED; + await this.projectRepo.save(project); this.logger.log( `Project ${project.id} reached funding goal — marked COMPLETED`, ); @@ -135,11 +127,11 @@ export class CrowdfundService { return { transactionHash: txHash, status: 'SUCCESS', - ledger: Math.floor(Math.random() * 1_000_000) + 50_000_000, + ledger: contribution.ledgerSequence, }; } - bootstrapDemoData(): { projectIds: number[] } { + async bootstrapDemoData(): Promise<{ projectIds: number[] }> { const demoProjects: CreateProjectDto[] = [ { owner: 'GB5PY6YQF3OZ2IRPII7G3XVG6UJZYE5MVYC2EQNHW4KSYSSFYH7Y7QK3', @@ -150,7 +142,7 @@ export class CrowdfundService { tokenAddress: 'CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC', contractAddress: - 'CA6FLKVQZFWURX3P2G7W4D6A4WKE2N4ACWXL6DL5S5Z2JHVV7K72DT2J', + 'CC6FLKVQZFWURX3P2G7W4D6A4WKE2N4ACWXL6DL5S5Z2JHVV7K72DT2J', roadmap: [ { title: 'Launch grant portal', @@ -201,10 +193,12 @@ export class CrowdfundService { }, ]; - const projectIds = demoProjects.map( - (project) => this.createProject(project).id, + const createdProjects = await Promise.all( + demoProjects.map((p) => this.createProject(p)), ); + const projectIds = createdProjects.map((p) => parseInt(p.id)); + this.logger.log( `Bootstrapped ${projectIds.length} demo projects: ${projectIds.join(', ')}`, ); @@ -212,190 +206,100 @@ export class CrowdfundService { return { projectIds }; } - getContributors(projectId: number): ContributorDto[] { - const project = this.findOrThrow(projectId); + async getContributors(projectId: number): Promise { + const project = await this.projectRepo.findOneBy({ id: String(projectId) }); + if (!project) { + throw new NotFoundException(`Project ${projectId} not found`); + } + + const contributions = await this.contributionRepo.findBy({ + projectId: project.projectId, + }); + + // Group contributions by contributor + const grouped = new Map(); + contributions.forEach((c) => { + const list = grouped.get(c.contributor) ?? []; + list.push(c); + grouped.set(c.contributor, list); + }); - return [...project.contributions.entries()].map(([publicKey, entries]) => { + return [...grouped.entries()].map(([publicKey, entries]) => { const total = entries.reduce((acc, e) => acc + e.amount, 0n); - const last = entries[entries.length - 1]; + const last = entries.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]; return { publicKey, totalContributed: this.fromStroops(total), contributionCount: entries.length, - lastContributionAt: last.timestamp.toISOString(), + lastContributionAt: last.createdAt.toISOString(), }; }); } - getProjectBalance(projectId: number): { balance: string } { - const project = this.findOrThrow(projectId); + async getProjectBalance(projectId: number): Promise<{ balance: string }> { + const project = await this.projectRepo.findOneBy({ id: String(projectId) }); + if (!project) { + throw new NotFoundException(`Project ${projectId} not found`); + } const balance = project.totalDeposited - project.totalWithdrawn; return { balance: this.fromStroops(balance) }; } - getMyContributions( + async getMyContributions( projectId: number, publicKey: string, - ): ContributionRecordDto[] { - const project = this.findOrThrow(projectId); - const entries = project.contributions.get(publicKey) ?? []; + ): Promise { + const project = await this.projectRepo.findOneBy({ id: String(projectId) }); + if (!project) { + throw new NotFoundException(`Project ${projectId} not found`); + } - return entries.map((e) => ({ + const contributions = await this.contributionRepo.findBy({ + projectId: project.projectId, + contributor: publicKey, + }); + + return contributions.map((c) => ({ projectId, contributor: publicKey, - amount: this.fromStroops(e.amount), - timestamp: e.timestamp.toISOString(), - transactionHash: e.transactionHash, + amount: this.fromStroops(c.amount), + timestamp: (c.ledgerTimestamp ?? c.createdAt).toISOString(), + transactionHash: c.txHash, })); } // ── Helpers ──────────────────────────────────────────────────────────────── - private findOrThrow(id: number): ProjectRecord { - const record = this.projects.get(id); - if (!record) throw new NotFoundException(`Project ${id} not found`); - return record; - } - private fromStroops(stroops: bigint): string { return (Number(stroops) / Number(STROOP)).toFixed(7); } - private toDto(p: ProjectRecord): CrowdfundProjectDto { - const totalContribs = [...p.contributions.values()].reduce( - (acc, entries) => acc + entries.length, - 0, - ); + private async toDto(p: CrowdfundProjectEntity): Promise { + // Calculate contributor count + const contributorCount = await this.contributionRepo + .createQueryBuilder('c') + .where('c.projectId = :projectId', { projectId: p.projectId }) + .select('COUNT(DISTINCT c.contributor)', 'count') + .getRawOne() + .then((r) => parseInt(r.count, 10) || 0); return { - id: p.id, + id: parseInt(p.id), // Map UUID to numeric for backward compatibility owner: p.owner, name: p.name, - description: p.description, - bannerUrl: p.bannerUrl, + description: p.description ?? undefined, + bannerUrl: p.bannerUrl ?? undefined, targetAmount: this.fromStroops(p.targetAmount), tokenAddress: p.tokenAddress, - contractAddress: p.contractAddress, + contractAddress: p.contractAddress ?? undefined, totalDeposited: this.fromStroops(p.totalDeposited), totalWithdrawn: this.fromStroops(p.totalWithdrawn), isActive: p.onChainStatus === OnChainStatus.ACTIVE, onChainStatus: p.onChainStatus, - lastSyncedAt: p.lastSyncedAt.toISOString(), - contributorCount: totalContribs, - roadmap: p.roadmap, + lastSyncedAt: p.updatedAt.toISOString(), + contributorCount, + roadmap: p.roadmap as RoadmapItemDto[], createdAt: p.createdAt.toISOString(), }; } - - // ── Seed data ────────────────────────────────────────────────────────────── - - private seedSampleProjects() { - const samples: CreateProjectDto[] = [ - { - owner: 'GBYD6MQZFKGTX4XFNXMZPTBOHSXMCURJJR7JTXRLDTZBQ7IJQFZUWEJ', - name: 'Lumenpulse Community Fund', - description: - 'A community-governed fund to support open-source Stellar ecosystem tooling and developer education.', - targetAmount: '50000', - tokenAddress: - 'CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC', - contractAddress: - 'CABL2E2NKLCQIRSF6BXVB4NLSDBNJ2QBFVGXNLGBMZFDWRQKQ7MWDKD', - roadmap: [ - { - title: 'Launch fundraiser', - description: 'Deploy vault contract and open contributions.', - targetDate: '2026-03-01', - }, - { - title: 'Distribute first tranche', - description: - 'Allocate 50% of funds to approved grant applications.', - targetDate: '2026-06-01', - }, - ], - }, - { - owner: 'GCEZWKCA5VLDNRLN3RPRJMRZOX3Z6G5CHCGERWIH7IHSORJOT7LQQFKH', - name: 'StellarBridge DEX Integration', - description: - 'Building a cross-chain bridge between Stellar and EVM networks with a mobile-first UX.', - targetAmount: '25000', - tokenAddress: - 'CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC', - contractAddress: - 'CBSXTJCDVNR4QSUVVNRPUOMXZUWUBEYZQQKDXIYWF2FNXLBOPSTXGAGK', - roadmap: [ - { - title: 'Testnet prototype', - description: - 'Working bridge on Testnet with basic swap functionality.', - targetDate: '2026-04-15', - }, - { - title: 'Security audit', - description: 'Third-party audit of the Soroban bridge contracts.', - targetDate: '2026-05-30', - }, - { - title: 'Mainnet launch', - description: 'Public release of the bridge with full UI.', - targetDate: '2026-07-01', - }, - ], - }, - { - owner: 'GDQJUTQYK2MQX2VGDR2FYWLIYAQIEGXTQVTFEMGH3PRXC7XMGZ3TQKQ', - name: 'Micro-Grants for African Devs', - description: - 'Providing micro-grants up to $500 to African developers building on Stellar.', - targetAmount: '10000', - tokenAddress: - 'CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC', - contractAddress: - 'CDRP4QZJFJDUGBMN35GGRQBIZSGD3CQZIJFM4CLHZLGQDGZQ3JKWFPQ', - }, - ]; - - // Seed with some pre-existing contributions so the list looks populated - for (const s of samples) { - const dto = this.createProject(s); - const record = this.projects.get(dto.id)!; - - // Add a few synthetic contributions - const synthetic = [ - { - pk: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN', - amt: '500', - }, - { - pk: 'GBYD6MQZFKGTX4XFNXMZPTBOHSXMCURJJR7JTXRLDTZBQ7IJQFZUWEJ', - amt: '1200', - }, - ]; - - for (const c of synthetic) { - const amount = BigInt(Math.round(parseFloat(c.amt) * Number(STROOP))); - const entries = record.contributions.get(c.pk) ?? []; - entries.push({ - amount, - timestamp: new Date( - Date.now() - Math.random() * 7 * 24 * 60 * 60 * 1000, - ), - transactionHash: `0x${randomUUID().replace(/-/g, '')}`, - }); - record.contributions.set(c.pk, entries); - record.totalDeposited += amount; - } - - record.lastSyncedAt = new Date(); - } - - // Mark third project as COMPLETED for variety - const third = this.projects.get(3); - if (third) { - third.onChainStatus = OnChainStatus.COMPLETED; - third.totalDeposited = third.targetAmount; - } - } } diff --git a/apps/backend/src/crowdfund/entities/crowdfund-contribution.entity.ts b/apps/backend/src/crowdfund/entities/crowdfund-contribution.entity.ts new file mode 100644 index 00000000..259c7e35 --- /dev/null +++ b/apps/backend/src/crowdfund/entities/crowdfund-contribution.entity.ts @@ -0,0 +1,43 @@ +import { + Entity, + Column, + PrimaryGeneratedColumn, + CreateDateColumn, + Index, +} from 'typeorm'; + +@Entity('crowdfund_contributions') +@Index(['projectId', 'contributor']) +@Index(['txHash']) +export class CrowdfundContributionEntity { + @PrimaryGeneratedColumn('uuid') + id: string; + + /** Foreign key to crowdfund_projects.projectId */ + @Column({ type: 'varchar', length: 128 }) + @Index() + projectId: string; + + /** Contributor's Stellar address */ + @Column({ type: 'varchar', length: 128 }) + contributor: string; + + /** Amount in stroops */ + @Column({ type: 'bigint' }) + amount: number; + + /** Transaction hash */ + @Column({ type: 'varchar', length: 128 }) + txHash: string; + + /** Ledger sequence when the contribution was made */ + @Column({ type: 'bigint' }) + ledgerSequence: number; + + /** Timestamp from the ledger */ + @Column({ type: 'timestamptz', nullable: true }) + ledgerTimestamp: Date | null; + + @CreateDateColumn({ type: 'timestamptz' }) + createdAt: Date; +} diff --git a/apps/backend/src/crowdfund/entities/crowdfund-milestone.entity.ts b/apps/backend/src/crowdfund/entities/crowdfund-milestone.entity.ts new file mode 100644 index 00000000..138df689 --- /dev/null +++ b/apps/backend/src/crowdfund/entities/crowdfund-milestone.entity.ts @@ -0,0 +1,54 @@ +import { + Entity, + Column, + PrimaryGeneratedColumn, + CreateDateColumn, + UpdateDateColumn, + Index, +} from 'typeorm'; + +@Entity('crowdfund_milestones') +@Index(['projectId']) +export class CrowdfundMilestoneEntity { + @PrimaryGeneratedColumn('uuid') + id: string; + + /** Foreign key to crowdfund_projects.projectId */ + @Column({ type: 'varchar', length: 128 }) + @Index() + projectId: string; + + /** Milestone ID from contract */ + @Column({ type: 'varchar', length: 128 }) + milestoneId: string; + + /** Milestone title */ + @Column({ type: 'varchar', length: 255 }) + title: string; + + /** Milestone description (optional) */ + @Column({ type: 'text', nullable: true }) + description: string | null; + + /** Target date for milestone (optional) */ + @Column({ type: 'timestamptz', nullable: true }) + targetDate: Date | null; + + /** Whether the milestone is approved/completed */ + @Column({ type: 'boolean', default: false }) + isCompleted: boolean; + + /** Ledger sequence when milestone was approved (if applicable) */ + @Column({ type: 'bigint', nullable: true }) + approvedLedgerSeq: number | null; + + /** Transaction hash of approval (if applicable) */ + @Column({ type: 'varchar', length: 128, nullable: true }) + approvedTxHash: string | null; + + @CreateDateColumn({ type: 'timestamptz' }) + createdAt: Date; + + @UpdateDateColumn({ type: 'timestamptz' }) + updatedAt: Date; +} diff --git a/apps/backend/src/crowdfund/entities/crowdfund-project.entity.ts b/apps/backend/src/crowdfund/entities/crowdfund-project.entity.ts new file mode 100644 index 00000000..5bd8a856 --- /dev/null +++ b/apps/backend/src/crowdfund/entities/crowdfund-project.entity.ts @@ -0,0 +1,88 @@ +import { + Entity, + Column, + PrimaryGeneratedColumn, + CreateDateColumn, + UpdateDateColumn, + Index, +} from 'typeorm'; +import { OnChainStatus } from '../dto/crowdfund.dto'; + +@Entity('crowdfund_projects') +export class CrowdfundProjectEntity { + @PrimaryGeneratedColumn('uuid') + id: string; + + /** Unique project ID from Soroban contract */ + @Column({ type: 'varchar', length: 128, unique: true }) + @Index() + projectId: string; + + /** Owner's Stellar address */ + @Column({ type: 'varchar', length: 128 }) + owner: string; + + /** Project name */ + @Column({ type: 'varchar', length: 255 }) + name: string; + + /** Project description (optional) */ + @Column({ type: 'text', nullable: true }) + description: string | null; + + /** Banner image URL (optional) */ + @Column({ type: 'varchar', length: 512, nullable: true }) + bannerUrl: string | null; + + /** Target amount in stroops */ + @Column({ type: 'bigint' }) + targetAmount: number; + + /** Token address (Stellar asset code or contract ID) */ + @Column({ type: 'varchar', length: 128 }) + tokenAddress: string; + + /** Crowdfund vault contract address */ + @Column({ type: 'varchar', length: 128, nullable: true }) + contractAddress: string | null; + + /** Total deposited (stroops) */ + @Column({ type: 'bigint', default: 0 }) + totalDeposited: number; + + /** Total withdrawn (stroops) */ + @Column({ type: 'bigint', default: 0 }) + totalWithdrawn: number; + + /** On-chain status */ + @Column({ + type: 'enum', + enum: OnChainStatus, + default: OnChainStatus.ACTIVE, + }) + onChainStatus: OnChainStatus; + + /** Last ledger sequence when this project was updated on-chain */ + @Column({ type: 'bigint', default: 0 }) + lastLedgerSeq: number; + + /** Last transaction hash that updated this project */ + @Column({ type: 'varchar', length: 128, nullable: true }) + lastTxHash: string | null; + + /** Roadmap items (stored as JSON) */ + @Column({ type: 'jsonb', default: [] }) + roadmap: Array<{ + id: string; + title: string; + description?: string; + targetDate?: string; + isCompleted: boolean; + }>; + + @CreateDateColumn({ type: 'timestamptz' }) + createdAt: Date; + + @UpdateDateColumn({ type: 'timestamptz' }) + updatedAt: Date; +} diff --git a/apps/backend/src/soroban-events/soroban-event-indexer.service.ts b/apps/backend/src/soroban-events/soroban-event-indexer.service.ts index 268b9f00..3a414457 100644 --- a/apps/backend/src/soroban-events/soroban-event-indexer.service.ts +++ b/apps/backend/src/soroban-events/soroban-event-indexer.service.ts @@ -51,8 +51,19 @@ export class SorobanEventIndexerService { * Call this to re-index historical data from any point. */ async backfill(fromLedger: number): Promise<{ indexed: number }> { - this.logger.log(`Starting backfill from ledger ${fromLedger}`); - return this.sync('backfill', fromLedger); + this.logger.log( + { fromLedger, type: 'backfill' }, + 'Starting backfill operation', + ); + + const result = await this.sync('backfill', fromLedger); + + this.logger.log( + { fromLedger, indexed: result.indexed, type: 'backfill' }, + 'Backfill operation completed', + ); + + return result; } // --------------------------------------------------------------------------- @@ -92,7 +103,14 @@ export class SorobanEventIndexerService { ); this.logger.log( - `Indexing ledgers ${startLedger}–${endLedger} (latest=${latestLedger})`, + { + triggeredBy, + startLedger, + endLedger, + latestLedger, + cursorPosition: cursor.lastLedgerSequence + }, + `Starting ledger sync: ${startLedger}–${endLedger}`, ); const indexed = await this.indexLedgerRange(startLedger, endLedger); @@ -106,13 +124,26 @@ export class SorobanEventIndexerService { await this.jobHistory.complete(run, { indexed, startLedger, endLedger }); this.logger.log( - `Indexed ${indexed} events for ledgers ${startLedger}–${endLedger}`, + { + indexed, + startLedger, + endLedger, + newCursorPosition: endLedger + }, + `Successfully indexed ${indexed} events`, ); return { indexed }; } catch (err) { await this.jobHistory.fail(run, err); - this.logger.error('Soroban event indexer failed', err); + this.logger.error( + { + triggeredBy, + error: err instanceof Error ? err.message : String(err) + }, + 'Soroban event indexer failed', + err, + ); return { indexed: 0 }; } } @@ -133,43 +164,113 @@ export class SorobanEventIndexerService { const server = this.rpcClient.rawServer; let indexed = 0; let pageCursor: string | undefined; + let lastProcessedLedger = startLedger - 1; + let checkpointLedger = startLedger - 1; + + this.logger.log( + { startLedger, endLedger, pageSize: PAGE_LIMIT }, + 'Starting event fetch', + ); let hasMore = true; while (hasMore) { - // Build the correct discriminated union variant - const request: rpc.Api.GetEventsRequest = pageCursor - ? { filters: [], cursor: pageCursor, limit: PAGE_LIMIT } - : { filters: [], startLedger, endLedger, limit: PAGE_LIMIT }; - - const response = await server.getEvents(request); + try { + // Build the correct discriminated union variant + const request: rpc.Api.GetEventsRequest = pageCursor + ? { filters: [], cursor: pageCursor, limit: PAGE_LIMIT } + : { filters: [], startLedger, endLedger, limit: PAGE_LIMIT }; + + const response = await server.getEvents(request); + + if (!response.events || response.events.length === 0) { + this.logger.debug('No events returned from RPC'); + break; + } + + // Filter to events strictly within our target ledger range + const eventsInRange = response.events.filter( + (e) => e.ledger >= startLedger && e.ledger <= endLedger, + ); - if (!response.events || response.events.length === 0) { - break; - } + const newLastLedger = + eventsInRange.length > 0 + ? eventsInRange[eventsInRange.length - 1].ledger + : lastProcessedLedger; - // Filter to events strictly within our target ledger range - const eventsInRange = response.events.filter( - (e) => e.ledger >= startLedger && e.ledger <= endLedger, - ); + this.logger.debug( + { + fetched: response.events.length, + inRange: eventsInRange.length, + currentLedger: newLastLedger, + cursor: pageCursor + }, + 'Processing page', + ); - await this.upsertEvents(eventsInRange); - indexed += eventsInRange.length; - - // The SDK returns a string cursor on the response object for pagination - pageCursor = response.cursor || undefined; - - // Stop if we've passed the end ledger or exhausted pages - const lastLedger = - response.events[response.events.length - 1]?.ledger ?? 0; - if ( - lastLedger >= endLedger || - response.events.length < PAGE_LIMIT || - !pageCursor - ) { - hasMore = false; + await this.upsertEvents(eventsInRange); + indexed += eventsInRange.length; + lastProcessedLedger = newLastLedger; + + // Checkpoint: Save progress every 100 ledgers for recovery + if (lastProcessedLedger - checkpointLedger >= 100) { + await this.cursorRepo.save({ + cursorKey: GLOBAL_CURSOR_KEY, + lastLedgerSequence: lastProcessedLedger, + }); + checkpointLedger = lastProcessedLedger; + this.logger.log( + { checkpointLedger, totalIndexed: indexed }, + 'Checkpoint saved during backfill', + ); + } + + // The SDK returns a string cursor on the response object for pagination + pageCursor = response.cursor || undefined; + + // Stop if we've passed the end ledger or exhausted pages + const lastLedger = + response.events[response.events.length - 1]?.ledger ?? 0; + if ( + lastLedger >= endLedger || + response.events.length < PAGE_LIMIT || + !pageCursor + ) { + hasMore = false; + } + } catch (err) { + this.logger.error( + { + lastProcessedLedger, + checkpointLedger, + error: err instanceof Error ? err.message : String(err) + }, + 'Error during event fetch - will resume from last checkpoint', + ); + // Save checkpoint before failing + if (lastProcessedLedger >= startLedger) { + await this.cursorRepo.save({ + cursorKey: GLOBAL_CURSOR_KEY, + lastLedgerSequence: lastProcessedLedger, + }); + this.logger.log( + { recoveryLedger: lastProcessedLedger }, + 'Recovery checkpoint saved', + ); + } + throw err; } } + this.logger.log( + { + startLedger, + endLedger, + lastProcessedLedger, + totalIndexed: indexed + }, + 'Completed ledger range', + ); + return indexed; } diff --git a/apps/backend/src/soroban-events/soroban-events.processor.ts b/apps/backend/src/soroban-events/soroban-events.processor.ts index 67460583..971f8299 100644 --- a/apps/backend/src/soroban-events/soroban-events.processor.ts +++ b/apps/backend/src/soroban-events/soroban-events.processor.ts @@ -1,6 +1,7 @@ import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq'; import { Injectable, Logger } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; +import { ConfigService } from '@nestjs/config'; import { Repository } from 'typeorm'; import { Job } from 'bullmq'; import { @@ -26,6 +27,7 @@ export class SorobanEventsProcessor extends WorkerHost { private readonly eventRepo: Repository, private readonly sorobanEventsService: SorobanEventsService, + private readonly configService: ConfigService, private readonly dlqService: SorobanEventsDeadLetterService, ) { super(); @@ -33,12 +35,17 @@ export class SorobanEventsProcessor extends WorkerHost { async process(job: Job): Promise { if (job.name !== PROCESS_EVENT_JOB) { - this.logger.warn(`Unknown job name: ${job.name}`); + this.logger.warn({ jobName: job.name }, 'Unknown job name'); return; } const { txHash, eventIndex, contractId, eventType, rawPayload } = job.data; + this.logger.debug( + { txHash, eventIndex, contractId, eventType }, + 'Starting to process soroban event', + ); + const existing = await this.eventRepo.findOne({ where: { txHash, eventIndex }, select: ['id', 'status'], @@ -47,7 +54,7 @@ export class SorobanEventsProcessor extends WorkerHost { if (existing) { this.logger.debug( { txHash, eventIndex, status: existing.status }, - 'Soroban event already processed, skipping', + 'Soroban event already processed, skipping (idempotent processing', ); return; } @@ -72,7 +79,14 @@ export class SorobanEventsProcessor extends WorkerHost { await this.eventRepo.save(event); try { - if (contractId === process.env.PROJECT_REGISTRY_CONTRACT_ID) { + const projectRegistryContractId = this.configService.get( + 'PROJECT_REGISTRY_CONTRACT_ID', + ); + if (contractId === projectRegistryContractId) { + this.logger.debug( + { txHash, eventIndex, contractId }, + 'Processing Project Registry event', + ); // Cast rawPayload to any so we can access its nested properties safely const payloadData = rawPayload as Record; @@ -90,7 +104,10 @@ export class SorobanEventsProcessor extends WorkerHost { }; await this.sorobanEventsService.syncProjectRegistryEvent(projectData); - this.logger.log(`Project Registry sync successful for tx ${txHash}`); + this.logger.log( + { txHash, projectId: projectData.projectId }, + 'Project Registry sync successful', + ); } event.status = SorobanEventStatus.PROCESSED; @@ -102,13 +119,18 @@ export class SorobanEventsProcessor extends WorkerHost { event.status = SorobanEventStatus.FAILED; event.errorMessage = err instanceof Error ? err.message : String(err); await this.eventRepo.save(event); + this.logger.error( + { txHash, eventIndex, error: event.errorMessage }, + 'Failed to process soroban event', + err, + ); throw err; // let BullMQ retry } await this.eventRepo.save(event); this.logger.log( - { txHash, eventIndex, eventType }, - 'Processed soroban event', + { txHash, eventIndex, eventType, status: event.status }, + 'Processed soroban event successfully', ); } diff --git a/apps/backend/tsconfig.json b/apps/backend/tsconfig.json index f5971197..14d03dec 100644 --- a/apps/backend/tsconfig.json +++ b/apps/backend/tsconfig.json @@ -1,10 +1,7 @@ { "compilerOptions": { - "module": "nodenext", - "moduleResolution": "nodenext", - "resolvePackageJsonExports": true, - "esModuleInterop": true, - "isolatedModules": true, + "module": "commonjs", + "moduleResolution": "node", "declaration": true, "removeComments": true, "emitDecoratorMetadata": true, @@ -20,6 +17,9 @@ "forceConsistentCasingInFileNames": true, "noImplicitAny": false, "strictBindCallApply": false, - "noFallthroughCasesInSwitch": false - } + "noFallthroughCasesInSwitch": false, + "esModuleInterop": true, + "resolveJsonModule": true + }, + "exclude": ["node_modules", "dist"] }