﻿/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.
 */

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Transactions;
using Apache.NMS.ActiveMQ.Commands;
using Apache.NMS.ActiveMQ.Transactions;
using Apache.NMS.ActiveMQ.Util.Synchronization;
using Apache.NMS.Util;

namespace Apache.NMS.ActiveMQ
{
    public sealed class NetTxTransactionContext : TransactionContext, ISinglePhaseNotification
    {
        private const int XA_OK = 0;
        private const int XA_READONLY = 3;

        private Enlistment currentEnlistment;

        private static readonly NmsSynchronizationMonitor recoveredResourceManagerIdsLock =
            new NmsSynchronizationMonitor();
        private static readonly Dictionary<string, bool> recoveredResourceManagerIds = new Dictionary<string, bool>();

        public NetTxTransactionContext(Session session) : base(session)
        {
        }

        /// <summary>
        /// DTC recovery is performed once for each AppDomain per default. In case you want to perform
        /// it again during execution of the application you can call this method before.
        /// But ensure in this case that no connection is active anymore.
        /// </summary>
        public static void ResetDtcRecovery()
        {
            recoveredResourceManagerIds.Clear();
        }

        public override bool InLocalTransaction
        {
            get { return this.TransactionId != null && this.currentEnlistment == null; }
        }

        public override void Begin()
        {
            throw new IllegalStateException("Local Transactions not supported in NetTx resources");
        }

        public override void Commit()
        {
            throw new IllegalStateException("Local Transactions not supported in NetTx resources");
        }

        public override void Rollback()
        {
            throw new IllegalStateException("Local Transactions not supported in NetTx resources");
        }

        #region Transaction Members used when dealing with .NET System Transactions.

        // When DTC calls prepare we must then wait for either the TX to commit, rollback or
        // be canceled because its in doubt.
        private readonly ManualResetEvent dtcControlEvent = new ManualResetEvent(true);

        // Once the DTC calls prepare we lock this object and don't unlock it again until
        // the TX has either completed or terminated, the users of this class should use
        // this sync point when the TX is a DTC version as opposed to a local one.
        private readonly NmsSynchronizationMonitor syncObject = new NmsSynchronizationMonitor(); //Mutex();

        public enum TxState
        {
            None = 0, Active = 1, Pending = 2
        }

        private TxState netTxState = TxState.None;

        public NmsSynchronizationMonitor SyncRoot
        {
            get { return this.syncObject; }
        }

        public bool InNetTransaction
        {
            get { return this.TransactionId != null && this.TransactionId is XATransactionId; }
        }

        public TxState NetTxState
        {
            get
            {
                return this.netTxState;
            }
        }

        public WaitHandle DtcWaitHandle
        {
            get { return dtcControlEvent; }
        }

        public void Begin(Transaction transaction)
        {
            using(syncObject.Lock())
            {
                dtcControlEvent.Reset();

                Tracer.Debug("Begin notification received");

                if (InNetTransaction)
                {
                    throw new TransactionInProgressException("A Transaction is already in Progress");
                }

                try
                {
                    Guid rmId = ResourceManagerGuid;
                     
                    // Enlist this object in the transaction.
                    this.currentEnlistment = transaction.EnlistDurable(rmId, this, EnlistmentOptions.None);

                    // In case of a exception in the current method the transaction will be rolled back.
                    // Until Begin Transaction is completed we consider to be in a rollback scenario.
                    this.netTxState = TxState.Pending;

                    Tracer.Debug("Enlisted in Durable Transaction with RM Id: " + rmId);

                    TransactionInformation txInfo = transaction.TransactionInformation;

                    XATransactionId xaId = new XATransactionId();
                    this.TransactionId = xaId;

                    if (txInfo.DistributedIdentifier != Guid.Empty)
                    {
                        xaId.GlobalTransactionId = txInfo.DistributedIdentifier.ToByteArray();
                        xaId.BranchQualifier = Encoding.UTF8.GetBytes(Guid.NewGuid().ToString());
                    }
                    else
                    {
                        xaId.GlobalTransactionId = Encoding.UTF8.GetBytes(txInfo.LocalIdentifier);
                        xaId.BranchQualifier = Encoding.UTF8.GetBytes(Guid.NewGuid().ToString());
                    }

                    // Now notify the broker that a new XA'ish transaction has started.
                    TransactionInfo info = new TransactionInfo();
                    info.ConnectionId = this.connection.ConnectionId;
                    info.TransactionId = this.TransactionId;
                    info.Type = (int)TransactionType.Begin;

                    this.session.Connection.Oneway(info);

                    // Begin Transaction is completed successfully. Change to transaction active state now.
                    this.netTxState = TxState.Active;

                    SignalTransactionStarted();

                    if (Tracer.IsDebugEnabled)
                    {
                        Tracer.Debug("Began XA'ish Transaction:" + xaId);
                    }
                }
                catch (Exception)
                {
                    // When in pending state the rollback will signal that a new transaction can be started. Otherwise do it here.
                    if (netTxState != TxState.Pending)
                    {
                        netTxState = TxState.None;
                        dtcControlEvent.Set();
                    }
                    throw;
                }
            }
        }

