//// Pure feed-skeleton collapse policy: folds a page of raw adoption rows
//// (already filtered to the intended audience, sorted `created_at` desc)
//// into the reason-tagged `FeedItem`s the lexicon defines. Precedence:
//// subject convergence beats same-actor batching, which beats a lone
//// passthrough. No IO, no clock -- every window comparison works off the
//// rows' own `created_at` strings.

import atproto/uri
import crate/gen/defs
import crate/gen/feed/get_feed_skeleton.{
  type FeedItem, EntryRef, FeedItem, FeedItemReasonReasonActorBatch,
  FeedItemReasonReasonImport, FeedItemReasonReasonSingle,
  FeedItemReasonReasonSubjectConverge, ReasonActorBatch, ReasonImport,
  ReasonSingle, ReasonSubjectConverge,
}
import crate_server/catalog_index.{type Adoption}
import gleam/dict
import gleam/list
import gleam/option.{None, Some}
import gleam/order
import gleam/time/duration
import gleam/time/timestamp.{type Timestamp}

/// The span within which same-actor or same-subject rows collapse together.
const batch_window_hours = 6

/// >= this many same-did rows in one window collapse to an actor batch.
const actor_batch_threshold = 2

/// >= this many same-did rows in one window is always an import, batch or not.
const import_count_threshold = 5

/// >= this many distinct dids on the same release in one window converge.
const converge_threshold = 2

const discogs_source = "discogs"

/// One adoption row paired with its parsed timestamp, so window comparisons
/// never re-parse `created_at`.
type Timed =
  #(Adoption, Timestamp)

pub fn collapse(adoptions: List(Adoption)) -> List(FeedItem) {
  let #(converge_groups, leftover_groups) =
    cluster_by_window(adoptions, fn(a) { a.release_uri })
    |> list.partition(fn(cluster) {
      list.length(unique_dids(cluster)) >= converge_threshold
    })
  let converge_items = list.map(converge_groups, to_converge_item)
  let actor_items =
    leftover_groups
    |> list.flatten
    |> list.map(fn(pair) { pair.0 })
    |> cluster_by_window(fn(a) { a.did })
    |> list.map(to_actor_item)
  list.append(converge_items, actor_items)
  |> list.sort(order.reverse(by_time))
  |> list.map(fn(pair) { pair.0 })
}

fn unique_dids(cluster: List(Timed)) -> List(String) {
  cluster |> list.map(fn(pair) { pair.0.did }) |> list.unique
}

fn by_time(a: #(t, Timestamp), b: #(t, Timestamp)) -> order.Order {
  timestamp.compare(a.1, b.1)
}

fn to_converge_item(cluster: List(Timed)) -> #(FeedItem, Timestamp) {
  let #(latest_row, latest_ts) = latest_of(cluster)
  let entries =
    cluster
    |> list.map(fn(pair) { pair.0.did })
    |> list.unique
    |> list.map(fn(did) {
      let #(row, _) =
        latest_of(list.filter(cluster, fn(pair) { pair.0.did == did }))
      EntryRef(actor: did, entry_id: entry_id_of(row))
    })
  // `cid` is only known once catalog.getRelease(uri) is hydrated downstream;
  // this placeholder mirrors the same pattern as BrowseRow's cover_cid
  // round-trip in catalog_index_postgres.
  let subject =
    defs.CatalogRef(cid: "", external_ids: None, uri: latest_row.release_uri)
  let item =
    FeedItem(
      entries:,
      reason: FeedItemReasonReasonSubjectConverge(ReasonSubjectConverge(
        action: reason_action(latest_row.status),
        subject:,
      )),
    )
  #(item, latest_ts)
}

fn to_actor_item(cluster: List(Timed)) -> #(FeedItem, Timestamp) {
  let #(latest_row, latest_ts) = latest_of(cluster)
  let #(earliest_row, _) = earliest_of(cluster)
  let count = list.length(cluster)
  let has_discogs =
    list.any(cluster, fn(pair) { pair.0.source == Some(discogs_source) })
  let entries =
    list.map(cluster, fn(pair) {
      EntryRef(actor: pair.0.did, entry_id: entry_id_of(pair.0))
    })
  let reason = case
    has_discogs,
    count >= import_count_threshold,
    count >= actor_batch_threshold
  {
    True, _, _ ->
      FeedItemReasonReasonImport(ReasonImport(source: Some(discogs_source)))
    False, True, _ -> FeedItemReasonReasonImport(ReasonImport(source: None))
    False, False, True ->
      FeedItemReasonReasonActorBatch(ReasonActorBatch(
        action: reason_action(latest_row.status),
        window_end: latest_row.created_at,
        window_start: earliest_row.created_at,
      ))
    False, False, False ->
      FeedItemReasonReasonSingle(
        ReasonSingle(action: reason_action(latest_row.status)),
      )
  }
  #(FeedItem(entries:, reason:), latest_ts)
}

/// The feed reason vocabulary is only "owned"/"wanted" -- everything but an
/// explicit `wanted` shelf.entry action (acquisitions, regrades, ...) reads
/// as owned, since gone rows never reach `collapse` (the store already
/// filters them out).
fn reason_action(status: String) -> String {
  case status {
    "wanted" -> "wanted"
    _ -> "owned"
  }
}

fn entry_id_of(a: Adoption) -> String {
  uri.rkey(a.entry_uri)
}

fn latest_of(cluster: List(Timed)) -> Timed {
  let assert Ok(row) = list.last(cluster)
  row
}

fn earliest_of(cluster: List(Timed)) -> Timed {
  let assert Ok(row) = list.first(cluster)
  row
}

fn parse_ts(a: Adoption) -> Timestamp {
  case timestamp.parse_rfc3339(a.created_at) {
    Ok(ts) -> ts
    Error(_) -> timestamp.from_unix_seconds(0)
  }
}

/// Groups `rows` by `key_of`, then splits each group into window-bounded
/// clusters (ascending by time, each cluster's span from its earliest row
/// capped at `batch_window_hours`).
fn cluster_by_window(
  rows: List(Adoption),
  key_of: fn(Adoption) -> String,
) -> List(List(Timed)) {
  rows
  |> list.map(fn(a) { #(a, parse_ts(a)) })
  |> list.group(fn(pair) { key_of(pair.0) })
  |> dict.values
  |> list.flat_map(fn(group) {
    group
    |> list.sort(by_time)
    |> split_into_windows
  })
}

/// Walks time-ascending rows once, opening a new cluster whenever a row
/// falls outside the window anchored at the current cluster's earliest row.
fn split_into_windows(rows: List(Timed)) -> List(List(Timed)) {
  case rows {
    [] -> []
    [first, ..rest] -> {
      let #(_, current, done) =
        list.fold(rest, #(first.1, [first], []), fn(acc, row) {
          let #(anchor, current, done) = acc
          case within_window(anchor, row.1) {
            True -> #(anchor, [row, ..current], done)
            False -> #(row.1, [row], [list.reverse(current), ..done])
          }
        })
      list.reverse([list.reverse(current), ..done])
    }
  }
}

fn within_window(start: Timestamp, candidate: Timestamp) -> Bool {
  // `difference(left, right)` is `right - left`, so this is `candidate - start`
  // (non-negative: candidate is always >= start, rows are sorted ascending).
  duration.compare(
    timestamp.difference(start, candidate),
    duration.hours(batch_window_hours),
  )
  != order.Gt
}