objective categorical abstract machine language personal data server
Make sqlitebs methods top level
+106
-104
+86
-103
pegasus/lib/sqlite_blockstore.ml
+86
-103
pegasus/lib/sqlite_blockstore.ml
···
1
1
open Util.Syntax
2
+
include Caqti_type.Std
3
+
include Caqti_request.Infix
4
+
include Lwt_result.Syntax
2
5
module Block_map = Mist.Storage.Block_map
3
-
4
-
let caqti_result_exn = function
5
-
| Ok x ->
6
-
Ok x
7
-
| Error caqti_err ->
8
-
Error (Caqti_error.Exn caqti_err)
9
6
10
7
module Cid : Rapper.CUSTOM with type t = Cid.t = struct
11
8
type t = Cid.t
···
29
26
in
30
27
Caqti_type.(custom ~encode ~decode string)
31
28
end
29
+
30
+
type t = {connection: Caqti_lwt.connection}
32
31
33
32
type block = {cid: Cid.t; data: Blob.t}
34
33
···
65
64
let put_block cid block =
66
65
[%rapper
67
66
get_opt
68
-
{sql| INSERT INTO blocks (cid, data) VALUES (%Cid{cid}, %Blob{block}) ON CONFLICT DO NOTHING RETURNING @Cid{cid} |sql}]
67
+
{sql| INSERT INTO mst (cid, data) VALUES (%Cid{cid}, %Blob{block}) ON CONFLICT DO NOTHING RETURNING @Cid{cid} |sql}]
69
68
~cid ~block
70
69
71
70
let delete_block cid =
72
-
[%rapper execute {sql| DELETE FROM blocks WHERE cid = %Cid{cid} |sql}] ~cid
71
+
[%rapper execute {sql| DELETE FROM mst WHERE cid = %Cid{cid} |sql}] ~cid
73
72
74
73
let delete_blocks cids =
75
74
[%rapper
76
75
get_many
77
-
{sql| DELETE FROM blocks WHERE cid IN (%list{%Cid{cids}}) RETURNING @Cid{cid} |sql}]
76
+
{sql| DELETE FROM mst WHERE cid IN (%list{%Cid{cids}}) RETURNING @Cid{cid} |sql}]
78
77
~cids
79
78
end
80
79
81
-
module S (C : Caqti_lwt.CONNECTION) : Mist.Storage.Writable_blockstore = struct
82
-
include Caqti_type.Std
83
-
include Caqti_request.Infix
84
-
include Lwt_result.Syntax
85
-
86
-
type t = {connection: Caqti_lwt.connection}
80
+
let caqti_result_exn = function
81
+
| Ok x ->
82
+
Ok x
83
+
| Error caqti_err ->
84
+
Error (Caqti_error.Exn caqti_err)
87
85
88
-
let multi_query (queries : (unit -> ('a, Caqti_error.t) Lwt_result.t) list) =
89
-
let$! () = C.start () in
90
-
let is_ignorable_error e =
91
-
match (e : Caqti_error.t) with
92
-
| `Request_failed qe | `Response_failed qe -> (
93
-
match Caqti_error.cause (`Request_failed qe) with
94
-
| `Not_null_violation | `Unique_violation ->
95
-
true
96
-
| _ ->
97
-
false )
86
+
let multi_query connection
87
+
(queries : (unit -> ('a, Caqti_error.t) Lwt_result.t) list) =
88
+
let module C = (val connection : Caqti_lwt.CONNECTION) in
89
+
let$! () = C.start () in
90
+
let is_ignorable_error e =
91
+
match (e : Caqti_error.t) with
92
+
| `Request_failed qe | `Response_failed qe -> (
93
+
match Caqti_error.cause (`Request_failed qe) with
94
+
| `Not_null_violation | `Unique_violation ->
95
+
true
98
96
| _ ->
99
-
false
100
-
in
101
-
let%lwt results =
102
-
Lwt_list.map_s
103
-
(fun query ->
104
-
match%lwt query () with
97
+
false )
98
+
| _ ->
99
+
false
100
+
in
101
+
let rec aux acc queries =
102
+
match acc with
103
+
| Error e ->
104
+
Lwt.return_error e
105
+
| Ok count -> (
106
+
match queries with
107
+
| [] ->
108
+
Lwt.return (Ok count)
109
+
| query :: rest -> (
110
+
let%lwt result = query () in
111
+
match result with
105
112
| Ok _ ->
106
-
Lwt.return true
113
+
aux (Ok (count + 1)) rest
107
114
| Error e ->
108
-
if is_ignorable_error e then Lwt.return false
109
-
else failwith (Caqti_error.show e) )
110
-
queries
111
-
in
112
-
let inserted =
113
-
List.fold_left
114
-
(fun acc success -> if success then acc + 1 else acc)
115
-
0 results
116
-
in
117
-
Lwt.return_ok inserted
115
+
if is_ignorable_error e then aux (Ok count) rest
116
+
else Lwt.return_error e ) )
117
+
in
118
+
aux (Ok 0) queries
118
119
119
-
let init t =
120
-
let$! () =
121
-
[%rapper execute {sql| PRAGMA journal_mode=WAL; |sql} syntax_off]
122
-
() t.connection
123
-
in
124
-
let$! () =
125
-
[%rapper execute {sql| PRAGMA synchronous=NORMAL; |sql} syntax_off]
126
-
() t.connection
127
-
in
128
-
let$! () = Queries.create_table t.connection in
129
-
Lwt.return_unit
120
+
let get_bytes t cid =
121
+
let$! b_opt = Queries.get_block cid t.connection in
122
+
match b_opt with
123
+
| Some {data; _} ->
124
+
Lwt.return_some data
125
+
| None ->
126
+
Lwt.return_none
130
127
131
-
let get_bytes t cid =
132
-
let$! b_opt = Queries.get_block cid t.connection in
133
-
match b_opt with
134
-
| Some {data; _} ->
135
-
Lwt.return_some data
136
-
| None ->
137
-
Lwt.return_none
128
+
let get_blocks t cids =
129
+
let$! blocks = Queries.get_blocks cids t.connection in
130
+
Lwt.return
131
+
(List.fold_left
132
+
(fun (acc : Block_map.with_missing) cid ->
133
+
match List.find_opt (fun b -> b.cid = cid) blocks with
134
+
| Some {data; _} ->
135
+
{acc with blocks= Block_map.set cid data acc.blocks}
136
+
| None ->
137
+
{acc with missing= cid :: acc.missing} )
138
+
{blocks= Block_map.empty; missing= []}
139
+
cids )
138
140
139
-
let get_blocks t cids =
140
-
let$! blocks = Queries.get_blocks cids t.connection in
141
-
Lwt.return
142
-
(List.fold_left
143
-
(fun (acc : Block_map.with_missing) cid ->
144
-
match List.find_opt (fun b -> b.cid = cid) blocks with
145
-
| Some {data; _} ->
146
-
{acc with blocks= Block_map.set cid data acc.blocks}
147
-
| None ->
148
-
{acc with missing= cid :: acc.missing} )
149
-
{blocks= Block_map.empty; missing= []}
150
-
cids )
141
+
let has t cid =
142
+
let$! b_opt = Queries.has_block cid t.connection in
143
+
match b_opt with Some _ -> Lwt.return true | None -> Lwt.return false
151
144
152
-
let has t cid =
153
-
let$! b_opt = Queries.has_block cid t.connection in
154
-
match b_opt with Some _ -> Lwt.return true | None -> Lwt.return false
155
-
156
-
let put_block t cid block =
157
-
Queries.put_block cid block t.connection
158
-
|> Lwt.map caqti_result_exn
159
-
|> Lwt.map (Result.map (function Some _ -> true | None -> false))
145
+
let put_block t cid block =
146
+
Queries.put_block cid block t.connection
147
+
|> Lwt.map caqti_result_exn
148
+
|> Lwt.map (Result.map (function Some _ -> true | None -> false))
160
149
161
-
let put_many t bm =
162
-
let$! inserted =
163
-
multi_query
164
-
(List.map
165
-
(fun (cid, block) ->
166
-
fun () -> Queries.put_block cid block t.connection )
167
-
(Block_map.entries bm) )
168
-
in
169
-
Lwt.return_ok inserted
150
+
let put_many t bm =
151
+
let$! inserted =
152
+
multi_query t.connection
153
+
(List.map
154
+
(fun (cid, block) -> fun () -> Queries.put_block cid block t.connection)
155
+
(Block_map.entries bm) )
156
+
in
157
+
Lwt.return_ok inserted
170
158
171
-
let delete_block t cid =
172
-
let$! () = Queries.delete_block cid t.connection in
173
-
Lwt.return_ok true
159
+
let delete_block t cid =
160
+
let$! () = Queries.delete_block cid t.connection in
161
+
Lwt.return_ok true
174
162
175
-
let delete_many t cids =
176
-
let$! deleted = Queries.delete_blocks cids t.connection in
177
-
Lwt.return_ok (List.length deleted)
178
-
end
163
+
let delete_many t cids =
164
+
let$! deleted = Queries.delete_blocks cids t.connection in
165
+
Lwt.return_ok (List.length deleted)
179
166
180
167
let connect db_uri =
181
-
match%lwt Caqti_lwt.connect (Uri.of_string db_uri) with
182
-
| Ok c ->
183
-
let module C = (val c : Caqti_lwt.CONNECTION) in
184
-
let module Store = S (C) in
185
-
Lwt.return (module Store : Mist.Storage.Writable_blockstore)
186
-
| Error e ->
187
-
failwith (Caqti_error.show e)
168
+
let%lwt connection = Util.connect_sqlite db_uri in
169
+
let$! () = Queries.create_table connection in
170
+
Lwt.return {connection}
+5
pegasus/lib/sqlite_blockstore.mli
+5
pegasus/lib/sqlite_blockstore.mli
+15
-1
pegasus/lib/util.ml
+15
-1
pegasus/lib/util.ml
···
1
1
module Syntax = struct
2
2
let ( let$! ) m f =
3
-
match%lwt m with Ok x -> f x | Error e -> failwith (Caqti_error.show e)
3
+
match%lwt m with Ok x -> f x | Error e -> raise (Caqti_error.Exn e)
4
4
end
5
+
6
+
let connect_sqlite db_uri =
7
+
let open Syntax in
8
+
match%lwt Caqti_lwt.connect (Uri.of_string db_uri) with
9
+
| Ok c ->
10
+
let$! () =
11
+
[%rapper execute {sql| PRAGMA journal_mode=WAL; |sql} syntax_off] () c
12
+
in
13
+
let$! () =
14
+
[%rapper execute {sql| PRAGMA synchronous=NORMAL; |sql} syntax_off] () c
15
+
in
16
+
Lwt.return c
17
+
| Error e ->
18
+
raise (Caqti_error.Exn e)