import {
  getHttpTriggerEndpoint,
  httpRequestTriggerDefinition,
} from "@automate.ax/catalog/triggers/core-http"
import * as z from "zod"
import { defineAction, type ActionObjectInput } from "../../automation/actions"
import {
  getAutomationInvocationEnvironment,
  getNextHookLocation,
} from "../../automation/runtime"
import {
  correlate,
  getCurrentSignalPrerequisites,
  group,
  withoutSignalPrerequisites,
  withPrerequisites,
} from "../../automation/signal-operators"
import { transform } from "../../automation/signal-protocol"
import { createSubscription } from "../../automation/subscription"
import {
  decodeHttpRequestEvent,
  type HttpRequestPayload,
} from "../core/http-request"
import { getFirecrawlApi } from "./lib"

const FIRECRAWL_FORMAT_SCHEMA = z.union([
  z.enum([
    "markdown",
    "html",
    "rawHtml",
    "links",
    "images",
    "screenshot",
    "summary",
    "changeTracking",
    "json",
    "attributes",
    "branding",
    "product",
    "menu",
    "audio",
    "video",
  ]),
  z.object({
    type: z.enum([
      "markdown",
      "html",
      "rawHtml",
      "links",
      "images",
      "summary",
      "branding",
      "product",
      "menu",
      "audio",
      "video",
    ]),
  }),
  z.object({
    /** Capture the entire page instead of the current viewport. */
    fullPage: z.boolean().optional(),

    /** JPEG quality from 1 through 100. */
    quality: z.number().int().min(1).max(100).optional(),

    type: z.literal("screenshot"),

    /** Explicit screenshot viewport. */
    viewport: z
      .object({
        height: z.number().int().positive(),
        width: z.number().int().positive(),
      })
      .optional(),
  }),
  z.object({
    /** Extraction instructions for Firecrawl. */
    prompt: z.string().min(1).optional(),

    /** JSON Schema describing the extracted value. */
    schema: z.record(z.string(), z.json()).optional(),

    type: z.literal("json"),
  }),
  z.object({
    /** Change representations to produce. */
    modes: z.enum(["git-diff", "json"]).array().min(1),

    /** Instructions used for JSON change tracking. */
    prompt: z.string().min(1).optional(),

    /** JSON Schema used for structured change tracking. */
    schema: z.record(z.string(), z.json()).optional(),

    /** Isolates comparison history under a named tag. */
    tag: z.string().min(1).optional(),

    type: z.literal("changeTracking"),
  }),
  z.object({
    /** CSS selectors and attributes to collect. */
    selectors: z
      .object({
        attribute: z.string().min(1),
        selector: z.string().min(1),
      })
      .array()
      .min(1),

    type: z.literal("attributes"),
  }),
  z.object({
    /** Question Firecrawl should answer from the page. */
    question: z.string().min(1),
    type: z.literal("question"),
  }),
  z.object({
    /** Query used to select relevant passages. */
    query: z.string().min(1),
    type: z.literal("highlights"),
  }),
])
const FIRECRAWL_BROWSER_ACTION_SCHEMA = z.discriminatedUnion("type", [
  z.object({
    /** Milliseconds to wait, when waiting by duration. */
    milliseconds: z.number().int().positive().optional(),

    /** CSS selector to wait for, when waiting for an element. */
    selector: z.string().min(1).optional(),

    type: z.literal("wait"),
  }),
  z.object({
    /** Capture the complete page. */
    fullPage: z.boolean().optional(),

    /** JPEG quality from 1 through 100. */
    quality: z.number().int().min(1).max(100).optional(),

    type: z.literal("screenshot"),

    /** Explicit screenshot viewport. */
    viewport: z
      .object({
        height: z.number().int().positive(),
        width: z.number().int().positive(),
      })
      .optional(),
  }),
  z.object({
    /** CSS selector for the element to click. */
    selector: z.string().min(1),
    type: z.literal("click"),
  }),
  z.object({
    /** Text typed into the currently focused element. */
    text: z.string(),
    type: z.literal("write"),
  }),
  z.object({
    /** Browser key name, such as `Enter`. */
    key: z.string().min(1),
    type: z.literal("press"),
  }),
  z.object({
    /** Direction to scroll. */
    direction: z.enum(["up", "down"]),

    /** Optional element to scroll instead of the page. */
    selector: z.string().min(1).optional(),

    type: z.literal("scroll"),
  }),
  z.object({ type: z.literal("scrape") }),
  z.object({
    /** JavaScript evaluated in the page. */
    script: z.string().min(1),
    type: z.literal("executeJavascript"),
  }),
  z.object({
    /** Generated PDF page size. */
    format: z
      .enum([
        "A0",
        "A1",
        "A2",
        "A3",
        "A4",
        "A5",
        "A6",
        "Letter",
        "Legal",
        "Tabloid",
        "Ledger",
      ])
      .optional(),

    /** Generate the PDF in landscape orientation. */
    landscape: z.boolean().optional(),

    /** PDF scale multiplier. */
    scale: z.number().positive().optional(),

    type: z.literal("pdf"),
  }),
])
const FIRECRAWL_LOCATION_SCHEMA = z.object({
  /** ISO 3166-1 alpha-2 country code. */
  country: z
    .string()
    .regex(/^[A-Z]{2}$/)
    .optional(),

  /** Preferred language or locale tags in priority order. */
  languages: z.string().min(1).array().min(1).optional(),
})
const FIRECRAWL_THREAT_PROTECTION_SCHEMA = z.object({
  /** Domain or glob patterns that should always be blocked. */
  blacklist: z.string().min(1).array().max(1_000).optional(),

  /** Lowercase top-level domains to block, without the leading dot. */
  blockedTlds: z.string().min(1).array().max(1_000).optional(),

  /** Whether unavailable scanning blocks or allows the request. */
  failurePolicy: z.enum(["open", "closed"]).optional(),

  /** Enable or disable threat scanning for this request. */
  mode: z.enum(["off", "normal"]).optional(),

  /** Minimum risk score that Firecrawl should block. */
  riskScoreThreshold: z.number().int().min(0).max(100).optional(),

  /** Domain or glob patterns that should always be allowed. */
  whitelist: z.string().min(1).array().max(1_000).optional(),
})
const FIRECRAWL_SCRAPE_OPTIONS_SCHEMA = z.object({
  /** Browser actions to run before extracting content. */
  actions: FIRECRAWL_BROWSER_ACTION_SCHEMA.array().min(1).optional(),

  /** User identity attached to Firecrawl audit logs. */
  auditMetadata: z.object({ username: z.string().min(1) }).optional(),

  /** Block advertising and cookie popups. */
  blockAds: z.boolean().optional(),

  /** HTML selectors or tags to exclude. */
  excludeTags: z.string().min(1).array().optional(),

  /** Use Firecrawl's faster scrape mode when available. */
  fastMode: z.boolean().optional(),

  /** Output representations to return. */
  formats: FIRECRAWL_FORMAT_SCHEMA.array().min(1).optional(),

  /** Headers sent to the target URL. */
  headers: z.record(z.string().min(1), z.string()).optional(),

  /** HTML selectors or tags to include. */
  includeTags: z.string().min(1).array().optional(),

  /** Serve exclusively from Firecrawl's cache without an outbound request. */
  lockdown: z.boolean().optional(),

  /** Geographic and language settings for the scrape. */
  location: FIRECRAWL_LOCATION_SCHEMA.optional(),

  /** Maximum acceptable cache age in milliseconds. */
  maxAge: z.number().int().nonnegative().optional(),

  /** Require a cached result at least this old without scraping afresh. */
  minAge: z.number().int().positive().optional(),

  /** Emulate a mobile device. */
  mobile: z.boolean().optional(),

  /** Remove page chrome such as headers, navigation, and footers. */
  onlyMainContent: z.boolean().optional(),

  /** File parsers used for supported documents. */
  parsers: z
    .union([
      z.string().min(1),
      z.object({
        /** Maximum PDF pages to parse. */
        maxPages: z.number().int().positive().optional(),

        /** PDF parsing strategy. */
        mode: z.enum(["fast", "auto", "ocr"]).optional(),

        type: z.literal("pdf"),
      }),
    ])
    .array()
    .optional(),

  /** Persistent browser profile shared by name. */
  profile: z
    .object({
      name: z.string().min(1),
      saveChanges: z.boolean().optional(),
    })
    .optional(),

  /** Proxy tier used for the target request. */
  proxy: z.enum(["basic", "stealth", "enhanced", "auto"]).optional(),

  /** Remove detected personal information from returned content. */
  redactPII: z
    .union([
      z.boolean(),
      z.object({
        /** Personal-information categories to redact. */
        entities: z
          .enum(["PERSON", "EMAIL", "PHONE", "LOCATION", "FINANCIAL", "SECRET"])
          .array()
          .min(1)
          .optional(),

        /** Redaction strategy. */
        mode: z.enum(["accurate", "aggressive", "fast"]).optional(),

        /** Replacement representation for redacted spans. */
        replaceStyle: z.enum(["tag", "mask", "remove"]).optional(),
      }),
    ])
    .optional(),

  /** Remove base64 image bodies while retaining alt text. */
  removeBase64Images: z.boolean().optional(),

  /** Skip target TLS certificate verification. */
  skipTlsVerification: z.boolean().optional(),

  /** Let Firecrawl reuse the result in its cache. */
  storeInCache: z.boolean().optional(),

  /** Maximum provider scrape time in milliseconds. */
  timeout: z.number().int().min(1_000).max(300_000).optional(),

  /** Enterprise threat-protection overrides. */
  threatProtection: FIRECRAWL_THREAT_PROTECTION_SCHEMA.optional(),

  /** Additional milliseconds to wait for dynamic content. */
  waitFor: z.number().int().nonnegative().optional(),
})
const FIRECRAWL_DOCUMENT_SCHEMA = z
  .object({
    actions: z.record(z.string(), z.json()).optional(),
    answer: z.string().optional(),
    attributes: z
      .object({
        attribute: z.string(),
        selector: z.string(),
        values: z.string().array(),
      })
      .array()
      .optional(),
    audio: z.string().optional(),
    branding: z.json().optional(),
    changeTracking: z.json().optional(),
    highlights: z.string().array().optional(),
    html: z.string().optional(),
    images: z.string().array().optional(),
    json: z.json().optional(),
    links: z.string().array().optional(),
    markdown: z.string().optional(),
    menu: z.json().optional(),
    metadata: z
      .object({
        cacheState: z.enum(["hit", "miss"]).optional(),
        contentType: z.string().optional(),
        creditsUsed: z.number().nonnegative().optional(),
        description: z.string().optional(),
        error: z.string().optional(),
        language: z.string().optional(),
        scrapeId: z.string().optional(),
        sourceURL: z.string().optional(),
        statusCode: z.number().int().optional(),
        title: z.string().optional(),
        url: z.string().optional(),
      })
      .catchall(z.json())
      .optional(),
    product: z.json().optional(),
    rawHtml: z.string().optional(),
    screenshot: z.string().optional(),
    summary: z.string().optional(),
    video: z.string().optional(),
    warning: z.string().optional(),
  })
  .catchall(z.json())
