use std::collections::HashMap;

use hashbrown::HashSet;
use nostrdb::Filter;

use crate::{
    relay::{
        websocket::WebsocketRelay, OutboxSubId, OutboxSubscriptions, QueuedTasks, RelayReqId,
        RelayReqStatus, RelayTask, SubPass, SubPassGuardian, SubPassRevocation,
    },
    ClientMessage,
};

/// CompactionData tracks every compaction REQ on a relay along with the
/// Outbox sub ids routed into it.
#[derive(Default)]
pub struct CompactionData {
    request_to_sid: HashMap<OutboxSubId, RelayReqId>, // we never split outbox subs over multiple REQs
    relay_subs: HashMap<RelayReqId, RelaySubData>,    // UUID
    queue: QueuedTasks,
}

impl CompactionData {
    #[allow(dead_code)]
    pub fn num_subs(&self) -> usize {
        self.relay_subs.len()
    }

    pub fn set_req_status(&mut self, sid: &str, status: RelayReqStatus) {
        let Some(data) = self.relay_subs.get_mut(sid) else {
            return;
        };

        data.status = status;
    }

    pub fn req_status(&self, id: &OutboxSubId) -> Option<RelayReqStatus> {
        let sid = self.request_to_sid.get(id)?;
        let data = self.relay_subs.get(sid)?;
        Some(data.status)
    }

    #[allow(dead_code)]
    pub fn has_eose(&self, id: &OutboxSubId) -> bool {
        self.req_status(id) == Some(RelayReqStatus::Eose)
    }

    /// Outbox subscription IDs currently placed in compaction requests.
    pub fn request_ids(&self) -> Vec<OutboxSubId> {
        self.request_to_sid.keys().copied().collect()
    }

    /// Returns the OutboxSubIds associated with the given relay subscription ID.
    pub fn ids(&self, sid: &RelayReqId) -> Option<&HashSet<OutboxSubId>> {
        self.relay_subs.get(sid).map(|d| &d.requests.requests)
    }

    /// Returns true when compaction has queued subscribe work waiting for capacity.
    pub fn has_queued_subs(&self) -> bool {
        !self.queue.is_empty()
    }

    /// Returns the current compaction request costs ordered from cheapest to most
    /// expensive for limit-downgrade planning.
    pub fn downgrade_revocation_costs(&self, subs: &OutboxSubscriptions) -> Vec<usize> {
        let mut costs = self
            .relay_subs
            .values()
            .map(|data| subs.json_size_sum(&data.requests.requests))
            .collect::<Vec<_>>();
        costs.sort_unstable();
        costs
    }
}

/// Ensures `max_subs` REQ to the websocket relay by "compacting" subscriptions (combining multiple requests into one)
pub struct CompactionRelay<'a> {
    ctx: CompactionCtx<'a>,
    sub_guardian: &'a mut SubPassGuardian,
    json_limit: usize,
}

/// CompactionRelay ensures multiple Outbox subscriptions are packed into as few
/// REQs as possible, respecting per-relay limits.
impl<'a> CompactionRelay<'a> {
    pub fn new(
        relay: Option<&'a mut WebsocketRelay>,
        data: &'a mut CompactionData,
        json_limit: usize,
        sub_guardian: &'a mut SubPassGuardian,
        subs: &'a OutboxSubscriptions,
    ) -> Self {
        let ctx = match relay {
            Some(relay) => CompactionCtx::Active(CompactionHandler::new(relay, data, subs)),
            None => CompactionCtx::Inactive {
                data,
                session: CompactionSubSession::default(),
                subs,
            },
        };
        Self {
            ctx,
            sub_guardian,
            json_limit,
        }
    }

    #[profiling::function]
    pub fn ingest_session(mut self, session: CompactionSession) -> HashSet<OutboxSubId> {
        self.execute_session(session, true);
        self.take_session_invalidations()
    }

    /// Applies the explicit compaction session without opportunistically draining
    /// queued compaction work afterward.
    #[profiling::function]
    pub fn ingest_session_without_queue_drain(
        mut self,
        session: CompactionSession,
    ) -> HashSet<OutboxSubId> {
        self.execute_session(session, false);
        self.take_session_invalidations()
    }

    /// Attempts to place queued compaction work using whatever pass capacity is
    /// currently free.
    #[profiling::function]
    pub fn drain_queue(mut self) -> HashSet<OutboxSubId> {
        self.drain_queue_internal();
        self.take_session_invalidations()
    }

