radicle: Implement migration callback mechanism — heartwood

Add the ability to have native migrations and progress callback functions in migration code for the COB cache.

 use radicle::patch::cache::Patches as _;
 use thiserror::Error;
 use radicle::cob::migrate;
 use radicle::git::raw;
 use radicle::identity::doc;
 use radicle::identity::doc::{DocError, RepoId};

     info.push([term::format::bold(proj.name()).into()]);
     info.push([term::format::italic(proj.description()).into()]);
     let issues = profile.issues(&repo)?.counts()?;
     let patches = profile.patches(&repo)?.counts()?;
     let issues = profile.issues(&repo, migrate::ignore)?.counts()?;
     let patches = profile.patches(&repo, migrate::ignore)?.counts()?;
     info.push([term::Line::spaced([
         term::format::tertiary(issues.open).into(),

 use chrono::prelude::*;
 use nonempty::NonEmpty;
 use radicle::cob;
 use radicle::cob::migrate;
 use radicle::cob::Op;
 use radicle::identity::Identity;
 use radicle::issue::cache::Issues;

             let oid = &oid.resolve(&repo.backend)?;
             if options.type_name == cob::patch::TYPENAME.clone() {
                 let patches = profile.patches(&repo)?;
                 let patches = profile.patches(&repo, migrate::ignore)?;
                 let Some(patch) = patches.get(oid)? else {
                     anyhow::bail!(cob::store::Error::NotFound(options.type_name, *oid))
                 };
                 serde_json::to_writer_pretty(std::io::stdout(), &patch)?
             } else if options.type_name == cob::issue::TYPENAME.clone() {
                 let issues = profile.issues(&repo)?;
                 let issues = profile.issues(&repo, migrate::ignore)?;
                 let Some(issue) = issues.get(oid)? else {
                     anyhow::bail!(cob::store::Error::NotFound(options.type_name, *oid))
                 };

 use git_ref_format::Qualified;
 use localtime::LocalTime;
 use radicle::cob::migrate;
 use radicle::cob::TypedId;
 use radicle::identity::Identity;
 use radicle::issue::cache::Issues as _;

     let (_, head) = repo.head()?;
     let doc = repo.identity_doc()?;
     let proj = doc.project()?;
     let issues = profile.issues(&repo)?;
     let patches = profile.patches(&repo)?;
     let issues = profile.issues(&repo, migrate::ignore)?;
     let patches = profile.patches(&repo, migrate::ignore)?;
     let mut notifs = notifs.by_repo(&rid, sort_by.field)?.collect::<Vec<_>>();
     if !sort_by.reverse {

     match n.kind {
         NotificationKind::Cob { typed_id } if typed_id.is_issue() => {
             let issues = profile.issues(&repo)?;
             let issues = profile.issues(&repo, migrate::ignore)?;
             let issue = issues.get(&typed_id.id)?.unwrap();
             term::issue::show(

             )?;
         }
         NotificationKind::Cob { typed_id } if typed_id.is_patch() => {
             let patches = profile.patches(&repo)?;
             let patches = profile.patches(&repo, migrate::ignore)?;
             let patch = patches.get(&typed_id.id)?.unwrap();
             term::patch::show(&patch, &typed_id.id, false, &repo, None, profile)?;

 use anyhow::{anyhow, Context as _};
 use radicle::cob::common::{Label, Reaction};
 use radicle::cob::issue;
 use radicle::cob::issue::{CloseReason, State};
 use radicle::cob::thread;
 use radicle::cob::{issue, migrate, thread};
 use radicle::crypto::Signer;
 use radicle::issue::cache::Issues as _;
 use radicle::prelude::{Did, RepoId};

                 | Operation::Comment { .. }
         );
     let mut issues = profile.issues_mut(&repo)?;
     let mut issues = profile.issues_mut(&repo, migrate::ignore)?;
     match options.op {
         Operation::Edit {

 use std::ops::ControlFlow;
 use radicle::cob::migrate;
 use radicle::issue::IssueId;
 use radicle::storage::git::Repository;
 use radicle::storage::ReadStorage as _;

 }
 fn cache(id: Option<IssueId>, repository: &Repository, profile: &Profile) -> anyhow::Result<()> {
     let mut issues = profile.issues_mut(repository)?;
     let mut issues = profile.issues_mut(repository, migrate::ignore)?;
     // FIXME: Should this change?
     match id {
         Some(id) => {
             issues.write(&id)?;

             match result {
                 Ok((id, _)) => term::success!(
                     "Successfully cached issue {id} ({}/{})",
                     progress.seen(),
                     progress.current(),
                     progress.total()
                 ),
                 Err(e) => term::warning(format!("Failed to retrieve issue: {e}")),

 use anyhow::anyhow;
 use radicle::cob::patch::PatchId;
 use radicle::cob::{patch, Label};
 use radicle::cob::{migrate, patch, Label};
 use radicle::git::RefString;
 use radicle::patch::cache::Patches as _;
 use radicle::storage::git::transport;

             label::run(&patch_id, add, delete, &profile, &repository)?;
         }
         Operation::Set { patch_id, remote } => {
             let patches = profile.patches(&repository)?;
             let patches = profile.patches(&repository, migrate::ignore)?;
             let patch_id = patch_id.resolve(&repository.backend)?;
             let patch = patches
                 .get(&patch_id)?

     repository: &Repository,
 ) -> anyhow::Result<()> {
     let signer = term::signer(profile)?;
     let mut patches = profile.patches_mut(repository)?;
     let mut patches = profile.patches_mut(repository, migrate::ignore)?;
     let Ok(mut patch) = patches.get_mut(patch_id) else {
         anyhow::bail!("Patch `{patch_id}` not found");
     };

     profile: &Profile,
     opts: Options,
 ) -> anyhow::Result<()> {
     let patches = profile.patches(stored)?;
     let patches = profile.patches(stored, migrate::ignore)?;
     let patch = patches
         .get(patch_id)?
         .ok_or_else(|| anyhow!("Patch `{patch_id}` not found"))?;

 use super::*;
 use radicle::cob;
 use radicle::cob::patch;
 use radicle::cob::thread::CommentId;
 use radicle::cob::{migrate, patch};
 use radicle::patch::ByRevision;
 use radicle::prelude::*;
 use radicle::storage::git::Repository;

     profile: &Profile,
 ) -> anyhow::Result<()> {
     let signer = term::signer(profile)?;
     let mut patches = profile.patches_mut(repo)?;
     let mut patches = profile.patches_mut(repo, migrate::ignore)?;
     let revision_id = revision_id.resolve::<cob::EntryId>(&repo.backend)?;
     let ByRevision {

 pub fn run(patch_id: &PatchId, profile: &Profile, repository: &Repository) -> anyhow::Result<()> {
     let signer = &term::signer(profile)?;
     let mut patches = profile.patches_mut(repository)?;
     let mut patches = profile.patches_mut(repository, migrate::ignore)?;
     patches.remove(patch_id, signer)?;
     Ok(())

     repository: &Repository,
     profile: &Profile,
 ) -> anyhow::Result<()> {
     let patches = profile.patches(repository)?;
     let patches = profile.patches(repository, migrate::ignore)?;
     let mut all = Vec::new();
     let iter = match filter {

 use anyhow::{anyhow, Context};
 use radicle::cob::migrate;
 use radicle::cob::patch::{PatchId, RevisionId, Verdict};
 use radicle::git;
 use radicle::prelude::*;

         "couldn't load repository {} from local state",
         repository.id
     ))?;
     let mut patches = profile.patches_mut(repository)?;
     let mut patches = profile.patches_mut(repository, migrate::ignore)?;
     let mut patch = patches
         .get_mut(&patch_id)
         .context(format!("couldn't find patch {patch_id} locally"))?;

     stored: &Repository,
     workdir: Option<&git::raw::Repository>,
 ) -> anyhow::Result<()> {
     let patches = profile.patches(stored)?;
     let patches = profile.patches(stored, migrate::ignore)?;
     let Some(patch) = patches.get(patch_id).map_err(|e| Error::WithHint {
         err: e.into(),
         hint: "reset the cache with `rad patch cache` and try again",

     profile: &Profile,
     stored: &R,
 ) -> Result<(), Error> {
     let patches = profile.patches(stored)?;
     let patches = profile.patches(stored, migrate::ignore)?;
     for patch in patches.list()? {
         let Ok((id, patch)) = patch else {
             // Ignore patches that fail to decode.

                         &nid,
                         &working,
                         stored,
                         profile.patches_mut(stored)?,
                         profile.patches_mut(stored, migrate::ignore)?,
                         &signer,
                         profile,
                         opts.clone(),

 use std::env;
 use anyhow::{anyhow, bail};
 use radicle::cob::migrate;
 use radicle::cob::patch::{PatchId, RevisionId};
 use radicle::git::Oid;
 use radicle::storage::ReadStorage;

     let profile = radicle::Profile::load()?;
     let (working, rid) = radicle::rad::cwd()?;
     let stored = profile.storage.repository(rid)?;
     let mut patches = profile.patches_mut(&stored)?;
     let mut patches = profile.patches_mut(&stored, migrate::ignore)?;
     let mut patch = patches.get_mut(&pid)?;
     if patch.is_merged() {

 #[cfg(test)]
 pub mod test;
 pub use cache::{migrate, MigrateCallback};
 pub use common::*;
 pub use op::{ActorId, Op};
 pub use radicle_cob::{

 /// Database migrations.
 /// The first migration is the creation of the initial tables.
 const MIGRATIONS: &[&str] = &[include_str!("cache/migrations/1.sql")];
 const MIGRATIONS: &[Migration] = &[Migration::Sql(include_str!("cache/migrations/1.sql"))];
 /// Function signature for native migrations.
 type MigrateFn = fn(&sql::Connection, Progress, &dyn MigrateCallback) -> Result<usize, Error>;
 /// A database migration.
 enum Migration {
     /// Migration written in SQL.
     Sql(&'static str),
     /// Migration function written in Rust.
     #[allow(dead_code)]
     Native(MigrateFn),
 }
 /// Something that can process migration progress.
 pub trait MigrateCallback {
     /// A migration has progressed.
     /// The first [`Progress`] parameter refers to the progress within the list of migrations.
     /// The second [`Progress`] parameter refers to the progress within the current migration.
     fn progress(&mut self, migration: Progress, item: Progress) -> Result<bool, Error>;
 }
 impl<F> MigrateCallback for F
 where
     F: Fn(Progress, Progress) -> Result<bool, Error>,
 {
     fn progress(&mut self, migration: Progress, item: Progress) -> Result<bool, Error> {
         (self)(migration, item)
     }
 }
 /// Migration functions that implement [`MigrateCallback`].
 pub mod migrate {
     use super::*;
     /// Log progress via installed logger at "info" level.
     pub fn log(migration: Progress, item: Progress) -> Result<bool, Error> {
         log::info!(
             target: "db",
             "Migration {}/{} in progress.. ({}%)",
             migration.current() + 1,
             migration.total(),
             item.percentage()
         );
         Ok(true)
     }
     /// Ignore progress, just migrate.
     pub fn ignore(_migration: Progress, _item: Progress) -> Result<bool, Error> {
         Ok(true)
     }
 }
 #[derive(Error, Debug)]
 pub enum Error {

     pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, Error> {
         let mut db = sql::Connection::open_thread_safe(path)?;
         db.set_busy_timeout(DB_WRITE_TIMEOUT.as_millis() as usize)?;
         migrate(&db)?;
         migrate(&db, migrate::ignore)?;
         Ok(Self {
             db: Arc::new(db),

 }
 /// Migrate the database to the latest schema.
 fn migrate(db: &sql::Connection) -> Result<usize, Error> {
 fn migrate<M>(db: &sql::Connection, mut callback: M) -> Result<usize, Error>
 where
     M: MigrateCallback,
 {
     let mut version = version(db)?;
     for (i, migration) in MIGRATIONS.iter().enumerate() {
         if i >= version {
             transaction(db, |db| {
                 db.execute(migration)?;
                 version = bump(db)?;
     let total = MIGRATIONS.len();
                 Ok::<_, Error>(())
             })?;
     for (i, migration) in MIGRATIONS.iter().enumerate() {
         if i < version {
             continue;
         }
         transaction(db, |db| {
             match migration {
                 Migration::Sql(query) => {
                     db.execute(query)?;
                     callback.progress(
                         Progress { total, current: i },
                         Progress::done(db.change_count()),
                     )?;
                 }
                 Migration::Native(migrate) => {
                     migrate(db, Progress { total, current: i }, &callback)?;
                 }
             }
             version = bump(db)?;
             Ok::<_, Error>(())
         })?;
     }
     Ok(version)
 }

 ///
 /// See [`crate::cob::issue::Cache::write_all`] and
 /// [`crate::cob::patch::Cache::write_all`].
 pub struct WriteAllProgress {
 pub struct Progress {
     current: usize,
     total: usize,
     seen: usize,
 }
 impl WriteAllProgress {
 impl Progress {
     /// Create a new progress tracker with the given `total` amount.
     pub fn new(total: usize) -> Self {
         Self { total, seen: 0 }
         Self { current: 0, total }
     }
     /// Create a new progress tracker that is "done".
     pub fn done(total: usize) -> Self {
         Self {
             current: total,
             total,
         }
     }
     /// Increment the [`WriteAllProgress::seen`] progress.
     /// Increment the [`Progress::current`] progress.
     pub fn inc(&mut self) {
         self.seen += 1;
         self.current += 1;
     }
     /// Return the `total` amount.

         self.total
     }
     /// Return the `seen` amount.
     pub fn seen(&self) -> usize {
         self.seen
     /// Return the `current` amount.
     pub fn current(&self) -> usize {
         self.current
     }
     /// Return the percentage of the progress made.

     ///
     /// If the `total` provided is `0`.
     pub fn percentage(&self) -> f32 {
         (self.seen as f32 / self.total as f32) * 100.0
         (self.current as f32 / self.total as f32) * 100.0
     }
 }