opusdev/vector-similarity-api

1import { type Document, Long, type ObjectId } from '../bson';2import { type MongoError, MongoRuntimeError } from '../error';3import { arrayStrictEqual, compareObjectId, errorStrictEqual, HostAddress, now } from '../utils';4import { type ClusterTime, ServerType } from './common';5 6const WRITABLE_SERVER_TYPES = new Set<ServerType>([7  ServerType.RSPrimary,8  ServerType.Standalone,9  ServerType.Mongos,10  ServerType.LoadBalancer11]);12 13const DATA_BEARING_SERVER_TYPES = new Set<ServerType>([14  ServerType.RSPrimary,15  ServerType.RSSecondary,16  ServerType.Mongos,17  ServerType.Standalone,18  ServerType.LoadBalancer19]);20 21/** @public */22export interface TopologyVersion {23  processId: ObjectId;24  counter: Long;25}26 27/** @public */28export type TagSet = { [key: string]: string };29 30/** @internal */31export interface ServerDescriptionOptions {32  /** An Error used for better reporting debugging */33  error?: MongoError;34 35  /** The average round trip time to ping this server (in ms) */36  roundTripTime?: number;37  /** The minimum round trip time to ping this server over the past 10 samples(in ms) */38  minRoundTripTime?: number;39 40  /** If the client is in load balancing mode. */41  loadBalanced?: boolean;42}43 44/**45 * The client's view of a single server, based on the most recent hello outcome.46 *47 * Internal type, not meant to be directly instantiated48 * @public49 */50export class ServerDescription {51  address: string;52  type: ServerType;53  hosts: string[];54  passives: string[];55  arbiters: string[];56  tags: TagSet;57  error: MongoError | null;58  topologyVersion: TopologyVersion | null;59  minWireVersion: number;60  maxWireVersion: number;61  roundTripTime: number;62  /** The minimum measurement of the last 10 measurements of roundTripTime that have been collected */63  minRoundTripTime: number;64  lastUpdateTime: number;65  lastWriteDate: number;66  me: string | null;67  primary: string | null;68  setName: string | null;69  setVersion: number | null;70  electionId: ObjectId | null;71  logicalSessionTimeoutMinutes: number | null;72  /** The max message size in bytes for the server. */73  maxMessageSizeBytes: number | null;74  /** The max number of writes in a bulk write command. */75  maxWriteBatchSize: number | null;76  /** The max bson object size. */77  maxBsonObjectSize: number | null;78  /** Indicates server is a mongocryptd instance. */79  iscryptd: boolean;80 81  // NOTE: does this belong here? It seems we should gossip the cluster time at the CMAP level82  $clusterTime?: ClusterTime;83 84  /**85   * Create a ServerDescription86   * @internal87   *88   * @param address - The address of the server89   * @param hello - An optional hello response for this server90   */91  constructor(92    address: HostAddress | string,93    hello?: Document,94    options: ServerDescriptionOptions = {}95  ) {96    if (address == null || address === '') {97      throw new MongoRuntimeError('ServerDescription must be provided with a non-empty address');98    }99 100    this.address =101      typeof address === 'string'102        ? HostAddress.fromString(address).toString() // Use HostAddress to normalize103        : address.toString();104    this.type = parseServerType(hello, options);105    this.hosts = hello?.hosts?.map((host: string) => host.toLowerCase()) ?? [];106    this.passives = hello?.passives?.map((host: string) => host.toLowerCase()) ?? [];107    this.arbiters = hello?.arbiters?.map((host: string) => host.toLowerCase()) ?? [];108    this.tags = hello?.tags ?? {};109    this.minWireVersion = hello?.minWireVersion ?? 0;110    this.maxWireVersion = hello?.maxWireVersion ?? 0;111    this.roundTripTime = options?.roundTripTime ?? -1;112    this.minRoundTripTime = options?.minRoundTripTime ?? 0;113    this.lastUpdateTime = now();114    this.lastWriteDate = hello?.lastWrite?.lastWriteDate ?? 0;115    // NOTE: This actually builds the stack string instead of holding onto the getter and all its116    // associated references. This is done to prevent a memory leak.117    this.error = options.error ?? null;118    this.error?.stack;119    // TODO(NODE-2674): Preserve int64 sent from MongoDB120    this.topologyVersion = this.error?.topologyVersion ?? hello?.topologyVersion ?? null;121    this.setName = hello?.setName ?? null;122    this.setVersion = hello?.setVersion ?? null;123    this.electionId = hello?.electionId ?? null;124    this.logicalSessionTimeoutMinutes = hello?.logicalSessionTimeoutMinutes ?? null;125    this.maxMessageSizeBytes = hello?.maxMessageSizeBytes ?? null;126    this.maxWriteBatchSize = hello?.maxWriteBatchSize ?? null;127    this.maxBsonObjectSize = hello?.maxBsonObjectSize ?? null;128    this.primary = hello?.primary ?? null;129    this.me = hello?.me?.toLowerCase() ?? null;130    this.$clusterTime = hello?.$clusterTime ?? null;131    this.iscryptd = Boolean(hello?.iscryptd);132  }133 134  get hostAddress(): HostAddress {135    return HostAddress.fromString(this.address);136  }137 138  get allHosts(): string[] {139    return this.hosts.concat(this.arbiters).concat(this.passives);140  }141 142  /** Is this server available for reads*/143  get isReadable(): boolean {144    return this.type === ServerType.RSSecondary || this.isWritable;145  }146 147  /** Is this server data bearing */148  get isDataBearing(): boolean {149    return DATA_BEARING_SERVER_TYPES.has(this.type);150  }151 152  /** Is this server available for writes */153  get isWritable(): boolean {154    return WRITABLE_SERVER_TYPES.has(this.type);155  }156 157  get host(): string {158    const chopLength = `:${this.port}`.length;159    return this.address.slice(0, -chopLength);160  }161 162  get port(): number {163    const port = this.address.split(':').pop();164    return port ? Number.parseInt(port, 10) : 27017;165  }166 167  /**168   * Determines if another `ServerDescription` is equal to this one per the rules defined in the SDAM specification.169   * @see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.md170   */171  equals(other?: ServerDescription | null): boolean {172    // Despite using the comparator that would determine a nullish topologyVersion as greater than173    // for equality we should only always perform direct equality comparison174    const topologyVersionsEqual =175      this.topologyVersion === other?.topologyVersion ||176      compareTopologyVersion(this.topologyVersion, other?.topologyVersion) === 0;177 178    const electionIdsEqual =179      this.electionId != null && other?.electionId != null180        ? compareObjectId(this.electionId, other.electionId) === 0181        : this.electionId === other?.electionId;182 183    return (184      other != null &&185      other.iscryptd === this.iscryptd &&186      errorStrictEqual(this.error, other.error) &&187      this.type === other.type &&188      this.minWireVersion === other.minWireVersion &&189      arrayStrictEqual(this.hosts, other.hosts) &&190      tagsStrictEqual(this.tags, other.tags) &&191      this.setName === other.setName &&192      this.setVersion === other.setVersion &&193      electionIdsEqual &&194      this.primary === other.primary &&195      this.logicalSessionTimeoutMinutes === other.logicalSessionTimeoutMinutes &&196      topologyVersionsEqual197    );198  }199}200 201// Parses a `hello` message and determines the server type202export function parseServerType(hello?: Document, options?: ServerDescriptionOptions): ServerType {203  if (options?.loadBalanced) {204    return ServerType.LoadBalancer;205  }206 207  if (!hello || !hello.ok) {208    return ServerType.Unknown;209  }210 211  if (hello.isreplicaset) {212    return ServerType.RSGhost;213  }214 215  if (hello.msg && hello.msg === 'isdbgrid') {216    return ServerType.Mongos;217  }218 219  if (hello.setName) {220    if (hello.hidden) {221      return ServerType.RSOther;222    } else if (hello.isWritablePrimary) {223      return ServerType.RSPrimary;224    } else if (hello.secondary) {225      return ServerType.RSSecondary;226    } else if (hello.arbiterOnly) {227      return ServerType.RSArbiter;228    } else {229      return ServerType.RSOther;230    }231  }232 233  return ServerType.Standalone;234}235 236function tagsStrictEqual(tags: TagSet, tags2: TagSet): boolean {237  const tagsKeys = Object.keys(tags);238  const tags2Keys = Object.keys(tags2);239 240  return (241    tagsKeys.length === tags2Keys.length &&242    tagsKeys.every((key: string) => tags2[key] === tags[key])243  );244}245 246/**247 * Compares two topology versions.248 *249 * 1. If the response topologyVersion is unset or the ServerDescription's250 *    topologyVersion is null, the client MUST assume the response is more recent.251 * 1. If the response's topologyVersion.processId is not equal to the252 *    ServerDescription's, the client MUST assume the response is more recent.253 * 1. If the response's topologyVersion.processId is equal to the254 *    ServerDescription's, the client MUST use the counter field to determine255 *    which topologyVersion is more recent.256 *257 * ```ts258 * currentTv <   newTv === -1259 * currentTv === newTv === 0260 * currentTv >   newTv === 1261 * ```262 */263export function compareTopologyVersion(264  currentTv?: TopologyVersion | null,265  newTv?: TopologyVersion | null266): 0 | -1 | 1 {267  if (currentTv == null || newTv == null) {268    return -1;269  }270 271  if (!currentTv.processId.equals(newTv.processId)) {272    return -1;273  }274 275  // TODO(NODE-2674): Preserve int64 sent from MongoDB276  const currentCounter =277    typeof currentTv.counter === 'bigint'278      ? Long.fromBigInt(currentTv.counter)279      : Long.isLong(currentTv.counter)280        ? currentTv.counter281        : Long.fromNumber(currentTv.counter);282 283  const newCounter =284    typeof newTv.counter === 'bigint'285      ? Long.fromBigInt(newTv.counter)286      : Long.isLong(newTv.counter)287        ? newTv.counter288        : Long.fromNumber(newTv.counter);289 290  return currentCounter.compare(newCounter);291}292