diff --git a/.changeset/sinks-seam.md b/.changeset/sinks-seam.md
new file mode 100644
index 0000000..4be8975
--- /dev/null
+++ b/.changeset/sinks-seam.md
@@ -0,0 +1,12 @@
+---
+"@atmo-dev/contrail-base": minor
+"@atmo-dev/contrail-appview": minor
+---
+
+Add a first-class `sinks` config option: write-only, post-commit observers of applied records.
+
+A `Sink` builds derived state (a search index, an audit log, a webhook fan-out) from every record contrail ingests. Each configured sink's `onRecords(events, { phase })` fires inside `applyEvents()` after the DB commit, on **both** the live and backfill paths, receiving one deduplicated `RecordEvent` per record. Failures are isolated — a throwing sink is logged via the configured logger and never blocks ingestion.
+
+Unlike `realtime.pubsub`, a sink is not a subscriber: it serves no reads, requires no ticket secret, and must see backfilled records (where realtime is intentionally silent). Public records only — space-scoped records publish via the publishing adapter and never reach the fan-out.
+
+Purely additive: `realtime` and all existing behavior are unchanged. Runs identically on D1 and Postgres (it is an in-process call after commit, not a database-log consumer).
diff --git a/packages/contrail-appview/src/core/backfill.ts b/packages/contrail-appview/src/core/backfill.ts
index e75eb62..f53a709 100644
--- a/packages/contrail-appview/src/core/backfill.ts
+++ b/packages/contrail-appview/src/core/backfill.ts
@@ -262,6 +262,10 @@ export async function backfillUser(
         await applyEvents(db, events, config, {
           skipReplayDetection: options?.skipReplayDetection,
           skipFeedFanout: true,
+          // Realtime pubsub stays off during backfill, but `config.sinks` fire
+          // so a rebuild repopulates derived indexes. Tagged so sinks can
+          // bulk-flush differently from live ingest.
+          phase: "backfill",
         });
       }
       totalInserted += events.length;
diff --git a/packages/contrail-appview/src/core/db/records.ts b/packages/contrail-appview/src/core/db/records.ts
index 57f696e..42ac3b3 100644
--- a/packages/contrail-appview/src/core/db/records.ts
+++ b/packages/contrail-appview/src/core/db/records.ts
@@ -7,6 +7,7 @@ import type {
   IngestEvent,
   RecordRow,
   RecordSource,
+  RecordEvent,
 } from "../types";
 import {
   getNestedValue,
@@ -563,6 +564,9 @@ export async function applyEvents(
      *  (see `realtime/publishing-adapter.ts`); public topics carry public
      *  records only, which is exactly the scope of this function. */
     pubsub?: import("../realtime/types").PubSub;
+    /** Ingest phase forwarded to `config.sinks`. `"live"` for jetstream /
+     *  persistent ingest (default), `"backfill"` for replay / rebuild. */
+    phase?: "live" | "backfill";
   }
 ): Promise<void> {
   if (events.length === 0) return;
@@ -683,6 +687,37 @@ export async function applyEvents(
       }
     }
   }
