using System.Data;
using System.Diagnostics;
using Microsoft.Extensions.Options;
using Npgsql;
using NpgsqlTypes;
using Tiku.Application.Jobs;
using Tiku.Application.PlatformAdmin;
using Tiku.Application.PlatformBilling;
using Tiku.Application.Security;
using Tiku.Application.Tenancy;
using Tiku.Infrastructure.Observability;

namespace Tiku.Worker;

public sealed class WorkerOptions
{
    public const string SectionName = "Worker";

    public bool Enabled { get; set; } = true;
    public int JobPollSeconds { get; set; } = 2;
    public int JobParallelism { get; set; } = 4;
    public int JobBatchSize { get; set; } = 5;

    public static bool BeValid(WorkerOptions options)
    {
        return options.JobPollSeconds is >= 1 and <= 3600 &&
               options.JobParallelism is >= 1 and <= 32 &&
               options.JobBatchSize is >= 1 and <= 100;
    }
}

internal interface IPeriodicProcessorLock
{
    Task<IAsyncDisposable?> TryAcquireAsync(string processor, CancellationToken cancellationToken);
}

internal sealed class WorkerStateReporter(NpgsqlDataSource dataSource)
{
    private readonly DateTimeOffset startedAt = DateTimeOffset.UtcNow;
    private readonly string workerId = $"{Environment.MachineName}:{Environment.ProcessId}";

    public Task StartedAsync(string processor, CancellationToken cancellationToken)
    {
        return UpsertAsync(processor, true, null, cancellationToken);
    }

    public Task CompletedAsync(string processor, Exception? error, CancellationToken cancellationToken)
    {
        return UpsertAsync(processor, false, error?.Message, cancellationToken);
    }

    private async Task UpsertAsync(
        string processor,
        bool running,
        string? error,
        CancellationToken cancellationToken)
    {
        var now = DateTimeOffset.UtcNow;
        await using var connection = await dataSource.OpenConnectionAsync(cancellationToken);
        await using var command = connection.CreateCommand();
        command.CommandText = """
                              INSERT INTO worker_heartbeats
                                  (id, worker_id, processor, started_at, last_heartbeat_at, last_iteration_started_at,
                                   last_iteration_completed_at, last_succeeded_at, last_error, is_running)
                              VALUES
                                  (gen_random_uuid(), @worker_id, @processor, @started_at, @now,
                                   CASE WHEN @running THEN @now ELSE NULL END,
                                   CASE WHEN @running THEN NULL ELSE @now END,
                                   CASE WHEN NOT @running AND @error IS NULL THEN @now ELSE NULL END,
                                   @error, @running)
                              ON CONFLICT (worker_id, processor) DO UPDATE SET
                                  last_heartbeat_at = EXCLUDED.last_heartbeat_at,
                                  last_iteration_started_at = CASE WHEN EXCLUDED.is_running THEN EXCLUDED.last_heartbeat_at ELSE worker_heartbeats.last_iteration_started_at END,
                                  last_iteration_completed_at = CASE WHEN EXCLUDED.is_running THEN worker_heartbeats.last_iteration_completed_at ELSE EXCLUDED.last_heartbeat_at END,
                                  last_succeeded_at = CASE WHEN NOT EXCLUDED.is_running AND EXCLUDED.last_error IS NULL THEN EXCLUDED.last_heartbeat_at ELSE worker_heartbeats.last_succeeded_at END,
                                  last_error = EXCLUDED.last_error,
                                  is_running = EXCLUDED.is_running
                              """;
        command.Parameters.AddWithValue("worker_id", workerId);
        command.Parameters.AddWithValue("processor", processor);
        command.Parameters.AddWithValue("started_at", startedAt);
        command.Parameters.AddWithValue("now", now);
        command.Parameters.AddWithValue("running", running);
        command.Parameters.Add("error", NpgsqlDbType.Text).Value = (object?)error ?? DBNull.Value;
        await command.ExecuteNonQueryAsync(cancellationToken);
    }
}

internal sealed class PostgresPeriodicProcessorLock(NpgsqlDataSource dataSource) : IPeriodicProcessorLock
{
    public async Task<IAsyncDisposable?> TryAcquireAsync(string processor, CancellationToken cancellationToken)
    {
        var connection = await dataSource.OpenConnectionAsync(cancellationToken);
        await using var command = connection.CreateCommand();
        command.CommandText = "SELECT pg_try_advisory_lock(hashtextextended(@processor, 0))";
        command.Parameters.AddWithValue("processor", processor);
        var acquired = (bool)(await command.ExecuteScalarAsync(cancellationToken) ?? false);
        if (!acquired)
        {
            await connection.DisposeAsync();
            return null;
        }

        return new AdvisoryLockLease(connection, processor);
    }