        public void Prepare(PreparingEnlistment preparingEnlistment)
        {
            PrepareAsync(preparingEnlistment).GetAsyncResult();
        }

        public async Task PrepareAsync(PreparingEnlistment preparingEnlistment)
        {
            using (await this.syncObject.LockAsync().Await())
            {
                this.netTxState = TxState.Pending;

                try
                {
                    Tracer.Debug("Prepare notification received for TX id: " + this.TransactionId);

                    await BeforeEndAsync().Await();

                    // Before sending the request to the broker, log the recovery bits, if
                    // this fails we can't prepare and the TX should be rolled back.
                    RecoveryLogger.LogRecoveryInfo(this.TransactionId as XATransactionId,
                                                   preparingEnlistment.RecoveryInformation());

                    // Inform the broker that work on the XA'sh TX Branch is complete.
                    TransactionInfo info = new TransactionInfo();
                    info.ConnectionId = this.connection.ConnectionId;
                    info.TransactionId = this.TransactionId;
                    info.Type = (int)TransactionType.End;

                    await this.connection.CheckConnectedAsync().Await();
                    await this.connection.SyncRequestAsync((TransactionInfo) info.Clone()).Await();

                    // Prepare the Transaction for commit.
                    info.Type = (int)TransactionType.Prepare;
                    IntegerResponse response = (IntegerResponse) await this.connection.SyncRequestAsync(info).Await();
                    if (response.Result == XA_READONLY)
                    {
                        Tracer.Debug("Transaction Prepare done and doesn't need a commit, TX id: " + this.TransactionId);

                        this.TransactionId = null;
                        this.currentEnlistment = null;

                        // Read Only means there's nothing to recover because there was no
                        // change on the broker.
                        RecoveryLogger.LogRecovered(this.TransactionId as XATransactionId);

                        // if server responds that nothing needs to be done, then reply done.
                        // otherwise the DTC will call Commit or Rollback but another transaction
                        // can already be in progress and this one would be commited or rolled back
                        // immediately.
                        preparingEnlistment.Done();

                        // Done so commit won't be called.
                        AfterCommit();

                        // A Read-Only TX is considered closed at this point, DTC won't call us again.
                        this.dtcControlEvent.Set();
                    }
                    else
                    {
                        Tracer.Debug("Transaction Prepare succeeded TX id: " + this.TransactionId);

                        // If work finished correctly, reply prepared
                        preparingEnlistment.Prepared();
                    }
                }
                catch (Exception ex)
                {
                    Tracer.DebugFormat("Transaction[{0}] Prepare failed with error: {1}",
                                       this.TransactionId, ex.Message);

                    AfterRollback();
                    preparingEnlistment.ForceRollback();
                    try
                    {
                        this.connection.OnAsyncException(ex);
                    }
                    catch (Exception error)
                    {
                        Tracer.Error(error.ToString());
                    }

                    this.currentEnlistment = null;
                    this.TransactionId = null;
                    this.netTxState = TxState.None;
                    this.dtcControlEvent.Set();
                }
            }
        }

