nexarch/ai/internal/queue/queue.go

// Package queue implementiert ENG-05: eine Postgres-gestuetzte Warteschlange
// fuer Inferenz-Aufrufe, damit lange KI-Anfragen den Anfrage-Pfad nicht
// blockieren (kein Redis, Techstack-Vorgabe). Struktur folgt dem etablierten
// Muster aus dms/internal/jobqueue und connect/internal/syncjob (Go-Module
// teilen keine internal/-Pakete, daher eigenstaendige Kopie).
package queue

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"gitea.perlbach24.de/scripte/nexarch/ai/internal/limits"
)

const (
	StatusPending    = "pending"
	StatusProcessing = "processing"
	StatusSucceeded  = "succeeded"
	StatusFailed     = "failed"
	StatusDeadLetter = "dead_letter"
)

var ErrNotFound = errors.New("queue: job nicht gefunden")
var ErrNoJobAvailable = errors.New("queue: kein job verfuegbar")

// ErrQueueFull ist QA-08s definiertes Ueberlastverhalten (AK3): Enqueue
// lehnt neue Jobs kontrolliert ab, statt die Warteschlange unbegrenzt
// anwachsen zu lassen, wenn bereits limits.MaxPendingJobs Jobs warten.
var ErrQueueFull = errors.New("queue: warteschlange ueberlastet, bitte spaeter erneut versuchen")

// Kind unterscheidet die Art des Inferenz-Aufrufs im Payload.
type Kind string

const (
	KindChat       Kind = "chat"
	KindCompletion Kind = "completion"
	KindEmbedding  Kind = "embedding"
)

type Job struct {
	ID          string
	JobType     Kind
	Payload     json.RawMessage
	Status      string
	Attempts    int
	MaxAttempts int
	Result      json.RawMessage
	LastError   *string
	TenantSlug  string
}

const DefaultMaxAttempts = 5

// QuotaEnforcer ist die schmale Schnittstelle zu GOV-04 (ai/internal/quota):
// blockiert Enqueue, wenn die Tenant-Quota bereits erreicht ist. Als
// Interface hier statt Import, damit queue (ENG-05) nicht von quota
// abhaengt — quota.Store erfuellt diese Schnittstelle strukturell.
type QuotaEnforcer interface {
	Enforce(ctx context.Context, tenantSlug string) error
}

type Queue struct {
	pool           *pgxpool.Pool
	staleLockAfter time.Duration
	limits         limits.Limits
	quota          QuotaEnforcer
}

func NewQueue(pool *pgxpool.Pool, staleLockAfter time.Duration) *Queue {
	return &Queue{pool: pool, staleLockAfter: staleLockAfter, limits: limits.FromEnv()}
}

// SetLimits ueberschreibt die aus der Umgebung geladenen Obergrenzen (ENG-07),
// vor allem fuer Tests mit kleinen, deterministischen Werten.
func (q *Queue) SetLimits(l limits.Limits) { q.limits = l }

// SetQuotaEnforcer verknuepft die Queue mit GOV-04s Kostenkontrolle; ohne
// Aufruf bleibt Enqueue unveraendert unbegrenzt (Rueckwaertskompatibilitaet).
func (q *Queue) SetQuotaEnforcer(e QuotaEnforcer) { q.quota = e }

type EnqueueOptions struct {
	IdempotencyKey string
	MaxAttempts    int
	TenantSlug     string
}

// Enqueue reiht einen Inferenz-Aufruf asynchron ein (AK1). Bei gesetztem
// IdempotencyKey liefert ein erneuter Aufruf mit demselben Key die ID des
// bereits vorhandenen Jobs statt eines Duplikats. Ist ein QuotaEnforcer
// gesetzt und TenantSlug angegeben, wird die Tenant-Quota VOR dem Anlegen
// des Jobs geprueft (GOV-04 AK2) — eine ueberschrittene Quota verhindert
// zuverlaessig, dass ueberhaupt erst neue Arbeit entsteht.
func (q *Queue) Enqueue(ctx context.Context, jobType Kind, payload any, opts EnqueueOptions) (string, error) {
	payloadJSON, err := json.Marshal(payload)
	if err != nil {
		return "", fmt.Errorf("queue: payload serialisieren: %w", err)
	}
	if err := q.limits.CheckSize(len(payloadJSON)); err != nil {
		return "", err
	}
	if q.quota != nil && opts.TenantSlug != "" {
		if err := q.quota.Enforce(ctx, opts.TenantSlug); err != nil {
			return "", err
		}
	}
	if q.limits.MaxPendingJobs > 0 {
		depth, err := q.Depth(ctx)
		if err != nil {
			return "", err
		}
		if depth.Pending >= int64(q.limits.MaxPendingJobs) {
			return "", ErrQueueFull
		}
	}
	maxAttempts := opts.MaxAttempts
	if maxAttempts <= 0 {
		maxAttempts = DefaultMaxAttempts
	}
	var idempotencyKey any
	if opts.IdempotencyKey != "" {
		idempotencyKey = opts.IdempotencyKey
	}
	var tenantSlug any
	if opts.TenantSlug != "" {
		tenantSlug = opts.TenantSlug
	}
	var id string
	err = q.pool.QueryRow(ctx, `
		INSERT INTO inference_jobs (job_type, payload, idempotency_key, max_attempts, tenant_slug)
		VALUES ($1, $2, $3, $4, $5)
		ON CONFLICT (idempotency_key) DO UPDATE SET job_type = inference_jobs.job_type
		RETURNING id
	`, string(jobType), payloadJSON, idempotencyKey, maxAttempts, tenantSlug).Scan(&id)
	if err != nil {
		return "", fmt.Errorf("queue: job einreihen: %w", err)
	}
	return id, nil
}

