# Copyright 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Continuous Integration Database Library."""

import datetime
import glob
import logging
import os
import re
import sqlalchemy
import sqlalchemy.exc
import sqlalchemy.interfaces
from sqlalchemy import MetaData
import time

from chromite.cbuildbot import constants

CIDB_MIGRATIONS_DIR = os.path.join(constants.CHROMITE_DIR, 'cidb',
                                   'migrations')

class DBException(Exception):
  """General exception class for this module."""


class UnsupportedMethodException(DBException):
  """Raised when a call is made that the database does not support."""


def minimum_schema(min_version):
  """Generate a decorator to specify a minimum schema version for a method.

  This decorator should be applied only to instance methods of
  SchemaVersionedMySQLConnection objects.
  """

  def decorator(f):
    def wrapper(self, *args, **kwargs):
      if self.schema_version < min_version:
        raise UnsupportedMethodException()
      return f(self, *args, **kwargs)
    return wrapper
  return decorator


class StrictModeListener(sqlalchemy.interfaces.PoolListener):
  """This listener ensures that STRICT_ALL_TABLES for all connections."""
  # pylint: disable-msg=W0613
  def connect(self, dbapi_con, *args, **kwargs):
    cur = dbapi_con.cursor()
    cur.execute("SET SESSION sql_mode='STRICT_ALL_TABLES'")
    cur.close()


