diff --git a/packages/trailbase-db-collection/e2e/Dockerfile b/packages/trailbase-db-collection/e2e/Dockerfile
new file mode 100644
index 000000000..caea23727
--- /dev/null
+++ b/packages/trailbase-db-collection/e2e/Dockerfile
@@ -0,0 +1,6 @@
+FROM trailbase/trailbase:latest
+
+COPY traildepot /app/traildepot/
+EXPOSE 4000
+
+CMD ["/app/trail", "--data-dir", "/app/traildepot", "run", "--address", "0.0.0.0:4000"]
diff --git a/packages/trailbase-db-collection/e2e/global-setup.ts b/packages/trailbase-db-collection/e2e/global-setup.ts
new file mode 100644
index 000000000..6d3f9ccb8
--- /dev/null
+++ b/packages/trailbase-db-collection/e2e/global-setup.ts
@@ -0,0 +1,166 @@
+import { execSync, spawn } from 'node:child_process'
+import { dirname } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import type { ChildProcess } from 'node:child_process'
+import type { TestProject } from 'vitest/node'
+
+const CONTAINER_NAME = 'trailbase-e2e-test'
+const TRAILBASE_PORT = process.env.TRAILBASE_PORT ?? '4047'
+const TRAILBASE_URL =
+  process.env.TRAILBASE_URL ?? `http://localhost:${TRAILBASE_PORT}`
+
+// Module augmentation for type-safe context injection
+declare module 'vitest' {
+  export interface ProvidedContext {
+    baseUrl: string
+  }
+}
+
+function isDockerAvailable(): boolean {
+  try {
+    execSync('docker --version', { stdio: 'pipe' })
+    return true
+  } catch {}
+
+  return false
+}
+
+async function isTrailBaseRunning(url: string): Promise<boolean> {
+  try {
+    const res = await fetch(`${url}/api/healthcheck`)
+    return res.ok
+  } catch {}
+
+  return false
+}
+
+function buildDockerImage(): void {
+  const DOCKER_DIR = dirname(fileURLToPath(import.meta.url))
+
+  console.log('🔨 Building TrailBase Docker image...')
+  execSync(`docker build -t trailbase-e2e ${DOCKER_DIR}`, {
+    stdio: 'inherit',
+  })
+  console.log('✓ Docker image built')
+}
+
+function cleanupExistingContainer(): void {
+  try {
+    execSync(`docker stop ${CONTAINER_NAME}`, {
+      stdio: 'pipe',
+    })
+    execSync(`docker rm ${CONTAINER_NAME}`, {
+      stdio: 'pipe',
+    })
+  } catch {
+    // Ignore errors - container might not exist
+  }
+}
+
+function startDockerContainer(): ChildProcess {
+  console.log('🚀 Starting TrailBase container...')
+
+  const proc = spawn(
+    'docker',
+    [
+      'run',
+      '--rm',
+      '--name',
+      CONTAINER_NAME,
+      '-p',
+      `${TRAILBASE_PORT}:4000`,
+      'trailbase-e2e',
+    ],
+    {
+      stdio: ['ignore', 'pipe', 'pipe'],
+    },
+  )
+
+  proc.stdout.on('data', (data) => {
+    console.log(`[trailbase] ${data.toString().trim()}`)
+  })
+
+  proc.stderr.on('data', (data) => {
+    console.error(`[trailbase] ${data.toString().trim()}`)
+  })
+
+  proc.on('error', (error) => {
+    console.error('Failed to start TrailBase container:', error)
+  })
+
+  return proc
+}
+
+async function waitForTrailBase(url: string): Promise<void> {
+  return new Promise<void>((resolve, reject) => {
+    const timeout = setTimeout(() => {
+      reject(
+        new Error(`Timed out waiting for TrailBase to be active at ${url}`),
+      )
+    }, 60000) // 60 seconds timeout for startup
+
+    const check = async (): Promise<void> => {
+      try {
+        // Try the healthz endpoint first, then fall back to root
+        const res = await fetch(`${url}/api/healthcheck`)
+        if (res.ok) {
+          clearTimeout(timeout)
+          return resolve()
+        }
+      } catch {}
+
+      setTimeout(() => void check(), 500)
+    }
+
+    void check()
+  })
+}
+
+/**
+ * Global setup for TrailBase e2e test suite
+ */
+export default async function ({ provide }: TestProject) {
+  let serverProcess: ChildProcess | null = null
+
+  // Check if TrailBase is already running
+  if (await isTrailBaseRunning(TRAILBASE_URL)) {
+    console.log(`✓ TrailBase already running at ${TRAILBASE_URL}`)
+  } else {
+    if (!isDockerAvailable()) {
+      throw new Error(
+        `TrailBase is not running at ${TRAILBASE_URL} and no startup method is available.\n` +
+          `Please either:\n` +
+          `  1. Start TrailBase manually at ${TRAILBASE_URL}\n` +
+          `  2. Install Docker and run the tests again\n`,
+      )
+    }
+
+    // Clean up any existing container
+    cleanupExistingContainer()
+    // Build Docker image
+    buildDockerImage()
+    // Start container
+    serverProcess = startDockerContainer()
+  }
+
+  // Wait for TrailBase server to be ready
+  console.log(`⏳ Waiting for TrailBase at ${TRAILBASE_URL}...`)
+  await waitForTrailBase(TRAILBASE_URL)
+  console.log('✓ TrailBase is ready')
+
+  // Provide context values to all tests
+  provide('baseUrl', TRAILBASE_URL)
+
+  console.log('✓ Global setup complete\n')
+
+  // Return cleanup function (runs once after all tests)
+  return () => {
+    console.log('\n🧹 Running global teardown...')
+    if (serverProcess !== null) {
+      cleanupExistingContainer()
+      serverProcess.kill()
+      serverProcess = null
+    }
+    console.log('✅ Global teardown complete')
+  }
+}
diff --git a/packages/trailbase-db-collection/e2e/trailbase.e2e.test.ts b/packages/trailbase-db-collection/e2e/trailbase.e2e.test.ts
new file mode 100644
index 000000000..af0da8449
--- /dev/null
+++ b/packages/trailbase-db-collection/e2e/trailbase.e2e.test.ts
@@ -0,0 +1,415 @@
+/**
+ * TrailBase Collection E2E Tests
+ *
+ * End-to-end tests using actual TrailBase server with sync.
+ * Uses shared test suites from @tanstack/db-collection-e2e.
+ */
+
+import { describe, expect, inject } from 'vitest'
+import { parse as uuidParse, stringify as uuidStringify } from 'uuid'
+import { createCollection } from '@tanstack/db'
+import { initClient } from 'trailbase'
+import { trailBaseCollectionOptions } from '../src/trailbase'
+import {
+  createCollationTestSuite,
+  createDeduplicationTestSuite,
+  createJoinsTestSuite,
+  createLiveUpdatesTestSuite,
+  createMutationsTestSuite,
+  createPaginationTestSuite,
+  createPredicatesTestSuite,
+  createProgressiveTestSuite,
+  generateSeedData,
+} from '../../db-collection-e2e/src/index'
+import type { Client } from 'trailbase'
+import type { SeedDataResult } from '../../db-collection-e2e/src/index'
+import type { TrailBaseSyncMode } from '../src/trailbase'
+import type {
+  Comment,
+  E2ETestConfig,
+  Post,
+  User,
+} from '../../db-collection-e2e/src/types'
+import type { Collection } from '@tanstack/db'
+
+declare module 'vitest' {
+  export interface ProvidedContext {
+    baseUrl: string
+  }
+}
+
+// / Decode a "url-safe" base64 string to bytes.
+function urlSafeBase64Decode(base64: string): Uint8Array {
+  return Uint8Array.from(
+    atob(base64.replace(/_/g, '/').replace(/-/g, '+')),
+    (c) => c.charCodeAt(0),
+  )
+}
+
+// / Encode an arbitrary string input as a "url-safe" base64 string.
+function urlSafeBase64Encode(bytes: Uint8Array): string {
+  return btoa(String.fromCharCode(...bytes))
+    .replace(/\//g, '_')
+    .replace(/\+/g, '-')
+}
+
+function parseTrailBaseId(id: string): string {
+  return uuidStringify(urlSafeBase64Decode(id))
+}
+
+function toTrailBaseId(id: string): string {
+  return urlSafeBase64Encode(uuidParse(id))
+}
+
+/**
+ * TrailBase record types matching the camelCase schema
+ * Column names match the app types, only types differ for storage
+ */
+interface UserRecord {
+  id: string // base64 encoded UUID
+  name: string
+  email: string | null
+  age: number
+  isActive: number // SQLite INTEGER (0/1) for boolean
+  createdAt: string // ISO date string
+  metadata: string | null // JSON stored as string
+  deletedAt: string | null // ISO date string
+}
+
+interface PostRecord {
+  id: string
+  userId: string
+  title: string
+  content: string | null
+  viewCount: number
+  largeViewCount: string // BigInt as string
+  publishedAt: string | null
+  deletedAt: string | null
+}
+
+interface CommentRecord {
+  id: string
+  postId: string
+  userId: string
+  text: string
+  createdAt: string
+  deletedAt: string | null
+}
+
+/**
+ * Serialize functions - transform app types to DB storage types
+ * ID is base64 encoded for TrailBase BLOB storage
+ */
+const serializeUser = (user: User): UserRecord => ({
+  ...user,
+  isActive: user.isActive ? 1 : 0,
+  createdAt: user.createdAt.toISOString(),
+  metadata: user.metadata ? JSON.stringify(user.metadata) : null,
+  deletedAt: user.deletedAt ? user.deletedAt.toISOString() : null,
+})
+
+const serializePost = (post: Post): PostRecord => ({
+  ...post,
+  largeViewCount: post.largeViewCount.toString(),
+  publishedAt: post.publishedAt ? post.publishedAt.toISOString() : null,
+  deletedAt: post.deletedAt ? post.deletedAt.toISOString() : null,
+})
+
+const serializeComment = (comment: Comment): CommentRecord => ({
+  ...comment,
+  createdAt: comment.createdAt.toISOString(),
+  deletedAt: comment.deletedAt ? comment.deletedAt.toISOString() : null,
+})
+
+/**
+ * Helper to create a set of collections for a given sync mode
+ */
+function createCollectionsForSyncMode(
+  client: Client,
+  testId: string,
+  syncMode: TrailBaseSyncMode,
+  suffix: string,
+) {
+  const usersRecordApi = client.records<UserRecord>(`users_e2e`)
+  const postsRecordApi = client.records<PostRecord>(`posts_e2e`)
+  const commentsRecordApi = client.records<CommentRecord>(`comments_e2e`)
+
+  const usersCollection = createCollection(
+    trailBaseCollectionOptions({
+      id: `trailbase-e2e-users-${suffix}-${testId}`,
+      recordApi: usersRecordApi,
+      getKey: (item: User) => item.id,
+      startSync: true,
+      syncMode,
+      parse: {
+        id: parseTrailBaseId,
+        isActive: (isActive) => Boolean(isActive),
+        createdAt: (createdAt) => new Date(createdAt),
+        metadata: (m) => (m ? JSON.parse(m) : null),
+        deletedAt: (d) => (d ? new Date(d) : null),
+      },
+      serialize: {
+        id: toTrailBaseId,
+        isActive: (a) => (a ? 1 : 0),
+        createdAt: (c) => c.toISOString(),
+        metadata: (m) => (m ? JSON.stringify(m) : null),
+        deletedAt: (d) => (d ? d.toISOString() : null),
+      },
+    }),
+  )
+
+  const postsCollection = createCollection(
+    trailBaseCollectionOptions({
+      id: `trailbase-e2e-posts-${suffix}-${testId}`,
+      recordApi: postsRecordApi,
+      getKey: (item: Post) => item.id,
+      startSync: true,
+      syncMode,
+      parse: {
+        id: parseTrailBaseId,
+        largeViewCount: (l) => BigInt(l),
+        publishedAt: (v) => (v ? new Date(v) : null),
+        deletedAt: (d) => (d ? new Date(d) : null),
+      },
+      serialize: {
+        id: toTrailBaseId,
+        largeViewCount: (v) => v.toString(),
+        publishedAt: (v) => (v ? v.toISOString() : null),
+        deletedAt: (d) => (d ? d.toISOString() : null),
+      },
+    }),
+  )
+
+  const commentsCollection = createCollection(
+    trailBaseCollectionOptions({
+      id: `trailbase-e2e-comments-${suffix}-${testId}`,
+      recordApi: commentsRecordApi,
+      getKey: (item: Comment) => item.id,
+      startSync: true,
+      syncMode,
+      parse: {
+        id: parseTrailBaseId,
+        createdAt: (v) => new Date(v),
+        deletedAt: (d) => (d ? new Date(d) : null),
+      },
+      serialize: {
+        id: toTrailBaseId,
+        createdAt: (v) => v.toISOString(),
+        deletedAt: (d) => (d ? d.toISOString() : null),
+      },
+    }),
+  )
+
+  return {
+    users: usersCollection as Collection<User>,
+    posts: postsCollection as Collection<Post>,
+    comments: commentsCollection as Collection<Comment>,
+  }
+}
+
+async function initialCleanup(client: Client) {
+  console.log(`Cleaning up existing records...`)
+
+  const commentsRecordApi = client.records<CommentRecord>(`comments_e2e`)
+  const existingComments = await commentsRecordApi.list({})
+  for (const comment of existingComments.records) {
+    try {
+      await commentsRecordApi.delete(comment.id)
+    } catch {
+      /* ignore */
+    }
+  }
+
+  const postsRecordApi = client.records<PostRecord>(`posts_e2e`)
+  const existingPosts = await postsRecordApi.list({})
+  for (const post of existingPosts.records) {
+    try {
+      await postsRecordApi.delete(post.id)
+    } catch {
+      /* ignore */
+    }
+  }
+
+  const usersRecordApi = client.records<UserRecord>(`users_e2e`)
+  const existingUsers = await usersRecordApi.list({})
+  for (const user of existingUsers.records) {
+    try {
+      await usersRecordApi.delete(user.id)
+    } catch {
+      /* ignore */
+    }
+  }
+
+  console.log(`Cleanup complete`)
+}
+
+async function setupInitialData(client: Client, seedData: SeedDataResult) {
+  const usersRecordApi = client.records<UserRecord>(`users_e2e`)
+  const postsRecordApi = client.records<PostRecord>(`posts_e2e`)
+  const commentsRecordApi = client.records<CommentRecord>(`comments_e2e`)
+
+  // Insert seed data - we provide the ID so the original UUIDs are preserved
+  console.log(`Inserting ${seedData.users.length} users...`)
+  let userErrors = 0
+  for (const user of seedData.users) {
+    try {
+      const serialized = serializeUser(user)
+      if (userErrors === 0)
+        console.log('First user data:', JSON.stringify(serialized))
+      await usersRecordApi.create(serialized)
+    } catch (e) {
+      userErrors++
+      if (userErrors <= 3) console.error('User insert error:', e)
+    }
+  }
+  console.log(
+    `Inserted users: ${seedData.users.length - userErrors} success, ${userErrors} errors`,
+  )
+  console.log(`First user ID: ${seedData.users.at(0)?.id}`)
+
+  console.log(`Inserting ${seedData.posts.length} posts...`)
+  let postErrors = 0
+  for (const post of seedData.posts) {
+    try {
+      await postsRecordApi.create(serializePost(post))
+    } catch (e) {
+      postErrors++
+      if (postErrors <= 3) console.error('Post insert error:', e)
+    }
+  }
+  console.log(
+    `Inserted posts: ${seedData.posts.length - postErrors} success, ${postErrors} errors`,
+  )
+
+  console.log(`Inserting ${seedData.comments.length} comments...`)
+  let commentErrors = 0
+  for (const comment of seedData.comments) {
+    try {
+      await commentsRecordApi.create(serializeComment(comment))
+    } catch (e) {
+      commentErrors++
+      if (commentErrors <= 3) console.error('Comment insert error:', e)
+    }
+  }
+  console.log(
+    `Inserted comments: ${seedData.comments.length - commentErrors} success, ${commentErrors} errors`,
+  )
+}
+
+describe(`TrailBase Collection E2E Tests`, async () => {
+  const baseUrl = inject(`baseUrl`)
+  const client = initClient(baseUrl)
+
+  // Wipe all pre-existing data, e.g. when using a persistent TB instance.
+  await initialCleanup(client)
+
+  const seedData = generateSeedData()
+  await setupInitialData(client, seedData)
+
+  async function getConfig(): Promise<E2ETestConfig> {
+    // Create collections with different sync modes
+    const testId = Date.now().toString(16)
+
+    const onDemandCollections = createCollectionsForSyncMode(
+      client,
+      testId,
+      `on-demand`,
+      `ondemand`,
+    )
+
+    // On-demand collections are marked ready immediately
+    await Promise.all([
+      onDemandCollections.users.preload(),
+      onDemandCollections.posts.preload(),
+      onDemandCollections.comments.preload(),
+    ])
+
+    const eagerCollections = createCollectionsForSyncMode(
+      client,
+      testId,
+      `eager`,
+      `eager`,
+    )
+
+    // Wait for eager collections to sync (they need to fetch all data before marking ready)
+    // console.log('Calling preload on eager collections...')
+    await Promise.all([
+      eagerCollections.users.preload(),
+      eagerCollections.posts.preload(),
+      eagerCollections.comments.preload(),
+    ])
+    expect(eagerCollections.posts.size).toEqual(seedData.posts.length)
+    expect(eagerCollections.comments.size).toEqual(seedData.comments.length)
+
+    // NOTE: One of the tests deletes a user :/
+    expect(eagerCollections.users.size).toBeGreaterThanOrEqual(
+      seedData.users.length - 1,
+    )
+
+    const usersRecordApi = client.records<UserRecord>(`users_e2e`)
+    const postsRecordApi = client.records<PostRecord>(`posts_e2e`)
+
+    return {
+      collections: {
+        eager: {
+          users: eagerCollections.users,
+          posts: eagerCollections.posts,
+          comments: eagerCollections.comments,
+        },
+        onDemand: {
+          users: onDemandCollections.users,
+          posts: onDemandCollections.posts,
+          comments: onDemandCollections.comments,
+        },
+      },
+      hasReplicationLag: true, // TrailBase has async subscription-based sync
+      // Note: progressiveTestControl is not provided because the explicit snapshot/swap
+      // transition tests require Electric-specific sync behavior that TrailBase doesn't support.
+      // Tests that require this will be skipped.
+      mutations: {
+        insertUser: async (user) => {
+          // Insert with the provided ID (base64-encoded UUID)
+          await usersRecordApi.create(serializeUser(user))
+          // ID is preserved from the user object
+        },
+        updateUser: async (id, updates) => {
+          const partialRecord: Partial<UserRecord> = {}
+          if (updates.age !== undefined) partialRecord.age = updates.age
+          if (updates.name !== undefined) partialRecord.name = updates.name
+          if (updates.email !== undefined) partialRecord.email = updates.email
+          if (updates.isActive !== undefined)
+            partialRecord.isActive = updates.isActive ? 1 : 0
+          await usersRecordApi.update(id, partialRecord)
+        },
+        deleteUser: async (id) => {
+          await usersRecordApi.delete(id)
+        },
+        insertPost: async (post) => {
+          // Insert with the provided ID
+          await postsRecordApi.create(serializePost(post))
+        },
+      },
+      setup: async () => {},
+      teardown: async () => {
+        await Promise.all([
+          eagerCollections.users.cleanup(),
+          eagerCollections.posts.cleanup(),
+          eagerCollections.comments.cleanup(),
+          onDemandCollections.users.cleanup(),
+          onDemandCollections.posts.cleanup(),
+          onDemandCollections.comments.cleanup(),
+        ])
+      },
+    }
+  }
+
+  // Run all shared test suites
+  createPredicatesTestSuite(getConfig)
+  createPaginationTestSuite(getConfig)
+  createJoinsTestSuite(getConfig)
+  createDeduplicationTestSuite(getConfig)
+  createCollationTestSuite(getConfig)
+  createMutationsTestSuite(getConfig)
+  createLiveUpdatesTestSuite(getConfig)
+  createProgressiveTestSuite(getConfig)
+})
diff --git a/packages/trailbase-db-collection/e2e/traildepot/config.textproto b/packages/trailbase-db-collection/e2e/traildepot/config.textproto
new file mode 100644
index 000000000..230a61465
--- /dev/null
+++ b/packages/trailbase-db-collection/e2e/traildepot/config.textproto
@@ -0,0 +1,26 @@
+email {}
+server {
+  application_name: "TrailBase E2E"
+}
+auth {}
+jobs {}
+record_apis: [
+  {
+    name: "users_e2e"
+    table_name: "users_e2e"
+    acl_world: [CREATE, READ, UPDATE, DELETE]
+    enable_subscriptions: true
+  },
+  {
+    name: "posts_e2e"
+    table_name: "posts_e2e"
+    acl_world: [CREATE, READ, UPDATE, DELETE]
+    enable_subscriptions: true
+  },
+  {
+    name: "comments_e2e"
+    table_name: "comments_e2e"
+    acl_world: [CREATE, READ, UPDATE, DELETE]
+    enable_subscriptions: true
+  }
+]
diff --git a/packages/trailbase-db-collection/e2e/traildepot/migrations/main/V10__init.sql b/packages/trailbase-db-collection/e2e/traildepot/migrations/main/V10__init.sql
new file mode 100644
index 000000000..28b8158cc
--- /dev/null
+++ b/packages/trailbase-db-collection/e2e/traildepot/migrations/main/V10__init.sql
@@ -0,0 +1,35 @@
+-- E2E Test Tables for TrailBase
+-- Using BLOB UUID PRIMARY KEY with auto-generated uuid_v7()
+-- Using is_uuid() check to accept both v4 and v7 UUIDs
+-- Using camelCase column names to match @tanstack/db-collection-e2e types
+
+CREATE TABLE "users_e2e" (
+  "id" BLOB PRIMARY KEY NOT NULL CHECK(is_uuid(id)) DEFAULT (uuid_v7()),
+  "name" TEXT NOT NULL,
+  "email" TEXT,
+  "age" INTEGER NOT NULL,
+  "isActive" INTEGER NOT NULL DEFAULT 1,
+  "createdAt" TEXT NOT NULL,
+  "metadata" TEXT,
+  "deletedAt" TEXT
+) STRICT;
+
+CREATE TABLE "posts_e2e" (
+  "id" BLOB PRIMARY KEY NOT NULL CHECK(is_uuid(id)) DEFAULT (uuid_v7()),
+  "userId" TEXT NOT NULL,
+  "title" TEXT NOT NULL,
+  "content" TEXT,
+  "viewCount" INTEGER NOT NULL DEFAULT 0,
+  "largeViewCount" TEXT NOT NULL,
+  "publishedAt" TEXT,
+  "deletedAt" TEXT
+) STRICT;
+
+CREATE TABLE "comments_e2e" (
+  "id" BLOB PRIMARY KEY NOT NULL CHECK(is_uuid(id)) DEFAULT (uuid_v7()),
+  "postId" TEXT NOT NULL,
+  "userId" TEXT NOT NULL,
+  "text" TEXT NOT NULL,
+  "createdAt" TEXT NOT NULL,
+  "deletedAt" TEXT
+) STRICT;
diff --git a/packages/trailbase-db-collection/package.json b/packages/trailbase-db-collection/package.json
index f1837cfd7..a2bf19933 100644
--- a/packages/trailbase-db-collection/package.json
+++ b/packages/trailbase-db-collection/package.json
@@ -20,7 +20,8 @@
     "build": "vite build",
     "dev": "vite build --watch",
     "lint": "eslint . --fix",
-    "test": "vitest --run"
+    "test": "vitest --run",
+    "test:e2e": "vitest --run --config vitest.e2e.config.ts"
   },
   "type": "module",
   "main": "dist/cjs/index.cjs",
