descrybe/apps/api/internal/processing/pipeline_process_test.go

package processing

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgconn"
)

func TestStopOnCancel(t *testing.T) {
	if err := stopOnCancel(false, nil); err != nil {
		t.Fatalf("continue: %v", err)
	}
	if err := stopOnCancel(true, nil); !errors.Is(err, errJobCancelled) {
		t.Fatalf("cancelled: %v", err)
	}
	lookup := errors.New("db down")
	if err := stopOnCancel(false, lookup); !errors.Is(err, lookup) {
		t.Fatalf("lookup err: %v", err)
	}
	// Prefer fail-closed on lookup error even if cancelled was true.
	if err := stopOnCancel(true, lookup); !errors.Is(err, lookup) {
		t.Fatalf("prefer lookup err: %v", err)
	}
}

func TestMarshalStepProgress_roundTrip(t *testing.T) {
	progress := InitialStepProgress("full")
	if len(progress) == 0 {
		t.Fatal("expected steps")
	}
	progress[0].Status = "running"
	b, err := marshalStepProgress(progress)
	if err != nil {
		t.Fatal(err)
	}
	var got []StepProgress
	if err := json.Unmarshal(b, &got); err != nil {
		t.Fatal(err)
	}
	if len(got) != len(progress) || got[0].Status != "running" {
		t.Fatalf("got=%v", got)
	}

	b, err = marshalStepProgress(nil)
	if err != nil {
		t.Fatal(err)
	}
	if string(b) != "null" {
		t.Fatalf("nil progress=%s", b)
	}
}

func TestCancelPendingJobProducts_failClosed(t *testing.T) {
	jobID := uuid.MustParse("11111111-1111-1111-1111-111111111111")
	errDB := errors.New("db down")
	err := cancelPendingJobProducts(context.Background(), func(context.Context, string, ...any) (pgconn.CommandTag, error) {
		return pgconn.CommandTag{}, errDB
	}, jobID)
	if err == nil {
		t.Fatal("expected product-cancel Exec error")
	}
	if !errors.Is(err, errDB) {
		t.Fatalf("wrap: %v", err)
	}
	if !strings.Contains(err.Error(), "cancel job products") {
		t.Fatalf("missing context: %v", err)
	}
	if !strings.Contains(err.Error(), jobID.String()) {
		t.Fatalf("missing job id: %v", err)
	}
}

func TestCancelPendingJobProducts_ok(t *testing.T) {
	called := false
	err := cancelPendingJobProducts(context.Background(), func(_ context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
		called = true
		if !strings.Contains(sql, "processing_job_products") {
			t.Fatalf("sql=%q", sql)
		}
		if len(args) != 1 {
			t.Fatalf("args=%v", args)
		}
		return pgconn.CommandTag{}, nil
	}, uuid.MustParse("22222222-2222-2222-2222-222222222222"))
	if err != nil {
		t.Fatal(err)
	}
	if !called {
		t.Fatal("exec not called")
	}
}

func TestMarshalProcessOnePayload_roundTrip(t *testing.T) {
	result := StepResult{
		Attributes:          map[string]any{"color": "red"},
		ProcessedAttributes: map[string]any{"color": "crimson"},
		GPTResponse:         map[string]any{"ok": true},
		FieldSources:        map[string]any{"color": "ai"},
	}
	attrs, proc, gpt, sources, err := marshalProcessOnePayload(result)
	if err != nil {
		t.Fatal(err)
	}
	if string(attrs) == "" || string(proc) == "" || string(gpt) == "" || string(sources) == "" {
		t.Fatalf("empty payload attrs=%s proc=%s gpt=%s sources=%s", attrs, proc, gpt, sources)
	}
}

