package observability

import (
	"errors"
	"fmt"
	"io"
	"math"
	"nudgebee/services/common"
	"nudgebee/services/integrations"
	"nudgebee/services/query"
	"nudgebee/services/security"
	"strconv"
	"strings"
	"time"
)

// DynatraceLogSource implements LogSource interface for Dynatrace Grail
type DynatraceLogSource struct{}

// dynatraceLogLabelMapping maps standard NudgeBee field names to Dynatrace Grail (OTel) field names.
var dynatraceLogLabelMapping = map[string]string{
	"timestamp": "timestamp",
	"message":   "content",
	"severity":  "status",
	"host":      "host.name",
	"service":   "service.name",
	"namespace": "k8s.namespace.name",
	"pod":       "k8s.pod.name",
	"container": "k8s.container.name",
	"app":       "k8s.workload.name",
}

// GetIgnoredQueryRequestKeys returns keys that should be stripped from
// QueryRequest.Where before query execution. trace_id is not a filterable
// field in New Relic log queries and must be excluded.
func (s *DynatraceLogSource) GetIgnoredQueryRequestKeys() []string {
	return []string{"trace_id"}
}

// QueryLogs fetches logs from Dynatrace Grail using DQL.
// Handles time-range scoping and optional user-supplied filter expressions.
func (s *DynatraceLogSource) QueryLogs(ctx *security.RequestContext, req FetchLogRequest) ([]OutputLog, error) {
	apiToken, baseURL, err := integrations.GetDynatraceConfigs(ctx, req.AccountId)
	if err != nil {
		ctx.GetLogger().Error("DynatraceLogSource.QueryLogs: failed to get configs", "error", err)
		return nil, fmt.Errorf("failed to get Dynatrace configs: %w", err)
	}

	limit := req.Limit
	if limit <= 0 || limit > 1000 {
		limit = 1000
	}

	from, to := s.getTimeRange(req.StartTime, req.EndTime)
	dql, err := s.buildLogDQL(req, from, to, limit)
	if err != nil {
		return nil, fmt.Errorf("failed to build DQL: %w", err)
	}
	ctx.GetLogger().Info("DynatraceLogSource.QueryLogs", "dql", dql)

	result, err := executeDQLQuery(baseURL, apiToken, dql)
	if err != nil {
		return nil, fmt.Errorf("dynatrace logs query failed: %w", err)
	}

	return s.convertToOutputLogs(result.Records), nil
}

// buildLogDQL constructs the DQL fetch logs statement.
// Prefers req.Query (raw DQL filter) over req.QueryRequest.Where (structured conditions).
func (s *DynatraceLogSource) buildLogDQL(req FetchLogRequest, from, to string, limit int) (string, error) {
	var filterExpr string
	if req.Query != "" {
		// If the query is already a complete DQL statement (generated by GetQuery and
		// stored back into req.Query by FetchLogs), return it directly to avoid
		// wrapping it a second time inside "| filter ...".
		if strings.HasPrefix(strings.TrimSpace(req.Query), "fetch ") {
			return req.Query, nil
		}
		filterExpr = req.Query
	} else if hasWhereConditions(req.QueryRequest.Where) {
		var err error
		filterExpr, err = buildDQLWhereClause(req.QueryRequest.Where)
		if err != nil {
			return "", fmt.Errorf("failed to build DQL filter: %w", err)
		}
	}

	dql := fmt.Sprintf(`fetch logs, from: "%s", to: "%s"`, from, to)
	if filterExpr != "" {
		dql += " | filter " + filterExpr
	}
	dql += " | sort timestamp desc"
	dql += fmt.Sprintf(" | limit %d", limit)
	return dql, nil
}

// buildDQLWhereClause is a log-specific wrapper around buildDQLWhereClauseWithMapping.
func buildDQLWhereClause(where query.QueryWhereClause) (string, error) {
	return buildDQLWhereClauseWithMapping(where, dynatraceLogLabelMapping, nil)
}

