package net.corda.node.services.statemachine

import net.corda.core.CordaRuntimeException
import net.corda.core.messaging.startFlow
import net.corda.core.utilities.getOrThrow
import net.corda.core.utilities.seconds
import net.corda.testing.core.ALICE_NAME
import net.corda.testing.core.CHARLIE_NAME
import net.corda.testing.core.singleIdentity
import net.corda.testing.driver.internal.OutOfProcessImpl
import org.junit.Test
import java.sql.Connection
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import kotlin.test.assertEquals
import kotlin.test.assertFailsWith
import kotlin.test.assertTrue

@Suppress("MaxLineLength") // Byteman rules cannot be easily wrapped
class StateMachineFlowInitErrorHandlingTest : StateMachineErrorHandlingTest() {

    private companion object {
        val executor: ExecutorService = Executors.newSingleThreadExecutor()
    }

    /**
     * Throws an exception when performing an [Action.CommitTransaction] event before the flow has initialised and saved its first checkpoint
     * (remains in an unstarted state).
     *
     * The exception is thrown 3 times.
     *
     * This causes the transition to be discharged from the hospital 3 times (retries 3 times). On the final retry the transition
     * succeeds and the flow finishes.
     *
     * Each time the flow retries, it starts from the beginning of the flow (due to being in an unstarted state).
     *
     */
    @Test(timeout = 300_000)
    fun `error during transition with CommitTransaction action that occurs during flow initialisation will retry and complete successfully`() {
        startDriver {
            val charlie = createNode(CHARLIE_NAME)
            val (alice, port) = createBytemanNode(ALICE_NAME)

            val rules = """
                RULE Create Counter
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF createCounter("counter", $counter)
                DO traceln("Counter created")
                ENDRULE

                RULE Throw exception on executeCommitTransaction action
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF readCounter("counter") < 3
                DO incrementCounter("counter"); traceln("Throwing exception"); throw new java.sql.SQLException("die dammit die", "1")
                ENDRULE
            """.trimIndent()

            submitBytemanRules(rules, port)

            alice.rpc.startFlow(
                StateMachineErrorHandlingTest::SendAMessageFlow,
                charlie.nodeInfo.singleIdentity()
            ).returnValue.getOrThrow(
                30.seconds
            )

            alice.rpc.assertHospitalCounts(discharged = 3)
            assertEquals(0, alice.rpc.stateMachinesSnapshot().size)
            alice.rpc.assertNumberOfCheckpoints(0)
        }
    }

    /**
     * Throws an exception when calling [FlowStateMachineImpl.processEvent].
     *
     * This is not an expected place for an exception to occur, but allows us to test what happens when a random exception is propagated
     * up to [FlowStateMachineImpl.run] during flow initialisation.
     *
     * A "Transaction context is missing" exception is thrown due to where the exception is thrown (no transaction is created so this is
     * thrown when leaving [FlowStateMachineImpl.processEventsUntilFlowIsResumed] due to the finally block).
     */
    @Test(timeout = 300_000)
    fun `unexpected error during flow initialisation throws exception to client`() {
        startDriver {
            val charlie = createNode(CHARLIE_NAME)
            val (alice, port) = createBytemanNode(ALICE_NAME)
            val rules = """
                RULE Create Counter
                CLASS ${FlowStateMachineImpl::class.java.name}
                METHOD processEvent
                AT ENTRY
                IF createCounter("counter", $counter)
                DO traceln("Counter created")
                ENDRULE
                
                RULE Throw exception
                CLASS ${FlowStateMachineImpl::class.java.name}
                METHOD processEvent
                AT ENTRY
                IF readCounter("counter") < 1
                DO incrementCounter("counter"); traceln("Throwing exception"); throw new java.lang.RuntimeException("die dammit die")
                ENDRULE
            """.trimIndent()

            submitBytemanRules(rules, port)

            assertFailsWith<CordaRuntimeException> {
                alice.rpc.startFlow(
                    StateMachineErrorHandlingTest::SendAMessageFlow,
                    charlie.nodeInfo.singleIdentity()
                ).returnValue.getOrThrow(30.seconds)
            }

            alice.rpc.assertHospitalCounts(propagated = 1)
            assertEquals(0, alice.rpc.stateMachinesSnapshot().size)
            alice.rpc.assertNumberOfCheckpoints(0)
        }
    }

