feat: implement indexer foundation

indexer/common/package.json

@@ -13,5 +13,14 @@
     "lint": "biome check .",
     "test": "vitest run src",
     "type-check": "tsc -p tsconfig.json --noEmit"
+  },
+  "dependencies": {
+    "@stellar/stellar-sdk": "^12.0.0",
+    "pg": "^8.11.3",
+    "typeorm": "^0.3.20",
+    "zod": "^3.22.4"
+  },
+  "devDependencies": {
+    "@types/pg": "^8.11.0"
   }
 }

indexer/common/src/config/index.ts

@@ -0,0 +1,18 @@
+import { z } from "zod";
+
+export const ConfigSchema = z.object({
+  RPC_URL: z.string().url().default("https://soroban-testnet.stellar.org"),
+  NETWORK_PASSPHRASE: z.string().default("Test SDF Network ; September 2015"),
+});
+
+export type Config = z.infer<typeof ConfigSchema>;
+
+export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config {
+  const result = ConfigSchema.safeParse(env);
+  if (!result.success) {
+    throw new Error(`Invalid configuration: ${result.error.message}`);
+  }
+  return result.data;
+}
+
+export const config = loadConfig();

indexer/common/src/db/entity/IndexedEvent.ts

@@ -0,0 +1,23 @@
+import { Column, CreateDateColumn, Entity, PrimaryGeneratedColumn, Unique } from "typeorm";
+
+@Entity("indexed_event")
+@Unique("uq_indexed_event_identity", ["contractId", "ledgerNumber", "txHash", "eventIndex"])
+export class IndexedEvent {
+  @PrimaryGeneratedColumn("uuid")
+  id!: string;
+
+  @Column({ type: "varchar" })
+  contractId!: string;
+
+  @Column({ type: "int" })
+  ledgerNumber!: number;
+
+  @Column({ type: "varchar" })
+  txHash!: string;
+
+  @Column({ type: "int" })
+  eventIndex!: number;
+
+  @CreateDateColumn()
+  processedAt!: Date;
+}

indexer/common/src/db/repository.test.ts

@@ -0,0 +1,72 @@
+import { type DataSource, InsertQueryBuilder, Repository } from "typeorm";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { IndexedEvent } from "./entity/IndexedEvent.js";
+import { EventRepository } from "./repository.js";
+
+describe("EventRepository", () => {
+  // biome-ignore lint/suspicious/noExplicitAny: mock objects
+  let mockDataSource: any;
+  // biome-ignore lint/suspicious/noExplicitAny: mock objects
+  let mockRepo: any;
+  // biome-ignore lint/suspicious/noExplicitAny: mock objects
+  let mockQueryBuilder: any;
+
+  beforeEach(() => {
+    mockQueryBuilder = {
+      insert: vi.fn().mockReturnThis(),
+      into: vi.fn().mockReturnThis(),
+      values: vi.fn().mockReturnThis(),
+      orIgnore: vi.fn().mockReturnThis(),
+      execute: vi.fn(),
+    };
+
+    mockRepo = {
+      count: vi.fn(),
+      createQueryBuilder: vi.fn(() => mockQueryBuilder),
+    };
+
+    mockDataSource = {
+      getRepository: vi.fn(() => mockRepo),
+    };
+  });
+
+  it("should return true if event is processed", async () => {
+    mockRepo.count.mockResolvedValue(1);
+    const repo = new EventRepository(mockDataSource as unknown as DataSource);
+    const isProcessed = await repo.isEventProcessed("contract1", 100, "txHash1", 0);
+
+    expect(isProcessed).toBe(true);
+    expect(mockRepo.count).toHaveBeenCalledWith({
+      where: {
+        contractId: "contract1",
+        ledgerNumber: 100,
+        txHash: "txHash1",
+        eventIndex: 0,
+      },
+    });
+  });
+
+  it("should record event successfully", async () => {
+    const repo = new EventRepository(mockDataSource as unknown as DataSource);
+    mockQueryBuilder.execute.mockResolvedValue({});
+
+    const result = await repo.recordEventProcessed("contract1", 100, "txHash1", 0);
+
+    expect(result).toBe(true);
+    expect(mockQueryBuilder.values).toHaveBeenCalledWith({
+      contractId: "contract1",
+      ledgerNumber: 100,
+      txHash: "txHash1",
+      eventIndex: 0,
+    });
+    expect(mockQueryBuilder.orIgnore).toHaveBeenCalledWith("uq_indexed_event_identity");
+  });
+
+  it("should catch unique constraint errors and return false", async () => {
+    const repo = new EventRepository(mockDataSource as unknown as DataSource);
+    mockQueryBuilder.execute.mockRejectedValue(new Error("unique constraint violation"));
+
+    const result = await repo.recordEventProcessed("contract1", 100, "txHash1", 0);
+    expect(result).toBe(false);
+  });
+});