const FIRECRAWL_WEB_SEARCH_ITEM_SCHEMA = FIRECRAWL_DOCUMENT_SCHEMA.extend({
  /** Search category that produced the result. */
  category: z.string().optional(),

  /** Provider description for the result. */
  description: z.string().optional(),

  /** Provider result position. */
  position: z.number().int().optional(),

  /** Provider title for the result. */
  title: z.string().optional(),

  /** Result URL. */
  url: z.string().optional(),
})
const FIRECRAWL_NEWS_SEARCH_RESULT_SCHEMA = FIRECRAWL_DOCUMENT_SCHEMA.extend({
  /** Search category that produced the result. */
  category: z.string().optional(),

  /** Provider-reported publication date. */
  date: z.string().optional(),

  /** Associated news image URL. */
  imageUrl: z.string().optional(),

  /** Provider result position. */
  position: z.number().int().optional(),

  /** Search-result snippet. */
  snippet: z.string().optional(),

  /** Provider title for the result. */
  title: z.string().optional(),

  /** Result URL. */
  url: z.string().optional(),
})
const FIRECRAWL_IMAGE_SEARCH_RESULT_SCHEMA = FIRECRAWL_DOCUMENT_SCHEMA.extend({
  /** Image height in pixels. */
  imageHeight: z.number().int().optional(),

  /** Result image URL. */
  imageUrl: z.string().optional(),

  /** Image width in pixels. */
  imageWidth: z.number().int().optional(),

  /** Provider result position. */
  position: z.number().int().optional(),

  /** Provider title for the image. */
  title: z.string().optional(),

  /** Page containing the image. */
  url: z.string().optional(),
})
const FIRECRAWL_SEARCH_OUTPUT_SCHEMA = z.object({
  /** Developer-focused search results, when requested. */
  developer: FIRECRAWL_WEB_SEARCH_ITEM_SCHEMA.array().optional(),

  /** Image-search results, when requested. */
  images: FIRECRAWL_IMAGE_SEARCH_RESULT_SCHEMA.array().optional(),

  /** News-search results, when requested. */
  news: FIRECRAWL_NEWS_SEARCH_RESULT_SCHEMA.array().optional(),

  /** Ordinary web-search results, returned by default. */
  web: FIRECRAWL_WEB_SEARCH_ITEM_SCHEMA.array().optional(),
})
const FIRECRAWL_SEARCH_INPUT_SCHEMA = z
  .object({
    /** Web-search category filters. */
    categories: z
      .union([
        z.enum(["github", "research", "pdf", "developer"]),
        z.object({
          type: z.enum(["github", "research", "pdf", "developer"]),
        }),
      ])
      .array()
      .min(1)
      .optional(),

    /** Domains to exclude, without protocols or paths. */
    excludeDomains: z.string().min(1).array().min(1).optional(),

    /** Enterprise search privacy mode. */
    enterprise: z.enum(["default", "anon", "zdr"]).array().min(1).optional(),

    /** Generate query-relevant highlights for scraped results. */
    highlights: z.boolean().optional(),

    /** Ignore results that other Firecrawl endpoints cannot process. */
    ignoreInvalidURLs: z.boolean().optional(),

    /** Domains to include exclusively, without protocols or paths. */
    includeDomains: z.string().min(1).array().min(1).optional(),

    /** Maximum results per selected source. */
    limit: z.number().int().min(1).max(100).prefault(10),

    /** Geographic search location. */
    location: z.string().min(1).optional(),

    /** Search query. */
    query: z.string().trim().min(1).max(500),

    /** Options used to scrape each search result. */
    scrapeOptions: FIRECRAWL_SCRAPE_OPTIONS_SCHEMA.optional(),

    /** Search result families to request. */
    sources: z
      .union([
        z.enum(["web", "news", "images"]),
        z.object({ type: z.enum(["web", "news", "images"]) }),
      ])
      .array()
      .min(1)
      .prefault(["web"]),

    /** Firecrawl time-based search expression such as `qdr:w`. */
    tbs: z.string().min(1).optional(),

    /** Maximum provider search time in milliseconds. */
    timeout: z.number().int().positive().optional(),

    /** Enterprise threat-protection overrides. */
    threatProtection: FIRECRAWL_THREAT_PROTECTION_SCHEMA.optional(),
  })
  .superRefine((input, context) => {
    if (input.includeDomains && input.excludeDomains) {
      context.addIssue({
        code: "custom",
        message: "includeDomains and excludeDomains cannot both be provided.",
      })
    }
  })
