diff --git a/indexer/common/src/handlers/index.ts b/indexer/common/src/handlers/index.ts
new file mode 100644
index 0000000..5d7d50a
--- /dev/null
+++ b/indexer/common/src/handlers/index.ts
@@ -0,0 +1,7 @@
+export { HandlerRegistry } from "./registry.js";
+export type {
+  EventHandler,
+  HandlerFilter,
+  HandlerResult,
+  SorobanEventInput,
+} from "./types.js";
diff --git a/indexer/common/src/handlers/registry.test.ts b/indexer/common/src/handlers/registry.test.ts
new file mode 100644
index 0000000..41cd21c
--- /dev/null
+++ b/indexer/common/src/handlers/registry.test.ts
@@ -0,0 +1,123 @@
+import { describe, expect, test, vi } from "vitest";
+
+import { HandlerRegistry } from "./registry.js";
+import type { EventHandler, HandlerResult, SorobanEventInput } from "./types.js";
+
+const baseEvent: SorobanEventInput = {
+  contractId: "CABC123",
+  ledger: 100,
+  ledgerClosedAt: "2024-01-01T00:00:00Z",
+  topic: ["stream_created"],
+  data: { amount: 1000 },
+  id: "event-1",
+  pagingToken: "paging-1",
+};
+
+const ok: HandlerResult = { ok: true };
+const makeHandler = (result: HandlerResult = ok): EventHandler =>
+  vi.fn().mockResolvedValue(result);
+
+describe("HandlerRegistry", () => {
+  describe("register and matches", () => {
+    test("matches on contractId", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({ contractId: "CABC123" }, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+    });
+
+    test("does not match wrong contractId", () => {
+      const registry = new HandlerRegistry();
+      registry.register({ contractId: "COTHER" }, makeHandler());
+
+      expect(registry.matches(baseEvent)).toHaveLength(0);
+    });
+
+    test("matches on topic", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({ topic: "stream_created" }, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+    });
+
+    test("does not match absent topic", () => {
+      const registry = new HandlerRegistry();
+      registry.register({ topic: "stream_cancelled" }, makeHandler());
+
+      expect(registry.matches(baseEvent)).toHaveLength(0);
+    });
+
+    test("matches on eventName alias", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({ eventName: "stream_created" }, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+    });
+
+    test("matches combined contractId + topic filter", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({ contractId: "CABC123", topic: "stream_created" }, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+      expect(
+        registry.matches({ ...baseEvent, contractId: "COTHER" }),
+      ).toHaveLength(0);
+    });
+
+    test("returns multiple handlers when several match", () => {
+      const registry = new HandlerRegistry();
+      const h1 = makeHandler();
+      const h2 = makeHandler();
+      registry.register({ contractId: "CABC123" }, h1);
+      registry.register({ topic: "stream_created" }, h2);
+
+      expect(registry.matches(baseEvent)).toEqual([h1, h2]);
+    });
+
+    test("empty filter matches every event", () => {
+      const registry = new HandlerRegistry();
+      const handler = makeHandler();
+      registry.register({}, handler);
+
+      expect(registry.matches(baseEvent)).toEqual([handler]);
+    });
+  });
+
+  describe("dispatch", () => {
+    test("calls all matched handlers and returns their results", async () => {
+      const registry = new HandlerRegistry();
+      const h1 = makeHandler({ ok: true });
+      const h2 = makeHandler({ ok: false, error: "boom", retriable: true });
+      registry.register({}, h1);
+      registry.register({}, h2);
+
+      const results = await registry.dispatch(baseEvent);
+
+      expect(h1).toHaveBeenCalledWith(baseEvent);
+      expect(h2).toHaveBeenCalledWith(baseEvent);
+      expect(results).toEqual([
+        { ok: true },
+        { ok: false, error: "boom", retriable: true },
+      ]);
+    });
+
+    test("returns empty array when no handlers match", async () => {
+      const registry = new HandlerRegistry();
+      const results = await registry.dispatch(baseEvent);
+      expect(results).toEqual([]);
+    });
+  });
+
+  describe("fluent API", () => {
+    test("register returns the registry for chaining", () => {
+      const registry = new HandlerRegistry();
+      const returned = registry.register({}, makeHandler());
+      expect(returned).toBe(registry);
+    });
+  });
+});
+
diff --git a/indexer/common/src/handlers/registry.ts b/indexer/common/src/handlers/registry.ts
new file mode 100644
index 0000000..07a965e
--- /dev/null
+++ b/indexer/common/src/handlers/registry.ts
@@ -0,0 +1,50 @@
+import type {
+  EventHandler,
+  HandlerFilter,
+  HandlerResult,
+  SorobanEventInput,
+} from "./types.js";
+
+interface RegisteredHandler {
+  filter: HandlerFilter;
+  handler: EventHandler;
+}
+
+export class HandlerRegistry {
+  private readonly entries: RegisteredHandler[] = [];
+
+  register(filter: HandlerFilter, handler: EventHandler): this {
+    this.entries.push({ filter, handler });
+    return this;
+  }
+
+  matches(event: SorobanEventInput): EventHandler[] {
+    return this.entries
+      .filter(({ filter }) => {
+        if (filter.contractId && filter.contractId !== event.contractId) {
+          return false;
+        }
+        if (filter.topic && !event.topic.includes(filter.topic)) {
+          return false;
+        }
+        if (filter.eventName && !event.topic.includes(filter.eventName)) {
+          return false;
+        }
+        return true;
+      })
+      .map(({ handler }) => handler);
+  }
+
+  async dispatch(event: SorobanEventInput): Promise<HandlerResult[]> {
+    const handlers = this.matches(event);
+    return Promise.all(
+      handlers.map((h) =>
+        h(event).catch((err) => ({
+          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
new file mode 100644
index 0000000..444c418
--- /dev/null
+++ b/indexer/common/src/handlers/types.ts
@@ -0,0 +1,21 @@
+export interface SorobanEventInput {
+  contractId: string;
+  ledger: number;
+  ledgerClosedAt: string;
+  topic: string[];
+  data: unknown;
+  id: string;
+  pagingToken: string;
+}
+
+export type HandlerResult =
+  | { ok: true }
+  | { ok: false; error: string; retriable: boolean };
+
+export type EventHandler = (event: SorobanEventInput) => Promise<HandlerResult>;
+
+export interface HandlerFilter {
+  contractId?: string;
+  topic?: string;
+  eventName?: string;
+}
diff --git a/indexer/common/src/index.ts b/indexer/common/src/index.ts
index cbe5437..ad2f597 100644
--- a/indexer/common/src/index.ts
+++ b/indexer/common/src/index.ts
@@ -8,3 +8,4 @@ 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";
+export * from "./handlers/index.js";
diff --git a/indexer/distributions/src/handlers/distribution-created.handler.ts b/indexer/distributions/src/handlers/distribution-created.handler.ts
new file mode 100644
index 0000000..ea8cee0
--- /dev/null
+++ b/indexer/distributions/src/handlers/distribution-created.handler.ts
@@ -0,0 +1,76 @@
+import type {
+  EventHandler,
+  HandlerResult,
+  SorobanEventInput,
+} from "@fundable-indexer/common";
+import { parseDistributionCreated } from "./types.js";
+
+export const distributionCreatedHandler: EventHandler = async (
+  event: SorobanEventInput,
+): Promise<HandlerResult> => {
+  try {
+    const payload = parseDistributionCreated(event.data);
+
+    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.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.transactionHash) {
+      return {
+        ok: false,
+        error: "Missing transactionHash in distribution_created event",
+        retriable: false,
+      };
+    }
+
+    if (payload.recipientCount <= 0) {
+      return {
+        ok: false,
+        error: "Invalid recipientCount in distribution_created event",
+        retriable: false,
+      };
+    }
+
+    // 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
new file mode 100644
index 0000000..6c68d32
--- /dev/null
+++ b/indexer/distributions/src/handlers/distribution-handlers.test.ts
@@ -0,0 +1,148 @@
+import { describe, expect, test } from "vitest";
+
+import type { SorobanEventInput } from "@fundable-indexer/common";
+import { distributionCreatedHandler } from "./distribution-created.handler.js";
+import {
+  distributionPausedHandler,
+  distributionResumedHandler,
+} from "./distribution-pause.handler.js";
+import { tokensClaimedHandler } from "./tokens-claimed.handler.js";
+
+const baseEvent: SorobanEventInput = {
+  contractId: "CDIST456",
+  ledger: 300,
+  ledgerClosedAt: "2024-06-15T00:00:00Z",
+  topic: ["distribution_created"],
+  data: {},
+  id: "event-3",
+  pagingToken: "paging-3",
+};
+
+describe("distributionCreatedHandler", () => {
+  test("returns ok for valid created payload", async () => {
+    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 distributionCreatedHandler(event);
+    expect(result).toEqual({ ok: true });
+  });
+
+  test("returns error when distributionId is missing", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      data: { creator: "GCREATOR", token: "USDC" },
+    };
+
+    const result = await distributionCreatedHandler(event);
+    expect(result).toMatchObject({ ok: false, retriable: false });
+  });
+
+  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",
+    };
+
+    const results = await Promise.all([
+      distributionCreatedHandler({ ...baseEvent, data: payload }),
+      distributionCreatedHandler({ ...baseEvent, data: payload }),
+    ]);
+
+    expect(results.every((r) => r.ok)).toBe(true);
+  });
+});
+
+describe("tokensClaimedHandler", () => {
+  test("returns ok for valid claim payload", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      topic: ["tokens_claimed"],
+      data: {
+        distribution_id: "dist-1",
+        claimant: "GCLAIMANT",
+        amount: "2000",
+        tx_hash: "txclaim",
+      },
+    };
+
+    const result = await tokensClaimedHandler(event);
+    expect(result).toEqual({ ok: true });
+  });
+
+  test("returns error when distributionId is missing", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      data: { claimant: "GCLAIM", amount: "10" },
+    };
+
+    const result = await tokensClaimedHandler(event);
+    expect(result).toMatchObject({ ok: false, retriable: false });
+  });
+});
+
+describe("distributionPausedHandler", () => {
+  test("returns ok for valid paused payload", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      topic: ["distribution_paused"],
+      data: {
+        distribution_id: "dist-1",
+        paused_by: "GADMIN",
+        tx_hash: "txpause",
+      },
+    };
+
+    const result = await distributionPausedHandler(event);
+    expect(result).toEqual({ ok: true });
+  });
+
+  test("returns error when distributionId is missing", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      data: { paused_by: "GADMIN" },
+    };
+
+    const result = await distributionPausedHandler(event);
+    expect(result).toMatchObject({ ok: false, retriable: false });
+  });
+});
+
+describe("distributionResumedHandler", () => {
+  test("returns ok for valid resumed payload", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      topic: ["distribution_resumed"],
+      data: {
+        distribution_id: "dist-1",
+        resumed_by: "GADMIN",
+        tx_hash: "txresume",
+      },
+    };
+
+    const result = await distributionResumedHandler(event);
+    expect(result).toEqual({ ok: true });
+  });
+
+  test("returns error when distributionId is missing", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      data: { resumed_by: "GADMIN" },
+    };
+
+    const result = await distributionResumedHandler(event);
+    expect(result).toMatchObject({ ok: false, retriable: false });
+  });
+});
diff --git a/indexer/distributions/src/handlers/distribution-pause.handler.ts b/indexer/distributions/src/handlers/distribution-pause.handler.ts
new file mode 100644
index 0000000..2f73ff3
--- /dev/null
+++ b/indexer/distributions/src/handlers/distribution-pause.handler.ts
@@ -0,0 +1,96 @@
+import type {
+  EventHandler,
+  HandlerResult,
+  SorobanEventInput,
+} from "@fundable-indexer/common";
+import { parseDistributionPaused, parseDistributionResumed } from "./types.js";
+
+export const distributionPausedHandler: EventHandler = async (
+  event: SorobanEventInput,
+): Promise<HandlerResult> => {
+  try {
+    const payload = parseDistributionPaused(event.data);
+
+    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.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,
+    };
+  }
+};
+
+export const distributionResumedHandler: EventHandler = async (
+  event: SorobanEventInput,
+): Promise<HandlerResult> => {
+  try {
+    const payload = parseDistributionResumed(event.data);
+
+    if (!payload.distributionId) {
+      return {
+        ok: false,
+        error: "Missing distributionId in distribution_resumed event",
+        retriable: false,
+      };
+    }
+
+    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,
+      };
+    }
+
+    // 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
new file mode 100644
index 0000000..cad4931
--- /dev/null
+++ b/indexer/distributions/src/handlers/index.ts
@@ -0,0 +1,7 @@
+export { distributionCreatedHandler } from "./distribution-created.handler.js";
+export { tokensClaimedHandler } from "./tokens-claimed.handler.js";
+export {
+  distributionPausedHandler,
+  distributionResumedHandler,
+} from "./distribution-pause.handler.js";
+export * from "./types.js";
diff --git a/indexer/distributions/src/handlers/tokens-claimed.handler.ts b/indexer/distributions/src/handlers/tokens-claimed.handler.ts
new file mode 100644
index 0000000..10b5302
--- /dev/null
+++ b/indexer/distributions/src/handlers/tokens-claimed.handler.ts
@@ -0,0 +1,60 @@
+import type {
+  EventHandler,
+  HandlerResult,
+  SorobanEventInput,
+} from "@fundable-indexer/common";
+import { parseTokensClaimed } from "./types.js";
+
+export const tokensClaimedHandler: EventHandler = async (
+  event: SorobanEventInput,
+): Promise<HandlerResult> => {
+  try {
+    const payload = parseTokensClaimed(event.data);
+
+    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.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,
+      };
+    }
+
+    // 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
new file mode 100644
index 0000000..7aa6f4b
--- /dev/null
+++ b/indexer/distributions/src/handlers/types.ts
@@ -0,0 +1,86 @@
+export interface DistributionCreatedPayload {
+  distributionId: string;
+  creator: string;
+  token: string;
+  totalAmount: string;
+  recipientCount: number;
+  transactionHash: string;
+}
+
+export interface TokensClaimedPayload {
+  distributionId: string;
+  claimant: string;
+  amount: string;
+  transactionHash: string;
+}
+
+export interface DistributionPausedPayload {
+  distributionId: string;
+  pausedBy: string;
+  transactionHash: string;
+}
+
+export interface DistributionResumedPayload {
+  distributionId: string;
+  resumedBy: string;
+  transactionHash: string;
+}
+
+function record(v: unknown): Record<string, unknown> {
+  return v !== null && typeof v === "object" ? (v as Record<string, unknown>) : {};
+}
+
+function str(v: unknown): string {
+  return v !== undefined && v !== null ? String(v) : "";
+}
+
+function num(v: unknown): number {
+  const n = Number(v);
+  return Number.isFinite(n) ? n : 0;
+}
+
+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"),
+    recipientCount: num(d.recipientCount ?? d.recipient_count),
+    transactionHash: str(d.transactionHash ?? d.tx_hash),
+  };
+}
+
+export function parseTokensClaimed(data: unknown): TokensClaimedPayload {
+  const d = record(data);
+  return {
+    distributionId: str(d.distributionId ?? d.distribution_id),
+    claimant: str(d.claimant),
+    amount: str(d.amount ?? "0"),
+    transactionHash: str(d.transactionHash ?? d.tx_hash),
+  };
+}
+
+export function parseDistributionPaused(
+  data: unknown,
+): DistributionPausedPayload {
+  const d = record(data);
+  return {
+    distributionId: str(d.distributionId ?? d.distribution_id),
+    pausedBy: str(d.pausedBy ?? d.paused_by),
+    transactionHash: str(d.transactionHash ?? d.tx_hash),
+  };
+}
+
+export function parseDistributionResumed(
+  data: unknown,
+): DistributionResumedPayload {
+  const d = record(data);
+  return {
+    distributionId: str(d.distributionId ?? d.distribution_id),
+    resumedBy: str(d.resumedBy ?? d.resumed_by),
+    transactionHash: str(d.transactionHash ?? d.tx_hash),
+  };
+}
diff --git a/indexer/distributions/src/index.ts b/indexer/distributions/src/index.ts
index 51d59b6..f231142 100644
--- a/indexer/distributions/src/index.ts
+++ b/indexer/distributions/src/index.ts
@@ -5,3 +5,5 @@ export const distributionsPackage = {
   role: "distribution-indexer",
   common: commonPackage.name,
 } as const;
