Merge branch 'master' into dev

Resolved conflicts:
- adb.ts: keep both goToMachineHome() (dev) and executeStreamingCmd() (master)
- messageHandler.ts / ws_messageSender.ts: adopt master's semver.satisfies
  version check (forward-correct vs startsWith('0.0.2')); keep dev's additive
  bits (raw_stream_end_priceslot handler, ciphertext/iv guard) + master's
  (announce handler, 'upload-log' command); drop now-unused isSecuredAppVersion
- AnnouncementDialog.svelte: drop invalid export on interface (Svelte 5)
- install @types/semver + semver (bun.lock)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
thanawat saiyota 2026-06-30 17:55:37 +07:00
commit 7a406fd409
20 changed files with 2615 additions and 57 deletions

View file

@ -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';
@ -371,6 +372,83 @@ export async function goToMachineHome() {
}
}
/**
* 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) {
@ -535,12 +613,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);
@ -646,6 +744,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);
}
}
@ -653,6 +752,7 @@ async function connectToAndroidRecipeMenuServerOnce(notifyFailure = true, retryO
const remainingMessage = messageBuffer.trim();
if (remainingMessage) {
GlobalEventBus.emit('adb:raw-payload', remainingMessage);
handleAdbPayload(remainingMessage);
}
} catch (e) {

File diff suppressed because it is too large Load diff

View file

@ -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()
});
}
}

View file

@ -52,6 +52,8 @@ import { v4 as uuidv4 } from 'uuid';
import { handleSheetResponseFromNoti } from './sheetNotiHandler';
import { env } from '$env/dynamic/public';
import { WebCryptoHelper } from '../utils/crypto';
import { GlobalEventBus } from '../utils/eventBus';
import * as semver from 'semver';
export const messages = writable<string[]>([]);
@ -552,13 +554,13 @@ const handlers: Record<string, (payload: any) => void> = {
},
raw_stream_end_priceslot: (p) => {
handleRawStreamEnd('priceslot', p);
},
announce: (p) => {
// Server-pushed announcement (e.g., closing maintenance)
GlobalEventBus.emit('announce', p);
}
};
function isSecuredAppVersion(version: string | undefined) {
return version?.startsWith('0.0.2') ?? false;
}
export async function handleIncomingMessages(raw: string, clientPrivateKey?: CryptoKey) {
const APP_VERSION = env.PUBLIC_APP_SEMVER;
const parsedMessage = JSON.parse(raw);
@ -574,7 +576,7 @@ export async function handleIncomingMessages(raw: string, clientPrivateKey?: Cry
return;
}
if (isSecuredAppVersion(APP_VERSION) && parsedMessage.ciphertext && parsedMessage.iv) {
if (semver.satisfies(APP_VERSION, '>=0.0.2') && parsedMessage.ciphertext && parsedMessage.iv) {
// secured message decryption
let sharedKeyStore = get(sharedKey);
if (sharedKeyStore) {

View file

@ -5,14 +5,11 @@ import { addNotification } from '../stores/noti';
import { auth } from '../stores/auth';
import { WebCryptoHelper } from '../utils/crypto';
import { env } from '$env/dynamic/public';
import * as semver from 'semver';
export const queue = writable<string[]>([]);
function isSecuredAppVersion(version: string | undefined) {
return version?.startsWith('0.0.2') ?? false;
}
type CommandRequest = 'sheet' | 'command';
type CommandRequest = 'sheet' | 'command' | 'upload-log';
function getServiceName(cmdReq: CommandRequest) {
switch (cmdReq) {
@ -20,6 +17,8 @@ function getServiceName(cmdReq: CommandRequest) {
return 'sheet-service';
case 'command':
return 'command';
case 'upload-log':
return 'upload-log';
}
}
@ -108,8 +107,8 @@ export async function sendMessage(
// console.log('send v2', APP_VERSION, isSecuredAppVersion(APP_VERSION));
if (isSecuredAppVersion(APP_VERSION)) {
console.log('sending secured');
if (semver.satisfies(APP_VERSION, '>=0.0.2')) {
// console.log('sending secured');
let sharedKeyRes = get(sharedKey);
// do encrypt

View file

@ -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;

View file

@ -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);
}

View file

@ -108,7 +108,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
})
);

View file

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

View file

@ -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
};