1	import Base
2	import ../ByteMemTable.bend as ByteMemTable
3	import ../Fs.bend as Fs
4	import ./BackupIo.bend as BackupIo
5	import ./FormatInspect.bend as FormatInspect
6	import ./FormatInspectIo.bend as FormatInspectIo
7	import ./Migrate.bend as Migrate
8	import ./MigrateIo.bend as MigrateIo
9	import ./RestoreIo.bend as RestoreIo
10	
11	# Separates inspection, conversion, bundle, and destination publication failures.
12	type Error is Data:
13	  Inspection{error: FormatInspectIo.Error}
14	  Source{error: MigrateIo.Error}
15	  Rejected{reason: Migrate.Rejection}
16	  Conversion{code: U32, message: String}
17	  Backup{error: BackupIo.Error}
18	  Restore{error: RestoreIo.Error}
19	  Host{code: U32, message: String}
20	  PublicationStateInvalid{}
21	
22	def path.clone.pair.go(path: String, left: String, right: String) -> String & String:
23	  match path:
24	    case SNil{}: (String.reverse(left), String.reverse(right))
25	    case SCon{+character, rest}:
26	      path.clone.pair.go(rest, SCon{character, left}, SCon{character, right})
27	
28	def path.clone.pair(path: String) -> String & String:
29	  path.clone.pair.go(path, SNil{}, SNil{})
30	
31	def path.clone.triple.second(
32	  first: String,
33	  pair: String & String
34	) -> String & String & String:
35	  match pair:
36	    case (second, third): (first, second, third)
37	
38	def path.clone.triple.first(pair: String & String) -> String & String & String:
39	  match pair:
40	    case (first, suffix):
41	      path.clone.triple.second(first, path.clone.pair(suffix))
42	
43	def path.clone.triple(path: String) -> String & String & String:
44	  path.clone.triple.first(path.clone.pair(path))
45	
46	def migration.restored(
47	  result: Result<&1, &1, RestoreIo.Error, Unit>
48	) -> IO(Result<&1, &1, Error, Unit>):
49	  match result:
50	    case Fail{error}:
51	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Restore{error}})
52	    case Done{Unit{}}:
53	      IO.pure(Result<&1, &1, Error, Unit>, Done{Unit{}})
54	
55	def cleanup.bundle(paths: String & String) -> IO(Unit):
56	  match paths:
57	    case (backup_path, directory):
58	      do IO<Unit>:
59	        _backup : Result<&1, &1, U32 & String, Unit> <-
60	          Fs.remove(backup_path ++ "/BACKUP")
61	        _directory : Result<&1, &1, U32 & String, Unit> <- Fs.remove(directory)
62	        return Unit{}
63	
64	def migration.restore.paths(
65	  destination: String,
66	  paths: String & String & String
67	) -> IO(Result<&1, &1, Error, Unit>):
68	  match paths:
69	    case (restore_path, backup_path, directory):
70	      do IO<Result<&1, &1, Error, Unit>>:
71	        restored : Result<&1, &1, RestoreIo.Error, Unit> <-
72	          RestoreIo.restore_from(destination, restore_path)
73	        cleanup.bundle((backup_path, directory))
74	        migration.restored(restored)
75	
76	def migration.restore(
77	  destination: String,
78	  bundle: String
79	) -> IO(Result<&1, &1, Error, Unit>):
80	  migration.restore.paths(destination, path.clone.triple(bundle))
81	
82	def migration.exported(
83	  destination: String,
84	  bundle: String,
85	  result: Result<&1, &1, BackupIo.Error, Unit>
86	) -> IO(Result<&1, &1, Error, Unit>):
87	  match result:
88	    case Fail{error}:
89	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Backup{error}})
90	    case Done{Unit{}}:
91	      migration.restore(destination, bundle)
92	
93	def migration.publication.export(
94	  destination: String,
95	  paths: String & String,
96	  entries: List<&1, ByteMemTable.Entry>
97	) -> IO(Result<&1, &1, Error, Unit>):
98	  match paths:
99	    case (export_path, retained_path):
100	      do IO<Result<&1, &1, Error, Unit>>:
101	        exported : Result<&1, &1, BackupIo.Error, Unit> <-
102	          BackupIo.export(export_path, entries)
103	        migration.exported(destination, retained_path, exported)
104	
105	def migration.publication.started(
106	  result: Migrate.PublicationDecision,
107	  destination: String,
108	  bundle: String,
109	  entries: List<&1, ByteMemTable.Entry>
110	) -> IO(Result<&1, &1, Error, Unit>):
111	  match result:
112	    case Migrate.PublicationAdvanced{Migrate.SourceValidated{}}:
113	      migration.publication.export(destination, path.clone.pair(bundle),
114	        entries)
115	    case _:
116	      IO.pure(Result<&1, &1, Error, Unit>, Fail{PublicationStateInvalid{}})
117	
118	def migration.admitted(
119	  decision: Migrate.Decision,
120	  destination: String,
121	  bundle: String,
122	  entries: List<&1, ByteMemTable.Entry>
123	) -> IO(Result<&1, &1, Error, Unit>):
124	  match decision:
125	    case Migrate.MigrateRejected{reason}:
126	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Rejected{reason}})
127	    case Migrate.MigrateReady{_, _}:
128	      migration.publication.started(Migrate.begin(decision), destination,
129	        bundle, entries)
130	
131	def migration.measured(
132	  inspection: FormatInspect.Inspection,
133	  destination: String,
134	  bundle: String,
135	  encoded_bytes: Nat,
136	  measured: List<&1, ByteMemTable.Entry> & Nat
137	) -> IO(Result<&1, &1, Error, Unit>):
138	  match measured:
139	    case (entries, count):
140	      decision = Migrate.plan(inspection,
141	        Migrate.LegacyFilesVerified{count, encoded_bytes}, True{},
142	        1000000n, 134217728n)
143	      migration.admitted(decision, destination, bundle, entries)
144	
145	def migration.converted(
146	  inspection: FormatInspect.Inspection,
147	  destination: String,
148	  bundle: String,
149	  encoded_bytes: Nat,
150	  result: Result<&1, &1, U32 & String, List<&1, ByteMemTable.Entry>>
151	) -> IO(Result<&1, &1, Error, Unit>):
152	  match result:
153	    case Fail{(code, message)}:
154	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Conversion{code, message}})
155	    case Done{entries}:
156	      migration.measured(inspection, destination, bundle, encoded_bytes,
157	        ByteMemTable.count.keep(entries))
158	
159	def migration.snapshot(
160	  inspection: FormatInspect.Inspection,
161	  destination: String,
162	  bundle: String,
163	  result: Result<&1, &1, MigrateIo.Error, MigrateIo.SourceSnapshot>
164	) -> IO(Result<&1, &1, Error, Unit>):
165	  match result:
166	    case Fail{error}:
167	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Source{error}})
168	    case Done{MigrateIo.LegacySourceSnapshot{sst_records, wal_batches,
169	        _, _, encoded_bytes}}:
170	      migration.converted(inspection, destination, bundle, encoded_bytes,
171	        Migrate.legacy.snapshot.entries(sst_records, wal_batches))
172	
173	def migration.destination.checked(
174	  source: String,
175	  destination: String,
176	  bundle: String,
177	  inspection: FormatInspect.Inspection,
178	  result: Result<&1, &1, U32 & String, Bool>
179	) -> IO(Result<&1, &1, Error, Unit>):
180	  match result:
181	    case Fail{(code, message)}:
182	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Host{code, message}})
183	    case Done{True{}}:
184	      IO.pure(Result<&1, &1, Error, Unit>,
185	        Fail{Rejected{Migrate.DestinationExists{}}})
186	    case Done{False{}}:
187	      do IO<Result<&1, &1, Error, Unit>>:
188	        snapshot : Result<&1, &1, MigrateIo.Error, MigrateIo.SourceSnapshot> <-
189	          MigrateIo.read_legacy_source(source)
190	        migration.snapshot(inspection, destination, bundle, snapshot)
191	
192	def inspection.checked(
193	  paths: String & String & String & String,
194	  result: Result<&1, &1, FormatInspectIo.Error, FormatInspect.Inspection>
195	) -> IO(Result<&1, &1, Error, Unit>):
196	  match paths result:
197	    case (source_read, destination_check, destination, bundle), Fail{error}:
198	      IO.pure(Result<&1, &1, Error, Unit>, Fail{Inspection{error}})
199	    case (source_read, destination_check, destination, bundle), Done{inspection}:
200	      do IO<Result<&1, &1, Error, Unit>>:
201	        exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(destination_check)
202	        migration.destination.checked(source_read, destination, bundle,
203	          inspection, exists)
204	
205	def migrate.paths.destination(
206	  source_inspect: String,
207	  source_read: String,
208	  destination_paths: String & String & String
209	) -> IO(Result<&1, &1, Error, Unit>):
210	  match destination_paths:
211	    case (destination_check, destination, bundle_base):
212	      do IO<Result<&1, &1, Error, Unit>>:
213	        inspection : Result<&1, &1, FormatInspectIo.Error,
214	          FormatInspect.Inspection> <- FormatInspectIo.inspect(source_inspect)
215	        inspection.checked((source_read, destination_check, destination,
216	          bundle_base ++ ".migrate-060.bundle"), inspection)
217	
218	def migrate.paths(
219	  source_paths: String & String,
220	  destination_paths: String & String & String
221	) -> IO(Result<&1, &1, Error, Unit>):
222	  match source_paths:
223	    case (source_inspect, source_read):
224	      migrate.paths.destination(source_inspect, source_read, destination_paths)
225	
226	# Converts and atomically publishes one verified legacy source snapshot.
227	def migrate_legacy_source(
228	  +source: String,
229	  +destination: String
230	) -> IO(Result<&1, &1, Error, Unit>):
231	  migrate.paths(path.clone.pair(source), path.clone.triple(destination))