//! Lease-bound command service over the transport-neutral configuration core.

use core::num::NonZeroU32;

use kaptan_config::{
    BeginError, CommitError, ConfigManager, ConfigModel, ConfigState, FieldWriteError, Generation,
    ProfileStore, SetError,
};

use super::{
    import::{CAPABILITY_CBOR_IMPORT, ImportBuffer, NoSnapshotImporter, SnapshotImporter},
    query::ServiceResponse,
    replay::{REQUEST_CACHE_CAPACITY, RememberedResult},
};
#[cfg(test)]
use super::{
    import::{IMPORT_CHUNK_BYTES, crc32c},
    query::{ConfigInfo, DescriptorPage, ValueResponse},
};
use crate::{
    command::{Authorized, ConfigCommand, Target},
    protocol::{ConfigReply, Operation, ProtocolError, ResultKind},
};

#[derive(Clone, Copy)]
pub(super) struct Session {
    pub(super) id: u32,
    pub(super) owner: u64,
    pub(super) revision: u32,
}

/// Owns only the one GCS transaction lease and bounded protocol state.
///
/// A future signing/replay dispatcher creates [`Authorized`] commands. This
/// protocol never accepts raw frames and can be expired by the composition
/// owner on lease or link loss. The same composition owner retains the one
/// [`ConfigManager`] and one runtime log hub.
pub struct ConfigProtocol<I = NoSnapshotImporter> {
    pub(super) session: Option<Session>,
    pub(super) next_session_id: Option<NonZeroU32>,
    pub(super) next_transfer_id: Option<NonZeroU32>,
    pub(super) target: Target,
    pub(super) remembered: [Option<RememberedResult>; REQUEST_CACHE_CAPACITY],
    pub(super) next_remembered: usize,
    pub(super) importer: I,
    pub(super) import: Option<ImportBuffer>,
    pub(super) generation: u64,
    pub(super) state: ConfigState,
}

impl ConfigProtocol<NoSnapshotImporter> {
    /// Creates a custom-protocol state machine without CBOR import support.
    pub const fn new(target: Target) -> Self {
        Self::with_importer(target, NoSnapshotImporter)
    }
}

impl<I> ConfigProtocol<I> {
    /// Creates a custom-protocol state machine with a vehicle-specific CBOR
    /// importer. Profile/store ownership stays outside this adapter.
    pub const fn with_importer(target: Target, importer: I) -> Self {
        Self {
            session: None,
            next_session_id: Some(NonZeroU32::MIN),
            next_transfer_id: Some(NonZeroU32::MIN),
            target,
            remembered: [None; REQUEST_CACHE_CAPACITY],
            next_remembered: 0,
            importer,
            import: None,
            generation: 0,
            state: ConfigState::Active,
        }
    }
}

impl<I> ConfigProtocol<I> {
    /// Feature bits for `KAPTAN_CONFIG_INFO` for a concrete candidate type.
    pub const fn capability_flags_for<C>(&self) -> u8
    where
        I: SnapshotImporter<C>,
    {
        if I::IMPORT_SUPPORTED {
            CAPABILITY_CBOR_IMPORT
        } else {
            0
        }
    }

    /// Dispatches one authorized configuration command. The authoritative
    /// `KAPTAN_CONFIG_RESULT` response remains the complete configuration
    /// audit; generic human logging is intentionally independent.
    pub fn dispatch<C, S>(
        &mut self,
        manager: &mut ConfigManager<C, S>,
        authorized: Authorized<ConfigCommand>,
    ) -> ServiceResponse
    where
        C: ConfigModel,
        S: ProfileStore<C>,
        I: SnapshotImporter<C>,
    {
        self.refresh(manager);
        let (authority, command) = authorized.into_parts();
        self.dispatch_inner(manager, authority.principal(), command)
    }

    /// Expires the bounded session and drops only its RAM transaction.
    pub fn expire<C, S>(&mut self, manager: &mut ConfigManager<C, S>) -> bool
    where
        C: ConfigModel,
        S: ProfileStore<C>,
    {
        let expired = self.session.take().is_some();
        if expired {
            manager.abort();
            self.import = None;
            self.remembered = [None; REQUEST_CACHE_CAPACITY];
            self.next_remembered = 0;
        }
        expired
    }

