objective categorical abstract machine language personal data server
Apply migrations in transaction
+33
-56
+1
-1
pegasus/lib/data_store.ml
+1
-1
pegasus/lib/data_store.ml
···
269
269
Util.mkfile_p Util.Constants.pegasus_db_filepath ~perm:0o644 ;
270
270
Util.connect_sqlite ?create ?write Util.Constants.pegasus_db_location
271
271
272
-
let init conn : unit Lwt.t = Migrations.run_ds_migrations conn
272
+
let init conn : unit Lwt.t = Migrations.run_migrations Data_store conn
273
273
274
274
let create_actor ~did ~handle ~email ~password ~signing_key conn =
275
275
let password_hash = Bcrypt.hash password |> Bcrypt.string_of_hash in
+31
-54
pegasus/lib/migrations/migrations.ml
+31
-54
pegasus/lib/migrations/migrations.ml
···
1
-
[@@@ocaml.warning "-33"]
2
-
3
1
open Lwt.Infix
4
2
5
3
type migration = {id: int; name: string; applied_at: int}
6
4
7
5
module Queries = struct
8
-
open Util.Rapper
9
-
open Util.Syntax
10
-
11
6
let create_migrations_table =
12
7
[%rapper
13
8
execute
···
62
57
else None
63
58
with _ -> None
64
59
65
-
let run_migration conn path (id, name, sql) =
66
-
let%lwt () = Lwt_io.printlf "running migration %03d: %s" id name in
67
-
let%lwt result = execute_raw path sql in
68
-
let%lwt () =
69
-
match result with Ok () -> Lwt.return_unit | Error e -> raise e
70
-
in
71
-
let applied_at = Util.now_ms () in
72
-
let%lwt () =
73
-
Util.use_pool conn (Queries.record_migration ~id ~name ~applied_at)
74
-
in
75
-
Lwt_io.printlf "migration %03d applied successfully" id
60
+
let run_migration db (id, name, sql) =
61
+
(* I think it's better to do a transaction per migration, no harm in applying the ones that don't error *)
62
+
Util.use_pool db (fun conn ->
63
+
Util.transact conn (fun () ->
64
+
let module C = (val conn : Caqti_lwt.CONNECTION) in
65
+
let query =
66
+
Caqti_request.Infix.( ->. ) Caqti_type.unit Caqti_type.unit sql
67
+
in
68
+
let result = C.exec query () in
69
+
Lwt_result.map
70
+
(fun _ ->
71
+
let applied_at = Util.now_ms () in
72
+
Queries.record_migration ~id ~name ~applied_at conn )
73
+
result ) )
76
74
77
-
let run_ds_migrations conn =
78
-
let%lwt () = Util.use_pool conn Queries.create_migrations_table in
79
-
let%lwt applied =
80
-
Util.use_pool conn Queries.get_applied_migrations
81
-
>|= List.map (fun m -> m.id)
75
+
type migration_type = Data_store | User_store
76
+
77
+
let run_migrations typ conn =
78
+
let read_migration, file_list =
79
+
match typ with
80
+
| Data_store ->
81
+
Data_store_migrations_sql.(read, file_list)
82
+
| User_store ->
83
+
User_store_migrations_sql.(read, file_list)
82
84
in
83
-
let pending =
84
-
List.filter_map
85
-
(fun filename ->
86
-
match parse_migration_filename filename with
87
-
| Some (id, name) when not (List.mem id applied) -> begin
88
-
match Data_store_migrations_sql.read filename with
89
-
| Some sql ->
90
-
Some (id, name, sql)
91
-
| None ->
92
-
None
93
-
end
94
-
| _ ->
95
-
None )
96
-
Data_store_migrations_sql.file_list
97
-
in
98
-
match pending with
99
-
| [] ->
100
-
Lwt.return_unit
101
-
| _ ->
102
-
let%lwt () =
103
-
Lwt_io.printlf "found %d pending migrations" (List.length pending)
104
-
in
105
-
Lwt_list.iter_s
106
-
(run_migration conn Util.Constants.pegasus_db_filepath)
107
-
pending
108
-
109
-
let run_us_migrations conn did =
110
85
let%lwt () = Util.use_pool conn Queries.create_migrations_table in
111
86
let%lwt applied =
112
87
Util.use_pool conn Queries.get_applied_migrations
···
117
92
(fun filename ->
118
93
match parse_migration_filename filename with
119
94
| Some (id, name) when not (List.mem id applied) -> begin
120
-
match User_store_migrations_sql.read filename with
95
+
match read_migration filename with
121
96
| Some sql ->
122
97
Some (id, name, sql)
123
98
| None ->
···
125
100
end
126
101
| _ ->
127
102
None )
128
-
User_store_migrations_sql.file_list
103
+
file_list
129
104
in
130
105
match pending with
131
106
| [] ->
132
107
Lwt.return_unit
133
-
| _ ->
134
-
Lwt_list.iter_s
135
-
(run_migration conn (Util.Constants.user_db_filepath did))
136
-
pending
108
+
| _ -> (
109
+
try%lwt Lwt_list.iter_s (run_migration conn) pending with
110
+
| Caqti_error.Exn e ->
111
+
failwith ("failed to run migrations: " ^ Caqti_error.show e)
112
+
| exn ->
113
+
failwith ("failed to run migrations: " ^ Printexc.to_string exn) )
+1
-1
pegasus/lib/user_store.ml
+1
-1
pegasus/lib/user_store.ml