[+] Streaming post in frontend

AquaNet/src/libs/sdk.ts

@@ -8,13 +8,14 @@ import type {
   TrendEntry,
   AquaNetUser, GameOption,
   UserBox,
-  UserItem
+  UserItem,
+  Dict
 } from './generalTypes'
 import type { GameName } from './scoring'
 
-interface RequestInitWithParams extends RequestInit {
+interface ExtReqInit extends RequestInit {
   params?: { [index: string]: string }
-  localCache?: boolean
+  json?: any
 }
 
 /**
@@ -37,33 +38,18 @@ export function reconstructUrl(input: URL | RequestInfo, callback: (url: URL) =>
 /**
  * Fetch with url parameters
  */
-export function fetchWithParams(input: URL | RequestInfo, init?: RequestInitWithParams): Promise<Response> {
+export function fetchWithParams(input: URL | RequestInfo, init?: ExtReqInit): Promise<Response> {
   return fetch(reconstructUrl(input, u => {
     u.search = new URLSearchParams(init?.params ?? {}).toString()
   }), init)
 }
 
-const cache: { [index: string]: any } = {}
+/**
-
+ * Do something with the response when it's not ok
-export async function post(endpoint: string, params: Record<string, any> = {}, init?: RequestInitWithParams): Promise<any> {
+ *
-  // Add token if exists
+ * @param res Response object
-  const token = localStorage.getItem('token')
+ */
-  if (token && !('token' in params)) params = { ...(params ?? {}), token }
+async function ensureOk(res: Response) {
-
-  if (init?.localCache) {
-    const cached = cache[endpoint + JSON.stringify(params) + JSON.stringify(init)]
-    if (cached) return cached
-  }
-
-  const res = await fetchWithParams(AQUA_HOST + endpoint, {
-    method: 'POST',
-    params,
-    ...init
-  }).catch(e => {
-    console.error(e)
-    throw new Error('Network error')
-  })
-
   if (!res.ok) {
     const text = await res.text()
     console.error(`${res.status}: ${text}`)
@@ -83,11 +69,82 @@ export async function post(endpoint: string, params: Record<string, any> = {}, i
     }
     if (json.error) throw new Error(json.error)
   }
+}
 
-  const ret = res.json()
+/**
-  cache[endpoint + JSON.stringify(params) + JSON.stringify(init)] = ret
+ * Post to an endpoint and return the response in JSON while doing error checks
+ * and handling token (and token expiry) automatically.
+ *
+ * @param endpoint The endpoint to post to (e.g., '/pull')
+ * @param params An object containing the request body or any necessary parameters
+ * @param init Additional fetch/init configuration
+ * @returns The JSON response from the server
+ */
+export async function post(endpoint: string, params: Dict = {}, init?: ExtReqInit): Promise<any> {
+  return postHelper(endpoint, params, init).then(it => it.json())
+}
 
-  return ret
+/**
+ * Actual impl of post(). This does not return JSON but returns response object.
+ */
+async function postHelper(endpoint: string, params: Dict = {}, init?: ExtReqInit): Promise<any> {
+  // Add token if exists
+  const token = localStorage.getItem('token')
+  if (token && !('token' in params)) params = { ...(params ?? {}), token }
+
+  if (init?.json) {
+    init.body = JSON.stringify(init.json)
+    init.headers = { 'Content-Type': 'application/json', ...init.headers }
+    init.json = undefined
+  }
+
+  const res = await fetchWithParams(AQUA_HOST + endpoint, { method: 'POST', params, ...init })
+    .catch(e => { console.error(e); throw new Error("Network error") })
+  await ensureOk(res)
+
+  return res
+}
+
+const decoder = new TextDecoder()
+
+/**
+ * Post with a stream response. Similar to post(), but the response will stream messages to onChunk.
+ */
+export async function postStream(endpoint: string, params: Dict = {}, onChunk: (data: any) => void, init?: ExtReqInit): Promise<void> {
+  const res = await postHelper(endpoint, params, init)
+  if (!res.body) {
+    console.error('Response body is not a stream')
+    return
+  }
+
+  // The response body is a ReadableStream. We'll read chunks as they arrive.
+  const reader = res.body?.getReader()
+  if (!reader) return
+  let buffer = ''
+
+  try {
+    while (true) {
+      const { done, value } = await reader.read()
+      if (done) break
+
+      // Decode any new data, parse full lines, keep the rest in buffer
+      buffer += decoder.decode(value, { stream: true })
+      let fullLines = buffer.split('\n')
+      buffer = fullLines.pop() ?? ''
+
+      for (const line of fullLines) {
+        if (!line.trim()) continue // skip empty lines
+        onChunk(JSON.parse(line))
+      }
+    }
+
+    // If there's leftover data in 'buffer' after stream ends, parse
+    if (buffer.trim())
+      onChunk(JSON.parse(buffer.trim()))
+
+  } finally {
+    reader.releaseLock()
+  }
 }
 
 /**
@@ -98,6 +155,7 @@ export async function post(endpoint: string, params: Record<string, any> = {}, i
 async function register(user: { username: string, email: string, password: string, turnstile: string }) {
   return await post('/api/v2/user/register', user)
 }
+
 async function login(user: { email: string, password: string, turnstile: string }) {
   const data = await post('/api/v2/user/login', user)
 
@@ -164,9 +222,9 @@ export const GAME = {
   export: (game: GameName): Promise<Record<string, any>> =>
     post(`/api/v2/game/${game}/export`),
   import: (game: GameName, data: any): Promise<Record<string, any>> =>
-    post(`/api/v2/game/${game}/import`, {}, { body: JSON.stringify(data) }),
+    post(`/api/v2/game/${game}/import`, {}, { json: data }),
   importMusicDetail: (game: GameName, data: any): Promise<Record<string, any>> =>
-    post(`/api/v2/game/${game}/import-music-detail`, {}, {body: JSON.stringify(data), headers: {'Content-Type': 'application/json'}}),
+    post(`/api/v2/game/${game}/import-music-detail`, {}, { json: data }),
   setRival: (game: GameName, rivalUserName: string, isAdd: boolean) =>
     post(`/api/v2/game/${game}/set-rival`, { rivalUserName, isAdd }),
 }
@@ -189,6 +247,9 @@ export const SETTING = {
 
 export const TRANSFER = {
   check: (d: AllNetClient): Promise<TrCheckGood> =>
-    post('/api/v2/transfer/check', {}, { body: JSON.stringify(d), headers: { 'Content-Type': 'application/json' } }),
+    post('/api/v2/transfer/check', {}, { json: d }),
-
+  pull: (d: AllNetClient, callback: (data: TrStreamMessage) => void) =>
+    postStream('/api/v2/transfer/pull', {}, callback, { json: d }),
+  push: (d: AllNetClient, data: string) =>
+    post('/api/v2/transfer/push', {}, { json: { client: d, data } }),
 }