// SPDX-License-Identifier: AGPL-3.0-or-later
// Copyright (C) 2022-2026 Aykut Alparslan KOC

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Net;
using Autofac;
using Ferrite.Core.Connection;
using Ferrite.Services.Calls;
using Ferrite.Services.Scheduling;
using Ferrite.Services.Sessions;
using Ferrite.Core.Execution;
using Ferrite.Transport;
using Ferrite.Utils;

namespace Ferrite.Core;

public class FerriteServer : IFerriteServer
{
    private enum ServerState
    {
        Created,
        Running,
        Stopped
    }

    private readonly ILifetimeScope _scope;
    private readonly IConnectionListener _socketListener;
    private readonly IMessagePipe _pipe;
    private readonly DeliveredPtsRecorder _deliveredPts;
    private Task? _pipeReceiveTask;
    private readonly ISessionService _sessionManager;
    private readonly ISecretChatMaintenance _secretChatMaintenance;
    private readonly ICallMediaRelay _callMediaRelay;
    private readonly GroupCallDisconnectMonitor _groupCallDisconnects;
    private readonly GroupCallSourcesChangedMonitor _groupCallSourcesChanged;
    private readonly GroupCallMediaRuntime _groupCallMediaRuntime;
    private readonly GroupCallBroadcastRuntime _groupCallBroadcastRuntime;
    private readonly GroupCallRecordingRuntime _groupCallRecordingRuntime;
    private readonly ScheduledMessageRuntime _scheduledMessages;
    private readonly MessageExpiryRuntime _messageExpiry;
    private readonly ICallRegistry _callRegistry;
    private readonly CallTerminator _callTerminator;
    private readonly IMTProtoTime _time;
    private readonly ILogger _log;
    private readonly ConcurrentDictionary<MTProtoConnection, byte> _connections =
        new();
    private readonly SemaphoreSlim _lifecycleGate = new(1, 1);
    private volatile ServerState _state = ServerState.Created;
    private CancellationTokenSource? _stopping;
    private Task? _acceptLoop;
    public FerriteServer(ILifetimeScope scope)
    {
        _scope = scope;
        _socketListener = _scope.Resolve<IConnectionListener>();
        _sessionManager = _scope.Resolve<ISessionService>();
        _secretChatMaintenance = _scope.Resolve<ISecretChatMaintenance>();
        _callMediaRelay = _scope.Resolve<ICallMediaRelay>();
        _groupCallDisconnects = _scope.Resolve<GroupCallDisconnectMonitor>();
        _groupCallSourcesChanged = _scope.Resolve<GroupCallSourcesChangedMonitor>();
        _groupCallMediaRuntime = _scope.Resolve<GroupCallMediaRuntime>();
        _groupCallBroadcastRuntime = _scope.Resolve<GroupCallBroadcastRuntime>();
        _groupCallRecordingRuntime = _scope.Resolve<GroupCallRecordingRuntime>();
        _scheduledMessages = _scope.Resolve<ScheduledMessageRuntime>();
        _messageExpiry = _scope.Resolve<MessageExpiryRuntime>();
        _callRegistry = _scope.Resolve<ICallRegistry>();
        _callTerminator = _scope.Resolve<CallTerminator>();
        _time = _scope.Resolve<IMTProtoTime>();
        _pipe = _scope.Resolve<IMessagePipe>();
        _deliveredPts = _scope.Resolve<DeliveredPtsRecorder>();
        _log = _scope.Resolve<ILogger>();
    }

    public IPEndPoint? BoundEndpoint => _state == ServerState.Running
        ? _socketListener.EndPoint as IPEndPoint
        : null;

    public IPEndPoint? CallMediaBoundEndpoint => _state == ServerState.Running
        ? _callMediaRelay.BoundEndpoint
        : null;

    public CallMediaRelaySnapshot CallMediaRelaySnapshot => new(
        _callMediaRelay.IsReady, _callMediaRelay.AllocationCount,
        _callMediaRelay.ForwardedPackets, _callMediaRelay.ForwardedBytes,
        _callMediaRelay.DroppedPackets);

