/* KInterbasDB Python Package - Support for Conversion of Nonstandard Blobs
 *
 * Version 3.3
 *
 * The following contributors hold Copyright (C) over their respective
 * portions of code (see license.txt for details):
 *
 * [Original Author (maintained through version 2.0-0.3.1):]
 *   1998-2001 [alex]  Alexander Kuznetsov   <alexan@users.sourceforge.net>
 * [Maintainers (after version 2.0-0.3.1):]
 *   2001-2002 [maz]   Marek Isalski         <kinterbasdb@maz.nu>
 *   2002-2007 [dsr]   David Rushby          <woodsplitter@rocketmail.com>
 * [Contributors:]
 *   2001      [eac]   Evgeny A. Cherkashin  <eugeneai@icc.ru>
 *   2001-2002 [janez] Janez Jere            <janez.jere@void.si>
 */

/* This source file is designed to be directly included in _kiconversion.c,
 * without the involvement of a header file. */

/* The isc_blob_lookup_desc function, at least in FB 1.5 and 2.0, has a
 * concurrency bug that causes severe problems if the function is called from
 * multiple threads at once.  (In the function's definition in blob.epp, a
 * global database handle is set to the one the client code has passed in, and
 * then some GPRE-generated code is run.  Even synchronizing all calls to
 * isc_blob_lookup_desc doesn't fix the problem, because the function tries to
 * clear open requests its previous handle.)
 *
 * On 2006.01.30, DSR reimplemented the functionality of isc_blob_lookup_desc
 * in this file in order to bypass the concurrency bug.  Defining
 * USE_OFFICIAL_ISC_BLOB_LOOKUP_DESC will restore the old code. */
/* #define USE_OFFICIAL_ISC_BLOB_LOOKUP_DESC */


/* Infinitely peristent global "constants" that support streaming blob
 * operations (initialized by init_kidb_nonstandard_blob_support): */
static PyObject *blob_nonstandard__config_slot_name__mode;
static PyObject *blob_nonstandard__config_slot_name__treat_subtype_text_as_text;

static PyObject *blob_nonstandard__config_value__stream;
static PyObject *blob_nonstandard__config_value__materialize;

static PyObject *blob_streaming__reader_mode__rb;

static PyObject *blob_streaming__method_name_read;
static PyObject *blob_streaming__method_name__iter_read_chunk;

static PyObject *blob_streaming__MAX_BLOB_SEGMENT_SIZE;
static PyObject *blob_streaming__1Tuple_containing_MAX_BLOB_SEGMENT_SIZE;

#define DEFAULT_BLOB_STREAM_BUFFER_SIZE USHRT_MAX

#define DTT_BLOB_CHUNK_SIZE_UNSPECIFIED -1


/******************** UTILITY FUNCTIONS:BEGIN ********************/

static int init_kidb_nonstandard_blob_support(void) {
  ISWEC(blob_nonstandard__config_slot_name__mode, "mode");

  ISWEC(blob_nonstandard__config_value__stream, "stream");
  ISWEC(blob_nonstandard__config_value__materialize, "materialize");

  ISWEC(blob_nonstandard__config_slot_name__treat_subtype_text_as_text,
      "treat_subtype_text_as_text"
    );

  ISWEC(blob_streaming__reader_mode__rb, "rb");

  ISWEC(blob_streaming__method_name_read, "read");
  ISWEC(blob_streaming__method_name__iter_read_chunk, "_iter_read_chunk");

  blob_streaming__MAX_BLOB_SEGMENT_SIZE = PyInt_FromLong(MAX_BLOB_SEGMENT_SIZE);
  if (blob_streaming__MAX_BLOB_SEGMENT_SIZE == NULL) { goto fail; }

  blob_streaming__1Tuple_containing_MAX_BLOB_SEGMENT_SIZE = PyTuple_New(1);
  if (blob_streaming__1Tuple_containing_MAX_BLOB_SEGMENT_SIZE == NULL) {
    goto fail;
  }
  Py_INCREF(blob_streaming__MAX_BLOB_SEGMENT_SIZE);
  PyTuple_SET_ITEM(blob_streaming__1Tuple_containing_MAX_BLOB_SEGMENT_SIZE,
      0, blob_streaming__MAX_BLOB_SEGMENT_SIZE
    );

  /* BlobReaderType is a new-style class, so PyType_Ready must be called before
   * its getters and setters will function. */
  if (PyType_Ready(&BlobReaderType) < 0) { goto fail; }

  return 0;

  fail:
    /* This function is indirectly called by the module loader, which makes no
     * provision for error recovery, so we don't attempt to reclaim any
     * allocated memory. */
    return -1;
} /* init_kidb_nonstandard_blob_support */

