mod.rs

/*
 * Parseable Server (C) 2022 - 2025 Parseable, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *
 */

use std::{
    collections::{HashMap, HashSet},
    num::NonZeroU32,
    path::PathBuf,
    str::FromStr,
    sync::{Arc, RwLock},
};

use actix_web::http::{
    StatusCode,
    header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue},
};
use arrow_schema::{Field, Schema};
use bytes::Bytes;
use chrono::Utc;
use clap::{Parser, error::ErrorKind};
use once_cell::sync::Lazy;
use relative_path::RelativePathBuf;
pub use staging::StagingError;
use streams::StreamRef;
pub use streams::{Stream, StreamNotFound, Streams};
use tokio::try_join;
use tracing::error;

pub const DEFAULT_TENANT: &str = "DEFAULT_TENANT";

#[cfg(feature = "kafka")]
use crate::connectors::kafka::config::KafkaConfig;
use crate::{
    cli::{Cli, Options, StorageOptions},
    event::{
        commit_schema,
        format::{LogSource, LogSourceEntry},
    },
    handlers::{
        DatasetTag, STREAM_TYPE_KEY, TelemetryType,
        http::{
            cluster::{PMETA_STREAM_NAME, sync_streams_with_ingestors},
            ingest::PostError,
            logstream::error::{CreateStreamError, StreamError},
            middleware::{CLUSTER_SECRET, CLUSTER_SECRET_HEADER},
            modal::{
                ingest_server::INGESTOR_META,
                utils::{logstream_utils::PutStreamHeaders, rbac_utils::get_metadata},
            },
        },
    },
    metadata::{LogStreamMetadata, SchemaVersion},
    metastore::{
        metastore_traits::Metastore, metastores::object_store_metastore::ObjectStoreMetastore,
    },
    option::Mode,
    rbac::{
        Users,
        map::{mut_roles, mut_users, write_user_groups},
    },
    static_schema::{StaticSchema, convert_static_schema_to_arrow_schema},
    storage::{
        ObjectStorageError, ObjectStorageProvider, ObjectStoreFormat, Owner, Permisssion,
        StorageMetadata, StreamType, put_remote_metadata,
    },
    tenants::{Service, TENANT_METADATA},
    validator,
};

mod staging;
mod streams;

/// File extension for arrow files in staging
const ARROW_FILE_EXTENSION: &str = "arrows";

/// File extension for incomplete arrow files
const PART_FILE_EXTENSION: &str = "part";

/// Name of a Stream
/// NOTE: this used to be a struct, flattened out for simplicity
pub type LogStream = String;

pub const JOIN_COMMUNITY: &str =
    "Join us on Parseable Slack community for questions : https://logg.ing/community";
pub const STREAM_EXISTS: &str = "Stream exists";

/// Shared state of the Parseable server.
pub static PARSEABLE: Lazy<Parseable> = Lazy::new(|| match Cli::parse().storage {
    StorageOptions::Local(args) => {
        if args.options.staging_dir() == &args.storage.root {
            clap::Error::raw(
                ErrorKind::ValueValidation,
                "Cannot use same path for storage and staging",
            )
            .exit();
        }

        if args.options.hot_tier_storage_path.is_some() {
            clap::Error::raw(
                ErrorKind::ValueValidation,
                "Cannot use hot tier with local-store subcommand.",
            )
            .exit();
        }

        // for now create a metastore without using a CLI arg
        let metastore = ObjectStoreMetastore {
            storage: args.storage.construct_client(),
        };

        Parseable::new(
            args.options,
            #[cfg(feature = "kafka")]
            args.kafka,
            Arc::new(args.storage),
            Arc::new(metastore),
        )
    }
    StorageOptions::S3(args) => {
        // for now create a metastore without using a CLI arg
        let metastore = ObjectStoreMetastore {
            storage: args.storage.construct_client(),
        };
        Parseable::new(
            args.options,
            #[cfg(feature = "kafka")]
            args.kafka,
            Arc::new(args.storage),
            Arc::new(metastore),
        )
    }
    StorageOptions::Blob(args) => {
        // for now create a metastore without using a CLI arg
        let metastore = ObjectStoreMetastore {
            storage: args.storage.construct_client(),
        };
        Parseable::new(
            args.options,
            #[cfg(feature = "kafka")]
            args.kafka,
            Arc::new(args.storage),
            Arc::new(metastore),
        )
    }
    StorageOptions::Gcs(args) => {
        // for now create a metastore without using a CLI arg
        let metastore = ObjectStoreMetastore {
            storage: args.storage.construct_client(),
        };
        Parseable::new(
            args.options,
            #[cfg(feature = "kafka")]
            args.kafka,
            Arc::new(args.storage),
            Arc::new(metastore),
        )
    }
});

