michael@0: /* This Source Code Form is subject to the terms of the Mozilla Public
michael@0:  * License, v. 2.0. If a copy of the MPL was not distributed with this
michael@0:  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
michael@0: 
michael@0: "use strict";
michael@0: 
michael@0: this.EXPORTED_SYMBOLS = [
michael@0:   "Sqlite",
michael@0: ];
michael@0: 
michael@0: const {classes: Cc, interfaces: Ci, utils: Cu} = Components;
michael@0: 
michael@0: Cu.import("resource://gre/modules/Promise.jsm");
michael@0: Cu.import("resource://gre/modules/osfile.jsm");
michael@0: Cu.import("resource://gre/modules/Services.jsm");
michael@0: Cu.import("resource://gre/modules/XPCOMUtils.jsm");
michael@0: Cu.import("resource://gre/modules/Log.jsm");
michael@0: 
michael@0: XPCOMUtils.defineLazyModuleGetter(this, "AsyncShutdown",
michael@0:                                   "resource://gre/modules/AsyncShutdown.jsm");
michael@0: XPCOMUtils.defineLazyModuleGetter(this, "CommonUtils",
michael@0:                                   "resource://services-common/utils.js");
michael@0: XPCOMUtils.defineLazyModuleGetter(this, "FileUtils",
michael@0:                                   "resource://gre/modules/FileUtils.jsm");
michael@0: XPCOMUtils.defineLazyModuleGetter(this, "Task",
michael@0:                                   "resource://gre/modules/Task.jsm");
michael@0: 
michael@0: 
michael@0: // Counts the number of created connections per database basename(). This is
michael@0: // used for logging to distinguish connection instances.
michael@0: let connectionCounters = new Map();
michael@0: 
michael@0: 
michael@0: /**
michael@0:  * Opens a connection to a SQLite database.
michael@0:  *
michael@0:  * The following parameters can control the connection:
michael@0:  *
michael@0:  *   path -- (string) The filesystem path of the database file to open. If the
michael@0:  *       file does not exist, a new database will be created.
michael@0:  *
michael@0:  *   sharedMemoryCache -- (bool) Whether multiple connections to the database
michael@0:  *       share the same memory cache. Sharing the memory cache likely results
michael@0:  *       in less memory utilization. However, sharing also requires connections
michael@0:  *       to obtain a lock, possibly making database access slower. Defaults to
michael@0:  *       true.
michael@0:  *
michael@0:  *   shrinkMemoryOnConnectionIdleMS -- (integer) If defined, the connection
michael@0:  *       will attempt to minimize its memory usage after this many
michael@0:  *       milliseconds of connection idle. The connection is idle when no
michael@0:  *       statements are executing. There is no default value which means no
michael@0:  *       automatic memory minimization will occur. Please note that this is
michael@0:  *       *not* a timer on the idle service and this could fire while the
michael@0:  *       application is active.
michael@0:  *
michael@0:  * FUTURE options to control:
michael@0:  *
michael@0:  *   special named databases
michael@0:  *   pragma TEMP STORE = MEMORY
michael@0:  *   TRUNCATE JOURNAL
michael@0:  *   SYNCHRONOUS = full
michael@0:  *
michael@0:  * @param options
michael@0:  *        (Object) Parameters to control connection and open options.
michael@0:  *
michael@0:  * @return Promise<OpenedConnection>
michael@0:  */
michael@0: function openConnection(options) {
michael@0:   let log = Log.repository.getLogger("Sqlite.ConnectionOpener");
michael@0: 
michael@0:   if (!options.path) {
michael@0:     throw new Error("path not specified in connection options.");
michael@0:   }
michael@0: 
michael@0:   // Retains absolute paths and normalizes relative as relative to profile.
michael@0:   let path = OS.Path.join(OS.Constants.Path.profileDir, options.path);
michael@0: 
michael@0:   let sharedMemoryCache = "sharedMemoryCache" in options ?
michael@0:                             options.sharedMemoryCache : true;
michael@0: 
michael@0:   let openedOptions = {};
michael@0: 
michael@0:   if ("shrinkMemoryOnConnectionIdleMS" in options) {
michael@0:     if (!Number.isInteger(options.shrinkMemoryOnConnectionIdleMS)) {
michael@0:       throw new Error("shrinkMemoryOnConnectionIdleMS must be an integer. " +
michael@0:                       "Got: " + options.shrinkMemoryOnConnectionIdleMS);
michael@0:     }
michael@0: 
michael@0:     openedOptions.shrinkMemoryOnConnectionIdleMS =
michael@0:       options.shrinkMemoryOnConnectionIdleMS;
michael@0:   }
michael@0: 
michael@0:   let file = FileUtils.File(path);
michael@0: 
michael@0:   let basename = OS.Path.basename(path);
michael@0:   let number = connectionCounters.get(basename) || 0;
michael@0:   connectionCounters.set(basename, number + 1);
michael@0: 
michael@0:   let identifier = basename + "#" + number;
michael@0: 
michael@0:   log.info("Opening database: " + path + " (" + identifier + ")");
michael@0:   let deferred = Promise.defer();
michael@0:   let options = null;
michael@0:   if (!sharedMemoryCache) {
michael@0:     options = Cc["@mozilla.org/hash-property-bag;1"].
michael@0:       createInstance(Ci.nsIWritablePropertyBag);
michael@0:     options.setProperty("shared", false);
michael@0:   }
michael@0:   Services.storage.openAsyncDatabase(file, options, function(status, connection) {
michael@0:     if (!connection) {
michael@0:       log.warn("Could not open connection: " + status);
michael@0:       deferred.reject(new Error("Could not open connection: " + status));
michael@0:       return;
michael@0:     }
michael@0:     log.info("Connection opened");
michael@0:     try {
michael@0:       deferred.resolve(
michael@0:         new OpenedConnection(connection.QueryInterface(Ci.mozIStorageAsyncConnection), basename, number,
michael@0:         openedOptions));
michael@0:     } catch (ex) {
michael@0:       log.warn("Could not open database: " + CommonUtils.exceptionStr(ex));
michael@0:       deferred.reject(ex);
michael@0:     }
michael@0:   });
michael@0:   return deferred.promise;
michael@0: }
michael@0: 
michael@0: /**
michael@0:  * Creates a clone of an existing and open Storage connection.  The clone has
michael@0:  * the same underlying characteristics of the original connection and is
michael@0:  * returned in form of on OpenedConnection handle.
michael@0:  *
michael@0:  * The following parameters can control the cloned connection:
michael@0:  *
michael@0:  *   connection -- (mozIStorageAsyncConnection) The original Storage connection
michael@0:  *       to clone.  It's not possible to clone connections to memory databases.
michael@0:  *
michael@0:  *   readOnly -- (boolean) - If true the clone will be read-only.  If the
michael@0:  *       original connection is already read-only, the clone will be, regardless
michael@0:  *       of this option.  If the original connection is using the shared cache,
michael@0:  *       this parameter will be ignored and the clone will be as privileged as
michael@0:  *       the original connection.
michael@0:  *   shrinkMemoryOnConnectionIdleMS -- (integer) If defined, the connection
michael@0:  *       will attempt to minimize its memory usage after this many
michael@0:  *       milliseconds of connection idle. The connection is idle when no
michael@0:  *       statements are executing. There is no default value which means no
michael@0:  *       automatic memory minimization will occur. Please note that this is
michael@0:  *       *not* a timer on the idle service and this could fire while the
michael@0:  *       application is active.
michael@0:  *
michael@0:  *
michael@0:  * @param options
michael@0:  *        (Object) Parameters to control connection and clone options.
michael@0:  *
michael@0:  * @return Promise<OpenedConnection>
michael@0:  */
michael@0: function cloneStorageConnection(options) {
michael@0:   let log = Log.repository.getLogger("Sqlite.ConnectionCloner");
michael@0: 
michael@0:   let source = options && options.connection;
michael@0:   if (!source) {
michael@0:     throw new TypeError("connection not specified in clone options.");
michael@0:   }
michael@0:   if (!source instanceof Ci.mozIStorageAsyncConnection) {
michael@0:     throw new TypeError("Connection must be a valid Storage connection.")
michael@0:   }
michael@0: 
michael@0:   let openedOptions = {};
michael@0: 
michael@0:   if ("shrinkMemoryOnConnectionIdleMS" in options) {
michael@0:     if (!Number.isInteger(options.shrinkMemoryOnConnectionIdleMS)) {
michael@0:       throw new TypeError("shrinkMemoryOnConnectionIdleMS must be an integer. " +
michael@0:                           "Got: " + options.shrinkMemoryOnConnectionIdleMS);
michael@0:     }
michael@0:     openedOptions.shrinkMemoryOnConnectionIdleMS =
michael@0:       options.shrinkMemoryOnConnectionIdleMS;
michael@0:   }
michael@0: 
michael@0:   let path = source.databaseFile.path;
michael@0:   let basename = OS.Path.basename(path);
michael@0:   let number = connectionCounters.get(basename) || 0;
michael@0:   connectionCounters.set(basename, number + 1);
michael@0:   let identifier = basename + "#" + number;
michael@0: 
michael@0:   log.info("Cloning database: " + path + " (" + identifier + ")");
michael@0:   let deferred = Promise.defer();
michael@0: 
michael@0:   source.asyncClone(!!options.readOnly, (status, connection) => {
michael@0:     if (!connection) {
michael@0:       log.warn("Could not clone connection: " + status);
michael@0:       deferred.reject(new Error("Could not clone connection: " + status));
michael@0:     }
michael@0:     log.info("Connection cloned");
michael@0:     try {
michael@0:       let conn = connection.QueryInterface(Ci.mozIStorageAsyncConnection);
michael@0:       deferred.resolve(new OpenedConnection(conn, basename, number,
michael@0:                                             openedOptions));
michael@0:     } catch (ex) {
michael@0:       log.warn("Could not clone database: " + CommonUtils.exceptionStr(ex));
michael@0:       deferred.reject(ex);
michael@0:     }
michael@0:   });
michael@0:   return deferred.promise;
michael@0: }
michael@0: 
michael@0: /**
michael@0:  * Handle on an opened SQLite database.
michael@0:  *
michael@0:  * This is essentially a glorified wrapper around mozIStorageConnection.
michael@0:  * However, it offers some compelling advantages.
michael@0:  *
michael@0:  * The main functions on this type are `execute` and `executeCached`. These are
michael@0:  * ultimately how all SQL statements are executed. It's worth explaining their
michael@0:  * differences.
michael@0:  *
michael@0:  * `execute` is used to execute one-shot SQL statements. These are SQL
michael@0:  * statements that are executed one time and then thrown away. They are useful
michael@0:  * for dynamically generated SQL statements and clients who don't care about
michael@0:  * performance (either their own or wasting resources in the overall
michael@0:  * application). Because of the performance considerations, it is recommended
michael@0:  * to avoid `execute` unless the statement you are executing will only be
michael@0:  * executed once or seldomly.
michael@0:  *
michael@0:  * `executeCached` is used to execute a statement that will presumably be
michael@0:  * executed multiple times. The statement is parsed once and stuffed away
michael@0:  * inside the connection instance. Subsequent calls to `executeCached` will not
michael@0:  * incur the overhead of creating a new statement object. This should be used
michael@0:  * in preference to `execute` when a specific SQL statement will be executed
michael@0:  * multiple times.
michael@0:  *
michael@0:  * Instances of this type are not meant to be created outside of this file.
michael@0:  * Instead, first open an instance of `UnopenedSqliteConnection` and obtain
michael@0:  * an instance of this type by calling `open`.
michael@0:  *
michael@0:  * FUTURE IMPROVEMENTS
michael@0:  *
michael@0:  *   Ability to enqueue operations. Currently there can be race conditions,
michael@0:  *   especially as far as transactions are concerned. It would be nice to have
michael@0:  *   an enqueueOperation(func) API that serially executes passed functions.
michael@0:  *
michael@0:  *   Support for SAVEPOINT (named/nested transactions) might be useful.
michael@0:  *
michael@0:  * @param connection
michael@0:  *        (mozIStorageConnection) Underlying SQLite connection.
michael@0:  * @param basename
michael@0:  *        (string) The basename of this database name. Used for logging.
michael@0:  * @param number
michael@0:  *        (Number) The connection number to this database.
michael@0:  * @param options
michael@0:  *        (object) Options to control behavior of connection. See
michael@0:  *        `openConnection`.
michael@0:  */
michael@0: function OpenedConnection(connection, basename, number, options) {
michael@0:   this._log = Log.repository.getLoggerWithMessagePrefix("Sqlite.Connection." + basename,
michael@0:                                                         "Conn #" + number + ": ");
michael@0: 
michael@0:   this._log.info("Opened");
michael@0: 
michael@0:   this._connection = connection;
michael@0:   this._connectionIdentifier = basename + " Conn #" + number;
michael@0:   this._open = true;
michael@0: 
michael@0:   this._cachedStatements = new Map();
michael@0:   this._anonymousStatements = new Map();
michael@0:   this._anonymousCounter = 0;
michael@0: 
michael@0:   // A map from statement index to mozIStoragePendingStatement, to allow for
michael@0:   // canceling prior to finalizing the mozIStorageStatements.
michael@0:   this._pendingStatements = new Map();
michael@0: 
michael@0:   // Increments for each executed statement for the life of the connection.
michael@0:   this._statementCounter = 0;
michael@0: 
michael@0:   this._inProgressTransaction = null;
michael@0: 
michael@0:   this._idleShrinkMS = options.shrinkMemoryOnConnectionIdleMS;
michael@0:   if (this._idleShrinkMS) {
michael@0:     this._idleShrinkTimer = Cc["@mozilla.org/timer;1"]
michael@0:                               .createInstance(Ci.nsITimer);
michael@0:     // We wait for the first statement execute to start the timer because
michael@0:     // shrinking now would not do anything.
michael@0:   }
michael@0: }
michael@0: 
michael@0: OpenedConnection.prototype = Object.freeze({
michael@0:   TRANSACTION_DEFERRED: "DEFERRED",
michael@0:   TRANSACTION_IMMEDIATE: "IMMEDIATE",
michael@0:   TRANSACTION_EXCLUSIVE: "EXCLUSIVE",
michael@0: 
michael@0:   TRANSACTION_TYPES: ["DEFERRED", "IMMEDIATE", "EXCLUSIVE"],
michael@0: 
michael@0:   /**
michael@0:    * The integer schema version of the database.
michael@0:    *
michael@0:    * This is 0 if not schema version has been set.
michael@0:    *
michael@0:    * @return Promise<int>
michael@0:    */
michael@0:   getSchemaVersion: function() {
michael@0:     let self = this;
michael@0:     return this.execute("PRAGMA user_version").then(
michael@0:       function onSuccess(result) {
michael@0:         if (result == null) {
michael@0:           return 0;
michael@0:         }
michael@0:         return JSON.stringify(result[0].getInt32(0));
michael@0:       }
michael@0:     );
michael@0:   },
michael@0: 
michael@0:   setSchemaVersion: function(value) {
michael@0:     if (!Number.isInteger(value)) {
michael@0:       // Guarding against accidental SQLi
michael@0:       throw new TypeError("Schema version must be an integer. Got " + value);
michael@0:     }
michael@0:     this._ensureOpen();
michael@0:     return this.execute("PRAGMA user_version = " + value);
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Close the database connection.
michael@0:    *
michael@0:    * This must be performed when you are finished with the database.
michael@0:    *
michael@0:    * Closing the database connection has the side effect of forcefully
michael@0:    * cancelling all active statements. Therefore, callers should ensure that
michael@0:    * all active statements have completed before closing the connection, if
michael@0:    * possible.
michael@0:    *
michael@0:    * The returned promise will be resolved once the connection is closed.
michael@0:    *
michael@0:    * IMPROVEMENT: Resolve the promise to a closed connection which can be
michael@0:    * reopened.
michael@0:    *
michael@0:    * @return Promise<>
michael@0:    */
michael@0:   close: function () {
michael@0:     if (!this._connection) {
michael@0:       return Promise.resolve();
michael@0:     }
michael@0: 
michael@0:     this._log.debug("Request to close connection.");
michael@0:     this._clearIdleShrinkTimer();
michael@0:     let deferred = Promise.defer();
michael@0: 
michael@0:     AsyncShutdown.profileBeforeChange.addBlocker(
michael@0:       "Sqlite.jsm: " + this._connectionIdentifier,
michael@0:       deferred.promise
michael@0:     );
michael@0: 
michael@0:     // We need to take extra care with transactions during shutdown.
michael@0:     //
michael@0:     // If we don't have a transaction in progress, we can proceed with shutdown
michael@0:     // immediately.
michael@0:     if (!this._inProgressTransaction) {
michael@0:       this._finalize(deferred);
michael@0:       return deferred.promise;
michael@0:     }
michael@0: 
michael@0:     // Else if we do have a transaction in progress, we forcefully roll it
michael@0:     // back. This is an async task, so we wait on it to finish before
michael@0:     // performing finalization.
michael@0:     this._log.warn("Transaction in progress at time of close. Rolling back.");
michael@0: 
michael@0:     let onRollback = this._finalize.bind(this, deferred);
michael@0: 
michael@0:     this.execute("ROLLBACK TRANSACTION").then(onRollback, onRollback);
michael@0:     this._inProgressTransaction.reject(new Error("Connection being closed."));
michael@0:     this._inProgressTransaction = null;
michael@0: 
michael@0:     return deferred.promise;
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Clones this connection to a new Sqlite one.
michael@0:    *
michael@0:    * The following parameters can control the cloned connection:
michael@0:    *
michael@0:    * @param readOnly
michael@0:    *        (boolean) - If true the clone will be read-only.  If the original
michael@0:    *        connection is already read-only, the clone will be, regardless of
michael@0:    *        this option.  If the original connection is using the shared cache,
michael@0:    *        this parameter will be ignored and the clone will be as privileged as
michael@0:    *        the original connection.
michael@0:    *
michael@0:    * @return Promise<OpenedConnection>
michael@0:    */
michael@0:   clone: function (readOnly=false) {
michael@0:     this._ensureOpen();
michael@0: 
michael@0:     this._log.debug("Request to clone connection.");
michael@0: 
michael@0:     let options = {
michael@0:       connection: this._connection,
michael@0:       readOnly: readOnly,
michael@0:     };
michael@0:     if (this._idleShrinkMS)
michael@0:       options.shrinkMemoryOnConnectionIdleMS = this._idleShrinkMS;
michael@0: 
michael@0:     return cloneStorageConnection(options);
michael@0:   },
michael@0: 
michael@0:   _finalize: function (deferred) {
michael@0:     this._log.debug("Finalizing connection.");
michael@0:     // Cancel any pending statements.
michael@0:     for (let [k, statement] of this._pendingStatements) {
michael@0:       statement.cancel();
michael@0:     }
michael@0:     this._pendingStatements.clear();
michael@0: 
michael@0:     // We no longer need to track these.
michael@0:     this._statementCounter = 0;
michael@0: 
michael@0:     // Next we finalize all active statements.
michael@0:     for (let [k, statement] of this._anonymousStatements) {
michael@0:       statement.finalize();
michael@0:     }
michael@0:     this._anonymousStatements.clear();
michael@0: 
michael@0:     for (let [k, statement] of this._cachedStatements) {
michael@0:       statement.finalize();
michael@0:     }
michael@0:     this._cachedStatements.clear();
michael@0: 
michael@0:     // This guards against operations performed between the call to this
michael@0:     // function and asyncClose() finishing. See also bug 726990.
michael@0:     this._open = false;
michael@0: 
michael@0:     this._log.debug("Calling asyncClose().");
michael@0:     this._connection.asyncClose({
michael@0:       complete: function () {
michael@0:         this._log.info("Closed");
michael@0:         this._connection = null;
michael@0:         deferred.resolve();
michael@0:       }.bind(this),
michael@0:     });
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Execute a SQL statement and cache the underlying statement object.
michael@0:    *
michael@0:    * This function executes a SQL statement and also caches the underlying
michael@0:    * derived statement object so subsequent executions are faster and use
michael@0:    * less resources.
michael@0:    *
michael@0:    * This function optionally binds parameters to the statement as well as
michael@0:    * optionally invokes a callback for every row retrieved.
michael@0:    *
michael@0:    * By default, no parameters are bound and no callback will be invoked for
michael@0:    * every row.
michael@0:    *
michael@0:    * Bound parameters can be defined as an Array of positional arguments or
michael@0:    * an object mapping named parameters to their values. If there are no bound
michael@0:    * parameters, the caller can pass nothing or null for this argument.
michael@0:    *
michael@0:    * Callers are encouraged to pass objects rather than Arrays for bound
michael@0:    * parameters because they prevent foot guns. With positional arguments, it
michael@0:    * is simple to modify the parameter count or positions without fixing all
michael@0:    * users of the statement. Objects/named parameters are a little safer
michael@0:    * because changes in order alone won't result in bad things happening.
michael@0:    *
michael@0:    * When `onRow` is not specified, all returned rows are buffered before the
michael@0:    * returned promise is resolved. For INSERT or UPDATE statements, this has
michael@0:    * no effect because no rows are returned from these. However, it has
michael@0:    * implications for SELECT statements.
michael@0:    *
michael@0:    * If your SELECT statement could return many rows or rows with large amounts
michael@0:    * of data, for performance reasons it is recommended to pass an `onRow`
michael@0:    * handler. Otherwise, the buffering may consume unacceptable amounts of
michael@0:    * resources.
michael@0:    *
michael@0:    * If a `StopIteration` is thrown during execution of an `onRow` handler,
michael@0:    * the execution of the statement is immediately cancelled. Subsequent
michael@0:    * rows will not be processed and no more `onRow` invocations will be made.
michael@0:    * The promise is resolved immediately.
michael@0:    *
michael@0:    * If a non-`StopIteration` exception is thrown by the `onRow` handler, the
michael@0:    * exception is logged and processing of subsequent rows occurs as if nothing
michael@0:    * happened. The promise is still resolved (not rejected).
michael@0:    *
michael@0:    * The return value is a promise that will be resolved when the statement
michael@0:    * has completed fully.
michael@0:    *
michael@0:    * The promise will be rejected with an `Error` instance if the statement
michael@0:    * did not finish execution fully. The `Error` may have an `errors` property.
michael@0:    * If defined, it will be an Array of objects describing individual errors.
michael@0:    * Each object has the properties `result` and `message`. `result` is a
michael@0:    * numeric error code and `message` is a string description of the problem.
michael@0:    *
michael@0:    * @param name
michael@0:    *        (string) The name of the registered statement to execute.
michael@0:    * @param params optional
michael@0:    *        (Array or object) Parameters to bind.
michael@0:    * @param onRow optional
michael@0:    *        (function) Callback to receive each row from result.
michael@0:    */
michael@0:   executeCached: function (sql, params=null, onRow=null) {
michael@0:     this._ensureOpen();
michael@0: 
michael@0:     if (!sql) {
michael@0:       throw new Error("sql argument is empty.");
michael@0:     }
michael@0: 
michael@0:     let statement = this._cachedStatements.get(sql);
michael@0:     if (!statement) {
michael@0:       statement = this._connection.createAsyncStatement(sql);
michael@0:       this._cachedStatements.set(sql, statement);
michael@0:     }
michael@0: 
michael@0:     this._clearIdleShrinkTimer();
michael@0: 
michael@0:     let deferred = Promise.defer();
michael@0: 
michael@0:     try {
michael@0:       this._executeStatement(sql, statement, params, onRow).then(
michael@0:         function onResult(result) {
michael@0:           this._startIdleShrinkTimer();
michael@0:           deferred.resolve(result);
michael@0:         }.bind(this),
michael@0:         function onError(error) {
michael@0:           this._startIdleShrinkTimer();
michael@0:           deferred.reject(error);
michael@0:         }.bind(this)
michael@0:       );
michael@0:     } catch (ex) {
michael@0:       this._startIdleShrinkTimer();
michael@0:       throw ex;
michael@0:     }
michael@0: 
michael@0:     return deferred.promise;
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Execute a one-shot SQL statement.
michael@0:    *
michael@0:    * If you find yourself feeding the same SQL string in this function, you
michael@0:    * should *not* use this function and instead use `executeCached`.
michael@0:    *
michael@0:    * See `executeCached` for the meaning of the arguments and extended usage info.
michael@0:    *
michael@0:    * @param sql
michael@0:    *        (string) SQL to execute.
michael@0:    * @param params optional
michael@0:    *        (Array or Object) Parameters to bind to the statement.
michael@0:    * @param onRow optional
michael@0:    *        (function) Callback to receive result of a single row.
michael@0:    */
michael@0:   execute: function (sql, params=null, onRow=null) {
michael@0:     if (typeof(sql) != "string") {
michael@0:       throw new Error("Must define SQL to execute as a string: " + sql);
michael@0:     }
michael@0: 
michael@0:     this._ensureOpen();
michael@0: 
michael@0:     let statement = this._connection.createAsyncStatement(sql);
michael@0:     let index = this._anonymousCounter++;
michael@0: 
michael@0:     this._anonymousStatements.set(index, statement);
michael@0:     this._clearIdleShrinkTimer();
michael@0: 
michael@0:     let onFinished = function () {
michael@0:       this._anonymousStatements.delete(index);
michael@0:       statement.finalize();
michael@0:       this._startIdleShrinkTimer();
michael@0:     }.bind(this);
michael@0: 
michael@0:     let deferred = Promise.defer();
michael@0: 
michael@0:     try {
michael@0:       this._executeStatement(sql, statement, params, onRow).then(
michael@0:         function onResult(rows) {
michael@0:           onFinished();
michael@0:           deferred.resolve(rows);
michael@0:         }.bind(this),
michael@0: 
michael@0:         function onError(error) {
michael@0:           onFinished();
michael@0:           deferred.reject(error);
michael@0:         }.bind(this)
michael@0:       );
michael@0:     } catch (ex) {
michael@0:       onFinished();
michael@0:       throw ex;
michael@0:     }
michael@0: 
michael@0:     return deferred.promise;
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Whether a transaction is currently in progress.
michael@0:    */
michael@0:   get transactionInProgress() {
michael@0:     return this._open && !!this._inProgressTransaction;
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Perform a transaction.
michael@0:    *
michael@0:    * A transaction is specified by a user-supplied function that is a
michael@0:    * generator function which can be used by Task.jsm's Task.spawn(). The
michael@0:    * function receives this connection instance as its argument.
michael@0:    *
michael@0:    * The supplied function is expected to yield promises. These are often
michael@0:    * promises created by calling `execute` and `executeCached`. If the
michael@0:    * generator is exhausted without any errors being thrown, the
michael@0:    * transaction is committed. If an error occurs, the transaction is
michael@0:    * rolled back.
michael@0:    *
michael@0:    * The returned value from this function is a promise that will be resolved
michael@0:    * once the transaction has been committed or rolled back. The promise will
michael@0:    * be resolved to whatever value the supplied function resolves to. If
michael@0:    * the transaction is rolled back, the promise is rejected.
michael@0:    *
michael@0:    * @param func
michael@0:    *        (function) What to perform as part of the transaction.
michael@0:    * @param type optional
michael@0:    *        One of the TRANSACTION_* constants attached to this type.
michael@0:    */
michael@0:   executeTransaction: function (func, type=this.TRANSACTION_DEFERRED) {
michael@0:     if (this.TRANSACTION_TYPES.indexOf(type) == -1) {
michael@0:       throw new Error("Unknown transaction type: " + type);
michael@0:     }
michael@0: 
michael@0:     this._ensureOpen();
michael@0: 
michael@0:     if (this._inProgressTransaction) {
michael@0:       throw new Error("A transaction is already active. Only one transaction " +
michael@0:                       "can be active at a time.");
michael@0:     }
michael@0: 
michael@0:     this._log.debug("Beginning transaction");
michael@0:     let deferred = Promise.defer();
michael@0:     this._inProgressTransaction = deferred;
michael@0:     Task.spawn(function doTransaction() {
michael@0:       // It's tempting to not yield here and rely on the implicit serial
michael@0:       // execution of issued statements. However, the yield serves an important
michael@0:       // purpose: catching errors in statement execution.
michael@0:       yield this.execute("BEGIN " + type + " TRANSACTION");
michael@0: 
michael@0:       let result;
michael@0:       try {
michael@0:         result = yield Task.spawn(func(this));
michael@0:       } catch (ex) {
michael@0:         // It's possible that a request to close the connection caused the
michael@0:         // error.
michael@0:         // Assertion: close() will unset this._inProgressTransaction when
michael@0:         // called.
michael@0:         if (!this._inProgressTransaction) {
michael@0:           this._log.warn("Connection was closed while performing transaction. " +
michael@0:                          "Received error should be due to closed connection: " +
michael@0:                          CommonUtils.exceptionStr(ex));
michael@0:           throw ex;
michael@0:         }
michael@0: 
michael@0:         this._log.warn("Error during transaction. Rolling back: " +
michael@0:                        CommonUtils.exceptionStr(ex));
michael@0:         try {
michael@0:           yield this.execute("ROLLBACK TRANSACTION");
michael@0:         } catch (inner) {
michael@0:           this._log.warn("Could not roll back transaction. This is weird: " +
michael@0:                          CommonUtils.exceptionStr(inner));
michael@0:         }
michael@0: 
michael@0:         throw ex;
michael@0:       }
michael@0: 
michael@0:       // See comment above about connection being closed during transaction.
michael@0:       if (!this._inProgressTransaction) {
michael@0:         this._log.warn("Connection was closed while performing transaction. " +
michael@0:                        "Unable to commit.");
michael@0:         throw new Error("Connection closed before transaction committed.");
michael@0:       }
michael@0: 
michael@0:       try {
michael@0:         yield this.execute("COMMIT TRANSACTION");
michael@0:       } catch (ex) {
michael@0:         this._log.warn("Error committing transaction: " +
michael@0:                        CommonUtils.exceptionStr(ex));
michael@0:         throw ex;
michael@0:       }
michael@0: 
michael@0:       throw new Task.Result(result);
michael@0:     }.bind(this)).then(
michael@0:       function onSuccess(result) {
michael@0:         this._inProgressTransaction = null;
michael@0:         deferred.resolve(result);
michael@0:       }.bind(this),
michael@0:       function onError(error) {
michael@0:         this._inProgressTransaction = null;
michael@0:         deferred.reject(error);
michael@0:       }.bind(this)
michael@0:     );
michael@0: 
michael@0:     return deferred.promise;
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Whether a table exists in the database (both persistent and temporary tables).
michael@0:    *
michael@0:    * @param name
michael@0:    *        (string) Name of the table.
michael@0:    *
michael@0:    * @return Promise<bool>
michael@0:    */
michael@0:   tableExists: function (name) {
michael@0:     return this.execute(
michael@0:       "SELECT name FROM (SELECT * FROM sqlite_master UNION ALL " +
michael@0:                         "SELECT * FROM sqlite_temp_master) " +
michael@0:       "WHERE type = 'table' AND name=?",
michael@0:       [name])
michael@0:       .then(function onResult(rows) {
michael@0:         return Promise.resolve(rows.length > 0);
michael@0:       }
michael@0:     );
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Whether a named index exists (both persistent and temporary tables).
michael@0:    *
michael@0:    * @param name
michael@0:    *        (string) Name of the index.
michael@0:    *
michael@0:    * @return Promise<bool>
michael@0:    */
michael@0:   indexExists: function (name) {
michael@0:     return this.execute(
michael@0:       "SELECT name FROM (SELECT * FROM sqlite_master UNION ALL " +
michael@0:                         "SELECT * FROM sqlite_temp_master) " +
michael@0:       "WHERE type = 'index' AND name=?",
michael@0:       [name])
michael@0:       .then(function onResult(rows) {
michael@0:         return Promise.resolve(rows.length > 0);
michael@0:       }
michael@0:     );
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Free up as much memory from the underlying database connection as possible.
michael@0:    *
michael@0:    * @return Promise<>
michael@0:    */
michael@0:   shrinkMemory: function () {
michael@0:     this._log.info("Shrinking memory usage.");
michael@0: 
michael@0:     let onShrunk = this._clearIdleShrinkTimer.bind(this);
michael@0: 
michael@0:     return this.execute("PRAGMA shrink_memory").then(onShrunk, onShrunk);
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Discard all cached statements.
michael@0:    *
michael@0:    * Note that this relies on us being non-interruptible between
michael@0:    * the insertion or retrieval of a statement in the cache and its
michael@0:    * execution: we finalize all statements, which is only safe if
michael@0:    * they will not be executed again.
michael@0:    *
michael@0:    * @return (integer) the number of statements discarded.
michael@0:    */
michael@0:   discardCachedStatements: function () {
michael@0:     let count = 0;
michael@0:     for (let [k, statement] of this._cachedStatements) {
michael@0:       ++count;
michael@0:       statement.finalize();
michael@0:     }
michael@0:     this._cachedStatements.clear();
michael@0:     this._log.debug("Discarded " + count + " cached statements.");
michael@0:     return count;
michael@0:   },
michael@0: 
michael@0:   /**
michael@0:    * Helper method to bind parameters of various kinds through
michael@0:    * reflection.
michael@0:    */
michael@0:   _bindParameters: function (statement, params) {
michael@0:     if (!params) {
michael@0:       return;
michael@0:     }
michael@0: 
michael@0:     if (Array.isArray(params)) {
michael@0:       // It's an array of separate params.
michael@0:       if (params.length && (typeof(params[0]) == "object")) {
michael@0:         let paramsArray = statement.newBindingParamsArray();
michael@0:         for (let p of params) {
michael@0:           let bindings = paramsArray.newBindingParams();
michael@0:           for (let [key, value] of Iterator(p)) {
michael@0:             bindings.bindByName(key, value);
michael@0:           }
michael@0:           paramsArray.addParams(bindings);
michael@0:         }
michael@0: 
michael@0:         statement.bindParameters(paramsArray);
michael@0:         return;
michael@0:       }
michael@0: 
michael@0:       // Indexed params.
michael@0:       for (let i = 0; i < params.length; i++) {
michael@0:         statement.bindByIndex(i, params[i]);
michael@0:       }
michael@0:       return;
michael@0:     }
michael@0: 
michael@0:     // Named params.
michael@0:     if (params && typeof(params) == "object") {
michael@0:       for (let k in params) {
michael@0:         statement.bindByName(k, params[k]);
michael@0:       }
michael@0:       return;
michael@0:     }
michael@0: 
michael@0:     throw new Error("Invalid type for bound parameters. Expected Array or " +
michael@0:                     "object. Got: " + params);
michael@0:   },
michael@0: 
michael@0:   _executeStatement: function (sql, statement, params, onRow) {
michael@0:     if (statement.state != statement.MOZ_STORAGE_STATEMENT_READY) {
michael@0:       throw new Error("Statement is not ready for execution.");
michael@0:     }
michael@0: 
michael@0:     if (onRow && typeof(onRow) != "function") {
michael@0:       throw new Error("onRow must be a function. Got: " + onRow);
michael@0:     }
michael@0: 
michael@0:     this._bindParameters(statement, params);
michael@0: 
michael@0:     let index = this._statementCounter++;
michael@0: 
michael@0:     let deferred = Promise.defer();
michael@0:     let userCancelled = false;
michael@0:     let errors = [];
michael@0:     let rows = [];
michael@0: 
michael@0:     // Don't incur overhead for serializing params unless the messages go
michael@0:     // somewhere.
michael@0:     if (this._log.level <= Log.Level.Trace) {
michael@0:       let msg = "Stmt #" + index + " " + sql;
michael@0: 
michael@0:       if (params) {
michael@0:         msg += " - " + JSON.stringify(params);
michael@0:       }
michael@0:       this._log.trace(msg);
michael@0:     } else {
michael@0:       this._log.debug("Stmt #" + index + " starting");
michael@0:     }
michael@0: 
michael@0:     let self = this;
michael@0:     let pending = statement.executeAsync({
michael@0:       handleResult: function (resultSet) {
michael@0:         // .cancel() may not be immediate and handleResult() could be called
michael@0:         // after a .cancel().
michael@0:         for (let row = resultSet.getNextRow(); row && !userCancelled; row = resultSet.getNextRow()) {
michael@0:           if (!onRow) {
michael@0:             rows.push(row);
michael@0:             continue;
michael@0:           }
michael@0: 
michael@0:           try {
michael@0:             onRow(row);
michael@0:           } catch (e if e instanceof StopIteration) {
michael@0:             userCancelled = true;
michael@0:             pending.cancel();
michael@0:             break;
michael@0:           } catch (ex) {
michael@0:             self._log.warn("Exception when calling onRow callback: " +
michael@0:                            CommonUtils.exceptionStr(ex));
michael@0:           }
michael@0:         }
michael@0:       },
michael@0: 
michael@0:       handleError: function (error) {
michael@0:         self._log.info("Error when executing SQL (" + error.result + "): " +
michael@0:                        error.message);
michael@0:         errors.push(error);
michael@0:       },
michael@0: 
michael@0:       handleCompletion: function (reason) {
michael@0:         self._log.debug("Stmt #" + index + " finished.");
michael@0:         self._pendingStatements.delete(index);
michael@0: 
michael@0:         switch (reason) {
michael@0:           case Ci.mozIStorageStatementCallback.REASON_FINISHED:
michael@0:             // If there is an onRow handler, we always resolve to null.
michael@0:             let result = onRow ? null : rows;
michael@0:             deferred.resolve(result);
michael@0:             break;
michael@0: 
michael@0:           case Ci.mozIStorageStatementCallback.REASON_CANCELLED:
michael@0:             // It is not an error if the user explicitly requested cancel via
michael@0:             // the onRow handler.
michael@0:             if (userCancelled) {
michael@0:               let result = onRow ? null : rows;
michael@0:               deferred.resolve(result);
michael@0:             } else {
michael@0:               deferred.reject(new Error("Statement was cancelled."));
michael@0:             }
michael@0: 
michael@0:             break;
michael@0: 
michael@0:           case Ci.mozIStorageStatementCallback.REASON_ERROR:
michael@0:             let error = new Error("Error(s) encountered during statement execution.");
michael@0:             error.errors = errors;
michael@0:             deferred.reject(error);
michael@0:             break;
michael@0: 
michael@0:           default:
michael@0:             deferred.reject(new Error("Unknown completion reason code: " +
michael@0:                                       reason));
michael@0:             break;
michael@0:         }
michael@0:       },
michael@0:     });
michael@0: 
michael@0:     this._pendingStatements.set(index, pending);
michael@0:     return deferred.promise;
michael@0:   },
michael@0: 
michael@0:   _ensureOpen: function () {
michael@0:     if (!this._open) {
michael@0:       throw new Error("Connection is not open.");
michael@0:     }
michael@0:   },
michael@0: 
michael@0:   _clearIdleShrinkTimer: function () {
michael@0:     if (!this._idleShrinkTimer) {
michael@0:       return;
michael@0:     }
michael@0: 
michael@0:     this._idleShrinkTimer.cancel();
michael@0:   },
michael@0: 
michael@0:   _startIdleShrinkTimer: function () {
michael@0:     if (!this._idleShrinkTimer) {
michael@0:       return;
michael@0:     }
michael@0: 
michael@0:     this._idleShrinkTimer.initWithCallback(this.shrinkMemory.bind(this),
michael@0:                                            this._idleShrinkMS,
michael@0:                                            this._idleShrinkTimer.TYPE_ONE_SHOT);
michael@0:   },
michael@0: });
michael@0: 
michael@0: this.Sqlite = {
michael@0:   openConnection: openConnection,
michael@0:   cloneStorageConnection: cloneStorageConnection
michael@0: };