1 (* Applies the versioned schema migrations. The migrations themselves live in 2 {!Migration_data}: the SQL and the ledger table. This module holds only the 3 logic that runs them. That logic is the ledger check that makes applying 4 idempotent, and the per-migration transaction. *) 5 6 type migration = Migration_data.t = { 7 version : int; 8 name : string; 9 statements : string list; 10 } 11 12 let migrations = Migration_data.migrations 13 let ledger = Migration_data.ledger 14 15 let statements = 16 List.concat_map (fun migration -> migration.statements) migrations 17 18 (* --- request helpers --- *) 19 20 let exec_sql (module Db : Caqti_lwt.CONNECTION) sql = 21 let request = 22 let open Caqti_request.Infix in 23 let open Caqti_type.Std in 24 (unit ->. unit) ~oneshot:true sql 25 in 26 Db.exec request () 27 28 let is_applied (module Db : Caqti_lwt.CONNECTION) version = 29 let request = 30 let open Caqti_request.Infix in 31 let open Caqti_type.Std in 32 (int ->! int) ~oneshot:true 33 "SELECT COUNT(*) FROM schema_migrations WHERE version = ?" 34 in 35 Db.find request version 36 37 let record_applied (module Db : Caqti_lwt.CONNECTION) migration = 38 let request = 39 let open Caqti_request.Infix in 40 let open Caqti_type.Std in 41 (t2 int string ->. unit) 42 ~oneshot:true 43 "INSERT INTO schema_migrations (version, name, applied_at) VALUES (?, ?, \ 44 strftime('%s','now'))" 45 in 46 Db.exec request (migration.version, migration.name) 47 48 (* Apply one migration inside a transaction, then record it. The ledger check 49 makes this idempotent. The [IF NOT EXISTS] clauses keep an already-populated 50 file from a pre-ledger build safe. *) 51 let apply_migration ((module Db : Caqti_lwt.CONNECTION) as db) migration = 52 let open Lwt_result.Syntax in 53 let* applied = is_applied db migration.version in 54 if applied > 0 then Lwt_result.return () 55 else 56 Db.with_transaction (fun () -> 57 let rec go = function 58 | [] -> record_applied db migration 59 | sql :: tl -> 60 let* () = exec_sql db sql in 61 go tl 62 in 63 go migration.statements) 64 65 let apply (module Db : Caqti_lwt.CONNECTION) = 66 let db = (module Db : Caqti_lwt.CONNECTION) in 67 let open Lwt_result.Syntax in 68 let run = 69 let* () = exec_sql db ledger in 70 let rec go = function 71 | [] -> Lwt_result.return () 72 | migration :: tl -> 73 let* () = apply_migration db migration in 74 go tl 75 in 76 go migrations 77 in 78 Lwt.map 79 (function Ok () -> Ok () | Error e -> Error (e :> Caqti_error.t)) 80 run 81