    fn execute_session(&mut self, session: CompactionSession, drain_queue: bool) {
        let request_free = session.request_free;
        let mut reserved: Vec<SubPass> = Vec::new();

        // Reserve passes - take from guardian or compact to free them
        while reserved.len() < request_free {
            if let Some(pass) = self.sub_guardian.take_pass() {
                reserved.push(pass);
            } else if let Some(ejected_pass) = self.compact() {
                reserved.push(ejected_pass);
            } else {
                break;
            }
        }

        // Process session (can't touch reserved passes)
        self.ingest_session_internal(session);

        if drain_queue {
            self.drain_queue_internal();
        }

        // Return reserved passes
        for pass in reserved {
            self.sub_guardian.return_pass(pass);
        }
    }

    fn drain_queue_internal(&mut self) {
        profiling::scope!("drain queue");
        let attempts = self.ctx.data().queue.len();
        for _ in 0..attempts {
            let Some(id) = self.ctx.data().queue.pop() else {
                break;
            };
            self.subscribe(id);
        }
    }

    #[profiling::function]
    fn ingest_session_internal(&mut self, session: CompactionSession) {
        for (id, task) in session.tasks {
            match task {
                RelayTask::Unsubscribe => {
                    self.unsubscribe(id);
                }
                RelayTask::Subscribe => {
                    self.subscribe(id);
                }
            }
        }
    }

    #[profiling::function]
    pub fn handle_relay_open(&mut self) -> HashSet<OutboxSubId> {
        let CompactionCtx::Active(handler) = &mut self.ctx else {
            return HashSet::new();
        };

        if !handler.relay.is_connected() {
            return HashSet::new();
        }

        let mut invalidated = HashSet::new();
        for (sid, sub_data) in &mut handler.data.relay_subs {
            let filters = handler
                .subs
                .filters_all_for_compaction(&sub_data.requests.requests);
            if are_filters_empty(&filters) {
                continue;
            }

            sub_data.status = RelayReqStatus::InitialQuery;
            handler
                .relay
                .conn
                .send(&ClientMessage::req(sid.to_string(), filters));
            invalidated.extend(sub_data.requests.requests.iter().copied());
        }

        invalidated
    }

    #[allow(dead_code)]
    pub fn revocate(&mut self, mut revocation: SubPassRevocation) -> HashSet<OutboxSubId> {
        let Some(pass) = self.compact() else {
            // this shouldn't be possible
            return HashSet::new();
        };

        revocation.revocate(pass);
        self.take_session_invalidations()
    }

    #[allow(dead_code)]
    pub fn revocate_all(&mut self, revocations: Vec<SubPassRevocation>) -> HashSet<OutboxSubId> {
        let mut invalidated = HashSet::new();
        for revocation in revocations {
            invalidated.extend(self.revocate(revocation));
        }
        invalidated
    }

    #[profiling::function]
    fn compact(&mut self) -> Option<SubPass> {
        let SharedCtx {
            data,
            session,
            subs,
        } = self.ctx.shared();

        let (id, smallest) = take_smallest_sub_reqs(subs, &mut data.relay_subs)?;

        session.tasks.insert(id, SubSessionTask::Removed);
        session
            .invalidated_sub_ids
            .extend(smallest.requests.requests.iter().copied());
        for id in smallest.requests.requests {
            self.ctx.data().request_to_sid.remove(&id);
            self.place(id);
        }

        Some(smallest.sub_pass)
    }

    #[profiling::function]
    fn new_sub(&mut self, id: OutboxSubId) -> PlaceResult {
        let Some(new_pass) = self.sub_guardian.take_pass() else {
            // pass not available, try to place on an existing sub
            return self.place(id);
        };

        let relay_id = RelayReqId::default();
        let mut requests = SubRequests::default();
        requests.add(id);

        let SharedCtx {
            data,
            session,
            subs: _,
        } = self.ctx.shared();
        data.relay_subs.insert(
            relay_id.clone(),
            RelaySubData {
                requests,
                status: RelayReqStatus::InitialQuery,
                sub_pass: new_pass,
            },
        );
        data.request_to_sid.insert(id, relay_id.clone());
        session.tasks.insert(relay_id, SubSessionTask::New);
        PlaceResult::Placed
    }

