package indexer

import (
    "context"
    "fmt"
    "sync"
)

// Document representa un documento a procesar por el pipeline de IA.
type Document struct {
    ID      string
    Content string
}

// IndexResult encapsula el resultado (embedding + score) o un error.
type IndexResult struct {
    DocID     string
    Embedding []float32
    Score     float64
    Err       error
}

// RunPool lanza numWorkers goroutines que consumen docs y escriben en results.
func RunPool(ctx context.Context, numWorkers int, docs <-chan Document) <-chan IndexResult {
    results := make(chan IndexResult, numWorkers*2)

    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(workerID int) {
            defer wg.Done()
            for doc := range docs {
                select {
                case <-ctx.Done():
                    return // contexto cancelado: salir limpiamente
                default:
                    result := embedDocument(doc)
                    results <- result
                }
            }
        }(i)
    }

    // Cerrar results cuando TODOS los workers terminen.
    go func() {
        wg.Wait()
        close(results)
    }()

    return results
}

// embedDocument llama al microservicio de embeddings (simplificado).
func embedDocument(doc Document) IndexResult {
    // En producción: llamada gRPC al embedder con timeout propio.
    return IndexResult{
        DocID:     doc.ID,
        Embedding: []float32{0.12, 0.87, 0.34},
        Score:     0.95,
    }
}

// Uso desde main o handler:
//   docs := make(chan Document, 500)
//   go feedDocs(docs, batch)            // productor cierra el canal
//   for r := range RunPool(ctx, 8, docs) {
//       store(r)                         // consumidor
//   }

package inference

import (
    "context"
    "sync"
)

type ServiceResult struct {
    Service string
    Payload any
    Err     error
}

// callService simula una llamada gRPC a un microservicio.
func callService(ctx context.Context, name string, docID string) <-chan ServiceResult {
    out := make(chan ServiceResult, 1)
    go func() {
        defer close(out)
        select {
        case <-ctx.Done():
            out <- ServiceResult{Service: name, Err: ctx.Err()}
        default:
            // Llamada real al servicio (omitida para claridad)
            out <- ServiceResult{Service: name, Payload: map[string]any{"doc": docID}}
        }
    }()
    return out
}

// merge fusiona N canales en uno solo (fan-in).
func merge(ctx context.Context, channels ...<-chan ServiceResult) <-chan ServiceResult {
    var wg sync.WaitGroup
    out := make(chan ServiceResult, len(channels))

    forward := func(ch <-chan ServiceResult) {
        defer wg.Done()
        for r := range ch {
            select {
            case <-ctx.Done():
                return
            case out <- r:
            }
        }
    }

    wg.Add(len(channels))
    for _, ch := range channels {
        go forward(ch)
    }

    go func() { wg.Wait(); close(out) }()
    return out
}

// ProcessDocument hace fan-out a los 3 servicios y fan-in de resultados.
func ProcessDocument(ctx context.Context, docID string) []ServiceResult {
    // Fan-out: lanzar los 3 servicios simultáneamente
    embedCh  := callService(ctx, "embedder",   docID)
    classCh  := callService(ctx, "classifier", docID)
    rerankCh := callService(ctx, "reranker",   docID)

    // Fan-in: recoger todos los resultados
    var results []ServiceResult
    for r := range merge(ctx, embedCh, classCh, rerankCh) {
        results = append(results, r)
    }
    return results
}

package main

import (
    "context"
    "fmt"
    "log/slog"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)

type Engine struct {
    wg sync.WaitGroup
}

func (e *Engine) Start(ctx context.Context) {
    for i := 0; i < 8; i++ {
        e.wg.Add(1)
        go e.inferenceWorker(ctx, i)
    }
}

func (e *Engine) inferenceWorker(ctx context.Context, id int) {
    defer e.wg.Done()
    slog.Info("worker started", "id", id)

    ticker := time.NewTicker(200 * time.Millisecond)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            // Cleanup: flush buffer, cerrar conexión gRPC, etc.
            slog.Info("worker draining", "id", id)
            time.Sleep(100 * time.Millisecond) // flush simulado
            slog.Info("worker stopped", "id", id)
            return
        case <-ticker.C:
            // Procesar siguiente documento de la cola
            fmt.Printf("[w%d] processing...\n", id)
        }
    }
}