    /**
     * Throws an exception when performing an [Action.CommitTransaction] event before the flow has initialised and saved its first checkpoint
     * (remains in an unstarted state).
     *
     * A [SQLException] is then thrown when trying to rollback the flow's database transaction.
     *
     * The [SQLException] should be suppressed and the flow should continue to retry and complete successfully.
     */
    @Test(timeout = 300_000)
    fun `error during initialisation when trying to rollback the flow's database transaction the flow is able to retry and complete successfully`() {
        startDriver {
            val charlie = createNode(CHARLIE_NAME)
            val (alice, port) = createBytemanNode(ALICE_NAME)

            val rules = """
                RULE Create Counter
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF createCounter("counter", $counter)
                DO traceln("Counter created")
                ENDRULE

                RULE Throw exception on executeCommitTransaction action
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF readCounter("counter") == 0
                DO incrementCounter("counter"); traceln("Throwing exception"); throw new java.lang.RuntimeException("die dammit die")
                ENDRULE
                
                RULE Throw exception when rolling back transaction in transition executor
                INTERFACE ${Connection::class.java.name}
                METHOD rollback
                AT ENTRY
                IF readCounter("counter") == 1
                DO incrementCounter("counter"); traceln("Throwing exception in transition executor"); throw new java.sql.SQLException("could not reach db", "1")
                ENDRULE
            """.trimIndent()

            submitBytemanRules(rules, port)

            alice.rpc.startFlow(
                StateMachineErrorHandlingTest::SendAMessageFlow,
                charlie.nodeInfo.singleIdentity()
            ).returnValue.getOrThrow(30.seconds)

            alice.rpc.assertHospitalCounts(discharged = 1)
            assertEquals(0, alice.rpc.stateMachinesSnapshot().size)
            alice.rpc.assertNumberOfCheckpoints(0)
        }
    }

    /**
     * Throws an exception when performing an [Action.CommitTransaction] event before the flow has initialised and saved its first checkpoint
     * (remains in an unstarted state).
     *
     * A [SQLException] is then thrown when trying to close the flow's database transaction.
     *
     * The [SQLException] should be suppressed and the flow should continue to retry and complete successfully.
     */
    @Test(timeout = 300_000)
    fun `error during initialisation when trying to close the flow's database transaction the flow is able to retry and complete successfully`() {
        startDriver {
            val charlie = createNode(CHARLIE_NAME)
            val (alice, port) = createBytemanNode(ALICE_NAME)

            val rules = """
                RULE Create Counter
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF createCounter("counter", $counter)
                DO traceln("Counter created")
                ENDRULE

                RULE Throw exception on executeCommitTransaction action
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF readCounter("counter") == 0
                DO incrementCounter("counter"); traceln("Throwing exception"); throw new java.lang.RuntimeException("die dammit die")
                ENDRULE
                
                RULE Throw exception when rolling back transaction in transition executor
                INTERFACE ${Connection::class.java.name}
                METHOD close
                AT ENTRY
                IF readCounter("counter") == 1
                DO incrementCounter("counter"); traceln("Throwing exception in transition executor"); throw new java.sql.SQLException("could not reach db", "1")
                ENDRULE
            """.trimIndent()

            submitBytemanRules(rules, port)

            alice.rpc.startFlow(
                StateMachineErrorHandlingTest::SendAMessageFlow,
                charlie.nodeInfo.singleIdentity()
            ).returnValue.getOrThrow(30.seconds)

            alice.rpc.assertHospitalCounts(discharged = 1)
            assertEquals(0, alice.rpc.stateMachinesSnapshot().size)
            alice.rpc.assertNumberOfCheckpoints(0)
        }
    }

    /**
     * Throws an exception when performing an [Action.CommitTransaction] event before the flow has initialised and saved its first checkpoint
     * (remains in an unstarted state).
     *
     * The exception is thrown 4 times.
     *
     * This causes the transition to be discharged from the hospital 3 times (retries 3 times) and then be kept in for observation.
     *
     * Each time the flow retries, it starts from the beginning of the flow (due to being in an unstarted state).
     */
    @Test(timeout = 300_000)
    fun `error during transition with CommitTransaction action that occurs during flow initialisation will retry and be kept for observation if error persists`() {
        startDriver {
            val charlie = createNode(CHARLIE_NAME)
            val (alice, port) = createBytemanNode(ALICE_NAME)

            val rules = """
                RULE Create Counter
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF createCounter("counter", $counter)
                DO traceln("Counter created")
                ENDRULE

                RULE Throw exception on executeCommitTransaction action
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF readCounter("counter") < 4
                DO incrementCounter("counter"); traceln("Throwing exception"); throw new java.sql.SQLException("die dammit die", "1")
                ENDRULE
            """.trimIndent()

            submitBytemanRules(rules, port)

            executor.execute {
                alice.rpc.startFlow(StateMachineErrorHandlingTest::SendAMessageFlow, charlie.nodeInfo.singleIdentity())
            }

            // flow is not signaled as started calls to [getOrThrow] will hang, sleeping instead
            Thread.sleep(30.seconds.toMillis())

            alice.rpc.assertHospitalCounts(
                discharged = 3,
                observation = 1
            )
            assertEquals(1, alice.rpc.stateMachinesSnapshot().size)
            alice.rpc.assertNumberOfCheckpoints(0)
            val terminated = (alice as OutOfProcessImpl).stop(60.seconds)
            assertTrue(terminated, "The node must be shutdown before it can be restarted")
            val (alice2, _) = createBytemanNode(ALICE_NAME)
            Thread.sleep(10.seconds.toMillis())
            alice2.rpc.assertNumberOfCheckpoints(0)
        }
    }

    /**
     * Throws an exception when performing an [Action.CommitTransaction] event before the flow has initialised and saved its first checkpoint
     * (remains in an unstarted state).
     *
     * An exception is thrown when committing a database transaction during a transition to trigger the retry of the flow. Another
     * exception is then thrown during the retry itself.
     *
     * The flow then retries the retry causing the flow to complete successfully.
     */
    @Test(timeout = 300_000)
    fun `error during retrying a flow that failed when committing its original checkpoint will retry the flow again and complete successfully`() {
        startDriver {
            val charlie = createNode(CHARLIE_NAME)
            val (alice, port) = createBytemanNode(ALICE_NAME)

            val rules = """
                RULE Throw exception on executeCommitTransaction action after first suspend + commit
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF !flagged("commit_exception_flag")
                DO flag("commit_exception_flag"); traceln("Throwing exception"); throw new java.sql.SQLException("die dammit die", "1")
                ENDRULE
                
                RULE Throw exception on retry
                CLASS ${SingleThreadedStateMachineManager::class.java.name}
                METHOD onExternalStartFlow
                AT ENTRY
                IF flagged("commit_exception_flag") && !flagged("retry_exception_flag")
                DO flag("retry_exception_flag"); traceln("Throwing retry exception"); throw new java.lang.RuntimeException("Here we go again")
                ENDRULE
            """.trimIndent()

            submitBytemanRules(rules, port)

            alice.rpc.startFlow(
                StateMachineErrorHandlingTest::SendAMessageFlow,
                charlie.nodeInfo.singleIdentity()
            ).returnValue.getOrThrow(
                30.seconds
            )

            alice.rpc.assertHospitalCounts(
                discharged = 1,
                dischargedRetry = 1
            )
            assertEquals(0, alice.rpc.stateMachinesSnapshot().size)
            alice.rpc.assertNumberOfCheckpoints(0)
        }
    }

