mirror of
https://github.com/rustfs/console.git
synced 2026-10-30 19:33:40 +08:00
refactor: harden task management and TypeScript safety (#162)
This commit is contained in:
@@ -63,9 +63,7 @@ jobs:
|
||||
run: |
|
||||
echo "## 🧪 Running Full Test Suite" >> $GITHUB_STEP_SUMMARY
|
||||
echo "" >> $GITHUB_STEP_SUMMARY
|
||||
if [ -d "tests" ] && [ -n "$(find tests -name '*.test.ts' -o -name '*.spec.ts' 2>/dev/null)" ]; then
|
||||
pnpm run test:run 2>/dev/null || true
|
||||
fi
|
||||
pnpm run test:run
|
||||
echo "✅ All tests passed!" >> $GITHUB_STEP_SUMMARY
|
||||
|
||||
- name: 🔍 Validate code quality
|
||||
|
||||
@@ -73,7 +73,7 @@ jobs:
|
||||
run: pnpm ci
|
||||
|
||||
- name: 🧪 Run unit tests
|
||||
run: pnpm run test:run 2>/dev/null || echo "No test:run script, skipping"
|
||||
run: pnpm run test:run
|
||||
|
||||
- name: 📊 Upload test results
|
||||
uses: actions/upload-artifact@v7
|
||||
|
||||
@@ -184,16 +184,19 @@ export default function AccessKeysPage() {
|
||||
positiveText: t("Confirm"),
|
||||
negativeText: t("Cancel"),
|
||||
onPositiveClick: async () => {
|
||||
try {
|
||||
await Promise.all(selectedKeys.map((key) => deleteServiceAccount(key)))
|
||||
const results = await Promise.allSettled(selectedKeys.map((key) => deleteServiceAccount(key)))
|
||||
const failures = results.filter((result) => result.status === "rejected")
|
||||
|
||||
if (failures.length === 0) {
|
||||
message.success(t("Delete Success"))
|
||||
table.resetRowSelection()
|
||||
await listUserAccounts()
|
||||
} catch (error) {
|
||||
} else {
|
||||
const firstFailure = failures[0]
|
||||
const error = firstFailure?.status === "rejected" ? firstFailure.reason : undefined
|
||||
console.error(error)
|
||||
const msg = (error as Error)?.message || t("Delete Failed")
|
||||
message.error(msg)
|
||||
message.error((error as Error)?.message || t("Delete Failed"))
|
||||
}
|
||||
await listUserAccounts()
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
@@ -199,16 +199,19 @@ export default function UsersPage() {
|
||||
positiveText: t("Confirm"),
|
||||
negativeText: t("Cancel"),
|
||||
onPositiveClick: async () => {
|
||||
try {
|
||||
await Promise.all(selectedKeys.map((key) => deleteUser(key)))
|
||||
const results = await Promise.allSettled(selectedKeys.map((key) => deleteUser(key)))
|
||||
const failures = results.filter((result) => result.status === "rejected")
|
||||
|
||||
if (failures.length === 0) {
|
||||
message.success(t("Delete Success"))
|
||||
table.resetRowSelection()
|
||||
await getDataList()
|
||||
} catch (error) {
|
||||
} else {
|
||||
const firstFailure = failures[0]
|
||||
const error = firstFailure?.status === "rejected" ? firstFailure.reason : undefined
|
||||
console.error(error)
|
||||
const msg = (error as Error)?.message || t("Delete Failed")
|
||||
message.error(msg)
|
||||
message.error((error as Error)?.message || t("Delete Failed"))
|
||||
}
|
||||
await getDataList()
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
@@ -210,6 +210,7 @@ export function LicenseEnterpriseSection() {
|
||||
window.open(
|
||||
"https://ww18.53kf.com/webCompany.php?arg=11003151&kf_sign=DA4MDMTc0Ng4MjE1MjEzODAyNDkyMDAyNzMwMDMxNTE%253D&style=2",
|
||||
"_blank",
|
||||
"noopener,noreferrer",
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -375,6 +375,7 @@ export function ObjectList({
|
||||
try {
|
||||
const url = await getSignedUrl(key)
|
||||
const response = await fetch(url)
|
||||
if (!response.ok) throw new Error(t("Download Failed"))
|
||||
const filename = key.split("/").pop() ?? ""
|
||||
const headers: Record<string, string> = {
|
||||
"content-type": getContentType(response.headers, filename),
|
||||
|
||||
@@ -160,13 +160,24 @@ export function ObjectPreviewModal({ show, onShowChange, object }: ObjectPreview
|
||||
|
||||
React.useEffect(() => {
|
||||
if (show && previewMode === "text" && previewUrl) {
|
||||
const controller = new AbortController()
|
||||
setLoading(true)
|
||||
setIsFormatted(true)
|
||||
fetch(previewUrl)
|
||||
.then((r) => r.text())
|
||||
setTextContent("")
|
||||
fetch(previewUrl, { signal: controller.signal })
|
||||
.then((response) => {
|
||||
if (!response.ok) throw new Error(`HTTP ${response.status}`)
|
||||
return response.text()
|
||||
})
|
||||
.then(setTextContent)
|
||||
.catch(() => setTextContent(t("Preview unavailable")))
|
||||
.finally(() => setLoading(false))
|
||||
.catch((error: unknown) => {
|
||||
if ((error as Error)?.name !== "AbortError") setTextContent(t("Preview unavailable"))
|
||||
})
|
||||
.finally(() => {
|
||||
if (!controller.signal.aborted) setLoading(false)
|
||||
})
|
||||
|
||||
return () => controller.abort()
|
||||
} else if (!show) {
|
||||
setTextContent("")
|
||||
setLoading(false)
|
||||
|
||||
+25
-11
@@ -2,7 +2,7 @@
|
||||
|
||||
import * as React from "react"
|
||||
import { useTranslation } from "react-i18next"
|
||||
import { RiDeleteBinLine } from "@remixicon/react"
|
||||
import { RiCloseCircleLine, RiDeleteBinLine } from "@remixicon/react"
|
||||
import { Button } from "@/components/ui/button"
|
||||
import { Progress } from "@/components/ui/progress"
|
||||
import { useTaskManager } from "@/contexts/task-context"
|
||||
@@ -17,7 +17,6 @@ const statusMap: Record<string, string> = {
|
||||
running: "in progress",
|
||||
completed: "success",
|
||||
failed: "failed",
|
||||
paused: "paused",
|
||||
canceled: "canceled",
|
||||
}
|
||||
|
||||
@@ -27,20 +26,35 @@ export function TaskItem({ task }: TaskItemProps) {
|
||||
|
||||
const statusKey = statusMap[task.status] ?? "in progress"
|
||||
const statusText = `${t(task.actionLabel)}${t(statusKey)}`
|
||||
const isActive = task.status === "pending" || task.status === "running"
|
||||
const isFinished = task.status === "completed" || task.status === "failed" || task.status === "canceled"
|
||||
|
||||
return (
|
||||
<div className="flex flex-col gap-1">
|
||||
<div className="flex items-center justify-between gap-3 group">
|
||||
<div className="min-w-0 flex-1 truncate text-sm font-medium text-foreground">{task.displayName}</div>
|
||||
<Button
|
||||
variant="ghost"
|
||||
size="sm"
|
||||
className="h-auto shrink-0 px-2 text-xs opacity-0 group-hover:opacity-100"
|
||||
aria-label={t("Delete Record")}
|
||||
onClick={() => taskManager.removeTask(task.id)}
|
||||
>
|
||||
<RiDeleteBinLine className="size-4 text-destructive" aria-hidden />
|
||||
</Button>
|
||||
{isActive && (
|
||||
<Button
|
||||
variant="ghost"
|
||||
size="sm"
|
||||
className="h-auto shrink-0 px-2 text-xs opacity-100 sm:opacity-0 sm:group-hover:opacity-100 sm:group-focus-within:opacity-100"
|
||||
aria-label={t("Cancel")}
|
||||
onClick={() => taskManager.cancelTask(task.id)}
|
||||
>
|
||||
<RiCloseCircleLine className="size-4" aria-hidden />
|
||||
</Button>
|
||||
)}
|
||||
{isFinished && (
|
||||
<Button
|
||||
variant="ghost"
|
||||
size="sm"
|
||||
className="h-auto shrink-0 px-2 text-xs opacity-100 sm:opacity-0 sm:group-hover:opacity-100 sm:group-focus-within:opacity-100"
|
||||
aria-label={t("Delete Record")}
|
||||
onClick={() => taskManager.removeFinishedTask(task.id)}
|
||||
>
|
||||
<RiDeleteBinLine className="size-4 text-destructive" aria-hidden />
|
||||
</Button>
|
||||
)}
|
||||
</div>
|
||||
<Progress value={task.progress} className="h-[2px]" />
|
||||
<div className="flex items-center justify-between text-xs text-muted-foreground">
|
||||
|
||||
+24
-16
@@ -10,11 +10,11 @@ import { TaskItem } from "./item"
|
||||
import type { AnyTask } from "@/contexts/task-context"
|
||||
|
||||
interface TaskPanelProps {
|
||||
tasks: AnyTask[]
|
||||
onClearTasks: () => void
|
||||
tasks: readonly AnyTask[]
|
||||
onClearFinishedTasks: () => void
|
||||
}
|
||||
|
||||
export function TaskPanel({ tasks, onClearTasks }: TaskPanelProps) {
|
||||
export function TaskPanel({ tasks, onClearFinishedTasks }: TaskPanelProps) {
|
||||
const { t } = useTranslation()
|
||||
const [tab, setTab] = React.useState("pending")
|
||||
|
||||
@@ -22,27 +22,30 @@ export function TaskPanel({ tasks, onClearTasks }: TaskPanelProps) {
|
||||
const processing = tasks.filter((task) => task.status === "running")
|
||||
const completed = tasks.filter((task) => task.status === "completed")
|
||||
const failed = tasks.filter((task) => task.status === "failed")
|
||||
const canceled = tasks.filter((task) => task.status === "canceled")
|
||||
|
||||
const total = tasks.length
|
||||
const percentage = total === 0 ? 100 : Math.floor((completed.length / total) * 100)
|
||||
const finished = completed.length + failed.length + canceled.length
|
||||
const percentage = total === 0 ? 100 : Math.floor((finished / total) * 100)
|
||||
|
||||
React.useEffect(() => {
|
||||
if (processing.length > 0) setTab("processing")
|
||||
else if (pending.length > 0) setTab("pending")
|
||||
else if (completed.length > 0) setTab("completed")
|
||||
else if (failed.length > 0) setTab("failed")
|
||||
else if (canceled.length > 0) setTab("canceled")
|
||||
else setTab("pending")
|
||||
}, [processing.length, pending.length, completed.length])
|
||||
|
||||
const withCount = (key: string, count: number) => t(key).replace(/\{[^}]*\}/g, String(count))
|
||||
}, [processing.length, pending.length, completed.length, failed.length, canceled.length])
|
||||
|
||||
const tabs = [
|
||||
{ value: "pending", label: withCount("Pending", pending.length) },
|
||||
{ value: "processing", label: withCount("Processing (with count)", processing.length) },
|
||||
{ value: "completed", label: withCount("Completed", completed.length) },
|
||||
{ value: "failed", label: withCount("Failed", failed.length) },
|
||||
{ value: "pending", label: t("Pending", { count: pending.length }) },
|
||||
{ value: "processing", label: t("Processing (with count)", { count: processing.length }) },
|
||||
{ value: "completed", label: t("Completed", { count: completed.length }) },
|
||||
{ value: "failed", label: t("Failed", { count: failed.length }) },
|
||||
{ value: "canceled", label: `${t("Canceled")} (${canceled.length})` },
|
||||
]
|
||||
|
||||
const renderList = (list: AnyTask[]) => (
|
||||
const renderList = (list: readonly AnyTask[]) => (
|
||||
<div className="flex flex-col gap-3">
|
||||
{list.length === 0 ? (
|
||||
<p className="py-4 text-center text-sm text-muted-foreground">{t("No Data")}</p>
|
||||
@@ -69,11 +72,13 @@ export function TaskPanel({ tasks, onClearTasks }: TaskPanelProps) {
|
||||
<div className="space-y-3">
|
||||
<div className="flex items-center justify-between text-sm text-muted-foreground">
|
||||
<div>
|
||||
{total - pending.length - processing.length}/{total}
|
||||
{finished}/{total}
|
||||
</div>
|
||||
<Button variant="ghost" size="sm" className="h-auto px-2" onClick={onClearTasks}>
|
||||
{t("Clear Records")}
|
||||
</Button>
|
||||
{finished > 0 && (
|
||||
<Button variant="ghost" size="sm" className="h-auto px-2" onClick={onClearFinishedTasks}>
|
||||
{t("Clear Records")}
|
||||
</Button>
|
||||
)}
|
||||
</div>
|
||||
<Progress value={percentage} className="h-[3px]" />
|
||||
</div>
|
||||
@@ -100,6 +105,9 @@ export function TaskPanel({ tasks, onClearTasks }: TaskPanelProps) {
|
||||
<TabsContent value="failed" className="mt-0">
|
||||
{renderList(failed)}
|
||||
</TabsContent>
|
||||
<TabsContent value="canceled" className="mt-0">
|
||||
{renderList(canceled)}
|
||||
</TabsContent>
|
||||
</Tabs>
|
||||
</div>
|
||||
)
|
||||
|
||||
@@ -15,9 +15,11 @@ export function TaskStatsButton() {
|
||||
const taskManager = useTaskManager()
|
||||
const { isTaskPanelOpen, setTaskPanelOpen } = useTaskPanelOpen()
|
||||
|
||||
const pending = tasks.filter((task) => task.status === "pending")
|
||||
const processing = tasks.filter((task) => task.status === "running")
|
||||
const completed = tasks.filter((task) => task.status === "completed")
|
||||
const failed = tasks.filter((task) => task.status === "failed")
|
||||
const active = pending.length + processing.length
|
||||
const total = tasks.length
|
||||
|
||||
if (total === 0) return null
|
||||
@@ -25,7 +27,7 @@ export function TaskStatsButton() {
|
||||
return (
|
||||
<Drawer open={isTaskPanelOpen} onOpenChange={setTaskPanelOpen} swipeDirection="right">
|
||||
<DrawerTrigger render={<Button variant="outline" />}>
|
||||
{processing.length > 0 ? (
|
||||
{active > 0 ? (
|
||||
<div className="flex items-center gap-2">
|
||||
<Spinner className="size-3 text-muted-foreground" />
|
||||
<span>
|
||||
@@ -48,7 +50,7 @@ export function TaskStatsButton() {
|
||||
<DrawerTitle>{t("Task Management")}</DrawerTitle>
|
||||
</DrawerHeader>
|
||||
<div className="overflow-auto p-4">
|
||||
<TaskPanel tasks={tasks} onClearTasks={() => taskManager.clearTasks()} />
|
||||
<TaskPanel tasks={tasks} onClearFinishedTasks={() => taskManager.clearFinishedTasks()} />
|
||||
</div>
|
||||
</DrawerContent>
|
||||
</Drawer>
|
||||
|
||||
+34
-44
@@ -2,45 +2,27 @@
|
||||
|
||||
import * as React from "react"
|
||||
import { useSyncExternalStore } from "react"
|
||||
import { TaskManager, type TaskHandler } from "@/lib/task-manager"
|
||||
import { createUploadTaskHelpers } from "@/lib/upload-task"
|
||||
import { createDeleteTaskHelpers } from "@/lib/delete-task"
|
||||
import { TaskManager, type TaskManagerApi } from "@/lib/task-manager"
|
||||
import { createUploadTaskHelpers, type UploadTask } from "@/lib/upload-task"
|
||||
import { createDeleteTaskHelpers, type AnyDeleteTask } from "@/lib/delete-task"
|
||||
import { useS3Optional } from "@/contexts/s3-context"
|
||||
|
||||
export type UploadTask = {
|
||||
id: string
|
||||
kind: "upload"
|
||||
status: string
|
||||
progress: number
|
||||
error?: string
|
||||
displayName: string
|
||||
subInfo: string
|
||||
actionLabel: string
|
||||
}
|
||||
export type AnyTask = UploadTask | AnyDeleteTask
|
||||
|
||||
export type AnyTask = {
|
||||
id: string
|
||||
kind: string
|
||||
status: string
|
||||
progress: number
|
||||
error?: string
|
||||
displayName: string
|
||||
subInfo: string
|
||||
actionLabel: string
|
||||
bucketName?: string
|
||||
}
|
||||
|
||||
const emptyTasks: AnyTask[] = []
|
||||
const emptyManager = {
|
||||
const emptyTasks: readonly AnyTask[] = []
|
||||
const emptyManager: TaskManagerApi<AnyTask> = {
|
||||
getTasks: () => emptyTasks,
|
||||
enqueue: () => {},
|
||||
removeTask: () => {},
|
||||
clearTasks: () => {},
|
||||
cancelTask: () => false,
|
||||
cancelAll: () => {},
|
||||
removeFinishedTask: () => false,
|
||||
clearFinishedTasks: () => {},
|
||||
dispose: () => {},
|
||||
subscribe: () => () => {},
|
||||
} as unknown as TaskManager<AnyTask, string>
|
||||
}
|
||||
|
||||
const TaskContext = React.createContext<{
|
||||
taskManager: TaskManager<AnyTask, string>
|
||||
taskManager: TaskManagerApi<AnyTask>
|
||||
addUploadFiles: (items: { file: File; key: string }[], bucketName: string) => void
|
||||
addDeleteKeys: (keys: string[], bucketName: string, prefix?: string, options?: { forceDelete?: boolean }) => void
|
||||
addDeleteFolder: (prefix: string, bucketName: string, options?: { forceDelete?: boolean }) => void
|
||||
@@ -56,7 +38,7 @@ const TaskContext = React.createContext<{
|
||||
})
|
||||
|
||||
type ManagerState = {
|
||||
manager: TaskManager<AnyTask, string>
|
||||
manager: TaskManager<AnyTask>
|
||||
uploadHelpers: ReturnType<typeof createUploadTaskHelpers>
|
||||
deleteHelpers: ReturnType<typeof createDeleteTaskHelpers>
|
||||
} | null
|
||||
@@ -67,7 +49,10 @@ export function TaskProvider({ children }: { children: React.ReactNode }) {
|
||||
const [managerState, setManagerState] = React.useState<ManagerState>(null)
|
||||
|
||||
React.useEffect(() => {
|
||||
if (!client) return
|
||||
if (!client) {
|
||||
setManagerState(null)
|
||||
return
|
||||
}
|
||||
const uploadHelpers = createUploadTaskHelpers(client, {
|
||||
chunkSize: 16,
|
||||
maxRetries: 3,
|
||||
@@ -77,17 +62,19 @@ export function TaskProvider({ children }: { children: React.ReactNode }) {
|
||||
maxRetries: 3,
|
||||
retryDelay: 1000,
|
||||
})
|
||||
const manager = new TaskManager<AnyTask, string>({
|
||||
const manager = new TaskManager<AnyTask>({
|
||||
handlers: {
|
||||
upload: uploadHelpers.handler as TaskHandler<unknown, string>,
|
||||
delete: deleteHelpers.handler as TaskHandler<unknown, string>,
|
||||
"delete-folder": deleteHelpers.folderHandler as TaskHandler<unknown, string>,
|
||||
upload: uploadHelpers.handler,
|
||||
delete: deleteHelpers.handler,
|
||||
"delete-folder": deleteHelpers.folderHandler,
|
||||
},
|
||||
maxConcurrent: 6,
|
||||
maxRetries: 3,
|
||||
retryDelay: 1000,
|
||||
})
|
||||
setManagerState({ manager, uploadHelpers, deleteHelpers })
|
||||
|
||||
return () => manager.dispose()
|
||||
}, [client])
|
||||
|
||||
const taskManager = managerState?.manager ?? emptyManager
|
||||
@@ -96,16 +83,19 @@ export function TaskProvider({ children }: { children: React.ReactNode }) {
|
||||
(items: { file: File; key: string }[], bucketName: string) => {
|
||||
if (!managerState) return
|
||||
const { manager, uploadHelpers } = managerState
|
||||
const allTasks = manager.getTasks() as Array<{ status: string; bucketName?: string; key?: string }>
|
||||
const allTasks = manager.getTasks()
|
||||
const activeKeys = new Set(
|
||||
allTasks
|
||||
.filter((t) => ["pending", "running", "failed", "paused"].includes(t.status))
|
||||
.map((t) => `${t.bucketName ?? ""}/${t.key ?? ""}`),
|
||||
.filter(
|
||||
(task): task is UploadTask =>
|
||||
task.kind === "upload" && (task.status === "pending" || task.status === "running"),
|
||||
)
|
||||
.map((task) => `${task.bucketName}/${task.key}`),
|
||||
)
|
||||
const newTasks = uploadHelpers
|
||||
.createTasks(items, bucketName)
|
||||
.filter((t) => !activeKeys.has(`${t.bucketName}/${t.key}`))
|
||||
manager.enqueue(newTasks as AnyTask[])
|
||||
manager.enqueue(newTasks)
|
||||
},
|
||||
[managerState],
|
||||
)
|
||||
@@ -115,7 +105,7 @@ export function TaskProvider({ children }: { children: React.ReactNode }) {
|
||||
if (!managerState) return
|
||||
const { manager, deleteHelpers } = managerState
|
||||
const newTasks = deleteHelpers.createTasks(keys, bucketName, prefix, options)
|
||||
manager.enqueue(newTasks as AnyTask[])
|
||||
manager.enqueue(newTasks)
|
||||
},
|
||||
[managerState],
|
||||
)
|
||||
@@ -125,7 +115,7 @@ export function TaskProvider({ children }: { children: React.ReactNode }) {
|
||||
if (!managerState) return
|
||||
const { manager, deleteHelpers } = managerState
|
||||
const task = deleteHelpers.createFolderDeleteTask(prefix, bucketName, options)
|
||||
manager.enqueue([task as AnyTask])
|
||||
manager.enqueue([task])
|
||||
},
|
||||
[managerState],
|
||||
)
|
||||
@@ -157,7 +147,7 @@ export function useTasks() {
|
||||
() => taskManager.getTasks(),
|
||||
() => taskManager.getTasks(),
|
||||
)
|
||||
return tasks as AnyTask[]
|
||||
return tasks
|
||||
}
|
||||
|
||||
export function useAddUploadFiles() {
|
||||
|
||||
@@ -31,7 +31,11 @@ export function useLocalStorage<T>(key: string, defaultValue: T) {
|
||||
|
||||
useEffect(() => {
|
||||
const handleStorage = (e: StorageEvent) => {
|
||||
if (e.key === key && e.newValue !== null) {
|
||||
if (e.key === key) {
|
||||
if (e.newValue === null) {
|
||||
setValue(defaultValue)
|
||||
return
|
||||
}
|
||||
try {
|
||||
setValue(JSON.parse(e.newValue) as T)
|
||||
} catch {
|
||||
@@ -42,7 +46,7 @@ export function useLocalStorage<T>(key: string, defaultValue: T) {
|
||||
|
||||
window.addEventListener("storage", handleStorage)
|
||||
return () => window.removeEventListener("storage", handleStorage)
|
||||
}, [key])
|
||||
}, [defaultValue, key])
|
||||
|
||||
const setStoredValue = useCallback(
|
||||
(newValue: T | ((prev: T) => T)) => {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"use client"
|
||||
|
||||
import { createContext, useContext, useState, useCallback, useEffect, useMemo } from "react"
|
||||
import { createContext, useContext, useState, useCallback, useEffect, useMemo, useRef } from "react"
|
||||
import { hasConsoleScopes, type ConsolePolicy } from "@/lib/console-policy-parser"
|
||||
import { hasConsoleCapability, type ConsoleCapability } from "@/lib/permission-capabilities"
|
||||
import type { PermissionResourceContext } from "@/lib/permission-resources"
|
||||
@@ -34,8 +34,10 @@ export function PermissionsProvider({ children }: { children: React.ReactNode })
|
||||
const [isLoading, setIsLoading] = useState(false)
|
||||
const [hasResolvedAdmin, setHasResolvedAdmin] = useState(false)
|
||||
const [hasFetchedPolicy, setHasFetchedPolicy] = useState(false)
|
||||
const requestEpochRef = useRef(0)
|
||||
|
||||
useEffect(() => {
|
||||
requestEpochRef.current += 1
|
||||
setUserInfo(null)
|
||||
setUserPolicy(null)
|
||||
setHasResolvedAdmin(false)
|
||||
@@ -43,28 +45,33 @@ export function PermissionsProvider({ children }: { children: React.ReactNode })
|
||||
if (isAuthenticated) {
|
||||
setIsAdmin(false)
|
||||
}
|
||||
}, [credentials?.AccessKeyId, credentials?.SessionToken, isAuthenticated, setIsAdmin])
|
||||
}, [api, credentials?.AccessKeyId, credentials?.SessionToken, isAuthenticated, setIsAdmin])
|
||||
|
||||
const fetchAdminStatus = useCallback(async () => {
|
||||
if (!api) return
|
||||
const requestEpoch = requestEpochRef.current
|
||||
|
||||
try {
|
||||
const info = (await api.get("/is-admin")) as { is_admin?: boolean }
|
||||
if (requestEpoch !== requestEpochRef.current) return
|
||||
setIsAdmin(info?.is_admin ?? false)
|
||||
} catch (e) {
|
||||
if (requestEpoch !== requestEpochRef.current) return
|
||||
console.error("Failed to resolve admin status", e)
|
||||
setIsAdmin(false)
|
||||
} finally {
|
||||
setHasResolvedAdmin(true)
|
||||
if (requestEpoch === requestEpochRef.current) setHasResolvedAdmin(true)
|
||||
}
|
||||
}, [api, setIsAdmin])
|
||||
|
||||
const fetchUserPolicy = useCallback(async () => {
|
||||
if (!api) return
|
||||
const requestEpoch = requestEpochRef.current
|
||||
|
||||
setIsLoading(true)
|
||||
try {
|
||||
const info = (await api.get("/accountinfo")) as Record<string, unknown>
|
||||
if (requestEpoch !== requestEpochRef.current) return
|
||||
setUserInfo(info)
|
||||
const policyStr = info?.policy as string
|
||||
if (policyStr) {
|
||||
@@ -73,11 +80,14 @@ export function PermissionsProvider({ children }: { children: React.ReactNode })
|
||||
setUserPolicy(null)
|
||||
}
|
||||
} catch (e) {
|
||||
if (requestEpoch !== requestEpochRef.current) return
|
||||
console.error("Failed to fetch user policy", e)
|
||||
setUserPolicy(null)
|
||||
} finally {
|
||||
setIsLoading(false)
|
||||
setHasFetchedPolicy(true)
|
||||
if (requestEpoch === requestEpochRef.current) {
|
||||
setIsLoading(false)
|
||||
setHasFetchedPolicy(true)
|
||||
}
|
||||
}
|
||||
}, [api])
|
||||
|
||||
|
||||
+1
-1
@@ -49,7 +49,7 @@ export function useTiers() {
|
||||
|
||||
const removeTiers = useCallback(
|
||||
async (name: string) => {
|
||||
return api.delete(`/tier/${encodeURIComponent(name)}?force=true`, {})
|
||||
return api.delete(`/tier/${encodeURIComponent(name)}?force=false`, {})
|
||||
},
|
||||
[api],
|
||||
)
|
||||
|
||||
+43
-62
@@ -1,8 +1,14 @@
|
||||
import { joinURL } from "ufo"
|
||||
import type { IApiErrorHandler } from "@/types/api"
|
||||
import { redactRequestOptionsForLog } from "./api-request-log"
|
||||
import { parseApiError } from "./error-handler"
|
||||
import { logger } from "./logger"
|
||||
import type { AwsClient } from "./aws4fetch"
|
||||
|
||||
type ApiRequestInit = RequestInit & { body?: BodyInit | null; aws?: Record<string, unknown> }
|
||||
|
||||
export interface ApiTransport {
|
||||
fetch(input: string | Request, init?: ApiRequestInit): Promise<Response>
|
||||
}
|
||||
|
||||
interface ApiClientOptions {
|
||||
baseUrl?: string
|
||||
@@ -24,43 +30,19 @@ interface RequestOptions {
|
||||
signal?: AbortSignal
|
||||
}
|
||||
|
||||
const SENSITIVE_LOG_KEY = /(?:token|secret|password|authorization|cookie|api[_-]?key|master[_-]?key)/i
|
||||
|
||||
function redactLogValue(value: unknown, key?: string): unknown {
|
||||
if (key && SENSITIVE_LOG_KEY.test(key)) return "[REDACTED]"
|
||||
if (Array.isArray(value)) return value.map((item) => redactLogValue(item))
|
||||
if (!value || typeof value !== "object") return value
|
||||
|
||||
return Object.fromEntries(
|
||||
Object.entries(value as Record<string, unknown>).map(([entryKey, entryValue]) => [
|
||||
entryKey,
|
||||
redactLogValue(entryValue, entryKey),
|
||||
]),
|
||||
)
|
||||
}
|
||||
|
||||
function redactRequestOptionsForLog(options: RequestOptions) {
|
||||
const body =
|
||||
typeof options.body === "string"
|
||||
? (() => {
|
||||
try {
|
||||
return JSON.stringify(redactLogValue(JSON.parse(options.body)))
|
||||
} catch {
|
||||
return options.body
|
||||
}
|
||||
})()
|
||||
: options.body
|
||||
|
||||
return { ...options, body }
|
||||
async function createHttpError(response: Response) {
|
||||
const error = new Error(await parseApiError(response)) as Error & { status: number }
|
||||
error.status = response.status
|
||||
return error
|
||||
}
|
||||
|
||||
export class ApiClient {
|
||||
private $api: AwsClient
|
||||
private $api: ApiTransport
|
||||
private config?: ApiClientOptions
|
||||
private errorHandler?: IApiErrorHandler
|
||||
private inflightGetRequests = new Map<string, Promise<unknown>>()
|
||||
|
||||
constructor(api: AwsClient, options?: ApiClientOptions) {
|
||||
constructor(api: ApiTransport, options?: ApiClientOptions) {
|
||||
this.$api = api
|
||||
this.config = options
|
||||
this.errorHandler = options?.errorHandler
|
||||
@@ -76,37 +58,42 @@ export class ApiClient {
|
||||
|
||||
async request(url: string, options: RequestOptions = {}, parseJson: boolean = true) {
|
||||
url = this.config?.baseUrl ? joinURL(this.config?.baseUrl, url) : url
|
||||
options.headers = { ...this.config?.headers, ...options.headers }
|
||||
const method = (options.method ?? "GET").toUpperCase()
|
||||
options.method = method
|
||||
const { params, ...providedOptions } = options
|
||||
const requestOptions: RequestOptions = {
|
||||
...providedOptions,
|
||||
headers: { ...this.config?.headers, ...providedOptions.headers },
|
||||
}
|
||||
const method = (requestOptions.method ?? "GET").toUpperCase()
|
||||
requestOptions.method = method
|
||||
if (
|
||||
options.body &&
|
||||
!(options.body instanceof FormData) &&
|
||||
!(options.body instanceof Blob) &&
|
||||
!(options.body instanceof ArrayBuffer) &&
|
||||
!(options.body instanceof Uint8Array) &&
|
||||
!(options.body instanceof File)
|
||||
requestOptions.body &&
|
||||
!(requestOptions.body instanceof FormData) &&
|
||||
!(requestOptions.body instanceof Blob) &&
|
||||
!(requestOptions.body instanceof ArrayBuffer) &&
|
||||
!(requestOptions.body instanceof Uint8Array) &&
|
||||
!(requestOptions.body instanceof File)
|
||||
) {
|
||||
options.body = JSON.stringify(options.body)
|
||||
requestOptions.body = JSON.stringify(requestOptions.body)
|
||||
}
|
||||
|
||||
if (options.params) {
|
||||
const queryString = new URLSearchParams(options.params).toString()
|
||||
url += `?${queryString}`
|
||||
delete options.params
|
||||
if (params) {
|
||||
const queryString = new URLSearchParams(params).toString()
|
||||
if (queryString) url += `${url.includes("?") ? "&" : "?"}${queryString}`
|
||||
}
|
||||
|
||||
const shouldDedupe = method === "GET" && parseJson && options.dedupe !== false
|
||||
const shouldDedupe =
|
||||
method === "GET" &&
|
||||
parseJson &&
|
||||
requestOptions.dedupe !== false &&
|
||||
!providedOptions.headers &&
|
||||
!requestOptions.signal
|
||||
const dedupeKey = shouldDedupe ? `${method}:${url}` : null
|
||||
|
||||
const executeRequest = async () => {
|
||||
logger.log("[request] url:", url)
|
||||
logger.log("[request] options:", redactRequestOptionsForLog(options))
|
||||
logger.log("[request] options:", redactRequestOptionsForLog(requestOptions))
|
||||
|
||||
const response = await this.$api.fetch(
|
||||
url,
|
||||
options as RequestInit & { body?: BodyInit | null; aws?: Record<string, unknown> },
|
||||
)
|
||||
const response = await this.$api.fetch(url, requestOptions as ApiRequestInit)
|
||||
|
||||
logger.log("[request] response:", response)
|
||||
|
||||
@@ -114,16 +101,13 @@ export class ApiClient {
|
||||
if (this.errorHandler) {
|
||||
await this.errorHandler.handle401(url)
|
||||
}
|
||||
return
|
||||
throw await createHttpError(response)
|
||||
}
|
||||
if (response.status === 403) {
|
||||
// If suppress403Redirect is true, throw error instead of triggering global handler
|
||||
// This allows components to handle permission errors gracefully
|
||||
if (options.suppress403Redirect) {
|
||||
const errorMsg = await parseApiError(response)
|
||||
const error = new Error(errorMsg) as Error & { status: number }
|
||||
error.status = response.status
|
||||
throw error
|
||||
if (requestOptions.suppress403Redirect) {
|
||||
throw await createHttpError(response)
|
||||
}
|
||||
|
||||
try {
|
||||
@@ -160,14 +144,11 @@ export class ApiClient {
|
||||
await this.errorHandler.handle403(url)
|
||||
}
|
||||
}
|
||||
return
|
||||
throw await createHttpError(response)
|
||||
}
|
||||
|
||||
if (!response.ok) {
|
||||
const errorMsg = await parseApiError(response)
|
||||
const error = new Error(errorMsg) as Error & { status: number }
|
||||
error.status = response.status
|
||||
throw error
|
||||
throw await createHttpError(response)
|
||||
}
|
||||
|
||||
if (response.status === 204 || response.headers.get("content-length") === "0" || !response.body) {
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
const SENSITIVE_LOG_KEY =
|
||||
/(?:token|secret|password|authorization|cookie|credential|creds|api[_-]?key|master[_-]?key|private[_-]?key)/i
|
||||
|
||||
function redactLogValue(value: unknown, key?: string): unknown {
|
||||
if (key && SENSITIVE_LOG_KEY.test(key)) return "[REDACTED]"
|
||||
if (Array.isArray(value)) return value.map((item) => redactLogValue(item))
|
||||
if (!value || typeof value !== "object") return value
|
||||
|
||||
return Object.fromEntries(
|
||||
Object.entries(value as Record<string, unknown>).map(([entryKey, entryValue]) => [
|
||||
entryKey,
|
||||
redactLogValue(entryValue, entryKey),
|
||||
]),
|
||||
)
|
||||
}
|
||||
|
||||
export function redactRequestOptionsForLog<T extends object>(options: T): unknown {
|
||||
const request = options as T & { body?: unknown }
|
||||
let body = request.body
|
||||
if (typeof body === "string") {
|
||||
try {
|
||||
const parsedBody = JSON.parse(body) as unknown
|
||||
body = parsedBody && typeof parsedBody === "object" ? JSON.stringify(redactLogValue(parsedBody)) : "[REDACTED]"
|
||||
} catch {
|
||||
body = "[REDACTED]"
|
||||
}
|
||||
} else if (body && typeof body === "object") {
|
||||
body = redactLogValue(body)
|
||||
}
|
||||
|
||||
return redactLogValue({ ...options, body })
|
||||
}
|
||||
+1
-1
@@ -166,5 +166,5 @@ export function validateBucketCorsJson(content: string): BucketCorsValidationRes
|
||||
}
|
||||
|
||||
export function stringifyBucketCorsConfig(config?: BucketCorsConfiguration | null): string {
|
||||
return JSON.stringify(config?.CORSRules ?? DEFAULT_BUCKET_CORS_CONFIGURATION.CORSRules, null, 2)
|
||||
return JSON.stringify(config ?? DEFAULT_BUCKET_CORS_CONFIGURATION, null, 2)
|
||||
}
|
||||
|
||||
@@ -144,6 +144,13 @@ const IMPLIED_SCOPES: Record<string, string[]> = {
|
||||
[CONSOLE_SCOPES.VIEW_LICENSE]: ["admin:ServerInfo", "admin:*"],
|
||||
}
|
||||
|
||||
function matchesImpliedConsoleAction(policyActions: string[] | undefined, action: string): boolean {
|
||||
return Object.entries(IMPLIED_SCOPES).some(
|
||||
([scope, impliedActions]) =>
|
||||
matchAction(policyActions, scope) && impliedActions.some((implied) => matchAction([implied], action)),
|
||||
)
|
||||
}
|
||||
|
||||
function isConsoleResource(resource: string): boolean {
|
||||
return resource === "console" || resource === "*" || resource.includes("console")
|
||||
}
|
||||
@@ -166,6 +173,14 @@ function matchStatementResource(statement: ConsoleStatement, requestResource: st
|
||||
return true
|
||||
}
|
||||
|
||||
if (
|
||||
!isConsoleScope(action) &&
|
||||
matchesImpliedConsoleAction(statement.Action, action) &&
|
||||
(statement.Resource ?? []).some(isConsoleResource)
|
||||
) {
|
||||
return true
|
||||
}
|
||||
|
||||
const resourceMatched =
|
||||
!statement.Resource || statement.Resource.length === 0
|
||||
? true
|
||||
@@ -189,7 +204,7 @@ function matchesRequestedAction(policyActions: string[] | undefined, action: str
|
||||
}
|
||||
|
||||
if (!isConsoleScope(action)) {
|
||||
return false
|
||||
return matchesImpliedConsoleAction(policyActions, action)
|
||||
}
|
||||
|
||||
if (
|
||||
|
||||
+17
-8
@@ -41,8 +41,8 @@ export interface DeleteTaskConfig {
|
||||
}
|
||||
|
||||
export interface DeleteTaskHelpers {
|
||||
handler: TaskHandler<DeleteTask, DeleteStatus>
|
||||
folderHandler: TaskHandler<FolderDeleteTask, DeleteStatus>
|
||||
handler: TaskHandler<DeleteTask>
|
||||
folderHandler: TaskHandler<FolderDeleteTask>
|
||||
createTasks: (
|
||||
keys: string[],
|
||||
bucketName: string,
|
||||
@@ -66,6 +66,13 @@ const lifecycle: TaskLifecycleStatus<DeleteStatus> = {
|
||||
canceled: "canceled",
|
||||
}
|
||||
|
||||
function assertDeleteObjectsSucceeded(errors: { Key?: string; Code?: string; Message?: string }[] | undefined) {
|
||||
if (!errors?.length) return
|
||||
const first = errors[0]
|
||||
const target = first?.Key ? ` for ${first.Key}` : ""
|
||||
throw new Error(first?.Message || first?.Code || `Failed to delete ${errors.length} object(s)${target}`)
|
||||
}
|
||||
|
||||
function attachForceDeleteHeader(command: DeleteObjectCommand | DeleteObjectsCommand) {
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
;(command.middlewareStack.add as any)(
|
||||
@@ -83,7 +90,7 @@ export function createDeleteTaskHelpers(s3Client: S3Client, config: DeleteTaskCo
|
||||
const maxRetries = config.maxRetries ?? 3
|
||||
const retryDelay = config.retryDelay ?? 1000
|
||||
|
||||
const perform: TaskHandler<DeleteTask, DeleteStatus>["perform"] = async (task) => {
|
||||
const perform: TaskHandler<DeleteTask>["perform"] = async (task) => {
|
||||
const { key, bucketName, prefix, versionId, forceDelete } = task
|
||||
const abortController = new AbortController()
|
||||
task.abortController = abortController
|
||||
@@ -100,7 +107,7 @@ export function createDeleteTaskHelpers(s3Client: S3Client, config: DeleteTaskCo
|
||||
task.progress = 100
|
||||
}
|
||||
|
||||
const performFolder: TaskHandler<FolderDeleteTask, DeleteStatus>["perform"] = async (task) => {
|
||||
const performFolder: TaskHandler<FolderDeleteTask>["perform"] = async (task) => {
|
||||
const { bucketName, prefix, forceDelete } = task
|
||||
const abortController = new AbortController()
|
||||
task.abortController = abortController
|
||||
@@ -137,7 +144,8 @@ export function createDeleteTaskHelpers(s3Client: S3Client, config: DeleteTaskCo
|
||||
Delete: { Objects: objectsToDelete, Quiet: true },
|
||||
})
|
||||
attachForceDeleteHeader(command)
|
||||
await s3Client.send(command, { abortSignal: abortController.signal })
|
||||
const result = await s3Client.send(command, { abortSignal: abortController.signal })
|
||||
assertDeleteObjectsSucceeded(result.Errors)
|
||||
}
|
||||
|
||||
isTruncated = data.IsTruncated ?? false
|
||||
@@ -167,7 +175,8 @@ export function createDeleteTaskHelpers(s3Client: S3Client, config: DeleteTaskCo
|
||||
Bucket: bucketName,
|
||||
Delete: { Objects: objectsToDelete, Quiet: true },
|
||||
})
|
||||
await s3Client.send(command, { abortSignal: abortController.signal })
|
||||
const result = await s3Client.send(command, { abortSignal: abortController.signal })
|
||||
assertDeleteObjectsSucceeded(result.Errors)
|
||||
}
|
||||
|
||||
isTruncated = data.IsTruncated ?? false
|
||||
@@ -185,7 +194,7 @@ export function createDeleteTaskHelpers(s3Client: S3Client, config: DeleteTaskCo
|
||||
return !nonRetryableErrors.some((msg) => errorMessage.includes(msg))
|
||||
}
|
||||
|
||||
const handler: TaskHandler<DeleteTask, DeleteStatus> = {
|
||||
const handler: TaskHandler<DeleteTask> = {
|
||||
lifecycle,
|
||||
perform,
|
||||
shouldRetry,
|
||||
@@ -195,7 +204,7 @@ export function createDeleteTaskHelpers(s3Client: S3Client, config: DeleteTaskCo
|
||||
retryDelay,
|
||||
}
|
||||
|
||||
const folderHandler: TaskHandler<FolderDeleteTask, DeleteStatus> = {
|
||||
const folderHandler: TaskHandler<FolderDeleteTask> = {
|
||||
lifecycle,
|
||||
perform: performFolder,
|
||||
shouldRetry,
|
||||
|
||||
@@ -54,5 +54,9 @@ export function buildObjectZipDownloadPayload({
|
||||
}
|
||||
|
||||
export function normalizeObjectZipDownloadUrl(downloadUrl: string, origin: string) {
|
||||
return new URL(downloadUrl, origin).toString()
|
||||
const url = new URL(downloadUrl, origin)
|
||||
if (url.protocol !== "http:" && url.protocol !== "https:") {
|
||||
throw new Error("Unsupported download URL protocol")
|
||||
}
|
||||
return url.toString()
|
||||
}
|
||||
|
||||
+239
-165
@@ -1,7 +1,4 @@
|
||||
export type TaskEventType = "enqueued" | "running" | "completed" | "failed" | "canceled" | "drained"
|
||||
|
||||
export type TaskEventHandler<TTask> = (task: TTask) => void
|
||||
export type AllCompletedEventHandler = () => void
|
||||
import { logger } from "./logger"
|
||||
|
||||
export interface TaskLifecycleStatus<TStatus extends string> {
|
||||
pending: TStatus
|
||||
@@ -9,10 +6,9 @@ export interface TaskLifecycleStatus<TStatus extends string> {
|
||||
completed: TStatus
|
||||
failed: TStatus
|
||||
canceled: TStatus
|
||||
paused?: TStatus
|
||||
}
|
||||
|
||||
export interface ManagedTask<TStatus extends string> {
|
||||
export interface ManagedTask<TStatus extends string = string> {
|
||||
id: string
|
||||
status: TStatus
|
||||
progress: number
|
||||
@@ -22,162 +18,228 @@ export interface ManagedTask<TStatus extends string> {
|
||||
kind: string
|
||||
}
|
||||
|
||||
export interface TaskHandler<TTask, TStatus extends string> {
|
||||
lifecycle: TaskLifecycleStatus<TStatus>
|
||||
export interface TaskHandler<TTask extends ManagedTask> {
|
||||
lifecycle: TaskLifecycleStatus<TTask["status"]>
|
||||
perform: (task: TTask) => Promise<void>
|
||||
shouldRetry?: (task: TTask, error: unknown) => boolean
|
||||
handleError?: (task: TTask, error: unknown) => Promise<"retry" | "handled" | "fail"> | "retry" | "handled" | "fail"
|
||||
isCanceledError?: (error: unknown) => boolean
|
||||
maxRetries?: number
|
||||
retryDelay?: number
|
||||
}
|
||||
|
||||
export interface TaskManagerOptions<TTask extends ManagedTask<TStatus>, TStatus extends string> {
|
||||
handlers: Record<string, TaskHandler<TTask, TStatus>>
|
||||
export type TaskHandlerMap<TTask extends ManagedTask> = {
|
||||
[TKind in TTask["kind"]]: TaskHandler<Extract<TTask, { kind: TKind }>>
|
||||
}
|
||||
|
||||
export interface TaskManagerOptions<TTask extends ManagedTask> {
|
||||
handlers: TaskHandlerMap<TTask>
|
||||
maxConcurrent?: number
|
||||
maxRetries?: number
|
||||
retryDelay?: number
|
||||
}
|
||||
|
||||
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))
|
||||
export interface TaskManagerApi<TTask extends ManagedTask> {
|
||||
subscribe(listener: () => void): () => void
|
||||
getTasks(): readonly TTask[]
|
||||
enqueue(tasks: readonly TTask[]): void
|
||||
cancelTask(taskId: string): boolean
|
||||
cancelAll(): void
|
||||
removeFinishedTask(taskId: string): boolean
|
||||
clearFinishedTasks(): void
|
||||
dispose(): void
|
||||
}
|
||||
|
||||
type Listener = () => void
|
||||
|
||||
export class TaskManager<TTask extends ManagedTask<TStatus>, TStatus extends string> {
|
||||
function assertIntegerOption(name: string, value: number, minimum: number) {
|
||||
if (!Number.isInteger(value) || value < minimum) {
|
||||
throw new Error(`${name} must be an integer greater than or equal to ${minimum}`)
|
||||
}
|
||||
}
|
||||
|
||||
function getErrorMessage(error: unknown): string {
|
||||
return error instanceof Error && error.message ? error.message : String(error || "Unknown error")
|
||||
}
|
||||
|
||||
export class TaskManager<TTask extends ManagedTask> implements TaskManagerApi<TTask> {
|
||||
private tasks: TTask[] = []
|
||||
private readonly handlers: Record<string, TaskHandler<TTask, TStatus>>
|
||||
private readonly handlers: TaskHandlerMap<TTask>
|
||||
readonly maxConcurrent: number
|
||||
readonly maxRetries: number
|
||||
readonly retryDelay: number
|
||||
private activeCount = 0
|
||||
private isStarted = false
|
||||
private allCompletedEmitted = false
|
||||
private disposed = false
|
||||
private listeners = new Set<Listener>()
|
||||
private snapshotCache: TTask[] | null = null
|
||||
private snapshotCache: readonly TTask[] | null = null
|
||||
private retryAvailableAt = new WeakMap<TTask, number>()
|
||||
private pumpTimer: ReturnType<typeof setTimeout> | null = null
|
||||
private pumpScheduledAt = Number.POSITIVE_INFINITY
|
||||
private progressWatchers = new Set<() => void>()
|
||||
|
||||
constructor(options: TaskManagerOptions<TTask, TStatus>) {
|
||||
constructor(options: TaskManagerOptions<TTask>) {
|
||||
this.handlers = options.handlers
|
||||
this.maxConcurrent = options.maxConcurrent ?? 6
|
||||
this.maxRetries = options.maxRetries ?? 3
|
||||
this.retryDelay = options.retryDelay ?? 1000
|
||||
|
||||
assertIntegerOption("maxConcurrent", this.maxConcurrent, 1)
|
||||
assertIntegerOption("maxRetries", this.maxRetries, 0)
|
||||
assertIntegerOption("retryDelay", this.retryDelay, 0)
|
||||
for (const handler of Object.values(options.handlers) as TaskHandler<TTask>[]) {
|
||||
if (handler.maxRetries !== undefined) assertIntegerOption("handler.maxRetries", handler.maxRetries, 0)
|
||||
if (handler.retryDelay !== undefined) assertIntegerOption("handler.retryDelay", handler.retryDelay, 0)
|
||||
}
|
||||
}
|
||||
|
||||
subscribe(listener: Listener): () => void {
|
||||
if (this.disposed) return () => {}
|
||||
this.listeners.add(listener)
|
||||
return () => this.listeners.delete(listener)
|
||||
}
|
||||
|
||||
private notify() {
|
||||
this.snapshotCache = null
|
||||
this.listeners.forEach((l) => l())
|
||||
}
|
||||
|
||||
enqueue(tasks: TTask[]) {
|
||||
if (!tasks.length) return
|
||||
this.allCompletedEmitted = false
|
||||
this.tasks.push(...tasks)
|
||||
this.notify()
|
||||
this.start()
|
||||
}
|
||||
|
||||
start() {
|
||||
if (!this.isStarted) this.isStarted = true
|
||||
this.processQueue()
|
||||
}
|
||||
|
||||
cancel() {
|
||||
this.tasks.forEach((task) => {
|
||||
const lifecycle = this.getLifecycle(task) as TaskLifecycleStatus<TStatus>
|
||||
if (this.isActive(task)) {
|
||||
;(task as TTask & { _pauseRequested?: boolean })._pauseRequested = false
|
||||
task.abortController?.abort()
|
||||
task.status = lifecycle.canceled
|
||||
}
|
||||
})
|
||||
this.notify()
|
||||
this.checkAllCompleted()
|
||||
}
|
||||
|
||||
cancelTask(taskId: string) {
|
||||
const task = this.tasks.find((t) => t.id === taskId)
|
||||
if (!task) return
|
||||
const lifecycle = this.getLifecycle(task) as TaskLifecycleStatus<TStatus>
|
||||
;(task as TTask & { _pauseRequested?: boolean })._pauseRequested = false
|
||||
task.abortController?.abort()
|
||||
task.status = lifecycle.canceled
|
||||
this.notify()
|
||||
this.checkAllCompleted()
|
||||
}
|
||||
|
||||
removeTask(taskId: string) {
|
||||
const index = this.tasks.findIndex((t) => t.id === taskId)
|
||||
if (index === -1) return
|
||||
const task = this.tasks[index]
|
||||
if (task && this.isActive(task)) {
|
||||
task.abortController?.abort()
|
||||
}
|
||||
this.tasks.splice(index, 1)
|
||||
this.notify()
|
||||
}
|
||||
|
||||
clearTasks() {
|
||||
this.cancel()
|
||||
this.tasks = []
|
||||
this.activeCount = 0
|
||||
this.isStarted = false
|
||||
this.notify()
|
||||
}
|
||||
|
||||
getTasks(): TTask[] {
|
||||
getTasks(): readonly TTask[] {
|
||||
if (this.snapshotCache) return this.snapshotCache
|
||||
this.snapshotCache = [...this.tasks]
|
||||
return this.snapshotCache
|
||||
}
|
||||
|
||||
private isActive(task: TTask) {
|
||||
const lifecycle = this.getLifecycle(task) as TaskLifecycleStatus<TStatus>
|
||||
const activeStatuses = [lifecycle.pending, lifecycle.running]
|
||||
if (lifecycle.paused) activeStatuses.push(lifecycle.paused)
|
||||
return activeStatuses.includes(task.status)
|
||||
enqueue(tasks: readonly TTask[]) {
|
||||
if (this.disposed) throw new Error("Cannot enqueue tasks after TaskManager is disposed")
|
||||
if (!tasks.length) return
|
||||
|
||||
this.validateEnqueue(tasks)
|
||||
this.tasks.push(...tasks)
|
||||
this.notify()
|
||||
this.schedulePump()
|
||||
}
|
||||
|
||||
private checkAllCompleted() {
|
||||
const hasActive = this.tasks.some((task) => this.isActive(task))
|
||||
if (!hasActive && this.tasks.length > 0 && !this.allCompletedEmitted) {
|
||||
this.allCompletedEmitted = true
|
||||
this.notify()
|
||||
cancelTask(taskId: string): boolean {
|
||||
const task = this.tasks.find((candidate) => candidate.id === taskId)
|
||||
if (!task || !this.isActive(task)) return false
|
||||
|
||||
this.cancelActiveTask(task)
|
||||
this.notify()
|
||||
this.schedulePump()
|
||||
return true
|
||||
}
|
||||
|
||||
cancelAll() {
|
||||
let changed = false
|
||||
for (const task of this.tasks) {
|
||||
if (!this.isActive(task)) continue
|
||||
this.cancelActiveTask(task)
|
||||
changed = true
|
||||
}
|
||||
if (changed) {
|
||||
this.notify()
|
||||
this.schedulePump()
|
||||
}
|
||||
}
|
||||
|
||||
removeFinishedTask(taskId: string): boolean {
|
||||
const index = this.tasks.findIndex((task) => task.id === taskId)
|
||||
if (index === -1 || !this.isFinished(this.tasks[index]!)) return false
|
||||
|
||||
this.tasks.splice(index, 1)
|
||||
this.notify()
|
||||
return true
|
||||
}
|
||||
|
||||
clearFinishedTasks() {
|
||||
const remaining = this.tasks.filter((task) => !this.isFinished(task))
|
||||
if (remaining.length === this.tasks.length) return
|
||||
this.tasks = remaining
|
||||
this.notify()
|
||||
}
|
||||
|
||||
dispose() {
|
||||
if (this.disposed) return
|
||||
this.disposed = true
|
||||
|
||||
for (const task of this.tasks) {
|
||||
if (this.isActive(task)) this.cancelActiveTask(task)
|
||||
}
|
||||
this.tasks = []
|
||||
this.clearPumpTimer()
|
||||
this.progressWatchers.forEach((stopWatching) => stopWatching())
|
||||
this.progressWatchers.clear()
|
||||
this.notify()
|
||||
this.listeners.clear()
|
||||
}
|
||||
|
||||
private validateEnqueue(tasks: readonly TTask[]) {
|
||||
const knownIds = new Set(this.tasks.map((task) => task.id))
|
||||
for (const task of tasks) {
|
||||
const handler = this.getHandler(task)
|
||||
if (task.status !== handler.lifecycle.pending) {
|
||||
throw new Error(`Task ${task.id} must be pending when enqueued`)
|
||||
}
|
||||
if (knownIds.has(task.id)) {
|
||||
throw new Error(`Task id already exists: ${task.id}`)
|
||||
}
|
||||
knownIds.add(task.id)
|
||||
}
|
||||
}
|
||||
|
||||
private notify() {
|
||||
this.snapshotCache = null
|
||||
this.listeners.forEach((listener) => {
|
||||
try {
|
||||
listener()
|
||||
} catch (error) {
|
||||
logger.error("TaskManager listener failed", error)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
private schedulePump(delay = 0) {
|
||||
if (this.disposed) return
|
||||
const scheduledAt = Date.now() + Math.max(0, delay)
|
||||
if (this.pumpTimer && this.pumpScheduledAt <= scheduledAt) return
|
||||
|
||||
this.clearPumpTimer()
|
||||
this.pumpScheduledAt = scheduledAt
|
||||
this.pumpTimer = setTimeout(
|
||||
() => {
|
||||
this.pumpTimer = null
|
||||
this.pumpScheduledAt = Number.POSITIVE_INFINITY
|
||||
this.processQueue()
|
||||
},
|
||||
Math.max(0, scheduledAt - Date.now()),
|
||||
)
|
||||
}
|
||||
|
||||
private clearPumpTimer() {
|
||||
if (this.pumpTimer) clearTimeout(this.pumpTimer)
|
||||
this.pumpTimer = null
|
||||
this.pumpScheduledAt = Number.POSITIVE_INFINITY
|
||||
}
|
||||
|
||||
private processQueue() {
|
||||
if (!this.isStarted) return
|
||||
if (this.disposed) return
|
||||
|
||||
const now = Date.now()
|
||||
while (this.activeCount < this.maxConcurrent) {
|
||||
const next = this.tasks.find((task) => {
|
||||
const handler = this.handlers[task.kind]
|
||||
return handler && task.status === (handler.lifecycle as TaskLifecycleStatus<TStatus>).pending
|
||||
})
|
||||
const next = this.tasks.find((task) => this.isPending(task) && (this.retryAvailableAt.get(task) ?? 0) <= now)
|
||||
if (!next) break
|
||||
this.activeCount++
|
||||
this.runTask(next)
|
||||
|
||||
this.retryAvailableAt.delete(next)
|
||||
this.activeCount += 1
|
||||
void this.runTask(next)
|
||||
}
|
||||
|
||||
const hasPending = this.tasks.some((task) => {
|
||||
const lifecycle = this.getLifecycle(task) as TaskLifecycleStatus<TStatus>
|
||||
return task.status === lifecycle.pending
|
||||
})
|
||||
if (hasPending) {
|
||||
setTimeout(() => this.processQueue(), 100)
|
||||
} else {
|
||||
setTimeout(() => this.checkAllCompleted(), 200)
|
||||
}
|
||||
const nextRetryAt = this.tasks.reduce((earliest, task) => {
|
||||
if (!this.isPending(task)) return earliest
|
||||
const availableAt = this.retryAvailableAt.get(task) ?? 0
|
||||
return availableAt > now ? Math.min(earliest, availableAt) : earliest
|
||||
}, Number.POSITIVE_INFINITY)
|
||||
if (Number.isFinite(nextRetryAt)) this.schedulePump(nextRetryAt - now)
|
||||
}
|
||||
|
||||
private watchProgress(task: TTask, lifecycle: TaskLifecycleStatus<TStatus>) {
|
||||
private watchProgress(task: TTask, lifecycle: TaskLifecycleStatus<TTask["status"]>) {
|
||||
let lastProgress = task.progress
|
||||
const timer = setInterval(() => {
|
||||
if (task.status !== lifecycle.running) return
|
||||
if (task.progress === lastProgress) return
|
||||
if (this.disposed || task.status !== lifecycle.running || task.progress === lastProgress) return
|
||||
lastProgress = task.progress
|
||||
this.notify()
|
||||
}, 120)
|
||||
@@ -185,89 +247,101 @@ export class TaskManager<TTask extends ManagedTask<TStatus>, TStatus extends str
|
||||
}
|
||||
|
||||
private async runTask(task: TTask) {
|
||||
const lifecycle = this.getLifecycle(task) as TaskLifecycleStatus<TStatus>
|
||||
const handler = this.getHandler(task)
|
||||
const lifecycle = handler.lifecycle
|
||||
task.status = lifecycle.running
|
||||
task.error = undefined
|
||||
this.notify()
|
||||
|
||||
const stopWatchingProgress = this.watchProgress(task, lifecycle)
|
||||
this.progressWatchers.add(stopWatchingProgress)
|
||||
|
||||
try {
|
||||
const handler = this.getHandler(task)
|
||||
await handler.perform(task)
|
||||
if (task.status === lifecycle.running) {
|
||||
task.status = lifecycle.completed
|
||||
task.progress = 100
|
||||
}
|
||||
this.notify()
|
||||
this.checkAllCompleted()
|
||||
} catch (error) {
|
||||
const action = await this.handleError(task, error)
|
||||
if (action === "retry") {
|
||||
await this.retryTask(task)
|
||||
} else {
|
||||
task.status = lifecycle.failed
|
||||
task.error = (error as Error)?.message ?? "Unknown error"
|
||||
}
|
||||
this.notify()
|
||||
this.checkAllCompleted()
|
||||
this.resolveTaskError(task, handler, error)
|
||||
} finally {
|
||||
stopWatchingProgress()
|
||||
this.activeCount--
|
||||
if (this.isStarted) setTimeout(() => this.processQueue(), 50)
|
||||
this.progressWatchers.delete(stopWatchingProgress)
|
||||
task.abortController = undefined
|
||||
this.activeCount = Math.max(0, this.activeCount - 1)
|
||||
this.schedulePump()
|
||||
}
|
||||
}
|
||||
|
||||
private async handleError(task: TTask, error: unknown): Promise<"retry" | "handled" | "fail"> {
|
||||
const handler = this.getHandler(task)
|
||||
if (handler.handleError) {
|
||||
const result = await handler.handleError(task, error)
|
||||
if (result !== "fail") return result
|
||||
private resolveTaskError(task: TTask, handler: TaskHandler<TTask>, error: unknown) {
|
||||
const lifecycle = handler.lifecycle
|
||||
try {
|
||||
if (task.status === lifecycle.canceled || this.isCanceledError(handler, error)) {
|
||||
task.status = lifecycle.canceled
|
||||
task.error = undefined
|
||||
this.retryAvailableAt.delete(task)
|
||||
} else if (this.canRetry(task, handler, error)) {
|
||||
const retryCount = (task.retryCount ?? 0) + 1
|
||||
task.retryCount = retryCount
|
||||
task.status = lifecycle.pending
|
||||
task.progress = 0
|
||||
task.error = undefined
|
||||
this.retryAvailableAt.set(task, Date.now() + this.getRetryDelay(handler) * retryCount)
|
||||
} else {
|
||||
task.status = lifecycle.failed
|
||||
task.error = getErrorMessage(error)
|
||||
}
|
||||
} catch (classificationError) {
|
||||
task.status = lifecycle.failed
|
||||
task.error = `Failed to classify task error: ${getErrorMessage(classificationError)}`
|
||||
}
|
||||
if (this.isCanceledError(task, error)) {
|
||||
const lifecycle = this.getLifecycle(task) as TaskLifecycleStatus<TStatus>
|
||||
task.status = lifecycle.canceled
|
||||
this.notify()
|
||||
this.checkAllCompleted()
|
||||
return "handled"
|
||||
}
|
||||
if (this.shouldRetry(task, error)) return "retry"
|
||||
return "fail"
|
||||
this.notify()
|
||||
}
|
||||
|
||||
private shouldRetry(task: TTask, error: unknown) {
|
||||
const handler = this.getHandler(task)
|
||||
const lifecycle = this.getLifecycle(task) as TaskLifecycleStatus<TStatus>
|
||||
private canRetry(task: TTask, handler: TaskHandler<TTask>, error: unknown): boolean {
|
||||
const maxRetries = handler.maxRetries ?? this.maxRetries
|
||||
if (task.status === lifecycle.canceled) return false
|
||||
if ((task.retryCount || 0) >= maxRetries) return false
|
||||
if (handler.shouldRetry) return handler.shouldRetry(task, error)
|
||||
return true
|
||||
assertIntegerOption("handler.maxRetries", maxRetries, 0)
|
||||
if ((task.retryCount ?? 0) >= maxRetries) return false
|
||||
return handler.shouldRetry ? handler.shouldRetry(task, error) : true
|
||||
}
|
||||
|
||||
private isCanceledError(task: TTask, error: unknown) {
|
||||
const handler = this.getHandler(task)
|
||||
if (handler.isCanceledError) return handler.isCanceledError(error)
|
||||
return (error as Error)?.name === "AbortError" || (error as Error)?.message?.includes("canceled")
|
||||
}
|
||||
|
||||
private async retryTask(task: TTask) {
|
||||
const handler = this.getHandler(task)
|
||||
const lifecycle = this.getLifecycle(task) as TaskLifecycleStatus<TStatus>
|
||||
private getRetryDelay(handler: TaskHandler<TTask>): number {
|
||||
const retryDelay = handler.retryDelay ?? this.retryDelay
|
||||
task.retryCount = (task.retryCount || 0) + 1
|
||||
task.status = lifecycle.pending
|
||||
task.progress = 0
|
||||
task.error = undefined
|
||||
const retryCount = task.retryCount || 1
|
||||
await sleep(retryDelay * retryCount)
|
||||
assertIntegerOption("handler.retryDelay", retryDelay, 0)
|
||||
return retryDelay
|
||||
}
|
||||
|
||||
private getHandler(task: TTask): TaskHandler<TTask, TStatus> {
|
||||
const handler = this.handlers[task.kind] as TaskHandler<TTask, TStatus> | undefined
|
||||
private isCanceledError(handler: TaskHandler<TTask>, error: unknown): boolean {
|
||||
if (handler.isCanceledError) return handler.isCanceledError(error)
|
||||
return error instanceof Error && (error.name === "AbortError" || error.message.toLowerCase().includes("canceled"))
|
||||
}
|
||||
|
||||
private cancelActiveTask(task: TTask) {
|
||||
const lifecycle = this.getHandler(task).lifecycle
|
||||
task.abortController?.abort()
|
||||
task.status = lifecycle.canceled
|
||||
task.error = undefined
|
||||
this.retryAvailableAt.delete(task)
|
||||
}
|
||||
|
||||
private isPending(task: TTask): boolean {
|
||||
return task.status === this.getHandler(task).lifecycle.pending
|
||||
}
|
||||
|
||||
private isActive(task: TTask): boolean {
|
||||
const lifecycle = this.getHandler(task).lifecycle
|
||||
return task.status === lifecycle.pending || task.status === lifecycle.running
|
||||
}
|
||||
|
||||
private isFinished(task: TTask): boolean {
|
||||
const lifecycle = this.getHandler(task).lifecycle
|
||||
return task.status === lifecycle.completed || task.status === lifecycle.failed || task.status === lifecycle.canceled
|
||||
}
|
||||
|
||||
private getHandler(task: TTask): TaskHandler<TTask> {
|
||||
const handler = (this.handlers as unknown as Record<string, TaskHandler<TTask> | undefined>)[task.kind]
|
||||
if (!handler) throw new Error(`No task handler registered for kind: ${task.kind}`)
|
||||
return handler
|
||||
}
|
||||
|
||||
private getLifecycle(task: TTask) {
|
||||
return this.getHandler(task).lifecycle
|
||||
}
|
||||
}
|
||||
|
||||
+15
-26
@@ -9,8 +9,9 @@ import type { ManagedTask, TaskHandler, TaskLifecycleStatus } from "./task-manag
|
||||
import { formatBytes } from "./functions"
|
||||
import { createTaskId } from "./task-id"
|
||||
import { getUploadContentType } from "./upload-content-type"
|
||||
import { logger } from "./logger"
|
||||
|
||||
export type UploadStatus = "pending" | "running" | "completed" | "failed" | "canceled" | "paused"
|
||||
export type UploadStatus = "pending" | "running" | "completed" | "failed" | "canceled"
|
||||
|
||||
export interface UploadTask extends ManagedTask<UploadStatus> {
|
||||
kind: "upload"
|
||||
@@ -23,7 +24,6 @@ export interface UploadTask extends ManagedTask<UploadStatus> {
|
||||
subInfo: string
|
||||
uploadId?: string
|
||||
completedParts?: { ETag: string; PartNumber: number }[]
|
||||
_pauseRequested?: boolean
|
||||
}
|
||||
|
||||
export interface UploadTaskConfig {
|
||||
@@ -33,7 +33,7 @@ export interface UploadTaskConfig {
|
||||
}
|
||||
|
||||
export interface UploadTaskHelpers {
|
||||
handler: TaskHandler<UploadTask, UploadStatus>
|
||||
handler: TaskHandler<UploadTask>
|
||||
createTasks: (items: { file: File; key: string }[], bucketName: string) => UploadTask[]
|
||||
}
|
||||
|
||||
@@ -43,7 +43,6 @@ const lifecycle: TaskLifecycleStatus<UploadStatus> = {
|
||||
completed: "completed",
|
||||
failed: "failed",
|
||||
canceled: "canceled",
|
||||
paused: "paused",
|
||||
}
|
||||
|
||||
export function createUploadTaskHelpers(s3Client: S3Client, config: UploadTaskConfig = {}): UploadTaskHelpers {
|
||||
@@ -51,7 +50,7 @@ export function createUploadTaskHelpers(s3Client: S3Client, config: UploadTaskCo
|
||||
const maxRetries = config.maxRetries ?? 3
|
||||
const retryDelay = config.retryDelay ?? 1000
|
||||
|
||||
const perform: TaskHandler<UploadTask, UploadStatus>["perform"] = async (task) => {
|
||||
const perform: TaskHandler<UploadTask>["perform"] = async (task) => {
|
||||
task.progress = 0
|
||||
if (task.file.size > chunkSize) {
|
||||
await multipartUpload(task, s3Client, chunkSize)
|
||||
@@ -60,12 +59,12 @@ export function createUploadTaskHelpers(s3Client: S3Client, config: UploadTaskCo
|
||||
}
|
||||
}
|
||||
|
||||
const shouldRetry: TaskHandler<UploadTask, UploadStatus>["shouldRetry"] = (task, error) => {
|
||||
const shouldRetry: TaskHandler<UploadTask>["shouldRetry"] = (task, error) => {
|
||||
if (task.status === lifecycle.canceled) return false
|
||||
if ((task.retryCount || 0) >= maxRetries) return false
|
||||
const err = error as { $metadata?: { httpStatusCode?: number }; message?: string }
|
||||
if (err?.$metadata?.httpStatusCode && err.$metadata.httpStatusCode >= 400 && err.$metadata.httpStatusCode < 500)
|
||||
return false
|
||||
const statusCode = err?.$metadata?.httpStatusCode
|
||||
if (statusCode && statusCode >= 400 && statusCode < 500 && statusCode !== 408 && statusCode !== 429) return false
|
||||
const errorMessage = (err?.message || "").toLowerCase()
|
||||
const nonRetryableErrors = [
|
||||
"access denied",
|
||||
@@ -77,22 +76,10 @@ export function createUploadTaskHelpers(s3Client: S3Client, config: UploadTaskCo
|
||||
return !nonRetryableErrors.some((msg) => errorMessage.includes(msg))
|
||||
}
|
||||
|
||||
const handleError: TaskHandler<UploadTask, UploadStatus>["handleError"] = async (task, error) => {
|
||||
if (
|
||||
task._pauseRequested &&
|
||||
((error as Error)?.message === "Upload paused" || (error as Error)?.message === "Request aborted")
|
||||
) {
|
||||
task.status = "paused"
|
||||
return "handled"
|
||||
}
|
||||
return "fail"
|
||||
}
|
||||
|
||||
const handler: TaskHandler<UploadTask, UploadStatus> = {
|
||||
const handler: TaskHandler<UploadTask> = {
|
||||
lifecycle,
|
||||
perform,
|
||||
shouldRetry,
|
||||
handleError,
|
||||
isCanceledError: (error) =>
|
||||
(error as Error)?.name === "AbortError" || (error as Error)?.message?.includes("canceled"),
|
||||
maxRetries,
|
||||
@@ -118,7 +105,6 @@ export function createUploadTaskHelpers(s3Client: S3Client, config: UploadTaskCo
|
||||
displayName,
|
||||
subInfo: formatBytes(item.file.size),
|
||||
retryCount: 0,
|
||||
_pauseRequested: false,
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -172,7 +158,7 @@ async function multipartUpload(task: UploadTask, s3Client: S3Client, chunkSize:
|
||||
|
||||
const totalChunks = Math.ceil(file.size / chunkSize)
|
||||
for (let partNumber = 1; partNumber <= totalChunks; partNumber++) {
|
||||
if (abortController.signal.aborted) throw new Error("Upload paused")
|
||||
if (abortController.signal.aborted) throw new DOMException("Upload canceled", "AbortError")
|
||||
if (completedParts.some((p) => p.PartNumber === partNumber)) continue
|
||||
|
||||
const start = (partNumber - 1) * chunkSize
|
||||
@@ -207,12 +193,15 @@ async function multipartUpload(task: UploadTask, s3Client: S3Client, chunkSize:
|
||||
task.uploadId = undefined
|
||||
task.completedParts = undefined
|
||||
} catch (error) {
|
||||
if (!task._pauseRequested && uploadId) {
|
||||
if (uploadId) {
|
||||
try {
|
||||
const { AbortMultipartUploadCommand } = await import("@aws-sdk/client-s3")
|
||||
await s3Client.send(new AbortMultipartUploadCommand({ Bucket: bucketName, Key: key, UploadId: uploadId }))
|
||||
} catch {
|
||||
// ignore cleanup errors
|
||||
} catch (cleanupError) {
|
||||
logger.warn("Failed to abort multipart upload", cleanupError)
|
||||
} finally {
|
||||
task.uploadId = undefined
|
||||
task.completedParts = undefined
|
||||
}
|
||||
}
|
||||
throw error
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
"start": "next start",
|
||||
"lint": "eslint",
|
||||
"lint:fix": "eslint --fix",
|
||||
"test:run": "node --experimental-strip-types --import ./scripts/register-typescript-loader.mjs --test tests/lib/*.test.{js,ts}",
|
||||
"type-check": "node scripts/apply-theme-overrides.js && tsc --noEmit",
|
||||
"prettier": "prettier",
|
||||
"format": "prettier --write .",
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
import { register } from "node:module"
|
||||
|
||||
register("./typescript-loader.mjs", import.meta.url)
|
||||
@@ -0,0 +1,36 @@
|
||||
import { access } from "node:fs/promises"
|
||||
import { extname } from "node:path"
|
||||
import { fileURLToPath } from "node:url"
|
||||
|
||||
const projectRoot = new URL("../", import.meta.url)
|
||||
const extensions = [".ts", ".tsx", ".js", ".mjs"]
|
||||
|
||||
async function resolveExistingModule(baseUrl, context, nextResolve) {
|
||||
for (const extension of extensions) {
|
||||
const candidate = new URL(`${baseUrl.href}${extension}`)
|
||||
try {
|
||||
await access(fileURLToPath(candidate))
|
||||
return nextResolve(candidate.href, context)
|
||||
} catch {
|
||||
// Try the next supported source extension.
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
export async function resolve(specifier, context, nextResolve) {
|
||||
if (specifier.startsWith("@/")) {
|
||||
const resolved = await resolveExistingModule(new URL(specifier.slice(2), projectRoot), context, nextResolve)
|
||||
if (resolved) return resolved
|
||||
}
|
||||
|
||||
if ((specifier.startsWith("./") || specifier.startsWith("../")) && context.parentURL) {
|
||||
const baseUrl = new URL(specifier, context.parentURL)
|
||||
if (!extname(baseUrl.pathname)) {
|
||||
const resolved = await resolveExistingModule(baseUrl, context, nextResolve)
|
||||
if (resolved) return resolved
|
||||
}
|
||||
}
|
||||
|
||||
return nextResolve(specifier, context)
|
||||
}
|
||||
@@ -28,9 +28,11 @@ test("object bulk download resolves relative download URLs through the API clien
|
||||
|
||||
test("ApiClient redacts credential fields before development logging", () => {
|
||||
const source = fs.readFileSync("lib/api-client.ts", "utf8")
|
||||
const redactionSource = fs.readFileSync("lib/api-request-log.ts", "utf8")
|
||||
|
||||
assert.match(source, /SENSITIVE_LOG_KEY/)
|
||||
assert.match(source, /master\[_-\]\?key/)
|
||||
assert.match(source, /redactRequestOptionsForLog\(options\)/)
|
||||
assert.match(source, /JSON\.stringify\(redactLogValue\(JSON\.parse\(options\.body\)\)\)/)
|
||||
assert.match(redactionSource, /SENSITIVE_LOG_KEY/)
|
||||
assert.match(redactionSource, /master\[_-\]\?key/)
|
||||
assert.match(redactionSource, /credential\|creds/)
|
||||
assert.match(source, /redactRequestOptionsForLog\(requestOptions\)/)
|
||||
assert.match(redactionSource, /redactLogValue\(\{ \.\.\.options, body \}\)/)
|
||||
})
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
import test from "node:test"
|
||||
import assert from "node:assert/strict"
|
||||
import { redactRequestOptionsForLog } from "../../lib/api-request-log"
|
||||
|
||||
const loadApiClient = () => import(new URL("../../lib/api-client.ts", import.meta.url).href)
|
||||
|
||||
test("ApiClient does not mutate request options and preserves existing query parameters", async () => {
|
||||
const { ApiClient } = await loadApiClient()
|
||||
const urls: string[] = []
|
||||
const transport = {
|
||||
fetch: async (input: string | Request) => {
|
||||
urls.push(String(input))
|
||||
return Response.json({ ok: true })
|
||||
},
|
||||
}
|
||||
const client = new ApiClient(transport, { baseUrl: "https://console.test/api" })
|
||||
const options = { params: { page: "2" }, headers: { "X-Request": "one" } }
|
||||
|
||||
await client.get("/objects?prefix=photos", options)
|
||||
await client.get("/objects?prefix=photos", options)
|
||||
|
||||
assert.deepEqual(options, { params: { page: "2" }, headers: { "X-Request": "one" } })
|
||||
assert.deepEqual(urls, [
|
||||
"https://console.test/api/objects?prefix=photos&page=2",
|
||||
"https://console.test/api/objects?prefix=photos&page=2",
|
||||
])
|
||||
})
|
||||
|
||||
test("ApiClient rejects 401 and 403 responses after invoking global handlers", async () => {
|
||||
const { ApiClient } = await loadApiClient()
|
||||
const handled: number[] = []
|
||||
const errorHandler = {
|
||||
handle401: async () => {
|
||||
handled.push(401)
|
||||
},
|
||||
handle403: async () => {
|
||||
handled.push(403)
|
||||
},
|
||||
handleServerError: async () => {},
|
||||
handleByStatus: async () => {},
|
||||
}
|
||||
const statuses = [401, 403]
|
||||
const client = new ApiClient(
|
||||
{
|
||||
fetch: async () => {
|
||||
const status = statuses.shift() ?? 500
|
||||
return Response.json({ message: `HTTP ${status}` }, { status })
|
||||
},
|
||||
},
|
||||
{ errorHandler },
|
||||
)
|
||||
|
||||
await assert.rejects(client.get("https://console.test/unauthorized"), { message: "HTTP 401", status: 401 })
|
||||
await assert.rejects(client.get("https://console.test/forbidden"), { message: "HTTP 403", status: 403 })
|
||||
assert.deepEqual(handled, [401, 403])
|
||||
})
|
||||
|
||||
test("ApiClient redacts sensitive headers and request bodies from development logs", () => {
|
||||
const redacted = redactRequestOptionsForLog({
|
||||
headers: { Authorization: "Bearer header-secret", Cookie: "session=cookie-secret" },
|
||||
body: JSON.stringify({
|
||||
password: "body-secret",
|
||||
profile: { apiKey: "nested-secret" },
|
||||
creds: '{"private_key":"gcs-private-key"}',
|
||||
}),
|
||||
})
|
||||
|
||||
const serialized = JSON.stringify(redacted)
|
||||
assert.doesNotMatch(serialized, /body-secret|nested-secret|header-secret|cookie-secret|gcs-private-key/)
|
||||
assert.match(serialized, /\[REDACTED\]/)
|
||||
})
|
||||
@@ -0,0 +1,25 @@
|
||||
import test from "node:test"
|
||||
import assert from "node:assert/strict"
|
||||
import { DeleteObjectsCommand, ListObjectsV2Command, type S3Client } from "@aws-sdk/client-s3"
|
||||
import { createDeleteTaskHelpers } from "../../lib/delete-task"
|
||||
|
||||
test("folder deletion rejects partial DeleteObjects failures", async () => {
|
||||
const client = {
|
||||
send: async (command: unknown) => {
|
||||
if (command instanceof ListObjectsV2Command) {
|
||||
return { Contents: [{ Key: "folder/locked.txt" }], IsTruncated: false }
|
||||
}
|
||||
if (command instanceof DeleteObjectsCommand) {
|
||||
return {
|
||||
Errors: [{ Key: "folder/locked.txt", Code: "AccessDenied", Message: "Object is locked" }],
|
||||
}
|
||||
}
|
||||
throw new Error("Unexpected command")
|
||||
},
|
||||
}
|
||||
const { folderHandler, createFolderDeleteTask } = createDeleteTaskHelpers(client as unknown as S3Client)
|
||||
const task = createFolderDeleteTask("folder/", "bucket")
|
||||
|
||||
await assert.rejects(folderHandler.perform(task), /Object is locked/)
|
||||
assert.equal(task.progress, 0)
|
||||
})
|
||||
@@ -43,3 +43,10 @@ test("object list shows fixed scroll shortcut buttons only when content overflow
|
||||
true,
|
||||
)
|
||||
})
|
||||
|
||||
test("object download rejects non-success responses before exporting a blob", () => {
|
||||
const source = fs.readFileSync("components/object/list.tsx", "utf8")
|
||||
|
||||
assert.match(source, /const response = await fetch\(url\)\s+if \(!response\.ok\) throw new Error/)
|
||||
assert.match(source, /finally \{\s+loadingMsg\.destroy\(\)/)
|
||||
})
|
||||
|
||||
@@ -4,7 +4,7 @@ import {
|
||||
createObjectListScope,
|
||||
shouldApplyObjectListResponse,
|
||||
shouldResetObjectListPagination,
|
||||
} from "../../lib/object-list-state.js"
|
||||
} from "../../lib/object-list-state"
|
||||
|
||||
test("shouldResetObjectListPagination returns false for the same listing scope", () => {
|
||||
const previousScope = createObjectListScope({
|
||||
|
||||
@@ -22,3 +22,25 @@ test("browser console scope allows editing object lock controls", () => {
|
||||
assert.equal(hasConsoleCapability(browserPolicy, "objects.legalHold.edit", context), true)
|
||||
assert.equal(hasConsoleCapability(browserPolicy, "objects.retention.edit", context), true)
|
||||
})
|
||||
|
||||
test("an explicit deny overrides an implied browser capability", () => {
|
||||
const policy: ConsolePolicy = {
|
||||
...browserPolicy,
|
||||
Statement: [
|
||||
...browserPolicy.Statement,
|
||||
{
|
||||
Effect: "Deny",
|
||||
Action: ["s3:PutObjectRetention"],
|
||||
Resource: ["arn:aws:s3:::locked-bucket/folder/*"],
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
assert.equal(
|
||||
hasConsoleCapability(policy, "objects.retention.edit", {
|
||||
bucket: "locked-bucket",
|
||||
objectKey: "folder/object.txt",
|
||||
}),
|
||||
false,
|
||||
)
|
||||
})
|
||||
|
||||
@@ -24,3 +24,12 @@ test("object preview modal only uses the PDF viewer for application/pdf content"
|
||||
assert.match(source, /isPdfPreview\(normalizedContentType\)/)
|
||||
assert.doesNotMatch(source, /keyLower\.endsWith\("\.pdf"\)/)
|
||||
})
|
||||
|
||||
test("object text preview aborts stale requests and rejects non-success responses", () => {
|
||||
const source = fs.readFileSync("components/object/preview-modal.tsx", "utf8")
|
||||
|
||||
assert.match(source, /const controller = new AbortController\(\)/)
|
||||
assert.match(source, /fetch\(previewUrl, \{ signal: controller\.signal \}\)/)
|
||||
assert.match(source, /if \(!response\.ok\) throw new Error/)
|
||||
assert.match(source, /return \(\) => controller\.abort\(\)/)
|
||||
})
|
||||
|
||||
@@ -4,7 +4,7 @@ import {
|
||||
buildObjectZipDownloadPayload,
|
||||
getObjectZipDownloadFilename,
|
||||
normalizeObjectZipDownloadUrl,
|
||||
} from "../../lib/object-zip-download.js"
|
||||
} from "../../lib/object-zip-download.ts"
|
||||
|
||||
test("buildObjectZipDownloadPayload separates object keys and prefixes", () => {
|
||||
const payload = buildObjectZipDownloadPayload({
|
||||
@@ -57,3 +57,9 @@ test("normalizeObjectZipDownloadUrl resolves relative URLs against the API base
|
||||
"https://api.example.com/rustfs/admin/v3/object-zip-downloads/abc.zip?token=abc",
|
||||
)
|
||||
})
|
||||
|
||||
test("normalizeObjectZipDownloadUrl rejects active and local URL protocols", () => {
|
||||
for (const url of ["javascript:alert(1)", "data:text/html,unsafe", "file:///tmp/unsafe.zip"]) {
|
||||
assert.throws(() => normalizeObjectZipDownloadUrl(url, "https://console.example.com"), /Unsupported download/)
|
||||
}
|
||||
})
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
import test from "node:test"
|
||||
import assert from "node:assert/strict"
|
||||
import fs from "node:fs"
|
||||
|
||||
test("credential changes invalidate stale permission requests and S3 task managers", () => {
|
||||
const permissionsSource = fs.readFileSync("hooks/use-permissions.tsx", "utf8")
|
||||
const taskContextSource = fs.readFileSync("contexts/task-context.tsx", "utf8")
|
||||
|
||||
assert.match(permissionsSource, /requestEpochRef\.current \+= 1/)
|
||||
assert.match(permissionsSource, /requestEpoch !== requestEpochRef\.current/)
|
||||
assert.match(taskContextSource, /setManagerState\(null\)/)
|
||||
assert.match(taskContextSource, /return \(\) => manager\.dispose\(\)/)
|
||||
})
|
||||
|
||||
test("tier deletion is non-forced by default", () => {
|
||||
const source = fs.readFileSync("hooks/use-tiers.ts", "utf8")
|
||||
|
||||
assert.match(source, /tier\/\$\{encodeURIComponent\(name\)\}\?force=false/)
|
||||
assert.doesNotMatch(source, /removeTiers[\s\S]*?force=true/)
|
||||
})
|
||||
|
||||
test("destructive batch actions refresh after partial failures", () => {
|
||||
for (const file of ["app/(dashboard)/users/page.tsx", "app/(dashboard)/access-keys/page.tsx"]) {
|
||||
const source = fs.readFileSync(file, "utf8")
|
||||
assert.match(source, /Promise\.allSettled/)
|
||||
assert.match(source, /failures\.length === 0/)
|
||||
}
|
||||
})
|
||||
@@ -0,0 +1,303 @@
|
||||
import test from "node:test"
|
||||
import assert from "node:assert/strict"
|
||||
import { TaskManager, type ManagedTask, type TaskHandler } from "../../lib/task-manager"
|
||||
|
||||
type Status = "pending" | "running" | "completed" | "failed" | "canceled"
|
||||
type Task = ManagedTask<Status> & { kind: "work" }
|
||||
|
||||
const lifecycle = {
|
||||
pending: "pending",
|
||||
running: "running",
|
||||
completed: "completed",
|
||||
failed: "failed",
|
||||
canceled: "canceled",
|
||||
} as const
|
||||
|
||||
function createTask(id: string): Task {
|
||||
return { id, kind: "work", status: "pending", progress: 0 }
|
||||
}
|
||||
|
||||
function createManager(
|
||||
handler: TaskHandler<Task>,
|
||||
options: { maxConcurrent?: number; maxRetries?: number; retryDelay?: number } = {},
|
||||
) {
|
||||
return new TaskManager<Task>({ handlers: { work: handler }, ...options })
|
||||
}
|
||||
|
||||
async function waitFor(predicate: () => boolean) {
|
||||
for (let attempt = 0; attempt < 200; attempt += 1) {
|
||||
if (predicate()) return
|
||||
await new Promise((resolve) => setTimeout(resolve, 5))
|
||||
}
|
||||
assert.fail("Timed out waiting for task state")
|
||||
}
|
||||
|
||||
test("completes tasks and reports progress snapshots", async () => {
|
||||
const task = createTask("completed")
|
||||
let notifications = 0
|
||||
const manager = createManager({
|
||||
lifecycle,
|
||||
perform: async (current) => {
|
||||
current.progress = 45
|
||||
},
|
||||
})
|
||||
manager.subscribe(() => {
|
||||
notifications += 1
|
||||
})
|
||||
|
||||
manager.enqueue([task])
|
||||
await waitFor(() => task.status === "completed")
|
||||
|
||||
assert.equal(task.progress, 100)
|
||||
assert.ok(notifications >= 3)
|
||||
assert.deepEqual(manager.getTasks(), [task])
|
||||
})
|
||||
|
||||
test("keeps canceled tasks canceled after AbortError", async () => {
|
||||
const manager = createManager({
|
||||
lifecycle,
|
||||
perform: (task) =>
|
||||
new Promise((_, reject) => {
|
||||
task.abortController = new AbortController()
|
||||
task.abortController.signal.addEventListener("abort", () => {
|
||||
reject(new DOMException("Canceled", "AbortError"))
|
||||
})
|
||||
}),
|
||||
})
|
||||
const task = createTask("canceled")
|
||||
|
||||
manager.enqueue([task])
|
||||
await waitFor(() => task.status === "running" && !!task.abortController)
|
||||
|
||||
assert.equal(manager.cancelTask(task.id), true)
|
||||
await waitFor(() => task.status === "canceled")
|
||||
|
||||
assert.equal(task.error, undefined)
|
||||
assert.equal(manager.cancelTask(task.id), false)
|
||||
})
|
||||
|
||||
test("cancelAll cancels both running and queued tasks", async () => {
|
||||
const manager = createManager(
|
||||
{
|
||||
lifecycle,
|
||||
perform: (task) =>
|
||||
new Promise((_, reject) => {
|
||||
task.abortController = new AbortController()
|
||||
task.abortController.signal.addEventListener("abort", () => {
|
||||
reject(new DOMException("Canceled", "AbortError"))
|
||||
})
|
||||
}),
|
||||
},
|
||||
{ maxConcurrent: 1 },
|
||||
)
|
||||
const running = createTask("running")
|
||||
const queued = createTask("queued")
|
||||
|
||||
manager.enqueue([running, queued])
|
||||
await waitFor(() => running.status === "running" && queued.status === "pending")
|
||||
manager.cancelAll()
|
||||
await waitFor(() => running.status === "canceled" && queued.status === "canceled")
|
||||
|
||||
assert.deepEqual(
|
||||
manager.getTasks().map((task) => task.status),
|
||||
["canceled", "canceled"],
|
||||
)
|
||||
})
|
||||
|
||||
test("enforces maxRetries even when the handler always retries", async () => {
|
||||
let attempts = 0
|
||||
const manager = createManager(
|
||||
{
|
||||
lifecycle,
|
||||
perform: async () => {
|
||||
attempts += 1
|
||||
throw new Error("temporary")
|
||||
},
|
||||
shouldRetry: () => true,
|
||||
},
|
||||
{ maxRetries: 1, retryDelay: 0 },
|
||||
)
|
||||
const task = createTask("retry-limit")
|
||||
|
||||
manager.enqueue([task])
|
||||
await waitFor(() => task.status === "failed")
|
||||
|
||||
assert.equal(attempts, 2)
|
||||
assert.equal(task.retryCount, 1)
|
||||
assert.equal(task.error, "temporary")
|
||||
})
|
||||
|
||||
test("fails safely when an error classifier throws", async () => {
|
||||
const manager = createManager({
|
||||
lifecycle,
|
||||
perform: async () => {
|
||||
throw new Error("operation failed")
|
||||
},
|
||||
shouldRetry: () => {
|
||||
throw new Error("classifier failed")
|
||||
},
|
||||
})
|
||||
const task = createTask("classifier")
|
||||
|
||||
manager.enqueue([task])
|
||||
await waitFor(() => task.status === "failed")
|
||||
|
||||
assert.equal(task.error, "Failed to classify task error: classifier failed")
|
||||
})
|
||||
|
||||
test("releases the concurrency slot during retry backoff", async () => {
|
||||
const executionOrder: string[] = []
|
||||
const manager = createManager(
|
||||
{
|
||||
lifecycle,
|
||||
perform: async (task) => {
|
||||
executionOrder.push(`${task.id}:${task.retryCount ?? 0}`)
|
||||
if (task.id === "first" && !task.retryCount) throw new Error("retry")
|
||||
},
|
||||
},
|
||||
{ maxConcurrent: 1, maxRetries: 1, retryDelay: 40 },
|
||||
)
|
||||
const first = createTask("first")
|
||||
const second = createTask("second")
|
||||
|
||||
manager.enqueue([first, second])
|
||||
await waitFor(() => first.status === "completed" && second.status === "completed")
|
||||
|
||||
assert.deepEqual(executionOrder, ["first:0", "second:0", "first:1"])
|
||||
})
|
||||
|
||||
test("never exceeds the configured concurrency", async () => {
|
||||
let active = 0
|
||||
let peak = 0
|
||||
const releases: Array<() => void> = []
|
||||
const manager = createManager(
|
||||
{
|
||||
lifecycle,
|
||||
perform: () =>
|
||||
new Promise<void>((resolve) => {
|
||||
active += 1
|
||||
peak = Math.max(peak, active)
|
||||
releases.push(() => {
|
||||
active -= 1
|
||||
resolve()
|
||||
})
|
||||
}),
|
||||
},
|
||||
{ maxConcurrent: 2 },
|
||||
)
|
||||
const tasks = [createTask("one"), createTask("two"), createTask("three")]
|
||||
|
||||
manager.enqueue(tasks)
|
||||
await waitFor(() => releases.length === 2)
|
||||
assert.equal(peak, 2)
|
||||
releases.shift()?.()
|
||||
await waitFor(() => releases.length === 2)
|
||||
releases.splice(0).forEach((release) => release())
|
||||
await waitFor(() => tasks.every((task) => task.status === "completed"))
|
||||
|
||||
assert.equal(peak, 2)
|
||||
})
|
||||
|
||||
test("clears only finished records and refuses to remove active work", async () => {
|
||||
let finishActive: (() => void) | undefined
|
||||
const manager = createManager({
|
||||
lifecycle,
|
||||
perform: (task) => {
|
||||
if (task.id === "active") {
|
||||
return new Promise<void>((resolve) => {
|
||||
finishActive = resolve
|
||||
})
|
||||
}
|
||||
return Promise.resolve()
|
||||
},
|
||||
})
|
||||
const completed = createTask("completed")
|
||||
const active = createTask("active")
|
||||
|
||||
manager.enqueue([completed])
|
||||
await waitFor(() => completed.status === "completed")
|
||||
manager.enqueue([active])
|
||||
await waitFor(() => active.status === "running" && !!finishActive)
|
||||
|
||||
assert.equal(manager.removeFinishedTask(active.id), false)
|
||||
manager.clearFinishedTasks()
|
||||
assert.deepEqual(manager.getTasks(), [active])
|
||||
|
||||
finishActive?.()
|
||||
await waitFor(() => active.status === "completed")
|
||||
assert.equal(manager.removeFinishedTask(active.id), true)
|
||||
assert.deepEqual(manager.getTasks(), [])
|
||||
})
|
||||
|
||||
test("validates an enqueue batch before mutating the queue", () => {
|
||||
const manager = createManager({ lifecycle, perform: async () => {} })
|
||||
const valid = createTask("valid")
|
||||
const invalid = { ...createTask("invalid"), status: "running" as const }
|
||||
|
||||
assert.throws(() => manager.enqueue([valid, invalid]), /must be pending/)
|
||||
assert.deepEqual(manager.getTasks(), [])
|
||||
|
||||
assert.throws(
|
||||
() => manager.enqueue([{ ...createTask("unknown"), kind: "unknown" } as unknown as Task]),
|
||||
/No task handler registered/,
|
||||
)
|
||||
assert.deepEqual(manager.getTasks(), [])
|
||||
})
|
||||
|
||||
test("rejects duplicate task ids within and across enqueue calls", () => {
|
||||
const manager = createManager({ lifecycle, perform: async () => {} })
|
||||
|
||||
assert.throws(() => manager.enqueue([createTask("same"), createTask("same")]), /already exists/)
|
||||
assert.deepEqual(manager.getTasks(), [])
|
||||
|
||||
manager.enqueue([createTask("existing")])
|
||||
assert.throws(() => manager.enqueue([createTask("existing")]), /already exists/)
|
||||
})
|
||||
|
||||
test("dispose aborts active work, clears state, and prevents reuse", async () => {
|
||||
let aborted = false
|
||||
const manager = createManager({
|
||||
lifecycle,
|
||||
perform: (task) =>
|
||||
new Promise((_, reject) => {
|
||||
task.abortController = new AbortController()
|
||||
task.abortController.signal.addEventListener("abort", () => {
|
||||
aborted = true
|
||||
reject(new DOMException("Canceled", "AbortError"))
|
||||
})
|
||||
}),
|
||||
})
|
||||
const task = createTask("active")
|
||||
|
||||
manager.enqueue([task])
|
||||
await waitFor(() => task.status === "running" && !!task.abortController)
|
||||
manager.dispose()
|
||||
await waitFor(() => aborted)
|
||||
|
||||
assert.deepEqual(manager.getTasks(), [])
|
||||
assert.throws(() => manager.enqueue([createTask("new")]), /disposed/)
|
||||
assert.throws(() => manager.enqueue([]), /disposed/)
|
||||
})
|
||||
|
||||
test("validates scheduler options", () => {
|
||||
const handler: TaskHandler<Task> = { lifecycle, perform: async () => {} }
|
||||
|
||||
assert.throws(() => createManager(handler, { maxConcurrent: 0 }), /maxConcurrent/)
|
||||
assert.throws(() => createManager(handler, { maxRetries: -1 }), /maxRetries/)
|
||||
assert.throws(() => createManager(handler, { retryDelay: 1.5 }), /retryDelay/)
|
||||
assert.throws(() => createManager({ ...handler, maxRetries: -1 }), /handler.maxRetries/)
|
||||
assert.throws(() => createManager({ ...handler, retryDelay: 1.5 }), /handler.retryDelay/)
|
||||
})
|
||||
|
||||
test("a failing subscriber cannot stall task execution", async () => {
|
||||
const manager = createManager({ lifecycle, perform: async () => {} })
|
||||
const task = createTask("subscriber")
|
||||
manager.subscribe(() => {
|
||||
throw new Error("render failed")
|
||||
})
|
||||
|
||||
manager.enqueue([task])
|
||||
await waitFor(() => task.status === "completed")
|
||||
|
||||
assert.equal(task.progress, 100)
|
||||
})
|
||||
@@ -7,3 +7,14 @@ test("task stats drawer trigger renders the Base UI button trigger", () => {
|
||||
|
||||
assert.match(source, /<DrawerTrigger\s+render={<Button variant="outline" \/>}>/)
|
||||
})
|
||||
|
||||
test("task controls separate cancellation from record cleanup", () => {
|
||||
const itemSource = fs.readFileSync("components/tasks/item.tsx", "utf8")
|
||||
const panelSource = fs.readFileSync("components/tasks/panel.tsx", "utf8")
|
||||
const statsSource = fs.readFileSync("components/tasks/stats-button.tsx", "utf8")
|
||||
|
||||
assert.match(itemSource, /taskManager\.cancelTask\(task\.id\)/)
|
||||
assert.match(itemSource, /taskManager\.removeFinishedTask\(task\.id\)/)
|
||||
assert.match(statsSource, /taskManager\.clearFinishedTasks\(\)/)
|
||||
assert.match(panelSource, /value="canceled"/)
|
||||
})
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
import test from "node:test"
|
||||
import assert from "node:assert/strict"
|
||||
import {
|
||||
AbortMultipartUploadCommand,
|
||||
CreateMultipartUploadCommand,
|
||||
UploadPartCommand,
|
||||
type S3Client,
|
||||
} from "@aws-sdk/client-s3"
|
||||
import { createUploadTaskHelpers } from "../../lib/upload-task"
|
||||
|
||||
test("multipart cleanup clears aborted upload state before retry", async () => {
|
||||
const commands: string[] = []
|
||||
const client = {
|
||||
send: async (command: unknown) => {
|
||||
commands.push(command?.constructor.name ?? "unknown")
|
||||
if (command instanceof CreateMultipartUploadCommand) return { UploadId: "aborted-upload" }
|
||||
if (command instanceof UploadPartCommand) throw new Error("temporary network error")
|
||||
if (command instanceof AbortMultipartUploadCommand) return {}
|
||||
throw new Error("Unexpected command")
|
||||
},
|
||||
}
|
||||
const { handler, createTasks } = createUploadTaskHelpers(client as unknown as S3Client, { chunkSize: 0.000001 })
|
||||
const task = createTasks([{ file: new File(["content"], "file.txt"), key: "file.txt" }], "bucket")[0]
|
||||
assert.ok(task)
|
||||
|
||||
await assert.rejects(handler.perform(task), /temporary network error/)
|
||||
|
||||
assert.equal(task.uploadId, undefined)
|
||||
assert.equal(task.completedParts, undefined)
|
||||
assert.deepEqual(commands, ["CreateMultipartUploadCommand", "UploadPartCommand", "AbortMultipartUploadCommand"])
|
||||
})
|
||||
|
||||
test("multipart cleanup resets local state when the abort request fails", async () => {
|
||||
const client = {
|
||||
send: async (command: unknown) => {
|
||||
if (command instanceof CreateMultipartUploadCommand) return { UploadId: "uncertain-upload" }
|
||||
if (command instanceof UploadPartCommand) throw new Error("original upload error")
|
||||
if (command instanceof AbortMultipartUploadCommand) throw new Error("cleanup error")
|
||||
throw new Error("Unexpected command")
|
||||
},
|
||||
}
|
||||
const { handler, createTasks } = createUploadTaskHelpers(client as unknown as S3Client, { chunkSize: 0.000001 })
|
||||
const task = createTasks([{ file: new File(["content"], "file.txt"), key: "file.txt" }], "bucket")[0]
|
||||
assert.ok(task)
|
||||
|
||||
await assert.rejects(handler.perform(task), /original upload error/)
|
||||
|
||||
assert.equal(task.uploadId, undefined)
|
||||
assert.equal(task.completedParts, undefined)
|
||||
})
|
||||
|
||||
test("upload retry policy retries throttling but not authorization failures", () => {
|
||||
const client = { send: async () => ({}) }
|
||||
const { handler, createTasks } = createUploadTaskHelpers(client as unknown as S3Client)
|
||||
const task = createTasks([{ file: new File(["content"], "file.txt"), key: "file.txt" }], "bucket")[0]
|
||||
assert.ok(task)
|
||||
|
||||
assert.equal(handler.shouldRetry?.(task, { $metadata: { httpStatusCode: 429 } }), true)
|
||||
assert.equal(handler.shouldRetry?.(task, { $metadata: { httpStatusCode: 403 } }), false)
|
||||
})
|
||||
Reference in New Issue
Block a user