@@ -55,7 +56,9 @@
     "typescript": ">=4.7"
   },
   "devDependencies": {
+    "@tanstack/db-collection-e2e": "workspace:*",
     "@types/debug": "^4.1.12",
-    "@vitest/coverage-istanbul": "^3.2.4"
+    "@vitest/coverage-istanbul": "^3.2.4",
+    "uuid": "^13.0.0"
   }
 }
diff --git a/packages/trailbase-db-collection/src/trailbase.ts b/packages/trailbase-db-collection/src/trailbase.ts
index 32b6755c5..5bdc84548 100644
--- a/packages/trailbase-db-collection/src/trailbase.ts
+++ b/packages/trailbase-db-collection/src/trailbase.ts
@@ -6,14 +6,16 @@ import {
   ExpectedUpdateTypeError,
   TimeoutWaitingForIdsError,
 } from './errors'
-import type { Event, RecordApi } from 'trailbase'
+import type { CompareOp, Event, FilterOrComposite, RecordApi } from 'trailbase'
 
 import type {
   BaseCollectionConfig,
   CollectionConfig,
   DeleteMutationFnParams,
   InsertMutationFnParams,
+  LoadSubsetOptions,
   SyncConfig,
+  SyncMode,
   UpdateMutationFnParams,
   UtilsRecord,
 } from '@tanstack/db'
