Quelle  mod.rs   Sprache: Rust

//! Backup creation task.
use core::{mem, ops::RangeFrom};

use educe::Educe;
use libthreema_macros::{DebugVariantNames, Name, VariantNames};
use tracing::debug;

use crate::{
    common::{
        ConversationType, DistributionListId, GroupIdentity, StreamOffset, ThreemaId,
        keys::{DataBackupLayoutKey, DataBackupRootKey, DataBackupSalt},
        task::TaskLoop,
    },
    data_backup::{
        DataBackupCreateError, DataBackupCreateInternalErrorCause, TARGET_INSTRUCTION_BUFFER_LENGTH,
        accountant::BackupAccountant,
        cipher::LayoutEncryptor,
        codec::container::ContainerEncoder as _,
        common::{
            Category, CategoryPosition, Header, HeaderAndLayout, Layout, LayoutBuilder, Offset, UserData,
        },
        task::{
            DataBackupProviders,
            create::writer::{PageWriter, PageWriterCheckpoint},
        },
    },
    model::user_profile::IdentityData,
    utils::debug::{Name as _, debug_slice_length},
};

mod writer;

/// Storage write instruction to execute as part of a [`DataBackupCreateInstruction`].
///
/// See each variant's steps.
#[derive(Educe)]
#[educe(Debug)]
#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
pub enum DataBackupCreateLocalWriteInstruction {
    /// 1. Append the bytes at the end of the stream.
    Append(#[educe(Debug(method(debug_slice_length)))] Vec<u8>),

    /// 1. Write `data` at `offset` within the stream.
    WriteAt {
        /// The _absolute_ byte offset.
        offset: StreamOffset,

        /// The bytes that must be written.
        #[educe(Debug(method(debug_slice_length)))]
        data: Vec<u8>,
    },
}

/// An instruction of what to do next.
///
/// See each variant's steps.
#[derive(Debug)]
#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
pub enum DataBackupCreateWriteInstruction {
    /// Apply a local write instruction. See [`DataBackupCreateLocalWriteInstruction`] for more details.
    Local(DataBackupCreateLocalWriteInstruction),
}

/// A state update indicating transition to the next state in the data backup create task.
///
/// When handling this state update, run the following steps:
///
/// 1. Let `state_update` be the [`DataBackupCreateStateUpdate`].
/// 2. Advance the displayed status indicator to `state_update`.
///
/// Note: State updates can be expected to appear in the declaration order below.
#[derive(Debug)]
#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
pub enum DataBackupCreateStateUpdate {
    /// Now backing up user data.
    UserData,

    /// Now backing up contacts.
    Contacts,

    /// Now backing up distribution lists.
    DistributionLists,

    /// Now backing up groups.
    Groups,

    /// Now backing up CSP nonces.
    CspNonces,

    /// Now backing up polls.
    Polls,

    /// Now backing up messages.
    Messages,

    /// Now finalizing the backup.
    Finalize,
}

/// An instruction of what to do next.
///
/// When handling this instruction, run the following steps:
///
/// 1. If `state_update` is present, handle it according to the description in
///    [`DataBackupCreateStateUpdate`].
/// 2. If `write_instruction` is present, handle it according to the steps defined in
///    [`DataBackupCreateWriteInstruction`].
#[derive(Debug)]
#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
pub struct DataBackupCreateInstruction {
    /// The state to which the data backup create task was advanced to.
    pub state_update: Option<DataBackupCreateStateUpdate>,

    /// A write instruction that must be handled.
    pub write_instruction: Option<DataBackupCreateWriteInstruction>,
}

/// Result of polling a [`DataBackupCreateTask`].
pub type DataBackupCreateLoop = TaskLoop<DataBackupCreateInstruction, ()>;

/// The initial data needed to create a backup.
pub struct DataBackupCreateInit {
    /// The data backup password (64 alphanumeric characters generated by the app).
    pub password: String,

    /// The user's identity data.
    pub identity_data: IdentityData,