/// All state related to parseable, in one place.
pub struct Parseable {
    /// Configuration variables for parseable
    pub options: Arc<Options>,
    /// Storage engine backing parseable
    pub storage: Arc<dyn ObjectStorageProvider>,
    // /// ObjectStorageProvider for each tenant
    // pub tenant_storage: Arc<DashMap<String, Arc<dyn ObjectStorageProvider>>>,
    /// Metadata and staging relating to each logstreams
    /// A globally shared mapping of `Streams` that parseable is aware of.
    pub streams: Streams,
    pub tenants: Arc<RwLock<Vec<String>>>,
    /// metastore
    pub metastore: Arc<dyn Metastore>,
    /// Used to configure the kafka connector
    #[cfg(feature = "kafka")]
    pub kafka_config: KafkaConfig,
}

impl Parseable {
    pub fn new(
        options: Options,
        #[cfg(feature = "kafka")] kafka_config: KafkaConfig,
        storage: Arc<dyn ObjectStorageProvider>,
        metastore: Arc<dyn Metastore>,
    ) -> Self {
        Parseable {
            options: Arc::new(options),
            storage,
            metastore,
            streams: Streams::default(),
            tenants: Arc::new(RwLock::new(vec![])),
            #[cfg(feature = "kafka")]
            kafka_config,
        }
    }
    /// Try to get the handle of a stream in staging, if it doesn't exist return `None`.
    pub fn get_stream(
        &self,
        stream_name: &str,
        tenant_id: &Option<String>,
    ) -> Result<StreamRef, StreamNotFound> {
        let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
        self.streams
            .read()
            .unwrap()
            .get(tenant_id)
            .ok_or_else(|| StreamNotFound(format!("{stream_name} with tenant {tenant_id}")))
            .map(|v| v.get(stream_name))?
            .ok_or_else(|| StreamNotFound(stream_name.to_owned()))
            .cloned()
    }

    /// Get the handle to a stream in staging, create one if it doesn't exist
    pub fn get_or_create_stream(&self, stream_name: &str, tenant_id: &Option<String>) -> StreamRef {
        if let Ok(staging) = self.get_stream(stream_name, tenant_id) {
            return staging;
        }

        let ingestor_id = INGESTOR_META
            .get()
            .map(|ingestor_metadata| ingestor_metadata.get_node_id());

        // Gets write privileges only for creating the stream when it doesn't already exist.
        self.streams.get_or_create(
            self.options.clone(),
            stream_name.to_owned(),
            LogStreamMetadata::default(),
            ingestor_id,
            tenant_id,
        )
    }

    /// Checks for the stream in memory, or loads it from storage when in distributed mode
    /// return true if stream exists in memory or loaded from storage
    /// return false if stream doesn't exist in memory and not loaded from storage
    pub async fn check_or_load_stream(
        &self,
        stream_name: &str,
        tenant_id: &Option<String>,
    ) -> bool {
        if self.streams.contains(stream_name, tenant_id) {
            return true;
        }
        (self.options.mode == Mode::Query || self.options.mode == Mode::Prism)
            && self
                .create_stream_and_schema_from_storage(stream_name, tenant_id)
                .await
                .unwrap_or_default()
    }

    // validate the storage, if the proper path for staging directory is provided
    // if the proper data directory is provided, or s3 bucket is provided etc
    pub async fn validate_storage(&self) -> Result<Option<Bytes>, ObjectStorageError> {
        let obj_store = self.storage.get_object_store();
        let mut has_parseable_json = false;
        let parseable_json_result = self
            .metastore
            .get_parseable_metadata(&None) // load the server meta
            .await
            .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;

        if parseable_json_result.is_some() {
            has_parseable_json = true;
        }

        // Lists all the directories in the root of the bucket/directory
        // can be a stream (if it contains .stream.json file) or not
        let has_dirs = match obj_store.list_dirs(&None).await {
            Ok(dirs) => !dirs.is_empty(),
            Err(_) => false,
        };

        if !has_dirs && !has_parseable_json {
            return Ok(None);
        }
        let has_stream = if let Some(tenants) = PARSEABLE.list_tenants() {
            let mut has_stream = true;
            for tenant in tenants {
                if let Err(e) = PARSEABLE.metastore.list_streams(&Some(tenant)).await {
                    tracing::error!("{e}");
                    has_stream = false;
                    break;
                };
            }
            has_stream
        } else {
            PARSEABLE.metastore.list_streams(&None).await.is_ok()
        };

        if has_stream {
            return Ok(parseable_json_result);
        }

        if self.storage.name() == "drive" {
            return Err(ObjectStorageError::Custom(format!(
                "Could not start the server because directory '{}' contains stale data, please use an empty directory, and restart the server.\n{}",
                self.storage.get_endpoint(),
                JOIN_COMMUNITY
            )));
        }

        // S3 bucket mode
        Err(ObjectStorageError::Custom(format!(
            "Could not start the server because bucket '{}' contains stale data, please use an empty bucket and restart the server.\n{}",
            self.storage.get_endpoint(),
            JOIN_COMMUNITY
        )))
    }