@@ -81,6 +83,8 @@ function convertPartial<
   ) as OutputType
 }
 
+export type TrailBaseSyncMode = SyncMode
+
 /**
  * Configuration interface for Trailbase Collection
  */
@@ -90,13 +94,19 @@ export interface TrailBaseCollectionConfig<
   TKey extends string | number = string | number,
 > extends Omit<
   BaseCollectionConfig<TItem, TKey>,
-  `onInsert` | `onUpdate` | `onDelete`
+  `onInsert` | `onUpdate` | `onDelete` | `syncMode`
 > {
   /**
    * Record API name
    */
   recordApi: RecordApi<TRecord>
 
+  /**
+   * The mode of sync to use for the collection.
+   * @default `eager`
+   */
+  syncMode?: TrailBaseSyncMode
+
   parse: Conversions<TRecord, TItem>
   serialize: Conversions<TItem, TRecord>
 }
@@ -125,6 +135,9 @@ export function trailBaseCollectionOptions<
 
   const seenIds = new Store(new Map<string, number>())
 
+  const internalSyncMode = config.syncMode ?? `eager`
+  let fullSyncCompleted = false
+
   const awaitIds = (
     ids: Array<string>,
     timeout: number = 120 * 1000,
@@ -165,44 +178,79 @@ export function trailBaseCollectionOptions<
     sync: (params: SyncParams) => {
       const { begin, write, commit, markReady } = params
 
-      // Initial fetch.
-      async function initialFetch() {
-        const limit = 256
-        let response = await config.recordApi.list({
-          pagination: {
-            limit,
-          },
-        })
-        let cursor = response.cursor
-        let got = 0
-
-        begin()
+      // NOTE: We cache cursors from prior fetches. TanStack/db expects that
+      // cursors can be derived from a key, which is not true for TB, since
+      // cursors are encrypted. This is leaky and therefore not ideal.
+      const cursors = new Map<string | number, string>()
+
+      // Load (more) data.
+      async function load(opts: LoadSubsetOptions) {
+        const lastKey = opts.cursor?.lastKey
+        let cursor: string | undefined =
+          lastKey !== undefined ? cursors.get(lastKey) : undefined
+        let offset: number | undefined =
+          (opts.offset ?? 0) > 0 ? opts.offset : undefined
+
+        const order: Array<string> | undefined = buildOrder(opts)
+        const filters: Array<FilterOrComposite> | undefined = buildFilters(
+          opts,
+          config,
+        )
+
+        let remaining: number = opts.limit ?? Number.MAX_VALUE
+        if (remaining <= 0) {
+          return
+        }
 
         while (true) {
+          const limit = Math.min(remaining, 256)
+          const response = await config.recordApi.list({
+            pagination: {
+              limit,
+              offset,
+              cursor,
+            },
+            order,
+            filters,
+          })
+
           const length = response.records.length
-          if (length === 0) break
+          if (length === 0) {
+            // Drained - read everything.
+            break
+          }
 
-          got = got + length
-          for (const item of response.records) {
+          begin()
+
+          for (let i = 0; i < Math.min(length, remaining); ++i) {
             write({
               type: `insert`,
-              value: parse(item),
+              value: parse(response.records[i]!),
             })
           }
 
-          if (length < limit) break
+          commit()
 
-          response = await config.recordApi.list({
-            pagination: {
-              limit,
-              cursor,
-              offset: cursor === undefined ? got : undefined,
-            },
-          })
-          cursor = response.cursor
-        }
+          remaining -= length
 
-        commit()
+          // Drained or read enough.
+          if (length < limit || remaining <= 0) {
+            if (response.cursor) {
+              cursors.set(
+                getKey(parse(response.records.at(-1)!)),
+                response.cursor,
+              )
+            }
+            break
+          }
+
+          // Update params for next iteration.
+          if (offset !== undefined) {
+            offset += length
+          } else {
+            cursor = response.cursor
+          }
+        }
       }
 
       // Afterwards subscribe.
@@ -251,7 +299,12 @@ export function trailBaseCollectionOptions<
         listen(reader)
 
         try {
-          await initialFetch()
+          // Eager mode: perform initial fetch to populate everything
+          if (internalSyncMode === `eager`) {
+            // Load everything on initial load.
+            await load({})
+            fullSyncCompleted = true
+          }
         } catch (e) {
           cancelEventReader()
           throw e
@@ -285,9 +338,26 @@ export function trailBaseCollectionOptions<
       }
 
       start()
+
+      // Eager mode doesn't need subset loading
+      if (internalSyncMode === `eager`) {
+        return
+      }
+
+      return {
+        loadSubset: load,
+        getSyncMetadata: () =>
+          ({
+            syncMode: internalSyncMode,
+          }) as const,
+      }
     },
     // Expose the getSyncMetadata function
-    getSyncMetadata: undefined,
+    getSyncMetadata: () =>
+      ({
+        syncMode: internalSyncMode,
+        fullSyncComplete: fullSyncCompleted,
+      }) as const,
   }
 
   return {
@@ -356,3 +426,107 @@ export function trailBaseCollectionOptions<
     },
   }
 }
