kinto-storage-adapter.sys.mjs

/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
import { Sqlite } from "resource://gre/modules/Sqlite.sys.mjs";

import { Kinto } from "resource://services-common/kinto-offline-client.sys.mjs";

/**
 * Filter and sort list against provided filters and order.
 *
 * @param  {Object} filters  The filters to apply.
 * @param  {String} order    The order to apply.
 * @param  {Array}  list     The list to reduce.
 * @return {Array}
 */
function reduceRecords(filters, order, list) {
  const filtered = filters ? filterObjects(filters, list) : list;
  return order ? sortObjects(order, filtered) : filtered;
}

/**
 * Checks if a value is undefined.
 *
 * This is a copy of `_isUndefined` from kinto.js/src/utils.js.
 * @param  {Any}  value
 * @return {Boolean}
 */
function _isUndefined(value) {
  return typeof value === "undefined";
}

/**
 * Sorts records in a list according to a given ordering.
 *
 * This is a copy of `sortObjects` from kinto.js/src/utils.js.
 *
 * @param  {String} order The ordering, eg. `-last_modified`.
 * @param  {Array}  list  The collection to order.
 * @return {Array}
 */
function sortObjects(order, list) {
  const hasDash = order[0] === "-";
  const field = hasDash ? order.slice(1) : order;
  const direction = hasDash ? -1 : 1;
  return list.slice().sort((a, b) => {
    if (a[field] && _isUndefined(b[field])) {
      return direction;
    }
    if (b[field] && _isUndefined(a[field])) {
      return -direction;
    }
    if (_isUndefined(a[field]) && _isUndefined(b[field])) {
      return 0;
    }
    return a[field] > b[field] ? direction : -direction;
  });
}

/**
 * Test if a single object matches all given filters.
 *
 * This is a copy of `filterObject` from kinto.js/src/utils.js.
 *
 * @param  {Object} filters  The filters object.
 * @param  {Object} entry    The object to filter.
 * @return {Function}
 */
function filterObject(filters, entry) {
  return Object.keys(filters).every(filter => {
    const value = filters[filter];
    if (Array.isArray(value)) {
      return value.some(candidate => candidate === entry[filter]);
    }
    return entry[filter] === value;
  });
}

/**
 * Filters records in a list matching all given filters.
 *
 * This is a copy of `filterObjects` from kinto.js/src/utils.js.
 *
 * @param  {Object} filters  The filters object.
 * @param  {Array}  list     The collection to filter.
 * @return {Array}
 */
function filterObjects(filters, list) {
  return list.filter(entry => {
    return filterObject(filters, entry);
  });
}

const statements = {
  createCollectionData: `
    CREATE TABLE collection_data (
      collection_name TEXT,
      record_id TEXT,
      record TEXT
    );`,

  createCollectionMetadata: `
    CREATE TABLE collection_metadata (
      collection_name TEXT PRIMARY KEY,
      last_modified INTEGER,
      metadata TEXT
    ) WITHOUT ROWID;`,

  createCollectionDataRecordIdIndex: `
    CREATE UNIQUE INDEX unique_collection_record
      ON collection_data(collection_name, record_id);`,

  clearData: `
    DELETE FROM collection_data
      WHERE collection_name = :collection_name;`,

  createData: `
    INSERT INTO collection_data (collection_name, record_id, record)
      VALUES (:collection_name, :record_id, :record);`,

  updateData: `
    INSERT OR REPLACE INTO collection_data (collection_name, record_id, record)
      VALUES (:collection_name, :record_id, :record);`,

  deleteData: `
    DELETE FROM collection_data
      WHERE collection_name = :collection_name
      AND record_id = :record_id;`,

  saveLastModified: `
    INSERT INTO collection_metadata(collection_name, last_modified)
      VALUES(:collection_name, :last_modified)
        ON CONFLICT(collection_name) DO UPDATE SET last_modified = :last_modified`,

  getLastModified: `
    SELECT last_modified
      FROM collection_metadata
        WHERE collection_name = :collection_name;`,

  saveMetadata: `
    INSERT INTO collection_metadata(collection_name, metadata)
      VALUES(:collection_name, :metadata)
        ON CONFLICT(collection_name) DO UPDATE SET metadata = :metadata`,

  getMetadata: `
    SELECT metadata
      FROM collection_metadata
        WHERE collection_name = :collection_name;`,

  getRecord: `
    SELECT record
      FROM collection_data
        WHERE collection_name = :collection_name
        AND record_id = :record_id;`,

  listRecords: `
    SELECT record
      FROM collection_data
        WHERE collection_name = :collection_name;`,

  // N.B. we have to have a dynamic number of placeholders, which you
  // can't do without building your own statement. See `execute` for details
  listRecordsById: `
    SELECT record_id, record
      FROM collection_data
        WHERE collection_name = ?
          AND record_id IN `,

  importData: `
    REPLACE INTO collection_data (collection_name, record_id, record)
      VALUES (:collection_name, :record_id, :record);`,

  scanAllRecords: `SELECT * FROM collection_data;`,

  clearCollectionMetadata: `DELETE FROM collection_metadata;`,

  calculateStorage: `
    SELECT collection_name, SUM(LENGTH(record)) as size, COUNT(record) as num_records
      FROM collection_data
        GROUP BY collection_name;`,

  addMetadataColumn: `
    ALTER TABLE collection_metadata
      ADD COLUMN metadata TEXT;`,
};

