| 233 | } |
| 234 | |
| 235 | async function begin(options, fn) { |
| 236 | !fn && (fn = options, options = '') |
| 237 | const queries = Queue() |
| 238 | let savepoints = 0 |
| 239 | , connection |
| 240 | , prepare = null |
| 241 | |
| 242 | try { |
| 243 | await sql.unsafe('begin ' + options.replace(/[^a-z ]/ig, ''), [], { onexecute }).execute() |
| 244 | return await Promise.race([ |
| 245 | scope(connection, fn), |
| 246 | new Promise((_, reject) => connection.onclose = reject) |
| 247 | ]) |
| 248 | } catch (error) { |
| 249 | throw error |
| 250 | } |
| 251 | |
| 252 | async function scope(c, fn, name) { |
| 253 | const sql = Sql(handler) |
| 254 | sql.savepoint = savepoint |
| 255 | sql.prepare = x => prepare = x.replace(/[^a-z0-9$-_. ]/gi) |
| 256 | let uncaughtError |
| 257 | , result |
| 258 | |
| 259 | name && await sql`savepoint ${ sql(name) }` |
| 260 | try { |
| 261 | result = await new Promise((resolve, reject) => { |
| 262 | const x = fn(sql) |
| 263 | Promise.resolve(Array.isArray(x) ? Promise.all(x) : x).then(resolve, reject) |
| 264 | }) |
| 265 | |
| 266 | if (uncaughtError) |
| 267 | throw uncaughtError |
| 268 | } catch (e) { |
| 269 | await (name |
| 270 | ? sql`rollback to ${ sql(name) }` |
| 271 | : sql`rollback` |
| 272 | ) |
| 273 | throw e instanceof PostgresError && e.code === '25P02' && uncaughtError || e |
| 274 | } |
| 275 | |
| 276 | if (!name) { |
| 277 | prepare |
| 278 | ? await sql`prepare transaction '${ sql.unsafe(prepare) }'` |
| 279 | : await sql`commit` |
| 280 | } |
| 281 | |
| 282 | return result |
| 283 | |
| 284 | function savepoint(name, fn) { |
| 285 | if (name && Array.isArray(name.raw)) |
| 286 | return savepoint(sql => sql.apply(sql, arguments)) |
| 287 | |
| 288 | arguments.length === 1 && (fn = name, name = null) |
| 289 | return scope(c, fn, 's' + savepoints++ + (name ? '_' + name : '')) |
| 290 | } |
| 291 | |
| 292 | function handler(q) { |