services/common/kinto-storage-adapter.sys.mjs

   1 /*
   2  * Licensed under the Apache License, Version 2.0 (the "License");
   3  * you may not use this file except in compliance with the License.
   4  * You may obtain a copy of the License at
   5  *
   6  *     http://www.apache.org/licenses/LICENSE-2.0
   7  *
   8  * Unless required by applicable law or agreed to in writing, software
   9  * distributed under the License is distributed on an "AS IS" BASIS,
  10  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11  * See the License for the specific language governing permissions and
  12  * limitations under the License.
  13  */
  14 import { Sqlite } from "resource://gre/modules/Sqlite.sys.mjs";
  15
  16 import { Kinto } from "resource://services-common/kinto-offline-client.sys.mjs";
  17
  18 /**
  19  * Filter and sort list against provided filters and order.
  20  *
  21  * @param  {Object} filters  The filters to apply.
  22  * @param  {String} order    The order to apply.
  23  * @param  {Array}  list     The list to reduce.
  24  * @return {Array}
  25  */
  26 function reduceRecords(filters, order, list) {
  27   const filtered = filters ? filterObjects(filters, list) : list;
  28   return order ? sortObjects(order, filtered) : filtered;
  29 }
  30
  31 /**
  32  * Checks if a value is undefined.
  33  *
  34  * This is a copy of `_isUndefined` from kinto.js/src/utils.js.
  35  * @param  {Any}  value
  36  * @return {Boolean}
  37  */
  38 function _isUndefined(value) {
  39   return typeof value === "undefined";
  40 }
  41
  42 /**
  43  * Sorts records in a list according to a given ordering.
  44  *
  45  * This is a copy of `sortObjects` from kinto.js/src/utils.js.
  46  *
  47  * @param  {String} order The ordering, eg. `-last_modified`.
  48  * @param  {Array}  list  The collection to order.
  49  * @return {Array}
  50  */
  51 function sortObjects(order, list) {
  52   const hasDash = order[0] === "-";
  53   const field = hasDash ? order.slice(1) : order;
  54   const direction = hasDash ? -1 : 1;
  55   return list.slice().sort((a, b) => {
  56     if (a[field] && _isUndefined(b[field])) {
  57       return direction;
  58     }
  59     if (b[field] && _isUndefined(a[field])) {
  60       return -direction;
  61     }
  62     if (_isUndefined(a[field]) && _isUndefined(b[field])) {
  63       return 0;
  64     }
  65     return a[field] > b[field] ? direction : -direction;
  66   });
  67 }
  68
  69 /**
  70  * Test if a single object matches all given filters.
  71  *
  72  * This is a copy of `filterObject` from kinto.js/src/utils.js.
  73  *
  74  * @param  {Object} filters  The filters object.
  75  * @param  {Object} entry    The object to filter.
  76  * @return {Function}
  77  */
  78 function filterObject(filters, entry) {
  79   return Object.keys(filters).every(filter => {
  80     const value = filters[filter];
  81     if (Array.isArray(value)) {
  82       return value.some(candidate => candidate === entry[filter]);
  83     }
  84     return entry[filter] === value;
  85   });
  86 }
  87
  88 /**
  89  * Filters records in a list matching all given filters.
  90  *
  91  * This is a copy of `filterObjects` from kinto.js/src/utils.js.
  92  *
  93  * @param  {Object} filters  The filters object.
  94  * @param  {Array}  list     The collection to filter.
  95  * @return {Array}
  96  */
  97 function filterObjects(filters, list) {
  98   return list.filter(entry => {
  99     return filterObject(filters, entry);
 100   });
 101 }
 102
 103 const statements = {
 104   createCollectionData: `
 105     CREATE TABLE collection_data (
 106       collection_name TEXT,
 107       record_id TEXT,
 108       record TEXT
 109     );`,
 110
 111   createCollectionMetadata: `
 112     CREATE TABLE collection_metadata (
 113       collection_name TEXT PRIMARY KEY,
 114       last_modified INTEGER,
 115       metadata TEXT
 116     ) WITHOUT ROWID;`,
 117
 118   createCollectionDataRecordIdIndex: `
 119     CREATE UNIQUE INDEX unique_collection_record
 120       ON collection_data(collection_name, record_id);`,
 121
 122   clearData: `
 123     DELETE FROM collection_data
 124       WHERE collection_name = :collection_name;`,
 125
 126   createData: `
 127     INSERT INTO collection_data (collection_name, record_id, record)
 128       VALUES (:collection_name, :record_id, :record);`,
 129
 130   updateData: `
 131     INSERT OR REPLACE INTO collection_data (collection_name, record_id, record)
 132       VALUES (:collection_name, :record_id, :record);`,
 133
 134   deleteData: `
 135     DELETE FROM collection_data
 136       WHERE collection_name = :collection_name
 137       AND record_id = :record_id;`,
 138
 139   saveLastModified: `
 140     INSERT INTO collection_metadata(collection_name, last_modified)
 141       VALUES(:collection_name, :last_modified)
 142         ON CONFLICT(collection_name) DO UPDATE SET last_modified = :last_modified`,
 143
 144   getLastModified: `
 145     SELECT last_modified
 146       FROM collection_metadata
 147         WHERE collection_name = :collection_name;`,
 148
 149   saveMetadata: `
 150     INSERT INTO collection_metadata(collection_name, metadata)
 151       VALUES(:collection_name, :metadata)
 152         ON CONFLICT(collection_name) DO UPDATE SET metadata = :metadata`,
 153
 154   getMetadata: `
 155     SELECT metadata
 156       FROM collection_metadata
 157         WHERE collection_name = :collection_name;`,
 158
 159   getRecord: `
 160     SELECT record
 161       FROM collection_data
 162         WHERE collection_name = :collection_name
 163         AND record_id = :record_id;`,
 164
 165   listRecords: `
 166     SELECT record
 167       FROM collection_data
 168         WHERE collection_name = :collection_name;`,
 169
 170   // N.B. we have to have a dynamic number of placeholders, which you
 171   // can't do without building your own statement. See `execute` for details
 172   listRecordsById: `
 173     SELECT record_id, record
 174       FROM collection_data
 175         WHERE collection_name = ?
 176           AND record_id IN `,
 177
 178   importData: `
 179     REPLACE INTO collection_data (collection_name, record_id, record)
 180       VALUES (:collection_name, :record_id, :record);`,
 181
 182   scanAllRecords: `SELECT * FROM collection_data;`,
 183
 184   clearCollectionMetadata: `DELETE FROM collection_metadata;`,
 185
 186   calculateStorage: `
 187     SELECT collection_name, SUM(LENGTH(record)) as size, COUNT(record) as num_records
 188       FROM collection_data
 189         GROUP BY collection_name;`,
 190
 191   addMetadataColumn: `
 192     ALTER TABLE collection_metadata
 193       ADD COLUMN metadata TEXT;`,
 194 };
 195
 196 const createStatements = [
 197   "createCollectionData",
 198   "createCollectionMetadata",
 199   "createCollectionDataRecordIdIndex",
 200 ];
 201
 202 const currentSchemaVersion = 2;
 203
 204 /**
 205  * Firefox adapter.
 206  *
 207  * Uses Sqlite as a backing store.
 208  *
 209  * Options:
 210  *  - sqliteHandle: a handle to the Sqlite database this adapter will
 211  *    use as its backing store. To open such a handle, use the
 212  *    static openConnection() method.
 213  */
 214 export class FirefoxAdapter extends Kinto.adapters.BaseAdapter {
 215   constructor(collection, options = {}) {
 216     super();
 217     const { sqliteHandle = null } = options;
 218     this.collection = collection;
 219     this._connection = sqliteHandle;
 220     this._options = options;
 221   }
 222
 223   /**
 224    * Initialize a Sqlite connection to be suitable for use with Kinto.
 225    *
 226    * This will be called automatically by open().
 227    */
 228   static async _init(connection) {
 229     await connection.executeTransaction(async function doSetup() {
 230       const schema = await connection.getSchemaVersion();
 231
 232       if (schema == 0) {
 233         for (let statementName of createStatements) {
 234           await connection.execute(statements[statementName]);
 235         }
 236         await connection.setSchemaVersion(currentSchemaVersion);
 237       } else if (schema == 1) {
 238         await connection.execute(statements.addMetadataColumn);
 239         await connection.setSchemaVersion(currentSchemaVersion);
 240       } else if (schema != 2) {
 241         throw new Error("Unknown database schema: " + schema);
 242       }
 243     });
 244     return connection;
 245   }
 246
 247   _executeStatement(statement, params) {
 248     return this._connection.executeCached(statement, params);
 249   }
 250
 251   /**
 252    * Open and initialize a Sqlite connection to a database that Kinto
 253    * can use. When you are done with this connection, close it by
 254    * calling close().
 255    *
 256    * Options:
 257    *   - path: The path for the Sqlite database
 258    *
 259    * @returns SqliteConnection
 260    */
 261   static async openConnection(options) {
 262     const opts = Object.assign({}, { sharedMemoryCache: false }, options);
 263     const conn = await Sqlite.openConnection(opts).then(this._init);
 264     try {
 265       Sqlite.shutdown.addBlocker(
 266         "Kinto storage adapter connection closing",
 267         () => conn.close()
 268       );
 269     } catch (e) {
 270       // It's too late to block shutdown, just close the connection.
 271       await conn.close();
 272       throw e;
 273     }
 274     return conn;
 275   }
 276
 277   clear() {
 278     const params = { collection_name: this.collection };
 279     return this._executeStatement(statements.clearData, params);
 280   }
 281
 282   execute(callback, options = { preload: [] }) {
 283     let result;
 284     const conn = this._connection;
 285     const collection = this.collection;
 286
 287     return conn
 288       .executeTransaction(async function doExecuteTransaction() {
 289         // Preload specified records from DB, within transaction.
 290
 291         // if options.preload has more elements than the sqlite variable
 292         // limit, split it up.
 293         const limit = 100;
 294         let preloaded = {};
 295         let preload;
 296         let more = options.preload;
 297
 298         while (more.length) {
 299           preload = more.slice(0, limit);
 300           more = more.slice(limit, more.length);
 301
 302           const parameters = [collection, ...preload];
 303           const placeholders = preload.map(_ => "?");
 304           const stmt =
 305             statements.listRecordsById + "(" + placeholders.join(",") + ");";
 306           const rows = await conn.execute(stmt, parameters);
 307
 308           rows.reduce((acc, row) => {
 309             const record = JSON.parse(row.getResultByName("record"));
 310             acc[row.getResultByName("record_id")] = record;
 311             return acc;
 312           }, preloaded);
 313         }
 314         const proxy = transactionProxy(collection, preloaded);
 315         result = callback(proxy);
 316
 317         for (let { statement, params } of proxy.operations) {
 318           await conn.executeCached(statement, params);
 319         }
 320       }, conn.TRANSACTION_EXCLUSIVE)
 321       .then(_ => result);
 322   }
 323
 324   get(id) {
 325     const params = {
 326       collection_name: this.collection,
 327       record_id: id,
 328     };
 329     return this._executeStatement(statements.getRecord, params).then(result => {
 330       if (!result.length) {
 331         return null;
 332       }
 333       return JSON.parse(result[0].getResultByName("record"));
 334     });
 335   }
 336
 337   list(params = { filters: {}, order: "" }) {
 338     const parameters = {
 339       collection_name: this.collection,
 340     };
 341     return this._executeStatement(statements.listRecords, parameters)
 342       .then(result => {
 343         const records = [];
 344         for (let k = 0; k < result.length; k++) {
 345           const row = result[k];
 346           records.push(JSON.parse(row.getResultByName("record")));
 347         }
 348         return records;
 349       })
 350       .then(results => {
 351         // The resulting list of records is filtered and sorted.
 352         // XXX: with some efforts, this could be implemented using SQL.
 353         return reduceRecords(params.filters, params.order, results);
 354       });
 355   }
 356
 357   async loadDump(records) {
 358     return this.importBulk(records);
 359   }
 360
 361   /**
 362    * Load a list of records into the local database.
 363    *
 364    * Note: The adapter is not in charge of filtering the already imported
 365    * records. This is done in `Collection#loadDump()`, as a common behaviour
 366    * between every adapters.
 367    *
 368    * @param  {Array} records.
 369    * @return {Array} imported records.
 370    */
 371   async importBulk(records) {
 372     const connection = this._connection;
 373     const collection_name = this.collection;
 374     await connection.executeTransaction(async function doImport() {
 375       for (let record of records) {
 376         const params = {
 377           collection_name,
 378           record_id: record.id,
 379           record: JSON.stringify(record),
 380         };
 381         await connection.execute(statements.importData, params);
 382       }
 383       const lastModified = Math.max(
 384         ...records.map(record => record.last_modified)
 385       );
 386       const params = {
 387         collection_name,
 388       };
 389       const previousLastModified = await connection
 390         .execute(statements.getLastModified, params)
 391         .then(result => {
 392           return result.length
 393             ? result[0].getResultByName("last_modified")
 394             : -1;
 395         });
 396       if (lastModified > previousLastModified) {
 397         const params = {
 398           collection_name,
 399           last_modified: lastModified,
 400         };
 401         await connection.execute(statements.saveLastModified, params);
 402       }
 403     });
 404     return records;
 405   }
 406
 407   saveLastModified(lastModified) {
 408     const parsedLastModified = parseInt(lastModified, 10) || null;
 409     const params = {
 410       collection_name: this.collection,
 411       last_modified: parsedLastModified,
 412     };
 413     return this._executeStatement(statements.saveLastModified, params).then(
 414       () => parsedLastModified
 415     );
 416   }
 417
 418   getLastModified() {
 419     const params = {
 420       collection_name: this.collection,
 421     };
 422     return this._executeStatement(statements.getLastModified, params).then(
 423       result => {
 424         if (!result.length) {
 425           return 0;
 426         }
 427         return result[0].getResultByName("last_modified");
 428       }
 429     );
 430   }
 431
 432   async saveMetadata(metadata) {
 433     const params = {
 434       collection_name: this.collection,
 435       metadata: JSON.stringify(metadata),
 436     };
 437     await this._executeStatement(statements.saveMetadata, params);
 438     return metadata;
 439   }
 440
 441   async getMetadata() {
 442     const params = {
 443       collection_name: this.collection,
 444     };
 445     const result = await this._executeStatement(statements.getMetadata, params);
 446     if (!result.length) {
 447       return null;
 448     }
 449     return JSON.parse(result[0].getResultByName("metadata"));
 450   }
 451
 452   calculateStorage() {
 453     return this._executeStatement(statements.calculateStorage, {}).then(
 454       result => {
 455         return Array.from(result, row => ({
 456           collectionName: row.getResultByName("collection_name"),
 457           size: row.getResultByName("size"),
 458           numRecords: row.getResultByName("num_records"),
 459         }));
 460       }
 461     );
 462   }
 463
 464   /**
 465    * Reset the sync status of every record and collection we have
 466    * access to.
 467    */
 468   resetSyncStatus() {
 469     // We're going to use execute instead of executeCached, so build
 470     // in our own sanity check
 471     if (!this._connection) {
 472       throw new Error("The storage adapter is not open");
 473     }
 474
 475     return this._connection.executeTransaction(async function (conn) {
 476       const promises = [];
 477       await conn.execute(statements.scanAllRecords, null, function (row) {
 478         const record = JSON.parse(row.getResultByName("record"));
 479         const record_id = row.getResultByName("record_id");
 480         const collection_name = row.getResultByName("collection_name");
 481         if (record._status === "deleted") {
 482           // Garbage collect deleted records.
 483           promises.push(
 484             conn.execute(statements.deleteData, { collection_name, record_id })
 485           );
 486         } else {
 487           const newRecord = Object.assign({}, record, {
 488             _status: "created",
 489             last_modified: undefined,
 490           });
 491           promises.push(
 492             conn.execute(statements.updateData, {
 493               record: JSON.stringify(newRecord),
 494               record_id,
 495               collection_name,
 496             })
 497           );
 498         }
 499       });
 500       await Promise.all(promises);
 501       await conn.execute(statements.clearCollectionMetadata);
 502     });
 503   }
 504 }
 505
 506 function transactionProxy(collection, preloaded) {
 507   const _operations = [];
 508
 509   return {
 510     get operations() {
 511       return _operations;
 512     },
 513
 514     create(record) {
 515       _operations.push({
 516         statement: statements.createData,
 517         params: {
 518           collection_name: collection,
 519           record_id: record.id,
 520           record: JSON.stringify(record),
 521         },
 522       });
 523     },
 524
 525     update(record) {
 526       _operations.push({
 527         statement: statements.updateData,
 528         params: {
 529           collection_name: collection,
 530           record_id: record.id,
 531           record: JSON.stringify(record),
 532         },
 533       });
 534     },
 535
 536     delete(id) {
 537       _operations.push({
 538         statement: statements.deleteData,
 539         params: {
 540           collection_name: collection,
 541           record_id: id,
 542         },
 543       });
 544     },
 545
 546     get(id) {
 547       // Gecko JS engine outputs undesired warnings if id is not in preloaded.
 548       return id in preloaded ? preloaded[id] : undefined;
 549     },
 550   };
 551 }