MeshBot/src/opencode/opencode.service.ts

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<void> {
    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<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }

  private authHeaders(): Record<string, string> {
    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<WorkspaceEntry[]>(
      `${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<SessionEntry[]>(`${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<T>(url: string): Promise<T> {
    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<void> {
    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;
    }
  }
}