const createStatements = [
  "createCollectionData",
  "createCollectionMetadata",
  "createCollectionDataRecordIdIndex",
];

const currentSchemaVersion = 2;

/**
 * Firefox adapter.
 *
 * Uses Sqlite as a backing store.
 *
 * Options:
 *  - sqliteHandle: a handle to the Sqlite database this adapter will
 *    use as its backing store. To open such a handle, use the
 *    static openConnection() method.
 */
export class FirefoxAdapter extends Kinto.adapters.BaseAdapter {
  constructor(collection, options = {}) {
    super();
    const { sqliteHandle = null } = options;
    this.collection = collection;
    this._connection = sqliteHandle;
    this._options = options;
  }

  /**
   * Initialize a Sqlite connection to be suitable for use with Kinto.
   *
   * This will be called automatically by open().
   */
  static async _init(connection) {
    await connection.executeTransaction(async function doSetup() {
      const schema = await connection.getSchemaVersion();

      if (schema == 0) {
        for (let statementName of createStatements) {
          await connection.execute(statements[statementName]);
        }
        await connection.setSchemaVersion(currentSchemaVersion);
      } else if (schema == 1) {
        await connection.execute(statements.addMetadataColumn);
        await connection.setSchemaVersion(currentSchemaVersion);
      } else if (schema != 2) {
        throw new Error("Unknown database schema: " + schema);
      }
    });
    return connection;
  }

  _executeStatement(statement, params) {
    return this._connection.executeCached(statement, params);
  }

  /**
   * Open and initialize a Sqlite connection to a database that Kinto
   * can use. When you are done with this connection, close it by
   * calling close().
   *
   * Options:
   *   - path: The path for the Sqlite database
   *
   * @returns SqliteConnection
   */
  static async openConnection(options) {
    const opts = Object.assign({}, { sharedMemoryCache: false }, options);
    const conn = await Sqlite.openConnection(opts).then(this._init);
    try {
      Sqlite.shutdown.addBlocker(
        "Kinto storage adapter connection closing",
        () => conn.close()
      );
    } catch (e) {
      // It's too late to block shutdown, just close the connection.
      await conn.close();
      throw e;
    }
    return conn;
  }

  clear() {
    const params = { collection_name: this.collection };
    return this._executeStatement(statements.clearData, params);
  }

  execute(callback, options = { preload: [] }) {
    let result;
    const conn = this._connection;
    const collection = this.collection;

    return conn
      .executeTransaction(async function doExecuteTransaction() {
        // Preload specified records from DB, within transaction.

        // if options.preload has more elements than the sqlite variable
        // limit, split it up.
        const limit = 100;
        let preloaded = {};
        let preload;
        let more = options.preload;

        while (more.length) {
          preload = more.slice(0, limit);
          more = more.slice(limit, more.length);

          const parameters = [collection, ...preload];
          const placeholders = preload.map(_ => "?");
          const stmt =
            statements.listRecordsById + "(" + placeholders.join(",") + ");";
          const rows = await conn.execute(stmt, parameters);

          rows.reduce((acc, row) => {
            const record = JSON.parse(row.getResultByName("record"));
            acc[row.getResultByName("record_id")] = record;
            return acc;
          }, preloaded);
        }
        const proxy = transactionProxy(collection, preloaded);
        result = callback(proxy);

        for (let { statement, params } of proxy.operations) {
          await conn.executeCached(statement, params);
        }
      }, conn.TRANSACTION_EXCLUSIVE)
      .then(_ => result);
  }

  get(id) {
    const params = {
      collection_name: this.collection,
      record_id: id,
    };
    return this._executeStatement(statements.getRecord, params).then(result => {
      if (!result.length) {
        return null;
      }
      return JSON.parse(result[0].getResultByName("record"));
    });
  }

