diff --git a/bun.lock b/bun.lock index 0cc31a2..1661f96 100644 --- a/bun.lock +++ b/bun.lock @@ -62,6 +62,7 @@ "version": "0.1.0", "dependencies": { "@fundable-indexer/common": "workspace:*", + "typeorm": "^0.3.20", }, }, "indexer/streams": { diff --git a/indexer/common/src/handlers/registry.test.ts b/indexer/common/src/handlers/registry.test.ts index 41cd21c..dfc8ae0 100644 --- a/indexer/common/src/handlers/registry.test.ts +++ b/indexer/common/src/handlers/registry.test.ts @@ -14,8 +14,7 @@ const baseEvent: SorobanEventInput = { }; const ok: HandlerResult = { ok: true }; -const makeHandler = (result: HandlerResult = ok): EventHandler => - vi.fn().mockResolvedValue(result); +const makeHandler = (result: HandlerResult = ok): EventHandler => vi.fn().mockResolvedValue(result); describe("HandlerRegistry", () => { describe("register and matches", () => { @@ -63,9 +62,7 @@ describe("HandlerRegistry", () => { registry.register({ contractId: "CABC123", topic: "stream_created" }, handler); expect(registry.matches(baseEvent)).toEqual([handler]); - expect( - registry.matches({ ...baseEvent, contractId: "COTHER" }), - ).toHaveLength(0); + expect(registry.matches({ ...baseEvent, contractId: "COTHER" })).toHaveLength(0); }); test("returns multiple handlers when several match", () => { @@ -99,10 +96,7 @@ describe("HandlerRegistry", () => { expect(h1).toHaveBeenCalledWith(baseEvent); expect(h2).toHaveBeenCalledWith(baseEvent); - expect(results).toEqual([ - { ok: true }, - { ok: false, error: "boom", retriable: true }, - ]); + expect(results).toEqual([{ ok: true }, { ok: false, error: "boom", retriable: true }]); }); test("returns empty array when no handlers match", async () => { @@ -120,4 +114,3 @@ describe("HandlerRegistry", () => { }); }); }); - diff --git a/indexer/common/src/handlers/registry.ts b/indexer/common/src/handlers/registry.ts index 07a965e..324b4e1 100644 --- a/indexer/common/src/handlers/registry.ts +++ b/indexer/common/src/handlers/registry.ts @@ -1,9 +1,4 @@ -import type { - EventHandler, - HandlerFilter, - HandlerResult, - SorobanEventInput, -} from "./types.js"; +import type { EventHandler, HandlerFilter, HandlerResult, SorobanEventInput } from "./types.js"; interface RegisteredHandler { filter: HandlerFilter; @@ -43,8 +38,8 @@ export class HandlerRegistry { ok: false as const, error: err instanceof Error ? err.message : String(err), retriable: true, - })) - ) + })), + ), ); } } diff --git a/indexer/common/src/handlers/types.ts b/indexer/common/src/handlers/types.ts index 444c418..8bb5dbe 100644 --- a/indexer/common/src/handlers/types.ts +++ b/indexer/common/src/handlers/types.ts @@ -8,9 +8,7 @@ export interface SorobanEventInput { pagingToken: string; } -export type HandlerResult = - | { ok: true } - | { ok: false; error: string; retriable: boolean }; +export type HandlerResult = { ok: true } | { ok: false; error: string; retriable: boolean }; export type EventHandler = (event: SorobanEventInput) => Promise; diff --git a/indexer/distributions/package.json b/indexer/distributions/package.json index 40a5dcc..1abd73c 100644 --- a/indexer/distributions/package.json +++ b/indexer/distributions/package.json @@ -15,6 +15,7 @@ "type-check": "tsc -p tsconfig.json --noEmit" }, "dependencies": { - "@fundable-indexer/common": "workspace:*" + "@fundable-indexer/common": "workspace:*", + "typeorm": "^0.3.20" } } diff --git a/indexer/distributions/src/db/entity/ClaimAction.ts b/indexer/distributions/src/db/entity/ClaimAction.ts new file mode 100644 index 0000000..f4c9c2a --- /dev/null +++ b/indexer/distributions/src/db/entity/ClaimAction.ts @@ -0,0 +1,55 @@ +import { + Column, + CreateDateColumn, + Entity, + Index, + JoinColumn, + ManyToOne, + PrimaryGeneratedColumn, + Unique, +} from "typeorm"; +import { DistributionBatch } from "./DistributionBatch.js"; + +/** + * A single token claim against a distribution batch. + * + * The `(txHash, ledgerNumber, eventIndex)` triple is unique so a replayed + * `tokens_claimed` event cannot create a duplicate claim row, which keeps the + * batch's `claimedAmount` from being double-counted. + */ +@Entity("distribution_claim_action") +@Unique("uq_claim_event_identity", ["txHash", "ledgerNumber", "eventIndex"]) +export class ClaimAction { + @PrimaryGeneratedColumn("uuid") + id!: string; + + @Index() + @Column({ type: "varchar", comment: "The ID of the distribution batch this claim belongs to" }) + batchId!: string; + + @ManyToOne(() => DistributionBatch, { onDelete: "CASCADE" }) + @JoinColumn({ name: "batchId" }) + batch!: DistributionBatch; + + @Index() + @Column({ type: "varchar", comment: "The address that claimed tokens" }) + claimant!: string; + + @Column({ type: "bigint", comment: "The amount claimed" }) + amount!: string; + + @Column({ type: "varchar", comment: "Transaction hash where the claim occurred" }) + txHash!: string; + + @Column({ type: "int", comment: "Ledger number the claim event was indexed at" }) + ledgerNumber!: number; + + @Column({ type: "int", comment: "Deterministic event position within the ledger" }) + eventIndex!: number; + + @Column({ type: "varchar", comment: "Timestamp the claim event closed at" }) + eventTimestamp!: string; + + @CreateDateColumn() + createdAt!: Date; +} diff --git a/indexer/distributions/src/db/entity/DistributionBatch.ts b/indexer/distributions/src/db/entity/DistributionBatch.ts new file mode 100644 index 0000000..b593cea --- /dev/null +++ b/indexer/distributions/src/db/entity/DistributionBatch.ts @@ -0,0 +1,86 @@ +import { Column, CreateDateColumn, Entity, Index, PrimaryColumn, UpdateDateColumn } from "typeorm"; + +/** + * Lifecycle status of a distribution batch. Mirrors the `DistributionStatus` + * enum exposed in the GraphQL schema. + */ +export enum DistributionStatus { + ACTIVE = "ACTIVE", + PAUSED = "PAUSED", + COMPLETED = "COMPLETED", + CANCELLED = "CANCELLED", +} + +/** + * Durable state for a distribution created on-chain. The primary key is the + * deterministic on-chain distribution ID so re-processing a `distribution_created` + * event maps to the same row. + */ +@Entity("distribution_batch") +export class DistributionBatch { + @PrimaryColumn({ + type: "varchar", + comment: "The deterministic on-chain distribution ID", + }) + id!: string; + + @Index() + @Column({ type: "varchar", comment: "The distribution contract address" }) + contractId!: string; + + @Index() + @Column({ type: "varchar", comment: "The address that created the distribution" }) + distributor!: string; + + @Column({ type: "varchar", comment: "The token asset address" }) + token!: string; + + @Column({ type: "bigint", comment: "The total amount of tokens in the distribution" }) + totalAmount!: string; + + @Column({ type: "bigint", default: "0", comment: "Total amount claimed so far" }) + claimedAmount!: string; + + @Column({ type: "int", comment: "Number of recipients in the distribution" }) + recipientCount!: number; + + @Column({ + type: "enum", + enum: DistributionStatus, + default: DistributionStatus.ACTIVE, + comment: "Current lifecycle status of the distribution", + }) + status!: DistributionStatus; + + @Column({ type: "varchar", nullable: true, comment: "Timestamp the distribution was paused" }) + pausedAt!: string | null; + + @Column({ type: "varchar", nullable: true, comment: "Timestamp the distribution was resumed" }) + resumedAt!: string | null; + + @Column({ + type: "int", + nullable: true, + comment: "Ledger of the last applied status change, used to reject stale pause/resume writes", + }) + statusLedger!: number | null; + + @Column({ + type: "varchar", + unique: true, + comment: "Unique reference for the batch (on-chain distribution ID)", + }) + uniqueRef!: string; + + @Column({ type: "int", comment: "Ledger number the creation event was indexed at" }) + ledgerNumber!: number; + + @Column({ type: "varchar", comment: "Transaction hash where the distribution was created" }) + txHash!: string; + + @CreateDateColumn() + createdAt!: Date; + + @UpdateDateColumn() + updatedAt!: Date; +} diff --git a/indexer/distributions/src/db/migrations/00001_InitialDistributionsSchema.ts b/indexer/distributions/src/db/migrations/00001_InitialDistributionsSchema.ts new file mode 100644 index 0000000..5d393f7 --- /dev/null +++ b/indexer/distributions/src/db/migrations/00001_InitialDistributionsSchema.ts @@ -0,0 +1,61 @@ +import type { MigrationInterface, QueryRunner } from "typeorm"; + +export class InitialDistributionsSchema00001 implements MigrationInterface { + public async up(queryRunner: QueryRunner): Promise { + // Note: In a real environment, this can be auto-generated by TypeORM. + // These tables match the DistributionBatch and ClaimAction entities. + await queryRunner.query(` + CREATE EXTENSION IF NOT EXISTS "uuid-ossp"; + + DO $$ BEGIN + CREATE TYPE "distribution_batch_status_enum" AS ENUM ('ACTIVE', 'PAUSED', 'COMPLETED', 'CANCELLED'); + EXCEPTION + WHEN duplicate_object THEN null; + END $$; + + CREATE TABLE "distribution_batch" ( + "id" varchar PRIMARY KEY NOT NULL, + "contractId" varchar NOT NULL, + "distributor" varchar NOT NULL, + "token" varchar NOT NULL, + "totalAmount" bigint NOT NULL, + "claimedAmount" bigint NOT NULL DEFAULT '0', + "recipientCount" integer NOT NULL, + "status" "distribution_batch_status_enum" NOT NULL DEFAULT 'ACTIVE', + "pausedAt" varchar, + "resumedAt" varchar, + "statusLedger" integer, + "uniqueRef" varchar NOT NULL, + "ledgerNumber" integer NOT NULL, + "txHash" varchar NOT NULL, + "createdAt" TIMESTAMP NOT NULL DEFAULT now(), + "updatedAt" TIMESTAMP NOT NULL DEFAULT now(), + CONSTRAINT "uq_distribution_batch_unique_ref" UNIQUE ("uniqueRef") + ); + CREATE INDEX "IDX_distribution_batch_contractId" ON "distribution_batch" ("contractId"); + CREATE INDEX "IDX_distribution_batch_distributor" ON "distribution_batch" ("distributor"); + + CREATE TABLE "distribution_claim_action" ( + "id" uuid PRIMARY KEY DEFAULT uuid_generate_v4(), + "batchId" varchar NOT NULL, + "claimant" varchar NOT NULL, + "amount" bigint NOT NULL, + "txHash" varchar NOT NULL, + "ledgerNumber" integer NOT NULL, + "eventIndex" integer NOT NULL, + "eventTimestamp" varchar NOT NULL, + "createdAt" TIMESTAMP NOT NULL DEFAULT now(), + CONSTRAINT "FK_claim_batch" FOREIGN KEY ("batchId") REFERENCES "distribution_batch"("id") ON DELETE CASCADE, + CONSTRAINT "uq_claim_event_identity" UNIQUE ("txHash", "ledgerNumber", "eventIndex") + ); + CREATE INDEX "IDX_claim_batchId" ON "distribution_claim_action" ("batchId"); + CREATE INDEX "IDX_claim_claimant" ON "distribution_claim_action" ("claimant"); + `); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP TABLE "distribution_claim_action"`); + await queryRunner.query(`DROP TABLE "distribution_batch"`); + await queryRunner.query(`DROP TYPE "distribution_batch_status_enum"`); + } +} diff --git a/indexer/distributions/src/db/repository.test.ts b/indexer/distributions/src/db/repository.test.ts new file mode 100644 index 0000000..8feadf4 --- /dev/null +++ b/indexer/distributions/src/db/repository.test.ts @@ -0,0 +1,165 @@ +import type { DataSource } from "typeorm"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { DistributionRepository } from "./repository.js"; + +// biome-ignore lint/suspicious/noExplicitAny: chainable query-builder mock +function makeQueryBuilder(executeResult: any) { + // biome-ignore lint/suspicious/noExplicitAny: chainable query-builder mock + const qb: any = { + insert: vi.fn(() => qb), + into: vi.fn(() => qb), + values: vi.fn(() => qb), + orIgnore: vi.fn(() => qb), + update: vi.fn(() => qb), + set: vi.fn(() => qb), + where: vi.fn(() => qb), + setParameter: vi.fn(() => qb), + execute: vi.fn(async () => executeResult), + }; + return qb; +} + +describe("DistributionRepository", () => { + // biome-ignore lint/suspicious/noExplicitAny: mock objects + let mockDataSource: any; + // biome-ignore lint/suspicious/noExplicitAny: mock objects + let mockBatchRepo: any; + // biome-ignore lint/suspicious/noExplicitAny: mock objects + let createQb: any; + + beforeEach(() => { + createQb = makeQueryBuilder({ identifiers: [] }); + mockBatchRepo = { + createQueryBuilder: vi.fn(() => createQb), + update: vi.fn(async () => ({ affected: 1 })), + }; + mockDataSource = { + getRepository: vi.fn(() => mockBatchRepo), + transaction: vi.fn(), + }; + }); + + it("inserts a batch ignoring conflicts", async () => { + const repo = new DistributionRepository(mockDataSource as unknown as DataSource); + await repo.createBatch({ + distributionId: "dist-1", + contractId: "CDIST", + distributor: "GCREATOR", + token: "USDC", + totalAmount: "1000", + recipientCount: 10, + ledgerNumber: 42, + txHash: "txabc", + }); + + expect(createQb.orIgnore).toHaveBeenCalled(); + expect(createQb.values).toHaveBeenCalledWith( + expect.objectContaining({ + id: "dist-1", + uniqueRef: "dist-1", + claimedAmount: "0", + recipientCount: 10, + }), + ); + expect(createQb.execute).toHaveBeenCalled(); + }); + + it("records a claim and increments the batch when a row is inserted", async () => { + const insertQb = makeQueryBuilder({ identifiers: [{ id: "claim-uuid" }] }); + const updateQb = makeQueryBuilder({ affected: 1 }); + const manager = { + createQueryBuilder: vi.fn().mockReturnValueOnce(insertQb).mockReturnValueOnce(updateQb), + }; + mockDataSource.transaction = vi.fn(async (cb: (m: unknown) => Promise) => cb(manager)); + + const repo = new DistributionRepository(mockDataSource as unknown as DataSource); + await repo.recordClaim({ + distributionId: "dist-1", + claimant: "GCLAIMANT", + amount: "500", + txHash: "txclaim", + ledgerNumber: 42, + eventIndex: 3, + eventTimestamp: "2024-06-15T00:00:00Z", + }); + + expect(insertQb.orIgnore).toHaveBeenCalled(); + expect(updateQb.update).toHaveBeenCalled(); + expect(updateQb.setParameter).toHaveBeenCalledWith("claimAmount", "500"); + expect(updateQb.execute).toHaveBeenCalled(); + }); + + it("does not increment the batch when the claim was a duplicate", async () => { + const insertQb = makeQueryBuilder({ identifiers: [] }); + const updateQb = makeQueryBuilder({ affected: 1 }); + const manager = { + createQueryBuilder: vi.fn().mockReturnValueOnce(insertQb).mockReturnValueOnce(updateQb), + }; + mockDataSource.transaction = vi.fn(async (cb: (m: unknown) => Promise) => cb(manager)); + + const repo = new DistributionRepository(mockDataSource as unknown as DataSource); + await repo.recordClaim({ + distributionId: "dist-1", + claimant: "GCLAIMANT", + amount: "500", + txHash: "txclaim", + ledgerNumber: 42, + eventIndex: 3, + eventTimestamp: "2024-06-15T00:00:00Z", + }); + + expect(insertQb.execute).toHaveBeenCalled(); + expect(updateQb.update).not.toHaveBeenCalled(); + }); + + it("applies a paused status with pausedAt and statusLedger", async () => { + const repo = new DistributionRepository(mockDataSource as unknown as DataSource); + await repo.setStatus({ + distributionId: "dist-1", + // biome-ignore lint/suspicious/noExplicitAny: enum imported indirectly + status: "PAUSED" as any, + ledgerNumber: 50, + changedAt: "2024-06-15T00:00:00Z", + }); + + expect(createQb.set).toHaveBeenCalledWith({ + status: "PAUSED", + pausedAt: "2024-06-15T00:00:00Z", + statusLedger: 50, + }); + expect(createQb.execute).toHaveBeenCalled(); + }); + + it("applies a resumed status with resumedAt and statusLedger", async () => { + const repo = new DistributionRepository(mockDataSource as unknown as DataSource); + await repo.setStatus({ + distributionId: "dist-1", + // biome-ignore lint/suspicious/noExplicitAny: enum imported indirectly + status: "ACTIVE" as any, + ledgerNumber: 51, + changedAt: "2024-06-15T01:00:00Z", + }); + + expect(createQb.set).toHaveBeenCalledWith({ + status: "ACTIVE", + resumedAt: "2024-06-15T01:00:00Z", + statusLedger: 51, + }); + expect(createQb.execute).toHaveBeenCalled(); + }); + + it("guards status updates against stale ledgers", async () => { + const repo = new DistributionRepository(mockDataSource as unknown as DataSource); + await repo.setStatus({ + distributionId: "dist-1", + // biome-ignore lint/suspicious/noExplicitAny: enum imported indirectly + status: "PAUSED" as any, + ledgerNumber: 50, + changedAt: "2024-06-15T00:00:00Z", + }); + + const [clause, params] = createQb.where.mock.calls[0]; + expect(clause).toContain('"statusLedger" IS NULL OR "statusLedger" <= :ledgerNumber'); + expect(params).toEqual({ distributionId: "dist-1", ledgerNumber: 50 }); + }); +}); diff --git a/indexer/distributions/src/db/repository.ts b/indexer/distributions/src/db/repository.ts new file mode 100644 index 0000000..44c3e70 --- /dev/null +++ b/indexer/distributions/src/db/repository.ts @@ -0,0 +1,151 @@ +import type { DataSource } from "typeorm"; +import { ClaimAction } from "./entity/ClaimAction.js"; +import { DistributionBatch, DistributionStatus } from "./entity/DistributionBatch.js"; + +/** Payload required to persist a newly created distribution batch. */ +export interface CreateBatchInput { + distributionId: string; + contractId: string; + distributor: string; + token: string; + totalAmount: string; + recipientCount: number; + ledgerNumber: number; + txHash: string; +} + +/** Payload required to record a token claim against a batch. */ +export interface RecordClaimInput { + distributionId: string; + claimant: string; + amount: string; + txHash: string; + ledgerNumber: number; + eventIndex: number; + eventTimestamp: string; +} + +/** The only status transitions the pause/resume handlers drive. */ +export type PauseResumeStatus = DistributionStatus.ACTIVE | DistributionStatus.PAUSED; + +/** Payload required to apply a pause/resume status change. */ +export interface SetStatusInput { + distributionId: string; + status: PauseResumeStatus; + /** Ledger the status-change event was observed at, used to reject stale writes. */ + ledgerNumber: number; + /** ISO timestamp the event closed at, recorded on the matching status field. */ + changedAt: string; +} + +/** + * Explicit persistence API for distribution batches and claims. + * + * Methods are written to be safe under event replay: batch creation ignores + * conflicting inserts, claim recording is keyed on a unique event identity, and + * the derived `claimedAmount` is only incremented when a new claim row is + * actually inserted. + */ +export interface DistributionPersistence { + createBatch(input: CreateBatchInput): Promise; + recordClaim(input: RecordClaimInput): Promise; + setStatus(input: SetStatusInput): Promise; +} + +export class DistributionRepository implements DistributionPersistence { + constructor(private readonly dataSource: DataSource) {} + + /** + * Inserts a distribution batch. A repeated `distribution_created` event for + * the same on-chain ID is ignored rather than overwriting existing state. + */ + async createBatch(input: CreateBatchInput): Promise { + await this.dataSource + .getRepository(DistributionBatch) + .createQueryBuilder() + .insert() + .into(DistributionBatch) + .values({ + id: input.distributionId, + uniqueRef: input.distributionId, + contractId: input.contractId, + distributor: input.distributor, + token: input.token, + totalAmount: input.totalAmount, + claimedAmount: "0", + recipientCount: input.recipientCount, + status: DistributionStatus.ACTIVE, + ledgerNumber: input.ledgerNumber, + txHash: input.txHash, + }) + .orIgnore() + .execute(); + } + + /** + * Records a claim and increments the batch's claimed total in one transaction. + * The claim insert ignores duplicates on the unique event identity, and the + * batch total is only advanced when a row is actually written, so replays do + * not double-count. + */ + async recordClaim(input: RecordClaimInput): Promise { + await this.dataSource.transaction(async (manager) => { + const result = await manager + .createQueryBuilder() + .insert() + .into(ClaimAction) + .values({ + batchId: input.distributionId, + claimant: input.claimant, + amount: input.amount, + txHash: input.txHash, + ledgerNumber: input.ledgerNumber, + eventIndex: input.eventIndex, + eventTimestamp: input.eventTimestamp, + }) + .orIgnore() + .execute(); + + const inserted = (result.identifiers ?? []).some((id) => id != null); + if (!inserted) { + return; + } + + await manager + .createQueryBuilder() + .update(DistributionBatch) + .set({ + claimedAmount: () => `"claimedAmount" + :claimAmount`, + }) + .where("id = :distributionId", { distributionId: input.distributionId }) + .setParameter("claimAmount", input.amount) + .execute(); + }); + } + + /** + * Applies a paused/resumed status change, recording the timestamp on the + * matching field. No-op if the batch does not exist. + * + * The update is guarded by `statusLedger` so a stale (out-of-order) pause or + * resume cannot overwrite state produced by a newer event — important when an + * earlier event is retried after its sibling already advanced the status. + */ + async setStatus(input: SetStatusInput): Promise { + const patch: Partial = + input.status === DistributionStatus.PAUSED + ? { status: input.status, pausedAt: input.changedAt, statusLedger: input.ledgerNumber } + : { status: input.status, resumedAt: input.changedAt, statusLedger: input.ledgerNumber }; + + await this.dataSource + .getRepository(DistributionBatch) + .createQueryBuilder() + .update(DistributionBatch) + .set(patch) + .where( + '"id" = :distributionId AND ("statusLedger" IS NULL OR "statusLedger" <= :ledgerNumber)', + { distributionId: input.distributionId, ledgerNumber: input.ledgerNumber }, + ) + .execute(); + } +} diff --git a/indexer/distributions/src/handlers/distribution-created.handler.ts b/indexer/distributions/src/handlers/distribution-created.handler.ts index ea8cee0..ad5883a 100644 --- a/indexer/distributions/src/handlers/distribution-created.handler.ts +++ b/indexer/distributions/src/handlers/distribution-created.handler.ts @@ -1,76 +1,102 @@ -import type { - EventHandler, - HandlerResult, - SorobanEventInput, -} from "@fundable-indexer/common"; +import type { EventHandler, HandlerResult, SorobanEventInput } from "@fundable-indexer/common"; +import { type DistributionHandlerDeps, deriveEventIndex } from "./persistence.js"; import { parseDistributionCreated } from "./types.js"; -export const distributionCreatedHandler: EventHandler = async ( - event: SorobanEventInput, -): Promise => { - try { - const payload = parseDistributionCreated(event.data); +/** + * Builds a `distribution_created` handler that persists a distribution batch. + * + * Required payload fields are validated before any write. The event identity is + * checked against the shared event store so a replayed event is a no-op. + */ +export const createDistributionCreatedHandler = (deps: DistributionHandlerDeps): EventHandler => { + return async (event: SorobanEventInput): Promise => { + try { + const payload = parseDistributionCreated(event.data); - if (!payload.distributionId) { - return { - ok: false, - error: "Missing distributionId in distribution_created event", - retriable: false, - }; - } + if (!payload.distributionId) { + return { + ok: false, + error: "Missing distributionId in distribution_created event", + retriable: false, + }; + } - if (!payload.creator) { - return { - ok: false, - error: "Missing creator in distribution_created event", - retriable: false, - }; - } + if (!payload.creator) { + return { + ok: false, + error: "Missing creator in distribution_created event", + retriable: false, + }; + } - if (!payload.token) { - return { - ok: false, - error: "Missing token in distribution_created event", - retriable: false, - }; - } + if (!payload.token) { + return { + ok: false, + error: "Missing token in distribution_created event", + retriable: false, + }; + } - if (!payload.totalAmount) { - return { - ok: false, - error: "Missing totalAmount in distribution_created event", - retriable: false, - }; - } + if (!payload.totalAmount) { + return { + ok: false, + error: "Missing totalAmount in distribution_created event", + retriable: false, + }; + } - if (!payload.transactionHash) { - return { - ok: false, - error: "Missing transactionHash in distribution_created event", - retriable: false, - }; - } + if (!payload.transactionHash) { + return { + ok: false, + error: "Missing transactionHash in distribution_created event", + retriable: false, + }; + } - if (payload.recipientCount <= 0) { + if (payload.recipientCount <= 0) { + return { + ok: false, + error: "Invalid recipientCount in distribution_created event", + retriable: false, + }; + } + + const eventIndex = deriveEventIndex(event); + const alreadyProcessed = await deps.events.isEventProcessed( + event.contractId, + event.ledger, + payload.transactionHash, + eventIndex, + ); + if (alreadyProcessed) { + return { ok: true }; + } + + await deps.distributions.createBatch({ + distributionId: payload.distributionId, + contractId: event.contractId, + distributor: payload.creator, + token: payload.token, + totalAmount: payload.totalAmount, + recipientCount: payload.recipientCount, + ledgerNumber: event.ledger, + txHash: payload.transactionHash, + }); + + await deps.events.recordEventProcessed( + event.contractId, + event.ledger, + payload.transactionHash, + eventIndex, + ); + + return { ok: true }; + } catch (err) { return { ok: false, - error: "Invalid recipientCount in distribution_created event", - retriable: false, + error: err instanceof Error ? err.message : String(err), + retriable: true, }; } - - // TODO(#36): persist distribution batch via repository once DB schema is merged - // TODO(#27): record indexed event identity via event repository - console.info( - `[distribution-created] id=${payload.distributionId} creator=${payload.creator} token=${payload.token} total=${payload.totalAmount} ledger=${event.ledger}`, - ); - - return { ok: true }; - } catch (err) { - return { - ok: false, - error: err instanceof Error ? err.message : String(err), - retriable: true, - }; - } + }; }; diff --git a/indexer/distributions/src/handlers/distribution-handlers.test.ts b/indexer/distributions/src/handlers/distribution-handlers.test.ts index 6c68d32..050fc53 100644 --- a/indexer/distributions/src/handlers/distribution-handlers.test.ts +++ b/indexer/distributions/src/handlers/distribution-handlers.test.ts @@ -1,25 +1,90 @@ -import { describe, expect, test } from "vitest"; +import { beforeEach, describe, expect, test } from "vitest"; import type { SorobanEventInput } from "@fundable-indexer/common"; -import { distributionCreatedHandler } from "./distribution-created.handler.js"; +import type { CreateBatchInput, RecordClaimInput, SetStatusInput } from "../db/repository.js"; +import { createDistributionCreatedHandler } from "./distribution-created.handler.js"; import { - distributionPausedHandler, - distributionResumedHandler, + createDistributionPausedHandler, + createDistributionResumedHandler, } from "./distribution-pause.handler.js"; -import { tokensClaimedHandler } from "./tokens-claimed.handler.js"; +import type { DistributionHandlerDeps } from "./persistence.js"; +import { createTokensClaimedHandler } from "./tokens-claimed.handler.js"; +interface RecordedCalls { + createBatch: CreateBatchInput[]; + recordClaim: RecordClaimInput[]; + setStatus: SetStatusInput[]; + recordedEvents: string[]; +} + +function identityKey( + contractId: string, + ledger: number, + txHash: string, + eventIndex: number, +): string { + return `${contractId}|${ledger}|${txHash}|${eventIndex}`; +} + +function makeDeps(): { deps: DistributionHandlerDeps; calls: RecordedCalls } { + const processed = new Set(); + const calls: RecordedCalls = { + createBatch: [], + recordClaim: [], + setStatus: [], + recordedEvents: [], + }; + + const deps: DistributionHandlerDeps = { + events: { + async isEventProcessed(contractId, ledger, txHash, eventIndex) { + return processed.has(identityKey(contractId, ledger, txHash, eventIndex)); + }, + async recordEventProcessed(contractId, ledger, txHash, eventIndex) { + const key = identityKey(contractId, ledger, txHash, eventIndex); + const isNew = !processed.has(key); + processed.add(key); + calls.recordedEvents.push(key); + return isNew; + }, + }, + distributions: { + async createBatch(input) { + calls.createBatch.push(input); + }, + async recordClaim(input) { + calls.recordClaim.push(input); + }, + async setStatus(input) { + calls.setStatus.push(input); + }, + }, + }; + + return { deps, calls }; +} + +// Realistic Soroban event envelope. Event IDs encode the ledger sequence and +// the trailing event position used to derive a deterministic event index. const baseEvent: SorobanEventInput = { contractId: "CDIST456", ledger: 300, ledgerClosedAt: "2024-06-15T00:00:00Z", topic: ["distribution_created"], data: {}, - id: "event-3", - pagingToken: "paging-3", + id: "0000000523986165760-0000000007", + pagingToken: "0000000523986165760-0000000007", }; describe("distributionCreatedHandler", () => { - test("returns ok for valid created payload", async () => { + let deps: DistributionHandlerDeps; + let calls: RecordedCalls; + + beforeEach(() => { + ({ deps, calls } = makeDeps()); + }); + + test("persists a batch for a valid created payload", async () => { const event: SorobanEventInput = { ...baseEvent, data: { @@ -32,41 +97,123 @@ describe("distributionCreatedHandler", () => { }, }; - const result = await distributionCreatedHandler(event); + const result = await createDistributionCreatedHandler(deps)(event); + expect(result).toEqual({ ok: true }); + expect(calls.createBatch).toHaveLength(1); + expect(calls.createBatch[0]).toMatchObject({ + distributionId: "dist-1", + contractId: "CDIST456", + distributor: "GCREATOR", + token: "USDC", + totalAmount: "100000", + recipientCount: 50, + ledgerNumber: 300, + txHash: "txabc", + }); + expect(calls.recordedEvents).toHaveLength(1); }); - test("returns error when distributionId is missing", async () => { + test("returns error and does not persist when distributionId is missing", async () => { const event: SorobanEventInput = { ...baseEvent, data: { creator: "GCREATOR", token: "USDC" }, }; - const result = await distributionCreatedHandler(event); + const result = await createDistributionCreatedHandler(deps)(event); + expect(result).toMatchObject({ ok: false, retriable: false }); + expect(calls.createBatch).toHaveLength(0); }); - test("is idempotent — processes same payload without throwing", async () => { - const payload = { - distribution_id: "dist-idempotent", - creator: "G123", - token: "XLM", - total_amount: "500", - recipient_count: 5, - tx_hash: "txidem", + test("returns error and does not persist when total_amount is missing", async () => { + const event: SorobanEventInput = { + ...baseEvent, + data: { + distribution_id: "dist-1", + creator: "GCREATOR", + token: "USDC", + recipient_count: 50, + tx_hash: "txabc", + }, }; - const results = await Promise.all([ - distributionCreatedHandler({ ...baseEvent, data: payload }), - distributionCreatedHandler({ ...baseEvent, data: payload }), - ]); + const result = await createDistributionCreatedHandler(deps)(event); + expect(result).toMatchObject({ ok: false, retriable: false }); + expect(calls.createBatch).toHaveLength(0); + }); - expect(results.every((r) => r.ok)).toBe(true); + test("returns error when recipientCount is not positive", async () => { + const event: SorobanEventInput = { + ...baseEvent, + data: { + distribution_id: "dist-1", + creator: "GCREATOR", + token: "USDC", + total_amount: "100000", + recipient_count: 0, + tx_hash: "txabc", + }, + }; + + const result = await createDistributionCreatedHandler(deps)(event); + expect(result).toMatchObject({ ok: false, retriable: false }); + expect(calls.createBatch).toHaveLength(0); + }); + + test("is idempotent — replaying the same event persists the batch once", async () => { + const event: SorobanEventInput = { + ...baseEvent, + data: { + distribution_id: "dist-idempotent", + creator: "G123", + token: "XLM", + total_amount: "500", + recipient_count: 5, + tx_hash: "txidem", + }, + }; + + const handler = createDistributionCreatedHandler(deps); + const first = await handler(event); + const second = await handler(event); + + expect(first).toEqual({ ok: true }); + expect(second).toEqual({ ok: true }); + expect(calls.createBatch).toHaveLength(1); + expect(calls.recordedEvents).toHaveLength(1); + }); + + test("surfaces repository failures as retriable", async () => { + deps.distributions.createBatch = async () => { + throw new Error("connection reset"); + }; + const event: SorobanEventInput = { + ...baseEvent, + data: { + distribution_id: "dist-1", + creator: "GCREATOR", + token: "USDC", + total_amount: "100000", + recipient_count: 50, + tx_hash: "txabc", + }, + }; + + const result = await createDistributionCreatedHandler(deps)(event); + expect(result).toMatchObject({ ok: false, retriable: true }); }); }); describe("tokensClaimedHandler", () => { - test("returns ok for valid claim payload", async () => { + let deps: DistributionHandlerDeps; + let calls: RecordedCalls; + + beforeEach(() => { + ({ deps, calls } = makeDeps()); + }); + + test("records a claim for a valid payload", async () => { const event: SorobanEventInput = { ...baseEvent, topic: ["tokens_claimed"], @@ -78,23 +225,78 @@ describe("tokensClaimedHandler", () => { }, }; - const result = await tokensClaimedHandler(event); + const result = await createTokensClaimedHandler(deps)(event); + expect(result).toEqual({ ok: true }); + expect(calls.recordClaim).toHaveLength(1); + expect(calls.recordClaim[0]).toMatchObject({ + distributionId: "dist-1", + claimant: "GCLAIMANT", + amount: "2000", + txHash: "txclaim", + ledgerNumber: 300, + eventIndex: 7, + eventTimestamp: "2024-06-15T00:00:00Z", + }); }); - test("returns error when distributionId is missing", async () => { + test("returns error and does not record when distributionId is missing", async () => { const event: SorobanEventInput = { ...baseEvent, data: { claimant: "GCLAIM", amount: "10" }, }; - const result = await tokensClaimedHandler(event); + const result = await createTokensClaimedHandler(deps)(event); + expect(result).toMatchObject({ ok: false, retriable: false }); + expect(calls.recordClaim).toHaveLength(0); + }); + + test("returns error when amount is zero", async () => { + const event: SorobanEventInput = { + ...baseEvent, + data: { + distribution_id: "dist-1", + claimant: "GCLAIMANT", + amount: "0", + tx_hash: "txclaim", + }, + }; + + const result = await createTokensClaimedHandler(deps)(event); expect(result).toMatchObject({ ok: false, retriable: false }); + expect(calls.recordClaim).toHaveLength(0); + }); + + test("does not double-count a replayed claim event", async () => { + const event: SorobanEventInput = { + ...baseEvent, + topic: ["tokens_claimed"], + data: { + distribution_id: "dist-1", + claimant: "GCLAIMANT", + amount: "2000", + tx_hash: "txclaim", + }, + }; + + const handler = createTokensClaimedHandler(deps); + await handler(event); + await handler(event); + + expect(calls.recordClaim).toHaveLength(1); + expect(calls.recordedEvents).toHaveLength(1); }); }); describe("distributionPausedHandler", () => { - test("returns ok for valid paused payload", async () => { + let deps: DistributionHandlerDeps; + let calls: RecordedCalls; + + beforeEach(() => { + ({ deps, calls } = makeDeps()); + }); + + test("sets status to PAUSED for a valid payload", async () => { const event: SorobanEventInput = { ...baseEvent, topic: ["distribution_paused"], @@ -105,23 +307,57 @@ describe("distributionPausedHandler", () => { }, }; - const result = await distributionPausedHandler(event); + const result = await createDistributionPausedHandler(deps)(event); + expect(result).toEqual({ ok: true }); + expect(calls.setStatus).toHaveLength(1); + expect(calls.setStatus[0]).toMatchObject({ + distributionId: "dist-1", + status: "PAUSED", + ledgerNumber: 300, + changedAt: "2024-06-15T00:00:00Z", + }); }); - test("returns error when distributionId is missing", async () => { + test("returns error and does not update when distributionId is missing", async () => { const event: SorobanEventInput = { ...baseEvent, data: { paused_by: "GADMIN" }, }; - const result = await distributionPausedHandler(event); + const result = await createDistributionPausedHandler(deps)(event); expect(result).toMatchObject({ ok: false, retriable: false }); + expect(calls.setStatus).toHaveLength(0); + }); + + test("is idempotent on replay", async () => { + const event: SorobanEventInput = { + ...baseEvent, + topic: ["distribution_paused"], + data: { + distribution_id: "dist-1", + paused_by: "GADMIN", + tx_hash: "txpause", + }, + }; + + const handler = createDistributionPausedHandler(deps); + await handler(event); + await handler(event); + + expect(calls.setStatus).toHaveLength(1); }); }); describe("distributionResumedHandler", () => { - test("returns ok for valid resumed payload", async () => { + let deps: DistributionHandlerDeps; + let calls: RecordedCalls; + + beforeEach(() => { + ({ deps, calls } = makeDeps()); + }); + + test("sets status to ACTIVE for a valid payload", async () => { const event: SorobanEventInput = { ...baseEvent, topic: ["distribution_resumed"], @@ -132,17 +368,44 @@ describe("distributionResumedHandler", () => { }, }; - const result = await distributionResumedHandler(event); + const result = await createDistributionResumedHandler(deps)(event); + expect(result).toEqual({ ok: true }); + expect(calls.setStatus).toHaveLength(1); + expect(calls.setStatus[0]).toMatchObject({ + distributionId: "dist-1", + status: "ACTIVE", + ledgerNumber: 300, + changedAt: "2024-06-15T00:00:00Z", + }); }); - test("returns error when distributionId is missing", async () => { + test("returns error and does not update when resumedBy is missing", async () => { const event: SorobanEventInput = { ...baseEvent, - data: { resumed_by: "GADMIN" }, + data: { distribution_id: "dist-1", tx_hash: "txresume" }, }; - const result = await distributionResumedHandler(event); + const result = await createDistributionResumedHandler(deps)(event); expect(result).toMatchObject({ ok: false, retriable: false }); + expect(calls.setStatus).toHaveLength(0); + }); + + test("is idempotent on replay", async () => { + const event: SorobanEventInput = { + ...baseEvent, + topic: ["distribution_resumed"], + data: { + distribution_id: "dist-1", + resumed_by: "GADMIN", + tx_hash: "txresume", + }, + }; + + const handler = createDistributionResumedHandler(deps); + await handler(event); + await handler(event); + + expect(calls.setStatus).toHaveLength(1); }); }); diff --git a/indexer/distributions/src/handlers/distribution-pause.handler.ts b/indexer/distributions/src/handlers/distribution-pause.handler.ts index 2f73ff3..62fa8b3 100644 --- a/indexer/distributions/src/handlers/distribution-pause.handler.ts +++ b/indexer/distributions/src/handlers/distribution-pause.handler.ts @@ -1,96 +1,142 @@ -import type { - EventHandler, - HandlerResult, - SorobanEventInput, -} from "@fundable-indexer/common"; +import type { EventHandler, HandlerResult, SorobanEventInput } from "@fundable-indexer/common"; +import { DistributionStatus } from "../db/entity/DistributionBatch.js"; +import { type DistributionHandlerDeps, deriveEventIndex } from "./persistence.js"; import { parseDistributionPaused, parseDistributionResumed } from "./types.js"; -export const distributionPausedHandler: EventHandler = async ( - event: SorobanEventInput, -): Promise => { - try { - const payload = parseDistributionPaused(event.data); +/** + * Builds a `distribution_paused` handler that flips a batch to PAUSED and + * records when it happened. Replayed events are ignored via the event store. + */ +export const createDistributionPausedHandler = (deps: DistributionHandlerDeps): EventHandler => { + return async (event: SorobanEventInput): Promise => { + try { + const payload = parseDistributionPaused(event.data); - if (!payload.distributionId) { - return { - ok: false, - error: "Missing distributionId in distribution_paused event", - retriable: false, - }; - } + if (!payload.distributionId) { + return { + ok: false, + error: "Missing distributionId in distribution_paused event", + retriable: false, + }; + } - if (!payload.pausedBy) { - return { - ok: false, - error: "Missing pausedBy in distribution_paused event", - retriable: false, - }; - } + if (!payload.pausedBy) { + return { + ok: false, + error: "Missing pausedBy in distribution_paused event", + retriable: false, + }; + } - if (!payload.transactionHash) { - return { - ok: false, - error: "Missing transactionHash in distribution_paused event", - retriable: false, - }; - } + if (!payload.transactionHash) { + return { + ok: false, + error: "Missing transactionHash in distribution_paused event", + retriable: false, + }; + } - // TODO(#36): update distribution status to PAUSED via repository - console.info( - `[distribution-paused] id=${payload.distributionId} by=${payload.pausedBy} ledger=${event.ledger}`, - ); - - return { ok: true }; - } catch (err) { - return { - ok: false, - error: err instanceof Error ? err.message : String(err), - retriable: true, - }; - } -}; + const eventIndex = deriveEventIndex(event); + const alreadyProcessed = await deps.events.isEventProcessed( + event.contractId, + event.ledger, + payload.transactionHash, + eventIndex, + ); + if (alreadyProcessed) { + return { ok: true }; + } -export const distributionResumedHandler: EventHandler = async ( - event: SorobanEventInput, -): Promise => { - try { - const payload = parseDistributionResumed(event.data); + await deps.distributions.setStatus({ + distributionId: payload.distributionId, + status: DistributionStatus.PAUSED, + ledgerNumber: event.ledger, + changedAt: event.ledgerClosedAt, + }); - if (!payload.distributionId) { - return { - ok: false, - error: "Missing distributionId in distribution_resumed event", - retriable: false, - }; - } + await deps.events.recordEventProcessed( + event.contractId, + event.ledger, + payload.transactionHash, + eventIndex, + ); - if (!payload.resumedBy) { + return { ok: true }; + } catch (err) { return { ok: false, - error: "Missing resumedBy in distribution_resumed event", - retriable: false, + error: err instanceof Error ? err.message : String(err), + retriable: true, }; } + }; +}; + +/** + * Builds a `distribution_resumed` handler that flips a batch back to ACTIVE and + * records when it happened. Replayed events are ignored via the event store. + */ +export const createDistributionResumedHandler = (deps: DistributionHandlerDeps): EventHandler => { + return async (event: SorobanEventInput): Promise => { + try { + const payload = parseDistributionResumed(event.data); + + if (!payload.distributionId) { + return { + ok: false, + error: "Missing distributionId in distribution_resumed event", + retriable: false, + }; + } - if (!payload.transactionHash) { + if (!payload.resumedBy) { + return { + ok: false, + error: "Missing resumedBy in distribution_resumed event", + retriable: false, + }; + } + + if (!payload.transactionHash) { + return { + ok: false, + error: "Missing transactionHash in distribution_resumed event", + retriable: false, + }; + } + + const eventIndex = deriveEventIndex(event); + const alreadyProcessed = await deps.events.isEventProcessed( + event.contractId, + event.ledger, + payload.transactionHash, + eventIndex, + ); + if (alreadyProcessed) { + return { ok: true }; + } + + await deps.distributions.setStatus({ + distributionId: payload.distributionId, + status: DistributionStatus.ACTIVE, + ledgerNumber: event.ledger, + changedAt: event.ledgerClosedAt, + }); + + await deps.events.recordEventProcessed( + event.contractId, + event.ledger, + payload.transactionHash, + eventIndex, + ); + + return { ok: true }; + } catch (err) { return { ok: false, - error: "Missing transactionHash in distribution_resumed event", - retriable: false, + error: err instanceof Error ? err.message : String(err), + retriable: true, }; } - - // TODO(#36): update distribution status to ACTIVE via repository - console.info( - `[distribution-resumed] id=${payload.distributionId} by=${payload.resumedBy} ledger=${event.ledger}`, - ); - - return { ok: true }; - } catch (err) { - return { - ok: false, - error: err instanceof Error ? err.message : String(err), - retriable: true, - }; - } + }; }; diff --git a/indexer/distributions/src/handlers/index.ts b/indexer/distributions/src/handlers/index.ts index cad4931..538c147 100644 --- a/indexer/distributions/src/handlers/index.ts +++ b/indexer/distributions/src/handlers/index.ts @@ -1,7 +1,12 @@ -export { distributionCreatedHandler } from "./distribution-created.handler.js"; -export { tokensClaimedHandler } from "./tokens-claimed.handler.js"; +export { createDistributionCreatedHandler } from "./distribution-created.handler.js"; +export { createTokensClaimedHandler } from "./tokens-claimed.handler.js"; export { - distributionPausedHandler, - distributionResumedHandler, + createDistributionPausedHandler, + createDistributionResumedHandler, } from "./distribution-pause.handler.js"; +export { + type DistributionHandlerDeps, + type EventIdentityStore, + deriveEventIndex, +} from "./persistence.js"; export * from "./types.js"; diff --git a/indexer/distributions/src/handlers/persistence.ts b/indexer/distributions/src/handlers/persistence.ts new file mode 100644 index 0000000..bcde105 --- /dev/null +++ b/indexer/distributions/src/handlers/persistence.ts @@ -0,0 +1,43 @@ +import type { SorobanEventInput } from "@fundable-indexer/common"; +import type { DistributionPersistence } from "../db/repository.js"; + +/** + * Subset of the shared `EventRepository` the distribution handlers depend on to + * coordinate indexed event identity. Declared as an interface so handlers can be + * tested with lightweight mocks. + */ +export interface EventIdentityStore { + isEventProcessed( + contractId: string, + ledgerNumber: number, + txHash: string, + eventIndex: number, + ): Promise; + recordEventProcessed( + contractId: string, + ledgerNumber: number, + txHash: string, + eventIndex: number, + ): Promise; +} + +/** Dependencies injected into each distribution event handler. */ +export interface DistributionHandlerDeps { + distributions: DistributionPersistence; + events: EventIdentityStore; +} + +/** + * Derives a deterministic event position from a Soroban event. + * + * Soroban event IDs encode the ledger sequence and the event's position within + * that ledger, separated by a dash (e.g. `0000000523986165760-0000000001`). We + * use the trailing segment as the event index; if the ID is not in that form we + * fall back to `0`, which is still deterministic for a single event per tx. + */ +export function deriveEventIndex(event: SorobanEventInput): number { + const segments = event.id.split("-"); + const last = segments[segments.length - 1]; + const parsed = Number.parseInt(last ?? "", 10); + return Number.isFinite(parsed) ? parsed : 0; +} diff --git a/indexer/distributions/src/handlers/tokens-claimed.handler.ts b/indexer/distributions/src/handlers/tokens-claimed.handler.ts index 10b5302..df17f92 100644 --- a/indexer/distributions/src/handlers/tokens-claimed.handler.ts +++ b/indexer/distributions/src/handlers/tokens-claimed.handler.ts @@ -1,60 +1,86 @@ -import type { - EventHandler, - HandlerResult, - SorobanEventInput, -} from "@fundable-indexer/common"; +import type { EventHandler, HandlerResult, SorobanEventInput } from "@fundable-indexer/common"; +import { type DistributionHandlerDeps, deriveEventIndex } from "./persistence.js"; import { parseTokensClaimed } from "./types.js"; -export const tokensClaimedHandler: EventHandler = async ( - event: SorobanEventInput, -): Promise => { - try { - const payload = parseTokensClaimed(event.data); +/** + * Builds a `tokens_claimed` handler that records a claim against a batch. + * + * The event identity gate plus the unique claim identity in the repository keep + * repeated events from creating duplicate claim rows or double-counting the + * batch's claimed total. + */ +export const createTokensClaimedHandler = (deps: DistributionHandlerDeps): EventHandler => { + return async (event: SorobanEventInput): Promise => { + try { + const payload = parseTokensClaimed(event.data); - if (!payload.distributionId) { - return { - ok: false, - error: "Missing distributionId in tokens_claimed event", - retriable: false, - }; - } + if (!payload.distributionId) { + return { + ok: false, + error: "Missing distributionId in tokens_claimed event", + retriable: false, + }; + } - if (!payload.claimant) { - return { - ok: false, - error: "Missing claimant in tokens_claimed event", - retriable: false, - }; - } + if (!payload.claimant) { + return { + ok: false, + error: "Missing claimant in tokens_claimed event", + retriable: false, + }; + } - if (!payload.transactionHash) { - return { - ok: false, - error: "Missing transactionHash in tokens_claimed event", - retriable: false, - }; - } + if (!payload.transactionHash) { + return { + ok: false, + error: "Missing transactionHash in tokens_claimed event", + retriable: false, + }; + } + + if (!payload.amount || payload.amount === "0") { + return { + ok: false, + error: "Missing or zero amount in tokens_claimed event", + retriable: false, + }; + } - if (!payload.amount || payload.amount === "0") { + const eventIndex = deriveEventIndex(event); + const alreadyProcessed = await deps.events.isEventProcessed( + event.contractId, + event.ledger, + payload.transactionHash, + eventIndex, + ); + if (alreadyProcessed) { + return { ok: true }; + } + + await deps.distributions.recordClaim({ + distributionId: payload.distributionId, + claimant: payload.claimant, + amount: payload.amount, + txHash: payload.transactionHash, + ledgerNumber: event.ledger, + eventIndex, + eventTimestamp: event.ledgerClosedAt, + }); + + await deps.events.recordEventProcessed( + event.contractId, + event.ledger, + payload.transactionHash, + eventIndex, + ); + + return { ok: true }; + } catch (err) { return { ok: false, - error: "Missing or zero amount in tokens_claimed event", - retriable: false, + error: err instanceof Error ? err.message : String(err), + retriable: true, }; } - - // TODO(#36): record claim action via repository once DB schema is merged - // TODO(#27): record indexed event identity via event repository - console.info( - `[tokens-claimed] distributionId=${payload.distributionId} claimant=${payload.claimant} amount=${payload.amount} ledger=${event.ledger}`, - ); - - return { ok: true }; - } catch (err) { - return { - ok: false, - error: err instanceof Error ? err.message : String(err), - retriable: true, - }; - } + }; }; diff --git a/indexer/distributions/src/handlers/types.ts b/indexer/distributions/src/handlers/types.ts index 7aa6f4b..ada4b9f 100644 --- a/indexer/distributions/src/handlers/types.ts +++ b/indexer/distributions/src/handlers/types.ts @@ -39,15 +39,13 @@ function num(v: unknown): number { return Number.isFinite(n) ? n : 0; } -export function parseDistributionCreated( - data: unknown, -): DistributionCreatedPayload { +export function parseDistributionCreated(data: unknown): DistributionCreatedPayload { const d = record(data); return { distributionId: str(d.distributionId ?? d.distribution_id), creator: str(d.creator), token: str(d.token), - totalAmount: str(d.totalAmount ?? d.total_amount ?? "0"), + totalAmount: str(d.totalAmount ?? d.total_amount), recipientCount: num(d.recipientCount ?? d.recipient_count), transactionHash: str(d.transactionHash ?? d.tx_hash), }; @@ -63,9 +61,7 @@ export function parseTokensClaimed(data: unknown): TokensClaimedPayload { }; } -export function parseDistributionPaused( - data: unknown, -): DistributionPausedPayload { +export function parseDistributionPaused(data: unknown): DistributionPausedPayload { const d = record(data); return { distributionId: str(d.distributionId ?? d.distribution_id), @@ -74,9 +70,7 @@ export function parseDistributionPaused( }; } -export function parseDistributionResumed( - data: unknown, -): DistributionResumedPayload { +export function parseDistributionResumed(data: unknown): DistributionResumedPayload { const d = record(data); return { distributionId: str(d.distributionId ?? d.distribution_id), diff --git a/indexer/distributions/src/index.ts b/indexer/distributions/src/index.ts index f231142..4763a6d 100644 --- a/indexer/distributions/src/index.ts +++ b/indexer/distributions/src/index.ts @@ -6,4 +6,13 @@ export const distributionsPackage = { common: commonPackage.name, } as const; +export { ClaimAction } from "./db/entity/ClaimAction.js"; +export { DistributionBatch, DistributionStatus } from "./db/entity/DistributionBatch.js"; +export { + type CreateBatchInput, + type DistributionPersistence, + DistributionRepository, + type RecordClaimInput, + type SetStatusInput, +} from "./db/repository.js"; export * from "./handlers/index.js"; diff --git a/indexer/streams/src/handlers/stream-cancel.handler.ts b/indexer/streams/src/handlers/stream-cancel.handler.ts index 2b3a916..6c01c2b 100644 --- a/indexer/streams/src/handlers/stream-cancel.handler.ts +++ b/indexer/streams/src/handlers/stream-cancel.handler.ts @@ -1,8 +1,4 @@ -import type { - EventHandler, - HandlerResult, - SorobanEventInput, -} from "@fundable-indexer/common"; +import type { EventHandler, HandlerResult, SorobanEventInput } from "@fundable-indexer/common"; import { parseStreamCancel } from "./types.js"; export const streamCancelHandler: EventHandler = async ( diff --git a/indexer/streams/src/handlers/stream-funded.handler.ts b/indexer/streams/src/handlers/stream-funded.handler.ts index c0e951c..893663e 100644 --- a/indexer/streams/src/handlers/stream-funded.handler.ts +++ b/indexer/streams/src/handlers/stream-funded.handler.ts @@ -1,8 +1,4 @@ -import type { - EventHandler, - HandlerResult, - SorobanEventInput, -} from "@fundable-indexer/common"; +import type { EventHandler, HandlerResult, SorobanEventInput } from "@fundable-indexer/common"; import { parseStreamFunded } from "./types.js"; export const streamFundedHandler: EventHandler = async ( diff --git a/indexer/streams/src/handlers/stream-withdrawal.handler.ts b/indexer/streams/src/handlers/stream-withdrawal.handler.ts index 1e03219..d258ff4 100644 --- a/indexer/streams/src/handlers/stream-withdrawal.handler.ts +++ b/indexer/streams/src/handlers/stream-withdrawal.handler.ts @@ -1,8 +1,4 @@ -import type { - EventHandler, - HandlerResult, - SorobanEventInput, -} from "@fundable-indexer/common"; +import type { EventHandler, HandlerResult, SorobanEventInput } from "@fundable-indexer/common"; import { parseStreamWithdrawal } from "./types.js"; export const streamWithdrawalHandler: EventHandler = async ( diff --git a/src/__tests__/distribution.controller.test.ts b/src/__tests__/distribution.controller.test.ts new file mode 100644 index 0000000..4e4f5b0 --- /dev/null +++ b/src/__tests__/distribution.controller.test.ts @@ -0,0 +1,189 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +// The controller imports the TypeORM data source, which validates DB env vars at +// module load (via the eagerly-built app config). Provide dummy values BEFORE any +// module that reads config is imported. The data source is never initialized, so +// isInitialized stays false. Imports that pull in config are done dynamically +// inside the tests so these assignments run first. +process.env.DATABASE_HOST ||= 'localhost'; +process.env.DATABASE_PORT ||= '5432'; +process.env.DATABASE_USERNAME ||= 'test'; +process.env.DATABASE_PASSWORD ||= 'test'; +process.env.DATABASE_NAME ||= 'test'; + +type MockRes = { + statusCode: number; + body: any; + status: (code: number) => MockRes; + json: (payload: any) => MockRes; +}; + +const makeRes = (): MockRes => { + const res: MockRes = { + statusCode: 0, + body: undefined, + status(code: number) { + res.statusCode = code; + return res; + }, + json(payload: any) { + res.body = payload; + return res; + }, + }; + return res; +}; + +type Controller = + typeof import('../components/v1/distribution/distribution.controller'); + +let cached: Controller | null = null; +const loadController = async (): Promise => { + if (!cached) { + cached = await import( + '../components/v1/distribution/distribution.controller' + ); + } + return cached; +}; + +const sampleDistribution = { + id: '550e8400-e29b-41d4-a716-446655440000', + userAddress: '0xuser', + transactionHash: null, + tokenAddress: '0xtoken', + tokenSymbol: 'USDC', + tokenDecimals: 6, + totalAmount: '1000', + feeAmount: '10', + usdRate: '0', + totalUsdAmount: '0', + totalRecipients: 5, + distributionType: 'airdrop', + chainName: '', + status: 'pending', + blockNumber: null, + blockTimestamp: null, + network: 'mainnet', + createdAt: new Date(), + metadata: null, +}; + +test('createDistribution returns 201 with shared success shape', async () => { + const { createDistribution } = await loadController(); + const service = { + createDistribution: async () => sampleDistribution, + }; + const req: any = { body: {} }; + const res = makeRes(); + + await createDistribution(() => service as any)(req, res as any); + + assert.equal(res.statusCode, 201); + assert.equal(res.body.success, true); + assert.equal(res.body.data.id, sampleDistribution.id); +}); + +test('createDistribution returns 503 DB_NOT_READY when data source is not initialized', async () => { + const { createDistribution } = await loadController(); + const req: any = { body: {} }; + const res = makeRes(); + + // Use the default resolver, which checks the (uninitialized) data source. + await createDistribution()(req, res as any); + + assert.equal(res.statusCode, 503); + assert.equal(res.body.success, false); + assert.equal(res.body.error.code, 'DB_NOT_READY'); +}); + +test('createDistribution returns 500 INTERNAL_ERROR for unexpected failures', async () => { + const { createDistribution } = await loadController(); + const service = { + createDistribution: async () => { + throw new Error('boom'); + }, + }; + const req: any = { body: {} }; + const res = makeRes(); + + await createDistribution(() => service as any)(req, res as any); + + assert.equal(res.statusCode, 500); + assert.equal(res.body.error.code, 'INTERNAL_ERROR'); +}); + +test('updateDistribution returns 404 when the distribution is missing', async () => { + const { updateDistribution } = await loadController(); + const service = { + updateDistribution: async () => { + throw Object.assign(new Error('Distribution not found'), { + code: 'DISTRIBUTION_NOT_FOUND', + }); + }, + }; + const req: any = { params: { id: 'missing' }, body: {} }; + const res = makeRes(); + + await updateDistribution(() => service as any)(req, res as any); + + assert.equal(res.statusCode, 404); + assert.equal(res.body.success, false); + assert.equal(res.body.error.code, 'DISTRIBUTION_NOT_FOUND'); +}); + +test('updateDistribution returns 200 on success', async () => { + const { updateDistribution } = await loadController(); + const service = { + updateDistribution: async () => ({ + ...sampleDistribution, + status: 'completed', + }), + }; + const req: any = { params: { id: sampleDistribution.id }, body: {} }; + const res = makeRes(); + + await updateDistribution(() => service as any)(req, res as any); + + assert.equal(res.statusCode, 200); + assert.equal(res.body.success, true); + assert.equal(res.body.data.status, 'completed'); +}); + +test('listDistributions returns 200 with an array payload', async () => { + const { listDistributions } = await loadController(); + const service = { + listDistributions: async () => [sampleDistribution], + }; + const req: any = {}; + const res = makeRes(); + + await listDistributions(() => service as any)(req, res as any); + + assert.equal(res.statusCode, 200); + assert.equal(res.body.success, true); + assert.equal(res.body.data.length, 1); +}); + +test('policyMiddleware forwards a 400 error for an invalid create body', async () => { + const { default: policyMiddleware } = await import( + '../appMiddlewares/policy.middleware' + ); + const { createDistributionSchema } = await import( + '../components/v1/distribution/distribution.validation' + ); + + let captured: any; + const req: any = { body: { userAddress: 'invalid' } }; + const res: any = {}; + const next = (err: any) => { + captured = err; + }; + + policyMiddleware(createDistributionSchema)(req, res, next); + + assert.ok(captured); + assert.equal(captured.httpCode, 400); + assert.equal(captured.type, 'API'); +}); diff --git a/src/__tests__/distribution.service.test.ts b/src/__tests__/distribution.service.test.ts new file mode 100644 index 0000000..ee0b392 --- /dev/null +++ b/src/__tests__/distribution.service.test.ts @@ -0,0 +1,172 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { + DistributionNotFoundError, + DistributionService, +} from '../components/v1/distribution/distribution.service'; +import type DistributionEntity from '../components/v1/distribution/distribution.entity'; +import { DistributionStatus, DistributionType, Network } from '../types/enums'; + +type Repo = { + data: T[]; + findOne: (_arg: any) => Promise; + find: (_arg?: any) => Promise; + create: (_partial: Partial) => T; + save: (_entity: any) => Promise; +}; + +const makeRepo = >(): Repo => { + const repo: Repo = { + data: [], + async findOne(arg: any) { + const where = arg?.where ?? {}; + return ( + repo.data.find((d) => + Object.keys(where).every((k) => (d as any)[k] === where[k]) + ) ?? null + ); + }, + async find(arg?: any) { + const take = arg?.take ?? repo.data.length; + return repo.data.slice(0, take); + }, + create(partial: Partial) { + return { ...partial } as T; + }, + async save(entity: any) { + const e = { ...entity } as any; + if (!e.id) e.id = `dist_${repo.data.length + 1}`; + if (!e.createdAt) e.createdAt = new Date(); + const idx = repo.data.findIndex((d) => (d as any).id === e.id); + if (idx >= 0) repo.data[idx] = e; + else repo.data.push(e); + return e; + }, + }; + return repo; +}; + +const validUser = '0xABCDEF1234567890ABCDEF1234567890ABCDEF12'; +const validToken = '0x1234567890ABCDEF1234567890ABCDEF12345678'; + +const baseCreate = { + userAddress: validUser, + tokenAddress: validToken, + tokenSymbol: 'usdc', + tokenDecimals: 6, + totalAmount: '1000', + feeAmount: '10', + totalRecipients: 5, + distributionType: DistributionType.AIRDROP, +}; + +test('DistributionService.createDistribution normalizes addresses and symbol', async () => { + const repo = makeRepo(); + const service = new DistributionService(repo as any); + + const result = await service.createDistribution(baseCreate as any); + + assert.equal(result.userAddress, validUser.toLowerCase()); + assert.equal(result.tokenAddress, validToken.toLowerCase()); + assert.equal(result.tokenSymbol, 'USDC'); + assert.equal(result.status, DistributionStatus.PENDING); + assert.equal(result.network, Network.MAINNET); + assert.equal(repo.data.length, 1); +}); + +test('DistributionService.createDistribution computes totalUsdAmount from usdRate', async () => { + const repo = makeRepo(); + const service = new DistributionService(repo as any); + + const result = await service.createDistribution({ + ...baseCreate, + usdRate: '2', + } as any); + + assert.equal(result.usdRate, '2'); + assert.equal(result.totalUsdAmount, '2000'); +}); + +test('DistributionService.createDistribution defaults usd values to zero', async () => { + const repo = makeRepo(); + const service = new DistributionService(repo as any); + + const result = await service.createDistribution(baseCreate as any); + + assert.equal(result.usdRate, '0'); + assert.equal(result.totalUsdAmount, '0'); +}); + +test('DistributionService.createDistribution wraps repository failures', async () => { + const repo = makeRepo(); + repo.save = async () => { + throw new Error('db down'); + }; + const service = new DistributionService(repo as any); + + await assert.rejects( + () => service.createDistribution(baseCreate as any), + /Failed to create distribution/ + ); +}); + +test('DistributionService.updateDistribution throws DistributionNotFoundError for missing id', async () => { + const repo = makeRepo(); + const service = new DistributionService(repo as any); + + let err: any; + try { + await service.updateDistribution('missing', { tokenSymbol: 'dai' } as any); + } catch (e) { + err = e; + } + + assert.ok(err instanceof DistributionNotFoundError); + assert.equal(err.code, 'DISTRIBUTION_NOT_FOUND'); +}); + +test('DistributionService.updateDistribution normalizes and recomputes usd amount', async () => { + const repo = makeRepo(); + const service = new DistributionService(repo as any); + + const created = await service.createDistribution(baseCreate as any); + + const updated = await service.updateDistribution(created.id, { + tokenSymbol: 'dai', + userAddress: validUser, + totalAmount: '500', + usdRate: '3', + status: DistributionStatus.COMPLETED, + } as any); + + assert.equal(updated.tokenSymbol, 'DAI'); + assert.equal(updated.userAddress, validUser.toLowerCase()); + assert.equal(updated.totalUsdAmount, '1500'); + assert.equal(updated.status, DistributionStatus.COMPLETED); +}); + +test('DistributionService.listDistributions returns formatted responses', async () => { + const repo = makeRepo(); + const service = new DistributionService(repo as any); + + await service.createDistribution(baseCreate as any); + await service.createDistribution(baseCreate as any); + + const result = await service.listDistributions(); + assert.equal(result.length, 2); + assert.equal(result[0].tokenSymbol, 'USDC'); +}); + +test('DistributionService.listDistributions wraps repository failures', async () => { + const repo = makeRepo(); + repo.find = async () => { + throw new Error('db down'); + }; + const service = new DistributionService(repo as any); + + await assert.rejects( + () => service.listDistributions(), + /Failed to list distributions/ + ); +}); diff --git a/src/__tests__/distribution.validation.test.ts b/src/__tests__/distribution.validation.test.ts new file mode 100644 index 0000000..62212e8 --- /dev/null +++ b/src/__tests__/distribution.validation.test.ts @@ -0,0 +1,119 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; + +import { + createDistributionSchema, + updateDistributionSchema, + updateDistributionParamsSchema, +} from '../components/v1/distribution/distribution.validation'; +import { DistributionType, DistributionStatus, Network } from '../types/enums'; + +const validAddress = '0x1234567890abcdef1234567890abcdef12345678'; +const validTokenAddress = '0xabcdef1234567890abcdef1234567890abcdef12'; + +const baseCreate = { + userAddress: validAddress, + tokenAddress: validTokenAddress, + tokenSymbol: 'USDC', + tokenDecimals: 6, + totalAmount: '1000', + feeAmount: '10', + totalRecipients: 5, + distributionType: DistributionType.AIRDROP, +}; + +test('createDistributionSchema accepts a valid payload', () => { + const result = createDistributionSchema.parse(baseCreate); + assert.equal(result.userAddress, validAddress); + assert.equal(result.distributionType, DistributionType.AIRDROP); +}); + +test('createDistributionSchema accepts optional fields', () => { + const result = createDistributionSchema.parse({ + ...baseCreate, + usdRate: '1.5', + chainName: 'ethereum', + network: Network.TESTNET, + metadata: { note: 'q3 rewards' }, + }); + assert.equal(result.usdRate, '1.5'); + assert.equal(result.network, Network.TESTNET); + assert.deepEqual(result.metadata, { note: 'q3 rewards' }); +}); + +test('createDistributionSchema rejects invalid Ethereum addresses', () => { + assert.throws( + () => createDistributionSchema.parse({ ...baseCreate, userAddress: 'invalid' }), + /userAddress/ + ); + assert.throws( + () => createDistributionSchema.parse({ ...baseCreate, tokenAddress: 'invalid' }), + /tokenAddress/ + ); +}); + +test('createDistributionSchema rejects invalid decimal strings', () => { + assert.throws( + () => createDistributionSchema.parse({ ...baseCreate, totalAmount: 'abc' }), + /totalAmount/ + ); + assert.throws( + () => createDistributionSchema.parse({ ...baseCreate, feeAmount: 'xyz' }), + /feeAmount/ + ); +}); + +test('createDistributionSchema rejects out-of-range tokenDecimals', () => { + assert.throws( + () => createDistributionSchema.parse({ ...baseCreate, tokenDecimals: -1 }), + /tokenDecimals/ + ); + assert.throws( + () => createDistributionSchema.parse({ ...baseCreate, tokenDecimals: 31 }), + /tokenDecimals/ + ); +}); + +test('createDistributionSchema rejects non-positive totalRecipients', () => { + assert.throws( + () => createDistributionSchema.parse({ ...baseCreate, totalRecipients: 0 }), + /totalRecipients/ + ); +}); + +test('createDistributionSchema rejects an invalid distributionType', () => { + assert.throws( + () => + createDistributionSchema.parse({ + ...baseCreate, + distributionType: 'not_a_type', + }), + /distributionType/ + ); +}); + +test('updateDistributionSchema accepts a partial payload with status', () => { + const result = updateDistributionSchema.parse({ + status: DistributionStatus.COMPLETED, + }); + assert.equal(result.status, DistributionStatus.COMPLETED); +}); + +test('updateDistributionSchema rejects an invalid status', () => { + assert.throws( + () => updateDistributionSchema.parse({ status: 'bogus' }), + /status/ + ); +}); + +test('updateDistributionParamsSchema validates UUID format', () => { + assert.throws( + () => updateDistributionParamsSchema.parse({ id: 'not-a-uuid' }), + /uuid/ + ); + assert.doesNotThrow(() => + updateDistributionParamsSchema.parse({ + id: '550e8400-e29b-41d4-a716-446655440000', + }) + ); +}); diff --git a/src/components/v1/distribution/distribution.controller.ts b/src/components/v1/distribution/distribution.controller.ts index f365bdf..80f7823 100644 --- a/src/components/v1/distribution/distribution.controller.ts +++ b/src/components/v1/distribution/distribution.controller.ts @@ -1,96 +1,95 @@ import type { Request, Response } from "express" import AppDataSource from "../../../config/persistence/data-source" +import { sendError, sendSuccess } from "../../../utils/apiResponse" +import logger from "../../../utils/logger" import { DistributionEntity } from "./distribution.entity" -import { DistributionService } from "./distribution.service" -import type { ApiResponse, DistributionResponseDto, CreateDistributionDto, UpdateDistributionDto } from "./distribution.dto" - -const getDistributionService = () => { - if (!AppDataSource.isInitialized) { - throw new Error("Database not initialized") +import { DistributionNotFoundError, DistributionService } from "./distribution.service" +import type { CreateDistributionDto, UpdateDistributionDto } from "./distribution.dto" + +/** + * Raised when the data source has not finished initializing. Carries a stable + * `code` so it maps to a predictable 503 payload instead of a generic 500. + */ +export class DatabaseNotReadyError extends Error { + public readonly code = "DB_NOT_READY" + constructor() { + super("Database not initialized") + this.name = "DatabaseNotReadyError" } - - const distributionRepository = AppDataSource.getRepository(DistributionEntity) - return new DistributionService(distributionRepository) } -export const createDistribution = async (req: Request, res: Response): Promise => { - try { - const distributionService = getDistributionService() - - const validatedData = req.body as CreateDistributionDto - - const distribution = await distributionService.createDistribution(validatedData) - - const response: ApiResponse = { - data: distribution, - success: true, - message: "Distribution created successfully", - } - - res.status(201).json(response) - } catch (error) { - console.error("Error in createDistribution:", error) - - const errorResponse: ApiResponse = { - data: null, - success: false, - message: error instanceof Error ? error.message : "Internal server error", - } +export type DistributionServiceResolver = () => DistributionService - res.status(500).json(errorResponse) +/** + * Default resolver used by the routes. Tests inject their own resolver so the + * controller logic can be exercised without a live database connection. + */ +export const defaultDistributionServiceResolver: DistributionServiceResolver = () => { + if (!AppDataSource.isInitialized) { + throw new DatabaseNotReadyError() } -} -export const updateDistribution = async (req: Request, res: Response): Promise => { - try { - const distributionService = getDistributionService() - const { id } = req.params - const validatedData = req.body as UpdateDistributionDto + return new DistributionService(AppDataSource.getRepository(DistributionEntity)) +} - const distribution = await distributionService.updateDistribution(id, validatedData) +const handleControllerError = (res: Response, error: unknown, context: string): void => { + if (error instanceof DatabaseNotReadyError) { + sendError(res, 503, { + code: "DB_NOT_READY", + message: "Database not initialized", + }) + return + } - const response: ApiResponse = { - data: distribution, - success: true, - message: "Distribution updated successfully", - } + const code = (error as { code?: unknown })?.code + if (error instanceof DistributionNotFoundError || code === "DISTRIBUTION_NOT_FOUND") { + sendError(res, 404, { + code: "DISTRIBUTION_NOT_FOUND", + message: error instanceof Error ? error.message : "Distribution not found", + }) + return + } - res.status(200).json(response) - } catch (error) { - console.error("Error in updateDistribution:", error) + logger.error(`${context}: ${error instanceof Error ? error.message : String(error)}`) + sendError(res, 500, { + code: "INTERNAL_ERROR", + message: "Internal server error", + }) +} - const isNotFound = error instanceof Error && error.message === "Distribution not found" - const status = isNotFound ? 404 : 500 - const errorResponse: ApiResponse = { - data: null, - success: false, - message: isNotFound && error instanceof Error ? error.message : "Internal server error", +export const createDistribution = + (resolve: DistributionServiceResolver = defaultDistributionServiceResolver) => + async (req: Request, res: Response): Promise => { + try { + const service = resolve() + const distribution = await service.createDistribution(req.body as CreateDistributionDto) + sendSuccess(res, distribution, 201) + } catch (error) { + handleControllerError(res, error, "Error in createDistribution") } - res.status(status).json(errorResponse) } -} -export const listDistributions = async (_req: Request, res: Response): Promise => { - try { - const distributionService = getDistributionService() - const distributions = await distributionService.listDistributions() - - const response: ApiResponse = { - data: distributions, - success: true, - message: "Distributions fetched successfully", +export const updateDistribution = + (resolve: DistributionServiceResolver = defaultDistributionServiceResolver) => + async (req: Request, res: Response): Promise => { + try { + const service = resolve() + const { id } = req.params + const distribution = await service.updateDistribution(id, req.body as UpdateDistributionDto) + sendSuccess(res, distribution) + } catch (error) { + handleControllerError(res, error, "Error in updateDistribution") } + } - res.status(200).json(response) - } catch (error) { - console.error("Error in listDistributions:", error) - - const errorResponse: ApiResponse = { - data: null, - success: false, - message: error instanceof Error ? error.message : "Internal server error", +export const listDistributions = + (resolve: DistributionServiceResolver = defaultDistributionServiceResolver) => + async (_req: Request, res: Response): Promise => { + try { + const service = resolve() + const distributions = await service.listDistributions() + sendSuccess(res, distributions) + } catch (error) { + handleControllerError(res, error, "Error in listDistributions") } - - res.status(500).json(errorResponse) } -} diff --git a/src/components/v1/distribution/distribution.service.ts b/src/components/v1/distribution/distribution.service.ts index c4f4a06..169dad8 100644 --- a/src/components/v1/distribution/distribution.service.ts +++ b/src/components/v1/distribution/distribution.service.ts @@ -3,6 +3,19 @@ import type { Repository } from "typeorm" import type { DistributionEntity } from "./distribution.entity" import type { CreateDistributionDto, DistributionResponseDto, UpdateDistributionDto } from "./distribution.dto" import { DistributionStatus, Network } from "../../../types/enums" +import logger from "../../../utils/logger" + +/** + * Raised when an update targets a distribution that does not exist. Carries a + * stable `code` so controllers can map it to a 404 without matching on message text. + */ +export class DistributionNotFoundError extends Error { + public readonly code = "DISTRIBUTION_NOT_FOUND" + constructor(message = "Distribution not found") { + super(message) + this.name = "DistributionNotFoundError" + } +} export class DistributionService { constructor(private readonly distributionRepository: Repository) {} @@ -16,7 +29,7 @@ export class DistributionService { return this.formatDistributionResponse(savedDistribution) } catch (error) { - console.error("Error creating distribution:", error) + logger.error(`Error creating distribution: ${error instanceof Error ? error.message : String(error)}`) throw new Error("Failed to create distribution") } } @@ -25,7 +38,7 @@ export class DistributionService { try { const distribution = await this.distributionRepository.findOne({ where: { id } }) if (!distribution) { - throw new Error("Distribution not found") + throw new DistributionNotFoundError() } const updatedFields: Partial = { ...updateData } @@ -57,7 +70,10 @@ export class DistributionService { return this.formatDistributionResponse(savedDistribution) } catch (error) { - console.error("Error updating distribution:", error) + if (error instanceof DistributionNotFoundError) { + throw error + } + logger.error(`Error updating distribution: ${error instanceof Error ? error.message : String(error)}`) throw error instanceof Error ? error : new Error("Failed to update distribution") } } @@ -71,7 +87,7 @@ export class DistributionService { return distributions.map((d) => this.formatDistributionResponse(d)) } catch (error) { - console.error("Error listing distributions:", error) + logger.error(`Error listing distributions: ${error instanceof Error ? error.message : String(error)}`) throw new Error("Failed to list distributions") } } @@ -115,7 +131,7 @@ export class DistributionService { const rate = new Decimal(usdRate) return amount.mul(rate).toString() } catch (error) { - console.warn("Error calculating total USD amount:", error) + logger.warn(`Error calculating total USD amount: ${error instanceof Error ? error.message : String(error)}`) return "0" } } diff --git a/src/components/v1/distribution/distrubtion.routes.ts b/src/components/v1/distribution/distrubtion.routes.ts index f9f8ed3..b78ee68 100644 --- a/src/components/v1/distribution/distrubtion.routes.ts +++ b/src/components/v1/distribution/distrubtion.routes.ts @@ -13,13 +13,13 @@ import { const distributionRouter = new EnhancedRouter() -distributionRouter.get("/", listDistributions) -distributionRouter.post("/", policyMiddleware(createDistributionSchema), createDistribution) +distributionRouter.get("/", listDistributions()) +distributionRouter.post("/", policyMiddleware(createDistributionSchema), createDistribution()) distributionRouter.patch( "/:id", policyMiddleware(updateDistributionParamsSchema, "params"), policyMiddleware(updateDistributionSchema), - updateDistribution, + updateDistribution(), ) export default distributionRouter.getRouter()