    #[profiling::function]
    pub fn subscribe(&mut self, id: OutboxSubId) -> PlaceResult {
        let SharedCtx {
            data,
            session,
            subs: _,
        } = self.ctx.shared();
        let Some(relay_id) = data.request_to_sid.get(&id) else {
            return self.new_sub(id);
        };

        let Some(sub_data) = data.relay_subs.get_mut(relay_id) else {
            return self.new_sub(id);
        };

        // modifying a filter
        sub_data.requests.add(id);

        sub_data.status = RelayReqStatus::InitialQuery;

        session
            .tasks
            .insert(relay_id.clone(), SubSessionTask::Touched);
        tracing::debug!("Placed {id:?} on an existing subscription: {relay_id:?}");
        PlaceResult::Placed
    }

    #[profiling::function]
    pub fn unsubscribe(&mut self, id: OutboxSubId) {
        let SharedCtx {
            data: compaction_data,
            session,
            subs: _,
        } = self.ctx.shared();
        let Some(relay_id) = compaction_data.request_to_sid.remove(&id) else {
            compaction_data.queue.cancel(id);
            return;
        };
        session.invalidated_sub_ids.insert(id);

        let Some(data) = compaction_data.relay_subs.get_mut(&relay_id) else {
            compaction_data.queue.cancel(id);
            return;
        };

        data.status = RelayReqStatus::InitialQuery;

        if !data.requests.remove(&id) {
            return;
        }

        if !data.requests.is_empty() {
            session
                .tasks
                .insert(relay_id.clone(), SubSessionTask::Touched);
            return;
        }

        let Some(data) = compaction_data.relay_subs.remove(&relay_id) else {
            return;
        };

        self.sub_guardian.return_pass(data.sub_pass);
        tracing::debug!("Unsubed from last internal id in REQ, returning pass");
        session
            .tasks
            .insert(relay_id.clone(), SubSessionTask::Removed);
    }

    #[profiling::function]
    fn place(&mut self, id: OutboxSubId) -> PlaceResult {
        let SharedCtx {
            data,
            session,
            subs,
        } = self.ctx.shared();
        let placed_on = 'place: {
            for (relay_id, relay_data) in &mut data.relay_subs {
                if !relay_data.requests.can_fit(subs, &id, self.json_limit) {
                    continue;
                }

                session
                    .tasks
                    .insert(relay_id.clone(), SubSessionTask::Touched);
                relay_data.requests.add(id);
                break 'place Some(relay_id.clone());
            }

            None
        };

        if let Some(relay_id) = placed_on {
            data.request_to_sid.insert(id, relay_id);
            return PlaceResult::Placed;
        }

        data.queue.enqueue(id);
        PlaceResult::Queued
    }

    /// Returns and clears sub IDs whose compaction relay legs were reset by
    /// the current relay session.
    pub fn take_session_invalidations(&mut self) -> HashSet<OutboxSubId> {
        let SharedCtx {
            data,
            session,
            subs: _,
        } = self.ctx.shared();
        let mut invalidated = std::mem::take(&mut session.invalidated_sub_ids);

        for (sid, task) in &session.tasks {
            if !matches!(task, SubSessionTask::Touched | SubSessionTask::New) {
                continue;
            }

            let Some(relay_sub) = data.relay_subs.get(sid) else {
                continue;
            };
            invalidated.extend(relay_sub.requests.requests.iter().copied());
        }

        invalidated
    }
}

#[derive(Debug, PartialEq, Eq)]
pub enum PlaceResult {
    Placed,
    Queued,
}

fn take_smallest_sub_reqs(
    subs: &OutboxSubscriptions,
    data: &mut HashMap<RelayReqId, RelaySubData>,
) -> Option<(RelayReqId, RelaySubData)> {
    let mut smallest = usize::MAX;
    let mut res = None;

    for (id, d) in data.iter() {
        let cur_size = subs.json_size_sum(&d.requests.requests);
        if cur_size < smallest {
            smallest = cur_size;
            res = Some(id.clone());
        }
    }

    let id = res?;

    data.remove(&id).map(|r| (id, r))
}

#[derive(Default)]
struct CompactionSubSession {
    tasks: HashMap<RelayReqId, SubSessionTask>,
    invalidated_sub_ids: HashSet<OutboxSubId>,
}

enum SubSessionTask {
    New,
    Touched,
    Removed,
}

enum CompactionCtx<'a> {
    Active(CompactionHandler<'a>),
    Inactive {
        data: &'a mut CompactionData,
        session: CompactionSubSession,
        subs: &'a OutboxSubscriptions,
    },
}

