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