objective categorical abstract machine language personal data server
Implement record store
+243
-157
+16
-6
mist/lib/lex.ml
+16
-6
mist/lib/lex.ml
···
6
6
| `LexArray of value Array.t
7
7
| `LexMap of value StringMap.t ]
8
8
9
-
type repo_record = value StringMap.t
10
-
11
9
let rec to_ipld (v : value) : Dag_cbor.value =
12
10
match v with
13
11
| `BlobRef r -> (
···
78
76
let cid = Cid.create Dcbor encoded in
79
77
(cid, encoded)
80
78
79
+
let of_yojson (v : Yojson.Safe.t) : value = of_ipld (Dag_cbor.of_yojson v)
80
+
81
+
let to_yojson (v : value) : Yojson.Safe.t = Dag_cbor.to_yojson (to_ipld v)
82
+
83
+
type repo_record =
84
+
(value StringMap.t
85
+
[@of_yojson
86
+
fun v ->
87
+
match of_yojson v with
88
+
| `LexMap m ->
89
+
Ok m
90
+
| _ ->
91
+
Error "decoded non-map value"]
92
+
[@to_yojson fun v -> to_yojson (`LexMap v)] )
93
+
[@@deriving yojson]
94
+
81
95
let of_cbor encoded : repo_record =
82
96
let decoded = Dag_cbor.decode encoded in
83
97
match of_ipld decoded with
···
85
99
m
86
100
| _ ->
87
101
raise (Failure "Decoded non-record value")
88
-
89
-
let of_yojson (v : Yojson.Safe.t) : value = of_ipld (Dag_cbor.of_yojson v)
90
-
91
-
let to_yojson (v : value) : Yojson.Safe.t = Dag_cbor.to_yojson (to_ipld v)
+99
pegasus/lib/mst_store.ml
+99
pegasus/lib/mst_store.ml
···
1
+
open Util.Rapper
2
+
open Util.Syntax
3
+
module Block_map = Mist.Storage.Block_map
4
+
5
+
type t = Caqti_lwt.connection
6
+
7
+
type block = {cid: Cid.t; data: Blob.t}
8
+
9
+
module Queries = struct
10
+
let create_table =
11
+
[%rapper
12
+
execute
13
+
{sql| CREATE TABLE IF NOT EXISTS mst (
14
+
cid TEXT NOT NULL PRIMARY KEY,
15
+
data BLOB NOT NULL
16
+
);
17
+
|sql}]
18
+
()
19
+
20
+
let get_block cid =
21
+
[%rapper
22
+
get_opt
23
+
{sql| SELECT @CID{cid}, @Blob{data} FROM mst WHERE cid = %CID{cid} |sql}
24
+
record_out]
25
+
~cid
26
+
27
+
let get_blocks cids =
28
+
[%rapper
29
+
get_many
30
+
{sql| SELECT @CID{cid}, @Blob{data} FROM mst WHERE cid IN (%list{%CID{cids}}) |sql}
31
+
record_out]
32
+
~cids
33
+
34
+
let has_block cid =
35
+
[%rapper
36
+
get_opt {sql| SELECT @CID{cid} FROM mst WHERE cid = %CID{cid} |sql}]
37
+
~cid
38
+
39
+
let put_block cid block =
40
+
[%rapper
41
+
get_opt
42
+
{sql| INSERT INTO mst (cid, data) VALUES (%CID{cid}, %Blob{block}) ON CONFLICT DO NOTHING RETURNING @CID{cid} |sql}]
43
+
~cid ~block
44
+
45
+
let delete_block cid =
46
+
[%rapper execute {sql| DELETE FROM mst WHERE cid = %CID{cid} |sql}] ~cid
47
+
48
+
let delete_blocks cids =
49
+
[%rapper
50
+
get_many
51
+
{sql| DELETE FROM mst WHERE cid IN (%list{%CID{cids}}) RETURNING @CID{cid} |sql}]
52
+
~cids
53
+
end
54
+
55
+
let init connection = Queries.create_table connection
56
+
57
+
let get_bytes t cid =
58
+
Queries.get_block cid t
59
+
>$! function
60
+
| Some {data; _} ->
61
+
Lwt.return_some data
62
+
| None ->
63
+
Lwt.return_none
64
+
65
+
let get_blocks t cids =
66
+
let$! blocks = Queries.get_blocks cids t in
67
+
Lwt.return
68
+
(List.fold_left
69
+
(fun (acc : Block_map.with_missing) cid ->
70
+
match List.find_opt (fun b -> b.cid = cid) blocks with
71
+
| Some {data; _} ->
72
+
{acc with blocks= Block_map.set cid data acc.blocks}
73
+
| None ->
74
+
{acc with missing= cid :: acc.missing} )
75
+
{blocks= Block_map.empty; missing= []}
76
+
cids )
77
+
78
+
let has t cid =
79
+
Queries.has_block cid t
80
+
>$! function Some _ -> Lwt.return true | None -> Lwt.return false
81
+
82
+
let put_block t cid block =
83
+
Queries.put_block cid block t
84
+
|> Lwt.map Util.caqti_result_exn
85
+
|> Lwt.map (Result.map (function Some _ -> true | None -> false))
86
+
87
+
let put_many t bm =
88
+
Util.multi_query t
89
+
(List.map
90
+
(fun (cid, block) -> fun () -> Queries.put_block cid block t)
91
+
(Block_map.entries bm) )
92
+
>$! Lwt.return_ok
93
+
94
+
let delete_block t cid =
95
+
let$! () = Queries.delete_block cid t in
96
+
Lwt.return_ok true
97
+
98
+
let delete_many t cids =
99
+
Queries.delete_blocks cids t >$! List.length |> Lwt.return_ok
+74
pegasus/lib/record_store.ml
+74
pegasus/lib/record_store.ml
···
1
+
open Lwt.Infix
2
+
open Util.Rapper
3
+
open Util.Syntax
4
+
module Lex = Mist.Lex
5
+
module Tid = Mist.Tid
6
+
7
+
type t = Caqti_lwt.connection
8
+
9
+
type record = {path: string; cid: Cid.t; value: Lex.repo_record; since: Tid.t}
10
+
11
+
module Queries = struct
12
+
let create_table =
13
+
[%rapper
14
+
execute
15
+
{sql| CREATE TABLE IF NOT EXISTS records (
16
+
path TEXT NOT NULL PRIMARY KEY,
17
+
cid TEXT NOT NULL,
18
+
data BLOB NOT NULL,
19
+
since TEXT NOT NULL
20
+
);
21
+
|sql}]
22
+
()
23
+
24
+
let get_record_by_path =
25
+
[%rapper
26
+
get_opt
27
+
{sql| SELECT @CID{cid}, @Blob{data}, @string{since} FROM records WHERE path = %string{path}
28
+
|sql}]
29
+
30
+
let get_record_by_cid =
31
+
[%rapper
32
+
get_opt
33
+
{sql| SELECT @string{path}, @Blob{data}, @string{since} FROM records WHERE cid = %CID{cid}
34
+
|sql}]
35
+
36
+
let list_records =
37
+
[%rapper
38
+
get_many
39
+
{sql| SELECT @string{path}, @CID{cid}, @Blob{data}, @string{since} FROM records WHERE path LIKE %string{collection}/%
40
+
|sql}]
41
+
42
+
let write_record =
43
+
[%rapper
44
+
execute
45
+
{sql| INSERT INTO records (path, cid, data, since)
46
+
VALUES (%string{path}, %CID{cid}, %Blob{data}, %string{since})
47
+
|sql}]
48
+
end
49
+
50
+
let init connection = Queries.create_table connection
51
+
52
+
let get_record_by_path t path : record option Lwt.t =
53
+
Queries.get_record_by_path ~path t
54
+
>$! Option.map (fun (cid, data, since) ->
55
+
{path; cid; value= Lex.of_cbor data; since} )
56
+
>>= Lwt.return
57
+
58
+
let get_record_by_cid t cid : record option Lwt.t =
59
+
Queries.get_record_by_cid ~cid t
60
+
>$! Option.map (fun (path, data, since) ->
61
+
{path; cid; value= Lex.of_cbor data; since} )
62
+
>>= Lwt.return
63
+
64
+
let list_records t collection : record list Lwt.t =
65
+
Queries.list_records ~collection t
66
+
>$! List.map (fun (path, cid, data, since) ->
67
+
{path; cid; value= Lex.of_cbor data; since} )
68
+
>>= Lwt.return
69
+
70
+
let write_record t record path : unit Lwt.t =
71
+
let cid, data = Lex.to_cbor_block record in
72
+
let since = Tid.now () in
73
+
let$! () = Queries.write_record ~path ~cid ~data ~since t in
74
+
Lwt.return_unit
-145
pegasus/lib/sqlite_blockstore.ml
-145
pegasus/lib/sqlite_blockstore.ml
···
1
-
open Util.Rapper
2
-
open Util.Syntax
3
-
module Block_map = Mist.Storage.Block_map
4
-
5
-
type t = {connection: Caqti_lwt.connection}
6
-
7
-
type block = {cid: Cid.t; data: Blob.t}
8
-
9
-
module Queries = struct
10
-
let create_table =
11
-
[%rapper
12
-
execute
13
-
{sql| CREATE TABLE IF NOT EXISTS mst (
14
-
cid TEXT NOT NULL PRIMARY KEY,
15
-
data BLOB NOT NULL
16
-
);
17
-
|sql}]
18
-
()
19
-
20
-
let get_block cid =
21
-
[%rapper
22
-
get_opt
23
-
{sql| SELECT @Cid{cid}, @Blob{data} FROM mst WHERE cid = %Cid{cid} |sql}
24
-
record_out]
25
-
~cid
26
-
27
-
let get_blocks cids =
28
-
[%rapper
29
-
get_many
30
-
{sql| SELECT @Cid{cid}, @Blob{data} FROM mst WHERE cid IN (%list{%Cid{cids}}) |sql}
31
-
record_out]
32
-
~cids
33
-
34
-
let has_block cid =
35
-
[%rapper
36
-
get_opt {sql| SELECT @Cid{cid} FROM mst WHERE cid = %Cid{cid} |sql}]
37
-
~cid
38
-
39
-
let put_block cid block =
40
-
[%rapper
41
-
get_opt
42
-
{sql| INSERT INTO mst (cid, data) VALUES (%Cid{cid}, %Blob{block}) ON CONFLICT DO NOTHING RETURNING @Cid{cid} |sql}]
43
-
~cid ~block
44
-
45
-
let delete_block cid =
46
-
[%rapper execute {sql| DELETE FROM mst WHERE cid = %Cid{cid} |sql}] ~cid
47
-
48
-
let delete_blocks cids =
49
-
[%rapper
50
-
get_many
51
-
{sql| DELETE FROM mst WHERE cid IN (%list{%Cid{cids}}) RETURNING @Cid{cid} |sql}]
52
-
~cids
53
-
end
54
-
55
-
let caqti_result_exn = function
56
-
| Ok x ->
57
-
Ok x
58
-
| Error caqti_err ->
59
-
Error (Caqti_error.Exn caqti_err)
60
-
61
-
let multi_query connection
62
-
(queries : (unit -> ('a, Caqti_error.t) Lwt_result.t) list) =
63
-
let module C = (val connection : Caqti_lwt.CONNECTION) in
64
-
let$! () = C.start () in
65
-
let is_ignorable_error e =
66
-
match (e : Caqti_error.t) with
67
-
| `Request_failed qe | `Response_failed qe -> (
68
-
match Caqti_error.cause (`Request_failed qe) with
69
-
| `Not_null_violation | `Unique_violation ->
70
-
true
71
-
| _ ->
72
-
false )
73
-
| _ ->
74
-
false
75
-
in
76
-
let rec aux acc queries =
77
-
match acc with
78
-
| Error e ->
79
-
Lwt.return_error e
80
-
| Ok count -> (
81
-
match queries with
82
-
| [] ->
83
-
Lwt.return (Ok count)
84
-
| query :: rest -> (
85
-
let%lwt result = query () in
86
-
match result with
87
-
| Ok _ ->
88
-
aux (Ok (count + 1)) rest
89
-
| Error e ->
90
-
if is_ignorable_error e then aux (Ok count) rest
91
-
else Lwt.return_error e ) )
92
-
in
93
-
aux (Ok 0) queries
94
-
95
-
let connect db_uri =
96
-
let%lwt connection = Util.connect_sqlite db_uri in
97
-
let$! () = Queries.create_table connection in
98
-
Lwt.return {connection}
99
-
100
-
let get_bytes t cid =
101
-
let$! b_opt = Queries.get_block cid t.connection in
102
-
match b_opt with
103
-
| Some {data; _} ->
104
-
Lwt.return_some data
105
-
| None ->
106
-
Lwt.return_none
107
-
108
-
let get_blocks t cids =
109
-
let$! blocks = Queries.get_blocks cids t.connection in
110
-
Lwt.return
111
-
(List.fold_left
112
-
(fun (acc : Block_map.with_missing) cid ->
113
-
match List.find_opt (fun b -> b.cid = cid) blocks with
114
-
| Some {data; _} ->
115
-
{acc with blocks= Block_map.set cid data acc.blocks}
116
-
| None ->
117
-
{acc with missing= cid :: acc.missing} )
118
-
{blocks= Block_map.empty; missing= []}
119
-
cids )
120
-
121
-
let has t cid =
122
-
let$! b_opt = Queries.has_block cid t.connection in
123
-
match b_opt with Some _ -> Lwt.return true | None -> Lwt.return false
124
-
125
-
let put_block t cid block =
126
-
Queries.put_block cid block t.connection
127
-
|> Lwt.map caqti_result_exn
128
-
|> Lwt.map (Result.map (function Some _ -> true | None -> false))
129
-
130
-
let put_many t bm =
131
-
let$! inserted =
132
-
multi_query t.connection
133
-
(List.map
134
-
(fun (cid, block) -> fun () -> Queries.put_block cid block t.connection)
135
-
(Block_map.entries bm) )
136
-
in
137
-
Lwt.return_ok inserted
138
-
139
-
let delete_block t cid =
140
-
let$! () = Queries.delete_block cid t.connection in
141
-
Lwt.return_ok true
142
-
143
-
let delete_many t cids =
144
-
let$! deleted = Queries.delete_blocks cids t.connection in
145
-
Lwt.return_ok (List.length deleted)
-5
pegasus/lib/sqlite_blockstore.mli
-5
pegasus/lib/sqlite_blockstore.mli
+54
-1
pegasus/lib/util.ml
+54
-1
pegasus/lib/util.ml
···
1
1
module Syntax = struct
2
+
(* unwraps an Lwt result, raising an exception if there's an error *)
2
3
let ( let$! ) m f =
3
4
match%lwt m with Ok x -> f x | Error e -> raise (Caqti_error.Exn e)
5
+
6
+
(* unwraps an Lwt result, raising an exception if there's an error *)
7
+
let ( >$! ) m f =
8
+
match%lwt m with
9
+
| Ok x ->
10
+
Lwt.return (f x)
11
+
| Error e ->
12
+
raise (Caqti_error.Exn e)
4
13
end
5
14
6
15
module Rapper = struct
7
-
module Cid : Rapper.CUSTOM with type t = Cid.t = struct
16
+
module CID : Rapper.CUSTOM with type t = Cid.t = struct
8
17
type t = Cid.t
9
18
10
19
let t =
···
28
37
end
29
38
end
30
39
40
+
(* turns a caqti error into an exception *)
41
+
let caqti_result_exn = function
42
+
| Ok x ->
43
+
Ok x
44
+
| Error caqti_err ->
45
+
Error (Caqti_error.Exn caqti_err)
46
+
47
+
(* runs a bunch of queries and catches duplicate insertion, returning how many succeeded *)
48
+
let multi_query connection
49
+
(queries : (unit -> ('a, Caqti_error.t) Lwt_result.t) list) =
50
+
let open Syntax in
51
+
let module C = (val connection : Caqti_lwt.CONNECTION) in
52
+
let$! () = C.start () in
53
+
let is_ignorable_error e =
54
+
match (e : Caqti_error.t) with
55
+
| `Request_failed qe | `Response_failed qe -> (
56
+
match Caqti_error.cause (`Request_failed qe) with
57
+
| `Not_null_violation | `Unique_violation ->
58
+
true
59
+
| _ ->
60
+
false )
61
+
| _ ->
62
+
false
63
+
in
64
+
let rec aux acc queries =
65
+
match acc with
66
+
| Error e ->
67
+
Lwt.return_error e
68
+
| Ok count -> (
69
+
match queries with
70
+
| [] ->
71
+
Lwt.return (Ok count)
72
+
| query :: rest -> (
73
+
let%lwt result = query () in
74
+
match result with
75
+
| Ok _ ->
76
+
aux (Ok (count + 1)) rest
77
+
| Error e ->
78
+
if is_ignorable_error e then aux (Ok count) rest
79
+
else Lwt.return_error e ) )
80
+
in
81
+
aux (Ok 0) queries
82
+
83
+
(* opens an sqlite connection *)
31
84
let connect_sqlite db_uri =
32
85
let open Syntax in
33
86
match%lwt Caqti_lwt.connect (Uri.of_string db_uri) with