impl<'a> CompactionCtx<'a> {
    #[profiling::function]
    pub fn shared(&mut self) -> SharedCtx<'_> {
        match self {
            CompactionCtx::Active(compaction_handler) => SharedCtx {
                data: compaction_handler.data,
                session: &mut compaction_handler.session,
                subs: compaction_handler.subs,
            },
            CompactionCtx::Inactive {
                data,
                session,
                subs,
            } => SharedCtx {
                data,
                session,
                subs,
            },
        }
    }

    pub fn data(&mut self) -> &mut CompactionData {
        match self {
            CompactionCtx::Active(compaction_handler) => compaction_handler.data,
            CompactionCtx::Inactive {
                data,
                session: _,
                subs: _,
            } => data,
        }
    }
}
struct SharedCtx<'a> {
    data: &'a mut CompactionData,
    session: &'a mut CompactionSubSession,
    subs: &'a OutboxSubscriptions,
}

struct CompactionHandler<'a> {
    relay: &'a mut WebsocketRelay,
    data: &'a mut CompactionData,
    subs: &'a OutboxSubscriptions,
    pub session: CompactionSubSession,
}

impl<'a> Drop for CompactionHandler<'a> {
    #[profiling::function]
    fn drop(&mut self) {
        for (id, task) in &self.session.tasks {
            match task {
                SubSessionTask::Touched => {
                    let Some(data) = self.data.relay_subs.get_mut(id) else {
                        continue;
                    };

                    let filters = self
                        .subs
                        .filters_all_for_compaction(&data.requests.requests);

                    if filters.is_empty() {
                        self.relay.conn.send(&ClientMessage::close(id.0.clone()));
                    } else {
                        self.relay
                            .conn
                            .send(&ClientMessage::req(id.0.clone(), filters));
                    }
                }
                SubSessionTask::Removed => {
                    self.relay.conn.send(&ClientMessage::close(id.0.clone()));
                }
                SubSessionTask::New => {
                    let Some(data) = self.data.relay_subs.get(id) else {
                        continue;
                    };

                    let filters = self
                        .subs
                        .filters_all_for_compaction(&data.requests.requests);
                    self.relay
                        .conn
                        .send(&ClientMessage::req(id.0.clone(), filters));
                }
            }
        }
    }
}

fn are_filters_empty(filters: &Vec<Filter>) -> bool {
    if filters.is_empty() {
        return true;
    }

    for filter in filters {
        if filter.num_elements() != 0 {
            return false;
        }
    }

    true
}

impl<'a> CompactionHandler<'a> {
    pub fn new(
        relay: &'a mut WebsocketRelay,
        data: &'a mut CompactionData,
        subs: &'a OutboxSubscriptions,
    ) -> Self {
        Self {
            relay,
            data,
            session: CompactionSubSession::default(),
            subs,
        }
    }
}

/// Represents a singular REQ to a relay
struct RelaySubData {
    requests: SubRequests,
    status: RelayReqStatus,
    sub_pass: SubPass,
}

#[derive(Default)]
struct SubRequests {
    pub requests: HashSet<OutboxSubId>,
}

impl SubRequests {
    #[profiling::function]
    pub fn add(&mut self, id: OutboxSubId) {
        self.requests.insert(id);
    }

    pub fn remove(&mut self, id: &OutboxSubId) -> bool {
        self.requests.remove(id)
    }

    pub fn is_empty(&self) -> bool {
        self.requests.is_empty()
    }

    pub fn can_fit(
        &self,
        subs: &OutboxSubscriptions,
        new: &OutboxSubId,
        json_limit: usize,
    ) -> bool {
        let Some(new_size) = subs.json_size(new) else {
            return true;
        };

        let cur_json_size = subs.json_size_sum(&self.requests);

        // `["REQ","abc...123"]`;
        //  12345678  ...    90 -> 10 characters excluding the UUID
        cur_json_size + new_size + 10 + RelayReqId::byte_len() <= json_limit
    }
}

#[derive(Default)]
pub struct CompactionSession {
    // Number of subs which should be free after ingestion. Subs will compact enough to free up that number of subs
    // OR as much as possible without dropping any existing subs
    request_free: usize,
    tasks: HashMap<OutboxSubId, RelayTask>,
}

impl CompactionSession {
    pub fn request_free_subs(&mut self, num_free: usize) {
        self.request_free = num_free;
    }

    pub fn unsub(&mut self, unsub: OutboxSubId) {
        self.tasks.insert(unsub, RelayTask::Unsubscribe);
    }

    pub fn sub(&mut self, id: OutboxSubId) {
        self.tasks.insert(id, RelayTask::Subscribe);
    }

    pub fn is_empty(&self) -> bool {
        self.tasks.is_empty() && self.request_free == 0
    }

