feat: MySQLKeyValueDB

Author: kirillgroshkov

src/index.ts

@@ -1,7 +1,8 @@
 import { getDBAdapter } from './dbAdapter'
 import { MysqlDB, MysqlDBCfg, MysqlDBOptions, MysqlDBSaveOptions } from './mysql.db'
+import { MySQLKeyValueDB } from './mysqlKeyValueDB'
 import { jsonSchemaToMySQLDDL } from './schema/mysql.schema.util'
 
 export type { MysqlDBCfg, MysqlDBOptions, MysqlDBSaveOptions }
 
-export { MysqlDB, jsonSchemaToMySQLDDL, getDBAdapter }
+export { MysqlDB, MySQLKeyValueDB, jsonSchemaToMySQLDDL, getDBAdapter }

src/mysql.db.ts

@@ -60,7 +60,7 @@ export interface MysqlDBCfg extends PoolConfig {
 const BOOLEAN_TYPES = new Set(['TINY', 'TINYINT', 'INT'])
 const BOOLEAN_BIT_TYPES = new Set(['BIT'])
 
-const typeCast: TypeCast = (field, next) => {
+export const typeCast: TypeCast = (field, next) => {
   // cast to boolean
   if (field.length === 1) {
     if (BOOLEAN_BIT_TYPES.has(field.type)) {

src/mysqlKeyValueDB.ts

@@ -0,0 +1,155 @@
+import { Transform } from 'stream'
+import { CommonDBCreateOptions, CommonKeyValueDB, KeyValueDBTuple } from '@naturalcycles/db-lib'
+import { pMap } from '@naturalcycles/js-lib'
+import { ReadableTyped } from '@naturalcycles/nodejs-lib'
+import { QueryOptions } from 'mysql'
+import { MysqlDB, MysqlDBCfg, typeCast } from './mysql.db'
+
+interface KeyValueObject {
+  id: string
+  v: Buffer
+}
+
+export class MySQLKeyValueDB implements CommonKeyValueDB {
+  constructor(cfg: MysqlDBCfg = {}) {
+    this.cfg = {
+      typeCast,
+      ...cfg,
+    }
+
+    this.db = new MysqlDB(this.cfg)
+  }
+
+  cfg: MysqlDBCfg
+
+  db: MysqlDB
+
+  async ping(): Promise<void> {
+    await this.db.ping()
+  }
+
+  async close(): Promise<void> {
+    await this.db.close()
+  }
+
+  async createTable(table: string, opt: CommonDBCreateOptions = {}): Promise<void> {
+    if (opt.dropIfExists) await this.dropTable(table)
+
+    // On blob sizes: https://tableplus.com/blog/2019/10/tinyblob-blob-mediumblob-longblob.html
+    // LONGBLOB supports up to 4gb
+    // MEDIUMBLOB up to 16Mb
+    const sql = `create table ${table} (id VARCHAR(64) PRIMARY KEY, v LONGBLOB NOT NULL)`
+    this.db.cfg.logger.log(sql)
+    await this.db.runSQL({ sql })
+  }
+
+  async getByIds(table: string, ids: string[]): Promise<KeyValueDBTuple[]> {
+    if (!ids.length) return []
+
+    const sql = `SELECT id,v FROM ${table} where id in (${ids.map(id => `"${id}"`).join(',')})`
+
+    const rows = await this.db.runSQL<KeyValueObject[]>({ sql })
+
+    return rows.map(({ id, v }) => [id, v])
+  }
+
+  /**
+   * Use with caution!
+   */
+  async dropTable(table: string): Promise<void> {
+    await this.db.runSQL({ sql: `DROP TABLE IF EXISTS ${table}` })
+  }
+
+  async deleteByIds(table: string, ids: string[]): Promise<void> {
+    const sql = `DELETE FROM ${table} WHERE id in (${ids.map(id => `"${id}"`).join(',')})`
+    if (this.cfg.logSQL) this.db.cfg.logger.log(sql)
+    await this.db.runSQL({ sql })
+  }
+
+  async saveBatch(table: string, entries: KeyValueDBTuple[]): Promise<void> {
+    const statements: QueryOptions[] = entries.map(([id, buf]) => {
+      return {
+        sql: `INSERT INTO ${table} (id, v) VALUES (?, ?)`,
+        values: [id, buf],
+      }
+    })
+
+    await pMap(statements, async statement => {
+      if (this.cfg.debug) this.db.cfg.logger.log(statement.sql)
+      await this.db.runSQL(statement)
+    })
+  }
+
+  streamIds(table: string, limit?: number): ReadableTyped<string> {
+    let sql = `SELECT id FROM ${table}`
+    if (limit) sql += ` LIMIT ${limit}`
+    if (this.cfg.logSQL) this.db.cfg.logger.log(`stream: ${sql}`)
+
+    return this.db
+      .pool()
+      .query(sql)
+      .stream()
+      .pipe(
+        new Transform({
+          objectMode: true,
+          transform(row, _, cb) {
+            cb(null, row.id)
+          },
+        }),
+      )
+  }
+
+  streamValues(table: string, limit?: number): ReadableTyped<Buffer> {
+    let sql = `SELECT v FROM ${table}`
+    if (limit) sql += ` LIMIT ${limit}`
+    if (this.cfg.logSQL) this.db.cfg.logger.log(`stream: ${sql}`)
+
+    return this.db
+      .pool()
+      .query(sql)
+      .stream()
+      .pipe(
+        new Transform({
+          objectMode: true,
+          transform(row, _, cb) {
+            cb(null, row.v)
+          },
+        }),
+      )
+  }
+
+  streamEntries(table: string, limit?: number): ReadableTyped<KeyValueDBTuple> {
+    let sql = `SELECT id,v FROM ${table}`
+    if (limit) sql += ` LIMIT ${limit}`
+    if (this.cfg.logSQL) this.db.cfg.logger.log(`stream: ${sql}`)
+
+    return this.db
+      .pool()
+      .query(sql)
+      .stream()
+      .pipe(
+        new Transform({
+          objectMode: true,
+          transform(row, _, cb) {
+            cb(null, [row.id, row.v])
+          },
+        }),
+      )
+  }
+
+  async beginTransaction(): Promise<void> {
+    await this.db.runSQL({ sql: `BEGIN TRANSACTION` })
+  }
+
+  async endTransaction(): Promise<void> {
+    await this.db.runSQL({ sql: `END TRANSACTION` })
+  }
+
+  async count(table: string): Promise<number> {
+    const sql = `SELECT count(*) as cnt FROM ${table}`
+    if (this.cfg.logSQL) this.db.cfg.logger.log(sql)
+
+    const rows = await this.db.runSQL<{ cnt: number }[]>({ sql })
+    return rows[0]!.cnt
+  }
+}

src/test/mysqlKeyValueDB.manual.test.ts

@@ -0,0 +1,45 @@
+import { runCommonKeyValueDBTest, TEST_TABLE } from '@naturalcycles/db-lib/dist/testing'
+import { requireEnvKeys } from '@naturalcycles/nodejs-lib'
+import { MySQLKeyValueDB } from '../mysqlKeyValueDB'
+
+require('dotenv').config()
+
+const { MYSQL_HOST, MYSQL_USER, MYSQL_PW, MYSQL_DB } = requireEnvKeys(
+  'MYSQL_HOST',
+  'MYSQL_USER',
+  'MYSQL_PW',
+  'MYSQL_DB',
+)
+
+const db = new MySQLKeyValueDB({
+  host: MYSQL_HOST,
+  user: MYSQL_USER,
+  password: MYSQL_PW,
+  database: MYSQL_DB,
+  charset: 'utf8mb4',
+  logSQL: true,
+  // debug: true,
+})
+
+beforeAll(async () => {
+  await db.createTable(TEST_TABLE, { dropIfExists: true })
+})
+
+afterAll(async () => {
+  await db.close()
+})
+
+describe('runCommonKeyValueDBTest', () => runCommonKeyValueDBTest(db))
+
+test('count', async () => {
+  const count = await db.count(TEST_TABLE)
+  expect(count).toBe(0)
+})
+
+// test('test1', async () => {
+//   await db.deleteByIds(TEST_TABLE, ['id1', 'id2'])
+//   await db.saveBatch(TEST_TABLE, {
+//     k1: Buffer.from('hello1'),
+//     k2: Buffer.from('hello2'),
+//   })
+// })