+
+function buildOrder(opts: LoadSubsetOptions): undefined | Array<string> {
+  return opts.orderBy
+    ?.map((o) => {
+      switch (o.expression.type) {
+        case 'ref': {
+          const field = o.expression.path[0]
+          if (o.compareOptions.direction == 'asc') {
+            return `+${field}`
+          }
+          return `-${field}`
+        }
+        default: {
+          console.warn(
+            'Skipping unsupported order clause:',
+            JSON.stringify(o.expression),
+          )
+          return undefined
+        }
+      }
+    })
+    .filter((f) => f !== undefined)
+}
+
+function buildCompareOp(name: string): CompareOp | undefined {
+  switch (name) {
+    case 'eq':
+      return 'equal'
+    case 'ne':
+      return 'notEqual'
+    case 'gt':
+      return 'greaterThan'
+    case 'gte':
+      return 'greaterThanEqual'
+    case 'lt':
+      return 'lessThan'
+    case 'lte':
+      return 'lessThanEqual'
+    default:
+      return undefined
+  }
+}
+
+function buildFilters<
+  TItem extends ShapeOf<TRecord>,
+  TRecord extends ShapeOf<TItem> = TItem,
+  TKey extends string | number = string | number,
+>(
+  opts: LoadSubsetOptions,
+  config: TrailBaseCollectionConfig<TItem, TRecord, TKey>,
+): undefined | Array<FilterOrComposite> {
+  const where = opts.where
+  if (where === undefined) {
+    return undefined
+  }
+
+  function serializeValue<T = any>(column: string, value: T): string {
+    const convert = (config.serialize as any)[column]
+    if (convert) {
+      return `${convert(value)}`
+    }
+
+    if (typeof value === 'boolean') {
+      return value ? '1' : '0'
+    }
+
+    return `${value}`
+  }
+
+  switch (where.type) {
+    case 'func': {
+      const field = where.args[0]
+      const val = where.args[1]
+
+      const op = buildCompareOp(where.name)
+      if (op === undefined) {
+        break
+      }
+
+      if (field?.type === 'ref' && val?.type === 'val') {
+        const column = field.path.at(0)
+        if (column) {
+          const f = [
+            {
+              column: field.path.at(0) ?? '',
+              op,
+              value: serializeValue(column, val.value),
+            },
+          ]
+
+          return f
+        }
+      }
+      break
+    }
+    case 'ref':
+    case 'val':
+      break
+  }
+
+  console.warn('where clause which is not (yet) supported', opts.where)
+
+  return undefined
+}
diff --git a/packages/trailbase-db-collection/tsconfig.json b/packages/trailbase-db-collection/tsconfig.json
index f3c0ea369..eac958767 100644
--- a/packages/trailbase-db-collection/tsconfig.json
+++ b/packages/trailbase-db-collection/tsconfig.json
@@ -16,6 +16,6 @@
       "@tanstack/store": ["../store/src"]
     }
   },
