anomalyco_opencode/packages/opencode/src/file/ripgrep.ts

// Ripgrep utility functions
import path from "path"
import { Global } from "../global"
import fs from "fs/promises"
import z from "zod"
import { Array as Arr, Effect, Layer, Context, Schema } from "effect"
import * as Stream from "effect/Stream"
import { ChildProcess } from "effect/unstable/process"
import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"
import * as CrossSpawnSpawner from "@/effect/cross-spawn-spawner"
import type { PlatformError } from "effect/PlatformError"
import { NamedError } from "@opencode-ai/util/error"
import { lazy } from "../util/lazy"

import { Filesystem } from "../util/filesystem"
import { AppFileSystem } from "../filesystem"
import { Process } from "../util/process"
import { which } from "../util/which"
import { text } from "node:stream/consumers"

import { ZipReader, BlobReader, BlobWriter } from "@zip.js/zip.js"
import { Log } from "@/util/log"

export namespace Ripgrep {
  const log = Log.create({ service: "ripgrep" })
  const Stats = z.object({
    elapsed: z.object({
      secs: z.number(),
      nanos: z.number(),
      human: z.string(),
    }),
    searches: z.number(),
    searches_with_match: z.number(),
    bytes_searched: z.number(),
    bytes_printed: z.number(),
    matched_lines: z.number(),
    matches: z.number(),
  })

  const Begin = z.object({
    type: z.literal("begin"),
    data: z.object({
      path: z.object({
        text: z.string(),
      }),
    }),
  })

  export const Match = z.object({
    type: z.literal("match"),
    data: z.object({
      path: z.object({
        text: z.string(),
      }),
      lines: z.object({
        text: z.string(),
      }),
      line_number: z.number(),
      absolute_offset: z.number(),
      submatches: z.array(
        z.object({
          match: z.object({
            text: z.string(),
          }),
          start: z.number(),
          end: z.number(),
        }),
      ),
    }),
  })

  const End = z.object({
    type: z.literal("end"),
    data: z.object({
      path: z.object({
        text: z.string(),
      }),
      binary_offset: z.number().nullable(),
      stats: Stats,
    }),
  })

  const Summary = z.object({
    type: z.literal("summary"),
    data: z.object({
      elapsed_total: z.object({
        human: z.string(),
        nanos: z.number(),
        secs: z.number(),
      }),
      stats: Stats,
    }),
  })

  const Result = z.union([Begin, Match, End, Summary])

  const Hit = Schema.Struct({
    type: Schema.Literal("match"),
    data: Schema.Struct({
      path: Schema.Struct({
        text: Schema.String,
      }),
      lines: Schema.Struct({
        text: Schema.String,
      }),
      line_number: Schema.Number,
      absolute_offset: Schema.Number,
      submatches: Schema.mutable(
        Schema.Array(
          Schema.Struct({
            match: Schema.Struct({
              text: Schema.String,
            }),
            start: Schema.Number,
            end: Schema.Number,
          }),
        ),
      ),
    }),
  })

  const Row = Schema.Union([
    Schema.Struct({ type: Schema.Literal("begin"), data: Schema.Unknown }),
    Hit,
    Schema.Struct({ type: Schema.Literal("end"), data: Schema.Unknown }),
    Schema.Struct({ type: Schema.Literal("summary"), data: Schema.Unknown }),
  ])

  const decode = Schema.decodeUnknownSync(Schema.fromJsonString(Row))

  export type Result = z.infer<typeof Result>
  export type Match = z.infer<typeof Match>
  export type Item = Match["data"]
  export type Begin = z.infer<typeof Begin>
  export type End = z.infer<typeof End>
  export type Summary = z.infer<typeof Summary>
  const PLATFORM = {
    "arm64-darwin": { platform: "aarch64-apple-darwin", extension: "tar.gz" },
    "arm64-linux": {
      platform: "aarch64-unknown-linux-gnu",
      extension: "tar.gz",
    },
    "x64-darwin": { platform: "x86_64-apple-darwin", extension: "tar.gz" },
    "x64-linux": { platform: "x86_64-unknown-linux-musl", extension: "tar.gz" },
    "arm64-win32": { platform: "aarch64-pc-windows-msvc", extension: "zip" },
    "x64-win32": { platform: "x86_64-pc-windows-msvc", extension: "zip" },
  } as const

