repo_description: A fast, local first, reactive Database for JavaScript Applications https://rxdb.info/
repo_topics: angular browser-database couchdb crdt database firebase graphql indexeddb ionic local-first localstorage nodejs nosql postgres pouchdb react-native realtime realtime-database rxdb sqlite
code: import { firstValueFrom, filter } from 'rxjs';
  import {
      getChangedDocumentsSince,
      getWrittenDocumentsFromBulkWriteResponse,
      stackCheckpoints
  } from '../rx-storage-helper.ts';
  import type {
      BulkWriteRow,
      BulkWriteRowById,
      ById,
      EventBulk,
      RxDocumentData,
      RxError,
      RxReplicationWriteToMasterRow,
      RxStorageChangeEvent,
      RxStorageInstanceReplicationState,
      RxStorageReplicationMeta,
      WithDeleted
  } from '../types/index.d.ts';
  import {
      batchArray,
      clone,
      ensureNotFalsy,
      getHeightOfRevision,
      PROMISE_RESOLVE_FALSE
  } from '../plugins/utils/index.ts';
  import {
      getLastCheckpointDoc,
      setCheckpoint
  } from './checkpoint.ts';
  import {
      resolveConflictError
  } from './conflicts.ts';
  import {
      stripAttachmentsDataFromMetaWriteRows,
      writeDocToDocState
  } from './helper.ts';
  import {
      getAssumedMasterState,
      getMetaWriteRow
  } from './meta-instance.ts';
  import { fillWriteDataForAttachmentsChange } from '../plugins/attachments/index.ts';
  import { newRxError } from '../rx-error.ts';
  
  /**
   * Writes all document changes from the fork to the master.
   * The upstream runs on two modes:
   * - For initial replication, a checkpoint-iteration is used
   * - For ongoing local writes, we just subscribe to the changeStream of the fork.
   *   In contrast to the master, the fork can be assumed to never loose connection,
   *   so we do not have to prepare for missed out events.
   */
  export async function startReplicationUpstream<RxDocType, CheckpointType>(
      state: RxStorageInstanceReplicationState<RxDocType>
  ) {
      if (
          state.input.initialCheckpoint &&
          state.input.initialCheckpoint.upstream
      ) {
          const checkpointDoc = await getLastCheckpointDoc(state, 'up');
          if (!checkpointDoc) {
              await setCheckpoint(
                  state,
                  'up',
                  state.input.initialCheckpoint.upstream
              );
          }
      }
  
      const replicationHandler = state.input.replicationHandler;
      state.streamQueue.up = state.streamQueue.up.then(() => {
          return upstreamInitialSync().then(() => {
              return processTasks();
          });
      });
  
      // used to detect which tasks etc can in it at which order.
      let timer = 0;
      let initialSyncStartTime = -1;
  
      type Task = EventBulk<RxStorageChangeEvent<RxDocType>, any> | 'RESYNC';
      type TaskWithTime = {
          task: Task;
          time: number;
      };
      const openTasks: TaskWithTime[] = [];
      let persistenceQueue: Promise<boolean> = PROMISE_RESOLVE_FALSE;
      const nonPersistedFromMaster: {
          checkpoint?: CheckpointType;
          docs: ById<RxDocumentData<RxDocType>>;
      } = {
          docs: {}
      };
  
      const sub = state.input.forkInstance.changeStream()
          .subscribe((eventBulk: EventBulk<RxStorageChangeEvent<RxDocType>, any>) => {
              if (state.events.paused.getValue()) {
                  return;
              }
  
  
              state.stats.up.forkChangeStreamEmit = state.stats.up.forkChangeStreamEmit + 1;
              openTasks.push({
                  task: eventBulk,
                  time: timer++
              });
              if (!state.events.active.up.getValue()) {
                  state.events.active.up.next(true);
              }
              if (state.input.waitBeforePersist) {
                  return state.input.waitBeforePersist()
                      .then(() => processTasks());
              } else {
                  return processTasks();
              }
          });
      const subResync = replicationHandler
          .masterChangeStream$
          .pipe(
              filter((ev: any) => ev === 'RESYNC')
          )
          .subscribe(() => {
              openTasks.push({
                  task: 'RESYNC',
                  time: timer++
              });
              processTasks();
          });
  
      // unsubscribe when replication is canceled
      firstValueFrom(
          state.events.canceled.pipe(
              filter((canceled: boolean) => !!canceled)
          )
      ).then(() => {
          sub.unsubscribe();
          subResync.unsubscribe();
      });
  
  
      async function upstreamInitialSync() {
          state.stats.up.upstreamInitialSync = state.stats.up.upstreamInitialSync + 1;
          if (state.events.canceled.getValue()) {
              return;
          }
  
          state.checkpointQueue = state.checkpointQueue.then(() => getLastCheckpointDoc(state, 'up'));
          let lastCheckpoint: CheckpointType = await state.checkpointQueue;
  
          const promises: Set<Promise<any>> = new Set();
  
          while (!state.events.canceled.getValue()) {
              initialSyncStartTime = timer++;
  
              /**
               * Throttle the calls to
               * forkInstance.getChangedDocumentsSince() so that
               * if the pushing to the remote is slower compared to the
               * pulling out of forkInstance, we do not block the UI too much
               * and have a big memory spike with all forkInstance documents.
               */
              if (promises.size > 3) {
                  await Promise.race(Array.from(promises));
              }
              const upResult = await getChangedDocumentsSince(
                  state.input.forkInstance,
                  state.input.pushBatchSize,
                  lastCheckpoint
              );
              if (upResult.documents.length === 0) {
                  break;
              }
  
              lastCheckpoint = stackCheckpoints([lastCheckpoint, upResult.checkpoint]);
  
              const promise = persistToMaster(
                  upResult.documents,
                  ensureNotFalsy(lastCheckpoint)
              );
              promises.add(promise);
              promise.catch().then(() => promises.delete(promise));
          }
  
          /**
           * If we had conflicts during the initial sync,
           * it means that we likely have new writes to the fork
           * and so we have to run the initial sync again to upstream these new writes.
           */
          const resolvedPromises = await Promise.all(promises);
          const hadConflicts = resolvedPromises.find(r => !!r);
          if (hadConflicts) {
              await upstreamInitialSync();
          } else if (
              !state.firstSyncDone.up.getValue() &&
              !state.events.canceled.getValue()
          ) {
              state.firstSyncDone.up.next(true);
          }
      }
  
  
      /**
       * Takes all open tasks an processes them at once.
       */
      function processTasks() {
          if (
              state.events.canceled.getValue() ||
              openTasks.length === 0
          ) {
              state.events.active.up.next(false);
              return;
          }
          state.stats.up.processTasks = state.stats.up.processTasks + 1;
          state.events.active.up.next(true);
          state.streamQueue.up = state.streamQueue.up.then(async () => {
              /**
               * Merge/filter all open tasks
               */
              let docs: RxDocumentData<RxDocType>[] = [];
              let checkpoint: CheckpointType | undefined;
              while (openTasks.length > 0) {
                  const taskWithTime = ensureNotFalsy(openTasks.shift());
                  /**
                   * If the task came in before the last time the initial sync fetching
                   * has run, we can ignore the task because the initial sync already processed
                   * these documents.
                   */
                  if (taskWithTime.time < initialSyncStartTime) {
                      continue;
                  }
  
                  if (taskWithTime.task === 'RESYNC') {
                      state.events.active.up.next(false);
                      await upstreamInitialSync();
                      return;
                  }
  
                  /**
                   * If the task came from the downstream, we can ignore these documents
                   * because we know they are replicated already.
                   * But even if they can be ignored, we later have to call persistToMaster()
                   * to have the correct checkpoint set.
                   */
                  if (taskWithTime.task.context !== await state.downstreamBulkWriteFlag) {
                      docs = docs.concat(taskWithTime.task.events.map(r => {
                          return r.documentData as any;
                      }));
                  }
                  checkpoint = stackCheckpoints([checkpoint, taskWithTime.task.checkpoint]);
              }
  
              await persistToMaster(
                  docs,
                  checkpoint as any
              );
  
              // might have got more tasks while running persistToMaster()
              if (openTasks.length === 0) {
                  state.events.active.up.next(false);
              } else {
                  return processTasks();
              }
          });
      }
  
      /**
       * Returns true if had conflicts,
       * false if not.
       */
      function persistToMaster(
          docs: RxDocumentData<RxDocType>[],
          checkpoint: CheckpointType
      ): Promise<boolean> {
          state.stats.up.persistToMaster = state.stats.up.persistToMaster + 1;
  
          /**
           * Add the new docs to the non-persistent list
           */
          docs.forEach(docData => {
              const docId: string = (docData as any)[state.primaryPath];
              nonPersistedFromMaster.docs[docId] = docData;
          });
          nonPersistedFromMaster.checkpoint = checkpoint;
  
          persistenceQueue = persistenceQueue.then(async () => {
              if (state.events.canceled.getValue()) {
                  return false;
              }
  
              const upDocsById: ById<RxDocumentData<RxDocType>> = nonPersistedFromMaster.docs;
              nonPersistedFromMaster.docs = {};
              const useCheckpoint = nonPersistedFromMaster.checkpoint;
              const docIds = Object.keys(upDocsById);
              /**
               * Even if we do not have anything to push,
               * we still have to store the up-checkpoint.
               * This ensures that when many documents have been pulled
               * from the remote (that do not have to be pushed again),
               * we continue at the correct position and do not have to load
               * these documents from the storage again when the replication is restarted.
               */
              function rememberCheckpointBeforeReturn() {
                  return setCheckpoint(
                      state,
                      'up',
                      useCheckpoint
                  );
              };
  
  
              if (docIds.length === 0) {
                  rememberCheckpointBeforeReturn();
                  return false;
              }
  
              const assumedMasterState = await getAssumedMasterState(
                  state,
                  docIds
              );
  
              const writeRowsToMaster: ById<RxReplicationWriteToMasterRow<RxDocType>> = {};
              const writeRowsToMasterIds: string[] = [];
              const writeRowsToMeta: BulkWriteRowById<RxStorageReplicationMeta<RxDocType, any>> = {};
              const forkStateById: ById<RxDocumentData<RxDocType>> = {};
  
              await Promise.all(
                  docIds.map(async (docId) => {
                      const fullDocData: RxDocumentData<RxDocType> = upDocsById[docId];
                      forkStateById[docId] = fullDocData;
                      const docData: WithDeleted<RxDocType> = writeDocToDocState(fullDocData, state.hasAttachments, !!state.input.keepMeta);
                      const assumedMasterDoc = assumedMasterState[docId];
  
                      /**
                       * If the master state is equal to the
                       * fork state, we can assume that the document state is already
                       * replicated.
                       */
                      if (
                          (
                              assumedMasterDoc &&
                              // if the isResolvedConflict is correct, we do not have to compare the documents.
                              assumedMasterDoc.metaDocument.isResolvedConflict !== fullDocData._rev
                              &&
                              (
                                  state.input.conflictHandler.isEqual(
                                      assumedMasterDoc.docData,
                                      docData,
                                      'upstream-check-if-equal'
                                  )
                              )
                          )
                          ||
                          /**
                           * If the master works with _rev fields,
                           * we use that to check if our current doc state
                           * is different from the assumedMasterDoc.
                           */
                          (
                              assumedMasterDoc &&
                              (assumedMasterDoc.docData as any)._rev &&
                              getHeightOfRevision(fullDocData._rev) === fullDocData._meta[state.input.identifier]
                          )
                      ) {
                          return;
                      }
  
                      writeRowsToMasterIds.push(docId);
  
                      writeRowsToMaster[docId] = {
                          assumedMasterState: assumedMasterDoc ? assumedMasterDoc.docData : undefined,
                          newDocumentState: docData
                      };
                      writeRowsToMeta[docId] = await getMetaWriteRow(
                          state,
                          docData,
                          assumedMasterDoc ? assumedMasterDoc.metaDocument : undefined
                      );
                  })
              );
  
              if (writeRowsToMasterIds.length === 0) {
                  rememberCheckpointBeforeReturn();
                  return false;
              }
  
  
              const writeRowsArray = Object.values(writeRowsToMaster);
              const conflictIds: Set<string> = new Set();
              const conflictsById: ById<WithDeleted<RxDocType>> = {};
  
              /**
               * To always respect the push.batchSize,
               * we have to split the write rows into batches
               * to ensure that replicationHandler.masterWrite() is never
               * called with more documents than what the batchSize limits.
               */
              const writeBatches = batchArray(writeRowsArray, state.input.pushBatchSize);
              await Promise.all(
                  writeBatches.map(async (writeBatch) => {
  
                      // enhance docs with attachments
                      if (state.hasAttachments) {
                          await Promise.all(
                              writeBatch.map(async (row) => {
                                  row.newDocumentState = await fillWriteDataForAttachmentsChange(
                                      state.primaryPath,
                                      state.input.forkInstance,
                                      clone(row.newDocumentState),
                                      row.assumedMasterState
                                  );
                              })
                          );
                      }
                      const masterWriteResult = await replicationHandler.masterWrite(writeBatch);
                      masterWriteResult.forEach(conflictDoc => {
                          const id = (conflictDoc as any)[state.primaryPath];
                          conflictIds.add(id);
                          conflictsById[id] = conflictDoc;
                      });
                  })
              );
  
              /**
               * Check for canceled or paused state before marking documents
               * as successfully pushed. When the replication is paused during
               * a push retry, masterWrite() returns [] even though no documents
               * were actually sent. Without this check the meta instance would
               * be updated, causing the documents to never be retried on resume.
               */
              if (state.events.canceled.getValue() || state.events.paused.getValue()) {
                  return false;
              }
  
              const useWriteRowsToMeta: BulkWriteRow<RxStorageReplicationMeta<RxDocType, any>>[] = [];
  
              writeRowsToMasterIds.forEach(docId => {
                  if (!conflictIds.has(docId)) {
                      /**
                       * Skip the processed.up emission for rows that were
                       * filtered out by the replicationHandler.masterWrite()
                       * wrapper (e.g. via a push modifier returning null).
                       * Such rows have their newDocumentState set to null to
                       * signal they were not actually sent to the master.
                       * We still persist their meta state so that the upstream
                       * does not keep retrying them.
                       */
                      if (writeRowsToMaster[docId].newDocumentState !== null) {
                          state.events.processed.up.next(writeRowsToMaster[docId]);
                      }
                      useWriteRowsToMeta.push(writeRowsToMeta[docId]);
                  }
              });
  
              if (!state.skipStoringPullMeta && useWriteRowsToMeta.length > 0) {
                  await state.input.metaInstance.bulkWrite(
                      stripAttachmentsDataFromMetaWriteRows(state, useWriteRowsToMeta),
                      'replication-up-write-meta'
                  );
                  // TODO what happens when we have conflicts here?
              }
  
              /**
               * Resolve conflicts by writing a new document
               * state to the fork instance and the 'real' master state
               * to the meta instance.
               * Non-409 errors will be detected by resolveConflictError()
               */
              let hadConflictWrites = false;
              if (conflictIds.size > 0) {
                  state.stats.up.persistToMasterHadConflicts = state.stats.up.persistToMasterHadConflicts + 1;
                  const conflictWriteFork: BulkWriteRow<RxDocType>[] = [];
                  const conflictWriteMeta: BulkWriteRowById<RxStorageReplicationMeta<RxDocType, any>> = {};
                  await Promise.all(
                      Object
                          .entries(conflictsById)
                          .map(([docId, realMasterState]) => {
                              const writeToMasterRow = writeRowsToMaster[docId];
                              const input = {
                                  newDocumentState: writeToMasterRow.newDocumentState,
                                  assumedMasterState: writeToMasterRow.assumedMasterState,
                                  realMasterState
                              };
                              return resolveConflictError(
                                  state,
                                  input,
                                  forkStateById[docId]
                              ).then(async (resolved) => {
                                  if (resolved) {
                                      state.events.resolvedConflicts.next({
                                          input,
                                          output: resolved
                                      });
                                      conflictWriteFork.push({
                                          previous: forkStateById[docId],
                                          document: resolved
                                      });
                                      const assumedMasterDoc = assumedMasterState[docId];
                                      conflictWriteMeta[docId] = await getMetaWriteRow(
                                          state,
                                          ensureNotFalsy(realMasterState),
                                          assumedMasterDoc ? assumedMasterDoc.metaDocument : undefined,
                                          resolved._rev
                                      );
                                  }
                              });
                          })
                  );
  
                  if (conflictWriteFork.length > 0) {
                      hadConflictWrites = true;
  
                      state.stats.up.persistToMasterConflictWrites = state.stats.up.persistToMasterConflictWrites + 1;
                      const forkWriteResult = await state.input.forkInstance.bulkWrite(
                          conflictWriteFork,
                          'replication-up-write-conflict'
                      );
  
                      let mustThrow: RxError | undefined;
                      forkWriteResult.error.forEach(error => {
                          /**
                           * Conflict-Errors in the forkWriteResult must not be handled
                           * because they have been caused by a write to the forkInstance
                           * in between which will anyway trigger a new upstream cycle
                           * that will then resolved the conflict again.
                           */
                          if (error.status === 409) {
                              return;
                          }
                          // other non-conflict errors must be handled
                          const throwMe = newRxError('RC_PUSH', {
                              writeError: error
                          });
                          state.events.error.next(throwMe);
                          mustThrow = throwMe;
                      });
                      if (mustThrow) {
                          throw mustThrow;
                      }
  
                      const useMetaWrites: BulkWriteRow<RxStorageReplicationMeta<RxDocType, any>>[] = [];
                      const success = getWrittenDocumentsFromBulkWriteResponse(
                          state.primaryPath,
                          conflictWriteFork,
                          forkWriteResult
                      );
                      success
                          .forEach(docData => {
                              const docId = (docData as any)[state.primaryPath];
                              useMetaWrites.push(
                                  conflictWriteMeta[docId]
                              );
                          });
                      if (useMetaWrites.length > 0) {
                          await state.input.metaInstance.bulkWrite(
                              stripAttachmentsDataFromMetaWriteRows(state, useMetaWrites),
                              'replication-up-write-conflict-meta'
                          );
                      }
                      // TODO what to do with conflicts while writing to the metaInstance?
                  }
              }
  
              /**
               * For better performance we do not await checkpoint writes,
               * but to ensure order on parallel checkpoint writes,
               * we have to use a queue.
               */
              rememberCheckpointBeforeReturn();
  
              return hadConflictWrites;
          }).catch(unhandledError => {
              state.events.error.next(unhandledError);
              return false;
          });
  
          return persistenceQueue;
      }
  }
  