const FIRECRAWL_MAP_LINK_SCHEMA = z.object({
  /** Provider description for the discovered page. */
  description: z.string().optional(),

  /** Provider title for the discovered page. */
  title: z.string().optional(),

  /** Discovered page URL. */
  url: z.string(),
})
const FIRECRAWL_WEBHOOK_SCHEMA = z.object({
  /** Firecrawl event families sent to the callback. */
  events: z
    .enum(["completed", "failed", "page", "started"])
    .array()
    .min(1)
    .optional(),

  /** Additional headers sent with every callback. */
  headers: z.record(z.string().min(1), z.string()).optional(),

  /** String metadata copied into every callback payload. */
  metadata: z.record(z.string().min(1), z.string()).optional(),

  /** Public callback URL. */
  url: z.url(),
})
const FIRECRAWL_CRAWL_OPTIONS_SCHEMA = z.object({
  /** Follow links to external websites. */
  allowExternalLinks: z.boolean().optional(),

  /** Follow links on subdomains of the starting domain. */
  allowSubdomains: z.boolean().optional(),

  /** Follow sibling and parent paths on the starting domain. */
  crawlEntireDomain: z.boolean().optional(),

  /** Delay in seconds between page scrapes. */
  delay: z.number().nonnegative().optional(),

  /** Deduplicate URLs that Firecrawl considers materially similar. */
  deduplicateSimilarURLs: z.boolean().optional(),

  /** URL pathname regexes to exclude. */
  excludePaths: z.string().min(1).array().optional(),

  /** Ignore query parameters when deduplicating pages. */
  ignoreQueryParameters: z.boolean().optional(),

  /** Crawl pages even when robots.txt disallows them. */
  ignoreRobotsTxt: z.boolean().optional(),

  /** URL pathname regexes to include. */
  includePaths: z.string().min(1).array().optional(),

  /** Maximum pages to crawl. */
  limit: z.number().int().positive().prefault(100),

  /** Maximum simultaneous page scrapes. */
  maxConcurrency: z.number().int().positive().optional(),

  /** Maximum link-discovery depth from the root. */
  maxDiscoveryDepth: z.number().int().nonnegative().optional(),

  /** Natural-language instructions from which Firecrawl derives crawl options. */
  prompt: z.string().min(1).optional(),

  /** Match include and exclude expressions against the complete URL. */
  regexOnFullURL: z.boolean().optional(),

  /** User-Agent used while evaluating robots.txt. */
  robotsUserAgent: z.string().min(1).optional(),

  /** Per-page scrape options. */
  scrapeOptions: FIRECRAWL_SCRAPE_OPTIONS_SCHEMA.optional(),

  /** How Firecrawl should use the target site's sitemap. */
  sitemap: z.enum(["skip", "include", "only"]).optional(),

  /** Request Firecrawl's enterprise zero-data-retention mode. */
  zeroDataRetention: z.boolean().optional(),
})
const FIRECRAWL_CRAWL_INPUT_SCHEMA = FIRECRAWL_CRAWL_OPTIONS_SCHEMA.extend({
  /** Website root URL. */
  url: z.url(),
})
const FIRECRAWL_START_CRAWL_INPUT_SCHEMA = FIRECRAWL_CRAWL_INPUT_SCHEMA.extend({
  /** Optional callback configuration for crawl lifecycle events. */
  webhook: z.union([z.url(), FIRECRAWL_WEBHOOK_SCHEMA]).optional(),
})
const FIRECRAWL_CRAWL_RESULT_SCHEMA = z.object({
  /** Pages Firecrawl completed successfully. */
  completed: z.number().int().nonnegative(),

  /** Credits billed for the crawl, when reported. */
  creditsUsed: z.number().nonnegative().optional(),

  /** Firecrawl result expiration time, when reported. */
  expiresAt: z.date().optional(),

  /** Firecrawl crawl job ID. */
  id: z.string(),

  /** Content returned for every completed page. */
  data: FIRECRAWL_DOCUMENT_SCHEMA.array(),

  /** Next result page, when automatic pagination is disabled. */
  next: z.string().nullable().optional(),

  /** Current Firecrawl crawl state. */
  status: z.enum(["scraping", "completed", "failed", "cancelled"]),

  /** Total pages Firecrawl attempted. */
  total: z.number().int().nonnegative(),
})
const FIRECRAWL_CRAWL_CALLBACK_SCHEMA = z.object({
  error: z.string().optional(),
  id: z.string().min(1),
  success: z.boolean(),
  type: z.enum(["crawl.completed", "crawl.failed"]),
  webhookId: z.string().min(1),
})