    #[cfg(test)]
    pub(super) fn task_for_test(&self, id: &OutboxSubId) -> Option<RelayTask> {
        self.tasks.get(id).copied()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        relay::{test_utils::MockWakeup, RelayStatus, RelayUrlPkgs, SubscribeTask},
        WebsocketConn,
    };
    use hashbrown::HashSet;

    // ==================== CompactionData tests ====================

    #[test]
    fn compaction_data_default_empty() {
        let data = CompactionData::default();
        assert_eq!(data.num_subs(), 0);
    }

    #[test]
    fn compaction_data_req_status_none_for_unknown() {
        let data = CompactionData::default();
        assert!(data.req_status(&OutboxSubId(999)).is_none());
    }

    #[test]
    fn compaction_data_has_eose_false_for_unknown() {
        let data = CompactionData::default();
        assert!(!data.has_eose(&OutboxSubId(999)));
    }

    #[test]
    fn compaction_data_set_req_status_ignores_unknown_sid() {
        let mut data = CompactionData::default();
        // Should not panic or error when setting status for unknown sid
        data.set_req_status("unknown-sid", RelayReqStatus::Eose);
    }

    #[test]
    fn compaction_data_ids_returns_sub_ids() {
        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(1);
        let pass = guardian.take_pass().unwrap();

        let id = OutboxSubId(7);
        let relay_id = RelayReqId::from("req-123");
        let mut requests = SubRequests::default();
        requests.add(id);
        data.relay_subs.insert(
            relay_id.clone(),
            RelaySubData {
                requests,
                status: RelayReqStatus::InitialQuery,
                sub_pass: pass,
            },
        );

        let ids = data.ids(&relay_id);
        assert!(ids.is_some());
        assert!(ids.unwrap().contains(&id));
    }

    #[test]
    fn compaction_data_set_req_status_updates_status() {
        let mut data = CompactionData::default();

        // Manually set up a relay subscription
        let relay_id = RelayReqId::from("test-sid");
        let mut guardian = SubPassGuardian::new(1);
        let pass = guardian.take_pass().unwrap();

        data.relay_subs.insert(
            relay_id.clone(),
            RelaySubData {
                requests: SubRequests::default(),
                status: RelayReqStatus::InitialQuery,
                sub_pass: pass,
            },
        );

        // Set EOSE should update status
        data.set_req_status("test-sid", RelayReqStatus::Eose);

        // Verify status was set
        let sub_data = data.relay_subs.get(&relay_id).unwrap();
        assert_eq!(sub_data.status, RelayReqStatus::Eose);
    }

    // ==================== SubRequests tests ====================

    /// can_fit returns true when combined JSON size is under the limit.
    #[test]
    fn sub_requests_can_fit() {
        use crate::relay::{RelayUrlPkgs, SubscribeTask};
        use hashbrown::HashSet;

        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            OutboxSubId(0),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        let requests = SubRequests::default();

        assert!(requests.can_fit(&subs, &OutboxSubId(0), 1_000_000));
        assert!(!requests.can_fit(&subs, &OutboxSubId(0), 5));
    }

    // ==================== CompactionSession tests ====================

    #[test]
    fn compaction_session_default() {
        let session = CompactionSession::default();
        assert_eq!(session.request_free, 0);
        assert!(session.tasks.is_empty());
    }

    #[test]
    fn compaction_session_unsub() {
        let mut session = CompactionSession::default();
        session.unsub(OutboxSubId(42));

        assert!(session.tasks.contains_key(&OutboxSubId(42)));
        match session.tasks.get(&OutboxSubId(42)) {
            Some(RelayTask::Unsubscribe) => (),
            _ => panic!("Expected Unsubscribe task"),
        }
    }

    #[test]
    fn compaction_session_sub() {
        let mut session = CompactionSession::default();
        session.sub(OutboxSubId(1));

        assert!(session.tasks.contains_key(&OutboxSubId(1)));
        assert!(matches!(
            session.tasks.get(&OutboxSubId(1)),
            Some(RelayTask::Subscribe)
        ));
    }

    // ==================== take_smallest_sub_reqs tests ====================

    #[test]
    fn take_smallest_returns_none_for_empty() {
        let subs = OutboxSubscriptions::default();
        let mut data: HashMap<RelayReqId, RelaySubData> = HashMap::new();
        assert!(take_smallest_sub_reqs(&subs, &mut data).is_none());
    }

