Repository navigation
Expand file tree
/
Copy pathmysql.ts
More file actions
1535 lines (1452 loc) · 52.3 KB
/
Copy pathmysql.ts
File metadata and controls
1535 lines (1452 loc) · 52.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/**
* MySQL data access layer
* @category mysql-backend
* @module mysql
*/
import mysql from "mysql2/promise";
import type { Pool, PoolConnection } from "mysql2/promise";
import { createRequire } from "module";
const require = createRequire(import.meta.url);
import {
sqlsanitize,
sqlsanitizeAllowDots,
mkWhereForDialect,
mkSelectOptionsForDialect,
ftsFieldsSqlExpr,
} from "@saltcorn/db-common/internal";
import type {
Value,
Where,
SelectOptions,
Row,
SqlDialect,
} from "@saltcorn/db-common/internal";
import tenantsModule from "@saltcorn/db-common/tenants";
import { reprAsJson } from "@saltcorn/db-common/sqlite-commons";
const ISO_DATETIME_RE =
/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+-]\d{2}:\d{2})$/;
/**
* mysql2 does not auto-serialize bound parameters into JSON (unlike `pg`,
* which handles JS objects for jsonb columns itself) - plain objects/arrays
* need an explicit JSON.stringify before binding. Dates and PlainDate
* instances are left untouched so mysql2 can bind them as proper
* DATETIME/DATE values.
*/
const mkVal = (v: any): any => {
if (typeof v === "number" && Number.isNaN(v)) return null;
if (Buffer.isBuffer(v)) return v;
if (
v?.constructor?.name === "PlainDate" &&
typeof v.toISOString === "function"
)
return v.toISOString();
if (reprAsJson(v)) return JSON.stringify(v);
if (typeof v === "string" && ISO_DATETIME_RE.test(v)) {
const d = new Date(v);
if (!isNaN(d.getTime())) return d;
}
return v;
};
const normalizePlaceholders = (
text: string,
params?: any[],
): { text: string; params?: any[] } => {
if (!params || !text.match(/\$\d+/)) return { text, params };
const orderedParams: any[] = [];
const normalizedText = text.replace(/\$(\d+)/g, (_match, index) => {
orderedParams.push(params[+index - 1]);
return "?";
});
return { text: normalizedText, params: orderedParams };
};
let getTenantSchema: () => string;
let getRequestContext: () => any;
let getConnectObject: ((connObj?: any) => any) | null = null;
export let pool: Pool | null = null;
let sessionPool: Pool | null = null;
let log_sql_enabled = false;
const quote = (s: string): string =>
s.includes(".")
? s.split(".").map(quote).join(".")
: s.includes('"')
? s
: `"${s}"`;
const ppPK = (pk?: string): string => (pk ? quote(pk) : "id");
/*
* A SqlDialect implementation for MySQL.
*
* The mysql2 client uses positional "?" placeholders (like SQLite), and
* MySQL run with ANSI_QUOTES + PIPES_AS_CONCAT (set once per connection in
* buildPool() below) accepts double-quoted identifiers and "||" string
* concatenation exactly like Postgres/SQLite, so the shared
* quote()/sqlsanitize()/"'%' || ? || '%'"-style machinery in db-common
* needs no changes.
*/
export const mysqlPlaceHolderStack = (): SqlDialect => {
let values: Value[] = [];
const push = (x: Value): string => {
values.push(x);
return "?";
};
const self: SqlDialect = {
name: "mysql",
is_sqlite: false,
push,
getValues() {
return values;
},
placeholderAt() {
return "?";
},
like() {
// MySQL's default collation (*_ci) is already case-insensitive, so a
// plain LIKE gives ILIKE-equivalent semantics - MySQL has no ILIKE
// keyword at all.
return "LIKE";
},
regexOperator() {
return "REGEXP";
},
castDateExpr(doCast: boolean, s: string) {
return !doCast ? s : `DATE(${s})`;
},
jsonExtractExpr(fieldExpr: string, jsonPath: string, asText: boolean) {
const p = jsonPath.replace(/\\/g, "\\\\");
return asText
? `JSON_UNQUOTE(JSON_EXTRACT(${fieldExpr}, '${p}'))`
: `JSON_EXTRACT(${fieldExpr}, '${p}')`;
},
ftsWhereClause(v) {
// A true MATCH()...AGAINST() implementation needs a FULLTEXT index on
// the exact column list being searched, but Saltcorn's FTS expression
// (ftsFieldsSqlExpr) is a computed coalesce/concat across possibly
// several columns, which MySQL cannot back with a FULLTEXT index (no
// expression indexes there). Until the (deferred) generic table-DDL
// work adds MySQL-aware FTS index creation, fall back to LIKE - same
// degraded-but-correct behaviour already used for the disable_fts case
// on Postgres and always on SQLite.
const { fields, table, schema } = v;
const flds = ftsFieldsSqlExpr(fields, table, schema);
return `${flds} LIKE '%' || ${push(v.searchTerm)} || '%'`;
},
arrayInClause(vals: Value[]) {
return `IN (${vals.map((v) => push(v)).join(", ")})`;
},
slugifyWhereClause(k: string, s: string) {
return `REGEXP_REPLACE(REPLACE(LOWER(${quote(
sqlsanitizeAllowDots(k),
)}),' ','-'),'[^\\\\w-]','')=${push(s)}`;
},
textCastSuffix() {
return "";
},
};
// MySQL's random function is RAND(); postgres/sqlite use RANDOM().
(self as SqlDialect & { randomOrderExpr?: () => string }).randomOrderExpr =
() => "RAND()";
return self;
};
const mkWhere = (
whereObj: Where | undefined,
): { where: string; values: Value[] } =>
mkWhereForDialect(whereObj, mysqlPlaceHolderStack());
const mkSelectOptions = (selopts: SelectOptions, values: any[]): string =>
mkSelectOptionsForDialect(selopts, values, mysqlPlaceHolderStack());
/**
* Control Logging sql statements to console
* @param {boolean} [val = true] - if true then log sql statements to console
*/
export function set_sql_logging(val: boolean = true): void {
log_sql_enabled = val;
}
/**
* Get sql logging state
* @returns {boolean} if true then sql logging eabled
*/
export function get_sql_logging(): boolean {
return log_sql_enabled;
}
/**
* Log SQL statement to console
* @param {string} sql - SQL statement
* @param {object} [vs] - any additional parameter
*/
export function sql_log(sql: string, vs?: any): void {
if (log_sql_enabled)
if (typeof vs === "undefined") console.log(sql);
else console.log(sql, vs);
}
const boolTypeCast = (field: any, next: () => any): any => {
if (field.type === "TINY" && field.length === 1) {
const v = field.string();
return v === null ? null : v === "1";
}
return next();
};
const buildPool = (connectObj: any): Pool => {
// DATABASE_URL takes priority over discrete host/user/password/database in
// connect.ts's getConnectObject() (which deletes the discrete fields when
// it's set), so a "mysql://user:pass@host:port/db" DATABASE_URL must be
// handled here too - mysql2's createPool() accepts a plain URI string as
// an alternative to an options object, same idea as `pg`'s Pool accepting
// `{ connectionString }`.
const newPool = connectObj.connectionString
? mysql.createPool({
uri: `${connectObj.connectionString}${
connectObj.connectionString.includes("?") ? "&" : "?"
}multipleStatements=true`,
typeCast: boolTypeCast,
timezone: "Z",
})
: mysql.createPool({
host: connectObj.host,
port: connectObj.port ? +connectObj.port : undefined,
user: connectObj.user,
password: connectObj.password,
database: connectObj.database || connectObj.default_schema,
ssl: connectObj.ssl,
dateStrings: false,
// Treat all DATETIME values as UTC (matching postgres's timestamptz).
// Combined with `SET time_zone='+00:00'` per connection, NOW(3)/
// FROM_UNIXTIME and Date round-trips stay consistent regardless of the
// server's or node process's local timezone (otherwise db.time() drifts
// by the offset between them and the sync time-window comparisons break).
timezone: "Z",
// MySQL has no boolean type - Saltcorn's Bool maps to tinyint(1). Coerce
// it back to a JS boolean so reads match postgres even where the type's
// own read is bypassed (joined fields, stored calculated JSON).
typeCast: boolTypeCast,
// migrations and some bootstrap SQL bundle several ";"-separated
// statements in one query() call, which pg/sqlite3 both tolerate -
// mysql2 needs this explicitly enabled to match.
multipleStatements: true,
});
// Every query in this codebase double-quotes identifiers (as Postgres and
// SQLite both do); ANSI_QUOTES makes MySQL accept that same syntax, so
// none of the "schema"."table" string-building already spread across
// saltcorn-data needs to change for MySQL. Similarly, "||" is used
// throughout as the ANSI string-concatenation operator (e.g. the ilike
// "'%' || ? || '%'" pattern in db-common/internal.ts) - by default MySQL
// treats "||" as logical OR instead, so PIPES_AS_CONCAT is required too.
newPool.on("connection", (connection: any) => {
connection.query(
"SET SESSION sql_mode = (SELECT CONCAT(@@sql_mode, ',ANSI_QUOTES,PIPES_AS_CONCAT'))",
);
connection.query("SET collation_connection = @@collation_server");
connection.query("SET time_zone = '+00:00'");
});
return newPool;
};
/**
* Close database connection
* @returns {Promise<void>}
*/
export const close = async (): Promise<void> => {
if (pool) await pool.end();
pool = null;
if (sessionPool) await sessionPool.end();
sessionPool = null;
};
/**
* Change connection (close connection and open new connection from connObj)
* @param {object} [connObj = {}] - connection object
* @returns {Promise<void>}
*/
export const changeConnection = async (
connObj: any = Object.create(null),
): Promise<void> => {
await close();
pool = buildPool(getConnectObject!(connObj));
};
export const begin = async (): Promise<void> => {
await query("BEGIN");
};
export const commit = async (): Promise<void> => {
await query("COMMIT");
};
export const rollback = async (): Promise<void> => {
await query("ROLLBACK");
};
const getMyClient = (selopts?: any): any => {
return selopts?.client || getRequestContext()?.client || pool;
};
/**
* Execute Select statement
* @param {string} tbl - table name
* @param {object} whereObj - where object
* @param {object} [selectopts = {}] - select options
* @returns {Promise<*>} return rows
*/
export const select = async (
tbl: string,
whereObj: Where,
selectopts: SelectOptions & { [key: string]: any } = Object.create(null),
): Promise<Row[]> => {
const { where, values } = mkWhere(whereObj);
const schema = selectopts.schema || getTenantSchema();
const qtbl = `"${schema}"."${sqlsanitize(tbl)}"`;
if (selectopts.tree_field && !(whereObj as any)?.[selectopts.tree_field]) {
const tf = sqlsanitize(selectopts.tree_field);
const pk = sqlsanitize(selectopts.pk_name || "id");
const ob =
typeof selectopts.orderBy === "string"
? sqlsanitize(selectopts.orderBy)
: pk;
const cmp = selectopts.orderDesc ? ">" : "<";
const rank = (alias: string, rootScope: boolean) =>
`(select count(*) from ${qtbl} sib where ${
rootScope ? `sib."${tf}" is null` : `sib."${tf}" = ${alias}."${tf}"`
} and (sib."${ob}" ${cmp} ${alias}."${ob}" or (sib."${ob}" = ${alias}."${ob}" and sib."${pk}" < ${alias}."${pk}")))`;
const anchorFields = selectopts.fields
? selectopts.fields.map((f) => `"${sqlsanitize(f)}"`).join(", ")
: "*";
const recurFields = selectopts.fields
? selectopts.fields.map((f) => `p."${sqlsanitize(f)}"`).join(", ")
: "p.*";
const finalFields = selectopts.fields
? `${selectopts.fields.map((f) => `"${sqlsanitize(f)}"`).join(", ")}, _level`
: "*";
const whereAnd = where ? `and ${where.replace(/^\s*where\s+/i, "")}` : "";
const treeSql = `WITH RECURSIVE _tree AS (
SELECT ${anchorFields}, 0 as _level,
CAST(LPAD(${rank("p", true)}, 10, '0') AS CHAR(2000)) as _sort_path
FROM ${qtbl} p
WHERE p."${tf}" IS NULL ${whereAnd}
UNION ALL
SELECT ${recurFields}, pt._level+1,
CONCAT(pt._sort_path, '.', LPAD(${rank("p", false)}, 10, '0'))
FROM ${qtbl} p
JOIN _tree pt ON p."${tf}" = pt."${pk}"
)
SELECT ${finalFields} FROM _tree ${where} ORDER BY _sort_path`;
const treeValues = where ? [...values, ...values] : values;
sql_log(treeSql, treeValues);
const [treeRows] = await getMyClient(selectopts).query(treeSql, treeValues);
return treeRows as Row[];
}
const sql = `SELECT ${
selectopts.fields ? selectopts.fields.join(", ") : `*`
} FROM ${qtbl} ${where} ${mkSelectOptions(selectopts, values)}`;
sql_log(sql, values);
const [rows] = await getMyClient(selectopts).query(sql, values);
return rows as Row[];
};
/**
* Reset the tenant's database (DROP DATABASE + CREATE DATABASE): the MySQL
* analogue of Postgres's schema drop/recreate under the one-database-per-
* tenant model.
* @param {string} schema - db/tenant name
* @returns {Promise<void>} no result
*/
export const drop_reset_schema = async (schema: string): Promise<void> => {
const name = sqlsanitize(schema);
const dropSql = `DROP DATABASE IF EXISTS "${name}";`;
const createSql = `CREATE DATABASE "${name}";`;
sql_log(dropSql);
await getMyClient().query(dropSql);
sql_log(createSql);
await getMyClient().query(createSql);
};
/**
* Get count of rows in table
* @param {string} - tbl - table name
* @param {object} - whereObj - where object
* @returns {Promise<number>} count of tables
*/
export const count = async (
tbl: string,
whereObj: Where,
opts?: SelectOptions & { [key: string]: any },
): Promise<number> => {
const { where, values } = mkWhere(whereObj);
const core_sql = `FROM "${opts?.schema || getTenantSchema()}"."${sqlsanitize(
tbl,
)}" ${where}`;
const sql = opts?.limit
? `SELECT count(*) AS count FROM (
SELECT 1 ${core_sql} limit ${+opts?.limit}) limited_count`
: `SELECT COUNT(*) AS count ${core_sql}`;
sql_log(sql, values);
const [rows]: any = await getMyClient(opts).query(sql, values);
const countValue = rows[0].count ?? Object.values(rows[0])[0];
return parseInt(String(countValue));
};
/**
* Get version of MySQL
* @param {boolean} short - if true return short version info else full version info
* @returns {Promise<string>} returns version
*/
export const getVersion = async (short?: boolean): Promise<string> => {
const sql = `SELECT VERSION() AS version;`;
sql_log(sql);
const [rows]: any = await getMyClient().query(sql);
const v = rows[0].version;
if (short) return v.split("-")[0];
return v;
};
/**
* Delete rows in table
* @param {string} tbl - table name
* @param {object} whereObj - where object
* @param {object} [opts = {}]
* @returns {Promise<object[]>} result of delete execution
*/
export const deleteWhere = async (
tbl: string,
whereObj: Where,
opts: { schema?: string; client?: any } = Object.create(null),
): Promise<any> => {
const { where, values } = mkWhere(whereObj);
const sql = `delete FROM "${opts.schema || getTenantSchema()}"."${sqlsanitize(
tbl,
)}" ${where}`;
sql_log(sql, values);
const [result] = await getMyClient(opts).query(sql, values);
return result;
};
export const truncate = async (tbl: string): Promise<any> => {
const sql = `truncate "${getTenantSchema()}"."${sqlsanitize(tbl)}"`;
sql_log(sql, []);
const [result] = await getMyClient().query(sql, []);
return result;
};
/**
* Insert rows into table
* @param {string} tbl - table name
* @param {object} obj - columns names and data
* @param {object} [opts = {}] - columns attributes
* @returns {Promise<any>} returns primary key column value, unless opts.noid
*/
export const insert = async (
tbl: string,
obj: Row,
opts: {
schema?: string;
onConflictDoNothing?: boolean;
noid?: boolean;
pk_name?: string;
client?: any;
} = Object.create(null),
): Promise<any> => {
const kvs = Object.entries(obj);
const fnameList = kvs.map(([k, v]) => `"${sqlsanitize(k)}"`).join();
const schema = opts.schema || getTenantSchema();
const client = getMyClient(opts);
var valPosList: string[] = [];
var valList: any[] = [];
kvs.forEach(([k, v]: [string, any]) => {
if (v && v.next_version_by_id) {
valList.push(v.next_version_by_id);
valPosList.push(
`coalesce((select max(\`_version\`) from (select \`_version\` from "${schema}"."${sqlsanitize(
tbl,
)}" where "${v.pk_name || "id"}"=?) as _h), 0)+1`,
);
} else {
valList.push(mkVal(v));
valPosList.push("?");
}
});
const insertIgnore = opts.onConflictDoNothing ? "ignore " : "";
const sql =
valPosList.length > 0
? `insert ${insertIgnore}into "${schema}"."${sqlsanitize(
tbl,
)}"(${fnameList}) values(${valPosList.join()})`
: `insert ${insertIgnore}into "${schema}"."${sqlsanitize(
tbl,
)}" () values()`;
sql_log(sql, valList);
const [result]: any = await client.query(sql, valList);
if (opts.noid) return;
else if (opts.onConflictDoNothing && !result.affectedRows) return;
else return result.insertId || obj[opts.pk_name || "id"];
};
/**
* Update table records
* @param {string} tbl - table name
* @param {object} obj - columns names and data
* @param {number|undefined} id - id of record (primary key column value)
* @param {object} [opts = {}] - columns attributes
* @returns {Promise<void>} no result
*/
export const update = async (
tbl: string,
obj: Row,
id: any,
opts: { schema?: string; pk_name?: string; client?: any } = Object.create(
null,
),
): Promise<void> => {
const kvs = Object.entries(obj);
if (kvs.length === 0) return;
const assigns = kvs.map(([k, v]) => `"${sqlsanitize(k)}"=?`).join();
let valList = kvs.map(([k, v]) => mkVal(v));
let whereS;
if (id && typeof id == "object") {
const whereStrs: string[] = [];
Object.keys(id).forEach((k) => {
valList.push(id[k]);
whereStrs.push(`"${k}"=?`);
});
whereS = whereStrs.join(" and ");
} else {
valList.push(id === "undefined" ? obj[opts.pk_name || "id"] : id);
whereS = `${ppPK(opts.pk_name)}=?`;
}
const q = `update "${opts.schema || getTenantSchema()}"."${sqlsanitize(
tbl,
)}" set ${assigns} where ${whereS}`;
sql_log(q, valList);
await getMyClient(opts).query(q, valList);
};
/**
* Update table records matching a where clause
* @param {string} tbl - table name
* @param {object} obj - columns names and data
* @param {object} whereObj - where object
* @param {object} opts - can contain a db client for transactions
* @returns {Promise<void>} no result
*/
export const updateWhere = async (
tbl: string,
obj: Row,
whereObj: Where,
opts: { client?: any } = Object.create(null),
): Promise<void> => {
const kvs = Object.entries(obj);
if (kvs.length === 0) return;
const { where, values } = mkWhere(whereObj);
const assigns = kvs.map(([k, v]) => `"${sqlsanitize(k)}"=?`).join();
let valList = [...kvs.map(([k, v]) => mkVal(v)), ...values];
const q = `update "${getTenantSchema()}"."${sqlsanitize(
tbl,
)}" set ${assigns} ${where}`;
sql_log(q, valList);
await getMyClient(opts).query(q, valList);
};
/**
* Select one record
* @param {string} tbl - table name
* @param {object} where - where object
* @param {object} [selectopts = {}] - select options
* @returns {Promise<object>} return first record from sql result
* @throws {Error}
*/
export const selectOne = async (
tbl: string,
where: Where,
selectopts: SelectOptions = Object.create(null),
): Promise<Row> => {
const rows = await select(tbl, where, { ...selectopts, limit: 1 });
if (rows.length === 0) {
const w = mkWhere(where);
throw new Error(`no ${tbl} ${w.where} are ${w.values}`);
} else return rows[0];
};
/**
* Select one record or null if no records
* @param {string} tbl - table name
* @param {object} where - where object
* @param {object} [selectopts = {}] - select options
* @returns {Promise<null|object>} - null if no record or first record data
*/
export const selectMaybeOne = async (
tbl: string,
where: Where,
selectopts: SelectOptions = Object.create(null),
): Promise<Row | null> => {
const rows = await select(tbl, where, { ...selectopts, limit: 1 });
if (rows.length === 0) return null;
else return rows[0];
};
/**
* Open db connection
* @returns {Promise<*>} db connection object
*/
export const getClient = async (): Promise<PoolConnection> =>
await pool!.getConnection();
/**
* Reset auto-increment counter (the MySQL analogue of Postgres's sequence reset)
* @param {string} tblname - table name
* @returns {Promise<void>} no result
*/
export const reset_sequence = async (
tblname: string,
pkname: string = "id",
): Promise<void> => {
const schema = getTenantSchema();
const selSql = `SELECT COALESCE(MAX("${sqlsanitize(
pkname,
)}"), 0) + 1 AS next_id FROM "${schema}"."${sqlsanitize(tblname)}"`;
const [rows]: any = await getMyClient().query(selSql);
const nextId = rows[0].next_id;
const alterSql = `ALTER TABLE "${schema}"."${sqlsanitize(
tblname,
)}" AUTO_INCREMENT = ${+nextId};`;
sql_log(alterSql);
await getMyClient().query(alterSql);
};
/*
* MySQL cannot index/unique-constrain a TEXT/BLOB column without an explicit
* key prefix length ("BLOB/TEXT column used in key specification without a
* key length"). Saltcorn's String type maps to TEXT, so index/unique targets
* frequently are TEXT columns - look their types up and add a prefix. 191
* utf8mb4 chars fits the classic 767-byte InnoDB key limit.
*/
const keyColumnExprs = async (
table_name: string,
field_names: string[],
): Promise<string> => {
const [rows]: any = await getMyClient().query(
`SHOW COLUMNS FROM "${getTenantSchema()}"."${sqlsanitize(table_name)}"`,
);
const typeOf: Record<string, string> = {};
for (const r of rows)
typeOf[r.Field || r.field] = (r.Type || r.type || "").toLowerCase();
const strLen = (t: string): number | null => {
if (/text|blob/.test(t)) return Infinity;
const m = t.match(/^(?:var)?char\((\d+)\)/);
return m ? parseInt(m[1], 10) : null; // null => not a string column
};
const cols = field_names.map((f) => {
const s = sqlsanitize(f);
return { s, len: strLen(typeOf[s] || "") };
});
const strCols = cols.filter((c) => c.len !== null);
const otherBytes = (cols.length - strCols.length) * 8;
const budgetChars = Math.max(1, Math.floor((3000 - otherBytes) / 4));
const perColChars = Math.max(
1,
Math.floor(budgetChars / Math.max(1, strCols.length)),
);
return cols
.map(({ s, len }) => {
if (len === null) return `"${s}"`;
const cap = Math.min(len, perColChars);
return cap < len ? `"${s}"(${cap})` : `"${s}"`;
})
.join(",");
};
/**
* Add unique constraint
* @param {string} table_name - table name
* @param {string[]} field_names - list of columns (members of constraint)
* @returns {Promise<void>} no result
*/
export const add_unique_constraint = async (
table_name: string,
field_names: string[],
): Promise<void> => {
const sql = `alter table "${getTenantSchema()}"."${sqlsanitize(
table_name,
)}" add CONSTRAINT "${sqlsanitize(table_name)}_${field_names
.map((f) => sqlsanitize(f))
.join("_")}_unique" UNIQUE (${await keyColumnExprs(
table_name,
field_names,
)});`;
sql_log(sql);
await getMyClient().query(sql);
};
/**
* Drop unique constraint
* @param {string} table_name - table name
* @param {string[]} field_names - list of columns (members of constraint)
* @returns {Promise<void>} no results
*/
export const drop_unique_constraint = async (
table_name: string,
field_names: string[],
): Promise<void> => {
const sql = `alter table "${getTenantSchema()}"."${sqlsanitize(
table_name,
)}" drop index "${sqlsanitize(table_name)}_${field_names
.map((f) => sqlsanitize(f))
.join("_")}_unique";`;
sql_log(sql);
try {
await getMyClient().query(sql);
} catch (e: any) {
// MySQL has no "DROP INDEX IF EXISTS" equivalent for this form - swallow
// "unknown key" errors to match the IF EXISTS semantics postgres.ts has
if (!/check that column\/key exists|doesn't exist/i.test(e?.message || ""))
throw e;
}
};
/**
* Add index
* @param {string} table_name - table name
* @param {string} field_name - column name
* @returns {Promise<void>} no result
*/
export const add_index = async (
table_name: string,
field_name: string,
): Promise<void> => {
const sql = `create index "${sqlsanitize(table_name)}_${sqlsanitize(
field_name,
)}_index" on "${getTenantSchema()}"."${sqlsanitize(
table_name,
)}" (${await keyColumnExprs(table_name, [field_name])});`;
sql_log(sql);
await getMyClient().query(sql);
};
/**
* Add Full-text search index
*
* Not currently implemented for MySQL - Saltcorn's FTS expression is a
* computed coalesce/concat across possibly several columns
* (ftsFieldsSqlExpr), and MySQL FULLTEXT indexes can only be built on real
* column lists, not arbitrary expressions. Search still works via the LIKE
* fallback in mysqlPlaceHolderStack().ftsWhereClause - it's just not backed
* by an index. Revisit once MySQL-aware FTS index creation is added
* alongside the generic table-DDL work.
* @returns {Promise<void>} no result
*/
export const add_fts_index = async (
table_name: string,
field_expression: string,
language?: string,
disable_fts?: boolean,
): Promise<void> => {
console.warn(
`add_fts_index: full-text indexing is not yet implemented for MySQL (table ${table_name}) - search will use LIKE without an index`,
);
};
export const drop_fts_index = async (table_name: string): Promise<void> => {};
/**
* Drop index
* @param {string} table_name - table name
* @param {string} field_name - column name
* @returns {Promise<void>} no result
*/
export const drop_index = async (
table_name: string,
field_name: string,
): Promise<void> => {
const sql = `drop index "${sqlsanitize(table_name)}_${sqlsanitize(
field_name,
)}_index" on "${getTenantSchema()}"."${sqlsanitize(table_name)}";`;
sql_log(sql);
await getMyClient().query(sql);
};
/**
* Bulk-load rows from a CSV stream into a table.
*
* MySQL has no streaming COPY equivalent that doesn't require server-side
* file access (LOAD DATA INFILE needs FILE privilege / secure_file_priv
* configuration that isn't guaranteed to be available), so this reads the
* stream and issues batched multi-row INSERTs instead. Slower than
* Postgres's COPY for very large files, but correct.
* @param {object} fileStream - file stream (CSV, with header row)
* @param {string} tableName - table name
* @param {string[]} fieldNames - list of columns
* @param {object} client - db connection
* @returns {Promise<void>} no result
*/
export const copyFrom = async (
fileStream: any,
tableName: string,
fieldNames: string[],
client: any,
): Promise<any> => {
const { parse } = require("csv-parse/sync");
const chunks: Buffer[] = [];
for await (const chunk of fileStream) chunks.push(chunk);
const text = Buffer.concat(chunks).toString("utf8");
const records: string[][] = parse(text, { columns: false, from_line: 2 });
const schema = getTenantSchema();
const fnameList = fieldNames.map((f) => `"${sqlsanitize(f)}"`).join();
const BATCH = 500;
for (let i = 0; i < records.length; i += BATCH) {
const batch = records.slice(i, i + BATCH);
if (batch.length === 0) continue;
const valPosList = batch
.map((row) => `(${row.map(() => "?").join(",")})`)
.join(",");
const sql = `insert into "${schema}"."${sqlsanitize(
tableName,
)}"(${fnameList}) values ${valPosList}`;
await (client || getMyClient()).query(sql, batch.flat());
}
};
export const copyToJson = async (
fileStream: any,
tableName: string,
client?: any,
): Promise<any> => {
const { Readable } = require("stream");
const { pipeline } = require("stream/promises");
const schema = getTenantSchema();
const sql = `SELECT * FROM "${schema}"."${sqlsanitize(tableName)}"`;
const [rows]: any = await (client || getMyClient()).query(sql);
await pipeline(
Readable.from(rows.map((row: any) => JSON.stringify(row) + ",\n")),
fileStream,
);
};
export const slugify = (s: string): string =>
s
.toLowerCase()
.replace(/\s+/g, "-")
.replace(/[^\w-]/g, "");
export const time = async (): Promise<Date> => {
const [rows]: any = await getMyClient().query("select now(3) as now");
return new Date(rows[0].now);
};
export const listTables = async (): Promise<{ name: string }[]> => {
const [rows]: any = await getMyClient().query(
`SELECT table_name FROM information_schema.tables WHERE table_schema = ?`,
[getTenantSchema()],
);
return rows.map((row: any) => ({ name: row.table_name || row.TABLE_NAME }));
};
export const listUserDefinedTables = async (): Promise<{ name: string }[]> => {
const [rows]: any = await getMyClient().query(
`SELECT table_name FROM information_schema.tables WHERE table_schema = ? AND table_name NOT LIKE '\\_sc\\_%'`,
[getTenantSchema()],
);
return rows.map((row: any) => ({ name: row.table_name || row.TABLE_NAME }));
};
export const listScTables = async (): Promise<{ name: string }[]> => {
const [rows]: any = await getMyClient().query(
`SELECT table_name FROM information_schema.tables WHERE table_schema = ? AND table_name LIKE '\\_sc\\_%'`,
[getTenantSchema()],
);
return rows.map((row: any) => ({ name: row.table_name || row.TABLE_NAME }));
};
/**
* Run work after COMMIT, so slow or unrollbackable effects (sending email,
* calling a remote service) don't hold the transaction's connection open.
* Dropped on rollback; errors are logged, not thrown. Runs inline if there is
* no transaction, or no async context (runWithTenant) to defer against.
*
* @param f work to run after COMMIT
*/
export const afterCommit = async (f: () => Promise<void>): Promise<void> => {
const reqCon = getRequestContext();
if (!reqCon?.inTransaction) return await f();
if (!reqCon.afterCommit) reqCon.afterCommit = [];
reqCon.afterCommit.push(f);
};
const runAfterCommit = async (queue?: Array<() => Promise<void>>) => {
for (const f of queue || []) {
try {
await f();
} catch (e) {
console.error("Error in afterCommit callback", e);
}
}
};
/* rules of using this:
- no try catch inside unless you rethrow: wouldnt roll back
- no state.refresh_*() inside: other works wouldnt see updates as they are in transactioon
- you can use state.refresh_*(true) for update on own worker only
Note: unlike Postgres, MySQL's DDL statements are not transactional (they
implicitly commit any open transaction). withTransaction() is still correct
for the DML this driver issues today (select/insert/update/delete); it would
NOT roll back DDL, same caveat that applies to the (deferred) generic
table-DDL work.
*/
export const withTransaction = async (
f: (rollback: () => Promise<void>) => Promise<any>,
onError?: (e: Error) => any,
): Promise<any> => {
const client = await getClient();
const reqCon = getRequestContext();
if (reqCon) {
reqCon.client = client;
reqCon.inTransaction = true;
}
sql_log("BEGIN;");
await client.query("BEGIN;");
let aborted = false;
let committed = false;
const rollback = async () => {
aborted = true;
sql_log("ROLLBACK;");
await client.query("ROLLBACK;");
};
try {
const result = await f(rollback);
if (!aborted) {
sql_log("COMMIT;");
await client.query("COMMIT;");
committed = true;
}
return result;
} catch (error) {
if (!aborted) {
sql_log("ROLLBACK;");
await client.query("ROLLBACK;");
}
if (onError) return onError(error as Error);
else throw error;
} finally {
const queued = reqCon?.afterCommit;
if (reqCon) {
reqCon.client = null;
reqCon.inTransaction = false;
reqCon.afterCommit = undefined;
}
client.release();
// only once the client is back in the pool, and never after a rollback
if (committed) await runAfterCommit(queued);
}
};
export const commitAndBeginNewTransaction = async (): Promise<void> => {
const client = await getMyClient();
sql_log("COMMIT;");
await client.query("COMMIT;");
sql_log("BEGIN;");
await client.query("BEGIN;");
};
export const tryCatchInTransaction = async (
f: () => Promise<any>,
onError?: (e: Error) => any,
): Promise<any> => {
const rndid = Math.floor(Math.random() * 16777215).toString(16);
const reqCon = getRequestContext();
const savepoint = async (stmt: string) => {
if (!reqCon?.client) return;
try {
await query(stmt);
} catch (e: any) {
if (e?.code !== "ER_SP_DOES_NOT_EXIST") throw e;
}
};
if (reqCon?.client) await query(`SAVEPOINT sp${rndid}`);
try {
return await f();
} catch (error) {
await savepoint(`ROLLBACK TO SAVEPOINT sp${rndid}`);
if (onError) return await onError(error as Error);
} finally {
await savepoint(`RELEASE SAVEPOINT sp${rndid}`);
}
};
/**
* Should be used for code that is sometimes called from within a withTransaction block
* and sometimes not.
* @param {Function} f logic to execute
* @param {Function} onError error handler
* @returns
*/
export const openOrUseTransaction = async (
f: (rollback?: () => Promise<void>) => Promise<any>,
onError?: (e: Error) => any,
): Promise<any> => {
const reqCon = getRequestContext();
if (reqCon?.client) return await f();
else return await withTransaction(f, onError);