diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index e9addc7..a63d69e 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -38,7 +38,7 @@ type WorkerConfig struct { Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"` NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"` DeleteProcessedNotices bool `env:"WORKER_DELETE_PROCESSED_NOTICES" envDefault:"true"` - NoticeWorkerConcurrency int `env:"WORKER_NOTICE_CONCURRENCY" envDefault:"8"` + NoticeWorkerConcurrency int `env:"WORKER_NOTICE_CONCURRENCY" envDefault:"1"` NoticeFetchBatchSize int `env:"WORKER_NOTICE_FETCH_BATCH" envDefault:"50"` // NoticeBatchPause is slept after each notice batch finishes, before loading the next batch (0 = no pause). NoticeBatchPause time.Duration `env:"WORKER_NOTICE_BATCH_PAUSE" envDefault:"0s"` diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index dc069a9..804fd7b 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -21,6 +21,10 @@ import ( "golang.org/x/sync/errgroup" ) +// noticeRunSerialMu ensures only one NoticeWorker.Run executes at a time (startup +// goroutine + cron ticks can otherwise overlap and double load on MongoDB). +var noticeRunSerialMu sync.Mutex + const ( // tenderIDCountersCollection holds per-(sourceCode+year) atomic counters used to allocate // unique tender_id values (e.g. PTD26827). Using a Mongo counter document is the standard way @@ -39,7 +43,7 @@ type NoticeWorker struct { TenderRepo tender.TenderRepository ProcessingLimit int DeleteProcessedNotices bool - // Concurrency is the maximum number of notices processed in parallel per batch (default 8). + // Concurrency is the maximum number of notices processed in parallel per batch (default 1). Concurrency int // FetchBatchSize is how many unprocessed notices to load from MongoDB per batch (default 50). FetchBatchSize int @@ -60,7 +64,7 @@ func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notif processingLimit = 5 // Default limit } if concurrency < 1 { - concurrency = 8 + concurrency = 1 } if fetchBatchSize < 1 { fetchBatchSize = 50 @@ -141,6 +145,9 @@ func (w *NoticeWorker) cleanupProcessedNotices() { } func (w *NoticeWorker) Run() { + noticeRunSerialMu.Lock() + defer noticeRunSerialMu.Unlock() + w.Logger.Info("Notice worker started", map[string]interface{}{ "concurrency": w.Concurrency, "fetch_batch": w.FetchBatchSize, diff --git a/pkg/ai_summarizer/storage.go b/pkg/ai_summarizer/storage.go index a9e3727..d946d3d 100644 --- a/pkg/ai_summarizer/storage.go +++ b/pkg/ai_summarizer/storage.go @@ -126,10 +126,14 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde "path": objectPath, }) - // Attempt to get the object - object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectPath, minio.GetObjectOptions{}) + return s.getSummaryFromObjectKey(ctx, objectPath, contractFolderID, noticePublicationID) +} + +// getSummaryFromObjectKey reads tender.json at objectKey and returns overall_summary when ready. +// contractFolderID and noticePublicationID are used for logging only (may be empty when unknown). +func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey, contractFolderID, noticePublicationID string) (string, error) { + object, err := s.client.GetObject(ctx, s.config.MinioBucket, objectKey, minio.GetObjectOptions{}) if err != nil { - // minio-go returns an error response that we can inspect var minioErr minio.ErrorResponse if errors.As(err, &minioErr) { switch minioErr.StatusCode { @@ -146,24 +150,20 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde } defer object.Close() - // Read the object content data, err := io.ReadAll(object) if err != nil { return "", fmt.Errorf("ai_summarizer: failed to read tender.json: %w", err) } - // Parse the JSON var tenderJSON TenderJSON if err := json.Unmarshal(data, &tenderJSON); err != nil { return "", fmt.Errorf("ai_summarizer: failed to parse tender.json: %w", err) } - // Check if summarization is complete if !tenderJSON.SummarizeStatus.Done { return "", ErrSummaryNotReady } - // Check if overall_summary is available if tenderJSON.OverallSummary == nil || *tenderJSON.OverallSummary == "" { return "", ErrNoOverallSummary } @@ -171,6 +171,7 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde s.logger.Info("Retrieved overall_summary from storage", map[string]interface{}{ "contract_folder_id": contractFolderID, "notice_id": noticePublicationID, + "object_key": objectKey, "summary_length": len(*tenderJSON.OverallSummary), }) @@ -302,65 +303,86 @@ func (s *StorageClient) DownloadDocument(ctx context.Context, objectName string) return object, nil } -// GetAllSummaries walks the MinIO bucket for the AI team's procedure-scoped -// layout (PROC_//tender.json) and -// returns one TenderSummaryItem per notice. Tenders whose summarize_status.done -// is true are returned with their overall_summary; tenders still in progress -// are returned with Done=false and an empty summary. +// GetAllSummaries walks the MinIO bucket for tender.json objects. It first lists +// under the PROC_ prefix; if none match, it falls back to a full-bucket recursive +// scan (slower on large buckets) so alternate key layouts still work. func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryItem, error) { - s.logger.Debug("Listing all tender.json files from MinIO", map[string]interface{}{ - "bucket": s.config.MinioBucket, - }) + results := make([]TenderSummaryItem, 0) + seen := make(map[string]struct{}) - objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{ - Prefix: procedurePrefix, - Recursive: true, - }) - - var results []TenderSummaryItem - - for object := range objectCh { - if object.Err != nil { - s.logger.Error("Error listing MinIO objects", map[string]interface{}{ - "error": object.Err.Error(), - }) - continue - } - - // Expected key shape: PROC_//tender.json - if !strings.HasSuffix(object.Key, "/tender.json") { - continue - } - - contractFolderID, noticePublicationID, ok := parseTenderJSONKey(object.Key) - if !ok { - continue - } - - summary, err := s.GetSummaryFromStorage(ctx, contractFolderID, noticePublicationID) - if err != nil { - s.logger.Debug("Skipping tender summary", map[string]interface{}{ - "contract_folder_id": contractFolderID, - "notice_id": noticePublicationID, - "error": err.Error(), - }) - - if errors.Is(err, ErrSummaryNotReady) { - results = append(results, TenderSummaryItem{ - ContractFolderID: contractFolderID, - NoticeID: noticePublicationID, - Done: false, - }) - } - continue - } - - results = append(results, TenderSummaryItem{ - ContractFolderID: contractFolderID, - NoticeID: noticePublicationID, - OverallSummary: summary, - Done: true, + appendFromPrefix := func(prefix string) { + s.logger.Debug("Listing tender.json files from MinIO", map[string]interface{}{ + "bucket": s.config.MinioBucket, + "prefix": prefix, }) + + objectCh := s.client.ListObjects(ctx, s.config.MinioBucket, minio.ListObjectsOptions{ + Prefix: prefix, + Recursive: true, + }) + + for object := range objectCh { + if object.Err != nil { + s.logger.Error("Error listing MinIO objects", map[string]interface{}{ + "prefix": prefix, + "error": object.Err.Error(), + }) + continue + } + key := object.Key + if !strings.HasSuffix(key, "/tender.json") { + continue + } + if _, dup := seen[key]; dup { + continue + } + + contractFolderID, noticePublicationID, ok := parseTenderJSONKey(key) + if !ok { + s.logger.Debug("Skipping unparseable tender.json key", map[string]interface{}{ + "key": key, + }) + continue + } + seen[key] = struct{}{} + + summary, err := s.GetSummaryFromStorage(ctx, contractFolderID, noticePublicationID) + if errors.Is(err, ErrObjectNotFound) { + summary, err = s.getSummaryFromObjectKey(ctx, key, contractFolderID, noticePublicationID) + } + if err != nil { + s.logger.Debug("Skipping tender summary", map[string]interface{}{ + "object_key": key, + "contract_folder_id": contractFolderID, + "notice_id": noticePublicationID, + "error": err.Error(), + }) + + if errors.Is(err, ErrSummaryNotReady) { + results = append(results, TenderSummaryItem{ + ContractFolderID: contractFolderID, + NoticeID: noticePublicationID, + Done: false, + }) + } + continue + } + + results = append(results, TenderSummaryItem{ + ContractFolderID: contractFolderID, + NoticeID: noticePublicationID, + OverallSummary: summary, + Done: true, + }) + } + } + + appendFromPrefix(procedurePrefix) + if len(seen) == 0 { + s.logger.Info("No tender.json keys under PROC_; listing entire bucket for */tender.json (may be slower on large buckets)", map[string]interface{}{ + "bucket": s.config.MinioBucket, + }) + appendFromPrefix("") } s.logger.Info("Retrieved all summaries from storage", map[string]interface{}{ @@ -371,26 +393,38 @@ func (s *StorageClient) GetAllSummaries(ctx context.Context) ([]TenderSummaryIte return results, nil } -// parseTenderJSONKey splits a key of the form -// PROC_//tender.json into its components. -// The boolean is false if the key does not match the expected layout. +// parseTenderJSONKey splits an object key ending in /tender.json into contract folder +// id and notice publication id. Accepts PROC_/proc_ prefix or plain //tender.json. func parseTenderJSONKey(key string) (contractFolderID, noticePublicationID string, ok bool) { - if !strings.HasPrefix(key, procedurePrefix) { + key = strings.TrimPrefix(strings.TrimSpace(key), "/") + if !strings.HasSuffix(key, "/tender.json") { return "", "", false } trimmed := strings.TrimSuffix(key, "/tender.json") - if trimmed == key { + if trimmed == "" { return "", "", false } - parts := strings.SplitN(trimmed, "/", 3) - if len(parts) != 2 { + last := strings.LastIndex(trimmed, "/") + if last < 0 { return "", "", false } - cfid := strings.TrimPrefix(parts[0], procedurePrefix) - if cfid == "" || parts[1] == "" { + noticePublicationID = strings.TrimSpace(trimmed[last+1:]) + parent := strings.TrimSpace(trimmed[:last]) + if noticePublicationID == "" || parent == "" { return "", "", false } - return cfid, parts[1], true + contractFolderID = stripLeadingProcPrefix(parent) + if contractFolderID == "" { + return "", "", false + } + return contractFolderID, noticePublicationID, true +} + +func stripLeadingProcPrefix(parent string) string { + if len(parent) >= 5 && strings.EqualFold(parent[:5], "proc_") { + return strings.TrimSpace(parent[5:]) + } + return parent } // countDone counts how many items have Done=true