From 718f5ba514101dbd29d4c6feb37a33c2023cf4c6 Mon Sep 17 00:00:00 2001 From: DecDuck Date: Thu, 10 Oct 2024 16:13:10 +1100 Subject: [PATCH] task API The Task API allows for an easy way to create long-lived tasks that require reporting back to user with progress/logs. It will be used in the upcoming game importing. --- composables/task.ts | 64 +++++++++++++ layouts/default.vue | 3 + nuxt.config.ts | 6 ++ package.json | 2 +- prisma/schema.prisma | 15 ++++ server/api/v1/task/index.get.ts | 44 +++++++++ server/internal/library/README.md | 11 +++ server/internal/tasks/index.ts | 145 ++++++++++++++++++++++++++++++ 8 files changed, 289 insertions(+), 1 deletion(-) create mode 100644 composables/task.ts create mode 100644 server/api/v1/task/index.get.ts create mode 100644 server/internal/library/README.md create mode 100644 server/internal/tasks/index.ts diff --git a/composables/task.ts b/composables/task.ts new file mode 100644 index 0000000..e508b3a --- /dev/null +++ b/composables/task.ts @@ -0,0 +1,64 @@ +import type { TaskMessage } from "~/server/internal/tasks"; + +let ws: WebSocket | undefined; +const msgQueue: Array = []; + +const useTaskStates = () => + useState<{ [key: string]: Ref }>("task-states", () => ({ + connect: useState("task-connect", () => ({ + id: "connect", + success: false, + progress: 0, + log: [], + error: undefined, + })), + })); + +function initWs() { + const isSecure = location.protocol === "https:"; + const url = (isSecure ? "wss://" : "ws://") + location.host + "/api/v1/task"; + ws = new WebSocket(url); + ws.onmessage = (e) => { + const msg = JSON.parse(e.data) as TaskMessage; + console.log(msg); + const taskStates = useTaskStates(); + const state = taskStates.value[msg.id]; + if (!state) return; + state.value = msg; + }; + ws.onopen = () => { + for (const message of msgQueue) { + ws?.send(message); + } + }; + + return ws; +} + +function sendMessage(msg: string) { + if (!ws) return msgQueue.push(msg); + if (ws.readyState == 0) return msgQueue.push(msg); + return ws.send(msg); +} + +export const useTaskReady = () => { + const taskStates = useTaskStates(); + return taskStates.value["connect"]; +}; + +export const useTask = (taskId: string): Ref => { + if (import.meta.server) return {} as unknown as Ref; + const taskStates = useTaskStates(); + if (taskStates.value[taskId]) return taskStates.value[taskId]; + + if (!ws) initWs(); + taskStates.value[taskId] = useState(`task-${taskId}`, () => ({ + id: taskId, + success: false, + progress: 0, + error: undefined, + log: [], + })); + sendMessage(`connect/${taskId}`); + return taskStates.value[taskId]; +}; diff --git a/layouts/default.vue b/layouts/default.vue index 78a45f8..078a6b9 100644 --- a/layouts/default.vue +++ b/layouts/default.vue @@ -3,6 +3,7 @@
+ {{ test }}
@@ -15,4 +16,6 @@ useHead({ return `Drop`; }, }); + +const test = useTask("test"); diff --git a/nuxt.config.ts b/nuxt.config.ts index 5833054..f031654 100644 --- a/nuxt.config.ts +++ b/nuxt.config.ts @@ -16,4 +16,10 @@ export default defineNuxtConfig({ link: [{ rel: "icon", href: "/favicon.ico" }], }, }, + + nitro: { + experimental: { + websocket: true, + }, + }, }); diff --git a/package.json b/package.json index d1917d8..d78ae2c 100644 --- a/package.json +++ b/package.json @@ -35,7 +35,7 @@ "@types/turndown": "^5.0.5", "@types/uuid": "^10.0.0", "autoprefixer": "^10.4.20", - "h3": "^1.12.0", + "nitropack": "^2.9.7", "postcss": "^8.4.47", "sass": "^1.79.4", "tailwindcss": "^3.4.13" diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4e71eb3..f091ed4 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -86,9 +86,24 @@ model Game { mArt String[] // linked to objects in s3 mScreenshots String[] // linked to objects in s3 + versionOrder String + versions GameVersion[] + libraryBasePath String // Base dir for all the game versions + @@unique([metadataSource, metadataId], name: "metadataKey") } +// A particular set of files that relate to the version +model GameVersion { + gameId String + game Game @relation(fields: [gameId], references: [id]) + versionName String // Sub directory for the game files + + + + @@id([gameId, versionName]) +} + model Developer { id String @id @default(uuid()) diff --git a/server/api/v1/task/index.get.ts b/server/api/v1/task/index.get.ts new file mode 100644 index 0000000..7420aba --- /dev/null +++ b/server/api/v1/task/index.get.ts @@ -0,0 +1,44 @@ +import { H3Event } from "h3"; +import session from "~/server/internal/session"; +import { v4 as uuidv4 } from "uuid"; +import taskHandler, { TaskMessage } from "~/server/internal/tasks"; + +export default defineWebSocketHandler({ + open(peer) { + const dummyEvent = { + node: { + req: { + headers: peer.headers, + }, + }, + } as unknown as H3Event; + const userId = session.getUserId(dummyEvent); + if (!userId) { + peer.send("unauthenticated"); + return; + } + const peerId = uuidv4(); + peer.ctx.id = peerId; + + const rtMsg: TaskMessage = { + id: "connect", + success: true, + progress: 0, + error: undefined, + log: [], + }; + peer.send(rtMsg); + }, + message(peer, message) { + if (!peer.ctx.id) return; + const text = message.text(); + if (text.startsWith("connect/")) { + const id = text.substring("connect/".length); + taskHandler.connect(peer.ctx.id, id, peer); + return; + } + }, + close(peer, details) { + if (!peer.ctx.id) return; + }, +}); diff --git a/server/internal/library/README.md b/server/internal/library/README.md new file mode 100644 index 0000000..e33b8ad --- /dev/null +++ b/server/internal/library/README.md @@ -0,0 +1,11 @@ +# Library Format + +Drop uses a filesystem-based library format, as it targets homelabs and not enterprise-grade solutions. The format works as follows: + +## /{game name} + +The game name is only used for initial matching, and doesn't affect actual metadata. Metadata is linked to the game's database entry, which is linked to it's filesystem name (they, however, can be completely different). + +## /{game name}/{version name} + +The version name can be anything. Versions have to manually imported within the web UI. There, you can change the order of the updates and mark them as deltas. Delta updates apply files over the previous versions. \ No newline at end of file diff --git a/server/internal/tasks/index.ts b/server/internal/tasks/index.ts new file mode 100644 index 0000000..1384489 --- /dev/null +++ b/server/internal/tasks/index.ts @@ -0,0 +1,145 @@ +/** + * The TaskHandler setups up two-way connections to web clients and manages the state for them + * This allows long-running tasks (like game imports and such) to report progress, success and error states + * easily without re-inventing the wheel every time. + */ + +type TaskRegistryEntry = { + runPromise: Promise; + success: boolean; + progress: number; + log: string[]; + error: string | undefined; + clients: { [key: string]: boolean }; + name: string; +}; + +class TaskHandler { + private taskRegistry: { [key: string]: TaskRegistryEntry } = {}; + private clientRegistry: { [key: string]: PeerImpl } = {}; + + constructor() {} + + create(task: Task) { + let updateCollectTimeout: NodeJS.Timeout | undefined; + + const updateAllClients = () => { + if (updateCollectTimeout) return; + updateCollectTimeout = setTimeout(() => { + const taskEntry = this.taskRegistry[task.id]; + if (!taskEntry) return; + const taskMessage: TaskMessage = { + id: task.id, + success: taskEntry.success, + progress: taskEntry.progress, + error: taskEntry.error, + log: taskEntry.log, + }; + for (const client of Object.keys(taskEntry.clients)) { + if (!this.clientRegistry[client]) continue; + this.clientRegistry[client].send(taskMessage); + } + updateCollectTimeout = undefined; + }, 500); + }; + + const progress = (progress: number) => { + const taskEntry = this.taskRegistry[task.id]; + if (!taskEntry) return; + this.taskRegistry[task.id].progress = progress; + updateAllClients(); + }; + + const log = (entry: string) => { + const taskEntry = this.taskRegistry[task.id]; + if (!taskEntry) return; + this.taskRegistry[task.id].log.push(entry); + updateAllClients(); + }; + + const promiseRun = task.run({ progress, log }); + promiseRun.then(() => { + const taskEntry = this.taskRegistry[task.id]; + if (!taskEntry) return; + this.taskRegistry[task.id].success = true; + updateAllClients(); + }); + promiseRun.catch((error) => { + const taskEntry = this.taskRegistry[task.id]; + if (!taskEntry) return; + this.taskRegistry[task.id].success = false; + this.taskRegistry[task.id].error = error; + updateAllClients(); + }); + this.taskRegistry[task.id] = { + name: task.name, + runPromise: promiseRun, + success: false, + progress: 0, + error: undefined, + log: [], + clients: {}, + }; + } + + connect(id: string, taskId: string, peer: PeerImpl) { + const task = this.taskRegistry[taskId]; + if (!task) return false; + + this.clientRegistry[id] = peer; + this.taskRegistry[taskId].clients[id] = true; // Uniquely insert client to avoid sending duplicate traffic + + const catchupMessage: TaskMessage = { + id: taskId, + success: task.success, + error: task.error, + log: task.log, + progress: task.progress, + }; + peer.send(catchupMessage); + + return true; + } + + disconnect(id: string, taskId: string) { + if (!this.taskRegistry[taskId]) return false; + + delete this.taskRegistry[taskId].clients[id]; + + const allClientIds = Object.values(this.taskRegistry) + .map((_) => Object.keys(_.clients)) + .flat(); + + if (!allClientIds.includes(id)) { + delete this.clientRegistry[id]; + } + + return true; + } +} + +export type TaskRunContext = { + progress: (progress: number) => void; + log: (message: string) => void; +}; + +export interface Task { + id: string; + name: string; + run: (context: TaskRunContext) => Promise; +} + +export type TaskMessage = { + id: string; + success: boolean; + progress: number; + error: undefined | string; + log: string[]; +}; + +export type PeerImpl = { + send: (message: TaskMessage) => void; +}; + +export const taskHandler = new TaskHandler(); +export default taskHandler;