// buildDQLWhereClauseWithMapping converts a QueryWhereClause to a DQL filter expression.
// labelMapping maps NudgeBee field names to their Grail DQL equivalents.
// uidFields is the set of DQL field names (after mapping) whose equality comparisons require touid().
// Returns an empty string (no error) when the clause is empty.
func buildDQLWhereClauseWithMapping(where query.QueryWhereClause, labelMapping map[string]string, uidFields map[string]bool) (string, error) {
	if len(where.Binary) > 0 {
		return buildDQLBinaryClauseWithMapping(where.Binary, labelMapping, uidFields)
	}

	if len(where.And) > 0 {
		return buildDQLBooleanClauseWithMapping(where.And, "AND", labelMapping, uidFields)
	}

	if len(where.Or) > 0 {
		return buildDQLBooleanClauseWithMapping(where.Or, "OR", labelMapping, uidFields)
	}

	if where.Not != nil {
		notPart, err := buildDQLWhereClauseWithMapping(*where.Not, labelMapping, uidFields)
		if err != nil {
			return "", err
		}
		if notPart != "" {
			return "not(" + notPart + ")", nil
		}
	}

	return "", nil
}

// buildDQLBooleanClauseWithMapping joins multiple where clauses with the given operator ("AND" or "OR").
func buildDQLBooleanClauseWithMapping(clauses []query.QueryWhereClause, operator string, labelMapping map[string]string, uidFields map[string]bool) (string, error) {
	var parts []string
	for _, c := range clauses {
		part, err := buildDQLWhereClauseWithMapping(c, labelMapping, uidFields)
		if err != nil {
			return "", err
		}
		if part != "" {
			parts = append(parts, part)
		}
	}
	if len(parts) == 0 {
		return "", nil
	}
	if len(parts) == 1 {
		return parts[0], nil
	}
	return "(" + strings.Join(parts, " "+operator+" ") + ")", nil
}

// buildDQLBinaryClauseWithMapping handles individual field comparisons, applying labelMapping
// for standard NudgeBee field names and wrapping uidFields equality values in touid().
func buildDQLBinaryClauseWithMapping(binary query.BinaryWhereClause, labelMapping map[string]string, uidFields map[string]bool) (string, error) {
	var parts []string
	for field, ops := range binary {
		if mapped, ok := labelMapping[field]; ok {
			field = mapped
		}
		for op, val := range ops {
			var clause string
			var err error
			if uidFields != nil && uidFields[field] && op == query.Eq {
				// uid-typed fields in Grail require touid() for equality comparison.
				clause = fmt.Sprintf(`%s == touid("%s")`, field, escapeDQLValue(val))
			} else {
				clause, err = buildDQLOperatorClause(field, op, val)
				if err != nil {
					return "", err
				}
			}
			if clause != "" {
				parts = append(parts, clause)
			}
		}
	}
	return strings.Join(parts, " AND "), nil
}

