text: # Copyright (c) Facebook, Inc. and its affiliates.
  #
  # This source code is licensed under the MIT license found in the
  # LICENSE file in the root directory of this source tree.
  
  import contextlib
  import dataclasses
  import datetime
  import enum
  import logging
  import os
  import subprocess
  import tempfile
  from pathlib import Path
  from typing import (
      TextIO,
      Any,
      Dict,
      Iterator,
      List,
      Mapping,
      Optional,
      Sequence,
      Tuple,
      Union,
  )
  
  from ... import (
      command_arguments,
      commands,
      configuration as configuration_module,
      find_directories,
      log,
  )
  from . import backend_arguments, server_connection, server_event, stop, remote_logging
  
  
  LOG: logging.Logger = logging.getLogger(__name__)
  
  SERVER_LOG_FILE_FORMAT: str = "server.stderr.%Y_%m_%d_%H_%M_%S_%f"
  
  
  class MatchPolicy(enum.Enum):
      BASE_NAME = "base_name"
      EXTENSION = "extension"
      FULL_PATH = "full_path"
  
      def __str__(self) -> str:
          return self.value
  
  
  @dataclasses.dataclass(frozen=True)
  class CriticalFile:
      policy: MatchPolicy
      path: str
  
      def serialize(self) -> Dict[str, str]:
          return {str(self.policy): self.path}
  
  
  @dataclasses.dataclass(frozen=True)
  class LoadSavedStateFromFile:
      shared_memory_path: str
      changed_files_path: Optional[str] = None
  
      def serialize(self) -> Tuple[str, Dict[str, str]]:
          return (
              "load_from_file",
              {
                  "shared_memory_path": self.shared_memory_path,
                  **(
                      {}
                      if self.changed_files_path is None
                      else {"changed_files_path": self.changed_files_path}
                  ),
              },
          )
  
  
  @dataclasses.dataclass(frozen=True)
  class LoadSavedStateFromProject:
      project_name: str
      project_metadata: Optional[str] = None
  
      def serialize(self) -> Tuple[str, Dict[str, str]]:
          return (
              "load_from_project",
              {
                  "project_name": self.project_name,
                  **(
                      {}
                      if self.project_metadata is None
                      else {"project_metadata": self.project_metadata}
                  ),
              },
          )
  
  
  @dataclasses.dataclass(frozen=True)
  class StoreSavedStateToFile:
      shared_memory_path: str
  
      def serialize(self) -> Tuple[str, Dict[str, str]]:
          return (
              "save_to_file",
              {
                  "shared_memory_path": self.shared_memory_path,
              },
          )
  
  
  SavedStateAction = Union[
      LoadSavedStateFromFile, LoadSavedStateFromProject, StoreSavedStateToFile
  ]
  
  
  @dataclasses.dataclass(frozen=True)
  class Arguments:
      """
      Data structure for configuration options the backend server can recognize.
      Need to keep in sync with `pyre/command/newServerCommand.ml`
      """
  
      base_arguments: backend_arguments.BaseArguments
  
      strict: bool = False
      show_error_traces: bool = False
      additional_logging_sections: Sequence[str] = dataclasses.field(default_factory=list)
      watchman_root: Optional[Path] = None
      taint_models_path: Sequence[str] = dataclasses.field(default_factory=list)
      store_type_check_resolution: bool = False
      critical_files: Sequence[CriticalFile] = dataclasses.field(default_factory=list)
      saved_state_action: Optional[SavedStateAction] = None
  
      def serialize(self) -> Dict[str, Any]:
          return {
              **self.base_arguments.serialize(),
              "strict": self.strict,
              "show_error_traces": self.show_error_traces,
              "additional_logging_sections": self.additional_logging_sections,
              **(
                  {}
                  if self.watchman_root is None
                  else {"watchman_root": str(self.watchman_root)}
              ),
              "taint_model_paths": self.taint_models_path,
              "store_type_check_resolution": self.store_type_check_resolution,
              "critical_files": [
                  critical_file.serialize() for critical_file in self.critical_files
              ],
              **(
                  {}
                  if self.saved_state_action is None
                  else {"saved_state_action": self.saved_state_action.serialize()}
              ),
          }
  
  
  def get_critical_files(
      configuration: configuration_module.Configuration,
  ) -> List[CriticalFile]:
      def get_full_path(root: str, relative: str) -> str:
          full_path = (Path(root) / relative).resolve(strict=False)
          if not full_path.exists():
              LOG.warning(f"Critical file does not exist: {full_path}")
          return str(full_path)
  
      local_root = configuration.local_root
      return [
          CriticalFile(
              policy=MatchPolicy.FULL_PATH,
              path=get_full_path(
                  root=configuration.project_root,
                  relative=find_directories.CONFIGURATION_FILE,
              ),
          ),
          *(
              []
              if local_root is None
              else [
                  CriticalFile(
                      policy=MatchPolicy.FULL_PATH,
                      path=get_full_path(
                          root=local_root,
                          relative=find_directories.LOCAL_CONFIGURATION_FILE,
                      ),
                  )
              ]
          ),
          *(
              # TODO(T92070475): This is a temporary hack until generated code can be
              # fully supported.
              []
              if configuration.targets is None
              else [CriticalFile(policy=MatchPolicy.EXTENSION, path="thrift")]
          ),
          *(
              [
                  CriticalFile(
                      policy=MatchPolicy.FULL_PATH,
                      path=get_full_path(root=path, relative=""),
                  )
                  for path in configuration.other_critical_files
              ]
          ),
      ]
  
  
  def get_saved_state_action(
      start_arguments: command_arguments.StartArguments,
      relative_local_root: Optional[str] = None,
  ) -> Optional[SavedStateAction]:
      saved_state_output_path = start_arguments.save_initial_state_to
      if saved_state_output_path is not None:
          return StoreSavedStateToFile(shared_memory_path=saved_state_output_path)
  
      saved_state_input_path = start_arguments.load_initial_state_from
      if saved_state_input_path is not None:
          return LoadSavedStateFromFile(
              shared_memory_path=saved_state_input_path,
              changed_files_path=start_arguments.changed_files_path,
          )
  
      saved_state_project = start_arguments.saved_state_project
      if saved_state_project is not None:
          return LoadSavedStateFromProject(
              project_name=saved_state_project,
              project_metadata=relative_local_root.replace("/", "$")
              if relative_local_root is not None
              else None,
          )
  
      return None
  
  
  def create_server_arguments(
      configuration: configuration_module.Configuration,
      start_arguments: command_arguments.StartArguments,
  ) -> Arguments:
      """
      Translate client configurations and command-line flags to server
      configurations.
  
      This API is not pure since it needs to access filesystem to filter out
      nonexistent directories. It is idempotent though, since it does not alter
      any filesystem state.
      """
      source_paths = backend_arguments.get_source_path_for_server(configuration)
  
      logging_sections = start_arguments.logging_sections
      additional_logging_sections = (
          [] if logging_sections is None else logging_sections.split(",")
      )
      if start_arguments.noninteractive:
          additional_logging_sections.append("-progress")
      # Server section is usually useful when Pyre server is involved
      additional_logging_sections.append("server")
  
      profiling_output = (
          backend_arguments.get_profiling_log_path(Path(configuration.log_directory))
          if start_arguments.enable_profiling
          else None
      )
      memory_profiling_output = (
          backend_arguments.get_profiling_log_path(Path(configuration.log_directory))
          if start_arguments.enable_memory_profiling
          else None
      )
  
      return Arguments(
          base_arguments=backend_arguments.BaseArguments(
              log_path=configuration.log_directory,
              global_root=configuration.project_root,
              checked_directory_allowlist=(
                  list(source_paths.get_checked_directory_allowlist())
                  + configuration.get_existent_do_not_ignore_errors_in_paths()
              ),
              checked_directory_blocklist=(
                  configuration.get_existent_ignore_all_errors_paths()
              ),
              debug=start_arguments.debug,
              excludes=configuration.excludes,
              extensions=configuration.get_valid_extension_suffixes(),
              relative_local_root=configuration.relative_local_root,
              memory_profiling_output=memory_profiling_output,
              number_of_workers=configuration.get_number_of_workers(),
              parallel=not start_arguments.sequential,
              profiling_output=profiling_output,
              python_version=configuration.get_python_version(),
              shared_memory=configuration.shared_memory,
              remote_logging=backend_arguments.RemoteLogging.create(
                  configuration.logger, start_arguments.log_identifier
              ),
              search_paths=configuration.expand_and_get_existent_search_paths(),
              source_paths=source_paths,
          ),
          strict=configuration.strict,
          show_error_traces=start_arguments.show_error_traces,
          additional_logging_sections=additional_logging_sections,
          watchman_root=None
          if start_arguments.no_watchman
          else backend_arguments.find_watchman_root(Path(configuration.project_root)),
          taint_models_path=configuration.taint_models_path,
          store_type_check_resolution=start_arguments.store_type_check_resolution,
          critical_files=get_critical_files(configuration),
          saved_state_action=None
          if start_arguments.no_saved_state
          else get_saved_state_action(
              start_arguments, relative_local_root=configuration.relative_local_root
          ),
      )
  
  
  def get_server_identifier(configuration: configuration_module.Configuration) -> str:
      global_identifier = Path(configuration.project_root).name
      relative_local_root = configuration.relative_local_root
      if relative_local_root is None:
          return global_identifier
      return f"{global_identifier}/{relative_local_root}"
  
  
  def _run_in_foreground(
      command: Sequence[str], environment: Mapping[str, str]
  ) -> commands.ExitCode:
      # In foreground mode, we shell out to the backend server and block on it.
      # Server stdout/stderr will be forwarded to the current terminal.
      return_code = 0
      try:
          LOG.warning("Starting server in the foreground...")
          result = subprocess.run(
              command,
              env=environment,
              stdout=subprocess.DEVNULL,
              stderr=None,
              universal_newlines=True,
          )
          return_code = result.returncode
      except KeyboardInterrupt:
          # Backend server will exit cleanly when receiving SIGINT.
          pass
  
      if return_code == 0:
          return commands.ExitCode.SUCCESS
      else:
          LOG.error(f"Server exited with non-zero return code: {return_code}")
          return commands.ExitCode.FAILURE
  
  
  @contextlib.contextmanager
  def background_logging(log_file: Path) -> Iterator[None]:
      with log.file_tailer(log_file) as log_stream:
          with log.StreamLogger(log_stream) as logger:
              yield
      logger.join()
  
  
  def _create_symbolic_link(source: Path, target: Path) -> None:
      try:
          source.unlink()
      except FileNotFoundError:
          pass
      source.symlink_to(target)
  
  
  @contextlib.contextmanager
  def background_server_log_file(log_directory: Path) -> Iterator[TextIO]:
      new_server_log_directory = log_directory / "new_server"
      new_server_log_directory.mkdir(parents=True, exist_ok=True)
      log_file_path = new_server_log_directory / datetime.datetime.now().strftime(
          SERVER_LOG_FILE_FORMAT
      )
      with open(str(log_file_path), "a") as log_file:
          yield log_file
      # Symlink the log file to a known location for subsequent `pyre incremental`
      # to find.
      _create_symbolic_link(new_server_log_directory / "server.stderr", log_file_path)
  
  
  def _run_in_background(
      command: Sequence[str],
      environment: Mapping[str, str],
      log_directory: Path,
      event_waiter: server_event.Waiter,
  ) -> commands.ExitCode:
      # In background mode, we asynchronously start the server with `Popen` and
      # detach it from the current process immediately with `start_new_session`.
      # Do not call `wait()` on the Popen object to avoid blocking.
      # Server stderr will be forwarded to dedicated log files.
      # Server stdout will be used as additional communication channel for status
      # updates.
      with background_server_log_file(log_directory) as server_stderr:
          log_file = Path(server_stderr.name)
          server_process = subprocess.Popen(
              command,
              stdout=subprocess.PIPE,
              stderr=server_stderr,
              env=environment,
              start_new_session=True,
              universal_newlines=True,
          )
  
      LOG.info("Server is starting in the background.\n")
      server_stdout = server_process.stdout
      if server_stdout is None:
          raise RuntimeError("subprocess.Popen failed to set up a pipe for server stdout")
  
      try:
          # Block until an expected server event is obtained from stdout
          with background_logging(log_file):
              event_waiter.wait_on(server_stdout)
              server_stdout.close()
              return commands.ExitCode.SUCCESS
      except KeyboardInterrupt:
          LOG.info("SIGINT received. Terminating background server...")
  
          # If a background server has spawned and the user hits Ctrl-C, bring down
          # the spwaned server as well.
          server_stdout.close()
          server_process.terminate()
          server_process.wait()
  
          # Since we abruptly terminate the background server, it may not have the
          # chance to clean up the socket file properly. Make sure the file is
          # removed on our side.
          stop.remove_socket_if_exists(
              server_connection.get_default_socket_path(log_directory)
          )
  
          raise commands.ClientException("Interrupted by user. No server is spawned.")
  
  
  def run_start(
      configuration: configuration_module.Configuration,
      start_arguments: command_arguments.StartArguments,
  ) -> commands.ExitCode:
      binary_location = configuration.get_binary_respecting_override()
      if binary_location is None:
          raise configuration_module.InvalidConfiguration(
              "Cannot locate a Pyre binary to run."
          )
  
      log_directory = Path(configuration.log_directory)
  
      server_arguments = create_server_arguments(configuration, start_arguments)
      if not start_arguments.no_watchman and server_arguments.watchman_root is None:
          LOG.warning(
              "Cannot find a watchman root. Incremental check will not be functional"
              " since no filesystem updates will be sent to the Pyre server."
          )
  
      LOG.info(f"Starting server at `{get_server_identifier(configuration)}`...")
      with backend_arguments.temporary_argument_file(
          server_arguments
      ) as argument_file_path:
          server_command = [binary_location, "newserver", str(argument_file_path)]
          server_environment = {
              **os.environ,
              # This is to make sure that backend server shares the socket root
              # directory with the client.
              # TODO(T77556312): It might be cleaner to turn this into a
              # configuration option instead.
              "TMPDIR": tempfile.gettempdir(),
          }
          if start_arguments.terminal:
              return _run_in_foreground(server_command, server_environment)
          else:
              return _run_in_background(
                  server_command,
                  server_environment,
                  log_directory,
                  server_event.Waiter(
                      wait_on_initialization=start_arguments.wait_on_initialization
                  ),
              )
  
  
  @remote_logging.log_usage(command_name="start")
  def run(
      configuration: configuration_module.Configuration,
      start_arguments: command_arguments.StartArguments,
  ) -> commands.ExitCode:
      return run_start(configuration, start_arguments)
  