<commit_before>import logger from '../logger';
import * as lodash from 'lodash';
import { CoinStorage } from './coin';
import { WalletAddressStorage, IWalletAddress } from './walletAddress';
import { ObjectID } from 'bson';
import { TransformOptions } from '../types/TransformOptions';
import { LoggifyClass } from '../decorators/Loggify';
import { Bitcoin } from '../types/namespaces/Bitcoin';
import { MongoBound } from './base';
import { StorageService } from '../services/storage';
import { TransactionJSON } from '../types/Transaction';
import { SpentHeightIndicators } from '../types/Coin';
import { Config } from '../services/config';
import { EventStorage } from './events';
import { Libs } from '../providers/libs';
import { BaseTransaction, ITransaction } from './baseTransaction';
import { Readable, Transform } from 'stream';
import { Collection } from 'mongodb';
import { partition } from '../utils/partition';

export { ITransaction };

const { onlyWalletEvents } = Config.get().services.event;
function shouldFire(obj: { wallets?: Array<ObjectID> }) {
  return !onlyWalletEvents || (onlyWalletEvents && obj.wallets && obj.wallets.length > 0);
}
const MAX_BATCH_SIZE = 50000;

export type IBtcTransaction = ITransaction & {
  coinbase: boolean;
  locktime: number;
  inputCount: number;
  outputCount: number;
  size: number;
};

export type TaggedBitcoinTx = Bitcoin.Transaction & { wallets: Array<ObjectID> };

export type MintOp = {
  updateOne: {
    filter: {
      mintTxid: string;
      mintIndex: number;
      chain: string;
      network: string;
    };
    update: {
      $set: {
        chain: string;
        network: string;
        address: string;
        mintHeight: number;
        coinbase: boolean;
        value: number;
        script: Buffer;
        spentTxid?: string;
        spentHeight?: SpentHeightIndicators;
        wallets?: Array<ObjectID>;
      };
      $setOnInsert: {
        spentHeight: SpentHeightIndicators;
        wallets: Array<ObjectID>;
      };
    };
    upsert: true;
    forceServerObjectId: true;
  };
};

export type SpendOp = {
  updateOne: {
    filter: {
      mintTxid: string;
      mintIndex: number;
      spentHeight: { $lt: SpentHeightIndicators };
      chain: string;
      network: string;
    };
    update: { $set: { spentTxid: string; spentHeight: number } };
  };
};

export interface TxOp {
  updateOne: {
    filter: { txid: string; chain: string; network: string };
    update: {
      $set: {
        chain: string;
        network: string;
        blockHeight: number;
        blockHash?: string;
        blockTime?: Date;
        blockTimeNormalized?: Date;
        coinbase: boolean;
        fee: number;
        size: number;
        locktime: number;
        inputCount: number;
        outputCount: number;
        value: number;
        wallets: Array<ObjectID>;
        mempoolTime?: Date;
      };
      $setOnInsert?: TxOp['updateOne']['update']['$set'];
    };
    upsert: true;
    forceServerObjectId: true;
  };
}

const getUpdatedBatchIfMempool = (batch, height) =>
  height >= SpentHeightIndicators.minimum ? batch : batch.map(op => TransactionStorage.toMempoolSafeUpsert(op, height));

export class MempoolSafeTransform extends Transform {
  constructor(private height: number) {
    super({ objectMode: true });
  }

  async _transform(
    coinBatch: Array<{ updateOne: { filter: any; update: { $set: any; $setOnInsert?: any } } }>,
    _,
    done
  ) {
    done(null, getUpdatedBatchIfMempool(coinBatch, this.height));
  }
}

export class MempoolCoinEventTransform extends Transform {
  constructor(private height: number) {
    super({ objectMode: true });
  }

  _transform(coinBatch: Array<MintOp>, _, done) {
    if (this.height < SpentHeightIndicators.minimum) {
      const eventPayload = coinBatch
        .map(coinOp => {
          const coin = {
            ...coinOp.updateOne.update.$set,
            ...coinOp.updateOne.filter,
            ...coinOp.updateOne.update.$setOnInsert
          };
          const address = coin.address;
          return { address, coin };
        })
        .filter(({ coin }) => shouldFire(coin));
      EventStorage.signalAddressCoins(eventPayload);
    }
    done(null, coinBatch);
  }
}

export class MempoolTxEventTransform extends Transform {
  constructor(private height: number) {
    super({ objectMode: true });
  }

  _transform(txBatch: Array<TxOp>, _, done) {
    if (this.height < SpentHeightIndicators.minimum) {
      const eventPayload = txBatch
        .map(op => ({ ...op.updateOne.update.$set, ...op.updateOne.filter, ...op.updateOne.update.$setOnInsert }))
        .filter(shouldFire);
      EventStorage.signalTxs(eventPayload);
    }
    done(null, txBatch);
  }
}

export class MongoWriteStream extends Transform {
  constructor(private collection: Collection) {
    super({ objectMode: true });
  }

  async _transform(data: Array<any>, _, done) {
    await Promise.all(
      partition([...data], data.length / Config.get().maxPoolSize).map(batch => this.collection.bulkWrite(batch))
    );
    done(null, data);
  }
}

export class PruneMempoolStream extends Transform {
  constructor(private chain: string, private network: string, private initialSyncComplete: boolean) {
    super({ objectMode: true });
  }

