eks_relay/src/clients/flowiseClient.ts

import { config } from '../config.js';
import { logger } from '../logger.js';
import { upstream } from '../errors.js';
import { request, safeLabel } from './httpClient.js';

export interface FlowisePayload {
  question: string;
  overrideConfig: Record<string, unknown>;
  metadata: Record<string, unknown>;
}

export interface FlowiseResult {
  type: 'reply' | 'handoff' | 'unknown';
  text: string | null;
  actions: unknown[] | null;
}

export class FlowiseClient {
  private readonly predictUrl: string;
  private readonly apiKey: string;
  private readonly timeoutMs: number;

  constructor() {
    const cfg = config();
    this.predictUrl = cfg.FLOWISE_PREDICT_URL;
    this.apiKey = cfg.FLOWISE_API_KEY;
    this.timeoutMs = cfg.FLOWISE_TIMEOUT_MS;
  }

  async predict(payload: FlowisePayload): Promise<FlowiseResult> {
    const headers: Record<string, string> = {};
    // The PHP relay shipped a literal `Bearer ***` placeholder here; send the
    // real key (and only when one is configured).
    if (this.apiKey) headers.Authorization = `Bearer ${this.apiKey}`;

    logger.info('Calling Flowise', { target: safeLabel(this.predictUrl) });

    const res = await request<Record<string, unknown>>(this.predictUrl, {
      method: 'POST',
      headers,
      json: payload,
      timeoutMs: this.timeoutMs,
      label: 'flowise predict',
    });

    if (res.status >= 400) {
      logger.error('Flowise error response', { status: res.status });
      throw upstream('FLOWISE_ERROR', `Flowise returned HTTP ${res.status}.`);
    }

    return normaliseFlowiseResponse(res.json, res.body);
  }

  /** Health probe against the Flowise API root; used by /ready only. */
  async ping(): Promise<{ ok: boolean; status: number; target: string }> {
    const cfg = config();
    const base = cfg.FLOWISE_BASE_URL || deriveBaseUrl(this.predictUrl);
    if (!base) return { ok: false, status: 0, target: 'not-configured' };
    const url = `${base.replace(/\/+$/, '')}/api/v1/ping`;
    try {
      const headers: Record<string, string> = {};
      if (this.apiKey) headers.Authorization = `Bearer ${this.apiKey}`;
      const res = await request(url, { headers, label: 'flowise ping', timeoutMs: 5_000 });
      return { ok: res.status < 400, status: res.status, target: safeLabel(url) };
    } catch {
      return { ok: false, status: 0, target: safeLabel(url) };
    }
  }
}

function deriveBaseUrl(predictUrl: string): string {
  try {
    const u = new URL(predictUrl);
    return `${u.protocol}//${u.host}`;
  } catch {
    return '';
  }
}

/**
 * Flowise answers in several shapes depending on chatflow/version:
 *  - a bare text body,
 *  - `{ text | response | answer }`,
 *  - the same plus `actions: [{ type: 'handoff' }]`.
 */
export function normaliseFlowiseResponse(json: unknown, rawBody: string): FlowiseResult {
  if (json && typeof json === 'object' && !Array.isArray(json)) {
    const obj = json as Record<string, unknown>;
    const candidate = obj.text ?? obj.response ?? obj.answer;
    const text = typeof candidate === 'string' ? candidate : null;
    const actions = Array.isArray(obj.actions) ? (obj.actions as unknown[]) : null;

    if (actions) {
      const handoff = actions.some(
        (a) =>
          a !== null &&
          typeof a === 'object' &&
          (a as Record<string, unknown>).type === 'handoff',
      );
      if (handoff) return { type: 'handoff', text, actions };
    }
    if (text !== null && text.trim() !== '') return { type: 'reply', text, actions };
  }

  const body = rawBody.trim();
  // A plain-text body is the answer itself, but only when it is not JSON we
  // already failed to make sense of.
  if (body !== '' && json === null) return { type: 'reply', text: body, actions: null };

  return { type: 'unknown', text: null, actions: null };
}