/** Scrapes one URL into LLM-ready content with Firecrawl. */
export const scrapeUrlWithFirecrawl = defineAction("Scrape URL with Firecrawl")
  .describe(
    "Scrapes one URL and returns requested content, metadata, screenshots, and extraction results.",
  )
  .account("firecrawl")
  .input(
    FIRECRAWL_SCRAPE_OPTIONS_SCHEMA.extend({
      /** Complete URL to scrape. */
      url: z.url(),
    }),
  )
  .output(FIRECRAWL_DOCUMENT_SCHEMA)
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    const { url, ...options } = input
    return FIRECRAWL_DOCUMENT_SCHEMA.parse(
      await getFirecrawlApi(account.secret).scrape(url, options),
    )
  })

/** Searches the web and optionally scrapes each result with Firecrawl. */
export const searchWebWithFirecrawl = defineAction(
  "Search the web with Firecrawl",
)
  .describe(
    "Searches web, news, image, or developer sources and can scrape matching pages in the same request.",
  )
  .account("firecrawl")
  .input(FIRECRAWL_SEARCH_INPUT_SCHEMA)
  .output(FIRECRAWL_SEARCH_OUTPUT_SCHEMA)
  .retry({ replaySafety: "safe" })
  .handler(async ({ account, input }) => {
    const { query, ...options } = input
    return FIRECRAWL_SEARCH_OUTPUT_SCHEMA.parse(
      await getFirecrawlApi(account.secret).search(query, options),
    )
  })