    private sealed class AdvisoryLockLease(NpgsqlConnection connection, string processor) : IAsyncDisposable
    {
        public async ValueTask DisposeAsync()
        {
            if (connection.State == ConnectionState.Open)
            {
                await using var command = connection.CreateCommand();
                command.CommandText = "SELECT pg_advisory_unlock(hashtextextended(@processor, 0))";
                command.Parameters.AddWithValue("processor", processor);
                await command.ExecuteScalarAsync();
            }

            await connection.DisposeAsync();
        }
    }
}

internal abstract class PeriodicWorker(
    ILogger logger,
    IPeriodicProcessorLock processorLock,
    WorkerStateReporter stateReporter,
    string processorName,
    TimeSpan interval,
    bool enabled) : BackgroundService
{
    protected abstract Task<int> ProcessAsync(CancellationToken cancellationToken);

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        if (!enabled) return;

        while (!stoppingToken.IsCancellationRequested)
            try
            {
                await using var lease = await processorLock.TryAcquireAsync(processorName, stoppingToken);
                if (lease is not null)
                {
                    var iterationTimestamp = Stopwatch.GetTimestamp();
                    await stateReporter.StartedAsync(processorName, stoppingToken);
                    try
                    {
                        var processed = await ProcessAsync(stoppingToken);
                        await stateReporter.CompletedAsync(processorName, null, stoppingToken);
                        WorkerTelemetry.RecordIteration(processorName, true,
                            Stopwatch.GetElapsedTime(iterationTimestamp).TotalMilliseconds);
                        if (processed > 0)
                            logger.LogInformation("{Worker} processed {Count} items.", GetType().Name, processed);
                    }
                    catch (Exception exception)
                    {
                        await stateReporter.CompletedAsync(processorName, exception, CancellationToken.None);
                        WorkerTelemetry.RecordIteration(processorName, false,
                            Stopwatch.GetElapsedTime(iterationTimestamp).TotalMilliseconds);
                        throw;
                    }
                }

                await Task.Delay(interval, stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception exception)
            {
                logger.LogError(exception, "{Worker} iteration failed.", GetType().Name);
                try
                {
                    await Task.Delay(interval, stoppingToken);
                }
                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
                {
                    break;
                }
            }
    }

    protected static void InitializeSystem(IServiceProvider services, string reason)
    {
        services.GetRequiredService<ITenantContextInitializer>().InitializeSystem(null, reason);
    }
}

internal sealed class TenantDomainWorker(
    IServiceScopeFactory scopeFactory,
    IPeriodicProcessorLock processorLock,
    WorkerStateReporter stateReporter,
    IOptions<DomainLifecycleOptions> domainOptions,
    IOptions<WorkerOptions> workerOptions,
    ILogger<TenantDomainWorker> logger)
    : PeriodicWorker(logger, processorLock, stateReporter, "tenant-domain-lifecycle",
        TimeSpan.FromSeconds(Math.Clamp(domainOptions.Value.PollSeconds, 10, 3600)), workerOptions.Value.Enabled)
{
    protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        InitializeSystem(scope.ServiceProvider, "Tenant domain DNS and TLS lifecycle worker");
        return await scope.ServiceProvider.GetRequiredService<ITenantDomainLifecycleService>()
            .ProcessPendingAsync(cancellationToken);
    }
}

internal sealed class SaasSubscriptionWorker(
    IServiceScopeFactory scopeFactory,
    IPeriodicProcessorLock processorLock,
    WorkerStateReporter stateReporter,
    IOptions<WorkerOptions> options,
    ILogger<SaasSubscriptionWorker> logger)
    : PeriodicWorker(logger, processorLock, stateReporter, "saas-subscription-lifecycle", TimeSpan.FromSeconds(60),
        options.Value.Enabled)
{
    protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        InitializeSystem(scope.ServiceProvider, "SaaS subscription lifecycle worker");
        return await scope.ServiceProvider.GetRequiredService<ISaasSubscriptionLifecycleService>()
            .ProcessDueAsync(cancellationToken: cancellationToken);
    }
}

