1	import Base
2	import ../Fs.bend as Fs
3	import ../scan/VisibleCursor.bend as VisibleCursor
4	import ./BackupCursor.bend as BackupCursor
5	import ./BackupIo.bend as BackupIo
6	import ./Backup.bend as Backup
7	# Carries one pending read, bounded chunk write, or completed file write.
8	type State is Type:
9	  Need{file: File, stream: BackupCursor.Stream}
10	  Output{file: File, step: BackupCursor.Step}
11	  Written{pair: File & Result<&1, &1, U32 & String, Unit>, stream: BackupCursor.Stream}
12	
13	def path.reverse(path: String, reversed: String) -> String:
14	  match path:
15	    case SNil{}: reversed
16	    case SCon{character, rest}: path.reverse(rest, SCon{character, reversed})
17	
18	def path.clone(path: String) -> String:
19	  path.reverse(path.reverse(path, SNil{}), SNil{})
20	
21	def error.host(error: Backup.Error) -> U32 & String:
22	  match error:
23	    case Backup.UnsupportedPlatform{}:
24	      (95, "backup source transport is unsupported")
25	    case Backup.SourceFailure{code, message}: (code, message)
26	    case Backup.ResourceLimit{}: (7, "backup cursor exceeded its bounded policy")
27	    case Backup.InvalidOrder{}: (22, "backup cursor rejected noncanonical order")
28	    case Backup.CodecFailure{}: (22, "backup cursor encoder rejected its source")
29	
30	def error.io.copied(pair: U32 & String) -> BackupIo.Error:
31	  match pair:
32	    case (code, message): BackupIo.Io{code, message}
33	
34	def error.io(error: Backup.Error) -> BackupIo.Error:
35	  match error:
36	    case Backup.UnsupportedPlatform{}: BackupIo.UnsupportedPlatform{}
37	    case _: error.io.copied(error.host(error))
38	
39	# Removes the known staging file and directory after failed publication.
40	# Removes the known staging file and directory after failed publication.
41	def cleanup(+directory: String) -> IO(Unit):
42	  do IO<Unit>:
43	    _file : Result<&1, &1, U32 & String, Unit> <- Fs.remove(directory ++ "/BACKUP")
44	    _directory : Result<&1, &1, U32 & String, Unit> <- Fs.remove(directory)
45	    return Unit{}
46	
47	def close.source(
48	  file: File,
49	  +directory: String,
50	  cursor: VisibleCursor.Cursor,
51	  error: U32 & String
52	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
53	  do IO<Result<&1, &1, BackupIo.Error, Unit>>:
54	    _closed_cursor : Result<&1, &1, U32 & String, Unit> <- VisibleCursor.close(cursor)
55	    BackupIo.close.failed(file, directory, error)
56	
57	def close.stream(
58	  file: File,
59	  +directory: String,
60	  stream: BackupCursor.Stream,
61	  error: U32 & String
62	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
63	  match stream:
64	    case BackupCursor.Stream{_, Some{cursor}, _, _, _, _, _, _, _}:
65	      close.source(file, directory, cursor, error)
66	    case BackupCursor.Stream{_, None{}, _, _, _, _, _, _, _}:
67	      BackupIo.close.failed(file, directory, error)
68	
69	def close.optional(
70	  file: File,
71	  +directory: String,
72	  cursor: Maybe<&1, VisibleCursor.Cursor>,
73	  error: U32 & String
74	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
75	  match cursor:
76	    case Some{reader}: close.source(file, directory, reader, error)
77	    case None{}: BackupIo.close.failed(file, directory, error)
78	
79	def close.resource(
80	  file: File,
81	  +directory: String,
82	  cursor: Maybe<&1, VisibleCursor.Cursor>
83	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
84	  match cursor:
85	    case Some{reader}:
86	      do IO<Result<&1, &1, BackupIo.Error, Unit>>:
87	        _closed_cursor : Result<&1, &1, U32 & String, Unit> <- VisibleCursor.close(reader)
88	        _closed_file : Unit <- File.close(file)
89	        cleanup(directory)
90	        return Fail{BackupIo.ResourceLimit{}}
91	    case None{}:
92	      do IO<Result<&1, &1, BackupIo.Error, Unit>>:
93	        _closed_file : Unit <- File.close(file)
94	        cleanup(directory)
95	        return Fail{BackupIo.ResourceLimit{}}
96	
97	def close.resource.stream(
98	  file: File,
99	  +directory: String,
100	  stream: BackupCursor.Stream
101	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
102	  match stream:
103	    case BackupCursor.Stream{_, cursor, _, _, _, _, _, _, _}:
104	      close.resource(file, directory, cursor)
105	
106	# Closes backup resources and preserves an unsupported source platform error.
107	def close.unsupported(
108	  file: File,
109	  +directory: String,
110	  cursor: Maybe<&1, VisibleCursor.Cursor>
111	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
112	  match cursor:
113	    case Some{reader}:
114	      do IO<Result<&1, &1, BackupIo.Error, Unit>>:
115	        _closed_cursor : Result<&1, &1, U32 & String, Unit> <-
116	          VisibleCursor.close(reader)
117	        _closed_file : Unit <- File.close(file)
118	        cleanup(directory)
119	        return Fail{BackupIo.UnsupportedPlatform{}}
120	    case None{}:
121	      do IO<Result<&1, &1, BackupIo.Error, Unit>>:
122	        _closed_file : Unit <- File.close(file)
123	        cleanup(directory)
124	        return Fail{BackupIo.UnsupportedPlatform{}}
125	
126	def close.error.host(
127	  file: File,
128	  +directory: String,
129	  cursor: Maybe<&1, VisibleCursor.Cursor>,
130	  pair: U32 & String
131	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
132	  match pair:
133	    case (code, message): close.optional(file, directory, cursor, (code, message))
134	
135	def close.error(
136	  file: File,
137	  +directory: String,
138	  cursor: Maybe<&1, VisibleCursor.Cursor>,
139	  error: Backup.Error
140	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
141	  match error:
142	    case Backup.ResourceLimit{}: close.resource(file, directory, cursor)
143	    case Backup.UnsupportedPlatform{}:
144	      close.unsupported(file, directory, cursor)
145	    case _: close.error.host(file, directory, cursor, error.host(error))
146	
147	def close.cursor(cursor: Maybe<&1, VisibleCursor.Cursor>) -> IO(Unit):
148	  match cursor:
149	    case Some{reader}:
150	      do IO<Unit>:
151	        _closed : Result<&1, &1, U32 & String, Unit> <- VisibleCursor.close(reader)
152	        return Unit{}
153	    case None{}: IO.pure(Unit, Unit{})
154	
155	def close.initial.stream(stream: BackupCursor.Stream) -> IO(Unit):
156	  match stream:
157	    case BackupCursor.Stream{_, cursor, _, _, _, _, _, _, _}:
158	      close.cursor(cursor)
159	
160	def close.initial(initial: BackupCursor.Step) -> IO(Unit):
161	  match initial:
162	    case BackupCursor.Chunk{stream, _}: close.initial.stream(stream)
163	    case BackupCursor.Continue{stream}: close.initial.stream(stream)
164	    case BackupCursor.Failure{cursor, _}: close.cursor(cursor)
165	    case BackupCursor.End{}: IO.pure(Unit, Unit{})
166	
167	def initial.failed(
168	  initial: BackupCursor.Step,
169	  error: BackupIo.Error
170	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
171	  do IO<Result<&1, &1, BackupIo.Error, Unit>>:
172	    _closed : Unit <- close.initial(initial)
173	    return Fail{error}
174	
175	def close.file(
176	  +destination: String,
177	  +temporary: String,
178	  file: File
179	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
180	  BackupIo.publish.closed(destination, temporary, file, Done{Unit{}})
181	
182	def source.step(stream: BackupCursor.Stream) -> IO(BackupCursor.Step):
183	  match stream:
184	    case BackupCursor.Stream{BackupCursor.CursorBlocks{}, Some{cursor}, pending,
185	        pending_count, seen, expected, entry_count, block_count, block_index}:
186	      metadata : BackupCursor.Stream = BackupCursor.Stream{BackupCursor.CursorBlocks{}, None{},
187	        pending, pending_count, seen, expected, entry_count, block_count,
188	        block_index}
189	      do IO<BackupCursor.Step>:
190	        next : VisibleCursor.Step <- VisibleCursor.next(cursor)
191	        return BackupCursor.advance(metadata, next)
192	    case _:
193	      IO.pure(BackupCursor.Step, BackupCursor.advance(stream, VisibleCursor.End{}))
194	
195	def loop.abort(
196	  +temporary: String,
197	  state: State
198	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
199	  match state:
200	    case Need{file, stream}:
201	      close.resource.stream(file, temporary, stream)
202	    case Output{file, step}:
203	      match step:
204	        case BackupCursor.Failure{cursor, error}:
205	          close.error(file, temporary, cursor, error)
206	        case BackupCursor.Chunk{stream, _}:
207	          close.resource.stream(file, temporary, stream)
208	        case BackupCursor.Continue{stream}:
209	          close.resource.stream(file, temporary, stream)
210	        case BackupCursor.End{}:
211	          close.resource(file, temporary, None{})
212	    case Written{pair, stream}:
213	      match pair:
214	        case (file, Fail{error}): close.stream(file, temporary, stream, error)
215	        case (file, Done{Unit{}}):
216	          close.resource.stream(file, temporary, stream)
217	
218	# Pulls at most one visible cursor result and writes at most one encoded chunk.
219	def loop(
220	  fuel: Nat,
221	  +destination: String,
222	  +temporary: String,
223	  state: State
224	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
225	  match fuel:
226	    case 0n: loop.abort(temporary, state)
227	    case 1n+rest:
228	      match state:
229	        case Need{file, stream}:
230	          do IO<Result<&1, &1, BackupIo.Error, Unit>>:
231	            step : BackupCursor.Step <- source.step(stream)
232	            loop(rest, destination, temporary, Output{file, step})
233	        case Output{file, step}:
234	          match step:
235	            case BackupCursor.Failure{cursor, error}:
236	              close.error(file, temporary, cursor, error)
237	            case BackupCursor.Continue{stream}:
238	              loop(rest, destination, temporary, Need{file, stream})
239	            case BackupCursor.Chunk{stream, bytes}:
240	              do IO<Result<&1, &1, BackupIo.Error, Unit>>:
241	                written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, bytes)
242	                loop(rest, destination, temporary, Written{written, stream})
243	            case BackupCursor.End{}:
244	              close.file(destination, temporary, file)
245	        case Written{pair, stream}:
246	          match pair:
247	            case (file, Fail{error}):
248	              close.stream(file, temporary, stream, error)
249	            case (file, Done{Unit{}}):
250	              loop(rest, destination, temporary, Need{file, stream})
251	
252	def file.opened(
253	  +destination: String,
254	  +temporary: String,
255	  initial: BackupCursor.Step,
256	  fuel: Nat,
257	  opened: Result<&1, &1, U32 & String, File>
258	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
259	  match opened:
260	    case Fail{(code, message)}:
261	      do IO<Result<&1, &1, BackupIo.Error, Unit>>:
262	        _closed : Unit <- close.initial(initial)
263	        cleanup(temporary)
264	        return Fail{BackupIo.Io{code, message}}
265	    case Done{file}:
266	      loop(fuel, destination, temporary, Output{file, initial})
267	
268	def directory.created(
269	  +destination: String,
270	  +temporary: String,
271	  initial: BackupCursor.Step,
272	  fuel: Nat,
273	  created: Result<&1, &1, U32 & String, Unit>
274	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
275	  match created:
276	    case Fail{(code, message)}:
277	      match code:
278	        case 17: initial.failed(initial, BackupIo.AlreadyExists{})
279	        case _: initial.failed(initial, BackupIo.Io{code, message})
280	    case Done{Unit{}}:
281	      do IO<Result<&1, &1, BackupIo.Error, Unit>>:
282	        opened : Result<&1, &1, U32 & String, File> <- File.open(path.clone(temporary ++ "/BACKUP"), "w")
283	        file.opened(destination, temporary, initial, fuel, opened)
284	
285	def destination.checked(
286	  +destination: String,
287	  +temporary: String,
288	  initial: BackupCursor.Step,
289	  fuel: Nat,
290	  exists: Result<&1, &1, U32 & String, Bool>
291	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
292	  match exists:
293	    case Fail{(code, message)}:
294	      initial.failed(initial, BackupIo.Io{code, message})
295	    case Done{True{}}:
296	      initial.failed(initial, BackupIo.AlreadyExists{})
297	    case Done{False{}}:
298	      do IO<Result<&1, &1, BackupIo.Error, Unit>>:
299	        created : Result<&1, &1, U32 & String, Unit> <- Fs.make_dir(path.clone(temporary))
300	        directory.created(destination, temporary, initial, fuel, created)
301	
302	def initial.checked(
303	  +destination: String,
304	  +temporary: String,
305	  initial: BackupCursor.Step,
306	  fuel: Nat,
307	  exists: Result<&1, &1, U32 & String, Bool>
308	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
309	  match initial:
310	    case BackupCursor.Failure{cursor, error}:
311	      do IO<Result<&1, &1, BackupIo.Error, Unit>>:
312	        _closed : Unit <- close.cursor(cursor)
313	        return Fail{error.io(error)}
314	    case BackupCursor.Chunk{_, _}:
315	      destination.checked(destination, temporary, initial, fuel, exists)
316	    case _:
317	      initial.failed(initial, BackupIo.Io{22, "backup cursor failed to encode its header"})
318	
319	# Exports a stable visible cursor through bounded checksummed chunks.
320	def export(
321	  +destination: String,
322	  cursor: VisibleCursor.Cursor,
323	  entry_count: Nat,
324	  fuel: Nat
325	) -> IO(Result<&1, &1, BackupIo.Error, Unit>):
326	  +temporary = destination ++ ".backup-060.tmp"
327	  do IO<Result<&1, &1, BackupIo.Error, Unit>>:
328	    initial : BackupCursor.Step <- IO.pure(BackupCursor.Step,
329	      BackupCursor.start(cursor, entry_count))
330	    exists : Result<&1, &1, U32 & String, Bool> <- Fs.exists(path.clone(destination))
331	    initial.checked(destination, temporary, initial, fuel, exists)