    fn dispatch_inner<C, S>(
        &mut self,
        manager: &mut ConfigManager<C, S>,
        owner: u64,
        command: ConfigCommand,
    ) -> ServiceResponse
    where
        C: ConfigModel,
        S: ProfileStore<C>,
        I: SnapshotImporter<C>,
    {
        self.refresh(manager);
        if command.target() != self.target {
            return ServiceResponse::Result(self.rejected(command, ProtocolError::BadTarget));
        }
        if let Some(cached) = self.cached(owner, command) {
            return ServiceResponse::Result(cached);
        }
        let response = match command {
            ConfigCommand::InfoRequest { request_id, .. } => {
                ServiceResponse::Info(self.info(manager, request_id))
            }
            ConfigCommand::DescriptorRequest {
                request_id,
                first,
                count,
                ..
            } => ServiceResponse::Descriptors(self.descriptors(manager, request_id, first, count)),
            ConfigCommand::ValueRequest {
                request_id,
                session_id,
                field,
                source,
                ..
            } => self.value_response(manager, owner, request_id, session_id, field, source),
            ConfigCommand::ImportBegin { .. }
            | ConfigCommand::ImportChunk { .. }
            | ConfigCommand::ImportFinish { .. } => {
                ServiceResponse::Result(self.in_session(manager, owner, command))
            }
            ConfigCommand::Begin { .. } => {
                ServiceResponse::Result(self.begin(manager, owner, command))
            }
            ConfigCommand::Set { .. }
            | ConfigCommand::Validate { .. }
            | ConfigCommand::Commit { .. }
            | ConfigCommand::Abort { .. } => {
                ServiceResponse::Result(self.in_session(manager, owner, command))
            }
        };
        self.refresh(manager);
        match response {
            ServiceResponse::Result(reply) => {
                self.remember(RememberedResult {
                    owner,
                    command,
                    reply,
                });
                ServiceResponse::Result(reply)
            }
            response => response,
        }
    }

    fn begin<C, S>(
        &mut self,
        manager: &mut ConfigManager<C, S>,
        owner: u64,
        command: ConfigCommand,
    ) -> ConfigReply
    where
        C: ConfigModel,
        S: ProfileStore<C>,
    {
        let ConfigCommand::Begin {
            request_id,
            expected_generation,
            ..
        } = command
        else {
            return self.rejected(command, ProtocolError::Internal);
        };
        if self.session.is_some() {
            return self.rejected(command, ProtocolError::SessionBusy);
        }
        if expected_generation != self.generation() {
            return self.rejected(command, ProtocolError::GenerationConflict);
        }
        let Some(id) = self.allocate_session_id() else {
            return self.rejected(command, ProtocolError::Internal);
        };
        match manager.begin() {
            Ok(()) => {
                self.session = Some(Session {
                    id,
                    owner,
                    revision: 0,
                });
                self.reply(
                    request_id,
                    id,
                    0,
                    Operation::Begin,
                    ResultKind::Accepted,
                    None,
                )
            }
            Err(BeginError::TransactionActive) => {
                self.rejected(command, ProtocolError::SessionBusy)
            }
        }
    }

