// Copyright 2020-2022 The NATS 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.

//! Support for Object Store.
//! This feature is experimental and the API may change.

use crate::header::HeaderMap;
use crate::jetstream::{
    DateTime, DiscardPolicy, JetStream, PushSubscription, StorageType, StreamConfig,
    SubscribeOptions,
};
use crate::Message;
use base64::URL_SAFE;
use lazy_static::lazy_static;
use regex::Regex;
use ring::digest::SHA256;
use serde::{Deserialize, Serialize};
use std::cmp;
use std::io::{self, ErrorKind};
use std::time::Duration;
use time::serde::rfc3339;
use time::OffsetDateTime;

const DEFAULT_CHUNK_SIZE: usize = 128 * 1024;
const NATS_ROLLUP: &str = "Nats-Rollup";
const ROLLUP_SUBJECT: &str = "sub";

lazy_static! {
    static ref BUCKET_NAME_RE: Regex = Regex::new(r#"\A[a-zA-Z0-9_-]+\z"#).unwrap();
    static ref OBJECT_NAME_RE: Regex = Regex::new(r#"\A[-/_=\.a-zA-Z0-9]+\z"#).unwrap();
}

fn is_valid_bucket_name(bucket_name: &str) -> bool {
    BUCKET_NAME_RE.is_match(bucket_name)
}

fn is_valid_object_name(object_name: &str) -> bool {
    if object_name.is_empty() || object_name.starts_with('.') || object_name.ends_with('.') {
        return false;
    }

    OBJECT_NAME_RE.is_match(object_name)
}

fn encode_object_name(object_name: &str) -> String {
    base64::encode_config(object_name, URL_SAFE)
}

/// Configuration values for object store buckets.
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct Config {
    /// Name of the storage bucket.
    pub bucket: String,
    /// A short description of the purpose of this storage bucket.
    pub description: Option<String>,
    /// Maximum age of any value in the bucket, expressed in nanoseconds
    pub max_age: Duration,
    /// The type of storage backend, `File` (default) and `Memory`
    pub storage: StorageType,
    /// How many replicas to keep for each value in a cluster, maximum 5.
    pub num_replicas: usize,
}

impl JetStream {
    /// Creates a new object store bucket.
    ///
    /// # Example
    ///
    /// ```
    /// # use nats::object_store::Config;
    /// # fn main() -> std::io::Result<()> {
    /// # let client = nats::connect("demo.nats.io")?;
    /// # let context = nats::jetstream::new(client);
    /// #
    /// let bucket = context.create_object_store(&Config {
    ///   bucket: "create_object_store".to_string(),
    ///   ..Default::default()
    /// })?;
    ///
    /// # context.delete_object_store("create_object_store")?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn create_object_store(&self, config: &Config) -> io::Result<ObjectStore> {
        if !self.connection.is_server_compatible_version(2, 6, 2) {
            return Err(io::Error::new(
                io::ErrorKind::Other,
                "object-store requires at least server version 2.6.2",
            ));
        }

        if !is_valid_bucket_name(&config.bucket) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid bucket name",
            ));
        }

        let bucket_name = config.bucket.clone();
        let stream_name = format!("OBJ_{}", bucket_name);
        let chunk_subject = format!("$O.{}.C.>", bucket_name);
        let meta_subject = format!("$O.{}.M.>", bucket_name);

        self.add_stream(&StreamConfig {
            name: stream_name,
            description: config.description.clone(),
            subjects: vec![chunk_subject, meta_subject],
            max_age: config.max_age,
            storage: config.storage,
            num_replicas: config.num_replicas,
            discard: DiscardPolicy::New,
            allow_rollup: true,
            ..Default::default()
        })?;

        Ok(ObjectStore::new(bucket_name, self.clone()))
    }

    /// Bind to an existing object store bucket.
    ///
    /// # Example
    ///
    /// ```
    /// # use nats::object_store::Config;
    /// # fn main() -> std::io::Result<()> {
    /// # let client = nats::connect("demo.nats.io")?;
    /// # let context = nats::jetstream::new(client);
    /// #
    /// context.create_object_store(&Config {
    ///   bucket: "object_store".to_string(),
    ///   ..Default::default()
    /// })?;
    ///
    /// let bucket = context.object_store("object_store")?;
    ///
    /// # context.delete_object_store("object_store")?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn object_store(&self, bucket_name: &str) -> io::Result<ObjectStore> {
        if !self.connection.is_server_compatible_version(2, 6, 2) {
            return Err(io::Error::new(
                io::ErrorKind::Other,
                "object-store requires at least server version 2.6.2",
            ));
        }

        if !is_valid_bucket_name(bucket_name) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid bucket name",
            ));
        }

        let stream_name = format!("OBJ_{}", bucket_name);
        self.stream_info(stream_name)?;

        Ok(ObjectStore::new(bucket_name.to_string(), self.clone()))
    }

    /// Delete the underlying stream for the named object.
    ///
    /// # Example
    ///
    /// ```
    /// use nats::object_store::Config;
    /// # fn main() -> std::io::Result<()> {
    /// # let client = nats::connect("demo.nats.io")?;
    /// # let context = nats::jetstream::new(client);
    /// #
    /// # let bucket = context.create_object_store(&Config {
    /// #  bucket: "delete_object_store".to_string(),
    /// #  ..Default::default()
    /// # })?;
    ///
    /// context.delete_object_store("delete_object_store")?;
    ///
    /// # Ok(())
    /// # }
    /// ```
    ///
    pub fn delete_object_store(&self, bucket_name: &str) -> io::Result<()> {
        let stream_name = format!("OBJ_{}", bucket_name);
        self.delete_stream(stream_name)?;

        Ok(())
    }
}

