fix: stream method was broken, since readable-stream in mysql is not real

Author: kirillgroshkov

src/mysql.db.ts

@@ -1,4 +1,4 @@
-import { Readable } from 'node:stream'
+import { Readable, Transform } from 'node:stream'
 import { promisify } from 'node:util'
 import {
   BaseCommonDB,
@@ -277,10 +277,21 @@ export class MysqlDB extends BaseCommonDB implements CommonDB {
 
     if (this.cfg.logSQL) this.cfg.logger.log(`stream: ${sql}`)
 
-    // return this.streamSQL(sql, opt)
-    return (this.pool().query(sql).stream() as ReadableTyped<ROW>).map(row =>
-      _filterUndefinedValues(row, true),
-    )
+    // todo: this is nice, but `mysql` package uses `readable-stream@2` which is not compatible with `node:stream` iterable helpers
+    // return (this.pool().query(sql).stream() as ReadableTyped<ROW>).map(row =>
+    //   _filterUndefinedValues(row, true),
+    // )
+    return this.pool()
+      .query(sql)
+      .stream()
+      .pipe(
+        new Transform({
+          objectMode: true,
+          transform(row: ROW, _encoding, cb) {
+            cb(null, _filterUndefinedValues(row, true))
+          },
+        }),
+      )
   }
 
   // SAVE

src/mysqlKeyValueDB.ts

@@ -1,3 +1,4 @@
+import { Transform } from 'node:stream'
 import {
   CommonDBCreateOptions,
   CommonKeyValueDB,
@@ -88,7 +89,20 @@ export class MySQLKeyValueDB implements CommonKeyValueDB {
     if (limit) sql += ` LIMIT ${limit}`
     if (this.cfg.logSQL) this.db.cfg.logger.log(`stream: ${sql}`)
 
-    return (this.db.pool().query(sql).stream() as ReadableTyped<ObjectWithId>).map(row => row.id)
+    // todo: this is nice, but `mysql` package uses `readable-stream@2` which is not compatible with `node:stream` iterable helpers
+    // return (this.db.pool().query(sql).stream() as ReadableTyped<ObjectWithId>).map(row => row.id)
+    return this.db
+      .pool()
+      .query(sql)
+      .stream()
+      .pipe(
+        new Transform({
+          objectMode: true,
+          transform(row: ObjectWithId, _encoding, cb) {
+            cb(null, row.id)
+          },
+        }),
+      )
   }
 
   streamValues(table: string, limit?: number): ReadableTyped<Buffer> {