    /// Data providers.
    pub providers: DataBackupProviders,
}

struct DataBackupCreateContext {
    providers: DataBackupProviders,
}

/// Acts as glue between writing pages and placing their location in the layout. Keeps track of the current
/// offset.
struct LayoutTracker {
    salt: DataBackupSalt,
    encryptor: LayoutEncryptor,
    builder: LayoutBuilder,
}
impl LayoutTracker {
    fn new(salt: DataBackupSalt, key: DataBackupLayoutKey) -> Self {
        Self {
            salt,
            encryptor: LayoutEncryptor::new(key),
            builder: LayoutBuilder::default(),
        }
    }

    /// Create a page writer for a specific category, starting at the current offset.
    fn next_writer(&mut self, checkpoint: PageWriterCheckpoint, category: Category) -> PageWriter {
        self.builder.set(
            category,
            CategoryPosition {
                offset: checkpoint.offset,
                length: category.page_length(),
            },
        );
        PageWriter::new(checkpoint, category)
    }

    /// Build the layout and seal it with the snapshot seal.
    #[expect(clippy::needless_pass_by_value, reason = "Prevent reuse")]
    fn finalize(
        mut self,
        page_writer_checkpoint: PageWriterCheckpoint,
    ) -> Result<(DataBackupSalt, LayoutEncryptor, Layout), DataBackupCreateInternalErrorCause> {
        let layout = self
            .builder
            .snapshot_seal(page_writer_checkpoint.offset)
            .build()
            .map_err(|error| DataBackupCreateInternalErrorCause::from(error.to_string()))?;
        Ok((self.salt, self.encryptor, layout))
    }
}

struct InitState {
    identity_data: IdentityData,
    layout_tracker: LayoutTracker,
    page_writer_checkpoint: PageWriterCheckpoint,
}

struct UserDataState {
    identity_data: IdentityData,
    layout_tracker: LayoutTracker,
    page_writer: PageWriter,
}

struct ContactsState {
    layout_tracker: LayoutTracker,
    accountant: BackupAccountant,
    remaining_contacts: Vec<ThreemaId>,
    page_writer: PageWriter,
}

struct DistributionListsState {
    layout_tracker: LayoutTracker,
    accountant: BackupAccountant,
    remaining_distribution_lists: Vec<DistributionListId>,
    page_writer: PageWriter,
}

struct GroupsState {
    layout_tracker: LayoutTracker,
    accountant: BackupAccountant,
    remaining_groups: Vec<GroupIdentity>,
    page_writer: PageWriter,
}

struct CspNoncesState {
    layout_tracker: LayoutTracker,
    accountant: BackupAccountant,
    batch_index: RangeFrom<usize>,
    page_writer: PageWriter,
}

struct PollsState {
    layout_tracker: LayoutTracker,
    accountant: BackupAccountant,
    batch_index: RangeFrom<usize>,
    page_writer: PageWriter,
}

struct MessagesState {
    layout_tracker: LayoutTracker,
    accountant: BackupAccountant,
    conversation_type: ConversationType,
    batch_index: RangeFrom<usize>,
    page_writer: PageWriter,
}

struct FinalizeState {
    layout_tracker: LayoutTracker,
    page_writer_checkpoint: PageWriterCheckpoint,
}

#[derive(DebugVariantNames, VariantNames)]
enum State {
    Init(InitState),
    UserData(UserDataState),
    Contacts(ContactsState),
    DistributionLists(DistributionListsState),
    Groups(GroupsState),
    CspNonces(CspNoncesState),
    Polls(PollsState),
    Messages(MessagesState),
    Finalize(FinalizeState),
    Dissolve,
    Done,
    Error(DataBackupCreateError),
}
impl State {
    #[tracing::instrument(skip_all)]
    fn poll_init(mut state: InitState) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        // Write placeholder for header and layout.
        let instruction = DataBackupCreateLocalWriteInstruction::Append(HeaderAndLayout::stub().to_vec());

