// Copyright 2020 MaidSafe.net limited.
//
// This SAFE Network Software is licensed to you under the MIT license <LICENSE-MIT
// https://opensource.org/licenses/MIT> or the Modified BSD license <LICENSE-BSD
// https://opensource.org/licenses/BSD-3-Clause>, at your option. This file may not be copied,
// modified, or distributed except according to those terms. Please review the Licences for the
// specific language governing permissions and limitations relating to use of the SAFE Network
// Software.

mod metadata;
mod seq_crdt;

use crate::{Error, PublicKey, Result};
use crdts::lseq::ident::Identifier;
pub use metadata::{
    Action, Address, Entries, Entry, Index, Kind, Perm, Permissions, Policy, PrivatePermissions,
    PrivatePolicy, PublicPermissions, PublicPolicy, User,
};
use seq_crdt::{CrdtDataOperation, CrdtPolicyOperation, Op, SequenceCrdt};
use serde::{Deserialize, Serialize};
use std::{
    collections::BTreeMap,
    fmt::{self, Debug, Formatter},
    hash::Hash,
};
use xor_name::XorName;

// Type of data used for the 'Actor' in CRDT vector clocks
type ActorType = String;

/// Data mutation operation to apply to Sequence.
pub type DataWriteOp<T> = CrdtDataOperation<ActorType, T>;

/// Policy mutation operation to apply to Sequence.
pub type PolicyWriteOp<T> = CrdtPolicyOperation<ActorType, T>;

/// Public Sequence.
pub type PublicSeqData = SequenceCrdt<ActorType, PublicPolicy>;
/// Private Sequence.
pub type PrivateSeqData = SequenceCrdt<ActorType, PrivatePolicy>;

impl Debug for PublicSeqData {
    fn fmt(&self, formatter: &mut Formatter) -> fmt::Result {
        write!(formatter, "PubSequence {:?}", self.address().name())
    }
}

impl Debug for PrivateSeqData {
    fn fmt(&self, formatter: &mut Formatter) -> fmt::Result {
        write!(formatter, "PrivSequence {:?}", self.address().name())
    }
}

macro_rules! check_perm {
    ($policy: ident, $requester: ident, $action: ident) => {
        match $policy {
            None => {
                if $action == Action::Admin {
                    Ok(())
                } else {
                    Err(Error::AccessDenied)
                }
            }
            Some(policy) => policy.is_action_allowed($requester, $action),
        }
    };
}

/// Object storing a Sequence variant.
#[derive(Clone, Eq, PartialEq, PartialOrd, Hash, Serialize, Deserialize, Debug)]
enum SeqData {
    /// Public Sequence Data.
    Public(PublicSeqData),
    /// Private Sequence Data.
    Private(PrivateSeqData),
}

/// Object storing the Sequence
#[derive(Clone, Eq, PartialEq, PartialOrd, Hash, Serialize, Deserialize, Debug)]
pub struct Data {
    operations_pk: PublicKey,
    data: SeqData,
}

impl Data {
    /// Constructs a new Public Sequence Data.
    /// The 'operations_pk' is assumed to be the PK which the messages were and will be
    /// signed with, whilst the 'actor' is a unique identifier to be used as the
    /// clock's Dot in all operations generated by this instance.
    pub fn new_public(operations_pk: PublicKey, actor: ActorType, name: XorName, tag: u64) -> Self {
        Self {
            operations_pk,
            data: SeqData::Public(PublicSeqData::new(actor, Address::Public { name, tag })),
        }
    }

    /// Constructs a new Private Sequence Data.
    /// The 'operations_pk' is assumed to be the PK which the messages were and will be
    /// signed with, whilst the 'actor' is a unique identifier to be used as the
    /// clock's Dot in all operations generated by this instance.
    pub fn new_private(
        operations_pk: PublicKey,
        actor: ActorType,
        name: XorName,
        tag: u64,
    ) -> Self {
        Self {
            operations_pk,
            data: SeqData::Private(PrivateSeqData::new(actor, Address::Private { name, tag })),
        }
    }

    /// Returns the address.
    pub fn address(&self) -> &Address {
        match &self.data {
            SeqData::Public(data) => data.address(),
            SeqData::Private(data) => data.address(),
        }
    }

    /// Returns the kind.
    pub fn kind(&self) -> Kind {
        self.address().kind()
    }

    /// Returns the name.
    pub fn name(&self) -> &XorName {
        self.address().name()
    }

    /// Returns the tag.
    pub fn tag(&self) -> u64 {
        self.address().tag()
    }

    /// Returns `true` if public.
    pub fn is_pub(&self) -> bool {
        self.kind().is_pub()
    }

    /// Returns `true` if private.
    pub fn is_private(&self) -> bool {
        self.kind().is_private()
    }

    /// Returns the length of the sequence, optionally
    /// verifying read permissions if a pk is provided
    pub fn len(&self, requester: Option<PublicKey>) -> Result<u64> {
        self.check_permission(Action::Read, requester, None)?;

        Ok(match &self.data {
            SeqData::Public(data) => data.len(),
            SeqData::Private(data) => data.len(),
        })
    }

    /// Returns true if the sequence is empty.
    pub fn is_empty(&self, requester: Option<PublicKey>) -> Result<bool> {
        self.check_permission(Action::Read, requester, None)?;

        Ok(self.len(None)? == 0)
    }

    /// Returns the version of last Policy.
    pub fn policy_version(&self, requester: Option<PublicKey>) -> Result<Option<u64>> {
        self.check_permission(Action::Read, requester, None)?;

        Ok(match &self.data {
            SeqData::Public(data) => data.policy_index(),
            SeqData::Private(data) => data.policy_index(),
        })
    }

    /// Gets a list of items which are within the given indices.
    pub fn in_range(
        &self,
        start: Index,
        end: Index,
        requester: Option<PublicKey>,
    ) -> Result<Option<Entries>> {
        self.check_permission(Action::Read, requester, None)?;

        Ok(match &self.data {
            SeqData::Public(data) => data.in_range(start, end),
            SeqData::Private(data) => data.in_range(start, end),
        })
    }

    /// Returns a value at 'index', if present.
    pub fn get(&self, index: Index, requester: Option<PublicKey>) -> Result<Option<&Vec<u8>>> {
        self.check_permission(Action::Read, requester, None)?;

        Ok(match &self.data {
            SeqData::Public(data) => data.get(index),
            SeqData::Private(data) => data.get(index),
        })
    }

    /// Returns the last entry, if present.
    pub fn last_entry(&self, requester: Option<PublicKey>) -> Result<Option<&Entry>> {
        self.check_permission(Action::Read, requester, None)?;

        Ok(match &self.data {
            SeqData::Public(data) => data.last_entry(),
            SeqData::Private(data) => data.last_entry(),
        })
    }

    /// Generate unsigned crdt op, adding the new entry.
    pub fn create_unsigned_append_op(&mut self, entry: Entry) -> Result<DataWriteOp<Entry>> {
        self.check_permission(Action::Append, None, None)?;

        match &mut self.data {
            SeqData::Public(data) => data.create_append_op(entry, self.operations_pk),
            SeqData::Private(data) => data.create_append_op(entry, self.operations_pk),
        }
    }

    /// Apply a signed data CRDT operation.
    pub fn apply_data_op(&mut self, op: DataWriteOp<Entry>) -> Result<()> {
        self.check_permission(Action::Append, Some(op.source), Some(&op.ctx))?;

        match &mut self.data {
            SeqData::Public(data) => data.apply_data_op(op),
            SeqData::Private(data) => data.apply_data_op(op),
        }
    }

    /// Sets the new policy for Public Sequence.
    pub fn create_unsigned_public_policy_op(
        &mut self,
        owner: PublicKey,
        permissions: BTreeMap<User, PublicPermissions>,
    ) -> Result<PolicyWriteOp<PublicPolicy>> {
        self.check_permission(Action::Admin, None, None)?;

        match &mut self.data {
            SeqData::Public(data) => {
                data.create_policy_op(PublicPolicy { owner, permissions }, self.operations_pk)
            }
            SeqData::Private(_) => Err(Error::InvalidOperation),
        }
    }

    /// Sets the new policy for Private Sequence.
    pub fn create_unsigned_private_policy_op(
        &mut self,
        owner: PublicKey,
        permissions: BTreeMap<PublicKey, PrivatePermissions>,
    ) -> Result<PolicyWriteOp<PrivatePolicy>> {
        self.check_permission(Action::Admin, None, None)?;

        match &mut self.data {
            SeqData::Private(data) => {
                data.create_policy_op(PrivatePolicy { owner, permissions }, self.operations_pk)
            }
            SeqData::Public(_) => Err(Error::InvalidOperation),
        }
    }

    /// Apply Public Policy CRDT operation.
    pub fn apply_public_policy_op(&mut self, op: PolicyWriteOp<PublicPolicy>) -> Result<()> {
        // TODO: review if this permissions check is too laxed
        match op.ctx {
            Some((ref policy_id, _)) => {
                self.check_permission(Action::Admin, Some(op.source), Some(policy_id))?
            }
            None => self.check_permission(Action::Admin, Some(op.source), None)?,
        }

        match (&mut self.data, &op.crdt_op) {
            (SeqData::Public(data), Op::Insert { .. }) => data.apply_policy_op(op),
            _ => Err(Error::InvalidOperation),
        }
    }

