//// Round-trip checks for the Postgres backend, gated on `DATABASE_URL` (a
//// no-op when unset); run with a live Postgres to regression-test it.

import crate/gen/repo/strong_ref.{type RepoStrongRef}
import crate/gen/shelf/entry.{type ShelfEntry, ShelfEntry}
import crate_server/oauth/sessions_postgres
import crate_server/shelf_index
import crate_server/shelf_index_postgres
import envoy
import exception
import gleam/erlang/process
import gleam/option.{None, Some}

// A single-connection pool, killed as soon as this call returns (even if
// `run` itself crashes): each test opens its own fresh pool, and a normal
// `connect_pool` here would leak a whole `pool_size`-10 pool per test for
// the rest of the suite's run, exhausting `max_connections` well before the
// last postgres-gated test finishes -- the actual CI failure this guards
// against (see `shelf_index_postgres.gleam`'s migration docblock for detail
// on the shape that took here specifically).
fn with_store(run: fn(shelf_index.Store) -> Nil) -> Nil {
  case envoy.get("DATABASE_URL") {
    Error(Nil) -> Nil
    Ok(url) -> {
      let assert Ok(#(conn, pid)) = sessions_postgres.connect_test_pool(url)
      // `pog.start` links the pool to its caller; `kill`'s exit signal is
      // untrappable and propagates through links regardless of trap_exit,
      // so unlink first or this test process goes down with its own pool.
      use <- exception.defer(fn() {
        process.unlink(pid)
        process.kill(pid)
      })
      let assert Ok(store) = shelf_index_postgres.table_store(conn)
      run(store)
    }
  }
}

fn blank_entry(
  action action: String,
  created_at created_at: String,
  subject subject: option.Option(RepoStrongRef),
) -> ShelfEntry {
  ShelfEntry(
    action:,
    counterparty: Some("discogs:seller123"),
    created_at:,
    external_ids: None,
    folder: None,
    media_grade: Some("VG+"),
    notes: Some("first pressing"),
    price: None,
    rating: None,
    release: None,
    sleeve_grade: None,
    snapshot: None,
    source: None,
    subject:,
  )
}

pub fn event_round_trip_test() {
  use store <- with_store()
  let entry_uri = "at://did:plc:pgtest/dev.mokkenstorm.crate.shelf.entry/pg1"
  let event =
    shelf_index.event_row(
      event_uri: entry_uri,
      entry_uri:,
      did: "did:plc:pgtest",
      rkey: "pg1",
      entry: blank_entry(
        action: "acquired",
        created_at: "2024-01-01T00:00:00Z",
        subject: None,
      ),
    )
  store.events.upsert(event)
  let assert Some(found) = store.events.get(entry_uri)
  assert found.entry_uri == entry_uri
  assert found.did == "did:plc:pgtest"
  assert found.action == "acquired"
  let assert [only] = store.events.for_entry(entry_uri)
  assert only.event_uri == entry_uri
  store.events.delete(entry_uri)
  assert store.events.get(entry_uri) == None
}

pub fn record_and_fold_round_trips_flattened_columns_test() {
  use store <- with_store()
  let entry_uri = "at://did:plc:pgtest/dev.mokkenstorm.crate.shelf.entry/pg2"
  shelf_index.record_and_fold(
    store,
    shelf_index.event_row(
      event_uri: entry_uri,
      entry_uri:,
      did: "did:plc:pgtest",
      rkey: "pg2",
      entry: blank_entry(
        action: "acquired",
        created_at: "2024-01-01T00:00:00Z",
        subject: None,
      ),
    ),
  )
  let assert Some(found) = store.entries.get(entry_uri)
  assert found.did == "did:plc:pgtest"
  assert found.status == "owned"
  assert found.media_grade == Some("VG+")
  assert found.notes == Some("first pressing")
  assert found.counterparty == Some("discogs:seller123")
  assert found.created_at == "2024-01-01T00:00:00Z"

  let assert [in_list] = store.entries.list_for_did("did:plc:pgtest")
  assert in_list.entry_uri == entry_uri

  shelf_index.delete_and_fold(store, entry_uri, entry_uri)
  assert store.entries.get(entry_uri) == None
  assert store.events.for_entry(entry_uri) == []
}

pub fn seen_round_trip_test() {
  use store <- with_store()
  let did = "did:plc:pgtest-seen"
  assert store.seen.has_seen(did) == False
  store.seen.mark_seen(did)
  assert store.seen.has_seen(did) == True
}

pub fn counts_reflect_stored_rows_and_go_back_down_on_delete_test() {
  use store <- with_store()
  let did = "did:plc:pgtest-counts"
  let entries_before = store.entries.count()
  let seen_before = store.seen.count()
  let entry_uri = "at://" <> did <> "/dev.mokkenstorm.crate.shelf.entry/pgc1"
  shelf_index.record_and_fold(
    store,
    shelf_index.event_row(
      event_uri: entry_uri,
      entry_uri:,
      did:,
      rkey: "pgc1",
      entry: blank_entry(
        action: "acquired",
        created_at: "2024-01-01T00:00:00Z",
        subject: None,
      ),
    ),
  )
  store.seen.mark_seen(did)
  assert store.events.count_for_did(did) == 1
  assert store.entries.count() == entries_before + 1
  assert store.seen.count() == seen_before + 1
  store.entries.delete_for_did(did)
  assert store.events.count_for_did(did) == 0
  assert store.entries.count() == entries_before
}

pub fn delete_for_did_round_trip_test() {
  use store <- with_store()
  let did = "did:plc:pgtest-purge"
  let entry_uri = "at://" <> did <> "/dev.mokkenstorm.crate.shelf.entry/pg3"
  shelf_index.record_and_fold(
    store,
    shelf_index.event_row(
      event_uri: entry_uri,
      entry_uri:,
      did:,
      rkey: "pg3",
      entry: blank_entry(
        action: "acquired",
        created_at: "2024-01-01T00:00:00Z",
        subject: None,
      ),
    ),
  )
  store.seen.mark_seen(did)
  store.entries.delete_for_did(did)
  assert store.entries.get(entry_uri) == None
  assert store.events.for_entry(entry_uri) == []
  assert store.seen.has_seen(did) == False
}