        // Advance to the next state.
        let page_writer = state
            .layout_tracker
            .next_writer(state.page_writer_checkpoint, Category::UserData);
        Ok((
            Self::UserData(UserDataState {
                identity_data: state.identity_data,
                layout_tracker: state.layout_tracker,
                page_writer,
            }),
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                state_update: Some(DataBackupCreateStateUpdate::UserData),
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(instruction)),
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_user_data(
        context: &DataBackupCreateContext,
        mut state: UserDataState,
    ) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        let accountant = BackupAccountant::new(state.identity_data.user_identity);

        // Get and write the user data.
        {
            // Get and sanitize the settings.
            let settings = accountant.sanitize_settings(context.providers.settings.lock()?.get()?);

            // Write the user data.
            let user_data = UserData {
                identity_data: state.identity_data,
                user_profile: context.providers.user_profile.lock()?.get()?,
                settings,
            };
            state.page_writer.add_data(&user_data.encode_u32_le_delimited()?);
        };
        let (pages, page_writer_checkpoint) = state.page_writer.drain_all()?;

        // Advance to the next state.
        let page_writer = state
            .layout_tracker
            .next_writer(page_writer_checkpoint, Category::Contacts);
        Ok((
            Self::Contacts(ContactsState {
                layout_tracker: state.layout_tracker,
                accountant,
                remaining_contacts: context
                    .providers
                    .contacts
                    .lock()?
                    .get_identities()?
                    .into_iter()
                    .collect(),
                page_writer,
            }),
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                state_update: Some(DataBackupCreateStateUpdate::Contacts),
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                    DataBackupCreateLocalWriteInstruction::Append(pages),
                )),
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_contacts(
        context: &DataBackupCreateContext,
        mut state: ContactsState,
    ) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        // Add contacts until the target byte length per instruction has been reached.
        {
            let contact_provider = context.providers.contacts.lock()?;
            while let Some(identity) = state.remaining_contacts.pop() {
                let contact = contact_provider.get_with_profile_pictures(identity)?.ok_or(
                    DataBackupCreateError::ContractViolation(
                        "No contact for previously queried identity".to_owned(),
                    ),
                )?;
                if state.accountant.check_and_add_contact(&contact.contact) {
                    state.page_writer.add_data(&contact.encode_u32_le_delimited()?);
                }
                if state.page_writer.buffered_length() >= TARGET_INSTRUCTION_BUFFER_LENGTH as usize {
                    break;
                }
            }
        }

        // Drain until all contacts have been written.
        if !state.remaining_contacts.is_empty() {
            let pages = state.page_writer.drain_complete()?;
            return Ok((
                Self::Contacts(state),
                DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                    state_update: None,
                    write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                        DataBackupCreateLocalWriteInstruction::Append(pages),
                    )),
                }),
            ));
        }
        let (pages, page_writer_checkpoint) = state.page_writer.drain_all()?;

        // Advance to the next state.
        let page_writer = state
            .layout_tracker
            .next_writer(page_writer_checkpoint, Category::DistributionLists);
        Ok((
            Self::DistributionLists(DistributionListsState {
                layout_tracker: state.layout_tracker,
                accountant: state.accountant,
                remaining_distribution_lists: context
                    .providers
                    .distribution_lists
                    .lock()?
                    .get_ids()?
                    .into_iter()
                    .collect(),
                page_writer,
            }),
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                state_update: Some(DataBackupCreateStateUpdate::DistributionLists),
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                    DataBackupCreateLocalWriteInstruction::Append(pages),
                )),
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_distribution_lists(
        context: &DataBackupCreateContext,
        mut state: DistributionListsState,
    ) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        // Add distribution lists until the target byte length per instruction has been reached.
        {
            let distribution_list_provider = context.providers.distribution_lists.lock()?;
            while let Some(distribution_list_id) = state.remaining_distribution_lists.pop() {
                let distribution_list = distribution_list_provider.get(distribution_list_id)?.ok_or(
                    DataBackupCreateError::ContractViolation(
                        "No distribution list for previously queried distribution list ID".to_owned(),
                    ),
                )?;
                if let Some(distribution_list) = state
                    .accountant
                    .sanitize_and_add_distribution_list(distribution_list)
                {
                    state
                        .page_writer
                        .add_data(&distribution_list.encode_u32_le_delimited()?);
                }
                if state.page_writer.buffered_length() >= TARGET_INSTRUCTION_BUFFER_LENGTH as usize {
                    break;
                }
            }
        }

        // Drain until all distribution lists have been written.
        if !state.remaining_distribution_lists.is_empty() {
            let pages = state.page_writer.drain_complete()?;
            return Ok((
                Self::DistributionLists(state),
                DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                    state_update: None,
                    write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                        DataBackupCreateLocalWriteInstruction::Append(pages),
                    )),
                }),
            ));
        }
        let (pages, page_writer_checkpoint) = state.page_writer.drain_all()?;

        // Advance to the next state.
        let page_writer = state
            .layout_tracker
            .next_writer(page_writer_checkpoint, Category::Groups);
        Ok((
            Self::Groups(GroupsState {
                layout_tracker: state.layout_tracker,
                accountant: state.accountant,
                remaining_groups: context
                    .providers
                    .groups
                    .lock()?
                    .get_identities()?
                    .into_iter()
                    .collect(),
                page_writer,
            }),
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                state_update: Some(DataBackupCreateStateUpdate::Groups),
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                    DataBackupCreateLocalWriteInstruction::Append(pages),
                )),
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_groups(
        context: &DataBackupCreateContext,
        mut state: GroupsState,
    ) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        // Add groups until the target byte length per instruction has been reached.
        {
            let group_provider = context.providers.groups.lock()?;
            while let Some(group_identity) = state.remaining_groups.pop() {
                let group = group_provider.get_with_profile_picture(group_identity)?.ok_or(
                    DataBackupCreateError::ContractViolation(
                        "No group for previously queried group identity".to_owned(),
                    ),
                )?;
                if let Some(group) = state.accountant.sanitize_and_add_group(group) {
                    state.page_writer.add_data(&group.encode_u32_le_delimited()?);
                }
                if state.page_writer.buffered_length() >= TARGET_INSTRUCTION_BUFFER_LENGTH as usize {
                    break;
                }
            }
        }

        // Drain until all groups have been written.
        if !state.remaining_groups.is_empty() {
            let pages = state.page_writer.drain_complete()?;
            return Ok((
                Self::Groups(state),
                DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                    state_update: None,
                    write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                        DataBackupCreateLocalWriteInstruction::Append(pages),
                    )),
                }),
            ));
        }
        let (pages, page_writer_checkpoint) = state.page_writer.drain_all()?;

        // Advance to the next state.
        let page_writer = state
            .layout_tracker
            .next_writer(page_writer_checkpoint, Category::CspNonces);
        Ok((
            Self::CspNonces(CspNoncesState {
                layout_tracker: state.layout_tracker,
                accountant: state.accountant,
                batch_index: 0..,
                page_writer,
            }),
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                state_update: Some(DataBackupCreateStateUpdate::CspNonces),
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                    DataBackupCreateLocalWriteInstruction::Append(pages),
                )),
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_csp_nonces(
        context: &DataBackupCreateContext,
        mut state: CspNoncesState,
    ) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        // Add CSP nonces, one batch per instruction.
        let nonces = {
            let batch_index = state
                .batch_index
                .next()
                .ok_or(DataBackupCreateInternalErrorCause::Other(
                    "Reached maximum CSP nonce batch index".to_owned(),
                ))?;
            context.providers.csp_nonces.lock()?.get_many(batch_index)?
        };

        // Write and drain until all CSP nonces have been written.
        if !nonces.is_empty() {
            for nonce in nonces {
                state.page_writer.add_data(&nonce.0);
            }
            let pages = state.page_writer.drain_complete()?;
            return Ok((
                Self::CspNonces(state),
                DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                    state_update: None,
                    write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                        DataBackupCreateLocalWriteInstruction::Append(pages),
                    )),
                }),
            ));
        }
        let (pages, page_writer_checkpoint) = state.page_writer.drain_all()?;

        // Advance to the next state.
        let page_writer = state
            .layout_tracker
            .next_writer(page_writer_checkpoint, Category::Polls);
        Ok((
            Self::Polls(PollsState {
                layout_tracker: state.layout_tracker,
                accountant: state.accountant,
                batch_index: 0..,
                page_writer,
            }),
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                state_update: Some(DataBackupCreateStateUpdate::Polls),
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                    DataBackupCreateLocalWriteInstruction::Append(pages),
                )),
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_polls(
        context: &DataBackupCreateContext,
        mut state: PollsState,
    ) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        // Add polls, one batch per instruction.
        let polls = {
            let batch_index = state
                .batch_index
                .next()
                .ok_or(DataBackupCreateInternalErrorCause::Other(
                    "Reached maximum poll batch index".to_owned(),
                ))?;
            context.providers.polls.lock()?.get_many(batch_index)?
        };

        // Write and drain until all polls have been written.
        if !polls.is_empty() {
            for poll in polls {
                if state.accountant.check_and_add_poll(&poll) {
                    state.page_writer.add_data(&poll.encode_u32_le_delimited()?);
                }
            }
            let pages = state.page_writer.drain_complete()?;
            return Ok((
                Self::Polls(state),
                DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                    state_update: None,
                    write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                        DataBackupCreateLocalWriteInstruction::Append(pages),
                    )),
                }),
            ));
        }
        let (pages, page_writer_checkpoint) = state.page_writer.drain_all()?;

        // Advance to the next state.
        let page_writer = state
            .layout_tracker
            .next_writer(page_writer_checkpoint, Category::Messages);
        Ok((
            Self::Messages(MessagesState {
                layout_tracker: state.layout_tracker,
                accountant: state.accountant,
                conversation_type: ConversationType::Contact,
                batch_index: 0..,
                page_writer,
            }),
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                state_update: Some(DataBackupCreateStateUpdate::Messages),
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                    DataBackupCreateLocalWriteInstruction::Append(pages),
                )),
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_messages(
        context: &DataBackupCreateContext,
        mut state: MessagesState,
    ) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        // Add messages of the current conversation type, one batch per instruction.
        let messages = {
            let batch_index = state
                .batch_index
                .next()
                .ok_or(DataBackupCreateInternalErrorCause::Other(
                    "Reached maximum message batch index".to_owned(),
                ))?;
            context
                .providers
                .conversations
                .lock()?
                .get_many_messages(state.conversation_type, batch_index)?
        };

        // Write and drain until all messages of all conversation types have been written.
        if !messages.is_empty() {
            for message in messages {
                if let Some(message) = state.accountant.sanitize_and_add_message(message) {
                    state.page_writer.add_data(&message.encode_u32_le_delimited()?);
                }
            }
            let pages = state.page_writer.drain_complete()?;
            return Ok((
                Self::Messages(state),
                DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                    state_update: None,
                    write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                        DataBackupCreateLocalWriteInstruction::Append(pages),
                    )),
                }),
            ));
        }

        // We have gotten all messages of this conversation type. Move to the next message category if there
        // are subsequent categories left.
        if let Some(conversation_type) = match state.conversation_type {
            ConversationType::Contact => Some(ConversationType::DistributionList),
            ConversationType::DistributionList => Some(ConversationType::Group),
            ConversationType::Group => None,
        } {
            return Ok((
                Self::Messages(MessagesState {
                    layout_tracker: state.layout_tracker,
                    accountant: state.accountant,
                    conversation_type,
                    batch_index: 0..,
                    page_writer: state.page_writer,
                }),
                DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                    state_update: None,
                    write_instruction: None,
                }),
            ));
        }
        let (pages, page_writer_checkpoint) = state.page_writer.drain_all()?;

        // Advance to the next state.
        Ok((
            Self::Finalize(FinalizeState {
                layout_tracker: state.layout_tracker,
                page_writer_checkpoint,
            }),
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                state_update: Some(DataBackupCreateStateUpdate::Finalize),
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(
                    DataBackupCreateLocalWriteInstruction::Append(pages),
                )),
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_finalize(state: FinalizeState) -> Result<(Self, DataBackupCreateLoop), DataBackupCreateError> {
        // Determine the final layout.
        let (salt, layout_encryptor, layout) = state.layout_tracker.finalize(state.page_writer_checkpoint)?;

        // Write the header and the encrypted layout.
        let instruction = {
            let header = Header::encode(salt);
            let layout = layout_encryptor.encrypt(&header, layout.encode())?;
            DataBackupCreateLocalWriteInstruction::WriteAt {
                offset: StreamOffset(0),
                data: HeaderAndLayout::encode(header, layout).to_vec(),
            }
        };

        // Advance to the next state.
        Ok((
            Self::Dissolve,
            DataBackupCreateLoop::Instruction(DataBackupCreateInstruction {
                write_instruction: Some(DataBackupCreateWriteInstruction::Local(instruction)),
                state_update: None,
            }),
        ))
    }

    #[tracing::instrument(skip_all)]
    fn poll_dissolve() -> (Self, DataBackupCreateLoop) {
        (Self::Done, DataBackupCreateLoop::Done(()))
    }
}