/// Meta and instance information about an object.
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
pub struct ObjectInfo {
    /// Name of the object
    pub name: String,
    /// A short human readable description of the object.
    pub description: Option<String>,
    /// Link this object points to, if any.
    pub link: Option<ObjectLink>,
    /// Name of the bucket the object is stored in.
    pub bucket: String,
    /// Unique identifier used to uniquely identify this version of the object.
    pub nuid: String,
    /// Size in bytes of the object.
    pub size: usize,
    /// Number of chunks the object is stored in.
    pub chunks: usize,
    /// Date and time the object was last modified.
    #[serde(with = "rfc3339")]
    pub modified: DateTime,
    /// Digest of the object stream.
    pub digest: String,
    /// Set to true if the object has been deleted.
    pub deleted: bool,
}

/// Meta information about an object.
#[derive(Debug, Default, Clone, Serialize, Deserialize, Eq, PartialEq)]
pub struct ObjectMeta {
    /// Name of the object
    pub name: String,
    /// A short human readable description of the object.
    pub description: Option<String>,
    /// Link this object points to, if any.
    pub link: Option<ObjectLink>,
}

impl From<&str> for ObjectMeta {
    fn from(s: &str) -> ObjectMeta {
        ObjectMeta {
            name: s.to_string(),
            ..Default::default()
        }
    }
}

/// A link to another object, potentially in another bucket.
#[derive(Debug, Default, Clone, Serialize, Deserialize, Eq, PartialEq)]
pub struct ObjectLink {
    /// Name of the object
    pub name: String,
    /// Name of the bucket the object is stored in.
    pub bucket: Option<String>,
}

/// A blob store capable of storing large objects efficiently in streams.
pub struct ObjectStore {
    name: String,
    context: JetStream,
}

/// Represents an object stored in a bucket.
pub struct Object {
    /// Information about given object.
    pub info: ObjectInfo,
    subscription: PushSubscription,
    remaining_bytes: Vec<u8>,
    has_pending_messages: bool,
    digest: Option<ring::digest::Context>,
}