static int validate_nonstandard_blob_config_dict(PyObject *config,
    BlobMode *mode, boolean *treat_subtype_text_as_text
  )
{
  PyObject *py_mode;
  PyObject *py_treat_subtype_text_as_text;

  /* Note that the py_err_ variables are XDECREFed in the universal error
   * block at the end of the function, so they needn't be explicitly DECREFed
   * in specific error handling blocks. */
  PyObject *py_err_msg = NULL;
  PyObject *py_err_support = NULL;

  int mode_comp;

  assert (config != NULL);
  assert (PyDict_Check(config));

  /* Validate 'mode' (must be present and must equal either 'materialize' or
   * 'stream'): */
  py_mode = PyDict_GetItem(config, blob_nonstandard__config_slot_name__mode);
  if (py_mode == NULL) {
    raise_exception(ProgrammingError, "'BLOB' dynamic type translator"
        " configuration dictionary must include 'mode' setting."
      );
    goto fail;
  }

  mode_comp = PyObject_Compare(py_mode, blob_nonstandard__config_value__stream);
  if (PyErr_Occurred()) {
    /* The comparison operation raised an exception. */
    goto fail;
  }
  if (mode_comp == 0) {
    *mode = blob_mode_stream;
  } else {
    mode_comp = PyObject_Compare(py_mode,
        blob_nonstandard__config_value__materialize
      );
    if (mode_comp == 0) {
      *mode = blob_mode_materialize;
    } else {
      py_err_support = PyObject_Str(py_mode);
      if (py_err_support == NULL) { goto fail; }
      py_err_msg = PyString_FromFormat("'BLOB' dynamic type translator"
          " configuration dictionary includes unrecognized 'mode' setting"
          " '%s'.  At present, 'mode' must be one of"
          " ('materialize', 'stream').",
          PyString_AS_STRING(py_err_support)
        );
      if (py_err_msg == NULL) { goto fail; }
      raise_exception(ProgrammingError, PyString_AS_STRING(py_err_msg));
      goto fail;
    }
  }

  /* Validate 'treat_subtype_text_as_text' (optional boolean; defaults to
   * false). */
  py_treat_subtype_text_as_text = PyDict_GetItem(config,
      blob_nonstandard__config_slot_name__treat_subtype_text_as_text
    );
  if (py_treat_subtype_text_as_text == NULL) {
    /* The value was missing; default to false (since the engine (prior to FB
     * 2.1) doesn't provide the supporting values required to determine the
     * character set of incoming blobs, we obviously don't want to enable that
     * feature by default). */
    *treat_subtype_text_as_text = FALSE;
  } else {
    const int py_treat_subtype_text_as_text__as_bool =
      PyObject_IsTrue(py_treat_subtype_text_as_text);
    if (py_treat_subtype_text_as_text__as_bool == -1) { goto fail; }
    *treat_subtype_text_as_text = (boolean) py_treat_subtype_text_as_text__as_bool;
  }

  assert (py_err_msg == NULL);
  assert (py_err_support == NULL);
  return DTT_BLOB_CONFIG_VALID;

  fail:
    Py_XDECREF(py_err_msg);
    Py_XDECREF(py_err_support);

    return DTT_BLOB_CONFIG_INVALID;
} /* validate_nonstandard_blob_config_dict */

