//// Existing transcript reference and image BLOB migrations, in that order.

import albedo/daemon/image_payloads
import albedo/daemon/images
import albedo/daemon/migrations/backup as snapshot
import albedo/daemon/store
import gleam/dynamic/decode
import gleam/list
import gleam/result
import sqlight

/// Moves inline images in existing transcript rows into `images`, once per
/// database. The database is first copied to `backup` with VACUUM INTO. Rows
/// are rewritten a page at a time, each page in its own transaction, so an
/// interrupted run resumes where it stopped (a rewritten row has nothing inline).
pub fn run(ledger: store.Store, backup: String) -> Result(Int, String) {
  use moved <- result.try(migrate_legacy(ledger, backup))
  use _ <- result.try(migrate_blobs(ledger, backup))
  Ok(moved)
}

fn migrate_legacy(ledger: store.Store, backup: String) -> Result(Int, String) {
  let read = images.reader(ledger)
  use applied <- result.try(
    store.read(
      ledger,
      "SELECT 1 FROM migrations WHERE name='image_store'",
      [],
      decode.dynamic,
    )
    |> result.map(fn(rows) { rows != [] }),
  )
  case applied {
    True -> Ok(0)
    False -> {
      use pending <- result.try(store.read(
        ledger,
        "SELECT count(*) FROM transcript WHERE instr(payload,CAST('user_image' AS BLOB))>0 OR instr(payload,CAST('tool_output' AS BLOB))>0",
        [],
        decode.field(0, decode.int, decode.success),
      ))
      use _ <- result.try(case pending {
        [0] -> Ok(Nil)
        _ -> {
          snapshot.image_store(ledger, backup)
        }
      })
      use moved <- result.try(migrate_pages(ledger, read, -1, 0))
      use _ <- result.try(
        store.write(
          ledger,
          "INSERT OR IGNORE INTO migrations(name,applied_at) VALUES('image_store',unixepoch())",
          [],
        ),
      )
      Ok(moved)
    }
  }
}

/// Existing installations used TEXT for the base64 data. Convert in bounded
/// transactions after taking a copy of the database. SQLite's BLOB affinity
/// does not convert an existing TEXT value on its own.
fn migrate_blobs(ledger: store.Store, backup: String) -> Result(Nil, String) {
  use pending <- result.try(store.read(
    ledger,
    "SELECT count(*) FROM images WHERE typeof(data)='text'",
    [],
    decode.field(0, decode.int, decode.success),
  ))
  case pending {
    [0] -> Ok(Nil)
    _ -> {
      use _ <- result.try(snapshot.image_store(ledger, backup))
      migrate_blob_pages(ledger)
    }
  }
}

fn migrate_blob_pages(ledger: store.Store) -> Result(Nil, String) {
  let page =
    store.query(ledger, fn(db) {
      use rows <- result.try(
        store.rows(
          db,
          "SELECT hash,data FROM images WHERE typeof(data)='text' LIMIT ?",
          [sqlight.int(migrate_page_rows)],
          {
            use hash <- decode.field(0, decode.string)
            use data <- decode.field(1, decode.string)
            decode.success(#(hash, data))
          },
        ),
      )
      use _ <- result.try(
        store.transaction(db, fn() {
          list.try_each(rows, fn(row) {
            case decode_legacy_base64(row.1) {
              Ok(bytes) ->
                store.run(
                  db,
                  "UPDATE images SET data=? WHERE hash=? AND typeof(data)='text'",
                  [sqlight.blob(bytes), sqlight.text(row.0)],
                )
              Error(_) -> Error("invalid base64 image at hash " <> row.0)
            }
          })
        }),
      )
      Ok(list.length(rows))
    })
  case page {
    Ok(0) -> Ok(Nil)
    Ok(_) -> migrate_blob_pages(ledger)
    Error(e) -> Error("image blob migration failed: " <> e)
  }
}

/// Rows per migration transaction; a page of screenshot rows is tens of MB.
const migrate_page_rows = 16

fn migrate_pages(
  ledger: store.Store,
  read: fn(String) -> Result(String, Nil),
  after: Int,
  moved: Int,
) -> Result(Int, String) {
  let page =
    store.query(ledger, fn(db) {
      use rows <- result.try(
        store.rows(
          db,
          "SELECT seq,payload FROM transcript WHERE seq>? AND (instr(payload,CAST('user_image' AS BLOB))>0 OR instr(payload,CAST('tool_output' AS BLOB))>0) ORDER BY seq LIMIT ?",
          [sqlight.int(after), sqlight.int(migrate_page_rows)],
          {
            use seq <- decode.field(0, decode.int)
            use payload <- decode.field(1, decode.bit_array)
            decode.success(#(seq, payload))
          },
        ),
      )
      use count <- result.try(
        store.transaction(db, fn() {
          list.try_fold(rows, 0, fn(count, row) {
            case migrate_row(row.1, read) {
              Keep -> Ok(count)
              Rewrite(payload, blobs) -> {
                use _ <- result.try(image_payloads.insert(db, blobs))
                store.run(db, "UPDATE transcript SET payload=? WHERE seq=?", [
                  sqlight.blob(payload),
                  sqlight.int(row.0),
                ])
                |> result.replace(count + 1)
              }
            }
          })
        }),
      )
      Ok(#(count, list.last(rows) |> result.map(fn(row) { row.0 })))
    })
  case page {
    Error(e) -> Error("image store migration failed: " <> e)
    Ok(#(count, Ok(last))) -> migrate_pages(ledger, read, last, moved + count)
    Ok(#(count, Error(_))) -> Ok(moved + count)
  }
}

type Migration {
  Keep
  Rewrite(payload: BitArray, blobs: List(#(String, String)))
}

@external(erlang, "albedo_images", "migrate")
fn migrate_row(
  payload: BitArray,
  read: fn(String) -> Result(String, Nil),
) -> Migration

@external(erlang, "albedo_images", "decode_legacy_base64")
fn decode_legacy_base64(data: String) -> Result(BitArray, Nil)