import { Injectable, Logger } from '@nestjs/common'; // ponytail: hardcoded backoff [2s, 10s, 30s] (~45 s window) — no env knob; tune here if outages run longer const RETRY_DELAYS_MS = [2_000, 10_000, 30_000]; // opencode accepts a session message immediately but holds the HTTP response until the agent turn finishes const TIMEOUT_MS = 15_000; /** Entry of GET {api}/api/workspaces (only the fields we use). */ interface WorkspaceEntry { id?: string; path?: string; name?: string; status?: string; proxyPath?: string; } /** Entry of GET {base}/session (only the fields we use). */ interface SessionEntry { id: string; time?: { updated?: number }; } @Injectable() export class OpencodeService { private readonly logger = new Logger(OpencodeService.name); // read at construction, like MeshService's env config private readonly api = process.env.OPENCODE_API ?? 'http://172.0.0.1:9898'; private readonly workspacePath = process.env.OPENCODE_WORKSPACE_PATH ?? '/home/krzyw/projects/sandbox/mesh-bot'; // when set, skips session discovery private readonly sessionOverride = process.env.OPENCODE_SESSION ?? ''; private readonly user = process.env.OPENCODE_USER ?? ''; private readonly pass = process.env.OPENCODE_PASS ?? ''; /** Notifier disabled without credentials — the app must keep running (and notify nothing) when they are missing. */ private readonly enabled = this.user !== '' && this.pass !== ''; constructor() { if (!this.enabled) this.logger.warn( 'opencode notifier disabled: OPENCODE_USER/OPENCODE_PASS not set', ); } /** * Push an rx mesh message into the opencode session as a text part. Resolves when delivered — * including on a final-POST read timeout, which means DELIVERED (opencode holds the response until * the agent turn finishes) and is never retried. Workspace/session resolution failures are retried * up to RETRY_DELAYS_MS.length + 1 attempts total; the last failure logs `opencode notify dropped * after N attempts`. No-op when disabled (no credentials). */ async notify(id: number, text: string, from?: number): Promise { if (!this.enabled) return; const msg = `New mesh message (id ${id})` + (from != null ? ` from node ${from}` : '') + `: "${text}" — please react to it.`; const maxAttempts = RETRY_DELAYS_MS.length + 1; for (let attempt = 1; ; attempt++) { try { const { base, sid } = await this.resolveTarget(); await this.postMessage(base, sid, msg); return; // success on any attempt — a POST timeout already resolved as delivered inside postMessage } catch (err: unknown) { const message = err instanceof Error ? err.message : String(err); if (attempt === maxAttempts) { this.logger.warn( `opencode notify dropped after ${attempt} attempts: ${message}`, ); return; } this.logger.warn( `opencode notify failed (attempt ${attempt}/${maxAttempts}): ${message}`, ); await this.sleep(RETRY_DELAYS_MS[attempt - 1]); } } } /** Injectable so specs can skip the backoff delays. */ private sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } private authHeaders(): Record { return { authorization: 'Basic ' + Buffer.from(`${this.user}:${this.pass}`).toString('base64'), }; } /** Resolve (base_url, session_id) fresh per request; throws with a clear reason on API/workspace/session failure. */ private async resolveTarget(): Promise<{ base: string; sid: string }> { const workspaces = await this.getJson( `${this.api}/api/workspaces`, ); // match by path first, fall back to name — the workspace UUID is regenerated on host restarts const matches = workspaces.filter((w) => w.path === this.workspacePath).length > 0 ? workspaces.filter((w) => w.path === this.workspacePath) : workspaces.filter((w) => w.name === 'mesh-bot'); // prefer ready entries when several match const pool = matches.filter((w) => w.status === 'ready').length > 0 ? matches.filter((w) => w.status === 'ready') : matches; if (pool.length === 0) throw new Error( `workspace not found: path=${this.workspacePath} name=mesh-bot`, ); const ws = pool[0]; const base = this.api + (ws.proxyPath ?? ''); let sid = this.sessionOverride; if (!sid) { const sessions = await this.getJson(`${base}/session`); if (sessions.length === 0) throw new Error( `no sessions in workspace ${ws.id} (${this.workspacePath})`, ); // most recently updated session; first entry when none carry time.updated const updated = (s: SessionEntry): number => s.time?.updated ?? 0; sid = sessions.some((s) => updated(s)) ? sessions.reduce((a, b) => (updated(b) > updated(a) ? b : a)).id : sessions[0].id; } return { base, sid }; } private async getJson(url: string): Promise { let res: Response; try { res = await fetch(url, { headers: this.authHeaders(), signal: AbortSignal.timeout(TIMEOUT_MS), }); } catch (err) { throw new Error( `opencode API unreachable at ${url}: ${(err as Error).message}`, ); } if (!res.ok) throw new Error(`opencode API HTTP ${res.status} at ${url}`); return (await res.json()) as T; } private async postMessage( base: string, sid: string, msg: string, ): Promise { try { const res = await fetch(`${base}/session/${sid}/message`, { method: 'POST', headers: { ...this.authHeaders(), 'content-type': 'application/json' }, body: JSON.stringify({ parts: [{ type: 'text', text: msg }] }), signal: AbortSignal.timeout(TIMEOUT_MS), }); if (!res.ok) throw new Error(`HTTP ${res.status}`); // HTTP errors retry like network failures } catch (err) { // a read timeout here means DELIVERED, not failed — opencode accepted the message and is holding // the response until the agent turn finishes; never retried if ((err as Error).name === 'TimeoutError') { this.logger.log( `opencode notify accepted (response still pending after ${TIMEOUT_MS / 1000} s)`, ); return; } throw err; } } }