// 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::{ColumnSchema, Row};
use common_event_recorder::Event;
use common_event_recorder::error::{Result, SerializeEventSnafu};
use common_event_recorder::event_table::{
    ACTOR_COLUMN, EVENT_CONTEXT_COLUMN, PROCEDURE_ERROR_COLUMN, PROCEDURE_ID_COLUMN,
    PROCEDURE_STATE_COLUMN, PROCEDURE_TRIGGER_COLUMN, jsonb_value, nullable_json, nullable_string,
    procedure_event_column_schemas,
};
use common_time::timestamp::{TimeUnit, Timestamp};
use snafu::ResultExt;

use crate::{EventTrigger, ProcedureContext, ProcedureId, ProcedureState};

pub const EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME: &str = PROCEDURE_ID_COLUMN.name();
pub const EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME: &str = PROCEDURE_STATE_COLUMN.name();
pub const EVENTS_TABLE_PROCEDURE_ERROR_COLUMN_NAME: &str = PROCEDURE_ERROR_COLUMN.name();
pub const EVENTS_TABLE_PROCEDURE_TRIGGER_COLUMN_NAME: &str = PROCEDURE_TRIGGER_COLUMN.name();

/// `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 lifecycle trigger that caused the event to be emitted.
    pub trigger: EventTrigger,
    /// Context associated with the root submission.
    pub context: ProcedureContext,
    /// 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,
        trigger: EventTrigger,
    ) -> Self {
        Self::new_with_context(
            procedure_id,
            internal_event,
            state,
            trigger,
            ProcedureContext::default(),
        )
    }

    pub fn new_with_context(
        procedure_id: ProcedureId,
        internal_event: Box<dyn Event>,
        state: ProcedureState,
        trigger: EventTrigger,
        context: ProcedureContext,
    ) -> Self {
        Self {
            procedure_id,
            internal_event,
            timestamp: Timestamp::current_time(TimeUnit::Nanosecond),
            state,
            trigger,
            context,
        }
    }
}

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 = procedure_event_column_schemas();
        let mut internal_schema = self.internal_event.extra_schema();
        schema.append(&mut internal_schema);
        schema.push(ACTOR_COLUMN.column_schema());
        schema.push(EVENT_CONTEXT_COLUMN.column_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());
        let procedure_id = self.procedure_id.to_string();
        let state = self.state.as_str_name().to_string();
        let error = match &self.state {
            ProcedureState::Failed { error }
            | ProcedureState::PrepareRollback { error }
            | ProcedureState::RollingBack { error }
            | ProcedureState::Retrying { error }
            | ProcedureState::Poisoned { error, .. } => format!("{error:?}"),
            _ => String::new(),
        };
        let trigger = serde_json::to_value(&self.trigger).context(SerializeEventSnafu)?;
        let event_context = matches!(self.trigger, EventTrigger::Submitted)
            .then_some(self.context.event_context.as_ref())
            .flatten()
            .map(serde_json::to_value)
            .transpose()
            .context(SerializeEventSnafu)?;
        let actor = self.context.actor.as_deref();

        for internal_event_extra_row in internal_event_extra_rows.iter_mut() {
            let mut values = Vec::with_capacity(6 + internal_event_extra_row.values.len());
            values.extend([
                ValueData::StringValue(procedure_id.clone()).into(),
                ValueData::StringValue(state.clone()).into(),
                ValueData::StringValue(error.clone()).into(),
                jsonb_value(&trigger),
            ]);
            values.append(&mut internal_event_extra_row.values);
            values.push(nullable_string(actor));
            values.push(nullable_json(event_context.as_ref()));
            rows.push(Row { values });
        }

        Ok(rows)
    }

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

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use api::v1::value::ValueData;
    use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType, Value};
    use common_error::mock::MockError;
    use common_error::status_code::StatusCode;
    use common_event_recorder::event_table::{
        ACTOR_COLUMN, EVENT_CONTEXT_COLUMN, PROCEDURE_TRIGGER_COLUMN, jsonb_value,
    };
    use common_event_recorder::{Event, PersistentEventContext, TriggerReason};
    use serde_json::json;

    use crate::{
        ChildSubmissionOutcome, Error, EventTrigger, ProcedureContext, ProcedureEvent, ProcedureId,
        ProcedureState, RetryPhase,
    };

    #[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()],
                },
                Row {
                    values: vec![Value { value_data: None }],
                },
            ])
        }

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

    #[test]
    fn procedure_event_extra_rows_preserve_envelope_values_and_internal_nulls() {
        let procedure_id = ProcedureId::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
        let procedure_event = ProcedureEvent::new(
            procedure_id,
            Box::new(TestEvent {}),
            ProcedureState::Running,
            EventTrigger::Submitted,
        );

        let procedure_event_extra_rows = procedure_event.extra_rows().unwrap();
        assert_eq!(
            procedure_event_extra_rows,
            vec![
                Row {
                    values: vec![
                        ValueData::StringValue(procedure_id.to_string()).into(),
                        ValueData::StringValue("Running".to_string()).into(),
                        ValueData::StringValue(String::new()).into(),
                        jsonb_value(&json!({"type": "Submitted"})),
                        ValueData::StringValue("test_event1".to_string()).into(),
                        Value { value_data: None },
                        Value { value_data: None },
                    ],
                },
                Row {
                    values: vec![
                        ValueData::StringValue(procedure_id.to_string()).into(),
                        ValueData::StringValue("Running".to_string()).into(),
                        ValueData::StringValue(String::new()).into(),
                        jsonb_value(&json!({"type": "Submitted"})),
                        ValueData::StringValue("test_event2".to_string()).into(),
                        Value { value_data: None },
                        Value { value_data: None },
                    ],
                },
                Row {
                    values: vec![
                        ValueData::StringValue(procedure_id.to_string()).into(),
                        ValueData::StringValue("Running".to_string()).into(),
                        ValueData::StringValue(String::new()).into(),
                        jsonb_value(&json!({"type": "Submitted"})),
                        Value { value_data: None },
                        Value { value_data: None },
                        Value { value_data: None },
                    ],
                },
            ]
        );
    }

    #[test]
    fn procedure_event_extra_rows_include_error_for_failed_state() {
        let error = Arc::new(Error::external(MockError::new(StatusCode::Unexpected)));
        let procedure_event = ProcedureEvent::new(
            ProcedureId::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
            Box::new(TestEvent {}),
            ProcedureState::failed(error.clone()),
            EventTrigger::Failed,
        );

        let procedure_event_extra_rows = procedure_event.extra_rows().unwrap();

        assert_eq!(procedure_event_extra_rows.len(), 3);
        assert_eq!(
            procedure_event_extra_rows[0].values[1],
            ValueData::StringValue("Failed".to_string()).into()
        );
        assert_eq!(
            procedure_event_extra_rows[0].values[2],
            ValueData::StringValue(format!("{error:?}")).into()
        );
        assert_eq!(
            procedure_event_extra_rows[0].values[3],
            jsonb_value(&json!({"type": "Failed"}))
        );
    }

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

        assert_eq!(
            procedure_event.extra_schema(),
            vec![
                ColumnSchema {
                    column_name: "procedure_id".to_string(),
                    datatype: ColumnDataType::String.into(),
                    semantic_type: SemanticType::Field.into(),
                    ..Default::default()
                },
                ColumnSchema {
                    column_name: "procedure_state".to_string(),
                    datatype: ColumnDataType::String.into(),
                    semantic_type: SemanticType::Field.into(),
                    ..Default::default()
                },
                ColumnSchema {
                    column_name: "procedure_error".to_string(),
                    datatype: ColumnDataType::String.into(),
                    semantic_type: SemanticType::Field.into(),
                    ..Default::default()
                },
                PROCEDURE_TRIGGER_COLUMN.column_schema(),
                ColumnSchema {
                    column_name: "test_event_column".to_string(),
                    datatype: ColumnDataType::String.into(),
                    semantic_type: SemanticType::Field.into(),
                    ..Default::default()
                },
                ACTOR_COLUMN.column_schema(),
                EVENT_CONTEXT_COLUMN.column_schema(),
            ]
        );
    }

    #[test]
    fn procedure_event_records_actor_for_all_lifecycle_triggers() {
        let mut context = ProcedureContext::from_event_context(PersistentEventContext::new(
            TriggerReason::AutoRepartition,
        ));
        context.actor = Some("alice".to_string());
        let submitted = ProcedureEvent::new_with_context(
            ProcedureId::random(),
            Box::new(TestEvent {}),
            ProcedureState::Running,
            EventTrigger::Submitted,
            context.clone(),
        );

        assert_eq!(
            submitted.extra_rows().unwrap()[0].values[5],
            ValueData::StringValue("alice".to_string()).into()
        );
        assert_eq!(
            submitted.extra_rows().unwrap()[0].values[6],
            jsonb_value(&json!({"reason": "auto_repartition"}))
        );

        let child_id = ProcedureId::random();
        for trigger in [
            EventTrigger::Recovered,
            EventTrigger::ChildSubmitted {
                procedure_id: child_id,
                outcome: ChildSubmissionOutcome::Accepted,
            },
            EventTrigger::Retrying {
                phase: RetryPhase::Execute,
                attempt: 1,
            },
            EventTrigger::RollingBack,
            EventTrigger::Succeeded,
            EventTrigger::Failed,
            EventTrigger::Poisoned,
        ] {
            let event = ProcedureEvent::new_with_context(
                ProcedureId::random(),
                Box::new(TestEvent {}),
                ProcedureState::Running,
                trigger,
                context.clone(),
            );
            let values = &event.extra_rows().unwrap()[0].values;
            assert_eq!(
                values[5],
                ValueData::StringValue("alice".to_string()).into()
            );
            assert_eq!(values[6], Value { value_data: None });
        }
    }

    #[test]
    fn test_event_trigger_serialization() {
        let procedure_id = ProcedureId::parse_str("00000000-0000-0000-0000-000000000001").unwrap();

        for (trigger, name) in [
            (EventTrigger::Submitted, "Submitted"),
            (EventTrigger::Recovered, "Recovered"),
            (EventTrigger::RollingBack, "RollingBack"),
            (EventTrigger::Succeeded, "Succeeded"),
            (EventTrigger::Failed, "Failed"),
            (EventTrigger::Poisoned, "Poisoned"),
        ] {
            assert_eq!(
                serde_json::to_value(trigger).unwrap(),
                json!({"type": name})
            );
        }
        assert_eq!(
            serde_json::to_value(EventTrigger::ChildSubmitted {
                procedure_id,
                outcome: ChildSubmissionOutcome::Accepted,
            })
            .unwrap(),
            json!({
                "type": "ChildSubmitted",
                "procedure_id": "00000000-0000-0000-0000-000000000001",
                "outcome": "Accepted",
            })
        );
        assert_eq!(
            serde_json::to_value(EventTrigger::Retrying {
                phase: RetryPhase::Execute,
                attempt: 2,
            })
            .unwrap(),
            json!({"type": "Retrying", "phase": "Execute", "attempt": 2})
        );
    }
}
