MySQL2 to handle MongoDB connector BI

lib/mysql.js

@@ -1,47 +1,189 @@
-import {json} from "micro";
-import {createPool} from "mysql";
-import connectionConfig from "mysql/lib/ConnectionConfig.js";
-import types from "mysql/lib/protocol/constants/types.js";
 import JSONStream from "JSONStream";
+import {json} from "micro";
+import mysql, {createConnection} from "mysql2";
+import {failedCheck} from "./errors.js";
 
-const {parseUrl} = connectionConfig;
+const {Types, ConnectionConfig} = mysql;
 
-export default (url) => {
-  const pool = createPool(parseUrl(url));
+export async function query(req, res, pool) {
+  const {sql, params} = await json(req);
+  const keepAlive = setInterval(() => res.write("\n"), 25e3);
 
-  return async function query(req, res) {
-    const {sql, params} = await json(req);
+  let fields;
+  let rowCount = 0;
+  let bytes = 0;
+  try {
+    await new Promise((resolve, reject) => {
+      const stream = pool
+        .query({sql, timeout: 240e3}, params)
+        .once("fields", (f) => {
+          res.write(`{"schema":${JSON.stringify(schema((fields = f)))}`);
+        })
+        .stream()
+        .on("end", resolve)
+        .on("error", (error) => {
+          if (!stream.destroyed) stream.destroy();
+          reject(error);
+        })
+        .once("readable", () => clearInterval(keepAlive))
+        .pipe(JSONStream.stringify(`,"data":[`, ",", "]}"))
+        .on("data", (chunk) => {
+          bytes += chunk.length;
+          rowCount++;
+          if (rowCount && rowCount % 2e3 === 0)
+            req.log({
+              progress: {
+                rows: rowCount,
+                fields: fields.length,
+                bytes,
+                done: false,
+              },
+            });
+        });
+      stream.pipe(res, {end: false});
+    });
+  } catch (error) {
+    if (!error.statusCode) error.statusCode = 400;
+    throw error;
+  } finally {
+    clearInterval(keepAlive);
+  }
 
-    let fields;
+  req.log({
+    progress: {
+      rows: rowCount,
+      fields: fields ? fields.length : 0,
+      bytes,
+      done: true,
+    },
+  });
+
+  res.end();
+}
+
+export async function queryStream(req, res, pool) {
+  const {sql, params} = await json(req);
+  res.setHeader("Content-Type", "text/plain");
+  const keepAlive = setInterval(() => res.write("\n"), 25e3);
+
+  let fields;
+  let rowCount = 0;
+  let bytes = 0;
+
+  try {
     await new Promise((resolve, reject) => {
       const stream = pool
-        .query({sql, timeout: 30e3}, params)
-        .on("fields", (f) => (fields = f))
+        .query({sql, timeout: 240e3}, params)
+        .once("fields", (f) => {
+          res.write(JSON.stringify(schema((fields = f))));
+          res.write("\n");
+        })
         .stream()
         .on("end", resolve)
         .on("error", (error) => {
-          stream.destroy();
+          if (!stream.destroyed) stream.destroy();
           reject(error);
         })
-        .pipe(JSONStream.stringify(`{"data":[`, ",", "]"));
+        .once("readable", () => clearInterval(keepAlive))
+        .pipe(JSONStream.stringify("", "\n", "\n"))
+        .on("data", (chunk) => {
+          bytes += chunk.length;
+          rowCount++;
+          if (rowCount % 2e3 === 0)
+            req.log({
+              progress: {
+                rows: rowCount,
+                fields: fields.length,
+                bytes,
+                done: false,
+              },
+            });
+        });
       stream.pipe(res, {end: false});
     });
+  } catch (error) {
+    if (!error.statusCode) error.statusCode = 400;
+    throw error;
+  } finally {
+    clearInterval(keepAlive);
+  }
+
+  req.log({
+    progress: {
+      rows: rowCount,
+      fields: fields ? fields.length : 0,
+      bytes,
+      done: true,
+    },
+  });
+
+  res.end();
+}
+
+const READ_ONLY = new Set(["SELECT", "SHOW DATABASES", "SHOW VIEW", "USAGE"]);
+export async function check(req, res, pool) {
+  const rows = await new Promise((resolve, reject) => {
+    pool.query("SHOW GRANTS FOR CURRENT_USER", (error, results) => {
+      error ? reject(failedCheck(error.message)) : resolve(results);
+    });
+  });
+  const grants = [].concat(
+    ...rows.map((grant) =>
+      Object.values(grant)[0]
+        .match(/^GRANT (.+) ON/)[1]
+        .split(", ")
+    )
+  );
+  const permissive = grants.filter((g) => !READ_ONLY.has(g));
+  if (permissive.length)
+    throw failedCheck(
+      `User has too permissive grants: ${permissive.join(", ")}`
+    );
+
+  return {ok: true};
+}
 
-    const schema = {
-      type: "array",
-      items: {
-        type: "object",
-        properties: fields.reduce(
-          (schema, {name, type, charsetNr}) => (
-            (schema[name] = dataTypeSchema({type, charsetNr})), schema
-          ),
-          {}
+export default (url) => async (req, res) => {
+  const config = ConnectionConfig.parseUrl(url);
+
+  // Unless specified as a property of the url connection string, ssl is used with the default.
+  // See https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-connp-props-security.html#cj-conn-prop_sslMode
+  if (config.sslMode !== "DISABLED") {
+    config.ssl = {};
+  }
+
+  // the mysql2.createConnection method is not happy if we pass any extra properties not recognized by it.
+  delete config.sslMode;
+
+  const connection = createConnection({
+    ...config,
+    decimalNumbers: true,
+  });
+
+  if (req.method === "POST") {
+    if (req.url === "/query") return query(req, res, connection);
+    if (req.url === "/query-stream") return queryStream(req, res, connection);
+    if (req.url === "/check") return check(req, res, connection);
+  }
+
+  throw notFound();
+};
+
+function schema(fields) {
+  return {
+    type: "array",
+    items: {
+      type: "object",
+      properties: fields.reduce(
+        (schema, {name, type, characterSet}) => (
+          (schema[name] = dataTypeSchema({type, charsetNr: characterSet})),
+          schema
         ),
-      },
-    };
-    res.end(`,"schema":${JSON.stringify(schema)}}`);
+        {}
+      ),
+    },
   };
-};
+}
 
 // https://github.com/mysqljs/mysql/blob/5569e02ad72789f4b396d9a901f0390fe11b5b4e/lib/protocol/constants/types.js
 // https://github.com/mysqljs/mysql/blob/5569e02ad72789f4b396d9a901f0390fe11b5b4e/lib/protocol/packets/RowDataPacket.js#L53
@@ -52,46 +194,53 @@ const boolean = ["null", "boolean"],
   string = ["null", "string"];
 function dataTypeSchema({type, charsetNr}) {
   switch (type) {
-    case types.BIT:
+    case Types.BIT:
       return {type: boolean};
-    case types.TINY:
-    case types.SHORT:
-    case types.LONG:
-      return {type: integer};
-    case types.INT24:
-    case types.YEAR:
-    case types.FLOAT:
-    case types.DOUBLE:
-    case types.DECIMAL:
-    case types.NEWDECIMAL:
-      return {type: number};
-    case types.TIMESTAMP:
-    case types.DATE:
-    case types.DATETIME:
-    case types.NEWDATE:
-    case types.TIMESTAMP2:
-    case types.DATETIME2:
-    case types.TIME2:
+    case Types.TINY:
+      return {type: integer, tiny: true};
+    case Types.SHORT:
+      return {type: integer, short: true};
+    case Types.LONG:
+      return {type: integer, long: true};
+    case Types.INT24:
+      return {type: number, int24: true};
+    case Types.YEAR:
+      return {type: number, year: true};
+    case Types.FLOAT:
+      return {type: number, float: true};
+    case Types.DOUBLE:
+      return {type: number, double: true};
+    case Types.DECIMAL:
+      return {type: number, decimal: true};
+    case Types.NEWDECIMAL:
+      return {type: number, newdecimal: true};
+    case Types.TIMESTAMP:
+    case Types.DATE:
+    case Types.DATETIME:
+    case Types.NEWDATE:
+    case Types.TIMESTAMP2:
+    case Types.DATETIME2:
+    case Types.TIME2:
       return {type: string, date: true};
-    case types.LONGLONG: // TODO
+    case Types.LONGLONG:
       return {type: string, bigint: true};
-    case types.TINY_BLOB:
-    case types.MEDIUM_BLOB:
-    case types.LONG_BLOB:
-    case types.BLOB:
-    case types.VAR_STRING:
-    case types.VARCHAR:
-    case types.STRING:
+    case Types.TINY_BLOB:
+    case Types.MEDIUM_BLOB:
+    case Types.LONG_BLOB:
+    case Types.BLOB:
+    case Types.VAR_STRING:
+    case Types.VARCHAR:
+    case Types.STRING:
       return charsetNr === 63 // binary
         ? {type: object, buffer: true}
         : {type: string};
-    case types.JSON:
-      return {type: object};
-    case types.TIME: // TODO
-    case types.ENUM: // TODO
-    case types.SET: // TODO
-    case types.GEOMETRY: // TODO
-    case types.NULL: // TODO
+    case Types.JSON:
+      return {type: object, json: true};
+    case Types.TIME: // TODO
+    case Types.ENUM: // TODO
+    case Types.SET: // TODO
+    case Types.GEOMETRY: // TODO
+    case Types.NULL: // TODO
     default:
       return {type: string};
   }

lib/server.js

@@ -6,14 +6,15 @@ import https from "https";
 import {readFileSync} from "fs";
 import {run} from "micro";
 import {createHmac, timingSafeEqual} from "crypto";
-import serializeErrors from "./serialize-errors.js";
+import serializeErrors from "../middleware/serialize-errors.js";
 import {notFound, unauthorized, exit} from "./errors.js";
 import mysql from "./mysql.js";
 import postgres from "./postgres.js";
 import snowflake from "./snowflake.js";
 import mssql from "./mssql.js";
 import oracle from "./oracle.js";
 import databricks from "./databricks.js";
+import logger from "../middleware/logger.js";
 
 export async function server(config, argv) {
   const development = process.env.NODE_ENV === "development";
@@ -34,7 +35,7 @@ export async function server(config, argv) {
   } = config;
 
   const handler =
-    type === "mysql"
+    type === "mysql" || type === "mongosql"
       ? mysql(url)
       : type === "postgres"
       ? postgres(url)
@@ -63,11 +64,11 @@ export async function server(config, argv) {
     const sslcert = readFileSync(argv.sslcert);
     const sslkey = readFileSync(argv.sslkey);
     server = https.createServer({cert: sslcert, key: sslkey}, (req, res) =>
-      run(req, res, serializeErrors(index))
+      run(req, res, logger(serializeErrors(index)))
     );
   } else {
     server = http.createServer((req, res) =>
-      run(req, res, serializeErrors(index))
+      run(req, res, logger(serializeErrors(index)))
     );
   }
 

middleware/logger.js

@@ -0,0 +1,22 @@
+// Recursively flatten key names separated by dots.
+function* entries(data, prefix = []) {
+  if (
+    data instanceof Object &&
+    Object.getPrototypeOf(data) === Object.prototype
+  ) {
+    for (const [key, value] of Object.entries(data))
+      yield* entries(value, prefix.concat(key));
+  } else {
+    yield [prefix.join("."), data];
+  }
+}
+
+export default (handler) => (req, res) => {
+  req.log = function log(data) {
+    const requestId = req.headers["x-request-id"];
+    const parts = requestId ? [`http.request_id=${requestId}`] : [];
+    for (const [key, value] of entries(data)) parts.push(`${key}=${value}`);
+    console.log(parts.join(" "));
+  };
+  return handler(req, res);
+};

package.json

@@ -23,7 +23,7 @@
     "ajv": "^8.11.0",
     "micro": "^9.3.4",
     "mssql": "^9.0.1",
-    "mysql": "^2.17.1",
+    "mysql2": "^3.0.1",
     "open": "^6.3.0",
     "pg": "^8.7.1",
     "pg-query-stream": "^4.2.1",