objective categorical abstract machine language personal data server
Align User_store with Blockstore module type
+70
-25
+1
-1
mist/lib/blob_ref.ml
+1
-1
mist/lib/blob_ref.ml
+40
-7
pegasus/lib/user_store.ml
+40
-7
pegasus/lib/user_store.ml
···
11
11
12
12
type blob = {id: int; cid: Cid.t; mimetype: string; data: Blob.t}
13
13
14
+
type t = (module Rapper_helper.CONNECTION)
15
+
14
16
module Queries = struct
15
17
(* mst storage *)
16
18
let create_mst_table =
···
90
92
{sql| SELECT @string{path}, @CID{cid}, @Blob{data}, @string{since} FROM records WHERE path LIKE %string{collection}/%
91
93
|sql}]
92
94
93
-
let write_record =
95
+
let put_record =
94
96
[%rapper
95
97
execute
96
98
{sql| INSERT INTO records (path, cid, data, since)
···
148
150
{sql| SELECT @CID{cid} FROM blobs WHERE id > %int{cursor} ORDER BY id LIMIT %int{limit} |sql}]
149
151
~limit ~cursor
150
152
151
-
let write_blob cid mimetype data =
153
+
let put_blob cid mimetype data =
152
154
[%rapper
153
155
get_one
154
156
{sql| INSERT INTO blobs (cid, mimetype, data) VALUES (%CID{cid}, %string{mimetype}, %Blob{data}) RETURNING @int{id} |sql}]
155
157
~cid ~mimetype ~data
158
+
159
+
let list_blob_refs path =
160
+
[%rapper
161
+
get_many
162
+
{sql| SELECT @CID{cid} FROM blobs WHERE path LIKE %string{path} |sql}]
163
+
~path
164
+
165
+
let put_blob_ref cid path =
166
+
[%rapper
167
+
execute
168
+
{sql| INSERT INTO blobs_records (blob_id, record_path) VALUES (
169
+
(SELECT id FROM blobs WHERE cid = %CID{cid} LIMIT 1),
170
+
%string{path}
171
+
)
172
+
|sql}]
173
+
~cid ~path
156
174
end
157
175
158
176
let init conn : unit Lwt.t =
···
225
243
{path; cid; value= Lex.of_cbor data; since} )
226
244
>>= Lwt.return
227
245
228
-
let write_record conn record path : unit Lwt.t =
246
+
let put_record conn record path : Cid.t Lwt.t =
229
247
let cid, data = Lex.to_cbor_block record in
230
248
let since = Tid.now () in
231
-
let$! () = Queries.write_record ~path ~cid ~data ~since conn in
232
-
Lwt.return_unit
249
+
let$! () = Queries.put_record ~path ~cid ~data ~since conn in
250
+
Lwt.return cid
233
251
234
252
(* blobs *)
235
253
···
241
259
let$! blobs = Queries.list_blobs conn ~limit ~cursor in
242
260
Lwt.return blobs
243
261
244
-
let write_blob conn cid mimetype data : int Lwt.t =
245
-
let$! blob_id = Queries.write_blob conn cid mimetype data in
262
+
let put_blob conn cid mimetype data : int Lwt.t =
263
+
let$! blob_id = Queries.put_blob cid mimetype data conn in
246
264
Lwt.return blob_id
265
+
266
+
let list_blob_refs conn path : Cid.t list Lwt.t =
267
+
let$! blob_refs = Queries.list_blob_refs path conn in
268
+
Lwt.return blob_refs
269
+
270
+
let put_blob_ref conn path cid : unit Lwt.t =
271
+
let$! () = Queries.put_blob_ref path cid conn in
272
+
Lwt.return_unit
273
+
274
+
let put_blob_refs conn path cids : (unit, exn) Lwt_result.t =
275
+
Lwt_result.map (fun _ -> ())
276
+
@@ Util.multi_query conn
277
+
(List.map
278
+
(fun cid -> fun () -> Queries.put_blob_ref cid path conn)
279
+
cids )
+29
-17
pegasus/lib/util.ml
+29
-17
pegasus/lib/util.ml
···
1
+
module Exceptions = struct
2
+
exception XrpcError of (string * string)
3
+
end
4
+
1
5
module Syntax = struct
2
6
(* unwraps an Lwt result, raising an exception if there's an error *)
3
7
let ( let$! ) m f =
···
44
48
| Error caqti_err ->
45
49
Error (Caqti_error.Exn caqti_err)
46
50
51
+
(* opens an sqlite connection *)
52
+
let connect_sqlite db_uri =
53
+
let open Syntax in
54
+
match%lwt Caqti_lwt.connect (Uri.of_string db_uri) with
55
+
| Ok c ->
56
+
let$! () =
57
+
[%rapper execute {sql| PRAGMA journal_mode=WAL; |sql} syntax_off] () c
58
+
in
59
+
let$! () =
60
+
[%rapper execute {sql| PRAGMA synchronous=NORMAL; |sql} syntax_off] () c
61
+
in
62
+
let$! () =
63
+
[%rapper execute {sql| PRAGMA foreign_keys=ON; |sql} syntax_off] () c
64
+
in
65
+
Lwt.return c
66
+
| Error e ->
67
+
raise (Caqti_error.Exn e)
68
+
47
69
(* runs a bunch of queries and catches duplicate insertion, returning how many succeeded *)
48
70
let multi_query connection
49
71
(queries : (unit -> ('a, Caqti_error.t) Lwt_result.t) list) :
···
81
103
in
82
104
aux (Ok 0) queries
83
105
84
-
(* opens an sqlite connection *)
85
-
let connect_sqlite db_uri =
86
-
let open Syntax in
87
-
match%lwt Caqti_lwt.connect (Uri.of_string db_uri) with
88
-
| Ok c ->
89
-
let$! () =
90
-
[%rapper execute {sql| PRAGMA journal_mode=WAL; |sql} syntax_off] () c
91
-
in
92
-
let$! () =
93
-
[%rapper execute {sql| PRAGMA synchronous=NORMAL; |sql} syntax_off] () c
94
-
in
95
-
let$! () =
96
-
[%rapper execute {sql| PRAGMA foreign_keys=ON; |sql} syntax_off] () c
97
-
in
98
-
Lwt.return c
99
-
| Error e ->
100
-
raise (Caqti_error.Exn e)
106
+
(* returns all blob refs in a record *)
107
+
let find_blob_refs (record : Mist.Lex.repo_record) : Mist.Blob_ref.t list =
108
+
List.fold_left
109
+
(fun acc (_, value) ->
110
+
match value with `BlobRef blob -> blob :: acc | _ -> acc )
111
+
[]
112
+
(Mist.Lex.StringMap.bindings record)