package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/go-chi/chi/v5/middleware"
	"github.com/segmentio/kafka-go"
)

type Event struct {
	EventID         string                 `json:"event_id"`
	EventType       string                 `json:"event_type"`
	Timestamp       time.Time              `json:"timestamp"`
	UserID          string                 `json:"user_id,omitempty"`
	UserEmail       string                 `json:"user_email,omitempty"`
	CompanyID       string                 `json:"company_id,omitempty"`
	JobID           string                 `json:"job_id,omitempty"`
	Amount          float64                `json:"amount,omitempty"`
	Currency        string                 `json:"currency,omitempty"`
	Status          string                 `json:"status"`
	Country         string                 `json:"country,omitempty"`
	IPAddress       string                 `json:"ip_address,omitempty"`
	UserAgent       string                 `json:"user_agent,omitempty"`
	Metadata        map[string]interface{} `json:"metadata,omitempty"`
	RiskScore       float64                `json:"risk_score,omitempty"`
	AnomalyDetected bool                   `json:"anomaly_detected,omitempty"`
}

type EventStore struct {
	kafkaWriter *kafka.Writer
}

func NewEventStore(brokers []string) *EventStore {
	// Use explicit dialer to avoid DNS issues
	dialer := &kafka.Dialer{
		Timeout:   10 * time.Second,
		DualStack: true,
	}
	
	return &EventStore{
		kafkaWriter: &kafka.Writer{
			Addr:     kafka.TCP(brokers...),
			Topic:    "quixzoom.events",
			Balancer: &kafka.LeastBytes{},
			Dialer:   dialer,
		},
	}
}

func (s *EventStore) PublishEvent(ctx context.Context, event Event) error {
	event.Timestamp = time.Now().UTC()
	if event.EventID == "" {
		event.EventID = fmt.Sprintf("evt_%d", time.Now().UnixNano())
	}

	data, err := json.Marshal(event)
	if err != nil {
		return err
	}

	return s.kafkaWriter.WriteMessages(ctx, kafka.Message{
		Key:   []byte(event.EventType),
		Value: data,
	})
}

func (s *EventStore) Close() error {
	return s.kafkaWriter.Close()
}

func main() {
	brokers := os.Getenv("KAFKA_BROKERS")
	if brokers == "" {
		brokers = "172.24.0.5:29092"
	}

	store := NewEventStore([]string{brokers})
	defer store.Close()

	r := chi.NewRouter()
	r.Use(middleware.Logger)
	r.Use(middleware.Recoverer)
	r.Use(middleware.RequestID)

	// Health check
	r.Get("/health", func(w http.ResponseWriter, r *http.Request) {
		json.NewEncoder(w).Encode(map[string]string{"status": "ok"})
	})

	// Ingest event
	r.Post("/api/v1/events", func(w http.ResponseWriter, r *http.Request) {
		var event Event
		if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
			http.Error(w, `{"error":"invalid json"}`, http.StatusBadRequest)
			return
		}

		if event.EventType == "" {
			http.Error(w, `{"error":"event_type required"}`, http.StatusBadRequest)
			return
		}

		// Anomaly detection (simple rule-based)
		if event.Amount > 100000 {
			event.RiskScore = 0.8
			event.AnomalyDetected = true
		}

		ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
		defer cancel()

		if err := store.PublishEvent(ctx, event); err != nil {
			log.Printf("Error publishing event: %v", err)
			http.Error(w, `{"error":"failed to publish"}`, http.StatusInternalServerError)
			return
		}

		w.WriteHeader(http.StatusCreated)
		json.NewEncoder(w).Encode(map[string]string{
			"status":   "published",
			"event_id": event.EventID,
		})
	})

	// Search events (placeholder - would query Elasticsearch)
	r.Get("/api/v1/events/search", func(w http.ResponseWriter, r *http.Request) {
		query := r.URL.Query().Get("q")
		eventType := r.URL.Query().Get("type")
		userID := r.URL.Query().Get("user_id")

		// TODO: Query Elasticsearch
		json.NewEncoder(w).Encode(map[string]interface{}{
			"query":      query,
			"type":       eventType,
			"user_id":    userID,
			"results":    []Event{},
			"total":      0,
			"note":       "Elasticsearch integration pending",
		})
	})

	port := os.Getenv("PORT")
	if port == "" {
		port = "9097"
	}

	log.Printf("Event streaming server starting on :%s", port)
	log.Fatal(http.ListenAndServe(":"+port, r))
}