Merge pull request #37 from oslabs-beta/ms/valid-input-error

Ms/valid input error

Author: matthewxing1

.gitignore

@@ -103,5 +103,10 @@ dist
 # TernJS port file
 .tern-port
 
+# Local
+ksqljsTest.js
+package-lock.json
+local_ignore/
+.gitignore
 
 

ksqldb/customErrors.js

@@ -10,12 +10,13 @@ class ksqlDBError extends Error {
 
     // Ensure the name of this error is the same as the class name
     this.name = this.constructor.name
-
+    
     // capturing the stack trace keeps the reference to your error class
     Error.captureStackTrace(this, this.constructor);
-
+    
     // you may also assign additional properties to your error
     //this.status = 404
+    Object.keys(error).forEach(property => {this[property] = error[property]});
   }
 }
 
@@ -46,10 +47,22 @@ class InappropriateStringParamError extends QueryBuilderError {
   }
 }
 
+class invalidArgumentTypes extends Error {
+  constructor(message) {
+    super(message);
+
+    this.name = this.constructor.name;
+    // necessary?
+    Error.captureStackTrace(this, this.constructor);
+
+  }
+}
+
 module.exports = {
   ksqlDBError,
   QueryBuilderError,
   EmptyQueryError,
   NumParamsError,
-  InappropriateStringParamError
+  InappropriateStringParamError,
+  invalidArgumentTypes
 };

ksqldb/ksqldb.js

@@ -2,6 +2,7 @@ const axios = require("axios");
 const https = require('node:https');
 const http2 = require("http2");
 const { ksqlDBError } = require("./customErrors.js");
+const validateInputs = require('./validateInputs.js');
 const queryBuilder = require('./queryBuilder.js');
 const builder = new queryBuilder();
 
@@ -48,10 +49,14 @@ class ksqldb {
    *         Example: [{object that contains the metadata}, [data], [data], ...}]
    */
   pull = (query) => {
+    validateInputs([query, 'string', 'query']);
+
+    const validatedQuery = builder.build(query);
+    
     return axios
       .post(this.ksqldbURL + "/query-stream",
         {
-          sql: query,
+          sql: validatedQuery,
         },
         {
           headers:
@@ -65,8 +70,7 @@ class ksqldb {
         })
       .then((res) => res.data)
       .catch((error) => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
       });
   }
 