/** Discovers pages on one website without scraping their content. */
export const mapWebsiteWithFirecrawl = defineAction(
  "Map website with Firecrawl",
)
  .describe(
    "Discovers website URLs from sitemaps and links without scraping every page.",
  )
  .account("firecrawl")
  .input(
    z.object({
      /** User identity attached to Firecrawl audit logs. */
      auditMetadata: z.object({ username: z.string().min(1) }).optional(),

      /** Include pages on subdomains. */
      includeSubdomains: z.boolean().optional(),

      /** Exclude URL variants that differ only by query parameters. */
      ignoreQueryParameters: z.boolean().optional(),

      /** Maximum links to return. */
      limit: z.number().int().min(1).max(100_000).optional(),

      /** Geographic and language settings for discovery. */
      location: FIRECRAWL_LOCATION_SCHEMA.optional(),

      /** Rank discovered URLs against this search text. */
      search: z.string().min(1).optional(),

      /** How Firecrawl should use the target site's sitemap. */
      sitemap: z.enum(["only", "include", "skip"]).optional(),

      /** Maximum provider map time in milliseconds. */
      timeout: z.number().int().positive().optional(),

      /** Enterprise threat-protection overrides. */
      threatProtection: FIRECRAWL_THREAT_PROTECTION_SCHEMA.optional(),

      /** Website root URL. */
      url: z.url(),
    }),
  )
  .output(
    z.object({
      /** Firecrawl map job ID, when supplied. */
      id: z.string().optional(),

      /** Discovered website pages. */
      links: FIRECRAWL_MAP_LINK_SCHEMA.array(),
    }),
  )
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    const { url, ...options } = input
    return await getFirecrawlApi(account.secret).map(url, options)
  })