+
+export * from "./handlers/index.js";
diff --git a/indexer/streams/schema.graphql b/indexer/streams/schema.graphql
index 6fe88ce..1303db3 100644
--- a/indexer/streams/schema.graphql
+++ b/indexer/streams/schema.graphql
@@ -1,11 +1,78 @@
+enum StreamStatus {
+  ACTIVE
+  CANCELLED
+  COMPLETED
+}
+
 type Stream {
   id: ID!
+  contractId: String!
+  sender: String!
+  recipient: String!
+  token: String!
+  depositedAmount: String!
+  withdrawnAmount: String!
+  startTime: String!
+  stopTime: String!
+  status: StreamStatus!
+  ledger: Int!
+  ledgerClosedAt: String!
+  createdAt: String!
+  updatedAt: String!
+  withdrawals: [WithdrawalAction!]!
+  cancelAction: CancelAction
 }
 
 type WithdrawalAction {
   id: ID!
+  streamId: ID!
+  recipient: String!
+  amount: String!
+  ledger: Int!
+  ledgerClosedAt: String!
+  transactionHash: String!
+  createdAt: String!
 }
 
 type CancelAction {
   id: ID!
+  streamId: ID!
+  cancelledBy: String!
+  senderBalance: String!
+  recipientBalance: String!
+  ledger: Int!
+  ledgerClosedAt: String!
+  transactionHash: String!
+  createdAt: String!
+}
+
+input StreamFilterInput {
+  sender: String
+  recipient: String
+  token: String
+  status: StreamStatus
+  contractId: String
+}
+
+input PaginationInput {
+  first: Int
+  after: String
+}
+
+type StreamConnection {
+  nodes: [Stream!]!
+  pageInfo: PageInfo!
+  totalCount: Int!
+}
+
+type PageInfo {
+  hasNextPage: Boolean!
+  endCursor: String
+}
+
+type Query {
+  stream(id: ID!): Stream
+  streams(filter: StreamFilterInput, pagination: PaginationInput): StreamConnection!
+  streamsByRecipient(recipient: String!, pagination: PaginationInput): StreamConnection!
+  streamsBySender(sender: String!, pagination: PaginationInput): StreamConnection!
 }
