// 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.

use std::any::Any;

use api::v1::value::ValueData;
use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType};
use common_event_recorder::Event;
use common_event_recorder::error::Result;
use common_time::timestamp::{TimeUnit, Timestamp};

use crate::{ProcedureId, ProcedureState};

pub const EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME: &str = "procedure_id";
pub const EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME: &str = "procedure_state";
pub const EVENTS_TABLE_PROCEDURE_ERROR_COLUMN_NAME: &str = "procedure_error";

/// `ProcedureEvent` represents an event emitted by a procedure during its execution lifecycle.
#[derive(Debug)]
pub struct ProcedureEvent {
    /// Unique identifier associated with the originating procedure instance.
    pub procedure_id: ProcedureId,
    /// The timestamp of the event.
    pub timestamp: Timestamp,
    /// The state of the procedure.
    pub state: ProcedureState,
    /// The event emitted by the procedure. It's generated by [Procedure::event].
    pub internal_event: Box<dyn Event>,
}

impl ProcedureEvent {
    pub fn new(
        procedure_id: ProcedureId,
        internal_event: Box<dyn Event>,
        state: ProcedureState,
    ) -> Self {
        Self {
            procedure_id,
            internal_event,
            timestamp: Timestamp::current_time(TimeUnit::Nanosecond),
            state,
        }
    }
}

impl Event for ProcedureEvent {
    fn event_type(&self) -> &str {
        self.internal_event.event_type()
    }

    fn timestamp(&self) -> Timestamp {
        self.timestamp
    }

    fn json_payload(&self) -> Result<serde_json::Value> {
        self.internal_event.json_payload()
    }

    fn extra_schema(&self) -> Vec<ColumnSchema> {
        let mut schema = vec![
            ColumnSchema {
                column_name: EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME.to_string(),
                datatype: ColumnDataType::String.into(),
                semantic_type: SemanticType::Field.into(),
                ..Default::default()
            },
            ColumnSchema {
                column_name: EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME.to_string(),
                datatype: ColumnDataType::String.into(),
                semantic_type: SemanticType::Field.into(),
                ..Default::default()
            },
            ColumnSchema {
                column_name: EVENTS_TABLE_PROCEDURE_ERROR_COLUMN_NAME.to_string(),
                datatype: ColumnDataType::String.into(),
                semantic_type: SemanticType::Field.into(),
                ..Default::default()
            },
        ];
        schema.append(&mut self.internal_event.extra_schema());
        schema
    }

    fn extra_rows(&self) -> Result<Vec<Row>> {
        let mut internal_event_extra_rows = self.internal_event.extra_rows()?;
        let mut rows = Vec::with_capacity(internal_event_extra_rows.len());
        for internal_event_extra_row in internal_event_extra_rows.iter_mut() {
            let error_str = match &self.state {
                ProcedureState::Failed { error } => format!("{:?}", error),
                ProcedureState::PrepareRollback { error } => format!("{:?}", error),
                ProcedureState::RollingBack { error } => format!("{:?}", error),
                ProcedureState::Retrying { error } => format!("{:?}", error),
                ProcedureState::Poisoned { error, .. } => format!("{:?}", error),
                _ => "".to_string(),
            };
            let mut values = Vec::with_capacity(3 + internal_event_extra_row.values.len());
            values.extend([
                ValueData::StringValue(self.procedure_id.to_string()).into(),
                ValueData::StringValue(self.state.as_str_name().to_string()).into(),
                ValueData::StringValue(error_str).into(),
            ]);
            values.append(&mut internal_event_extra_row.values);
            rows.push(Row { values });
        }

        Ok(rows)
    }

    fn as_any(&self) -> &dyn Any {
        self
    }
}

#[cfg(test)]
mod tests {
    use api::v1::value::ValueData;
    use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType};
    use common_event_recorder::Event;

    use crate::{ProcedureEvent, ProcedureId, ProcedureState};

    #[derive(Debug)]
    struct TestEvent;

    impl Event for TestEvent {
        fn event_type(&self) -> &str {
            "test_event"
        }

        fn extra_schema(&self) -> Vec<ColumnSchema> {
            vec![ColumnSchema {
                column_name: "test_event_column".to_string(),
                datatype: ColumnDataType::String.into(),
                semantic_type: SemanticType::Field.into(),
                ..Default::default()
            }]
        }

        fn extra_rows(&self) -> common_event_recorder::error::Result<Vec<Row>> {
            Ok(vec![
                Row {
                    values: vec![ValueData::StringValue("test_event1".to_string()).into()],
                },
                Row {
                    values: vec![ValueData::StringValue("test_event2".to_string()).into()],
                },
            ])
        }

        fn as_any(&self) -> &dyn std::any::Any {
            self
        }
    }

    #[test]
    fn test_procedure_event_extra_rows() {
        let procedure_event = ProcedureEvent::new(
            ProcedureId::random(),
            Box::new(TestEvent {}),
            ProcedureState::Running,
        );

        let procedure_event_extra_rows = procedure_event.extra_rows().unwrap();
        assert_eq!(procedure_event_extra_rows.len(), 2);
        assert_eq!(procedure_event_extra_rows[0].values.len(), 4);
        assert_eq!(
            procedure_event_extra_rows[0].values[3],
            ValueData::StringValue("test_event1".to_string()).into()
        );
        assert_eq!(procedure_event_extra_rows[1].values.len(), 4);
        assert_eq!(
            procedure_event_extra_rows[1].values[3],
            ValueData::StringValue("test_event2".to_string()).into()
        );
    }
}