// buildDQLOperatorClause builds a single DQL comparison expression for a field/operator/value triple.
func buildDQLOperatorClause(field string, op query.BinaryWhereClauseType, val any) (string, error) {
	switch op {
	case query.Eq:
		if isDQLNumeric(val) {
			return fmt.Sprintf("%s == %v", field, val), nil
		}
		return fmt.Sprintf(`%s == "%s"`, field, escapeDQLValue(val)), nil

	case query.Nq:
		if isDQLNumeric(val) {
			return fmt.Sprintf("%s != %v", field, val), nil
		}
		return fmt.Sprintf(`%s != "%s"`, field, escapeDQLValue(val)), nil

	case query.Gt:
		if isDQLNumeric(val) {
			return fmt.Sprintf("%s > %v", field, val), nil
		}
		return fmt.Sprintf(`%s > "%s"`, field, escapeDQLValue(val)), nil

	case query.Lt:
		if isDQLNumeric(val) {
			return fmt.Sprintf("%s < %v", field, val), nil
		}
		return fmt.Sprintf(`%s < "%s"`, field, escapeDQLValue(val)), nil

	case query.Gte:
		if isDQLNumeric(val) {
			return fmt.Sprintf("%s >= %v", field, val), nil
		}
		return fmt.Sprintf(`%s >= "%s"`, field, escapeDQLValue(val)), nil

	case query.Lte:
		if isDQLNumeric(val) {
			return fmt.Sprintf("%s <= %v", field, val), nil
		}
		return fmt.Sprintf(`%s <= "%s"`, field, escapeDQLValue(val)), nil

	case query.In:
		arr, ok := val.([]any)
		if !ok {
			// single non-array value — degrade to equality
			if isDQLNumeric(val) {
				return fmt.Sprintf("%s == %v", field, val), nil
			}
			return fmt.Sprintf(`%s == "%s"`, field, escapeDQLValue(val)), nil
		}
		if len(arr) == 0 {
			return "", fmt.Errorf("IN operator requires non-empty array for field %q", field)
		}
		if len(arr) == 1 {
			if isDQLNumeric(arr[0]) {
				return fmt.Sprintf("%s == %v", field, arr[0]), nil
			}
			return fmt.Sprintf(`%s == "%s"`, field, escapeDQLValue(arr[0])), nil
		}
		var vals []string
		for _, v := range arr {
			if isDQLNumeric(v) {
				vals = append(vals, fmt.Sprintf("%v", v))
			} else {
				vals = append(vals, fmt.Sprintf(`"%s"`, escapeDQLValue(v)))
			}
		}
		return fmt.Sprintf("in(%s, %s)", field, strings.Join(vals, ", ")), nil

	case query.NotIn:
		arr, ok := val.([]any)
		if !ok {
			return "", fmt.Errorf("NOT IN operator requires array value for field %q", field)
		}
		var vals []string
		for _, v := range arr {
			if isDQLNumeric(v) {
				vals = append(vals, fmt.Sprintf("%v", v))
			} else {
				vals = append(vals, fmt.Sprintf(`"%s"`, escapeDQLValue(v)))
			}
		}
		return fmt.Sprintf("not(in(%s, %s))", field, strings.Join(vals, ", ")), nil

	case query.Like, query.NLike:
		// Strip SQL-style % wildcards; use DQL contains() for case-sensitive substring match.
		s := strings.Trim(fmt.Sprintf("%v", val), "%")
		if op == query.NLike {
			return fmt.Sprintf(`not(contains(%s, "%s"))`, field, escapeDQLValue(s)), nil
		}
		return fmt.Sprintf(`contains(%s, "%s")`, field, escapeDQLValue(s)), nil

	case query.ILike, query.IContains:
		// Case-insensitive: use DQL matchesPhrase().
		s := strings.Trim(fmt.Sprintf("%v", val), "%")
		return fmt.Sprintf(`matchesPhrase(%s, "%s")`, field, escapeDQLValue(s)), nil

	case query.Contains:
		return fmt.Sprintf(`contains(%s, "%s")`, field, escapeDQLValue(val)), nil

	case query.Regex:
		return fmt.Sprintf(`matches(%s, "%s")`, field, escapeDQLValue(val)), nil

	case query.NRegex:
		return fmt.Sprintf(`not(matches(%s, "%s"))`, field, escapeDQLValue(val)), nil

	default:
		return "", fmt.Errorf("unsupported DQL operator %q for field %q", op, field)
	}
}

// isDQLNumeric reports whether val is a numeric type (or a string that parses as a number)
// that should be rendered without quotes in DQL.
func isDQLNumeric(val any) bool {
	switch v := val.(type) {
	case int, int8, int16, int32, int64,
		uint, uint8, uint16, uint32, uint64,
		float32, float64:
		return true
	case string:
		_, err := strconv.ParseFloat(v, 64)
		return err == nil
	}
	return false
}

