//// Pure collapse-policy tests for `feed_skeleton.collapse`: the reason
//// precedence (converge > import > actorBatch > single), the 6h window
//// boundary, and that ordering by recency survives collapsing.

import at_record/gen/feed/get_feed_skeleton.{
  FeedItem, FeedItemReasonReasonActorBatch, FeedItemReasonReasonImport,
  FeedItemReasonReasonSingle, FeedItemReasonReasonSubjectConverge,
  ReasonActorBatch, ReasonImport, ReasonSingle,
}
import at_record_server/catalog_index.{type Adoption, Adoption}
import at_record_server/feed_skeleton
import gleam/int
import gleam/list
import gleam/option.{None, Some}
import gleam/string

const release_a = "at://did:plc:pub/dev.mokkenstorm.crate.catalog.release/ra"

const release_b = "at://did:plc:pub/dev.mokkenstorm.crate.catalog.release/rb"

fn entry_uri(did: String, rkey: String) -> String {
  "at://" <> did <> "/dev.mokkenstorm.crate.shelf.entry/" <> rkey
}

fn row(
  did did: String,
  rkey rkey: String,
  release_uri release_uri: String,
  status status: String,
  created_at created_at: String,
  source source: option.Option(String),
) -> Adoption {
  Adoption(
    entry_uri: entry_uri(did, rkey),
    did:,
    release_uri:,
    status:,
    created_at:,
    source:,
  )
}

/// An "acquired" row with no import source -- the shape of almost every row
/// in this file that isn't specifically exercising status or source.
fn acquired_row(
  did did: String,
  rkey rkey: String,
  release_uri release_uri: String,
  created_at created_at: String,
) -> Adoption {
  row(did:, rkey:, release_uri:, status: "acquired", created_at:, source: None)
}

pub fn single_passthrough_reflects_the_row_status_test() {
  [#("acquired", "owned"), #("wanted", "wanted")]
  |> list.each(fn(c) {
    let #(status, expected_action) = c
    let rows = [
      row(
        did: "did:plc:a",
        rkey: "e1",
        release_uri: release_a,
        status:,
        created_at: "2026-01-01T00:00:00Z",
        source: None,
      ),
    ]
    let assert [FeedItem(entries:, reason: FeedItemReasonReasonSingle(inner))] =
      feed_skeleton.collapse(rows)
    assert entries
      == [get_feed_skeleton.EntryRef(actor: "did:plc:a", entry_id: "e1")]
    assert inner == ReasonSingle(action: expected_action)
  })
}

pub fn actor_batch_window_boundary_test() {
  [#("2026-01-01T06:00:00Z", True), #("2026-01-01T06:00:01Z", False)]
  |> list.each(fn(c) {
    let #(second_created_at, batches) = c
    let rows = [
      acquired_row(
        did: "did:plc:a",
        rkey: "e1",
        release_uri: release_a,
        created_at: "2026-01-01T00:00:00Z",
      ),
      acquired_row(
        did: "did:plc:a",
        rkey: "e2",
        release_uri: release_b,
        created_at: second_created_at,
      ),
    ]
    let items = feed_skeleton.collapse(rows)
    case batches {
      True -> {
        let assert [
          FeedItem(entries:, reason: FeedItemReasonReasonActorBatch(inner)),
        ] = items
        assert list.length(entries) == 2
        assert inner
          == ReasonActorBatch(
            action: "owned",
            window_end: second_created_at,
            window_start: "2026-01-01T00:00:00Z",
          )
      }
      False -> {
        assert list.length(items) == 2
        assert list.all(items, fn(item) {
          case item.reason {
            FeedItemReasonReasonSingle(_) -> True
            _ -> False
          }
        })
      }
    }
  })
}

pub fn import_via_count_collapses_five_same_did_rows_test() {
  let rows =
    list.repeat(0, 5)
    |> list.index_map(fn(_, i) { i + 1 })
    |> list.map(fn(n) {
      acquired_row(
        did: "did:plc:a",
        rkey: "e" <> int.to_string(n),
        release_uri: release_a,
        created_at: "2026-01-01T0" <> int.to_string(n) <> ":00:00Z",
      )
    })
  let assert [FeedItem(entries:, reason: FeedItemReasonReasonImport(inner))] =
    feed_skeleton.collapse(rows)
  assert list.length(entries) == 5
  assert inner == ReasonImport(source: None)
}

pub fn import_via_source_wins_regardless_of_cluster_size_test() {
  let imported_row =
    row(
      did: "did:plc:a",
      rkey: "e1",
      release_uri: release_a,
      status: "acquired",
      created_at: "2026-01-01T00:00:00Z",
      source: Some("discogs"),
    )
  let same_did_neighbor =
    acquired_row(
      did: "did:plc:a",
      rkey: "e2",
      release_uri: release_b,
      created_at: "2026-01-01T01:00:00Z",
    )
  [[imported_row], [imported_row, same_did_neighbor]]
  |> list.each(fn(rows) {
    let assert [FeedItem(reason: FeedItemReasonReasonImport(inner), ..)] =
      feed_skeleton.collapse(rows)
    assert inner == ReasonImport(source: Some("discogs"))
  })
}

pub fn subject_converge_with_two_actors_test() {
  let rows = [
    acquired_row(
      did: "did:plc:a",
      rkey: "e1",
      release_uri: release_a,
      created_at: "2026-01-01T00:00:00Z",
    ),
    row(
      did: "did:plc:b",
      rkey: "e2",
      release_uri: release_a,
      status: "wanted",
      created_at: "2026-01-01T01:00:00Z",
      source: None,
    ),
  ]
  let assert [
    FeedItem(entries:, reason: FeedItemReasonReasonSubjectConverge(inner)),
  ] = feed_skeleton.collapse(rows)
  assert list.length(entries) == 2
  assert inner.subject.uri == release_a
  // The freshest row's action ("wanted" at 01:00) drives the single action label.
  assert inner.action == "wanted"
}

pub fn subject_converge_with_three_actors_test() {
  let rows = [
    acquired_row(
      did: "did:plc:a",
      rkey: "e1",
      release_uri: release_a,
      created_at: "2026-01-01T00:00:00Z",
    ),
    acquired_row(
      did: "did:plc:b",
      rkey: "e2",
      release_uri: release_a,
      created_at: "2026-01-01T01:00:00Z",
    ),
    acquired_row(
      did: "did:plc:c",
      rkey: "e3",
      release_uri: release_a,
      created_at: "2026-01-01T02:00:00Z",
    ),
  ]
  let assert [FeedItem(entries:, ..)] = feed_skeleton.collapse(rows)
  assert list.length(entries) == 3
}

pub fn converge_takes_precedence_over_the_same_did_also_batching_test() {
  // a+b converge on release_a; a's second, unrelated row stays a lone single.
  let rows = [
    acquired_row(
      did: "did:plc:a",
      rkey: "e1",
      release_uri: release_a,
      created_at: "2026-01-01T00:00:00Z",
    ),
    acquired_row(
      did: "did:plc:b",
      rkey: "e2",
      release_uri: release_a,
      created_at: "2026-01-01T01:00:00Z",
    ),
    acquired_row(
      did: "did:plc:a",
      rkey: "e3",
      release_uri: release_b,
      created_at: "2026-01-01T02:00:00Z",
    ),
  ]
  let items = feed_skeleton.collapse(rows)
  assert list.length(items) == 2
  let assert Ok(converge) =
    list.find(items, fn(item) {
      case item.reason {
        FeedItemReasonReasonSubjectConverge(_) -> True
        _ -> False
      }
    })
  assert list.length(converge.entries) == 2
  let assert Ok(single) =
    list.find(items, fn(item) {
      case item.reason {
        FeedItemReasonReasonSingle(_) -> True
        _ -> False
      }
    })
  assert single.entries
    == [get_feed_skeleton.EntryRef(actor: "did:plc:a", entry_id: "e3")]
}

pub fn ordering_by_recency_is_preserved_for_unrelated_singles_test() {
  let rows = [
    acquired_row(
      did: "did:plc:c",
      rkey: "e3",
      release_uri: release_a,
      created_at: "2026-01-03T00:00:00Z",
    ),
    acquired_row(
      did: "did:plc:b",
      rkey: "e2",
      release_uri: release_b,
      created_at: "2026-01-02T00:00:00Z",
    ),
    acquired_row(
      did: "did:plc:a",
      rkey: "e1",
      release_uri: "at://did:plc:pub/dev.mokkenstorm.crate.catalog.release/rc",
      created_at: "2026-01-01T00:00:00Z",
    ),
  ]
  let items = feed_skeleton.collapse(rows)
  let dids =
    list.map(items, fn(item) {
      let assert [only] = item.entries
      only.actor
    })
  assert dids == ["did:plc:c", "did:plc:b", "did:plc:a"]
}

fn devnet_release_uri(n: Int) -> String {
  "at://did:plc:pub/dev.mokkenstorm.crate.catalog.release/r" <> int.to_string(n)
}

fn devnet_row(
  did: String,
  rkey: String,
  release_n: Int,
  status: String,
  created_at: String,
) -> Adoption {
  row(
    did:,
    rkey:,
    release_uri: devnet_release_uri(release_n),
    status:,
    created_at:,
    source: None,
  )
}

/// Mirrors e2e/devnet/seed-social.sh's adoption shape exactly: same dids,
/// same release ids, same T0/T_PAST timestamps. Only a genesis shelf.entry
/// (one that carries a `release` ref) becomes an adoption row -- follow-up
/// rated/annotated/regraded events reference their entry via `subject`
/// instead, so the T1/T2 timestamps the seed also stamps never show up
/// here. This is the highest-value check on the seed's own choreography
/// claims: it makes them true without spinning up a devnet.
pub fn devnet_seed_choreography_produces_the_claimed_reason_mix_test() {
  let t0 = "2026-01-01T00:00:00Z"
  let t_past = "2025-12-30T23:00:00Z"
  let rows = [
    // alice: jazz genesis rows carrying a release (R1, R2, R6).
    devnet_row("did:plc:alice", "a1", 1, "acquired", t0),
    devnet_row("did:plc:alice", "a2", 2, "acquired", t0),
    devnet_row("did:plc:alice", "a3", 6, "wanted", t0),
    // bob: krautrock rows (R3, R7, R8).
    devnet_row("did:plc:bob", "b1", 3, "acquired", t0),
    devnet_row("did:plc:bob", "b2", 7, "acquired", t0),
    devnet_row("did:plc:bob", "b3", 8, "wanted", t0),
    // carol: electronic rows (R5, R8, R7) plus the deliberate R3 converge pair.
    devnet_row("did:plc:carol", "c1", 5, "acquired", t0),
    devnet_row("did:plc:carol", "c2", 8, "acquired", t0),
    devnet_row("did:plc:carol", "c3", 7, "wanted", t0),
    devnet_row("did:plc:carol", "c4", 3, "acquired", t0),
    // dave: R4 (dave-exclusive), R2/R8 (shared -- converge away), R9/R10
    // (dave-exclusive -- actorBatch), R6 at T_PAST (reasonSingle).
    devnet_row("did:plc:dave", "d1", 4, "acquired", t0),
    devnet_row("did:plc:dave", "d2", 2, "wanted", t0),
    devnet_row("did:plc:dave", "d3", 8, "wanted", t0),
    devnet_row("did:plc:dave", "d4", 9, "acquired", t0),
    devnet_row("did:plc:dave", "d5", 10, "acquired", t0),
    devnet_row("did:plc:dave", "d6", 6, "acquired", t_past),
    // erin: R6/R8 (shared -- converge away), R11-R15 (erin-exclusive -- reasonImport).
    devnet_row("did:plc:erin", "e1", 6, "acquired", t0),
    devnet_row("did:plc:erin", "e2", 8, "acquired", t0),
    devnet_row("did:plc:erin", "e3", 11, "wanted", t0),
    devnet_row("did:plc:erin", "e4", 12, "wanted", t0),
    devnet_row("did:plc:erin", "e5", 13, "wanted", t0),
    devnet_row("did:plc:erin", "e6", 14, "wanted", t0),
    devnet_row("did:plc:erin", "e7", 15, "wanted", t0),
  ]

  let items = feed_skeleton.collapse(rows)

  let converge =
    list.filter(items, fn(item) {
      case item.reason {
        FeedItemReasonReasonSubjectConverge(_) -> True
        _ -> False
      }
    })
  let actor_batches =
    list.filter(items, fn(item) {
      case item.reason {
        FeedItemReasonReasonActorBatch(_) -> True
        _ -> False
      }
    })
  let imports =
    list.filter(items, fn(item) {
      case item.reason {
        FeedItemReasonReasonImport(_) -> True
        _ -> False
      }
    })
  let singles =
    list.filter(items, fn(item) {
      case item.reason {
        FeedItemReasonReasonSingle(_) -> True
        _ -> False
      }
    })

  // R2 (alice+dave), R3 (bob+carol), R6@T0 (alice+erin), R7 (bob+carol), and
  // R8 (bob+carol+dave+erin) converge: five genuine groups survive.
  assert list.length(converge) == 5

  // Donuts (R4) plus the two dedicated pressings (R9, R10) are the only
  // releases nobody else touches in dave's T0 window: exactly one
  // actorBatch card, 3 rows (>= actor_batch_threshold, < import_count_threshold).
  let assert [actor_batch] = actor_batches
  assert list.all(actor_batch.entries, fn(entry) {
    entry.actor == "did:plc:dave"
  })
  assert list.map(actor_batch.entries, fn(entry) { entry.entry_id })
    |> list.sort(string.compare)
    == ["d1", "d4", "d5"]

  // erin's five dedicated wanted rows (R11-R15) clear the import floor.
  let assert [import_item] = imports
  assert list.all(import_item.entries, fn(entry) {
    entry.actor == "did:plc:erin"
  })
  assert list.length(import_item.entries) == 5

  // alice (R1), carol (R5), and dave's T_PAST row (R6) all stay standalone.
  assert list.length(singles) == 3
  assert list.any(singles, fn(item) {
    item.entries
    == [get_feed_skeleton.EntryRef(actor: "did:plc:dave", entry_id: "d6")]
  })
}