//
// Copyright Amazon.com Inc. or its affiliates.
// All Rights Reserved.
//
// SPDX-License-Identifier: Apache-2.0
//

import Foundation
import XCTest

@testable import Amplify
@testable import AmplifyTestCommon
@testable import AWSPluginsCore
@testable import AWSDataStoreCategoryPlugin

class AWSIncomingEventReconciliationQueueTests: XCTestCase {
    var storageAdapter: MockSQLiteStorageEngineAdapter!
    var apiPlugin: MockAPICategoryPlugin!

    override func setUp() {
        MockModelReconciliationQueue.reset()
        storageAdapter = MockSQLiteStorageEngineAdapter()
        storageAdapter.returnOnQuery(dataStoreResult: .none)
        storageAdapter.returnOnSave(dataStoreResult: .none)

        apiPlugin = MockAPICategoryPlugin()

        operationQueue = OperationQueue()
        operationQueue.name = "com.amazonaws.DataStore.UnitTestQueue"
        operationQueue.maxConcurrentOperationCount = 2
        operationQueue.underlyingQueue = DispatchQueue.global()
        operationQueue.isSuspended = true

    }
    var operationQueue: OperationQueue!

    //This test case attempts to hit a race condition, and may be required to execute multiple times
    // in order to demonstrate the bug
    func testTwoConnectionStatusUpdatesAtSameTime() {
        let expectInitialized = expectation(description: "eventQueue expected to send out initialized state")

        let modelReconciliationQueueFactory
            = MockModelReconciliationQueue.init(modelSchema:storageAdapter:api:modelPredicate:auth:incomingSubscriptionEvents:)
        let eventQueue = AWSIncomingEventReconciliationQueue(
            modelSchemas: [Post.schema, Comment.schema],
            api: apiPlugin,
            storageAdapter: storageAdapter,
            syncExpressions: [],
            modelReconciliationQueueFactory: modelReconciliationQueueFactory)
        eventQueue.start()

        // We need to keep this in scope for the duration of the test or else Combine will release the listeners.
        let sink = eventQueue.publisher.sink(receiveCompletion: { _ in
            XCTFail("Not expecting this to call")
        }, receiveValue: { event  in
            switch event {
            case .initialized:
                expectInitialized.fulfill()
            default:
                XCTFail("Should not expect any other state, received: \(event)")
            }
        })

        let reconciliationQueues = MockModelReconciliationQueue.mockModelReconciliationQueues
        for (queueName, queue) in reconciliationQueues {
            let cancellableOperation = CancelAwareBlockOperation {
                queue.modelReconciliationQueueSubject.send(.connected(modelName: queueName))
            }
            operationQueue.addOperation(cancellableOperation)
        }
        operationQueue.isSuspended = false
        waitForExpectations(timeout: 2)

        // Take action on the sink to prevent compiler warnings about unused variables.
        sink.cancel()
    }

    func testSubscriptionFailedWithSingleModelUnauthorizedError() {
        let expectInitialized = expectation(description: "eventQueue expected to send out initialized state")
        let modelReconciliationQueueFactory
            = MockModelReconciliationQueue.init(modelSchema:storageAdapter:api:modelPredicate:auth:incomingSubscriptionEvents:)
        let eventQueue = AWSIncomingEventReconciliationQueue(
            modelSchemas: [Post.schema],
            api: apiPlugin,
            storageAdapter: storageAdapter,
            syncExpressions: [],
            modelReconciliationQueueFactory: modelReconciliationQueueFactory)
        eventQueue.start()

        let sink = eventQueue.publisher.sink(receiveCompletion: { _ in
            XCTFail("Not expecting this to call")
        }, receiveValue: { event  in
            switch event {
            case .initialized:
                expectInitialized.fulfill()
            default:
                XCTFail("Should not expect any other state, received: \(event)")
            }
        })

        let reconciliationQueues = MockModelReconciliationQueue.mockModelReconciliationQueues
        for (queueName, queue) in reconciliationQueues {
            let cancellableOperation = CancelAwareBlockOperation {
                queue.modelReconciliationQueueSubject.send(.disconnected(modelName: queueName, reason: .unauthorized))
            }
            operationQueue.addOperation(cancellableOperation)
        }
        operationQueue.isSuspended = false
        waitForExpectations(timeout: 2)

        sink.cancel()
    }

    // This test case tests that initialized event is received even if only one
    // model subscriptions out of two failed - Post subscription will fail but Comment will succeed
    func testSubscriptionFailedWithMultipleModels() {
        let expectInitialized = expectation(description: "eventQueue expected to send out initialized state")
        let modelReconciliationQueueFactory
            = MockModelReconciliationQueue.init(modelSchema:storageAdapter:api:modelPredicate:auth:incomingSubscriptionEvents:)
        let eventQueue = AWSIncomingEventReconciliationQueue(
            modelSchemas: [Post.schema, Comment.schema],
            api: apiPlugin,
            storageAdapter: storageAdapter,
            syncExpressions: [],
            modelReconciliationQueueFactory: modelReconciliationQueueFactory)
        eventQueue.start()

        let sink = eventQueue.publisher.sink(receiveCompletion: { _ in
            XCTFail("Not expecting this to call")
        }, receiveValue: { event  in
            switch event {
            case .initialized:
                expectInitialized.fulfill()
            default:
                XCTFail("Should not expect any other state, received: \(event)")
            }
        })

        let reconciliationQueues = MockModelReconciliationQueue.mockModelReconciliationQueues
        for (queueName, queue) in reconciliationQueues {
            let cancellableOperation = CancelAwareBlockOperation {
                let event: ModelReconciliationQueueEvent = queueName == Post.modelName ?
                    .disconnected(modelName: queueName, reason: .unauthorized) :
                    .connected(modelName: queueName)
                queue.modelReconciliationQueueSubject.send(event)
            }
            operationQueue.addOperation(cancellableOperation)
        }
        operationQueue.isSuspended = false
        waitForExpectations(timeout: 2)

        sink.cancel()
    }
}
