//// Finding transcript rows by their text: the one search behind
//// transcript_grep, lcm_grep, `h.search_messages`, and `agents.sessions`.
////
//// SQLite narrows the stored rows to candidates one window of rows at a
//// time, so no search holds the store for long, and only candidates are
//// decoded. A candidate's text, as retrieval tools show it, then decides,
//// case-insensitively.

import albedo/daemon/conversation
import albedo/daemon/store
import albedo/harness/compaction
import albedo/harness/tool
import gleam/bool
import gleam/dynamic/decode
import gleam/int
import gleam/json
import gleam/list
import gleam/option.{None, Some}
import gleam/result
import gleam/string
import sqlight

pub type Scope {
  /// One session's rows.
  Within(session: String)
  /// The rows of every session but `caller`, and only of those opened in
  /// `cwd` unless it is "".
  Beyond(caller: String, cwd: String)
}

/// A row whose text contains the needle.
pub type Match {
  Match(session: String, seq: Int, text: String)
}

/// `pattern` trimmed, checked, and lowercased for matching.
pub fn needle(pattern: String) -> Result(String, String) {
  let pattern = string.trim(pattern)
  case pattern != "" && string.length(pattern) <= 200 {
    True -> Ok(string.lowercase(pattern))
    False -> Error("search pattern must be 1..200 characters")
  }
}

/// `step` over each row in `scope` whose text contains `needle`, newest
/// first. A row of a session for which `wants(acc, session)` is False is
/// passed over without being decoded.
pub fn fold(
  ledger: store.Store,
  scope: Scope,
  needle: String,
  acc: a,
  wants: fn(a, String) -> Bool,
  step: fn(a, Match) -> a,
) -> Result(a, String) {
  use #(low, high) <- result.try(bounds(ledger, scope))
  let scan =
    Scan(ledger, needle, probe(needle), scope, low, high, False, wants, step)
  pages(scan, high + 1, acc)
}

type Scan(a) {
  Scan(
    ledger: store.Store,
    needle: String,
    probe: String,
    scope: Scope,
    low: Int,
    high: Int,
    ascending: Bool,
    wants: fn(a, String) -> Bool,
    step: fn(a, Match) -> a,
  )
}

const window_rows = 2048

fn bounds(ledger: store.Store, scope: Scope) -> Result(#(Int, Int), String) {
  let #(sql, arguments) = case scope {
    Within(session) -> #(
      "SELECT (SELECT COALESCE(MIN(seq),1) FROM transcript WHERE session=?),(SELECT COALESCE(MAX(seq),0) FROM transcript WHERE session=?)",
      [sqlight.text(session), sqlight.text(session)],
    )
    Beyond(_, _) -> #(
      "SELECT (SELECT COALESCE(MIN(seq),1) FROM transcript),(SELECT COALESCE(MAX(seq),0) FROM transcript)",
      [],
    )
  }
  use rows <- result.try(
    store.read(ledger, sql, arguments, {
      use low <- decode.field(0, decode.int)
      use high <- decode.field(1, decode.int)
      decode.success(#(low, high))
    }),
  )
  case rows {
    [bounds] -> Ok(bounds)
    _ -> Error("could not read transcript bounds")
  }
}

fn scope_filter(scope: Scope) -> #(String, List(sqlight.Value)) {
  case scope {
    Within(session) -> #("t.session=?", [sqlight.text(session)])
    Beyond(caller, "") -> #("t.session<>?", [sqlight.text(caller)])
    Beyond(caller, cwd) -> #("t.session<>? AND s.cwd=?", [
      sqlight.text(caller),
      sqlight.text(cwd),
    ])
  }
}

/// Page actual rows before filtering candidates, so a rare probe cannot
/// make one store call scan the entire transcript. The cursor advances even
/// when no row in the page contains the probe. Cross-session scans page global
/// rows before scope filtering to bound work when most sessions are excluded.
fn pages(scan: Scan(a), cursor: Int, acc: a) -> Result(a, String) {
  let #(where, arguments) = scope_filter(scan.scope)
  let #(raw_scope, raw_arguments, hit_scope, hit_arguments) = case scan.scope {
    Within(_) -> #(where <> " AND ", arguments, "", [])
    Beyond(_, _) -> #("", [], where <> " AND ", arguments)
  }
  let #(comparison, order, boundary) = case scan.ascending {
    True -> #(">", "ASC", "MAX")
    False -> #("<", "DESC", "MIN")
  }
  use rows <- result.try(
    store.read(
      scan.ledger,
      "WITH page AS MATERIALIZED (SELECT t.session,t.seq,t.payload FROM transcript t WHERE "
        <> raw_scope
        <> "t.seq>=? AND t.seq<=? AND t.seq"
        <> comparison
        <> "? ORDER BY t.seq "
        <> order
        <> " LIMIT ?), hits AS (SELECT t.session,t.seq,t.payload FROM page t JOIN sessions s ON s.id=t.session WHERE "
        <> hit_scope
        <> "instr(lower(t.payload),?)>0) SELECT (SELECT COALESCE("
        <> boundary
        <> "(seq),-1) FROM page),hits.session,hits.seq,hits.payload FROM (SELECT 1) LEFT JOIN hits ON 1 ORDER BY hits.seq "
        <> order,
      list.append(raw_arguments, [
        sqlight.int(scan.low),
        sqlight.int(scan.high),
        sqlight.int(cursor),
        sqlight.int(window_rows),
      ])
        |> list.append(hit_arguments)
        |> list.append([sqlight.text(scan.probe)]),
      {
        use cursor <- decode.field(0, decode.int)
        use session <- decode.field(1, decode.optional(decode.string))
        use seq <- decode.field(2, decode.optional(decode.int))
        use payload <- decode.field(3, decode.optional(decode.bit_array))
        decode.success(
          #(cursor, case session, seq, payload {
            Some(session), Some(seq), Some(payload) ->
              Some(#(session, seq, payload))
            _, _, _ -> None
          }),
        )
      },
    ),
  )
  case rows {
    [#(-1, _)] -> Ok(acc)
    [#(next, _), ..] -> {
      let acc =
        list.fold(rows, acc, fn(acc, row) {
          case row.1 {
            Some(candidate) -> visit(scan, acc, candidate)
            None -> acc
          }
        })
      pages(scan, next, acc)
    }
    [] -> Error("could not read transcript search page")
  }
}