  export const ExtractionFailedError = NamedError.create(
    "RipgrepExtractionFailedError",
    z.object({
      filepath: z.string(),
      stderr: z.string(),
    }),
  )

  export const UnsupportedPlatformError = NamedError.create(
    "RipgrepUnsupportedPlatformError",
    z.object({
      platform: z.string(),
    }),
  )

  export const DownloadFailedError = NamedError.create(
    "RipgrepDownloadFailedError",
    z.object({
      url: z.string(),
      status: z.number(),
    }),
  )

  const state = lazy(async () => {
    const system = which("rg")
    if (system) {
      const stat = await fs.stat(system).catch(() => undefined)
      if (stat?.isFile()) return { filepath: system }
      log.warn("bun.which returned invalid rg path", { filepath: system })
    }
    const filepath = path.join(Global.Path.bin, "rg" + (process.platform === "win32" ? ".exe" : ""))

    if (!(await Filesystem.exists(filepath))) {
      const platformKey = `${process.arch}-${process.platform}` as keyof typeof PLATFORM
      const config = PLATFORM[platformKey]
      if (!config) throw new UnsupportedPlatformError({ platform: platformKey })

      const version = "14.1.1"
      const filename = `ripgrep-${version}-${config.platform}.${config.extension}`
      const url = `https://github.com/BurntSushi/ripgrep/releases/download/${version}/${filename}`

      const response = await fetch(url)
      if (!response.ok) throw new DownloadFailedError({ url, status: response.status })

      const arrayBuffer = await response.arrayBuffer()
      const archivePath = path.join(Global.Path.bin, filename)
      await Filesystem.write(archivePath, Buffer.from(arrayBuffer))
      if (config.extension === "tar.gz") {
        const args = ["tar", "-xzf", archivePath, "--strip-components=1"]

        if (platformKey.endsWith("-darwin")) args.push("--include=*/rg")
        if (platformKey.endsWith("-linux")) args.push("--wildcards", "*/rg")

        const proc = Process.spawn(args, {
          cwd: Global.Path.bin,
          stderr: "pipe",
          stdout: "pipe",
        })
        const exit = await proc.exited
        if (exit !== 0) {
          const stderr = proc.stderr ? await text(proc.stderr) : ""
          throw new ExtractionFailedError({
            filepath,
            stderr,
          })
        }
      }
      if (config.extension === "zip") {
        const zipFileReader = new ZipReader(new BlobReader(new Blob([arrayBuffer])))
        const entries = await zipFileReader.getEntries()
        let rgEntry: any
        for (const entry of entries) {
          if (entry.filename.endsWith("rg.exe")) {
            rgEntry = entry
            break
          }
        }

        if (!rgEntry) {
          throw new ExtractionFailedError({
            filepath: archivePath,
            stderr: "rg.exe not found in zip archive",
          })
        }

        const rgBlob = await rgEntry.getData(new BlobWriter())
        if (!rgBlob) {
          throw new ExtractionFailedError({
            filepath: archivePath,
            stderr: "Failed to extract rg.exe from zip archive",
          })
        }
        await Filesystem.write(filepath, Buffer.from(await rgBlob.arrayBuffer()))
        await zipFileReader.close()
      }
      await fs.unlink(archivePath)
      if (!platformKey.endsWith("-win32")) await fs.chmod(filepath, 0o755)
    }

    return {
      filepath,
    }
  })

  export async function filepath() {
    const { filepath } = await state()
    return filepath
  }

