// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Handling open request.

use std::sync::Arc;
use std::time::Instant;

use common_telemetry::{info, warn};
use object_store::util::{join_path, normalize_dir};
use snafu::{OptionExt, ResultExt};
use store_api::logstore::LogStore;
use store_api::region_request::{AffectedRows, RegionCleanUpRequest, RegionOpenRequest};
use store_api::storage::RegionId;
use table::requests::STORAGE_KEY;

use crate::access_layer::AccessLayer;
use crate::engine::region_hook::{RegionGcInfo, RegionHookRef};
use crate::error::{
    ObjectStoreNotFoundSnafu, OpenDalSnafu, OpenRegionSnafu, RegionBusySnafu, RegionNotFoundSnafu,
    Result,
};
use crate::region::opener::{
    RegionOpener, get_object_store, provider_from_wal_options, sanitize_open_request_options,
};
use crate::region::options::RegionOptions;
use crate::request::OptionOutputTx;
use crate::sst::location::region_dir_from_table_dir;
use crate::wal::entry_distributor::WalEntryReceiver;
use crate::worker::handle_drop::{
    cleanup_region_file_artifacts, remove_region_dir_for_full_drop, remove_region_dir_once,
};
use crate::worker::{DROPPING_MARKER_FILE, RegionWorkerLoop};

impl<S: LogStore> RegionWorkerLoop<S> {
    pub(crate) async fn handle_offline_cleanup_request(
        &mut self,
        region_id: RegionId,
        mut request: RegionCleanUpRequest,
    ) -> Result<AffectedRows> {
        info!(
            "Try to clean region {} offline, worker: {}",
            region_id, self.id
        );

        if self.regions.is_region_exists(region_id) {
            return RegionBusySnafu { region_id }.fail();
        }

        sanitize_open_request_options(&mut request.options);

        let options = RegionOptions::try_from_options(region_id, &request.options)?;
        let object_store = get_object_store(&options.storage, &self.object_store_manager)?;
        let provider = provider_from_wal_options::<S>(region_id, &options.wal_options)?;
        self.wal.obsolete_all(region_id, &provider).await?;

        let table_dir = normalize_dir(&request.table_dir);
        let region_dir = region_dir_from_table_dir(&table_dir, region_id, request.path_type);
        remove_region_dir_for_full_drop(&region_dir, &object_store).await?;

        // The region directory is gone. Offline cleanup (soft-drop PURGE) is a
        // terminal physical removal, identical in effect to the drop GC
        // worker's directory deletion — but unlike the drop worker it never
        // fires `on_region_files_removed`, because the region is offline and
        // carries no `RegionMetadataRef`. Fire `on_region_gc` instead: it
        // accepts an absent region and lets extensions with sidecar files
        // outside the region dir (e.g. an Iceberg export) reclaim their
        // per-region state.
        //
        // Propagate the hook's error so the caller retries the cleanup, mirroring
        // how the global GC worker keeps a region for the next pass when
        // `on_region_gc` fails. The whole handler is idempotent (the existing
        // `test_engine_offline_cleanup_closed_region` already drives it twice in
        // a row), so a retry re-runs the directory removal, WAL obsolete and the
        // hook safely. Runs inline, consistent with the blocking directory
        // removal above; offline cleanup is already a slow path.
        if let Some(hook) = self.plugins.get::<RegionHookRef>() {
            let access_layer = Arc::new(AccessLayer::new(
                table_dir.clone(),
                request.path_type,
                object_store.clone(),
                self.puffin_manager_factory.clone(),
                self.intermediate_manager.clone(),
            ));
            let gc_info = RegionGcInfo {
                removed_files: &[],
                is_region_dropped: true,
                full_file_listing: true,
            };
            if let Err(err) = hook
                .on_region_gc(region_id, None, &access_layer, &gc_info)
                .await
            {
                warn!(
                    err;
                    "Region hook on_region_gc failed during offline cleanup for region {}; \
                     returning the error so the caller retries the cleanup",
                    region_id,
                );
                return Err(err);
            }
        }

        self.cleanup_dropped_region_runtime_state(region_id).await;
        self.dropping_regions.remove_region(region_id);
        cleanup_region_file_artifacts(
            region_id,
            &table_dir,
            &self.intermediate_manager,
            &self.cache_manager,
        )
        .await;

        Ok(0)
    }

    async fn check_and_cleanup_region(
        &self,
        region_id: RegionId,
        request: &RegionOpenRequest,
    ) -> Result<()> {
        let object_store = if let Some(storage_name) = request.options.get(STORAGE_KEY) {
            self.object_store_manager
                .find(storage_name)
                .with_context(|| ObjectStoreNotFoundSnafu {
                    object_store: storage_name.clone(),
                })?
        } else {
            self.object_store_manager.default_object_store()
        };
        // Check if this region is pending drop. And clean the entire dir if so.
        let region_dir =
            region_dir_from_table_dir(&request.table_dir, region_id, request.path_type);
        if !self.dropping_regions.is_region_exists(region_id)
            && object_store
                .exists(&join_path(&region_dir, DROPPING_MARKER_FILE))
                .await
                .context(OpenDalSnafu)?
        {
            let result = remove_region_dir_once(&region_dir, object_store, true).await;
            info!(
                "Region {} is dropped, worker: {}, result: {:?}",
                region_id, self.id, result
            );
            return RegionNotFoundSnafu { region_id }.fail();
        }

        Ok(())
    }

    pub(crate) async fn handle_open_request(
        &mut self,
        region_id: RegionId,
        mut request: RegionOpenRequest,
        wal_entry_receiver: Option<WalEntryReceiver>,
        sender: OptionOutputTx,
    ) {
        if self.regions.is_region_exists(region_id) {
            sender.send(Ok(0));
            return;
        }
        let Some(sender) = self
            .opening_regions
            .wait_for_opening_region(region_id, sender)
        else {
            return;
        };
        info!("Try to open region {}, worker: {}", region_id, self.id);
        sanitize_open_request_options(&mut request.options);

        // Open region from specific region dir.
        let requirements = request.requirements;
        let opener = match RegionOpener::new(
            region_id,
            &request.table_dir,
            request.path_type,
            self.memtable_builder_provider.clone(),
            self.object_store_manager.clone(),
            self.purge_scheduler.clone(),
            self.puffin_manager_factory.clone(),
            self.intermediate_manager.clone(),
            self.time_provider.clone(),
            self.file_ref_manager.clone(),
            self.partition_expr_fetcher.clone(),
        )
        .skip_wal_replay(request.skip_wal_replay)
        .cache(Some(self.cache_manager.clone()))
        .hook(self.plugins.get())
        .wal_entry_reader(wal_entry_receiver.map(|receiver| Box::new(receiver) as _))
        .replay_checkpoint(request.checkpoint.map(|checkpoint| checkpoint.entry_id))
        .parse_options(request.options.clone())
        {
            Ok(opener) => opener,
            Err(err) => {
                sender.send(Err(err));
                return;
            }
        };

        if let Err(err) = opener.ensure_region_requirements(requirements) {
            sender.send(Err(err));
            return;
        }

        if let Err(err) = self.check_and_cleanup_region(region_id, &request).await {
            sender.send(Err(err));
            return;
        }

        let now = Instant::now();
        let regions = self.regions.clone();
        let wal = self.wal.clone();
        let config = self.config.clone();
        let opening_regions = self.opening_regions.clone();
        let region_count = self.region_count.clone();
        let worker_id = self.id;
        opening_regions.insert_sender(region_id, sender);
        common_runtime::spawn_global(async move {
            match opener.open(&config, &wal).await {
                Ok(region) => {
                    info!(
                        "Region {} is opened with requirements {:?}, worker: {}, elapsed: {:?}",
                        region_id,
                        requirements,
                        worker_id,
                        now.elapsed()
                    );
                    region_count.inc();

                    // Notify the region hook that the region has been opened.
                    // Fires before registration; allocates nothing when no hook
                    // is registered.
                    if let Some(hook) = region.manifest_ctx.hook() {
                        hook.on_region_opened(region_id, &region.metadata()).await;
                    }

                    // Insert the Region into the RegionMap.
                    regions.insert_region(region);

                    let senders = opening_regions.remove_sender(region_id);
                    for sender in senders {
                        sender.send(Ok(0));
                    }
                }
                Err(err) => {
                    let senders = opening_regions.remove_sender(region_id);
                    let err = Arc::new(err);
                    for sender in senders {
                        sender.send(Err(err.clone()).context(OpenRegionSnafu));
                    }
                }
            }
        });
    }
}