    public FerriteReadinessSnapshot Readiness => new(
        _state == ServerState.Running,
        ToPipelineSnapshot(_groupCallMediaRuntime.Snapshot),
        ToPipelineSnapshot(_groupCallBroadcastRuntime.Snapshot),
        ToPipelineSnapshot(_groupCallRecordingRuntime.Snapshot));

    public async Task StartAsync(IPEndPoint endPoint, CancellationToken token)
    {
        Task acceptLoop = await StartCoreAsync(endPoint, token);
        await acceptLoop;
    }

    private async Task<Task> StartCoreAsync(IPEndPoint endPoint,
        CancellationToken token)
    {
        await _lifecycleGate.WaitAsync(token);
        try
        {
            if (_state != ServerState.Created)
            {
                throw new InvalidOperationException(
                    $"The server cannot start from the {_state} state.");
            }

            var rollback = new List<Func<Task>>();
            try
            {
                token.ThrowIfCancellationRequested();
                await _pipe.SubscribeAsync(
                    MessagePipeChannels.ForNode(_sessionManager.NodeId));
                rollback.Add(async () => await _pipe.UnSubscribeAsync());

                token.ThrowIfCancellationRequested();
                var stopping = new CancellationTokenSource();
                _stopping = stopping;
                Task pipeReceiveTask = DoReceive(stopping.Token);
                _pipeReceiveTask = pipeReceiveTask;
                rollback.Add(async () =>
                {
                    stopping.Cancel();
                    await pipeReceiveTask;
                    stopping.Dispose();
                    _stopping = null;
                    _pipeReceiveTask = null;
                });

                token.ThrowIfCancellationRequested();
                await _secretChatMaintenance.StartAsync(token);
                rollback.Add(async () => await _secretChatMaintenance
                    .StopAsync(CancellationToken.None));

                token.ThrowIfCancellationRequested();
                await _callMediaRelay.StartAsync(token);
                rollback.Add(() => _callMediaRelay.StopAsync());

                token.ThrowIfCancellationRequested();
                await _groupCallDisconnects.StartAsync(token);
                rollback.Add(() => _groupCallDisconnects
                    .StopAsync(CancellationToken.None));

                token.ThrowIfCancellationRequested();
                await _groupCallSourcesChanged.StartAsync(token);
                rollback.Add(() => _groupCallSourcesChanged
                    .StopAsync(CancellationToken.None));

                token.ThrowIfCancellationRequested();
                await _groupCallMediaRuntime.StartAsync(token);
                rollback.Add(() => _groupCallMediaRuntime
                    .StopAsync(CancellationToken.None));

                token.ThrowIfCancellationRequested();
                await _groupCallBroadcastRuntime.StartAsync(token);
                rollback.Add(() => _groupCallBroadcastRuntime
                    .StopAsync(CancellationToken.None));

                token.ThrowIfCancellationRequested();
                await _groupCallRecordingRuntime.StartAsync(token);
                rollback.Add(() => _groupCallRecordingRuntime
                    .StopAsync(CancellationToken.None));

                _callRegistry.SetDeadlineExpiredHandler(OnCallDeadlineExpired);
                rollback.Add(() =>
                {
                    _callRegistry.SetDeadlineExpiredHandler(null);
                    return Task.CompletedTask;
                });

                token.ThrowIfCancellationRequested();
                await _scheduledMessages.StartAsync(token);
                rollback.Add(async () => await _scheduledMessages
                    .StopAsync(CancellationToken.None));

                token.ThrowIfCancellationRequested();
                await _messageExpiry.StartAsync(token);
                rollback.Add(async () => await _messageExpiry
                    .StopAsync(CancellationToken.None));

                token.ThrowIfCancellationRequested();
                _socketListener.Bind(endPoint);
            }
            catch
            {
                await RollbackAsync(rollback);
                throw;
            }

            _state = ServerState.Running;
            _acceptLoop = StartAccept(_socketListener, token);
            return _acceptLoop;
        }
        finally
        {
            _lifecycleGate.Release();
        }
    }

