package parquet

import (
	"strings"
	"time"

	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/types"
)

// LogRow matches o11y.general_logs Parquet data columns.
// component is a Glue partition key only (path component=...); do NOT write it into
// Parquet — Athena/Hive binds SELECT component from the injected partition path.
// tenant_id is a data column only (not a Glue partition).
type LogRow struct {
	EventTime string `parquet:"event_time"`
	Timestamp int64  `parquet:"timestamp"`
	Host      string `parquet:"host"`
	Pod       string `parquet:"pod"`
	Level     string `parquet:"level"`
	Position  string `parquet:"position"`
	Message   string `parquet:"message"`
	Cluster   string `parquet:"cluster"`
	Instance  string `parquet:"instance"`
	TenantID  string `parquet:"tenant_id"`
}

func LogToRow(item *types.Log, cluster, tenantID string) LogRow {
	host, pod := vectorHostPod(item.RawLine, item.Instance)
	return LogRow{
		EventTime: formatEventTime(item),
		Timestamp: item.Timestamp,
		Host:      host,
		Pod:       pod,
		Level:     item.Level,
		Position:  item.Position,
		Message:   item.Message,
		Cluster:   cluster,
		Instance:  item.Instance,
		TenantID:  tenantID,
	}
}

// formatEventTime prefers full-precision EventTime; falls back to unix-second Timestamp.
func formatEventTime(item *types.Log) string {
	if !item.EventTime.IsZero() {
		return item.EventTime.UTC().Format(time.RFC3339Nano)
	}
	return time.Unix(item.Timestamp, 0).UTC().Format(time.RFC3339Nano)
}

func vectorHostPod(rawLine, fallbackInstance string) (host, pod string) {
	pod = fallbackInstance
	line := strings.TrimSpace(rawLine)
	if line == "" {
		return "", pod
	}
	header := strings.SplitN(line, " ", 2)[0]
	parts := strings.SplitN(header, ";", 5)
	if len(parts) >= 4 {
		host = parts[1]
		if parts[3] != "" {
			pod = parts[3]
		}
	}
	return host, pod
}