objective categorical abstract machine language personal data server
xrpc getRecord, listRecords
+126
-13
+5
-1
bin/main.ml
+5
-1
bin/main.ml
···
20
20
; ( post
21
21
, "/xrpc/com.atproto.server.createSession"
22
22
, Api.Server.CreateSession.handler )
23
+
; (get, "/xrpc/com.atproto.repo.getRecord", Api.Repo.GetRecord.handler)
24
+
; (get, "/xrpc/com.atproto.repo.listRecords", Api.Repo.ListRecords.handler)
23
25
; (* account *)
24
26
(get, "/xrpc/com.atproto.server.getSession", Api.Server.GetSession.handler)
25
27
; ( post
···
34
36
; (* repo *)
35
37
(post, "/xrpc/com.atproto.repo.applyWrites", Api.Repo.ApplyWrites.handler)
36
38
; (post, "/xrpc/com.atproto.repo.createRecord", Api.Repo.CreateRecord.handler)
39
+
; (post, "/xrpc/com.atproto.repo.putRecord", Api.Repo.PutRecord.handler)
40
+
; (post, "/xrpc/com.atproto.repo.deleteRecord", Api.Repo.DeleteRecord.handler)
37
41
; (* preferences *)
38
42
( get
39
43
, "/xrpc/com.atproto.actor.getPreferences"
···
43
47
, Api.Actor.PutPreferences.handler ) ]
44
48
45
49
let main =
46
-
let%lwt db = Util.connect_sqlite Util.Constants.pegasus_db_location in
50
+
let%lwt db = Data_store.connect () in
47
51
let%lwt () = Data_store.init db in
48
52
Dream.serve ~interface:"0.0.0.0" ~port:8008
49
53
@@ Dream.logger
+30
pegasus/lib/api/repo/getRecord.ml
+30
pegasus/lib/api/repo/getRecord.ml
···
1
+
type query = {repo: string; collection: string; rkey: string; cid: Cid.t option}
2
+
[@@deriving yojson]
3
+
4
+
type response = {uri: string; cid: Cid.t; value: Mist.Lex.repo_record}
5
+
[@@deriving yojson]
6
+
7
+
let handler =
8
+
Xrpc.handler (fun ctx ->
9
+
let%lwt input = Xrpc.parse_query ctx.req query_of_yojson in
10
+
let%lwt input_did =
11
+
if String.starts_with ~prefix:"did:" input.repo then
12
+
Lwt.return input.repo
13
+
else
14
+
match%lwt Data_store.get_actor_by_identifier input.repo ctx.db with
15
+
| Some {did; _} ->
16
+
Lwt.return did
17
+
| None ->
18
+
Errors.invalid_request "target repository not found"
19
+
in
20
+
let%lwt repo = Repository.load input_did in
21
+
let path = input.collection ^ "/" ^ input.rkey in
22
+
let uri = "at://" ^ input_did ^ "/" ^ path in
23
+
match%lwt Repository.get_record repo path with
24
+
| Some {cid; value; _} when input.cid = None || input.cid = Some cid ->
25
+
Dream.json @@ Yojson.Safe.to_string
26
+
@@ response_to_yojson {uri; cid; value}
27
+
| _ ->
28
+
Errors.internal_error ~name:"RecordNotFound"
29
+
~msg:("could not find record " ^ uri)
30
+
() )
+49
pegasus/lib/api/repo/listRecords.ml
+49
pegasus/lib/api/repo/listRecords.ml
···
1
+
type query =
2
+
{ repo: string
3
+
; collection: string
4
+
; limit: int option
5
+
; cursor: string option
6
+
; reverse: bool option }
7
+
[@@deriving yojson]
8
+
9
+
type response = {cursor: string option; records: response_record list}
10
+
[@@deriving yojson]
11
+
12
+
and response_record = {uri: string; cid: Cid.t; value: Mist.Lex.repo_record}
13
+
[@@deriving yojson]
14
+
15
+
let handler =
16
+
Xrpc.handler (fun ctx ->
17
+
let%lwt input = Xrpc.parse_query ctx.req query_of_yojson in
18
+
let limit =
19
+
match input.limit with
20
+
| Some limit when limit > 0 && limit <= 100 ->
21
+
limit
22
+
| _ ->
23
+
100
24
+
in
25
+
let%lwt input_did =
26
+
if String.starts_with ~prefix:"did:" input.repo then
27
+
Lwt.return input.repo
28
+
else
29
+
match%lwt Data_store.get_actor_by_identifier input.repo ctx.db with
30
+
| Some {did; _} ->
31
+
Lwt.return did
32
+
| None ->
33
+
Errors.invalid_request "target repository not found"
34
+
in
35
+
let%lwt db = User_store.connect input_did in
36
+
let%lwt results =
37
+
User_store.list_records db ~limit ?cursor:input.cursor
38
+
?reverse:input.reverse input.collection
39
+
in
40
+
let cursor, results_rev =
41
+
List.fold_left
42
+
(fun (_cursor, results_rev) (record : User_store.Types.record) ->
43
+
let uri = "at://" ^ input_did ^ "/" ^ record.path in
44
+
( record.since
45
+
, {uri; cid= record.cid; value= record.value} :: results_rev ) )
46
+
("", []) results
47
+
in
48
+
Dream.json @@ Yojson.Safe.to_string
49
+
@@ response_to_yojson {cursor= Some cursor; records= List.rev results_rev} )
+3
pegasus/lib/data_store.ml
+3
pegasus/lib/data_store.ml
···
187
187
188
188
type t = (module Rapper_helper.CONNECTION)
189
189
190
+
let connect () : t Lwt.t =
191
+
Util.connect_sqlite Util.Constants.pegasus_db_location
192
+
190
193
let init conn : unit Lwt.t = unwrap @@ Queries.create_tables conn
191
194
192
195
let create_actor ~did ~handle ~email ~password ~signing_key conn =
+3
-5
pegasus/lib/repository.ml
+3
-5
pegasus/lib/repository.ml
···
180
180
Set.empty
181
181
|> Set.to_list |> Lwt.return
182
182
183
-
let list_records t collection : (string * Cid.t * record) list Lwt.t =
183
+
let list_all_records t collection : (string * Cid.t * record) list Lwt.t =
184
184
let%lwt map = get_map t in
185
185
StringMap.bindings map
186
186
|> List.filter (fun (path, _) ->
···
436
436
Lwt.return {commit= new_commit; results}
437
437
438
438
let load did : t Lwt.t =
439
-
let%lwt data_store_conn =
440
-
Util.connect_sqlite Util.Constants.pegasus_db_location
441
-
in
442
-
let%lwt user_db = Util.connect_sqlite (Util.Constants.user_db_location did) in
439
+
let%lwt data_store_conn = Data_store.connect () in
440
+
let%lwt user_db = User_store.connect did in
443
441
let%lwt () = User_store.init user_db in
444
442
let%lwt {signing_key; _} =
445
443
match%lwt Data_store.get_actor_by_identifier did data_store_conn with
+26
-7
pegasus/lib/user_store.ml
+26
-7
pegasus/lib/user_store.ml
···
150
150
data BLOB NOT NULL
151
151
);
152
152
CREATE INDEX IF NOT EXISTS records_cid_idx ON records (cid);
153
+
CREATE INDEX IF NOT EXISTS records_since_idx ON records (since);
153
154
|sql}]
154
155
()
155
156
···
168
169
get_many
169
170
{sql| SELECT @string{path}, @CID{cid}, @Blob{data}, @string{since} FROM records
170
171
WHERE path LIKE %string{collection}/%
171
-
ORDER BY since DESC LIMIT %int{limit} OFFSET %int{offset}
172
+
AND since < %string{cursor}
173
+
ORDER BY since DESC LIMIT %int{limit}
174
+
|sql}]
175
+
176
+
let list_records_reverse =
177
+
[%rapper
178
+
get_many
179
+
{sql| SELECT @string{path}, @CID{cid}, @Blob{data}, @string{since} FROM records
180
+
WHERE path LIKE %string{collection}/%
181
+
AND since > %string{cursor}
182
+
ORDER BY since ASC LIMIT %int{limit}
172
183
|sql}]
173
184
174
185
let put_record =
···
260
271
let clear_blob_refs path cids =
261
272
[%rapper
262
273
execute
263
-
{sql| DELETE FROM blobs_records WHERE record_path LIKE %string{path} AND blob_id IN (
264
-
SELECT id FROM blobs WHERE cid IN (%list{%CID{cids}})
265
-
)
274
+
{sql| DELETE FROM blobs_records
275
+
WHERE record_path LIKE %string{path}
276
+
AND blob_id IN (
277
+
SELECT id FROM blobs WHERE cid IN (%list{%CID{cids}})
278
+
)
266
279
|sql}]
267
280
~path ~cids
268
281
end
269
282
270
283
type t = (module Rapper_helper.CONNECTION)
284
+
285
+
let connect did : t Lwt.t =
286
+
Util.connect_sqlite (Util.Constants.user_db_location did)
271
287
272
288
let init conn : unit Lwt.t =
273
289
let$! () = Queries.create_blocks_tables conn in
···
348
364
{path; cid; value= Lex.of_cbor data; since} )
349
365
>>= Lwt.return
350
366
351
-
let list_records conn ?(limit = 100) ?(offset = 0) collection :
352
-
record list Lwt.t =
353
-
Queries.list_records ~collection ~limit ~offset conn
367
+
let list_records conn ?(limit = 100) ?(cursor = "") ?(reverse = false)
368
+
collection : record list Lwt.t =
369
+
let fn =
370
+
if reverse then Queries.list_records_reverse else Queries.list_records
371
+
in
372
+
fn ~collection ~limit ~cursor conn
354
373
>$! List.map (fun (path, cid, data, since) ->
355
374
{path; cid; value= Lex.of_cbor data; since} )
356
375
>>= Lwt.return
+10
pegasus/lib/xrpc.ml
+10
pegasus/lib/xrpc.ml
···
23
23
| Error e ->
24
24
exn_to_response e
25
25
26
+
let parse_query (req : Dream.request)
27
+
(of_yojson : Yojson.Safe.t -> ('a, string) result) : 'a Lwt.t =
28
+
try%lwt
29
+
let queries = Dream.all_queries req in
30
+
let query_json =
31
+
`Assoc (List.map (fun (k, v) -> (k, Yojson.Safe.from_string v)) queries)
32
+
in
33
+
query_json |> of_yojson |> Result.get_ok |> Lwt.return
34
+
with _ -> Errors.invalid_request "Invalid query string"
35
+
26
36
let parse_body (req : Dream.request)
27
37
(of_yojson : Yojson.Safe.t -> ('a, string) result) : 'a Lwt.t =
28
38
try%lwt