impl Object {
    pub(crate) fn new(subscription: PushSubscription, info: ObjectInfo) -> Self {
        Object {
            subscription,
            info,
            remaining_bytes: Vec::new(),
            has_pending_messages: true,
            digest: Some(ring::digest::Context::new(&SHA256)),
        }
    }

    /// Returns information about the object.
    pub fn info(&self) -> &ObjectInfo {
        &self.info
    }
}

impl io::Read for Object {
    /// Read the data chunks for a given Object from attached subscription and copy it to provided buffer.
    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
        // read data accumulated in remaining bytes into the buffer.
        if !self.remaining_bytes.is_empty() {
            let len = cmp::min(buffer.len(), self.remaining_bytes.len());
            buffer[..len].copy_from_slice(&self.remaining_bytes[..len]);
            self.remaining_bytes = self.remaining_bytes[len..].to_vec();
            return Ok(len);
        }

        // fetch messages from subject.
        // Run at each `read` call until there are no more pending messages for a given Object.
        if self.has_pending_messages {
            let maybe_message = self.subscription.next();
            if let Some(message) = maybe_message {
                let len = cmp::min(buffer.len(), message.data.len());
                buffer[..len].copy_from_slice(&message.data[..len]);
                if let Some(context) = &mut self.digest {
                    context.update(&message.data);
                }
                self.remaining_bytes.extend_from_slice(&message.data[len..]);

                if let Some(message_info) = message.jetstream_message_info() {
                    if message_info.pending == 0 {
                        let digest = self.digest.take().map(|context| context.finish());
                        if let Some(digest) = digest {
                            if format!("SHA-256={}", base64::encode_config(digest, URL_SAFE))
                                != self.info.digest
                            {
                                return Err(io::Error::new(ErrorKind::InvalidData, "wrong digest"));
                            }
                        } else {
                            return Err(io::Error::new(
                                ErrorKind::InvalidData,
                                "digest should be Some",
                            ));
                        }
                        self.has_pending_messages = false;
                    }
                }
                return Ok(len);
            }
        }

        Ok(0)
    }
}

impl ObjectStore {
    /// Instantiates a new object store
    pub(crate) fn new(name: String, context: JetStream) -> Self {
        ObjectStore { name, context }
    }

