diff --git a/integration-tests/tests/src/tests/sync/asymmetric.ts b/integration-tests/tests/src/tests/sync/asymmetric.ts
index e314e6b128..4ee60f2dea 100644
--- a/integration-tests/tests/src/tests/sync/asymmetric.ts
+++ b/integration-tests/tests/src/tests/sync/asymmetric.ts
@@ -45,24 +45,41 @@ describe.skipIf(environment.missingServer, "Asymmetric sync", function () {
       },
     });
 
-    it("Schema with asymmetric = true and embedded = false", function () {
+    it("Schema with asymmetric = true and embedded = false", function (this: RealmContext) {
       const schema = this.realm.schema;
       expect(schema.length).to.equal(1);
-      expect(schema[0].asymmetric).to.equal(true);
-      expect(schema[0].embedded).to.equal(false);
+      expect(schema[0].asymmetric).to.be.true;
+      expect(schema[0].embedded).to.be.false;
     });
 
-    it("creating an object for an asymmetric schema returns undefined", function () {
+    it("creating an object for an asymmetric schema returns undefined", function (this: RealmContext) {
       this.realm.write(() => {
-        const returnValue = this.realm.create(PersonSchema.name, { _id: new BSON.ObjectId(), name: "Joe", age: 12 });
-        expect(returnValue).to.equal(undefined);
+        const returnValue = this.realm.create(PersonSchema.name, {
+          _id: new BSON.ObjectId(),
+          name: "Joe",
+          age: 12,
+        });
+        expect(returnValue).to.be.undefined;
       });
     });
 
-    it("an asymmetric schema cannot be queried", function () {
+    it("an asymmetric schema cannot be queried through 'objects()'", function (this: RealmContext) {
       expect(() => {
         this.realm.objects(PersonSchema.name);
-      }).to.throw("You cannot query an asymmetric class.");
+      }).to.throw("You cannot query an asymmetric object.");
+    });
+
+    it("an asymmetric schema cannot be queried through 'objectForPrimaryKey()'", function (this: RealmContext) {
+      expect(() => {
+        this.realm.objectForPrimaryKey(PersonSchema.name, new BSON.ObjectId());
+      }).to.throw("You cannot query an asymmetric object.");
+    });
+
+    it("an asymmetric schema cannot be queried through '_objectForObjectKey()'", function (this: RealmContext) {
+      expect(() => {
+        // A valid objectKey is not needed for this test
+        this.realm._objectForObjectKey(PersonSchema.name, "12345");
+      }).to.throw("You cannot query an asymmetric object.");
     });
   });
 });
diff --git a/integration-tests/tests/src/tests/sync/flexible.ts b/integration-tests/tests/src/tests/sync/flexible.ts
index 534103b72c..024b093b97 100644
--- a/integration-tests/tests/src/tests/sync/flexible.ts
+++ b/integration-tests/tests/src/tests/sync/flexible.ts
@@ -29,7 +29,7 @@
 // fraction too long.
 
 import { expect } from "chai";
-import Realm, { BSON, ClientResetMode, FlexibleSyncConfiguration, SessionStopPolicy } from "realm";
+import { BSON, ClientResetMode, FlexibleSyncConfiguration, Realm, SessionStopPolicy } from "realm";
 
 import { authenticateUserBefore, importAppBefore, openRealmBeforeEach } from "../../hooks";
 import { DogSchema, IPerson, PersonSchema } from "../../schemas/person-and-dog-with-object-ids";
@@ -121,6 +121,7 @@ async function addSubscriptionAndSync<T>(
 }
 
 describe.skipIf(environment.missingServer, "Flexible sync", function () {
+  this.timeout(60_000); // TODO: Temporarily hardcoded until envs are set up.
   importAppBefore("with-db-flx");
   authenticateUserBefore();
   openRealmBeforeEach({
@@ -164,7 +165,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
         }).to.throw("'partitionValue' cannot be specified when flexible sync is enabled");
       });
 
-      it("accepts { flexible: false } and a partition value", function () {
+      it("does not accept { flexible: false } and a partition value", function () {
         expect(() => {
           // @ts-expect-error This is not a compatible configuration anymore and will cause a typescript error
           new Realm({
@@ -175,7 +176,9 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
               partitionValue: "test",
             },
           });
-        }).to.not.throw();
+        }).to.throw(
+          "'flexible' can only be specified to enable flexible sync. To enable flexible sync, remove 'partitionValue' and set 'flexible' to true",
+        );
       });
 
       it("accepts { flexible: undefined } and a partition value", function () {
@@ -208,16 +211,12 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
         }
 
         describe("error", function () {
-          afterEach(function () {
-            Realm.deleteFile(this.config);
-          });
-
           it("throws an error if no update function is provided", async function (this: RealmContext) {
             // @ts-expect-error Intentionally testing the wrong type
             this.config = getConfig(this.user, {});
 
             await expect(Realm.open(this.config)).to.be.rejectedWith(
-              "update must be of type 'function', got (undefined)",
+              "Expected 'initialSubscriptions.update' on realm sync configuration to be a function, got undefined",
             );
           });
 
@@ -228,7 +227,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
             });
 
             await expect(Realm.open(this.config)).to.be.rejectedWith(
-              "update must be of type 'function', got (undefined)",
+              "Expected 'initialSubscriptions.update' on realm sync configuration to be a function, got undefined",
             );
           });
 
@@ -238,7 +237,9 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
               update: "Person",
             });
 
-            await expect(Realm.open(this.config)).to.be.rejectedWith("update must be of type 'function', got (Person)");
+            await expect(Realm.open(this.config)).to.be.rejectedWith(
+              "Expected 'initialSubscriptions.update' on realm sync configuration to be a function, got a string",
+            );
           });
 
           it("throws an error if `rerunOnOpen` is not a boolean", async function (this: RealmContext) {
@@ -250,7 +251,9 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
               rerunOnOpen: "yes please",
             });
 
-            await expect(Realm.open(this.config)).to.be.rejectedWith(/rerunOnOpen must be of type 'boolean', got.*/);
+            await expect(Realm.open(this.config)).to.be.rejectedWith(
+              "Expected 'initialSubscriptions.rerunOnOpen' on realm sync configuration to be a boolean, got a string",
+            );
           });
         });
 
@@ -587,20 +590,53 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
       });
 
       describe("array-like access", function () {
+        async function addThreeSubscriptions(this: RealmContext) {
+          addSubscriptionForPerson(this.realm);
+          await addSubscriptionAndSync(this.realm, this.realm.objects(FlexiblePersonSchema.name).filtered("age > 10"));
+          return await addSubscriptionAndSync(
+            this.realm,
+            this.realm.objects(FlexiblePersonSchema.name).filtered("age < 50"),
+          );
+        }
+
         it("returns an empty array if there are no subscriptions", function (this: RealmContext) {
           const subs = this.realm.subscriptions;
           expect(subs).to.have.length(0);
         });
 
-        it("returns an array of Subscription objects", async function (this: RealmContext) {
-          addSubscriptionForPerson(this.realm);
-          const { subs } = await addSubscriptionAndSync(
-            this.realm,
-            this.realm.objects(FlexiblePersonSchema.name).filtered("age > 10"),
-          );
+        it("accesses a SubscriptionSet using index operator", async function (this: RealmContext) {
+          const { subs } = await addThreeSubscriptions.call(this);
 
-          expect(subs).to.have.length(2);
-          expect(subs.every((s) => s instanceof Realm.App.Sync.Subscription)).to.be.true;
+          expect(subs).to.have.length(3);
+          expect(subs[0]).to.be.instanceOf(Realm.App.Sync.Subscription);
+          expect(subs[1]).to.be.instanceOf(Realm.App.Sync.Subscription);
+          expect(subs[2]).to.be.instanceOf(Realm.App.Sync.Subscription);
+        });
+
+        it("spreads a SubscriptionSet using spread operator", async function (this: RealmContext) {
+          const { subs } = await addThreeSubscriptions.call(this);
+
+          const spreadSubs = [...subs];
+          expect(spreadSubs).to.have.length(3);
+          expect(spreadSubs.every((s) => s instanceof Realm.App.Sync.Subscription)).to.be.true;
+        });
+
+        it("iterates over a SubscriptionSet using for-of loop", async function (this: RealmContext) {
+          const { subs } = await addThreeSubscriptions.call(this);
+
+          let numSubs = 0;
+          for (const sub of subs) {
+            expect(sub).to.be.an.instanceOf(Realm.App.Sync.Subscription);
+            numSubs++;
+          }
+          expect(numSubs).to.equal(3);
+        });
+
+        it("iterates over a SubscriptionSet using 'Object.keys()'", async function (this: RealmContext) {
+          const { subs } = await addThreeSubscriptions.call(this);
+
+          // Object.keys() always returns an array of strings.
+          expect(Object.keys(subs)).deep.equals(["0", "1", "2"]);
         });
 
         it("is an immutable snapshot of the subscriptions from when it was called", async function (this: RealmContext) {
@@ -862,7 +898,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
         it("passes a MutableSubscriptionSet instance as an argument", async function (this: RealmContext) {
           const subs = this.realm.subscriptions;
           await subs.update((mutableSubs) => {
-            expect(mutableSubs).to.be.instanceOf(Realm.App.Sync.MutableSubscriptionSet);
+            expect(mutableSubs).to.be.an.instanceOf(Realm.App.Sync.MutableSubscriptionSet);
           });
         });
 
@@ -1124,7 +1160,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
                 mutableSubs.add(query, { name: "test", throwOnUpdate: true });
               }),
             ).to.be.rejectedWith(
-              "A subscription with the name 'test' already exists but has a different query. If you meant to update it, remove `throwOnUpdate: true` from the subscription options.",
+              "A subscription with the name 'test' already exists but has a different query. If you meant to update it, remove 'throwOnUpdate: true' from the subscription options.",
             );
 
             expect(subs.findByQuery(query)).to.be.null;
@@ -1411,14 +1447,14 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
       const { id } = await addPersonAndWaitForUpload(realm);
 
       const newRealm = await closeAndReopenRealm(realm, config);
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.null;
 
       await newRealm.subscriptions.update((mutableSubs) => subsUpdateFn(mutableSubs, newRealm));
 
       return { id, newRealm };
     }
 
-    it.skip("syncs added items to a subscribed collection", async function (this: RealmContext) {
+    it("syncs added items to a subscribed collection", async function (this: RealmContext) {
       const { id, newRealm } = await addPersonAndResyncWithSubscription(
         this.realm,
         this.config,
@@ -1427,10 +1463,11 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
         },
       );
 
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.null;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.null;
     });
 
-    it.skip("syncs added items to a subscribed collection with a filter", async function (this: RealmContext) {
+    it("syncs added items to a subscribed collection with a filter", async function (this: RealmContext) {
       const { id, newRealm } = await addPersonAndResyncWithSubscription(
         this.realm,
         this.config,
@@ -1439,10 +1476,10 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
         },
       );
 
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.null;
     });
 
-    it.skip("does not sync added items not matching the filter", async function (this: RealmContext) {
+    it("does not sync added items not matching the filter", async function (this: RealmContext) {
       const { id, newRealm } = await addPersonAndResyncWithSubscription(
         this.realm,
         this.config,
@@ -1451,7 +1488,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
         },
       );
 
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.null;
     });
 
     // TODO: Probably remove this as it is testing old functionality
@@ -1463,7 +1500,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
           mutableSubs.add(realm.objects(FlexiblePersonSchema.name).filtered("age < 30"));
         },
       );
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.null;
 
       const subs = newRealm.subscriptions;
       await subs.update((mutableSubs) => {
@@ -1471,7 +1508,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
       });
 
       newRealm.addListener("change", () => {
-        expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.undefined;
+        expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.null;
       });
     });
 
@@ -1484,7 +1521,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
           mutableSubs.add(realm.objects(FlexiblePersonSchema.name).filtered("age < 30"));
         },
       );
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.null;
 
       const subs = newRealm.subscriptions;
       await subs.update((mutableSubs) => {
@@ -1493,11 +1530,11 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
       });
 
       newRealm.addListener("change", () => {
-        expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.undefined;
+        expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.null;
       });
     });
 