    /// this function only gets called from enterprise main
    /// If the server has traces of multi-tenancy AND is started with multi-tenant flag, then proceed
    /// otherwise fail with error
    ///
    /// if the server doesn't have traces of multi-tenancy AND is started without the flag, then proceed
    /// otherwise fail with error
    pub async fn validate_multi_tenancy(&self) -> Result<(), anyhow::Error> {
        self.load_tenants().await
    }

    pub fn storage(&self) -> Arc<dyn ObjectStorageProvider> {
        self.storage.clone()
    }

    pub fn hot_tier_dir(&self) -> &Option<PathBuf> {
        &self.options.hot_tier_storage_path
    }

    // returns the string representation of the storage mode
    // drive --> Local drive
    // s3 --> S3 bucket
    // azure_blob --> Azure Blob Storage
    pub fn get_storage_mode_string(&self) -> &str {
        if self.storage.name() == "drive" {
            return "Local drive";
        } else if self.storage.name() == "s3" {
            return "S3 bucket";
        } else if self.storage.name() == "blob_store" {
            return "Azure Blob Storage";
        } else if self.storage.name() == "gcs" {
            return "Google Object Store";
        }
        "Unknown"
    }

    pub fn get_server_mode_string(&self) -> &str {
        match self.options.mode {
            Mode::Query => "Distributed (Query)",
            Mode::Ingest => "Distributed (Ingest)",
            Mode::Index => "Distributed (Index)",
            Mode::Prism => "Distributed (Prism)",
            Mode::All => "Standalone",
        }
    }

    /// list all streams from storage
    /// if stream exists in storage, create stream and schema from storage
    /// and add it to the memory map
    pub async fn create_stream_and_schema_from_storage(
        &self,
        stream_name: &str,
        tenant_id: &Option<String>,
    ) -> Result<bool, StreamError> {
        // Proceed to create log stream if it doesn't exist
        let storage = self.storage.get_object_store();
        let streams = PARSEABLE.metastore.list_streams(tenant_id).await?;
        if !streams.contains(stream_name) {
            return Ok(false);
        }
        let (stream_metadata_bytes, schema_bytes) = try_join!(
            storage.create_stream_from_ingestor(stream_name, tenant_id),
            storage.create_schema_from_metastore(stream_name, tenant_id)
        )?;
        let stream_metadata = if stream_metadata_bytes.is_empty() {
            ObjectStoreFormat::default()
        } else {
            serde_json::from_slice::<ObjectStoreFormat>(&stream_metadata_bytes)?
        };

        let schema = if schema_bytes.is_empty() {
            Arc::new(Schema::empty())
        } else {
            serde_json::from_slice::<Arc<Schema>>(&schema_bytes)?
        };

        let static_schema: HashMap<String, Arc<Field>> = schema
            .fields
            .into_iter()
            .map(|field| (field.name().to_string(), field.clone()))
            .collect();

        let created_at = stream_metadata.created_at;
        let time_partition = stream_metadata.time_partition.unwrap_or_default();
        let time_partition_limit = stream_metadata
            .time_partition_limit
            .and_then(|limit| limit.parse().ok());
        let custom_partition = stream_metadata.custom_partition;
        let static_schema_flag = stream_metadata.static_schema_flag;
        let hot_tier_enabled = stream_metadata.hot_tier_enabled;
        let hot_tier = stream_metadata.hot_tier.clone();
        let stream_type = stream_metadata.stream_type;
        let schema_version = stream_metadata.schema_version;
        let log_source = stream_metadata.log_source;
        let telemetry_type = stream_metadata.telemetry_type;
        let dataset_tags = stream_metadata.dataset_tags;
        let dataset_labels = stream_metadata.dataset_labels;
        let mut metadata = LogStreamMetadata::new(
            created_at,
            time_partition,
            time_partition_limit,
            custom_partition,
            static_schema_flag,
            static_schema,
            stream_type,
            schema_version,
            log_source,
            telemetry_type,
            dataset_tags,
            dataset_labels,
        );

        // Set hot tier fields from the stored metadata
        metadata.hot_tier_enabled = hot_tier_enabled;
        metadata.hot_tier.clone_from(&hot_tier);

        let ingestor_id = INGESTOR_META
            .get()
            .map(|ingestor_metadata| ingestor_metadata.get_node_id());

        // Gets write privileges only for creating the stream when it doesn't already exist.
        let stream = self.streams.get_or_create(
            self.options.clone(),
            stream_name.to_owned(),
            metadata,
            ingestor_id,
            tenant_id,
        );

        // Set hot tier configuration in memory based on stored metadata
        if let Some(hot_tier_config) = hot_tier {
            stream.set_hot_tier(Some(hot_tier_config));
        }
        // commit schema in memory
        commit_schema(stream_name, schema, tenant_id).map_err(|e| StreamError::Anyhow(e.into()))?;

        Ok(true)
    }

