<commit_before>import {
  TOPIC,
  PARSER_ACTIONS,
  RECORD_ACTIONS as RA,
  RecordMessage,
  RecordWriteMessage,
  ListenMessage,
  Message,
} from '../constants'
import ListenerRegistry from '../listen/listener-registry'
import SubscriptionRegistry from '../utils/subscription-registry'
import RecordDeletion from './record-deletion'
import { recordRequestBinding } from './record-request'
import RecordTransition from './record-transition'
import { isExcluded } from '../utils/utils'

const WRITE_ACK_TO_ACTION: { [key: number]: RA } = {
  [RA.CREATEANDPATCH_WITH_WRITE_ACK]: RA.CREATEANDPATCH,
  [RA.CREATEANDUPDATE_WITH_WRITE_ACK]: RA.CREATEANDUPDATE,
  [RA.PATCH_WITH_WRITE_ACK]: RA.PATCH,
  [RA.UPDATE_WITH_WRITE_ACK]: RA.UPDATE,
  [RA.ERASE_WITH_WRITE_ACK]: RA.ERASE,
}

export default class RecordHandler {
  private metaData: any
  private config: DeepstreamConfig
  private services: DeepstreamServices
  private subscriptionRegistry: SubscriptionRegistry
  private listenerRegistry: ListenerRegistry
  private transitions: any
  private recordRequestsInProgress: any
  private recordRequest: Function

/**
 * The entry point for record related operations
 */
  constructor (config: DeepstreamConfig, services: DeepstreamServices, subscriptionRegistry?: SubscriptionRegistry, listenerRegistry?: ListenerRegistry, metaData?: any) {
    this.metaData = metaData
    this.config = config
    this.services = services
    this.subscriptionRegistry =
      subscriptionRegistry || new SubscriptionRegistry(config, services, TOPIC.RECORD, TOPIC.RECORD_SUBSCRIPTIONS)
    this.listenerRegistry =
      listenerRegistry || new ListenerRegistry(TOPIC.RECORD, config, services, this.subscriptionRegistry, null)
    this.subscriptionRegistry.setSubscriptionListener(this.listenerRegistry)
    this.transitions = {}
    this.recordRequestsInProgress = {}
    this.recordRequest = recordRequestBinding(config, services, this, metaData)
  }

/**
 * Handles incoming record requests.
 *
 * Please note that neither CREATE nor READ is supported as a
 * client send action. Instead the client sends CREATEORREAD
 * and deepstream works which one it will be
 */
  public handle (socketWrapper: SocketWrapper, message: RecordMessage): void {
    const action = message.isWriteAck ? WRITE_ACK_TO_ACTION[message.action] : message.action
    if (action === RA.SUBSCRIBECREATEANDREAD) {
    /*
     * Return the record's contents and subscribes for future updates.
     * Creates the record if it doesn't exist
     */
      this.createOrRead(socketWrapper, message)
    } else if (
      action === RA.CREATEANDUPDATE ||
      action === RA.CREATEANDPATCH
    ) {
    /*
     * Allows updates to the record without being subscribed, creates
     * the record if it doesn't exist
     */
      this.createAndUpdate(socketWrapper, message as RecordWriteMessage)
    } else if (action === RA.READ) {
    /*
     * Return the current state of the record in cache or db
     */
      this.snapshot(socketWrapper, message)
    } else if (action === RA.HEAD) {
    /*
     * Return the current version of the record or -1 if not found
     */
      this.head(socketWrapper, message)
    } else if (action === RA.SUBSCRIBEANDHEAD) {
    /*
     * Return the current version of the record or -1 if not found, subscribing either way
     */
      this.subscribeAndHead(socketWrapper, message)
    } else if (action === RA.UPDATE || action === RA.PATCH || action === RA.ERASE) {
    /*
     * Handle complete (UPDATE) or partial (PATCH/ERASE) updates
     */
      this.update(socketWrapper, message as RecordWriteMessage, message.isWriteAck || false)
    } else if (action === RA.DELETE) {
    /*
     * Deletes the record
     */
      this.delete(socketWrapper, message)
    } else if (action === RA.DELETE_SUCCESS) {
    /*
     * Handle delete acknowledgement from message bus
     * TODO: Different action
     */
      this.remoteDelete(socketWrapper, message)
    } else if (action === RA.UNSUBSCRIBE) {
  /*
   * Unsubscribes (discards) a record that was previously subscribed to
   * using read()
   */
      this.subscriptionRegistry.unsubscribe(message, socketWrapper)
    } else if (action === RA.LISTEN ||
  /*
   * Listen to requests for a particular record or records
   * whose names match a pattern
   */
    action === RA.UNLISTEN ||
    action === RA.LISTEN_ACCEPT ||
    action === RA.LISTEN_REJECT) {
      this.listenerRegistry.handle(socketWrapper, message as ListenMessage)
    } else {
      this.services.logger.error(PARSER_ACTIONS[PARSER_ACTIONS.UNKNOWN_ACTION], RA[action], this.metaData)
    }
  }

/**
 * Sends the records data current data once loaded from the cache, and null otherwise
 */
  private snapshot (socketWrapper: SocketWrapper, message: RecordMessage): void {
    this.recordRequest(message.name, socketWrapper, onSnapshotComplete, onRequestError, message)
  }

