#[derive(Debug)]
struct DataplaneOwnerShardRetireWorker {
    completed: VecDeque<CryptoCompletionBatch>,
}

impl DataplaneOwnerShardRetireWorker {
    fn new() -> Self {
        Self {
            completed: VecDeque::new(),
        }
    }

    fn queue_completion_batch(&mut self, batch: CryptoCompletionBatch) -> bool {
        if batch.is_empty() {
            return false;
        }
        let was_empty = self.completed.is_empty();
        self.completed.push_back(batch);
        was_empty
    }

    fn retire_queued_completions_into(
        &mut self,
        owner_shard: &mut DataplaneOwnerShard,
        limit: usize,
        retired: &mut DataplaneRetiredOutputSink<'_>,
        drops: &mut Vec<PacketDrop>,
        compact_endpoint_data: bool,
    ) -> usize {
        let mut retired_count = 0usize;
        while retired_count < limit {
            let Some(mut batch) = self.completed.pop_front() else {
                break;
            };
            let batch_limit = limit.saturating_sub(retired_count);
            let pending = if batch.len() > batch_limit {
                Some(batch.split_off(batch_limit))
            } else {
                None
            };
            let batch_len = batch.len();
            owner_shard.retire_completion_batch_into(
                batch,
                retired,
                drops,
                compact_endpoint_data,
            );
            retired_count = retired_count.saturating_add(batch_len);
            if let Some(pending) = pending {
                self.completed.push_front(pending);
                break;
            }
        }
        retired_count
    }

    fn has_queued_completions(&self) -> bool {
        !self.completed.is_empty()
    }
}

#[derive(Debug)]
struct DataplaneOwnerShard {
    index: usize,
    admission: AdmissionQueue,
    outbound_admission: OutboundAdmissionQueue,
    owners: HashMap<OwnerId, OwnerState>,
}

impl DataplaneOwnerShard {
    fn new(index: usize) -> Self {
        Self {
            index,
            admission: AdmissionQueue::new(),
            outbound_admission: OutboundAdmissionQueue::new(),
            owners: HashMap::new(),
        }
    }

    fn register_owner(&mut self, owner: OwnerId, config: OwnerConfig) {
        self.owners.insert(owner, OwnerState::new(owner, config));
    }

    fn unregister_owner(&mut self, owner: OwnerId) -> bool {
        self.owners.remove(&owner).is_some()
    }

    fn has_owner(&self, owner: OwnerId) -> bool {
        self.owners.contains_key(&owner)
    }

    fn fsp_owner_destinations(&self, destinations: &mut Vec<NodeAddr>) {
        destinations.extend(self.owners.keys().filter_map(|owner| {
            (owner.protocol() == PacketProtocol::Fsp).then_some(owner.node_addr())
        }));
    }

    fn owner_active_path(&self, owner: OwnerId) -> Option<TransportPath> {
        self.owners
            .get(&owner)
            .and_then(OwnerState::active_path)
    }

    fn owner_fsp_next_hop(&self, owner: OwnerId) -> Option<NodeAddr> {
        self.owners
            .get(&owner)
            .and_then(OwnerState::fsp_wrap_next_hop)
    }

    fn owner_mut(&mut self, owner: OwnerId) -> Option<&mut OwnerState> {
        self.owners.get_mut(&owner)
    }

    fn owner(&self, owner: OwnerId) -> Option<&OwnerState> {
        self.owners.get(&owner)
    }

    fn owner_fsp_activity(&self, owner: OwnerId) -> Option<DataplaneFspOwnerActivity> {
        self.owner(owner).and_then(OwnerState::fsp_activity)
    }

    fn owner_has_fsp_pending_receive_epoch(
        &self,
        owner: OwnerId,
        received_k_bit: bool,
    ) -> bool {
        self.owner(owner)
            .is_some_and(|owner| owner.has_fsp_pending_receive_epoch(received_k_bit))
    }

    fn owner_has_fmp_pending_receive_epoch(
        &self,
        owner: OwnerId,
        received_k_bit: bool,
    ) -> bool {
        self.owner(owner)
            .is_some_and(|owner| owner.has_fmp_pending_receive_epoch(received_k_bit))
    }

    fn owner_fsp_mmp_snapshot(&self, owner: OwnerId) -> Option<DataplaneFspMmpSnapshot> {
        self.owner(owner).and_then(OwnerState::fsp_mmp_snapshot)
    }

    fn owner_fsp_send_context(&self, owner: OwnerId) -> Option<DataplaneFspSendContext> {
        self.owner(owner).and_then(OwnerState::fsp_send_context)
    }

    fn owner_fmp_send_context(&self, owner: OwnerId) -> Option<DataplaneFmpSendContext> {
        self.owner(owner).and_then(OwnerState::fmp_send_context)
    }

