diff --git a/lib/OnyxSubscriptionManager.ts b/lib/OnyxSubscriptionManager.ts
new file mode 100644
index 000000000..7e61ec7e1
--- /dev/null
+++ b/lib/OnyxSubscriptionManager.ts
@@ -0,0 +1,196 @@
+import type {CollectionKeyBase, KeyValueMapping, OnyxCollection, OnyxKey, OnyxValue} from './types';
+
+import * as Logger from './Logger';
+import cache from './OnyxCache';
+import OnyxKeys from './OnyxKeys';
+
+/**
+ * Listener fired when an exact key's value changes.
+ */
+type Listener<TKey extends OnyxKey = OnyxKey> = (value: OnyxValue<TKey>, key: TKey) => void;
+
+/**
+ * Generic listener, without specific types for the keys or values.
+ * This way a single Map can hold listeners for every key type.
+ */
+type GenericListener = (value: unknown, key: OnyxKey) => void;
+
+/**
+ * OnyxSubscriptionManager is a registry for Onyx subscriptions.
+ * Subscriptions are stored in `keyListeners`, a flat map keyed by OnyxKey.
+ * Subscribers are notified on a per-key basis:
+ *    - `notifyKey` for individual keys or collection members
+ *    - `notifyCollection` for batch updates to collections
+ */
+class OnyxSubscriptionManager {
+    private keyListeners: Map<OnyxKey, Set<GenericListener>>;
+
+    constructor() {
+        this.keyListeners = new Map();
+    }
+
+    /**
+     * Returns the frozen collection object for collection
+     * keys, the cached value for single keys, or `undefined` if not in cache.
+     */
+    getState<TKey extends OnyxKey>(key: TKey): OnyxValue<TKey> {
+        if (OnyxKeys.isCollectionKey(key)) {
+            return cache.getCollectionData(key) as OnyxValue<TKey>;
+        }
+        return cache.get(key) as OnyxValue<TKey>;
+    }
+
+    /**
+     * Subscribe to an exact key. For a collection root key this is "collection mode": the
+     * listener fires with the frozen collection object whenever any member changes. For a
+     * collection member key or a regular key, the listener fires when that key's value changes.
+     *
+     * Returns an unsubscribe function.
+     */
+    subscribe<TKey extends OnyxKey>(key: TKey, listener: Listener<TKey>): () => void {
+        let listeners = this.keyListeners.get(key);
+        if (!listeners) {
+            listeners = new Set();
+            this.keyListeners.set(key, listeners);
+        }
+
+        listeners.add(listener as GenericListener);
+
+        return () => {
+            const set = this.keyListeners.get(key);
+            if (!set) {
+                return;
+            }
+
+            set.delete(listener as GenericListener);
+
+            if (set.size === 0) {
+                this.keyListeners.delete(key);
+            }
+        };
+    }
+
+    /**
+     * Notify of a single-key write.
+     *
+     * Dispatch:
+     *   1. keyListeners.get(key): exact-key subscribers (always fires).
+     *   2. If key is a collection member, keyListeners.get(collectionKey): collection
+     *      listeners for the parent collection.
+     */
+    notifyKey<TKey extends OnyxKey>(key: TKey, value: OnyxValue<TKey>): void {
+        // 1. Exact-key listeners
+        const exact = this.keyListeners.get(key);
+        if (exact && exact.size > 0) {
+            for (const listener of [...exact]) {
+                this.safeInvoke(() => listener(value, key), key);
+            }
+        }
+
+        // 2. Collection-level routing. Only fires when the write is to a member key.
+        // Direct writes to a collection root (e.g. `Onyx.merge(COLLECTION_KEY, ...)`) are an
+        // unsupported anti-pattern; treat them as opaque single-key writes.
+        const collectionKey = OnyxKeys.getCollectionKey(key);
+        const isCollectionMemberWrite = collectionKey !== undefined && collectionKey !== key;
+        if (isCollectionMemberWrite) {
+            const collectionListeners = this.keyListeners.get(collectionKey);
+            if (collectionListeners && collectionListeners.size > 0) {
+                const collectionData = cache.getCollectionData(collectionKey);
+                for (const listener of [...collectionListeners]) {
+                    this.safeInvoke(() => listener(collectionData, collectionKey), collectionKey);
+                }
+            }
+        }
+    }
+
+    /**
+     * Notify of a collection-level batch update.
+     *
+     * Dispatch:
+     *   1. keyListeners.get(collectionKey): fires once with the new collection object.
+     *   2. keyListeners.get(memberKey): fires per changed member whose value differs from
+     *      the previous, preserving ref-equality on unchanged members.
+     */
+    notifyCollection<TKey extends CollectionKeyBase>(
+        collectionKey: TKey,
+        partialCollection: OnyxCollection<KeyValueMapping[TKey]>,
+        partialPreviousCollection?: OnyxCollection<KeyValueMapping[TKey]>,
+    ): void {
+        const changedKeys = Object.keys(partialCollection ?? {});
+        if (changedKeys.length === 0) {
+            return;
+        }
+        // Read the merged collection once. `cache.getCollectionData()` returns the post-merge
+        // frozen object, which is what listeners should see (not the raw `partialCollection`
+        // input, which is just the delta and lacks fields preserved during merge).
+        const collectionData = cache.getCollectionData(collectionKey);
+
+        // 1. Collection listeners fire once with the new collection object.
+        const collectionListeners = this.keyListeners.get(collectionKey);
+        if (collectionListeners && collectionListeners.size > 0) {
+            for (const listener of [...collectionListeners]) {
+                this.safeInvoke(() => listener(collectionData, collectionKey), collectionKey);
+            }
+        }
+
+        // 2. Exact-member subscribers fire per changed key (skip if ref unchanged vs previous).
+        // Only treat a member as unchanged when `previous` actually carries it: when the
+        // previous collection is omitted, a removed member reads `undefined` on both sides
+        // and would otherwise be skipped even though it changed.
+        for (const memberKey of changedKeys) {
+            const value = collectionData?.[memberKey];
+            const prev = partialPreviousCollection?.[memberKey];
+            if (partialPreviousCollection && Object.prototype.hasOwnProperty.call(partialPreviousCollection, memberKey) && value === prev) {
+                continue;
+            }
+
+            const exact = this.keyListeners.get(memberKey);
+            if (!exact || exact.size === 0) {
+                continue;
+            }
+
+            for (const listener of [...exact]) {
+                this.safeInvoke(() => listener(value, memberKey), memberKey);
+            }
+        }
+    }
+
+    /**
+     * Wipe all subscriptions. Used by tests and `Onyx.clear()` follow-on.
+     */
+    clearAll(): void {
+        this.keyListeners.clear();
+    }
+
+    /**
+     * True if there are any subscribers for the given key (exact or parent collection).
+     */
+    hasListenersForKey(key: OnyxKey): boolean {
+        if ((this.keyListeners.get(key)?.size ?? 0) > 0) {
+            return true;
+        }
+
+        const collectionKey = OnyxKeys.getCollectionKey(key);
+        if (collectionKey && collectionKey !== key && (this.keyListeners.get(collectionKey)?.size ?? 0) > 0) {
+            return true;
+        }
+
+        return false;
+    }
+
+    /**
+     * Runs a listener, catching and logging any throw so one failing listener can't stop the rest.
+     */
+    private safeInvoke(fn: () => void, contextKey: OnyxKey): void {
+        try {
+            fn();
+        } catch (error) {
+            Logger.logAlert(`[OnyxSubscriptionManager] Listener threw an error for key '${contextKey}': ${error}`);
+        }
+    }
+}
+
+const onyxSubscriptionManager = new OnyxSubscriptionManager();
+
+export default onyxSubscriptionManager;
+export type {Listener};
diff --git a/tests/unit/OnyxSubscriptionManagerTest.ts b/tests/unit/OnyxSubscriptionManagerTest.ts
new file mode 100644
index 000000000..1b8ba7a09
--- /dev/null
+++ b/tests/unit/OnyxSubscriptionManagerTest.ts
@@ -0,0 +1,336 @@
+import type {OnyxKey} from '../../lib';
+import Onyx from '../../lib';
+import onyxSubscriptionManager from '../../lib/OnyxSubscriptionManager';
+import cache from '../../lib/OnyxCache';
+import * as Logger from '../../lib/Logger';
+
+// We need access to some internal properties of `onyxSubscriptionManager` during the tests but they are private,
+// so this workaround allows us to have access to them. The maps are created once in the constructor
+// and only ever `.clear()`ed (never reassigned), so capturing the references here stays valid.
+// eslint-disable-next-line dot-notation
+const keyListeners = onyxSubscriptionManager['keyListeners'];
+
+const ONYXKEYS = {
+    TEST_KEY: 'test',
+    OTHER_TEST: 'otherTest',
+    COLLECTION: {
+        TEST_KEY: 'test_',
+    },
+};
+
+const COLLECTION = ONYXKEYS.COLLECTION.TEST_KEY;
+const MEMBER_1 = `${COLLECTION}1`;
+const MEMBER_2 = `${COLLECTION}2`;
+
+Onyx.init({
+    keys: ONYXKEYS,
+});
+
+beforeEach(() => Onyx.clear());
+
+describe('OnyxSubscriptionManager', () => {
+    // Always start from a clean registry.
+    beforeEach(() => {
+        onyxSubscriptionManager.clearAll();
+    });
+
+    afterEach(() => {
+        jest.restoreAllMocks();
+    });
+
+    describe('subscribe / notifyKey', () => {
+        it('should fire the listener with (value, key) on notifyKey', () => {
+            const callback = jest.fn();
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback);
+
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'hello');
+
+            expect(callback).toHaveBeenCalledTimes(1);
+            expect(callback).toHaveBeenCalledWith('hello', ONYXKEYS.TEST_KEY);
+        });
+
+        it('should fire all listeners registered on the same key', () => {
+            const callback1 = jest.fn();
+            const callback2 = jest.fn();
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback1);
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback2);
+
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 1);
+
+            expect(callback1).toHaveBeenCalledTimes(1);
+            expect(callback2).toHaveBeenCalledTimes(1);
+        });
+
+        it('should not fire the listener after it unsubscribes', () => {
+            const callback = jest.fn();
+            const unsubscribe = onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback);
+
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'first');
+            unsubscribe();
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'second');
+
+            expect(callback).toHaveBeenCalledTimes(1);
+            expect(callback).toHaveBeenLastCalledWith('first', ONYXKEYS.TEST_KEY);
+        });
+
+        it('should only unsubscribe the specific listener, leaving others intact', () => {
+            const callback1 = jest.fn();
+            const callback2 = jest.fn();
+            const unsubscribe1 = onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback1);
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback2);
+
+            unsubscribe1();
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 1);
+
+            expect(callback1).not.toHaveBeenCalled();
+            expect(callback2).toHaveBeenCalledTimes(1);
+        });
+
+        it('should delete the key entry from the internal map once the last listener unsubscribes', () => {
+            const unsubscribe = onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, jest.fn());
+            expect(keyListeners.has(ONYXKEYS.TEST_KEY)).toBeTruthy();
+
+            unsubscribe();
+
+            expect(keyListeners.has(ONYXKEYS.TEST_KEY)).toBeFalsy();
+        });
+
+        it('should be a no-op to notify a key with no listeners', () => {
+            expect(() => onyxSubscriptionManager.notifyKey('keyWithNoListeners' as OnyxKey, 'x')).not.toThrow();
+        });
+
+        it('should be idempotent when unsubscribing more than once', () => {
+            const callback = jest.fn();
+            const unsubscribe = onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback);
+
+            unsubscribe();
+            expect(() => unsubscribe()).not.toThrow();
+
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 1);
+            expect(callback).not.toHaveBeenCalled();
+        });
+    });
+
+    describe('getState', () => {
+        it('should return the cached value for a single key', () => {
+            cache.set(ONYXKEYS.TEST_KEY, 'hello');
+
+            expect(onyxSubscriptionManager.getState(ONYXKEYS.TEST_KEY)).toBe('hello');
+        });
+
+        it('should return undefined for a single key that is not in the cache', () => {
+            expect(onyxSubscriptionManager.getState(ONYXKEYS.OTHER_TEST)).toBeUndefined();
+        });
+
+        it('should return the collection object for a collection key', () => {
+            const collectionData = {[MEMBER_1]: {id: 1}, [MEMBER_2]: {id: 2}};
+            jest.spyOn(cache, 'getCollectionData').mockReturnValue(collectionData);
+
+            expect(onyxSubscriptionManager.getState(COLLECTION)).toBe(collectionData);
+        });
+    });
+
+    describe('collection routing on notifyKey', () => {
+        it('should fire the collection-root listener with the cache collection object when a member is written', () => {
+            const collectionData = {[MEMBER_1]: {id: 1}, [MEMBER_2]: {id: 2}};
+            const getCollectionData = jest.spyOn(cache, 'getCollectionData').mockReturnValue(collectionData);
+
+            const callback = jest.fn();
+            onyxSubscriptionManager.subscribe(COLLECTION, callback);
+
+            onyxSubscriptionManager.notifyKey(MEMBER_1, {id: 1});
+
+            expect(getCollectionData).toHaveBeenCalledWith(COLLECTION);
+            expect(callback).toHaveBeenCalledTimes(1);
+            expect(callback).toHaveBeenCalledWith(collectionData, COLLECTION);
+        });
+
+        it('should fire both the exact-member listener and the collection-root listener', () => {
+            const collectionData = {[MEMBER_1]: {id: 1}};
+            jest.spyOn(cache, 'getCollectionData').mockReturnValue(collectionData);
+
+            const memberCallback = jest.fn();
+            const collectionCallback = jest.fn();
+            onyxSubscriptionManager.subscribe(MEMBER_1, memberCallback);
+            onyxSubscriptionManager.subscribe(COLLECTION, collectionCallback);
+
+            onyxSubscriptionManager.notifyKey(MEMBER_1, {id: 1});
+
+            expect(memberCallback).toHaveBeenCalledWith({id: 1}, MEMBER_1);
+            expect(collectionCallback).toHaveBeenCalledWith(collectionData, COLLECTION);
+        });
+
+        it('should not perform collection routing for a non-member single key', () => {
+            const getCollectionData = jest.spyOn(cache, 'getCollectionData');
+            const callback = jest.fn();
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback);
+
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'x');
+
+            expect(callback).toHaveBeenCalledTimes(1);
+            expect(getCollectionData).not.toHaveBeenCalled();
+        });
+    });
+
+    describe('notifyCollection', () => {
+        it('should fire the collection listener once with the cache collection object', () => {
+            const collectionData = {[MEMBER_1]: {id: 1}, [MEMBER_2]: {id: 2}};
+            jest.spyOn(cache, 'getCollectionData').mockReturnValue(collectionData);
+
+            const callback = jest.fn();
+            onyxSubscriptionManager.subscribe(COLLECTION, callback);
+
+            onyxSubscriptionManager.notifyCollection(COLLECTION, {[MEMBER_1]: {id: 1}, [MEMBER_2]: {id: 2}});
+
+            expect(callback).toHaveBeenCalledTimes(1);
+            expect(callback).toHaveBeenCalledWith(collectionData, COLLECTION);
+        });
+
+        it('should fire exact-member listeners only for members whose value reference changed', () => {
+            const shared = {id: 2}; // same reference in collection and previous, should be skipped
+            const collectionData = {[MEMBER_1]: {id: 1}, [MEMBER_2]: shared};
+            jest.spyOn(cache, 'getCollectionData').mockReturnValue(collectionData);
+
+            const member1Callback = jest.fn();
+            const member2Callback = jest.fn();
+            onyxSubscriptionManager.subscribe(MEMBER_1, member1Callback);
+            onyxSubscriptionManager.subscribe(MEMBER_2, member2Callback);
+
+            onyxSubscriptionManager.notifyCollection(
+                COLLECTION,
+                {[MEMBER_1]: {id: 1}, [MEMBER_2]: shared},
+                {[MEMBER_2]: shared}, // previous: member 2 unchanged by reference
+            );
+
+            expect(member1Callback).toHaveBeenCalledWith({id: 1}, MEMBER_1);
+            expect(member2Callback).not.toHaveBeenCalled();
+        });
+
+        it('should notify the exact-member listener when the member is deleted without a previous collection', () => {
+            // member1 has already been removed from the cache by the time notifyCollection runs.
+            jest.spyOn(cache, 'getCollectionData').mockReturnValue({[MEMBER_2]: {id: 2}});
+
+            const member1Callback = jest.fn();
+            onyxSubscriptionManager.subscribe(MEMBER_1, member1Callback);
+
+            // No partialPreviousCollection passed — a removed member reads undefined on both
+            // sides and must not be treated as unchanged.
+            onyxSubscriptionManager.notifyCollection(COLLECTION, {[MEMBER_1]: null});
+
+            expect(member1Callback).toHaveBeenCalledTimes(1);
+        });
+
+        it('should be a no-op when the partial collection is empty', () => {
+            const callback = jest.fn();
+            onyxSubscriptionManager.subscribe(COLLECTION, callback);
+
+            onyxSubscriptionManager.notifyCollection(COLLECTION, {});
+
+            expect(callback).not.toHaveBeenCalled();
+        });
+    });
+
+    describe('hasListenersForKey', () => {
+        it('should return true for an exact-key subscriber', () => {
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, jest.fn());
+            expect(onyxSubscriptionManager.hasListenersForKey(ONYXKEYS.TEST_KEY)).toBeTruthy();
+        });
+
+        it('should return true for a member key when its parent collection has a subscriber', () => {
+            onyxSubscriptionManager.subscribe(COLLECTION, jest.fn());
+            expect(onyxSubscriptionManager.hasListenersForKey(MEMBER_1)).toBeTruthy();
+        });
+
+        it('should return false when there are no relevant subscribers', () => {
+            expect(onyxSubscriptionManager.hasListenersForKey('someUnwatchedKey')).toBeFalsy();
+        });
+
+        it('should return false after the last listener unsubscribes', () => {
+            const unsubscribe = onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, jest.fn());
+            unsubscribe();
+            expect(onyxSubscriptionManager.hasListenersForKey(ONYXKEYS.TEST_KEY)).toBeFalsy();
+        });
+    });
+
+    describe('clearAll', () => {
+        it('should wipe key and collection subscriptions', () => {
+            const keyCallback = jest.fn();
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, keyCallback);
+
+            onyxSubscriptionManager.clearAll();
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'x');
+
+            expect(keyCallback).not.toHaveBeenCalled();
+            expect(onyxSubscriptionManager.hasListenersForKey(ONYXKEYS.TEST_KEY)).toBeFalsy();
+        });
+    });
+
+    describe('subscription mutation during dispatch', () => {
+        it('should fire a listener that unsubscribes and re-subscribes itself during dispatch only once', () => {
+            let unsubscribe: () => void = jest.fn();
+            const callback = jest.fn(() => {
+                unsubscribe();
+                unsubscribe = onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback);
+            });
+            unsubscribe = onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, callback);
+
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'x');
+
+            expect(callback).toHaveBeenCalledTimes(1);
+        });
+
+        it('should not deliver the in-flight notification to a listener added during dispatch', () => {
+            const lateCallback = jest.fn();
+            const firstCallback = jest.fn(() => {
+                onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, lateCallback);
+            });
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, firstCallback);
+
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'first');
+            expect(lateCallback).not.toHaveBeenCalled();
+
+            // It receives later notifications normally.
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'second');
+            expect(lateCallback).toHaveBeenCalledTimes(1);
+            expect(lateCallback).toHaveBeenCalledWith('second', ONYXKEYS.TEST_KEY);
+        });
+
+        it('should still fire a sibling unsubscribed during dispatch this round, but not on later notifications', () => {
+            const siblingCallback = jest.fn();
+            let unsubscribeSibling: () => void = jest.fn();
+            const firstCallback = jest.fn(() => {
+                unsubscribeSibling();
+            });
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, firstCallback);
+            unsubscribeSibling = onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, siblingCallback);
+
+            // The sibling was registered when dispatch began, so the snapshot still fires it.
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'first');
+            expect(siblingCallback).toHaveBeenCalledTimes(1);
+
+            // Now unsubscribed, it does not fire again.
+            onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'second');
+            expect(siblingCallback).toHaveBeenCalledTimes(1);
+        });
+    });
+
+    describe('listener error isolation', () => {
+        it('should log a throwing listener and still fire the other listeners', () => {
+            const logAlertSpy = jest.spyOn(Logger, 'logAlert').mockImplementation(() => {
+                /* empty */
+            });
+            const throwingCallback = jest.fn(() => {
+                throw new Error('boom');
+            });
+            const healthyCallback = jest.fn();
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, throwingCallback);
+            onyxSubscriptionManager.subscribe(ONYXKEYS.TEST_KEY, healthyCallback);
+
+            expect(() => onyxSubscriptionManager.notifyKey(ONYXKEYS.TEST_KEY, 'x')).not.toThrow();
+
+            expect(throwingCallback).toHaveBeenCalledTimes(1);
+            expect(healthyCallback).toHaveBeenCalledTimes(1);
+            expect(logAlertSpy).toHaveBeenCalled();
+        });
+    });
+});