/*
 * zap_cursor_retrieve read from current record.
 * to read bytes we need to call zap_lookup explicitly.
 */
int udmu_zap_cursor_retrieve_value(const struct lu_env *env,
		zap_cursor_t *zc,  char *buf,
		int buf_size, int *bytes_read)
{
	zap_attribute_t *za = &osd_oti_get(env)->oti_za;
	int err, actual_size;

	if ((err = zap_cursor_retrieve(zc, za)))
		return err;

	if (za->za_integer_length <= 0)
		return (ERANGE);

	actual_size = za->za_integer_length * za->za_num_integers;

	if (actual_size > buf_size) {
		actual_size = buf_size;
		buf_size = actual_size / za->za_integer_length;
	} else {
		buf_size = za->za_num_integers;
	}

	err = -zap_lookup(zc->zc_objset, zc->zc_zapobj,
			za->za_name, za->za_integer_length,
			buf_size, buf);

	if (!err)
		*bytes_read = actual_size;

	return err;
}

/**
 * to load a directory entry at a time and stored it in
 * iterator's in-memory data structure.
 *
 * \param di, struct osd_it_ea, iterator's in memory structure
 *
 * \retval +ve, iterator reached to end
 * \retval   0, iterator not reached to end
 * \retval -ve, on error
 */
static int osd_dir_it_next(const struct lu_env *env, struct dt_it *di)
{
	struct osd_zap_it *it = (struct osd_zap_it *)di;
	zap_attribute_t	  *za = &osd_oti_get(env)->oti_za;
	int		   rc;

	/* temp. storage should be enough for any key supported by ZFS */
	CLASSERT(sizeof(za->za_name) <= sizeof(it->ozi_name));

	/*
	 * the first ->next() moves the cursor to .
	 * the second ->next() moves the cursor to ..
	 * then we get to the real records and have to verify any exist
	 */
	if (it->ozi_pos <= 2) {
		it->ozi_pos++;
		if (it->ozi_pos <=2)
			RETURN(0);
	}

	zap_cursor_advance(it->ozi_zc);

	/*
	 * According to current API we need to return error if its last entry.
	 * zap_cursor_advance() does not return any value. So we need to call
	 * retrieve to check if there is any record.  We should make
	 * changes to Iterator API to not return status for this API
	 */
	rc = osd_index_retrieve_skip_dots(it, za);

	if (rc == -ENOENT) /* end of dir */
		RETURN(+1);

	RETURN(rc);
}

static int osd_index_it_load(const struct lu_env *env, const struct dt_it *di,
			     __u64 hash)
{
	struct osd_zap_it *it = (struct osd_zap_it *)di;
	struct osd_object *obj = it->ozi_obj;
	struct osd_device *osd = osd_obj2dev(obj);
	zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
	int                rc;
	ENTRY;

	/* close the current cursor */
	zap_cursor_fini(it->ozi_zc);

	/* create a new one starting at hash */
	memset(it->ozi_zc, 0, sizeof(*it->ozi_zc));
	zap_cursor_init_serialized(it->ozi_zc, osd->od_objset.os,
				   obj->oo_db->db_object, hash);
	it->ozi_reset = 0;

	rc = -zap_cursor_retrieve(it->ozi_zc, za);
	if (rc == 0)
		RETURN(+1);
	else if (rc == -ENOENT)
		RETURN(0);

	RETURN(rc);
}

static int osd_dir_it_key_size(const struct lu_env *env, const struct dt_it *di)
{
	struct osd_zap_it *it = (struct osd_zap_it *)di;
	zap_attribute_t	  *za = &osd_oti_get(env)->oti_za;
	int		   rc;
	ENTRY;

	if (it->ozi_pos <= 1) {
		it->ozi_pos = 1;
		RETURN(2);
	} else if (it->ozi_pos == 2) {
		RETURN(3);
	}

	if ((rc = -zap_cursor_retrieve(it->ozi_zc, za)) == 0)
		rc = strlen(za->za_name);

#if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 3, 90, 0)
	if (rc == 0 && za->za_name[0] == '.') {
		if (za->za_name[1] == 0 || (za->za_name[1] == '.' &&
		    za->za_name[2] == 0)) {
			/* we should not get onto . and ..
			 * stored in the directory. ->next() and
			 * other methods should prevent this
			 */
			LBUG();
		}
	}
#endif
	RETURN(rc);
}

