// Copyright 2023 RisingWave Labs
//
// 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.

use risingwave_common::id::UserId;
use risingwave_common::types::{Fields, Timestamptz};
use risingwave_frontend_macro::system_catalog;
use risingwave_pb::id::RelationId;
use serde_json::json;

use crate::catalog::system_catalog::SysCatalogReaderImpl;
use crate::error::Result;

// TODO: `rw_relation_info` contains some extra streaming meta info that's only meaningful for
// streaming jobs, we'd better query relation infos from `rw_relations` and move these streaming
// infos into anther system table.
#[derive(Fields)]
#[primary_key(schemaname, relationname)]
struct RwRelationInfo {
    schemaname: String,
    relationname: String,
    relationowner: UserId,
    definition: String,
    relationtype: String,
    relationid: RelationId,
    relationtimezone: String, // The timezone used to interpret ambiguous dates/timestamps as tstz
    fragments: Option<String>, // fragments is json encoded fragment infos.
    initialized_at: Option<Timestamptz>,
    created_at: Option<Timestamptz>,
    initialized_at_cluster_version: Option<String>,
    created_at_cluster_version: Option<String>,
}

#[system_catalog(table, "rw_catalog.rw_relation_info")]
async fn read_relation_info(reader: &SysCatalogReaderImpl) -> Result<Vec<RwRelationInfo>> {
    let mut job_ids = Vec::new();
    {
        let catalog_reader = reader.catalog_reader.read_guard();
        let schemas = catalog_reader.get_all_schema_names(&reader.auth_context.database)?;
        let user_reader = reader.user_info_reader.read_guard();
        let current_user = user_reader
            .get_user_by_name(&reader.auth_context.user_name)
            .expect("user not found");
        for schema in &schemas {
            let schema_catalog =
                catalog_reader.get_schema_by_name(&reader.auth_context.database, schema)?;

            schema_catalog
                .iter_created_mvs_with_acl(current_user)
                .for_each(|t| {
                    job_ids.push(t.id.as_job_id());
                });

            schema_catalog
                .iter_user_table_with_acl(current_user)
                .for_each(|t| {
                    job_ids.push(t.id.as_job_id());
                });

            schema_catalog
                .iter_source_with_acl(current_user)
                .filter(|s| s.info.is_shared())
                .for_each(|s| {
                    job_ids.push(s.id.as_share_source_job_id());
                });

            schema_catalog
                .iter_sink_with_acl(current_user)
                .for_each(|t| {
                    job_ids.push(t.id.as_job_id());
                });

            schema_catalog
                .iter_index_with_acl(current_user)
                .for_each(|t| {
                    job_ids.push(t.index_table().id.as_job_id());
                });
        }
    }

    let table_fragments = reader.meta_client.list_table_fragments(&job_ids).await?;
    let mut rows = Vec::new();
    let catalog_reader = reader.catalog_reader.read_guard();
    let schemas = catalog_reader.get_all_schema_names(&reader.auth_context.database)?;
    let user_reader = reader.user_info_reader.read_guard();
    let current_user = user_reader
        .get_user_by_name(&reader.auth_context.user_name)
        .expect("user not found");
    for schema in &schemas {
        let schema_catalog =
            catalog_reader.get_schema_by_name(&reader.auth_context.database, schema)?;
        schema_catalog
            .iter_created_mvs_with_acl(current_user)
            .for_each(|t| {
                if let Some(fragments) = table_fragments.get(&t.id.as_job_id()) {
                    rows.push(RwRelationInfo {
                        schemaname: schema.clone(),
                        relationname: t.name.clone(),
                        relationowner: t.owner,
                        definition: t.create_sql(),
                        relationtype: "MATERIALIZED VIEW".into(),
                        relationid: t.id.as_relation_id(),
                        relationtimezone: fragments.get_ctx().unwrap().get_timezone().clone(),
                        fragments: Some(json!(fragments.get_fragments()).to_string()),
                        initialized_at: t.initialized_at_epoch.map(|e| e.as_timestamptz()),
                        created_at: t.created_at_epoch.map(|e| e.as_timestamptz()),
                        initialized_at_cluster_version: t.initialized_at_cluster_version.clone(),
                        created_at_cluster_version: t.created_at_cluster_version.clone(),
                    });
                }
            });

        schema_catalog
            .iter_user_table_with_acl(current_user)
            .for_each(|t| {
                if let Some(fragments) = table_fragments.get(&t.id.as_job_id()) {
                    rows.push(RwRelationInfo {
                        schemaname: schema.clone(),
                        relationname: t.name.clone(),
                        relationowner: t.owner,
                        definition: t.create_sql_purified(),
                        relationtype: "TABLE".into(),
                        relationid: t.id.as_relation_id(),
                        relationtimezone: fragments.get_ctx().unwrap().get_timezone().clone(),
                        fragments: Some(json!(fragments.get_fragments()).to_string()),
                        initialized_at: t.initialized_at_epoch.map(|e| e.as_timestamptz()),
                        created_at: t.created_at_epoch.map(|e| e.as_timestamptz()),
                        initialized_at_cluster_version: t.initialized_at_cluster_version.clone(),
                        created_at_cluster_version: t.created_at_cluster_version.clone(),
                    });
                }
            });

        schema_catalog
            .iter_sink_with_acl(current_user)
            .for_each(|t| {
                if let Some(fragments) = table_fragments.get(&t.id.as_job_id()) {
                    rows.push(RwRelationInfo {
                        schemaname: schema.clone(),
                        relationname: t.name.clone(),
                        relationowner: t.owner,
                        definition: t.definition.clone(),
                        relationtype: "SINK".into(),
                        relationid: t.id.as_relation_id(),
                        relationtimezone: fragments.get_ctx().unwrap().get_timezone().clone(),
                        fragments: Some(json!(fragments.get_fragments()).to_string()),
                        initialized_at: t.initialized_at_epoch.map(|e| e.as_timestamptz()),
                        created_at: t.created_at_epoch.map(|e| e.as_timestamptz()),
                        initialized_at_cluster_version: t.initialized_at_cluster_version.clone(),
                        created_at_cluster_version: t.created_at_cluster_version.clone(),
                    });
                }
            });

        schema_catalog
            .iter_index_with_acl(current_user)
            .for_each(|t| {
                if let Some(fragments) = table_fragments.get(&t.index_table().id.as_job_id()) {
                    let index_table = t.index_table();
                    rows.push(RwRelationInfo {
                        schemaname: schema.clone(),
                        relationname: t.name.clone(),
                        relationowner: index_table.owner,
                        definition: index_table.create_sql(),
                        relationtype: "INDEX".into(),
                        relationid: index_table.id.as_relation_id(),
                        relationtimezone: fragments.get_ctx().unwrap().get_timezone().clone(),
                        fragments: Some(json!(fragments.get_fragments()).to_string()),
                        initialized_at: t.initialized_at_epoch.map(|e| e.as_timestamptz()),
                        created_at: t.created_at_epoch.map(|e| e.as_timestamptz()),
                        initialized_at_cluster_version: t.initialized_at_cluster_version.clone(),
                        created_at_cluster_version: t.created_at_cluster_version.clone(),
                    });
                }
            });

        // Sources have no fragments.
        schema_catalog
            .iter_source_with_acl(current_user)
            .for_each(|t| {
                let (timezone, fragments) = if t.info.is_shared()
                    && let Some(fragments) = table_fragments.get(&t.id.as_share_source_job_id())
                {
                    (
                        fragments.get_ctx().unwrap().get_timezone().clone(),
                        Some(json!(fragments.get_fragments()).to_string()),
                    )
                } else {
                    ("".into(), None)
                };

                rows.push(RwRelationInfo {
                    schemaname: schema.clone(),
                    relationname: t.name.clone(),
                    relationowner: t.owner,
                    definition: t.create_sql_purified(),
                    relationtype: "SOURCE".into(),
                    relationid: t.id.as_relation_id(),
                    relationtimezone: timezone,
                    fragments,
                    initialized_at: t.initialized_at_epoch.map(|e| e.as_timestamptz()),
                    created_at: t.created_at_epoch.map(|e| e.as_timestamptz()),
                    initialized_at_cluster_version: t.initialized_at_cluster_version.clone(),
                    created_at_cluster_version: t.created_at_cluster_version.clone(),
                });
            });

        schema_catalog
            .iter_subscription_with_acl(current_user)
            .for_each(|t| {
                rows.push(RwRelationInfo {
                    schemaname: schema.clone(),
                    relationname: t.name.clone(),
                    relationowner: t.owner,
                    definition: t.definition.clone(),
                    relationtype: "SUBSCRIPTION".into(),
                    relationid: t.id.as_relation_id(),
                    relationtimezone: "".into(),
                    fragments: None,
                    initialized_at: t.initialized_at_epoch.map(|e| e.as_timestamptz()),
                    created_at: t.created_at_epoch.map(|e| e.as_timestamptz()),
                    initialized_at_cluster_version: t.initialized_at_cluster_version.clone(),
                    created_at_cluster_version: t.created_at_cluster_version.clone(),
                });
            });
    }

    Ok(rows)
}