static PyObject *determine_blob_character_set_id(
    Cursor *cursor, XSQLVAR *sqlvar
  )
{
  /* Returns a new reference to a Python int containing the character set ID
   * on success; NULL on failure. */
  #ifdef USE_OFFICIAL_ISC_BLOB_LOOKUP_DESC
  ISC_BLOB_DESC blob_desc;
  #endif
  Transaction *trans;
  CConnection *con;
  boolean might_be_cached = TRUE;
  PyObject *py_reldict = NULL;
  PyObject *py_relname = NULL;
  PyObject *py_fieldname = NULL;
  PyObject *py_char_set_id = NULL;

  assert (cursor != NULL);
  assert (cursor->trans != NULL);
  assert (sqlvar != NULL);

  /* 2007.02.10: */
  /* Firebird 2.1 will store the blob's character set ID in the
   * sqlvar->sqlscale field, which will make this function's task trivial.
   * The relevant FB changelog entry is:
   *   2006-08-08 13:23  asfernandes
   *     firebird2/src/: dsql/dsql.cpp (1.178), isql/isql.epp (1.196):
   *     Put blob charset in XSQLVAR::sqlscale
   */
  if (sqlvar->sqlscale != 0) { /* For FB 2.1+ */
    return PyInt_FromLong(sqlvar->sqlscale);
  }

  trans = cursor->trans;
  con = Transaction_get_con(trans);
  assert (con != NULL);

  /* The fields XSQLVAR.relname and XSQLVAR.sqlname are not null-terminated,
   * but isc_blob_lookup_desc expects null-terminated strings.  Apparently the
   * folks who designed the API expected that client programmers would be using
   * C string constants for the relation name and field name.  Welcome to the
   * 1980s. */

  if (sqlvar->relname_length == 0 && sqlvar->sqlname_length == 0) {
    PyObject *py_char_set_id = PyObject_CallMethod(con->python_wrapper_obj,
        "_guessBlobCharSetIDFromConnectionCharSet", NULL
      );
    if (py_char_set_id != NULL && py_char_set_id != Py_None) {
      return py_char_set_id;
    } else {
      raise_exception(OperationalError, "At this stage of execution, the"
          " database client library has not provided enough information to"
          " allow KInterbasDB to automatically determine the blob's charset.\n"
          "Firebird 2.1 *does* provide the required information.\n"
          "If you call kinterbasdb.init(type_conv=300), KInterbasDB will:\n"
          " - work perfectly with FB 2.1+\n"
          " - dangerously assume that the character set of all textual blobs"
          " is the same as the Connection.charset with FB < 2.1"
        );
      goto fail;
    }
  }

  /* Python will intern most relation and field names, so the following string
   * creation ops should be very efficient. */
  py_relname = PyString_FromStringAndSize(sqlvar->relname, sqlvar->relname_length);
  if (py_relname == NULL) { goto fail; }
  py_fieldname = PyString_FromStringAndSize(sqlvar->sqlname, sqlvar->sqlname_length);
  if (py_fieldname == NULL) { goto fail; }

  /* Create a cache dictionary if none exists (kinterbasdb connections can be
   * used by no more than one thread at once, so this doesn't present any
   * thread-safety problems). */
  if (con->blob_charset_cache == NULL) {
    con->blob_charset_cache = PyDict_New();
    if (con->blob_charset_cache == NULL) { goto fail; }
    might_be_cached = FALSE;
  } else {
    /* PyDict_GetItem returns borrowed ref. */
    py_reldict = PyDict_GetItem(con->blob_charset_cache, py_relname);
    might_be_cached = (boolean) (py_reldict != NULL);
    if (might_be_cached) {
      py_char_set_id = PyDict_GetItem(py_reldict, py_fieldname);
      if (py_char_set_id != NULL) {
        might_be_cached = TRUE; /* Was cached. */
        Py_INCREF(py_char_set_id); /* PyDict_GetItem returned borrowed ref. */
      } else {
        might_be_cached = FALSE;
      }
    }
  }

  /* By this point in the function, might_be_cached is no longer tentative--
   * if might_be_cached is true, then the result *was* cached, and vice versa. */
  assert (might_be_cached ? py_char_set_id != NULL : py_char_set_id == NULL);
  if (!might_be_cached) {
    #ifdef USE_OFFICIAL_ISC_BLOB_LOOKUP_DESC
      /* Pre-fetch pointers to the internal char buffers of py_relname and
       * py_fieldname so as to avoid calling the Python API when the GIL is not
       * held. */
      unsigned char *relname_nullterminated =
        (unsigned char *) PyString_AS_STRING(py_relname);
      unsigned char *fieldname_nullterminated =
        (unsigned char *) PyString_AS_STRING(py_fieldname);
      /* Transaction_get_handle_p must be called while the GIL is held! */
      isc_tr_handle *trans_handle_p = Transaction_get_handle_p(trans);

      ENTER_GDAL
      isc_blob_lookup_desc(cursor->status_vector,
          &con->db_handle, trans_handle_p,
          relname_nullterminated, fieldname_nullterminated,
          &blob_desc,
          NULL
        );
      LEAVE_GDAL
      if (DB_API_ERROR(cursor->status_vector)) {
        raise_sql_exception(OperationalError,
            "determine_blob_character_set_id: ", cursor->status_vector
          );
        goto fail;
      }
      py_char_set_id = PyInt_FromLong(blob_desc.blob_desc_charset);
      if (py_char_set_id == NULL) { goto fail; }
    #else
    {
      ISC_STATUS *sv = con->status_vector;
      isc_db_handle *db_handle = &con->db_handle;
      isc_tr_handle *trans_handle = Transaction_get_handle_p(trans);

      const char *charset_determination_statement =
          "SELECT F.RDB$CHARACTER_SET_ID "
          "FROM  RDB$RELATION_FIELDS RF "
            "JOIN RDB$FIELDS F "
              "ON RF.RDB$FIELD_SOURCE = F.RDB$FIELD_NAME "
          "WHERE "
                "RF.RDB$RELATION_NAME = ? "
            "AND RF.RDB$FIELD_NAME = ?"
        ;

      char *in_da_mem[XSQLDA_LENGTH(2)];
      XSQLDA *in_da = (XSQLDA *) in_da_mem;
      char *out_da_mem[XSQLDA_LENGTH(1)];
      XSQLDA *out_da = (XSQLDA *) out_da_mem;
      XSQLVAR *in_var = NULL;
      XSQLVAR *out_var = NULL;
      short out_var_charset_id = 0;
      short out_var_sqlind = SQLIND_NULL;
      isc_stmt_handle stmt_handle = NULL_STMT_HANDLE;

      in_da->version = SQLDA_VERSION_KIDB;
      in_da->sqln = 2;
      in_da->sqld = 2;

      /* Set the names of the relation.field for which we're determining
       * character set. */
      in_var = in_da->sqlvar; /* First input variable. */
      in_var->sqltype = SQL_TEXT;
      /* The contents of py_relname were supplied by the DB engine, so no
       * danger of overflowing short: */
      in_var->sqllen = (short) PyString_GET_SIZE(py_relname);
      in_var->sqldata = PyString_AS_STRING(py_relname);

      in_var++; /* Second input variable. */
      in_var->sqltype = SQL_TEXT;
      in_var->sqllen = (short) PyString_GET_SIZE(py_fieldname);
      in_var->sqldata = PyString_AS_STRING(py_fieldname);

      /* Set up the output structures.  We know at design time exactly how they
       * should be configured. */
      out_da->version = SQLDA_VERSION_KIDB;
      out_da->sqln = 1;

      out_var = out_da->sqlvar;
      out_var->sqldata = (char *) &out_var_charset_id;
      out_var->sqlind = &out_var_sqlind;

      ENTER_GDAL
      isc_dsql_allocate_statement(sv, db_handle, &stmt_handle);
      LEAVE_GDAL
      if (DB_API_ERROR(sv)) {
        raise_sql_exception(OperationalError,
            "determine_blob_character_set_id: ", sv
          );
        goto fail;
      }

      ENTER_GDAL
      isc_dsql_prepare(sv, trans_handle, &stmt_handle,
          0, (char *) charset_determination_statement, SQL_DIALECT_DEFAULT,
          out_da
        );
      LEAVE_GDAL
      if (DB_API_ERROR(sv)) {
        raise_sql_exception(OperationalError,
            "determine_blob_character_set_id: ", sv
          );
        goto fail;
      }

      ENTER_GDAL
      isc_dsql_execute2(sv, trans_handle, &stmt_handle,
          SQL_DIALECT_DEFAULT, in_da, out_da
        );
      LEAVE_GDAL
      if (DB_API_ERROR(sv)) {
        raise_sql_exception(OperationalError,
            "determine_blob_character_set_id: ", sv
          );
        goto fail;
      }

      if (out_var_sqlind == SQLIND_NULL) {
        assert (out_var_charset_id == 0);
      } else {
        out_var_charset_id = *((short *) out_var->sqldata);
      }

      py_char_set_id = PyInt_FromLong(out_var_charset_id);
      if (py_char_set_id == NULL) { goto fail; }
    }
    #endif

    /* Write the result to the cache before returning it. */

    /* This should've been ensured earlier: */
    assert (con->blob_charset_cache != NULL);

    assert (py_char_set_id != NULL);
    py_reldict = PyDict_GetItem(con->blob_charset_cache, py_relname);
    if (py_reldict == NULL) {
      /* Create a dictionary for all fields of the relation named py_relname: */
      py_reldict = PyDict_New();
      if (py_reldict == NULL) { goto fail; }
      if (PyDict_SetItem(con->blob_charset_cache, py_relname, py_reldict) != 0) {
        Py_DECREF(py_reldict);
        goto fail;
      }
    }
    if (PyDict_SetItem(py_reldict, py_fieldname, py_char_set_id) != 0) {
      goto fail;
    }
  } /* end cache miss block */

  assert (py_char_set_id != NULL);
  goto clean;
  fail:
    assert (PyErr_Occurred());
    if (py_char_set_id != NULL) {
      Py_DECREF(py_char_set_id);
      py_char_set_id = NULL;
    }
    /* Fall through to clean: */
  clean:
    /* Note that py_reldict is not DECREFed; we only have borrowed ref to it. */
    Py_XDECREF(py_relname);
    Py_XDECREF(py_fieldname);

    return py_char_set_id; /* Returns new reference. */
} /* determine_blob_character_set_id */

