using System.Text.Json;
using Confluent.Kafka;
using Elastic.Clients.Elasticsearch;
using Microsoft.Extensions.Options;

public class EsIndexerService : BackgroundService
{
    private static readonly JsonSerializerOptions EventJsonOptions = new()
    {
        PropertyNameCaseInsensitive = true
    };

    private readonly ElasticsearchClient _elastic;
    private readonly KafkaOptions _kafkaOptions;
    private readonly ElasticsearchOptions _esOptions;
    private readonly ILogger<EsIndexerService> _logger;

    public EsIndexerService(
        ElasticsearchClient elastic,
        IOptions<KafkaOptions> kafkaOptions,
        IOptions<ElasticsearchOptions> esOptions,
        ILogger<EsIndexerService> logger)
    {
        _elastic = elastic;
        _kafkaOptions = kafkaOptions.Value;
        _esOptions = esOptions.Value;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = _kafkaOptions.BootstrapServers,
            GroupId = "es-indexer",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            // Manual commit: offset is only committed after a successful Elasticsearch write.
            // If the process crashes between ES write and commit, the message is reprocessed.
            // Consumers must be idempotent. See idempotency contract above.
            EnableAutoCommit = false,
            EnablePartitionEof = false
        };

        using var consumer = new ConsumerBuilder<string, string>(config).Build();

        consumer.Subscribe(new[]
        {
            _kafkaOptions.Topics.ObservationRecorded,
            _kafkaOptions.Topics.AlertGenerated,
            _kafkaOptions.Topics.EncounterStatusChanged,
            _kafkaOptions.Topics.SepsisBundleCreated,
            _kafkaOptions.Topics.SepsisBundleUpdated
        });

        _logger.LogInformation("EsIndexerService started. Subscribed to 5 topics.");