-    it.skip("stops syncing items when a subscription is removed (but other subscriptions still exist)", async function (this: RealmContext) {
+    it("stops syncing items when a subscription is removed (but other subscriptions still exist)", async function (this: RealmContext) {
       const { id, newRealm } = await addPersonAndResyncWithSubscription(
         this.realm,
         this.config,
@@ -1506,7 +1543,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
           mutableSubs.add(realm.objects(FlexiblePersonSchema.name).filtered("age > 50"));
         },
       );
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.null;
 
       const subs = newRealm.subscriptions;
       await subs.update((mutableSubs) => {
@@ -1514,11 +1551,11 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
       });
 
       newRealm.addListener("change", () => {
-        expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.undefined;
+        expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.null;
       });
     });
 
-    it.skip("stops syncing items when all subscriptions are removed", async function (this: RealmContext) {
+    it("stops syncing items when all subscriptions are removed", async function (this: RealmContext) {
       const { id, newRealm } = await addPersonAndResyncWithSubscription(
         this.realm,
         this.config,
@@ -1526,7 +1563,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
           mutableSubs.add(realm.objects(FlexiblePersonSchema.name));
         },
       );
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.null;
 
       const subs = newRealm.subscriptions;
       await subs.update((mutableSubs) => {
@@ -1534,11 +1571,11 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
       });
 
       newRealm.addListener("change", () => {
-        expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.undefined;
+        expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.null;
       });
     });
 
-    it.skip("stops syncing items if the filter changes to not match some items", async function (this: RealmContext) {
+    it("stops syncing items if the filter changes to not match some items", async function (this: RealmContext) {
       const { id, newRealm } = await addPersonAndResyncWithSubscription(
         this.realm,
         this.config,
@@ -1546,7 +1583,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
           mutableSubs.add(realm.objects(FlexiblePersonSchema.name).filtered("age > 30"));
         },
       );
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.not.be.null;
 
       const subs = newRealm.subscriptions;
       await subs.update((mutableSubs) => {
@@ -1554,7 +1591,7 @@ describe.skipIf(environment.missingServer, "Flexible sync", function () {
         mutableSubs.add(newRealm.objects(FlexiblePersonSchema.name).filtered("age < 30"));
       });
 
-      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.undefined;
+      expect(newRealm.objectForPrimaryKey(FlexiblePersonSchema.name, id)).to.be.null;
     });
 
     // TODO test more complex integration scenarios, e.g. links, embedded objects, collections, complex queries
diff --git a/integration-tests/tests/src/tests/sync/sync-as-local.ts b/integration-tests/tests/src/tests/sync/sync-as-local.ts
index 017a316afc..8835537085 100644
--- a/integration-tests/tests/src/tests/sync/sync-as-local.ts
+++ b/integration-tests/tests/src/tests/sync/sync-as-local.ts
@@ -22,7 +22,8 @@ import { Realm } from "realm";
 import { PersonSchema, IPerson } from "../../schemas/person-and-dog-with-object-ids";
 import { authenticateUserBefore, importAppBefore, openRealmBefore } from "../../hooks";
 
-describe.skipIf(environment.missingServer, "Synced Realm as local", () => {
+describe.skipIf(environment.missingServer, "Synced Realm as local", function () {
+  this.timeout(60_000); // TODO: Temporarily hardcoded until envs are set up.
   importAppBefore("with-db-flx");
   authenticateUserBefore();
   openRealmBefore({
@@ -48,13 +49,13 @@ describe.skipIf(environment.missingServer, "Synced Realm as local", () => {
     });
   });
 
-  it("opens when `sync: true`", function (this: RealmContext) {
+  it("opens when `openSyncedRealmLocally: true`", function (this: RealmContext) {
     // Close the synced Realm
     const realmPath = this.realm.path;
     this.realm.close();
     // Re-open as local Realm
-    // @ts-expect-error Using sync: true is an undocumented API
-    this.realm = new Realm({ path: realmPath, sync: true });
+    // @ts-expect-error Using `openSyncedRealmLocally: true` is an internal API
+    this.realm = new Realm({ path: realmPath, openSyncedRealmLocally: true });
     expect(this.realm.schema[0].name).equals("Person");
     console.log(this.realm.objects<IPerson>("Person").length);
     expect(this.realm.objects<IPerson>("Person")[0].name).equals("Alice");
diff --git a/packages/bindgen/spec.yml b/packages/bindgen/spec.yml
index 46e875f231..cf434023b5 100644
--- a/packages/bindgen/spec.yml
+++ b/packages/bindgen/spec.yml
@@ -381,6 +381,10 @@ records:
         type: bool
         default: false
       sync_config: Nullable<std::shared_ptr<SyncConfig>>
+      # Used internally as a flag for opening a synced realm locally.
+      force_sync_history:
+        type: bool
+        default: false
       # Unclear if migration_function and initialization_function should be marked as off_thread.
       # Existing JS sdk treats them as not.
       migration_function: 'Nullable<std::function<(old_realm: SharedRealm, new_realm: SharedRealm, new_schema_handle: IgnoreArgument<Schema&>) -> void>>'
@@ -615,6 +619,9 @@ classes:
       simulate_sync_error: '(session: SyncSession&, code: int, message: StringData, type: StringData, is_fatal: bool)'
       consume_thread_safe_reference_to_shared_realm: '(tsr: ThreadSafeReference) -> SharedRealm'
       file_exists: '(path: StringData) -> bool'
+      erase_subscription: '(subs: MutableSyncSubscriptionSet&, sub_to_remove: const SyncSubscription&) -> bool'
+      # This is added due to DescriptorOrdering not being exposed
+      get_results_description: '(results: const Results&) -> StringData'
 
   ConstTableRef:
     needsDeref: true
@@ -715,6 +722,7 @@ classes:
   Query:
     properties:
       get_table: ConstTableRef
+      get_description: StringData
     methods:
       count: () const -> count_t
 
@@ -1211,10 +1219,9 @@ classes:
       state: SyncSubscriptionSetState
       error_str: StringData
       size: count_t
-
     methods:
       make_mutable_copy: () const -> MutableSyncSubscriptionSet
-      get_state_change_notification: '(notify_when: SyncSubscriptionSetState, callback: AsyncCallback<(new_state: util::Optional<SyncSubscriptionSetState>, err: util::Optional<Status>)>) off_thread const'
+      get_state_change_notification: '(notify_when: SyncSubscriptionSetState, callback: AsyncCallback<(new_state: util::Optional<SyncSubscriptionSetState>, err: util::Optional<Status>) off_thread>) const'
       at: '(index: count_t) const -> const SyncSubscription&'
       find:
         - sig: '(name: StringData) const -> Nullable<const SyncSubscription*>'
diff --git a/packages/bindgen/src/realm_js_helpers.h b/packages/bindgen/src/realm_js_helpers.h
index bd75d54cab..12e529b149 100644
--- a/packages/bindgen/src/realm_js_helpers.h
+++ b/packages/bindgen/src/realm_js_helpers.h
@@ -3,9 +3,11 @@
 #include "realm/binary_data.hpp"
 #include "realm/object-store/object_store.hpp"
 #include "realm/object-store/sync/sync_session.hpp"
+#include "realm/object_id.hpp"
 #include "realm/query.hpp"
 #include "realm/sync/client_base.hpp"
 #include "realm/sync/protocol.hpp"
+#include "realm/sync/subscriptions.hpp"
 #include "realm/util/base64.hpp"
 #include "realm/util/file.hpp"
 #include "realm/util/logger.hpp"
@@ -228,6 +230,25 @@ struct Helpers {
     static bool file_exists(const StringData& path) {
         return realm::util::File::exists(path);
     }
+
+    static bool erase_subscription(sync::MutableSubscriptionSet& subs, const sync::Subscription& sub_to_remove) {
+        auto it = std::find_if(subs.begin(), subs.end(), [&](const auto& sub) {
+            return sub.id == sub_to_remove.id;
+        });
+
+        if (it == subs.end()) {
+            return false;
+        }
+        subs.erase(it);
+
+        return true;
+    }
+
+    static std::string get_results_description(const Results& results) {
+        const auto& query = results.get_query();
+
+        return query.get_description() + ' ' + results.get_descriptor_ordering().get_description(query.get_table());
+    }
 };
 
 struct ObjectChangeSet {
diff --git a/packages/realm/src/Configuration.ts b/packages/realm/src/Configuration.ts
index 976f6787c4..717a7a55f1 100644
--- a/packages/realm/src/Configuration.ts
+++ b/packages/realm/src/Configuration.ts
@@ -17,19 +17,14 @@
 ////////////////////////////////////////////////////////////////////////////
 
 import {
-  CanonicalObjectSchema,
-  CanonicalObjectSchemaProperty,
-  DefaultObject,
   ObjectSchema,
-  ObjectSchemaProperty,
   Realm,
-  RealmObject,
   RealmObjectConstructor,
   SyncConfiguration,
+  TypeAssertionError,
   assert,
-  ClientResetConfig,
-  ClientResetMode,
-  ErrorCallback,
+  validateRealmSchema,
+  validateSyncConfiguration,
 } from "./internal";
 
 // export type Configuration = ConfigurationWithSync | ConfigurationWithoutSync;
@@ -47,6 +42,7 @@ type BaseConfiguration = {
   readOnly?: boolean;
   fifoFilesFallbackPath?: string;
   sync?: SyncConfiguration;
+  /**@internal */ openSyncedRealmLocally?: true;
   shouldCompact?: (totalBytes: number, usedBytes: number) => boolean;
   deleteRealmIfMigrationNeeded?: boolean;
   disableFormatUpgrade?: boolean;
@@ -197,155 +193,78 @@ type BaseConfiguration = {
 //   disableFormatUpgrade?: boolean;
 // };
 
-// Need to use `CanonicalObjectSchema` rather than `ObjectSchema` due to some
-// integration tests using `openRealmHook()`. That function sets `this.realm`
-// to the opened realm whose schema is a `CanonicalObjectSchema[]`. Consequently,
-// the key `"ctor"` (which doesn't exist on `ObjectSchema`) also needs to be allowed.
-const OBJECT_SCHEMA_KEYS = new Set<keyof CanonicalObjectSchema>([
-  "name",
-  "primaryKey",
-  "embedded",
-  "asymmetric",
-  "properties",
-  // Not part of `ObjectSchema`
-  "ctor",
-]);
-
-// Need to use `CanonicalObjectSchemaProperty` rather than `ObjectSchemaProperty`
-// due to the same reasons as above.
-const PROPERTY_SCHEMA_KEYS = new Set<keyof CanonicalObjectSchemaProperty>([
-  "type",
-  "objectType",
-  "property",
-  "default",
-  "optional",
-  "indexed",
-  "mapTo",
-  // Not part of `ObjectSchemaProperty`
-  "name",
-]);
-
 /**
  * Validate the fields of a user-provided Realm configuration.
+ * @internal
  */
 export function validateConfiguration(config: unknown): asserts config is Configuration {
-  assert.object(config);
-  const { path, schema, onMigration, sync } = config;
-  if (typeof onMigration !== "undefined") {
-    assert.function(onMigration, "migration");
+  assert.object(config, "realm configuration", { allowArrays: false });
+  const {
+    path,
+    schema,
+    schemaVersion,
+    inMemory,
+    readOnly,
+    fifoFilesFallbackPath,
+    sync,
+    openSyncedRealmLocally,
+    shouldCompact,
+    deleteRealmIfMigrationNeeded,
+    disableFormatUpgrade,
+    encryptionKey,
+    onMigration,
+  } = config;
+
+  if (path !== undefined) {
+    assert.string(path, "'path' on realm configuration");
+    assert(path.length > 0, "The path cannot be empty. Provide a path or remove the field.");
   }
-  if (typeof path === "string") {
-    assert(path.length > 0, "Expected a non-empty path or none at all");
+  if (schema !== undefined) {
+    validateRealmSchema(schema);
   }
-  if (onMigration && sync) {
-    throw new Error("Options 'onMigration' and 'sync' are mutually exclusive");
+  if (schemaVersion !== undefined) {
+    assert.number(schemaVersion, "'schemaVersion' on realm configuration");
+    assert(
+      schemaVersion >= 0 && Number.isInteger(schemaVersion),
+      "'schemaVersion' on realm configuration must be 0 or a positive integer.",
+    );
   }
-  if (schema) {
-    validateRealmSchema(schema);
+  if (inMemory !== undefined) {
+    assert.boolean(inMemory, "'inMemory' on realm configuration");
   }
-}
-
-/**
- * Validate the data types of the fields of a user-provided realm schema.
- */
-export function validateRealmSchema(realmSchema: unknown): asserts realmSchema is Configuration["schema"][] {
-  assert.array(realmSchema, "realm schema");
-  for (const objectSchema of realmSchema) {
-    validateObjectSchema(objectSchema);
+  if (readOnly !== undefined) {
+    assert.boolean(readOnly, "'readOnly' on realm configuration");
   }
-  // TODO: Assert that backlinks point to object schemas that are actually declared
-}
-
-/**
- * Validate the data types of the fields of a user-provided object schema.
- */
-export function validateObjectSchema(
-  objectSchema: unknown,
-): asserts objectSchema is RealmObjectConstructor | ObjectSchema {
-  // Schema is passed via a class based model (RealmObjectConstructor)
-  if (typeof objectSchema === "function") {
-    const clazz = objectSchema as unknown as DefaultObject;
-    // We assert this later, but want a custom error message
-    if (!(objectSchema.prototype instanceof RealmObject)) {
-      const schemaName = clazz.schema && (clazz.schema as DefaultObject).name;
-      if (typeof schemaName === "string" && schemaName !== objectSchema.name) {
-        throw new TypeError(`Class '${objectSchema.name}' (declaring '${schemaName}' schema) must extend Realm.Object`);
-      } else {
-        throw new TypeError(`Class '${objectSchema.name}' must extend Realm.Object`);
-      }
-    }
-    assert.object(clazz.schema, "schema static");
-    validateObjectSchema(clazz.schema);
+  if (fifoFilesFallbackPath !== undefined) {
+    assert.string(fifoFilesFallbackPath, "'fifoFilesFallbackPath' on realm configuration");
   }
-  // Schema is passed as an object (ObjectSchema)
-  else {
-    assert.object(objectSchema, "object schema", { allowArrays: false });
-    const { name: objectName, properties, primaryKey, asymmetric, embedded } = objectSchema;
-    assert.string(objectName, "'name' on object schema");
-    assert.object(properties, `'properties' on '${objectName}'`, { allowArrays: false });
-    if (primaryKey !== undefined) {
-      assert.string(primaryKey, `'primaryKey' on '${objectName}'`);
-    }
-    if (embedded !== undefined) {
-      assert.boolean(embedded, `'embedded' on '${objectName}'`);
-    }
-    if (asymmetric !== undefined) {
-      assert.boolean(asymmetric, `'asymmetric' on '${objectName}'`);
-    }
-
-    const invalidKeysUsed = filterInvalidKeys(objectSchema, OBJECT_SCHEMA_KEYS);
+  if (onMigration !== undefined) {
+    assert.function(onMigration, "'onMigration' on realm configuration");
+  }
+  if (sync !== undefined) {
+    assert(!onMigration, "The realm configuration options 'onMigration' and 'sync' cannot both be defined.");
+    validateSyncConfiguration(sync);
+  }
+  if (openSyncedRealmLocally !== undefined) {
+    // Internal use
     assert(
-      !invalidKeysUsed.length,
-      `Unexpected field(s) found on the schema for object '${objectName}': '${invalidKeysUsed.join("', '")}'.`,
+      openSyncedRealmLocally === true,
+      "'openSyncedRealmLocally' on realm configuration is only used internally and must be true if defined.",
     );
-
-    for (const propertyName in properties) {
-      const propertySchema = properties[propertyName];
-      const isUsingShorthand = typeof propertySchema === "string";
-      if (!isUsingShorthand) {
-        validatePropertySchema(objectName, propertyName, propertySchema);
-      }
-    }
   }
-}
-
-/**
- * Validate the data types of a user-provided property schema that ought to use the
- * relaxed object notation.
- */
-export function validatePropertySchema(
-  objectName: string,
-  propertyName: string,
-  propertySchema: unknown,
-): asserts propertySchema is ObjectSchemaProperty {
-  assert.object(propertySchema, `'${propertyName}' on '${objectName}'`, { allowArrays: false });
-  const { type, objectType, optional, property, indexed, mapTo } = propertySchema;
-  assert.string(type, `'${propertyName}.type' on '${objectName}'`);
-  if (objectType !== undefined) {
-    assert.string(objectType, `'${propertyName}.objectType' on '${objectName}'`);
+  if (shouldCompact !== undefined) {
+    assert.function(shouldCompact, "'shouldCompact' on realm configuration");
   }
-  if (optional !== undefined) {
-    assert.boolean(optional, `'${propertyName}.optional' on '${objectName}'`);
+  if (deleteRealmIfMigrationNeeded !== undefined) {
+    assert.boolean(deleteRealmIfMigrationNeeded, "'deleteRealmIfMigrationNeeded' on realm configuration");
   }
-  if (property !== undefined) {
-    assert.string(property, `'${propertyName}.property' on '${objectName}'`);
+  if (disableFormatUpgrade !== undefined) {
+    assert.boolean(disableFormatUpgrade, "'disableFormatUpgrade' on realm configuration");
   }
-  if (indexed !== undefined) {
-    assert.boolean(indexed, `'${propertyName}.indexed' on '${objectName}'`);
-  }
-  if (mapTo !== undefined) {
-    assert.string(mapTo, `'${propertyName}.mapTo' on '${objectName}'`);
+  if (encryptionKey !== undefined) {
+    assert(
+      encryptionKey instanceof ArrayBuffer || ArrayBuffer.isView(encryptionKey) || encryptionKey instanceof Int8Array,
+      `Expected 'encryptionKey' on realm configuration to be an ArrayBuffer, ArrayBufferView (Uint8Array), or Int8Array, got ${TypeAssertionError.deriveType(encryptionKey)}.`,
+    );
   }
-  const invalidKeysUsed = filterInvalidKeys(propertySchema, PROPERTY_SCHEMA_KEYS);
-  assert(
-    !invalidKeysUsed.length,
-    `Unexpected field(s) found on the schema for property '${propertyName}' on '${objectName}': '${invalidKeysUsed.join("', '")}'.`,
-  );
-}
-
-/**
- * Get the keys of an object that are not part of the provided valid keys.
- */
-function filterInvalidKeys(object: Record<string, unknown>, validKeys: Set<string>): string[] {
-  return Object.keys(object).filter((key) => !validKeys.has(key));
 }
diff --git a/packages/realm/src/ProgressRealmPromise.ts b/packages/realm/src/ProgressRealmPromise.ts
index 0795404c7a..f95a076347 100644
--- a/packages/realm/src/ProgressRealmPromise.ts
+++ b/packages/realm/src/ProgressRealmPromise.ts
@@ -23,6 +23,7 @@ import {
   ProgressNotificationCallback,
   PromiseHandle,
   Realm,
+  SubscriptionsState,
   TimeoutError,
   TimeoutPromise,
   assert,
@@ -36,12 +37,12 @@ type OpenBehavior = {
   timeOutBehavior?: OpenRealmTimeOutBehavior;
 };
 
-function determineBehavior(config: Configuration): OpenBehavior {
-  const { sync } = config;
-  if (!sync) {
+function determineBehavior(config: Configuration, realmExists: boolean): OpenBehavior {
+  const { sync, openSyncedRealmLocally } = config;
+  if (!sync || openSyncedRealmLocally) {
     return { openBehavior: OpenRealmBehaviorType.OpenImmediately };
   } else {
-    const configProperty = Realm.exists(config) ? "existingRealmFileBehavior" : "newRealmFileBehavior";
+    const configProperty = realmExists ? "existingRealmFileBehavior" : "newRealmFileBehavior";
     const configBehavior = sync[configProperty];
     if (configBehavior) {
       const { type, timeOut, timeOutBehavior } = configBehavior;
@@ -69,7 +70,11 @@ export class ProgressRealmPromise implements Promise<Realm> {
   constructor(config: Configuration) {
     try {
       validateConfiguration(config);
-      const { openBehavior: openBehavior, timeOut, timeOutBehavior } = determineBehavior(config);
+      // Calling `Realm.exists()` before `binding.Realm.getSynchronizedRealm()` is necessary to capture
+      // the correct value when this constructor was called since `binding.Realm.getSynchronizedRealm()`
+      // will open the realm. This is needed when calling the Realm constructor.
+      const realmExists = Realm.exists(config);
+      const { openBehavior, timeOut, timeOutBehavior } = determineBehavior(config, realmExists);
       if (openBehavior === OpenRealmBehaviorType.OpenImmediately) {
         const realm = new Realm(config);
         this.handle.resolve(realm);
@@ -79,17 +84,17 @@ export class ProgressRealmPromise implements Promise<Realm> {
         this.task
           .start()
           .then(async (tsr) => {
-            const realm = new Realm(config, binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr));
-
-            const initialSubscriptions = config.sync && config.sync.flexible ? config.sync.initialSubscriptions : false;
-            const realmExists = Realm.exists(config);
-            if (!initialSubscriptions || (!initialSubscriptions.rerunOnOpen && realmExists)) {
-              return realm;
+            const realm = new Realm(config, {
+              internal: binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr),
+              // Do not call `Realm.exists()` here in case the realm has been opened by this point in time.
+              realmExists,
+            });
+            if (config.sync?.flexible && !config.openSyncedRealmLocally) {
+              const { subscriptions } = realm;
+              if (subscriptions.state === SubscriptionsState.Pending) {
+                await subscriptions.waitForSynchronization();
+              }
             }
-            // TODO: Implement this once flexible sync gets implemented
-            // await realm.subscriptions.waitForSynchronization();
-            // TODO: Consider implementing adding the subscriptions here as well
-            throw new Error("'initialSubscriptions' is not yet supported");
             return realm;
           })
           .then(this.handle.resolve, this.handle.reject);
diff --git a/packages/realm/src/Realm.ts b/packages/realm/src/Realm.ts
index ba6ab77fcc..765ef77d9a 100644
--- a/packages/realm/src/Realm.ts
+++ b/packages/realm/src/Realm.ts
@@ -30,7 +30,9 @@ import {
   DefaultObject,
   Dictionary,
   EmailPasswordAuthClient,
+  FlexibleSyncConfiguration,
   INTERNAL,
+  InitialSubscriptions,
   List,
   MigrationCallback,
   ObjectSchema,
@@ -44,6 +46,7 @@ import {
   RealmObjectConstructor,
   RealmSet,
   Results,
+  SubscriptionSet,
   SyncSession,
   TypeAssertionError,
   Types,
@@ -93,6 +96,13 @@ function assertRealmEvent(name: RealmEventName): asserts name is RealmEvent {
   }
 }
 
+/** @internal */
+type InternalConfig = {
+  internal?: binding.Realm;
+  schemaExtras?: RealmSchemaExtra;
+  realmExists?: boolean;
+};
+
 export class Realm {
   public static Object = RealmObject;
   public static Collection = Collection;
@@ -139,6 +149,7 @@ export class Realm {
    * @throws {@link Error} If anything in the provided {@link config} is invalid.
    */
   public static deleteFile(config: Configuration): void {
+    validateConfiguration(config);
     const path = Realm.determinePath(config);
     fs.removeFile(path);
     fs.removeFile(path + ".lock");
@@ -292,7 +303,7 @@ export class Realm {
   }
 
   private static determinePath(config: Configuration): string {
-    if (config.path || !config.sync) {
+    if (config.path || !config.sync || config.openSyncedRealmLocally) {
       return Realm.normalizePath(config.path);
     } else {
       // TODO: Determine if it's okay to get the syncManager through the app instead of the user:
@@ -356,6 +367,7 @@ export class Realm {
         disableFormatUpgrade: config.disableFormatUpgrade,
         encryptionKey: Realm.determineEncryptionKey(config.encryptionKey),
         syncConfig: config.sync ? toBindingSyncConfig(config.sync) : undefined,
+        forceSyncHistory: config.openSyncedRealmLocally,
       },
     };
   }
@@ -387,8 +399,8 @@ export class Realm {
   ): binding.RealmConfig_Relaxed["migrationFunction"] {
     return (oldRealmInternal: binding.Realm, newRealmInternal: binding.Realm) => {
       try {
-        const oldRealm = new Realm(oldRealmInternal, schemaExtras);
-        const newRealm = new Realm(newRealmInternal, schemaExtras);
+        const oldRealm = new Realm(null, { internal: oldRealmInternal, schemaExtras });
+        const newRealm = new Realm(null, { internal: newRealmInternal, schemaExtras });
         onMigration(oldRealm, newRealm);
       } finally {
         oldRealmInternal.close();
@@ -439,21 +451,21 @@ export class Realm {
    */
   constructor(config: Configuration);
   /** @internal */
-  constructor(config: Configuration, internal: binding.Realm);
-  /** @internal */
-  constructor(internal: binding.Realm, schemaExtras?: RealmSchemaExtra);
-  constructor(arg: Configuration | binding.Realm | string = {}, secondArg?: object) {
-    if (arg instanceof binding.Realm) {
-      this.schemaExtras = (secondArg ?? {}) as RealmSchemaExtra;
-      this.internal = arg;
-    } else {
-      const config = typeof arg === "string" ? { path: arg } : arg;
+  constructor(config: Configuration | null, internalConfig: InternalConfig);
+  constructor(arg?: Configuration | string | null, internalConfig: InternalConfig = {}) {
+    const config = typeof arg === "string" ? { path: arg } : arg || {};
+    // Calling `Realm.exists()` before `binding.Realm.getSharedRealm()` is necessary to capture
+    // the correct value when this constructor was called since `binding.Realm.getSharedRealm()`
+    // will open the realm. This is needed when deciding whether to update initial subscriptions.
+    const realmExists = internalConfig.realmExists ?? Realm.exists(config);
+    if (arg !== null) {
+      assert(!internalConfig.schemaExtras, "Expected either a configuration or schemaExtras");
       validateConfiguration(config);
       const { bindingConfig, schemaExtras } = Realm.transformConfig(config);
       debug("open", bindingConfig);
       this.schemaExtras = schemaExtras;
-      assert(!secondArg || secondArg instanceof binding.Realm, "The realm constructor only takes a single argument");
-      this.internal = secondArg ?? binding.Realm.getSharedRealm(bindingConfig);
+
+      this.internal = internalConfig.internal ?? binding.Realm.getSharedRealm(bindingConfig);
 
       binding.Helpers.setBindingContext(this.internal, {
         didChange: (r) => {
@@ -470,6 +482,11 @@ export class Realm {
         },
       });
       RETURNED_REALMS.add(new binding.WeakRef(this.internal));
+    } else {
+      const { internal, schemaExtras } = internalConfig;
+      assert.instanceOf(internal, binding.Realm, "internal");
+      this.internal = internal;
+      this.schemaExtras = schemaExtras || {};
     }
 
     Object.defineProperties(this, {
@@ -489,6 +506,12 @@ export class Realm {
 
     const syncSession = this.internal.syncSession;
     this.syncSession = syncSession ? new SyncSession(syncSession) : null;
+
+    const initialSubscriptions = config.sync?.initialSubscriptions;
+    if (initialSubscriptions && !config.openSyncedRealmLocally) {
+      // Do not call `Realm.exists()` here in case the realm has been opened by this point in time.
+      this.handleInitialSubscriptions(initialSubscriptions, realmExists);
+    }
   }
 
   /**
@@ -577,10 +600,24 @@ export class Realm {
 
   /**
    * The latest set of flexible sync subscriptions.
-   * @throws {@link Error} If flexible sync is not enabled for this app
+   * @throws {@link Error} If flexible sync is not enabled for this app.
    */
-  get subscriptions(): any {
-    throw new Error("Not yet implemented");
+  get subscriptions(): SubscriptionSet {
+    const { syncConfig } = this.internal.config;
+    assert(
+      syncConfig,
+      "`subscriptions` can only be accessed if flexible sync is enabled, but sync is " +
+        "currently disabled for your app. Add a flexible sync config when opening the " +
+        "Realm, for example: { sync: { user, flexible: true } }.",
+    );
+    assert(
+      syncConfig.flxSyncRequested,
+      "`subscriptions` can only be accessed if flexible sync is enabled, but partition " +
+        "based sync is currently enabled for your Realm. Modify your sync config to remove any `partitionValue` " +
+        "and enable flexible sync, for example: { sync: { user, flexible: true } }",
+    );
+
+    return new SubscriptionSet(this, this.internal.latestSubscriptionSet);
   }
 
   /**
@@ -593,11 +630,11 @@ export class Realm {
     this.syncSession?.resetInternal();
   }
 
-  // TODO: Support embedded objects and asymmetric sync
+  // TODO: Support embedded objects
   // TODO: Rollback by deleting the object if any property assignment fails (fixing #2638)
   /**
-   * Create a new Realm object of the given type and with the specified properties. For object schemas annotated
-   * as asymmetric, no object is returned. The API for asymmetric object schema is subject to changes in the future.
+   * Create a new {@link Realm.Object} of the given type and with the specified properties. For objects marked asymmetric,
+   * `undefined` is returned. The API for asymmetric objects is subject to changes in the future.
    * @param type The type of Realm object to create.
    * @param values Property values for all required properties without a
    *   default value.
@@ -611,6 +648,7 @@ export class Realm {
    *       across devices are merged. For most use cases, the behavior will match the intuitive behavior of how
    *       changes should be merged, but if updating an entire object is considered an atomic operation, this mode
    *       should not be used.
+   * @returns A {@link Realm.Object} or `undefined` if the object is asymmetric.
    */
   create<T = DefaultObject>(type: string, values: RealmInsertionModel<T>, mode?: UpdateMode.Never): RealmObject<T> & T;
   create<T = DefaultObject>(
@@ -644,7 +682,9 @@ export class Realm {
     }
     this.internal.verifyOpen();
     const helpers = this.classes.getHelpers(type);
-    return RealmObject.create(this, values, mode, { helpers });
+    const realmObject = RealmObject.create(this, values, mode, { helpers });
+
+    return isAsymmetric(helpers.objectSchema) ? undefined : realmObject;
   }
 
   /**
@@ -707,9 +747,9 @@ export class Realm {
    * Searches for a Realm object by its primary key.
    * @param type The type of Realm object to search for.
    * @param primaryKey The primary key value of the object to search for.
-   * @throws {@link Error} If type passed into this method is invalid or if the object type did
-   * not have a {@link primaryKey} specified in the schema.
-   * @returns A Realm.Object or undefined if no object is found.
+   * @throws {@link Error} If type passed into this method is invalid, or if the object type did
+   *  not have a {@link primaryKey} specified in the schema, or if it was marked asymmetric.
+   * @returns A {@link Realm.Object} or `null` if no object is found.
    * @since 0.14.0
    */
   objectForPrimaryKey<T = DefaultObject>(type: string, primaryKey: T[keyof T]): (RealmObject<T> & T) | null;
@@ -720,6 +760,9 @@ export class Realm {
     if (!objectSchema.primaryKey) {
       throw new Error(`Expected a primary key on '${objectSchema.name}'`);
     }
+    if (isAsymmetric(objectSchema)) {
+      throw new Error("You cannot query an asymmetric object.");
+    }
     const table = binding.Helpers.getTable(this.internal, objectSchema.tableKey);
     const value = properties.get(objectSchema.primaryKey).toBinding(primaryKey, undefined);
     try {
@@ -742,17 +785,22 @@ export class Realm {
 
   /**
    * Returns all objects of the given {@link type} in the Realm.
-   * @param type The type of Realm objects to retrieve.
+   * @param type The type of Realm object to search for.
+   * @param objectKey The object key of the Realm object to search for.
    * @throws {@link Error} If type passed into this method is invalid or if the type is marked embedded or asymmetric.
-   * @returns Realm.Results that will live-update as objects are created and destroyed.
-   */
-  /**
+   * @returns A {@link Realm.Object} or `undefined` if the object key is not found.
    * @internal
    */
   _objectForObjectKey<T = DefaultObject>(type: string, objectKey: string): (RealmObject<T> & T) | undefined;
   _objectForObjectKey<T extends RealmObject>(type: Constructor<T>, objectKey: string): T | undefined;
   _objectForObjectKey<T extends RealmObject>(type: string | Constructor<T>, objectKey: string): T | undefined {
     const { objectSchema, wrapObject } = this.classes.getHelpers(type);
+    if (isEmbedded(objectSchema)) {
+      throw new Error("You cannot query an embedded object.");
+    } else if (isAsymmetric(objectSchema)) {
+      throw new Error("You cannot query an asymmetric object.");
+    }
+
     const table = binding.Helpers.getTable(this.internal, objectSchema.tableKey);
     try {
       const objKey = binding.stringToObjKey(objectKey);
@@ -768,14 +816,20 @@ export class Realm {
     }
   }
 
+  /**
+   * Returns all objects of the given {@link type} in the Realm.
+   * @param type The type of Realm objects to retrieve.
+   * @throws {@link Error} If type passed into this method is invalid or if the type is marked embedded or asymmetric.
+   * @returns Realm.Results that will live-update as objects are created, modified, and destroyed.
+   */
   objects<T>(type: string): Results<RealmObject & T>;
   objects<T extends RealmObject = RealmObject>(type: Constructor<T>): Results<T>;
   objects<T extends RealmObject = RealmObject>(type: string | Constructor<T>): Results<T> {
     const { objectSchema, wrapObject } = this.classes.getHelpers(type);
-    if (objectSchema.tableType === binding.TableType.Embedded) {
+    if (isEmbedded(objectSchema)) {
       throw new Error("You cannot query an embedded object.");
-    } else if (objectSchema.tableType === binding.TableType.TopLevelAsymmetric) {
-      throw new Error("You cannot query an asymmetric class.");
+    } else if (isAsymmetric(objectSchema)) {
+      throw new Error("You cannot query an asymmetric object.");
     }
 
     const table = binding.Helpers.getTable(this.internal, objectSchema.tableKey);
@@ -991,6 +1045,35 @@ export class Realm {
   ): ClassHelpers {
     return this.classes.getHelpers<T>(arg);
   }
+
+  /**
+   * Update subscriptions with the initial subscriptions if needed.
+   *
+   * @param initialSubscriptions The initial subscriptions.
+   * @param realmExists Whether the realm already exists.
+   */
+  private handleInitialSubscriptions(initialSubscriptions: InitialSubscriptions, realmExists: boolean): void {
+    const shouldUpdateSubscriptions = initialSubscriptions.rerunOnOpen || !realmExists;
+    if (shouldUpdateSubscriptions) {
+      this.subscriptions.updateNoWait(initialSubscriptions.update);
+    }
+  }
+}
+
+/**
+ * @param objectSchema The schema of the object.
+ * @returns `true` if the object is marked for asymmetric sync, otherwise `false`.
+ */
+function isAsymmetric(objectSchema: binding.ObjectSchema): boolean {
+  return objectSchema.tableType === binding.TableType.TopLevelAsymmetric;
+}
+
+/**
+ * @param objectSchema The schema of the object.
+ * @returns `true` if the object is marked as embedded, otherwise `false`.
+ */
+function isEmbedded(objectSchema: binding.ObjectSchema): boolean {
+  return objectSchema.tableType === binding.TableType.Embedded;
 }
 
 // Declare the Realm namespace for backwards compatibility
@@ -1009,6 +1092,8 @@ type BSONType = typeof BSON;
 type TypesType = typeof Types;
 type UserType = typeof User;
 type CredentialsType = typeof Credentials;
+type ConfigurationType = Configuration;
+type FlexibleSyncConfigurationType = FlexibleSyncConfiguration;
 
 // eslint-disable-next-line @typescript-eslint/no-namespace
 export namespace Realm {
@@ -1027,4 +1112,6 @@ export namespace Realm {
   export type Types = TypesType;
   export type User = UserType;
   export type Credentials = CredentialsType;
+  export type Configuration = ConfigurationType;
+  export type FlexibleSyncConfiguration = FlexibleSyncConfigurationType;
 }
diff --git a/packages/realm/src/Results.ts b/packages/realm/src/Results.ts
index 8e1afac56e..381f92ef9e 100644
--- a/packages/realm/src/Results.ts
+++ b/packages/realm/src/Results.ts
@@ -75,6 +75,10 @@ export class Results<T = unknown> extends OrderedCollection<T> {
     return this.internal.size();
   }
 
+  description(): string {
+    return binding.Helpers.getResultsDescription(this.internal);
+  }
+
   /**
    * Bulk update objects in the collection.
    * @param propertyName The name of the property.
diff --git a/packages/realm/src/app-services/BaseSubscriptionSet.ts b/packages/realm/src/app-services/BaseSubscriptionSet.ts
new file mode 100644
index 0000000000..1f714e021c
--- /dev/null
+++ b/packages/realm/src/app-services/BaseSubscriptionSet.ts
@@ -0,0 +1,234 @@
+////////////////////////////////////////////////////////////////////////////
+//
+// Copyright 2023 Realm Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+////////////////////////////////////////////////////////////////////////////
+
+import { MutableSubscriptionSet, Realm, Subscription, SubscriptionSet, assert, binding } from "../internal";
+
+/**
+ * Enum representing the state of a {@link SubscriptionSet}.
+ */
+export enum SubscriptionsState {
+  /**
+   * The subscription update has been persisted locally, but the server hasn't
+   * yet returned all the data that matched the updated subscription queries.
+   */
+  Pending = "pending",
+
+  /**
+   * The server has acknowledged the subscription and sent all the data that
+   * matched the subscription queries at the time the SubscriptionSet was
+   * updated. The server is now in steady-state synchronization mode where it
+   * will stream updates as they come.
+   */
+  Complete = "complete",
+
+  /**
+   * The server has returned an error and synchronization is paused for this
+   * Realm. To view the actual error, use `Subscriptions.error`.
+   *
+   * You can still use {@link SubscriptionSet.update} to update the subscriptions,
+   * and if the new update doesn't trigger an error, synchronization will be restarted.
+   */
+  Error = "error",
+
+  /**
+   * The SubscriptionSet has been superseded by an updated one. This typically means
+   * that someone has called {@link SubscriptionSet.update} on a different instance
+   * of the {@link SubscriptionSet}. You should not use a superseded SubscriptionSet,
+   * and instead obtain a new instance from {@link Realm.subscriptions}.
+   */
+  Superseded = "superseded",
+}
+
+const DEFAULT_PROPERTY_DESCRIPTOR: PropertyDescriptor = { configurable: true, enumerable: true, writable: false };
+const PROXY_HANDLER: ProxyHandler<BaseSubscriptionSet> = {
+  get(target, prop) {
+    if (Reflect.has(target, prop)) {
+      return Reflect.get(target, prop);
+    }
+    if (typeof prop === "string") {
+      const BASE = 10;
+      const index = Number.parseInt(prop, BASE);
+      // TODO: Consider catching an error from access out of bounds, instead of checking the length, to optimize for the hot path
+      if (index >= 0 && index < target.length) {
+        return target.get(index);
+      }
+    }
+  },
+  ownKeys(target) {
+    return Reflect.ownKeys(target).concat([...target.keys()].map(String));
+  },
+  getOwnPropertyDescriptor(target, prop) {
+    if (Reflect.has(target, prop)) {
+      return Reflect.getOwnPropertyDescriptor(target, prop);
+    }
+    if (typeof prop === "string") {
+      const BASE = 10;
+      const index = Number.parseInt(prop, BASE);
+      if (index >= 0 && index < target.length) {
+        return DEFAULT_PROPERTY_DESCRIPTOR;
+      }
+    }
+  },
+  // Not defining `set()` here will make e.g. `mySubscriptions[0] = someValue` a no-op
+  // if strict mode (`"use strict"`) is used, or throw a TypeError if it is not used.
+};
+
+/**
+ * Class representing the common functionality for the {@link SubscriptionSet} and
+ * {@link MutableSubscriptionSet} classes.
+ *
+ * The {@link Subscription}s in a SubscriptionSet can be accessed as an array, e.g.
+ * `realm.subscriptions[0]`. This array is readonly – SubscriptionSets can only be
+ * modified inside a {@link SubscriptionSet.update} callback.
+ */
+export abstract class BaseSubscriptionSet {
+  /**@internal */
+  protected constructor(/**@internal */ protected internal: binding.SyncSubscriptionSet) {
+    Object.defineProperties(this, {
+      internal: {
+        enumerable: false,
+        configurable: false,
+        // `internal` needs to be writable due to `SubscriptionSet.updateNoWait()`
+        // overwriting `this.internal` with the new committed set.
+        writable: true,
+      },
+    });
+    return new Proxy<BaseSubscriptionSet>(this, PROXY_HANDLER);
+  }
+
+  /**
+   * Whether there are no subscriptions in the set.
+   */
+  get isEmpty(): boolean {
+    return this.internal.size === 0;
+  }
+
+  /**
+   * The version of the SubscriptionSet. This is incremented every time a
+   * {@link SubscriptionSet.update} is applied.
+   */
+  get version(): number {
+    return Number(this.internal.version);
+  }
+
+  /**
+   * The state of the SubscriptionSet.
+   */
+  get state(): SubscriptionsState {
+    const state = this.internal.state;
+    switch (state) {
+      case binding.SyncSubscriptionSetState.Uncommitted:
+      case binding.SyncSubscriptionSetState.Pending:
+      case binding.SyncSubscriptionSetState.Bootstrapping:
+      case binding.SyncSubscriptionSetState.AwaitingMark:
+        return SubscriptionsState.Pending;
+      case binding.SyncSubscriptionSetState.Complete:
+        return SubscriptionsState.Complete;
+      case binding.SyncSubscriptionSetState.Error:
+        return SubscriptionsState.Error;
+      case binding.SyncSubscriptionSetState.Superseded:
+        return SubscriptionsState.Superseded;
+      default:
+        throw new Error(`Unsupported SubscriptionsState value: ${state}`);
+    }
+  }
+
+  /**
+   * If `state` is {@link SubscriptionsState.Error}, this will be a `string` representing
+   * why the SubscriptionSet is in an error state. It will be `null` if there is no error.
+   */
+  get error(): string | null {
+    return this.state === SubscriptionsState.Error ? this.internal.errorStr : null;
+  }
+
+  /**
+   * The number of subscriptions in the set.
+   */
+  get length(): number {
+    return this.internal.size;
+  }
+
+  /**
+   * Get a Subscription by index.
+   * (Needed by the ProxyHandler when the subscription set is accessed by index.)
+   *
+   * @param index The index.
+   * @returns The subscription.
+   * @internal
+   */
+  get(index: number): Subscription {
+    return new Subscription(this.internal.at(index));
+  }
+
+  /**
+   * Find a subscription by name.
+   *
+   * @param name The name to search for.
+   * @returns The named subscription, or `null` if the subscription is not found.
+   */
+  findByName(name: string): Subscription | null {
+    assert.string(name, "name");
+
+    const subscription = this.internal.findByName(name);
+    return subscription ? new Subscription(subscription) : null;
+  }
+
+  /**
+   * Find a subscription by query. Will match both named and unnamed subscriptions.
+   *
+   * @param query The query to search for, represented as a {@link Realm.Results} instance,
+   *  e.g. `Realm.objects("Cat").filtered("age > 10")`.
+   * @returns The subscription with the specified query, or `null` if the subscription is not found.
+   */
+  findByQuery<Subscription>(query: Realm.Results<Subscription & Realm.Object>): Subscription | null {
+    assert.instanceOf(query, Realm.Results, "query");
+
+    const subscription = this.internal.findByQuery(query.internal.query);
+    return subscription ? (new Subscription(subscription) as Subscription) : null; // TODO: Remove the type assertion into Subscription
+  }
+
+  /**
+   * Enables index accessing when combined with the proxy handler's `get()` method.
+   */
+  readonly [n: number]: Subscription;
+
+  /**
+   * Makes the subscription set iterable.
+   *
+   * @returns Iterable of each value in the set.
+   * @example
+   * for (const subscription of subscriptions) {
+   *   // ...
+   * }
+   */
+  *[Symbol.iterator](): IterableIterator<Subscription> {
+    for (const subscription of this.internal) {
+      yield new Subscription(subscription);
+    }
+  }
+
+  /**
+   * Get an iterator that contains each index in the subscription set.
+   */
+  *keys() {
+    const size = this.length;
+    for (let i = 0; i < size; i++) {
+      yield i;
+    }
+  }
+}
diff --git a/packages/realm/src/app-services/MutableSubscriptionSet.ts b/packages/realm/src/app-services/MutableSubscriptionSet.ts
new file mode 100644
index 0000000000..4a682460b0
--- /dev/null
+++ b/packages/realm/src/app-services/MutableSubscriptionSet.ts
@@ -0,0 +1,190 @@
+////////////////////////////////////////////////////////////////////////////
+//
+// Copyright 2023 Realm Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+////////////////////////////////////////////////////////////////////////////
+
+import { BaseSubscriptionSet, Realm, Subscription, SubscriptionSet, assert, binding } from "../internal";
+
+/**
+ * Options for {@link MutableSubscriptionSet.add}.
+ */
+export interface SubscriptionOptions {
+  /**
+   * Sets the name of the subscription being added. This allows you to later refer
+   * to the subscription by name, e.g. when calling {@link MutableSubscriptionSet.removeByName}.
+   */
+  name?: string;
+
+  /**
+   * By default, adding a subscription with the same name as an existing one
+   * but a different query will update the existing subscription with the new
+   * query. If `throwOnUpdate` is set to true, adding a subscription with the
+   * same name but a different query will instead throw an exception.
+   * Adding a subscription with the same name and query is always a no-op.
+   */
+  throwOnUpdate?: boolean;
+}
+
+/**
+ * The mutable version of a given SubscriptionSet. The {@link MutableSubscriptionSet}
+ * instance can only be used from inside the {@link SubscriptionSet.update} callback.
+ */
+export class MutableSubscriptionSet extends BaseSubscriptionSet {
+  // This class overrides the BaseSubscriptionSet's `internal` field (by having
+  // `declare internal`) in order to be able to write `this.internal.someOwnMember`
+  // rather than `(this.internal as binding.MutableSyncSubscriptionSet).someOwnMember`.
+  // (`this.internal = internal` cannot be used in the constructor due to the proxy
+  // handler in BaseSubscriptionSet making it non-writable.)
+  /**@internal */
+  declare internal: binding.MutableSyncSubscriptionSet;
+
+  /**@internal */
+  constructor(/**@internal */ internal: binding.MutableSyncSubscriptionSet) {
+    super(internal);
+  }
+
+  /**
+   * Add a query to the set of active subscriptions. The query will be joined via
+   * an `OR` operator with any existing queries for the same type.
+   *
+   * A query is represented by a {@link Realm.Results} instance returned from {@link Realm.objects},
+   * for example: `mutableSubs.add(realm.objects("Cat").filtered("age > 10"));`.
+   *
+   * @param query A {@link Realm.Results} instance representing the query to subscribe to.
+   * @param options An optional {@link SubscriptionOptions} object containing options to
+   *  use when adding this subscription (e.g. to give the subscription a name).
+   * @returns A `Subscription` instance for the new subscription.
+   */
+  add(query: Realm.Results<unknown>, options?: SubscriptionOptions): Subscription {
+    assert.instanceOf(query, Realm.Results, "query");
+    if (options) {
+      assertIsSubscriptionOptions(options);
+    }
+
+    const subscriptions = this.internal;
+    const results = query.internal;
+    const queryInternal = results.query;
+
+    if (options?.throwOnUpdate && options.name) {
+      const existingSubscription = subscriptions.findByName(options.name);
+      if (existingSubscription) {
+        const isSameQuery =
+          existingSubscription.queryString === queryInternal.description &&
+          existingSubscription.objectClassName === results.objectType;
+        assert(
+          isSameQuery,
+          `A subscription with the name '${options.name}' already exists but has a different query. If you meant to update it, remove 'throwOnUpdate: true' from the subscription options.`,
+        );
+      }
+    }
+
+    const [subscription] = options?.name
+      ? subscriptions.insertOrAssignByName(options.name, queryInternal)
+      : subscriptions.insertOrAssignByQuery(queryInternal);
+
+    return new Subscription(subscription);
+  }
+
+  /**
+   * Remove a subscription with the given query from the SubscriptionSet.
+   *
+   * @param query A {@link Realm.Results} instance representing the query to remove a subscription to.
+   * @returns `true` if the subscription was removed, `false` if it was not found.
+   */
+  remove(query: Realm.Results<unknown>): boolean {
+    assert.instanceOf(query, Realm.Results, "query");
+
+    return this.internal.eraseByQuery(query.internal.query);
+  }
+
+  /**
+   * Remove a subscription with the given name from the SubscriptionSet.
+   *
+   * @param name The name of the subscription to remove.
+   * @returns `true` if the subscription was removed, `false` if it was not found.
+   */
+  removeByName(name: string): boolean {
+    assert.string(name, "name");
+
+    return this.internal.eraseByName(name);
+  }
+
+  /**
+   * Remove the specified subscription from the SubscriptionSet.
+   *
+   * @param subscription The {@link Subscription} instance to remove.
+   * @returns `true` if the subscription was removed, `false` if it was not found.
+   */
+  removeSubscription(subscription: Subscription): boolean {
+    assert.instanceOf(subscription, Subscription, "subscription");
+
+    return binding.Helpers.eraseSubscription(this.internal, subscription.internal);
+  }
+
+  /**
+   * Remove all subscriptions for the specified object type from the SubscriptionSet.
+   *
+   * @param objectType The string name of the object type to remove all subscriptions for.
+   * @returns The number of subscriptions removed.
+   */
+  removeByObjectType(objectType: string): number {
+    // TODO: This is currently O(n^2) because each erase call is O(n). Once Core has
+    //       fixed https://github.com/realm/realm-core/issues/6241, we can update this.
+
+    assert.string(objectType, "objectType");
+
+    // Removing the subscription (calling `eraseSubscription()`) invalidates all current
+    // iterators, so it would be illegal to continue iterating. Instead, we push it to an
+    // array to remove later.
+    const subscriptionsToRemove: binding.SyncSubscription[] = [];
+    for (const subscription of this.internal) {
+      if (subscription.objectClassName === objectType) {
+        subscriptionsToRemove.push(subscription);
+      }
+    }
+    let numRemoved = 0;
+    for (const subscription of subscriptionsToRemove) {
+      const isRemoved = binding.Helpers.eraseSubscription(this.internal, subscription);
+      if (isRemoved) {
+        numRemoved++;
+      }
+    }
+
+    return numRemoved;
+  }
+
+  /**
+   * Remove all subscriptions from the SubscriptionSet.
+   *
+   * @returns The number of subscriptions removed.
+   */
+  removeAll(): number {
+    const numSubscriptions = this.internal.size;
+    this.internal.clear();
+
+    return numSubscriptions;
+  }
+}
+
+function assertIsSubscriptionOptions(input: unknown): asserts input is SubscriptionOptions {
+  assert.object(input, "options", { allowArrays: false });
+  if (input.name !== undefined) {
+    assert.string(input.name, "'name' on 'SubscriptionOptions'");
+  }
+  if (input.throwOnUpdate !== undefined) {
+    assert.boolean(input.throwOnUpdate, "'throwOnUpdate' on 'SubscriptionOptions'");
+  }
+}
diff --git a/packages/realm/src/app-services/Subscription.ts b/packages/realm/src/app-services/Subscription.ts
new file mode 100644
index 0000000000..9ac1fdbc89
--- /dev/null
+++ b/packages/realm/src/app-services/Subscription.ts
@@ -0,0 +1,77 @@
+////////////////////////////////////////////////////////////////////////////
+//
+// Copyright 2023 Realm Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+////////////////////////////////////////////////////////////////////////////
+
+import { BSON } from "../bson";
+import { binding } from "../internal";
+
+/**
+ * Class representing a single query subscription in a set of flexible sync
+ * {@link SubscriptionSet}. This class contains readonly information about the
+ * subscription – any changes to the set of subscriptions must be carried out
+ * in a {@link SubscriptionSet.update} callback.
+ */
+export class Subscription {
+  /**@internal */
+  constructor(/**@internal */ public internal: binding.SyncSubscription) {
+    this.internal = internal;
+  }
+
+  /**
+   * The ObjectId of the subscription.
+   */
+  get id(): BSON.ObjectId {
+    return this.internal.id;
+  }
+
+  /**
+   * The date when this subscription was created.
+   */
+  get createdAt(): Date {
+    return this.internal.createdAt.toDate();
+  }
+
+  /**
+   * The date when this subscription was last updated.
+   */
+  get updatedAt(): Date {
+    return this.internal.updatedAt.toDate();
+  }
+
+  /**
+   * The name given to this subscription when it was created.
+   * If no name was set, this will be `null`.
+   */
+  get name(): string | null {
+    return this.internal.name || null;
+  }
+
+  /**
+   * The type of objects the subscription refers to.
+   */
+  get objectType(): string {
+    return this.internal.objectClassName;
+  }
+
+  /**
+   * The string representation of the query the subscription was created with.
+   * If no filter or sort was specified, this will be `"TRUEPREDICATE"`.
+   */
+  get queryString(): string {
+    return this.internal.queryString;
+  }
+}
diff --git a/packages/realm/src/app-services/SubscriptionSet.ts b/packages/realm/src/app-services/SubscriptionSet.ts
new file mode 100644
index 0000000000..ba08f3676c
--- /dev/null
+++ b/packages/realm/src/app-services/SubscriptionSet.ts
@@ -0,0 +1,125 @@
+////////////////////////////////////////////////////////////////////////////
+//
+// Copyright 2023 Realm Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+////////////////////////////////////////////////////////////////////////////
+
+import {
+  BaseSubscriptionSet,
+  FlexibleSyncConfiguration,
+  MutableSubscriptionSet,
+  Realm,
+  SubscriptionsState,
+  assert,
+  binding,
+} from "../internal";
+
+/**
+ * Represents the set of all active flexible sync subscriptions for a Realm instance.
+ *
+ * The server will continuously evaluate the queries that the instance is subscribed to
+ * and will send data that matches them, as well as remove data that no longer does.
+ *
+ * The set of subscriptions can only be modifed inside a {@link SubscriptionSet.update} callback,
+ * by calling methods on the corresponding {@link MutableSubscriptionSet} instance.
+ */
+export class SubscriptionSet extends BaseSubscriptionSet {
+  /**@internal */
+  constructor(/**@internal */ private realm: Realm, internal: binding.SyncSubscriptionSet) {
+    super(internal);
+
+    Object.defineProperties(this, {
+      realm: {
+        enumerable: false,
+        configurable: false,
+        writable: false,
+      },
+    });
+  }
+
+  /**
+   * Wait for the server to acknowledge this set of subscriptions and return the
+   * matching objects.
+   *
+   * If `state` is {@link SubscriptionsState.Complete}, the promise will be resolved immediately.
+   *
+   * If `state` is {@link SubscriptionsState.Error}, the promise will be rejected immediately.
+   *
+   * @returns A promise which is resolved when synchronization is complete, or is
+   *  rejected if there is an error during synchronisation.
+   */
+  async waitForSynchronization(): Promise<void> {
+    try {
+      const state = await this.internal.getStateChangeNotification(binding.SyncSubscriptionSetState.Complete);
+      if (state === binding.SyncSubscriptionSetState.Error) {
+        throw new Error(this.error || "Encountered an error when waiting for synchronization.");
+      }
+    } finally {
+      if (!this.realm.isClosed) {
+        this.internal.refresh();
+      }
+    }
+  }
+
+  /**
+   * Call this to make changes to this SubscriptionSet from inside the callback,
+   * such as adding or removing subscriptions from the set.
+   *
+   * The MutableSubscriptonSet argument can only be used from the callback and must
+   * not be used after it returns.
+   *
+   * All changes done by the callback will be batched and sent to the server. You can either
+   * `await` the call to `update`, or call {@link SubscriptionSet.waitForSynchronization}
+   * to wait for the new data to be available.
+   *
+   * @param callback A callback function which receives a {@link MutableSubscriptionSet}
+   *  instance as the first argument, which can be used to add or remove subscriptions
+   *  from the set, and the {@link Realm} associated with the SubscriptionSet as the
+   *  second argument (mainly useful when working with `initialSubscriptions` in
+   *  {@link FlexibleSyncConfiguration}).
+   *
+   * @returns A promise which resolves when the SubscriptionSet is synchronized, or is rejected
+   *  if there was an error during synchronization (see {@link SubscriptionSet.waitForSynchronization})
+   *
+   * @example
+   * await realm.subscriptions.update(mutableSubscriptions => {
+   *   mutableSubscriptions.add(realm.objects("Cat").filtered("age > 10"));
+   *   mutableSubscriptions.add(realm.objects("Dog").filtered("age > 20"), { name: "oldDogs" });
+   *   mutableSubscriptions.removeByName("youngDogs");
+   * });
+   * // `realm` will now return the expected results based on the updated subscriptions
+   */
+  async update(callback: (mutableSubscriptions: MutableSubscriptionSet, realm: Realm) => void): Promise<void> {
+    this.updateNoWait(callback);
+    await this.waitForSynchronization();
+  }
+
+  /**@internal */
+  updateNoWait(callback: (mutableSubscriptions: MutableSubscriptionSet, realm: Realm) => void): void {
+    assert.function(callback, "callback");
+
+    // Create a mutable copy of this instance (which copies the original and upgrades
+    // its internal transaction to a write transaction) so that we can make updates to it.
+    const mutableSubscriptions = this.internal.makeMutableCopy();
+
+    callback(new MutableSubscriptionSet(mutableSubscriptions), this.realm);
+
+    // Commit the mutation, which downgrades its internal transaction to a read transaction
+    // so no more changes can be made to it, and returns a new (immutable) SubscriptionSet
+    // with the changes we made. Then update this SubscriptionSet instance to point to the
+    // updated version.
+    this.internal = mutableSubscriptions.commit();
+  }
+}
diff --git a/packages/realm/src/app-services/Sync.ts b/packages/realm/src/app-services/Sync.ts
index 8a9ae68eae..6e70197e37 100644
--- a/packages/realm/src/app-services/Sync.ts
+++ b/packages/realm/src/app-services/Sync.ts
@@ -27,6 +27,7 @@ import {
   assert,
   binding,
   toBindingSyncConfig,
+  validateSyncConfiguration,
 } from "../internal";
 
 import * as internal from "../internal";
@@ -66,6 +67,11 @@ function fromBindingLoggerLevel(arg: binding.LoggerLevel): NumericLogLevel {
 export namespace Sync {
   export const Session = SyncSession;
   export const ConnectionState = internal.ConnectionState;
+  export const Subscription = internal.Subscription;
+  export const SubscriptionSet = internal.SubscriptionSet;
+  export const MutableSubscriptionSet = internal.MutableSubscriptionSet;
+  export const SubscriptionsState = internal.SubscriptionsState;
+  export type SubscriptionOptions = internal.SubscriptionOptions;
   export function setLogLevel(app: App, level: LogLevel) {
     const numericLevel = toBindingLoggerLevel(level);
     app.internal.syncManager.setLogLevel(numericLevel);
@@ -80,6 +86,7 @@ export namespace Sync {
     throw new Error("Not yet implemented");
   }
   export function getSyncSession(user: User, partitionValue: PartitionValue): SyncSession | null {
+    validateSyncConfiguration({ user, partitionValue });
     const config = toBindingSyncConfig({ user, partitionValue });
     const path = user.app.internal.syncManager.pathForRealm(config);
     const session = user.internal.sessionForOnDiskPath(path);
diff --git a/packages/realm/src/app-services/SyncConfiguration.ts b/packages/realm/src/app-services/SyncConfiguration.ts
index 59b3346722..4be9343b65 100644
--- a/packages/realm/src/app-services/SyncConfiguration.ts
+++ b/packages/realm/src/app-services/SyncConfiguration.ts
@@ -21,9 +21,12 @@ import { EJSON, ObjectId, UUID } from "bson";
 import {
   BSON,
   ClientResetError,
+  MutableSubscriptionSet,
   Realm,
+  SubscriptionSet,
   SyncError,
   SyncSession,
+  TypeAssertionError,
   User,
   assert,
   binding,
@@ -114,26 +117,25 @@ export type BaseSyncConfiguration = {
   clientReset?: ClientResetConfig;
 };
 
-// TODO: Delete once the flexible sync API gets implemented
-type MutableSubscriptionSet = unknown;
+export type InitialSubscriptions = {
+  /**
+   * A callback to make changes to a SubscriptionSet.
+   *
+   * @see {@link SubscriptionSet.update} for more information.
+   */
+  update: (mutableSubscriptions: MutableSubscriptionSet, realm: Realm) => void;
+  /**
+   * If `true`, the {@link update} callback will be rerun every time the Realm is
+   * opened (e.g. every time a user opens your app), otherwise (by default) it
+   * will only be run if the Realm does not yet exist.
+   */
+  rerunOnOpen?: boolean;
+};
 
 export type FlexibleSyncConfiguration = BaseSyncConfiguration & {
   flexible: true;
   partitionValue?: never;
-  initialSubscriptions?: {
-    /**
-     * Callback called with the {@link Realm} instance to allow you to setup the
-     * initial set of subscriptions by calling `realm.subscriptions.update`.
-     * See {@link Realm.App.Sync.SubscriptionSet.update} for more information.
-     */
-    update: (subs: MutableSubscriptionSet, realm: Realm) => void;
-    /**
-     * If `true`, the {@link update} callback will be rerun every time the Realm is
-     * opened (e.g. every time a user opens your app), otherwise (by default) it
-     * will only be run if the Realm does not yet exist.
-     */
-    rerunOnOpen?: boolean;
-  };
+  initialSubscriptions?: InitialSubscriptions;
 };
 
 export type PartitionSyncConfiguration = BaseSyncConfiguration & {
@@ -146,20 +148,16 @@ export type SyncConfiguration = FlexibleSyncConfiguration | PartitionSyncConfigu
 
 /** @internal */
 export function toBindingSyncConfig(config: SyncConfiguration): binding.SyncConfig_Relaxed {
-  if (config.flexible) {
-    throw new Error("Flexible sync has not been implemented yet");
-  }
-  const { user, onError, _sessionStopPolicy, customHttpHeaders, clientReset } = config;
-  assert.instanceOf(user, User, "user");
-  validatePartitionValue(config.partitionValue);
-  const partitionValue = EJSON.stringify(config.partitionValue as EJSON.SerializableTypes);
+  const { user, flexible, partitionValue, onError, _sessionStopPolicy, customHttpHeaders, clientReset } = config;
+
   return {
-    user: config.user.internal,
-    partitionValue,
+    user: user.internal,
+    partitionValue: flexible ? undefined : EJSON.stringify(partitionValue),
     stopPolicy: _sessionStopPolicy
       ? toBindingStopPolicy(_sessionStopPolicy)
       : binding.SyncSessionStopPolicy.AfterChangesUploaded,
-    customHttpHeaders: customHttpHeaders,
+    customHttpHeaders,
+    flxSyncRequested: !!flexible,
     ...parseClientResetConfig(clientReset, onError),
   };
 }
@@ -238,26 +236,155 @@ function parseRecoverOrDiscardUnsyncedChanges(clientReset: ClientResetRecoverOrD
   };
 }
 
-/** @internal */
-function validatePartitionValue(pv: unknown) {
-  if (typeof pv === "number") {
-    validateNumberValue(pv);
-    return;
+/**
+ * Validate the fields of a user-provided realm sync configuration.
+ * @internal
+ */
+export function validateSyncConfiguration(config: unknown): asserts config is SyncConfiguration {
+  assert.object(config, "'sync' on realm configuration", { allowArrays: false });
+  const { user, newRealmFileBehavior, existingRealmFileBehavior, onError, customHttpHeaders, clientReset, flexible } =
+    config;
+
+  assert.instanceOf(user, User, "'user' on realm sync configuration");
+  if (newRealmFileBehavior !== undefined) {
+    validateOpenRealmBehaviorConfiguration(newRealmFileBehavior, "newRealmFileBehavior");
+  }
+  if (existingRealmFileBehavior !== undefined) {
+    validateOpenRealmBehaviorConfiguration(existingRealmFileBehavior, "existingRealmFileBehavior");
   }
-  if (!(pv instanceof ObjectId || pv instanceof UUID || typeof pv === "string" || pv === null)) {
-    throw new Error(pv + " is not an allowed PartitionValue");
+  if (onError !== undefined) {
+    assert.function(onError, "'onError' on realm sync configuration");
+  }
+  if (customHttpHeaders !== undefined) {
+    assert.object(customHttpHeaders, "'customHttpHeaders' on realm sync configuration", { allowArrays: false });
+    for (const key in customHttpHeaders) {
+      assert.string(customHttpHeaders[key], "all property values of 'customHttpHeaders' on realm sync configuration");
+    }
+  }
+  if (clientReset !== undefined) {
+    validateClientResetConfiguration(clientReset);
+  }
+  // Assume the user intends to use Flexible Sync for all truthy values provided.
+  if (flexible) {
+    validateFlexibleSyncConfiguration(config);
+  } else {
+    validatePartitionSyncConfiguration(config);
   }
 }
 
-/** @internal */
-function validateNumberValue(numberValue: number) {
-  if (!Number.isInteger(numberValue)) {
-    throw new Error("PartitionValue " + numberValue + " must be of type integer");
+/**
+ * Validate the fields of a user-provided open realm behavior configuration.
+ */
+function validateOpenRealmBehaviorConfiguration(
+  config: unknown,
+  target: string,
+): asserts config is OpenRealmBehaviorConfiguration {
+  assert.object(config, `'${target}' on realm sync configuration`, { allowArrays: false });
+  assert(
+    config.type === OpenRealmBehaviorType.DownloadBeforeOpen || config.type === OpenRealmBehaviorType.OpenImmediately,
+    `'${target}.type' on realm sync configuration must be either '${OpenRealmBehaviorType.DownloadBeforeOpen}' or '${OpenRealmBehaviorType.OpenImmediately}'.`,
+  );
+  if (config.timeOut !== undefined) {
+    assert.number(config.timeOut, `'${target}.timeOut' on realm sync configuration`);
+  }
+  if (config.timeOutBehavior !== undefined) {
+    assert(
+      config.timeOutBehavior === OpenRealmTimeOutBehavior.OpenLocalRealm ||
+        config.timeOutBehavior === OpenRealmTimeOutBehavior.ThrowException,
+      `'${target}.timeOutBehavior' on realm sync configuration must be either '${OpenRealmTimeOutBehavior.OpenLocalRealm}' or '${OpenRealmTimeOutBehavior.ThrowException}'.`,
+    );
+  }
+}
+
+/**
+ * Validate the fields of a user-provided client reset configuration.
+ */
+function validateClientResetConfiguration(config: unknown): asserts config is ClientResetConfig {
+  assert.object(config, "'clientReset' on realm sync configuration", { allowArrays: false });
+  const modes = Object.values(ClientResetMode);
+  assert(
+    modes.includes(config.mode as ClientResetMode),
+    `'clientReset' on realm sync configuration must be one of the following: '${modes.join("', '")}'`,
+  );
+  if (config.onManual !== undefined) {
+    assert.function(config.onManual, "'clientReset.onManual' on realm sync configuration");
+  }
+  if (config.onAfter !== undefined) {
+    assert.function(config.onAfter, "'clientReset.onAfter' on realm sync configuration");
+  }
+  if (config.onBefore !== undefined) {
+    assert.function(config.onBefore, "'clientReset.onBefore' on realm sync configuration");
   }
-  if (numberValue > Number.MAX_SAFE_INTEGER) {
-    throw new Error("PartitionValue " + numberValue + " is greater than Number.MAX_SAFE_INTEGER");
+  if (config.onFallback !== undefined) {
+    assert.function(config.onFallback, "'clientReset.onFallback' on realm sync configuration");
+  }
+}
+
+/**
+ * Validate the fields of a user-provided realm flexible sync configuration.
+ */
+function validateFlexibleSyncConfiguration(
+  config: Record<string, unknown>,
+): asserts config is FlexibleSyncConfiguration {
+  const { flexible, partitionValue, initialSubscriptions } = config;
+
+  assert(
+    flexible === true,
+    "'flexible' must always be true for realms using flexible sync. To enable partition-based sync, remove 'flexible' and specify 'partitionValue'.",
+  );
+  if (initialSubscriptions !== undefined) {
+    assert.object(initialSubscriptions, "'initialSubscriptions' on realm sync configuration", { allowArrays: false });
+    assert.function(initialSubscriptions.update, "'initialSubscriptions.update' on realm sync configuration");
+    if (initialSubscriptions.rerunOnOpen !== undefined) {
+      assert.boolean(
+        initialSubscriptions.rerunOnOpen,
+        "'initialSubscriptions.rerunOnOpen' on realm sync configuration",
+      );
+    }
   }
-  if (numberValue < Number.MIN_SAFE_INTEGER) {
-    throw new Error("PartitionValue " + numberValue + " is lesser than Number.MIN_SAFE_INTEGER");
+  assert(
+    partitionValue === undefined,
+    "'partitionValue' cannot be specified when flexible sync is enabled. To enable partition-based sync, remove 'flexible' and specify 'partitionValue'.",
+  );
+}
+
+/**
+ * Validate the fields of a user-provided realm partition sync configuration.
+ */
+function validatePartitionSyncConfiguration(
+  config: Record<string, unknown>,
+): asserts config is PartitionSyncConfiguration {
+  const { flexible, partitionValue, initialSubscriptions } = config;
+
+  validatePartitionValue(partitionValue);
+  // We only allow `flexible` to be `true` (for Flexible Sync) or `undefined` (for Partition Sync).
+  // `{ flexible: false }` is not allowed because TypeScript cannot discriminate that type correctly
+  // with `strictNullChecks` disabled, and there is no real use case for `{ flexible: false }`.
+  assert(
+    flexible === undefined,
+    "'flexible' can only be specified to enable flexible sync. To enable flexible sync, remove 'partitionValue' and set 'flexible' to true.",
+  );
+  assert(
+    initialSubscriptions === undefined,
+    "'initialSubscriptions' can only be specified when flexible sync is enabled. To enable flexible sync, remove 'partitionValue' and set 'flexible' to true.",
+  );
+}
+
+/**
+ * Validate the user-provided partition value of a realm sync configuration.
+ */
+function validatePartitionValue(value: unknown): asserts value is PartitionValue {
+  if (typeof value === "number") {
+    assert(
+      Number.isSafeInteger(value),
+      `Expected 'partitionValue' on realm sync configuration to be an integer, got ${value}.`,
+    );
+  } else {
+    assert(
+      typeof value === "string" || value instanceof ObjectId || value instanceof UUID || value === null,
+      `Expected 'partitionValue' on realm sync configuration to be an integer, string, ObjectId, UUID, or null, got ${TypeAssertionError.deriveType(
+        value,
+      )}.`,
+    );
   }
 }
diff --git a/packages/realm/src/app-services/SyncSession.ts b/packages/realm/src/app-services/SyncSession.ts
index b201f391d1..e76436590d 100644
--- a/packages/realm/src/app-services/SyncSession.ts
+++ b/packages/realm/src/app-services/SyncSession.ts
@@ -141,15 +141,18 @@ export function toBindingErrorHandlerWithOnManual(
 
 /** @internal */
 export function toBindingNotifyBeforeClientReset(onBefore: ClientResetBeforeCallback) {
-  return (localRealmInternal: binding.Realm) => {
-    onBefore(new Realm(localRealmInternal));
+  return (internal: binding.Realm) => {
+    onBefore(new Realm(null, { internal }));
   };
 }
 
 /** @internal */
 export function toBindingNotifyAfterClientReset(onAfter: ClientResetAfterCallback) {
-  return (localRealmInternal: binding.Realm, tsr: binding.ThreadSafeReference) => {
-    onAfter(new Realm(localRealmInternal), new Realm(binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr)));
+  return (internal: binding.Realm, tsr: binding.ThreadSafeReference) => {
+    onAfter(
+      new Realm(null, { internal }),
+      new Realm(null, { internal: binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr) }),
+    );
   };
 }
 
@@ -158,11 +161,14 @@ export function toBindingNotifyAfterClientResetWithFallback(
   onAfter: ClientResetAfterCallback,
   onFallback: ClientResetFallbackCallback | undefined,
 ) {
-  return (localRealmInternal: binding.Realm, tsr: binding.ThreadSafeReference, didRecover: boolean) => {
+  return (internal: binding.Realm, tsr: binding.ThreadSafeReference, didRecover: boolean) => {
     if (didRecover) {
-      onAfter(new Realm(localRealmInternal), new Realm(binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr)));
+      onAfter(
+        new Realm(null, { internal }),
+        new Realm(null, { internal: binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr) }),
+      );
     } else {
-      const realm = new Realm(binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr));
+      const realm = new Realm(null, { internal: binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr) });
       if (onFallback) {
         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
         onFallback(realm.syncSession!, realm.path);
diff --git a/packages/realm/src/errors.ts b/packages/realm/src/errors.ts
index fd4ad910ed..45bbc2a818 100644
--- a/packages/realm/src/errors.ts
+++ b/packages/realm/src/errors.ts
@@ -26,7 +26,7 @@ export class AssertionError extends Error {
 
 export class TypeAssertionError extends AssertionError {
   /** @internal */
-  private static deriveType(value: unknown) {
+  public static deriveType(value: unknown) {
     if (typeof value === "object") {
       if (value === null) {
         return "null";
diff --git a/packages/realm/src/index.ts b/packages/realm/src/index.ts
index 60299e688e..481177e7cd 100644
--- a/packages/realm/src/index.ts
+++ b/packages/realm/src/index.ts
@@ -24,6 +24,7 @@ export {
   Credentials,
   Dictionary,
   List,
+  MutableSubscriptionSet,
   OrderedCollection,
   ProgressRealmPromise,
   Realm,
@@ -31,6 +32,9 @@ export {
   RealmSet as Set,
   Results,
   SessionStopPolicy,
+  Subscription,
+  SubscriptionSet,
+  SubscriptionsState,
   UpdateMode,
   User,
   Types,
@@ -41,10 +45,13 @@ export type {
   CollectionChangeCallback,
   CollectionChangeSet,
   Configuration,
+  FlexibleSyncConfiguration,
   ObjectChangeCallback,
   ObjectChangeSet,
+  PartitionSyncConfiguration,
   RealmEventName,
   RealmListenerCallback,
+  SubscriptionOptions,
 } from "./internal";
 
 // Exporting default for backwards compatibility
diff --git a/packages/realm/src/internal.ts b/packages/realm/src/internal.ts
index 825d0ac2e2..d050f3cddb 100644
--- a/packages/realm/src/internal.ts
+++ b/packages/realm/src/internal.ts
@@ -79,6 +79,10 @@ export * from "./app-services/PushClient";
 export * from "./app-services/MongoClient";
 export * from "./app-services/FunctionsFactory";
 export * from "./app-services/UserProfile";
+export * from "./app-services/BaseSubscriptionSet";
+export * from "./app-services/MutableSubscriptionSet";
+export * from "./app-services/SubscriptionSet";
+export * from "./app-services/Subscription";
 export * from "./app-services/Sync";
 export * from "./app-services/App";
 
diff --git a/packages/realm/src/schema.ts b/packages/realm/src/schema.ts
index 44686d12fc..95ae0e4ac5 100644
--- a/packages/realm/src/schema.ts
+++ b/packages/realm/src/schema.ts
@@ -20,3 +20,4 @@ export * from "./schema/from-binding";
 export * from "./schema/to-binding";
 export * from "./schema/normalize";
 export * from "./schema/types";
+export * from "./schema/validate";
diff --git a/packages/realm/src/schema/validate.ts b/packages/realm/src/schema/validate.ts
new file mode 100644
index 0000000000..3b313ae95c
--- /dev/null
+++ b/packages/realm/src/schema/validate.ts
@@ -0,0 +1,162 @@
+////////////////////////////////////////////////////////////////////////////
+//
+// Copyright 2023 Realm Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+////////////////////////////////////////////////////////////////////////////
+
+import {
+  CanonicalObjectSchema,
+  CanonicalObjectSchemaProperty,
+  Configuration,
+  DefaultObject,
+  ObjectSchema,
+  ObjectSchemaProperty,
+  RealmObject,
+  RealmObjectConstructor,
+  assert,
+} from "../internal";
+
+// Need to use `CanonicalObjectSchema` rather than `ObjectSchema` due to some
+// integration tests using `openRealmHook()`. That function sets `this.realm`
+// to the opened realm whose schema is a `CanonicalObjectSchema[]`. Consequently,
+// the key `"ctor"` (which doesn't exist on `ObjectSchema`) also needs to be allowed.
+const OBJECT_SCHEMA_KEYS = new Set<keyof CanonicalObjectSchema>([
+  "name",
+  "primaryKey",
+  "embedded",
+  "asymmetric",
+  "properties",
+  // Not part of `ObjectSchema`
+  "ctor",
+]);
+
+// Need to use `CanonicalObjectSchemaProperty` rather than `ObjectSchemaProperty`
+// due to the same reasons as above.
+const PROPERTY_SCHEMA_KEYS = new Set<keyof CanonicalObjectSchemaProperty>([
+  "type",
+  "objectType",
+  "property",
+  "default",
+  "optional",
+  "indexed",
+  "mapTo",
+  // Not part of `ObjectSchemaProperty`
+  "name",
+]);
+
+/**
+ * Validate the data types of the fields of a user-provided realm schema.
+ */
+export function validateRealmSchema(realmSchema: unknown): asserts realmSchema is Configuration["schema"][] {
+  assert.array(realmSchema, "realm schema");
+  for (const objectSchema of realmSchema) {
+    validateObjectSchema(objectSchema);
+  }
+  // TODO: Assert that backlinks point to object schemas that are actually declared
+}
+
+/**
+ * Validate the data types of the fields of a user-provided object schema.
+ */
+export function validateObjectSchema(
+  objectSchema: unknown,
+): asserts objectSchema is RealmObjectConstructor | ObjectSchema {
+  // Schema is passed via a class based model (RealmObjectConstructor)
+  if (typeof objectSchema === "function") {
+    const clazz = objectSchema as unknown as DefaultObject;
+    // We assert this later, but want a custom error message
+    if (!(objectSchema.prototype instanceof RealmObject)) {
+      const schemaName = clazz.schema && (clazz.schema as DefaultObject).name;
+      if (typeof schemaName === "string" && schemaName !== objectSchema.name) {
+        throw new TypeError(`Class '${objectSchema.name}' (declaring '${schemaName}' schema) must extend Realm.Object`);
+      } else {
+        throw new TypeError(`Class '${objectSchema.name}' must extend Realm.Object`);
+      }
+    }
+    assert.object(clazz.schema, "schema static");
+    validateObjectSchema(clazz.schema);
+  }
+  // Schema is passed as an object (ObjectSchema)
+  else {
+    assert.object(objectSchema, "object schema", { allowArrays: false });
+    const { name: objectName, properties, primaryKey, asymmetric, embedded } = objectSchema;
+    assert.string(objectName, "'name' on object schema");
+    assert.object(properties, `'properties' on '${objectName}'`, { allowArrays: false });
+    if (primaryKey !== undefined) {
+      assert.string(primaryKey, `'primaryKey' on '${objectName}'`);
+    }
+    if (embedded !== undefined) {
+      assert.boolean(embedded, `'embedded' on '${objectName}'`);
+    }
+    if (asymmetric !== undefined) {
+      assert.boolean(asymmetric, `'asymmetric' on '${objectName}'`);
+    }
+
+    const invalidKeysUsed = filterInvalidKeys(objectSchema, OBJECT_SCHEMA_KEYS);
+    assert(
+      !invalidKeysUsed.length,
+      `Unexpected field(s) found on the schema for object '${objectName}': '${invalidKeysUsed.join("', '")}'.`,
+    );
+
+    for (const propertyName in properties) {
+      const propertySchema = properties[propertyName];
+      const isUsingShorthand = typeof propertySchema === "string";
+      if (!isUsingShorthand) {
+        validatePropertySchema(objectName, propertyName, propertySchema);
+      }
+    }
+  }
+}
+
+/**
+ * Validate the data types of a user-provided property schema that ought to use the
+ * relaxed object notation.
+ */
+export function validatePropertySchema(
+  objectName: string,
+  propertyName: string,
+  propertySchema: unknown,
+): asserts propertySchema is ObjectSchemaProperty {
+  assert.object(propertySchema, `'${propertyName}' on '${objectName}'`, { allowArrays: false });
+  const { type, objectType, optional, property, indexed, mapTo } = propertySchema;
+  assert.string(type, `'${propertyName}.type' on '${objectName}'`);
+  if (objectType !== undefined) {
+    assert.string(objectType, `'${propertyName}.objectType' on '${objectName}'`);
+  }
+  if (optional !== undefined) {
+    assert.boolean(optional, `'${propertyName}.optional' on '${objectName}'`);
+  }
+  if (property !== undefined) {
+    assert.string(property, `'${propertyName}.property' on '${objectName}'`);
+  }
+  if (indexed !== undefined) {
+    assert.boolean(indexed, `'${propertyName}.indexed' on '${objectName}'`);
+  }
+  if (mapTo !== undefined) {
+    assert.string(mapTo, `'${propertyName}.mapTo' on '${objectName}'`);
+  }
+  const invalidKeysUsed = filterInvalidKeys(propertySchema, PROPERTY_SCHEMA_KEYS);
+  assert(
+    !invalidKeysUsed.length,
+    `Unexpected field(s) found on the schema for property '${propertyName}' on '${objectName}': '${invalidKeysUsed.join("', '")}'.`,
+  );
+}
+
+/**
+ * Get the keys of an object that are not part of the provided valid keys.
+ */
+function filterInvalidKeys(object: Record<string, unknown>, validKeys: Set<string>): string[] {
+  return Object.keys(object).filter((key) => !validKeys.has(key));
+}
diff --git a/packages/realm/src/tests/schema-validation.test.ts b/packages/realm/src/tests/schema-validation.test.ts
index 724a10a8a0..7f2d43fa6e 100644
--- a/packages/realm/src/tests/schema-validation.test.ts
+++ b/packages/realm/src/tests/schema-validation.test.ts
@@ -18,7 +18,7 @@
 
 import { expect } from "chai";
 
-import { validateObjectSchema, validatePropertySchema } from "../Configuration";
+import { validateObjectSchema, validatePropertySchema } from "../internal";
 
 const OBJECT_NAME = "MyObject";
 const PROPERTY_NAME = "prop";