    /// Apply Private Policy CRDT operation.
    pub fn apply_private_policy_op(&mut self, op: PolicyWriteOp<PrivatePolicy>) -> Result<()> {
        // TODO: review if this permissions check is too laxed
        match op.ctx {
            Some((ref policy_id, _)) => {
                self.check_permission(Action::Admin, Some(op.source), Some(policy_id))?
            }
            None => self.check_permission(Action::Admin, Some(op.source), None)?,
        }

        match &mut self.data {
            SeqData::Private(data) => data.apply_policy_op(op),
            _ => Err(Error::InvalidOperation),
        }
    }

    /// Returns user permissions, if applicable.
    pub fn permissions(&self, user: User, requester: Option<PublicKey>) -> Result<Permissions> {
        let version = self.policy_version(None)?.ok_or(Error::PolicyNotSet)?;
        self.permissions_at(user, version, requester)
    }

    /// Returns user permissions at provided index in the history, if applicable.
    pub fn permissions_at(
        &self,
        user: User,
        version: impl Into<Index>,
        requester: Option<PublicKey>,
    ) -> Result<Permissions> {
        self.check_permission(Action::Read, requester, None)?;

        let user_perm = match &self.data {
            SeqData::Public(data) => data
                .policy_at(version)
                .ok_or(Error::NoSuchEntry)?
                .permissions(user)
                .ok_or(Error::NoSuchEntry)?,
            SeqData::Private(data) => data
                .policy_at(version)
                .ok_or(Error::NoSuchEntry)?
                .permissions(user)
                .ok_or(Error::NoSuchEntry)?,
        };

        Ok(user_perm)
    }

    /// Returns the current public policy, if applicable.
    pub fn public_policy(&self) -> Result<&PublicPolicy> {
        let version = self.policy_version(None)?.ok_or(Error::PolicyNotSet)?;
        self.public_policy_at(version)
    }

    /// Returns the public policy at provided index in the history, if applicable.
    pub fn public_policy_at(&self, version: impl Into<Index>) -> Result<&PublicPolicy> {
        let perms = match &self.data {
            SeqData::Public(data) => data.policy_at(version),
            SeqData::Private(_) => return Err(Error::InvalidOperation),
        };
        perms.ok_or(Error::NoSuchEntry)
    }

    /// Returns the current private policy, if applicable.
    pub fn private_policy(&self, requester: Option<PublicKey>) -> Result<&PrivatePolicy> {
        let version = self.policy_version(None)?.ok_or(Error::PolicyNotSet)?;
        self.private_policy_at(version, requester)
    }

    /// Returns the private policy at provided index in the history, if applicable.
    pub fn private_policy_at(
        &self,
        version: impl Into<Index>,
        requester: Option<PublicKey>,
    ) -> Result<&PrivatePolicy> {
        self.check_permission(Action::Read, requester, None)?;
        let perms = match &self.data {
            SeqData::Private(data) => data.policy_at(version),
            SeqData::Public(_) => return Err(Error::InvalidOperation),
        };
        perms.ok_or(Error::NoSuchEntry)
    }

    /// Helper to check permissions for given `action`
    /// for the given requester's public key.
    ///
    /// Returns:
    /// `Ok(())` if the permissions are valid,
    /// `Err::InvalidOwners` if the last owner is invalid,
    /// `Err::AccessDenied` if the action is not allowed.
    pub fn check_permission(
        &self,
        action: Action,
        requester: Option<PublicKey>,
        policy_id: Option<&Identifier<ActorType>>,
    ) -> Result<()> {
        macro_rules! get_policy {
            ($policy_id: ident, $data: ident) => {
                match $policy_id {
                    Some(id) => match $data.policy_by_id(id) {
                        Some(policy) => Ok(Some(policy)),
                        None => Err(Error::OpNotCausallyReady),
                    },
                    None => Ok($data.policy()),
                }
            };
        }

        match (requester, &self.data) {
            (Some(requester), SeqData::Public(data)) => {
                let policy = get_policy!(policy_id, data)?;
                check_perm!(policy, requester, action)
            }
            (Some(requester), SeqData::Private(data)) => {
                let policy = get_policy!(policy_id, data)?;
                check_perm!(policy, requester, action)
            }
            (None, SeqData::Public(data)) => {
                let policy = get_policy!(policy_id, data)?;
                let requester = self.operations_pk;
                check_perm!(policy, requester, action)
            }
            (None, SeqData::Private(data)) => {
                let policy = get_policy!(policy_id, data)?;
                let requester = self.operations_pk;
                check_perm!(policy, requester, action)
            }
        }
    }

    /// Returns the owner of the data.
    pub fn owner(&self) -> PublicKey {
        self.operations_pk
    }
}

#[cfg(test)]
mod tests {
    use crate::{
        utils, Error, Keypair, PublicKey, Result, Sequence, SequenceAddress, SequenceDataWriteOp,
        SequenceEntry, SequenceIndex, SequenceKind, SequencePermissions, SequencePolicyWriteOp,
        SequencePrivatePermissions, SequencePrivatePolicy, SequencePublicPermissions,
        SequencePublicPolicy, SequenceUser,
    };

    use proptest::prelude::*;
    use rand::rngs::OsRng;
    use rand::seq::SliceRandom;
    use std::collections::BTreeMap;
    use std::sync::Arc;
    use xor_name::XorName;

    pub fn sign_sequence_data_op(
        mut op: SequenceDataWriteOp<SequenceEntry>,
        keypair: &Keypair,
    ) -> Result<SequenceDataWriteOp<SequenceEntry>> {
        let bytes = utils::serialise(&op.crdt_op)?;
        let signature = keypair.sign(&bytes);
        op.signature = Some(signature);
        Ok(op)
    }

    pub fn sign_sequence_public_policy_op(
        mut op: SequencePolicyWriteOp<SequencePublicPolicy>,
        keypair: &Keypair,
    ) -> Result<SequencePolicyWriteOp<SequencePublicPolicy>> {
        let bytes = utils::serialise(&op.crdt_op)?;
        let signature = keypair.sign(&bytes);
        op.signature = Some(signature);
        Ok(op)
    }

    pub fn sign_sequence_private_policy_op(
        mut op: SequencePolicyWriteOp<SequencePrivatePolicy>,
        keypair: &Keypair,
    ) -> Result<SequencePolicyWriteOp<SequencePrivatePolicy>> {
        let bytes = utils::serialise(&op.crdt_op).map_err(|_| "Could not serialize op")?;
        let signature = keypair.sign(&bytes);
        op.signature = Some(signature);
        Ok(op)
    }

    #[test]
    fn sequence_create_public() {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let actor = op_source_pk_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let sequence =
            Sequence::new_public(actor, "actor".to_string(), sequence_name, sequence_tag);
        assert_eq!(sequence.kind(), SequenceKind::Public);
        assert_eq!(*sequence.name(), sequence_name);
        assert_eq!(sequence.tag(), sequence_tag);
        assert!(sequence.is_pub());
        assert!(!sequence.is_private());

        let sequence_address =
            SequenceAddress::from_kind(SequenceKind::Public, sequence_name, sequence_tag);
        assert_eq!(*sequence.address(), sequence_address);
    }

    #[test]
    fn sequence_create_private() {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let actor = op_source_pk_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let sequence =
            Sequence::new_private(actor, "actor".to_string(), sequence_name, sequence_tag);
        assert_eq!(sequence.kind(), SequenceKind::Private);
        assert_eq!(*sequence.name(), sequence_name);
        assert_eq!(sequence.tag(), sequence_tag);
        assert!(!sequence.is_pub());
        assert!(sequence.is_private());

        let sequence_address =
            SequenceAddress::from_kind(SequenceKind::Private, sequence_name, sequence_tag);
        assert_eq!(*sequence.address(), sequence_address);
    }

    #[test]
    fn sequence_append_entry_and_apply() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let actor = op_source_pk_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 =
            Sequence::new_public(actor, "actor".to_string(), sequence_name, sequence_tag);
        let mut replica2 =
            Sequence::new_public(actor, "actor".to_string(), sequence_name, sequence_tag);

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePublicPermissions::new(true, false);
        let _ = perms1.insert(SequenceUser::Anyone, user_perms1);
        let policy_op = replica1.create_unsigned_public_policy_op(actor, perms1)?;
        let signed_op = sign_sequence_public_policy_op(policy_op, &op_source_pk_keypair)?;
        replica1.apply_public_policy_op(signed_op.clone())?;
        replica2.apply_public_policy_op(signed_op)?;

        let entry1 = b"value0".to_vec();
        let entry2 = b"value1".to_vec();