    pub async fn create_internal_stream_if_not_exists(&self) -> Result<(), StreamError> {
        let log_source_entry = LogSourceEntry::new(LogSource::Pmeta, HashSet::new());
        let tenants = if let Some(tenants) = PARSEABLE.list_tenants() {
            tenants.into_iter().map(Some).collect()
        } else {
            vec![None]
        };
        for tenant_id in tenants {
            let internal_stream_result = self
                .create_stream_if_not_exists(
                    PMETA_STREAM_NAME,
                    StreamType::Internal,
                    None,
                    vec![log_source_entry.clone()],
                    TelemetryType::Logs,
                    &tenant_id,
                    vec![],
                    vec![],
                )
                .await;

            // Check if either stream creation failed
            if let Err(e) = &internal_stream_result {
                tracing::error!("Failed to create pmeta stream: {:?}", e);
            }

            // Check if both streams already existed
            if matches!(internal_stream_result, Ok(true)) {
                continue;
            }

            let mut header_map = HeaderMap::new();
            header_map.insert(
                HeaderName::from_str(STREAM_TYPE_KEY).unwrap(),
                HeaderValue::from_str(&StreamType::Internal.to_string()).unwrap(),
            );
            header_map.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));

            if let Some((_, hash)) = CLUSTER_SECRET.get() {
                header_map.insert(
                    HeaderName::from_static(CLUSTER_SECRET_HEADER),
                    HeaderValue::from_str(hash).unwrap(),
                );
                header_map.insert(
                    HeaderName::from_static("intra-cluster-tenant"),
                    HeaderValue::from_str(tenant_id.as_deref().unwrap_or(DEFAULT_TENANT)).unwrap(),
                );
            }