+
+  // Fan out to write-only sinks (derived indexes, audit logs, webhooks).
+  // Unlike the realtime pubsub above this fires on BOTH the live and backfill
+  // paths (driven by `options.phase`), carries one deduplicated event per
+  // record, and isolates failures so a throwing sink never blocks ingestion.
+  const sinks = config?.sinks;
+  if (sinks && sinks.length > 0) {
+    const records: RecordEvent[] = events.map((e) =>
+      e.operation === "delete"
+        ? { kind: "deleted", uri: e.uri, did: e.did, collection: e.collection, rkey: e.rkey }
+        : {
+            kind: "created",
+            uri: e.uri,
+            did: e.did,
+            collection: e.collection,
+            rkey: e.rkey,
+            cid: e.cid,
+            record: e.record ? safeParseJson(e.record) : {},
+            time_us: e.time_us,
+          }
+    );
+    const ctx = { phase: options?.phase ?? "live" } as const;
+    const logger = config?.logger ?? console;
+    for (const sink of sinks) {
+      try {
+        await sink.onRecords(records, ctx);
+      } catch (err) {
+        logger.error("[sink] onRecords failed", err);
+      }
+    }
+  }
 }
 
 function safeParseJson(s: string): Record<string, unknown> {
diff --git a/packages/contrail-base/src/index.ts b/packages/contrail-base/src/index.ts
index fcf47b0..9d46c95 100644
--- a/packages/contrail-base/src/index.ts
+++ b/packages/contrail-base/src/index.ts
@@ -59,6 +59,9 @@ export * from "./community-integration";
 // Labels types
 export * from "./labels/types";
 
+// Sinks (write-only, post-commit observers of applied records)
+export * from "./sinks/types";
+
 // Realtime infrastructure
 export * from "./realtime/types";
 export * from "./realtime/in-memory";
diff --git a/packages/contrail-base/src/sinks/types.ts b/packages/contrail-base/src/sinks/types.ts
new file mode 100644
index 0000000..a12ccb3
--- /dev/null
+++ b/packages/contrail-base/src/sinks/types.ts
@@ -0,0 +1,55 @@
+/** Sinks — write-only, post-commit observers of applied records.
+ *
+ *  A sink builds *derived state* from the records contrail ingests: a search
+ *  index, an audit log, a webhook fan-out. It is NOT a realtime subscriber —
+ *  it never serves reads, and unlike `realtime.pubsub` (a lossy, drop-oldest
+ *  delivery channel for live UI feeds) it must see every applied record,
+ *  including during backfill. Contrail invokes each configured sink after every
+ *  `applyEvents()` commit, on BOTH the live and backfill paths, and isolates
+ *  failures: a throwing sink is logged and never blocks ingestion.
+ *
+ *  Scope — public records only. The fan-out lives in `applyEvents`, so sinks
+ *  see exactly what that path carries: public records. Space-scoped (private)
+ *  records publish through the separate publishing-adapter and never reach
+ *  `applyEvents`, so a sink cannot accidentally observe them. */
+
+export interface SinkContext {
+  /** `"live"` for jetstream / persistent ingest, `"backfill"` for replay or a
+   *  rebuild-after-wipe. A sink can buffer and bulk-flush differently during a
+   *  large backfill (sinks are expected to batch). */
+  phase: "live" | "backfill";
+}
+
+/** One event per applied record — deduplicated. Unlike the realtime
+ *  `RealtimeEvent`, a record is NOT split across `collection:` and `actor:`
+ *  topics (that split is a delivery concern), and there are no `member.*`
+ *  kinds. `created` covers both create and update — i.e. an upsert, matching
+ *  how the realtime path already collapses them. `deleted` carries identity
+ *  only. */
+export type RecordEvent =
+  | {
+      kind: "created";
+      uri: string;
+      did: string;
+      collection: string;
+      rkey: string;
+      cid: string | null;
+      record: Record<string, unknown>;
+      time_us: number;
+    }
+  | {
+      kind: "deleted";
+      uri: string;
+      did: string;
+      collection: string;
+      rkey: string;
+    };
+
+/** A write-only, post-commit observer of applied records. */
+export interface Sink {
+  /** Called once per `applyEvents()` batch, after the DB commit. Receives the
+   *  deduplicated records from that batch and the ingest `phase`. May be async;
+   *  contrail awaits it. A thrown error is caught, logged via the configured
+   *  logger, and never propagated — ingestion continues. */
+  onRecords(events: RecordEvent[], ctx: SinkContext): Promise<void> | void;
+}
diff --git a/packages/contrail-base/src/types.ts b/packages/contrail-base/src/types.ts
index 386cb1a..f6e20d3 100644
--- a/packages/contrail-base/src/types.ts
+++ b/packages/contrail-base/src/types.ts
@@ -228,6 +228,13 @@ export interface ContrailConfig {
   /** Realtime module configuration. When set, the service exposes ticket + SSE/WS
    *  subscribe XRPCs, and wraps the spaces adapter to publish events after writes. */
   realtime?: import("./realtime/types").RealtimeConfig;
+  /** Write-only, post-commit observers of applied records — derived indexes,
+   *  audit logs, webhook fan-outs. Each fires after every `applyEvents()` commit
+   *  on BOTH the live and backfill paths, with failures isolated so a throwing
+   *  sink never blocks ingestion. Distinct from `realtime`, which serves live
+   *  subscribers over a lossy delivery channel and is intentionally silent during
+   *  backfill. Public records only. */
+  sinks?: import("./sinks/types").Sink[];
   /** Labels module configuration. When set, contrail subscribes to the
    *  configured labelers, indexes their labels into a single `labels` table,
    *  and hydrates `record.labels` onto `listRecords` / `getRecord` / profile
diff --git a/packages/contrail/tests/sinks.test.ts b/packages/contrail/tests/sinks.test.ts
new file mode 100644
index 0000000..0e397c3
--- /dev/null
+++ b/packages/contrail/tests/sinks.test.ts
@@ -0,0 +1,166 @@
+/** Sinks — write-only, post-commit observers fanned out by `applyEvents`.
+ *
+ *   - One deduplicated event per record (not the realtime collection:/actor: pair).
+ *   - Fire on the live path by default and on backfill when `phase` says so.
+ *   - Failures are isolated: a throwing sink blocks neither the DB commit nor
+ *     the other sinks, and is logged.
+ *   - Public-record scope: space-scoped records publish via the publishing
+ *     adapter and never reach `applyEvents`, so a sink cannot observe them. */
+
+import { describe, it, expect } from "vitest";
+import { createSqliteDatabase } from "../src/adapters/sqlite";
+import { initSchema } from "../src/core/db/schema";
+import { applyEvents, queryRecords } from "../src/core/db/records";
+import { resolveConfig } from "../src/core/types";
+import type {
+  ContrailConfig,
+  IngestEvent,
+  RecordEvent,
+  Sink,
+  SinkContext,
+} from "../src/core/types";
+
+const ALICE = "did:plc:alice";
+const EVENT_NSID = "community.lexicon.calendar.event";
+const EVENT_URI = `at://${ALICE}/${EVENT_NSID}/abc`;
+
+/** Captures every onRecords call for assertions. */
+class RecordingSink implements Sink {
+  calls: { events: RecordEvent[]; ctx: SinkContext }[] = [];
+  async onRecords(events: RecordEvent[], ctx: SinkContext): Promise<void> {
+    this.calls.push({ events, ctx });
+  }
+}
+
+function configWithSinks(sinks: Sink[]): ContrailConfig {
+  return resolveConfig({
+    namespace: "test.sinks",
+    collections: { event: { collection: EVENT_NSID } },
+    sinks,
+  });
+}
+
+function createEvent(overrides: Partial<IngestEvent> = {}): IngestEvent {
+  return {
+    uri: EVENT_URI,
+    did: ALICE,
+    collection: EVENT_NSID,
+    rkey: "abc",
+    operation: "create",
+    cid: "bafytest",
+    record: JSON.stringify({ name: "Launch party" }),
+    time_us: 1_700_000_000_000_000,
+    indexed_at: 1_700_000_000_000,
+    ...overrides,
+  };
+}
+
+async function freshDb(config: ContrailConfig) {
+  const db = createSqliteDatabase(":memory:");
+  await initSchema(db, config);
+  return db;
+}
+
+describe("sinks — applyEvents fan-out", () => {
+  it("delivers one deduplicated created event per record, phase=live by default", async () => {
+    const sink = new RecordingSink();
+    const config = configWithSinks([sink]);
+    const db = await freshDb(config);
+
+    await applyEvents(db, [createEvent()], config);
+
+    expect(sink.calls).toHaveLength(1);
+    expect(sink.calls[0].ctx.phase).toBe("live");
+    // One event per record — not the collection:/actor: pair the realtime
+    // pubsub emits.
+    expect(sink.calls[0].events).toHaveLength(1);
+    expect(sink.calls[0].events[0]).toMatchObject({
+      kind: "created",
+      uri: EVENT_URI,
+      did: ALICE,
+      collection: EVENT_NSID,
+      rkey: "abc",
+      cid: "bafytest",
+      record: { name: "Launch party" },
+    });
+  });
+
+  it("delivers deleted events carrying identity only", async () => {
+    const sink = new RecordingSink();
+    const config = configWithSinks([sink]);
+    const db = await freshDb(config);
+
+    await applyEvents(db, [createEvent()], config);
+    await applyEvents(db, [createEvent({ operation: "delete" })], config);
+
+    expect(sink.calls).toHaveLength(2);
+    expect(sink.calls[1].events[0]).toEqual({
+      kind: "deleted",
+      uri: EVENT_URI,
+      did: ALICE,
+      collection: EVENT_NSID,
+      rkey: "abc",
+    });
+  });
+
+  it("forwards phase=backfill", async () => {
+    const sink = new RecordingSink();
+    const config = configWithSinks([sink]);
+    const db = await freshDb(config);
+
+    await applyEvents(db, [createEvent()], config, { phase: "backfill" });
+
+    expect(sink.calls[0].ctx.phase).toBe("backfill");
+  });
+
+  it("fans out to multiple sinks without nesting", async () => {
+    const a = new RecordingSink();
+    const b = new RecordingSink();
+    const config = configWithSinks([a, b]);
+    const db = await freshDb(config);
+
+    await applyEvents(db, [createEvent()], config);
+
+    expect(a.calls).toHaveLength(1);
+    expect(b.calls).toHaveLength(1);
+  });
+
+  it("isolates a throwing sink: the record still commits and other sinks still fire", async () => {
+    const errors: unknown[][] = [];
+    const throwing: Sink = {
+      onRecords() {
+        throw new Error("boom");
+      },
+    };
+    const after = new RecordingSink();
+    const config = configWithSinks([throwing, after]);
+    config.logger = {
+      log() {},
+      warn() {},
+      error: (...args: unknown[]) => {
+        errors.push(args);
+      },
+    };
+    const db = await freshDb(config);
+
+    await applyEvents(db, [createEvent()], config);
+
+    // The throw blocked neither the later sink...
+    expect(after.calls).toHaveLength(1);
+    // ...nor the DB commit...
+    const result = await queryRecords(db, config, { collection: EVENT_NSID });
+    expect(result.records).toHaveLength(1);
+    // ...and the failure was logged, not propagated.
+    expect(errors).toHaveLength(1);
+  });
+
+  it("does not fire when there are no events", async () => {
+    const sink = new RecordingSink();
+    const config = configWithSinks([sink]);
+    const db = await freshDb(config);
+
+    await applyEvents(db, [], config);
+
+    expect(sink.calls).toHaveLength(0);
+  });
+});