admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor Gmail email handling and improve HTTP client management

Browse Source 
- Updated `gmailSlice` to replace `GmailEmailSummary` with `GmailEmailBatch`, enhancing email data structure.
- Modified `SkillProvider` to dispatch emails as `GmailEmailBatch | null`.
- Improved HTTP client management in `ops_net.rs` by implementing a shared client for fetch calls, optimizing connection handling and timeout management.
- Enhanced deep link handling in `desktopDeepLinkListener.ts` to trigger synchronization after OAuth completion.
This commit is contained in:
M3gA-Mind
2026-03-14 04:33:03 +05:30
parent a2a81ee9c1
commit 5d2fbc9e47
5 changed files with 183 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
package.json
+1 -1
View File
@@ -12,7 +12,7 @@
"compile": "tsc --noEmit",
"preview": "vite preview",
"tauri": "tauri",
"tauri:dev": "source scripts/load-dotenv.sh 2>/dev/null; (cd src-tauri && bash scripts/download-tdlib.sh) && LOCAL_TDLIB_PATH=$(pwd)/src-tauri/tdlib-cache/macos-$(uname -m | sed 's/arm64/aarch64/; s/x86_64/x86_64/') tauri dev",
"tauri:dev": "RUST_LOG=debug source scripts/load-dotenv.sh 2>/dev/null; (cd src-tauri && bash scripts/download-tdlib.sh) && LOCAL_TDLIB_PATH=$(pwd)/src-tauri/tdlib-cache/macos-$(uname -m | sed 's/arm64/aarch64/; s/x86_64/x86_64/') tauri dev",
"macos:build:debug": "yarn macos:build:release --debug",
"macos:build:release": "source scripts/load-dotenv.sh 2>/dev/null; (cd src-tauri && bash scripts/download-tdlib.sh) && LOCAL_TDLIB_PATH=$(pwd)/src-tauri/tdlib-cache/macos-$(uname -m | sed 's/arm64/aarch64/; s/x86_64/x86_64/') tauri build --bundles app dmg",
"macos:build:sign:release": "source scripts/load-env.sh && tauri build --bundles app dmg",
src-tauri/src/services/quickjs-libs/qjs_ops/ops_net.rs
+138 -85
View File
@@ -3,84 +3,123 @@
use parking_lot::RwLock;
use rquickjs::{function::Async, Ctx, Function, Object};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::{Arc, OnceLock};
use super::types::{js_err, WebSocketConnection, WebSocketState};
pub fn register<'js>(ctx: &Ctx<'js>, ops: &Object<'js>, ws_state: Arc<RwLock<WebSocketState>>) -> rquickjs::Result<()> {
/// Shared HTTP client — built once, reused across all fetch calls.
/// Using a shared client enables connection pooling, persistent TLS sessions,
/// and prevents per-request TLS handshake overhead that can cause hangs.
static HTTP_CLIENT: OnceLock<reqwest::Client> = OnceLock::new();
fn get_http_client() -> &'static reqwest::Client {
HTTP_CLIENT.get_or_init(|| {
reqwest::Client::builder()
.use_rustls_tls()
.connect_timeout(std::time::Duration::from_secs(10))
.pool_idle_timeout(std::time::Duration::from_secs(90))
.pool_max_idle_per_host(10)
.build()
.expect("failed to build shared HTTP client")
})
}
pub fn register<'js>(
ctx: &Ctx<'js>,
ops: &Object<'js>,
ws_state: Arc<RwLock<WebSocketState>>,
) -> rquickjs::Result<()> {
// ========================================================================
// Fetch (1) - ASYNC
// ========================================================================
ops.set("fetch", Function::new(ctx.clone(),
Async(move |url: String, options: String| async move {
let opts: serde_json::Value =
serde_json::from_str(&options).map_err(|e| js_err(e.to_string()))?;
ops.set(
"fetch",
Function::new(
ctx.clone(),
Async(move |url: String, options: String| async move {
let opts: serde_json::Value =
serde_json::from_str(&options).map_err(|e| js_err(e.to_string()))?;
let method = opts["method"].as_str().unwrap_or("GET");
let headers_obj = opts["headers"].as_object();
let body = opts["body"].as_str();
let timeout_secs = opts["timeout"]
.as_u64()
.or_else(|| opts["timeout"].as_f64().map(|f| f as u64))
.unwrap_or(30);
let method = opts["method"].as_str().unwrap_or("GET");
let headers_obj = opts["headers"].as_object();
let body = opts["body"].as_str();
let timeout_secs = opts["timeout"]
.as_u64()
.or_else(|| opts["timeout"].as_f64().map(|f| f as u64))
.unwrap_or(30);
let client = reqwest::Client::builder()
.use_rustls_tls()
.build()
.map_err(|e| js_err(e.to_string()))?;
let mut req = match method {
"GET" => client.get(&url),
"POST" => client.post(&url),
"PUT" => client.put(&url),
"PATCH" => client.patch(&url),
"DELETE" => client.delete(&url),
_ => client.get(&url),
};
let client = get_http_client();
let mut req = match method {
"GET" => client.get(&url),
"POST" => client.post(&url),
"PUT" => client.put(&url),
"PATCH" => client.patch(&url),
"DELETE" => client.delete(&url),
_ => client.get(&url),
};
req = req.timeout(std::time::Duration::from_secs(timeout_secs));
req = req.timeout(std::time::Duration::from_secs(timeout_secs));
if let Some(h) = headers_obj {
for (k, v) in h {
if let Some(val_str) = v.as_str() {
req = req.header(k, val_str);
if let Some(h) = headers_obj {
for (k, v) in h {
if let Some(val_str) = v.as_str() {
req = req.header(k, val_str);
}
}
}
}
if let Some(b) = body {
req = req.body(b.to_string());
}
let response = req.send().await.map_err(|e| {
let mut msg = e.to_string();
let mut source = std::error::Error::source(&e);
while let Some(cause) = source {
msg.push_str(&format!(" | caused by: {cause}"));
source = std::error::Error::source(cause);
if let Some(b) = body {
req = req.body(b.to_string());
}
js_err(msg)
})?;
let status = response.status().as_u16();
let status_text = response.status().canonical_reason().unwrap_or("").to_string();
let headers: HashMap<String, String> = response
.headers()
.iter()
.map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
.collect();
let body_text = response.text().await.map_err(|e| js_err(e.to_string()))?;
// Hard safety net: tokio timeout wraps send+body read so a stalled
// connection cannot block the QuickJS event loop indefinitely even
// if the per-request timeout on the reqwest builder fails to fire.
let total_deadline = std::time::Duration::from_secs(timeout_secs + 5);
let response = tokio::time::timeout(total_deadline, req.send())
.await
.map_err(|_| js_err(format!("request timed out after {}s", timeout_secs + 5)))?
.map_err(|e| {
let mut msg = e.to_string();
let mut source = std::error::Error::source(&e);
while let Some(cause) = source {
msg.push_str(&format!(" | caused by: {cause}"));
source = std::error::Error::source(cause);
}
js_err(msg)
})?;
let result = serde_json::json!({
"status": status,
"statusText": status_text,
"headers": headers,
"body": body_text,
});
let status = response.status().as_u16();
let status_text = response
.status()
.canonical_reason()
.unwrap_or("")
.to_string();
let headers: HashMap<String, String> = response
.headers()
.iter()
.map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
.collect();
let body_text = tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs + 5),
response.text(),
)
.await
.map_err(|_| js_err(format!("body read timed out after {}s", timeout_secs + 5)))?
.map_err(|e| js_err(e.to_string()))?;
Ok::<String, rquickjs::Error>(result.to_string())
}),
))?;
let result = serde_json::json!({
"status": status,
"statusText": status_text,
"headers": headers,
"body": body_text,
});
Ok::<String, rquickjs::Error>(result.to_string())
}),
),
)?;
// ========================================================================
// WebSocket (4) - placeholders
@@ -88,43 +127,57 @@ pub fn register<'js>(ctx: &Ctx<'js>, ops: &Object<'js>, ws_state: Arc<RwLock<Web
{
let ws = ws_state.clone();
ops.set("ws_connect", Function::new(ctx.clone(),
Async(move |url: String| {
let ws = ws.clone();
async move {
let mut state = ws.write();
let id = state.next_id;
state.next_id += 1;
state.connections.insert(id, WebSocketConnection { url });
Ok::<u32, rquickjs::Error>(id)
}
}),
))?;
ops.set(
"ws_connect",
Function::new(
ctx.clone(),
Async(move |url: String| {
let ws = ws.clone();
async move {
let mut state = ws.write();
let id = state.next_id;
state.next_id += 1;
state.connections.insert(id, WebSocketConnection { url });
Ok::<u32, rquickjs::Error>(id)
}
}),
),
)?;
}
{
let ws = ws_state.clone();
ops.set("ws_send", Function::new(ctx.clone(), move |_id: u32, _data: String| {
let _state = ws.read();
}))?;
ops.set(
"ws_send",
Function::new(ctx.clone(), move |_id: u32, _data: String| {
let _state = ws.read();
}),
)?;
}
{
let ws = ws_state.clone();
ops.set("ws_recv", Function::new(ctx.clone(),
Async(move |_id: u32| {
let _ws = ws.clone();
async move { Ok::<Option<String>, rquickjs::Error>(None) }
}),
))?;
ops.set(
"ws_recv",
Function::new(
ctx.clone(),
Async(move |_id: u32| {
let _ws = ws.clone();
async move { Ok::<Option<String>, rquickjs::Error>(None) }
}),
),
)?;
}
{
let ws = ws_state;
ops.set("ws_close", Function::new(ctx.clone(), move |id: u32, _code: u16, _reason: String| {
let mut state = ws.write();
state.connections.remove(&id);
}))?;
ops.set(
"ws_close",
Function::new(ctx.clone(), move |id: u32, _code: u16, _reason: String| {
let mut state = ws.write();
state.connections.remove(&id);
}),
)?;
}
Ok(())
}
src/providers/SkillProvider.tsx
+2 -6
View File
@@ -20,7 +20,7 @@ import { skillManager } from '../lib/skills/manager';
import type { SkillManifest } from '../lib/skills/types';
import { buildManualSentryEvent, enqueueError } from '../services/errorReportQueue';
import {
GmailEmailSummary,
GmailEmailBatch,
type GmailProfile,
setGmailEmails,
setGmailProfile,
@@ -100,11 +100,7 @@ function syncGmailStateToSlice(
: null
)
);
dispatch(
setGmailEmails(
Array.isArray(gmailState.emails) ? (gmailState.emails as GmailEmailSummary[]) : []
)
);
dispatch(setGmailEmails(gmailState.emails as GmailEmailBatch | null));
syncGmailMetadataToBackend(gmailState as GmailStateForSync);
}
src/store/gmailSlice.ts
+28 -10
View File
@@ -1,12 +1,30 @@
import { createSlice, type PayloadAction } from '@reduxjs/toolkit';
export interface GmailEmailSummary {
id: string;
export interface GmailEmailEntity {
identifier: string;
kind: 'sender' | 'recipient' | 'recipient_cc' | string;
name: string;
}
export interface GmailEmailMetadata {
emailId: string;
threadId: string;
snippet?: string;
subject?: string;
from?: string;
date?: string;
date: number;
}
export interface GmailEmailChunk {
content: string;
entities: GmailEmailEntity[];
labels: string[];
metadata: GmailEmailMetadata;
title: string;
}
export interface GmailEmailBatch {
chunks: GmailEmailChunk[]; // up to 20 in your example
createdAt: number; // when this batch was generated
emailIds: string[]; // same ids as in chunks[*].emailId
total: number; // total emails in this batch
}
export interface GmailProfile {
@@ -18,22 +36,22 @@ export interface GmailProfile {
interface GmailState {
/** Emails fetched after OAuth connection (from Gmail skill) */
emails: GmailEmailSummary[];
emails: GmailEmailBatch | null;
/** Profile of the connected Gmail user (from Gmail skill) */
profile: GmailProfile | null;
}
const initialState: GmailState = { emails: [], profile: null };
const initialState: GmailState = { emails: null, profile: null };
const gmailSlice = createSlice({
name: 'gmail',
initialState,
reducers: {
setGmailEmails(state, action: PayloadAction<GmailEmailSummary[]>) {
setGmailEmails(state, action: PayloadAction<GmailEmailBatch | null>) {
state.emails = action.payload;
},
clearGmailEmails(state) {
state.emails = [];
state.emails = null;
},
setGmailProfile(state, action: PayloadAction<GmailProfile | null>) {
state.profile = action.payload;
src/utils/desktopDeepLinkListener.ts
+14 -12
View File
@@ -1,13 +1,17 @@
import {invoke, isTauri as coreIsTauri} from '@tauri-apps/api/core';
import {getCurrent, onOpenUrl} from '@tauri-apps/plugin-deep-link';
import { isTauri as coreIsTauri, invoke } from '@tauri-apps/api/core';
import { getCurrent, onOpenUrl } from '@tauri-apps/plugin-deep-link';
import {skillManager} from '../lib/skills/manager';
import {consumeLoginToken, fetchIntegrationTokens} from '../services/api/authApi';
import {buildManualSentryEvent, enqueueError} from '../services/errorReportQueue';
import {store} from '../store';
import {setToken} from '../store/authSlice';
import {setSkillState} from '../store/skillsSlice';
import {decryptIntegrationTokens, hexToBase64, type IntegrationTokensPayload,} from './integrationTokensCrypto';
import { skillManager } from '../lib/skills/manager';
import { consumeLoginToken, fetchIntegrationTokens } from '../services/api/authApi';
import { buildManualSentryEvent, enqueueError } from '../services/errorReportQueue';
import { store } from '../store';
import { setToken } from '../store/authSlice';
import { setSkillState } from '../store/skillsSlice';
import {
decryptIntegrationTokens,
hexToBase64,
type IntegrationTokensPayload,
} from './integrationTokensCrypto';
function getCurrentUserId(): string | null {
const state = store.getState();
@@ -42,7 +46,6 @@ const handleAuthDeepLink = async (parsed: URL) => {
return;
}
console.log('[DeepLink] Received auth token');
try {
@@ -59,8 +62,6 @@ const handleAuthDeepLink = async (parsed: URL) => {
store.dispatch(setToken(jwtToken));
window.location.hash = '/onboarding';
}
};
/**
@@ -239,6 +240,7 @@ const handleOAuthDeepLink = async (parsed: URL) => {
}
await skillManager.notifyOAuthComplete(skillId, integrationId, undefined, extraCredential);
await skillManager.triggerSync(skillId);
} catch (err) {
console.error('[DeepLink] Failed to notify OAuth complete:', err);
}