int __osd_obj2dbuf(const struct lu_env *env, objset_t *os,
		   uint64_t oid, dmu_buf_t **dbp)
{
	dmu_object_info_t *doi = &osd_oti_get(env)->oti_doi;
	int rc;

	rc = -sa_buf_hold(os, oid, osd_obj_tag, dbp);
	if (rc)
		return rc;

	dmu_object_info_from_db(*dbp, doi);
	if (unlikely (oid != DMU_USERUSED_OBJECT &&
	    oid != DMU_GROUPUSED_OBJECT && doi->doi_bonus_type != DMU_OT_SA)) {
		sa_buf_rele(*dbp, osd_obj_tag);
		*dbp = NULL;
		return -EINVAL;
	}

	LASSERT(*dbp);
	LASSERT((*dbp)->db_object == oid);
	LASSERT((*dbp)->db_offset == -1);
	LASSERT((*dbp)->db_data != NULL);

	return 0;
}

/*
 * return status :
 *  rc == 0 -> end of directory.
 *  rc >  0 -> ok, proceed.
 *  rc <  0 -> error.  ( EOVERFLOW  can be masked.)
 */
static int osd_dir_it_load(const struct lu_env *env,
			const struct dt_it *di, __u64 hash)
{
	struct osd_zap_it *it = (struct osd_zap_it *)di;
	struct osd_object *obj = it->ozi_obj;
	struct osd_device *osd = osd_obj2dev(obj);
	zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
	int		   rc;
	ENTRY;

	udmu_zap_cursor_fini(it->ozi_zc);
	if (udmu_zap_cursor_init(&it->ozi_zc, &osd->od_objset,
				 obj->oo_db->db_object, hash))
		RETURN(-ENOMEM);

	if (hash <= 2) {
		it->ozi_pos = hash;
		rc = +1;
	} else {
		it->ozi_pos = 3;
		/* to return whether the end has been reached */
		rc = osd_index_retrieve_skip_dots(it, za);
		if (rc == 0)
			rc = +1;
		else if (rc == -ENOENT)
			rc = 0;
	}

	RETURN(rc);
}

static int osd_check_lma(const struct lu_env *env, struct osd_object *obj)
{
	struct osd_thread_info	*info = osd_oti_get(env);
	struct lu_buf		buf;
	int			rc;
	struct lustre_mdt_attrs	*lma;
	ENTRY;

	CLASSERT(sizeof(info->oti_buf) >= sizeof(*lma));
	lma = (struct lustre_mdt_attrs *)info->oti_buf;
	buf.lb_buf = lma;
	buf.lb_len = sizeof(info->oti_buf);

	rc = osd_xattr_get(env, &obj->oo_dt, &buf, XATTR_NAME_LMA);
	if (rc > 0) {
		rc = 0;
		lustre_lma_swab(lma);
		if (unlikely((lma->lma_incompat & ~LMA_INCOMPAT_SUPP) ||
			     CFS_FAIL_CHECK(OBD_FAIL_OSD_LMA_INCOMPAT))) {
			CWARN("%s: unsupported incompat LMA feature(s) %#x for "
			      "fid = "DFID"\n", osd_obj2dev(obj)->od_svname,
			      lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
			      PFID(lu_object_fid(&obj->oo_dt.do_lu)));
			rc = -EOPNOTSUPP;
		}
	} else if (rc == -ENODATA) {
		/* haven't initialize LMA xattr */
		rc = 0;
	}

	RETURN(rc);
}