indexer/common/src/db/repository.ts

@@ -0,0 +1,62 @@
+import type { DataSource, Repository } from "typeorm";
+import { IndexedEvent } from "./entity/IndexedEvent.js";
+
+export class EventRepository {
+  private repo: Repository<IndexedEvent>;
+
+  constructor(private dataSource: DataSource) {
+    this.repo = this.dataSource.getRepository(IndexedEvent);
+  }
+
+  /**
+   * Checks if an event has already been processed based on its deterministic identity.
+   */
+  async isEventProcessed(
+    contractId: string,
+    ledgerNumber: number,
+    txHash: string,
+    eventIndex: number,
+  ): Promise<boolean> {
+    const count = await this.repo.count({
+      where: {
+        contractId,
+        ledgerNumber,
+        txHash,
+        eventIndex,
+      },
+    });
+    return count > 0;
+  }
+
+  /**
+   * Records an event as processed safely handling duplicates without throwing.
+   */
+  async recordEventProcessed(
+    contractId: string,
+    ledgerNumber: number,
+    txHash: string,
+    eventIndex: number,
+  ): Promise<boolean> {
+    try {
+      await this.repo
+        .createQueryBuilder()
+        .insert()
+        .into(IndexedEvent)
+        .values({
+          contractId,
+          ledgerNumber,
+          txHash,
+          eventIndex,
+        })
+        .orIgnore("uq_indexed_event_identity") // Safe duplicate insert
+        .execute();
+      return true;
+    } catch (err) {
+      // If the driver doesn't support orIgnore, fallback to simple insert and catch unique violation.
+      if (err instanceof Error && err.message.includes("unique constraint")) {
+        return false;
+      }
+      throw err;
+    }
+  }
+}

indexer/common/src/index.ts

@@ -2,3 +2,9 @@ export const commonPackage = {
   name: "@fundable-indexer/common",
   role: "shared-infrastructure",
 } as const;
+
+export { config, loadConfig } from "./config/index.js";
+export { createSorobanClient, sorobanClient } from "./rpc/client.js";
+export { IndexedEvent } from "./db/entity/IndexedEvent.js";
+export { EventRepository } from "./db/repository.js";
+export { SorobanPoller, type PollerOptions, type PollResult } from "./poller/index.js";

indexer/common/src/poller/index.test.ts

@@ -0,0 +1,52 @@
+import { describe, expect, it, vi } from "vitest";
+import { SorobanPoller } from "./index.js";
+
+describe("SorobanPoller", () => {
+  it("should successfully process events and advance cursor", async () => {
+    const poller = new SorobanPoller({ retryDelayMs: 1 });
+    const fetchEvents = vi.fn().mockResolvedValue([{ id: 1 }]);
+    const processEvent = vi.fn().mockResolvedValue(undefined);
+    const updateCursor = vi.fn().mockResolvedValue(undefined);
+
+    const result = await poller.processLedgerRange(10, 20, fetchEvents, processEvent, updateCursor);
+
+    expect(result.success).toBe(true);
+    expect(result.lastProcessedLedger).toBe(20);
+    expect(fetchEvents).toHaveBeenCalledWith(10, 20);
+    expect(processEvent).toHaveBeenCalledWith({ id: 1 });
+    expect(updateCursor).toHaveBeenCalledWith(20);
+  });
+
+  it("should retry on transient errors when fetching events", async () => {
+    const poller = new SorobanPoller({ retryDelayMs: 1, maxRetries: 2 });
+    let attempts = 0;
+    const fetchEvents = vi.fn().mockImplementation(() => {
+      attempts++;
+      if (attempts === 1) return Promise.reject(new Error("network timeout"));
+      return Promise.resolve([{ id: 2 }]);
+    });
+    const processEvent = vi.fn().mockResolvedValue(undefined);
+    const updateCursor = vi.fn().mockResolvedValue(undefined);
+
+    const result = await poller.processLedgerRange(10, 20, fetchEvents, processEvent, updateCursor);
+
+    expect(result.success).toBe(true);
+    expect(attempts).toBe(2);
+    expect(fetchEvents).toHaveBeenCalledTimes(2);
+    expect(processEvent).toHaveBeenCalledWith({ id: 2 });
+    expect(updateCursor).toHaveBeenCalledWith(20);
+  });
+
+  it("should not advance cursor if handler fails", async () => {
+    const poller = new SorobanPoller({ retryDelayMs: 1 });
+    const fetchEvents = vi.fn().mockResolvedValue([{ id: 3 }]);
+    const processEvent = vi.fn().mockRejectedValue(new Error("Handler failure"));
+    const updateCursor = vi.fn();
+
+    const result = await poller.processLedgerRange(10, 20, fetchEvents, processEvent, updateCursor);
+
+    expect(result.success).toBe(false);
+    expect(result.error?.message).toBe("Handler failure");
+    expect(updateCursor).not.toHaveBeenCalled();
+  });
+});

