// Copyright 2024 The Drasi Authors.
//
// 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.

#![allow(unexpected_cfgs)]

//! Garnet/Redis Index Backend for Drasi
//!
//! This crate provides a distributed storage backend for Drasi queries using Redis/Garnet.
//!
//! # Usage
//!
//! ```ignore
//! use drasi_index_garnet::GarnetIndexProvider;
//! use drasi_lib::DrasiLib;
//! use std::sync::Arc;
//!
//! let provider = GarnetIndexProvider::new("redis://localhost:6379", None, true);
//! let drasi = DrasiLib::builder()
//!     .with_index_provider("redis", Arc::new(provider))
//!     .build()?;
//! ```

use drasi_core::interface::IndexError;
use redis::{aio::MultiplexedConnection, cmd, AsyncCommands};

pub mod checkpoint;
#[cfg(feature = "plugin-descriptor")]
mod descriptor;
pub mod element_index;
pub mod future_queue;
pub mod live_results;
pub mod outbox;
mod plugin;
pub mod result_index;
pub(crate) mod session_state;
mod storage_models;

// Re-export the plugin provider for easy access
pub use checkpoint::GarnetCheckpointStore;
pub use live_results::GarnetLiveResultsWriter;
pub use outbox::GarnetOutboxWriter;
pub use plugin::GarnetIndexProvider;
pub use session_state::{GarnetSessionControl, GarnetSessionState};

#[cfg(feature = "plugin-descriptor")]
pub use descriptor::{GarnetIndexConfigDto, GarnetIndexDescriptor};

trait ClearByPattern {
    async fn clear(&self, pattern: String) -> Result<(), IndexError>;
}

impl ClearByPattern for MultiplexedConnection {
    async fn clear(&self, pattern: String) -> Result<(), IndexError> {
        let mut con = self.clone();
        let mut con2 = self.clone();

        let mut cursor = "0".to_string();
        loop {
            let mut cmd = cmd("SCAN");
            let cmd = cmd.arg(remove_surrounding_quotes(&cursor));
            let cmd = cmd.arg("MATCH");
            let cmd = cmd.arg(&pattern);
            let cmd = cmd.arg("COUNT");
            let cmd = cmd.arg(100);

            let result = match cmd
                .query_async::<MultiplexedConnection, Vec<redis::Value>>(&mut con)
                .await
            {
                Ok(v) => v,
                Err(e) => return Err(IndexError::other(e)),
            };

            if result.len() < 2 {
                break;
            }

            match &result[0] {
                redis::Value::Status(s) => {
                    cursor = s.clone();
                }
                redis::Value::Data(d) => {
                    if let Ok(s) = String::from_utf8(d.to_vec()) {
                        cursor = s;
                    }
                }
                _ => (),
            }

            if let redis::Value::Bulk(b) = &result[1] {
                for k in b {
                    if let redis::Value::Data(d) = k {
                        if let Ok(k) = String::from_utf8(d.to_vec()) {
                            match con2.del::<&str, ()>(remove_surrounding_quotes(&k)).await {
                                Ok(_) => (),
                                Err(e) => return Err(IndexError::other(e)),
                            }
                        }
                    }
                }
            }

            if cursor == "0" {
                break;
            }
        }
        Ok(())
    }
}

fn remove_surrounding_quotes(s: &str) -> &str {
    if s.len() >= 2
        && (s.starts_with('"') && s.ends_with('"') || s.starts_with('\'') && s.ends_with('\''))
    {
        &s[1..s.len() - 1]
    } else {
        s
    }
}