    fn in_session<C, S>(
        &mut self,
        manager: &mut ConfigManager<C, S>,
        owner: u64,
        command: ConfigCommand,
    ) -> ConfigReply
    where
        C: ConfigModel,
        S: ProfileStore<C>,
        I: SnapshotImporter<C>,
    {
        let Some(session) = self.session else {
            return self.rejected(command, ProtocolError::InvalidSession);
        };
        let session_id = command_session_id(command);
        if session.id != session_id || session.owner != owner {
            return self.rejected(command, ProtocolError::InvalidSession);
        }
        if let Some(expected_revision) = command_expected_revision(command)
            && expected_revision != session.revision
        {
            return self.rejected(command, ProtocolError::StaleRevision);
        }
        match command {
            ConfigCommand::ImportBegin { .. } => self.begin_import::<C>(owner, command, session),
            ConfigCommand::ImportChunk { .. } => self.import_chunk(command, session),
            ConfigCommand::ImportFinish {
                request_id,
                transfer_id,
                ..
            } => self.finish_import(manager, command, session, request_id, transfer_id),
            ConfigCommand::Set {
                request_id,
                field,
                value,
                ..
            } => {
                let Some(next_revision) = session.revision.checked_add(1) else {
                    return self.rejected(command, ProtocolError::Internal);
                };
                match manager.set(field, value) {
                    Ok(_) => {
                        let Some(session) = self.session.as_mut() else {
                            return self.rejected(command, ProtocolError::Internal);
                        };
                        session.revision = next_revision;
                        let (session_id, revision) = (session.id, session.revision);
                        self.reply(
                            request_id,
                            session_id,
                            revision,
                            Operation::Set,
                            ResultKind::Accepted,
                            None,
                        )
                    }
                    Err(SetError::NoTransaction) => self.rejected(command, ProtocolError::BadState),
                    Err(SetError::Field(error)) => self.rejected(command, field_error(error)),
                }
            }
            ConfigCommand::Validate { request_id, .. } => match manager.validate() {
                Ok(()) => self.reply(
                    request_id,
                    session.id,
                    session.revision,
                    Operation::Validate,
                    ResultKind::Accepted,
                    None,
                ),
                Err(CommitError::NoTransaction) => self.rejected(command, ProtocolError::BadState),
                Err(CommitError::InvalidCandidate(violation)) => {
                    self.rejected(command, ProtocolError::Violation(violation))
                }
                Err(CommitError::Store(_)) => self.rejected(command, ProtocolError::Internal),
            },
            ConfigCommand::Commit {
                request_id,
                expected_generation,
                ..
            } => {
                if expected_generation != self.generation() {
                    return self.rejected(command, ProtocolError::GenerationConflict);
                }
                match manager.commit() {
                    Ok(_) => {
                        self.session = None;
                        self.import = None;
                        self.refresh(manager);
                        self.reply(
                            request_id,
                            0,
                            session.revision,
                            Operation::Commit,
                            ResultKind::CommittedRestartRequired,
                            None,
                        )
                    }
                    Err(CommitError::NoTransaction) => {
                        self.rejected(command, ProtocolError::BadState)
                    }
                    Err(CommitError::InvalidCandidate(violation)) => {
                        self.rejected(command, ProtocolError::Violation(violation))
                    }
                    Err(CommitError::Store(_)) => {
                        self.rejected(command, ProtocolError::StorageFailure)
                    }
                }
            }
            ConfigCommand::Abort { request_id, .. } => {
                manager.abort();
                self.session = None;
                self.import = None;
                self.refresh(manager);
                self.reply(
                    request_id,
                    0,
                    session.revision,
                    Operation::Abort,
                    ResultKind::Accepted,
                    None,
                )
            }
            ConfigCommand::InfoRequest { .. }
            | ConfigCommand::DescriptorRequest { .. }
            | ConfigCommand::ValueRequest { .. }
            | ConfigCommand::Begin { .. } => self.rejected(command, ProtocolError::Internal),
        }
    }

    fn allocate_session_id(&mut self) -> Option<u32> {
        let id = self.next_session_id?;
        self.next_session_id = id.get().checked_add(1).and_then(NonZeroU32::new);
        Some(id.get())
    }

    pub(super) fn allocate_transfer_id(&mut self) -> Option<u32> {
        let id = self.next_transfer_id?;
        self.next_transfer_id = id.get().checked_add(1).and_then(NonZeroU32::new);
        Some(id.get())
    }

    fn refresh<C, S>(&mut self, manager: &ConfigManager<C, S>)
    where
        C: ConfigModel,
        S: ProfileStore<C>,
    {
        self.generation = manager.generation().map_or(0, Generation::get);
        self.state = manager.state();
    }

    fn generation(&self) -> u64 {
        self.generation
    }