/// Task for creating a data backup.
///
/// The task progresses through the states indicated by [`DataBackupCreateStateUpdate`].
#[derive(Name, educe::Educe)]
#[educe(Debug)]
pub struct DataBackupCreateTask {
    #[educe(Debug(ignore))]
    context: DataBackupCreateContext,

    state: State,
}
impl DataBackupCreateTask {
    /// Create a new task for backing up all user data and return a tuple containing task and filename that
    /// should be used when storing the backup.
    #[must_use]
    #[tracing::instrument(skip_all)]
    pub fn new(init: DataBackupCreateInit) -> (Self, String) {
        let salt = DataBackupSalt::random();
        let key = DataBackupRootKey::from_password(init.password, &salt);
        let filename = format!(
            "{}-threema.backup",
            chrono::Utc::now().format("%Y-%m-%dT%H-%M-%S")
        );

        // Create initial state.
        (
            Self {
                context: DataBackupCreateContext {
                    providers: init.providers,
                },
                state: State::Init(InitState {
                    identity_data: init.identity_data,
                    layout_tracker: LayoutTracker::new(salt, key.layout_key()),
                    page_writer_checkpoint: PageWriterCheckpoint {
                        key: key.page_key(),
                        offset: Offset(0),
                    },
                }),
            },
            filename,
        )
    }

