diff --git a/cmd/worker/workers/notice.go b/cmd/worker/workers/notice.go index 4686f3b..4830451 100644 --- a/cmd/worker/workers/notice.go +++ b/cmd/worker/workers/notice.go @@ -436,14 +436,44 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { } } + title := n.Title + description := n.Description + estimatedVal := n.EstimatedValue + awardedVal := n.AwardedValue + currency := n.Currency + + if !t.ID.IsZero() { + procurementLots = mergeProcurementLotsByLotID(t.ProcurementLots, procurementLots) + awardedEntities = mergeAwardedEntitiesByLotID(t.AwardedEntities, awardedEntities) + title, description = mergeTitleAndDescriptionForMultiLot(t.Title, t.Description, title, description) + + if strings.TrimSpace(previousNoticePublicationID) != "" && previousNoticePublicationID != n.NoticePublicationID { + if n.EstimatedValue > 0 { + estimatedVal = t.EstimatedValue + n.EstimatedValue + } else { + estimatedVal = t.EstimatedValue + } + if n.AwardedValue > 0 { + awardedVal = t.AwardedValue + n.AwardedValue + } else { + awardedVal = t.AwardedValue + } + if n.Currency != "" { + currency = n.Currency + } else { + currency = t.Currency + } + } + } + // Title and description stay in the notice language; English translation is handled by the AI service (translation worker). - t.Title = n.Title - t.Description = n.Description + t.Title = title + t.Description = description t.NoticePublicationID = n.NoticePublicationID t.ProcurementTypeCode = n.ProcurementTypeCode t.ProcedureCode = n.ProcedureCode - t.EstimatedValue = n.EstimatedValue - t.Currency = n.Currency + t.EstimatedValue = estimatedVal + t.Currency = currency t.TenderDeadline = n.TenderDeadline t.SubmissionURL = n.SubmissionURL t.BuyerOrganization = buyerOrg @@ -483,6 +513,7 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { t.Source = tender.TenderSource(n.Source) t.ContractNoticeID = n.ContractNoticeID t.ContractFolderID = n.ContractFolderID + t.ProcurementProjectID = n.ProcurementProjectID t.SourceFileURL = n.SourceFileURL t.SourceFileName = n.SourceFileName t.ContentXML = n.ContentXML @@ -500,13 +531,10 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { t.CancellationReason = n.CancellationReason t.CancellationDate = n.CancellationDate t.AwardDate = n.AwardDate - t.AwardedValue = n.AwardedValue + t.AwardedValue = awardedVal t.ContractNumber = n.ContractNumber t.SuspensionReason = n.SuspensionReason t.SuspensionDate = n.SuspensionDate - t.WinningTenderer = winningTenderer - t.Modifications = modifications - t.AwardedEntities = awardedEntities if preserveTenderID != "" { t.TenderID = preserveTenderID t.ProjectName = preserveProjectName @@ -526,13 +554,18 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { // resolveExistingTender finds the tender that this notice should update, preferring the procedure-level // ContractFolderID (so follow-up notices like a Result for an earlier Competition notice merge into the -// same tender) and falling back to NoticePublicationID for older rows or notices without a folder id. +// same tender), then UBL ProcurementProject/ID (merges per-lot CAN/CN notices), then NoticePublicationID. func (w *NoticeWorker) resolveExistingTender(ctx context.Context, n *notice.Notice) *tender.Tender { if strings.TrimSpace(n.ContractFolderID) != "" { if t, err := w.TenderRepo.GetByContractFolderID(ctx, n.ContractFolderID); err == nil && t != nil { return t } } + if strings.TrimSpace(n.ProcurementProjectID) != "" { + if t, err := w.TenderRepo.GetByProcurementProjectID(ctx, n.ProcurementProjectID); err == nil && t != nil { + return t + } + } if strings.TrimSpace(n.NoticePublicationID) != "" { if t, err := w.TenderRepo.GetByNoticePublicationID(ctx, n.NoticePublicationID); err == nil && t != nil { return t @@ -660,6 +693,80 @@ func mergeRelatedPublicationIDs(existing []string, previous, current string) []s return out } +func mergeProcurementLotsByLotID(existing, incoming []tender.ProcurementLot) []tender.ProcurementLot { + out := append([]tender.ProcurementLot(nil), existing...) + for _, l := range incoming { + id := strings.TrimSpace(l.LotID) + if id != "" { + replaced := false + for i := range out { + if strings.TrimSpace(out[i].LotID) == id { + out[i] = l + replaced = true + break + } + } + if !replaced { + out = append(out, l) + } + continue + } + out = append(out, l) + } + return out +} + +func mergeAwardedEntitiesByLotID(existing, incoming []tender.Awarded) []tender.Awarded { + out := append([]tender.Awarded(nil), existing...) + for _, a := range incoming { + id := strings.TrimSpace(a.LotID) + if id != "" { + replaced := false + for i := range out { + if strings.TrimSpace(out[i].LotID) == id { + out[i] = a + replaced = true + break + } + } + if !replaced { + out = append(out, a) + } + continue + } + out = append(out, a) + } + return out +} + +func mergeTitleAndDescriptionForMultiLot(existingTitle, existingDesc, incomingTitle, incomingDesc string) (string, string) { + et, it := strings.TrimSpace(existingTitle), strings.TrimSpace(incomingTitle) + title := existingTitle + if len(it) > len(et) { + title = incomingTitle + } + desc := mergeDescriptionsConcat(existingDesc, incomingDesc) + return title, desc +} + +func mergeDescriptionsConcat(a, b string) string { + a = strings.TrimSpace(a) + b = strings.TrimSpace(b) + if a == "" { + return b + } + if b == "" { + return a + } + if strings.Contains(a, b) { + return a + } + if strings.Contains(b, a) { + return b + } + return a + "\n\n---\n\n" + b +} + // GenerateTenderID generates a unique tender ID using the PBL naming convention: SCDYYNNN // SCD = Source Code (3 letters), YY = Year (2 digits), NNN = Sequential number (variable digits) func (w *NoticeWorker) GenerateTenderID(ctx context.Context, t *notice.Notice) string { @@ -866,6 +973,25 @@ func (w *NoticeWorker) persistTender(ctx context.Context, t *tender.Tender, n *n return nil } lastErr = err + + // Another goroutine inserted the same procedure/publication first: merge into the winner row. + if isDuplicateKeyError(err) && (isDuplicateTenderBusinessKeyError(err) || isDuplicateContractFolderIDKeyError(err) || isDuplicateProcurementProjectIDKeyError(err)) { + tMerged, convErr := w.ToTender(n) + if convErr != nil { + return convErr + } + if tMerged.ID.IsZero() { + return fmt.Errorf("duplicate key on insert but could not resolve existing tender: %w", err) + } + w.Logger.Info("Merged notice into existing tender after duplicate-key race", map[string]interface{}{ + "existing_tender_id": tMerged.TenderID, + "notice_id": n.ID.Hex(), + "notice_publication_id": n.NoticePublicationID, + "contract_folder_id": n.ContractFolderID, + }) + return w.TenderRepo.Update(ctx, tMerged) + } + if !isDuplicateTenderIDError(err) { return err } @@ -886,18 +1012,12 @@ func (w *NoticeWorker) persistTender(ctx context.Context, t *tender.Tender, n *n return fmt.Errorf("exhausted %d tender_id retries: %w", tenderIDMaxRetries, lastErr) } -// isDuplicateTenderIDError returns true if err is the Mongo "E11000 duplicate key" error on the -// tender_id_1 index. We retry only that specific case; other write errors propagate. -func isDuplicateTenderIDError(err error) bool { +// isDuplicateKeyError is true for Mongo duplicate key (E11000) write failures. +func isDuplicateKeyError(err error) bool { if err == nil { return false } - // Cheap, driver-version-agnostic check: look for the error code and the specific index. - msg := err.Error() - if !strings.Contains(msg, "E11000") { - return false - } - if strings.Contains(msg, "tender_id_1") || strings.Contains(msg, "tender_id:") { + if strings.Contains(err.Error(), "E11000") { return true } var we mongodriver.WriteException @@ -911,6 +1031,50 @@ func isDuplicateTenderIDError(err error) bool { return false } +func isDuplicateTenderBusinessKeyError(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "notice_publication_id") +} + +func isDuplicateContractFolderIDKeyError(err error) bool { + if err == nil { + return false + } + msg := err.Error() + // Field name in dup key (avoid matching unrelated "contract" substrings). + return strings.Contains(msg, "contract_folder_id") +} + +func isDuplicateProcurementProjectIDKeyError(err error) bool { + if err == nil { + return false + } + return strings.Contains(err.Error(), "procurement_project_id") +} + +// isDuplicateTenderIDError returns true if err is the Mongo "E11000 duplicate key" error on the +// tender_id index. We retry only that specific case; other write errors propagate. +func isDuplicateTenderIDError(err error) bool { + if err == nil { + return false + } + msg := err.Error() + if !strings.Contains(msg, "E11000") { + return false + } + if strings.Contains(msg, "tender_id_1") || strings.Contains(msg, "tender_id:") { + return true + } + // Some drivers only include "dup key" and the field in the message. + if strings.Contains(msg, "dup key") && strings.Contains(msg, "tender_id") { + return true + } + return false +} + // padOrTruncate pads a string with zeros or truncates it to the specified length func PadOrTruncate(str string, length int) string { // Remove any non-alphanumeric characters and convert to uppercase diff --git a/internal/notice/entity.go b/internal/notice/entity.go index ce67e11..eb2fef8 100644 --- a/internal/notice/entity.go +++ b/internal/notice/entity.go @@ -14,6 +14,8 @@ type Notice struct { ContractNoticeID string `bson:"contract_notice_id" json:"contract_notice_id"` NoticePublicationID string `bson:"notice_publication_id" json:"notice_publication_id"` ContractFolderID string `bson:"contract_folder_id" json:"contract_folder_id"` + // ProcurementProjectID is UBL ProcurementProject/ID — stable across per-lot notices for one procedure. + ProcurementProjectID string `bson:"procurement_project_id,omitempty" json:"procurement_project_id,omitempty"` NoticeTypeCode string `bson:"notice_type_code" json:"notice_type_code"` FormType string `bson:"form_type,omitempty" json:"form_type,omitempty"` NoticeSubTypeCode string `bson:"notice_sub_type_code" json:"notice_sub_type_code"` diff --git a/internal/notice/repository.go b/internal/notice/repository.go index 60b1cd8..8e9e2b2 100644 --- a/internal/notice/repository.go +++ b/internal/notice/repository.go @@ -41,6 +41,9 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Re *orm.NewIndex("processing_metadata_processed_idx", bson.D{{Key: "processing_metadata.processed", Value: 1}}), *orm.NewIndex("main_classification_idx", bson.D{{Key: "main_classification", Value: 1}}), *orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}), + // One row per TED ContractNoticeID when present (parallel scrape races). + *orm.CreateUniqueIndex("contract_notice_id_unique", bson.D{{Key: "contract_notice_id", Value: 1}}). + WithPartialFilterExpression(bson.M{"contract_notice_id": bson.M{"$type": "string", "$ne": ""}}), } // Create indexes diff --git a/internal/tender/entity.go b/internal/tender/entity.go index d332cac..acd9e78 100644 --- a/internal/tender/entity.go +++ b/internal/tender/entity.go @@ -39,6 +39,8 @@ type Tender struct { // (same procedure / contract_folder_id). Ordered oldest-first; the latest is also reflected in NoticePublicationID. RelatedNoticePublicationIDs []string `bson:"related_notice_publication_ids,omitempty" json:"related_notice_publication_ids,omitempty"` ContractFolderID string `bson:"contract_folder_id" json:"contract_folder_id"` + // ProcurementProjectID is UBL ProcurementProject/ID — merges per-lot CAN/CN notices into one tender. + ProcurementProjectID string `bson:"procurement_project_id,omitempty" json:"procurement_project_id,omitempty"` NoticeTypeCode string `bson:"notice_type_code" json:"notice_type_code"` FormType string `bson:"form_type,omitempty" json:"form_type,omitempty"` NoticeSubTypeCode string `bson:"notice_sub_type_code" json:"notice_sub_type_code"` diff --git a/internal/tender/repository.go b/internal/tender/repository.go index e2ef240..41e940e 100644 --- a/internal/tender/repository.go +++ b/internal/tender/repository.go @@ -21,6 +21,7 @@ type TenderRepository interface { GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Tender, error) GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error) GetByContractFolderID(ctx context.Context, contractFolderID string) (*Tender, error) + GetByProcurementProjectID(ctx context.Context, procurementProjectID string) (*Tender, error) FindTendersWithContentXML(ctx context.Context, limit, skip int) ([]Tender, error) GetTenderCountByCountry(ctx context.Context) (map[string]int64, error) GetTenderCountByType(ctx context.Context) (map[string]int64, error) @@ -54,6 +55,14 @@ func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Te *orm.NewIndex("project_name_idx", bson.D{{Key: "project_name", Value: 1}}), *orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}), *orm.NewIndex("contract_folder_id_idx", bson.D{{Key: "contract_folder_id", Value: 1}}), + // One tender row per TED publication / procedure when ids are present (prevents parallel worker races). + *orm.CreateUniqueIndex("notice_publication_id_unique", bson.D{{Key: "notice_publication_id", Value: 1}}). + WithPartialFilterExpression(bson.M{"notice_publication_id": bson.M{"$type": "string", "$ne": ""}}), + *orm.CreateUniqueIndex("contract_folder_id_unique", bson.D{{Key: "contract_folder_id", Value: 1}}). + WithPartialFilterExpression(bson.M{"contract_folder_id": bson.M{"$type": "string", "$ne": ""}}), + // One tender per UBL procurement project (merges per-lot notices that share ProcurementProject/ID). + *orm.CreateUniqueIndex("procurement_project_id_unique", bson.D{{Key: "procurement_project_id", Value: 1}}). + WithPartialFilterExpression(bson.M{"procurement_project_id": bson.M{"$type": "string", "$ne": ""}}), *orm.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}), *orm.NewIndex("source_idx", bson.D{{Key: "source", Value: 1}}), *orm.NewIndex("country_code_idx", bson.D{{Key: "country_code", Value: 1}}), @@ -216,6 +225,31 @@ func (r *tenderRepository) GetByContractFolderID(ctx context.Context, contractFo return &result.Items[0], nil } +// GetByProcurementProjectID finds a tender by UBL ProcurementProject ID (shared across per-lot notices). +func (r *tenderRepository) GetByProcurementProjectID(ctx context.Context, procurementProjectID string) (*Tender, error) { + filter := bson.M{"procurement_project_id": procurementProjectID} + pagination := orm.Pagination{ + Limit: 1, + SortField: "updated_at", + SortOrder: -1, + } + + result, err := r.ormRepo.FindAll(ctx, filter, pagination) + if err != nil { + r.logger.Error("Failed to get tender by procurement project ID", map[string]interface{}{ + "procurement_project_id": procurementProjectID, + "error": err.Error(), + }) + return nil, err + } + + if len(result.Items) == 0 { + return nil, orm.ErrDocumentNotFound + } + + return &result.Items[0], nil +} + // FindTendersWithContentXML returns tenders that still have TED XML (for backfill after notices were deleted). func (r *tenderRepository) FindTendersWithContentXML(ctx context.Context, limit, skip int) ([]Tender, error) { filter := bson.M{"content_xml": bson.M{"$exists": true, "$ne": ""}} diff --git a/ted/mapper.go b/ted/mapper.go index 6e05b03..e3e76f4 100644 --- a/ted/mapper.go +++ b/ted/mapper.go @@ -138,8 +138,9 @@ func (s *TEDScraper) mapToTender(cn *ContractNotice, sourceFileURL, sourceFileNa now := time.Now().Unix() t := ¬ice.Notice{ - ContractNoticeID: cn.ID, - ContractFolderID: cn.ContractFolderID, + ContractNoticeID: cn.ID, + ContractFolderID: cn.ContractFolderID, + ProcurementProjectID: procurementProjectIDFromContractNotice(cn), NoticeTypeCode: strings.TrimSpace(cn.GetNoticeType()), FormType: cn.GetNoticeFormType(), NoticeSubTypeCode: cn.GetNoticeSubType(), @@ -297,8 +298,9 @@ func (s *TEDScraper) mapContractAwardNoticeToTender(can *ContractAwardNotice, so now := time.Now().Unix() t := ¬ice.Notice{ - ContractNoticeID: can.ID, - ContractFolderID: can.ContractFolderID, + ContractNoticeID: can.ID, + ContractFolderID: can.ContractFolderID, + ProcurementProjectID: procurementProjectIDFromContractAwardNotice(can), NoticeTypeCode: strings.TrimSpace(can.GetNoticeType()), FormType: can.GetNoticeFormType(), NoticeSubTypeCode: can.GetNoticeSubType(), @@ -428,6 +430,36 @@ func (s *TEDScraper) mapContractAwardNoticeToTender(can *ContractAwardNotice, so return t } +func procurementProjectIDFromContractNotice(cn *ContractNotice) string { + if cn == nil { + return "" + } + if cn.ProcurementProject != nil { + if id := strings.TrimSpace(cn.ProcurementProject.ID); id != "" { + return id + } + } + if lot := cn.firstLot(); lot != nil && lot.ProcurementProject != nil { + return strings.TrimSpace(lot.ProcurementProject.ID) + } + return "" +} + +func procurementProjectIDFromContractAwardNotice(can *ContractAwardNotice) string { + if can == nil { + return "" + } + if can.ProcurementProject != nil { + if id := strings.TrimSpace(can.ProcurementProject.ID); id != "" { + return id + } + } + if lot := can.firstAwardLot(); lot != nil && lot.ProcurementProject != nil { + return strings.TrimSpace(lot.ProcurementProject.ID) + } + return "" +} + // mapOrganization maps a TED Organization to tender Organization func (s *TEDScraper) mapOrganization(tedOrg *Organization, role string) *notice.Organization { if tedOrg == nil || tedOrg.Company == nil { diff --git a/ted/scraper.go b/ted/scraper.go index f177b1c..4bbb054 100644 --- a/ted/scraper.go +++ b/ted/scraper.go @@ -304,6 +304,14 @@ func (s *TEDScraper) findNoticeByContractNoticeID(ctx context.Context, contractN return existing, nil } +func isMongoDupKeyOnContractNoticeID(err error) bool { + if err == nil { + return false + } + msg := err.Error() + return strings.Contains(msg, "E11000") && strings.Contains(msg, "contract_notice_id") +} + func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, fileName string, result *FileProcessingResult) error { s.logger.Debug("Processing XML content", map[string]interface{}{ "file_name": fileName, @@ -382,6 +390,29 @@ func (s *TEDScraper) processXMLContent(ctx context.Context, content []byte, file err = s.noticeRepo.Import(ctx, t) if err != nil { + if isMongoDupKeyOnContractNoticeID(err) { + existingAfterRace, findErr := s.findNoticeByContractNoticeID(ctx, t.ContractNoticeID) + if findErr != nil { + return fmt.Errorf("import duplicate key, re-fetch notice: %w", findErr) + } + if existingAfterRace == nil { + return fmt.Errorf("failed to import notice: %w", err) + } + MergeNoticeUpdate(existingAfterRace, t) + if upErr := s.noticeRepo.Update(ctx, existingAfterRace); upErr != nil { + s.logger.Error("Failed to update notice after duplicate-key race", map[string]interface{}{ + "contract_notice_id": t.ContractNoticeID, + "error": upErr.Error(), + }) + return fmt.Errorf("failed to update notice after duplicate key: %w", upErr) + } + s.logger.Info("Notice merged after duplicate-key race (parallel ingest)", map[string]interface{}{ + "contract_notice_id": t.ContractNoticeID, + "notice_doc_id": existingAfterRace.ID.Hex(), + }) + result.SuccessCount++ + return nil + } s.logger.Error("Failed to import notice", map[string]interface{}{ "contract_notice_id": t.ContractNoticeID, "error": err.Error(),