diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/CleanupOperationSummary.test.ts b/client/src/components/User/DiskUsage/Management/Cleanup/CleanupOperationSummary.test.ts
index ea56e1bb0731..dd6ef1ec9e0e 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/CleanupOperationSummary.test.ts
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/CleanupOperationSummary.test.ts
@@ -2,13 +2,18 @@ import { mount } from "@vue/test-utils";
 import flushPromises from "flush-promises";
 import { getLocalVue } from "@tests/jest/helpers";
 import CleanupOperationSummary from "./CleanupOperationSummary.vue";
-import { CleanableSummary, type CleanupOperation, CleanupResult } from "./model";
+import { CleanableSummary, type CleanupOperation, CleanupResult, type CleanableItem } from "./model";
 
 const localVue = getLocalVue();
 
 const REVIEW_ITEMS_LINK = '[data-test-id="review-link"]';
 const NO_ITEMS_INDICATOR = '[data-test-id="no-items-indicator"]';
 
+const EXPECTED_ITEMS: CleanableItem[] = [
+    { id: "1", name: "Item 1", size: 512, type: "dataset", update_time: new Date().toISOString() },
+    { id: "2", name: "Item 2", size: 512, type: "dataset", update_time: new Date().toISOString() },
+];
+
 /** Operation that can clean some items */
 const CLEANUP_OPERATION: CleanupOperation = {
     id: "operation-id",
@@ -16,16 +21,20 @@ const CLEANUP_OPERATION: CleanupOperation = {
     description: "operation description",
     fetchSummary: async () =>
         new CleanableSummary({
-            totalSize: 1024,
-            totalItems: 2,
+            total_size: 1024,
+            total_items: 2,
         }),
     fetchItems: async () => [],
     cleanupItems: async () =>
-        new CleanupResult({
-            totalItemCount: 2,
-            totalFreeBytes: 1024,
-            errors: [],
-        }),
+        new CleanupResult(
+            {
+                total_item_count: 2,
+                success_item_count: 2,
+                total_free_bytes: 1024,
+                errors: [],
+            },
+            EXPECTED_ITEMS
+        ),
 };
 /** Operation without items to clean*/
 const EMPTY_CLEANUP_OPERATION: CleanupOperation = {
@@ -34,8 +43,8 @@ const EMPTY_CLEANUP_OPERATION: CleanupOperation = {
     description: "operation that has no items to clean",
     fetchSummary: async () =>
         new CleanableSummary({
-            totalSize: 0,
-            totalItems: 0,
+            total_size: 0,
+            total_items: 0,
         }),
     fetchItems: async () => [],
     cleanupItems: async () => new CleanupResult(),
diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/CleanupResultDialog.test.ts b/client/src/components/User/DiskUsage/Management/Cleanup/CleanupResultDialog.test.ts
index c556f9e15826..e36462251f83 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/CleanupResultDialog.test.ts
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/CleanupResultDialog.test.ts
@@ -2,7 +2,7 @@ import { mount } from "@vue/test-utils";
 import flushPromises from "flush-promises";
 import { getLocalVue } from "tests/jest/helpers";
 import CleanupResultDialog from "./CleanupResultDialog.vue";
-import { CleanupResult } from "./model";
+import { CleanupResult, type CleanableItem } from "./model";
 
 const localVue = getLocalVue();
 
@@ -14,25 +14,41 @@ const ERRORS_TABLE = '[data-test-id="errors-table"]';
 
 const NO_RESULT_YET = undefined;
 const FAILED_RESULT = () => {
-    return new CleanupResult({
-        errorMessage: "The operation failed",
-        totalFreeBytes: 0,
-        totalItemCount: 0,
-        errors: [],
-    });
+    return new CleanupResult(
+        {
+            total_item_count: 0,
+            errors: [],
+            success_item_count: 0,
+            total_free_bytes: 0,
+        },
+        [],
+        "The operation failed"
+    );
 };
+const TEST_ITEMS: CleanableItem[] = [
+    { id: "1", name: "Dataset X", size: 512, type: "dataset", update_time: new Date().toISOString() },
+    { id: "2", name: "Dataset Y", size: 512, type: "dataset", update_time: new Date().toISOString() },
+    { id: "3", name: "Dataset Z", size: 512, type: "dataset", update_time: new Date().toISOString() },
+];
 const PARTIAL_SUCCESS_RESULT = () => {
-    return new CleanupResult({
-        totalItemCount: 3,
-        totalFreeBytes: 1,
-        errors: [
-            { name: "Dataset X", reason: "Failed because of X" },
-            { name: "Dataset Y", reason: "Failed because of Y" },
-        ],
-    });
+    return new CleanupResult(
+        {
+            total_item_count: 3,
+            success_item_count: 1,
+            total_free_bytes: 512,
+            errors: [
+                { item_id: "1", error: "Failed because of X" },
+                { item_id: "2", error: "Failed because of Y" },
+            ],
+        },
+        TEST_ITEMS
+    );
 };
 const SUCCESS_RESULT = () => {
-    return new CleanupResult({ totalItemCount: 2, totalFreeBytes: 2, errors: [] });
+    return new CleanupResult(
+        { total_item_count: 3, success_item_count: 3, total_free_bytes: 512 * 3, errors: [] },
+        TEST_ITEMS
+    );
 };
 async function mountCleanupResultDialogWith(result?: CleanupResult) {
     const wrapper = mount(CleanupResultDialog, { propsData: { result, show: true }, localVue });
diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/CleanupResultDialog.vue b/client/src/components/User/DiskUsage/Management/Cleanup/CleanupResultDialog.vue
index fe6e1e90844a..ad473d87af35 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/CleanupResultDialog.vue
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/CleanupResultDialog.vue
@@ -2,6 +2,7 @@
 import localize from "@/utils/localization";
 import { computed, ref } from "vue";
 import type { CleanupResult } from "./model";
+import Alert from "components/Alert.vue";
 
 interface CleanupResultDialogProps {
     result?: CleanupResult;
@@ -46,6 +47,9 @@ defineExpose({
 <template>
     <b-modal id="cleanup-result-modal" v-model="showModal" :title="title" title-tag="h2" hide-footer static>
         <div class="text-center">
+            <Alert
+                variant="info"
+                message="After the operation, the storage space that will be freed up will only be for the unique items. This means that some items may not free up any storage space because they are duplicates of other items." />
             <b-spinner v-if="isLoading" class="mx-auto" data-test-id="loading-spinner" />
             <div v-else>
                 <b-alert v-if="result.hasFailed" show variant="danger" data-test-id="error-alert">
diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/ReviewCleanupDialog.test.ts b/client/src/components/User/DiskUsage/Management/Cleanup/ReviewCleanupDialog.test.ts
index 8bcd23f75f24..bdb685b66923 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/ReviewCleanupDialog.test.ts
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/ReviewCleanupDialog.test.ts
@@ -1,7 +1,7 @@
-import { mount, type WrapperArray } from "@vue/test-utils";
+import { mount, type Wrapper, type WrapperArray } from "@vue/test-utils";
 import flushPromises from "flush-promises";
 import { getLocalVue } from "tests/jest/helpers";
-import { CleanableSummary, type CleanupOperation, CleanupResult } from "./model";
+import { CleanableSummary, type CleanupOperation, CleanupResult, type CleanableItem } from "./model";
 import ReviewCleanupDialog from "./ReviewCleanupDialog.vue";
 
 const localVue = getLocalVue();
@@ -12,26 +12,31 @@ const SELECT_ALL_CHECKBOX = '[data-test-id="select-all-checkbox"]';
 const AGREEMENT_CHECKBOX = '[data-test-id="agreement-checkbox"]';
 const CONFIRMATION_MODAL = "#confirmation-modal";
 
-const EXPECTED_TOTAL_ITEMS = 2;
+const EXPECTED_ITEMS: CleanableItem[] = [
+    { id: "1", name: "Item 1", size: 512, type: "dataset", update_time: new Date().toISOString() },
+    { id: "2", name: "Item 2", size: 512, type: "dataset", update_time: new Date().toISOString() },
+];
+const EXPECTED_TOTAL_ITEMS = EXPECTED_ITEMS.length;
 const FAKE_OPERATION: CleanupOperation = {
     id: "operation-id",
     name: "operation name",
     description: "operation description",
     fetchSummary: async () =>
         new CleanableSummary({
-            totalSize: 1024,
-            totalItems: EXPECTED_TOTAL_ITEMS,
+            total_size: 1024,
+            total_items: EXPECTED_TOTAL_ITEMS,
         }),
-    fetchItems: async () => [
-        { id: "1", name: "Item 1", size: 512, update_time: new Date().toISOString(), hda_ldda: "hda" },
-        { id: "2", name: "Item 2", size: 512, update_time: new Date().toISOString(), hda_ldda: "hda" },
-    ],
+    fetchItems: async () => EXPECTED_ITEMS,
     cleanupItems: async () =>
-        new CleanupResult({
-            totalItemCount: EXPECTED_TOTAL_ITEMS,
-            totalFreeBytes: 1024,
-            errors: [],
-        }),
+        new CleanupResult(
+            {
+                total_item_count: EXPECTED_TOTAL_ITEMS,
+                success_item_count: EXPECTED_TOTAL_ITEMS,
+                total_free_bytes: 1024,
+                errors: [],
+            },
+            EXPECTED_ITEMS
+        ),
 };
 
 async function mountReviewCleanupDialogWith(operation: CleanupOperation, totalItems = EXPECTED_TOTAL_ITEMS) {
@@ -43,6 +48,11 @@ async function mountReviewCleanupDialogWith(operation: CleanupOperation, totalIt
     return wrapper;
 }
 
+async function setAllItemsChecked(wrapper: Wrapper<Vue>) {
+    await wrapper.find(SELECT_ALL_CHECKBOX).setChecked();
+    await flushPromises();
+}
+
 describe("ReviewCleanupDialog.vue", () => {
     it("should display a table with items to review", async () => {
         const wrapper = await mountReviewCleanupDialogWith(FAKE_OPERATION);
@@ -54,12 +64,11 @@ describe("ReviewCleanupDialog.vue", () => {
     it("should disable the delete button if no items are selected", async () => {
         const wrapper = await mountReviewCleanupDialogWith(FAKE_OPERATION);
         const deleteButton = wrapper.find(DELETE_BUTTON);
-        const selectAllCheckbox = wrapper.find(SELECT_ALL_CHECKBOX);
 
         expect(deleteButton.attributes().disabled).toBeTruthy();
         // TODO: explicit any because the type of the vm is not correctly inferred, remove when fixed
         expect((wrapper.vm as any).selectedItems.length).toBe(0);
-        await selectAllCheckbox.setChecked();
+        await setAllItemsChecked(wrapper);
         // TODO: explicit any because the type of the vm is not correctly inferred, remove when fixed
         expect((wrapper.vm as any).selectedItems.length).toBe(EXPECTED_TOTAL_ITEMS);
         expect(deleteButton.attributes().disabled).toBeFalsy();
@@ -67,7 +76,7 @@ describe("ReviewCleanupDialog.vue", () => {
 
     it("should show a confirmation message when deleting items", async () => {
         const wrapper = await mountReviewCleanupDialogWith(FAKE_OPERATION);
-        await wrapper.find(SELECT_ALL_CHECKBOX).setChecked();
+        await setAllItemsChecked(wrapper);
         const confirmationModal = wrapper.find(CONFIRMATION_MODAL);
 
         expect(confirmationModal.attributes("aria-hidden")).toBeTruthy();
@@ -77,7 +86,7 @@ describe("ReviewCleanupDialog.vue", () => {
 
     it("should disable the confirmation button until the agreement has been accepted", async () => {
         const wrapper = await mountReviewCleanupDialogWith(FAKE_OPERATION);
-        await wrapper.find(SELECT_ALL_CHECKBOX).setChecked();
+        await setAllItemsChecked(wrapper);
         await wrapper.find(DELETE_BUTTON).trigger("click");
         const allButtons = wrapper.findAll(".btn");
         const permanentlyDeleteBtn = withNameFilter(allButtons).hasText("Permanently delete").at(0);
@@ -89,7 +98,7 @@ describe("ReviewCleanupDialog.vue", () => {
 
     it("should emit the confirmation event when the agreement and deletion has been confirmed", async () => {
         const wrapper = await mountReviewCleanupDialogWith(FAKE_OPERATION);
-        await wrapper.find(SELECT_ALL_CHECKBOX).setChecked();
+        await setAllItemsChecked(wrapper);
         await wrapper.find(DELETE_BUTTON).trigger("click");
         await wrapper.find(AGREEMENT_CHECKBOX).setChecked();
         const allButtons = wrapper.findAll(".btn");
diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/ReviewCleanupDialog.vue b/client/src/components/User/DiskUsage/Management/Cleanup/ReviewCleanupDialog.vue
index 02d3c9c6ac5e..04f058da2d54 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/ReviewCleanupDialog.vue
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/ReviewCleanupDialog.vue
@@ -1,9 +1,9 @@
 <script setup lang="ts">
 import localize from "@/utils/localization";
 import { bytesToString } from "@/utils/utils";
-import { BModal, BTable, BFormCheckbox, BLink, BPagination, BButton } from "bootstrap-vue";
+import { BModal, BTable, BFormCheckbox, BPagination, BButton } from "bootstrap-vue";
 import UtcDate from "@/components/UtcDate.vue";
-import type { CleanableItem, CleanupOperation } from "./model";
+import { type CleanableItem, type CleanupOperation, type SortableKey, PaginationOptions } from "./model";
 import { computed, ref, watch } from "vue";
 
 interface ReviewCleanupDialogProps {
@@ -22,7 +22,7 @@ const emit = defineEmits<{
 }>();
 
 const permanentlyDeleteText = localize("Permanently delete");
-const captionText = localize("To free up account space, review and select items to be permanently deleted or");
+const captionText = localize("To free up account space, review and select items to be permanently deleted here.");
 const agreementText = localize("I understand that once I delete the items, they cannot be recovered.");
 const fields = [
     {
@@ -46,11 +46,10 @@ const fields = [
     },
 ];
 
-type SortableKey = "name" | "size" | "update_time";
+const MAXIMUM_ITEMS_PER_PAGE = 25;
 
-const sortBy = ref(<SortableKey>"size");
+const sortBy = ref<SortableKey>("size");
 const sortDesc = ref(true);
-const perPage = ref(50);
 const currentPage = ref(1);
 const totalRows = ref(1);
 const allSelected = ref(false);
@@ -58,7 +57,6 @@ const indeterminate = ref(false);
 const showDialog = ref(false);
 const items = ref<CleanableItem[]>([]);
 const selectedItems = ref<CleanableItem[]>([]);
-const itemLimit = ref(500);
 const confirmChecked = ref(false);
 const isBusy = ref(false);
 
@@ -71,7 +69,7 @@ const hasItemsSelected = computed(() => {
 });
 
 const hasPages = computed(() => {
-    return totalRows.value > perPage.value;
+    return totalRows.value > MAXIMUM_ITEMS_PER_PAGE;
 });
 
 const title = computed(() => {
@@ -94,16 +92,6 @@ const confirmButtonVariant = computed(() => {
     return confirmChecked.value ? "danger" : "";
 });
 
-const rowLimitReached = computed(() => {
-    return totalRows.value >= itemLimit.value;
-});
-
-const rowLimitReachedText = computed(() => {
-    return localize(
-        `Displaying a maximum of ${itemLimit.value} items here. If there are more, you can rerun this operation after deleting some.`
-    );
-});
-
 watch(props, (newVal) => {
     currentPage.value = 1;
     totalRows.value = newVal.totalItems;
@@ -113,7 +101,7 @@ watch(selectedItems, (newVal) => {
     if (newVal.length === 0) {
         indeterminate.value = false;
         allSelected.value = false;
-    } else if (newVal.length === items.value.length) {
+    } else if (newVal.length === totalRows.value) {
         indeterminate.value = false;
         allSelected.value = true;
     } else {
@@ -126,8 +114,12 @@ function toNiceSize(sizeInBytes: number) {
     return bytesToString(sizeInBytes, true, undefined);
 }
 
-function toggleSelectAll(checked: boolean) {
-    selectedItems.value = checked ? items.value : [];
+async function toggleSelectAll(checked: boolean) {
+    if (checked) {
+        await selectAllItems();
+    } else {
+        unselectAllItems();
+    }
 }
 
 function openModal() {
@@ -143,7 +135,7 @@ function onShowModal() {
 }
 
 function resetModal() {
-    selectedItems.value = [];
+    unselectAllItems();
 }
 
 function resetConfirmationModal() {
@@ -164,12 +156,12 @@ async function itemsProvider(ctx: { currentPage: number; perPage: number }) {
     try {
         const page = ctx.currentPage > 0 ? ctx.currentPage - 1 : 0;
         const offset = page * ctx.perPage;
-        const options = {
+        const options = new PaginationOptions({
             offset: offset,
             limit: ctx.perPage,
             sortBy: sortBy.value,
             sortDesc: sortDesc.value,
-        };
+        });
         const result = await props.operation.fetchItems(options);
         return result;
     } catch (error) {
@@ -177,18 +169,25 @@ async function itemsProvider(ctx: { currentPage: number; perPage: number }) {
     }
 }
 
-async function onSelectAllItems() {
+async function selectAllItems() {
     isBusy.value = true;
-    const allItems = await props.operation.fetchItems({
-        offset: 0,
-        limit: totalRows.value,
-        sortBy: sortBy.value,
-        sortDesc: sortDesc.value,
-    });
+    const allItems = await props.operation.fetchItems(
+        new PaginationOptions({
+            offset: 0,
+            limit: totalRows.value,
+            sortBy: sortBy.value,
+            sortDesc: sortDesc.value,
+        })
+    );
+    items.value = allItems;
     selectedItems.value = allItems;
     isBusy.value = false;
 }
 
+function unselectAllItems() {
+    selectedItems.value = [];
+}
+
 defineExpose({
     openModal,
     selectedItems,
@@ -199,20 +198,17 @@ defineExpose({
     <b-modal v-model="showDialog" title-tag="h2" :static="modalStatic" centered @show="onShowModal">
         <template v-slot:modal-title>
             {{ title }}
-            <span class="text-primary h3">{{ totalItems }}<span v-if="rowLimitReached">+</span> items</span>
+            <span class="text-primary h3">{{ totalRows }} items</span>
         </template>
         <div>
             {{ captionText }}
-            <b>
-                <b-link @click="onSelectAllItems">select all {{ totalItems }} items</b-link>
-            </b>
         </div>
         <b-table
             v-if="operation"
             v-model="items"
             :fields="fields"
             :items="itemsProvider"
-            :per-page="perPage"
+            :per-page="MAXIMUM_ITEMS_PER_PAGE"
             :current-page="currentPage"
             :busy="isBusy"
             hover
@@ -237,8 +233,11 @@ defineExpose({
             </template>
         </b-table>
         <template v-slot:modal-footer>
-            <span v-if="rowLimitReached" class="font-italic">{{ rowLimitReachedText }}</span>
-            <b-pagination v-if="hasPages" v-model="currentPage" :total-rows="totalRows" :per-page="perPage" />
+            <b-pagination
+                v-if="hasPages"
+                v-model="currentPage"
+                :total-rows="totalRows"
+                :per-page="MAXIMUM_ITEMS_PER_PAGE" />
             <b-button
                 v-b-modal.confirmation-modal
                 :disabled="!hasItemsSelected"
diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanableSummary.ts b/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanableSummary.ts
index d61d37084605..c16de15cd1e0 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanableSummary.ts
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanableSummary.ts
@@ -1,32 +1,28 @@
 import { bytesToString } from "@/utils/utils";
+import type { components } from "@/schema";
 
-interface SummaryDataResponse {
-    totalSize: number;
-    totalItems: number;
-}
+type CleanableItemsSummaryResponse = components["schemas"]["CleanableItemsSummary"];
 
 /**
  * Contains summary information about how much storage space can be recovered by removing
  * a collection of items from it.
  */
 export class CleanableSummary {
-    private _data: SummaryDataResponse;
+    private _data: CleanableItemsSummaryResponse;
 
-    constructor(data: SummaryDataResponse) {
+    constructor(data: CleanableItemsSummaryResponse) {
         this._data = data;
     }
 
     /**
      * The total size in bytes that can be recovered by removing all the items.
-     * @returns {Number}
      */
     get totalSize(): number {
-        return this._data.totalSize;
+        return this._data.total_size;
     }
 
     /**
      * The human readable total amount of disk space that can be recovered.
-     * @returns {String}
      */
     get niceTotalSize(): string {
         return bytesToString(this.totalSize, true, undefined);
@@ -36,6 +32,6 @@ export class CleanableSummary {
      * The total number of items that could be removed.
      */
     get totalItems() {
-        return this._data.totalItems;
+        return this._data.total_items;
     }
 }
diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanupOperation.ts b/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanupOperation.ts
index e4b9055a76f2..83aa44b864f4 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanupOperation.ts
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanupOperation.ts
@@ -1,17 +1,27 @@
 import type { CleanableSummary } from "./CleanableSummary";
 import type { CleanupResult } from "./CleanupResult";
+import type { components } from "@/schema";
 
-export interface PaginationOptions {
+export type CleanableItem = components["schemas"]["StoredItem"];
+export type SortableKey = "name" | "size" | "update_time";
+type StoredItemOrderBy = components["schemas"]["StoredItemOrderBy"];
+
+export class PaginationOptions {
     limit?: number;
     offset?: number;
-    sortBy?: string;
+    sortBy?: SortableKey;
     sortDesc?: boolean;
-}
 
-export interface CleanableItem {
-    id: string;
-    name: string;
-    size: number;
+    constructor(props?: { limit?: number; offset?: number; sortBy?: SortableKey; sortDesc?: boolean }) {
+        this.limit = props?.limit;
+        this.offset = props?.offset;
+        this.sortBy = props?.sortBy;
+        this.sortDesc = props?.sortDesc;
+    }
+
+    get order(): StoredItemOrderBy | undefined {
+        return this.sortBy ? `${this.sortBy}${this.sortDesc ? "-dsc" : "-asc"}` : undefined;
+    }
 }
 
 /**
@@ -46,7 +56,7 @@ export interface CleanupOperation {
      * @param options The filter options for sorting and pagination of the items.
      * @returns An array of items that can be potentially `cleaned` and match the filtering params.
      */
-    fetchItems: (options: PaginationOptions) => Promise<CleanableItem[]>;
+    fetchItems: (options?: PaginationOptions) => Promise<CleanableItem[]>;
 
     /**
      * Processes the given items to free up some user storage space and provides a result
diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanupResult.ts b/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanupResult.ts
index 7d7be5dfa3b7..343bbc4fc070 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanupResult.ts
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/model/CleanupResult.ts
@@ -1,47 +1,79 @@
 import { bytesToString } from "@/utils/utils";
+import type { components } from "@/schema";
+import type { CleanableItem } from "./CleanupOperation";
 
+export type StorageItemsCleanupResult = components["schemas"]["StorageItemsCleanupResult"];
+
+/**
+ * Associates the name of an item with the error message when failed.
+ * Simplifies displaying information to the final user.
+ */
 export interface ItemError {
+    /** Name of the item. */
     name: string;
+    /** The reason why it couldn't be cleaned. */
     reason: string;
 }
 
-export interface CleanupResultResponse {
-    totalItemCount: number;
-    totalFreeBytes: number;
-    errors: ItemError[];
-    errorMessage?: string;
-}
-
 /**
- * Contains information about the result of the cleaning operation.
+ * Provides additional information about the result of the cleaning operation.
  */
 export class CleanupResult {
-    private _data: CleanupResultResponse;
+    private _data: StorageItemsCleanupResult;
+    private _item_id_map: Map<string, CleanableItem>;
+    private _errorMessage?: string;
 
-    constructor(data: CleanupResultResponse = { totalFreeBytes: 0, totalItemCount: 0, errors: [] }) {
-        this._data = data;
+    /**
+     * Creates a new CleanupResult.
+     * @param data The server response data for a cleanup operation.
+     * @param items The list of requested items to be cleaned.
+     * @param errorMessage A possible error message if the request completely failed.
+     */
+    constructor(data?: StorageItemsCleanupResult, items?: CleanableItem[], errorMessage?: string) {
+        this._data = data ?? { total_item_count: 0, success_item_count: 0, total_free_bytes: 0, errors: [] };
+        items = items ?? [];
+        this._item_id_map = new Map<string, CleanableItem>();
+        items.forEach((item) => {
+            this._item_id_map.set(item.id, item);
+        });
+        this._errorMessage = errorMessage;
     }
 
+    /**
+     * The total number of items requested to be cleaned.
+     */
     get totalItemCount(): number {
-        return this._data.totalItemCount;
+        return this._data.total_item_count;
     }
 
+    /**
+     * The total amount of space in bytes recovered after trying to cleanup all the requested items.
+     */
     get totalFreeBytes(): number {
-        return this._data.totalFreeBytes;
+        return this._data.total_free_bytes;
     }
 
+    /**
+     * List of error messages associated with a particular item ID.
+     * When an item cannot be cleaned it will be registered in this list.
+     */
     get errors(): ItemError[] {
-        return this._data.errors;
+        return this._data.errors.map((e) => {
+            return { name: this._item_id_map.get(e.item_id)?.name ?? "Unknown", reason: e.error };
+        });
     }
 
+    /**
+     * A general error message, usually indicating that the whole cleanup operation failed with
+     * no partial success.
+     */
     get errorMessage(): string | undefined {
-        return this._data.errorMessage;
+        return this._errorMessage;
     }
 
     /**
-     * Whether the cleanup operation yielded some errors.
-     * It doesn't mean the operation completely failed.
-     * @returns {boolean}
+     * Whether the cleanup operation could not success for every requested item.
+     * It doesn't mean the operation completely failed, some items could have been successfully cleaned.
      */
     get hasSomeErrors(): boolean {
         return this._data.errors.length > 0;
@@ -50,42 +82,44 @@ export class CleanupResult {
     /**
      * Whether the cleanup operation completely failed.
      * This means not even partial cleaning was made.
-     * @returns {boolean}
      */
     get hasFailed(): boolean {
-        return Boolean(this._data.errorMessage);
+        return Boolean(this._errorMessage);
     }
 
     /**
      * Whether the cleanup operation was executed without errors.
-     * @returns {boolean}
      */
     get success(): boolean {
-        return !this.hasSomeErrors && !this._data.errorMessage;
+        return !this.hasSomeErrors && !this.hasFailed;
     }
 
     /**
      * The number of items successfully cleaned.
-     * @returns {number}
      */
     get totalCleaned(): number {
-        return this._data.totalItemCount - this._data.errors.length;
+        return this._data.success_item_count;
     }
 
     /**
-     * Whether the cleanup operation managed to
-     * free some items but not all of them.
-     * @returns {boolean}
+     * Whether the cleanup operation managed to free some items but not all of them.
      */
     get isPartialSuccess(): boolean {
-        return this._data.errors.length > 0 && this.totalCleaned > 0;
+        return this.hasSomeErrors && this._data.success_item_count > 0;
+    }
+
+    /**
+     * Whether the cleanup operation managed to free some space or remove some items.
+     * It can happen that only "copy" items were removed effectively recovering 0 bytes.
+     */
+    get hasUpdatedResults(): boolean {
+        return this.totalFreeBytes > 0 || this.totalCleaned > 0;
     }
 
     /**
      * The total amount of disk space freed by the cleanup operation.
-     * @returns {String}
      */
     get niceTotalFreeBytes(): string {
-        return bytesToString(this._data.totalFreeBytes, true, undefined);
+        return bytesToString(this._data.total_free_bytes, true, undefined);
     }
 }
diff --git a/client/src/components/User/DiskUsage/Management/Cleanup/model/index.ts b/client/src/components/User/DiskUsage/Management/Cleanup/model/index.ts
index 74850c16e6ff..3f5e6b29c099 100644
--- a/client/src/components/User/DiskUsage/Management/Cleanup/model/index.ts
+++ b/client/src/components/User/DiskUsage/Management/Cleanup/model/index.ts
@@ -1,4 +1,4 @@
 export { CleanableSummary } from "./CleanableSummary";
 export type { CleanupCategory } from "./CleanupCategory";
-export type { CleanableItem, CleanupOperation, PaginationOptions } from "./CleanupOperation";
-export { CleanupResult, type CleanupResultResponse, type ItemError } from "./CleanupResult";
+export { type CleanableItem, type CleanupOperation, type SortableKey, PaginationOptions } from "./CleanupOperation";
+export { CleanupResult } from "./CleanupResult";
diff --git a/client/src/components/User/DiskUsage/Management/StorageManager.vue b/client/src/components/User/DiskUsage/Management/StorageManager.vue
index 927363f22236..7fccedc8acbb 100644
--- a/client/src/components/User/DiskUsage/Management/StorageManager.vue
+++ b/client/src/components/User/DiskUsage/Management/StorageManager.vue
@@ -39,7 +39,7 @@ async function onConfirmCleanupSelected(selectedItems: CleanableItem[]) {
     await delay(1000);
     if (currentOperation.value) {
         cleanupResult.value = await currentOperation.value.cleanupItems(selectedItems);
-        if (cleanupResult.value.success) {
+        if (cleanupResult.value.hasUpdatedResults) {
             refreshOperationId.value = currentOperation.value.id.toString();
         }
     }
diff --git a/client/src/components/User/DiskUsage/Management/services.ts b/client/src/components/User/DiskUsage/Management/services.ts
index 94d601487518..698548c546e8 100644
--- a/client/src/components/User/DiskUsage/Management/services.ts
+++ b/client/src/components/User/DiskUsage/Management/services.ts
@@ -1,240 +1,85 @@
-import { getAppRoot } from "@/onload/loadConfig";
 import { rethrowSimple } from "@/utils/simple-error";
-import axios from "axios";
-import {
-    CleanableSummary,
-    CleanupResult,
-    type CleanableItem,
-    type CleanupResultResponse,
-    type ItemError,
-    type PaginationOptions,
-} from "./Cleanup/model";
+import { CleanableSummary, CleanupResult, PaginationOptions, type CleanableItem } from "./Cleanup/model";
+import { fetcher } from "@/schema";
 
-const datasetKeys = "id,name,size,update_time,hda_ldda";
-const isDataset = "q=history_content_type-eq&qv=dataset";
-const isDeleted = "q=deleted-eq&qv=True";
-const isNotPurged = "q=purged-eq&qv=False";
-const maxItemsToFetch = 500;
-const discardedDatasetsQueryParams = `${isDataset}&${isDeleted}&${isNotPurged}&limit=${maxItemsToFetch}`;
+const _fetchDiscardedDatasetsSummary = fetcher.path("/api/storage/datasets/discarded/summary").method("get").create();
 
-const historyKeys = "id,name,size,update_time";
-const discardedHistoriesQueryParams = `&${isDeleted}&${isNotPurged}&limit=${maxItemsToFetch}`;
-
-interface DiscardedDataset extends CleanableItem {
-    hda_ldda: string;
-}
-
-interface DatasetSourceId {
-    id: string;
-    src: string;
-}
-
-interface DatasetErrorMessage {
-    dataset: DatasetSourceId;
-    error_message: string;
-}
-
-interface DeleteDatasetBatchResult {
-    success_count: number;
-    errors?: DatasetErrorMessage[];
-}
-
-type DiscardedHistory = CleanableItem;
-
-/**
- * Calculates the total amount of bytes that can be cleaned by permanently removing
- * deleted datasets.
- * @returns Object containing information about how much can be cleaned.
- */
 export async function fetchDiscardedDatasetsSummary(): Promise<CleanableSummary> {
-    //TODO: possible optimization -> moving this to specific API endpoint so we don't have to parse
-    //      potentially a huge number of items
-    const summaryKeys = "size";
-    const url = `${getAppRoot()}api/datasets?keys=${summaryKeys}&${discardedDatasetsQueryParams}`;
     try {
-        const { data } = await axios.get(url);
-        const totalSizeInBytes = data.reduce(
-            (partial_sum: number, item: DiscardedDataset) => partial_sum + item.size,
-            0
-        );
-        return new CleanableSummary({
-            totalSize: totalSizeInBytes,
-            totalItems: data.length,
-        });
+        const { data } = await _fetchDiscardedDatasetsSummary({});
+        return new CleanableSummary(data);
     } catch (e) {
         rethrowSimple(e);
     }
 }
 
-/**
- * Retrieves all deleted datasets of the current user that haven't been purged yet using pagination.
- * @param options Filtering options for pagination and sorting.
- * @returns Array of dataset objects with the fields defined in `datasetKeys` constant.
- */
-export async function fetchDiscardedDatasets(options: PaginationOptions = {}): Promise<DiscardedDataset[]> {
-    let params = "";
-    if (options.sortBy) {
-        const sortPostfix = options.sortDesc ? "-dsc" : "-asc";
-        params += `order=${options.sortBy}${sortPostfix}&`;
-    }
-    if (options.limit) {
-        params += `limit=${options.limit}&`;
-    }
-    if (options.offset) {
-        params += `offset=${options.offset}&`;
-    }
-    const url = `${getAppRoot()}api/datasets?keys=${datasetKeys}&${discardedDatasetsQueryParams}&${params}`;
-    try {
-        const { data } = await axios.get(url);
-        return data as DiscardedDataset[];
-    } catch (e) {
-        rethrowSimple(e);
-    }
-}
+const _fetchDiscardedDatasets = fetcher.path("/api/storage/datasets/discarded").method("get").create();
 
-/**
- * Purges a collection of datasets.
- * @param datasetSourceIds Array of objects with datasets {id, src} to be purged.
- * @returns Result object with `success_count` and `errors`.
- */
-export async function purgeDatasets(datasetSourceIds: DatasetSourceId[]): Promise<DeleteDatasetBatchResult> {
-    const payload = {
-        purge: true,
-        datasets: datasetSourceIds,
-    };
-    const url = `${getAppRoot()}api/datasets`;
+export async function fetchDiscardedDatasets(options?: PaginationOptions): Promise<CleanableItem[]> {
     try {
-        const { data } = await axios.delete(url, { data: payload });
+        options = options ?? new PaginationOptions();
+        const { data } = await _fetchDiscardedDatasets({
+            offset: options.offset,
+            limit: options.limit,
+            order: options.order,
+        });
         return data;
     } catch (e) {
         rethrowSimple(e);
     }
 }
 
-/**
- * Purges a set of datasets instances (HDA, LDDA, ...) from disk and returns the total space freed in bytes
- * taking into account possible datasets that couldn't be deleted.
- * @param items Array of datasets to be removed from disk.
- *                         Each dataset must contain `id` and `size`.
- * @returns Information about the result of the cleanup operation.
- */
+const _cleanupDiscardedDatasets = fetcher.path("/api/storage/datasets").method("delete").create();
+
 export async function cleanupDiscardedDatasets(items: CleanableItem[]): Promise<CleanupResult> {
-    const resultResponse: CleanupResultResponse = { errors: [], totalFreeBytes: 0, totalItemCount: 0 };
     try {
-        const datasetsTable = items.reduce(
-            (acc: { [key: string]: CleanableItem }, item: CleanableItem) => ((acc[item.id] = item), acc),
-            {}
-        );
-
-        const datasetSourceIds: DatasetSourceId[] = items.map((item: CleanableItem) => {
-            const dataset = item as DiscardedDataset;
-            return { id: dataset.id, src: dataset.hda_ldda };
+        const item_ids = items.map((item) => item.id);
+        const { data } = await _cleanupDiscardedDatasets({
+            item_ids,
         });
-
-        const requestResult = await purgeDatasets(datasetSourceIds);
-
-        resultResponse.totalItemCount = items.length;
-
-        if (requestResult.errors) {
-            resultResponse.errors = mapErrors(datasetsTable, requestResult.errors);
-
-            const erroredIds = requestResult.errors?.reduce((acc: string[], error) => [...acc, error.dataset.id], []);
-
-            resultResponse.totalFreeBytes = datasetSourceIds.reduce((partial_sum, item) => {
-                if (erroredIds?.includes(item.id)) {
-                    return partial_sum;
-                } else {
-                    return partial_sum + (datasetsTable[item.id]?.size ?? 0);
-                }
-            }, 0);
-        }
+        return new CleanupResult(data, items);
     } catch (error) {
-        resultResponse.errorMessage = error as string;
+        return new CleanupResult(undefined, items, error as string);
     }
-    return new CleanupResult(resultResponse);
 }
 
+const _fetchDiscardedHistoriesSummary = fetcher.path("/api/storage/histories/discarded/summary").method("get").create();
+
 export async function fetchDiscardedHistoriesSummary(): Promise<CleanableSummary> {
-    const summaryKeys = "size";
-    const url = `${getAppRoot()}api/histories?keys=${summaryKeys}&${discardedHistoriesQueryParams}`;
     try {
-        const { data } = await axios.get(url);
-        const totalSizeInBytes = data.reduce(
-            (partial_sum: number, item: DiscardedHistory) => partial_sum + item.size,
-            0
-        );
-        return new CleanableSummary({
-            totalSize: totalSizeInBytes,
-            totalItems: data.length,
-        });
+        const { data } = await _fetchDiscardedHistoriesSummary({});
+        return new CleanableSummary(data);
     } catch (e) {
         rethrowSimple(e);
     }
 }
 
-export async function fetchDiscardedHistories(options: PaginationOptions = {}): Promise<DiscardedHistory[]> {
-    let params = "";
-    if (options.sortBy) {
-        const sortPostfix = options.sortDesc ? "-dsc" : "-asc";
-        params += `order=${options.sortBy}${sortPostfix}&`;
-    }
-    if (options.limit) {
-        params += `limit=${options.limit}&`;
-    }
-    if (options.offset) {
-        params += `offset=${options.offset}&`;
-    }
-    const url = `${getAppRoot()}api/histories?keys=${historyKeys}&${discardedHistoriesQueryParams}&${params}`;
-    try {
-        const { data } = await axios.get(url);
-        return data as DiscardedHistory[];
-    } catch (e) {
-        rethrowSimple(e);
-    }
-}
+const _fetchDiscardedHistories = fetcher.path("/api/storage/histories/discarded").method("get").create();
 
-async function purgeHistory(historyId: string) {
-    const payload = {
-        purge: true,
-    };
-    const url = `${getAppRoot()}api/histories/${historyId}`;
+export async function fetchDiscardedHistories(options?: PaginationOptions): Promise<CleanableItem[]> {
     try {
-        const { data } = await axios.delete(url, { data: payload });
+        options = options ?? new PaginationOptions();
+        const { data } = await _fetchDiscardedHistories({
+            offset: options.offset,
+            limit: options.limit,
+            order: options.order,
+        });
         return data;
     } catch (e) {
         rethrowSimple(e);
     }
 }
 
-export async function cleanupDiscardedHistories(histories: DiscardedHistory[]) {
-    const resultResponse: CleanupResultResponse = { errors: [], totalFreeBytes: 0, totalItemCount: 0 };
-    const historiesTable = histories.reduce(
-        (acc: { [key: string]: DiscardedHistory }, item: DiscardedHistory) => ((acc[item.id] = item), acc),
-        {}
-    );
-    // TODO: Promise.all() and do this in parallel?  Or add a bulk delete endpoint?
+const _cleanupDiscardedHistories = fetcher.path("/api/storage/histories").method("delete").create();
+
+export async function cleanupDiscardedHistories(items: CleanableItem[]) {
     try {
-        for (const history of histories) {
-            await purgeHistory(history.id);
-            resultResponse.totalFreeBytes += historiesTable[history.id]?.size ?? 0;
-            resultResponse.totalItemCount += 1;
-        }
+        const item_ids = items.map((item) => item.id);
+        const { data } = await _cleanupDiscardedHistories({
+            item_ids,
+        });
+        return new CleanupResult(data, items);
     } catch (error) {
-        resultResponse.errorMessage = error as string;
+        return new CleanupResult(undefined, items, error as string);
     }
-
-    return new CleanupResult(resultResponse);
-}
-
-/**
- * Maps the error messages with the dataset name for user display.
- * @param datasetsTable Datasets dictionary indexed by ID
- * @param errors List of errors associated with each dataset ID
- * @returns A list with the name of the dataset and the associated error message.
- */
-function mapErrors(datasetsTable: { [key: string]: CleanableItem }, errors: DatasetErrorMessage[]): ItemError[] {
-    return errors.map((error) => {
-        const name = datasetsTable[error.dataset.id]?.name ?? "Unknown Dataset";
-        return { name: name, reason: error.error_message };
-    });
 }
diff --git a/client/src/schema/schema.ts b/client/src/schema/schema.ts
index 955f7fb91184..2b7213671e84 100644
--- a/client/src/schema/schema.ts
+++ b/client/src/schema/schema.ts
@@ -1070,6 +1070,36 @@ export interface paths {
         /** Determine if specified storage request ID is ready for download. */
         get: operations["is_ready_api_short_term_storage__storage_request_id__ready_get"];
     };
+    "/api/storage/datasets": {
+        /**
+         * Purges a set of datasets by ID from disk. The datasets must be owned by the user.
+         * @description **Warning**: This operation cannot be undone. All objects will be deleted permanently from the disk.
+         */
+        delete: operations["cleanup_datasets_api_storage_datasets_delete"];
+    };
+    "/api/storage/datasets/discarded": {
+        /** Returns discarded datasets owned by the given user. The results can be paginated. */
+        get: operations["discarded_datasets_api_storage_datasets_discarded_get"];
+    };
+    "/api/storage/datasets/discarded/summary": {
+        /** Returns information with the total storage space taken by discarded datasets owned by the given user. */
+        get: operations["discarded_datasets_summary_api_storage_datasets_discarded_summary_get"];
+    };
+    "/api/storage/histories": {
+        /**
+         * Purges a set of histories by ID. The histories must be owned by the user.
+         * @description **Warning**: This operation cannot be undone. All objects will be deleted permanently from the disk.
+         */
+        delete: operations["cleanup_histories_api_storage_histories_delete"];
+    };
+    "/api/storage/histories/discarded": {
+        /** Returns all discarded histories associated with the given user. */
+        get: operations["discarded_histories_api_storage_histories_discarded_get"];
+    };
+    "/api/storage/histories/discarded/summary": {
+        /** Returns information with the total storage space taken by discarded histories associated with the given user. */
+        get: operations["discarded_histories_summary_api_storage_histories_discarded_summary_get"];
+    };
     "/api/tags": {
         /**
          * Apply a new set of tags to an item.
@@ -1588,6 +1618,30 @@ export interface components {
              */
             type: string;
         };
+        /**
+         * CleanableItemsSummary
+         * @description Base model definition with common configuration used by all derived models.
+         */
+        CleanableItemsSummary: {
+            /**
+             * Total Items
+             * @description The total number of items that could be purged.
+             */
+            total_items: number;
+            /**
+             * Total Size
+             * @description The total size in bytes that can be recovered by purging all the items.
+             */
+            total_size: number;
+        };
+        /**
+         * CleanupStorageItemsRequest
+         * @description Base model definition with common configuration used by all derived models.
+         */
+        CleanupStorageItemsRequest: {
+            /** Item Ids */
+            item_ids: string[];
+        };
         /**
          * CollectionElementIdentifier
          * @description Base model definition with common configuration used by all derived models.
@@ -6659,6 +6713,33 @@ export interface components {
          * @enum {string}
          */
         Src: "url" | "pasted" | "files" | "path" | "composite" | "ftp_import" | "server_dir";
+        /**
+         * StorageItemCleanupError
+         * @description Base model definition with common configuration used by all derived models.
+         */
+        StorageItemCleanupError: {
+            /** Error */
+            error: string;
+            /**
+             * Item Id
+             * @example 0123456789ABCDEF
+             */
+            item_id: string;
+        };
+        /**
+         * StorageItemsCleanupResult
+         * @description Base model definition with common configuration used by all derived models.
+         */
+        StorageItemsCleanupResult: {
+            /** Errors */
+            errors: components["schemas"]["StorageItemCleanupError"][];
+            /** Success Item Count */
+            success_item_count: number;
+            /** Total Free Bytes */
+            total_free_bytes: number;
+            /** Total Item Count */
+            total_item_count: number;
+        };
         /**
          * StoreExportPayload
          * @description Base model definition with common configuration used by all derived models.
@@ -6688,6 +6769,35 @@ export interface components {
              */
             model_store_format?: components["schemas"]["ModelStoreFormat"];
         };
+        /**
+         * StoredItem
+         * @description Base model definition with common configuration used by all derived models.
+         */
+        StoredItem: {
+            /**
+             * Id
+             * @example 0123456789ABCDEF
+             */
+            id: string;
+            /** Name */
+            name: string;
+            /** Size */
+            size: number;
+            /** Type */
+            type: "history" | "dataset";
+            /**
+             * Update Time
+             * Format: date-time
+             * @description The last time and date this item was updated.
+             */
+            update_time: string;
+        };
+        /**
+         * StoredItemOrderBy
+         * @description Available options for sorting Stored Items results.
+         * @enum {string}
+         */
+        StoredItemOrderBy: "name-asc" | "name-dsc" | "size-asc" | "size-dsc" | "update_time-asc" | "update_time-dsc";
         /**
          * SuitableConverter
          * @description Base model definition with common configuration used by all derived models.
@@ -13375,6 +13485,176 @@ export interface operations {
             };
         };
     };
+    cleanup_datasets_api_storage_datasets_delete: {
+        /**
+         * Purges a set of datasets by ID from disk. The datasets must be owned by the user.
+         * @description **Warning**: This operation cannot be undone. All objects will be deleted permanently from the disk.
+         */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string;
+            };
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["CleanupStorageItemsRequest"];
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["StorageItemsCleanupResult"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    discarded_datasets_api_storage_datasets_discarded_get: {
+        /** Returns discarded datasets owned by the given user. The results can be paginated. */
+        parameters?: {
+            /** @description Starts at the beginning skip the first ( offset - 1 ) items and begin returning at the Nth item */
+            /** @description The maximum number of items to return. */
+            /** @description String containing one of the valid ordering attributes followed by '-asc' or '-dsc' for ascending and descending order respectively. */
+            query?: {
+                offset?: number;
+                limit?: number;
+                order?: components["schemas"]["StoredItemOrderBy"];
+            };
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string;
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["StoredItem"][];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    discarded_datasets_summary_api_storage_datasets_discarded_summary_get: {
+        /** Returns information with the total storage space taken by discarded datasets owned by the given user. */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string;
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["CleanableItemsSummary"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    cleanup_histories_api_storage_histories_delete: {
+        /**
+         * Purges a set of histories by ID. The histories must be owned by the user.
+         * @description **Warning**: This operation cannot be undone. All objects will be deleted permanently from the disk.
+         */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string;
+            };
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["CleanupStorageItemsRequest"];
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["StorageItemsCleanupResult"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    discarded_histories_api_storage_histories_discarded_get: {
+        /** Returns all discarded histories associated with the given user. */
+        parameters?: {
+            /** @description Starts at the beginning skip the first ( offset - 1 ) items and begin returning at the Nth item */
+            /** @description The maximum number of items to return. */
+            /** @description String containing one of the valid ordering attributes followed by '-asc' or '-dsc' for ascending and descending order respectively. */
+            query?: {
+                offset?: number;
+                limit?: number;
+                order?: components["schemas"]["StoredItemOrderBy"];
+            };
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string;
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["StoredItem"][];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    discarded_histories_summary_api_storage_histories_discarded_summary_get: {
+        /** Returns information with the total storage space taken by discarded histories associated with the given user. */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string;
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["CleanableItemsSummary"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     update_api_tags_put: {
         /**
          * Apply a new set of tags to an item.
diff --git a/lib/galaxy/celery/tasks.py b/lib/galaxy/celery/tasks.py
index d452a29dfd80..35b52f308c07 100644
--- a/lib/galaxy/celery/tasks.py
+++ b/lib/galaxy/celery/tasks.py
@@ -43,6 +43,7 @@
     ImportModelStoreTaskRequest,
     MaterializeDatasetInstanceTaskRequest,
     PrepareDatasetCollectionDownload,
+    PurgeDatasetsTaskRequest,
     SetupHistoryExportJob,
     WriteHistoryContentTo,
     WriteHistoryTo,
@@ -84,6 +85,11 @@ def purge_hda(hda_manager: HDAManager, hda_id: int):
     hda_manager._purge(hda)
 
 
+@galaxy_task(ignore_result=True, action="completely removes a set of datasets from the object_store")
+def purge_datasets(dataset_manager: DatasetManager, request: PurgeDatasetsTaskRequest):
+    dataset_manager.purge_datasets(request)
+
+
 @galaxy_task(ignore_result=True, action="materializing dataset instance")
 def materialize(
     hda_manager: HDAManager,
diff --git a/lib/galaxy/managers/base.py b/lib/galaxy/managers/base.py
index f46599fa2175..a0779c4380af 100644
--- a/lib/galaxy/managers/base.py
+++ b/lib/galaxy/managers/base.py
@@ -56,6 +56,12 @@
 from galaxy.model import tool_shed_install
 from galaxy.schema import ValueFilterQueryParams
 from galaxy.schema.fields import DecodedDatabaseIdField
+from galaxy.schema.storage_cleaner import (
+    CleanableItemsSummary,
+    StorageItemsCleanupResult,
+    StoredItem,
+    StoredItemOrderBy,
+)
 from galaxy.security.idencoding import IdEncodingHelper
 from galaxy.structured_app import (
     BasicSharedApp,
@@ -1298,3 +1304,32 @@ def parse_order_by(self, order_by_string, default=None):
         """Return an ORM compatible order_by clause using the given string (i.e.: 'name-dsc,create_time').
         This must be implemented by the manager."""
         raise NotImplementedError
+
+
+class StorageCleanerManager(Protocol):
+    """
+    Interface for monitoring storage usage and managing deletion/purging of objects that consume user's storage space.
+    """
+
+    sort_map: Dict[StoredItemOrderBy, Any]
+
+    def get_discarded_summary(self, user: model.User) -> CleanableItemsSummary:
+        """Returns information with the total storage space taken by discarded items for the given user.
+
+        Discarded items are those that are deleted but not purged yet.
+        """
+        raise NotImplementedError
+
+    def get_discarded(
+        self,
+        user: model.User,
+        offset: Optional[int],
+        limit: Optional[int],
+        order: Optional[StoredItemOrderBy],
+    ) -> List[StoredItem]:
+        """Returns a paginated list of items deleted by the given user that are not yet purged."""
+        raise NotImplementedError
+
+    def cleanup_items(self, user: model.User, item_ids: Set[int]) -> StorageItemsCleanupResult:
+        """Purges the given list of items by ID. The items must be owned by the user."""
+        raise NotImplementedError
diff --git a/lib/galaxy/managers/datasets.py b/lib/galaxy/managers/datasets.py
index 210862879098..0d361a781ae8 100644
--- a/lib/galaxy/managers/datasets.py
+++ b/lib/galaxy/managers/datasets.py
@@ -25,7 +25,10 @@
     secured,
     users,
 )
-from galaxy.schema.tasks import ComputeDatasetHashTaskRequest
+from galaxy.schema.tasks import (
+    ComputeDatasetHashTaskRequest,
+    PurgeDatasetsTaskRequest,
+)
 from galaxy.structured_app import MinimalManagerApp
 from galaxy.util.hash_util import memory_bound_hexdigest
 
@@ -85,6 +88,23 @@ def purge(self, dataset, flush=True):
             self.session().flush()
         return dataset
 
+    def purge_datasets(self, request: PurgeDatasetsTaskRequest):
+        """
+        Caution: any additional security checks must be done before executing this action.
+
+        Completely removes a set of object_store/files associated with the datasets from storage and marks them as purged.
+        They might not be removed if there are still un-purged associations to the dataset.
+        """
+        self.error_unless_dataset_purge_allowed()
+        with self.session().begin():
+            for dataset_id in request.dataset_ids:
+                dataset: model.Dataset = self.session().query(model.Dataset).get(dataset_id)
+                if dataset.user_can_purge:
+                    try:
+                        dataset.full_delete()
+                    except Exception:
+                        log.exception(f"Unable to purge dataset ({dataset.id})")
+
     # TODO: this may be more conv. somewhere else
     # TODO: how to allow admin bypass?
     def error_unless_dataset_purge_allowed(self, msg=None):
diff --git a/lib/galaxy/managers/hdas.py b/lib/galaxy/managers/hdas.py
index 476d9aa16a11..449007d817ed 100644
--- a/lib/galaxy/managers/hdas.py
+++ b/lib/galaxy/managers/hdas.py
@@ -12,8 +12,18 @@
     Dict,
     List,
     Optional,
+    Set,
 )
 
+from sqlalchemy import (
+    and_,
+    asc,
+    desc,
+    false,
+    func,
+    select,
+    true,
+)
 from sqlalchemy.orm.session import object_session
 
 from galaxy import (
@@ -33,8 +43,16 @@
 from galaxy.model.deferred import materializer_factory
 from galaxy.model.tags import GalaxyTagHandler
 from galaxy.schema.schema import DatasetSourceType
+from galaxy.schema.storage_cleaner import (
+    CleanableItemsSummary,
+    StorageItemCleanupError,
+    StorageItemsCleanupResult,
+    StoredItem,
+    StoredItemOrderBy,
+)
 from galaxy.schema.tasks import (
     MaterializeDatasetInstanceTaskRequest,
+    PurgeDatasetsTaskRequest,
     RequestUser,
 )
 from galaxy.structured_app import (
@@ -315,6 +333,117 @@ def _set_permissions(self, trans, hda, role_ids_dict):
                 raise exceptions.RequestParameterInvalidException(error)
 
 
+class HDAStorageCleanerManager(base.StorageCleanerManager):
+    def __init__(self, hda_manager: HDAManager, dataset_manager: datasets.DatasetManager):
+        self.hda_manager = hda_manager
+        self.dataset_manager = dataset_manager
+        self.sort_map = {
+            StoredItemOrderBy.NAME_ASC: asc(model.HistoryDatasetAssociation.name),
+            StoredItemOrderBy.NAME_DSC: desc(model.HistoryDatasetAssociation.name),
+            StoredItemOrderBy.SIZE_ASC: asc(model.Dataset.total_size),
+            StoredItemOrderBy.SIZE_DSC: desc(model.Dataset.total_size),
+            StoredItemOrderBy.UPDATE_TIME_ASC: asc(model.HistoryDatasetAssociation.update_time),
+            StoredItemOrderBy.UPDATE_TIME_DSC: desc(model.HistoryDatasetAssociation.update_time),
+        }
+
+    def get_discarded_summary(self, user: model.User) -> CleanableItemsSummary:
+        stmt = (
+            select([func.sum(model.Dataset.total_size), func.count(model.HistoryDatasetAssociation.id)])
+            .select_from(model.HistoryDatasetAssociation)
+            .join(model.Dataset, model.HistoryDatasetAssociation.table.c.dataset_id == model.Dataset.id)
+            .join(model.History, model.HistoryDatasetAssociation.table.c.history_id == model.History.id)
+            .where(
+                and_(
+                    model.HistoryDatasetAssociation.deleted == true(),
+                    model.HistoryDatasetAssociation.purged == false(),
+                    model.History.user_id == user.id,
+                )
+            )
+        )
+        result = self.hda_manager.session().execute(stmt).fetchone()
+        total_size = 0 if result[0] is None else result[0]
+        return CleanableItemsSummary(total_size=total_size, total_items=result[1])
+
+    def get_discarded(
+        self,
+        user: model.User,
+        offset: Optional[int],
+        limit: Optional[int],
+        order: Optional[StoredItemOrderBy],
+    ) -> List[StoredItem]:
+        stmt = (
+            select(
+                [
+                    model.HistoryDatasetAssociation.id,
+                    model.HistoryDatasetAssociation.name,
+                    model.HistoryDatasetAssociation.update_time,
+                    model.Dataset.total_size,
+                ]
+            )
+            .select_from(model.HistoryDatasetAssociation)
+            .join(model.Dataset, model.HistoryDatasetAssociation.table.c.dataset_id == model.Dataset.id)
+            .join(model.History, model.HistoryDatasetAssociation.table.c.history_id == model.History.id)
+            .where(
+                and_(
+                    model.HistoryDatasetAssociation.deleted == true(),
+                    model.HistoryDatasetAssociation.purged == false(),
+                    model.History.user_id == user.id,
+                )
+            )
+        )
+        if offset:
+            stmt = stmt.offset(offset)
+        if limit:
+            stmt = stmt.limit(limit)
+        if order:
+            stmt = stmt.order_by(self.sort_map[order])
+        result = self.hda_manager.session().execute(stmt)
+        discarded = [
+            StoredItem(id=row.id, name=row.name, type="dataset", size=row.total_size or 0, update_time=row.update_time)
+            for row in result
+        ]
+        return discarded
+
+    def cleanup_items(self, user: model.User, item_ids: Set[int]) -> StorageItemsCleanupResult:
+        success_item_count = 0
+        total_free_bytes = 0
+        errors: List[StorageItemCleanupError] = []
+        dataset_ids_to_remove: Set[int] = set()
+
+        with self.hda_manager.session().begin():
+            for hda_id in item_ids:
+                try:
+                    hda: model.HistoryDatasetAssociation = self.hda_manager.get_owned(hda_id, user)
+                    hda.deleted = True
+                    quota_amount = int(hda.quota_amount(user))
+                    hda.purge_usage_from_quota(user, hda.dataset.quota_source_info)
+                    hda.purged = True
+                    dataset_ids_to_remove.add(hda.dataset.id)
+                    success_item_count += 1
+                    total_free_bytes += quota_amount
+                except BaseException as e:
+                    errors.append(StorageItemCleanupError(item_id=hda_id, error=str(e)))
+
+        self._request_full_delete_all(dataset_ids_to_remove)
+
+        return StorageItemsCleanupResult(
+            total_item_count=len(item_ids),
+            success_item_count=success_item_count,
+            total_free_bytes=total_free_bytes,
+            errors=errors,
+        )
+
+    def _request_full_delete_all(self, dataset_ids_to_remove: Set[int]):
+        use_tasks = self.dataset_manager.app.config.enable_celery_tasks
+        request = PurgeDatasetsTaskRequest(dataset_ids=list(dataset_ids_to_remove))
+        if use_tasks:
+            from galaxy.celery.tasks import purge_datasets
+
+            purge_datasets.delay(request=request)
+        else:
+            self.dataset_manager.purge_datasets(request)
+
+
 class HDASerializer(  # datasets._UnflattenedMetadataDatasetAssociationSerializer,
     datasets.DatasetAssociationSerializer[HDAManager],
     taggable.TaggableSerializerMixin,
diff --git a/lib/galaxy/managers/histories.py b/lib/galaxy/managers/histories.py
index 1c976773f66a..bd66f59367ad 100644
--- a/lib/galaxy/managers/histories.py
+++ b/lib/galaxy/managers/histories.py
@@ -19,6 +19,10 @@
     and_,
     asc,
     desc,
+    false,
+    func,
+    select,
+    true,
 )
 from typing_extensions import Literal
 
@@ -36,6 +40,7 @@
     ModelDeserializingError,
     Serializer,
     SortableManager,
+    StorageCleanerManager,
 )
 from galaxy.managers.export_tracker import StoreExportTracker
 from galaxy.schema.fields import DecodedDatabaseIdField
@@ -45,6 +50,13 @@
     HDABasicInfo,
     ShareHistoryExtra,
 )
+from galaxy.schema.storage_cleaner import (
+    CleanableItemsSummary,
+    StorageItemCleanupError,
+    StorageItemsCleanupResult,
+    StoredItem,
+    StoredItemOrderBy,
+)
 from galaxy.security.validate_user_input import validate_preferred_object_store_id
 from galaxy.structured_app import MinimalManagerApp
 
@@ -355,6 +367,78 @@ def make_members_public(self, trans, item):
                     log.warning(f"User without permissions tried to make dataset with id: {dataset.id} public")
 
 
+class HistoryStorageCleanerManager(StorageCleanerManager):
+    def __init__(self, history_manager: HistoryManager):
+        self.history_manager = history_manager
+        self.sort_map = {
+            StoredItemOrderBy.NAME_ASC: asc(model.History.name),
+            StoredItemOrderBy.NAME_DSC: desc(model.History.name),
+            StoredItemOrderBy.SIZE_ASC: asc(model.History.disk_size),
+            StoredItemOrderBy.SIZE_DSC: desc(model.History.disk_size),
+            StoredItemOrderBy.UPDATE_TIME_ASC: asc(model.History.update_time),
+            StoredItemOrderBy.UPDATE_TIME_DSC: desc(model.History.update_time),
+        }
+
+    def get_discarded_summary(self, user: model.User) -> CleanableItemsSummary:
+        stmt = select([func.sum(model.History.disk_size), func.count(model.History.id)]).where(
+            model.History.user_id == user.id,
+            model.History.deleted == true(),
+            model.History.purged == false(),
+        )
+        result = self.history_manager.session().execute(stmt).fetchone()
+        total_size = 0 if result[0] is None else result[0]
+        return CleanableItemsSummary(total_size=total_size, total_items=result[1])
+
+    def get_discarded(
+        self,
+        user: model.User,
+        offset: Optional[int],
+        limit: Optional[int],
+        order: Optional[StoredItemOrderBy],
+    ) -> List[StoredItem]:
+        stmt = select(model.History).where(
+            model.History.user_id == user.id,
+            model.History.deleted == true(),
+            model.History.purged == false(),
+        )
+        if offset:
+            stmt = stmt.offset(offset)
+        if limit:
+            stmt = stmt.limit(limit)
+        if order:
+            stmt = stmt.order_by(self.sort_map[order])
+        result = self.history_manager.session().execute(stmt).scalars()
+        discarded = [self._history_to_stored_item(item) for item in result]
+        return discarded
+
+    def cleanup_items(self, user: model.User, item_ids: Set[int]) -> StorageItemsCleanupResult:
+        success_item_count = 0
+        total_free_bytes = 0
+        errors: List[StorageItemCleanupError] = []
+
+        with self.history_manager.session().begin():
+            for history_id in item_ids:
+                try:
+                    history = self.history_manager.get_owned(history_id, user)
+                    self.history_manager.purge(history, flush=False)
+                    success_item_count += 1
+                    total_free_bytes += int(history.disk_size)
+                except BaseException as e:
+                    errors.append(StorageItemCleanupError(item_id=history_id, error=str(e)))
+
+        return StorageItemsCleanupResult(
+            total_item_count=len(item_ids),
+            success_item_count=success_item_count,
+            total_free_bytes=total_free_bytes,
+            errors=errors,
+        )
+
+    def _history_to_stored_item(self, history: model.History) -> StoredItem:
+        return StoredItem(
+            id=history.id, name=history.name, type="history", size=history.disk_size, update_time=history.update_time
+        )
+
+
 class HistoryExportManager:
     export_object_type = ExportObjectType.HISTORY
 
diff --git a/lib/galaxy/schema/storage_cleaner.py b/lib/galaxy/schema/storage_cleaner.py
new file mode 100644
index 000000000000..69fb82771cb8
--- /dev/null
+++ b/lib/galaxy/schema/storage_cleaner.py
@@ -0,0 +1,69 @@
+from datetime import datetime
+from enum import Enum
+from typing import (
+    List,
+    Union,
+)
+
+from pydantic import Field
+from typing_extensions import Literal
+
+from galaxy.schema.fields import (
+    DecodedDatabaseIdField,
+    EncodedDatabaseIdField,
+)
+from galaxy.schema.schema import (
+    Model,
+    UpdateTimeField,
+)
+
+
+class CleanableItemsSummary(Model):
+    total_size: int = Field(
+        ...,
+        title="Total Size",
+        description="The total size in bytes that can be recovered by purging all the items.",
+    )
+    total_items: int = Field(
+        ...,
+        title="Total Items",
+        description="The total number of items that could be purged.",
+    )
+
+
+StoredItemType = Union[Literal["history"], Literal["dataset"]]
+
+
+class StoredItem(Model):
+    id: EncodedDatabaseIdField
+    name: str
+    type: StoredItemType
+    size: int
+    update_time: datetime = UpdateTimeField
+
+
+class StoredItemOrderBy(str, Enum):
+    """Available options for sorting Stored Items results."""
+
+    NAME_ASC = "name-asc"
+    NAME_DSC = "name-dsc"
+    SIZE_ASC = "size-asc"
+    SIZE_DSC = "size-dsc"
+    UPDATE_TIME_ASC = "update_time-asc"
+    UPDATE_TIME_DSC = "update_time-dsc"
+
+
+class StorageItemCleanupError(Model):
+    item_id: EncodedDatabaseIdField
+    error: str
+
+
+class CleanupStorageItemsRequest(Model):
+    item_ids: List[DecodedDatabaseIdField]
+
+
+class StorageItemsCleanupResult(Model):
+    total_item_count: int
+    success_item_count: int
+    total_free_bytes: int
+    errors: List[StorageItemCleanupError]
diff --git a/lib/galaxy/schema/tasks.py b/lib/galaxy/schema/tasks.py
index be120cf74b5b..3f8a96aca218 100644
--- a/lib/galaxy/schema/tasks.py
+++ b/lib/galaxy/schema/tasks.py
@@ -1,4 +1,7 @@
-from typing import Optional
+from typing import (
+    List,
+    Optional,
+)
 from uuid import UUID
 
 from pydantic import (
@@ -116,3 +119,7 @@ class ComputeDatasetHashTaskRequest(BaseModel):
     extra_files_path: Optional[str]
     hash_function: HashFunctionNameEnum
     user: Optional[RequestUser]  # access checks should be done pre-celery so this is optional
+
+
+class PurgeDatasetsTaskRequest(BaseModel):
+    dataset_ids: List[int]
diff --git a/lib/galaxy/webapps/galaxy/api/storage_cleaner.py b/lib/galaxy/webapps/galaxy/api/storage_cleaner.py
new file mode 100644
index 000000000000..8bf6d2af1c82
--- /dev/null
+++ b/lib/galaxy/webapps/galaxy/api/storage_cleaner.py
@@ -0,0 +1,121 @@
+"""
+API operations on User storage management.
+"""
+import logging
+from typing import (
+    List,
+    Optional,
+)
+
+from fastapi import (
+    Body,
+    Query,
+)
+
+from galaxy.managers.context import ProvidesHistoryContext
+from galaxy.schema.storage_cleaner import (
+    CleanableItemsSummary,
+    CleanupStorageItemsRequest,
+    StorageItemsCleanupResult,
+    StoredItem,
+    StoredItemOrderBy,
+)
+from galaxy.webapps.galaxy.api import (
+    depends,
+    DependsOnTrans,
+    Router,
+)
+from galaxy.webapps.galaxy.api.common import (
+    LimitQueryParam,
+    OffsetQueryParam,
+)
+from galaxy.webapps.galaxy.services.storage_cleaner import StorageCleanerService
+
+log = logging.getLogger(__name__)
+
+
+router = Router(tags=["storage management"])
+
+OrderQueryParam: Optional[StoredItemOrderBy] = Query(
+    default=None,
+    title="Order",
+    description=(
+        "String containing one of the valid ordering attributes followed "
+        "by '-asc' or '-dsc' for ascending and descending order respectively."
+    ),
+)
+
+
+@router.cbv
+class FastAPIStorageCleaner:
+    service: StorageCleanerService = depends(StorageCleanerService)
+
+    @router.get(
+        "/api/storage/histories/discarded/summary",
+        summary="Returns information with the total storage space taken by discarded histories associated with the given user.",
+    )
+    def discarded_histories_summary(
+        self,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> CleanableItemsSummary:
+        return self.service.get_discarded_summary(trans, stored_item_type="history")
+
+    @router.get(
+        "/api/storage/histories/discarded",
+        summary="Returns all discarded histories associated with the given user.",
+    )
+    def discarded_histories(
+        self,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+        offset: Optional[int] = OffsetQueryParam,
+        limit: Optional[int] = LimitQueryParam,
+        order: Optional[StoredItemOrderBy] = OrderQueryParam,
+    ) -> List[StoredItem]:
+        return self.service.get_discarded(trans, "history", offset, limit, order)
+
+    @router.delete(
+        "/api/storage/histories",
+        summary="Purges a set of histories by ID. The histories must be owned by the user.",
+    )
+    def cleanup_histories(
+        self, trans: ProvidesHistoryContext = DependsOnTrans, payload: CleanupStorageItemsRequest = Body(...)
+    ) -> StorageItemsCleanupResult:
+        """
+        **Warning**: This operation cannot be undone. All objects will be deleted permanently from the disk.
+        """
+        return self.service.cleanup_items(trans, stored_item_type="history", item_ids=set(payload.item_ids))
+
+    @router.get(
+        "/api/storage/datasets/discarded/summary",
+        summary="Returns information with the total storage space taken by discarded datasets owned by the given user.",
+    )
+    def discarded_datasets_summary(
+        self,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+    ) -> CleanableItemsSummary:
+        return self.service.get_discarded_summary(trans, stored_item_type="dataset")
+
+    @router.get(
+        "/api/storage/datasets/discarded",
+        summary="Returns discarded datasets owned by the given user. The results can be paginated.",
+    )
+    def discarded_datasets(
+        self,
+        trans: ProvidesHistoryContext = DependsOnTrans,
+        offset: Optional[int] = OffsetQueryParam,
+        limit: Optional[int] = LimitQueryParam,
+        order: Optional[StoredItemOrderBy] = OrderQueryParam,
+    ) -> List[StoredItem]:
+        return self.service.get_discarded(trans, "dataset", offset, limit, order)
+
+    @router.delete(
+        "/api/storage/datasets",
+        summary="Purges a set of datasets by ID from disk. The datasets must be owned by the user.",
+    )
+    def cleanup_datasets(
+        self, trans: ProvidesHistoryContext = DependsOnTrans, payload: CleanupStorageItemsRequest = Body(...)
+    ) -> StorageItemsCleanupResult:
+        """
+        **Warning**: This operation cannot be undone. All objects will be deleted permanently from the disk.
+        """
+        return self.service.cleanup_items(trans, stored_item_type="dataset", item_ids=set(payload.item_ids))
diff --git a/lib/galaxy/webapps/galaxy/services/base.py b/lib/galaxy/webapps/galaxy/services/base.py
index 2b302c89a910..752458090639 100644
--- a/lib/galaxy/webapps/galaxy/services/base.py
+++ b/lib/galaxy/webapps/galaxy/services/base.py
@@ -44,6 +44,10 @@ def ensure_celery_tasks_enabled(config):
         )
 
 
+class SecurityNotProvidedError(Exception):
+    pass
+
+
 class ServiceBase:
     """Base class with common logic and utils reused by other services.
 
@@ -56,8 +60,16 @@ class ServiceBase:
        the required parameters and outputs of each operation.
     """
 
-    def __init__(self, security: IdEncodingHelper):
-        self.security = security
+    def __init__(self, security: Optional[IdEncodingHelper] = None):
+        self._security = security
+
+    @property
+    def security(self) -> IdEncodingHelper:
+        if self._security is None:
+            raise SecurityNotProvidedError(
+                "Security encoding helper must be set in the service constructor to encode/decode ids."
+            )
+        return self._security
 
     def decode_id(self, id: EncodedDatabaseIdField, kind: Optional[str] = None) -> int:
         """Decodes a previously encoded database ID."""
diff --git a/lib/galaxy/webapps/galaxy/services/storage_cleaner.py b/lib/galaxy/webapps/galaxy/services/storage_cleaner.py
new file mode 100644
index 000000000000..d25b27f11f77
--- /dev/null
+++ b/lib/galaxy/webapps/galaxy/services/storage_cleaner.py
@@ -0,0 +1,56 @@
+import logging
+from typing import (
+    Dict,
+    Optional,
+    Set,
+)
+
+from galaxy.managers.base import StorageCleanerManager
+from galaxy.managers.context import ProvidesHistoryContext
+from galaxy.managers.hdas import HDAStorageCleanerManager
+from galaxy.managers.histories import HistoryStorageCleanerManager
+from galaxy.managers.users import UserManager
+from galaxy.schema.storage_cleaner import (
+    StoredItemOrderBy,
+    StoredItemType,
+)
+from galaxy.webapps.galaxy.services.base import ServiceBase
+
+log = logging.getLogger(__name__)
+
+
+class StorageCleanerService(ServiceBase):
+    """Service providing actions to monitor and recover storage space used by the user."""
+
+    def __init__(
+        self,
+        user_manager: UserManager,
+        history_cleaner: HistoryStorageCleanerManager,
+        hda_cleaner: HDAStorageCleanerManager,
+    ):
+        self.user_manager = user_manager
+        self.history_cleaner = history_cleaner
+        self.hda_cleaner = hda_cleaner
+        self.storage_cleaner_map: Dict[StoredItemType, StorageCleanerManager] = {
+            "history": self.history_cleaner,
+            "dataset": self.hda_cleaner,
+        }
+
+    def get_discarded_summary(self, trans: ProvidesHistoryContext, stored_item_type: StoredItemType):
+        user = self.get_authenticated_user(trans)
+        return self.storage_cleaner_map[stored_item_type].get_discarded_summary(user)
+
+    def get_discarded(
+        self,
+        trans: ProvidesHistoryContext,
+        stored_item_type: StoredItemType,
+        offset: Optional[int] = None,
+        limit: Optional[int] = None,
+        order: Optional[StoredItemOrderBy] = None,
+    ):
+        user = self.get_authenticated_user(trans)
+        return self.storage_cleaner_map[stored_item_type].get_discarded(user, offset, limit, order)
+
+    def cleanup_items(self, trans: ProvidesHistoryContext, stored_item_type: StoredItemType, item_ids: Set[int]):
+        user = self.get_authenticated_user(trans)
+        return self.storage_cleaner_map[stored_item_type].cleanup_items(user, item_ids)
diff --git a/test/integration/test_storage_cleaner.py b/test/integration/test_storage_cleaner.py
new file mode 100644
index 000000000000..37f59f671354
--- /dev/null
+++ b/test/integration/test_storage_cleaner.py
@@ -0,0 +1,156 @@
+from typing import (
+    List,
+    NamedTuple,
+    Optional,
+)
+from uuid import uuid4
+
+from galaxy_test.base.decorators import requires_new_history
+from galaxy_test.base.populators import DatasetPopulator
+from galaxy_test.driver import integration_util
+
+
+class StoredItemDataForTests(NamedTuple):
+    name: str
+    size: int
+
+
+class TestStorageCleaner(integration_util.IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @requires_new_history
+    def test_discarded_histories_monitoring_and_cleanup(self):
+        test_histories = self._build_test_items(resource_name="History")
+        history_ids = self._create_histories_with(test_histories)
+
+        self._assert_monitoring_and_cleanup_for_discarded_resource("histories", test_histories, history_ids)
+
+    @requires_new_history
+    def test_discarded_datasets_monitoring_and_cleanup(self):
+        history_id = self.dataset_populator.new_history(f"History for discarded datasets {uuid4()}")
+        test_datasets = self._build_test_items(resource_name="Dataset")
+        dataset_ids = self._create_datasets_in_history_with(history_id, test_datasets)
+
+        self._assert_monitoring_and_cleanup_for_discarded_resource(
+            "datasets", test_datasets, dataset_ids, delete_resource_uri=f"histories/{history_id}/contents"
+        )
+
+    def _build_test_items(self, resource_name: str):
+        return [
+            StoredItemDataForTests(name=f"Test{resource_name}01_{uuid4()}", size=10),
+            StoredItemDataForTests(name=f"Test{resource_name}02_{uuid4()}", size=25),
+            StoredItemDataForTests(name=f"Test{resource_name}03_{uuid4()}", size=50),
+        ]
+
+    def _assert_monitoring_and_cleanup_for_discarded_resource(
+        self,
+        resource: str,
+        test_items: List[StoredItemDataForTests],
+        item_ids: List[str],
+        delete_resource_uri: Optional[str] = None,
+    ):
+        """Tests the storage cleaner API for a particular resource (histories or datasets)"""
+        delete_resource_uri = delete_resource_uri if delete_resource_uri else resource
+        discarded_storage_items_uri = f"storage/{resource}/discarded"
+
+        # Initially, there shouldn't be any deleted and not purged (discarded) items
+        summary_response = self._get(f"{discarded_storage_items_uri}/summary")
+        self._assert_status_code_is_ok(summary_response)
+        summary = summary_response.json()
+        assert summary["total_items"] == 0
+        assert summary["total_size"] == 0
+
+        # Delete the items
+        for item_id in item_ids:
+            self._delete(f"{delete_resource_uri}/{item_id}", json=True)
+        expected_discarded_item_count = len(test_items)
+        expected_total_size = sum([item.size for item in test_items])
+
+        # All the `test_items` should be in the summary
+        summary_response = self._get(f"{discarded_storage_items_uri}/summary")
+        self._assert_status_code_is_ok(summary_response)
+        summary = summary_response.json()
+        assert summary["total_items"] == expected_discarded_item_count
+        assert summary["total_size"] == expected_total_size
+
+        # Check listing all the discarded items
+        paginated_items_response = self._get(discarded_storage_items_uri)
+        self._assert_status_code_is_ok(paginated_items_response)
+        paginated_items = paginated_items_response.json()
+        assert len(paginated_items) == expected_discarded_item_count
+        assert sum([item["size"] for item in paginated_items]) == expected_total_size
+
+        # Check pagination
+        offset = 1
+        limit = 1
+        paginated_items_response = self._get(f"{discarded_storage_items_uri}?offset={offset}&limit={limit}")
+        self._assert_status_code_is_ok(paginated_items_response)
+        paginated_items = paginated_items_response.json()
+        assert len(paginated_items) == 1
+        assert paginated_items[0]["name"] == test_items[1].name
+
+        # Check listing order by
+        item_names_forward_order = [test_items[0].name, test_items[1].name, test_items[2].name]
+        item_names_reverse_order = list(reversed(item_names_forward_order))
+        expected_order_by_map = {
+            "name-asc": item_names_forward_order,
+            "name-dsc": item_names_reverse_order,
+            "size-asc": item_names_forward_order,
+            "size-dsc": item_names_reverse_order,
+            "update_time-asc": item_names_forward_order,
+            "update_time-dsc": item_names_reverse_order,
+        }
+        for order_by, expected_ordered_names in expected_order_by_map.items():
+            self._assert_order_is_expected(discarded_storage_items_uri, order_by, expected_ordered_names)
+
+        # Cleanup all the items
+        payload = {"item_ids": item_ids}
+        cleanup_response = self._delete(f"storage/{resource}", data=payload, json=True)
+        self._assert_status_code_is_ok(cleanup_response)
+        cleanup_result = cleanup_response.json()
+        assert cleanup_result["total_item_count"] == expected_discarded_item_count
+        assert cleanup_result["success_item_count"] == expected_discarded_item_count
+        assert cleanup_result["total_free_bytes"] == expected_total_size
+        assert not cleanup_result["errors"]
+
+    def _create_histories_with(
+        self, test_histories: List[StoredItemDataForTests], wait_for_histories=True
+    ) -> List[str]:
+        history_ids = []
+        for history_data in test_histories:
+            post_data = dict(name=history_data.name)
+            create_response = self._post("histories", data=post_data).json()
+            self._assert_has_keys(create_response, "name", "id")
+            history_id = create_response["id"]
+            history_ids.append(history_id)
+            # Create a dataset with content equal to the expected size of the history
+            if history_data.size:
+                self.dataset_populator.new_dataset(history_id, content=f"{'0'*(history_data.size-1)}\n")
+        if wait_for_histories:
+            for history_id in history_ids:
+                self.dataset_populator.wait_for_history(history_id)
+        return history_ids
+
+    def _create_datasets_in_history_with(
+        self, history_id: str, test_datasets: List[StoredItemDataForTests], wait_for_history=True
+    ) -> List[str]:
+        dataset_ids = []
+        for dataset_data in test_datasets:
+            dataset = self.dataset_populator.new_dataset(
+                history_id, name=dataset_data.name, content=f"{'0'*(dataset_data.size-1)}\n"
+            )
+            dataset_ids.append(dataset["id"])
+        if wait_for_history:
+            self.dataset_populator.wait_for_history(history_id)
+        return dataset_ids
+
+    def _assert_order_is_expected(self, storage_items_url: str, order_by: str, expected_ordered_names: List[str]):
+        items_response = self._get(f"{storage_items_url}?order={order_by}")
+        self._assert_status_code_is_ok(items_response)
+        items = items_response.json()
+        for index, item in enumerate(items):
+            assert item["name"] == expected_ordered_names[index]