        let op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(entry1.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(op1.clone())?;
        let op2 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(entry2.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(op2.clone())?;

        // we apply the operations in different order, to verify that doesn't affect the result
        replica2.apply_data_op(op2)?;
        replica2.apply_data_op(op1)?;

        assert_eq!(replica1.len(None)?, 2);
        assert_eq!(replica2.len(None)?, 2);

        let index_0 = SequenceIndex::FromStart(0);
        let first_entry = replica1.get(index_0, None)?;
        assert_eq!(first_entry, Some(&entry1));
        assert_eq!(first_entry, replica2.get(index_0, None)?);

        let index_1 = SequenceIndex::FromStart(1);
        let second_entry = replica1.get(index_1, None)?;
        assert_eq!(second_entry, Some(&entry2));
        assert_eq!(second_entry, replica2.get(index_1, None)?);

        let last_entry = replica1.last_entry(None)?;
        assert_eq!(last_entry, Some(&entry2));
        assert_eq!(last_entry, replica2.last_entry(None)?);

        Ok(())
    }

    #[test]
    fn sequence_public_create_policy_op_and_apply() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let actor = op_source_pk_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 =
            Sequence::new_public(actor, "actor".to_string(), sequence_name, sequence_tag);
        let mut replica2 =
            Sequence::new_public(actor, "actor".to_string(), sequence_name, sequence_tag);

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePublicPermissions::new(true, false);
        let _ = perms1.insert(SequenceUser::Anyone, user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePublicPermissions::new(false, true);
        let _ = perms2.insert(SequenceUser::Key(actor), user_perms2);

        let op1 = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(actor, perms1.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(op1.clone())?;

        let op2 = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(actor, perms2.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(op2.clone())?;

        // if we apply the operations in different order it should fail
        // as op2 is not causally ready in replica2, it depends on op1
        check_not_causally_ready_failure(replica2.apply_public_policy_op(op2.clone()))?;

        // let's apply op1 first then
        replica2.apply_public_policy_op(op1)?;
        replica2.apply_public_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        assert_eq!(replica2.policy_version(None)?, Some(1));

        let index_0 = SequenceIndex::FromStart(0);
        let first_entry = replica1.public_policy_at(index_0)?;
        assert_eq!(first_entry.permissions, perms1);
        assert_eq!(first_entry.owner, actor);
        assert_eq!(first_entry, replica2.public_policy_at(index_0)?);
        assert_eq!(
            SequencePermissions::Public(user_perms1),
            replica1.permissions_at(SequenceUser::Anyone, index_0, None)?
        );

        let index_1 = SequenceIndex::FromStart(1);
        let second_entry = replica1.public_policy_at(index_1)?;
        assert_eq!(second_entry.permissions, perms2);
        assert_eq!(second_entry.owner, actor);
        assert_eq!(second_entry, replica2.public_policy_at(index_1)?);
        assert_eq!(
            SequencePermissions::Public(user_perms2),
            replica1.permissions_at(SequenceUser::Key(actor), index_1, None)?
        );

        Ok(())
    }

    #[test]
    fn sequence_private_create_policy_op_and_apply() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 = Sequence::new_private(
            op_source_pk,
            "some_identifying_str".to_string(),
            sequence_name,
            sequence_tag,
        );
        let mut replica2 = Sequence::new_private(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sequence_name,
            sequence_tag,
        );

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePrivatePermissions::new(true, false, true);
        let _ = perms1.insert(op_source_pk, user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePrivatePermissions::new(true, true, false);
        let _ = perms2.insert(op_source_pk2, user_perms2);

        let op1 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk2, perms1.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op1.clone())?;
        let op2 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk, perms2.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op2.clone())?;

        // if we apply the operations in different order it should fail
        // as op2 is not causally ready in replica2, it depends on op1
        check_not_causally_ready_failure(replica2.apply_private_policy_op(op2.clone()))?;

        // let's apply op1 first then
        replica2.apply_private_policy_op(op1)?;
        replica2.apply_private_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        assert_eq!(replica2.policy_version(None)?, Some(1));

        let index_0 = SequenceIndex::FromStart(0);
        let first_entry = replica1.private_policy_at(index_0, None)?;
        assert_eq!(first_entry.permissions, perms1);
        assert_eq!(first_entry.owner, op_source_pk2);
        assert_eq!(first_entry, replica2.private_policy_at(index_0, None)?);
        assert_eq!(
            SequencePermissions::Private(user_perms1),
            replica1.permissions_at(SequenceUser::Key(op_source_pk), index_0, None)?
        );

        let index_1 = SequenceIndex::FromStart(1);
        let second_entry = replica1.private_policy_at(index_1, None)?;
        assert_eq!(second_entry.permissions, perms2);
        assert_eq!(second_entry.owner, op_source_pk);
        assert_eq!(second_entry, replica2.private_policy_at(index_1, None)?);
        assert_eq!(
            SequencePermissions::Private(user_perms2),
            replica1.permissions_at(SequenceUser::Key(op_source_pk2), index_1, None)?
        );

        Ok(())
    }

    #[test]
    fn sequence_public_create_policy_op_and_append_fails_when_no_perms_for_actor() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 = Sequence::new_public(
            op_source_pk,
            "some_identifying_str".to_string(),
            sequence_name,
            sequence_tag,
        );
        let mut replica2 = Sequence::new_public(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sequence_name,
            sequence_tag,
        );

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePublicPermissions::new(true, false);
        let _ = perms1.insert(SequenceUser::Key(op_source_pk), user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePublicPermissions::new(false, false);
        let _ = perms2.insert(SequenceUser::Key(op_source_pk2), user_perms2);

        let op1 = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms1)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(op1.clone())?;

        let op2 = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms2)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(op2.clone())?;

        // let's apply op perms...
        replica2.apply_public_policy_op(op1)?;
        replica2.apply_public_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        assert_eq!(replica2.policy_version(None)?, Some(1));