/** Starts a Firecrawl crawl without waiting for the job to finish. */
export const startWebsiteCrawlWithFirecrawl = defineAction(
  "Start website crawl with Firecrawl",
)
  .describe(
    "Starts a Firecrawl crawl and immediately returns its job identity.",
  )
  .account("firecrawl")
  .input(FIRECRAWL_START_CRAWL_INPUT_SCHEMA)
  .output(
    z.object({
      /** Firecrawl crawl job ID. */
      id: z.string(),

      /** Firecrawl job status URL. */
      url: z.url(),
    }),
  )
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input }) => {
    const { url, ...options } = input
    return await getFirecrawlApi(account.secret).startCrawl(url, options)
  })

const startWebsiteCrawlForCompletion = defineAction(
  "Start website crawl for completion",
)
  .account("firecrawl")
  .input(
    FIRECRAWL_CRAWL_INPUT_SCHEMA.extend({
      callbackUrl: z.url(),
    }),
  )
  .output(
    z.object({
      id: z.string(),
      url: z.url(),
    }),
  )
  .retry({ replaySafety: "unsafe" })
  .handler(async ({ account, input, runtime }) => {
    const { callbackUrl, url, ...options } = input
    return await getFirecrawlApi(account.secret).startCrawl(url, {
      ...options,
      webhook: {
        events: ["completed", "failed"],
        url: await runtime.createCallbackUrl(callbackUrl),
      },
    })
  })