    /// Returns the relay sub with the smallest combined JSON size.
    #[test]
    fn take_smallest_returns_smallest_by_json_size() {
        use crate::relay::{RelayUrlPkgs, SubscribeTask};
        use hashbrown::HashSet;

        // Register subscriptions with different JSON sizes
        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            OutboxSubId(0),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            OutboxSubId(1),
            SubscribeTask {
                filters: vec![Filter::new()
                    .kinds(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
                    .build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        let mut guardian = SubPassGuardian::new(2);

        // Small relay sub contains id 0
        let mut small_requests = SubRequests::default();
        small_requests.add(OutboxSubId(0));

        // Large relay sub contains id 1
        let mut large_requests = SubRequests::default();
        large_requests.add(OutboxSubId(1));

        let mut data: HashMap<RelayReqId, RelaySubData> = HashMap::new();
        data.insert(
            RelayReqId::from("small"),
            RelaySubData {
                requests: small_requests,
                status: RelayReqStatus::InitialQuery,
                sub_pass: guardian.take_pass().unwrap(),
            },
        );
        data.insert(
            RelayReqId::from("large"),
            RelaySubData {
                requests: large_requests,
                status: RelayReqStatus::InitialQuery,
                sub_pass: guardian.take_pass().unwrap(),
            },
        );

        let (id, _) = take_smallest_sub_reqs(&subs, &mut data).unwrap();
        assert_eq!(id.0, "small");
        assert_eq!(data.len(), 1);
    }

    #[test]
    fn take_smallest_removes_from_map() {
        let subs = OutboxSubscriptions::default();
        let mut data: HashMap<RelayReqId, RelaySubData> = HashMap::new();
        let mut guardian = SubPassGuardian::new(1);

        data.insert(
            RelayReqId::from("only"),
            RelaySubData {
                requests: SubRequests::default(),
                status: RelayReqStatus::InitialQuery,
                sub_pass: guardian.take_pass().unwrap(),
            },
        );

        let result = take_smallest_sub_reqs(&subs, &mut data);
        assert!(result.is_some());
        assert!(data.is_empty());
    }

    // ==================== CompactionRelay tests ====================

    /// Requesting free subs when there's nothing to compact has no effect.
    #[test]
    fn compact_returns_none_when_no_subs() {
        let subs = OutboxSubscriptions::default();
        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(5);
        let json_limit = 100000;

        let initial_passes = guardian.available_passes();

        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        session.request_free_subs(1);
        relay.ingest_session(session);

        assert_eq!(guardian.available_passes(), initial_passes);
    }

    /// Compacting frees a pass and redistributes requests to remaining subs.
    #[test]
    fn compact_frees_pass_and_redistributes() {
        use crate::relay::{RelayUrlPkgs, SubscribeTask};
        use hashbrown::HashSet;

        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            OutboxSubId(0),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            OutboxSubId(1),
            SubscribeTask {
                filters: vec![Filter::new()
                    .kinds(vec![2, 3, 4, 5, 6, 7, 8, 9, 10])
                    .build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(5);
        let json_limit = 100000;

        // Create 2 relay subs
        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        session.sub(OutboxSubId(0));
        session.sub(OutboxSubId(1));
        relay.ingest_session(session);

        assert_eq!(data.relay_subs.len(), 2);
        assert_eq!(guardian.available_passes(), 3); // 5 - 2

        // Request 4 free passes - must compact 1
        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        session.request_free_subs(4);
        relay.ingest_session(session);

        assert_eq!(data.relay_subs.len(), 1);
        assert_eq!(guardian.available_passes(), 4);

        let remaining = data.relay_subs.values().next().unwrap();
        assert_eq!(remaining.requests.requests.len(), 2);
    }

    /// When compaction redistributes a request but the remaining sub
    /// doesn't have room, the request goes to the queue.
    #[test]
    fn place_queues_when_no_room() {
        use crate::relay::{RelayUrlPkgs, SubscribeTask};
        use hashbrown::HashSet;

        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            OutboxSubId(0),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            OutboxSubId(1),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![2]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        // Set limit so combined filters won't fit in one REQ
        let size0 = subs.json_size(&OutboxSubId(0)).unwrap();
        let size1 = subs.json_size(&OutboxSubId(1)).unwrap();
        let json_limit = size0 + size1 - 1;

        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(2);

        // Create 2 relay subs at capacity
        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        session.sub(OutboxSubId(0));
        session.sub(OutboxSubId(1));
        relay.ingest_session(session);

        assert_eq!(data.relay_subs.len(), 2);
        assert!(data.queue.is_empty());

        // Compact 1 - redistributed request won't fit
        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        session.request_free_subs(1);
        relay.ingest_session(session);

        assert_eq!(data.relay_subs.len(), 1);
        assert!(!data.queue.is_empty());
    }

    /// When no passes are available, requests are placed on existing relay subs.
    #[test]
    fn new_sub_places_on_existing_when_no_passes() {
        use crate::relay::{RelayUrlPkgs, SubscribeTask};
        use hashbrown::HashSet;

        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            OutboxSubId(0),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            OutboxSubId(1),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![2]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(1); // Only 1 pass
        let json_limit = 100000;

        // Add 2 requests with only 1 pass - second must go on existing
        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        session.sub(OutboxSubId(0));
        session.sub(OutboxSubId(1));
        relay.ingest_session(session);

        assert_eq!(data.relay_subs.len(), 1);
        let sub = data.relay_subs.values().next().unwrap();
        assert_eq!(sub.requests.requests.len(), 2);
    }

    /// Subscriptions placed onto an existing compacted REQ must register
    /// request-to-relay mapping so a later unsubscribe updates the correct REQ.
    #[test]
    fn unsubscribe_after_place_on_existing_removes_request() {
        use crate::relay::{RelayUrlPkgs, SubscribeTask};
        use hashbrown::HashSet;

        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            OutboxSubId(0),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            OutboxSubId(1),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![2]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(1); // Force second sub onto existing REQ
        let json_limit = 100000;

        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        session.sub(OutboxSubId(0));
        session.sub(OutboxSubId(1));
        relay.ingest_session(session);

        assert_eq!(data.relay_subs.len(), 1);
        let relay_id = data.relay_subs.keys().next().cloned().unwrap();
        assert_eq!(data.request_to_sid.get(&OutboxSubId(0)), Some(&relay_id));
        assert_eq!(data.request_to_sid.get(&OutboxSubId(1)), Some(&relay_id));

        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        session.unsub(OutboxSubId(1));
        relay.ingest_session(session);

        assert!(data.queue.is_empty());
        assert_eq!(data.relay_subs.len(), 1);
        let sub = data.relay_subs.get(&relay_id).unwrap();
        assert_eq!(sub.requests.requests.len(), 1);
        assert!(sub.requests.requests.contains(&OutboxSubId(0)));
        assert!(!sub.requests.requests.contains(&OutboxSubId(1)));
        assert_eq!(data.request_to_sid.get(&OutboxSubId(0)), Some(&relay_id));
        assert!(!data.request_to_sid.contains_key(&OutboxSubId(1)));
    }

    /// Touching one compacted REQ must invalidate every sub carried by that REQ.
    #[test]
    fn ingest_session_reports_invalidations_for_all_touched_compaction_subs() {
        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            OutboxSubId(0),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            OutboxSubId(1),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![2]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(1);
        let json_limit = 100_000;

        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut create = CompactionSession::default();
        create.sub(OutboxSubId(0));
        assert_eq!(
            relay.ingest_session(create),
            HashSet::from([OutboxSubId(0)])
        );

        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut touch = CompactionSession::default();
        touch.sub(OutboxSubId(1));
        let invalidated = relay.ingest_session(touch);

        assert_eq!(
            invalidated,
            HashSet::from([OutboxSubId(0), OutboxSubId(1)]),
            "adding one sub to an existing compacted REQ reissues that REQ for every carried sub"
        );
    }

    #[tokio::test]
    async fn handle_relay_open_reports_all_reissued_compaction_sub_ids() {
        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            OutboxSubId(0),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            OutboxSubId(1),
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![2]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(1);
        let json_limit = 100_000;
        let mut websocket = WebsocketRelay::new(
            WebsocketConn::from_wakeup(
                nostr::RelayUrl::parse("wss://compaction-replay.example.com").unwrap(),
                MockWakeup::default(),
            )
            .unwrap(),
        );
        websocket.conn.set_status(RelayStatus::Connected);

        let relay = CompactionRelay::new(
            Some(&mut websocket),
            &mut data,
            json_limit,
            &mut guardian,
            &subs,
        );
        let mut session = CompactionSession::default();
        session.sub(OutboxSubId(0));
        session.sub(OutboxSubId(1));
        let _ = relay.ingest_session(session);

        let sid0 = data
            .request_to_sid
            .get(&OutboxSubId(0))
            .expect("sid for first compaction sub")
            .clone();
        let sid1 = data
            .request_to_sid
            .get(&OutboxSubId(1))
            .expect("sid for second compaction sub")
            .clone();
        data.set_req_status(&sid0.to_string(), RelayReqStatus::Eose);
        data.set_req_status(&sid1.to_string(), RelayReqStatus::Eose);

        let invalidated = {
            let mut relay = CompactionRelay::new(
                Some(&mut websocket),
                &mut data,
                json_limit,
                &mut guardian,
                &subs,
            );
            relay.handle_relay_open()
        };

        assert_eq!(
            invalidated,
            HashSet::from([OutboxSubId(0), OutboxSubId(1)]),
            "relay-open replay should invalidate every sub carried by the reissued compaction REQ"
        );
        assert_eq!(
            data.req_status(&OutboxSubId(0)),
            Some(RelayReqStatus::InitialQuery),
            "relay-open replay must reset compaction req status to InitialQuery"
        );
        assert_eq!(
            data.req_status(&OutboxSubId(1)),
            Some(RelayReqStatus::InitialQuery),
            "relay-open replay must reset compaction req status to InitialQuery"
        );
    }

    /// When requesting multiple free passes, multiple subs are compacted
    /// and all requests are consolidated into fewer relay subs.
    #[test]
    fn compact_multiple_subs() {
        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(3);
        let json_limit = 100000;
        let mut subs = OutboxSubscriptions::default();
        for i in 0..3 {
            subs.new_subscription(
                OutboxSubId(i),
                SubscribeTask {
                    filters: vec![Filter::new().kinds(vec![i + 1]).build()],
                    relays: RelayUrlPkgs::new(HashSet::new()),
                },
                false,
            );
        }

        // Create 3 subs and request 2 free in same session
        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        let mut session = CompactionSession::default();
        for i in 0..3 {
            session.sub(OutboxSubId(i));
        }
        session.request_free_subs(2);
        relay.ingest_session(session);

        // Should compact down to 1 sub with all 3 requests
        assert_eq!(data.relay_subs.len(), 1);
        assert_eq!(guardian.available_passes(), 2);

        let sub = data.relay_subs.values().next().unwrap();
        assert_eq!(sub.requests.requests.len(), 3);
    }

    /// One unplaceable queued item should not block other queued items that can be placed.
    #[test]
    fn queue_drain_does_not_starve_placeable_items() {
        use crate::relay::{RelayUrlPkgs, SubscribeTask};
        use hashbrown::HashSet;

        let id_seed = OutboxSubId(0);
        let id_blocked = OutboxSubId(1);
        let id_placeable = OutboxSubId(2);

        let mut subs = OutboxSubscriptions::default();
        subs.new_subscription(
            id_seed,
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![1]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            id_blocked,
            SubscribeTask {
                filters: vec![Filter::new().kinds((2u64..40).collect::<Vec<_>>()).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );
        subs.new_subscription(
            id_placeable,
            SubscribeTask {
                filters: vec![Filter::new().kinds(vec![3]).build()],
                relays: RelayUrlPkgs::new(HashSet::new()),
            },
            false,
        );

        let overhead = 10 + RelayReqId::byte_len();
        let seed_size = subs.json_size(&id_seed).unwrap();
        let blocked_size = subs.json_size(&id_blocked).unwrap();
        let placeable_size = subs.json_size(&id_placeable).unwrap();
        assert!(blocked_size > placeable_size);
        let json_limit = seed_size + placeable_size + overhead;
        assert!(seed_size + blocked_size + overhead > json_limit);

        let mut data = CompactionData::default();
        let mut guardian = SubPassGuardian::new(1);
        let seed_pass = guardian.take_pass().unwrap();

        let relay_id = RelayReqId::from("seed");
        let mut requests = SubRequests::default();
        requests.add(id_seed);
        data.relay_subs.insert(
            relay_id.clone(),
            RelaySubData {
                requests,
                status: RelayReqStatus::InitialQuery,
                sub_pass: seed_pass,
            },
        );
        data.request_to_sid.insert(id_seed, relay_id.clone());

        data.queue.enqueue(id_blocked);
        data.queue.enqueue(id_placeable);

        let relay = CompactionRelay::new(None, &mut data, json_limit, &mut guardian, &subs);
        relay.ingest_session(CompactionSession::default());

        assert_eq!(data.request_to_sid.get(&id_seed), Some(&relay_id));
        assert_eq!(data.request_to_sid.get(&id_placeable), Some(&relay_id));
        assert!(!data.request_to_sid.contains_key(&id_blocked));
        assert_eq!(data.queue.len(), 1);
    }
}
