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) } } }