Skip to content

Commit a0056d0

Browse files
Otaiki1Otaiki1
authored
Merge pull request #379 from Jessepriase/feat/user-processor
feat: implement idempotent UserProcessor with atomic increments
2 parents 7c38df5 + 1436321 commit a0056d0

4 files changed

Lines changed: 120 additions & 66 deletions

File tree

‎indexer/src/database/entities/user.entity.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,14 @@ export class UserEntity {
3939
@Column({ type: "integer", name: "first_seen_ledger" })
4040
firstSeenLedger!: number;
4141

42+
/**
43+
* Transaction hash of the last event applied to this user row.
44+
* Acts as an idempotency key — processors skip re-applying an event
45+
* whose tx_hash matches this value.
46+
*/
47+
@Column({ type: "varchar", length: 64, nullable: true, name: "last_tx_hash" })
48+
lastTxHash!: string | null;
49+
4250
@UpdateDateColumn({ type: "timestamptz", name: "updated_at" })
4351
updatedAt!: Date;
4452
}

‎indexer/src/database/migrations/1720000000001-AddUserLastTxHash.ts‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
import { MigrationInterface, QueryRunner } from "typeorm";
2+
3+
/**
4+
* Adds `last_tx_hash` to the `users` table.
5+
* Used as an idempotency key in UserProcessor — prevents double-applying
6+
* the same event if the ingestion pipeline replays a transaction.
7+
*/
8+
export class AddUserLastTxHash1720000000001 implements MigrationInterface {
9+
public async up(queryRunner: QueryRunner): Promise<void> {
10+
await queryRunner.query(`
11+
ALTER TABLE users
12+
ADD COLUMN IF NOT EXISTS last_tx_hash VARCHAR(64) NULL
13+
`);
14+
}
15+
16+
public async down(queryRunner: QueryRunner): Promise<void> {
17+
await queryRunner.query(`
18+
ALTER TABLE users
19+
DROP COLUMN IF EXISTS last_tx_hash
20+
`);
21+
}
22+
}

