123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258
import { type IConnection, SuperbusMap } from "../../deps.ts";import { Peer } from "../peer/peer.ts";import { makeSyncerBag, SyncerBag } from "./_syncer-bag.ts";import { ShareAddress } from "../util/doc-types.ts";import {  ShareQueryRequest,  ShareState,  SyncSessionStatus,} from "./syncer-types.ts";
/** Orchestrates different requests in order to syncrhronise a Peer using a connection */export class SyncCoordinator {  private connection: IConnection<SyncerBag>;  private syncerBag: SyncerBag;  private shareStates: Record<ShareAddress, ShareState> = {};  private peerReplicaMapUnsub: () => void;  private pullTimeouts: Map<string, number> = new Map();
  /** A subscribable map of shares and the status of their synchronisation operations. */  syncStatuses: SuperbusMap<ShareAddress, SyncSessionStatus> =    new SuperbusMap();
  partnerLastSeenAt: number | null = null;  state: "ready" | "active" | "closed" = "ready";
  /** The shares which this SyncCoordinator has in common with the peer at the other end of the connection. */  get commonShares() {    return Array.from(this.syncStatuses.keys());  }
  constructor(peer: Peer, connection: IConnection<SyncerBag>) {    this.syncerBag = makeSyncerBag(peer);    this.connection = connection;
    this.peerReplicaMapUnsub = peer.replicaMap.bus.on("*", () => {      this.performSaltedHandshake().then(() => {        this.getShareStates();      });    });  }
  /** Start the coordinator - establish common shares and begin pulling   * @returns - A promise for an initial pull of all shares.   */  async start() {    this.state = "active";
    await this.performSaltedHandshake();
    // Get the share states from the partner
    this.connection.onClose(() => {      this.close();    });
    await this.getShareStates();  }
  async performSaltedHandshake() {    // Perform salted handshake
    const saltedHandshakeRes = await this.connection.request(      "serveSaltedHandshake",    );
    const { commonShares, partnerLastSeenAt } = await this.syncerBag      .processSaltedHandshake(saltedHandshakeRes);
    // Add a new sync status entry for each share    for (const share of commonShares) {      if (!this.syncStatuses.has(share)) {        this.syncStatuses.set(share, {          ingestedCount: 0,          pulledCount: 0,          isCaughtUp: false,          partnerIsCaughtUp: false,        });      }    }
    // Make sure to remove any status entries for shares no longer in common    for (const shareAddress in this.syncStatuses) {      if (!commonShares.includes(shareAddress)) {        await this.syncStatuses.delete(shareAddress);      }    }
    this.partnerLastSeenAt = partnerLastSeenAt;  }
  async pull(shareAddress: string) {    const state = this.shareStates[shareAddress];
    if (this.state === "closed" || !state) {      console.error(        `%c Could not find ${shareAddress} in share states...`,        "background-color: red;",      );      return;    }
    const queryResponse = await this.connection.request(      "serveShareQuery",      {        query: {          historyMode: "all",          orderBy: "localIndex ASC",          startAfter: {            localIndex: state.partnerMaxLocalIndexSoFar,          },          limit: 10,        },        storageId: state.partnerStorageId,        share: state.share,      },    );
    const { lastSeenAt, shareStates, ingested, pulled } = await this.syncerBag      .processShareQuery(        this.shareStates,        queryResponse,      );
    this.mergeShareStates(shareStates);    this.partnerLastSeenAt = lastSeenAt;
    const mergedShareState = this.shareStates[shareAddress];
    const syncStatus = this.syncStatuses.get(shareAddress);
    if (!syncStatus) {      return;    }
    const nextIsCaughtUp = mergedShareState.partnerMaxLocalIndexSoFar >=      mergedShareState.partnerMaxLocalIndexOverall;
    if (      ingested > 0 || pulled > 0 || nextIsCaughtUp !== syncStatus.isCaughtUp    ) {      await this.syncStatuses.set(shareAddress, {        ingestedCount: syncStatus.ingestedCount + ingested,        pulledCount: syncStatus.pulledCount + pulled,        isCaughtUp: nextIsCaughtUp,        partnerIsCaughtUp: syncStatus.partnerIsCaughtUp,      });    }
    if ((this.state as "ready" | "active" | "closed") === "closed") {      return;    }
    await this.connection.notify(      "notifyCaughtUpChange",      mergedShareState.storageId,      nextIsCaughtUp,    );
    clearTimeout(this.pullTimeouts.get(shareAddress));
    this.pullTimeouts.set(      shareAddress,      setTimeout(        () => {          this.pull(shareAddress);        },        nextIsCaughtUp ? 1000 : 0,      ),    );  }
  private async getShareStates() {    const shareStatesRequest = {      commonShares: Array.from(this.syncStatuses.keys()),    };
    const shareStatesResponse = await this.connection.request(      "serveAllShareStates",      shareStatesRequest,    );
    const { lastSeenAt, shareStates } = this.syncerBag      .processAllShareStates(        this.shareStates,        shareStatesRequest,        shareStatesResponse,      );
    this.partnerLastSeenAt = lastSeenAt;
    const prevShares = Object.keys(this.shareStates);    const nextShares = Object.keys(shareStates);
    this.shareStates = shareStates;
    // Start / end pulls for new / deleted shares, respectively.
    for (const nextShare in shareStates) {      if (!prevShares.includes(nextShare)) {        this.pull(nextShare);      }    }
    for (const oldShare of prevShares) {      if (!nextShares.includes(oldShare)) {        this.pullTimeouts.delete(oldShare);      }    }  }
  private mergeShareStates(newShareStates: Record<string, ShareState>) {    const nextShareStates: Record<string, ShareState> = {};
    for (const shareAddress in newShareStates) {      const newShareState = newShareStates[shareAddress];      const existingShareState = this.shareStates[shareAddress];
      if (!existingShareState) {        nextShareStates[shareAddress] = newShareState;        break;      }
      nextShareStates[shareAddress] = {        ...newShareState,        lastSeenAt: Math.max(          newShareState.lastSeenAt,        ),        partnerMaxLocalIndexOverall: newShareState.partnerMaxLocalIndexOverall,        partnerMaxLocalIndexSoFar: newShareState.partnerMaxLocalIndexSoFar,      };    }
    this.shareStates = nextShareStates;  }
  async storageCaughtUp(storageId: string, isCaughtUp: boolean) {    for (const shareAddress in this.shareStates) {      const shareState = this.shareStates[shareAddress];      const syncStatus = this.syncStatuses.get(shareAddress);
      if (shareState.partnerStorageId === storageId && syncStatus) {        await this.syncStatuses.set(shareState.share, {          ...syncStatus,          partnerIsCaughtUp: isCaughtUp,        });      }    }  }
  close() {    this.pullTimeouts.forEach((timeout) => clearTimeout(timeout));
    this.peerReplicaMapUnsub();
    this.state = "closed";  }}