// QueryLabels returns a fixed set of common Dynatrace Grail log field names.
// Dynatrace Grail does not expose a dedicated label enumeration endpoint.
// dynatraceLogLabelsFallback is returned when the autocomplete endpoint is unavailable.
var dynatraceLogLabelsFallback = []OutputLogLabel{
	{Label: "status", Attributes: map[string]any{}},
	{Label: "host.name", Attributes: map[string]any{}},
	{Label: "service.name", Attributes: map[string]any{}},
	{Label: "k8s.namespace.name", Attributes: map[string]any{}},
	{Label: "k8s.pod.name", Attributes: map[string]any{}},
	{Label: "k8s.container.name", Attributes: map[string]any{}},
	{Label: "k8s.cluster.name", Attributes: map[string]any{}},
}

// fetchDynatraceLogLabelsViaAutocomplete returns available Dynatrace log field names
// by querying the DQL autocomplete endpoint with "fetch logs | filter ".
// Only suggestions whose parts are all of type SIMPLE_IDENTIFIER are log field names.
func fetchDynatraceLogLabelsViaAutocomplete(baseURL, apiToken string) ([]OutputLogLabel, error) {
	autocompleteURL := strings.TrimRight(baseURL, "/") + "/platform/storage/query/v1/query:autocomplete"
	headers := map[string]string{
		"Authorization": "Bearer " + apiToken,
		"Content-Type":  "application/json",
		"Accept":        "application/json",
	}
	queryStr := "fetch logs | filter "
	body := dtAutocompleteRequest{Query: queryStr, Position: len(queryStr)}

	res, err := common.HttpPost(autocompleteURL,
		common.HttpWithHeaders(headers),
		common.HttpWithJsonBody(body),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to call Dynatrace autocomplete endpoint: %w", err)
	}
	bodyBytes, readErr := io.ReadAll(res.Body)
	_ = res.Body.Close()
	if readErr != nil {
		return nil, fmt.Errorf("failed to read Dynatrace autocomplete response: %w", readErr)
	}
	if res.StatusCode == 403 {
		return nil, errDtForbidden
	}
	if res.StatusCode != 200 {
		return nil, fmt.Errorf("dynatrace autocomplete returned status %d: %s", res.StatusCode, string(bodyBytes))
	}

	var resp dtAutocompleteResponse
	if err := common.UnmarshalJson(bodyBytes, &resp); err != nil {
		return nil, fmt.Errorf("failed to unmarshal Dynatrace autocomplete response: %w", err)
	}

	output := make([]OutputLogLabel, 0, len(resp.Suggestions))
	for _, s := range resp.Suggestions {
		if s.Suggestion == "" || len(s.Parts) == 0 {
			continue
		}
		allIdentifier := true
		for _, p := range s.Parts {
			if p.Type != "SIMPLE_IDENTIFIER" {
				allIdentifier = false
				break
			}
		}
		if allIdentifier {
			output = append(output, OutputLogLabel{Label: s.Suggestion, Attributes: map[string]any{}})
		}
	}
	return output, nil
}

func (s *DynatraceLogSource) QueryLabels(ctx *security.RequestContext, req FetchLogLabelRequest) ([]OutputLogLabel, error) {
	apiToken, baseURL, err := integrations.GetDynatraceConfigs(ctx, req.AccountId)
	if err == nil {
		labels, autoErr := fetchDynatraceLogLabelsViaAutocomplete(baseURL, apiToken)
		if autoErr == nil {
			return labels, nil
		}
		if ctx != nil {
			ctx.GetLogger().Warn("DynatraceLogSource.QueryLabels: autocomplete failed, using static list", "error", autoErr)
		}
	}
	return dynatraceLogLabelsFallback, nil
}