static int osd_index_insert(const struct lu_env *env, struct dt_object *dt,
			    const struct dt_rec *rec, const struct dt_key *key,
			    struct thandle *th, struct lustre_capa *capa,
			    int ignore_quota)
{
	struct osd_object  *obj = osd_dt_obj(dt);
	struct osd_device  *osd = osd_obj2dev(obj);
	struct osd_thandle *oh;
	__u64		   *k = osd_oti_get(env)->oti_key64;
	int                 rc;
	ENTRY;

	LASSERT(obj->oo_db);
	LASSERT(dt_object_exists(dt));
	LASSERT(osd_invariant(obj));
	LASSERT(th != NULL);

	oh = container_of0(th, struct osd_thandle, ot_super);

	rc = osd_prepare_key_uint64(obj, k, key);

	/* Insert (key,oid) into ZAP */
	rc = -zap_add_uint64(osd->od_objset.os, obj->oo_db->db_object,
			     k, rc, obj->oo_recusize, obj->oo_recsize,
			     (void *)rec, oh->ot_tx);
	RETURN(rc);
}

/**
 * Restore iterator from cookie. if the \a hash isn't found,
 * restore the first valid record.
 *
 * \param  di    - osd iterator
 * \param  hash  - iterator location cookie
 *
 * \retval +ve  - di points to exact matched key
 * \retval  0   - di points to the first valid record
 * \retval -ve  - failure
 */
static int osd_it_acct_load(const struct lu_env *env,
			    const struct dt_it *di, __u64 hash)
{
	struct osd_it_quota	*it  = (struct osd_it_quota *)di;
	struct osd_device	*osd = osd_obj2dev(it->oiq_obj);
	zap_attribute_t		*za = &osd_oti_get(env)->oti_za;
	zap_cursor_t		*zc;
	int			 rc;
	ENTRY;

	/* create new cursor pointing to the new hash */
	rc = osd_zap_cursor_init(&zc, osd->od_os, it->oiq_oid, hash);
	if (rc)
		RETURN(rc);
	osd_zap_cursor_fini(it->oiq_zc);
	it->oiq_zc = zc;
	it->oiq_reset = 0;

	rc = -zap_cursor_retrieve(it->oiq_zc, za);
	if (rc == 0)
		rc = 1;
	else if (rc == -ENOENT)
		rc = 0;

	RETURN(rc);
}

/*
 * zap_cursor_retrieve read from current record.
 * to read bytes we need to call zap_lookup explicitly.
 */
static int osd_zap_cursor_retrieve_value(const struct lu_env *env,
					 zap_cursor_t *zc,  char *buf,
					 int buf_size, int *bytes_read)
{
	zap_attribute_t *za = &osd_oti_get(env)->oti_za;
	int rc, actual_size;

	rc = -zap_cursor_retrieve(zc, za);
	if (unlikely(rc != 0))
		return -rc;

	if (unlikely(za->za_integer_length <= 0))
		return -ERANGE;

	actual_size = za->za_integer_length * za->za_num_integers;

	if (actual_size > buf_size) {
		actual_size = buf_size;
		buf_size = actual_size / za->za_integer_length;
	} else {
		buf_size = za->za_num_integers;
	}

	rc = -zap_lookup(zc->zc_objset, zc->zc_zapobj,
			 za->za_name, za->za_integer_length,
			 buf_size, buf);

	if (likely(rc == 0))
		*bytes_read = actual_size;

	return rc;
}

static struct dt_it *osd_index_it_init(const struct lu_env *env,
				       struct dt_object *dt,
				       __u32 unused,
				       struct lustre_capa *capa)
{
	struct osd_thread_info  *info = osd_oti_get(env);
	struct osd_zap_it       *it;
	struct osd_object       *obj = osd_dt_obj(dt);
	struct osd_device       *osd = osd_obj2dev(obj);
	struct lu_object        *lo  = &dt->do_lu;
	ENTRY;

	/* XXX: check capa ? */

	LASSERT(lu_object_exists(lo));
	LASSERT(obj->oo_db);
	LASSERT(udmu_object_is_zap(obj->oo_db));
	LASSERT(info);

	it = &info->oti_it_zap;

	if (udmu_zap_cursor_init(&it->ozi_zc, &osd->od_objset,
				 obj->oo_db->db_object, 0))
		RETURN(ERR_PTR(-ENOMEM));

