feat: terminal, eventbus, load recipe

- change: disable reboot android when logged out
- change: load recipe from android app's memory (requires ^0.0.3)
- fix: adb payload handler may get incomplete message

Signed-off-by: pakintada@gmail.com <Pakin>

src/lib/core/adb/adb.ts

@@ -11,6 +11,7 @@ import { Consumable, MaybeConsumable, ReadableStream } from '@yume-chan/stream-e
 import { AdbScrcpyClient } from '@yume-chan/adb-scrcpy';
 import { addNotification } from '../stores/noti';
 import { handleAdbPayload } from '../handlers/adbPayloadHandler';
+import { GlobalEventBus } from '../utils/eventBus';
 import { adbWriter } from '../stores/adbWriter';
 import { WritableStream } from '@yume-chan/stream-extra';
 import { env } from '$env/dynamic/public';
@@ -362,6 +363,83 @@ export async function executeCmd(command: string) {
 	}
 }
 
+/**
+ * Execute an ADB command and stream its output via callbacks.
+ * Used for commands like `logcat` that run indefinitely.
+ *
+ * Returns a cleanup function that can be called to abort the stream.
+ */
+export async function executeStreamingCmd(
+	command: string,
+	callbacks: {
+		onData?: (chunk: string) => void;
+		onError?: (error: string) => void;
+		onExit?: (exitCode: number | undefined) => void;
+	}
+): Promise<() => void> {
+	const instance = getAdbInstance();
+	let aborted = false;
+
+	if (!instance) {
+		callbacks.onError?.('No ADB device connected');
+		return () => {};
+	}
+
+	try {
+		// NOTE: Always use noneProtocol.spawn() for streaming.
+		// shellProtocol.spawnWaitText() waits for the process to exit — fine for
+		// batch commands like 'echo foo', but logcat runs indefinitely, so the
+		// Promise would never resolve and onData would never fire.
+		const process = await instance.subprocess.noneProtocol.spawn(command);
+
+		const reader = process.output.getReader();
+		const decoder = new TextDecoder();
+
+		// Start reading in the background
+		(async () => {
+			try {
+				while (!aborted) {
+					const { done, value } = await reader.read();
+					if (done) break;
+					const text = decoder.decode(value, { stream: true });
+					if (text && !aborted) {
+						callbacks.onData?.(text);
+					}
+				}
+			} catch (e: any) {
+				if (!aborted) {
+					callbacks.onError?.(e.message ?? 'Stream read error');
+				}
+			} finally {
+				if (!aborted) {
+					reader.releaseLock();
+					callbacks.onExit?.(0);
+				}
+			}
+		})();
+
+		// Return cleanup function to abort the stream
+		return () => {
+			aborted = true;
+			try {
+				reader.cancel();
+			} catch {
+				// reader may already be done
+			}
+		};
+	} catch (e: any) {
+		if (!aborted) {
+			if (e.message?.includes('ExactReadable ended')) {
+				callbacks.onError?.('Connection closed');
+			} else {
+				callbacks.onError?.(e.message ?? 'Unknown error');
+			}
+		}
+	}
+
+	return () => {};
+}
+
 export async function disconnect() {
 	let instance = getAdbInstance();
 	if (instance) {
@@ -526,12 +604,32 @@ async function connectToAndroidServer(maxRetries = 5) {
 			if (writer) {
 				addNotification('INFO:Enable Brewing Mode T on machine');
 
+				const textDecoder = new TextDecoder();
+				let buffer = '';
+
 				(async () => {
 					try {
 						while (true) {
 							const { value, done } = await reader.read();
 							if (done) break;
-							handleAdbPayload(new TextDecoder().decode(value));
+
+							// decode chunk
+							buffer += textDecoder.decode(value, { stream: true });
+
+							let lines = buffer.split('\n');
+
+							// save potential incomplete
+							buffer = lines.pop() ?? '';
+
+							for (const line of lines) {
+								if (line.trim() === '') continue;
+
+								GlobalEventBus.emit('adb:raw-payload', line);
+								handleAdbPayload(line);
+							}
+
+							// GlobalEventBus.emit('adb:raw-payload', new TextDecoder().decode(value));
+							// handleAdbPayload(new TextDecoder().decode(value));
 						}
 					} catch (e) {
 						console.error('read error', e);
@@ -637,6 +735,7 @@ async function connectToAndroidRecipeMenuServerOnce(notifyFailure = true, retryO
 						const trimmedMessage = message.trim();
 						if (trimmedMessage) {
 							console.log('[ADB Reader] Processing message:', trimmedMessage.slice(0, 200));
+							GlobalEventBus.emit('adb:raw-payload', trimmedMessage);
 							handleAdbPayload(trimmedMessage);
 						}
 					}
@@ -644,6 +743,7 @@ async function connectToAndroidRecipeMenuServerOnce(notifyFailure = true, retryO
 
 				const remainingMessage = messageBuffer.trim();
 				if (remainingMessage) {
+					GlobalEventBus.emit('adb:raw-payload', remainingMessage);
 					handleAdbPayload(remainingMessage);
 				}
 			} catch (e) {

src/lib/core/handlers/adbPayloadHandler.ts

@@ -7,16 +7,40 @@ import {
 } from '../services/androidRecipeExportService';
 import { handleIncomingMessages } from './messageHandler';
 import { setMenuSaved, setMenuSaveError } from '../stores/menuSaveStore';
-import { recipeFromMachineQuery } from '../stores/recipeStore';
+import { recipeFromMachine, recipeFromMachineQuery } from '../stores/recipeStore';
+import { buildTags, getMenuStatus } from '$lib/data/recipeService';
+import * as semver from 'semver';
+import { env } from '$env/dynamic/public';
+import { getContext } from 'svelte';
+import { GlobalEventBus, useEventBus } from '../utils/eventBus';
 
 type AdbPayload = { type: string; payload: any };
 
+let queuedPromises = new Array<Promise<void>>();
+
 async function handleAdbPayload(raw_payload: string) {
-	console.log('[ADB] Received payload:', raw_payload.slice(0, 300));
+	// console.log('[ADB] Received payload:', raw_payload.slice(0, 300));
+	const APP_VERSION = env.PUBLIC_APP_SEMVER;
+	// const bus = useEventBus();
+
 	try {
 		const payload: AdbPayload = JSON.parse(raw_payload);
-		console.log('[ADB] Parsed type:', payload.type, 'payload:', payload.payload);
-		switch (payload.type) {
+		// console.log('[ADB] Parsed type:', payload.type, 'payload:', payload.payload);
+
+		// Emit payload event for terminal drawer payload viewer
+		GlobalEventBus.emit('adb:payload', payload);
+
+		let payload_type = payload.type;
+		let sub_type = null;
+		// sub type handler
+		if (payload_type.includes('/')) {
+			//
+			let tspl = payload_type.split('/');
+			payload_type = tspl[0];
+			sub_type = tspl[1];
+		}
+
+		switch (payload_type) {
 			case 'log':
 				let log_level = payload.payload['level'] ?? 'INFO';
 				let log_message = payload.payload['msg'] ?? '';
@@ -162,10 +186,177 @@ async function handleAdbPayload(raw_payload: string) {
 						addNotification(`ERR:${error?.message ?? 'Unable to load recipe export from Android'}`)
 					);
 				break;
+			case 'get-recipe-response':
+				// only update recipe from memory of brew app
+				if (!semver.satisfies(APP_VERSION, '^0.0.3')) {
+					// reject
+					console.log('unsupported version');
+					break;
+				}
+
+				let stream_recipe_idx = -1;
+
+				try {
+					stream_recipe_idx = parseInt(sub_type ?? '-1');
+				} catch (_) {}
+
+				if (stream_recipe_idx > -1) {
+					// TODO: update both recipeFromMachine and recipeFromMachineQuery
+					//
+
+					queuedPromises.push(
+						new Promise((resolve, reject) => {
+							let recipeRawFromMachine = get(recipeFromMachine);
+							let recipeMachineQ = get(recipeFromMachineQuery);
+
+							if (!Object.keys(recipeMachineQ).includes('recipe')) {
+								recipeMachineQ = {
+									...recipeMachineQ,
+									recipe: {}
+								};
+							}
+
+							let current_r01_query = recipeMachineQ.recipe;
+
+							let rp_from_payload = JSON.parse(payload.payload);
+
+							try {
+								if (
+									recipeRawFromMachine != null &&
+									Object.keys(recipeRawFromMachine).includes('Recipe01')
+								) {
+									let is_update = false;
+									// update, assume that we already fetch
+									for (let rp of recipeRawFromMachine['Recipe01']) {
+										if (rp['productCode'] == rp_from_payload['productCode']) {
+											rp = rp_from_payload;
+											is_update = true;
+											break;
+										}
+									}
+
+									// new menu
+									if (!is_update) recipeRawFromMachine['Recipe01'].push(rp_from_payload);
+								} else {
+									// not initialize
+									recipeRawFromMachine = {
+										Recipe01: [rp_from_payload]
+									};
+								}
+
+								// build as overview style compatible
+
+								let overview_menu = {
+									productCode: rp_from_payload['productCode'] ?? '<not set>',
+									name: rp_from_payload['name']
+										? rp_from_payload['name']
+										: (rp_from_payload['otherName'] ?? '<not set>'),
+									description: rp_from_payload['desciption']
+										? rp_from_payload['desciption']
+										: (rp_from_payload['otherDescription'] ?? '<not set>'),
+									tags: buildTags(rp_from_payload),
+									status: getMenuStatus(rp_from_payload['MenuStatus'])
+								};
+
+								current_r01_query[overview_menu.productCode] = overview_menu;
+
+								recipeMachineQ = {
+									...recipeMachineQ,
+									recipe: current_r01_query
+								};
+
+								recipeFromMachineQuery.set(recipeMachineQ);
+								recipeFromMachine.set(recipeRawFromMachine);
+
+								if (stream_recipe_idx == 0) {
+									addNotification(`INFO:Loading recipes ...`);
+								}
+
+								resolve();
+							} catch (e) {
+								reject(e);
+							}
+						})
+					);
+				} else if (sub_type == 'end' && payload.payload.includes('finish')) {
+					let force_reload = payload.payload.includes('reload');
+
+					console.log('queued recipes: ', queuedPromises.length);
+					if (queuedPromises.length > 0) {
+						try {
+							await Promise.all(queuedPromises);
+							console.log('clear all recipe promises');
+							queuedPromises = new Array();
+
+							GlobalEventBus.emitUntilConsumed('recipe-event', {
+								type: 'load-recipe',
+								status: 'end',
+								reload: force_reload
+							});
+						} catch (e) {
+							console.error('some promise failed: ', e);
+
+							GlobalEventBus.emitUntilConsumed('recipe-event', {
+								type: 'load-recipe-fail',
+								status: 'end',
+								reload: force_reload
+							});
+						}
+					}
+				} else if (sub_type?.startsWith('mat')) {
+					let recipeRawFromMachine = get(recipeFromMachine);
+
+					if (sub_type == 'mat-0') {
+						addNotification('INFO:Loading materials ...');
+					}
+
+					if (
+						recipeRawFromMachine != null &&
+						Object.keys(recipeRawFromMachine).includes('MaterialSetting')
+					) {
+						recipeRawFromMachine.MaterialSetting.push(JSON.parse(payload.payload));
+					} else {
+						// not has field material yet
+						recipeRawFromMachine = {
+							...recipeRawFromMachine,
+							MaterialSetting: [JSON.parse(payload.payload)]
+						};
+					}
+
+					console.log('checking add mat', recipeRawFromMachine);
+					recipeFromMachine.set(recipeRawFromMachine);
+				} else if (sub_type == 'toppings') {
+					let recipeRawFromMachine = get(recipeFromMachine);
+
+					let topping_raw = JSON.parse(payload.payload);
+					console.log('receive topping', topping_raw);
+
+					//
+					recipeRawFromMachine = {
+						...recipeRawFromMachine,
+						Topping: topping_raw
+					};
+
+					// materialFromMachineQuery
+
+					addNotification('INFO:Loading toppings ...');
+
+					recipeFromMachine.set(recipeRawFromMachine);
+				} else {
+					// unhandled sub type
+					console.log('unhandled sub type', payload);
+				}
+
+				break;
 			default:
 		}
 	} catch (error: any) {
-		// invalid format
+		// invalid format — emit error event so listeners (e.g. terminal) can react
+		GlobalEventBus.emit('adb:payload-error', {
+			raw_payload,
+			error: error?.message ?? 'Unknown parse error',
+			timestamp: new Date().toISOString()
+		});
 	}
 }
 

src/lib/core/handlers/ws_messageSender.ts

@@ -9,7 +9,7 @@ import { env } from '$env/dynamic/public';
 
 export const queue = writable<string[]>([]);
 
-type CommandRequest = 'sheet' | 'command';
+type CommandRequest = 'sheet' | 'command' | 'upload-log';
 
 function getServiceName(cmdReq: CommandRequest) {
 	switch (cmdReq) {
@@ -17,6 +17,8 @@ function getServiceName(cmdReq: CommandRequest) {
 			return 'sheet-service';
 		case 'command':
 			return 'command';
+		case 'upload-log':
+			return 'upload-log';
 	}
 }
 

src/lib/core/stores/recipeStore.ts

@@ -51,6 +51,7 @@ export const toppingGroupFromServerQuery = writable<any>([]);
 export const latestRecipeToppingData = writable<any>([]);
 
 // edit data update
+/// NOTE: Will be obsolete in future, and replace with `EventBus` style.
 export const recipeDataEvent = writable<{
 	event_type: string;
 	payload: any;

src/lib/core/stores/terminalDrawer.ts

@@ -0,0 +1,28 @@
+import { writable } from 'svelte/store';
+
+/**
+ * Store for managing the terminal drawer open/closed state.
+ * Also holds command history and connection status.
+ */
+export const terminalDrawerOpen = writable<boolean>(false);
+
+/**
+ * Toggle the terminal drawer open/closed
+ */
+export function toggleTerminalDrawer() {
+	terminalDrawerOpen.update((v) => !v);
+}
+
+/**
+ * Open the terminal drawer
+ */
+export function openTerminalDrawer() {
+	terminalDrawerOpen.set(true);
+}
+
+/**
+ * Close the terminal drawer
+ */
+export function closeTerminalDrawer() {
+	terminalDrawerOpen.set(false);
+}

src/lib/core/stores/websocketStore.ts

@@ -81,7 +81,8 @@ export async function connectToWebsocket(id_token?: string) {
 					socket.send(
 						JSON.stringify({
 							token: id_token ?? '',
-							client_public_key: publicKeyBase64
+							client_public_key: publicKeyBase64,
+							client_version: env.PUBLIC_APP_SEMVER
 						})
 					);
 

src/lib/core/types/outMessage.ts

@@ -48,7 +48,7 @@ export type OutMessage =
 			payload: {};
 	  }
 	| {
-			type: 'sheet' | 'command';
+			type: 'sheet' | 'command' | 'upload-log';
 			payload: {
 				user_info: any;
 				srv_name: string;

src/lib/core/utils/eventBus.ts

@@ -0,0 +1,117 @@
+import { getContext, setContext } from 'svelte';
+
+const COMMON_BUS = Symbol('g-event');
+
+class EventBus {
+	#listeners = new Map();
+
+	/**
+	 * Register event with callback on this channel
+	 * @param event
+	 * @param callback
+	 * @returns unsubscribe function of this event, remove callback out of this event
+	 */
+	on(event: string, callback: any) {
+		if (!this.#listeners.has(event)) {
+			this.#listeners.set(event, new Set());
+		}
+
+		this.#listeners.get(event).add(callback);
+
+		// return unsubscribe
+		return () => this.#listeners.get(event).delete(callback);
+	}
+
+	/**
+	 * Emit data to this event, call every registered callbacks
+	 * @param event
+	 * @param data
+	 */
+	emit(event: string, data: any) {
+		if (this.#listeners.has(event)) {
+			this.#listeners.get(event).forEach((cb: any) => cb(data));
+		}
+	}
+
+	emitUntilConsumed(event: string, data: any, timeout?: number) {
+		if (this.#listeners.has(event)) {
+			let listener_count = this.#listeners.get(event).length;
+
+			if (listener_count == 0) {
+				setTimeout(
+					() => {
+						this.emitUntilConsumed(event, data);
+					},
+					(timeout ?? 1) * 1000
+				);
+			} else {
+				this.emit(event, data);
+			}
+		}
+	}
+
+	/**
+	 * Clear all listeners
+	 */
+	clear() {
+		this.#listeners.clear();
+	}
+
+	/**
+	 * Clear all callbacks on this event
+	 * @param event
+	 */
+	resetCallbackOnEvent(event: string) {
+		if (this.#listeners.has(event)) {
+			this.#listeners.set(event, new Set());
+		}
+	}
+}
+
+/**
+ * Initialize the common channel event bus
+ * @returns EventBus | undefined
+ */
+function setEventBus(): EventBus | undefined {
+	return setContext(COMMON_BUS, new EventBus());
+}
+
+/**
+ * Get common channel event bus, cannot be used in non-component
+ * @returns EventBus | undefined
+ */
+function useEventBus(): EventBus | undefined {
+	return getContext(COMMON_BUS);
+}
+
+/**
+ * Initialize the channel with name event bus
+ * @param name channel name
+ * @returns EventBus | undefined
+ */
+function setEventBusWithName(name: string): EventBus | undefined {
+	return setContext(name, new EventBus());
+}
+
+/**
+ * Get a specific channel event bus, cannot be used in non-component
+ * @param name channel name
+ * @returns EventBus | undefined
+ */
+function useEventBusWithName(name: string): EventBus | undefined {
+	return getContext(name);
+}
+
+/**
+ * Global type event bus, allow use without Svelte context
+ */
+const GlobalEventBus = new EventBus();
+
+export {
+	setEventBus,
+	useEventBus,
+	setEventBusWithName,
+	useEventBusWithName,
+	COMMON_BUS,
+	GlobalEventBus
+};