Wire distribution event handlers to persistence and add API tests

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", () => {
     });
   });
 });
-

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,
-        }))
-      )
+        })),
+      ),
     );
   }
 }

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<HandlerResult>;
 

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"
   }
 }

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;
+}

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;
+}

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<void> {
+    // 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<void> {
+    await queryRunner.query(`DROP TABLE "distribution_claim_action"`);
+    await queryRunner.query(`DROP TABLE "distribution_batch"`);
+    await queryRunner.query(`DROP TYPE "distribution_batch_status_enum"`);
+  }
+}