-  "include": ["src", "tests", "vite.config.ts"],
+  "include": ["src", "tests", "e2e", "vite.config.ts", "vitest.e2e.config.ts"],
   "exclude": ["node_modules", "dist"]
 }
diff --git a/packages/trailbase-db-collection/vitest.e2e.config.ts b/packages/trailbase-db-collection/vitest.e2e.config.ts
new file mode 100644
index 000000000..3418e292a
--- /dev/null
+++ b/packages/trailbase-db-collection/vitest.e2e.config.ts
@@ -0,0 +1,24 @@
+import { resolve } from 'node:path'
+import { defineConfig } from 'vitest/config'
+
+const packagesDir = resolve(__dirname, '..')
+
+export default defineConfig({
+  test: {
+    include: [`e2e/**/*.e2e.test.ts`],
+    globalSetup: `./e2e/global-setup.ts`,
+    fileParallelism: false, // Critical for shared database
+    testTimeout: 30000,
+    environment: `jsdom`,
+  },
+  resolve: {
+    alias: {
+      '@tanstack/db': resolve(packagesDir, 'db/src/index.ts'),
+      '@tanstack/db-ivm': resolve(packagesDir, 'db-ivm/src/index.ts'),
+      '@tanstack/db-collection-e2e': resolve(
+        packagesDir,
+        'db-collection-e2e/src/index.ts',
+      ),
+    },
+  },
+})
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 6f9ab0730..b9a874f2e 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -1014,12 +1014,18 @@ importers:
         specifier: '>=4.7'
         version: 5.9.3
     devDependencies:
+      '@tanstack/db-collection-e2e':
+        specifier: workspace:*
+        version: link:../db-collection-e2e
       '@types/debug':
         specifier: ^4.1.12
         version: 4.1.12
       '@vitest/coverage-istanbul':
         specifier: ^3.2.4
         version: 3.2.4(vitest@3.2.4)
+      uuid:
+        specifier: ^13.0.0
+        version: 13.0.0
 
   packages/vue-db:
     dependencies: