// Copyright 2026 PingCAP, Inc.

use super::*;

#[derive(Clone, Copy, Debug)]
struct UnsafeRecoverDestroyTask {
    region_id: u64,
    keyspace_id: Option<u32>,
}

impl MergedEngine {
    /// Validates that every requested region can be cleared immediately.
    ///
    /// The request must be dependency-closed over the current local replay
    /// state. Dependents that are already tombstoned are ignored.
    pub fn unsafe_recover_check_clear_regions(&self, requested_regions: &[u64]) -> Result<()> {
        if requested_regions.is_empty() {
            return Ok(());
        }

        let requested_set: HashSet<u64> = requested_regions.iter().copied().collect();
        for &region_id in requested_regions {
            let mut missing_dependents = Vec::new();
            self.raft.with_dependents(region_id, |dependents| {
                missing_dependents.extend(
                    dependents
                        .iter()
                        .copied()
                        .filter(|dependent_id| !requested_set.contains(dependent_id))
                        .filter(|dependent_id| {
                            self.raft.get_truncated_index(*dependent_id) != Some(TRUNCATE_ALL_INDEX)
                        }),
                );
            });
            if !missing_dependents.is_empty() {
                missing_dependents.sort_unstable();
                return Err(Error::Other(box_err!(
                    "region {} still has dependent regions outside request: {:?}",
                    region_id,
                    missing_dependents
                )));
            }
        }
        Ok(())
    }

    /// Clears the requested local replay regions and returns every region that
    /// was actually destroyed by the call.
    ///
    /// Requested regions are destroyed synchronously. Additional delayed
    /// tombstone parents may also be destroyed once their last dependent edge
    /// is removed.
    pub fn unsafe_recover_execute_clear_regions(
        &mut self,
        requested_regions: &[u64],
    ) -> Result<Vec<u64>> {
        if requested_regions.is_empty() {
            return Ok(Vec::new());
        }

        let mut pending = Vec::with_capacity(requested_regions.len());
        pending.extend(requested_regions.iter().copied().map(|region_id| {
            UnsafeRecoverDestroyTask {
                region_id,
                keyspace_id: None,
            }
        }));

        let mut destroyed_regions = Vec::with_capacity(requested_regions.len());
        let mut raft_wb = WriteBatch::new();
        let mut next = 0;

        // Intentionally destroy requested regions one by one in the caller-visible
        // order instead of reordering them dependents-first. An interrupted
        // admin clear is rare, and restart can rebuild the dependency edges
        // from replay/bootstrap state before apply resumes.
        while let Some(task) = pending.get(next).copied() {
            next += 1;
            let keyspace_id =
                self.unsafe_recover_take_destroy_keyspace_id(task.region_id, task.keyspace_id)?;
            let parent_id = self.unsafe_recover_parent_id(task.region_id);
            if let Some(parent_id) = parent_id {
                let remaining = self.raft.remove_dependent(parent_id, task.region_id);
                if remaining == 0 {
                    if let Some(parent_keyspace_id) = self.delay_destroy_regions.remove(&parent_id)
                    {
                        info!(
                            "unsafe_recover_execute_clear_regions: destroy delayed parent";
                            "parent_region" => parent_id,
                            "cleared_child_region" => task.region_id,
                        );
                        pending.push(UnsafeRecoverDestroyTask {
                            region_id: parent_id,
                            keyspace_id: Some(parent_keyspace_id),
                        });
                    }
                }
            }

            // Mirror the logic in `destroy_regions` for removing all state and metadata for
            // the region, except clear the region progress in additional.
            self.raft
                .iterate_peer_states(task.region_id, false, |k, _| {
                    raft_wb.set_state_bytes(
                        task.region_id,
                        task.region_id,
                        keyspace_id,
                        k.clone(),
                        Bytes::new(),
                    );
                    true
                });
            raft_wb.truncate_raft_log(
                task.region_id,
                task.region_id,
                keyspace_id,
                TRUNCATE_ALL_INDEX,
            );
            self.remove_shard(task.region_id);
            if let Some(preprocessor) = self.preprocessors.get(&task.region_id) {
                Self::notify_region_change(
                    &self.cop_host,
                    preprocessor.region(),
                    RegionChangeEvent::Destroy,
                );
                self.preprocessors.remove(&task.region_id);
            }
            self.clear_cached_bootstrap(task.region_id);
            self.pending_merge_states.remove(&task.region_id);
            if self.region_progresses.remove(&task.region_id).is_none() {
                warn!(
                    "unsafe recover clear region {} has no region progress",
                    task.region_id
                )
            }

            self.updated_regions.remove(&task.region_id);
            destroyed_regions.push(task.region_id);
        }

        self.manifest
            .update_region_progresses(&self.region_progresses);
        if !raft_wb.is_empty() {
            self.raft.write(raft_wb).expect("raft write");
        }
        self.manifest.persist().expect("persist manifest");
        Ok(destroyed_regions)
    }

    fn unsafe_recover_parent_id(&mut self, region_id: u64) -> Option<u64> {
        self.preprocessors
            .get_mut(&region_id)
            .and_then(|preprocessor| preprocessor.mut_shard_meta())
            .and_then(|meta| meta.parent.as_deref().map(|parent| parent.id))
    }

    fn unsafe_recover_take_destroy_keyspace_id(
        &mut self,
        region_id: u64,
        keyspace_id_hint: Option<u32>,
    ) -> Result<u32> {
        if let Some(keyspace_id) = keyspace_id_hint {
            return Ok(keyspace_id);
        }
        if let Some(keyspace_id) = self.delay_destroy_regions.remove(&region_id) {
            return Ok(keyspace_id);
        }
        self.region_progresses
            .get(&region_id)
            .map(|progress| progress.keyspace_id)
            .ok_or_else(|| {
                Error::Other(box_err!(
                    "unsafe recover clear region {} missing keyspace id",
                    region_id
                ))
            })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_util::{RegionSpec, TestEngineEnv, open_test_engine_with_seed};

    const MERGED_STORE_ID: u64 = 1024;

    fn open_test_engine(
        specs: &[RegionSpec],
        dependents: &[(u64, u64)],
        keyspaces: &[u32],
    ) -> TestEngineEnv {
        open_test_engine_with_seed(
            MERGED_STORE_ID,
            "merged_engine_unsafe_recover_",
            specs,
            dependents,
            keyspaces,
            &[],
        )
    }

    fn insert_uncommitted_entry(engine: &mut MergedEngine, region_id: u64, log_index: u64) {
        let mut entry = eraftpb::Entry::default();
        entry.set_index(log_index);
        entry.set_term(1);
        engine
            .region_progresses
            .get_mut(&region_id)
            .unwrap()
            .entries
            .insert(
                log_index,
                RaftLogOpWithCounter::from(RaftLogOp::new(&entry)),
            );
        engine
            .manifest
            .update_region_progresses(&engine.region_progresses);
    }

    #[test]
    fn test_unsafe_recover_check_clear_regions_rejects_external_dependent() {
        let mut env = open_test_engine(
            &[
                RegionSpec {
                    region_id: 10,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: None,
                },
                RegionSpec {
                    region_id: 11,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: Some(10),
                },
            ],
            &[(10, 11)],
            &[],
        );
        insert_uncommitted_entry(&mut env.engine, 10, 20);
        env.engine.updated_regions.insert(10);

        let err = env
            .engine
            .unsafe_recover_check_clear_regions(&[10])
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("dependent regions outside request")
        );
        assert!(env.engine.raft.has_dependents(10));
        assert!(env.engine.updated_regions.contains(&10));
        assert_eq!(
            env.engine.region_progresses.get(&10).unwrap().entries.len(),
            1
        );
        assert_eq!(
            env.engine
                .manifest
                .uncommitted_entries
                .get_region_entries(10)
                .unwrap()
                .len(),
            1,
        );
    }

