SqliteConnectionPool.js 5.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187
  1. const sqlite = require('sqlite');
  2. const SQL = require('sql-template-strings');
  3. class SqliteConnectionPool {
  4. constructor() {
  5. this.closed = true;
  6. }
  7. async open(poolConfig, dbFileName) {
  8. const connCount = poolConfig.connCount || 1;
  9. const busyTimeout = poolConfig.busyTimeout || 60*1000;
  10. const cacheSize = poolConfig.cacheSize || 2000;
  11. this.dbFileName = dbFileName;
  12. this.connections = [];
  13. this.freed = new Set();
  14. this.waitingQueue = [];
  15. for (let i = 0; i < connCount; i++) {
  16. let client = await sqlite.open(dbFileName);
  17. client.configure('busyTimeout', busyTimeout); //ms
  18. await client.exec(`PRAGMA cache_size = ${cacheSize}`);
  19. client.ret = () => {
  20. this.freed.add(i);
  21. if (this.waitingQueue.length) {
  22. this.waitingQueue.shift().onFreed(i);
  23. }
  24. };
  25. this.freed.add(i);
  26. this.connections[i] = client;
  27. }
  28. this.closed = false;
  29. }
  30. get() {
  31. return new Promise((resolve) => {
  32. if (this.closed)
  33. throw new Error('Connection pool closed');
  34. const freeConnIndex = this.freed.values().next().value;
  35. if (freeConnIndex !== undefined) {
  36. this.freed.delete(freeConnIndex);
  37. resolve(this.connections[freeConnIndex]);
  38. return;
  39. }
  40. this.waitingQueue.push({
  41. onFreed: (connIndex) => {
  42. this.freed.delete(connIndex);
  43. resolve(this.connections[connIndex]);
  44. },
  45. });
  46. });
  47. }
  48. async run(query) {
  49. const dbh = await this.get();
  50. try {
  51. let result = await dbh.run(query);
  52. dbh.ret();
  53. return result;
  54. } catch (e) {
  55. dbh.ret();
  56. throw e;
  57. }
  58. }
  59. async all(query) {
  60. const dbh = await this.get();
  61. try {
  62. let result = await dbh.all(query);
  63. dbh.ret();
  64. return result;
  65. } catch (e) {
  66. dbh.ret();
  67. throw e;
  68. }
  69. }
  70. async exec(query) {
  71. const dbh = await this.get();
  72. try {
  73. let result = await dbh.exec(query);
  74. dbh.ret();
  75. return result;
  76. } catch (e) {
  77. dbh.ret();
  78. throw e;
  79. }
  80. }
  81. async close() {
  82. for (let i = 0; i < this.connections.length; i++) {
  83. await this.connections[i].close();
  84. }
  85. this.closed = true;
  86. }
  87. // Modified from node-sqlite/.../src/Database.js
  88. async migrate(migs, table, force) {
  89. const migrations = migs.sort((a, b) => Math.sign(a.id - b.id));
  90. if (!migrations.length) {
  91. throw new Error('No migration data');
  92. }
  93. migrations.map(migration => {
  94. const data = migration.data;
  95. const [up, down] = data.split(/^--\s+?down\b/mi);
  96. if (!down) {
  97. const message = `The ${migration.filename} file does not contain '-- Down' separator.`;
  98. throw new Error(message);
  99. } else {
  100. /* eslint-disable no-param-reassign */
  101. migration.up = up.replace(/^-- .*?$/gm, '').trim();// Remove comments
  102. migration.down = down.trim(); // and trim whitespaces
  103. }
  104. });
  105. // Create a database table for migrations meta data if it doesn't exist
  106. await this.run(`CREATE TABLE IF NOT EXISTS "${table}" (
  107. id INTEGER PRIMARY KEY,
  108. name TEXT NOT NULL,
  109. up TEXT NOT NULL,
  110. down TEXT NOT NULL
  111. )`);
  112. // Get the list of already applied migrations
  113. let dbMigrations = await this.all(
  114. `SELECT id, name, up, down FROM "${table}" ORDER BY id ASC`,
  115. );
  116. // Undo migrations that exist only in the database but not in migs,
  117. // also undo the last migration if the `force` option was set to `last`.
  118. const lastMigration = migrations[migrations.length - 1];
  119. for (const migration of dbMigrations.slice().sort((a, b) => Math.sign(b.id - a.id))) {
  120. if (!migrations.some(x => x.id === migration.id) ||
  121. (force === 'last' && migration.id === lastMigration.id)) {
  122. const dbh = await this.get();
  123. await dbh.run('BEGIN');
  124. try {
  125. await dbh.exec(migration.down);
  126. await dbh.run(SQL`DELETE FROM "`.append(table).append(SQL`" WHERE id = ${migration.id}`));
  127. await dbh.run('COMMIT');
  128. dbMigrations = dbMigrations.filter(x => x.id !== migration.id);
  129. } catch (err) {
  130. await dbh.run('ROLLBACK');
  131. throw err;
  132. } finally {
  133. dbh.ret();
  134. }
  135. } else {
  136. break;
  137. }
  138. }
  139. // Apply pending migrations
  140. let applied = [];
  141. const lastMigrationId = dbMigrations.length ? dbMigrations[dbMigrations.length - 1].id : 0;
  142. for (const migration of migrations) {
  143. if (migration.id > lastMigrationId) {
  144. const dbh = await this.get();
  145. await dbh.run('BEGIN');
  146. try {
  147. await dbh.exec(migration.up);
  148. await dbh.run(SQL`INSERT INTO "`.append(table).append(
  149. SQL`" (id, name, up, down) VALUES (${migration.id}, ${migration.name}, ${migration.up}, ${migration.down})`)
  150. );
  151. await dbh.run('COMMIT');
  152. applied.push(migration.id);
  153. } catch (err) {
  154. await dbh.run('ROLLBACK');
  155. throw err;
  156. } finally {
  157. dbh.ret();
  158. }
  159. }
  160. }
  161. return applied;
  162. }
  163. }
  164. module.exports = SqliteConnectionPool;