@@ -85,6 +89,9 @@ class ksqldb {
    *         result if successful.
    */
   push(query, cb) {
+    validateInputs([query, 'string', 'query', true], [cb, 'function', 'cb', true]);
+    const validatedQuery = builder.build(query);
+
     return new Promise((resolve, reject) => {
       let sentQueryId = false;
       const session = http2.connect(
@@ -109,7 +116,7 @@ class ksqldb {
       );
 
       const reqBody = {
-        sql: query,
+        sql: validatedQuery,
         Accept: "application/json, application/vnd.ksqlapi.delimited.v1",
       };
 
@@ -118,6 +125,9 @@ class ksqldb {
       req.setEncoding("utf8");
 
       req.on("data", (chunk) => {
+        // check for chunk containing errors
+        if (JSON.parse(chunk)['@type']?.includes('error')) throw new ksqlDBError(JSON.parse(chunk));
+        // continue if chunk indicates a healthy response
         if (!sentQueryId) {
           sentQueryId = true;
           cb(chunk);
@@ -145,6 +155,8 @@ class ksqldb {
    *         if the termination was successful.
    */
   terminate(queryId) {
+    validateInputs([queryId, 'string', 'queryId']);
+
     return axios.post(this.ksqldbURL + '/ksql',
       {
         ksql: `TERMINATE ${queryId};`
@@ -161,8 +173,7 @@ class ksqldb {
       })
       .then(res => res.data[0])
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
       });
   }
 
@@ -177,9 +188,13 @@ class ksqldb {
    * @return {Promise} a promise that completes once the server response is received, and returns the requested data.
    */
   ksql(query) {
+    validateInputs([query, 'string', 'query']);
+
+    const validatedQuery = builder.build(query);
+    
     return axios.post(this.ksqldbURL + '/ksql',
       {
-        ksql: query
+        ksql: validatedQuery
       },
       {
         headers:
@@ -193,8 +208,7 @@ class ksqldb {
       })
       .then(res => res.data[0])
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
       });
   }
 
@@ -214,9 +228,8 @@ class ksqldb {
    * @return {Promise} a promise that completes once the server response is received, and returns a response object.
    */
   createStream(name, columnsType, topic, value_format = 'json', partitions = 1, key) {
-    if (typeof name !== 'string' || typeof columnsType !== 'object' || typeof topic !== 'string' || typeof partitions !== 'number') {
-      return console.log("invalid input(s)")
-    }
+    validateInputs([name, 'string', 'name', true], [columnsType, 'array', 'columnsType', true], [topic, 'string', 'topic'], [partitions, 'number', 'partitions']);
+
     const columnsTypeString = columnsType.reduce((result, currentType) => result + ', ' + currentType);
     const query = `CREATE STREAM ${name} (${columnsTypeString}) WITH (kafka_topic='${topic}', value_format='${value_format}', partitions=${partitions});`;
 
@@ -230,10 +243,8 @@ class ksqldb {
           {},
       httpsAgent: this.httpsAgentAxios ? this.httpsAgentAxios : null,
     })
-      .then(res => res)
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
       });
   }
 
@@ -248,6 +259,8 @@ class ksqldb {
    * @returns {Promise} - a promise that completes once the server response is received, and returns a query ID
    */
   createStreamAs = (streamName, selectColumns, sourceStream, propertiesObj, conditions, partitionBy) => {
+    validateInputs([streamName, 'string', 'streamName', true], [selectColumns, 'array', 'selectColumns', true], [sourceStream, 'string', 'sourceStream', true], [propertiesObj, 'object', 'propertiesObj'], [conditions, 'string', 'conditions'], [partitionBy, 'string', 'partitionBy']);
+
     const propertiesArgs = [];
     const selectColStr = selectColumns.reduce((result, current) => result + ', ' + current);
     // begin with first consistent portion of query
@@ -286,7 +299,9 @@ class ksqldb {
       httpsAgent: this.httpsAgentAxios ? this.httpsAgentAxios : null,
     })
       .then(res => res.data[0].commandStatus.queryId)
-      .catch(error => console.log(error));
+      .catch(error => {
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
+      });
   }
 
   //---------------------Create tables-----------------
@@ -305,6 +320,8 @@ class ksqldb {
    * @return {Promise} a promise that completes once the server response is received, and returns a response object.
    */
   createTable = (name, columnsType, topic, value_format = 'json', partitions) => {
+    validateInputs([name, 'string', 'name', true], [columnsType, 'array', 'columnsType', true], [topic, 'string', 'topic', true], [value_format, 'string', 'value_format', true], [partitions, 'number', 'partitions']);
+
     const columnsTypeString = columnsType.reduce((result, currentType) => result + ', ' + currentType);
     const query = `CREATE TABLE ${name} (${columnsTypeString}) WITH (kafka_topic='${topic}', value_format='${value_format}', partitions=${partitions});`
 
@@ -323,8 +340,7 @@ class ksqldb {
         httpsAgent: this.httpsAgentAxios ? this.httpsAgentAxios : null,
       })
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
       });
   }
 
@@ -344,6 +360,8 @@ class ksqldb {
    * @returns {Promise} a promise that completes once the server response is received, returning a response object
    */
   createTableAs = (tableName, source, selectArray, propertiesObj, conditionsObj) => {
+    validateInputs([tableName, 'string', 'tableName', true], [source, 'string', 'source', true], [selectArray, 'array', 'selectArray', true], [propertiesObj, 'object', 'propertiesObj'], [conditionsObj, 'object', 'conditionsObj']);
+
     let selectColStr = selectArray.reduce((result, current) => result + ', ' + current);
 
     // expect user to input properties object of format {topic: ... , value_format: ..., partitions: ...}
@@ -417,6 +435,8 @@ class ksqldb {
  */
   //---------------------Insert Rows Into Existing Streams-----------------
   insertStream = (stream, rows) => {
+    validateInputs([stream, 'string', 'stream', true], [rows, 'array', 'rows', true]);
+
     return new Promise((resolve, reject) => {
       const msgOutput = [];
 
@@ -452,6 +472,9 @@ class ksqldb {
       req.setEncoding("utf8");
 
       req.on("data", (chunk) => {
+        // check for chunk containing errors
+        if (JSON.parse(chunk)['@type']?.includes('error')) throw new ksqlDBError(JSON.parse(chunk));
+        // continue if chunk indicates a healthy response
         msgOutput.push(JSON.parse(chunk));
       });
 
@@ -480,13 +503,8 @@ class ksqldb {
    *         the end of the array that includes the time that the data was inserted into the ksqldb.
    */
   pullFromTo = async (streamName, timezone = 'Greenwich', from = [undefined, '00', '00', '00'], to = ['2200-03-14', '00', '00', '00']) => {
-    if (!streamName || typeof timezone !== 'string' || !from
-      || typeof from[0] !== 'string' || typeof from[1] !== 'string' || typeof from[2] !== 'string' || typeof from[3] !== 'string'
-      || typeof to[0] !== 'string' || typeof to[1] !== 'string' || typeof to[2] !== 'string' || typeof to[3] !== 'string'
-      || from[0].length !== 10 || to[0].length !== 10 || from[1].length !== 2 || to[1].length !== 2 || from[2].length !== 2 || to[2].length !== 2 || from[3].length !== 2 || to[3].length !== 2
-    ) {
-      return new Error('invalid inputs');
-    }
+    validateInputs([streamName, 'string', 'streamName', true], [timezone, 'string', 'timezone', true], [from, 'array', 'from', true], [to, 'array', 'to', true]);
+
     const userFrom = `${from[0]}T${from[1]}:${from[2]}:${from[3]}`;
     const userTo = `${to[0]}T${to[1]}:${to[2]}:${to[3]}`;
     const userFromUnix = new Date(userFrom).getTime();
@@ -518,11 +536,12 @@ class ksqldb {
    *         message (string): Detailed message regarding the status of the execution statement.
    */
   inspectQueryStatus(commandId) {
+    validateInputs([commandId, 'string', 'commandId', true]);
+
     return axios.get(this.ksqldbURL + `/status/${commandId}`)
       .then(response => response)
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
       });
   }
 
@@ -539,8 +558,7 @@ class ksqldb {
     return axios.get(this.ksqldbURL + `/info`)
       .then(response => response)
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -557,8 +575,7 @@ class ksqldb {
     return axios.get(this.ksqldbURL + `/healthcheck`)
       .then(response => response)
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -576,8 +593,7 @@ class ksqldb {
     return axios.get(this.ksqldbURL + `/clusterStatus`)
       .then(response => response)
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -592,6 +608,8 @@ class ksqldb {
    * @return {Promise} this method returns a promise that returns a response object.
    */
   terminateCluster(topicsToDelete = []) {
+    validateInputs([topicsToDelete, 'array', 'topicsToDelete', true]);
+
     return axios.post(this.ksqldbURL + `/ksql/terminate`, {
       "deleteTopicList": topicsToDelete
     }, {
@@ -604,8 +622,7 @@ class ksqldb {
     })
       .then(response => response)
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -623,14 +640,16 @@ class ksqldb {
  *   "message": "One or more properties overrides set locally are prohibited by the KSQL server (use UNSET to reset their default value): [ksql.service.id]"
  * }
  *
+ * @param {string} propertyName - the name of the property to validate
  * @return {Promise} this method returns a promise that resolves to a boolean true if the property is allowed to be changed.
  */
   isValidProperty(propertyName) {
+    validateInputs([propertyName, 'string', 'propertyName', true]);
+
     return axios.get(this.ksqldbURL + `/is_valid_property/${propertyName}`)
       .then(response => response)
       .catch(error => {
-        console.error(error);
-        throw new ksqlDBError(error);
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 };

ksqldb/queryBuilder.js

@@ -5,9 +5,11 @@ class queryBuilder {
   }
 
   build = (query, ...params) => {
-    // consider building custom errors
+    // check for empty query
     if (this._checkEmptyQuery(query)) throw new EmptyQueryError();
+
     let output = this._bind(query, ...params);
+
     return output;
   }
   _bind = (query, ...params) => {
@@ -33,10 +35,10 @@ class queryBuilder {
         return param;
       case "object":
         if (Array.isArray(param)) {
-          if (param[0].includes(";")) {
+          if (param[0]?.includes(";")) {
             throw new InappropriateStringParamError("string params not wrapped in quotes should not include semi-colons");
           }
-          return `${param[0].replaceAll("'", "''")}`
+          return `${param[0]?.replaceAll("'", "''")}`
         }
         throw new QueryBuilderError("object should not be passed in as query argument");
       case "function":