        public void Commit(Enlistment enlistment)
        {
            CommitAsync(enlistment).GetAsyncResult();
        }

        public async Task CommitAsync(Enlistment enlistment)
        {
            using(await this.syncObject.LockAsync().Await())
            {
                try
                {
                    Tracer.Debug("Commit notification received for TX id: " + this.TransactionId);

                    if (this.TransactionId != null)
                    {
                        // Now notify the broker that a new XA'ish transaction has completed.
                        TransactionInfo info = new TransactionInfo();
                        info.ConnectionId = this.connection.ConnectionId;
                        info.TransactionId = this.TransactionId;
                        info.Type = (int)TransactionType.CommitTwoPhase;

                        await this.connection.CheckConnectedAsync().Await();
                        await this.connection.SyncRequestAsync(info).Await();

                        Tracer.Debug("Transaction Commit Done TX id: " + this.TransactionId);

                        RecoveryLogger.LogRecovered(this.TransactionId as XATransactionId);

                        // if server responds that nothing needs to be done, then reply done.
                        enlistment.Done();

                        AfterCommit();
                    }
                }
                catch (Exception ex)
                {
                    Tracer.DebugFormat("Transaction[{0}] Commit failed with error: {1}",
                                       this.TransactionId, ex.Message);
                    try
                    {
                        this.connection.OnAsyncException(ex);
                    }
                    catch (Exception error)
                    {
                        Tracer.Error(error.ToString());
                    }
                }
                finally
                {
                    this.currentEnlistment = null;
                    this.TransactionId = null;
                    this.netTxState = TxState.None;

                    CountDownLatch latch = this.recoveryComplete;
                    if (latch != null)
                    {
                        latch.countDown();
                    }

                    this.dtcControlEvent.Set();
                }
            }
        }


        public void SinglePhaseCommit(SinglePhaseEnlistment enlistment)
        {
            SinglePhaseCommitAsync(enlistment).GetAsyncResult();
        }
        public async Task SinglePhaseCommitAsync(SinglePhaseEnlistment enlistment)
        {
            using(await this.syncObject.LockAsync().Await())
            {
                try
                {
                    Tracer.Debug("Single Phase Commit notification received for TX id: " + this.TransactionId);

                    if (this.TransactionId != null)
                    {
                        await BeforeEndAsync().Await();

                        // Now notify the broker that a new XA'ish transaction has completed.
                        TransactionInfo info = new TransactionInfo();
                        info.ConnectionId = this.connection.ConnectionId;
                        info.TransactionId = this.TransactionId;
                        info.Type = (int)TransactionType.CommitOnePhase;

                        await this.connection.CheckConnectedAsync().Await();
                        await this.connection.SyncRequestAsync(info).Await();

                        Tracer.Debug("Transaction Single Phase Commit Done TX id: " + this.TransactionId);

                        // if server responds that nothing needs to be done, then reply done.
                        enlistment.Done();

                        AfterCommit();
                    }
                }
                catch (Exception ex)
                {
                    Tracer.DebugFormat("Transaction[{0}] Single Phase Commit failed with error: {1}",
                                       this.TransactionId, ex.Message);
                    AfterRollback();
                    enlistment.Done();
                    try
                    {
                        this.connection.OnAsyncException(ex);
                    }
                    catch (Exception error)
                    {
                        Tracer.Error(error.ToString());
                    }
                }
                finally
                {
                    this.currentEnlistment = null;
                    this.TransactionId = null;
                    this.netTxState = TxState.None;

                    this.dtcControlEvent.Set();
                }
            }
        }

        public void Rollback(Enlistment enlistment)
        {
            RollbackAsync(enlistment).GetAsyncResult();
        }
        