‎indexer/src/processors/ticket.processor.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ export class TicketProcessor {
6969
await this.userProcessor.handleTicketPurchased(
7070
raffleId,
7171
buyer,
72+
ticketIds.length,
7273
ledger,
7374
txHash,
7475
queryRunner,

‎indexer/src/processors/user.processor.ts‎

Lines changed: 89 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@ import { Injectable, Logger } from '@nestjs/common';
22
import { DataSource, QueryRunner } from 'typeorm';
33
import { CacheService } from '../cache/cache.service';
44
import { UserEntity } from '../database/entities/user.entity';
5-
import { TicketEntity } from '../database/entities/ticket.entity';
6-
import { RaffleEntity } from '../database/entities/raffle.entity';
75

86
@Injectable()
97
export class UserProcessor {
@@ -13,78 +11,87 @@ export class UserProcessor {
1311

1412
/**
1513
* Called when a TicketPurchased event is indexed.
16-
* Invalidates the buyer's user profile cache and the raffle detail.
14+
*
15+
* Upserts the buyer row and atomically increments:
16+
* - total_tickets_bought by the number of tickets in this purchase
17+
* - total_raffles_entered by 1 (only when this is the buyer's first ticket in this raffle)
18+
*
19+
* Idempotent: skips the update if last_tx_hash already equals txHash.
20+
* Runs inside the caller's QueryRunner when provided (ticket.processor shares its tx).
1721
*/
1822
async handleTicketPurchased(
1923
raffleId: number,
2024
buyer: string,
25+
ticketCount: number,
2126
ledger: number,
2227
txHash: string,
2328
queryRunner?: QueryRunner,
2429
) {
2530
this.logger.log(`Handling TicketPurchased for ${buyer} in raffle ${raffleId}`);
2631
const runner = queryRunner ?? this.dataSource.createQueryRunner();
27-
const startedTx = !queryRunner;
28-
if (!queryRunner) {
32+
const ownTx = !queryRunner;
33+
if (ownTx) {
2934
await runner.connect();
3035
await runner.startTransaction();
3136
}
3237
try {
38+
// 1. Ensure the user row exists
3339
await runner.manager
3440
.createQueryBuilder()
3541
.insert()
3642
.into(UserEntity)
3743
.values({
3844
address: buyer,
3945
firstSeenLedger: ledger,
46+
lastTxHash: null,
4047
})
4148
.orIgnore()
4249
.execute();
4350

44-
await runner.manager
45-
.createQueryBuilder()
46-
.update(UserEntity)
47-
.set({
48-
firstSeenLedger: () => `LEAST(first_seen_ledger, ${ledger})`,
49-
})
50-
.where('address = :buyer', { buyer })
51-
.execute();
52-
53-
const ticketCounts = await runner.manager
54-
.createQueryBuilder(TicketEntity, 't')
55-
.select('COUNT(*)', 'total')
56-
.addSelect('COUNT(DISTINCT t.raffle_id)', 'distinctRaffles')
57-
.where('t.owner = :buyer', { buyer })
58-
.getRawOne();
51+
// 2. Idempotency check — skip if this tx was already applied
52+
const existing = await runner.manager.findOne(UserEntity, {
53+
where: { address: buyer },
54+
select: ['lastTxHash', 'firstSeenLedger'],
55+
});
56+
if (existing?.lastTxHash === txHash) {
57+
this.logger.debug(`TicketPurchased ${txHash} already applied for ${buyer}, skipping`);
58+
if (ownTx) await runner.commitTransaction();
59+
return;
60+
}
5961

60-
const totalTickets = Number(ticketCounts?.total ?? 0);
61-
const distinctRaffles = Number(ticketCounts?.distinctRaffles ?? 0);
62+
// 3. Determine whether this is the buyer's first ticket in this raffle
63+
// (within the same tx so the count is consistent)
64+
const priorTicketInRaffle = await runner.query(
65+
`SELECT 1 FROM tickets WHERE owner = $1 AND raffle_id = $2 AND purchase_tx_hash != $3 LIMIT 1`,
66+
[buyer, raffleId, txHash],
67+
);
68+
const isFirstEntryInRaffle = priorTicketInRaffle.length === 0;
6269

70+
// 4. Atomic increments + update first_seen_ledger + stamp last_tx_hash
6371
await runner.manager
6472
.createQueryBuilder()
6573
.update(UserEntity)
6674
.set({
67-
totalTicketsBought: totalTickets,
68-
totalRafflesEntered: distinctRaffles,
75+
totalTicketsBought: () => `total_tickets_bought + ${ticketCount}`,
76+
totalRafflesEntered: () =>
77+
isFirstEntryInRaffle
78+
? `total_raffles_entered + 1`
79+
: `total_raffles_entered`,
80+
firstSeenLedger: () => `LEAST(first_seen_ledger, ${ledger})`,
81+
lastTxHash: txHash,
6982
})
7083
.where('address = :buyer', { buyer })
7184
.execute();
7285

73-
if (startedTx) {
74-
await runner.commitTransaction();
75-
}
86+
if (ownTx) await runner.commitTransaction();
7687

7788
await this.cacheService.invalidateUserProfile(buyer);
7889
await this.cacheService.invalidateRaffleDetail(raffleId.toString());
7990
} catch (e) {
80-
if (startedTx) {
81-
await runner.rollbackTransaction();
82-
}
91+
if (ownTx) await runner.rollbackTransaction();
8392
throw e;
8493
} finally {
85-
if (startedTx) {
86-
await runner.release();
87-
}
94+
if (ownTx) await runner.release();
8895
}
8996
}
9097

@@ -99,78 +106,99 @@ export class UserProcessor {
99106
await this.cacheService.invalidateRaffleDetail(raffleId);
100107
}
101108

109+
/**
110+
* Called when a RaffleFinalized event is indexed.
111+
*
112+
* Upserts the winner row and atomically increments:
113+
* - total_raffles_won by 1
114+
* - total_prize_xlm by prizeAmount (bigint string addition via PostgreSQL numeric cast)
115+
*
116+
* Idempotent: keyed by a synthetic tx_hash derived from raffleId so a replay
117+
* of the same finalization is a no-op.
118+
* Runs inside the caller's QueryRunner when provided (raffle.processor shares its tx).
119+
*/
102120
async handleRaffleFinalized(
103121
raffleId: number,
104122
winner: string | null,
105123
prizeAmount: string,
106124
queryRunner?: QueryRunner,
107125
) {
108126
if (!winner) return;
109-
this.logger.log(`Handling RaffleFinalized for ${raffleId} winner ${winner}`);
127+
this.logger.log(`Handling RaffleFinalized for raffle ${raffleId}, winner ${winner}`);
128+
129+
// Synthetic idempotency key — one finalization per raffle
130+
const txHash = `finalized:${raffleId}`;
131+
110132
const runner = queryRunner ?? this.dataSource.createQueryRunner();
111-
const startedTx = !queryRunner;
112-
if (!queryRunner) {
133+
const ownTx = !queryRunner;
134+
if (ownTx) {
113135
await runner.connect();
114136
await runner.startTransaction();
115137
}
116138
try {
139+
// 1. Ensure the winner row exists
117140
await runner.manager
118141
.createQueryBuilder()
119142
.insert()
120143
.into(UserEntity)
121144
.values({
122145
address: winner,
123146
firstSeenLedger: 0,
147+
lastTxHash: null,
124148
})
125149
.orIgnore()
126150
.execute();
127151

128-
const raw = await runner.query(
129-
`SELECT COUNT(*)::int AS wins, COALESCE(SUM(prize_amount::numeric), 0)::text AS total_prize
130-
FROM raffles
131-
WHERE winner = $1`,
132-
[winner],
133-
);
134-
const wins = Number(raw?.[0]?.wins ?? 0);
135-
const totalPrize = String(raw?.[0]?.total_prize ?? '0');
152+
// 2. Idempotency check
153+
const existing = await runner.manager.findOne(UserEntity, {
154+
where: { address: winner },
155+
select: ['lastTxHash'],
156+
});
157+
if (existing?.lastTxHash === txHash) {
158+
this.logger.debug(`RaffleFinalized ${raffleId} already applied for ${winner}, skipping`);
159+
if (ownTx) await runner.commitTransaction();
160+
return;
161+
}
136162

163+
// 3. Atomic increments — add prize using PostgreSQL numeric arithmetic
164+
const safePrize = BigInt(prizeAmount || '0').toString(); // guard against non-numeric input
137165
await runner.manager
138166
.createQueryBuilder()
139167
.update(UserEntity)
140168
.set({
141-
totalRafflesWon: wins,
142-
totalPrizeXlm: totalPrize,
169+
totalRafflesWon: () => `total_raffles_won + 1`,
170+
totalPrizeXlm: () => `(total_prize_xlm::numeric + ${safePrize})::text`,
171+
lastTxHash: txHash,
143172
})
144173
.where('address = :winner', { winner })
145174
.execute();
146175

147-
if (startedTx) {
148-
await runner.commitTransaction();
149-
}
176+
if (ownTx) await runner.commitTransaction();
150177

151178
await this.cacheService.invalidateUserProfile(winner);
152179
await this.cacheService.invalidateLeaderboard();
153180
} catch (e) {
154-
if (startedTx) {
155-
await runner.rollbackTransaction();
156-
}
181+
if (ownTx) await runner.rollbackTransaction();
157182
throw e;
158183
} finally {
159-
if (startedTx) {
160-
await runner.release();
161-
}
184+
if (ownTx) await runner.release();
162185
}
163186
}
164187

188+
/**
189+
* Called when a RaffleCreated event is indexed.
190+
* Ensures the creator has a user row and updates first_seen_ledger.
191+
* No stats to increment — creation is tracked on the raffles table.
192+
*/
165193
async handleRaffleCreated(
166194
creator: string,
167195
createdLedger: number,
168196
queryRunner?: QueryRunner,
169197
) {
170198
this.logger.log(`Handling RaffleCreated by ${creator}`);
171199
const runner = queryRunner ?? this.dataSource.createQueryRunner();
172-
const startedTx = !queryRunner;
173-
if (!queryRunner) {
200+
const ownTx = !queryRunner;
201+
if (ownTx) {
174202
await runner.connect();
175203
await runner.startTransaction();
176204
}
@@ -182,6 +210,7 @@ export class UserProcessor {
182210
.values({
183211
address: creator,
184212
firstSeenLedger: createdLedger,
213+
lastTxHash: null,
185214
})
186215
.orIgnore()
187216
.execute();
@@ -195,18 +224,12 @@ export class UserProcessor {
195224
.where('address = :creator', { creator })
196225
.execute();
197226

198-
if (startedTx) {
199-
await runner.commitTransaction();
200-
}
227+
if (ownTx) await runner.commitTransaction();
201228
} catch (e) {
202-
if (startedTx) {
203-
await runner.rollbackTransaction();
204-
}
229+
if (ownTx) await runner.rollbackTransaction();
205230
throw e;
206231
} finally {
207-
if (startedTx) {
208-
await runner.release();
209-
}
232+
if (ownTx) await runner.release();
210233
}
211234
}
212235
}

0 commit comments

Comments
 (0)