diff --git a/indexer/streams/src/handlers/index.ts b/indexer/streams/src/handlers/index.ts
new file mode 100644
index 0000000..060e416
--- /dev/null
+++ b/indexer/streams/src/handlers/index.ts
@@ -0,0 +1,4 @@
+export { streamFundedHandler } from "./stream-funded.handler.js";
+export { streamWithdrawalHandler } from "./stream-withdrawal.handler.js";
+export { streamCancelHandler } from "./stream-cancel.handler.js";
+export * from "./types.js";
diff --git a/indexer/streams/src/handlers/stream-cancel.handler.ts b/indexer/streams/src/handlers/stream-cancel.handler.ts
new file mode 100644
index 0000000..2b3a916
--- /dev/null
+++ b/indexer/streams/src/handlers/stream-cancel.handler.ts
@@ -0,0 +1,39 @@
+import type {
+  EventHandler,
+  HandlerResult,
+  SorobanEventInput,
+} from "@fundable-indexer/common";
+import { parseStreamCancel } from "./types.js";
+
+export const streamCancelHandler: EventHandler = async (
+  event: SorobanEventInput,
+): Promise<HandlerResult> => {
+  try {
+    const payload = parseStreamCancel(event.data);
+
+    if (!payload.streamId) {
+      return { ok: false, error: "Missing streamId in cancel event", retriable: false };
+    }
+
+    if (!payload.cancelledBy) {
+      return { ok: false, error: "Missing cancelledBy in cancel event", retriable: false };
+    }
+
+    if (!payload.transactionHash) {
+      return { ok: false, error: "Missing transactionHash in cancel event", retriable: false };
+    }
+
+    // TODO(#32): update stream status to CANCELLED via repository once DB schema is merged
+    console.info(
+      `[stream-cancel] streamId=${payload.streamId} cancelledBy=${payload.cancelledBy} senderBalance=${payload.senderBalance} 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/streams/src/handlers/stream-funded.handler.ts b/indexer/streams/src/handlers/stream-funded.handler.ts
new file mode 100644
index 0000000..c0e951c
--- /dev/null
+++ b/indexer/streams/src/handlers/stream-funded.handler.ts
@@ -0,0 +1,47 @@
+import type {
+  EventHandler,
+  HandlerResult,
+  SorobanEventInput,
+} from "@fundable-indexer/common";
+import { parseStreamFunded } from "./types.js";
+
+export const streamFundedHandler: EventHandler = async (
+  event: SorobanEventInput,
+): Promise<HandlerResult> => {
+  try {
+    const payload = parseStreamFunded(event.data);
+
+    if (!payload.streamId) {
+      return { ok: false, error: "Missing streamId in funded event", retriable: false };
+    }
+
+    if (!payload.sender) {
+      return { ok: false, error: "Missing sender in funded event", retriable: false };
+    }
+
+    if (!payload.token) {
+      return { ok: false, error: "Missing token in funded event", retriable: false };
+    }
+
+    if (!payload.amount) {
+      return { ok: false, error: "Missing amount in funded event", retriable: false };
+    }
+
+    if (!payload.transactionHash) {
+      return { ok: false, error: "Missing transactionHash in funded event", retriable: false };
+    }
+
+    // TODO(#32): persist deposit via stream repository once DB schema is merged
+    console.info(
+      `[stream-funded] streamId=${payload.streamId} amount=${payload.amount} token=${payload.token} 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/streams/src/handlers/stream-handlers.test.ts b/indexer/streams/src/handlers/stream-handlers.test.ts
new file mode 100644
index 0000000..245831b
--- /dev/null
+++ b/indexer/streams/src/handlers/stream-handlers.test.ts
@@ -0,0 +1,112 @@
+import { describe, expect, test } from "vitest";
+
+import type { SorobanEventInput } from "@fundable-indexer/common";
+import { streamCancelHandler } from "./stream-cancel.handler.js";
+import { streamFundedHandler } from "./stream-funded.handler.js";
+import { streamWithdrawalHandler } from "./stream-withdrawal.handler.js";
+
+const baseEvent: SorobanEventInput = {
+  contractId: "CSTREAM123",
+  ledger: 200,
+  ledgerClosedAt: "2024-06-01T00:00:00Z",
+  topic: ["stream_funded"],
+  data: {},
+  id: "event-2",
+  pagingToken: "paging-2",
+};
+
+describe("streamFundedHandler", () => {
+  test("returns ok for valid funded payload", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      topic: ["stream_funded"],
+      data: {
+        stream_id: "stream-1",
+        sender: "GSENDER",
+        amount: "5000",
+        token: "USDC",
+        tx_hash: "abc123",
+      },
+    };
+
+    const result = await streamFundedHandler(event);
+    expect(result).toEqual({ ok: true });
+  });
+
+  test("returns error when streamId is missing", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      data: { amount: "100", token: "XLM", sender: "G123" },
+    };
+
+    const result = await streamFundedHandler(event);
+    expect(result).toMatchObject({ ok: false, retriable: false });
+  });
+
+  test("handles unexpected data shape without throwing", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      data: null,
+    };
+
+    const result = await streamFundedHandler(event);
+    expect(result.ok).toBe(false);
+  });
+});
+
+describe("streamWithdrawalHandler", () => {
+  test("returns ok for valid withdrawal payload", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      topic: ["stream_withdrawal"],
+      data: {
+        stream_id: "stream-1",
+        recipient: "GRECIPIENT",
+        amount: "250",
+        tx_hash: "def456",
+      },
+    };
+
+    const result = await streamWithdrawalHandler(event);
+    expect(result).toEqual({ ok: true });
+  });
+
+  test("returns error when streamId is missing", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      data: { recipient: "G123", amount: "50" },
+    };
+
+    const result = await streamWithdrawalHandler(event);
+    expect(result).toMatchObject({ ok: false, retriable: false });
+  });
+});
+
+describe("streamCancelHandler", () => {
+  test("returns ok for valid cancel payload", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      topic: ["stream_cancel"],
+      data: {
+        stream_id: "stream-1",
+        cancelled_by: "GSENDER",
+        sender_balance: "4750",
+        recipient_balance: "250",
+        tx_hash: "ghi789",
+      },
+    };
+
+    const result = await streamCancelHandler(event);
+    expect(result).toEqual({ ok: true });
+  });
+
+  test("returns error when streamId is missing", async () => {
+    const event: SorobanEventInput = {
+      ...baseEvent,
+      data: { cancelled_by: "G123" },
+    };
+
+    const result = await streamCancelHandler(event);
+    expect(result).toMatchObject({ ok: false, retriable: false });
+  });
+});
diff --git a/indexer/streams/src/handlers/stream-withdrawal.handler.ts b/indexer/streams/src/handlers/stream-withdrawal.handler.ts
new file mode 100644
index 0000000..1e03219
--- /dev/null
+++ b/indexer/streams/src/handlers/stream-withdrawal.handler.ts
@@ -0,0 +1,43 @@
+import type {
+  EventHandler,
+  HandlerResult,
+  SorobanEventInput,
+} from "@fundable-indexer/common";
+import { parseStreamWithdrawal } from "./types.js";
+
+export const streamWithdrawalHandler: EventHandler = async (
+  event: SorobanEventInput,
+): Promise<HandlerResult> => {
+  try {
+    const payload = parseStreamWithdrawal(event.data);
+
+    if (!payload.streamId) {
+      return { ok: false, error: "Missing streamId in withdrawal event", retriable: false };
+    }
+
+    if (!payload.recipient) {
+      return { ok: false, error: "Missing recipient in withdrawal event", retriable: false };
+    }
+
+    if (!payload.amount) {
+      return { ok: false, error: "Missing amount in withdrawal event", retriable: false };
+    }
+
+    if (!payload.transactionHash) {
+      return { ok: false, error: "Missing transactionHash in withdrawal event", retriable: false };
+    }
+
+    // TODO(#32): record withdrawal action via repository once DB schema is merged
+    console.info(
+      `[stream-withdrawal] streamId=${payload.streamId} recipient=${payload.recipient} 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/streams/src/handlers/types.ts b/indexer/streams/src/handlers/types.ts
new file mode 100644
index 0000000..92bb017
--- /dev/null
+++ b/indexer/streams/src/handlers/types.ts
@@ -0,0 +1,64 @@
+export interface StreamFundedPayload {
+  streamId: string | undefined;
+  sender: string | undefined;
+  amount: string | undefined;
+  token: string | undefined;
+  transactionHash: string | undefined;
+}
+
+export interface StreamWithdrawalPayload {
+  streamId: string | undefined;
+  recipient: string | undefined;
+  amount: string | undefined;
+  transactionHash: string | undefined;
+}
+
+export interface StreamCancelPayload {
+  streamId: string | undefined;
+  cancelledBy: string | undefined;
+  senderBalance: string | undefined;
+  recipientBalance: string | undefined;
+  transactionHash: string | undefined;
+}
+
+function record(v: unknown): Record<string, unknown> {
+  return v !== null && typeof v === "object" ? (v as Record<string, unknown>) : {};
+}
+
+function str(v: unknown): string | undefined {
+  if (v === undefined || v === null || v === "") return undefined;
+  const s = String(v);
+  return s === "" ? undefined : s;
+}
+
+export function parseStreamFunded(data: unknown): StreamFundedPayload {
+  const d = record(data);
+  return {
+    streamId: str(d.streamId ?? d.stream_id),
+    sender: str(d.sender),
+    amount: str(d.amount),
+    token: str(d.token),
+    transactionHash: str(d.transactionHash ?? d.tx_hash),
+  };
+}
+
+export function parseStreamWithdrawal(data: unknown): StreamWithdrawalPayload {
+  const d = record(data);
+  return {
+    streamId: str(d.streamId ?? d.stream_id),
+    recipient: str(d.recipient),
+    amount: str(d.amount),
+    transactionHash: str(d.transactionHash ?? d.tx_hash),
+  };
+}
+
+export function parseStreamCancel(data: unknown): StreamCancelPayload {
+  const d = record(data);
+  return {
+    streamId: str(d.streamId ?? d.stream_id),
+    cancelledBy: str(d.cancelledBy ?? d.cancelled_by),
+    senderBalance: str(d.senderBalance ?? d.sender_balance),
+    recipientBalance: str(d.recipientBalance ?? d.recipient_balance),
+    transactionHash: str(d.transactionHash ?? d.tx_hash),
+  };
+}
diff --git a/indexer/streams/src/index.ts b/indexer/streams/src/index.ts
index 2da94b9..94682ad 100644
--- a/indexer/streams/src/index.ts
+++ b/indexer/streams/src/index.ts
@@ -9,3 +9,4 @@ export const streamsPackage = {
 export { Stream } from "./db/entity/Stream.js";
 export { WithdrawalAction } from "./db/entity/WithdrawalAction.js";
 export { CancelAction } from "./db/entity/CancelAction.js";
+export * from "./handlers/index.js";