platform/seed/tools/apify/index.ts

import { mkdirSync, writeFileSync } from 'node:fs';
import { dirname } from 'node:path';

const BASE = 'https://api.apify.com/v2';

type RunStatus = 'READY' | 'RUNNING' | 'SUCCEEDED' | 'FAILED' | 'ABORTING' | 'ABORTED' | 'TIMING-OUT' | 'TIMED-OUT';

type RunData = {
  id: string;
  actId: string;
  status: RunStatus;
  statusMessage?: string;
  defaultDatasetId: string;
  defaultKeyValueStoreId: string;
  startedAt?: string;
  finishedAt?: string;
};

type ToolResult = {
  content: Array<{ type: string; text: string }>;
  isError?: boolean;
};

type OnUpdate = (partial: { content: Array<{ type: string; text: string }> }) => void;

type Params = {
  actor_id: string;
  input?: Record<string, unknown> | string;
  api_token?: string;
  output_path?: string;
  timeout_ms?: number;
  poll_interval_ms?: number;
};

function update(onUpdate: OnUpdate | undefined, text: string): void {
  onUpdate?.({ content: [{ type: 'text', text }] });
}

function resolveToken(params: Params): string | null {
  if (params.api_token) return params.api_token;
  return process.env.OFFICER_APIFY_TOKEN ?? null;
}

function apiUrl(path: string, token: string, extra?: Record<string, string>): string {
  const params = new URLSearchParams({ token, ...extra });
  return `${BASE}${path}?${params}`;
}

async function apiRequest<T>(url: string, init?: RequestInit): Promise<T> {
  const res = await fetch(url, init);
  if (!res.ok) {
    const body = await res.text();
    throw new Error(`Apify API error ${res.status}: ${body}`);
  }
  return res.json() as Promise<T>;
}

async function startRun(token: string, actorId: string, input: Record<string, unknown>): Promise<RunData> {
  const { data } = await apiRequest<{ data: RunData }>(apiUrl(`/acts/${actorId}/runs`, token), {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(input),
  });
  return data;
}

async function getRun(token: string, runId: string): Promise<RunData> {
  const { data } = await apiRequest<{ data: RunData }>(apiUrl(`/actor-runs/${runId}`, token));
  return data;
}

async function getDatasetItems<T>(token: string, datasetId: string): Promise<T[]> {
  return apiRequest<T[]>(apiUrl(`/datasets/${datasetId}/items`, token, { format: 'json' }));
}

const TERMINAL_STATUSES = new Set<RunStatus>(['SUCCEEDED', 'FAILED', 'ABORTED', 'TIMED-OUT']);

export async function execute(
  _toolCallId: string,
  params: Params,
  _signal: AbortSignal | undefined,
  onUpdate?: OnUpdate,
): Promise<ToolResult> {
  const token = resolveToken(params);
  if (!token) {
    return {
      content: [{ type: 'text', text: 'Apify API token not available. Configure it in Settings → Integrations → Apify, or pass api_token explicitly.' }],
      isError: true,
    };
  }

  const { actor_id } = params;
  if (!actor_id) {
    return {
      content: [{ type: 'text', text: 'actor_id is required.' }],
      isError: true,
    };
  }

  let input: Record<string, unknown> = {};
  if (params.input) {
    if (typeof params.input === 'string') {
      try {
        input = JSON.parse(params.input);
      } catch {
        return {
          content: [{ type: 'text', text: 'Invalid JSON in input parameter.' }],
          isError: true,
        };
      }
    } else {
      input = params.input;
    }
  }

  const timeoutMs = params.timeout_ms ?? 300_000;
  const pollIntervalMs = params.poll_interval_ms ?? 3_000;

  try {
    update(onUpdate, `Starting actor ${actor_id}...`);
    const run = await startRun(token, actor_id, input);
    update(onUpdate, `Run ${run.id} started. Waiting for completion...`);

    const start = Date.now();
    let finished = run;

    while (!TERMINAL_STATUSES.has(finished.status)) {
      if (Date.now() - start > timeoutMs) {
        return {
          content: [{ type: 'text', text: `Timeout after ${timeoutMs}ms waiting for run ${run.id}. Status: ${finished.status}` }],
          isError: true,
        };
      }
      await new Promise((r) => setTimeout(r, pollIntervalMs));
      finished = await getRun(token, run.id);
      update(onUpdate, `Status: ${finished.status}...`);
    }

    if (finished.status !== 'SUCCEEDED') {
      return {
        content: [{ type: 'text', text: `Actor run ${finished.status}: ${finished.statusMessage ?? 'unknown error'}` }],
        isError: true,
      };
    }

    update(onUpdate, `Run succeeded. Fetching dataset items...`);
    const items = await getDatasetItems(token, finished.defaultDatasetId);

    if (params.output_path) {
      mkdirSync(dirname(params.output_path), { recursive: true });
      writeFileSync(params.output_path, JSON.stringify(items, null, 2));
      return {
        content: [{ type: 'text', text: `${items.length} items saved to ${params.output_path}` }],
      };
    }

    return {
      content: [{ type: 'text', text: JSON.stringify(items) }],
    };
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    return {
      content: [{ type: 'text', text: `Apify error: ${message}` }],
      isError: true,
    };
  }
}