	it->ozi_obj   = obj;
	it->ozi_capa  = capa;
	it->ozi_reset = 1;
	lu_object_get(lo);

	RETURN((struct dt_it *)it);
}

static int osd_dir_lookup(const struct lu_env *env, struct dt_object *dt,
			  struct dt_rec *rec, const struct dt_key *key,
			  struct lustre_capa *capa)
{
	struct osd_thread_info *oti = osd_oti_get(env);
	struct osd_object  *obj = osd_dt_obj(dt);
	struct osd_device  *osd = osd_obj2dev(obj);
	char		   *name = (char *)key;
	int                 rc;
	ENTRY;

	LASSERT(udmu_object_is_zap(obj->oo_db));

	if (name[0] == '.') {
		if (name[1] == 0) {
			const struct lu_fid *f = lu_object_fid(&dt->do_lu);
			memcpy(rec, f, sizeof(*f));
			RETURN(1);
		} else if (name[1] == '.' && name[2] == 0) {
			rc = osd_find_parent_fid(env, dt, (struct lu_fid *)rec);
			RETURN(rc == 0 ? 1 : rc);
		}
	}

	rc = -zap_lookup(osd->od_objset.os, obj->oo_db->db_object,
			 (char *)key, 8, sizeof(oti->oti_zde) / 8,
			 (void *)&oti->oti_zde);
	memcpy(rec, &oti->oti_zde.lzd_fid, sizeof(struct lu_fid));

	RETURN(rc == 0 ? 1 : rc);
}

/**
 * Return space usage consumed by a given uid or gid.
 * Block usage is accurrate since it is maintained by DMU itself.
 * However, DMU does not provide inode accounting, so the #inodes in use
 * is estimated from the block usage and statfs information.
 *
 * \param env   - is the environment passed by the caller
 * \param dtobj - is the accounting object
 * \param dtrec - is the record to fill with space usage information
 * \param dtkey - is the id the of the user or group for which we would
 *                like to access disk usage.
 * \param capa - is the capability, not used.
 *
 * \retval +ve - success : exact match
 * \retval -ve - failure
 */
static int osd_acct_index_lookup(const struct lu_env *env,
				struct dt_object *dtobj,
				struct dt_rec *dtrec,
				const struct dt_key *dtkey,
				struct lustre_capa *capa)
{
	struct osd_thread_info	*info = osd_oti_get(env);
	char			*buf  = info->oti_buf;
	struct lquota_acct_rec	*rec  = (struct lquota_acct_rec *)dtrec;
	struct osd_object	*obj = osd_dt_obj(dtobj);
	struct osd_device	*osd = osd_obj2dev(obj);
	int			 rc;
	uint64_t		 oid;
	ENTRY;

	rec->bspace = rec->ispace = 0;

	/* convert the 64-bit uid/gid into a string */
	sprintf(buf, "%llx", *((__u64 *)dtkey));
	/* fetch DMU object ID (DMU_USERUSED_OBJECT/DMU_GROUPUSED_OBJECT) to be
	 * used */
	oid = osd_quota_fid2dmu(lu_object_fid(&dtobj->do_lu));

	/* disk usage (in bytes) is maintained by DMU.
	 * DMU_USERUSED_OBJECT/DMU_GROUPUSED_OBJECT are special objects which
	 * not associated with any dmu_but_t (see dnode_special_open()).
	 * As a consequence, we cannot use udmu_zap_lookup() here since it
	 * requires a valid oo_db. */
	rc = -zap_lookup(osd->od_objset.os, oid, buf, sizeof(uint64_t), 1,
			&rec->bspace);
	if (rc == -ENOENT)
		/* user/group has not created anything yet */
		CDEBUG(D_QUOTA, "%s: id %s not found in DMU accounting ZAP\n",
		       osd->od_svname, buf);
	else if (rc)
		RETURN(rc);

	if (osd->od_quota_iused_est) {
		if (rec->bspace != 0)
			/* estimate #inodes in use */
			rec->ispace = udmu_objset_user_iused(&osd->od_objset,
							     rec->bspace);
		RETURN(+1);
	}