    /// Poll to advance the state.
    ///
    /// If this method returns an error, the task is permanently in the error state. Subsequent calls will
    /// return the same error.
    ///
    /// # Errors
    ///
    /// Returns [`DataBackupCreateError`] for all possible reasons, including provider errors propagated from
    /// the client application.
    #[tracing::instrument(skip_all, fields(?self))]
    pub fn poll(&mut self) -> Result<DataBackupCreateLoop, DataBackupCreateError> {
        let result = match mem::replace(
            &mut self.state,
            State::Error(DataBackupCreateError::ContractViolation(format!(
                "{} in a transitional state",
                Self::NAME
            ))),
        ) {
            State::Error(error) => Err(error),
            State::Init(state) => State::poll_init(state),
            State::UserData(state) => State::poll_user_data(&self.context, state),
            State::Contacts(state) => State::poll_contacts(&self.context, state),
            State::DistributionLists(state) => State::poll_distribution_lists(&self.context, state),
            State::Groups(state) => State::poll_groups(&self.context, state),
            State::CspNonces(state) => State::poll_csp_nonces(&self.context, state),
            State::Polls(state) => State::poll_polls(&self.context, state),
            State::Messages(state) => State::poll_messages(&self.context, state),
            State::Finalize(state) => State::poll_finalize(state),
            State::Dissolve => Ok(State::poll_dissolve()),
            State::Done => Err(DataBackupCreateError::ContractViolation(format!(
                "{} already done",
                Self::NAME
            ))),
        };
        match result {
            Ok((state, instruction)) => {
                self.state = state;
                debug!(state = ?self.state, "Changed state");
                Ok(instruction)
            },
            Err(error) => {
                self.state = State::Error(error.clone());
                debug!(state = ?self.state, "Changed state to error");
                Err(error)
            },
        }
    }
}