        public async Task RollbackAsync(Enlistment enlistment)
        {
            using(await this.syncObject.LockAsync().Await())
            {
                try
                {
                    Tracer.Debug("Rollback notification received for TX id: " + this.TransactionId);

                    if (this.TransactionId != null)
                    {
                        await BeforeEndAsync().Await();

                        // Now notify the broker that a new XA'ish transaction has started.
                        TransactionInfo info = new TransactionInfo();
                        info.ConnectionId = this.connection.ConnectionId;
                        info.TransactionId = this.TransactionId;
                        info.Type = (int)TransactionType.End;

                        await this.connection.CheckConnectedAsync().Await();
                        await this.connection.SyncRequestAsync((TransactionInfo) info.Clone()).Await();

                        info.Type = (int)TransactionType.Rollback;
                        await this.connection.CheckConnectedAsync().Await();
                        await this.connection.SyncRequestAsync(info).Await();

                        Tracer.Debug("Transaction Rollback Done TX id: " + this.TransactionId);

                        RecoveryLogger.LogRecovered(this.TransactionId as XATransactionId);

                        // if server responds that nothing needs to be done, then reply done.
                        enlistment.Done();

                        AfterRollback();
                    }
                }
                catch (Exception ex)
                {
                    Tracer.DebugFormat("Transaction[{0}] Rollback failed with error: {1}",
                                       this.TransactionId, ex.Message);
                    AfterRollback();
                    try
                    {
                        this.connection.OnAsyncException(ex);
                    }
                    catch (Exception error)
                    {
                        Tracer.Error(error.ToString());
                    }
                }
                finally
                {
                    this.currentEnlistment = null;
                    this.TransactionId = null;
                    this.netTxState = TxState.None;

                    CountDownLatch latch = this.recoveryComplete;
                    if (latch != null)
                    {
                        latch.countDown();
                    }

                    this.dtcControlEvent.Set();
                }
            }
        }

        public void InDoubt(Enlistment enlistment)
        {
            
        }
        public async Task InDoubtAsync(Enlistment enlistment)
        {
            using(await syncObject.LockAsync().Await())
            {
                try
                {
                    Tracer.Debug("In Doubt notification received for TX id: " + this.TransactionId);

                    await BeforeEndAsync().Await();

                    // Now notify the broker that Rollback should be performed.
                    TransactionInfo info = new TransactionInfo();
                    info.ConnectionId = this.connection.ConnectionId;
                    info.TransactionId = this.TransactionId;
                    info.Type = (int)TransactionType.End;

                    await this.connection.CheckConnectedAsync().Await();
                    await this.connection.SyncRequestAsync((TransactionInfo) info.Clone()).Await();

                    info.Type = (int)TransactionType.Rollback;
                    await this.connection.CheckConnectedAsync().Await();
                    await this.connection.SyncRequestAsync(info).Await();

                    Tracer.Debug("InDoubt Transaction Rollback Done TX id: " + this.TransactionId);

                    RecoveryLogger.LogRecovered(this.TransactionId as XATransactionId);

                    // if server responds that nothing needs to be done, then reply done.
                    enlistment.Done();

                    AfterRollback();
                }
                finally
                {
                    this.currentEnlistment = null;
                    this.TransactionId = null;
                    this.netTxState = TxState.None;

                    CountDownLatch latch = this.recoveryComplete;
                    if (latch != null)
                    {
                        latch.countDown();
                    }

                    this.dtcControlEvent.Set();
                }
            }
        }

        #endregion

        #region Distributed Transaction Recovery Bits

        private volatile CountDownLatch recoveryComplete;