	/* as for inode accounting, it is not maintained by DMU, so we just
	 * use our own ZAP to track inode usage */
	rc = -zap_lookup(osd->od_objset.os, obj->oo_db->db_object,
			 buf, sizeof(uint64_t), 1, &rec->ispace);
	if (rc == -ENOENT)
		/* user/group has not created any file yet */
		CDEBUG(D_QUOTA, "%s: id %s not found in accounting ZAP\n",
		       osd->od_svname, buf);
	else if (rc)
		RETURN(rc);

	RETURN(+1);
}

static int osd_declare_object_destroy(const struct lu_env *env,
				      struct dt_object *dt,
				      struct thandle *th)
{
	char			*buf = osd_oti_get(env)->oti_str;
	const struct lu_fid	*fid = lu_object_fid(&dt->do_lu);
	struct osd_object	*obj = osd_dt_obj(dt);
	struct osd_device	*osd = osd_obj2dev(obj);
	struct osd_thandle	*oh;
	int			 rc;
	uint64_t		 zapid;
	ENTRY;

	LASSERT(th != NULL);
	LASSERT(dt_object_exists(dt));

	oh = container_of0(th, struct osd_thandle, ot_super);
	LASSERT(oh->ot_tx != NULL);

	/* declare that we'll remove object from fid-dnode mapping */
	zapid = osd_get_name_n_idx(env, osd, fid, buf);
	dmu_tx_hold_bonus(oh->ot_tx, zapid);
	dmu_tx_hold_zap(oh->ot_tx, zapid, FALSE, buf);

	osd_declare_xattrs_destroy(env, obj, oh);

	/* declare that we'll remove object from inode accounting ZAPs */
	dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
	dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, FALSE, buf);
	dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
	dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, FALSE, buf);

	/* one less inode */
	rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
			       obj->oo_attr.la_gid, -1, oh, false, NULL, false);
	if (rc)
		RETURN(rc);

	/* data to be truncated */
	rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
			       obj->oo_attr.la_gid, 0, oh, true, NULL, false);
	if (rc)
		RETURN(rc);

	osd_object_set_destroy_type(obj);
	if (obj->oo_destroy == OSD_DESTROY_SYNC)
		dmu_tx_hold_free(oh->ot_tx, obj->oo_db->db_object,
				 0, DMU_OBJECT_END);
	else
		dmu_tx_hold_zap(oh->ot_tx, osd->od_unlinkedid, TRUE, NULL);

	RETURN(0);
}

/**
 * Return pointer to the record under iterator.
 *
 * \param  di    - osd iterator
 * \param  attr  - not used
 */
static int osd_it_acct_rec(const struct lu_env *env,
			   const struct dt_it *di,
			   struct dt_rec *dtrec, __u32 attr)
{
	struct osd_thread_info	*info = osd_oti_get(env);
	char			*buf  = info->oti_buf;
	struct osd_it_quota	*it = (struct osd_it_quota *)di;
	struct lquota_acct_rec	*rec  = (struct lquota_acct_rec *)dtrec;
	struct osd_object	*obj = it->oiq_obj;
	struct osd_device	*osd = osd_obj2dev(obj);
	int			 bytes_read;
	int			 rc;
	ENTRY;

	it->oiq_reset = 0;
	rec->ispace = rec->bspace = 0;

	/* retrieve block usage from the DMU accounting object */
	rc = -udmu_zap_cursor_retrieve_value(env, it->oiq_zc,
					     (char *)&rec->bspace,
					     sizeof(uint64_t), &bytes_read);
	if (rc)
		RETURN(rc);

	if (osd->od_quota_iused_est) {
		if (rec->bspace != 0)
			/* estimate #inodes in use */
			rec->ispace = udmu_objset_user_iused(&osd->od_objset,
							     rec->bspace);
		RETURN(0);
	}

	/* retrieve key associated with the current cursor */
	rc = -udmu_zap_cursor_retrieve_key(env, it->oiq_zc, buf, 32);
	if (rc)
		RETURN(rc);