// QueryLabelValues samples recent logs and returns distinct values for the given field.
func (s *DynatraceLogSource) QueryLabelValues(ctx *security.RequestContext, req FetchLogLabelValuesRequest) ([]OutputLogLabelValue, error) {
	apiToken, baseURL, err := integrations.GetDynatraceConfigs(ctx, req.AccountId)
	if err != nil {
		return nil, fmt.Errorf("failed to get Dynatrace configs: %w", err)
	}

	labelName := req.LabelName
	if mapped, ok := dynatraceLogLabelMapping[labelName]; ok {
		labelName = mapped
	}

	from, to := s.getTimeRange(req.StartTime, req.EndTime)
	dql := fmt.Sprintf(`fetch logs, from: "%s", to: "%s" | sort timestamp desc | limit 500`, from, to)

	result, err := executeDQLQuery(baseURL, apiToken, dql)
	if err != nil {
		return nil, fmt.Errorf("dynatrace logs label values query failed: %w", err)
	}

	seen := make(map[string]bool)
	var values []OutputLogLabelValue
	for _, record := range result.Records {
		val := grailStr(record, labelName)
		if val != "" && !seen[val] {
			seen[val] = true
			values = append(values, OutputLogLabelValue{Value: val, Attributes: map[string]any{}})
			if len(values) >= 50 {
				break
			}
		}
	}
	return values, nil
}

// GetQuery returns the full Dynatrace Grail DQL query for the request.
func (s *DynatraceLogSource) GetQuery(_ *security.RequestContext, req FetchLogRequest) (string, error) {
	limit := req.Limit
	if limit <= 0 || limit > 1000 {
		limit = 1000
	}
	from, to := s.getTimeRange(req.StartTime, req.EndTime)
	return s.buildLogDQL(req, from, to, limit)
}

// GetLabelMapping returns the field name mapping from NudgeBee names to Dynatrace Grail field names.
func (s *DynatraceLogSource) GetLabelMapping() map[string]string {
	return dynatraceLogLabelMapping
}

func (s *DynatraceLogSource) GetSupportedOperators() []string {
	return []string{"_eq", "_neq", "_gt", "_lt", "_gte", "_lte", "_like", "_nlike", "_ilike", "_icontains", "_contains", "_regex", "_nregex"}
}

// convertToOutputLogs normalizes Dynatrace Grail log records to NudgeBee's OutputLog format.
//
// Mapping:
//
//	record["timestamp"]  → OutputLog.Timestamp
//	record["content"]    → OutputLog.Message
//	record["status"]     → OutputLog.Severity (fall back to "loglevel")
//	all record fields    → OutputLog.Labels
func (s *DynatraceLogSource) convertToOutputLogs(records []map[string]any) []OutputLog {
	logs := make([]OutputLog, 0, len(records))
	for _, record := range records {
		log := OutputLog{
			Timestamp: grailStr(record, "timestamp"),
			Message:   grailStr(record, "content"),
			Labels:    make(map[string]any, len(record)),
		}

		// Severity: prefer "status", fall back to "loglevel"
		if status := grailStr(record, "status"); status != "" {
			log.Severity = status
		} else if loglevel := grailStr(record, "loglevel"); loglevel != "" {
			log.Severity = loglevel
		}

		// Flatten all record fields into labels
		for k, v := range record {
			log.Labels[k] = v
		}

		logs = append(logs, log)
	}
	return logs
}

// getTimeRange converts epoch ms timestamps to ISO 8601 strings required by the Dynatrace Grail API.
func (s *DynatraceLogSource) getTimeRange(startTime, endTime int64) (string, string) {
	return dynatraceTimeRange(startTime, endTime)
}

// dynatraceTimeRange converts epoch timestamps to ISO 8601 strings for Dynatrace Grail API calls.
// Accepts epoch-seconds (<1e12), epoch-ms (1e12–1e15), or epoch-µs (>1e15) and normalizes to ms.
// Defaults to the last hour when a value is zero.
func dynatraceTimeRange(startTime, endTime int64) (string, string) {
	if startTime == 0 {
		startTime = time.Now().Add(-1 * time.Hour).UnixMilli()
	}
	if endTime == 0 {
		endTime = time.Now().UnixMilli()
	}
	// Normalize to milliseconds.
	if startTime < 1e12 {
		startTime *= 1000 // seconds → ms
	} else if startTime > 1e15 {
		startTime /= 1000 // microseconds → ms
	}
	if endTime < 1e12 {
		endTime *= 1000 // seconds → ms
	} else if endTime > 1e15 {
		endTime /= 1000 // microseconds → ms
	}
	return time.UnixMilli(startTime).UTC().Format(time.RFC3339),
		time.UnixMilli(endTime).UTC().Format(time.RFC3339)
}

