objective categorical abstract machine language personal data server
Implement optional S3 backups
+558
-49
+40
-2
bin/main.ml
+40
-2
bin/main.ml
···
197
197
let static_routes =
198
198
[Dream.get "/public/**" (Dream.static ~loader:public_loader "")]
199
199
200
-
let main =
200
+
let serve () =
201
201
Printexc.record_backtrace true ;
202
202
let%lwt db = Data_store.connect ~create:true () in
203
203
let%lwt () = Data_store.init db in
204
+
S3.Backup.start () ;
204
205
Dream.serve ~interface:"0.0.0.0" ~port:8008
205
206
@@ Dream.pipeline
206
207
[ Dream.logger
···
218
219
@ [ Dream.get "/xrpc/**" (Xrpc.service_proxy_handler db)
219
220
; Dream.post "/xrpc/**" (Xrpc.service_proxy_handler db) ]
220
221
221
-
let () = Lwt_main.run main
222
+
let migrate_blobs ?did () =
223
+
match did with
224
+
| Some did ->
225
+
print_endline ("migrating blobs for user " ^ did) ;
226
+
let%lwt _ = S3.Blob_migration.migrate_user ~did in
227
+
Lwt.return_unit
228
+
| None ->
229
+
print_endline "migrating all blobs to S3" ;
230
+
S3.Blob_migration.migrate_all ()
231
+
232
+
let print_usage () =
233
+
print_endline
234
+
@@ String.trim
235
+
{|
236
+
usage: pegasus [command]
237
+
238
+
commands:
239
+
serve start the PDS
240
+
migrate-blobs migrate all local blobs to S3
241
+
migrate-blobs <did> migrate blobs for a specific user to S3
242
+
|}
243
+
244
+
(** CLI entry point *)
245
+
let () =
246
+
let args = Array.to_list Sys.argv |> List.tl in
247
+
match args with
248
+
| [] | ["serve"] ->
249
+
Lwt_main.run (serve ())
250
+
| ["migrate-blobs"] ->
251
+
Lwt_main.run (migrate_blobs ())
252
+
| ["migrate-blobs"; did] ->
253
+
Lwt_main.run (migrate_blobs ~did ())
254
+
| ["help"] | ["--help"] | ["-h"] ->
255
+
print_usage ()
256
+
| cmd :: _ ->
257
+
print_endline ("unknown command: " ^ cmd) ;
258
+
print_usage () ;
259
+
exit 1
+1
dune-project
+1
dune-project
+1
pegasus.opam
+1
pegasus.opam
+36
-11
pegasus/lib/api/sync/getBlob.ml
+36
-11
pegasus/lib/api/sync/getBlob.ml
···
3
3
let handler =
4
4
Xrpc.handler (fun ctx ->
5
5
let {did; cid} = Xrpc.parse_query ctx.req query_of_yojson in
6
-
let cid = Cid.as_cid cid in
6
+
let cid_parsed = Cid.as_cid cid in
7
7
let%lwt {db; _} =
8
8
Repository.load did ~ensure_active:true ~write:false ~ds:ctx.db
9
9
in
10
-
let%lwt blob =
11
-
match%lwt User_store.get_blob db cid with
12
-
| Some blob ->
13
-
Lwt.return blob
10
+
let%lwt blob_meta = User_store.get_blob_metadata db cid_parsed in
11
+
match blob_meta with
12
+
| None ->
13
+
Errors.internal_error ~msg:"blob not found" ()
14
+
| Some {storage= Blob_store.S3; _} -> (
15
+
match Blob_store.cdn_redirect_url ~did ~cid with
16
+
| Some url ->
17
+
(* redirect to CDN *)
18
+
Lwt.return
19
+
@@ Dream.response ~status:`Found ~headers:[("Location", url)] ""
14
20
| None ->
15
-
Errors.internal_error ~msg:"blob not found" ()
16
-
in
17
-
Lwt.return
18
-
@@ Dream.response
19
-
~headers:[("Content-Type", blob.mimetype)]
20
-
(Bytes.to_string blob.data) )
21
+
(* no CDN, proxy from S3 *)
22
+
let%lwt blob =
23
+
match%lwt User_store.get_blob db cid_parsed with
24
+
| Some blob ->
25
+
Lwt.return blob
26
+
| None ->
27
+
Errors.internal_error ~msg:"blob data not found" ()
28
+
in
29
+
Lwt.return
30
+
@@ Dream.response
31
+
~headers:[("Content-Type", blob.mimetype)]
32
+
(Bytes.to_string blob.data) )
33
+
| Some _ ->
34
+
(* return from local storage *)
35
+
let%lwt blob =
36
+
match%lwt User_store.get_blob db cid_parsed with
37
+
| Some blob ->
38
+
Lwt.return blob
39
+
| None ->
40
+
Errors.internal_error ~msg:"blob not found" ()
41
+
in
42
+
Lwt.return
43
+
@@ Dream.response
44
+
~headers:[("Content-Type", blob.mimetype)]
45
+
(Bytes.to_string blob.data) )
+116
pegasus/lib/blob_store.ml
+116
pegasus/lib/blob_store.ml
···
1
+
type storage = Local | S3
2
+
3
+
let storage_of_string = function
4
+
| "local" ->
5
+
Local
6
+
| "s3" ->
7
+
S3
8
+
| s ->
9
+
failwith ("unknown storage type: " ^ s)
10
+
11
+
let storage_to_string = function Local -> "local" | S3 -> "s3"
12
+
13
+
let s3_key ~did ~cid = Printf.sprintf "blobs/%s/%s" did cid
14
+
15
+
let local_path ~did ~cid =
16
+
Filename.concat (Util.Constants.user_blobs_location did) cid
17
+
18
+
let cdn_redirect_url ~did ~cid : string option =
19
+
match Env.s3_config with
20
+
| Some {cdn_url= Some cdn_url; blobs_enabled= true; _} ->
21
+
Some (Printf.sprintf "%s/%s" cdn_url (s3_key ~did ~cid))
22
+
| _ ->
23
+
None
24
+
25
+
let put_s3 ~did ~cid ~data : unit Lwt.t =
26
+
match Env.s3_config with
27
+
| Some config when config.blobs_enabled -> (
28
+
let key = s3_key ~did ~cid in
29
+
let%lwt result =
30
+
Aws_s3_lwt.S3.put ~credentials:config.credentials_obj
31
+
~endpoint:config.endpoint_obj ~bucket:config.bucket ~key
32
+
~data:(Bytes.to_string data) ()
33
+
in
34
+
match result with
35
+
| Ok _ ->
36
+
Lwt.return_unit
37
+
| Error e ->
38
+
failwith
39
+
(Printf.sprintf "S3 put failed: %s" (Util.s3_error_to_string e)) )
40
+
| _ ->
41
+
failwith "S3 not configured for blob storage"
42
+
43
+
let get_s3 ~did ~cid : bytes option Lwt.t =
44
+
match Env.s3_config with
45
+
| Some config -> (
46
+
let key = s3_key ~did ~cid in
47
+
let%lwt result =
48
+
Aws_s3_lwt.S3.get ~credentials:config.credentials_obj
49
+
~endpoint:config.endpoint_obj ~bucket:config.bucket ~key ()
50
+
in
51
+
match result with
52
+
| Ok data ->
53
+
Lwt.return_some (Bytes.of_string data)
54
+
| Error _ ->
55
+
Lwt.return_none )
56
+
| None ->
57
+
Lwt.return_none
58
+
59
+
let delete_s3 ~did ~cid : unit Lwt.t =
60
+
match Env.s3_config with
61
+
| Some config -> (
62
+
let key = s3_key ~did ~cid in
63
+
let%lwt result =
64
+
Aws_s3_lwt.S3.delete ~credentials:config.credentials_obj
65
+
~endpoint:config.endpoint_obj ~bucket:config.bucket ~key ()
66
+
in
67
+
match result with
68
+
| Ok () ->
69
+
Lwt.return_unit
70
+
| Error e ->
71
+
Dream.error (fun log ->
72
+
log "S3 delete failed for %s: %s" key (Util.s3_error_to_string e) ) ;
73
+
Lwt.return_unit )
74
+
| None ->
75
+
Lwt.return_unit
76
+
77
+
let put_local ~did ~cid ~data : unit =
78
+
let file = local_path ~did ~cid in
79
+
Core_unix.mkdir_p (Filename.dirname file) ~perm:0o755 ;
80
+
Out_channel.with_open_bin file (fun oc -> Out_channel.output_bytes oc data)
81
+
82
+
let get_local ~did ~cid : bytes option =
83
+
let file = local_path ~did ~cid in
84
+
if Sys.file_exists file then
85
+
Some (In_channel.with_open_bin file In_channel.input_all |> Bytes.of_string)
86
+
else None
87
+
88
+
let delete_local ~did ~cid : unit =
89
+
let file = local_path ~did ~cid in
90
+
if Sys.file_exists file then Sys.remove file
91
+
92
+
let put ~did ~cid ~data : storage Lwt.t =
93
+
match Env.s3_config with
94
+
| Some {blobs_enabled= true; _} ->
95
+
let%lwt () = put_s3 ~did ~cid:(Cid.to_string cid) ~data in
96
+
Lwt.return S3
97
+
| _ ->
98
+
put_local ~did ~cid:(Cid.to_string cid) ~data ;
99
+
Lwt.return Local
100
+
101
+
let get ~did ~cid ~storage : bytes option Lwt.t =
102
+
let cid_str = Cid.to_string cid in
103
+
match storage with
104
+
| Local ->
105
+
Lwt.return (get_local ~did ~cid:cid_str)
106
+
| S3 ->
107
+
get_s3 ~did ~cid:cid_str
108
+
109
+
let delete ~did ~cid ~storage : unit Lwt.t =
110
+
let cid_str = Cid.to_string cid in
111
+
match storage with
112
+
| Local ->
113
+
delete_local ~did ~cid:cid_str ;
114
+
Lwt.return_unit
115
+
| S3 ->
116
+
delete_s3 ~did ~cid:cid_str
+1
pegasus/lib/dune
+1
pegasus/lib/dune
+69
pegasus/lib/env.ml
+69
pegasus/lib/env.ml
···
81
81
| None, None ->
82
82
(None, None)
83
83
end
84
+
85
+
type s3_config =
86
+
{ blobs_enabled: bool
87
+
; backups_enabled: bool
88
+
; backup_interval_s: float
89
+
; endpoint: string option
90
+
; region: string
91
+
; bucket: string
92
+
; access_key: string
93
+
; secret_key: string
94
+
; cdn_url: string option
95
+
; credentials_obj: Aws_s3.Credentials.t
96
+
; endpoint_obj: Aws_s3.Region.endpoint }
97
+
98
+
let s3_config =
99
+
begin
100
+
let default_backup_interval = 3600.0 in
101
+
let blobs_enabled =
102
+
Sys.getenv_opt "S3_BLOBS_ENABLED" |> Option.map (( = ) "true")
103
+
in
104
+
let backups_enabled =
105
+
Sys.getenv_opt "S3_BACKUPS_ENABLED" |> Option.map (( = ) "true")
106
+
in
107
+
let backup_interval_s =
108
+
Sys.getenv_opt "S3_BACKUP_INTERVAL_S"
109
+
|> Option.map float_of_string_opt
110
+
|> Option.join
111
+
in
112
+
let endpoint = Sys.getenv_opt "S3_ENDPOINT" in
113
+
match (blobs_enabled, backups_enabled) with
114
+
| Some true, _ | _, Some true -> (
115
+
match
116
+
( Sys.getenv_opt "S3_REGION"
117
+
, Sys.getenv_opt "S3_BUCKET"
118
+
, Sys.getenv_opt "S3_ACCESS_KEY"
119
+
, Sys.getenv_opt "S3_SECRET_KEY" )
120
+
with
121
+
| Some region, Some bucket, Some access_key, Some secret_key ->
122
+
let region_obj =
123
+
match endpoint with
124
+
| Some host ->
125
+
Aws_s3.Region.vendor ~region_name:region ~host ()
126
+
| None ->
127
+
Aws_s3.Region.of_string region
128
+
in
129
+
let endpoint_obj =
130
+
Aws_s3.Region.endpoint ~inet:`V4 ~scheme:`Https region_obj
131
+
in
132
+
Some
133
+
{ blobs_enabled= Option.value ~default:false blobs_enabled
134
+
; backups_enabled= Option.value ~default:false backups_enabled
135
+
; backup_interval_s=
136
+
Option.value ~default:default_backup_interval backup_interval_s
137
+
; endpoint
138
+
; region
139
+
; bucket
140
+
; access_key
141
+
; secret_key
142
+
; cdn_url= Sys.getenv_opt "S3_CDN_URL"
143
+
; credentials_obj=
144
+
Aws_s3.Credentials.make ~access_key ~secret_key ()
145
+
; endpoint_obj }
146
+
| _ ->
147
+
failwith
148
+
"S3 backups require the following environment variables: \
149
+
S3_REGION, S3_BUCKET, S3_ACCESS_KEY, S3_SECRET_KEY" )
150
+
| _ ->
151
+
None
152
+
end
+3
pegasus/lib/migrations/user_store/002_blob_storage_field.sql
+3
pegasus/lib/migrations/user_store/002_blob_storage_field.sql
+105
pegasus/lib/s3/backup.ml
+105
pegasus/lib/s3/backup.ml
···
1
+
let timestamp_string () =
2
+
let tm = Unix.gmtime (Unix.gettimeofday ()) in
3
+
Printf.sprintf "%04d-%02d-%02d_%02d-%02d-%02d" (tm.Unix.tm_year + 1900)
4
+
(tm.Unix.tm_mon + 1) tm.Unix.tm_mday tm.Unix.tm_hour tm.Unix.tm_min
5
+
tm.Unix.tm_sec
6
+
7
+
let upload_to_s3 ~(config : Env.s3_config) ~file_path ~key : unit Lwt.t =
8
+
let%lwt data =
9
+
Lwt_io.with_file ~mode:Lwt_io.Input file_path (fun ic -> Lwt_io.read ic)
10
+
in
11
+
let%lwt result =
12
+
Aws_s3_lwt.S3.put ~credentials:config.credentials_obj
13
+
~endpoint:config.endpoint_obj ~bucket:config.bucket ~key ~data ()
14
+
in
15
+
match result with
16
+
| Ok _ ->
17
+
Dream.info (fun log -> log "S3 backup uploaded: %s" key) ;
18
+
Lwt.return_unit
19
+
| Error e ->
20
+
Dream.error (fun log ->
21
+
log "S3 backup upload failed for %s: %s" key
22
+
(Util.s3_error_to_string e) ) ;
23
+
Lwt.return_unit
24
+
25
+
let backup_main_db ~config : unit Lwt.t =
26
+
let db_path = Util.Constants.pegasus_db_filepath in
27
+
if Sys.file_exists db_path then (
28
+
let timestamp = timestamp_string () in
29
+
let key = Printf.sprintf "backups/pegasus-%s.db" timestamp in
30
+
Dream.info (fun log -> log "starting main database backup") ;
31
+
upload_to_s3 ~config ~file_path:db_path ~key )
32
+
else (
33
+
Dream.warning (fun log -> log "main database not found: %s" db_path) ;
34
+
Lwt.return_unit )
35
+
36
+
let backup_user_db ~config ~did : unit Lwt.t =
37
+
let db_path = Util.Constants.user_db_filepath did in
38
+
if Sys.file_exists db_path then
39
+
let timestamp = timestamp_string () in
40
+
let did_safe = Str.global_replace (Str.regexp ":") "_" did in
41
+
let key = Printf.sprintf "backups/store/%s-%s.db" did_safe timestamp in
42
+
upload_to_s3 ~config ~file_path:db_path ~key
43
+
else Lwt.return_unit
44
+
45
+
let backup_all_user_dbs ~config ~ds : unit Lwt.t =
46
+
Dream.info (fun log -> log "starting backup of user databases") ;
47
+
let rec backup_batch cursor count =
48
+
let%lwt actors = Data_store.list_actors ~cursor ~limit:100 ds in
49
+
match actors with
50
+
| [] ->
51
+
Dream.info (fun log -> log "backed up %d user databases" count) ;
52
+
Lwt.return_unit
53
+
| actors ->
54
+
let%lwt () =
55
+
Lwt_list.iter_s
56
+
(fun (actor : Data_store.Types.actor) ->
57
+
backup_user_db ~config ~did:actor.did )
58
+
actors
59
+
in
60
+
let last = List.hd (List.rev actors) in
61
+
backup_batch last.did (count + List.length actors)
62
+
in
63
+
backup_batch "" 0
64
+
65
+
let do_backup () : unit Lwt.t =
66
+
match Env.s3_config with
67
+
| Some config when config.backups_enabled ->
68
+
Dream.info (fun log -> log "starting S3 backup") ;
69
+
let%lwt ds = Data_store.connect () in
70
+
let%lwt () = backup_main_db ~config in
71
+
let%lwt () = backup_all_user_dbs ~config ~ds in
72
+
Dream.info (fun log -> log "S3 backup completed") ;
73
+
Lwt.return_unit
74
+
| _ ->
75
+
Lwt.return_unit
76
+
77
+
let queue_backups () : unit Lwt.t =
78
+
match Env.s3_config with
79
+
| Some config when config.backups_enabled ->
80
+
Dream.info (fun log -> log "starting hourly S3 backups") ;
81
+
let%lwt () =
82
+
Lwt.catch
83
+
(fun () -> do_backup ())
84
+
(fun e ->
85
+
Dream.error (fun log ->
86
+
log "backup failed: %s" (Printexc.to_string e) ) ;
87
+
Lwt.return_unit )
88
+
in
89
+
let rec loop () =
90
+
let%lwt () = Lwt_unix.sleep config.backup_interval_s in
91
+
let%lwt () =
92
+
Lwt.catch
93
+
(fun () -> do_backup ())
94
+
(fun e ->
95
+
Dream.error (fun log ->
96
+
log "backup failed: %s" (Printexc.to_string e) ) ;
97
+
Lwt.return_unit )
98
+
in
99
+
loop ()
100
+
in
101
+
loop ()
102
+
| _ ->
103
+
Lwt.return_unit
104
+
105
+
let start () : unit = Lwt.async queue_backups
+96
pegasus/lib/s3/blob_migration.ml
+96
pegasus/lib/s3/blob_migration.ml
···
1
+
let migrate_user ~did : (int * int) Lwt.t =
2
+
match Env.s3_config with
3
+
| Some config when config.blobs_enabled ->
4
+
Dream.info (fun log -> log "migrating blobs for user %s" did) ;
5
+
let%lwt user_db = User_store.connect did in
6
+
let migrated = ref 0 in
7
+
let errors = ref 0 in
8
+
let rec migrate_batch cursor =
9
+
let%lwt blobs =
10
+
User_store.list_blobs_by_storage user_db ~storage:Blob_store.Local
11
+
~limit:100 ~cursor
12
+
in
13
+
match blobs with
14
+
| [] ->
15
+
Lwt.return_unit
16
+
| blobs ->
17
+
let%lwt () =
18
+
Lwt_list.iter_s
19
+
(fun (cid, _mimetype) ->
20
+
Lwt.catch
21
+
(fun () ->
22
+
let local_path =
23
+
Blob_store.local_path ~did ~cid:(Cid.to_string cid)
24
+
in
25
+
if Sys.file_exists local_path then (
26
+
let data =
27
+
In_channel.with_open_bin local_path
28
+
In_channel.input_all
29
+
|> Bytes.of_string
30
+
in
31
+
let%lwt () =
32
+
Blob_store.put_s3 ~did ~cid:(Cid.to_string cid) ~data
33
+
in
34
+
let%lwt () =
35
+
User_store.update_blob_storage user_db cid
36
+
Blob_store.S3
37
+
in
38
+
Sys.remove local_path ; incr migrated ; Lwt.return_unit
39
+
)
40
+
else (
41
+
Dream.warning (fun log ->
42
+
log "local blob file not found: %s" local_path ) ;
43
+
Lwt.return_unit ) )
44
+
(fun e ->
45
+
Dream.error (fun log ->
46
+
log "blob migration error for %s: %s"
47
+
(Cid.to_string cid) (Printexc.to_string e) ) ;
48
+
incr errors ;
49
+
Lwt.return_unit ) )
50
+
blobs
51
+
in
52
+
let last_cid, _ = List.hd (List.rev blobs) in
53
+
migrate_batch (Cid.to_string last_cid)
54
+
in
55
+
let%lwt () = migrate_batch "" in
56
+
Dream.info (fun log ->
57
+
log "blob migration complete for %s: %d migrated, %d errors" did
58
+
!migrated !errors ) ;
59
+
Lwt.return (!migrated, !errors)
60
+
| _ ->
61
+
Dream.error (fun log -> log "S3 blob storage not enabled") ;
62
+
Lwt.return (0, 0)
63
+
64
+
let migrate_all () : unit Lwt.t =
65
+
match Env.s3_config with
66
+
| Some config when config.blobs_enabled ->
67
+
Dream.info (fun log -> log "migrating all blobs to S3") ;
68
+
let%lwt ds = Data_store.connect () in
69
+
let total_migrated = ref 0 in
70
+
let total_errors = ref 0 in
71
+
let rec migrate_batch cursor =
72
+
let%lwt actors = Data_store.list_actors ~cursor ~limit:100 ds in
73
+
match actors with
74
+
| [] ->
75
+
Lwt.return_unit
76
+
| actors ->
77
+
let%lwt () =
78
+
Lwt_list.iter_s
79
+
(fun (actor : Data_store.Types.actor) ->
80
+
let%lwt migrated, errors = migrate_user ~did:actor.did in
81
+
total_migrated := !total_migrated + migrated ;
82
+
total_errors := !total_errors + errors ;
83
+
Lwt.return_unit )
84
+
actors
85
+
in
86
+
let last = List.hd (List.rev actors) in
87
+
migrate_batch last.did
88
+
in
89
+
let%lwt () = migrate_batch "" in
90
+
Dream.info (fun log ->
91
+
log "blob migration complete: %d total migrated, %d total errors"
92
+
!total_migrated !total_errors ) ;
93
+
Lwt.return_unit
94
+
| _ ->
95
+
Dream.error (fun log -> log "S3 blob storage not enabled") ;
96
+
Lwt.return_unit
+76
-36
pegasus/lib/user_store.ml
+76
-36
pegasus/lib/user_store.ml
···
5
5
module Lex = Mist.Lex
6
6
module Tid = Mist.Tid
7
7
8
-
let delete_blob_file ~did cid =
9
-
let file = Filename.concat (Util.Constants.user_blobs_location did) cid in
10
-
if Sys.file_exists file then Sys.remove file
8
+
let delete_blob_file ~did ~cid ~storage =
9
+
Lwt.async (fun () -> Blob_store.delete ~did ~cid ~storage)
11
10
12
11
module Types = struct
13
12
open struct
···
59
58
60
59
type record = {path: string; cid: Cid.t; value: Lex.repo_record; since: Tid.t}
61
60
62
-
type blob = {cid: Cid.t; mimetype: string}
61
+
type blob = {cid: Cid.t; mimetype: string; storage: Blob_store.storage}
63
62
64
-
type blob_with_contents = {cid: Cid.t; mimetype: string; data: Blob.t}
63
+
type blob_with_contents =
64
+
{cid: Cid.t; mimetype: string; data: Blob.t; storage: Blob_store.storage}
65
65
end
66
66
67
67
open Types
···
177
177
let get_blob =
178
178
[%rapper
179
179
get_opt
180
-
{sql| SELECT @CID{cid}, @string{mimetype} FROM blobs WHERE cid = %CID{cid} |sql}
181
-
record_out]
180
+
{sql| SELECT @CID{cid}, @string{mimetype}, @string{storage} FROM blobs WHERE cid = %CID{cid} |sql}]
182
181
183
182
let list_blobs =
184
183
[%rapper
···
227
226
)
228
227
|sql}]
229
228
230
-
let put_blob cid mimetype =
229
+
let put_blob cid mimetype storage =
231
230
[%rapper
232
231
get_one
233
-
{sql| INSERT INTO blobs (cid, mimetype)
234
-
VALUES (%CID{cid}, %string{mimetype})
235
-
ON CONFLICT (cid) DO UPDATE SET mimetype = excluded.mimetype
232
+
{sql| INSERT INTO blobs (cid, mimetype, storage)
233
+
VALUES (%CID{cid}, %string{mimetype}, %string{storage})
234
+
ON CONFLICT (cid) DO UPDATE SET mimetype = excluded.mimetype, storage = excluded.storage
236
235
RETURNING @CID{cid}
237
236
|sql}]
238
-
~cid ~mimetype
237
+
~cid ~mimetype ~storage
239
238
240
239
let delete_blob cid =
241
240
[%rapper execute {sql| DELETE FROM blobs WHERE cid = %CID{cid} |sql}] ~cid
242
241
242
+
let update_blob_storage cid storage =
243
+
[%rapper
244
+
execute
245
+
{sql| UPDATE blobs SET storage = %string{storage} WHERE cid = %CID{cid} |sql}]
246
+
~cid ~storage
247
+
248
+
let list_blobs_by_storage =
249
+
[%rapper
250
+
get_many
251
+
{sql| SELECT @CID{cid}, @string{mimetype} FROM blobs
252
+
WHERE storage = %string{storage}
253
+
AND cid > %string{cursor}
254
+
ORDER BY cid
255
+
LIMIT %int{limit}
256
+
|sql}]
257
+
243
258
let delete_orphaned_blobs_by_record_path path =
244
259
[%rapper
245
260
get_many
···
251
266
SELECT 1 FROM blobs_records
252
267
WHERE blob_cid = blobs.cid AND record_path != %string{path}
253
268
)
254
-
RETURNING @CID{cid}
269
+
RETURNING @CID{cid}, @string{storage}
255
270
|sql}]
256
271
~path
257
272
···
415
430
in
416
431
let () =
417
432
List.iter
418
-
(fun cid -> delete_blob_file ~did:t.did (Cid.to_string cid))
433
+
(fun (cid, storage_str) ->
434
+
let storage = Blob_store.storage_of_string storage_str in
435
+
delete_blob_file ~did:t.did ~cid ~storage )
419
436
deleted_blobs
420
437
in
421
438
del ) )
···
426
443
match%lwt Util.use_pool t.db @@ Queries.get_blob ~cid with
427
444
| None ->
428
445
Lwt.return_none
429
-
| Some blob ->
430
-
let {cid; mimetype} : blob = blob in
431
-
let file =
432
-
Filename.concat
433
-
(Util.Constants.user_blobs_location t.did)
434
-
(Cid.to_string cid)
435
-
in
436
-
let data =
437
-
In_channel.with_open_bin file In_channel.input_all |> Bytes.of_string
438
-
in
439
-
Lwt.return_some {cid; mimetype; data}
446
+
| Some (cid, mimetype, storage_str) -> (
447
+
let storage = Blob_store.storage_of_string storage_str in
448
+
let%lwt data_opt = Blob_store.get ~did:t.did ~cid ~storage in
449
+
match data_opt with
450
+
| Some data ->
451
+
Lwt.return_some {cid; mimetype; data; storage}
452
+
| None ->
453
+
Lwt.return_none )
454
+
455
+
let get_blob_metadata t cid : blob option Lwt.t =
456
+
match%lwt Util.use_pool t.db @@ Queries.get_blob ~cid with
457
+
| None ->
458
+
Lwt.return_none
459
+
| Some (cid, mimetype, storage_str) ->
460
+
let storage = Blob_store.storage_of_string storage_str in
461
+
Lwt.return_some {cid; mimetype; storage}
440
462
441
463
let list_blobs ?since t ~limit ~cursor : Cid.t list Lwt.t =
442
464
Util.use_pool t.db
···
457
479
Util.use_pool t.db @@ Queries.count_referenced_blobs ()
458
480
459
481
let put_blob t cid mimetype data : Cid.t Lwt.t =
460
-
let file =
461
-
Filename.concat
462
-
(Util.Constants.user_blobs_location t.did)
463
-
(Cid.to_string cid)
464
-
in
465
-
Core_unix.mkdir_p (Filename.dirname file) ~perm:0o755 ;
466
-
Out_channel.with_open_bin file (fun oc -> Out_channel.output_bytes oc data) ;
467
-
Util.use_pool t.db @@ Queries.put_blob cid mimetype
482
+
let%lwt storage = Blob_store.put ~did:t.did ~cid ~data in
483
+
let storage_str = Blob_store.storage_to_string storage in
484
+
Util.use_pool t.db @@ Queries.put_blob cid mimetype storage_str
468
485
469
486
let delete_blob t cid : unit Lwt.t =
470
-
delete_blob_file ~did:t.did (Cid.to_string cid) ;
487
+
let%lwt blob_opt = get_blob_metadata t cid in
488
+
( match blob_opt with
489
+
| Some {storage; _} ->
490
+
delete_blob_file ~did:t.did ~cid ~storage
491
+
| None ->
492
+
() ) ;
471
493
Util.use_pool t.db @@ Queries.delete_blob cid
472
494
473
-
let delete_orphaned_blobs_by_record_path t path : Cid.t list Lwt.t =
474
-
Util.use_pool t.db @@ Queries.delete_orphaned_blobs_by_record_path path
495
+
let delete_orphaned_blobs_by_record_path t path :
496
+
(Cid.t * Blob_store.storage) list Lwt.t =
497
+
let%lwt results =
498
+
Util.use_pool t.db @@ Queries.delete_orphaned_blobs_by_record_path path
499
+
in
500
+
Lwt.return
501
+
@@ List.map
502
+
(fun (cid, storage_str) ->
503
+
(cid, Blob_store.storage_of_string storage_str) )
504
+
results
475
505
476
506
let list_blob_refs t path : Cid.t list Lwt.t =
477
507
Util.use_pool t.db @@ Queries.list_blob_refs path
···
489
519
let clear_blob_refs t path cids : unit Lwt.t =
490
520
if List.is_empty cids then Lwt.return_unit
491
521
else Util.use_pool t.db @@ Queries.clear_blob_refs path cids
522
+
523
+
let update_blob_storage t cid storage : unit Lwt.t =
524
+
let storage_str = Blob_store.storage_to_string storage in
525
+
Util.use_pool t.db @@ Queries.update_blob_storage cid storage_str
526
+
527
+
let list_blobs_by_storage t ~storage ~limit ~cursor :
528
+
(Cid.t * string) list Lwt.t =
529
+
let storage_str = Blob_store.storage_to_string storage in
530
+
Util.use_pool t.db
531
+
@@ Queries.list_blobs_by_storage ~storage:storage_str ~limit ~cursor
+14
pegasus/lib/util.ml
+14
pegasus/lib/util.ml
···
530
530
with _ -> Lwt.return (log_email ()) ) )
531
531
| _ ->
532
532
Lwt.return (log_email ())
533
+
534
+
let s3_error_to_string : Aws_s3_lwt.S3.error -> string = function
535
+
| Redirect endpoint ->
536
+
"redirect to " ^ endpoint.host
537
+
| Throttled ->
538
+
"throttled"
539
+
| Unknown (code, msg) ->
540
+
Printf.sprintf "unknown error %d: %s" code msg
541
+
| Failed exn ->
542
+
Printf.sprintf "failed: %s" (Printexc.to_string exn)
543
+
| Forbidden ->
544
+
"forbidden"
545
+
| Not_found ->
546
+
"not found"