// Dequeue holt genau einen abholbaren Job und sperrt ihn atomar (FOR UPDATE
// SKIP LOCKED erlaubt mehrere gleichzeitige Worker ohne Doppelabholung).
func (q *Queue) Dequeue(ctx context.Context, workerID string) (*Job, error) {
	tx, err := q.pool.Begin(ctx)
	if err != nil {
		return nil, fmt.Errorf("queue: transaktion starten: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	row := tx.QueryRow(ctx, `
		SELECT id, job_type, payload, status, attempts, max_attempts, last_error, COALESCE(tenant_slug, '')
		FROM inference_jobs
		WHERE (
			(status = 'pending' AND available_at <= now())
			OR (status = 'processing' AND locked_at <= now() - ($1 * interval '1 second'))
		)
		ORDER BY available_at
		FOR UPDATE SKIP LOCKED
		LIMIT 1
	`, q.staleLockAfter.Seconds())

	var j Job
	var jobType string
	if err := row.Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.LastError, &j.TenantSlug); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrNoJobAvailable
		}
		return nil, fmt.Errorf("queue: naechsten job lesen: %w", err)
	}
	j.JobType = Kind(jobType)

	if _, err := tx.Exec(ctx, `
		UPDATE inference_jobs
		SET status = 'processing', attempts = attempts + 1, locked_at = now(), locked_by = $2, updated_at = now()
		WHERE id = $1
	`, j.ID, workerID); err != nil {
		return nil, fmt.Errorf("queue: job sperren: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, fmt.Errorf("queue: dequeue committen: %w", err)
	}
	j.Status = StatusProcessing
	j.Attempts++
	return &j, nil
}

func (q *Queue) Complete(ctx context.Context, jobID string, result any) error {
	resultJSON, err := json.Marshal(result)
	if err != nil {
		return fmt.Errorf("queue: ergebnis serialisieren: %w", err)
	}
	tag, err := q.pool.Exec(ctx, `
		UPDATE inference_jobs SET status = 'succeeded', result = $2, locked_at = NULL, locked_by = NULL, updated_at = now()
		WHERE id = $1
	`, jobID, resultJSON)
	if err != nil {
		return fmt.Errorf("queue: job abschliessen: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return ErrNotFound
	}
	return nil
}

// Fail markiert einen Job als fehlgeschlagen (AK2): sind die maximalen
// Versuche erreicht, wandert der Job in die Dead-Letter-Queue, sonst wird er
// mit Backoff erneut eingeplant. Die dabei anwachsende attempts-Spalte ist
// die Rueckverfolgung, die das Ticket fordert (jeder Versuch ist gezaehlt
// und der letzte Fehler bleibt bis zum naechsten Versuch sichtbar).
func (q *Queue) Fail(ctx context.Context, jobID string, cause error) error {
	errMsg := cause.Error()
	tag, err := q.pool.Exec(ctx, `
		UPDATE inference_jobs
		SET status = CASE WHEN attempts >= max_attempts THEN 'dead_letter' ELSE 'pending' END,
		    available_at = now() + (LEAST(attempts, 10) * interval '5 seconds'),
		    locked_at = NULL, locked_by = NULL, last_error = $2, updated_at = now()
		WHERE id = $1
	`, jobID, errMsg)
	if err != nil {
		return fmt.Errorf("queue: fehlschlag erfassen: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return ErrNotFound
	}
	return nil
}

func (q *Queue) Status(ctx context.Context, jobID string) (*Job, error) {
	var j Job
	var jobType string
	err := q.pool.QueryRow(ctx, `
		SELECT id, job_type, payload, status, attempts, max_attempts, result, last_error, COALESCE(tenant_slug, '')
		FROM inference_jobs WHERE id = $1
	`, jobID).Scan(&j.ID, &jobType, &j.Payload, &j.Status, &j.Attempts, &j.MaxAttempts, &j.Result, &j.LastError, &j.TenantSlug)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrNotFound
		}
		return nil, fmt.Errorf("queue: job-status lesen: %w", err)
	}
	j.JobType = Kind(jobType)
	return &j, nil
}

// Depth liefert den beobachtbaren Warteschlangenstand (AK3): Anzahl Jobs je
// Status, fuer ein Monitoring-Endpunkt oder Metrik-Export geeignet.
type Depth struct {
	Pending    int64
	Processing int64
	DeadLetter int64
}

func (q *Queue) Depth(ctx context.Context) (Depth, error) {
	var d Depth
	err := q.pool.QueryRow(ctx, `
		SELECT
			count(*) FILTER (WHERE status = 'pending'),
			count(*) FILTER (WHERE status = 'processing'),
			count(*) FILTER (WHERE status = 'dead_letter')
		FROM inference_jobs
	`).Scan(&d.Pending, &d.Processing, &d.DeadLetter)
	if err != nil {
		return Depth{}, fmt.Errorf("queue: warteschlangenstand lesen: %w", err)
	}
	return d, nil
}