    private async Task RollbackAsync(List<Func<Task>> rollback)
    {
        for (int i = rollback.Count - 1; i >= 0; i--)
        {
            try
            {
                await rollback[i]();
            }
            catch (Exception ex)
            {
                _log.Error(ex, "Server startup rollback step failed.");
            }
        }
    }

    public async Task StopAsync(CancellationToken token)
    {
        await _lifecycleGate.WaitAsync(token);
        try
        {
            if (_state == ServerState.Stopped)
            {
                return;
            }

            if (_state == ServerState.Running)
            {
                _callRegistry.SetDeadlineExpiredHandler(null);
                await _socketListener.UnbindAsync(token);
                if (_acceptLoop is not null)
                {
                    await _acceptLoop.WaitAsync(token);
                    _acceptLoop = null;
                }

                MTProtoConnection[] connections = _connections.Keys.ToArray();
                await Task.WhenAll(connections.Select(connection => connection
                    .StopAsync(new OperationCanceledException(
                        "Ferrite server is stopping.")).AsTask()));
                _connections.Clear();

                await _messageExpiry.StopAsync(token);
                await _scheduledMessages.StopAsync(token);
                await _secretChatMaintenance.StopAsync(token);
                await _callMediaRelay.StopAsync();
                await _groupCallRecordingRuntime.StopAsync(token);
                await _groupCallBroadcastRuntime.StopAsync(token);
                await _groupCallMediaRuntime.StopAsync(token);
                await _groupCallSourcesChanged.StopAsync(token);
                await _groupCallDisconnects.StopAsync(token);

                _stopping?.Cancel();
                if (_pipeReceiveTask is not null)
                {
                    try
                    {
                        await _pipeReceiveTask.WaitAsync(token);
                    }
                    catch (OperationCanceledException) when (!token.IsCancellationRequested)
                    {
                    }
                    _pipeReceiveTask = null;
                }
                await _pipe.UnSubscribeAsync();
                _stopping?.Dispose();
                _stopping = null;
            }

            _state = ServerState.Stopped;
        }
        finally
        {
            _lifecycleGate.Release();
        }

        await _scope.DisposeAsync();
    }

    private static FerritePipelineReadinessSnapshot ToPipelineSnapshot(
        GroupCallMediaRuntimeSnapshot snapshot) => new(
        snapshot.Status switch
        {
            GroupCallMediaRuntimeStatus.Starting =>
                FerritePipelineReadinessStatus.Starting,
            GroupCallMediaRuntimeStatus.Ready =>
                FerritePipelineReadinessStatus.Ready,
            GroupCallMediaRuntimeStatus.Degraded =>
                FerritePipelineReadinessStatus.Degraded,
            _ => FerritePipelineReadinessStatus.Stopped,
        }, snapshot.Failure);

    private static FerritePipelineReadinessSnapshot ToPipelineSnapshot(
        GroupCallBroadcastRuntimeSnapshot snapshot) => new(
        snapshot.Status switch
        {
            GroupCallBroadcastRuntimeStatus.Starting =>
                FerritePipelineReadinessStatus.Starting,
            GroupCallBroadcastRuntimeStatus.Ready =>
                FerritePipelineReadinessStatus.Ready,
            GroupCallBroadcastRuntimeStatus.Degraded =>
                FerritePipelineReadinessStatus.Degraded,
            _ => FerritePipelineReadinessStatus.Stopped,
        }, snapshot.Failure);