class SchemaVersionedMySQLConnection(object):
  """Connection to a database that is aware of its schema version."""

  SCHEMA_VERSION_TABLE_NAME = 'schemaVersionTable'
  SCHEMA_VERSION_COL = 'schemaVersion'

  def __init__(self, db_name, db_migrations_dir, db_credentials_dir):
    """SchemaVersionedMySQLConnection constructor.

    Args:
      db_name: Name of the database to connect to.
      db_migrations_dir: Absolute path to directory of migration scripts
                         for this database.
      db_credentials_dir: Absolute path to directory containing connection
                          information to the database. Specifically, this
                          directory should contain files names user.txt,
                          password.txt, host.txt, client-cert.pem,
                          client-key.pem, and server-ca.pem
    """
    # None, or a sqlalchemy.MetaData instance
    self._meta = None

    # pid of process on which _engine was created
    self._engine_pid = None

    self._engine = None

    self.db_migrations_dir = db_migrations_dir
    self.db_credentials_dir = db_credentials_dir
    self.db_name = db_name

    with open(os.path.join(db_credentials_dir, 'password.txt')) as f:
      password = f.read().strip()
    with open(os.path.join(db_credentials_dir, 'host.txt')) as f:
      host = f.read().strip()
    with open(os.path.join(db_credentials_dir, 'user.txt')) as f:
      user = f.read().strip()

    cert = os.path.join(db_credentials_dir, 'client-cert.pem')
    key = os.path.join(db_credentials_dir, 'client-key.pem')
    ca = os.path.join(db_credentials_dir, 'server-ca.pem')
    self._ssl_args = {'ssl': {'cert': cert, 'key': key, 'ca': ca}}

    connect_url = sqlalchemy.engine.url.URL('mysql', username=user,
                                            password=password,
                                            host=host)

    # Create a temporary engine to connect to the mysql instance, and check if
    # a database named |db_name| exists. If not, create one. We use a temporary
    # engine here because the real engine will be opened with a default
    # database name given by |db_name|.
    temp_engine = sqlalchemy.create_engine(connect_url,
                                           connect_args=self._ssl_args,
                                           listeners=[StrictModeListener()])
    databases = temp_engine.execute('SHOW DATABASES').fetchall()
    if (db_name,) not in databases:
      temp_engine.execute('CREATE DATABASE %s' % db_name)
      logging.info('Created database %s', db_name)

    temp_engine.dispose()

    # Now create the persistent connection to the database named |db_name|.
    # If there is a schema version table, read the current schema version
    # from it. Otherwise, assume schema_version 0.
    self._connect_url = sqlalchemy.engine.url.URL('mysql', username=user,
                                                  password=password,
                                                  host=host, database=db_name)

    self.schema_version = self.QuerySchemaVersion()

  def DropDatabase(self):
    """Delete all data and tables from database, and drop database.

    Use with caution. All data in database will be deleted. Invalidates
    this database connection instance.
    """
    self._meta = None
    self._GetEngine().execute('DROP DATABASE %s' % self.db_name)
    self._InvalidateEngine()

  def QuerySchemaVersion(self):
    """Query the database for its current schema version number.

    Returns:
      The current schema version from the database's schema version table,
      as an integer, or 0 if the table is empty or nonexistent.
    """
    tables = self._GetEngine().execute('SHOW TABLES').fetchall()
    if (self.SCHEMA_VERSION_TABLE_NAME,) in tables:
      r = self._GetEngine().execute('SELECT MAX(%s) from %s' %
          (self.SCHEMA_VERSION_COL, self.SCHEMA_VERSION_TABLE_NAME))
      return r.fetchone()[0] or 0
    else:
      return 0

  def _GetMigrationScripts(self):
    """Look for migration scripts and return their versions and paths."

    Returns:
      A list of (schema_version, script_path) tuples of the migration
      scripts for this database, sorted in ascending schema_version order.
    """
    # Look for migration script files in the migration script directory,
    # with names of the form [number]*.sql, and sort these by number.
    migration_scripts = glob.glob(os.path.join(self.db_migrations_dir, '*.sql'))
    migrations = []
    for script in migration_scripts:
      match = re.match(r'([0-9]*).*', os.path.basename(script))
      if match:
        migrations.append((int(match.group(1)), script))

    migrations.sort()
    return migrations

  def ApplySchemaMigrations(self, maxVersion=None):
    """Apply pending migration scripts to database, in order.

    Args:
      maxVersion: The highest version migration script to apply. If
                  unspecified, all migrations found will be applied.
    """
    migrations = self._GetMigrationScripts()

    # Execute the migration scripts in order, asserting that each one
    # updates the schema version to the expected number. If maxVersion
    # is specified stop early.
    for (number, script) in migrations:
      if maxVersion is not None and number > maxVersion:
        break

      if number > self.schema_version:
        # Invalidate self._meta, then run script and ensure that schema
        # version was increased.
        self._meta = None
        logging.info('Running migration script %s', script)
        self.RunQueryScript(script)
        self.schema_version = self.QuerySchemaVersion()
        if self.schema_version != number:
          raise DBException('Migration script %s did not update '
                            'schema version to %s as expected. ' % (number,
                                                                    script))

  def RunQueryScript(self, script_path):
    """Run a .sql script file located at |script_path| on the database."""
    with open(script_path, 'r') as f:
      script = f.read()
    queries = [q.strip() for q in script.split(';') if q.strip()]
    for q in queries:
      self._GetEngine().execute(q)

  def _ReflectToMetadata(self):
    """Use sqlalchemy reflection to construct MetaData model of database.

    If self._meta is already populated, this does nothing.
    """
    if self._meta is not None:
      return
    self._meta = MetaData()
    self._meta.reflect(bind=self._GetEngine())

  def _Insert(self, table, values):
    """Create and execute an INSERT query.

    Args:
      table: Table name to insert to.
      values: Dictionary of column values to insert. Or, list of
              value dictionaries to insert multiple rows.

    Returns:
      Integer primary key of the last inserted row.
    """
    self._ReflectToMetadata()
    ins = self._meta.tables[table].insert()
    r = self._Execute(ins, values)
    return r.inserted_primary_key[0]

  def _InsertMany(self, table, values):
    """Create and execute an multi-row INSERT query.

    Args:
      table: Table name to insert to.
      values: A list of value dictionaries to insert multiple rows.

    Returns:
      The number of inserted rows.
    """
    self._ReflectToMetadata()
    ins = self._meta.tables[table].insert()
    r = self._Execute(ins, values)
    return r.rowcount

  def _GetPrimaryKey(self, table):
    """Gets the primary key column of |table|.

    This function requires that the given table have a 1-column promary key.

    Args:
      table: Name of table to primary key for.

    Returns:
      A sqlalchemy.sql.schema.Column representing the primary key column.

    Raises:
      DBException if the table does not have a single column primary key.
   """
    self._ReflectToMetadata()
    t = self._meta.tables[table]
    key_columns = t.primary_key.columns.values()
    if len(key_columns) != 1:
      raise DBException('Table %s does not have a 1-column primary '
                        'key.' % table)
    return key_columns[0]

  def _Update(self, table, row_id, values):
    """Create and execute an UPDATE query by primary key.

    Args:
      table: Table name to update.
      row_id: Primary key value of row to update.
      values: Dictionary of column values to update.

    Returns:
      The number of rows that were updated (0 or 1).
    """
    self._ReflectToMetadata()
    primary_key = self._GetPrimaryKey(table)
    upd = self._meta.tables[table].update().where(primary_key==row_id)
    r = self._Execute(upd, values)
    return r.rowcount

  def _Execute(self, query, *args, **kwargs):
    """Execute a query using engine, with retires.

    This method wraps execution of a query in a single retry in case the
    engine's connection has been dropped.

    Args:
      query: Query to execute, of type string, or sqlalchemy.Executible,
             or other sqlalchemy-executible statement (see sqlalchemy
             docs).
      *args: Additional args passed along to .execute(...)
      **kwargs: Additional args passed along to .execute(...)

    Returns:
      The result of .execute(...)
    """
    try:
      return self._GetEngine().execute(query, *args, **kwargs)
    except sqlalchemy.exc.OperationalError as e:
      error_code = e.orig.args[0]
      # Error coded 2006 'MySQL server has gone away' indicates that the
      # connection used was closed or dropped.
      if error_code == 2006:
        logging.debug('Retrying a query on engine %s, due to dropped '
                      'connection.', self._GetEngine())
        return self._GetEngine().execute(query, *args, **kwargs)
      else:
        raise

  def _GetEngine(self):
    """Get the sqlalchemy engine for this process.

    This method creates a new sqlalchemy engine if necessary, and
    returns an engine that is unique to this process.

    Returns:
      An sqlalchemy.engine instance for this database.
    """
    pid = os.getpid()
    if pid == self._engine_pid and self._engine:
      return self._engine
    else:
      e = sqlalchemy.create_engine(self._connect_url,
                                   connect_args=self._ssl_args,
                                   listeners=[StrictModeListener()])
      self._engine = e
      self._engine_pid = pid
      logging.debug('Created cidb engine %s@%s for pid %s', e.url.username,
                    e.url.host, pid)
      return self._engine

  def _InvalidateEngine(self):
    """Dispose of an sqlalchemy engine."""
    try:
      pid = os.getpid()
      if pid == self._engine_pid and self._engine:
        self._engine.dispose()
    finally:
      self._engine = None
      self._meta = None


