descrybe/apps/api/internal/support/auto_jobs.go

package support

import (
	"context"
	"errors"
	"log/slog"
	"strings"
	"time"
	"unicode/utf8"

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

// AutoJob is one async AI fallback work item.
type AutoJob struct {
	ID        uuid.UUID
	TicketID  uuid.UUID
	CompanyID uuid.UUID
	Status    string
	Attempt   int
}

// EnqueueAIFallback queues Stage B after a FAQ miss (or FAQ disabled).
// Never awaits the LLM. Idempotent per active ticket job.
func (s *Service) EnqueueAIFallback(ctx context.Context, ticket Ticket) error {
	if s == nil || s.Pool == nil {
		return nil
	}
	cfg, err := s.GetAutoConfig(ctx)
	if err != nil {
		return err
	}
	if !cfg.Enabled || !cfg.AIEnabled {
		_ = s.markAutoAttempt(ctx, ticket.ID, AutoReplySkipped)
		return nil
	}
	if ticket.AutoReplyDisabled {
		return nil
	}
	switch ticket.AutoReplyStatus {
	case AutoReplyMatched, AutoReplyAISent, AutoReplyAIDraft, AutoReplyHandedOff:
		return nil
	}

	_, err = s.Pool.Exec(ctx, `
		INSERT INTO support_auto_jobs (ticket_id, company_id, status, attempt, created_at, updated_at)
		SELECT $1, $2, 'pending', 0, now(), now()
		WHERE NOT EXISTS (
			SELECT 1 FROM support_auto_jobs
			WHERE ticket_id = $1 AND status IN ('pending', 'running')
		)`,
		ticket.ID, ticket.CompanyID)
	if err != nil {
		if IsMissingRelation(err) {
			// Migration not applied — sync-with-timeout fallback so create still works.
			return s.TryAutoReplyLLM(ctx, ticket.ID)
		}
		var pgErr *pgconn.PgError
		if errors.As(err, &pgErr) && pgErr.Code == "23505" {
			return nil
		}
		return err
	}
	slog.Info("support_auto_ai_enqueued",
		"ticket_id", ticket.ID.String(),
		"company_id", ticket.CompanyID.String(),
	)
	_, _ = s.Pool.Exec(ctx, `SELECT pg_notify('support_auto_jobs', $1)`, ticket.ID.String())
	return nil
}

// ClaimNextAutoJob claims one pending AI job (SKIP LOCKED).
func (s *Service) ClaimNextAutoJob(ctx context.Context) (AutoJob, error) {
	var j AutoJob
	if s == nil || s.Pool == nil {
		return j, pgx.ErrNoRows
	}
	err := s.Pool.QueryRow(ctx, `
		UPDATE support_auto_jobs
		SET status = 'running', attempt = attempt + 1, updated_at = now()
		WHERE id = (
			SELECT id FROM support_auto_jobs
			WHERE status = 'pending'
			ORDER BY created_at ASC
			FOR UPDATE SKIP LOCKED
			LIMIT 1
		)
		RETURNING id, ticket_id, company_id, status, attempt`).Scan(
		&j.ID, &j.TicketID, &j.CompanyID, &j.Status, &j.Attempt,
	)
	if err != nil {
		return j, err
	}
	return j, nil
}

// ProcessAutoJob runs TryAutoReplyLLM for a claimed job and marks done/failed.
func (s *Service) ProcessAutoJob(ctx context.Context, job AutoJob) error {
	if s == nil {
		return ErrAIAutoReplyDisabled
	}
	err := s.TryAutoReplyLLM(ctx, job.TicketID)
	if err == nil ||
		errors.Is(err, ErrAutoReplyAlreadyPosted) ||
		errors.Is(err, ErrAutoReplyDisabled) ||
		errors.Is(err, ErrTicketClosed) {
		_ = s.finishAutoJob(ctx, job.ID, "done", "")
		return nil
	}
	if errors.Is(err, ErrAIAutoReplyDisabled) {
		_ = s.markAutoAttempt(ctx, job.TicketID, AutoReplySkipped)
		_ = s.finishAutoJob(ctx, job.ID, "done", "ai_disabled")
		return nil
	}
	if errors.Is(err, ErrAIRateLimited) {
		_ = s.markAutoHandOff(ctx, job.TicketID)
		_ = s.finishAutoJob(ctx, job.ID, "failed", "rate_limited")
		return err
	}
	msg := truncateJobError(RedactForAutoLog(err.Error()))
	_ = s.finishAutoJob(ctx, job.ID, "failed", msg)
	return err
}

func (s *Service) finishAutoJob(ctx context.Context, jobID uuid.UUID, status, lastErr string) error {
	if s == nil || s.Pool == nil {
		return nil
	}
	var errArg any
	if strings.TrimSpace(lastErr) == "" {
		errArg = nil
	} else {
		errArg = lastErr
	}
	_, err := s.Pool.Exec(ctx, `
		UPDATE support_auto_jobs
		SET status = $2, last_error = $3, updated_at = now()
		WHERE id = $1`, jobID, status, errArg)
	return err
}

// ProcessPendingAutoJobs claims and runs up to limit AI jobs (worker loop helper).
func (s *Service) ProcessPendingAutoJobs(ctx context.Context, limit int) (int, error) {
	if limit <= 0 {
		limit = 1
	}
	n := 0
	for i := 0; i < limit; i++ {
		job, err := s.ClaimNextAutoJob(ctx)
		if errors.Is(err, pgx.ErrNoRows) || IsMissingRelation(err) {
			return n, nil
		}
		if err != nil {
			return n, err
		}
		_ = s.ProcessAutoJob(ctx, job)
		n++
	}
	return n, nil
}

func truncateJobError(s string) string {
	s = strings.TrimSpace(s)
	const max = 500
	if utf8.RuneCountInString(s) <= max {
		return s
	}
	return string([]rune(s)[:max])
}

// RunAutoJobsLoop is a simple poller for tests / lightweight workers.
func (s *Service) RunAutoJobsLoop(ctx context.Context, every time.Duration, batch int) {
	if every <= 0 {
		every = 2 * time.Second
	}
	t := time.NewTicker(every)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			_, _ = s.ProcessPendingAutoJobs(ctx, batch)
		}
	}
}