	/* inode accounting is not maintained by DMU, so we use our own ZAP to
	 * track inode usage */
	rc = -zap_lookup(osd->od_objset.os, it->oiq_obj->oo_db->db_object,
			 buf, sizeof(uint64_t), 1, &rec->ispace);
	if (rc == -ENOENT)
		/* user/group has not created any file yet */
		CDEBUG(D_QUOTA, "%s: id %s not found in accounting ZAP\n",
		       osd->od_svname, buf);
	else if (rc)
		RETURN(rc);

	RETURN(0);
}

static int osd_index_it_rec(const struct lu_env *env, const struct dt_it *di,
			    struct dt_rec *rec, __u32 attr)
{
	zap_attribute_t   *za = &osd_oti_get(env)->oti_za;
	struct osd_zap_it *it = (struct osd_zap_it *)di;
	struct osd_object *obj = it->ozi_obj;
	struct osd_device *osd = osd_obj2dev(obj);
	__u64		  *k = osd_oti_get(env)->oti_key64;
	int                rc;
	ENTRY;

	it->ozi_reset = 0;
	rc = -zap_cursor_retrieve(it->ozi_zc, za);
	if (rc)
		RETURN(rc);

	rc = osd_prepare_key_uint64(obj, k, (const struct dt_key *)za->za_name);

	rc = -zap_lookup_uint64(osd->od_objset.os, obj->oo_db->db_object,
				k, rc, obj->oo_recusize, obj->oo_recsize,
				(void *)rec);
	RETURN(rc);
}

int udmu_zap_cursor_retrieve_key(const struct lu_env *env,
				 zap_cursor_t *zc, char *key, int max)
{
	zap_attribute_t *za = &osd_oti_get(env)->oti_za;
	int		 err;

	if ((err = zap_cursor_retrieve(zc, za)))
		return err;

	if (key)
		strcpy(key, za->za_name);

	return 0;
}

/**
 * Move on to the next valid entry.
 *
 * \param  di   - osd iterator
 *
 * \retval +ve  - iterator reached the end
 * \retval   0  - iterator has not reached the end yet
 * \retval -ve  - unexpected failure
 */
static int osd_it_acct_next(const struct lu_env *env, struct dt_it *di)
{
	struct osd_it_quota	*it = (struct osd_it_quota *)di;
	zap_attribute_t		*za = &osd_oti_get(env)->oti_za;
	int			 rc;
	ENTRY;

	if (it->oiq_reset == 0)
		zap_cursor_advance(it->oiq_zc);
	it->oiq_reset = 0;
	rc = -zap_cursor_retrieve(it->oiq_zc, za);
	if (rc == -ENOENT) /* reached the end */
		rc = 1;
	RETURN(rc);
}

/**
 * Free given iterator.
 *
 * \param  di   - osd iterator
 */
static void osd_it_acct_fini(const struct lu_env *env, struct dt_it *di)
{
	struct osd_thread_info	*info	= osd_oti_get(env);
	struct osd_it_quota	*it	= (struct osd_it_quota *)di;
	ENTRY;

	osd_zap_cursor_fini(it->oiq_zc);
	lu_object_put(env, &it->oiq_obj->oo_dt.do_lu);
	if (it != &info->oti_it_quota)
		OBD_FREE_PTR(it);
	else
		info->oti_it_inline = 0;

	EXIT;
}

/**
 * Return pointer to the key under iterator.
 *
 * \param  di   - osd iterator
 */
static struct dt_key *osd_it_acct_key(const struct lu_env *env,
				      const struct dt_it *di)
{
	struct osd_it_quota	*it = (struct osd_it_quota *)di;
	zap_attribute_t		*za = &osd_oti_get(env)->oti_za;
	int			 rc;
	ENTRY;

	it->oiq_reset = 0;
	rc = -zap_cursor_retrieve(it->oiq_zc, za);
	if (rc)
		RETURN(ERR_PTR(rc));
	rc = kstrtoull(za->za_name, 16, &it->oiq_id);

	RETURN((struct dt_key *) &it->oiq_id);
}