static int replmd_replicated_apply_add(struct replmd_replicated_request *ar)
{
	struct ldb_context *ldb;
	struct ldb_request *change_req;
	enum ndr_err_code ndr_err;
	struct ldb_message *msg;
	struct replPropertyMetaDataBlob *md;
	struct ldb_val md_value;
	unsigned int i;
	int ret;
	bool remote_isDeleted = false;
	bool is_schema_nc;
	NTTIME now;
	time_t t = time(NULL);
	const struct ldb_val *rdn_val;
	struct replmd_private *replmd_private =
		talloc_get_type(ldb_module_get_private(ar->module),
				struct replmd_private);
	unix_to_nt_time(&now, t);

	ldb = ldb_module_get_ctx(ar->module);
	msg = ar->objs->objects[ar->index_current].msg;
	md = ar->objs->objects[ar->index_current].meta_data;
	is_schema_nc = ldb_dn_compare_base(replmd_private->schema_dn, msg->dn) == 0;

	ret = ldb_sequence_number(ldb, LDB_SEQ_NEXT, &ar->seq_num);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	ret = dsdb_msg_add_guid(msg,
				&ar->objs->objects[ar->index_current].object_guid,
				"objectGUID");
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	ret = ldb_msg_add_string(msg, "whenChanged", ar->objs->objects[ar->index_current].when_changed);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	ret = samdb_msg_add_uint64(ldb, msg, msg, "uSNCreated", ar->seq_num);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	ret = samdb_msg_add_uint64(ldb, msg, msg, "uSNChanged", ar->seq_num);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	/* remove any message elements that have zero values */
	for (i=0; i<msg->num_elements; i++) {
		struct ldb_message_element *el = &msg->elements[i];

		if (el->num_values == 0) {
			if (ldb_attr_cmp(msg->elements[i].name, "objectClass") == 0) {
				ldb_asprintf_errstring(ldb, __location__
						       ": empty objectClass sent on %s, aborting replication\n",
						       ldb_dn_get_linearized(msg->dn));
				return replmd_replicated_request_error(ar, LDB_ERR_OBJECT_CLASS_VIOLATION);
			}

			DEBUG(4,(__location__ ": Removing attribute %s with num_values==0\n",
				 el->name));
			ldb_msg_remove_element(msg, &msg->elements[i]);
			i--;
			continue;
		}
	}

	if (DEBUGLVL(8)) {
		struct GUID_txt_buf guid_txt;

		char *s = ldb_ldif_message_redacted_string(ldb, ar,
							   LDB_CHANGETYPE_ADD,
							   msg);
		DEBUG(8, ("DRS replication add message of %s:\n%s\n",
			  GUID_buf_string(&ar->objs->objects[ar->index_current].object_guid, &guid_txt),
			  s));
		talloc_free(s);
	} else if (DEBUGLVL(4)) {
		struct GUID_txt_buf guid_txt;
		DEBUG(4, ("DRS replication add DN of %s is %s\n",
			  GUID_buf_string(&ar->objs->objects[ar->index_current].object_guid, &guid_txt),
			  ldb_dn_get_linearized(msg->dn)));
	}
	remote_isDeleted = ldb_msg_find_attr_as_bool(msg,
						     "isDeleted", false);

	/*
	 * the meta data array is already sorted by the caller, except
	 * for the RDN, which needs to be added.
	 */


	rdn_val = ldb_dn_get_rdn_val(msg->dn);
	ret = replmd_update_rpmd_rdn_attr(ldb, msg, rdn_val, NULL,
					  md, ar, now, is_schema_nc,
					  false);
	if (ret != LDB_SUCCESS) {
		ldb_asprintf_errstring(ldb, "%s: error during DRS repl ADD: %s", __func__, ldb_errstring(ldb));
		return replmd_replicated_request_error(ar, ret);
	}

	ret = replmd_replPropertyMetaDataCtr1_sort_and_verify(ldb, &md->ctr.ctr1, msg->dn);
	if (ret != LDB_SUCCESS) {
		ldb_asprintf_errstring(ldb, "%s: error during DRS repl ADD: %s", __func__, ldb_errstring(ldb));
		return replmd_replicated_request_error(ar, ret);
	}

	for (i=0; i < md->ctr.ctr1.count; i++) {
		md->ctr.ctr1.array[i].local_usn = ar->seq_num;
	}
	ndr_err = ndr_push_struct_blob(&md_value, msg, md,
				       (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
		NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
		return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
	}
	ret = ldb_msg_add_value(msg, "replPropertyMetaData", &md_value, NULL);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	replmd_ldb_message_sort(msg, ar->schema);

	if (!remote_isDeleted) {
		/*
		 * Ensure any local ACL inheritence is applied from
		 * the parent object.
		 *
		 * This is needed because descriptor is above
		 * repl_meta_data in the module stack, so this will
		 * not be trigered 'naturally' by the flow of
		 * operations.
		 */
		ret = dsdb_module_schedule_sd_propagation(ar->module,
							  ar->objs->partition_dn,
							  ar->objs->objects[ar->index_current].object_guid,
							  ar->objs->objects[ar->index_current].parent_guid ?
							  *ar->objs->objects[ar->index_current].parent_guid :
							  GUID_zero(),
							  true);
		if (ret != LDB_SUCCESS) {
			return replmd_replicated_request_error(ar, ret);
		}
	}

	ar->isDeleted = remote_isDeleted;

	ret = ldb_build_add_req(&change_req,
				ldb,
				ar,
				msg,
				ar->controls,
				ar,
				replmd_op_add_callback,
				ar->req);
	LDB_REQ_SET_LOCATION(change_req);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

	/* current partition control needed by "repmd_op_callback" */
	ret = ldb_request_add_control(change_req,
				      DSDB_CONTROL_CURRENT_PARTITION_OID,
				      false, NULL);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

	if (ar->objs->dsdb_repl_flags & DSDB_REPL_FLAG_PARTIAL_REPLICA) {
		/* this tells the partition module to make it a
		   partial replica if creating an NC */
		ret = ldb_request_add_control(change_req,
					      DSDB_CONTROL_PARTIAL_REPLICA,
					      false, NULL);
		if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
	}

	return ldb_next_request(ar->module, change_req);
}