  async _transform(spendOps: Array<SpendOp>, _, done) {
    await TransactionStorage.pruneMempool({
      chain: this.chain,
      network: this.network,
      initialSyncComplete: this.initialSyncComplete,
      spendOps
    });
    done(null, spendOps);<commit_msg>Remove clone<commit_after>import logger from '../logger';
import * as lodash from 'lodash';
import { CoinStorage } from './coin';
import { WalletAddressStorage, IWalletAddress } from './walletAddress';
import { ObjectID } from 'bson';
import { TransformOptions } from '../types/TransformOptions';
import { LoggifyClass } from '../decorators/Loggify';
import { Bitcoin } from '../types/namespaces/Bitcoin';
import { MongoBound } from './base';
import { StorageService } from '../services/storage';
import { TransactionJSON } from '../types/Transaction';
import { SpentHeightIndicators } from '../types/Coin';
import { Config } from '../services/config';
import { EventStorage } from './events';
import { Libs } from '../providers/libs';
import { BaseTransaction, ITransaction } from './baseTransaction';
import { Readable, Transform } from 'stream';
import { Collection } from 'mongodb';
import { partition } from '../utils/partition';

export { ITransaction };

const { onlyWalletEvents } = Config.get().services.event;
function shouldFire(obj: { wallets?: Array<ObjectID> }) {
  return !onlyWalletEvents || (onlyWalletEvents && obj.wallets && obj.wallets.length > 0);
}
const MAX_BATCH_SIZE = 50000;

export type IBtcTransaction = ITransaction & {
  coinbase: boolean;
  locktime: number;
  inputCount: number;
  outputCount: number;
  size: number;
};

export type TaggedBitcoinTx = Bitcoin.Transaction & { wallets: Array<ObjectID> };

export type MintOp = {
  updateOne: {
    filter: {
      mintTxid: string;
      mintIndex: number;
      chain: string;
      network: string;
    };
    update: {
      $set: {
        chain: string;
        network: string;
        address: string;
        mintHeight: number;
        coinbase: boolean;
        value: number;
        script: Buffer;
        spentTxid?: string;
        spentHeight?: SpentHeightIndicators;
        wallets?: Array<ObjectID>;
      };
      $setOnInsert: {
        spentHeight: SpentHeightIndicators;
        wallets: Array<ObjectID>;
      };
    };
    upsert: true;
    forceServerObjectId: true;
  };
};

export type SpendOp = {
  updateOne: {
    filter: {
      mintTxid: string;
      mintIndex: number;
      spentHeight: { $lt: SpentHeightIndicators };
      chain: string;
      network: string;
    };
    update: { $set: { spentTxid: string; spentHeight: number } };
  };
};

export interface TxOp {
  updateOne: {
    filter: { txid: string; chain: string; network: string };
    update: {
      $set: {
        chain: string;
        network: string;
        blockHeight: number;
        blockHash?: string;
        blockTime?: Date;
        blockTimeNormalized?: Date;
        coinbase: boolean;
        fee: number;
        size: number;
        locktime: number;
        inputCount: number;
        outputCount: number;
        value: number;
        wallets: Array<ObjectID>;
        mempoolTime?: Date;
      };
      $setOnInsert?: TxOp['updateOne']['update']['$set'];
    };
    upsert: true;
    forceServerObjectId: true;
  };
}

const getUpdatedBatchIfMempool = (batch, height) =>
  height >= SpentHeightIndicators.minimum ? batch : batch.map(op => TransactionStorage.toMempoolSafeUpsert(op, height));

export class MempoolSafeTransform extends Transform {
  constructor(private height: number) {
    super({ objectMode: true });
  }

  async _transform(
    coinBatch: Array<{ updateOne: { filter: any; update: { $set: any; $setOnInsert?: any } } }>,
    _,
    done
  ) {
    done(null, getUpdatedBatchIfMempool(coinBatch, this.height));
  }
}

export class MempoolCoinEventTransform extends Transform {
  constructor(private height: number) {
    super({ objectMode: true });
  }

  _transform(coinBatch: Array<MintOp>, _, done) {
    if (this.height < SpentHeightIndicators.minimum) {
      const eventPayload = coinBatch
        .map(coinOp => {
          const coin = {
            ...coinOp.updateOne.update.$set,
            ...coinOp.updateOne.filter,
            ...coinOp.updateOne.update.$setOnInsert
          };
          const address = coin.address;
          return { address, coin };
        })
        .filter(({ coin }) => shouldFire(coin));
      EventStorage.signalAddressCoins(eventPayload);
    }
    done(null, coinBatch);
  }
}

export class MempoolTxEventTransform extends Transform {
  constructor(private height: number) {
    super({ objectMode: true });
  }

  _transform(txBatch: Array<TxOp>, _, done) {
    if (this.height < SpentHeightIndicators.minimum) {
      const eventPayload = txBatch
        .map(op => ({ ...op.updateOne.update.$set, ...op.updateOne.filter, ...op.updateOne.update.$setOnInsert }))
        .filter(shouldFire);
      EventStorage.signalTxs(eventPayload);
    }
    done(null, txBatch);
  }
}

export class MongoWriteStream extends Transform {
  constructor(private collection: Collection) {
    super({ objectMode: true });
  }

  async _transform(data: Array<any>, _, done) {
    await Promise.all(
      partition(data, data.length / Config.get().maxPoolSize).map(batch => this.collection.bulkWrite(batch))
    );
    done(null, data);
  }
}

export class PruneMempoolStream extends Transform {
  constructor(private chain: string, private network: string, private initialSyncComplete: boolean) {
    super({ objectMode: true });
  }

  async _transform(spendOps: Array<SpendOp>, _, done) {
    await TransactionStorage.pruneMempool({
      chain: this.chain,
      network: this.network,
      initialSyncComplete: this.initialSyncComplete,
      spendOps
    });
    done(null, spendOps);
{'fraction_non_alphanumeric': 0.1295696020169278, 'fraction_numerical': 0.0010804970286331713, 'mean_word_length': 3.008300252616384, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1880194', 'n_tokens_mistral': 3297, 'n_tokens_neox': 3260, 'n_words': 782}