example-usage.ts

  private subscribeToDirtyChanges() {

    const decks = this.query.ui.getValue();
    if (DLP_DEBUG) {
      console.warn('DPR: decks state loaded', decks);
    }

    this.dirtyCheck = new RebaseableEntityDirtyCheckPlugin(this.query, {
      comparator: (head, curr) => {
        const result = DeckDiffUtil.diff(curr, head);
        const different = DiffUtil.isDifferent(result);
        if (different) {
          if (DLP_DEBUG) {
            console.error('DPR: Decks differ: ', result);
          }
        }
        return different;
      }
    }).setHead(decks.ids);

    const states: Map<string, Deck<Card>> = new Map();
    for (const id of Object.keys(decks.entities)) {
      if (decks.entities[id].head) {
        states.set(id, decks.entities[id].head);
      }
    }

    if (DLP_DEBUG) {
      console.log('DPR: states', states);
    }

    this.dirtyCheck.dirty$
      .pipe(take(1))
      .pipe(withTransaction<string[]>((dirty) => {
        if (DLP_DEBUG) {
          console.log('DPR: rebasing states', states);
        }
        // FIXME: This typing does not seem right
        this.dirtyCheck.rebaseAll(states as any);
      }))
      .subscribe();


    this._subs.sink = this.dirtyCheck.dirty$
      .pipe(withTransaction<string[]>((dirty) => {
        if (DLP_DEBUG) {
          console.warn('DPR: DIRTY', dirty);
        }
        for (const id of dirty) {
          const head: Deck<Card> = this.dirtyCheck.getHead(id) as any;
          if (DLP_DEBUG) {
            console.warn('DPR: saving, head', id, head);
          }
          this.updateDeckUI(id, {
            head
          });
        }
        for (const id of this.query.getValue().ids) {
          const uiHead: Deck<Card> = this.query.ui.getEntity(id).head;
          if (uiHead && !this.dirtyCheck.isDirty(id, false)) {
            if (DLP_DEBUG) {
              console.warn('DPR: clearing head', id, uiHead);
            }
            this.updateDeckUI(id, {
              head: null
            });
          }
        }
      })).subscribe();

    this._subs.sink = this.sync.synced$.subscribe(async event => {
      // TODO: Review code for updating HEAD state based on Sync events:
      // This is currently required so that the edit-details changes (and dirtyCheck) show the difference
      // between the server and the local copy in the NGRX store. It is also here so that we can
      // revert to the right thing. This *may* not be necessary if we keep separate UI state for sync
      // conflicts, but I think the current approach may be cleaner (what's below)

      for (const deck of event.updated) {
        await this.dirtyCheck.rebaseOne(deck.id, deck as any);
      }

      for (const deck of event.saved) {
        await this.dirtyCheck.rebaseOne(deck.id, deck as any);
      }

      for (const conflict of event.obsolete) {
        // FIXME: This typing does not seem right
        if (conflict.remote && conflict.remote.id) {
          await this.dirtyCheck.rebaseOne(conflict.remote.id, conflict.remote as any);
        }
      }

    });
  }

rebaseable-entity-changes.plugin.ts

import { BehaviorSubject, combineLatest, merge, Observable, Subject, Subscription } from 'rxjs';
import { auditTime, distinctUntilChanged, map, skip } from 'rxjs/operators';
import { environment } from 'src/environments/environment';

import {
  AkitaPlugin, coerceArray, EntityCollectionPlugin, EntityState, getEntityType, getIDType,
  getNestedPath, isFunction, isUndefined, logAction, OrArray, Queries, Query, QueryEntity
} from '@datorama/akita';

type Head<State = any> = State | Partial<State>;

export type DirtyCheckComparator<State> = (head: State, current: State) => boolean;

export interface DirtyCheckParams<StoreState = any> {
  comparator?: DirtyCheckComparator<StoreState>;
  watchProperty?: keyof StoreState | (keyof StoreState)[];
}

export const dirtyCheckDefaultParams = {
  comparator: (head, current) => JSON.stringify(head) !== JSON.stringify(current)
};

export interface DirtyCheckResetParams<StoreState = any> {
  updateFn?: StoreState | ((head: StoreState, current: StoreState) => any);
}

export interface DirtyCheckCollectionParams<State extends EntityState> {
  comparator?: DirtyCheckComparator<getEntityType<State>>;
  entityIds?: OrArray<getIDType<State>>;
}

const ECP_DEBUG = false && !environment.production;

// tslint:disable-next-line:max-line-length
export class RebaseableEntityDirtyCheckPlugin<State extends EntityState = any, P extends DirtyCheckPlugin<State> = DirtyCheckPlugin<State>> extends EntityCollectionPlugin<State, P> {
  private _someDirty = new Subject();

  dirty$: Observable<getIDType<State>[]> = merge(this.query.select(state => state.entities), this._someDirty.asObservable()).pipe(
    auditTime(0),
    map(() => this.getDirtyIds().sort()),
    distinctUntilChanged()
  );