    #[test]
    fn test_unsafe_recover_check_clear_regions_accepts_dependency_closed_request() {
        let env = open_test_engine(
            &[
                RegionSpec {
                    region_id: 20,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: None,
                },
                RegionSpec {
                    region_id: 21,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: Some(20),
                },
            ],
            &[(20, 21)],
            &[],
        );
        env.engine
            .unsafe_recover_check_clear_regions(&[20, 21])
            .unwrap();
    }

    #[test]
    fn test_unsafe_recover_execute_clear_regions_removes_parent_dependency_edge() {
        let mut env = open_test_engine(
            &[
                RegionSpec {
                    region_id: 30,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: None,
                },
                RegionSpec {
                    region_id: 31,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: Some(30),
                },
            ],
            &[(30, 31)],
            &[],
        );

        let destroyed = env
            .engine
            .unsafe_recover_execute_clear_regions(&[31])
            .unwrap();
        assert_eq!(destroyed, vec![31]);
        assert!(!env.engine.raft.has_dependents(30));
        assert_eq!(
            env.engine.raft.get_truncated_index(31),
            Some(TRUNCATE_ALL_INDEX),
        );
    }

    #[test]
    fn test_unsafe_recover_execute_clear_regions_destroys_delayed_parent_only() {
        let mut env = open_test_engine(
            &[
                RegionSpec {
                    region_id: 40,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Tombstone,
                    parent_id: None,
                },
                RegionSpec {
                    region_id: 41,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: Some(40),
                },
                RegionSpec {
                    region_id: 42,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: Some(43),
                },
                RegionSpec {
                    region_id: 43,
                    region_ver: 1,
                    keyspace_id: 1,
                    state: PeerState::Normal,
                    parent_id: None,
                },
            ],
            &[(40, 41), (43, 42)],
            &[],
        );

        assert!(env.engine.delay_destroy_regions.contains_key(&40));
        assert!(!env.engine.delay_destroy_regions.contains_key(&43));

        let destroyed = env
            .engine
            .unsafe_recover_execute_clear_regions(&[41, 42])
            .unwrap();
        assert_eq!(destroyed, vec![41, 42, 40]);
        assert_eq!(
            env.engine.raft.get_truncated_index(40),
            Some(TRUNCATE_ALL_INDEX),
        );
        assert_eq!(
            env.engine.raft.get_truncated_index(42),
            Some(TRUNCATE_ALL_INDEX),
        );
        assert_ne!(
            env.engine.raft.get_truncated_index(43),
            Some(TRUNCATE_ALL_INDEX),
        );
    }

    #[test]
    fn test_unsafe_recover_execute_clear_regions_removes_replay_state_and_manifest() {
        let mut env = open_test_engine(
            &[RegionSpec {
                region_id: 50,
                region_ver: 1,
                keyspace_id: 1,
                state: PeerState::Normal,
                parent_id: None,
            }],
            &[],
            &[],
        );
        insert_uncommitted_entry(&mut env.engine, 50, 60);
        env.engine.updated_regions.insert(50);
        {
            let progress = env.engine.region_progresses.get_mut(&50).unwrap();
            progress.synced_index = 55;
            progress.commit_index = 57;
            progress.last_index = 60;
            progress.safe_target_ts = TimeStamp::new(77);
            progress.pending_targets.push_back(RegionProgressTarget {
                ts: TimeStamp::new(77),
                log_idx: 60,
            });
        }

        let destroyed = env
            .engine
            .unsafe_recover_execute_clear_regions(&[50])
            .unwrap();
        assert_eq!(destroyed, vec![50]);
        assert!(!env.engine.updated_regions.contains(&50));
        assert!(!env.engine.merged_bootstrap_exists(50));

        assert!(!env.engine.region_progresses.contains_key(&50));
        assert!(
            env.engine
                .manifest
                .uncommitted_entries
                .get_region_entries(50)
                .is_none()
        );
    }
}