feat(sql): add SQL utility functions and refactor message insertion l…

…ogic (#1084)

apps/telegram-monitor/src/index.ts

@@ -1,4 +1,5 @@
 import { encodePath, formatTime } from '@yuants/data-model';
+import { buildInsertManyIntoTableSQL, requestSQL } from '@yuants/sql';
 import { listWatch } from '@yuants/utils';
 import {
   bufferTime,
@@ -150,18 +151,9 @@ message$
     bufferTime(1000),
     filter((messages) => messages.length > 0),
     mergeMap((messages) =>
-      defer(() =>
-        terminal.requestForResponse('SQL', {
-          query: `
-      insert into telegram_messages (message_id, created_at, chat_id, sender_id, message, raw_data) 
-      values ${messages
-        .map(
-          (temp1) =>
-            `('${temp1.message_id}', '${temp1.created_at}', '${temp1.chat_id}', '${temp1.sender_id}', '${temp1.message}', '${temp1.raw_data}')`,
-        )
-        .join(',')}`,
-        }),
-      ).pipe(retry({ delay: 5000 })),
+      defer(() => requestSQL(terminal, buildInsertManyIntoTableSQL(messages, 'telegram_messages'))).pipe(
+        retry({ delay: 5000 }),
+      ),
     ),
   )
   .subscribe();

common/changes/@yuants/app-telegram-monitor/2025-02-23-15-56.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@yuants/app-telegram-monitor",
+      "comment": "fix sql",
+      "type": "patch"
+    }
+  ],
+  "packageName": "@yuants/app-telegram-monitor"
+}

common/changes/@yuants/sql/2025-02-23-15-56.json

@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@yuants/sql",
+      "comment": "add some utils",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@yuants/sql"
+}

libraries/sql/etc/sql.api.md

@@ -9,6 +9,15 @@ import { Terminal } from '@yuants/protocol';
 // @public
 export const AddMigration: (migration: ISQLMigration) => void;
 
+// @public
+export const buildInsertManyIntoTableSQL: <T extends {}>(data: T[], tableName: string, options?: {
+    columns?: (keyof T)[] | undefined;
+} | undefined) => string;
+
+// @public
+const escape_2: (val: any, options?: {}) => string;
+export { escape_2 as escape }
+
 // @public
 export const ExecuteMigrations: (terminal: Terminal) => Promise<void>;
 
@@ -20,6 +29,9 @@ export interface ISQLMigration {
     statement: string;
 }
 
+// @public
+export const requestSQL: <T = unknown>(terminal: Terminal, query: string) => Promise<T>;
+
 // (No @packageDocumentation comment for this package)
 
 ```

libraries/sql/src/index.ts

@@ -153,3 +153,57 @@ export const ExecuteMigrations = async (terminal: Terminal) => {
 
   console.info(formatTime(Date.now()), `SetupMigrationEnd`);
 };
+
+/**
+ * 执行 SQL 语句
+ *
+ * @public
+ */
+export const requestSQL = async <T = unknown>(terminal: Terminal, query: string): Promise<T> => {
+  const result = await terminal.requestForResponse('SQL', {
+    query,
+  });
+
+  if (result.code !== 0) {
+    throw new Error(`Failed to run SQL query: ${query}, message: ${result.message}`);
+  }
+
+  return result.data as any as T;
+};
+
+/**
+ * 进行值的转义，防止 SQL 注入
+ *
+ * @public
+ */
+export const escape = (val: any, options: {} = {}): string => {
+  if (val === undefined || val === null) return 'NULL';
+  if (typeof val === 'number') return `${val}`;
+  if (typeof val === 'string') return `'${val.replace(/'/g, "''")}'`;
+  if (typeof val === 'boolean') return val ? 'TRUE' : 'FALSE';
+  // fallback to JSON
+  return escape(JSON.stringify(val));
+};
+
+const isValidColumnName = (name: string): boolean => {
+  return /^[a-zA-Z_][a-zA-Z0-9_]*$/.test(name);
+};
+
+/**
+ * 构造 Insert Many 模式的 SQL 查询语句 (INSERT INTO ... VALUES ...)
+ *
+ * @public
+ */
+export const buildInsertManyIntoTableSQL = <T extends {}>(
+  data: T[],
+  tableName: string,
+  options?: {
+    columns?: Array<keyof T>;
+  },
+): string => {
+  if (data.length === 0) throw 'Data is empty';
+  const columns = (options?.columns ?? Object.keys(data[0]).filter(isValidColumnName)) as string[];
+  return `INSERT INTO ${tableName} (${columns.join(',')}) VALUES ${data
+    .map((x) => `(${columns.map((c) => escape(x[c as keyof T])).join(',')})`)
+    .join(',')}`;
+};

ui/web/src/modules.ts

@@ -26,6 +26,7 @@ export * as Order from './modules/Order';
 export * as Pages from './modules/Pages';
 export * as Products from './modules/Products';
 export * as PullSourceRelations from './modules/PullSourceRelations';
+export * as SQL from './modules/SQL';
 export * as SupaBase from './modules/SupaBase';
 export * as System from './modules/System';
 export * as Terminals from './modules/Terminals';