  export async function* files(input: {
    cwd: string
    glob?: string[]
    hidden?: boolean
    follow?: boolean
    maxDepth?: number
    signal?: AbortSignal
  }) {
    input.signal?.throwIfAborted()

    const args = [await filepath(), "--files", "--glob=!.git/*"]
    if (input.follow) args.push("--follow")
    if (input.hidden !== false) args.push("--hidden")
    if (input.maxDepth !== undefined) args.push(`--max-depth=${input.maxDepth}`)
    if (input.glob) {
      for (const g of input.glob) {
        args.push(`--glob=${g}`)
      }
    }

    // Guard against invalid cwd to provide a consistent ENOENT error.
    if (!(await fs.stat(input.cwd).catch(() => undefined))?.isDirectory()) {
      throw Object.assign(new Error(`No such file or directory: '${input.cwd}'`), {
        code: "ENOENT",
        errno: -2,
        path: input.cwd,
      })
    }

    const proc = Process.spawn(args, {
      cwd: input.cwd,
      stdout: "pipe",
      stderr: "ignore",
      abort: input.signal,
    })

    if (!proc.stdout) {
      throw new Error("Process output not available")
    }

    let buffer = ""
    const stream = proc.stdout as AsyncIterable<Buffer | string>
    for await (const chunk of stream) {
      input.signal?.throwIfAborted()

      buffer += typeof chunk === "string" ? chunk : chunk.toString()
      // Handle both Unix (\n) and Windows (\r\n) line endings
      const lines = buffer.split(/\r?\n/)
      buffer = lines.pop() || ""

      for (const line of lines) {
        if (line) yield line
      }
    }

    if (buffer) yield buffer
    await proc.exited

    input.signal?.throwIfAborted()
  }

  export interface Interface {
    readonly files: (input: {
      cwd: string
      glob?: string[]
      hidden?: boolean
      follow?: boolean
      maxDepth?: number
    }) => Stream.Stream<string, PlatformError>
    readonly search: (input: {
      cwd: string
      pattern: string
      glob?: string[]
      limit?: number
      follow?: boolean
    }) => Effect.Effect<{ items: Item[]; partial: boolean }, PlatformError | Error>
  }

  export class Service extends Context.Service<Service, Interface>()("@opencode/Ripgrep") {}

