From ca35eb5f15d74cb18266cf696cd565ce13221d6e Mon Sep 17 00:00:00 2001 From: Mazyar Date: Sun, 14 Jun 2026 14:02:35 +0330 Subject: [PATCH] Enhance worker configuration and error handling for AI summarizer - Added `TranslationEnabled` and `TranslationInterval` fields to the worker configuration to manage automatic translation scheduling. - Updated the worker initialization to log when the translation worker is disabled. - Improved error handling in the AI summarizer client by introducing `APIStatusError` for better context on API failures, replacing direct error messages with structured error responses. This update enhances the configurability of the worker and improves error reporting for AI service interactions, contributing to better maintainability and user experience. --- cmd/worker/bootstrap/bootstrap.go | 8 ++++++- cmd/worker/bootstrap/config.go | 3 +++ internal/ai_pipeline/handler.go | 38 ++++++++++++++++++++++++++---- pkg/ai_summarizer/client.go | 39 ++++++++++++++++++------------- pkg/ai_summarizer/errors.go | 23 +++++++++++++++++- 5 files changed, 89 insertions(+), 22 deletions(-) diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index df337c3..1b3a4cc 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -85,6 +85,8 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "notice_processing_limit": config.Worker.NoticeProcessingLimit, "notice_batch_pause": config.Worker.NoticeBatchPause.String(), "notice_fetch_error_backoff": config.Worker.NoticeFetchErrorBackoff.String(), + "translation_enabled": config.Worker.TranslationEnabled, + "translation_interval": config.Worker.TranslationInterval, }) // Initialize repositories noticeRepo := notice.NewRepository(mongoManager, appLogger) @@ -138,7 +140,11 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger }) // Initialize translation backfill worker for configured languages - if aiClient != nil { + if !config.Worker.TranslationEnabled { + appLogger.Info("Tender translation worker disabled by configuration", map[string]interface{}{ + "translation_enabled": false, + }) + } else if aiClient != nil { targetLanguages := parseTranslationLanguages(config.AISummarizer) translationSuccessCounter := mongo.NewCounter(mongoManager) scheduler.AddJob(schedule.Job{ diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 47acc1b..31c4422 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -30,6 +30,9 @@ type WorkerConfig struct { NoticeFetchErrorBackoff time.Duration `env:"WORKER_NOTICE_FETCH_ERROR_BACKOFF" envDefault:"2s"` TranslationInterval string `env:"WORKER_TRANSLATION_INTERVAL" envDefault:"0 */30 * * * *"` TranslationBatchSize int `env:"WORKER_TRANSLATION_BATCH_SIZE" envDefault:"20"` + // TranslationEnabled schedules the automatic batch translation cron job on the worker. + // On-demand translation (admin/public tender endpoints and AI pipeline routes on web) is unaffected. + TranslationEnabled bool `env:"WORKER_TRANSLATION_ENABLED" envDefault:"true"` } // AISummarizerConfig holds configuration for the external AI summarizer service. diff --git a/internal/ai_pipeline/handler.go b/internal/ai_pipeline/handler.go index c497bd9..5057e0a 100644 --- a/internal/ai_pipeline/handler.go +++ b/internal/ai_pipeline/handler.go @@ -2,6 +2,7 @@ package ai_pipeline import ( "errors" + "fmt" "net/http" "tm/pkg/ai_summarizer" @@ -389,13 +390,42 @@ func mapPipelineHTTPError(c echo.Context, err error, action string) error { return response.Conflict(c, "A pipeline job is already running") case errors.Is(err, ai_summarizer.ErrObjectNotFound): return response.NotFound(c, "Tender not found in AI service") - default: - if errors.Is(err, ai_summarizer.ErrAPINonSuccess) { + } + + var upstream *ai_summarizer.APIStatusError + if errors.As(err, &upstream) { + details := upstream.Body + if details == "" { + details = fmt.Sprintf("upstream status %d", upstream.StatusCode) + } + + switch upstream.StatusCode { + case http.StatusNotFound: + return response.NotFound(c, "Tender not found in AI service") + case http.StatusUnprocessableEntity: + return response.ValidationError(c, "AI service rejected the request", details) + case http.StatusBadGateway: return c.JSON(http.StatusBadGateway, response.APIResponse{ Success: false, - Message: action + " failed: upstream AI service error", + Message: action + " failed", + Error: &response.APIError{ + Code: "UPSTREAM_AI_ERROR", + Message: "Opplens AI service returned status 502", + Details: details, + }, + }) + default: + return c.JSON(http.StatusBadGateway, response.APIResponse{ + Success: false, + Message: action + " failed", + Error: &response.APIError{ + Code: "UPSTREAM_AI_ERROR", + Message: fmt.Sprintf("Opplens AI service returned status %d", upstream.StatusCode), + Details: details, + }, }) } - return response.InternalServerError(c, action+" failed") } + + return response.InternalServerError(c, action+" failed") } diff --git a/pkg/ai_summarizer/client.go b/pkg/ai_summarizer/client.go index 5eae5e8..895d0a8 100644 --- a/pkg/ai_summarizer/client.go +++ b/pkg/ai_summarizer/client.go @@ -163,7 +163,7 @@ func (c *Client) ScrapeDocuments(ctx context.Context, reqBody ScrapeDocumentsReq return nil, fmt.Errorf("%w: status 404, body: %s", ErrObjectNotFound, string(bodyBytes)) } if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result ScrapeDocumentsResponse @@ -180,6 +180,13 @@ func (c *Client) ScrapeDocuments(ctx context.Context, reqBody ScrapeDocumentsReq return &result, nil } +func apiNonSuccessError(statusCode int, bodyBytes []byte) error { + return &APIStatusError{ + StatusCode: statusCode, + Body: string(bodyBytes), + } +} + // ScrapeDocumentsBatch calls POST /scrape/documents/batch. func (c *Client) ScrapeDocumentsBatch(ctx context.Context, tenders []TenderRef) (*BatchAcceptedResponse, error) { reqBody := ScrapeDocumentsBatchRequest{Tenders: tenders} @@ -198,7 +205,7 @@ func (c *Client) PipelineSync(ctx context.Context) (*PipelineSyncResponse, error defer httpResp.Body.Close() if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result PipelineSyncResponse @@ -233,7 +240,7 @@ func (c *Client) PipelineRun(ctx context.Context, reqBody PipelineRunRequest) (m return nil, fmt.Errorf("%w: status 404, body: %s", ErrObjectNotFound, string(bodyBytes)) } if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result map[string]interface{} @@ -308,10 +315,10 @@ func (c *Client) GetPipelineStats(ctx context.Context, window string) (*Pipeline defer httpResp.Body.Close() if httpResp.StatusCode == http.StatusBadRequest { - return nil, fmt.Errorf("%w: status 400, body: %s", ErrAPINonSuccess, string(bodyBytes)) + return nil, apiNonSuccessError(http.StatusBadRequest, bodyBytes) } if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result PipelineStatsResponse @@ -334,7 +341,7 @@ func (c *Client) GetPipelineMinioStats(ctx context.Context) (PipelineMinioStatsR defer httpResp.Body.Close() if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result PipelineMinioStatsResponse @@ -361,7 +368,7 @@ func (c *Client) postBatchAccepted(ctx context.Context, path string, reqBody any defer httpResp.Body.Close() if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result BatchAcceptedResponse @@ -396,7 +403,7 @@ func (c *Client) triggerPipelineStarted(ctx context.Context, path, label string) return nil, fmt.Errorf("%w: status 409, body: %s", ErrPipelineJobRunning, string(bodyBytes)) } if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result PipelineStartedResponse @@ -432,10 +439,10 @@ func (c *Client) getPipelineReportWithQuery(ctx context.Context, path, window st defer httpResp.Body.Close() if httpResp.StatusCode == http.StatusBadRequest { - return nil, fmt.Errorf("%w: status 400, body: %s", ErrAPINonSuccess, string(bodyBytes)) + return nil, apiNonSuccessError(http.StatusBadRequest, bodyBytes) } if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result PipelineReportResponse @@ -486,7 +493,7 @@ func (c *Client) FetchAnalyzeOnDemand(ctx context.Context, reqBody AnalyzeReques return nil, fmt.Errorf("%w: status 404, body: %s", ErrObjectNotFound, string(bodyBytes)) } if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result AnalyzeResponse @@ -518,7 +525,7 @@ func (c *Client) StartOnboarding(ctx context.Context, reqBody OnboardingRequest) defer httpResp.Body.Close() if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result OnboardingResponse @@ -551,7 +558,7 @@ func (c *Client) FetchRecommendations(ctx context.Context, reqBody RecommendRequ defer httpResp.Body.Close() if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result []RecommendedTender @@ -578,7 +585,7 @@ func (c *Client) triggerPipelineAction(ctx context.Context, path, label string) defer httpResp.Body.Close() if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result PipelineActionResponse @@ -650,7 +657,7 @@ func (c *Client) doPost(ctx context.Context, url string, jsonBody []byte) (*Summ } if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } var result SummarizeResponse @@ -692,7 +699,7 @@ func (c *Client) doTranslatePost(ctx context.Context, url string, jsonBody []byt } if httpResp.StatusCode >= 400 { - return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes)) + return nil, apiNonSuccessError(httpResp.StatusCode, bodyBytes) } result, err := decodeTranslateResponse(bodyBytes) diff --git a/pkg/ai_summarizer/errors.go b/pkg/ai_summarizer/errors.go index 8c62888..2d96933 100644 --- a/pkg/ai_summarizer/errors.go +++ b/pkg/ai_summarizer/errors.go @@ -1,6 +1,9 @@ package ai_summarizer -import "errors" +import ( + "errors" + "fmt" +) // Common sentinel errors for the AI summarizer package. var ( @@ -37,3 +40,21 @@ var ( // ErrPipelineJobRunning is returned when a single-flight pipeline job is already running (HTTP 409). ErrPipelineJobRunning = errors.New("ai_summarizer: pipeline job already running") ) + +// APIStatusError carries the HTTP status and response body from the Opplens AI service. +type APIStatusError struct { + StatusCode int + Body string +} + +func (e *APIStatusError) Error() string { + if e == nil { + return "" + } + return fmt.Sprintf("ai_summarizer: API returned status %d, body: %s", e.StatusCode, e.Body) +} + +// Is reports whether target is ErrAPINonSuccess. +func (e *APIStatusError) Is(target error) bool { + return target == ErrAPINonSuccess +}