func main() {
    // Contexto raíz cancelable por señal del SO
    ctx, stop := signal.NotifyContext(
        context.Background(),
        syscall.SIGINT, syscall.SIGTERM,
    )
    defer stop()

    engine := &Engine{}
    engine.Start(ctx)

    slog.Info("cultivaengine running, waiting for signal...")
    <-ctx.Done() // bloquea hasta SIGTERM / SIGINT

    slog.Info("signal received, shutting down (max 30s)")

    // Esperar workers con timeout de 30s (margen K8s)
    done := make(chan struct{})
    go func() { engine.wg.Wait(); close(done) }()

    select {
    case <-done:
        slog.Info("clean shutdown completed")
    case <-time.After(30 * time.Second):
        slog.Warn("shutdown timeout, forcing exit")
        os.Exit(1)
    }
}

package llm

import (
    "context"
    "fmt"
    "golang.org/x/sync/semaphore"
    "sync"
)

// LLMClient llama a una API externa con concurrencia limitada.
type LLMClient struct {
    sem *semaphore.Weighted
    mu  sync.Mutex
}

// NewLLMClient crea un cliente con maxConcurrent llamadas simultáneas.
func NewLLMClient(maxConcurrent int64) *LLMClient {
    return &LLMClient{
        sem: semaphore.NewWeighted(maxConcurrent),
    }
}

// Complete envía un prompt al LLM respetando el rate-limit.
func (c *LLMClient) Complete(ctx context.Context, prompt string) (string, error) {
    // Adquirir semáforo — bloquea si ya hay maxConcurrent llamadas activas
    if err := c.sem.Acquire(ctx, 1); err != nil {
        return "", fmt.Errorf("semaphore acquire: %w", err)
    }
    defer c.sem.Release(1) // liberar siempre, incluso con panic

    // Llamada real a la API (gRPC / HTTP)
    response, err := callExternalAPI(ctx, prompt)
    if err != nil {
        return "", fmt.Errorf("llm api: %w", err)
    }
    return response, nil
}

// BatchComplete procesa prompts concurrentemente con rate-limit aplicado.
func (c *LLMClient) BatchComplete(ctx context.Context, prompts []string) ([]string, []error) {
    results := make([]string, len(prompts))
    errors  := make([]error,  len(prompts))

    var wg sync.WaitGroup
    for i, p := range prompts {
        i, p := i, p
        wg.Add(1)
        go func() {
            defer wg.Done()
            results[i], errors[i] = c.Complete(ctx, p)
        }()
    }
    wg.Wait()
    return results, errors
}

func callExternalAPI(ctx context.Context, prompt string) (string, error) {
    // Stub: en producción, http.Post a api.openai.com con ctx
    return "respuesta generada para: " + prompt, nil
}

// Instanciación recomendada para CULTIVA Engine:
//   client := llm.NewLLMClient(10)  // 10 req concurrentes máx
//   responses, errs := client.BatchComplete(ctx, prompts)

package health

import (
    "context"
    "fmt"
    "golang.org/x/sync/errgroup"
    "time"
)

type Dependency struct {
    Name    string
    CheckFn func(ctx context.Context) error
}

// CheckAll verifica todas las dependencias en paralelo con timeout de 5s.
// Devuelve el primer error encontrado; cancela las demás goroutines.
func CheckAll(parentCtx context.Context, deps []Dependency) error {
    ctx, cancel := context.WithTimeout(parentCtx, 5*time.Second)
    defer cancel()

    // errgroup.WithContext cancela ctx automáticamente al primer error
    g, ctx := errgroup.WithContext(ctx)

    for _, dep := range deps {
        dep := dep // capturar variable del loop (pre-Go 1.22)
        g.Go(func() error {
            if err := dep.CheckFn(ctx); err != nil {
                return fmt.Errorf("%s unhealthy: %w", dep.Name, err)
            }
            return nil
        })
    }

    // Esperar: si alguna goroutine falla, g.Wait() devuelve ese error
    return g.Wait()
}

// Configuración en main / handler HTTP:
var cultivaDeps = []Dependency{
    {Name: "redis",      CheckFn: checkRedis},
    {Name: "postgres",   CheckFn: checkPostgres},
    {Name: "embedder",   CheckFn: checkEmbedderGRPC},
    {Name: "classifier", CheckFn: checkClassifierGRPC},
}

// HTTP handler para /healthz (K8s readiness probe)
func ReadinessHandler(w http.ResponseWriter, r *http.Request) {
    if err := CheckAll(r.Context(), cultivaDeps); err != nil {
        http.Error(w, err.Error(), http.StatusServiceUnavailable)
        return
    }
    w.WriteHeader(http.StatusOK)
    w.Write([]byte("ok"))
}