RxJS watchKeys operator for tracking added, removed, and persisted items within an object

watch-keys.operator.ts

import { Observable, Subscriber } from 'rxjs';

type Dictionary<T> = {
  [key: string]: T;
};

export type AddedEvent<T> = {
  type: 'added';
  value: T;
};

export type PersistEvent<T> = {
  type: 'persist';
  value: T[];
};

export type RemovedEvent<T> = {
  type: 'removed';
  value: T;
};

export type WatchKeysEvent<T>
  = AddedEvent<T>
  | PersistEvent<T>
  | RemovedEvent<T>;

export const watchKeys = <T>() => {
  const items = new Map<string, T>();

  return (source: Observable<Dictionary<T>>) =>
    new Observable((observer: Subscriber<WatchKeysEvent<T>>) =>
      source.subscribe(obj => {
        const keys = Object.keys(obj);
        const addedKeys = keys.filter(k => !items.has(k));
        const removedKeys = [...items.keys()].filter(k => !keys.includes(k));
        // Process the keys that were added since last event
        addedKeys.map(k => obj[k]).forEach(value => observer.next({ type: 'added', value }));

        // Process the keys that were removed since last event
        removedKeys.map(k => items.get(k) as T).forEach(value => observer.next({ type: 'removed', value }));

        // Emit the persisted objects
        observer.next({
          type: 'persist',
          value: keys.filter(k => !addedKeys.includes(k) && !removedKeys.includes(k)).map(k => obj[k]),
        });

        // Reset and update the tracking map
        items.clear();
        keys.forEach(id => items.set(id, obj[id]));
      })
    );
};