    private static FerritePipelineReadinessSnapshot ToPipelineSnapshot(
        GroupCallRecordingRuntimeSnapshot snapshot) => new(
        snapshot.Status switch
        {
            GroupCallRecordingRuntimeStatus.NotConfigured =>
                FerritePipelineReadinessStatus.NotConfigured,
            GroupCallRecordingRuntimeStatus.Starting =>
                FerritePipelineReadinessStatus.Starting,
            GroupCallRecordingRuntimeStatus.Ready =>
                FerritePipelineReadinessStatus.Ready,
            GroupCallRecordingRuntimeStatus.Degraded =>
                FerritePipelineReadinessStatus.Degraded,
            _ => FerritePipelineReadinessStatus.Stopped,
        }, snapshot.Failure);

    private async Task StartAccept(IConnectionListener socketListener,
        CancellationToken cancellationToken)
    {
        _log.Information(String.Format("Server is listening at {0}", socketListener.EndPoint));
        try
        {
            while (true)
            {
                var connection = await socketListener.AcceptAsync(cancellationToken);
                if (connection == null)
                {
                    break;
                }

                _log.Debug("New MTProto connection was created.");
                MTProtoConnection? mtProtoConnection = null;
                try
                {
                    connection.Start();
                    mtProtoConnection = _scope.Resolve<MTProtoConnection>(
                        new NamedParameter("connection", connection));
                    mtProtoConnection.Stopped += OnConnectionStopped;
                    _connections.TryAdd(mtProtoConnection, 0);
                    mtProtoConnection.Start();
                }
                catch (Exception e)
                {
                    if (mtProtoConnection is not null)
                    {
                        await mtProtoConnection.StopAsync(e);
                    }
                    else
                    {
                        connection.Abort(e);
                        await connection.DisposeAsync();
                    }
                    _log.Fatal(e, e.Message);
                }
            }
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
        }
    }

    private void OnConnectionStopped(MTProtoConnection connection)
    {
        connection.Stopped -= OnConnectionStopped;
        _connections.TryRemove(connection, out _);
    }

    private void OnCallDeadlineExpired(long callId, CallDeadlineKind kind)
    {
        int now = checked((int)_time.GetUnixTimeInSeconds());
        CallRegistryResult result = _callRegistry.TryExpire(callId, kind,
            Ferrite.TL.Constructors.baseLayer_PhoneCallDiscardReasonMissed, now);
        if (result.IsOk && result.Call is not null)
        {
            _ = FinalizeExpiredCallAsync(result.Call);
        }
    }

    private async Task FinalizeExpiredCallAsync(CallSnapshot call)
    {
        try
        {
            await _callTerminator.FinalizeAsync(call, invokerUserId: null,
                invokerAuthKeyId: null);
        }
        catch (Exception ex)
        {
            _log.Error(ex, "Call timeout finalization failed.");
        }
    }

    private async Task DoReceive(CancellationToken cancellationToken)
    {
        try
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                var result = await _pipe.ReadMessageAsync(cancellationToken);
                try
                {
                    var message = MTProtoMessageEnvelope.Deserialize(result);
                    if (message is { MessageType: MTProtoMessageType.Unencrypted } &&
                        message.Nonce != null)
                    {
                        var sessionExists = _sessionManager.TryGetLocalAuthSession(message.Nonce, out var protoSession);
                        if (sessionExists &&
                            protoSession.TryGetConnection(out var connection) &&
                            !connection.IsEncrypted)
                        {
                            _ = connection.SendAsync(message);
                        }
                    }
                    else
                    {
                        var sessionExists = _sessionManager.TryGetLocalSession(message.SessionId, out var protoSession);
                        if (sessionExists &&
                            protoSession.TryGetConnection(out var connection))
                        {
                            _log.Debug($"==> delivered to session {message.SessionId} ==<");
                            await connection.SendAsync(message);
                            await _deliveredPts.RecordAsync(message);
                        }
                        else
                        {
                            _log.Debug($"==> session {message.SessionId} is not on this node " +
                                       $"(known:{sessionExists}) ==<");
                        }
                    }
                }
                catch (Exception ex)
                {
                    _log.Error(ex, ex.Message);
                }
            }
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
        }
    }
}