    fn owner_fmp_link_metrics(
        &self,
        owner: OwnerId,
        now: std::time::Instant,
    ) -> Option<DataplaneFmpLinkMetrics> {
        self.owner(owner)
            .and_then(|owner| owner.fmp_link_metrics(now))
    }

    fn owner_fmp_link_cost(&self, owner: OwnerId) -> Option<f64> {
        self.owner(owner).and_then(OwnerState::fmp_link_cost)
    }

    fn owner_fmp_has_srtt(&self, owner: OwnerId) -> bool {
        self.owner(owner).is_some_and(OwnerState::fmp_has_srtt)
    }

    fn collect_fmp_mmp_reports(
        &mut self,
        now: std::time::Instant,
        batch: &mut DataplaneFmpMmpReportBatch,
    ) {
        for owner in self.owners.values_mut() {
            owner.collect_fmp_mmp_reports(now, batch);
        }
    }

    fn collect_fsp_mmp_reports(
        &mut self,
        now: std::time::Instant,
        batch: &mut DataplaneFspMmpReportBatch,
    ) {
        for owner in self.owners.values_mut() {
            owner.collect_fsp_mmp_reports(now, batch);
        }
    }

    fn record_fsp_mmp_send_result(
        &mut self,
        owner: OwnerId,
        success: bool,
    ) -> Option<DataplaneFspMmpReportingResumed> {
        self.owner_mut(owner)
            .and_then(|owner| owner.record_fsp_mmp_send_result(success))
    }

    fn seed_fsp_path_mtu(
        &mut self,
        owner: OwnerId,
        path_mtu: u16,
    ) -> Result<(), DataplaneFspMmpSkip> {
        self.owner_mut(owner)
            .ok_or(DataplaneFspMmpSkip::UnknownOwner)?
            .seed_fsp_path_mtu(path_mtu)
    }

    fn process_fsp_mmp_receiver_report(
        &mut self,
        owner: OwnerId,
        rr: &crate::mmp::report::ReceiverReport,
        last_outbound_next_hop: Option<NodeAddr>,
        now_ms: u64,
        now: std::time::Instant,
        min_loss_sample: u64,
    ) -> Result<DataplaneFspReceiverReportResult, DataplaneFspMmpSkip> {
        self.owner_mut(owner)
            .ok_or(DataplaneFspMmpSkip::UnknownOwner)?
            .process_fsp_mmp_receiver_report(
                rr,
                last_outbound_next_hop,
                now_ms,
                now,
                min_loss_sample,
            )
    }

    fn apply_fsp_path_mtu_signal(
        &mut self,
        owner: OwnerId,
        path_mtu: u16,
        now: std::time::Instant,
    ) -> Result<DataplaneFspPathMtuApplyResult, DataplaneFspMmpSkip> {
        self.owner_mut(owner)
            .ok_or(DataplaneFspMmpSkip::UnknownOwner)?
            .apply_fsp_path_mtu_signal(path_mtu, now)
    }

    fn min_fsp_rx_age_for_next_hop(&self, next_hop: &NodeAddr, now_ms: u64) -> Option<u64> {
        self.owners
            .values()
            .filter_map(OwnerState::fsp_activity)
            .filter(|activity| activity.tracks_next_hop(next_hop))
            .filter_map(|activity| activity.last_rx_age_ms(now_ms))
            .min()
    }

    fn min_fsp_data_rx_age_for_next_hop(&self, next_hop: &NodeAddr, now_ms: u64) -> Option<u64> {
        self.owners
            .values()
            .filter_map(OwnerState::fsp_activity)
            .filter(|activity| activity.tracks_data_next_hop(next_hop))
            .filter_map(|activity| activity.last_rx_data_age_ms(now_ms))
            .min()
    }

    fn any_fsp_recent_outbound_without_inbound_for_next_hop(
        &self,
        next_hop: &NodeAddr,
        now_ms: u64,
        timeout_ms: u64,
    ) -> bool {
        self.owners
            .values()
            .filter_map(OwnerState::fsp_activity)
            .filter(|activity| activity.tracks_next_hop(next_hop))
            .any(|activity| activity.has_recent_outbound_without_inbound(now_ms, timeout_ms))
    }

    fn submit_socket_packet_with_seq(
        &mut self,
        packet: SocketPacket,
        ingress_seq: u64,
    ) -> bool {
        self.admission.admit_with_seq(packet, ingress_seq)
    }

    fn submit_socket_packet_run_with_seq(
        &mut self,
        packets: Vec<SocketPacket>,
        first_seq: u64,
    ) -> bool {
        self.admission.admit_run_with_seq(packets, first_seq)
    }