static int get_blob_converter_override_for_direction(
    boolean direction_is_in, Cursor *cursor, XSQLVAR *sqlvar,
    PyObject **py_converter_override,
    PyObject **py_blob_charset_id,
    boolean *is_unicode_charset
  )
{
  /* py_converter_override will hold only a borrowed ref, so client code need
   * not worries about DECREFing it.  py_blob_charset_id will hold a new ref,
   * however. */
  int blob_charset_id;

  *py_blob_charset_id = determine_blob_character_set_id(cursor, sqlvar);
  if (*py_blob_charset_id == NULL) { goto fail; }
  blob_charset_id = PyInt_AS_LONG(*py_blob_charset_id);
  *is_unicode_charset = (boolean) (blob_charset_id < 0 || blob_charset_id > 2);

  {
    PyObject *type_name =
        *is_unicode_charset ?
            cached_type_name_TEXT_UNICODE
          : cached_type_name_TEXT
      ;
    if (direction_is_in) {
      *py_converter_override = cursor_get_in_converter_for_type_name(cursor,
          type_name
        );
    } else {
      *py_converter_override = cursor_get_out_converter_for_type_name(cursor,
          type_name
        );
    }
  }
  if (*py_converter_override == NULL) { goto fail; }

  return 0;

  fail:
    assert (PyErr_Occurred());

    return -1;
} /* get_blob_converter_override_for_direction */


/******************** UTILITY FUNCTIONS:END ********************/