const completeWebsiteCrawlWithFirecrawl = defineAction(
  "Complete website crawl with Firecrawl",
)
  .account("firecrawl")
  .input(FIRECRAWL_CRAWL_CALLBACK_SCHEMA)
  .output(FIRECRAWL_CRAWL_RESULT_SCHEMA)
  .retry({ replaySafety: "safe" })
  .handler(async ({ account, input }) => {
    if (input.type === "crawl.failed") {
      throw new Error(input.error ?? `Firecrawl crawl ${input.id} failed.`)
    }
    const crawl = await getFirecrawlApi(account.secret).getCrawlStatus(input.id)
    if (crawl.status !== "completed") {
      throw new Error(`Firecrawl crawl ${crawl.id} ${crawl.status}.`)
    }
    return FIRECRAWL_CRAWL_RESULT_SCHEMA.parse({
      ...crawl,
      expiresAt: crawl.expiresAt ? new Date(crawl.expiresAt) : undefined,
    })
  })

/**
 * Crawls a website and continues in a correlated child after Firecrawl
 * finishes.
 *
 * @param input - Crawl configuration.
 * @param options - Firecrawl account selection.
 */
export function crawlWebsiteWithFirecrawl(
  input: ActionObjectInput<typeof FIRECRAWL_CRAWL_INPUT_SCHEMA>,
  options?: Parameters<typeof startWebsiteCrawlWithFirecrawl>[1],
) {
  const prerequisites = getCurrentSignalPrerequisites()
  return group(
    { name: "Crawl website with Firecrawl", presentation: "hidden" },
    () =>
      withoutSignalPrerequisites(() => {
        const environment = getAutomationInvocationEnvironment()
        const callbackLocation = getNextHookLocation()
        if (!environment || !callbackLocation) {
          throw new Error(
            "Firecrawl crawl callbacks must be declared inside an automation invocation.",
          )
        }
        const callbackRequest = createSubscription<
          HttpRequestPayload,
          { readonly endpoint: string }
        >(
          httpRequestTriggerDefinition,
          { protection: "callback" },
          {
            endpoint: getHttpTriggerEndpoint({
              ...environment,
              hookSlot: callbackLocation.slot,
              scope: "trigger",
              scopePath: callbackLocation.scopePath,
            }),
          },
          { inferActionBoundary: false },
        )
        const start = () =>
          startWebsiteCrawlForCompletion(
            {
              ...input,
              callbackUrl: callbackRequest.endpoint,
            },
            options,
          )
        const callback = transform(
          callbackRequest,
          (event) =>
            decodeHttpRequestEvent(event, FIRECRAWL_CRAWL_CALLBACK_SCHEMA).body,
        )
        const completed = transform(
          [
            correlate(
              [
                (prerequisites
                  ? withPrerequisites(prerequisites, start)
                  : start()
                ).keyBy(({ id }) => id),
                callback.keyBy(({ id }) => id),
              ],
              { occurrenceTtl: "7d" },
            ),
            callback,
          ],
          (_resumed, callback) => callback,
        )
        return completeWebsiteCrawlWithFirecrawl(
          {
            error: completed.error,
            id: completed.id,
            success: completed.success,
            type: completed.type,
            webhookId: completed.webhookId,
          },
          options,
        )
      }),
  )
}