    fn submit_outbound_packet_with_seq(
        &mut self,
        packet: OutboundPacket,
        ingress_seq: u64,
    ) -> bool {
        self.outbound_admission.admit_with_seq(packet, ingress_seq)
    }

    fn submit_outbound_packet_run_with_seq(
        &mut self,
        packets: Vec<OutboundPacket>,
        first_seq: u64,
    ) -> bool {
        self.outbound_admission
            .admit_run_with_seq(packets, first_seq)
    }

    fn dispatch_ingress_prepared_into(
        &mut self,
        limit: usize,
        prepared: &mut Vec<PreparedCryptoWork>,
        completion_batches: &mut Vec<CryptoCompletionBatch>,
        priority_only: bool,
        record_fsp_path_open: bool,
        fsp_path_open: &mut u64,
        fsp_path_open_bulk: &mut u64,
        drops: &mut Vec<PacketDrop>,
    ) -> usize {
        let mut dispatched = 0usize;
        let mut attempts_remaining = self.admission.len();
        while dispatched < limit && attempts_remaining > 0 {
            let run_limit = limit.saturating_sub(dispatched);
            let Some(mut run) = self.admission.pop_next_run(priority_only, run_limit) else {
                if !priority_only && limit > 0 {
                    crate::perf_profile::record_event(
                        crate::perf_profile::Event::DataplaneDispatchNoIngress,
                    );
                }
                break;
            };
            attempts_remaining = attempts_remaining.saturating_sub(run.items.len());
            if run.items.len() > 1 {
                crate::perf_profile::record_event_count(
                    crate::perf_profile::Event::DataplaneIngressOwnerRunContinue,
                    run.items.len().saturating_sub(1) as u64,
                );
            }
            let owner_id = run.cursor.owner;

            let Some(owner) = self.owners.get_mut(&owner_id) else {
                for queued in &run.items {
                    drops.push(PacketDrop::from_queued(
                        queued,
                        PacketDropReason::UnknownOwner,
                    ));
                }
                self.admission.continue_owner_lane(run.cursor);
                continue;
            };

            let mut remaining = Vec::new();
            let mut items = std::mem::take(&mut run.items).into_iter();
            while let Some(queued) = items.next() {
                if !owner.can_reserve_class(queued.packet.class) {
                    record_ingress_owner_blocked(owner.reserve_block_reason(queued.packet.class));
                    remaining.push(queued);
                    remaining.extend(items);
                    break;
                }

                match owner.reserve(&queued.packet, queued.ingress_seq) {
                    Ok((reservation, open_key)) => {
                        let packet_owner = queued.packet.owner;
                        let packet_counter = queued.packet.counter;
                        let packet_lane = queued.packet.lane();
                        let reservation = reservation.with_owner_shard(self.index);
                        if record_fsp_path_open {
                            count_fsp_path_open_dispatch(
                                &reservation,
                                fsp_path_open,
                                fsp_path_open_bulk,
                            );
                        }
                        match open_key {
                            Some(open_key) => prepared.push(PreparedCryptoWork::open(
                                CryptoWork {
                                    reservation,
                                    packet: queued.packet,
                                },
                                open_key,
                            )),
                            None => CryptoCompletionBatch::push_grouped(
                                failed_crypto_completion(reservation, CryptoFailureKind::Open),
                                completion_batches,
                            ),
                        }
                        tracing::debug!(
                            owner = ?packet_owner,
                            counter = packet_counter,
                            lane = ?packet_lane,
                            "dataplane inbound dispatched"
                        );
                        dispatched = dispatched.saturating_add(1);
                        attempts_remaining = self.admission.len();
                    }
                    Err(error) => {
                        tracing::debug!(
                            owner = ?queued.packet.owner,
                            counter = queued.packet.counter,
                            generation = queued.packet.generation,
                            class = ?queued.packet.class,
                            lane = ?queued.packet.lane(),
                            wire_flags = queued.packet.wire_flags,
                            receive_epoch = ?queued.packet.receive_epoch,
                            ingress_seq = queued.ingress_seq,
                            reason = ?error,
                            "dataplane inbound reservation failed"
                        );
                        drops.push(PacketDrop::from_queued(&queued, error.into()));
                    }
                }
            }

            if remaining.is_empty() {
                self.admission.continue_owner_lane(run.cursor);
            } else {
                run.items = remaining;
                self.admission.defer_owner_run(run);
            }
        }

        if !priority_only && limit > 0 && dispatched >= limit {
            crate::perf_profile::record_event(
                crate::perf_profile::Event::DataplaneDispatchLimitHit,
            );
        }
        dispatched
    }