        var guard = new PoisonPillGuard("es-indexer", _kafkaOptions.MaxPoisonRetries, _logger);

        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                ConsumeResult<string, string>? result = null;
                try
                {
                    result = consumer.Consume(stoppingToken);
                    await DispatchAsync(result.Topic, result.Message.Value, stoppingToken);
                    consumer.Commit(result);
                    guard.OnSuccess();
                }
                catch (OperationCanceledException)
                {
                    break;
                }
                catch (Exception ex)
                {
                    if (result is not null && guard.ShouldSkip(result, ex))
                    {
                        consumer.Commit(result);
                        continue;
                    }

                    _logger.LogError(ex,
                        "EsIndexer failed processing topic={Topic} offset={Offset} — will retry",
                        result?.Topic, result?.Offset.Value);
                    await Task.Delay(1000, stoppingToken);
                }
            }
        }
        finally
        {
            consumer.Close();
        }
    }

    private Task DispatchAsync(string topic, string payload, CancellationToken ct) => topic switch
    {
        var t when t == _kafkaOptions.Topics.EncounterStatusChanged =>
            HandleEncounterStatusChangedAsync(payload, ct),
        var t when t == _kafkaOptions.Topics.ObservationRecorded =>
            HandleObservationRecordedAsync(payload, ct),
        var t when t == _kafkaOptions.Topics.AlertGenerated =>
            HandleAlertGeneratedAsync(payload, ct),
        var t when t == _kafkaOptions.Topics.SepsisBundleCreated =>
            HandleSepsisBundleEventAsync(payload, ct),
        var t when t == _kafkaOptions.Topics.SepsisBundleUpdated =>
            HandleSepsisBundleEventAsync(payload, ct),
        _ => Task.CompletedTask
    };

    // --- encounter.status.changed ---
    // Upserts the patient_encounters document. DocAsUpsert=true means:
    // if the document does not exist, it is created; if it exists, it is replaced.
    // Idempotent: applying the same event twice produces the same document.
    private async Task HandleEncounterStatusChangedAsync(string payload, CancellationToken ct)
    {
        var evt = JsonSerializer.Deserialize<EncounterStatusChangedEvent>(payload, EventJsonOptions)!;

        var doc = new PatientEncounterDocument
        {
            EncounterId = evt.EncounterId.ToString(),
            PatientId = evt.PatientId.ToString(),
            Mrn = evt.Mrn,
            PatientName = evt.PatientName,
            Department = evt.Department,
            Status = evt.NewStatus,
            AttendingPhysician = evt.AttendingPhysician,
            RoomBed = evt.RoomBed,
            AdmissionReason = evt.AdmissionReason,
            AdmittedAt = evt.AdmittedAt,            
            OpenAlertCount = 0,
            LastObservationAt = null
        };

        var resp = await _elastic.UpdateAsync<PatientEncounterDocument, PatientEncounterDocument>(
            _esOptions.Indices.PatientEncounters,
            evt.EncounterId.ToString(),
            u => u.Doc(doc).DocAsUpsert(true),
            ct);

        if (!resp.IsValidResponse)
            throw new InvalidOperationException(
                $"ES upsert failed for encounter {evt.EncounterId}: {resp.DebugInformation}");

        _logger.LogDebug("Upserted patient_encounters for encounter {Id} → status={Status}",
            evt.EncounterId, evt.NewStatus);
    }

    // --- observation.recorded ---
    // Indexes the observation by observationId — idempotent PUT.
    // Also updates lastObservationAt on the parent encounter document using a script.
    private async Task HandleObservationRecordedAsync(string payload, CancellationToken ct)
    {
        var evt = JsonSerializer.Deserialize<ObservationRecordedEvent>(payload, EventJsonOptions)!;

        // Index into observations — document ID is observationId
        var doc = new ObservationDocument
        {
            ObservationId = evt.ObservationId.ToString(),
            EncounterId = evt.EncounterId.ToString(),
            PatientId = evt.PatientId.ToString(),
            Mrn = evt.Mrn ?? string.Empty,
            ObservationCode = evt.ObservationCode,
            Value = (double)evt.Value,
            Unit = evt.Unit,
            Source = evt.Source,
            RecordedAt = evt.RecordedAt
        };

        var indexResp = await _elastic.IndexAsync(
            doc,
            i => i.Index(_esOptions.Indices.Observations).Id(doc.ObservationId),
            ct);

        if (!indexResp.IsValidResponse)
            throw new InvalidOperationException(
                $"ES index failed for observation {evt.ObservationId}: {indexResp.DebugInformation}");

        // Update lastObservationAt on patient_encounters using a conditional script:
        // only update if the new recordedAt is later than the stored value.
        // This handles out-of-order delivery: an older observation re-processed after a
        // newer one must not overwrite lastObservationAt with a stale timestamp.
        var updateResp = await _elastic.UpdateAsync<PatientEncounterDocument, object>(
            _esOptions.Indices.PatientEncounters,
            evt.EncounterId.ToString(),
            u => u
                .Script(new Script(new InlineScript
                {
                    Source = """
                        if (ctx._source.lastObservationAt == null ||
                            params.recordedAt > ctx._source.lastObservationAt) {
                            ctx._source.lastObservationAt = params.recordedAt;
                        }
                        """,
                    Language = ScriptLanguage.Painless,
                    Params = new Dictionary<string, object>
                    {
                        ["recordedAt"] = evt.RecordedAt.ToString("O")
                    }
                }))
                .RetryOnConflict(3),
            ct);

        if (!updateResp.IsValidResponse && updateResp.Result != Result.NotFound)
            _logger.LogWarning(
                "Could not update lastObservationAt for encounter {Id} — encounter document may not exist yet",
                evt.EncounterId);
    }

    // --- alert.generated ---
    // Indexes the alert by alertId. Also increments openAlertCount on patient_encounters.
    // TRADE-OFF: openAlertCount increment is not idempotent for partial reprocessing.
    // It is correct for full replay from offset 0 (the stated recovery procedure).
    // For production, use a set of counted alert IDs in the script to enforce idempotency.
    private async Task HandleAlertGeneratedAsync(string payload, CancellationToken ct)
    {
        var evt = JsonSerializer.Deserialize<AlertGeneratedEvent>(payload, EventJsonOptions)!;
        using var payloadDoc = JsonDocument.Parse(payload);
        var root = payloadDoc.RootElement;

        var alertDoc = new ClinicalAlertDocument
        {
            AlertId = evt.AlertId.ToString(),
            EncounterId = evt.EncounterId.ToString(),
            PatientId = evt.PatientId.ToString(),
            Department = evt.Department ?? string.Empty,
            AlertType = evt.AlertType,
            Severity = evt.Severity,
            Status = "Open",
            Details = root.TryGetProperty("details", out var detailsElem)
                ? detailsElem.GetString() ?? string.Empty
                : string.Empty,
            NarrativeSummary = root.TryGetProperty("explanation", out var explanationElem)
                && explanationElem.TryGetProperty("narrativeSummary", out var narrativeElem)
                ? narrativeElem.GetString()
                : null,
            TriggeredAt = evt.TriggeredAt
        };

        var indexResp = await _elastic.IndexAsync(
            alertDoc,
            i => i.Index(_esOptions.Indices.ClinicalAlerts).Id(alertDoc.AlertId),
            ct);

        if (!indexResp.IsValidResponse)
            throw new InvalidOperationException(
                $"ES index failed for alert {evt.AlertId}: {indexResp.DebugInformation}");

        // Increment openAlertCount on the parent encounter document.
        // NEWS2 alerts also carry news2Score/news2RiskLevel in the Kafka payload;
        // stamp those on patient_encounters so ward dashboards can filter by acuity.
        var scriptLines = new List<string> { "ctx._source.openAlertCount += 1" };
        Dictionary<string, object>? scriptParams = null;

        if (root.TryGetProperty("news2Score", out var scoreElem) &&
            root.TryGetProperty("news2RiskLevel", out var riskElem))
        {
            scriptLines.Add("ctx._source.news2Score = params.score");
            scriptLines.Add("ctx._source.news2RiskLevel = params.riskLevel");
            scriptParams = new Dictionary<string, object>
            {
                ["score"] = scoreElem.GetInt32(),
                ["riskLevel"] = riskElem.GetString()!
            };
        }

        var updateResp = await _elastic.UpdateAsync<PatientEncounterDocument, object>(
            _esOptions.Indices.PatientEncounters,
            evt.EncounterId.ToString(),
            u => u
                .Script(new Script(new InlineScript
                {
                    Source = string.Join(";\n", scriptLines),
                    Language = ScriptLanguage.Painless,
                    Params = scriptParams
                }))
                .RetryOnConflict(3),
            ct);

        if (!updateResp.IsValidResponse && updateResp.Result != Result.NotFound)
            _logger.LogWarning(
                "Could not update patient_encounters for alert on encounter {Id}", evt.EncounterId);
    }

    // --- sepsis.bundle.created / sepsis.bundle.updated ---
    // Denormalizes bundle status onto the patient_encounters document so department
    // acuity dashboards can filter/sort by sepsis bundle compliance state.
    private async Task HandleSepsisBundleEventAsync(string payload, CancellationToken ct)
    {
        using var doc = JsonDocument.Parse(payload);
        var root = doc.RootElement;

        var encounterId = root.GetProperty("encounterId").GetString()!;
        var complianceStatus = root.TryGetProperty("complianceStatus", out var cs)
            ? cs.GetString() ?? "IN_PROGRESS"
            : "IN_PROGRESS";
        var deadlineAt = root.TryGetProperty("deadlineAt", out var dl) ? dl.GetString() : null;

        var scriptLines = new List<string>
        {
            "ctx._source.sepsisBundleStatus = params.status",
            "ctx._source.sepsisBundleDeadlineAt = params.deadlineAt"
        };

        var scriptParams = new Dictionary<string, object?>
        {
            ["status"] = complianceStatus,
            ["deadlineAt"] = deadlineAt
        };

        if (root.TryGetProperty("elementStatus", out _))
        {
            scriptLines.Add("ctx._source.sepsisBundleElementsCompleted = params.elementsCompleted");
            scriptParams["elementsCompleted"] = root.TryGetProperty("elementsCompleted", out var ec)
                ? ec.GetInt32()
                : (complianceStatus is "COMPLIANT" or "NON_COMPLIANT" ? 4 : 0);
        }
        else
        {
            scriptLines.Add("ctx._source.sepsisBundleElementsCompleted = 0");
        }

        var updateResp = await _elastic.UpdateAsync<PatientEncounterDocument, object>(
            _esOptions.Indices.PatientEncounters,
            encounterId,
            u => u
                .Script(new Script(new InlineScript
                {
                    Source = string.Join(";\n", scriptLines),
                    Language = ScriptLanguage.Painless,
                    Params = scriptParams!
                }))
                .RetryOnConflict(3),
            ct);

        if (!updateResp.IsValidResponse && updateResp.Result != Result.NotFound)
            _logger.LogWarning(
                "Could not update patient_encounters bundle status for encounter {Id}", encounterId);

        _logger.LogDebug("Updated patient_encounters bundle status for encounter {Id} → {Status}",
            encounterId, complianceStatus);
    }
}