    /// Retrieve the current information for the object.
    ///
    /// # Examples
    ///
    /// ```
    /// # use nats::object_store::Config;
    /// # fn main() -> std::io::Result<()> {
    /// # let client = nats::connect("demo.nats.io")?;
    /// # let context = nats::jetstream::new(client);
    /// #
    /// let bucket = context.create_object_store(&Config {
    ///   bucket: "info".to_string(),
    ///   ..Default::default()
    /// })?;
    ///
    /// let bytes = vec![0];
    /// let info = bucket.put("foo", &mut bytes.as_slice())?;
    /// assert_eq!(info.name, "foo");
    /// assert_eq!(info.size, bytes.len());
    ///
    /// # context.delete_object_store("info")?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn info(&self, object_name: &str) -> io::Result<ObjectInfo> {
        // LoOkup the stream to get the bound subject.
        let object_name = encode_object_name(object_name);
        if !is_valid_object_name(&object_name) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid object name",
            ));
        }

        // Grab last meta value we have.
        let stream_name = format!("OBJ_{}", &self.name);
        let subject = format!("$O.{}.M.{}", &self.name, &object_name);

        let message = self.context.get_last_message(stream_name, &subject)?;
        let object_info = serde_json::from_slice::<ObjectInfo>(&message.data)?;

        Ok(object_info)
    }

    /// Seals the object store from further modifications.
    ///
    pub fn seal(&self) -> io::Result<()> {
        let stream_name = format!("OBJ_{}", self.name);
        let stream_info = self.context.stream_info(stream_name)?;

        let mut stream_config = stream_info.config;
        stream_config.sealed = true;

        self.context.update_stream(&stream_config)?;

        Ok(())
    }

    /// Put will place the contents from the given reader into this object-store.
    ///
    /// # Example
    ///
    /// ```
    /// # use nats::object_store::Config;
    /// # fn main() -> std::io::Result<()> {
    /// # let client = nats::connect("demo.nats.io")?;
    /// # let context = nats::jetstream::new(client);
    /// #
    /// let bucket = context.create_object_store(&Config {
    ///   bucket: "put".to_string(),
    ///   ..Default::default()
    /// })?;
    ///
    /// let bytes = vec![0, 1, 2, 3, 4];
    /// let info = bucket.put("foo", &mut bytes.as_slice())?;
    /// assert_eq!(bucket.info("foo").unwrap(), info);
    ///
    /// # context.delete_object_store("put")?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn put<T>(&self, meta: T, data: &mut impl io::Read) -> io::Result<ObjectInfo>
    where
        ObjectMeta: From<T>,
    {
        let object_meta: ObjectMeta = meta.into();
        let object_name = encode_object_name(&object_meta.name);
        if !is_valid_object_name(&object_name) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid object name",
            ));
        }

        // Fetch any existing object info, if ther is any for later use.
        let maybe_existing_object_info = match self.info(&object_name) {
            Ok(object_info) => Some(object_info),
            Err(_) => None,
        };

        let object_nuid = nuid::next();
        let chunk_subject = format!("$O.{}.C.{}", &self.name, &object_nuid);

        let mut object_chunks = 0;
        let mut object_size = 0;

        let mut context = ring::digest::Context::new(&SHA256);
        let mut buffer = [0; DEFAULT_CHUNK_SIZE];

        loop {
            let n = data.read(&mut buffer)?;
            if n == 0 {
                break;
            }
            context.update(&buffer[..n]);

            object_size += n;
            object_chunks += 1;

            self.context.publish(&chunk_subject, &buffer[..n])?;
        }

        let digest = context.finish();
        // Create a random subject prefixed with the object stream name.
        let subject = format!("$O.{}.M.{}", &self.name, &object_name);
        let object_info = ObjectInfo {
            name: object_meta.name,
            description: object_meta.description,
            link: object_meta.link,
            bucket: self.name.clone(),
            nuid: object_nuid,
            chunks: object_chunks,
            size: object_size,
            digest: format!("SHA-256={}", base64::encode_config(digest, URL_SAFE)),
            modified: OffsetDateTime::now_utc(),
            deleted: false,
        };

        let data = serde_json::to_vec(&object_info)?;
        let mut headers = HeaderMap::default();
        headers.insert(NATS_ROLLUP, ROLLUP_SUBJECT.to_string());

        let message = Message::new(&subject, None, data, Some(headers));

        // Publish metadata
        self.context.publish_message(&message)?;

        // Purge any old chunks.
        if let Some(existing_object_info) = maybe_existing_object_info {
            let stream_name = format!("OBJ_{}", self.name);
            let chunk_subject = format!("$O.{}.C.{}", &self.name, &existing_object_info.nuid);

            self.context
                .purge_stream_subject(stream_name, &chunk_subject)?;
        }

        Ok(object_info)
    }

    /// Get an existing object by name.
    ///
    /// # Example
    ///
    /// ```
    /// use std::io::Read;
    /// # use nats::object_store::Config;
    /// # fn main() -> std::io::Result<()> {
    /// # let client = nats::connect("demo.nats.io")?;
    /// # let context = nats::jetstream::new(client);
    /// #
    /// let bucket = context.create_object_store(&Config {
    ///   bucket: "get".to_string(),
    ///   ..Default::default()
    /// })?;
    ///
    /// let bytes = vec![0, 1, 2, 3, 4];
    /// let info = bucket.put("foo", &mut bytes.as_slice())?;
    ///
    /// let mut result = Vec::new();
    /// bucket.get("foo").unwrap().read_to_end(&mut result)?;
    ///
    /// # context.delete_object_store("get")?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn get(&self, object_name: &str) -> io::Result<Object> {
        let object_info = self.info(object_name)?;
        if let Some(link) = object_info.link {
            return self.get(&link.name);
        }

        let chunk_subject = format!("$O.{}.C.{}", self.name, object_info.nuid);
        let subscription = self
            .context
            .subscribe_with_options(&chunk_subject, &SubscribeOptions::ordered())?;

        Ok(Object::new(subscription, object_info))
    }

    /// Places a delete marker and purges the data stream associated with the key.
    ///
    /// # Example
    ///
    /// ```
    /// use std::io::Read;
    /// # use nats::object_store::Config;
    /// # fn main() -> std::io::Result<()> {
    /// # let client = nats::connect("demo.nats.io")?;
    /// # let context = nats::jetstream::new(client);
    /// #
    /// let bucket = context.create_object_store(&Config {
    ///   bucket: "delete".to_string(),
    ///   ..Default::default()
    /// })?;
    ///
    /// let bytes = vec![0, 1, 2, 3, 4];
    /// bucket.put("foo", &mut bytes.as_slice())?;
    ///
    /// bucket.delete("foo")?;
    ///
    /// let info = bucket.info("foo")?;
    /// assert!(info.deleted);
    /// assert_eq!(info.size, 0);
    /// assert_eq!(info.chunks, 0);
    ///
    /// # context.delete_object_store("delete")?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn delete(&self, object_name: &str) -> io::Result<()> {
        let mut object_info = self.info(object_name)?;
        object_info.chunks = 0;
        object_info.size = 0;
        object_info.deleted = true;

        let data = serde_json::to_vec(&object_info)?;

        let mut headers = HeaderMap::default();
        headers.insert(NATS_ROLLUP, ROLLUP_SUBJECT.to_string());

        let subject = format!("$O.{}.M.{}", &self.name, &encode_object_name(object_name));
        let message = Message::new(&subject, None, data, Some(headers));

        self.context.publish_message(&message)?;

        let stream_name = format!("OBJ_{}", self.name);
        let chunk_subject = format!("$O.{}.C.{}", self.name, object_info.nuid);

        self.context
            .purge_stream_subject(stream_name, &chunk_subject)?;

        Ok(())
    }

    /// Watch for changes in the underlying store and receive meta information updates.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use std::io::Read;
    /// # use nats::object_store::Config;
    /// # fn main() -> std::io::Result<()> {
    /// # let client = nats::connect("demo.nats.io")?;
    /// # let context = nats::jetstream::new(client);
    /// #
    /// let bucket = context.create_object_store(&Config {
    ///   bucket: "watch".to_string(),
    ///   ..Default::default()
    /// })?;
    ///
    /// let mut watch = bucket.watch()?;
    ///
    /// let bytes = vec![0, 1, 2, 3, 4];
    /// bucket.put("foo", &mut bytes.as_slice())?;
    ///
    /// let info = watch.next().unwrap();
    /// assert_eq!(info.name, "foo");
    /// assert_eq!(info.size, bytes.len());
    ///
    /// let bytes = vec![0];
    /// bucket.put("bar", &mut bytes.as_slice())?;
    ///
    /// let info = watch.next().unwrap();
    /// assert_eq!(info.name, "bar");
    /// assert_eq!(info.size, bytes.len());
    ///
    /// # context.delete_object_store("watch")?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn watch(&self) -> io::Result<Watch> {
        let subject = format!("$O.{}.M.>", &self.name);
        let subscription = self.context.subscribe_with_options(
            &subject,
            &SubscribeOptions::ordered().deliver_last_per_subject(),
        )?;

        Ok(Watch { subscription })
    }
}

/// Iterator returned by `watch`
pub struct Watch {
    subscription: PushSubscription,
}

impl Iterator for Watch {
    type Item = ObjectInfo;

    fn next(&mut self) -> Option<Self::Item> {
        match self.subscription.next() {
            Some(message) => Some(serde_json::from_slice(&message.data).unwrap()),
            None => None,
        }
    }
}