func TestMarshalProcessOnePayload_failClosed(t *testing.T) {
	bad := map[string]any{"ch": make(chan int)}
	_, _, _, _, err := marshalProcessOnePayload(StepResult{Attributes: bad})
	if err == nil {
		t.Fatal("expected marshal attributes error")
	}
	_, _, _, _, err = marshalProcessOnePayload(StepResult{
		Attributes:          map[string]any{"ok": 1},
		ProcessedAttributes: bad,
	})
	if err == nil {
		t.Fatal("expected marshal processed_attributes error")
	}
	_, _, _, _, err = marshalProcessOnePayload(StepResult{
		Attributes:          map[string]any{"ok": 1},
		ProcessedAttributes: map[string]any{"ok": 1},
		GPTResponse:         bad,
	})
	if err == nil {
		t.Fatal("expected marshal gpt_response error")
	}
	_, _, _, _, err = marshalProcessOnePayload(StepResult{
		Attributes:          map[string]any{"ok": 1},
		ProcessedAttributes: map[string]any{"ok": 1},
		GPTResponse:         map[string]any{"ok": 1},
		FieldSources:        bad,
	})
	if err == nil {
		t.Fatal("expected marshal field_sources error")
	}
}

func TestStuckAgeIntervalAligned(t *testing.T) {
	if StuckAgeInterval != "2 hours" {
		t.Fatalf("StuckAgeInterval=%q want 2 hours (CleanupStuck + claim reclaim)", StuckAgeInterval)
	}
}

func TestResolveBatchSize(t *testing.T) {
	if got := resolveBatchSize(0); got != defaultBatchSize {
		t.Fatalf("0 -> %d want %d", got, defaultBatchSize)
	}
	if got := resolveBatchSize(-1); got != defaultBatchSize {
		t.Fatalf("-1 -> %d want %d", got, defaultBatchSize)
	}
	if got := resolveBatchSize(50); got != 50 {
		t.Fatalf("50 -> %d", got)
	}
	if got := resolveBatchSize(maxBatchSize + 10); got != maxBatchSize {
		t.Fatalf("over max -> %d want %d", got, maxBatchSize)
	}
}

func TestShouldFlushJobProgress(t *testing.T) {
	if shouldFlushJobProgress(0, 25, true) {
		t.Fatal("no successes should not flush")
	}
	if shouldFlushJobProgress(3, 25, false) {
		t.Fatal("below threshold should not flush")
	}
	if !shouldFlushJobProgress(25, 25, false) {
		t.Fatal("at threshold should flush")
	}
	if !shouldFlushJobProgress(3, 25, true) {
		t.Fatal("batch done should flush pending successes")
	}
	if shouldFlushJobProgress(1, 0, false) {
		t.Fatal("1 < default progressEvery should not flush")
	}
}

func TestShouldFlushJobProgress_defaultEvery(t *testing.T) {
	// progressEvery<=0 resolves to defaultProgressEvery (25).
	if shouldFlushJobProgress(24, 0, false) {
		t.Fatal("24 < default 25")
	}
	if !shouldFlushJobProgress(25, 0, false) {
		t.Fatal("25 == default")
	}
}

func TestShouldDebitProductProcessing(t *testing.T) {
	run := StepResult{TotalTokens: 12}
	skip := StepResult{SkipCreditDebit: true, TotalTokens: 0}
	if !shouldDebitProductProcessing(false, run) {
		t.Fatal("normal AI run must debit")
	}
	if shouldDebitProductProcessing(true, run) {
		t.Fatal("BYOK must not debit")
	}
	if shouldDebitProductProcessing(false, skip) {
		t.Fatal("hash-skip must not debit flat product_processing")
	}
	if shouldDebitProductProcessing(true, skip) {
		t.Fatal("BYOK + hash-skip must not debit")
	}
	// Flat 0-token without hash-skip still debits (paid processing fee path).
	if !shouldDebitProductProcessing(false, StepResult{TotalTokens: 0}) {
		t.Fatal("0-token without SkipCreditDebit must still debit")
	}
}

func TestNewPipeline_defaultBatchAndProgress(t *testing.T) {
	p := NewPipeline(nil)
	if p.BatchSize != defaultBatchSize {
		t.Fatalf("BatchSize=%d want %d", p.BatchSize, defaultBatchSize)
	}
	if p.ProgressEvery != defaultProgressEvery {
		t.Fatalf("ProgressEvery=%d want %d", p.ProgressEvery, defaultProgressEvery)
	}
}