    fn dispatch_outbound_prepared_into(
        &mut self,
        limit: usize,
        prepared: &mut Vec<PreparedCryptoWork>,
        completion_batches: &mut Vec<CryptoCompletionBatch>,
        priority_only: bool,
        drops: &mut Vec<PacketDrop>,
    ) -> usize {
        let mut dispatched = 0usize;
        let mut attempts_remaining = self.outbound_admission.len();
        while dispatched < limit && attempts_remaining > 0 {
            let run_limit = limit.saturating_sub(dispatched);
            let Some(mut run) = self
                .outbound_admission
                .pop_next_run(priority_only, run_limit)
            else {
                break;
            };
            attempts_remaining = attempts_remaining.saturating_sub(run.items.len());
            if run.items.len() > 1 {
                crate::perf_profile::record_event_count(
                    crate::perf_profile::Event::DataplaneOutboundOwnerRunContinue,
                    run.items.len().saturating_sub(1) as u64,
                );
            }
            let owner_id = run.cursor.owner;

            let Some(owner) = self.owners.get_mut(&owner_id) else {
                for queued in &run.items {
                    drops.push(PacketDrop::from_queued_outbound(
                        queued,
                        PacketDropReason::UnknownOwner,
                    ));
                }
                self.outbound_admission.continue_owner_lane(run.cursor);
                continue;
            };

            let mut remaining = Vec::new();
            let mut items = std::mem::take(&mut run.items).into_iter();
            while let Some(queued) = items.next() {
                let class = queued.packet.class;
                let ingress_seq = queued.ingress_seq;
                if !owner.can_reserve_class(class) {
                    record_outbound_owner_blocked(owner.reserve_block_reason(class));
                    remaining.push(queued);
                    remaining.extend(items);
                    break;
                }

                let send_token = queued.packet.send_token;
                match owner.reserve_outbound(queued.packet, ingress_seq) {
                    Ok((reservation, packet)) => {
                        let reservation = reservation.with_owner_shard(self.index);
                        match owner.seal_key() {
                            Some(seal_key) => prepared.push(PreparedCryptoWork::seal(
                                OutboundCryptoWork {
                                    reservation,
                                    packet,
                                },
                                seal_key,
                            )),
                            None => CryptoCompletionBatch::push_grouped(
                                failed_crypto_completion(reservation, CryptoFailureKind::Seal),
                                completion_batches,
                            ),
                        }
                        dispatched = dispatched.saturating_add(1);
                    }
                    Err(error) => {
                        drops.push(PacketDrop {
                            owner: owner_id,
                            counter: None,
                            send_token,
                            reason: error.into(),
                            crypto_failure: None,
                            wire_flags: None,
                            authenticated_counter_highest: None,
                        });
                    }
                }
                attempts_remaining = self.outbound_admission.len();
            }

            if remaining.is_empty() {
                self.outbound_admission.continue_owner_lane(run.cursor);
            } else {
                run.items = remaining;
                self.outbound_admission.defer_owner_run(run);
            }
        }
        dispatched
    }

    pub(crate) fn retire_completion_batch_into(
        &mut self,
        batch: CryptoCompletionBatch,
        retired: &mut DataplaneRetiredOutputSink<'_>,
        drops: &mut Vec<PacketDrop>,
        compact_endpoint_data: bool,
    ) {
        let _timer =
            crate::perf_profile::Timer::start(crate::perf_profile::Stage::DataplaneRetire);
        let owner_id = batch.owner();
        let Some(owner) = self.owners.get_mut(&owner_id) else {
            for completion in batch.into_completions() {
                let drop = PacketDrop::from_completion(
                    &completion,
                    PacketDropReason::UnknownOwner,
                    None,
                );
                drops.push(drop);
            }
            return;
        };
        let before_in_flight = owner.in_flight;
        owner.retire_batch_outputs_into(batch, retired, drops, compact_endpoint_data);
        if owner.in_flight < before_in_flight {
            self.admission.wake_owner(owner_id);
            self.outbound_admission.wake_owner(owner_id);
        }
    }

    fn admission_queue_lens(&self) -> (usize, usize) {
        self.admission.lens()
    }

    fn admission_ready_lens(&self) -> (usize, usize) {
        self.admission.ready_lens()
    }

    fn outbound_admission_queue_lens(&self) -> (usize, usize) {
        self.outbound_admission.lens()
    }

    fn outbound_admission_ready_lens(&self) -> (usize, usize) {
        self.outbound_admission.ready_lens()
    }

    fn record_authenticated_fsp_session(
        &mut self,
        session: DataplaneAuthenticatedFspSession,
    ) -> Option<bool> {
        self.owner_mut(session.owner)
            .and_then(|owner| owner.record_authenticated_fsp_session(session))
    }

    fn record_fsp_decrypt_failure(&mut self, owner: OwnerId) -> Option<u32> {
        self.owner_mut(owner)
            .and_then(OwnerState::record_fsp_decrypt_failure)
    }

}
