Skip to content

Commit 2484f8c

Browse files
committed
chore: tidy up app.ts and soroban-event-worker.ts
1 parent 8003365 commit 2484f8c

2 files changed

Lines changed: 54 additions & 20 deletions

File tree

backend/src/app.ts

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@ import { sandboxMiddleware } from "./middleware/sandbox.middleware.js";
1414
import { globalRateLimiter } from "./middleware/rate-limiter.middleware.js";
1515
import { requestIdMiddleware } from "./middleware/requestId.js";
1616
import v1Routes from "./routes/v1/index.js";
17-
1817
import healthRoutes from "./routes/health.routes.js";
1918
import "./lib/stream-id.js";
2019

backend/src/workers/soroban-event-worker.ts

Lines changed: 54 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,6 @@ import { Prisma } from "../generated/prisma/index.js";
88
import "../lib/stream-id.js";
99
import { rpcPool } from "../lib/rpc-pool.js";
1010

11-
// ─── Config ──────────────────────────────────────────────────────────────────
12-
1311
// ─── XDR Decoding Helpers ────────────────────────────────────────────────────
1412

1513
/** Decode an ScVal symbol to a string. */
@@ -116,7 +114,8 @@ export class SorobanEventWorker {
116114

117115
constructor() {
118116
this.contractId = process.env.STREAM_CONTRACT_ID ?? "";
119-
const rpcUrl = process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org";
117+
const rpcUrl =
118+
process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org";
120119
this.server = new rpc.Server(rpcUrl, { allowHttp: true });
121120
this.pollIntervalMs = parseInt(
122121
process.env.INDEXER_POLL_INTERVAL_MS ?? "5000",
@@ -205,16 +204,22 @@ export class SorobanEventWorker {
205204
*/
206205
async triggerPoll(customRequestId?: string): Promise<string> {
207206
if (!this.isRunning) {
208-
return customRequestId || requestContext?.getStore?.()?.requestId || randomUUID();
207+
return (
208+
customRequestId ||
209+
requestContext?.getStore?.()?.requestId ||
210+
randomUUID()
211+
);
209212
}
210213

211214
const requestId =
212-
customRequestId || requestContext?.getStore?.()?.requestId || randomUUID();
215+
customRequestId ||
216+
requestContext?.getStore?.()?.requestId ||
217+
randomUUID();
213218

214219
try {
215220
await this.runExclusive(() => {
216221
const runBatch = () => this.fetchAndProcessEvents();
217-
return requestContext && typeof requestContext.run === 'function'
222+
return requestContext && typeof requestContext.run === "function"
218223
? requestContext.run({ requestId }, runBatch)
219224
: runBatch();
220225
});
@@ -292,7 +297,7 @@ export class SorobanEventWorker {
292297
this.fetchAndProcessEvents().catch((err) => {
293298
logger.error("[SorobanWorker] Unhandled error during poll:", err);
294299
});
295-
return requestContext && typeof requestContext.run === 'function'
300+
return requestContext && typeof requestContext.run === "function"
296301
? requestContext.run({ requestId }, execute)
297302
: execute();
298303
});
@@ -307,7 +312,11 @@ export class SorobanEventWorker {
307312
*/
308313
private async fetchAndProcessEvents(): Promise<void> {
309314
const currentCtx = requestContext?.getStore?.();
310-
if (!currentCtx?.requestId && requestContext && typeof requestContext.run === 'function') {
315+
if (
316+
!currentCtx?.requestId &&
317+
requestContext &&
318+
typeof requestContext.run === "function"
319+
) {
311320
const requestId = randomUUID();
312321
return requestContext.run({ requestId }, () =>
313322
this.fetchAndProcessEvents(),
@@ -336,7 +345,9 @@ export class SorobanEventWorker {
336345
? { ...baseFilter, cursor: state.lastCursor }
337346
: { ...baseFilter, startLedger: state.lastLedger || this.startLedger };
338347

339-
const response = await (this.server ? this.server.getEvents(params) : rpcPool.execute("getEvents", (server) => server.getEvents(params)));
348+
const response = await (this.server
349+
? this.server.getEvents(params)
350+
: rpcPool.execute("getEvents", (server) => server.getEvents(params)));
340351

341352
if (response.events.length === 0) return;
342353

@@ -384,7 +395,7 @@ export class SorobanEventWorker {
384395
// Use the response's final cursor if provided and no error occurred, otherwise the last valid event's ID
385396
const finalCursor = hasError
386397
? lastCursor
387-
: ((response as any).latestCursor || lastCursor);
398+
: (response as any).latestCursor || lastCursor;
388399

389400
await prisma.indexerState.upsert({
390401
where: { id: INDEXER_STATE_ID },
@@ -717,11 +728,18 @@ export class SorobanEventWorker {
717728
// Check for a duplicate BEFORE mutating any Stream fields so that a
718729
// replayed event never re-applies the top-up.
719730
const existingEvent = await tx.streamEvent.findUnique({
720-
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
731+
where: {
732+
transactionHash_eventType: {
733+
transactionHash: event.txHash,
734+
eventType: "TOPPED_UP",
735+
},
736+
},
721737
select: { id: true },
722738
});
723739
if (existingEvent) {
724-
logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=TOPPED_UP`);
740+
logger.warn(
741+
`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=TOPPED_UP`,
742+
);
725743
return;
726744
}
727745

@@ -739,7 +757,7 @@ export class SorobanEventWorker {
739757
ratePerSecondBigInt === 0n
740758
? null
741759
: BigInt(stream.startTime) +
742-
(BigInt(newDepositedAmount) / ratePerSecondBigInt) +
760+
BigInt(newDepositedAmount) / ratePerSecondBigInt +
743761
BigInt(stream.totalPausedDuration);
744762

745763
await tx.stream.update({
@@ -752,10 +770,15 @@ export class SorobanEventWorker {
752770
});
753771

754772
await tx.streamEvent.upsert({
755-
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
773+
where: {
774+
transactionHash_eventType: {
775+
transactionHash: event.txHash,
776+
eventType: "TOPPED_UP",
777+
},
778+
},
756779
create: {
757780
streamId,
758-
eventType: 'TOPPED_UP',
781+
eventType: "TOPPED_UP",
759782
amount,
760783
transactionHash: event.txHash,
761784
ledgerSequence: event.ledger,
@@ -798,11 +821,18 @@ export class SorobanEventWorker {
798821
// Check for a duplicate BEFORE mutating any Stream fields so that a
799822
// replayed event never double-increments withdrawnAmount.
800823
const existingEvent = await tx.streamEvent.findUnique({
801-
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
824+
where: {
825+
transactionHash_eventType: {
826+
transactionHash: event.txHash,
827+
eventType: "WITHDRAWN",
828+
},
829+
},
802830
select: { id: true },
803831
});
804832
if (existingEvent) {
805-
logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`);
833+
logger.warn(
834+
`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`,
835+
);
806836
return;
807837
}
808838

@@ -824,10 +854,15 @@ export class SorobanEventWorker {
824854
});
825855

826856
await tx.streamEvent.upsert({
827-
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
857+
where: {
858+
transactionHash_eventType: {
859+
transactionHash: event.txHash,
860+
eventType: "WITHDRAWN",
861+
},
862+
},
828863
create: {
829864
streamId,
830-
eventType: 'WITHDRAWN',
865+
eventType: "WITHDRAWN",
831866
amount,
832867
transactionHash: event.txHash,
833868
ledgerSequence: event.ledger,

0 commit comments

Comments
 (0)