  list(params = { filters: {}, order: "" }) {
    const parameters = {
      collection_name: this.collection,
    };
    return this._executeStatement(statements.listRecords, parameters)
      .then(result => {
        const records = [];
        for (let k = 0; k < result.length; k++) {
          const row = result[k];
          records.push(JSON.parse(row.getResultByName("record")));
        }
        return records;
      })
      .then(results => {
        // The resulting list of records is filtered and sorted.
        // XXX: with some efforts, this could be implemented using SQL.
        return reduceRecords(params.filters, params.order, results);
      });
  }

  async loadDump(records) {
    return this.importBulk(records);
  }

  /**
   * Load a list of records into the local database.
   *
   * Note: The adapter is not in charge of filtering the already imported
   * records. This is done in `Collection#loadDump()`, as a common behaviour
   * between every adapters.
   *
   * @param  {Array} records.
   * @return {Array} imported records.
   */
  async importBulk(records) {
    const connection = this._connection;
    const collection_name = this.collection;
    await connection.executeTransaction(async function doImport() {
      for (let record of records) {
        const params = {
          collection_name,
          record_id: record.id,
          record: JSON.stringify(record),
        };
        await connection.execute(statements.importData, params);
      }
      const lastModified = Math.max(
        ...records.map(record => record.last_modified)
      );
      const params = {
        collection_name,
      };
      const previousLastModified = await connection
        .execute(statements.getLastModified, params)
        .then(result => {
          return result.length
            ? result[0].getResultByName("last_modified")
            : -1;
        });
      if (lastModified > previousLastModified) {
        const params = {
          collection_name,
          last_modified: lastModified,
        };
        await connection.execute(statements.saveLastModified, params);
      }
    });
    return records;
  }

  saveLastModified(lastModified) {
    const parsedLastModified = parseInt(lastModified, 10) || null;
    const params = {
      collection_name: this.collection,
      last_modified: parsedLastModified,
    };
    return this._executeStatement(statements.saveLastModified, params).then(
      () => parsedLastModified
    );
  }

  getLastModified() {
    const params = {
      collection_name: this.collection,
    };
    return this._executeStatement(statements.getLastModified, params).then(
      result => {
        if (!result.length) {
          return 0;
        }
        return result[0].getResultByName("last_modified");
      }
    );
  }

  async saveMetadata(metadata) {
    const params = {
      collection_name: this.collection,
      metadata: JSON.stringify(metadata),
    };
    await this._executeStatement(statements.saveMetadata, params);
    return metadata;
  }

  async getMetadata() {
    const params = {
      collection_name: this.collection,
    };
    const result = await this._executeStatement(statements.getMetadata, params);
    if (!result.length) {
      return null;
    }
    return JSON.parse(result[0].getResultByName("metadata"));
  }

  calculateStorage() {
    return this._executeStatement(statements.calculateStorage, {}).then(
      result => {
        return Array.from(result, row => ({
          collectionName: row.getResultByName("collection_name"),
          size: row.getResultByName("size"),
          numRecords: row.getResultByName("num_records"),
        }));
      }
    );
  }

  /**
   * Reset the sync status of every record and collection we have
   * access to.
   */
  resetSyncStatus() {
    // We're going to use execute instead of executeCached, so build
    // in our own sanity check
    if (!this._connection) {
      throw new Error("The storage adapter is not open");
    }

    return this._connection.executeTransaction(async function (conn) {
      const promises = [];
      await conn.execute(statements.scanAllRecords, null, function (row) {
        const record = JSON.parse(row.getResultByName("record"));
        const record_id = row.getResultByName("record_id");
        const collection_name = row.getResultByName("collection_name");
        if (record._status === "deleted") {
          // Garbage collect deleted records.
          promises.push(
            conn.execute(statements.deleteData, { collection_name, record_id })
          );
        } else {
          const newRecord = Object.assign({}, record, {
            _status: "created",
            last_modified: undefined,
          });
          promises.push(
            conn.execute(statements.updateData, {
              record: JSON.stringify(newRecord),
              record_id,
              collection_name,
            })
          );
        }
      });
      await Promise.all(promises);
      await conn.execute(statements.clearCollectionMetadata);
    });
  }
}

function transactionProxy(collection, preloaded) {
  const _operations = [];

  return {
    get operations() {
      return _operations;
    },

    create(record) {
      _operations.push({
        statement: statements.createData,
        params: {
          collection_name: collection,
          record_id: record.id,
          record: JSON.stringify(record),
        },
      });
    },

    update(record) {
      _operations.push({
        statement: statements.updateData,
        params: {
          collection_name: collection,
          record_id: record.id,
          record: JSON.stringify(record),
        },
      });
    },

    delete(id) {
      _operations.push({
        statement: statements.deleteData,
        params: {
          collection_name: collection,
          record_id: id,
        },
      });
    },

    get(id) {
      // Gecko JS engine outputs undesired warnings if id is not in preloaded.
      return id in preloaded ? preloaded[id] : undefined;
    },
  };
}