    pub(super) fn reply(
        &self,
        request_id: u32,
        session_id: u32,
        revision: u32,
        operation: Operation,
        result: ResultKind,
        error: Option<ProtocolError>,
    ) -> ConfigReply {
        ConfigReply {
            request_id,
            session_id,
            transfer_id: 0,
            revision,
            generation: self.generation,
            operation,
            result,
            error,
            state: self.state,
        }
    }

    pub(super) fn rejected(&self, command: ConfigCommand, error: ProtocolError) -> ConfigReply {
        self.reply(
            command.request_id(),
            command_session_id(command),
            self.session.map_or(0, |session| session.revision),
            operation(command),
            ResultKind::Rejected,
            Some(error),
        )
    }

    pub(super) fn rejected_for(
        &self,
        request_id: u32,
        session_id: u32,
        error: ProtocolError,
    ) -> ConfigReply {
        self.reply(
            request_id,
            session_id,
            self.session.map_or(0, |session| session.revision),
            Operation::Query,
            ResultKind::Rejected,
            Some(error),
        )
    }
}

fn command_session_id(command: ConfigCommand) -> u32 {
    match command {
        ConfigCommand::InfoRequest { .. }
        | ConfigCommand::DescriptorRequest { .. }
        | ConfigCommand::Begin { .. } => 0,
        ConfigCommand::ValueRequest { session_id, .. } => session_id,
        ConfigCommand::Set { session_id, .. }
        | ConfigCommand::ImportBegin { session_id, .. }
        | ConfigCommand::ImportChunk { session_id, .. }
        | ConfigCommand::ImportFinish { session_id, .. }
        | ConfigCommand::Validate { session_id, .. }
        | ConfigCommand::Commit { session_id, .. }
        | ConfigCommand::Abort { session_id, .. } => session_id,
    }
}

fn command_expected_revision(command: ConfigCommand) -> Option<u32> {
    match command {
        ConfigCommand::Set {
            expected_revision, ..
        }
        | ConfigCommand::ImportBegin {
            expected_revision, ..
        }
        | ConfigCommand::ImportFinish {
            expected_revision, ..
        }
        | ConfigCommand::Validate {
            expected_revision, ..
        }
        | ConfigCommand::Commit {
            expected_revision, ..
        } => Some(expected_revision),
        ConfigCommand::InfoRequest { .. }
        | ConfigCommand::DescriptorRequest { .. }
        | ConfigCommand::ValueRequest { .. }
        | ConfigCommand::ImportChunk { .. }
        | ConfigCommand::Begin { .. }
        | ConfigCommand::Abort { .. } => None,
    }
}

const fn operation(command: ConfigCommand) -> Operation {
    match command {
        ConfigCommand::InfoRequest { .. }
        | ConfigCommand::DescriptorRequest { .. }
        | ConfigCommand::ValueRequest { .. } => Operation::Query,
        ConfigCommand::ImportBegin { .. }
        | ConfigCommand::ImportChunk { .. }
        | ConfigCommand::ImportFinish { .. } => Operation::Import,
        ConfigCommand::Begin { .. } => Operation::Begin,
        ConfigCommand::Set { .. } => Operation::Set,
        ConfigCommand::Validate { .. } => Operation::Validate,
        ConfigCommand::Commit { .. } => Operation::Commit,
        ConfigCommand::Abort { .. } => Operation::Abort,
    }
}

fn field_error(error: FieldWriteError) -> ProtocolError {
    match error {
        FieldWriteError::UnknownField(field) => ProtocolError::UnknownField(field),
        FieldWriteError::WrongType { field, .. } => ProtocolError::TypeMismatch(field),
        FieldWriteError::Violation(violation) => ProtocolError::Violation(violation),
    }
}

#[cfg(test)]
mod tests {
    use kaptan_config::{
        ConfigManager, ConfigModel, ConfigValue, ConfigViolation, FieldDescriptor, FieldId,
        FieldWriteError, Generation, ProfileStore, ValidationRule,
    };

    use super::*;
    use crate::{Authorized, ConfigCommand, ProtocolError, ResultKind, Target, ValueSource};