  someDirty$: Observable<boolean> = this.dirty$.pipe(map(ids => ids.length > 0));

  constructor(protected query: QueryEntity<State>, private readonly params: DirtyCheckCollectionParams<State> = {}) {
    super(query, params.entityIds);
    this.params = { ...dirtyCheckDefaultParams, ...params };
    // TODO lazy activate?
    this.activate();
    this.selectIds()
      .pipe(skip(1))
      .subscribe(ids => {
        super.rebase(ids, { afterAdd: plugin => plugin.setHead() });
      });
  }

  getHead(id: getIDType<State>) {
    if (this.entities.has(id)) {
      const entity = this.getEntity(id);
      return entity.getHead();
    }
    return undefined;
  }

  rebaseOne(id: getIDType<State>, state: State) {
    if (this.entities.has(id)) {
      const entity = this.getEntity(id);
      if (ECP_DEBUG) {
        console.log('ECP: Setting head - rebaseOne', state);
      }
      entity.setHead(state);
    }
    this._someDirty.next();
    return this;
  }

  rebaseAll(states: Map<getIDType<State>, State>) {
    for (const id of states.keys()) {
      if (this.entities.has(id)) {
        const entity = this.getEntity(id);
        const head = states.get(id);
        if (ECP_DEBUG) {
          console.log('ECP: Setting head - rebaseAll', head);
        }
        entity.setHead(head);
      }
    }
    this._someDirty.next();
    return this;
  }

  setHead(ids?: OrArray<getIDType<State>>) {
    if (this.params.entityIds && ids) {
      const toArray = coerceArray(ids) as getIDType<State>[];
      const someAreWatched = coerceArray(this.params.entityIds).some(id => toArray.indexOf(id) > -1);
      if (someAreWatched === false) {
        return this;
      }
    }
    if (ECP_DEBUG) {
      console.log('ECP: Setting all heads', ids);
    }
    this.forEachId(ids, e => e.setHead());
    this._someDirty.next();
    return this;
  }

  hasHead(id: getIDType<State>): boolean {
    if (this.entities.has(id)) {
      const entity = this.getEntity(id);
      return entity.hasHead();
    }

    return false;
  }

  reset(ids?: OrArray<getIDType<State>>, params: DirtyCheckResetParams = {}) {
    this.forEachId(ids, e => e.reset(params));
  }

  isDirty(id: getIDType<State>): Observable<boolean>;
  // tslint:disable-next-line:unified-signatures
  isDirty(id: getIDType<State>, asObservable: true): Observable<boolean>;
  isDirty(id: getIDType<State>, asObservable: false): boolean;
  isDirty(id: getIDType<State>, asObservable = true): Observable<boolean> | boolean {
    if (this.entities.has(id)) {
      const entity = this.getEntity(id);
      return asObservable ? entity.isDirty$ : entity.isDirty();
    }

    return false;
  }

  someDirty(): boolean {
    return this.checkSomeDirty();
  }

  isPathDirty(id: getIDType<State>, path: string) {
    if (this.entities.has(id)) {
      const head = (this.getEntity(id) as any).getHead();
      if (ECP_DEBUG) {
        console.log('ECP: Checking dirty state. Head:', head);
      }
      const current = this.query.getEntity(id);
      const currentPathValue = getNestedPath(current, path);
      const headPathValue = getNestedPath(head, path);

      if (ECP_DEBUG) {
        console.log('ECP: Checking dirty state. Head path:', headPathValue);
      }
      return this.params.comparator(currentPathValue, headPathValue);
    }

    return null;
  }

  destroy(ids?: OrArray<getIDType<State>>) {
    this.forEachId(ids, e => e.destroy());
    /** complete only when the plugin destroys */
    if (!ids) {
      this._someDirty.complete();
    }
  }

  protected instantiatePlugin(id: getIDType<State>): P {
    return new DirtyCheckPlugin(this.query, this.params as any, id) as P;
  }

  private checkSomeDirty(): boolean {
    const entitiesIds = this.resolvedIds();
    for (const id of entitiesIds) {
      if (this.getEntity(id).isDirty()) {
        return true;
      }
    }
    return false;
  }

  private getDirtyIds(): getIDType<State>[] {
    return this.resolvedIds().filter(id => this.getEntity(id).isDirty());
  }
}



/**
 * OVERRIDING THIS SO WE CAN EXPOSE getHead() publicly.
 */
class DirtyCheckPlugin<State = any> extends AkitaPlugin<State> {
  private head: Head<State>;
  private dirty = new BehaviorSubject(false);
  private subscription: Subscription;
  private active = false;
  private _reset = new Subject();

  isDirty$: Observable<boolean> = this.dirty.asObservable().pipe(distinctUntilChanged());
  reset$ = this._reset.asObservable();