        // And let's append to both replicas with one first item
        let item1 = b"item1";
        let item2 = b"item2";
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1.to_vec())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        replica2.apply_data_op(append_op1)?;

        check_op_not_allowed_failure(replica2.create_unsigned_append_op(item2.to_vec()))?;

        assert_eq!(replica1.len(None)?, replica2.len(None)?);

        Ok(())
    }

    #[test]
    fn sequence_public_create_policy_op_and_append_succeeds_when_perms_updated_for_actor(
    ) -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 = Sequence::new_public(
            op_source_pk,
            "some_identifying_str".to_string(),
            sequence_name,
            sequence_tag,
        );
        let mut replica2 = Sequence::new_public(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sequence_name,
            sequence_tag,
        );

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePublicPermissions::new(true, false);
        let _ = perms1.insert(SequenceUser::Key(op_source_pk), user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePublicPermissions::new(false, false);
        let _ = perms2.insert(SequenceUser::Key(op_source_pk2), user_perms2);

        let op1 = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms1)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(op1.clone())?;
        let op2 = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms2)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(op2.clone())?;

        // let's apply op perms...
        replica2.apply_public_policy_op(op1)?;
        replica2.apply_public_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        assert_eq!(replica2.policy_version(None)?, Some(1));

        // And let's append to both replicas with one first item
        let item1 = b"item1";
        let item2 = b"item2";
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1.to_vec())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        replica2.apply_data_op(append_op1)?;

        check_op_not_allowed_failure(replica2.create_unsigned_append_op(item2.to_vec()))?;

        // Now lets update the policy and try appending again

        let mut perms3 = BTreeMap::default();
        let user_perms3 = SequencePublicPermissions::new(true, false);
        let _ = perms3.insert(SequenceUser::Key(op_source_pk2), user_perms3);

        let op1 = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms3)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(op1.clone())?;
        replica2.apply_public_policy_op(op1)?;

        let append_op = sign_sequence_data_op(
            replica2.create_unsigned_append_op(item2.to_vec())?,
            &op_source_pk2_keypair,
        )?;
        replica2.apply_data_op(append_op.clone())?;
        replica1.apply_data_op(append_op)?;
        assert_eq!(replica1.len(None)?, replica2.len(None)?);

        Ok(())
    }

    #[test]
    fn sequence_private_create_policy_op_and_append_fails_when_no_perms_for_actor() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 = Sequence::new_private(
            op_source_pk,
            "some_identifying_str".to_string(),
            sequence_name,
            sequence_tag,
        );
        let mut replica2 = Sequence::new_private(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sequence_name,
            sequence_tag,
        );

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePrivatePermissions::new(true, false, true);
        let _ = perms1.insert(op_source_pk, user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePrivatePermissions::new(false, false, false);
        let _ = perms2.insert(op_source_pk2, user_perms2);

        let op1 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk2, perms1)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op1.clone())?;

        let op2 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk, perms2)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op2.clone())?;

        // let's apply op perms...
        replica2.apply_private_policy_op(op1)?;
        replica2.apply_private_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        check_op_not_allowed_failure(replica2.policy_version(None))?;
        assert_eq!(replica2.policy_version(Some(op_source_pk))?, Some(1));

        // And let's append to both replicas with one first item
        let item1 = b"item1";
        let item2 = b"item2";
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1.to_vec())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        replica2.apply_data_op(append_op1)?;

        check_op_not_allowed_failure(replica2.create_unsigned_append_op(item2.to_vec()))?;

        Ok(())
    }

    #[test]
    fn sequence_private_create_policy_op_and_get_read_fails_when_no_perms_for_actor() -> Result<()>
    {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 = Sequence::new_private(
            op_source_pk,
            "some_identifying_str".to_string(),
            sequence_name,
            sequence_tag,
        );
        let mut replica2 = Sequence::new_private(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sequence_name,
            sequence_tag,
        );

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePrivatePermissions::new(true, false, true);
        let _ = perms1.insert(op_source_pk, user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePrivatePermissions::new(false, false, false);
        let _ = perms2.insert(op_source_pk2, user_perms2);

        let op1 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk2, perms1)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op1.clone())?;
        let op2 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk, perms2)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op2.clone())?;

        // let's apply op perms...
        replica2.apply_private_policy_op(op1)?;
        replica2.apply_private_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        check_op_not_allowed_failure(replica2.policy_version(None))?;
        assert_eq!(replica2.policy_version(Some(op_source_pk))?, Some(1));

        // And let's append to both replicas with one first item
        let item1 = b"item1".to_vec();
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        replica2.apply_data_op(append_op1)?;

        // lets check replica1 can read that
        let data = replica1.get(SequenceIndex::FromStart(0), None)?;
        assert_eq!(data, Some(&item1));

        // replica2 cannot read using op_source_pk2 pk
        check_op_not_allowed_failure(
            replica2.get(SequenceIndex::FromStart(0), Some(op_source_pk2)),
        )?;
        // but it can read with pk from op_source_pk
        let data = replica2.get(SequenceIndex::FromStart(0), Some(op_source_pk))?;
        assert_eq!(data, Some(&item1));

        Ok(())
    }

    #[test]
    fn sequence_private_create_policy_op_and_last_entry_read_fails_when_no_perms_for_actor(
    ) -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 = Sequence::new_private(
            op_source_pk,
            "some_identifying_str".to_string(),
            sequence_name,
            sequence_tag,
        );
        let mut replica2 = Sequence::new_private(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sequence_name,
            sequence_tag,
        );

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePrivatePermissions::new(true, false, true);
        let _ = perms1.insert(op_source_pk, user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePrivatePermissions::new(false, false, false);
        let _ = perms2.insert(op_source_pk2, user_perms2);

        let op1 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk2, perms1)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op1.clone())?;

        let op2 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk, perms2)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op2.clone())?;

        // let's apply op perms...
        replica2.apply_private_policy_op(op1)?;
        replica2.apply_private_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        check_op_not_allowed_failure(replica2.policy_version(None))?;
        assert_eq!(replica2.policy_version(Some(op_source_pk))?, Some(1));

        // And let's append to both replicas with one first item
        let item1 = b"item1".to_vec();
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        replica2.apply_data_op(append_op1)?;

        // lets check replica1 can read that, and replica2 not...
        let data = replica1.last_entry(None)?;
        assert_eq!(data, Some(&item1));

        check_op_not_allowed_failure(replica2.last_entry(Some(op_source_pk2)))?;
        let data = replica2.last_entry(Some(op_source_pk))?;
        assert_eq!(data, Some(&item1));

        Ok(())
    }

    #[test]
    fn sequence_private_create_policy_op_and_range_read_fails_when_no_perms_for_actor() -> Result<()>
    {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 = Sequence::new_private(
            op_source_pk,
            "some_identifying_str".to_string(),
            sequence_name,
            sequence_tag,
        );
        let mut replica2 = Sequence::new_private(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sequence_name,
            sequence_tag,
        );

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePrivatePermissions::new(true, false, true);
        let _ = perms1.insert(op_source_pk, user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePrivatePermissions::new(false, false, false);
        let _ = perms2.insert(op_source_pk2, user_perms2);

        let op1 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk2, perms1)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op1.clone())?;

        let op2 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk, perms2)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op2.clone())?;

        // let's apply op perms...
        replica2.apply_private_policy_op(op1)?;
        replica2.apply_private_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        check_op_not_allowed_failure(replica2.policy_version(None))?;
        assert_eq!(replica2.policy_version(Some(op_source_pk))?, Some(1));

        // And let's append to both replicas with one first item
        let item1 = b"item1".to_vec();
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        replica2.apply_data_op(append_op1)?;

        // lets check replica1 can read that, and replica2 not...
        let data = replica1.in_range(
            SequenceIndex::FromStart(0),
            SequenceIndex::FromStart(1),
            None,
        )?;
        assert_eq!(data, Some(vec![item1.clone()]));

        check_op_not_allowed_failure(replica2.in_range(
            SequenceIndex::FromStart(0),
            SequenceIndex::FromStart(1),
            Some(op_source_pk2),
        ))?;
        let data = replica2.in_range(
            SequenceIndex::FromStart(0),
            SequenceIndex::FromStart(1),
            Some(op_source_pk),
        )?;
        assert_eq!(data, Some(vec![item1]));

        Ok(())
    }

    #[test]
    fn sequence_private_create_policy_op_and_read_possible_after_update() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sequence_name = XorName::random();
        let sequence_tag = 43_000;
        let mut replica1 = Sequence::new_private(
            op_source_pk,
            "some_identifying_str".to_string(),
            sequence_name,
            sequence_tag,
        );
        let mut replica2 = Sequence::new_private(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sequence_name,
            sequence_tag,
        );

        let mut perms1 = BTreeMap::default();
        let user_perms1 = SequencePrivatePermissions::new(true, false, true);
        let _ = perms1.insert(op_source_pk, user_perms1);

        let mut perms2 = BTreeMap::default();
        let user_perms2 = SequencePrivatePermissions::new(false, false, false);
        let _ = perms2.insert(op_source_pk2, user_perms2);

        let op1 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk2, perms1)?,
            &op_source_pk_keypair,
        )?;

        replica1.apply_private_policy_op(op1.clone())?;

        let op2 = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk, perms2)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_private_policy_op(op2.clone())?;

        // let's apply op perms...
        replica2.apply_private_policy_op(op1)?;
        replica2.apply_private_policy_op(op2)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        check_op_not_allowed_failure(replica2.policy_version(None))?;
        assert_eq!(replica2.policy_version(Some(op_source_pk))?, Some(1));

        // And let's append to both replicas with one first item
        let item1 = b"item1".to_vec();
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        replica2.apply_data_op(append_op1)?;

        // lets check replica1 can read that, and replica2 not...
        let data = replica1.get(SequenceIndex::FromStart(0), None)?;
        assert_eq!(data, Some(&item1));

        check_op_not_allowed_failure(
            replica2.get(SequenceIndex::FromStart(0), Some(op_source_pk2)),
        )?;

        // set readable for replica 2 once more
        let mut perms3 = BTreeMap::default();
        let user_perms3 = SequencePrivatePermissions::new(true, false, false);
        let _ = perms3.insert(op_source_pk2, user_perms3);

        let updated_perms_op = sign_sequence_private_policy_op(
            replica1.create_unsigned_private_policy_op(op_source_pk, perms3)?,
            &op_source_pk_keypair,
        )?;

        // let's apply op perms...
        replica1.apply_private_policy_op(updated_perms_op.clone())?;
        replica2.apply_private_policy_op(updated_perms_op)?;

        let _ = replica1.get(SequenceIndex::FromStart(0), None)?;
        let data = replica2.get(SequenceIndex::FromStart(0), None)?;

        assert_eq!(data, replica2.last_entry(None)?);
        assert_eq!(data, Some(&item1));

        let data = replica2.in_range(
            SequenceIndex::FromStart(0),
            SequenceIndex::FromStart(1),
            None,
        )?;
        assert_eq!(data, Some(vec![item1]));

        // finally check the policy has been updated
        assert_eq!(replica1.policy_version(None)?, Some(2));
        assert_eq!(replica2.policy_version(None)?, Some(2));

        Ok(())
    }

    #[test]
    fn sequence_concurrent_policy_and_data_ops() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sdata_name: XorName = rand::random();
        let sdata_tag = 43_000u64;

        // Instantiate the same Sequence on two replicas with two diff actors
        let mut replica1 = Sequence::new_public(
            op_source_pk,
            "some_identifying_str".to_string(),
            sdata_name,
            sdata_tag,
        );
        let mut replica2 = Sequence::new_public(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sdata_name,
            sdata_tag,
        );

        // Set Actor1 as the owner in both replicas and
        // grant authorisation for Append to Actor2 in both replicas
        let mut perms = BTreeMap::default();
        let user_perms =
            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ false);
        let _ = perms.insert(SequenceUser::Key(op_source_pk2), user_perms);
        let grant_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(grant_op.clone())?;
        replica2.apply_public_policy_op(grant_op)?;

        // And let's append to both replicas with one first item
        let item1 = b"item1";
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1.to_vec())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        replica2.apply_data_op(append_op1)?;

        // Let's assert initial state on both replicas
        assert_eq!(replica1.len(None)?, 1);
        assert_eq!(replica1.policy_version(None)?, Some(0));
        assert_eq!(replica2.len(None)?, 1);
        assert_eq!(replica2.policy_version(None)?, Some(0));

        // We revoke authorisation for Actor2 locally on replica1
        let revoke_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, BTreeMap::default())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(revoke_op.clone())?;
        // New Policy should have been set on replica1
        assert_eq!(replica1.policy_version(None)?, Some(1));

        // Concurrently append an item with Actor2 on replica2
        let item2 = b"item2";
        let append_op2 = sign_sequence_data_op(
            replica2.create_unsigned_append_op(item2.to_vec())?,
            &op_source_pk2_keypair,
        )?;
        replica2.apply_data_op(append_op2.clone())?;
        // Item should be appended on replica2
        assert_eq!(replica2.len(None)?, 2);

        // Append operation is broadcasted and applied on replica1 using old Policy
        replica1.apply_data_op(append_op2)?;
        assert_eq!(replica1.len(None)?, 1);

        // Now revoke operation is broadcasted and applied on replica2
        replica2.apply_public_policy_op(revoke_op)?;
        assert_eq!(replica2.policy_version(None)?, Some(1));
        assert_eq!(replica2.len(None)?, 1);

        // Let's assert that append_op2 created a branch of data on both replicas
        // due to new policy having been applied concurrently, thus only first
        // item shall be returned from main branch of data
        verify_data_convergence(vec![replica1, replica2], 1)?;

        Ok(())
    }

    #[test]
    fn sequence_causality_between_data_and_policy_ops() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let op_source_pk3_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk3 = op_source_pk3_keypair.public_key();
        let sdata_name: XorName = rand::random();
        let sdata_tag = 43_001u64;

        // Instantiate the same Sequence on three replicas with three diff actors
        let mut replica1 = Sequence::new_public(
            op_source_pk,
            "some_identifying_str".to_string(),
            sdata_name,
            sdata_tag,
        );
        let mut replica2 = Sequence::new_public(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sdata_name,
            sdata_tag,
        );
        let mut replica3 = Sequence::new_public(
            op_source_pk3,
            "some_identifying_str3".to_string(),
            sdata_name,
            sdata_tag,
        );

        // Set Actor1 as the owner in all replicas, with Append perms for Actor3
        let mut perms = BTreeMap::default();
        let user_perms =
            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ false);
        let _ = perms.insert(SequenceUser::Key(op_source_pk3), user_perms);
        let owner_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(owner_op.clone())?;
        replica2.apply_public_policy_op(owner_op.clone())?;
        replica3.apply_public_policy_op(owner_op)?;

        // Grant authorisation for Append and Admin to Actor3 in replica1,
        // and apply it to replica3 too
        let mut perms = BTreeMap::default();
        let user_perms =
            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ true);
        let _ = perms.insert(SequenceUser::Key(op_source_pk3), user_perms);
        let grant_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(grant_op.clone())?;
        replica3.apply_public_policy_op(grant_op.clone())?;

        // Let's assert the state on three replicas
        assert_eq!(replica1.len(None)?, 0);
        assert_eq!(replica1.policy_version(None)?, Some(1));
        assert_eq!(replica2.len(None)?, 0);
        assert_eq!(replica2.policy_version(None)?, Some(0));
        assert_eq!(replica3.len(None)?, 0);
        assert_eq!(replica3.policy_version(None)?, Some(1));

        // We append an item with Actor3 on replica3
        let item = b"item0";
        let append_op = sign_sequence_data_op(
            replica3.create_unsigned_append_op(item.to_vec())?,
            &op_source_pk3_keypair,
        )?;
        replica3.apply_data_op(append_op.clone())?;
        assert_eq!(replica3.len(None)?, 1);

        // Append op is broadcasted and applied on replica1
        replica1.apply_data_op(append_op.clone())?;
        assert_eq!(replica1.len(None)?, 1);

        // And now append op is broadcasted and applied on replica2
        // It should be rejected on replica2 as it's not causally ready
        check_not_causally_ready_failure(replica2.apply_data_op(append_op.clone()))?;
        assert_eq!(replica2.len(None)?, 0);

        // So let's apply grant operation to replica2
        replica2.apply_public_policy_op(grant_op)?;
        assert_eq!(replica2.policy_version(None)?, Some(1));

        // Retrying to apply append op to replica2 should be successful, due
        // to now being causally ready with the new policy
        replica2.apply_data_op(append_op)?;
        verify_data_convergence(vec![replica1, replica2, replica3], 1)?;

        Ok(())
    }

    #[test]
    fn sequence_concurrent_policy_ops() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sdata_name: XorName = rand::random();
        let sdata_tag = 43_001u64;

        // Instantiate the same Sequence on two replicas with two diff actors
        let mut replica1 = Sequence::new_public(
            op_source_pk,
            "some_identifying_str".to_string(),
            sdata_name,
            sdata_tag,
        );
        let mut replica2 = Sequence::new_public(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sdata_name,
            sdata_tag,
        );

        // Set Actor1 as the owner and Actor2 with Append and Admin perms in all replicas
        let mut perms = BTreeMap::default();
        let user_perms =
            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ true);
        let _ = perms.insert(SequenceUser::Key(op_source_pk2), user_perms);
        let owner_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(owner_op.clone())?;
        replica2.apply_public_policy_op(owner_op)?;

        // Append item on replica1, and apply it to replica2
        let item0 = b"item0".to_vec();
        let append_op = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item0)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op.clone())?;
        replica2.apply_data_op(append_op)?;

        // Let's assert the state on both replicas
        assert_eq!(replica1.len(None)?, 1);
        assert_eq!(replica1.policy_version(None)?, Some(0));
        assert_eq!(replica2.len(None)?, 1);
        assert_eq!(replica2.policy_version(None)?, Some(0));

        // Concurrently set new policy (new random owner) with Append and Admin perms
        // for both actors on both replicas
        let _ = perms.insert(SequenceUser::Key(op_source_pk), user_perms);
        let owner_op_1 = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(generate_public_key(), perms.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(owner_op_1.clone())?;

        let owner_op_2 = sign_sequence_public_policy_op(
            replica2.create_unsigned_public_policy_op(generate_public_key(), perms)?,
            &op_source_pk2_keypair,
        )?;
        replica2.apply_public_policy_op(owner_op_2.clone())?;

        // ...and concurrently append a new item on top of their own respective new policies
        let item1_r1 = b"item1_replica1".to_vec();
        let item1_r2 = b"item1_replica2".to_vec();
        let append_op1 = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item1_r1)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_data_op(append_op1.clone())?;
        let append_op2 = sign_sequence_data_op(
            replica2.create_unsigned_append_op(item1_r2)?,
            &op_source_pk2_keypair,
        )?;
        replica2.apply_data_op(append_op2.clone())?;

        assert_eq!(replica1.len(None)?, 2);
        assert_eq!(replica2.len(None)?, 2);

        // Let's now apply the policy op to the other replica
        replica1.apply_public_policy_op(owner_op_2)?;
        replica2.apply_public_policy_op(owner_op_1)?;

        assert_eq!(replica1.policy_version(None)?, Some(2));
        assert_eq!(replica2.policy_version(None)?, Some(2));

        // Let's now apply the append ops on the other replica
        replica1.apply_data_op(append_op2)?;
        replica2.apply_data_op(append_op1)?;

        // Let's assert the state on all replicas to assure convergence
        // One of the items appended concurrently should not belong to
        // the master branch of the data thus we should see only 2 items
        verify_data_convergence(vec![replica1, replica2], 2)?;

        Ok(())
    }

    #[test]
    fn sequence_old_data_op() -> Result<()> {
        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sdata_name: XorName = rand::random();
        let sdata_tag = 43_001u64;

        // Instantiate the same Sequence on two replicas with two diff actors
        let mut replica1 = Sequence::new_public(
            op_source_pk,
            "some_identifying_str".to_string(),
            sdata_name,
            sdata_tag,
        );
        let mut replica2 = Sequence::new_public(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sdata_name,
            sdata_tag,
        );

        // Set Actor1 as the owner and Actor2 with append perms in all replicas
        let mut perms = BTreeMap::default();
        let user_perms =
            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ false);
        let _ = perms.insert(SequenceUser::Key(op_source_pk2), user_perms);
        let owner_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(owner_op.clone())?;
        replica2.apply_public_policy_op(owner_op)?;

        // Append an item on replica1
        let item0 = b"item0".to_vec();
        let append_op = sign_sequence_data_op(
            replica1.create_unsigned_append_op(item0)?,
            &op_source_pk_keypair,
        )?;

        replica1.apply_data_op(append_op.clone())?;

        // A new Policy is set in replica1 and applied to replica2
        let policy_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, BTreeMap::default())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(policy_op.clone())?;
        replica2.apply_public_policy_op(policy_op)?;

        // Now the old append op is applied to replica2
        replica2.apply_data_op(append_op)?;

        assert_eq!(replica1.policy_version(None)?, Some(1));
        assert_eq!(replica2.policy_version(None)?, Some(1));

        verify_data_convergence(vec![replica1, replica2], 1)?;

        Ok(())
    }

    #[test]
    fn sequence_old_policy_op() -> Result<()> {
        // Assuming the following scenario:
        // - op_source_pk is the owner of the Seq as per first policy (policy1),
        // - op_source_pk2 is then owner of the Seq as per second/new policy (policy2)
        // Then replica1 sends a policy op (note it's not the owner anymore)
        // which was generated before applying the policy2 op, thus it can still be applied
        // but as an old policy between policy1 and policy2 in the policies history.

        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sdata_name: XorName = rand::random();
        let sdata_tag = 43_001u64;

        // Instantiate the same Sequence on two replicas with two diff actors
        let mut replica1 = Sequence::new_public(
            op_source_pk,
            "some_identifying_str".to_string(),
            sdata_name,
            sdata_tag,
        );
        let mut replica2 = Sequence::new_public(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sdata_name,
            sdata_tag,
        );

        // Set Actor1 as the owner in both replicas (policy1)
        let mut perms = BTreeMap::default();
        let user_perms =
            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ true);
        let _ = perms.insert(SequenceUser::Key(op_source_pk2), user_perms);
        let owner_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms)?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(owner_op.clone())?;
        replica2.apply_public_policy_op(owner_op)?;

        // Let's create a second policy op on replica1, but don't apply it to replica2 yet
        let perms = BTreeMap::default();
        let op_source3_pk = Keypair::new_ed25519(&mut OsRng).public_key();
        let old_owner_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source3_pk, perms.clone())?,
            &op_source_pk_keypair,
        )?;

        replica1.apply_public_policy_op(old_owner_op.clone())?;

        // Set Actor2 as the new owner in replica2 (policy2)
        let owner_op = sign_sequence_public_policy_op(
            replica2.create_unsigned_public_policy_op(op_source_pk2, perms)?,
            &op_source_pk2_keypair,
        )?;

        replica2.apply_public_policy_op(owner_op.clone())?;

        // Now apply the old policy op to replica2, which should be applied as
        // an old policy even if the current/latest policy doesn't allow op_source_pk to change policy
        replica2.apply_public_policy_op(old_owner_op)?;

        // and finally apply the latest owner op to replica1
        replica1.apply_public_policy_op(owner_op)?;

        // Let's assert the state on both replicas
        assert_eq!(replica1.policy_version(None)?, Some(2));
        assert_eq!(replica2.policy_version(None)?, Some(2));

        // Let's assert the owners set in policy1 and policy2
        // are op_source_pk and op_source_pk2 respectivelly
        let policy1 = replica1.public_policy_at(0)?;
        assert_eq!(policy1.owner, op_source_pk);
        assert_eq!(policy1.owner, replica2.public_policy_at(0)?.owner);

        let policy2 = replica1.public_policy_at(1)?;
        if policy2.owner == op_source3_pk {
            assert_eq!(policy2.owner, replica2.public_policy_at(1)?.owner);
            let policy3 = replica1.public_policy_at(2)?;
            assert_eq!(policy3.owner, op_source_pk2);
            assert_eq!(policy3.owner, replica2.public_policy_at(2)?.owner);
        } else {
            assert_eq!(policy2.owner, op_source_pk2);
            assert_eq!(policy2.owner, replica2.public_policy_at(1)?.owner);

            let policy3 = replica1.public_policy_at(2)?;
            assert_eq!(policy3.owner, op_source3_pk);
            assert_eq!(policy3.owner, replica2.public_policy_at(2)?.owner);
        }

        Ok(())
    }

    #[test]
    fn sequence_falsified_policy_op() -> Result<()> {
        // Assuming the following scenario:
        // - op_source_pk is the owner of the Seq as per first policy (policy1),
        // - op_source_pk2 is then owner of the Seq as per second/new policy (policy2)
        // Then replica1 tries to cheat by sending a policy op (note it
        // is not the owner anymore, thus shouldn't be allowed) by setting
        // the correct context/dependency on policy1, but setting the identifier
        // of the policy to be appended after policy2
        // Note: this is currently a known issue, once an entity was given
        // Admin permissions it can always send policy changes on top of any policy.

        let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk = op_source_pk_keypair.public_key();
        let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
        let op_source_pk2 = op_source_pk2_keypair.public_key();
        let sdata_name: XorName = rand::random();
        let sdata_tag = 43_001u64;

        // Instantiate the same Sequence on two replicas with two diff actors
        let mut replica1 = Sequence::new_public(
            op_source_pk,
            "some_identifying_str".to_string(),
            sdata_name,
            sdata_tag,
        );
        let mut replica2 = Sequence::new_public(
            op_source_pk2,
            "some_identifying_str2".to_string(),
            sdata_name,
            sdata_tag,
        );

        // Set Actor1 as the owner in both replicas (policy1)
        let perms = BTreeMap::default();
        let owner_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk, perms.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(owner_op.clone())?;
        replica2.apply_public_policy_op(owner_op)?;

        // Let's create a clone of replica1 we'll use later on to falsify a policy op
        let mut temp_replica = replica1.clone();

        // Set Actor2 as the new owner in both replicas (policy2)
        let owner_op = sign_sequence_public_policy_op(
            replica1.create_unsigned_public_policy_op(op_source_pk2, perms.clone())?,
            &op_source_pk_keypair,
        )?;
        replica1.apply_public_policy_op(owner_op.clone())?;
        replica2.apply_public_policy_op(owner_op)?;

        // Let's assert the state on both replicas
        assert_eq!(replica1.policy_version(None)?, Some(1));
        assert_eq!(replica2.policy_version(None)?, Some(1));

        // Now replica1 shouldn't be allowed to set a new policy since
        // it's not the owner anymore, thus we use the cloned replica
        // we created above (where replica1 is still the owner)
        // in order to falsify a new policy operation
        let mut owner_op = sign_sequence_public_policy_op(
            temp_replica.create_unsigned_public_policy_op(op_source_pk, perms)?,
            &op_source_pk_keypair,
        )?;

        // Let's falsify the op by changing the policy CRDT identifier to be the last
        use crdts::lseq::{ident::IdentGen, Op};
        let upper_ident = IdentGen::new("some_identifying_str".to_string()).upper();
        owner_op.crdt_op = match owner_op.crdt_op {
            Op::Insert { dot, val, .. } => Op::Insert {
                id: upper_ident,
                dot,
                val,
            },
            Op::Delete { remote, dot, .. } => Op::Delete {
                remote,
                id: upper_ident,
                dot,
            },
        };

        owner_op = sign_sequence_public_policy_op(owner_op, &op_source_pk_keypair)?;

        // and now we overwrite the sig
        replica1.apply_public_policy_op(owner_op.clone())?;

        replica2.apply_public_policy_op(owner_op)?;

        // Let's assert the state on both replicas
        assert_eq!(replica1.policy_version(None)?, Some(2));
        assert_eq!(replica2.policy_version(None)?, Some(2));

        // Let's assert the owners set in policy1 and policy2 converge
        assert_eq!(
            replica1.public_policy_at(0)?.owner,
            replica2.public_policy_at(0)?.owner
        );
        assert_eq!(
            replica1.public_policy_at(1)?.owner,
            replica2.public_policy_at(1)?.owner
        );

        Ok(())
    }

    // Helpers for tests

    fn generate_public_key() -> PublicKey {
        let keypair = Keypair::new_ed25519(&mut OsRng);
        keypair.public_key()
    }

    // check it fails due to not being causally ready
    fn check_not_causally_ready_failure(result: Result<()>) -> Result<()> {
        match result {
            Err(Error::OpNotCausallyReady) => Ok(()),
            Err(err) => Err(Error::Unexpected(format!(
                "Error returned was the unexpected one: {}",
                err
            ))),
            Ok(()) => Err(Error::Unexpected(
                "Data op applied unexpectedly".to_string(),
            )),
        }
    }

    // check it fails due to not having permissions
    fn check_op_not_allowed_failure<T>(result: Result<T>) -> Result<()> {
        match result {
            Err(Error::AccessDenied) => Ok(()),
            Err(err) => Err(Error::Unexpected(format!(
                "Error returned was the unexpected one for a non-allowed op: {}",
                err
            ))),
            Ok(_) => Err(Error::Unexpected(
                "Data operation succeded unexpectedly, an AccessDenied error was expected"
                    .to_string(),
            )),
        }
    }

    // verify data convergence on a set of replicas and with the expected length
    fn verify_data_convergence(replicas: Vec<Sequence>, expected_len: u64) -> Result<()> {
        // verify replicas have the expected length
        // also verify replicas failed to get with index beyond reported length
        let index_beyond = SequenceIndex::FromStart(expected_len);
        for r in &replicas {
            assert_eq!(r.len(None)?, expected_len);
            assert_eq!(r.get(index_beyond, None)?, None);
        }

        // now verify that the items are the same in all replicas
        for i in 0..expected_len {
            let index = SequenceIndex::FromStart(i);
            let r0_entry = replicas[0].get(index, None)?;
            for r in &replicas {
                assert_eq!(r0_entry, r.get(index, None)?);
            }
        }

        Ok(())
    }

    // Generate a vec of Sequence replicas of some length, with corresponding vec of keypairs for signing, and the overall owner of the sequence
    fn generate_replicas(
        max_quantity: usize,
    ) -> impl Strategy<Value = Result<(Vec<Sequence>, Arc<Keypair>)>> {
        let xorname = XorName::random();
        let tag = 45_000u64;
        let owner_keypair = Arc::new(Keypair::new_ed25519(&mut OsRng));
        let owner = owner_keypair.public_key();
        (1..max_quantity + 1).prop_map(move |quantity| {
            let mut replicas = Vec::with_capacity(quantity);
            for _ in 0..quantity {
                let actor = Keypair::new_ed25519(&mut OsRng).public_key().to_string();

                let replica = Sequence::new_public(owner, actor, xorname, tag);
                replicas.push(replica);
            }

            // set the same owner in all replicas
            let perms = BTreeMap::default();
            let mut owner_op = replicas[0].create_unsigned_public_policy_op(owner, perms)?;
            let bytes = utils::serialise(&owner_op.crdt_op)?;
            let signature = owner_keypair.sign(&bytes);

            owner_op.signature = Some(signature);
            for r in replicas.iter_mut() {
                r.apply_public_policy_op(owner_op.clone())?;
            }
            Ok((replicas, owner_keypair.clone()))
        })
    }

    // Generate a Sequence entry
    fn generate_seq_entry() -> impl Strategy<Value = Vec<u8>> {
        "\\PC*".prop_map(|s| s.into_bytes())
    }

    // Generate a vec of Sequence entries
    fn generate_dataset(max_quantity: usize) -> impl Strategy<Value = Vec<Vec<u8>>> {
        prop::collection::vec(generate_seq_entry(), 1..max_quantity + 1)
    }

    // Generates a vec of Sequence entries each with a value suggesting
    // the delivery chance of the op that gets created with the entry
    fn generate_dataset_and_probability(
        max_quantity: usize,
    ) -> impl Strategy<Value = Vec<(Vec<u8>, u8)>> {
        prop::collection::vec((generate_seq_entry(), any::<u8>()), 1..max_quantity + 1)
    }

    #[derive(Clone)]
    enum OpType {
        Owner(SequencePolicyWriteOp<SequencePublicPolicy>),
        Data(SequenceDataWriteOp<SequenceEntry>),
    }

    proptest! {
        #[test]
        fn proptest_seq_doesnt_crash_with_random_data(
            s in generate_seq_entry()
        ) {
            let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
            let op_source_pk = op_source_pk_keypair.public_key();
            let sequence_name = XorName::random();

            let sdata_tag = 43_001u64;

            // Instantiate the same Sequence on two replicas
            let mut replica1 = Sequence::new_public(op_source_pk, "some_identifying_str".to_string(), sequence_name, sdata_tag);
            let mut replica2 = Sequence::new_public(op_source_pk, "some_identifying_str".to_string(), sequence_name, sdata_tag);

            // Set Actor1 as the owner
            let perms = BTreeMap::default();
            let owner_op = sign_sequence_public_policy_op( replica1.create_unsigned_public_policy_op(op_source_pk, perms)?, &op_source_pk_keypair)?;
            replica1.apply_public_policy_op(owner_op.clone())?;

            replica2.apply_public_policy_op(owner_op)?;

            // Append an item on replicas
            let append_op = sign_sequence_data_op(replica1.create_unsigned_append_op(s)?, &op_source_pk_keypair)?;
            replica1.apply_data_op(append_op.clone())?;

            replica2.apply_data_op(append_op)?;

            verify_data_convergence(vec![replica1, replica2], 1)?;

        }

        #[test]
        fn proptest_seq_converge_with_many_random_data(
            dataset in generate_dataset(1000)
        ) {

            let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
            let op_source_pk = op_source_pk_keypair.public_key();
            let sequence_name = XorName::random();

            let sdata_tag = 43_001u64;

            // Instantiate the same Sequence on two replicas
            let mut replica1 = Sequence::new_public(op_source_pk, "some_identifying_str".to_string(), sequence_name, sdata_tag);
            let mut replica2 = Sequence::new_public(op_source_pk, "some_identifying_str".to_string(), sequence_name, sdata_tag);

            // Set Actor1 as the owner
            let perms = BTreeMap::default();
            let owner_op = sign_sequence_public_policy_op( replica1.create_unsigned_public_policy_op(op_source_pk, perms)?, &op_source_pk_keypair)?;
            replica1.apply_public_policy_op(owner_op.clone())?;
            replica2.apply_public_policy_op(owner_op)?;

            let dataset_length = dataset.len() as u64;

            // insert our data at replicas
            for data in dataset {
                // Append an item on replica1
                let append_op = sign_sequence_data_op( replica1.create_unsigned_append_op(data)?, &op_source_pk_keypair)?;
                replica1.apply_data_op(append_op.clone())?;
                // now apply that op to replica 2
                replica2.apply_data_op(append_op)?;
            }

            verify_data_convergence(vec![replica1, replica2], dataset_length)?;

        }

        #[test]
        fn proptest_seq_converge_with_many_random_data_across_arbitrary_number_of_replicas(
            dataset in generate_dataset(500),
            res in generate_replicas(50)
        ) {
            let (mut replicas, owner_keypair) = res?;
            let dataset_length = dataset.len() as u64;

            // insert our data at replicas
            for data in dataset {
                // first generate an op from one replica...
                let op = sign_sequence_data_op( replicas[0].create_unsigned_append_op(data)?, &owner_keypair )?;

                // then apply this to all replicas
                for replica in &mut replicas {
                    replica.apply_data_op(op.clone())?;
                }
            }

            verify_data_convergence(replicas, dataset_length)?;

        }


        #[test]
        fn proptest_converge_with_shuffled_op_set_across_arbitrary_number_of_replicas(
            dataset in generate_dataset(100),
            res in generate_replicas(500)
        ) {
            let (mut replicas, owner_keypair) = res?;
            let dataset_length = dataset.len() as u64;

            // generate an ops set from one replica
            let mut ops = vec![];

            for data in dataset {
                let op = sign_sequence_data_op( replicas[0].create_unsigned_append_op(data)?, &owner_keypair )?;
                replicas[0].apply_data_op(op.clone())?;
                ops.push(op);
            }

            // now we randomly shuffle ops and apply at each replica
            for replica in &mut replicas {
                let mut ops = ops.clone();
                ops.shuffle(&mut OsRng);

                for op in ops {

                    replica.apply_data_op(op)?;
                }

            }

            verify_data_convergence(replicas, dataset_length)?;

        }



        #[test]
        fn proptest_converge_with_shuffled_ops_from_many_replicas_across_arbitrary_number_of_replicas(
            dataset in generate_dataset(1000),
            res in generate_replicas(100)
        ) {
            let (mut replicas, owner_keypair) = res?;
            let dataset_length = dataset.len() as u64;

            // generate an ops set using random replica for each data
            let mut ops = vec![];
            for data in dataset {
                if let Some(replica) = replicas.choose_mut(&mut OsRng)
                {
                    let op = sign_sequence_data_op( replica.create_unsigned_append_op(data)?, &owner_keypair )?;
                    replica.apply_data_op(op.clone())?;

                    ops.push(op);
                }
            }

            let opslen = ops.len() as u64;
            prop_assert_eq!(dataset_length, opslen);

            // now we randomly shuffle ops and apply at each replica
            for replica in &mut replicas {

                let mut ops = ops.clone();
                ops.shuffle(&mut OsRng);

                for op in ops {
                    replica.apply_data_op(op)?;
                }
            }

            verify_data_convergence(replicas, dataset_length)?;
        }


        #[test]
        fn proptest_we_converge_with_ownership_changes(
            dataset in generate_dataset_and_probability(1000),
        ) {

            let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
            let op_source_pk = op_source_pk_keypair.public_key();
            let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
            let op_source_pk2 = op_source_pk2_keypair.public_key();
            let sequence_name = XorName::random();

            let sdata_tag = 43_001u64;

            // Instantiate the same Sequence on two replicas
            let mut replica1 = Sequence::new_public(op_source_pk, "some_identifying_str".to_string(), sequence_name, sdata_tag);
            let mut replica2 = Sequence::new_public(op_source_pk2, "some_identifying_str2".to_string(), sequence_name, sdata_tag);

            // Set Actor1 as the owner
            let perms = BTreeMap::default();
            let owner_op = sign_sequence_public_policy_op( replica1.create_unsigned_public_policy_op(op_source_pk, perms)?, &op_source_pk_keypair)?;
            replica1.apply_public_policy_op(owner_op.clone())?;

            replica2.apply_public_policy_op(owner_op)?;


            let dataset_length = dataset.len() as u64;

            let mut ops = vec![];
            for (data, policy_change_chance) in dataset {
                let op = sign_sequence_data_op(replica1.create_unsigned_append_op(data)?, &op_source_pk_keypair)? ;
                replica1.apply_data_op(op.clone())?;
                    ops.push(OpType::Data(op));

                    if policy_change_chance < u8::MAX / 3 {
                        let new_owner = Keypair::new_ed25519(&mut OsRng).public_key();

                        let mut perms = BTreeMap::default();
                        let user_perms =
                            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ false);
                        let _ = perms.insert(SequenceUser::Key(op_source_pk), user_perms);
                        if let Ok(mut op) = replica1.create_unsigned_public_policy_op(new_owner, perms) {
                            let bytes = utils::serialise( &op.crdt_op )?;
                            let signature = op_source_pk_keypair.sign(&bytes);
                            op.signature = Some(signature);

                            replica1.apply_public_policy_op(op.clone())?;
                            ops.push(OpType::Owner(op));
                        };


                    }
            }

            let mut suffled_ops = ops.clone();
            suffled_ops.shuffle(&mut OsRng);

            for op in suffled_ops.clone() {
                match op {
                    OpType::Data(op) => {
                        // TODO: current out of order ops fail...
                        // 1. there is no permission check on if the op is allowed
                        // 2. The permission depends on the prev one, which _if it is missing_, we should get back an error stating
                        // missing policy requirements (lazy messaging) to resend those ops.
                        let _ = replica2.apply_data_op(op);
                    },
                    OpType::Owner(op) => {

                        let _ = replica2.apply_public_policy_op(op);
                    },
                }
            }

            // reapply any potentially failed ops
            for op in ops.clone() {

                match op {
                    OpType::Data(op) => {

                        replica2.apply_data_op(op)?;
                    },
                    OpType::Owner(op) => {

                        replica2.apply_public_policy_op(op)?;
                    },
                }
            }

            // now we converge
            verify_data_convergence(vec![replica1, replica2], dataset_length)?;

        }


        #[test]
        #[ignore]
        fn proptest_we_converge_with_ownership_and_perms_changes(
            // 100 min to ensure we get some other ops in shuffle (statistically)
            dataset in prop::collection::vec((generate_seq_entry(), any::<u8>()), 100..1000)
        ) {

            let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
            let op_source_pk = op_source_pk_keypair.public_key();
            let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
            let op_source_pk2 = op_source_pk2_keypair.public_key();
            let sequence_name = XorName::random();

            let sdata_tag = 43_001u64;

            // Instantiate the same Sequence on two replicas
            let mut replica1 = Sequence::new_public(op_source_pk, "some_identifying_str".to_string(), sequence_name, sdata_tag);
            let mut replica2 = Sequence::new_public(op_source_pk2, "some_identifying_str2".to_string(), sequence_name, sdata_tag);

            // Set Actor1 as the owner
            let perms = BTreeMap::default();
            let owner_op = sign_sequence_public_policy_op( replica1.create_unsigned_public_policy_op(op_source_pk, perms)?, &op_source_pk_keypair)?;
            replica1.apply_public_policy_op(owner_op.clone())?;
            replica2.apply_public_policy_op(owner_op)?;

            let dataset_length = dataset.len() as u64;

            let mut ops = vec![];
            for (data, policy_change_chance) in dataset {
                    let op = sign_sequence_data_op(replica1.create_unsigned_append_op(data)?, &op_source_pk_keypair )?;
                    replica1.apply_data_op(op.clone())?;
                    ops.push(OpType::Data(op));
                    let new_owner = Keypair::new_ed25519(&mut OsRng).public_key();

                    if policy_change_chance < u8::MAX / 8 {
                        let mut perms = BTreeMap::default();
                        let user_perms =
                            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ false);
                        let _ = perms.insert(SequenceUser::Key(op_source_pk), user_perms);
                        let owner_op = sign_sequence_public_policy_op( replica1.create_unsigned_public_policy_op(new_owner, perms)?, &op_source_pk_keypair)?;
                        replica1.apply_public_policy_op(owner_op.clone())?;
                        ops.push(OpType::Owner(owner_op));

                    }

                    // randomly disallow writes from our original key
                    if policy_change_chance < u8::MAX / 4 && policy_change_chance > u8::MAX / 8 {
                        let mut perms = BTreeMap::default();
                        let user_perms =
                            SequencePublicPermissions::new(/*append=*/ false, /*admin=*/ false);
                        let _ = perms.insert(SequenceUser::Key(new_owner), user_perms);

                        let owner_op = sign_sequence_public_policy_op( replica1.create_unsigned_public_policy_op(new_owner, perms)?, &op_source_pk_keypair)?;
                        replica1.apply_public_policy_op(owner_op.clone())?;

                        ops.push(OpType::Owner(owner_op));

                    }

                    // randomly allow writes from our original key
                    if policy_change_chance < u8::MAX / 2 && policy_change_chance > u8::MAX / 4 {
                        let mut perms = BTreeMap::default();
                        let user_perms =
                            SequencePublicPermissions::new(/*append=*/ true, /*admin=*/ false);
                        let _ = perms.insert(SequenceUser::Key(op_source_pk), user_perms);
                        let owner_op = sign_sequence_public_policy_op( replica1.create_unsigned_public_policy_op(new_owner, perms)?, &op_source_pk_keypair)?;
                        replica1.apply_public_policy_op(owner_op.clone())?;

                        ops.push(OpType::Owner(owner_op));

                    }
            }
            let mut suffled_ops = ops.clone();
            suffled_ops.shuffle(&mut OsRng);

            for op in suffled_ops.clone() {
                match op {
                    OpType::Data(op) => {
                        // TODO: current out of order ops fail...
                        // 1. there is no permission check on if the op is allowed
                        // 2. The permission depends on the prev one, which _if it is missing_, we should get back an error stating
                        // missing policy requirements (lazy messaging) to resend those ops.
                        let _ = replica2.apply_data_op(op);
                    },
                    OpType::Owner(op) => {
                        // Don't care about failed ops just now due to causality changes... the solution at the
                        // app layer is to request missing ops + reapply them (which we effectively do below)
                        let _ = replica2.apply_public_policy_op(op);
                    },
                }
            }

            // be sure we're missing ops at the moment...
            assert_ne!(replica2.len(None), replica1.len(None));

            // reapply any potentially failed ops
            for op in ops.clone() {
                match op {
                    OpType::Data(op) => {
                        replica2.apply_data_op(op)?;
                    },
                    OpType::Owner(op) => {
                        replica2.apply_public_policy_op(op)?;
                    },
                }
            }

            // now we converge
            verify_data_convergence(vec![replica1, replica2], dataset_length)?;

        }


        #[test]
        fn proptest_dropped_data_can_be_reapplied_and_we_converge(
            dataset in generate_dataset_and_probability(1000),
        ) {

            let op_source_pk_keypair = Keypair::new_ed25519(&mut OsRng);
            let op_source_pk = op_source_pk_keypair.public_key();
            let op_source_pk2_keypair = Keypair::new_ed25519(&mut OsRng);
            let op_source_pk2 = op_source_pk2_keypair.public_key();
            let sequence_name = XorName::random();

            let sdata_tag = 43_001u64;

            // Instantiate the same Sequence on two replicas
            let mut replica1 = Sequence::new_public(op_source_pk, "some_identifying_str".to_string(), sequence_name, sdata_tag);
            let mut replica2 = Sequence::new_public(op_source_pk2, "some_identifying_str2".to_string(), sequence_name, sdata_tag);

            // Set Actor1 as the owner
            let perms = BTreeMap::default();
            let owner_op = sign_sequence_public_policy_op( replica1.create_unsigned_public_policy_op(op_source_pk, perms)?, &op_source_pk_keypair)?;
            replica1.apply_public_policy_op(owner_op.clone())?;

            replica2.apply_public_policy_op(owner_op)?;

            let dataset_length = dataset.len() as u64;

            let mut ops = vec![];
            for (data, delivery_chance) in dataset {
                    let op = sign_sequence_data_op(replica1.create_unsigned_append_op(data)?, &op_source_pk_keypair)?;
                    replica1.apply_data_op(op.clone())?;

                    ops.push((op, delivery_chance));
            }

            for (op, delivery_chance) in ops.clone() {
                if delivery_chance < u8::MAX / 3 {
                    replica2.apply_data_op(op)?;
                }
            }

            // here we statistically should have dropped some messages
            if dataset_length > 50 {
                assert_ne!(replica2.len(None), replica1.len(None));
            }

            // reapply all ops
            for (op, _) in ops {
                replica2.apply_data_op(op)?;
            }

            // now we converge
            verify_data_convergence(vec![replica1, replica2], dataset_length)?;

        }

        #[test]
        fn proptest_converge_with_shuffled_ops_from_many_while_dropping_some_at_random(
            dataset in generate_dataset_and_probability(1000),
            res in generate_replicas(100),
        ) {
            let (mut replicas, owner_keypair) = res?;
            let dataset_length = dataset.len() as u64;

            // generate an ops set using random replica for each data
            let mut ops = vec![];
            for (data, delivery_chance) in dataset {

                // a random index within the replicas range
                let index: usize = OsRng.gen_range( 0, replicas.len());
                let replica = &mut replicas[index];

                let op = sign_sequence_data_op( replica.create_unsigned_append_op(data)?, &owner_keypair)?;
                replica.apply_data_op(op.clone())?;
                ops.push((op, delivery_chance));
            }

            let opslen = ops.len() as u64;
            prop_assert_eq!(dataset_length, opslen);

            // now we randomly shuffle ops and apply at each replica
            for replica in &mut replicas {

                let mut ops = ops.clone();
                ops.shuffle(&mut OsRng);

                for (op, delivery_chance) in ops.clone() {
                    if delivery_chance > u8::MAX / 3 {
                        replica.apply_data_op(op)?;
                    }
                }

                // reapply all ops, simulating lazy messaging filling in the gaps
                for (op, _) in ops {
                    replica.apply_data_op(op)?;
                }
            }

            verify_data_convergence(replicas, dataset_length)?;
        }

        #[test]
        fn proptest_converge_with_shuffled_ops_including_bad_ops_which_error_and_are_not_applied(
            dataset in generate_dataset(10),
            // should be same number as dataset
            bogus_dataset in generate_dataset(10),
            res in generate_replicas(10),

        ) {
            let (mut replicas, owner_keypair) = res?;
            let dataset_length = dataset.len();
            let bogus_dataset_length = bogus_dataset.len();
            let number_replicas = replicas.len();

            // set up a replica that has nothing to do with the rest, random xor... different owner...
            let xorname = XorName::random();
            let tag = 45_000u64;

            // ops are bogus as this is for another random xorname
            let mut bogus_replica = Sequence::new_public(owner_keypair.public_key(), "actor".to_string(), xorname, tag);
            let perms = BTreeMap::default();

            let bogus_op = sign_sequence_public_policy_op(bogus_replica.create_unsigned_public_policy_op(owner_keypair.public_key(), perms)?, &owner_keypair )?;
            bogus_replica.apply_public_policy_op(bogus_op)?;

            // generate the real ops set using random replica for each data
            let mut ops = vec![];
            for data in dataset {
                if let Some(replica) = replicas.choose_mut(&mut OsRng)
                {
                    let op = sign_sequence_data_op(replica.create_unsigned_append_op(data)?, &owner_keypair)?;

                    replica.apply_data_op(op.clone())?;
                    ops.push(op);
                }
            }

            // add bogus ops from bogus replica + bogus data
            for data in bogus_dataset {
                let bogus_op = sign_sequence_data_op( bogus_replica.create_unsigned_append_op(data)?, &owner_keypair)?;
                bogus_replica.apply_data_op(bogus_op.clone())?;
                ops.push(bogus_op);
            }

            let opslen = ops.len();
            prop_assert_eq!(dataset_length + bogus_dataset_length, opslen);

            let mut err_count = vec![];
            // now we randomly shuffle ops and apply at each replica
            for replica in &mut replicas {
                let mut ops = ops.clone();
                ops.shuffle(&mut OsRng);

                for op in ops {
                    match replica.apply_data_op(op) {
                        Ok(_) => {},
                        // record all errors to check this matches bogus data
                        Err(error) => {err_count.push(error)},

                    }
                }
            }

            // check we get an error per bogus datum per replica
            assert_eq!(err_count.len(), bogus_dataset_length * number_replicas);

            verify_data_convergence(replicas, dataset_length as u64)?;

        }
    }
}