    /**
     * Throws an exception when performing an [Action.CommitTransaction] event on a responding node before the flow has initialised and
     * saved its first checkpoint (remains in an unstarted state).
     *
     * The exception is thrown 3 times.
     *
     * This causes the transition to be discharged from the hospital 3 times (retries 3 times). On the final retry the transition
     * succeeds and the flow finishes.
     *
     * Each time the flow retries, it starts from the beginning of the flow (due to being in an unstarted state).
     */
    @Test(timeout = 300_000)
    fun `responding flow - error during transition with CommitTransaction action that occurs during flow initialisation will retry and complete successfully`() {
        startDriver {
            val (charlie, port) = createBytemanNode(CHARLIE_NAME)
            val alice = createNode(ALICE_NAME)

            val rules = """
                RULE Create Counter
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF createCounter("counter", $counter)
                DO traceln("Counter created")
                ENDRULE

                RULE Throw exception on executeCommitTransaction action
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF readCounter("counter") < 3
                DO incrementCounter("counter"); traceln("Throwing exception"); throw new java.sql.SQLException("die dammit die", "1")
                ENDRULE
            """.trimIndent()

            submitBytemanRules(rules, port)

            alice.rpc.startFlow(
                StateMachineErrorHandlingTest::SendAMessageFlow,
                charlie.nodeInfo.singleIdentity()
            ).returnValue.getOrThrow(
                30.seconds
            )

            charlie.rpc.assertHospitalCounts(discharged = 3)
            alice.rpc.assertNumberOfCheckpoints(0)
            charlie.rpc.assertNumberOfCheckpoints(0)
        }
    }

    /**
     * Throws an exception when performing an [Action.CommitTransaction] event on a responding node before the flow has initialised and
     * saved its first checkpoint (remains in an unstarted state).
     *
     * The exception is thrown 4 times.
     *
     * This causes the transition to be discharged from the hospital 3 times (retries 3 times) and then be kept in for observation.
     *
     * Each time the flow retries, it starts from the beginning of the flow (due to being in an unstarted state).
     */
    @Test(timeout = 300_000)
    fun `responding flow - error during transition with CommitTransaction action that occurs during flow initialisation will retry and be kept for observation if error persists`() {
        startDriver {
            val (charlie, port) = createBytemanNode(CHARLIE_NAME)
            val alice = createNode(ALICE_NAME)

            val rules = """
                RULE Create Counter
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF createCounter("counter", $counter)
                DO traceln("Counter created")
                ENDRULE

                RULE Throw exception on executeCommitTransaction action
                CLASS ${ActionExecutorImpl::class.java.name}
                METHOD executeCommitTransaction
                AT ENTRY
                IF readCounter("counter") < 4
                DO incrementCounter("counter"); traceln("Throwing exception"); throw new java.sql.SQLException("die dammit die", "1")
                ENDRULE
            """.trimIndent()

            submitBytemanRules(rules, port)

            executor.execute {
                alice.rpc.startFlow(StateMachineErrorHandlingTest::SendAMessageFlow, charlie.nodeInfo.singleIdentity())
            }

            // flow is not signaled as started calls to [getOrThrow] will hang, sleeping instead
            Thread.sleep(30.seconds.toMillis())

            charlie.rpc.assertHospitalCounts(
                discharged = 3,
                observation = 1
            )
            assertEquals(1, alice.rpc.stateMachinesSnapshot().size)
            assertEquals(1, charlie.rpc.stateMachinesSnapshot().size)
            alice.rpc.assertNumberOfCheckpoints(1)
            charlie.rpc.assertNumberOfCheckpoints(0)
            val terminated = (charlie as OutOfProcessImpl).stop(60.seconds)
            assertTrue(terminated, "The node must be shutdown before it can be restarted")
            val (charlie2, _) = createBytemanNode(CHARLIE_NAME)
            Thread.sleep(10.seconds.toMillis())
            alice.rpc.assertNumberOfCheckpoints(0)
            charlie2.rpc.assertNumberOfCheckpoints(0)
        }
    }
}