//// One connection owner. All queries and transactions run in this process.

import gleam/dynamic/decode
import gleam/erlang/process.{type Subject}
import gleam/list
import gleam/otp/actor
import gleam/result
import sqlight

pub opaque type Store {
  Store(Subject(Message))
}

type Message {
  Run(fn(sqlight.Connection) -> Nil)
  Close(Subject(Nil))
}

pub fn start(path: String, schema: String) -> Result(Store, actor.StartError) {
  actor.new_with_initialiser(5000, fn(subject) {
    label("albedo_store", path)
    use db <- result.try(sqlight.open(path) |> result.map_error(message))
    case exec(db, schema) {
      Ok(_) -> Ok(actor.initialised(db) |> actor.returning(Store(subject)))
      Error(error) -> {
        let _ = sqlight.close(db)
        Error(error)
      }
    }
  })
  |> actor.on_message(fn(db, message) {
    case message {
      Run(run) -> {
        run(db)
        // A transcript load leaves every decoded row on this heap; the store
        // then idles and would hold that garbage indefinitely.
        collect_over(131_072)
        actor.continue(db)
      }
      Close(reply) -> {
        let _ = sqlight.close(db)
        process.send(reply, Nil)
        actor.stop()
      }
    }
  })
  |> actor.start
  |> result.map(fn(started) { started.data })
}

pub fn owner(store: Store) -> process.Pid {
  let Store(subject) = store
  let assert Ok(pid) = process.subject_owner(subject)
  pid
}

pub fn close(store: Store) -> Nil {
  let Store(subject) = store
  actor.call(subject, 5000, Close)
}

pub fn query(store: Store, run: fn(sqlight.Connection) -> a) -> a {
  let Store(subject) = store
  actor.call(subject, 10_000, fn(reply) {
    Run(fn(db) { process.send(reply, run(db)) })
  })
}

/// Runs one statement in the store for its effect.
pub fn write(
  store: Store,
  sql: String,
  arguments: List(sqlight.Value),
) -> Result(Nil, String) {
  query(store, run(_, sql, arguments))
}

/// Reads rows in the store.
pub fn read(
  store: Store,
  sql: String,
  arguments: List(sqlight.Value),
  decoder: decode.Decoder(a),
) -> Result(List(a), String) {
  query(store, rows(_, sql, arguments, decoder))
}

/// Runs one or more statements that return nothing, such as a schema.
pub fn exec(db: sqlight.Connection, sql: String) -> Result(Nil, String) {
  sqlight.exec(sql, db) |> result.map_error(message)
}

/// Runs one statement for its effect, discarding any rows it returns.
pub fn run(
  db: sqlight.Connection,
  sql: String,
  arguments: List(sqlight.Value),
) -> Result(Nil, String) {
  rows(db, sql, arguments, decode.dynamic) |> result.replace(Nil)
}

/// Deletes the rows `tables` keep for `session`: the cleanup an extension
/// that owns session-keyed tables registers with a `CleanPlugin`.
pub fn forget_session(
  db: sqlight.Connection,
  tables: List(String),
  session: String,
) -> Result(Nil, String) {
  list.try_each(tables, fn(table) {
    run(db, "DELETE FROM " <> table <> " WHERE session=?", [
      sqlight.text(session),
    ])
  })
}

pub fn rows(
  db: sqlight.Connection,
  sql: String,
  arguments: List(sqlight.Value),
  decoder: decode.Decoder(a),
) -> Result(List(a), String) {
  sqlight.query(sql, db, arguments, decoder) |> result.map_error(message)
}

/// The first row, or `missing` when there is none.
pub fn one(
  db: sqlight.Connection,
  sql: String,
  arguments: List(sqlight.Value),
  decoder: decode.Decoder(a),
  missing: String,
) -> Result(a, String) {
  use found <- result.try(rows(db, sql, arguments, decoder))
  list.first(found) |> result.replace_error(missing)
}

/// Runs `body` in one immediate transaction: committed when it succeeds,
/// rolled back when it fails.
pub fn transaction(
  db: sqlight.Connection,
  body: fn() -> Result(a, String),
) -> Result(a, String) {
  use _ <- result.try(exec(db, "BEGIN IMMEDIATE"))
  case body() {
    Ok(value) ->
      case exec(db, "COMMIT") {
        Ok(_) -> Ok(value)
        Error(error) -> {
          let _ = sqlight.exec("ROLLBACK", db)
          Error(error)
        }
      }
    Error(error) -> {
      let _ = sqlight.exec("ROLLBACK", db)
      Error(error)
    }
  }
}

/// Adds each `#(name, type)` column the table lacks: the only migration
/// these tables need.
pub fn add_columns(
  db: sqlight.Connection,
  table: String,
  columns: List(#(String, String)),
) -> Result(Nil, String) {
  use present <- result.try(rows(
    db,
    "PRAGMA table_info(" <> table <> ")",
    [],
    decode.field(1, decode.string, decode.success),
  ))
  columns
  |> list.filter(fn(column) { !list.contains(present, column.0) })
  |> list.try_each(fn(column) {
    exec(
      db,
      "ALTER TABLE " <> table <> " ADD COLUMN " <> column.0 <> " " <> column.1,
    )
  })
}

fn message(error: sqlight.Error) -> String {
  error.message
}

@external(erlang, "albedo_inspect", "label")
fn label(kind: String, id: String) -> Nil

@external(erlang, "albedo_session", "collect_over")
fn collect_over(words: Int) -> Nil