  constructor(protected query: Queries<State>, private params?: DirtyCheckParams<State>, private _entityId?: any) {
    super(query);
    this.params = { ...dirtyCheckDefaultParams, ...params };
    if (this.params.watchProperty) {
      const watchProp = coerceArray(this.params.watchProperty) as any[];
      if (query instanceof QueryEntity && watchProp.includes('entities') && !watchProp.includes('ids')) {
        watchProp.push('ids');
      }
      this.params.watchProperty = watchProp;
    }
    if (ECP_DEBUG) {
      console.log('ECP: Entity based?', this._entityId, this.isEntityBased(this._entityId));
    }
  }

  reset(params: DirtyCheckResetParams = {}) {
    let currentValue = this.head;
    if (isFunction(params.updateFn)) {
      if (this.isEntityBased(this._entityId)) {
        currentValue = params.updateFn(this.head, (this.getQuery() as QueryEntity<State>).getEntity(this._entityId));
      } else {
        currentValue = params.updateFn(this.head, (this.getQuery() as Query<State>).getValue());
      }
    }
    logAction(`@DirtyCheck - Revert`);
    this.updateStore(currentValue, this._entityId);
    this._reset.next();
  }

  setHead(state?: State) {
    if (!this.active) {
      this.activate(state);
      this.active = true;
      if (ECP_DEBUG) {
        console.log('ECP: setting head -- activated', state, this.head);
      }
    } else {
      this.head = state ? state : this._getHead();
      if (ECP_DEBUG) {
        console.log('ECP: setting head -- not active', this.head);
      }
    }

    let currentValue: State = null;
    if (this.isEntityBased(this._entityId)) {
      currentValue = (this.getQuery() as QueryEntity<State>).getEntity(this._entityId) as State;
    } else {
      currentValue = (this.getQuery() as Query<State>).getValue();
    }
    const head = this._getHead() as State;
    const dirty = state ? this.params.comparator(head, currentValue) : false;
    if (ECP_DEBUG) {
      console.log('ECP: got head -- updateDirtiness', dirty, head, currentValue);
    }
    this.updateDirtiness(dirty);
    return this;
  }

  isDirty(): boolean {
    return !!this.dirty.value;
  }

  hasHead() {
    return !!this.getHead();
  }

  destroy() {
    if (ECP_DEBUG) {
      console.log('ECP: Clearing head - rebaseOne', this.head);
    }
    this.head = null;
    // tslint:disable-next-line:no-unused-expression
    this.subscription && this.subscription.unsubscribe();
    // tslint:disable-next-line:no-unused-expression
    this._reset && this._reset.complete();
  }

  isPathDirty(path: string) {
    const head = this.getHead();
    if (ECP_DEBUG) {
      console.log('ECP: Checking dirty state. Head:', head);
    }
    const current = (this.getQuery() as Query<State>).getValue();
    const currentPathValue = getNestedPath(current, path);
    const headPathValue = getNestedPath(head, path);
    if (ECP_DEBUG) {
      console.log('ECP: Checking dirty state. Head path:', headPathValue);
    }

    return this.params.comparator(currentPathValue, headPathValue);
  }

  public getHead() {
    return this.head;
  }

  private activate(initialHead?: State) {
    if (initialHead) {
      this.head = initialHead;
      if (ECP_DEBUG) {
        console.log('ECP: Set initial head:', this.head);
      }
    } else {
      this.head = this._getHead();
      if (ECP_DEBUG) {
        console.log('ECP: Set initial head -- default:', this.head);
      }
    }
    /** if we are tracking specific properties select only the relevant ones */
    const source = this.params.watchProperty
      ? (this.params.watchProperty as (keyof State)[]).map(prop =>
        this.query
          .select(state => state[prop])
          .pipe(
            map(val => ({
              val,
              __akitaKey: prop
            }))
          )
      )
      : [this.selectSource(this._entityId)];
    this.subscription = combineLatest(...source)
      .pipe(skip(1))
      .subscribe((currentState: any[]) => {
        if (isUndefined(this.head)) { return; }
        /** __akitaKey is used to determine if we are tracking a specific property or a store change */
        const isChange = currentState.some(state => {
          const head = state.__akitaKey ? this.head[state.__akitaKey as any] : this.head;
          if (ECP_DEBUG) {
            console.log('ECP: Got head to compare:', head);
          }
          const compareTo = state.__akitaKey ? state.val : state;

          // console.error('COMPARE TO', state.__akitaKey, '<<<' , compareTo );
          return this.params.comparator(head, compareTo);
        });

        this.updateDirtiness(isChange);
      });
  }

  private updateDirtiness(isDirty: boolean) {
    this.dirty.next(isDirty);
  }

  private _getHead(): Head<State> {
    let head: Head<State> = this.getSource(this._entityId);
    if (this.params.watchProperty) {
      head = this.getWatchedValues(head as State);
    }
    return head;
  }

  private getWatchedValues(source: State): Partial<State> {
    return (this.params.watchProperty as (keyof State)[]).reduce(
      (watched, prop) => {
        watched[prop] = source[prop];
        return watched;
      },
      {} as Partial<State>
    );
  }
}