  export const layer: Layer.Layer<Service, never, ChildProcessSpawner | AppFileSystem.Service> = Layer.effect(
    Service,
    Effect.gen(function* () {
      const spawner = yield* ChildProcessSpawner
      const afs = yield* AppFileSystem.Service
      const bin = Effect.fn("Ripgrep.path")(function* () {
        return yield* Effect.promise(() => filepath())
      })
      const args = Effect.fn("Ripgrep.args")(function* (input: {
        mode: "files" | "search"
        glob?: string[]
        hidden?: boolean
        follow?: boolean
        maxDepth?: number
        limit?: number
        pattern?: string
      }) {
        const out = [yield* bin(), input.mode === "search" ? "--json" : "--files", "--glob=!.git/*"]
        if (input.follow) out.push("--follow")
        if (input.hidden !== false) out.push("--hidden")
        if (input.maxDepth !== undefined) out.push(`--max-depth=${input.maxDepth}`)
        if (input.glob) {
          for (const g of input.glob) {
            out.push(`--glob=${g}`)
          }
        }
        if (input.limit) out.push(`--max-count=${input.limit}`)
        if (input.mode === "search") out.push("--no-messages")
        if (input.pattern) out.push("--", input.pattern)
        return out
      })

      const files = Effect.fn("Ripgrep.files")(function* (input: {
        cwd: string
        glob?: string[]
        hidden?: boolean
        follow?: boolean
        maxDepth?: number
      }) {
        const rgPath = yield* bin()
        const isDir = yield* afs.isDir(input.cwd)
        if (!isDir) {
          return yield* Effect.die(
            Object.assign(new Error(`No such file or directory: '${input.cwd}'`), {
              code: "ENOENT" as const,
              errno: -2,
              path: input.cwd,
            }),
          )
        }

        const cmd = yield* args({
          mode: "files",
          glob: input.glob,
          hidden: input.hidden,
          follow: input.follow,
          maxDepth: input.maxDepth,
        })

        return spawner
          .streamLines(ChildProcess.make(cmd[0], cmd.slice(1), { cwd: input.cwd }))
          .pipe(Stream.filter((line: string) => line.length > 0))
      })

      const search = Effect.fn("Ripgrep.search")(function* (input: {
        cwd: string
        pattern: string
        glob?: string[]
        limit?: number
        follow?: boolean
      }) {
        return yield* Effect.scoped(
          Effect.gen(function* () {
            const cmd = yield* args({
              mode: "search",
              glob: input.glob,
              follow: input.follow,
              limit: input.limit,
              pattern: input.pattern,
            })

            const handle = yield* spawner.spawn(
              ChildProcess.make(cmd[0], cmd.slice(1), {
                cwd: input.cwd,
                stdin: "ignore",
              }),
            )

            const [items, stderr, code] = yield* Effect.all(
              [
                Stream.decodeText(handle.stdout).pipe(
                  Stream.splitLines,
                  Stream.filter((line) => line.length > 0),
                  Stream.mapArrayEffect((lines) =>
                    Effect.try({
                      try: () => Arr.map(lines, (line) => decode(line)),
                      catch: (cause) => new Error("invalid ripgrep output", { cause }),
                    }),
                  ),
                  Stream.filter((row): row is Schema.Schema.Type<typeof Hit> => row.type === "match"),
                  Stream.map((row): Item => row.data),
                  Stream.runCollect,
                  Effect.map((chunk) => [...chunk]),
                ),
                Stream.mkString(Stream.decodeText(handle.stderr)),
                handle.exitCode,
              ],
              { concurrency: "unbounded" },
            )

            if (code !== 0 && code !== 1 && code !== 2) {
              return yield* Effect.fail(new Error(`ripgrep failed: ${stderr}`))
            }

            return {
              items,
              partial: code === 2,
            }
          }),
        )
      })

      return Service.of({
        files: (input) => Stream.unwrap(files(input)),
        search,
      })
    }),
  )

  export const defaultLayer = layer.pipe(
    Layer.provide(AppFileSystem.defaultLayer),
    Layer.provide(CrossSpawnSpawner.defaultLayer),
  )

  export async function tree(input: { cwd: string; limit?: number; signal?: AbortSignal }) {
    log.info("tree", input)
    const files = await Array.fromAsync(Ripgrep.files({ cwd: input.cwd, signal: input.signal }))
    interface Node {
      name: string
      children: Map<string, Node>
    }

    function dir(node: Node, name: string) {
      const existing = node.children.get(name)
      if (existing) return existing
      const next = { name, children: new Map() }
      node.children.set(name, next)
      return next
    }

    const root: Node = { name: "", children: new Map() }
    for (const file of files) {
      if (file.includes(".opencode")) continue
      const parts = file.split(path.sep)
      if (parts.length < 2) continue
      let node = root
      for (const part of parts.slice(0, -1)) {
        node = dir(node, part)
      }
    }

    function count(node: Node): number {
      let total = 0
      for (const child of node.children.values()) {
        total += 1 + count(child)
      }
      return total
    }

    const total = count(root)
    const limit = input.limit ?? total
    const lines: string[] = []
    const queue: { node: Node; path: string }[] = []
    for (const child of Array.from(root.children.values()).sort((a, b) => a.name.localeCompare(b.name))) {
      queue.push({ node: child, path: child.name })
    }

    let used = 0
    for (let i = 0; i < queue.length && used < limit; i++) {
      const { node, path } = queue[i]
      lines.push(path)
      used++
      for (const child of Array.from(node.children.values()).sort((a, b) => a.name.localeCompare(b.name))) {
        queue.push({ node: child, path: `${path}/${child.name}` })
      }
    }

    if (total > used) lines.push(`[${total - used} truncated]`)

    return lines.join("\n")
  }
}