// ---------------------------------------------------------------------------
// DynatraceLogGroupSource — implements LogGroupSource for Dynatrace Grail
// ---------------------------------------------------------------------------

// DynatraceLogGroupSource queries Dynatrace Grail for aggregated error/critical
// log patterns grouped by message, namespace, workload, and container.
type DynatraceLogGroupSource struct{}

// QueryLogGroup fetches log group aggregations from Dynatrace Grail using DQL summarize.
func (s *DynatraceLogGroupSource) QueryLogGroup(ctx *security.RequestContext, req FetchLogGroupRequest) (LogGroupOutput, error) {
	apiToken, baseURL, err := integrations.GetDynatraceConfigs(ctx, req.AccountId)
	if err != nil {
		ctx.GetLogger().Error("DynatraceLogGroupSource.QueryLogGroup: failed to get configs", "error", err)
		return LogGroupOutput{}, fmt.Errorf("failed to get Dynatrace configs: %w", err)
	}

	from, to := dynatraceTimeRange(req.StartTime, req.EndTime)
	dql := s.buildLogGroupDQL(req, from, to)
	ctx.GetLogger().Info("DynatraceLogGroupSource.QueryLogGroup", "dql", dql)

	result, err := executeDQLQuery(baseURL, apiToken, dql)
	if err != nil {
		ctx.GetLogger().Error("DynatraceLogGroupSource.QueryLogGroup: DQL query failed", "error", err)
		if errors.Is(err, integrations.ErrDQLQueryTimeout) {
			return LogGroupOutput{}, fmt.Errorf(
				"log group query timed out — the selected time range contains too many logs. " +
					"Please apply more filters: select a specific Namespace or Workload to narrow the scope",
			)
		}
		return LogGroupOutput{}, fmt.Errorf("dynatrace log group query failed: %w", err)
	}

	// Use the end of the query window as the single timestamp for all groups.
	// Return epoch-seconds to match all other log group providers (frontend multiplies by 1000).
	var endTimeSec int64
	if req.EndTime <= 0 {
		endTimeSec = time.Now().Unix()
	} else if req.EndTime >= 1e12 {
		// EndTime is in milliseconds — convert to seconds.
		endTimeSec = req.EndTime / 1000
	} else {
		endTimeSec = req.EndTime
	}

	groups := make([]LogGroup, 0, len(result.Records))
	for _, record := range result.Records {
		r, ok := s.convertToLogGroup(record, endTimeSec)
		if ok {
			groups = append(groups, r)
		}
	}

	return LogGroupOutput{Groups: groups}, nil
}