indexer/common/src/poller/index.ts

@@ -0,0 +1,92 @@
+export interface PollerOptions {
+  maxRetries?: number;
+  retryDelayMs?: number;
+}
+
+export interface PollResult {
+  success: boolean;
+  error?: Error;
+  lastProcessedLedger?: number;
+}
+
+export class SorobanPoller {
+  private maxRetries: number;
+  private retryDelayMs: number;
+  private isRunning = false;
+
+  constructor(options: PollerOptions = {}) {
+    this.maxRetries = options.maxRetries ?? 3;
+    this.retryDelayMs = options.retryDelayMs ?? 1000;
+  }
+
+  /**
+   * Utility to wait for a given amount of time.
+   */
+  private delay(ms: number): Promise<void> {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+  }
+
+  /**
+   * Executes a block of logic with retry for transient RPC errors.
+   */
+  public async withRetry<T>(operation: () => Promise<T>): Promise<T> {
+    let attempt = 0;
+    while (true) {
+      try {
+        return await operation();
+      } catch (error) {
+        attempt++;
+        if (attempt > this.maxRetries || !this.isTransientError(error)) {
+          throw error;
+        }
+        await this.delay(this.retryDelayMs);
+      }
+    }
+  }
+
+  /**
+   * Determines if an error is transient and should be retried.
+   */
+  private isTransientError(error: unknown): boolean {
+    // Basic transient error checking - expand based on specific RPC error formats
+    const message = error instanceof Error ? error.message : String(error);
+    return (
+      message.includes("timeout") ||
+      message.includes("rate limit") ||
+      message.includes("ECONNRESET") ||
+      message.includes("503") ||
+      message.includes("504") ||
+      message.includes("502")
+    );
+  }
+
+  /**
+   * Processes a ledger range. Returns success=true if processing completes,
+   * otherwise returns an error and the operation stops without advancing cursor.
+   */
+  public async processLedgerRange<TEvent = unknown>(
+    startLedger: number,
+    endLedger: number,
+    fetchEvents: (start: number, end: number) => Promise<TEvent[]>,
+    processEvent: (event: TEvent) => Promise<void>,
+    updateCursor: (ledger: number) => Promise<void>,
+  ): Promise<PollResult> {
+    try {
+      // 1. Fetch events with retry logic for RPC
+      const events = await this.withRetry(() => fetchEvents(startLedger, endLedger));
+
+      // 2. Process events sequentially
+      for (const event of events) {
+        // If a handler fails, it throws, skipping the updateCursor step
+        await processEvent(event);
+      }
+
+      // 3. Update cursor ONLY if all events in the range succeeded
+      await updateCursor(endLedger);
+      return { success: true, lastProcessedLedger: endLedger };
+    } catch (error) {
+      // Return the error to surface it. Cursor is intentionally not advanced.
+      return { success: false, error: error instanceof Error ? error : new Error(String(error)) };
+    }
+  }
+}

indexer/common/src/rpc/client.test.ts

@@ -0,0 +1,15 @@
+import { describe, expect, it } from "vitest";
+import { createSorobanClient } from "./client.js";
+
+describe("RPC Client Wrapper", () => {
+  it("should create a client with the provided URL", () => {
+    const url = "https://soroban-testnet.stellar.org";
+    const client = createSorobanClient(url);
+    expect(client.serverURL.toString()).toBe("https://soroban-testnet.stellar.org/");
+  });
+
+  it("should use the default configuration if no URL is provided", () => {
+    const client = createSorobanClient();
+    expect(client.serverURL.toString()).toBe("https://soroban-testnet.stellar.org/");
+  });
+});

indexer/common/src/rpc/client.ts

@@ -0,0 +1,12 @@
+import { rpc } from "@stellar/stellar-sdk";
+import { config } from "../config/index.js";
+
+/**
+ * Creates and configures a Soroban Server instance.
+ * Reads the RPC URL from the validated configuration.
+ */
+export function createSorobanClient(rpcUrl: string = config.RPC_URL): rpc.Server {
+  return new rpc.Server(rpcUrl);
+}
+
+export const sorobanClient = createSorobanClient();

indexer/streams/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"
   }
 }