  /**
   * Returns just the current version number of a record
   * Results in a HEAD_RESPONSE
   * If the record is not found, the version number will be -1
   */
  private head (socketWrapper: SocketWrapper, message: RecordMessage): void {
    this.recordRequest(message.name, socketWrapper, onHeadComplete, onRequestError, message)
  }

  /**
   * Same as head, and also subscribes the client to record updates.
   * Always results in SUBSCRIBE_ACK
   */
  private subscribeAndHead (socketWrapper: SocketWrapper, message: RecordMessage): void {
    this.head(socketWrapper, message)
    this.subscriptionRegistry.subscribe(
      Object.assign({}, message, { action: RA.SUBSCRIBE }),
      socketWrapper
    )
  }

  private onCreateOrReadComplete (recordName, version, data, socket, message) {
    if (data) {
      this.readAndSubscribe(message, version, data, socket)
    } else {
      this.permissionAction(
        RA.CREATE,
        message,
        message.action,
        socket,
        this.create.bind(this, message, socket),
      )
    }
  }

/**
 * Tries to retrieve the record and creates it if it doesn't exist. Please
 * note that create also triggers a read once done
 */
  private createOrRead (socketWrapper: SocketWrapper, message: RecordMessage): void {
    this.recordRequest(message.name, socketWrapper, this.onCreateOrReadComplete, onRequestError, message)
  }

/**
 * An upsert operation where the record will be created and written to
 * with the data in the message. Important to note that each operation,
 * the create and the write are permissioned separately.
 *
 * This method also takes note of the storageHotPathPatterns option, when a record
 * with a name that matches one of the storageHotPathPatterns is written to with
 * the CREATEANDUPDATE action, it will be permissioned for both CREATE and UPDATE, then
 * inserted into the cache and storage.
 */
  private createAndUpdate (socketWrapper: SocketWrapper, message: RecordWriteMessage): void {
    const recordName = message.name
    const isPatch = message.path !== undefined
    const originalAction = message.action
    Object.assign(message, { action: isPatch ? RA.PATCH : RA.UPDATE })

    // allow writes on the hot path to bypass the record transition
    // and be written directly to cache and storage
    for (let i = 0; i < this.config.storageHotPathPrefixes.length; i++) {
      const pattern = this.config.storageHotPathPrefixes[i]
      if (recordName.indexOf(pattern) === 0) {
        if (isPatch) {
          const errorMessage = {
            topic: TOPIC.RECORD,
            action: RA.INVALID_PATCH_ON_HOTPATH,
            originalAction: message.action,
            name: recordName
          } as RecordMessage
          if (message.correlationId) {
            errorMessage.correlationId = message.correlationId
          }
          socketWrapper.sendMessage(errorMessage)
          return
        }

        this.permissionAction(RA.CREATE, message, originalAction, socketWrapper, () => {
          this.permissionAction(RA.UPDATE, message, originalAction, socketWrapper, () => {
            this.forceWrite(recordName, message, socketWrapper)
          })
        })
        return
      }
    }

    const transition = this.transitions[recordName]
    if (transition) {
      this.permissionAction(message.action, message, originalAction, socketWrapper, () => {
        transition.add(socketWrapper, message)
      })
      return
    }

    this.permissionAction(RA.CREATE, message, originalAction, socketWrapper, () => {
      this.permissionAction(RA.UPDATE, message, originalAction, socketWrapper, () => {
        this.update(socketWrapper, message, true)
      })<commit_msg>fix: use correct original action<commit_after>import {
  TOPIC,
  PARSER_ACTIONS,
  RECORD_ACTIONS as RA,
  RecordMessage,
  RecordWriteMessage,
  ListenMessage,
  Message,
} from '../constants'
import ListenerRegistry from '../listen/listener-registry'
import SubscriptionRegistry from '../utils/subscription-registry'
import RecordDeletion from './record-deletion'
import { recordRequestBinding } from './record-request'
import RecordTransition from './record-transition'
import { isExcluded } from '../utils/utils'

const WRITE_ACK_TO_ACTION: { [key: number]: RA } = {
  [RA.CREATEANDPATCH_WITH_WRITE_ACK]: RA.CREATEANDPATCH,
  [RA.CREATEANDUPDATE_WITH_WRITE_ACK]: RA.CREATEANDUPDATE,
  [RA.PATCH_WITH_WRITE_ACK]: RA.PATCH,
  [RA.UPDATE_WITH_WRITE_ACK]: RA.UPDATE,
  [RA.ERASE_WITH_WRITE_ACK]: RA.ERASE,
}

export default class RecordHandler {
  private metaData: any
  private config: DeepstreamConfig
  private services: DeepstreamServices
  private subscriptionRegistry: SubscriptionRegistry
  private listenerRegistry: ListenerRegistry
  private transitions: any
  private recordRequestsInProgress: any
  private recordRequest: Function

/**
 * The entry point for record related operations
 */
  constructor (config: DeepstreamConfig, services: DeepstreamServices, subscriptionRegistry?: SubscriptionRegistry, listenerRegistry?: ListenerRegistry, metaData?: any) {
    this.metaData = metaData
    this.config = config
    this.services = services
    this.subscriptionRegistry =
      subscriptionRegistry || new SubscriptionRegistry(config, services, TOPIC.RECORD, TOPIC.RECORD_SUBSCRIPTIONS)
    this.listenerRegistry =
      listenerRegistry || new ListenerRegistry(TOPIC.RECORD, config, services, this.subscriptionRegistry, null)
    this.subscriptionRegistry.setSubscriptionListener(this.listenerRegistry)
    this.transitions = {}
    this.recordRequestsInProgress = {}
    this.recordRequest = recordRequestBinding(config, services, this, metaData)
  }

/**
 * Handles incoming record requests.
 *
 * Please note that neither CREATE nor READ is supported as a
 * client send action. Instead the client sends CREATEORREAD
 * and deepstream works which one it will be
 */
  public handle (socketWrapper: SocketWrapper, message: RecordMessage): void {
    const action = message.isWriteAck ? WRITE_ACK_TO_ACTION[message.action] : message.action
    if (action === RA.SUBSCRIBECREATEANDREAD) {
    /*
     * Return the record's contents and subscribes for future updates.
     * Creates the record if it doesn't exist
     */
      this.createOrRead(socketWrapper, message)
    } else if (
      action === RA.CREATEANDUPDATE ||
      action === RA.CREATEANDPATCH
    ) {
    /*
     * Allows updates to the record without being subscribed, creates
     * the record if it doesn't exist
     */
      this.createAndUpdate(socketWrapper, message as RecordWriteMessage)
    } else if (action === RA.READ) {
    /*
     * Return the current state of the record in cache or db
     */
      this.snapshot(socketWrapper, message)
    } else if (action === RA.HEAD) {
    /*
     * Return the current version of the record or -1 if not found
     */
      this.head(socketWrapper, message)
    } else if (action === RA.SUBSCRIBEANDHEAD) {
    /*
     * Return the current version of the record or -1 if not found, subscribing either way
     */
      this.subscribeAndHead(socketWrapper, message)
    } else if (action === RA.UPDATE || action === RA.PATCH || action === RA.ERASE) {
    /*
     * Handle complete (UPDATE) or partial (PATCH/ERASE) updates
     */
      this.update(socketWrapper, message as RecordWriteMessage, message.isWriteAck || false)
    } else if (action === RA.DELETE) {
    /*
     * Deletes the record
     */
      this.delete(socketWrapper, message)
    } else if (action === RA.DELETE_SUCCESS) {
    /*
     * Handle delete acknowledgement from message bus
     * TODO: Different action
     */
      this.remoteDelete(socketWrapper, message)
    } else if (action === RA.UNSUBSCRIBE) {
  /*
   * Unsubscribes (discards) a record that was previously subscribed to
   * using read()
   */
      this.subscriptionRegistry.unsubscribe(message, socketWrapper)
    } else if (action === RA.LISTEN ||
  /*
   * Listen to requests for a particular record or records
   * whose names match a pattern
   */
    action === RA.UNLISTEN ||
    action === RA.LISTEN_ACCEPT ||
    action === RA.LISTEN_REJECT) {
      this.listenerRegistry.handle(socketWrapper, message as ListenMessage)
    } else {
      this.services.logger.error(PARSER_ACTIONS[PARSER_ACTIONS.UNKNOWN_ACTION], RA[action], this.metaData)
    }
  }

/**
 * Sends the records data current data once loaded from the cache, and null otherwise
 */
  private snapshot (socketWrapper: SocketWrapper, message: RecordMessage): void {
    this.recordRequest(message.name, socketWrapper, onSnapshotComplete, onRequestError, message)
  }

  /**
   * Returns just the current version number of a record
   * Results in a HEAD_RESPONSE
   * If the record is not found, the version number will be -1
   */
  private head (socketWrapper: SocketWrapper, message: RecordMessage): void {
    this.recordRequest(message.name, socketWrapper, onHeadComplete, onRequestError, message)
  }

  /**
   * Same as head, and also subscribes the client to record updates.
   * Always results in SUBSCRIBE_ACK
   */
  private subscribeAndHead (socketWrapper: SocketWrapper, message: RecordMessage): void {
    this.head(socketWrapper, message)
    this.subscriptionRegistry.subscribe(
      Object.assign({}, message, { action: RA.SUBSCRIBE }),
      socketWrapper
    )
  }

  private onCreateOrReadComplete (recordName, version, data, socket, message) {
    if (data) {
      this.readAndSubscribe(message, version, data, socket)
    } else {
      this.permissionAction(
        RA.CREATE,
        message,
        message.action,
        socket,
        this.create.bind(this, message, socket),
      )
    }
  }

/**
 * Tries to retrieve the record and creates it if it doesn't exist. Please
 * note that create also triggers a read once done
 */
  private createOrRead (socketWrapper: SocketWrapper, message: RecordMessage): void {
    this.recordRequest(message.name, socketWrapper, this.onCreateOrReadComplete, onRequestError, message)
  }

/**
 * An upsert operation where the record will be created and written to
 * with the data in the message. Important to note that each operation,
 * the create and the write are permissioned separately.
 *
 * This method also takes note of the storageHotPathPatterns option, when a record
 * with a name that matches one of the storageHotPathPatterns is written to with
 * the CREATEANDUPDATE action, it will be permissioned for both CREATE and UPDATE, then
 * inserted into the cache and storage.
 */
  private createAndUpdate (socketWrapper: SocketWrapper, message: RecordWriteMessage): void {
    const recordName = message.name
    const isPatch = message.path !== undefined
    const originalAction = message.action
    Object.assign(message, { action: isPatch ? RA.PATCH : RA.UPDATE })

    // allow writes on the hot path to bypass the record transition
    // and be written directly to cache and storage
    for (let i = 0; i < this.config.storageHotPathPrefixes.length; i++) {
      const pattern = this.config.storageHotPathPrefixes[i]
      if (recordName.indexOf(pattern) === 0) {
        if (isPatch) {
          const errorMessage = {
            topic: TOPIC.RECORD,
            action: RA.INVALID_PATCH_ON_HOTPATH,
            originalAction,
            name: recordName
          } as RecordMessage
          if (message.correlationId) {
            errorMessage.correlationId = message.correlationId
          }
          socketWrapper.sendMessage(errorMessage)
          return
        }

        this.permissionAction(RA.CREATE, message, originalAction, socketWrapper, () => {
          this.permissionAction(RA.UPDATE, message, originalAction, socketWrapper, () => {
            this.forceWrite(recordName, message, socketWrapper)
          })
        })
        return
      }
    }

    const transition = this.transitions[recordName]
    if (transition) {
      this.permissionAction(message.action, message, originalAction, socketWrapper, () => {
        transition.add(socketWrapper, message)
      })
      return
    }

    this.permissionAction(RA.CREATE, message, originalAction, socketWrapper, () => {
      this.permissionAction(RA.UPDATE, message, originalAction, socketWrapper, () => {
        this.update(socketWrapper, message, true)
      })
{'fraction_non_alphanumeric': 0.0929839883551674, 'fraction_numerical': 0.0005822416302765648, 'mean_word_length': 3.318833291425698, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 13, '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': '5974923', 'n_tokens_mistral': 4774, 'n_tokens_neox': 4663, 'n_words': 1401}