pub type Page {
  Page(count: Int, matches: List(Match))
}

/// Count every confirmed match while retaining only the chronological page.
/// Exact counts still require scanning every candidate inside the range.
pub fn page(
  ledger: store.Store,
  session: String,
  needle: String,
  first: Int,
  last: Int,
  offset: Int,
  limit: Int,
) -> Result(Page, String) {
  use snapshot <- result.try(conversation.snapshot(ledger, session))
  let scan =
    Scan(
      ledger,
      needle,
      probe(needle),
      Within(session),
      first,
      int.min(last, snapshot.upper),
      True,
      fn(_, _) { True },
      fn(page: Page, match) {
        let matches = case page.count >= offset && page.count - offset < limit {
          True -> [match, ..page.matches]
          False -> page.matches
        }
        Page(page.count + 1, matches)
      },
    )
  pages(scan, first - 1, Page(0, []))
  |> result.map(fn(page) { Page(page.count, list.reverse(page.matches)) })
}

/// `acc` stepped with `row` when its session is wanted and its text holds
/// the needle. Packed rows also match on field names and tags, so the text
/// decides.
fn visit(scan: Scan(a), acc: a, row: #(String, Int, BitArray)) -> a {
  let #(session, seq, payload) = row
  use <- bool.guard(!scan.wants(acc, session), acc)
  let text =
    conversation.read_input(scan.ledger, payload)
    |> result.map(tool.row_text)
    |> result.unwrap("")
  case string.contains(string.lowercase(text), scan.needle) {
    True -> scan.step(acc, Match(session, seq, text))
    False -> acc
  }
}

/// The longest ASCII stretch of `needle`: every row that contains the needle
/// contains it, and SQLite folds its case. "" (every row) when there is none.
fn probe(needle: String) -> String {
  let #(longest, _) =
    list.fold(string.to_graphemes(needle), #("", ""), fn(runs, grapheme) {
      let #(longest, current) = runs
      case string.byte_size(grapheme) == 1 {
        False -> #(longest, "")
        True -> {
          let current = current <> grapheme
          case string.length(current) > string.length(longest) {
            True -> #(current, current)
            False -> #(longest, current)
          }
        }
      }
    })
  longest
}

/// Up to 400 graphemes of `text` from a little before the first place it
/// contains `needle`, case-insensitively.
pub fn preview(text: String, needle: String) -> String {
  let at = case string.split_once(string.lowercase(text), needle) {
    Ok(#(before, _)) -> string.length(before)
    Error(_) -> 0
  }
  case at > 120 {
    True -> "…" <> tool.excerpt(string.drop_start(text, at - 100), 400)
    False -> tool.excerpt(text, 400)
  }
}

/// The rows of `session`'s durable transcript whose text contains `pattern`,
/// case-insensitively: a page of at most 20 seqs with previews, oldest first.
pub fn transcript_grep(
  ledger: store.Store,
  session: String,
  pattern: String,
  limit: Int,
  offset: Int,
) -> Result(json.Json, String) {
  use needle <- result.try(needle(pattern))
  use _ <- result.try(compaction.require(
    offset >= 0,
    "transcript search offset must be nonnegative",
  ))
  let limit = int.clamp(limit, 1, 20)
  use page <- result.try(page(
    ledger,
    session,
    needle,
    0,
    9_223_372_036_854_775_807,
    offset,
    limit,
  ))
  let next = offset + limit
  Ok(
    json.object([
      #("pattern", json.string(string.trim(pattern))),
      #("offset", json.int(offset)),
      #("count", json.int(page.count)),
      #("next_offset", tool.next_offset(next, next < page.count)),
      #(
        "rows",
        json.array(page.matches, fn(match) {
          json.object([
            #("seq", json.int(match.seq)),
            #("preview", json.string(preview(match.text, needle))),
          ])
        }),
      ),
    ]),
  )
}