    const VALUE: FieldId = FieldId::new(1);
    const DESCRIPTOR: FieldDescriptor = FieldDescriptor {
        id: VALUE,
        display_key: "test.value",
        kind: kaptan_config::ValueKind::U32,
        minimum: Some(ConfigValue::U32(0)),
        maximum: Some(ConfigValue::U32(10)),
        default: ConfigValue::U32(2),
        unit: "count",
        apply_policy: kaptan_config::ApplyPolicy::RestartRequired,
    };

    #[derive(Clone)]
    struct Candidate(u32);

    impl ConfigModel for Candidate {
        type Active = u32;
        type Context = ();
        const SCHEMA_ID: [u8; 16] = *b"CONFIGSERVICE-01";

        fn descriptor(index: u16) -> Option<&'static FieldDescriptor> {
            (index == 0).then_some(&DESCRIPTOR)
        }
        fn field_count() -> u16 {
            1
        }
        fn field_value(&self, id: FieldId) -> Option<ConfigValue> {
            (id == VALUE).then_some(ConfigValue::U32(self.0))
        }
        fn active_field_value(active: &Self::Active, id: FieldId) -> Option<ConfigValue> {
            (id == VALUE).then_some(ConfigValue::U32(*active))
        }
        fn set_field(&mut self, id: FieldId, value: ConfigValue) -> Result<(), FieldWriteError> {
            if id != VALUE {
                return Err(FieldWriteError::UnknownField(id));
            }
            let ConfigValue::U32(value) = value else {
                return Err(FieldWriteError::WrongType {
                    field: id,
                    expected: kaptan_config::ValueKind::U32,
                    actual: value.kind(),
                });
            };
            if value > 10 {
                return Err(FieldWriteError::Violation(ConfigViolation::field(
                    id,
                    ValidationRule::AboveMaximum,
                )));
            }
            self.0 = value;
            Ok(())
        }
        fn validate(&self, _: &Self::Context) -> Result<Self::Active, ConfigViolation> {
            Ok(self.0)
        }
    }

    struct Store;
    impl ProfileStore<Candidate> for Store {
        type Error = ();
        fn load(&mut self) -> Result<Option<(Generation, Candidate)>, Self::Error> {
            Ok(None)
        }
        fn persist(&mut self, _: &Candidate) -> Result<Generation, Self::Error> {
            Ok(Generation::new(1))
        }
    }

    struct Importer;

    impl SnapshotImporter<Candidate> for Importer {
        const IMPORT_SUPPORTED: bool = true;
        const PROFILE_FORMAT: u16 = 1;

        fn decode(&self, bytes: &[u8]) -> Result<Candidate, ProtocolError> {
            if bytes != [7]
                && !(bytes.len() == IMPORT_CHUNK_BYTES + 1
                    && bytes[0] == 7
                    && bytes[IMPORT_CHUNK_BYTES] == 9)
            {
                return Err(ProtocolError::CborMalformed);
            }
            Ok(Candidate(7))
        }
    }

    /// Test-only composition root. Production code must retain these two
    /// objects separately; this wrapper keeps protocol examples compact while
    /// exercising the same externally-owned-manager dispatch API.
    struct TestService<I = NoSnapshotImporter> {
        manager: ConfigManager<Candidate, Store>,
        protocol: ConfigProtocol<I>,
    }

    impl TestService<NoSnapshotImporter> {
        fn new(manager: ConfigManager<Candidate, Store>, target: Target) -> Self {
            Self {
                manager,
                protocol: ConfigProtocol::new(target),
            }
        }
    }

    impl<I> TestService<I>
    where
        I: SnapshotImporter<Candidate>,
    {
        fn with_importer(
            manager: ConfigManager<Candidate, Store>,
            target: Target,
            importer: I,
        ) -> Self {
            Self {
                manager,
                protocol: ConfigProtocol::with_importer(target, importer),
            }
        }

        fn dispatch(&mut self, command: Authorized<ConfigCommand>) -> ServiceResponse {
            self.protocol.dispatch(&mut self.manager, command)
        }

        fn manager(&self) -> &ConfigManager<Candidate, Store> {
            &self.manager
        }

        fn capability_flags(&self) -> u8 {
            self.protocol.capability_flags_for::<Candidate>()
        }
    }

    fn result(response: ServiceResponse) -> ConfigReply {
        match response {
            ServiceResponse::Result(reply) => reply,
            ServiceResponse::Info(_)
            | ServiceResponse::Descriptors(_)
            | ServiceResponse::Value(_) => {
                panic!("test expected a result response")
            }
        }
    }

    #[test]
    fn retries_are_idempotent_and_wrong_owner_cannot_mutate() {
        let manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut service = TestService::new(manager, Target::new(1, 1));
        let begin = ConfigCommand::Begin {
            request_id: 1,
            expected_generation: 0,
            target: Target::new(1, 1),
        };
        let started = result(service.dispatch(Authorized::host(7, begin)));
        assert_eq!(started.result, ResultKind::Accepted);
        let set = ConfigCommand::Set {
            request_id: 2,
            session_id: started.session_id,
            expected_revision: 0,
            field: VALUE,
            value: ConfigValue::U32(3),
            target: Target::new(1, 1),
        };
        let accepted = result(service.dispatch(Authorized::host(7, set)));
        assert_eq!(accepted.result, ResultKind::Accepted);
        assert_eq!(result(service.dispatch(Authorized::host(7, set))), accepted);
        let foreign = result(service.dispatch(Authorized::host(
            8,
            ConfigCommand::Set {
                request_id: 3,
                session_id: started.session_id,
                expected_revision: 1,
                field: VALUE,
                value: ConfigValue::U32(3),
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(foreign.error, Some(ProtocolError::InvalidSession));
    }

    #[test]
    fn exact_validation_issue_reaches_result() {
        let manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut service = TestService::new(manager, Target::new(1, 1));
        let begin = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::Begin {
                request_id: 1,
                expected_generation: 0,
                target: Target::new(1, 1),
            },
        )));
        let rejected = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::Set {
                request_id: 2,
                session_id: begin.session_id,
                expected_revision: 0,
                field: VALUE,
                value: ConfigValue::U32(11),
                target: Target::new(1, 1),
            },
        )));
        assert!(
            matches!(rejected.error, Some(ProtocolError::Violation(issue)) if issue.rule() == ValidationRule::AboveMaximum)
        );
    }

    #[test]
    fn discovery_and_value_reads_are_service_responses() {
        let manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut service = TestService::new(manager, Target::new(1, 1));
        let descriptors = service.dispatch(Authorized::host(
            7,
            ConfigCommand::DescriptorRequest {
                request_id: 1,
                first: 0,
                count: 4,
                target: Target::new(1, 1),
            },
        ));
        assert!(matches!(
            descriptors,
            ServiceResponse::Descriptors(DescriptorPage {
                len: 1,
                descriptors: [Some(&DESCRIPTOR), None, None, None],
                ..
            })
        ));
        let value = service.dispatch(Authorized::host(
            7,
            ConfigCommand::ValueRequest {
                request_id: 2,
                session_id: 0,
                field: VALUE,
                source: ValueSource::Active,
                target: Target::new(1, 1),
            },
        ));
        assert!(matches!(
            value,
            ServiceResponse::Value(ValueResponse {
                value: ConfigValue::U32(2),
                ..
            })
        ));

        let info = service.dispatch(Authorized::host(
            7,
            ConfigCommand::InfoRequest {
                request_id: 3,
                target: Target::new(1, 1),
            },
        ));
        assert!(matches!(
            info,
            ServiceResponse::Info(ConfigInfo {
                request_id: 3,
                field_count: 1,
                profile_format: 0,
                capability_flags: 0,
                ..
            })
        ));
    }

    #[test]
    fn complete_cbor_import_replaces_only_the_open_transaction() {
        let manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut service = TestService::with_importer(manager, Target::new(1, 1), Importer);
        let begin = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::Begin {
                request_id: 1,
                expected_generation: 0,
                target: Target::new(1, 1),
            },
        )));
        let bytes = [7];
        let mut chunk = [0; IMPORT_CHUNK_BYTES];
        chunk[0] = 7;
        let import_begin = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportBegin {
                request_id: 2,
                session_id: begin.session_id,
                expected_revision: 0,
                length: 1,
                schema_id: Candidate::SCHEMA_ID,
                crc32c: crc32c(&bytes),
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(import_begin.result, ResultKind::Accepted);
        assert_ne!(import_begin.transfer_id, 0);
        assert_eq!(
            result(service.dispatch(Authorized::host(
                7,
                ConfigCommand::ImportChunk {
                    request_id: 3,
                    session_id: begin.session_id,
                    transfer_id: import_begin.transfer_id,
                    offset: 0,
                    length: 1,
                    data: chunk,
                    target: Target::new(1, 1),
                },
            )))
            .result,
            ResultKind::Accepted
        );
        let complete = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportFinish {
                request_id: 4,
                session_id: begin.session_id,
                transfer_id: import_begin.transfer_id,
                expected_revision: 0,
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(complete.result, ResultKind::Accepted);
        assert_eq!(complete.revision, 1);
        assert_eq!(
            service.manager().active_value(VALUE),
            Some(ConfigValue::U32(2))
        );
        assert_eq!(
            service.manager().transaction_value(VALUE),
            Some(ConfigValue::U32(7))
        );
    }

    #[test]
    fn stale_import_transfer_cannot_feed_a_new_import() {
        let manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut service = TestService::with_importer(manager, Target::new(1, 1), Importer);
        let session = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::Begin {
                request_id: 1,
                expected_generation: 0,
                target: Target::new(1, 1),
            },
        )));
        let first = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportBegin {
                request_id: 2,
                session_id: session.session_id,
                expected_revision: 0,
                length: 1,
                schema_id: Candidate::SCHEMA_ID,
                crc32c: crc32c(&[7]),
                target: Target::new(1, 1),
            },
        )));
        let second = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportBegin {
                request_id: 3,
                session_id: session.session_id,
                expected_revision: 0,
                length: 1,
                schema_id: Candidate::SCHEMA_ID,
                crc32c: crc32c(&[7]),
                target: Target::new(1, 1),
            },
        )));
        assert_ne!(first.transfer_id, second.transfer_id);
        let mut chunk = [0; IMPORT_CHUNK_BYTES];
        chunk[0] = 7;
        let stale = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportChunk {
                request_id: 4,
                session_id: session.session_id,
                transfer_id: first.transfer_id,
                offset: 0,
                length: 1,
                data: chunk,
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(stale.error, Some(ProtocolError::InvalidSession));
    }

    #[test]
    fn import_accepts_aligned_out_of_order_idempotent_chunks_and_rejects_conflicts() {
        let manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut service = TestService::with_importer(manager, Target::new(1, 1), Importer);
        let session = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::Begin {
                request_id: 1,
                expected_generation: 0,
                target: Target::new(1, 1),
            },
        )));
        let mut bytes = [0_u8; IMPORT_CHUNK_BYTES + 1];
        bytes[0] = 7;
        bytes[IMPORT_CHUNK_BYTES] = 9;
        let imported = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportBegin {
                request_id: 2,
                session_id: session.session_id,
                expected_revision: 0,
                length: bytes.len() as u16,
                schema_id: Candidate::SCHEMA_ID,
                crc32c: crc32c(&bytes),
                target: Target::new(1, 1),
            },
        )));
        let mut tail = [0_u8; IMPORT_CHUNK_BYTES];
        tail[0] = 9;
        let tail_command = ConfigCommand::ImportChunk {
            request_id: 3,
            session_id: session.session_id,
            transfer_id: imported.transfer_id,
            offset: IMPORT_CHUNK_BYTES as u16,
            length: 1,
            data: tail,
            target: Target::new(1, 1),
        };
        assert_eq!(
            result(service.dispatch(Authorized::host(7, tail_command))).result,
            ResultKind::Accepted
        );

        let mut head = [0_u8; IMPORT_CHUNK_BYTES];
        head[0] = 7;
        let head_command = ConfigCommand::ImportChunk {
            request_id: 4,
            session_id: session.session_id,
            transfer_id: imported.transfer_id,
            offset: 0,
            length: IMPORT_CHUNK_BYTES as u8,
            data: head,
            target: Target::new(1, 1),
        };
        assert_eq!(
            result(service.dispatch(Authorized::host(7, head_command))).result,
            ResultKind::Accepted
        );
        let retry = ConfigCommand::ImportChunk {
            request_id: 5,
            session_id: session.session_id,
            transfer_id: imported.transfer_id,
            offset: 0,
            length: IMPORT_CHUNK_BYTES as u8,
            data: head,
            target: Target::new(1, 1),
        };
        assert_eq!(
            result(service.dispatch(Authorized::host(7, retry))).result,
            ResultKind::Accepted
        );
        let mut conflicting = head;
        conflicting[0] = 8;
        let conflict = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportChunk {
                request_id: 6,
                data: conflicting,
                session_id: session.session_id,
                transfer_id: imported.transfer_id,
                offset: 0,
                length: IMPORT_CHUNK_BYTES as u8,
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(conflict.error, Some(ProtocolError::RequestConflict));
        let complete = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportFinish {
                request_id: 7,
                session_id: session.session_id,
                transfer_id: imported.transfer_id,
                expected_revision: 0,
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(complete.result, ResultKind::Accepted);
        assert_eq!(
            service.manager().transaction_value(VALUE),
            Some(ConfigValue::U32(7))
        );
    }

    #[test]
    fn configuration_result_remains_the_structured_failure_authority() {
        let manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut service = TestService::new(manager, Target::new(1, 1));
        let begin = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::Begin {
                request_id: 1,
                expected_generation: 0,
                target: Target::new(1, 1),
            },
        )));
        let rejected = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::Set {
                request_id: 2,
                session_id: begin.session_id,
                expected_revision: 0,
                field: FieldId::new(2),
                value: ConfigValue::U32(3),
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(
            rejected.error,
            Some(ProtocolError::UnknownField(FieldId::new(2)))
        );
        assert_eq!(rejected.request_id, 2);
        assert_eq!(rejected.primary_field(), FieldId::new(2).get());
    }

    #[test]
    fn import_capability_truthfully_reflects_the_installed_importer() {
        let default_manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut default_service = TestService::new(default_manager, Target::new(1, 1));
        assert_eq!(default_service.capability_flags(), 0);
        let begin = result(default_service.dispatch(Authorized::host(
            7,
            ConfigCommand::Begin {
                request_id: 1,
                expected_generation: 0,
                target: Target::new(1, 1),
            },
        )));
        let unavailable = result(default_service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportBegin {
                request_id: 2,
                session_id: begin.session_id,
                expected_revision: 0,
                length: 1,
                schema_id: [0; 16],
                crc32c: 0,
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(unavailable.error, Some(ProtocolError::ImportUnavailable));
        let importing_manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let importing_service =
            TestService::with_importer(importing_manager, Target::new(1, 1), Importer);
        assert_eq!(importing_service.capability_flags(), CAPABILITY_CBOR_IMPORT);
    }

    #[test]
    fn import_requires_the_discovered_schema_id_before_receiving_bytes() {
        let manager = ConfigManager::new(Candidate(2), (), Store).unwrap();
        let mut service = TestService::with_importer(manager, Target::new(1, 1), Importer);
        let begin = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::Begin {
                request_id: 1,
                expected_generation: 0,
                target: Target::new(1, 1),
            },
        )));
        let rejected = result(service.dispatch(Authorized::host(
            7,
            ConfigCommand::ImportBegin {
                request_id: 2,
                session_id: begin.session_id,
                expected_revision: 0,
                length: 1,
                schema_id: [0; 16],
                crc32c: crc32c(&[7]),
                target: Target::new(1, 1),
            },
        )));
        assert_eq!(rejected.error, Some(ProtocolError::SchemaMismatch));
    }
}