            // Sync only the streams that were created successfully
            if matches!(internal_stream_result, Ok(false))
                && let Err(e) = sync_streams_with_ingestors(
                    header_map.clone(),
                    Bytes::new(),
                    PMETA_STREAM_NAME,
                    &tenant_id,
                )
                .await
            {
                tracing::error!("Failed to sync pmeta stream with ingestors: {:?}", e);
            }
        }

        Ok(())
    }

    // Check if the stream exists and create a new stream if doesn't exist
    #[allow(clippy::too_many_arguments)]
    pub async fn create_stream_if_not_exists(
        &self,
        stream_name: &str,
        stream_type: StreamType,
        custom_partition: Option<&String>,
        log_source: Vec<LogSourceEntry>,
        telemetry_type: TelemetryType,
        tenant_id: &Option<String>,
        dataset_tags: Vec<DatasetTag>,
        dataset_labels: Vec<String>,
    ) -> Result<bool, PostError> {
        if self.streams.contains(stream_name, tenant_id) {
            return Ok(true);
        }

        // validate custom partition if provided
        if let Some(partition) = custom_partition {
            validate_custom_partition(partition)?;
        }

        // For distributed deployments, if the stream not found in memory map,
        // check if it exists in the storage
        // create stream and schema from storage
        if self.options.mode != Mode::All
            && self
                .create_stream_and_schema_from_storage(stream_name, tenant_id)
                .await?
        {
            return Ok(true);
        }

        self.create_stream(
            stream_name.to_string(),
            "",
            None,
            custom_partition,
            false,
            Arc::new(Schema::empty()),
            stream_type,
            log_source,
            telemetry_type,
            tenant_id,
            dataset_tags,
            dataset_labels,
        )
        .await?;

        Ok(false)
    }

    pub async fn add_update_log_source(
        &self,
        stream_name: &str,
        log_source: LogSourceEntry,
        tenant_id: &Option<String>,
    ) -> Result<(), StreamError> {
        let stream = self
            .get_stream(stream_name, tenant_id)
            .expect(STREAM_EXISTS);
        let mut log_sources = stream.get_log_source();
        let mut changed = false;

        // Try to find existing log source with the same format
        if let Some(stream_log_source) = log_sources
            .iter_mut()
            .find(|source| source.log_source_format == log_source.log_source_format)
        {
            // Use a HashSet to efficiently track only new fields
            let existing_fields: HashSet<String> =
                stream_log_source.fields.iter().cloned().collect();
            let new_fields: HashSet<String> = log_source
                .fields
                .iter()
                .filter(|field| !existing_fields.contains(*field))
                .cloned()
                .collect();

            // Only update if there are new fields to add
            if !new_fields.is_empty() {
                stream_log_source.fields.extend(new_fields);
                changed = true;
            }
        } else {
            // If no matching log source found, add the new one
            log_sources.push(log_source);
            changed = true;
        }

        // Only persist to storage if we made changes
        if changed {
            stream.set_log_source(log_sources.clone());

            let storage = self.storage.get_object_store();
            if let Err(err) = storage
                .update_log_source_in_stream(stream_name, &log_sources, tenant_id)
                .await
            {
                return Err(StreamError::Storage(err));
            }
        }

        Ok(())
    }

    pub async fn create_update_stream(
        &self,
        headers: &HeaderMap,
        body: &Bytes,
        stream_name: &str,
        tenant_id: &Option<String>,
    ) -> Result<HeaderMap, StreamError> {
        let PutStreamHeaders {
            time_partition,
            time_partition_limit,
            custom_partition,
            static_schema_flag,
            update_stream_flag,
            stream_type,
            log_source,
            telemetry_type,
            dataset_tags,
            dataset_labels,
        } = headers.into();

        let stream_in_memory_dont_update =
            self.streams.contains(stream_name, tenant_id) && !update_stream_flag;

        // check if stream in storage only if not in memory
        // for Parseable OSS, create_update_stream is called only from query node
        // for Parseable Enterprise, create_update_stream is called from prism node
        let stream_in_storage_only_for_query_node = !self.streams.contains(stream_name, tenant_id)
            && (self.options.mode == Mode::Query || self.options.mode == Mode::Prism)
            && self
                .create_stream_and_schema_from_storage(stream_name, tenant_id)
                .await?;
        if stream_in_memory_dont_update || stream_in_storage_only_for_query_node {
            return Err(StreamError::Custom {
                msg: format!(
                    "Logstream {stream_name} already exists, please create a new log stream with unique name"
                ),
                status: StatusCode::BAD_REQUEST,
            });
        }

        if update_stream_flag {
            return self
                .update_stream(
                    headers,
                    stream_name,
                    &time_partition,
                    static_schema_flag,
                    &time_partition_limit,
                    custom_partition.as_ref(),
                    tenant_id,
                )
                .await;
        }

        let time_partition_in_days = if !time_partition_limit.is_empty() {
            Some(validate_time_partition_limit(&time_partition_limit)?)
        } else {
            None
        };

        if let Some(custom_partition) = &custom_partition {
            validate_custom_partition(custom_partition)?;
        }

        if !time_partition.is_empty() && custom_partition.is_some() {
            return Err(StreamError::Custom {
                msg: "Cannot set both time partition and custom partition".to_string(),
                status: StatusCode::BAD_REQUEST,
            });
        }

        let schema = validate_static_schema(
            body,
            stream_name,
            &time_partition,
            custom_partition.as_ref(),
            static_schema_flag,
        )?;

        let log_source_entry = LogSourceEntry::new(log_source, HashSet::new());
        self.create_stream(
            stream_name.to_string(),
            &time_partition,
            time_partition_in_days,
            custom_partition.as_ref(),
            static_schema_flag,
            schema,
            stream_type,
            vec![log_source_entry],
            telemetry_type,
            tenant_id,
            dataset_tags,
            dataset_labels,
        )
        .await?;

        Ok(headers.clone())
    }

    #[allow(clippy::too_many_arguments)]
    async fn update_stream(
        &self,
        headers: &HeaderMap,
        stream_name: &str,
        time_partition: &str,
        static_schema_flag: bool,
        time_partition_limit: &str,
        custom_partition: Option<&String>,
        tenant_id: &Option<String>,
    ) -> Result<HeaderMap, StreamError> {
        if !self.streams.contains(stream_name, tenant_id) {
            return Err(StreamNotFound(stream_name.to_string()).into());
        }
        if !time_partition.is_empty() {
            return Err(StreamError::Custom {
                msg: "Altering the time partition of an existing stream is restricted.".to_string(),
                status: StatusCode::BAD_REQUEST,
            });
        }
        if static_schema_flag {
            return Err(StreamError::Custom {
                msg: "Altering the schema of an existing stream is restricted.".to_string(),
                status: StatusCode::BAD_REQUEST,
            });
        }
        if !time_partition_limit.is_empty() {
            let time_partition_days = validate_time_partition_limit(time_partition_limit)?;
            self.update_time_partition_limit_in_stream(
                stream_name.to_string(),
                time_partition_days,
                tenant_id,
            )
            .await?;
            return Ok(headers.clone());
        }
        self.validate_and_update_custom_partition(stream_name, custom_partition, tenant_id)
            .await?;

        Ok(headers.clone())
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn create_stream(
        &self,
        stream_name: String,
        time_partition: &str,
        time_partition_limit: Option<NonZeroU32>,
        custom_partition: Option<&String>,
        static_schema_flag: bool,
        schema: Arc<Schema>,
        stream_type: StreamType,
        log_source: Vec<LogSourceEntry>,
        telemetry_type: TelemetryType,
        tenant_id: &Option<String>,
        dataset_tags: Vec<DatasetTag>,
        dataset_labels: Vec<String>,
    ) -> Result<(), CreateStreamError> {
        // fail to proceed if invalid stream name
        if stream_type != StreamType::Internal {
            validator::stream_name(&stream_name, stream_type)?;
        }
        // Proceed to create log stream if it doesn't exist
        let storage = self.storage.get_object_store();

        // update owner and permissions
        let meta = ObjectStoreFormat {
            created_at: Utc::now().to_rfc3339(),
            permissions: vec![Permisssion::new(PARSEABLE.options.username.clone())],
            stream_type,
            time_partition: (!time_partition.is_empty()).then(|| time_partition.to_string()),
            time_partition_limit: time_partition_limit.map(|limit| limit.to_string()),
            custom_partition: custom_partition.cloned(),
            static_schema_flag,
            schema_version: SchemaVersion::V1, // NOTE: Newly created streams are all V1
            owner: Owner {
                id: PARSEABLE.options.username.clone(),
                group: PARSEABLE.options.username.clone(),
            },
            log_source: log_source.clone(),
            telemetry_type,
            dataset_tags: dataset_tags.clone(),
            dataset_labels: dataset_labels.clone(),
            ..Default::default()
        };

        match storage
            .create_stream(&stream_name, meta, schema.clone(), tenant_id)
            .await
        {
            Ok(created_at) => {
                let mut static_schema: HashMap<String, Arc<Field>> = HashMap::new();

                for (field_name, field) in schema
                    .fields()
                    .iter()
                    .map(|field| (field.name().to_string(), field.clone()))
                {
                    static_schema.insert(field_name, field);
                }

                let metadata = LogStreamMetadata::new(
                    created_at,
                    time_partition.to_owned(),
                    time_partition_limit,
                    custom_partition.cloned(),
                    static_schema_flag,
                    static_schema,
                    stream_type,
                    SchemaVersion::V1, // New stream
                    log_source,
                    telemetry_type,
                    dataset_tags,
                    dataset_labels,
                );
                let ingestor_id = INGESTOR_META
                    .get()
                    .map(|ingestor_metadata| ingestor_metadata.get_node_id());

                // Gets write privileges only for creating the stream when it doesn't already exist.
                self.streams.get_or_create(
                    self.options.clone(),
                    stream_name.to_owned(),
                    metadata,
                    ingestor_id,
                    tenant_id,
                );
            }
            Err(err) => {
                return Err(CreateStreamError::Storage { stream_name, err });
            }
        }
        Ok(())
    }

    async fn validate_and_update_custom_partition(
        &self,
        stream_name: &str,
        custom_partition: Option<&String>,
        tenant_id: &Option<String>,
    ) -> Result<(), StreamError> {
        let stream = self
            .get_stream(stream_name, tenant_id)
            .expect(STREAM_EXISTS);
        if stream.get_time_partition().is_some() {
            return Err(StreamError::Custom {
                msg: "Cannot set both time partition and custom partition".to_string(),
                status: StatusCode::BAD_REQUEST,
            });
        }
        if let Some(custom_partition) = custom_partition {
            validate_custom_partition(custom_partition)?;
        }

        self.update_custom_partition_in_stream(
            stream_name.to_string(),
            custom_partition,
            tenant_id,
        )
        .await?;

        Ok(())
    }

    pub async fn update_time_partition_limit_in_stream(
        &self,
        stream_name: String,
        time_partition_limit: NonZeroU32,
        tenant_id: &Option<String>,
    ) -> Result<(), CreateStreamError> {
        let storage = self.storage.get_object_store();
        if let Err(err) = storage
            .update_time_partition_limit_in_stream(&stream_name, time_partition_limit, tenant_id)
            .await
        {
            return Err(CreateStreamError::Storage { stream_name, err });
        }

        if let Ok(stream) = self.get_stream(&stream_name, tenant_id) {
            stream.set_time_partition_limit(time_partition_limit)
        } else {
            return Err(CreateStreamError::Custom {
                msg: "failed to update time partition limit in metadata".to_string(),
                status: StatusCode::EXPECTATION_FAILED,
            });
        }

        Ok(())
    }

    pub async fn update_custom_partition_in_stream(
        &self,
        stream_name: String,
        custom_partition: Option<&String>,
        tenant_id: &Option<String>,
    ) -> Result<(), CreateStreamError> {
        let stream = self
            .get_stream(&stream_name, tenant_id)
            .expect(STREAM_EXISTS);
        let static_schema_flag = stream.get_static_schema_flag();
        let time_partition = stream.get_time_partition();
        if static_schema_flag {
            let schema = stream.get_schema();

            if let Some(custom_partition) = custom_partition {
                let custom_partition_list = custom_partition.split(',').collect::<Vec<&str>>();
                for partition in custom_partition_list.iter() {
                    if !schema
                        .fields()
                        .iter()
                        .any(|field| field.name() == partition)
                    {
                        return Err(CreateStreamError::Custom {
                            msg: format!(
                                "custom partition field {partition} does not exist in the schema for the stream {stream_name}"
                            ),
                            status: StatusCode::BAD_REQUEST,
                        });
                    }
                }

                for partition in custom_partition_list {
                    if time_partition
                        .as_ref()
                        .is_some_and(|time| time == partition)
                    {
                        return Err(CreateStreamError::Custom {
                            msg: format!(
                                "time partition {partition} cannot be set as custom partition"
                            ),
                            status: StatusCode::BAD_REQUEST,
                        });
                    }
                }
            }
        }
        let storage = self.storage.get_object_store();
        if let Err(err) = storage
            .update_custom_partition_in_stream(&stream_name, custom_partition, tenant_id)
            .await
        {
            return Err(CreateStreamError::Storage { stream_name, err });
        }

        stream.set_custom_partition(custom_partition);

        Ok(())
    }

    /// Updates the first-event-at in storage and logstream metadata for the specified stream.
    ///
    /// This function updates the `first-event-at` in both the object store and the stream info metadata.
    /// If either update fails, an error is logged, but the function will still return the `first-event-at`.
    ///
    /// # Arguments
    ///
    /// * `stream_name` - The name of the stream to update.
    /// * `first_event_at` - The value of first-event-at.
    ///
    /// # Returns
    ///
    /// * `Option<String>` - Returns `Some(String)` with the provided timestamp if the update is successful,
    ///   or `None` if an error occurs.
    ///
    /// # Errors
    ///
    /// This function logs an error if:
    /// * The `first-event-at` cannot be updated in the object store.
    /// * The `first-event-at` cannot be updated in the stream info.
    ///
    /// # Examples
    ///```ignore
    /// ```rust
    /// use parseable::handlers::http::modal::utils::logstream_utils::update_first_event_at;
    /// let result = update_first_event_at("my_stream", "2023-01-01T00:00:00Z").await;
    /// match result {
    ///     Some(timestamp) => println!("first-event-at: {}", timestamp),
    ///     None => eprintln!("Failed to update first-event-at"),
    /// }
    /// ```
    pub async fn update_first_event_at(
        &self,
        stream_name: &str,
        first_event_at: &str,
        tenant_id: &Option<String>,
    ) -> Option<String> {
        let storage = self.storage.get_object_store();
        if let Err(err) = storage
            .update_first_event_in_stream(stream_name, first_event_at, tenant_id)
            .await
        {
            error!(
                "Failed to update first_event_at in storage for stream {stream_name:?}: {err:?}"
            );
        }

        match self.get_stream(stream_name, tenant_id) {
            Ok(stream) => stream.set_first_event_at(first_event_at),
            Err(err) => error!(
                "Failed to update first_event_at in stream info for stream {stream_name:?}: {err:?}"
            ),
        }

        Some(first_event_at.to_string())
    }

    pub async fn update_log_source(
        &self,
        stream_name: &str,
        log_source: Vec<LogSourceEntry>,
        tenant_id: &Option<String>,
    ) -> Result<(), StreamError> {
        let storage = self.storage.get_object_store();
        if let Err(err) = storage
            .update_log_source_in_stream(stream_name, &log_source, tenant_id)
            .await
        {
            return Err(StreamError::Storage(err));
        }

        let stream = self
            .get_stream(stream_name, tenant_id)
            .expect(STREAM_EXISTS);
        stream.set_log_source(log_source);

        Ok(())
    }

    pub fn add_tenant(
        &self,
        tenant_id: String,
        tenant_meta: StorageMetadata,
    ) -> Result<(), anyhow::Error> {
        if !self.options.is_multi_tenant() {
            return Err(anyhow::Error::msg("P_MULTI_TENANCY is set to false"));
        }

        let mut tenants = self.tenants.write().unwrap();
        if tenants.contains(&tenant_id) {
            return Err(anyhow::Error::msg(format!(
                "Tenant with id- {tenant_id} already exists"
            )));
        }
        tenants.push(tenant_id.clone());
        drop(tenants);
        TENANT_METADATA.insert_tenant(tenant_id, tenant_meta);

        Ok(())
    }

    pub async fn suspend_tenant_service(
        &self,
        tenant_id: &str,
        service: &Service,
    ) -> Result<(), anyhow::Error> {
        TENANT_METADATA.suspend_service(tenant_id, service);

        // write to disk
        let tenant_id = &Some(tenant_id.to_owned());
        let mut meta = get_metadata(tenant_id).await?;
        if let Some(sus) = meta.suspended_services.as_mut() {
            sus.insert(service.clone());
        } else {
            meta.suspended_services = Some(HashSet::from_iter([service.clone()]));
        }

        put_remote_metadata(&meta, tenant_id).await?;
        Ok(())
    }

    pub async fn resume_tenant_service(
        &self,
        tenant_id: &str,
        service: &Service,
    ) -> Result<(), anyhow::Error> {
        TENANT_METADATA.resume_service(tenant_id, service);

        // write to disk
        let tenant_id = &Some(tenant_id.to_owned());
        let mut meta = get_metadata(tenant_id).await?;
        if let Some(sus) = meta.suspended_services.as_mut() {
            sus.remove(service);
        }

        put_remote_metadata(&meta, tenant_id).await?;
        Ok(())
    }

    pub async fn delete_tenant(&self, tenant_id: &str) -> Result<(), anyhow::Error> {
        // delete users and sessions
        let users = mut_users().remove(tenant_id);
        if let Some(users) = users {
            for (userid, user) in users {
                Users.delete_user(&userid, &user.tenant);
            }
        }

        // delete roles
        mut_roles().remove(tenant_id);

        // delete user group
        write_user_groups().remove(tenant_id);

        // delete streams
        if let Ok(mut streams) = PARSEABLE.streams.write() {
            streams.remove(tenant_id);
        }

        // delete from in-mem
        if let Ok(mut tenants) = self.tenants.write() {
            tenants.retain(|t| t != tenant_id);
        }
        TENANT_METADATA.delete_tenant(tenant_id);
        Ok(())
    }

    async fn load_tenants(&self) -> Result<(), anyhow::Error> {
        let is_multi_tenant = self.options.is_multi_tenant();

        let obj_store = self.storage().get_object_store();
        let dirs: Vec<String> = obj_store
            .list_dirs_relative(&RelativePathBuf::from_iter([""]), &None)
            .await?
            .into_iter()
            .filter(|d| !d.starts_with("."))
            .collect();

        let mut recognized_tenants = vec![];

        // validate the possible presence of tenant storage metadata
        for tenant_id in dirs.into_iter() {
            if let Some(meta) = PARSEABLE
                .metastore
                .get_parseable_metadata(&Some(tenant_id.clone()))
                .await?
                && is_multi_tenant
            {
                let metadata: StorageMetadata = serde_json::from_slice(&meta)?;

                TENANT_METADATA.insert_tenant(tenant_id.clone(), metadata.clone());
                recognized_tenants.push(tenant_id);
            } else if !is_multi_tenant {
            } else {
                // found an invalid multi-tenant directory, delete
                tracing::warn!(
                    "Found invalid tenant directory with multi-tenant mode- {tenant_id}.\nDeleting."
                );

                // delete from storage
                if let Err(e) = obj_store
                    .delete_prefix(&RelativePathBuf::from_iter([&tenant_id]), &None)
                    .await
                {
                    tracing::error!(
                        "Unable to delete invalid directory- {tenant_id} with error-\n{e}"
                    );
                };
            }
        }

        if is_multi_tenant {
            let mut t = self
                .tenants
                .write()
                .expect("Unable to write to in-memory tenant map");
            t.extend(recognized_tenants);
            Ok(())
        } else {
            Ok(())
        }
    }

    pub fn list_tenants(&self) -> Option<Vec<String>> {
        if let Ok(t) = self.tenants.as_ref().read()
            && !t.is_empty()
        {
            let t = t.clone();
            Some(t)
        } else {
            None
        }
    }
}