// buildLogGroupDQL constructs the DQL aggregation query for error/critical log groups.
// Optional filters for selectedNamespace and selectedWorkload are appended when present.
func (s *DynatraceLogGroupSource) buildLogGroupDQL(req FetchLogGroupRequest, from, to string) string {
	var sb strings.Builder

	fmt.Fprintf(&sb, `fetch logs, from: "%s", to: "%s"`, from, to)

	// Severity filter: match on status OR loglevel; fall back to message-content patterns
	// when both fields are absent (handles non-DT-normalised log sources).
	sb.WriteString("\n| filter (")
	sb.WriteString("in(`status`, \"ERROR\", \"CRITICAL\", \"error\", \"critical\", \"Error\", \"Critical\", \"fatal\", \"FATAL\")")
	sb.WriteString(" OR in(`loglevel`, \"error\", \"ERROR\", \"critical\", \"CRITICAL\", \"fatal\", \"FATAL\")")
	sb.WriteString(" OR ((isNull(`status`) OR `status` == \"\") AND (isNull(`loglevel`) OR `loglevel` == \"\")")
	sb.WriteString(" AND (matchesPhrase(content, \"ERROR\") OR matchesPhrase(content, \"FATAL\") OR matchesPhrase(content, \"CRITICAL\"))))")

	// Always exclude known system/infrastructure containers.
	sb.WriteString("\n| filter not(in(`k8s.container.name`, \"prometheus\", \"grafana\", \"nudgebee-agent\"))")

	// Optional namespace filter.
	if ns := common.GetString(req.Request, "selectedNamespace"); ns != "" {
		fmt.Fprintf(&sb, "\n| filter `k8s.namespace.name` == \"%s\"", escapeDQLValue(ns))
	}

	// Optional workload filter: exact match on workload name when present; fall back to
	// pod name substring only when k8s.workload.name is absent — avoids false positives
	// where a pod name substring matches but the workload label points to a different workload.
	if wl := common.GetString(req.Request, "selectedWorkload"); wl != "" {
		escaped := escapeDQLValue(wl)
		fmt.Fprintf(&sb,
			"\n| filter `k8s.workload.name` == \"%s\" OR (isNull(`k8s.workload.name`) AND matchesPhrase(`k8s.pod.name`, \"%s\"))",
			escaped, escaped,
		)
	}

	// Aggregate: one row per unique (message, namespace, workload, container) combination.
	// Using k8s.workload.name (not pod name) ensures all replicas of a workload are merged
	// natively by DQL without any post-processing deduplication.
	sb.WriteString("\n| summarize value = count(), by: {content, `k8s.namespace.name`, `k8s.workload.name`, `k8s.container.name`}")
	sb.WriteString("\n| sort value desc")
	sb.WriteString("\n| limit 100")

	return sb.String()
}

// convertToLogGroup converts a single DQL summarize record into a LogGroup.
// Returns (LogGroup{}, false) for records that are missing required fields.
func (s *DynatraceLogGroupSource) convertToLogGroup(record map[string]any, endTimeSec int64) (LogGroup, bool) {
	// count is a float64 from DQL summarize — must NOT use grailStr.
	// Some DQL responses may return the value as a string; handle both.
	var count float64
	switch v := record["value"].(type) {
	case float64:
		count = v
	case string:
		parsed, err := strconv.ParseFloat(v, 64)
		if err != nil {
			return LogGroup{}, false
		}
		count = parsed
	default:
		return LogGroup{}, false
	}

	// Skip records without a log message — they produce useless empty log groups.
	content := grailStr(record, "content")
	if content == "" {
		return LogGroup{}, false
	}

	// Unicode-safe truncation for display sample (keeps hashes stable by using full content).
	sample := content
	if runes := []rune(content); len(runes) > 500 {
		sample = string(runes[:500])
	}

	namespace := grailStr(record, "k8s.namespace.name")
	workload := grailStr(record, "k8s.workload.name")
	container := grailStr(record, "k8s.container.name")

	// Build Prometheus-compatible container_id for ticket reference linking.
	containerID := ""
	if namespace != "" && workload != "" {
		containerID = fmt.Sprintf("/k8s/%s/%s", namespace, workload)
	}

	patternHash := generatePatternHash(content)

	// Extract severity: prefer "status", fall back to "loglevel", default to "error".
	level := grailStr(record, "status")
	if level == "" {
		level = grailStr(record, "loglevel")
	}
	if level == "" {
		level = "error"
	}

	return LogGroup{
		Sample:      sample,
		Namespace:   namespace,
		Workload:    workload,
		Container:   container,
		ContainerID: containerID,
		PatternHash: patternHash,
		Level:       level,
		Count:       int64(math.Round(count)),
		Timestamps:  []int64{endTimeSec},
		Values:      []float64{count},
	}, true
}
