objective categorical abstract machine language personal data server
Switch to connection pooling
+142
-118
+22
-25
pegasus/lib/data_store.ml
+22
-25
pegasus/lib/data_store.ml
···
218
218
{sql| INSERT INTO revoked_tokens (did, jti, revoked_at) VALUES (%string{did}, %string{jti}, %int{now}) |sql}]
219
219
end
220
220
221
-
type t = (module Rapper_helper.CONNECTION)
221
+
type t = Util.caqti_pool
222
222
223
223
let connect ?create ?write () : t Lwt.t =
224
224
Util.connect_sqlite ?create ?write Util.Constants.pegasus_db_location
225
225
226
-
let init conn : unit Lwt.t = unwrap @@ Queries.create_tables conn
226
+
let init conn : unit Lwt.t = Util.use_pool conn Queries.create_tables
227
227
228
228
let create_actor ~did ~handle ~email ~password ~signing_key conn =
229
229
let password_hash = Bcrypt.hash password |> Bcrypt.string_of_hash in
230
230
let now = Util.now_ms () in
231
-
let$! () =
232
-
Queries.create_actor ~did ~handle ~email ~password_hash ~signing_key
233
-
~created_at:now
234
-
~preferences:(Yojson.Safe.from_string "{}")
235
-
conn
236
-
in
237
-
Lwt.return_unit
231
+
Util.use_pool conn
232
+
@@ Queries.create_actor ~did ~handle ~email ~password_hash ~signing_key
233
+
~created_at:now
234
+
~preferences:(Yojson.Safe.from_string "{}")
238
235
239
236
let get_actor_by_identifier id conn =
240
-
unwrap @@ Queries.get_actor_by_identifier ~id conn
237
+
Util.use_pool conn @@ Queries.get_actor_by_identifier ~id
241
238
242
239
let update_actor_handle ~did ~handle conn =
243
-
unwrap @@ Queries.update_actor_handle ~did ~handle conn
240
+
Util.use_pool conn @@ Queries.update_actor_handle ~did ~handle
244
241
245
242
let try_login ~id ~password conn =
246
243
match%lwt get_actor_by_identifier id conn with
···
255
252
Lwt.return_none )
256
253
257
254
let list_actors ?(cursor = "") ?(limit = 100) conn =
258
-
unwrap @@ Queries.list_actors ~cursor ~limit conn
255
+
Util.use_pool conn @@ Queries.list_actors ~cursor ~limit
259
256
260
257
let put_preferences ~did ~prefs conn =
261
-
unwrap @@ Queries.put_preferences ~did ~preferences:prefs conn
258
+
Util.use_pool conn @@ Queries.put_preferences ~did ~preferences:prefs
262
259
263
260
(* invite codes *)
264
261
let create_invite ~code ~did ~remaining conn =
265
-
unwrap @@ Queries.create_invite ~code ~did ~remaining conn
262
+
Util.use_pool conn @@ Queries.create_invite ~code ~did ~remaining
266
263
267
-
let get_invite ~code conn = unwrap @@ Queries.get_invite ~code conn
264
+
let get_invite ~code conn = Util.use_pool conn @@ Queries.get_invite ~code
268
265
269
-
let use_invite ~code conn = unwrap @@ Queries.use_invite ~code conn
266
+
let use_invite ~code conn = Util.use_pool conn @@ Queries.use_invite ~code
270
267
271
268
(* firehose helpers *)
272
269
let append_firehose_event conn ~time ~t ~data : int Lwt.t =
273
-
unwrap @@ Queries.firehose_insert ~time ~t ~data conn
270
+
Util.use_pool conn @@ Queries.firehose_insert ~time ~t ~data
274
271
275
272
let list_firehose_since conn ~since ~limit : firehose_event list Lwt.t =
276
-
unwrap @@ Queries.firehose_since ~since ~limit conn
273
+
Util.use_pool conn @@ Queries.firehose_since ~since ~limit
277
274
278
275
let next_firehose_event conn ~cursor : firehose_event option Lwt.t =
279
-
unwrap @@ Queries.firehose_next ~cursor conn
276
+
Util.use_pool conn @@ Queries.firehose_next ~cursor
280
277
281
278
let earliest_firehose_after_time conn ~time : firehose_event option Lwt.t =
282
-
unwrap @@ Queries.firehose_earliest_after ~time conn
279
+
Util.use_pool conn @@ Queries.firehose_earliest_after ~time
283
280
284
281
let latest_firehose_seq conn : int option Lwt.t =
285
-
unwrap @@ Queries.firehose_latest_seq conn
282
+
Util.use_pool conn @@ Queries.firehose_latest_seq
286
283
287
284
let next_firehose_seq conn : int Lwt.t =
288
-
Queries.firehose_latest_seq conn
289
-
>$! fun s -> s |> Option.map succ |> Option.value ~default:0
285
+
let%lwt seq = Util.use_pool conn Queries.firehose_latest_seq in
286
+
Option.map succ seq |> Option.value ~default:0 |> Lwt.return
290
287
291
288
(* jwts *)
292
289
let is_token_revoked conn ~did ~jti =
293
-
unwrap @@ Queries.get_revoked_token conn ~did ~jti
290
+
Util.use_pool conn @@ Queries.get_revoked_token ~did ~jti
294
291
295
292
let revoke_token conn ~did ~jti =
296
-
unwrap @@ Queries.revoke_token conn ~did ~jti ~now:(Util.now_ms ())
293
+
Util.use_pool conn @@ Queries.revoke_token ~did ~jti ~now:(Util.now_ms ())
+51
-44
pegasus/lib/user_store.ml
+51
-44
pegasus/lib/user_store.ml
···
303
303
~path ~cids
304
304
end
305
305
306
-
type t = {did: string; db: (module Rapper_helper.CONNECTION)}
306
+
type t = {did: string; db: Util.caqti_pool}
307
307
308
308
let connect ?create ?write did : t Lwt.t =
309
309
let%lwt db =
···
312
312
Lwt.return {did; db}
313
313
314
314
let init t : unit Lwt.t =
315
-
let$! () = Queries.create_blocks_tables t.db in
316
-
let$! () = Queries.create_records_table t.db in
317
-
let$! () = Queries.create_blobs_tables t.db in
315
+
let%lwt () = Util.use_pool t.db Queries.create_blocks_tables in
316
+
let%lwt () = Util.use_pool t.db Queries.create_records_table in
317
+
let%lwt () = Util.use_pool t.db Queries.create_blobs_tables in
318
318
Lwt.return_unit
319
319
320
320
(* mst blocks; implements Writable_blockstore *)
321
321
322
322
let get_bytes t cid : Blob.t option Lwt.t =
323
-
Queries.get_block cid t.db
324
-
>$! function Some {data; _} -> Some data | None -> None
323
+
Util.use_pool t.db @@ Queries.get_block cid
324
+
>|= function Some {data; _} -> Some data | None -> None
325
325
326
326
let get_blocks t cids : Block_map.with_missing Lwt.t =
327
-
let$! blocks = Queries.get_blocks cids t.db in
327
+
let%lwt blocks = Util.use_pool t.db @@ Queries.get_blocks cids in
328
328
Lwt.return
329
329
(List.fold_left
330
330
(fun (acc : Block_map.with_missing) cid ->
···
337
337
cids )
338
338
339
339
let has t cid : bool Lwt.t =
340
-
Queries.has_block cid t.db >$! function Some _ -> true | None -> false
340
+
Util.use_pool t.db @@ Queries.has_block cid
341
+
>|= function Some _ -> true | None -> false
341
342
342
343
let put_block t cid block : (bool, exn) Lwt_result.t =
343
-
Queries.put_block cid block t.db
344
-
|> Lwt.map Util.caqti_result_exn
345
-
|> Lwt.map (Result.map (function Some _ -> true | None -> false))
344
+
Lwt_result.catch
345
+
@@ fun () ->
346
+
match%lwt Util.use_pool t.db @@ Queries.put_block cid block with
347
+
| Some _ ->
348
+
Lwt.return true
349
+
| None ->
350
+
Lwt.return false
346
351
347
352
let put_many t bm : (int, exn) Lwt_result.t =
348
353
Util.multi_query t.db
349
354
(List.map
350
-
(fun (cid, block) -> fun () -> Queries.put_block cid block t.db)
355
+
(fun (cid, block) -> Queries.put_block cid block)
351
356
(Block_map.entries bm) )
352
357
353
358
let delete_block t cid : (bool, exn) Lwt_result.t =
354
-
let$! () = Queries.delete_block cid t.db in
355
-
Lwt.return_ok true
359
+
Lwt_result.catch
360
+
@@ fun () -> Util.use_pool t.db @@ Queries.delete_block cid >|= fun _ -> true
356
361
357
362
let delete_many t cids : (int, exn) Lwt_result.t =
358
-
Queries.delete_blocks cids t.db >$! List.length >>= Lwt.return_ok
363
+
Lwt_result.catch
364
+
@@ fun () -> Util.use_pool t.db @@ Queries.delete_blocks cids >|= List.length
359
365
360
366
let clear_mst t : unit Lwt.t =
361
-
let$! () = Queries.clear_mst t.db in
367
+
let%lwt () = Util.use_pool t.db Queries.clear_mst in
362
368
Lwt.return_unit
363
369
364
370
(* repo commit *)
365
371
366
372
let get_commit t : (Cid.t * signed_commit) option Lwt.t =
367
-
Queries.get_commit t.db
368
-
>$! Option.map (fun (cid, data) ->
369
-
( cid
370
-
, data |> Dag_cbor.decode_to_yojson |> signed_commit_of_yojson
371
-
|> Result.get_ok ) )
373
+
let%lwt commit = Util.use_pool t.db Queries.get_commit in
374
+
Lwt.return
375
+
@@ Option.map
376
+
(fun (cid, data) ->
377
+
( cid
378
+
, data |> Dag_cbor.decode_to_yojson |> signed_commit_of_yojson
379
+
|> Result.get_ok ) )
380
+
commit
372
381
373
382
let put_commit t commit : (Cid.t, exn) Lwt_result.t =
374
383
let data = commit |> signed_commit_to_yojson |> Dag_cbor.encode_yojson in
375
384
let cid = Cid.create Dcbor data in
376
-
let$! () = Queries.put_commit cid data t.db in
377
-
Lwt.return_ok cid
385
+
( Lwt_result.catch
386
+
@@ fun () -> Util.use_pool t.db @@ Queries.put_commit cid data )
387
+
|> Lwt_result.map (fun () -> cid)
378
388
379
389
(* records *)
380
390
381
391
let get_record_by_path t path : record option Lwt.t =
382
-
Queries.get_record_by_path ~path t.db
383
-
>$! Option.map (fun (cid, data, since) ->
392
+
Util.use_pool t.db @@ Queries.get_record_by_path ~path
393
+
>|= Option.map (fun (cid, data, since) ->
384
394
{path; cid; value= Lex.of_cbor data; since} )
385
-
>>= Lwt.return
386
395
387
396
let get_record_by_cid t cid : record option Lwt.t =
388
-
Queries.get_record_by_cid ~cid t.db
389
-
>$! Option.map (fun (path, data, since) ->
397
+
Util.use_pool t.db @@ Queries.get_record_by_cid ~cid
398
+
>|= Option.map (fun (path, data, since) ->
390
399
{path; cid; value= Lex.of_cbor data; since} )
391
-
>>= Lwt.return
392
400
393
401
let list_records t ?(limit = 100) ?(cursor = "") ?(reverse = false) collection :
394
402
record list Lwt.t =
395
403
let fn =
396
404
if reverse then Queries.list_records_reverse else Queries.list_records
397
405
in
398
-
fn ~collection ~limit ~cursor t.db
399
-
>$! List.map (fun (path, cid, data, since) ->
406
+
Util.use_pool t.db @@ fn ~collection ~limit ~cursor
407
+
>|= List.map (fun (path, cid, data, since) ->
400
408
{path; cid; value= Lex.of_cbor data; since} )
401
-
>>= Lwt.return
402
409
403
410
let put_record t record path : (Cid.t * bytes) Lwt.t =
404
411
let cid, data = Lex.to_cbor_block record in
405
412
let since = Tid.now () in
406
-
let$! () = Queries.put_record ~path ~cid ~data ~since t.db in
413
+
let%lwt () =
414
+
Util.use_pool t.db @@ Queries.put_record ~path ~cid ~data ~since
415
+
in
407
416
Lwt.return (cid, data)
408
417
409
418
(* blobs *)
410
419
411
420
let get_blob t cid : blob_with_contents option Lwt.t =
412
-
match%lwt unwrap @@ Queries.get_blob t.db ~cid with
421
+
match%lwt Util.use_pool t.db @@ Queries.get_blob ~cid with
413
422
| None ->
414
423
Lwt.return_none
415
424
| Some blob ->
···
425
434
Lwt.return_some {id; cid; mimetype; data}
426
435
427
436
let list_blobs ?since t ~limit ~cursor : Cid.t list Lwt.t =
428
-
unwrap
437
+
Util.use_pool t.db
429
438
@@
430
439
match since with
431
440
| Some since ->
432
-
Queries.list_blobs_since t.db ~limit ~cursor ~since
441
+
Queries.list_blobs_since ~limit ~cursor ~since
433
442
| None ->
434
-
Queries.list_blobs t.db ~limit ~cursor
443
+
Queries.list_blobs ~limit ~cursor
435
444
436
445
let put_blob t cid mimetype data : int Lwt.t =
437
446
let file =
···
440
449
(Cid.to_string cid)
441
450
in
442
451
let _ = Out_channel.with_open_bin file Out_channel.output_bytes data in
443
-
unwrap @@ Queries.put_blob cid mimetype t.db
452
+
Util.use_pool t.db @@ Queries.put_blob cid mimetype
444
453
445
454
let list_blob_refs t path : Cid.t list Lwt.t =
446
-
unwrap @@ Queries.list_blob_refs path t.db
455
+
Util.use_pool t.db @@ Queries.list_blob_refs path
447
456
448
457
let put_blob_ref t path cid : unit Lwt.t =
449
-
unwrap @@ Queries.put_blob_ref path cid t.db
458
+
Util.use_pool t.db @@ Queries.put_blob_ref path cid
450
459
451
460
let put_blob_refs t path cids : (unit, exn) Lwt_result.t =
452
461
Lwt_result.map (fun _ -> ())
453
462
@@ Util.multi_query t.db
454
-
(List.map
455
-
(fun cid -> fun () -> Queries.put_blob_ref cid path t.db)
456
-
cids )
463
+
(List.map (fun cid -> Queries.put_blob_ref cid path) cids)
457
464
458
465
let clear_blob_refs t path cids : unit Lwt.t =
459
-
unwrap @@ Queries.clear_blob_refs path cids t.db
466
+
Util.use_pool t.db @@ Queries.clear_blob_refs path cids
+62
-43
pegasus/lib/util.ml
+62
-43
pegasus/lib/util.ml
···
164
164
Error "invalid field value"
165
165
end
166
166
167
+
type caqti_pool = (Caqti_lwt.connection, Caqti_error.t) Caqti_lwt_unix.Pool.t
168
+
167
169
(* turns a caqti error into an exception *)
168
170
let caqti_result_exn = function
169
171
| Ok x ->
···
172
174
Error (Caqti_error.Exn caqti_err)
173
175
174
176
let _init_connection conn =
175
-
let open Syntax in
176
-
let$! () =
177
+
match%lwt
177
178
[%rapper
178
179
execute
179
180
{sql|
···
183
184
|sql}
184
185
syntax_off]
185
186
() conn
186
-
in
187
-
Lwt.return conn
187
+
with
188
+
| Ok conn ->
189
+
Lwt.return conn
190
+
| Error e ->
191
+
raise (Caqti_error.Exn e)
188
192
189
-
(* opens an sqlite connection *)
190
-
let connect_sqlite ?(create = false) ?(write = true) db_uri =
193
+
(* creates an sqlite pool *)
194
+
let connect_sqlite ?(create = false) ?(write = true) db_uri : caqti_pool Lwt.t =
191
195
let uri =
192
196
Uri.add_query_params' db_uri
193
197
[("create", string_of_bool create); ("write", string_of_bool write)]
194
198
in
195
-
match%lwt Caqti_lwt_unix.connect uri with
196
-
| Ok c ->
197
-
_init_connection c
199
+
match
200
+
Caqti_lwt_unix.connect_pool
201
+
~post_connect:(fun conn -> Lwt_result.ok @@ _init_connection conn)
202
+
uri
203
+
with
204
+
| Ok pool ->
205
+
Lwt.return pool
198
206
| Error e ->
199
207
raise (Caqti_error.Exn e)
200
208
···
209
217
| Error e ->
210
218
raise (Caqti_error.Exn e)
211
219
220
+
let use_pool pool (f : Caqti_lwt.connection -> ('a, Caqti_error.t) Lwt_result.t)
221
+
: 'a Lwt.t =
222
+
match%lwt Caqti_lwt_unix.Pool.use f pool with
223
+
| Ok res ->
224
+
Lwt.return res
225
+
| Error e ->
226
+
raise (Caqti_error.Exn e)
227
+
212
228
(* runs a bunch of queries and catches duplicate insertion, returning how many succeeded *)
213
-
let multi_query connection
214
-
(queries : (unit -> ('a, Caqti_error.t) Lwt_result.t) list) :
215
-
(int, exn) Lwt_result.t =
229
+
let multi_query pool
230
+
(queries : (Caqti_lwt.connection -> ('a, Caqti_error.t) Lwt_result.t) list)
231
+
: (int, exn) Lwt_result.t =
216
232
let open Syntax in
217
-
let module C = (val connection : Caqti_lwt.CONNECTION) in
218
-
let$! () = C.start () in
219
-
let is_ignorable_error e =
220
-
match (e : Caqti_error.t) with
221
-
| `Request_failed qe | `Response_failed qe -> (
222
-
match Caqti_error.cause (`Request_failed qe) with
223
-
| `Not_null_violation | `Unique_violation ->
224
-
true
225
-
| _ ->
226
-
false )
227
-
| _ ->
228
-
false
229
-
in
230
-
let rec aux acc queries =
231
-
match acc with
232
-
| Error e ->
233
-
Lwt.return_error e
234
-
| Ok count -> (
235
-
match queries with
236
-
| [] ->
237
-
Lwt.return (Ok count)
238
-
| query :: rest -> (
239
-
let%lwt result = query () in
240
-
match result with
241
-
| Ok _ ->
242
-
aux (Ok (count + 1)) rest
243
-
| Error e ->
244
-
if is_ignorable_error e then aux (Ok count) rest
245
-
else Lwt.return_error (Caqti_error.Exn e) ) )
246
-
in
247
-
aux (Ok 0) queries
233
+
Lwt_result.catch (fun () ->
234
+
use_pool pool (fun connection ->
235
+
let module C = (val connection : Caqti_lwt.CONNECTION) in
236
+
let$! () = C.start () in
237
+
let is_ignorable_error e =
238
+
match (e : Caqti_error.t) with
239
+
| `Request_failed qe | `Response_failed qe -> (
240
+
match Caqti_error.cause (`Request_failed qe) with
241
+
| `Not_null_violation | `Unique_violation ->
242
+
true
243
+
| _ ->
244
+
false )
245
+
| _ ->
246
+
false
247
+
in
248
+
let rec aux acc queries =
249
+
match acc with
250
+
| Error e ->
251
+
Lwt.return_error e
252
+
| Ok count -> (
253
+
match queries with
254
+
| [] ->
255
+
Lwt.return (Ok count)
256
+
| query :: rest -> (
257
+
let%lwt result = query connection in
258
+
match result with
259
+
| Ok _ ->
260
+
aux (Ok (count + 1)) rest
261
+
| Error e ->
262
+
if is_ignorable_error e then aux (Ok count) rest
263
+
else Lwt.return_error e ) )
264
+
in
265
+
let%lwt result = aux (Ok 0) queries in
266
+
Lwt.return result ) )
248
267
249
268
(* unix timestamp *)
250
269
let now_ms () : int = int_of_float (Unix.gettimeofday () *. 1000.)
+7
-6
pegasus/test/test_sequencer.ml
+7
-6
pegasus/test/test_sequencer.ml
···
23
23
24
24
let with_db (f : Data_store.t -> unit Lwt.t) : unit Lwt.t =
25
25
let tmp = Filename.temp_file "pegasus_sequencer_test" ".db" in
26
-
Util.with_connection
27
-
(Uri.of_string ("sqlite3://" ^ tmp))
28
-
(fun conn ->
29
-
let%lwt () = Data_store.init conn in
30
-
let%lwt () = f conn in
31
-
Lwt.return_ok () )
26
+
let%lwt pool =
27
+
Util.connect_sqlite ~create:true ~write:true
28
+
(Uri.of_string ("sqlite3://" ^ tmp))
29
+
in
30
+
let%lwt () = Data_store.init pool in
31
+
let%lwt () = f pool in
32
+
Lwt.return ()
32
33
33
34
let mk_cid () =
34
35
let block =