1use std::collections::BTreeMap;
2use std::path::PathBuf;
3
4use tracing::log;
5
6use crate::Error;
7use crate::Res;
8use crate::error::LoginError;
9use crate::error::PackageOpError;
10use crate::flow;
11use crate::flow::PullOutcome;
12use crate::flow::UserMeta;
13use crate::flow::cache_remote_manifest;
14use crate::io::remote::HostConfig;
15use crate::io::remote::Remote;
16use crate::io::remote::RemoteS3;
17use crate::io::remote::WORKFLOWS_CONFIG_KEY;
18use crate::io::remote::WorkflowIntent;
19use crate::io::remote::WorkflowsConfig;
20use crate::io::remote::fetch_workflow_rules;
21use crate::io::remote::fetch_workflows_config;
22use crate::io::remote::resolve_workflow;
23use crate::io::remote::resolve_workflow_from_config;
24use crate::io::storage::LocalStorage;
25use crate::io::storage::Storage;
26use crate::lineage;
27use crate::lineage::CommitState;
28use crate::lineage::InstalledPackageStatus;
29use crate::lineage::LineagePaths;
30use crate::lineage::UpstreamState;
31use crate::manifest::Manifest;
32use crate::manifest::Workflow;
33use crate::paths;
34use crate::paths::copy_cached_to_installed;
35use crate::workflow::WorkflowRules;
36use quilt_uri::Host;
37use quilt_uri::ManifestUri;
38use quilt_uri::Namespace;
39use quilt_uri::S3Uri;
40use quilt_uri::UriError;
41
42pub struct PushOutcome {
44 pub manifest_uri: ManifestUri,
45 pub certified_latest: bool,
49}
50
51pub type PublishOutcome = flow::PublishOutcome<PushOutcome>;
55
56#[derive(Debug, Default)]
66pub struct SetRemoteOutcome {
67 pub resolution_warning: Option<String>,
68}
69
70#[derive(Debug)]
75pub struct InstalledPackage<S: Storage = LocalStorage, R: Remote = RemoteS3> {
76 pub lineage: lineage::PackageLineageIo,
77 pub paths: paths::DomainPaths,
78 pub remote: R,
79 pub storage: S,
80 pub namespace: Namespace,
81}
82
83impl std::fmt::Display for InstalledPackage {
84 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85 write!(f, r#"Installed package "{}""#, self.namespace)
86 }
87}
88
89impl<S: Storage + Sync, R: Remote> InstalledPackage<S, R> {
90 pub async fn scaffold_paths(&self) -> Res {
91 let home = self.lineage.domain_home(&self.storage).await?;
92 self.paths
93 .scaffold_for_installing(&self.storage, &home, &self.namespace)
94 .await
95 }
96
97 pub async fn scaffold_paths_for_caching(&self, bucket: &str) -> Res {
98 self.paths.scaffold_for_caching(&self.storage, bucket).await
99 }
100
101 pub async fn manifest(&self) -> Res<Manifest> {
102 let (_, lineage) = self.lineage.read(&self.storage).await?;
103 let Some(hash) = lineage.current_hash() else {
104 return Ok(Manifest::default());
105 };
106 let installed_path = self.paths.installed_manifest(&self.namespace, hash);
107 match Manifest::from_path(&self.storage, &installed_path).await {
108 Ok(manifest) => return Ok(manifest),
109
110 Err(e) => {
111 log::warn!(
112 "Failed to read installed manifest at {}: {}",
113 installed_path.display(),
114 e
115 );
116 }
117 }
118
119 match lineage.remote_uri.as_ref() {
121 Some(remote_uri) => {
122 log::info!("Attempting to recover from cache at {remote_uri}");
123 let cached_manifest =
124 cache_remote_manifest(&self.paths, &self.storage, &self.remote, remote_uri)
125 .await?;
126 copy_cached_to_installed(&self.paths, &self.storage, remote_uri).await?;
127 Ok(cached_manifest)
128 }
129 None => Err(Error::Uri(UriError::ManifestPath(
130 "No installed manifest and no remote to recover from".to_string(),
131 ))),
132 }
133 }
134
135 pub async fn lineage(&self) -> Res<lineage::PackageLineage> {
136 let (_, lineage) = self.lineage.read(&self.storage).await?;
137 Ok(lineage)
138 }
139
140 pub async fn package_home(&self) -> Res<PathBuf> {
141 self.lineage.package_home(&self.storage).await
142 }
143
144 pub async fn recompute_local_status(
150 &self,
151 host_config_opt: Option<HostConfig>,
152 ) -> Res<InstalledPackageStatus> {
153 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
154 let manifest = self.manifest().await?;
155
156 let host_config = match host_config_opt {
157 Some(hc) => hc,
158 None => match lineage.remote_uri.as_ref() {
159 Some(remote_uri) if !remote_uri.bucket.is_empty() => {
160 self.remote.host_config(remote_uri.origin.as_ref()).await?
161 }
162 _ => HostConfig::default(),
163 },
164 };
165
166 let (_, status) = flow::status(
167 lineage,
168 &self.storage,
169 &manifest,
170 &package_home,
171 host_config,
172 )
173 .await?;
174 Ok(status)
175 }
176
177 pub async fn status(&self, host_config_opt: Option<HostConfig>) -> Res<InstalledPackageStatus> {
178 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
179
180 let lineage = match lineage.remote_uri.as_ref() {
182 Some(_) => match flow::refresh_latest_hash(lineage.clone(), &self.remote).await {
183 Ok(lineage) => lineage,
184 Err(Error::Login(LoginError::Required(_))) => {
185 return Err(Error::Login(LoginError::Required(
186 lineage.remote_uri.as_ref().and_then(|r| r.origin.clone()),
187 )));
188 }
189 Err(err) if err.is_access_denied() => return Err(err),
198 Err(err) => {
199 log::warn!("Failed to refresh latest hash: {err}");
200 lineage
201 }
202 },
203 None => lineage,
204 };
205 let manifest = self.manifest().await?;
206
207 let host_config = match host_config_opt {
208 Some(hc) => hc,
209 None => match lineage.remote_uri.as_ref() {
210 Some(remote_uri) if !remote_uri.bucket.is_empty() => {
211 self.remote.host_config(remote_uri.origin.as_ref()).await?
212 }
213 _ => HostConfig::default(),
214 },
215 };
216
217 let (_, status) = flow::status(
218 lineage,
219 &self.storage,
220 &manifest,
221 &package_home,
222 host_config,
223 )
224 .await?;
225 Ok(status)
226 }
227
228 pub async fn install_paths(&self, paths: &[PathBuf]) -> Res<LineagePaths> {
229 if paths.is_empty() {
230 return Ok(BTreeMap::new());
231 }
232
233 self.scaffold_paths().await?;
234
235 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
236 let remote_uri = lineage.remote()?;
237
238 self.scaffold_paths_for_caching(&remote_uri.bucket).await?;
239
240 let mut manifest = self.manifest().await?;
241 let lineage = flow::install_paths(
242 lineage,
243 &mut manifest,
244 &self.paths,
245 package_home,
246 self.namespace.clone(),
247 &self.storage,
248 &self.remote,
249 &paths.iter().collect::<Vec<&PathBuf>>(),
250 )
251 .await?;
252 let lineage = self.lineage.write(&self.storage, lineage).await?;
253 Ok(lineage.paths)
254 }
255
256 pub async fn uninstall_paths(&self, paths: &Vec<PathBuf>) -> Res<LineagePaths> {
257 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
258 let lineage = flow::uninstall_paths(lineage, package_home, &self.storage, paths).await?;
259 let lineage = self.lineage.write(&self.storage, lineage).await?;
260 Ok(lineage.paths)
261 }
262
263 pub async fn revert_paths(&self, paths: &Vec<String>) -> Res {
264 log::debug!("revert_paths: {paths:?}");
265 unimplemented!()
266 }
267
268 pub async fn commit(
274 &self,
275 message: String,
276 user_meta: UserMeta,
277 workflow: Option<Workflow>,
278 host_config_opt: Option<HostConfig>,
279 ) -> Res<CommitState> {
280 self.scaffold_paths().await?;
281
282 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
283 let mut manifest = self.manifest().await?;
284
285 let host_config = match host_config_opt {
286 Some(hc) => hc,
287 None => match lineage.remote_uri.as_ref() {
288 Some(remote_uri) if !remote_uri.bucket.is_empty() => {
289 self.remote.host_config(remote_uri.origin.as_ref()).await?
290 }
291 _ => HostConfig::default(),
292 },
293 };
294
295 let host = host_config.host.clone();
299
300 let (lineage, status) = flow::status(
301 lineage,
302 &self.storage,
303 &manifest,
304 &package_home,
305 host_config,
306 )
307 .await?;
308
309 let (lineage, commit) = flow::commit(
310 lineage,
311 &mut manifest,
312 &self.paths,
313 &self.storage,
314 &self.remote,
315 host.as_ref(),
316 package_home,
317 status,
318 self.namespace.clone(),
319 message,
320 user_meta,
321 workflow,
322 )
323 .await?;
324 self.lineage.write(&self.storage, lineage).await?;
325 Ok(commit)
326 }
327
328 pub async fn publish(
340 &self,
341 message: String,
342 user_meta: UserMeta,
343 workflow: Option<Workflow>,
344 host_config_opt: Option<HostConfig>,
345 status_opt: Option<InstalledPackageStatus>,
346 ) -> Res<PublishOutcome> {
347 self.scaffold_paths().await?;
348
349 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
350 let remote_uri = match lineage.remote_uri.as_ref() {
351 Some(uri) if !uri.bucket.is_empty() => uri.clone(),
352 Some(_) => {
353 return Err(Error::PackageOp(PackageOpError::Publish(
354 "Remote bucket not set. Use set_remote first.".to_string(),
355 )));
356 }
357 None => {
358 return Err(Error::PackageOp(PackageOpError::Publish(
359 "No remote configured. Use set_remote first.".to_string(),
360 )));
361 }
362 };
363
364 self.scaffold_paths_for_caching(&remote_uri.bucket).await?;
365
366 let mut manifest = self.manifest().await?;
367 let host_config =
368 host_config_opt.unwrap_or(self.remote.host_config(remote_uri.origin.as_ref()).await?);
369
370 let (lineage, status) = match status_opt {
371 Some(status) => (lineage, status),
372 None => {
373 flow::status(
374 lineage,
375 &self.storage,
376 &manifest,
377 &package_home,
378 host_config.clone(),
379 )
380 .await?
381 }
382 };
383
384 let outcome = flow::publish(
385 lineage,
386 &mut manifest,
387 &self.paths,
388 &self.storage,
389 &self.remote,
390 package_home,
391 status,
392 self.namespace.clone(),
393 host_config,
394 flow::CommitOptions {
395 message,
396 user_meta,
397 workflow,
398 },
399 )
400 .await?;
401
402 let (committed, push_result) = match outcome {
403 flow::PublishOutcome::CommittedAndPushed(p) => (true, p),
404 flow::PublishOutcome::PushedOnly(p) => (false, p),
405 };
406 let certified_latest = push_result.certified_latest;
407 let lineage = self
408 .lineage
409 .write(&self.storage, push_result.lineage)
410 .await?;
411 let push = PushOutcome {
412 manifest_uri: lineage.remote()?.clone(),
413 certified_latest,
414 };
415 Ok(if committed {
416 PublishOutcome::CommittedAndPushed(push)
417 } else {
418 PublishOutcome::PushedOnly(push)
419 })
420 }
421
422 pub async fn push(&self, host_config_opt: Option<HostConfig>) -> Res<PushOutcome> {
424 self.scaffold_paths().await?;
425
426 let (_, lineage) = self.lineage.read(&self.storage).await?;
427 let remote_uri = match lineage.remote_uri.as_ref() {
428 Some(uri) if !uri.bucket.is_empty() => uri.clone(),
429 Some(_) => {
430 return Err(Error::PackageOp(PackageOpError::Push(
431 "Remote bucket not set. Use set_remote first.".to_string(),
432 )));
433 }
434 None => {
435 return Err(Error::PackageOp(PackageOpError::Push(
436 "No remote configured. Use set_remote first.".to_string(),
437 )));
438 }
439 };
440
441 if lineage.commit.is_none() {
442 return Err(Error::PackageOp(PackageOpError::Push(
443 "No commits to push".to_string(),
444 )));
445 }
446
447 self.scaffold_paths_for_caching(&remote_uri.bucket).await?;
448
449 let manifest = self.manifest().await?;
450
451 let host_config =
452 host_config_opt.unwrap_or(self.remote.host_config(remote_uri.origin.as_ref()).await?);
453
454 let result = flow::push(
455 lineage,
456 manifest,
457 &self.paths,
458 &self.storage,
459 &self.remote,
460 Some(self.namespace.clone()),
461 host_config,
462 )
463 .await?;
464 let certified_latest = result.certified_latest;
465 let lineage = self.lineage.write(&self.storage, result.lineage).await?;
466 Ok(PushOutcome {
467 manifest_uri: lineage.remote()?.clone(),
468 certified_latest,
469 })
470 }
471
472 pub async fn pull(&self, host_config_opt: Option<HostConfig>) -> Res<ManifestUri> {
473 self.scaffold_paths().await?;
474
475 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
476 let remote_uri = lineage.remote()?.clone();
477
478 self.scaffold_paths_for_caching(&remote_uri.bucket).await?;
479
480 let mut manifest = self.manifest().await?;
481
482 let host_config =
483 host_config_opt.unwrap_or(self.remote.host_config(remote_uri.origin.as_ref()).await?);
484
485 let (lineage, snapshot) = flow::snapshot_for_pull(
488 lineage,
489 &manifest,
490 &self.paths,
491 &self.storage,
492 &self.remote,
493 &package_home,
494 host_config,
495 )
496 .await?;
497 let lineage = flow::pull(
498 lineage,
499 &mut manifest,
500 &self.paths,
501 &self.storage,
502 &self.remote,
503 package_home,
504 snapshot,
505 self.namespace.clone(),
506 )
507 .await?;
508 let lineage = self.lineage.write(&self.storage, lineage).await?;
509 Ok(lineage.remote()?.clone())
510 }
511
512 pub async fn pull_outcome(&self, host_config_opt: Option<HostConfig>) -> Res<PullOutcome> {
539 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
540
541 if UpstreamState::from(lineage.clone()) == UpstreamState::Local {
548 return Ok(PullOutcome::UpToDate);
549 }
550
551 if UpstreamState::from(lineage.clone()) == UpstreamState::Diverged {
566 return Ok(PullOutcome::UpToDate);
567 }
568
569 let remote_uri = lineage.remote()?.clone();
570 let base = self.manifest().await?;
571 let host_config =
572 host_config_opt.unwrap_or(self.remote.host_config(remote_uri.origin.as_ref()).await?);
573
574 let snapshot = match flow::snapshot_for_pull(
579 lineage,
580 &base,
581 &self.paths,
582 &self.storage,
583 &self.remote,
584 &package_home,
585 host_config,
586 )
587 .await
588 {
589 Ok((_, snapshot)) => snapshot,
590 Err(Error::PackageOp(PackageOpError::AlreadyUpToDate)) => {
591 return Ok(PullOutcome::UpToDate);
592 }
593 Err(err) => return Err(err),
594 };
595
596 if snapshot.status.upstream_state != UpstreamState::Behind {
600 return Ok(PullOutcome::UpToDate);
601 }
602 Ok(flow::classify_pull(
603 &snapshot.status,
604 &base,
605 &snapshot.latest_manifest,
606 ))
607 }
608
609 pub async fn certify_latest(&self) -> Res<ManifestUri> {
615 let (_, lineage) = self.lineage.read(&self.storage).await?;
616
617 let lineage = if lineage.commit.is_some() {
620 self.push(None).await?;
621 self.lineage.read(&self.storage).await?.1
622 } else {
623 lineage
624 };
625
626 let pushed_manifest_uri = lineage.remote()?.clone();
627 let lineage = flow::certify_latest(lineage, &self.remote, pushed_manifest_uri).await?;
628 let lineage = self.lineage.write(&self.storage, lineage).await?;
629 Ok(lineage.remote()?.clone())
630 }
631
632 pub async fn reset_to_latest(&self) -> Res<ManifestUri> {
633 self.scaffold_paths().await?;
634
635 let (package_home, lineage) = self.lineage.read(&self.storage).await?;
636 let remote_uri = lineage.remote()?.clone();
637
638 self.scaffold_paths_for_caching(&remote_uri.bucket).await?;
639
640 let mut manifest = self.manifest().await?;
641 let lineage = flow::reset_to_latest(
642 lineage,
643 &mut manifest,
644 &self.paths,
645 &self.storage,
646 &self.remote,
647 package_home,
648 self.namespace.clone(),
649 )
650 .await?;
651 let lineage = self.lineage.write(&self.storage, lineage).await?;
652 Ok(lineage.remote()?.clone())
653 }
654
655 pub async fn set_remote(
656 &self,
657 bucket: String,
658 origin: Option<Host>,
659 workflow: WorkflowIntent,
660 ) -> Res<SetRemoteOutcome> {
661 if bucket.is_empty() {
662 return Err(Error::PackageOp(PackageOpError::Push(
663 "Bucket cannot be empty".to_string(),
664 )));
665 }
666 let (_, mut lineage) = self.lineage.read(&self.storage).await?;
667 if let Some(existing) = &lineage.remote_uri
668 && !existing.hash.is_empty()
669 {
670 let same_remote = existing.bucket == bucket && existing.origin == origin;
671 if same_remote {
672 return Ok(SetRemoteOutcome::default());
673 }
674 return Err(Error::PackageOp(PackageOpError::Push(
675 "Cannot change remote on a package that has already been pushed".to_string(),
676 )));
677 }
678 self.remote.verify_bucket(&bucket).await?;
683 lineage.remote_uri = Some(ManifestUri {
684 origin: origin.clone(),
685 bucket: bucket.clone(),
686 namespace: self.namespace.clone(),
687 hash: String::new(),
688 });
689
690 let explicit_workflow = !matches!(&workflow, WorkflowIntent::BucketDefault);
695
696 if let Some(origin) = origin
702 && lineage.commit.is_some()
703 {
704 return match self
705 .recommit_for_remote(lineage.clone(), origin, bucket, workflow)
706 .await
707 {
708 Ok(()) => Ok(SetRemoteOutcome::default()),
709 Err(err @ Error::WorkflowValidation(_)) => Err(err),
714 Err(err) => {
715 self.lineage.write(&self.storage, lineage).await?;
719 if explicit_workflow {
720 return Err(err);
725 }
726 let reason = match &err {
736 Error::RemoteCatalog(inner) => inner.to_string(),
737 _ => err.to_string(),
738 };
739 log::warn!(
740 "Remote saved but recommit failed ({reason}); re-run Set Remote (e.g. after logging in) to complete it before pushing."
741 );
742 Ok(SetRemoteOutcome {
743 resolution_warning: Some(reason),
744 })
745 }
746 };
747 }
748
749 self.lineage.write(&self.storage, lineage).await?;
751
752 Ok(SetRemoteOutcome::default())
753 }
754
755 async fn recommit_for_remote(
756 &self,
757 lineage: lineage::PackageLineage,
758 origin: Host,
759 bucket: String,
760 workflow: WorkflowIntent,
761 ) -> Res {
762 let host = Some(origin);
763 let host_config = self.remote.host_config(host.as_ref()).await?;
764 let workflows_config_uri = S3Uri {
765 key: WORKFLOWS_CONFIG_KEY.to_string(),
766 bucket,
767 version: None,
768 };
769 let (config_uri, workflows_config) =
774 fetch_workflows_config(&self.remote, host.as_ref(), &workflows_config_uri).await?;
775 let workflow = resolve_workflow_from_config(
781 &self.remote,
782 host.as_ref(),
783 workflow,
784 config_uri,
785 workflows_config.as_ref(),
786 )
787 .await?;
788 let manifest = self.manifest().await?;
789 let lineage = flow::recommit(
790 lineage,
791 &manifest,
792 &self.paths,
793 &self.storage,
794 &self.remote,
795 host.as_ref(),
796 self.namespace.clone(),
797 host_config,
798 workflow,
799 workflows_config.as_ref(),
800 )
801 .await?;
802 self.lineage.write(&self.storage, lineage).await?;
803 Ok(())
804 }
805
806 async fn workflows_config_location(&self) -> Res<Option<(Option<Host>, S3Uri)>> {
812 let (_, lineage) = self.lineage.read(&self.storage).await?;
813 let remote_uri = match lineage.remote_uri.as_ref() {
814 Some(uri) if !uri.bucket.is_empty() => uri.clone(),
815 _ => return Ok(None),
816 };
817 let config_uri = S3Uri {
818 key: WORKFLOWS_CONFIG_KEY.to_string(),
819 ..S3Uri::from(remote_uri.clone())
820 };
821 Ok(Some((remote_uri.origin, config_uri)))
822 }
823
824 pub async fn resolve_workflow(&self, intent: WorkflowIntent) -> Res<Option<Workflow>> {
825 let Some((origin, config_uri)) = self.workflows_config_location().await? else {
826 return Ok(None);
827 };
828 resolve_workflow(&self.remote, origin.as_ref(), intent, &config_uri).await
829 }
830
831 pub async fn workflows_config(&self) -> Res<Option<WorkflowsConfig>> {
839 let Some((origin, config_uri)) = self.workflows_config_location().await? else {
840 return Ok(None);
841 };
842 let (_, config) =
843 fetch_workflows_config(&self.remote, origin.as_ref(), &config_uri).await?;
844 Ok(config)
845 }
846
847 pub async fn workflow_rules(&self, workflow_id: &str) -> Res<Option<WorkflowRules>> {
861 let Some((origin, config_uri)) = self.workflows_config_location().await? else {
862 return Ok(None);
863 };
864 let (_, config) =
865 fetch_workflows_config(&self.remote, origin.as_ref(), &config_uri).await?;
866 let Some(config) = config else {
867 return Ok(None);
868 };
869 Ok(Some(
870 fetch_workflow_rules(&self.remote, origin.as_ref(), &config, workflow_id).await?,
871 ))
872 }
873}
874
875#[cfg(test)]
876mod set_remote_tests;
877#[cfg(test)]
878mod sync_flow_tests;