pub fn validate_static_schema(
    body: &Bytes,
    stream_name: &str,
    time_partition: &str,
    custom_partition: Option<&String>,
    static_schema_flag: bool,
) -> Result<Arc<Schema>, CreateStreamError> {
    if !static_schema_flag {
        return Ok(Arc::new(Schema::empty()));
    }

    if body.is_empty() {
        return Err(CreateStreamError::Custom {
            msg: format!(
                "Please provide schema in the request body for static schema logstream {stream_name}"
            ),
            status: StatusCode::BAD_REQUEST,
        });
    }

    let static_schema: StaticSchema = serde_json::from_slice(body)?;
    let parsed_schema =
        convert_static_schema_to_arrow_schema(static_schema, time_partition, custom_partition)
            .map_err(|_| CreateStreamError::Custom {
                msg: format!("Unable to commit static schema, logstream {stream_name} not created"),
                status: StatusCode::BAD_REQUEST,
            })?;

    Ok(parsed_schema)
}

pub fn validate_time_partition_limit(
    time_partition_limit: &str,
) -> Result<NonZeroU32, CreateStreamError> {
    if !time_partition_limit.ends_with('d') {
        return Err(CreateStreamError::Custom {
            msg: "Missing 'd' suffix for duration value".to_string(),
            status: StatusCode::BAD_REQUEST,
        });
    }
    let days = &time_partition_limit[0..time_partition_limit.len() - 1];
    let Ok(days) = days.parse::<NonZeroU32>() else {
        return Err(CreateStreamError::Custom {
            msg: "Could not convert duration to an unsigned number".to_string(),
            status: StatusCode::BAD_REQUEST,
        });
    };

    Ok(days)
}

pub fn validate_custom_partition(custom_partition: &str) -> Result<(), CreateStreamError> {
    let custom_partition_list = custom_partition.split(',').collect::<Vec<&str>>();
    if custom_partition_list.len() > 1 {
        return Err(CreateStreamError::Custom {
            msg: "Maximum 1 custom partition key is supported".to_string(),
            status: StatusCode::BAD_REQUEST,
        });
    }

    for partition in custom_partition_list.iter() {
        if partition.contains('.') {
            return Err(CreateStreamError::Custom {
                msg: format!("custom partition field {partition} must not contain '.'"),
                status: StatusCode::BAD_REQUEST,
            });
        }
    }

    Ok(())
}