internal sealed class FeatureUsageWorker(
    IServiceScopeFactory scopeFactory,
    IPeriodicProcessorLock processorLock,
    WorkerStateReporter stateReporter,
    IOptions<FeatureUsageReconciliationOptions> featureOptions,
    IOptions<WorkerOptions> workerOptions,
    ILogger<FeatureUsageWorker> logger)
    : PeriodicWorker(logger, processorLock, stateReporter, "feature-usage-reconciliation",
        TimeSpan.FromMinutes(Math.Clamp(featureOptions.Value.IntervalMinutes, 1, 1440)), workerOptions.Value.Enabled)
{
    protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        InitializeSystem(scope.ServiceProvider, "Tenant feature usage reconciliation worker");
        return await scope.ServiceProvider.GetRequiredService<IFeatureUsageReconciliationService>()
            .ProcessDueAsync(cancellationToken);
    }
}

internal sealed class BackgroundJobsWorker(
    IServiceScopeFactory scopeFactory,
    IPeriodicProcessorLock processorLock,
    WorkerStateReporter stateReporter,
    IOptions<WorkerOptions> options,
    ILogger<BackgroundJobsWorker> logger)
    : PeriodicWorker(logger, processorLock, stateReporter, "background-jobs",
        TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled)
{
    private readonly int batchSize = options.Value.JobBatchSize;
    private readonly int parallelism = options.Value.JobParallelism;
    private readonly string workerId = $"{Environment.MachineName}:{Guid.NewGuid():N}";

    protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
    {
        var workers = Enumerable.Range(0, parallelism)
            .Select(index => ProcessPartitionAsync(index, cancellationToken));
        return (await Task.WhenAll(workers)).Sum();
    }

    private async Task<int> ProcessPartitionAsync(int index, CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        InitializeSystem(scope.ServiceProvider, "Background job lease worker");
        return await scope.ServiceProvider.GetRequiredService<IBackgroundJobProcessor>()
            .ProcessPendingAsync($"{workerId}:{index}", batchSize, true, cancellationToken);
    }
}

internal sealed class AuthorizationCacheInvalidationWorker(
    IServiceScopeFactory scopeFactory,
    IPeriodicProcessorLock processorLock,
    WorkerStateReporter stateReporter,
    IOptions<WorkerOptions> options,
    ILogger<AuthorizationCacheInvalidationWorker> logger)
    : PeriodicWorker(logger, processorLock, stateReporter, "authorization-cache-invalidations",
        TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled)
{
    protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        InitializeSystem(scope.ServiceProvider, "Authorization cache invalidation worker");
        return await scope.ServiceProvider.GetRequiredService<IAuthorizationCacheInvalidationProcessor>()
            .ProcessPendingAsync(cancellationToken: cancellationToken);
    }
}

internal sealed class CommercialBillingWorker(
    IServiceScopeFactory scopeFactory,
    IPeriodicProcessorLock processorLock,
    WorkerStateReporter stateReporter,
    IOptions<WorkerOptions> options,
    ILogger<CommercialBillingWorker> logger)
    : PeriodicWorker(logger, processorLock, stateReporter, "commercial-billing",
        TimeSpan.FromSeconds(60), options.Value.Enabled)
{
    protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        InitializeSystem(scope.ServiceProvider, "Commercial renewal, receivable, dunning and refund worker");
        return await scope.ServiceProvider.GetRequiredService<ICommercialBillingProcessor>()
            .ProcessDueAsync(cancellationToken);
    }
}

internal sealed class PlatformApprovalWorker(
    IServiceScopeFactory scopeFactory,
    IPeriodicProcessorLock processorLock,
    WorkerStateReporter stateReporter,
    IOptions<WorkerOptions> options,
    ILogger<PlatformApprovalWorker> logger)
    : PeriodicWorker(logger, processorLock, stateReporter, "platform-approvals",
        TimeSpan.FromSeconds(options.Value.JobPollSeconds), options.Value.Enabled)
{
    protected override async Task<int> ProcessAsync(CancellationToken cancellationToken)
    {
        await using var scope = scopeFactory.CreateAsyncScope();
        InitializeSystem(scope.ServiceProvider, "Execute approved platform commands");
        return await scope.ServiceProvider.GetRequiredService<IPlatformApprovalService>()
            .ProcessApprovedAsync(cancellationToken: cancellationToken);
    }
}