ReviewOS
Reload

also looking at this

stacks/bun-query-builder

fix(client): a builder's writes belong to the builder's own connection

#1065
Merged glennmichael123 wants to merge fix/builder-local-connection into main
2 files +161 -29
packages/bun-query-builder/src/client.tsmodified+49-29
@@ -2740,6 +2740,28 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
27402740 // Single boundary cast: `state.sql` is `any` (allows mock/tx injection) and
27412741 // getOrCreateBunSql() returns Bun's `SQL`; both satisfy DriverConnection. With
27422742 // `_sql` typed, the downstream `.unsafe(...)` calls no longer need casts (#1044).
2743 //
2744 // EVERY statement this builder executes must go through `_sql`, never the
2745 // module-global `bunSql`. They are the same object only for a plain builder;
2746 // the moment one is created with its own connection — `db.transaction()`,
2747 // `savepoint()`, `beginDistributed()`, `reserve()`, or an explicit
2748 // `createQueryBuilder({ sql })` — they are two different connections, and the
2749 // global is a DIFFERENT pooled session from the one the builder's own reads
2750 // and writes use.
2751 //
2752 // That is not a cosmetic distinction. A write issued on the global while the
2753 // caller is inside `db.transaction()` autocommits on its own session, so it
2754 // outlives a ROLLBACK: the transaction fails, the caller sees the error, and
2755 // the row is still there. A read issued on the global cannot see the
2756 // transaction's uncommitted rows, so read-then-decide logic acts on stale
2757 // data. With `pool: { max: 1 }` it does not even get that far — the call
2758 // blocks forever waiting for the single connection its own transaction is
2759 // holding. All three are silent; none produces a driver error.
2760 //
2761 // The exceptions are the pool-lifecycle methods (`reserve`, `close`), which
2762 // act on the pool itself rather than on a session, and the two-phase
2763 // `commitDistributed`/`rollbackDistributed`, which by design run outside the
2764 // transaction that prepared them.
27432765 const _sql: DriverConnection = (state?.sql ?? getOrCreateBunSql()) as unknown as DriverConnection
27442766 const meta = state?.meta
27452767 const schema = state?.schema
@@ -6973,7 +6995,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
69736995 let hash = 7
69746996 for (let i = 0; i < s.length; i++) hash = (hash * 31 + s.charCodeAt(i)) | 0
69756997 const k = typeof key === 'number' ? key : Math.abs(hash)
6976 const q = bunSql`SELECT pg_advisory_lock(${k})`
6998 const q = _sql`SELECT pg_advisory_lock(${k})`
69776999 await runWithHooks<any[]>(q, 'raw')
69787000 return
69797001 }
@@ -6981,7 +7003,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
69817003 // MySQL has `GET_LOCK(name, timeout)`. Wait indefinitely
69827004 // (timeout=-1) to match Postgres `pg_advisory_lock` semantics.
69837005 const lockName = `bqb:${String(key)}`
6984 const q = bunSql`SELECT GET_LOCK(${lockName}, -1) AS ok`
7006 const q = _sql`SELECT GET_LOCK(${lockName}, -1) AS ok`
69857007 await runWithHooks<any[]>(q, 'raw')
69867008 return
69877009 }
@@ -6997,7 +7019,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
69977019 let hash = 7
69987020 for (let i = 0; i < s.length; i++) hash = (hash * 31 + s.charCodeAt(i)) | 0
69997021 const k = typeof key === 'number' ? key : Math.abs(hash)
7000 const q = bunSql`SELECT pg_try_advisory_lock(${k}) as ok`
7022 const q = _sql`SELECT pg_try_advisory_lock(${k}) as ok`
70017023 const rows = await runWithHooks<any[]>(q, 'raw')
70027024 return Boolean(rows?.[0]?.ok)
70037025 }
@@ -7005,7 +7027,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
70057027 // MySQL `GET_LOCK(name, 0)` returns 1 immediately if free, 0
70067028 // if held by another connection.
70077029 const lockName = `bqb:${String(key)}`
7008 const q = bunSql`SELECT GET_LOCK(${lockName}, 0) AS ok`
7030 const q = _sql`SELECT GET_LOCK(${lockName}, 0) AS ok`
70097031 const rows = await runWithHooks<any[]>(q, 'raw')
70107032 return Number(rows?.[0]?.ok) === 1
70117033 }
@@ -7023,7 +7045,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
70237045 return (_sql as any).unsafe(query, params)
70247046 },
70257047 file(path: string, params?: any[]) {
7026 return (bunSql as any).file(path, params)
7048 return (_sql as any).file(path, params)
70277049 },
70287050 async reserve() {
70297051 const reserved = await (bunSql as any).reserve()
@@ -7058,7 +7080,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
70587080 },
70597081 async ping() {
70607082 try {
7061 const q = bunSql`SELECT 1`
7083 const q = _sql`SELECT 1`
70627084 await runWithHooks<any[]>(q, 'select')
70637085 return true
70647086 }
renamed throughout · 34 hidden lines
@@ -7242,28 +7264,28 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti renamed throughout
72427264 const sqlText = isMysqlLike(config.dialect)
72437265 ? `INSERT IGNORE INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`
72447266 : `INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql} ON CONFLICT DO NOTHING`
7245 return (bunSql.unsafe(sqlText, params) as any).execute()
7267 return (_sql.unsafe(sqlText, params) as any).execute()
72467268 },
72477269 async insertGetId(table, values, idColumn = 'id' as any) {
72487270 const { colsSql, valuesSql, params } = buildInsertClause([values as Record<string, any>])
72497271 const tbl = quoteInsertIdent(String(table))
72507272 if (isMysqlLike(config.dialect)) {
72517273 // MySQL has no RETURNING — insert then read LAST_INSERT_ID().
7252 await (bunSql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7253 const [row] = await (bunSql.unsafe(`SELECT LAST_INSERT_ID() as id`) as any).execute()
7274 await (_sql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7275 const [row] = await (_sql.unsafe(`SELECT LAST_INSERT_ID() as id`) as any).execute()
72547276 return row?.id
72557277 }
72567278 if (config.dialect === 'sqlite') {
72577279 // The bun:sqlite wrapper returns { changes, lastInsertRowid } rather
72587280 // than RETURNING rows, so read the rowid directly.
7259 const res = await (bunSql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7281 const res = await (_sql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
72607282 if (res?.lastInsertRowid != null)
72617283 return res.lastInsertRowid
7262 const [row] = await (bunSql.unsafe(`SELECT last_insert_rowid() as id`) as any).execute()
7284 const [row] = await (_sql.unsafe(`SELECT last_insert_rowid() as id`) as any).execute()
72637285 return row?.id
72647286 }
72657287 // Postgres supports RETURNING.
7266 const [row] = await (bunSql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql} RETURNING ${quoteInsertIdent(String(idColumn))} as id`, params) as any).execute()
7288 const [row] = await (_sql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql} RETURNING ${quoteInsertIdent(String(idColumn))} as id`, params) as any).execute()
72677289 return row?.id
72687290 },
72697291 async updateOrInsert(table, match, values) {
renamed throughout · 21 hidden lines
@@ -7272,18 +7294,18 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti renamed throughout
72727294 let idx = 1
72737295 const whereSql = matchKeys.map(k => `${quoteInsertIdent(k)} = ${getPlaceholder(idx++)}`).join(' AND ')
72747296 const whereParams = matchKeys.map(k => (match as any)[k])
7275 const existsRows = await (bunSql.unsafe(`SELECT 1 FROM ${tbl} WHERE ${whereSql} LIMIT 1`, whereParams) as any).execute()
7297 const existsRows = await (_sql.unsafe(`SELECT 1 FROM ${tbl} WHERE ${whereSql} LIMIT 1`, whereParams) as any).execute()
72767298 if ((existsRows as any[]).length) {
72777299 const setKeys = Object.keys(values)
72787300 let i = 1
72797301 const setSql = setKeys.map(k => `${quoteInsertIdent(k)} = ${getPlaceholder(i++)}`).join(', ')
72807302 const whereSql2 = matchKeys.map(k => `${quoteInsertIdent(k)} = ${getPlaceholder(i++)}`).join(' AND ')
72817303 const params = [...setKeys.map(k => (values as any)[k]), ...matchKeys.map(k => (match as any)[k])]
7282 await (bunSql.unsafe(`UPDATE ${tbl} SET ${setSql} WHERE ${whereSql2}`, params) as any).execute()
7304 await (_sql.unsafe(`UPDATE ${tbl} SET ${setSql} WHERE ${whereSql2}`, params) as any).execute()
72837305 return true
72847306 }
72857307 const { colsSql, valuesSql, params } = buildInsertClause([{ ...match, ...values } as Record<string, any>])
7286 await (bunSql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7308 await (_sql.unsafe(`INSERT INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
72877309 return true
72887310 },
72897311 async upsert(table, rows, conflictColumns, mergeColumns) {
renamed throughout · 20 hidden lines
@@ -7301,16 +7323,16 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti renamed throughout
73017323 // "do nothing" form (an empty SET is a syntax error). See #1035, #1052.
73027324 if (isMysqlLike(config.dialect)) {
73037325 if (setCols.length === 0)
7304 return (bunSql.unsafe(`INSERT IGNORE INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
7326 return (_sql.unsafe(`INSERT IGNORE INTO ${tbl} (${colsSql}) VALUES ${valuesSql}`, params) as any).execute()
73057327 const updateList = setCols.map(c => `${quoteInsertIdent(c)} = VALUES(${quoteInsertIdent(c)})`).join(', ')
7306 return (bunSql.unsafe(`${insert} ON DUPLICATE KEY UPDATE ${updateList}`, params) as any).execute()
7328 return (_sql.unsafe(`${insert} ON DUPLICATE KEY UPDATE ${updateList}`, params) as any).execute()
73077329 }
73087330
73097331 const targets = targetCols.map(quoteInsertIdent).join(', ')
73107332 if (setCols.length === 0)
7311 return (bunSql.unsafe(`${insert} ON CONFLICT (${targets}) DO NOTHING`, params) as any).execute()
7333 return (_sql.unsafe(`${insert} ON CONFLICT (${targets}) DO NOTHING`, params) as any).execute()
73127334 const updateList = setCols.map(c => `${quoteInsertIdent(c)} = EXCLUDED.${quoteInsertIdent(c)}`).join(', ')
7313 return (bunSql.unsafe(`${insert} ON CONFLICT (${targets}) DO UPDATE SET ${updateList}`, params) as any).execute()
7335 return (_sql.unsafe(`${insert} ON CONFLICT (${targets}) DO UPDATE SET ${updateList}`, params) as any).execute()
73147336 },
73157337 async save(table, values) {
73167338 const pk = meta?.primaryKeys[String(table)] ?? 'id'
@@ -7361,7 +7383,7 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
73617383 const start = Date.now()
73627384 try {
73637385 config.hooks?.onQueryStart?.({ sql: query, kind: 'raw' })
7364 const res = await (bunSql as any).unsafe(query)
7386 const res = await (_sql as any).unsafe(query)
73657387 config.hooks?.onQueryEnd?.({ sql: query, durationMs: Date.now() - start, kind: 'raw' })
73667388 return res
73677389 }
@@ -7383,11 +7405,9 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
73837405
73847406 if (config.dialect === 'postgres') {
73857407 // For PostgreSQL, use RETURNING to get the ID, then fetch the full row
7386 const q = bunSql`INSERT INTO ${bunSql(String(table))} ${bunSql(values as any)} RETURNING ${bunSql(String(pk))} as id`
7408 const q = _sql`INSERT INTO ${_sql(String(table))} ${_sql(values as any)} RETURNING ${_sql(String(pk))} as id`
73877409 const [result] = await q.execute()
73887410
7389 console.log('resultId', result)
7390
73917411 if (!result?.id) {
73927412 console.error(`create() failed to get insert ID for table ${String(table)}`)
73937413 console.error('Inserted values:', values)
@@ -7600,28 +7620,28 @@ export function createQueryBuilder<DB extends DatabaseSchema<any>>(state?: Parti
76007620 return await (this as any).create(table, { ...(match as any), ...(values as any) })
76017621 },
76027622 async count(table, column) {
7603 const col = column ? bunSql(String(column)) : bunSql`*`
7604 const q = bunSql`SELECT COUNT(${col}) as c FROM ${bunSql(String(table))}`
7623 const col = column ? _sql(String(column)) : _sql`*`
7624 const q = _sql`SELECT COUNT(${col}) as c FROM ${_sql(String(table))}`
76057625 const [row] = await (q as any).execute()
76067626 return Number((row?.c ?? 0) as any)
76077627 },
76087628 async sum(table, column) {
7609 const q = bunSql`SELECT SUM(${bunSql(String(column))}) as s FROM ${bunSql(String(table))}`
7629 const q = _sql`SELECT SUM(${_sql(String(column))}) as s FROM ${_sql(String(table))}`
76107630 const [row] = await (q as any).execute()
76117631 return Number((row?.s ?? 0) as any)
76127632 },
76137633 async avg(table, column) {
7614 const q = bunSql`SELECT AVG(${bunSql(String(column))}) as a FROM ${bunSql(String(table))}`
7634 const q = _sql`SELECT AVG(${_sql(String(column))}) as a FROM ${_sql(String(table))}`
76157635 const [row] = await (q as any).execute()
76167636 return Number((row?.a ?? 0) as any)
76177637 },
76187638 async min(table, column) {
7619 const q = bunSql`SELECT MIN(${bunSql(String(column))}) as m FROM ${bunSql(String(table))}`
7639 const q = _sql`SELECT MIN(${_sql(String(column))}) as m FROM ${_sql(String(table))}`
76207640 const [row] = await (q as any).execute()
76217641 return (row?.m as any)
76227642 },
76237643 async max(table, column) {
7624 const q = bunSql`SELECT MAX(${bunSql(String(column))}) as m FROM ${bunSql(String(table))}`
7644 const q = _sql`SELECT MAX(${_sql(String(column))}) as m FROM ${_sql(String(table))}`
76257645 const [row] = await (q as any).execute()
76267646 return (row?.m as any)
76277647 },