class CIDBConnection(SchemaVersionedMySQLConnection):
  """Connection to a Continuous Integration database."""
  def __init__(self, db_credentials_dir):
    super(CIDBConnection, self).__init__('cidb', CIDB_MIGRATIONS_DIR,
                                         db_credentials_dir)

  @minimum_schema(2)
  def InsertBuild(self, builder_name, waterfall, build_number,
                  build_config, bot_hostname, start_time=None,
                  master_build_id=None):
    """Insert a build row.

    Args:
      builder_name: buildbot builder name.
      waterfall: buildbot waterfall name.
      build_number: buildbot build number.
      build_config: cbuildbot config of build
      bot_hostname: hostname of bot running the build
      start_time: (Optional) Unix timestamp of build start time. If None,
                  current time will be used.
      master_build_id: (Optional) primary key of master build to this build.
    """
    start_time = start_time or time.mktime()
    dt = datetime.datetime.fromtimestamp(start_time)

    return self._Insert('buildTable', {'builder_name': builder_name,
                                       'buildbot_generation':
                                         constants.BUILDBOT_GENERATION,
                                       'waterfall': waterfall,
                                       'build_number': build_number,
                                       'build_config' : build_config,
                                       'bot_hostname': bot_hostname,
                                       'start_time' : dt,
                                       'master_build_id' : master_build_id}
                        )

  @minimum_schema(3)
  def InsertCLActions(self, build_id, cl_actions):
    """Insert a list of |cl_actions|.

    If |cl_actions| is empty, this function does nothing.

    Args:
      build_id: primary key of build that performed these actions.
      cl_actions: A list of cl_action tuples.

    Returns:
      Number of actions inserted.
    """
    if not cl_actions:
      return 0

    values = []
    # TODO(akeshet): Refactor to use either cl action tuples out of the
    # metadata dict (as now) OR CLActionTuple objects.
    for cl_action in cl_actions:
      change_source = 'internal' if cl_action[0]['internal'] else 'external'
      change_number = cl_action[0]['gerrit_number']
      patch_number = cl_action[0]['patch_number']
      action = cl_action[1]
      timestamp = cl_action[2]
      reason = cl_action[3]
      values.append({
          'build_id' : build_id,
          'change_source' : change_source,
          'change_number': change_number,
          'patch_number' : patch_number,
          'action' : action,
          'timestamp' : datetime.datetime.fromtimestamp(timestamp),
          'reason' : reason})

    return self._InsertMany('clActionTable', values)

  @minimum_schema(4)
  def InsertBuildStage(self, build_id, stage_name, board, status,
                       log_url, duration_seconds, summary):
    """Insert a build stage into buildStageTable.

    Args:
      build_id: id of responsible build
      stage_name: name of stage
      board: board that stage ran for
      status: 'pass' or 'fail'
      log_url: URL of stage log
      duration_seconds: run time of stage, in seconds
      summary: summary message of stage

    Returns:
      Primary key of inserted stage.
    """
    return self._Insert('buildStageTable',
                        {'build_id': build_id,
                         'name': stage_name,
                         'board': board,
                         'status': status,
                         'log_url': log_url,
                         'duration_seconds': duration_seconds,
                         'summary': summary})

  @minimum_schema(4)
  def InsertBuildStages(self, stages):
    """For testing only. Insert multiple build stages into buildStageTable.

    This method allows integration tests to more quickly populate build
    stages into the database, from test data. Normal builder operations are
    expected to insert build stage rows one at a time, using InsertBuildStage.

    Args:
      stages: A list of dictionaries, each dictionary containing keys
              build_id, name, board, status, log_url, duration_seconds, and
              summary.

    Returns:
      The number of build stage rows inserted.
    """
    if not stages:
      return 0
    return self._InsertMany('buildStageTable',
                            stages)

  @minimum_schema(2)
  def UpdateMetadata(self, build_id, metadata):
    """Update the given metadata row in database.

    Args:
      build_id: id of row to update.
      metadata: CBuildbotMetadata instance to update with.

    Returns:
      The number of build rows that were updated (0 or 1).
    """
    d = metadata.GetDict()
    versions = d.get('version') or {}
    return self._Update('buildTable', build_id,
                        {'chrome_version': versions.get('chrome'),
                         'milestone_version': versions.get('milestone'),
                         'platform_version': versions.get('platform'),
                         'full_version': versions.get('full'),
                         'sdk_version': d.get('sdk-versions'),
                         'toolchain_url': d.get('toolchain-url'),
                         'build_type': d.get('build_type'),
                         'metadata_json': metadata.GetJSON()})

  @minimum_schema(2)
  def FinishBuild(self, build_id, finish_time=None, status=None,
                  status_pickle=None):
    """Update the given build row, marking it as finished.

    This should be called once per build, as the last update to the build.
    This will also mark the row's final=True.

    Args:
      build_id: id of row to update.
      finish_time: Unix timestamp of build finish time. If None, current time
                   will be used.
      status: Final build status, one of
              manifest_version.BuilderStatus.COMPLETED_STATUSES.
      status_pickle: Pickled manifest_version.BuilderStatus.
    """
    self._ReflectToMetadata()
    finish_time = finish_time or time.mktime()
    dt = datetime.datetime.fromtimestamp(finish_time)

    # TODO(akeshet) atomically update the final field of metadata to
    # True
    self._Update('buildTable', build_id, {'finish_time' : dt,
                                          'status' : status,
                                          'status_pickle' : status_pickle,
                                          'final' : True})