        /// <summary>
        /// Should be called from NetTxSession when created to check if any TX
        /// data is stored for recovery and whether the Broker has matching info
        /// stored.  If an Transaction is found that belongs to this client and is
        /// still alive on the Broker it will be recovered, otherwise the stored
        /// data should be cleared.
        /// </summary>
        public async Task InitializeDtcTxContextAsync()
        {
            string resourceManagerId = ResourceManagerId;

            // initialize the logger with the current Resource Manager Id
            RecoveryLogger.Initialize(resourceManagerId);

            using(await recoveredResourceManagerIdsLock.LockAsync().Await())
            {
                if (recoveredResourceManagerIds.ContainsKey(resourceManagerId))
                {
                    return;
                }

                recoveredResourceManagerIds[resourceManagerId] = true;

                KeyValuePair<XATransactionId, byte[]>[] localRecoverables = RecoveryLogger.GetRecoverables();
                if (localRecoverables.Length == 0)
                {
                    Tracer.Debug("Did not detect any open DTC transaction records on disk.");
                    // No local data so anything stored on the broker can't be recovered here.
                    return;
                }

                XATransactionId[] recoverables = await TryRecoverBrokerTXIdsAsync().Await();
                if (recoverables.Length == 0)
                {
                    Tracer.Debug("Did not detect any recoverable transactions at Broker.");
                    // Broker has no recoverable data so nothing to do here, delete the 
                    // old recovery log as its stale.
                    RecoveryLogger.Purge();
                    return;
                }

                List<KeyValuePair<XATransactionId, byte[]>> matches = new List<KeyValuePair<XATransactionId, byte[]>>();

                foreach (XATransactionId recoverable in recoverables)
                {
                    foreach (KeyValuePair<XATransactionId, byte[]> entry in localRecoverables)
                    {
                        if (entry.Key.Equals(recoverable))
                        {
                            Tracer.DebugFormat("Found a matching TX on Broker to stored Id: {0} reenlisting.", entry.Key);
                            matches.Add(entry);
                        }
                    }
                }

                if (matches.Count != 0)
                {
                    this.recoveryComplete = new CountDownLatch(matches.Count);

                    foreach (KeyValuePair<XATransactionId, byte[]> recoverable in matches)
                    {
                        this.TransactionId = recoverable.Key;
                        Tracer.Info("Reenlisting recovered TX with Id: " + this.TransactionId);
                        this.currentEnlistment =
                            TransactionManager.Reenlist(ResourceManagerGuid, recoverable.Value, this);
                    }

                    RecoveryCompleteAwait();
                    Tracer.Debug("All Recovered TX enlistments Reports complete, Recovery Complete.");
                    TransactionManager.RecoveryComplete(ResourceManagerGuid);
                    return;
                }

                // The old recovery information doesn't match what's on the broker so we
                // should discard it as its stale now.
                RecoveryLogger.Purge();
            }
        }

        private void RecoveryCompleteAwait()
        {
            this.recoveryComplete.@await();
        }
        
        private async Task<XATransactionId[]> TryRecoverBrokerTXIdsAsync()
        {
            Tracer.Debug("Checking for Recoverable Transactions on Broker.");

            TransactionInfo info = new TransactionInfo();
            info.ConnectionId = this.session.Connection.ConnectionId;
            info.Type = (int)TransactionType.Recover;

            await this.connection.CheckConnectedAsync().Await();
            DataArrayResponse response = await this.connection.SyncRequestAsync(info).Await() as DataArrayResponse;

            if (response != null && response.Data.Length > 0)
            {
                Tracer.DebugFormat("Broker reports there are {0} recoverable XA Transactions", response.Data.Length);

                List<XATransactionId> recovered = new List<XATransactionId>();

                foreach (DataStructure ds in response.Data)
                {
                    XATransactionId xid = ds as XATransactionId;
                    if (xid != null)
                    {
                        recovered.Add(xid);
                    }
                }

                return recovered.ToArray();
            }

            return new XATransactionId[0];
        }

        #endregion

        internal IRecoveryLogger RecoveryLogger
        {
            get { return (this.connection as NetTxConnection).RecoveryPolicy.RecoveryLogger; }
        }

        internal string ResourceManagerId
        {
            get { return (this.connection as NetTxConnection).ResourceManagerGuid.ToString(); }
        }

        internal Guid ResourceManagerGuid
        {
            get { return (this.connection as NetTxConnection).ResourceManagerGuid; }
        }

    }
}