class CIDBConnectionFactory(object):
  """Factory class used by builders to fetch the appropriate cidb connection"""

  # A call to one of the Setup methods below is necessary before using the
  # GetCIDBConnectionForBuilder Factory. This ensures that unit tests do not
  # accidentally use one of the live database instances.

  _ConnectionIsSetup = False
  _ConnectionType = None
  _ConnectionCredsPath = None
  _MockCIDB = None

  _CONNECTION_TYPE_PROD = 'prod'   # production database
  _CONNECTION_TYPE_DEBUG = 'debug' # debug database, used by --debug builds
  _CONNECTION_TYPE_MOCK = 'mock'   # mock connection, not backed by database
  _CONNECTION_TYPE_NONE = 'none'   # explicitly no connection


  @classmethod
  def IsCIDBSetup(cls):
    """Returns True iff GetCIDBConnectionForBuilder is ready to be called."""
    return cls._ConnectionIsSetup


  @classmethod
  def SetupProdCidb(cls):
    """Sets up CIDB to use the prod instance of the database.

    May be called only once, and may not be called after any other CIDB Setup
    method, otherwise it will raise an AssertionError.
    """
    assert not cls._ConnectionIsSetup, 'CIDB is already set up.'
    assert not cls._ConnectionType == cls._CONNECTION_TYPE_MOCK, (
        'CIDB set for mock use.')
    cls._ConnectionType = cls._CONNECTION_TYPE_PROD
    cls._ConnectionCredsPath = constants.CIDB_PROD_BOT_CREDS
    cls._ConnectionIsSetup = True


  @classmethod
  def SetupDebugCidb(cls):
    """Sets up CIDB to use the debug instance of the database.

    May be called only once, and may not be called after any other CIDB Setup
    method, otherwise it will raise an AssertionError.
    """
    assert not cls._ConnectionIsSetup, 'CIDB is already set up.'
    assert not cls._ConnectionType == cls._CONNECTION_TYPE_MOCK, (
        'CIDB set for mock use.')
    cls._ConnectionType = cls._CONNECTION_TYPE_DEBUG
    cls._ConnectionCredsPath = constants.CIDB_DEBUG_BOT_CREDS
    cls._ConnectionIsSetup = True


  @classmethod
  def SetupMockCidb(cls, mock_cidb=None):
    """Sets up CIDB to use a mock object. May be called more than once.

    Args:
      mock_cidb: (optional) The mock cidb object to be returned by
                 GetCIDBConnection. If not supplied, then CIDB will be
                 considered not set up, but future calls to set up a
                 non-(mock or None) connection will fail.
    """
    if cls._ConnectionIsSetup:
      assert cls._ConnectionType == cls._CONNECTION_TYPE_MOCK, (
          'A non-mock CIDB is already set up.')
    cls._ConnectionType = cls._CONNECTION_TYPE_MOCK
    if mock_cidb:
      cls._ConnectionIsSetup = True
      cls._MockCIDB = mock_cidb


  @classmethod
  def SetupNoCidb(cls):
    """Sets up CIDB to use an explicit None connection.

    May be called more than once, or after SetupMockCidb.
    """
    if cls._ConnectionIsSetup:
      assert (cls._ConnectionType == cls._CONNECTION_TYPE_MOCK or
              cls._ConnectionType == cls._CONNECTION_TYPE_NONE) , (
          'A non-mock CIDB is already set up.')
    cls._ConnectionType = cls._CONNECTION_TYPE_NONE
    cls._ConnectionIsSetup = True


  @classmethod
  def GetCIDBConnectionForBuilder(cls):
    """Get a CIDBConnection.

    A call to one of the CIDB Setup methods must have been made before calling
    this factory method.

    Returns:
      A CIDBConnection instance connected to either the prod or debug
      instance of the database, or a mock connection, depending on which
      Setup method was called. Returns None if CIDB has been explicitly
      set up for that using SetupNoCidb.
    """
    assert cls._ConnectionIsSetup, 'CIDB has not be set up with a Setup call.'
    if cls._ConnectionType == cls._CONNECTION_TYPE_MOCK:
      return cls._MockCIDB
    elif cls._ConnectionType == cls._CONNECTION_TYPE_NONE:
      return None
    else:
      return CIDBConnection(cls._ConnectionCredsPath)

  @classmethod
  def _ClearCIDBSetup(cls):
    """Clears the CIDB Setup state. For testing purposes only."""
    cls._ConnectionIsSetup = False
    cls._ConnectionType = None
    cls._ConnectionCredsPath = None
    cls._MockCIDB = None