package workers import ( "context" "errors" "fmt" "strconv" "strings" "sync" "sync/atomic" "time" "tm/internal/notice" "tm/internal/tender" "tm/pkg/logger" "tm/pkg/mongo" "tm/pkg/notification" "go.mongodb.org/mongo-driver/v2/bson" mongodriver "go.mongodb.org/mongo-driver/v2/mongo" "go.mongodb.org/mongo-driver/v2/mongo/options" "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 // to avoid race conditions when several goroutines (or processes) generate ids in parallel. tenderIDCountersCollection = "tender_id_counters" // tenderIDMaxRetries caps how many times we retry allocating a new tender id when an insert // fails with a duplicate-key error (e.g. counter lagging behind existing rows). tenderIDMaxRetries = 5 ) type NoticeWorker struct { Mongo *mongo.ConnectionManager Logger logger.Logger Notify *notification.SDK NoticeRepo notice.Repository TenderRepo tender.TenderRepository ProcessingLimit int DeleteProcessedNotices bool // 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 // BatchPause is slept after each batch completes, before fetching the next batch (0 = disabled). BatchPause time.Duration // FetchErrorBackoff is slept after a failed notice fetch before retrying (0 = disabled). FetchErrorBackoff time.Duration // tenderIDSeedMu / tenderIDSeeded track which (sourceCode+year) counters have been seeded // from the current tenders collection during this process, so we only run the seeding // $max-upsert once per key. tenderIDSeedMu sync.Mutex tenderIDSeeded map[string]bool } func NewNoticeWorker(mongo *mongo.ConnectionManager, logger logger.Logger, notify *notification.SDK, noticeRepo notice.Repository, tenderRepo tender.TenderRepository, processingLimit, concurrency, fetchBatchSize int, deleteProcessedNotices bool, batchPause, fetchErrorBackoff time.Duration) *NoticeWorker { if processingLimit < 0 { processingLimit = 5 // Default limit } if concurrency < 1 { concurrency = 1 } if fetchBatchSize < 1 { fetchBatchSize = 50 } if batchPause < 0 { batchPause = 0 } if fetchErrorBackoff < 0 { fetchErrorBackoff = 0 } return &NoticeWorker{ Mongo: mongo, Logger: logger, Notify: notify, NoticeRepo: noticeRepo, TenderRepo: tenderRepo, ProcessingLimit: processingLimit, DeleteProcessedNotices: deleteProcessedNotices, Concurrency: concurrency, FetchBatchSize: fetchBatchSize, BatchPause: batchPause, FetchErrorBackoff: fetchErrorBackoff, tenderIDSeeded: make(map[string]bool), } } // cleanupProcessedNotices removes notices that have been successfully processed func (w *NoticeWorker) cleanupProcessedNotices() { w.Logger.Info("Starting cleanup of processed notices", map[string]interface{}{}) // Get all processed notices (limit to avoid memory issues) limit := 100 skip := 0 for { notices, _, err := w.NoticeRepo.GetProcessedNotices(context.Background(), limit, skip) if err != nil { w.Logger.Error("Failed to get processed notices for cleanup", map[string]interface{}{ "error": err.Error(), }) break } if len(notices) == 0 { break } deletedCount := 0 for _, notice := range notices { if err := w.NoticeRepo.Delete(context.Background(), notice.ID.Hex()); err != nil { w.Logger.Warn("Failed to delete processed notice during cleanup", map[string]interface{}{ "notice_id": notice.ID.Hex(), "error": err.Error(), }) } else { deletedCount++ } } w.Logger.Info("Cleanup batch completed", map[string]interface{}{ "processed": len(notices), "deleted": deletedCount, "skip": skip, }) skip += limit // Safety limit to avoid infinite loops if skip > 10000 { w.Logger.Warn("Cleanup reached safety limit, stopping", map[string]interface{}{ "skip": skip, }) break } } w.Logger.Info("Notice cleanup completed", map[string]interface{}{}) } 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, "processing_limit": w.ProcessingLimit, "batch_pause": w.BatchPause.String(), "fetch_error_backoff": w.FetchErrorBackoff.String(), }) var processedTotal atomic.Int64 maxToProcess := w.ProcessingLimit for maxToProcess == 0 || int(processedTotal.Load()) < maxToProcess { fetchLimit := w.FetchBatchSize if maxToProcess > 0 { remaining := maxToProcess - int(processedTotal.Load()) if remaining < 1 { break } if remaining < fetchLimit { fetchLimit = remaining } } notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), fetchLimit, 0) if err != nil { w.Logger.Error("Failed to get notices", map[string]interface{}{ "error": err.Error(), }) if w.FetchErrorBackoff > 0 { time.Sleep(w.FetchErrorBackoff) } continue } if len(notices) == 0 { w.Logger.Info("No more unprocessed notices found", map[string]interface{}{ "processed_count": processedTotal.Load(), }) break } var g errgroup.Group g.SetLimit(w.Concurrency) for i := range notices { n := notices[i] g.Go(func() error { w.Logger.Debug("Processing notice", map[string]interface{}{ "notice_id": n.ID.Hex(), }) t, convErr := w.ToTender(&n) if convErr != nil { w.Logger.Error("Failed to convert notice to tender, skipping notice", map[string]interface{}{ "error": convErr.Error(), "notice_id": n.ID.Hex(), }) n.ProcessingMetadata.Processed = true n.ProcessingMetadata.ProcessedAt = time.Now().Unix() if updateErr := w.NoticeRepo.Update(context.Background(), &n); updateErr != nil { w.Logger.Error("Failed to update notice after conversion error", map[string]interface{}{ "error": updateErr.Error(), "notice_id": n.ID.Hex(), }) } return nil } persistErr := w.persistTender(context.Background(), t, &n) if persistErr != nil { // Leave the notice unprocessed so it is retried on the next worker tick. w.Logger.Error("Failed to persist tender, notice left unprocessed for retry", map[string]interface{}{ "error": persistErr.Error(), "notice_id": n.ID.Hex(), }) return nil } n.ProcessingMetadata.Processed = true n.ProcessingMetadata.ProcessedAt = time.Now().Unix() if updateErr := w.NoticeRepo.Update(context.Background(), &n); updateErr != nil { w.Logger.Error("Failed to update notice", map[string]interface{}{ "error": updateErr.Error(), "notice_id": n.ID.Hex(), }) } else { processedTotal.Add(1) w.Logger.Debug("Notice processed successfully", map[string]interface{}{ "notice": n.ID.Hex(), "processed_count": processedTotal.Load(), }) } return nil }) } if err := g.Wait(); err != nil { w.Logger.Error("Notice worker batch failed", map[string]interface{}{ "error": err.Error(), }) } if w.BatchPause > 0 && len(notices) > 0 { time.Sleep(w.BatchPause) } if maxToProcess > 0 && int(processedTotal.Load()) >= maxToProcess { w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{ "processed_count": processedTotal.Load(), "max_to_process": maxToProcess, }) break } } // Clean up: delete notices that were marked as processed in previous runs (optional to retain XML for audit/backfill) if w.DeleteProcessedNotices { w.cleanupProcessedNotices() } else { w.Logger.Info("Skipping deletion of processed notices (WORKER_DELETE_PROCESSED_NOTICES=false)", map[string]interface{}{}) } } func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) { t := w.resolveExistingTender(context.Background(), n) if t == nil { w.Logger.Info("New tender raised", map[string]interface{}{ "notice_publication_id": n.NoticePublicationID, "contract_folder_id": n.ContractFolderID, "country_code": n.CountryCode, }) t = new(tender.Tender) } // Guard against out-of-order or duplicate notice ingestion: if this notice is older than the data // already on the tender (same notice id with an older BT-757 version, or a strictly earlier issue // date), don't overwrite the richer record. The notice is still marked processed by the caller. if !t.ID.IsZero() && isStaleNotice(t, n) { w.Logger.Info("Stale notice ignored, preserving existing tender", map[string]interface{}{ "tender_id": t.GetID(), "contract_folder_id": n.ContractFolderID, "incoming_notice_id": n.ContractNoticeID, "incoming_version": n.NoticeVersion, "incoming_issue_date": n.IssueDate, "existing_notice_id": t.ContractNoticeID, "existing_version": t.NoticeVersion, "existing_issue_date": t.IssueDate, "existing_status": string(t.Status), }) return t, nil } var ( preserveTenderID string preserveProjectName string preserveRelatedPubIDs []string previousNoticePublicationID string previousStatus tender.TenderStatus ) if !t.ID.IsZero() { preserveTenderID = t.TenderID preserveProjectName = t.ProjectName preserveRelatedPubIDs = append(preserveRelatedPubIDs, t.RelatedNoticePublicationIDs...) previousNoticePublicationID = t.NoticePublicationID previousStatus = t.Status } // Buyer Organization buyerOrg := &tender.Organization{} if n.BuyerOrganization != nil { buyerOrg = &tender.Organization{ Name: n.BuyerOrganization.Name, CompanyID: n.BuyerOrganization.CompanyID, WebsiteURI: n.BuyerOrganization.WebsiteURI, ContactName: n.BuyerOrganization.ContactName, ContactTelephone: n.BuyerOrganization.ContactTelephone, ContactEmail: n.BuyerOrganization.ContactEmail, ContactFax: n.BuyerOrganization.ContactFax, Role: n.BuyerOrganization.Role, Address: tender.Address{ StreetName: n.BuyerOrganization.Address.StreetName, CityName: n.BuyerOrganization.Address.CityName, PostalZone: n.BuyerOrganization.Address.PostalZone, CountrySubentityCode: n.BuyerOrganization.Address.CountrySubentityCode, Department: n.BuyerOrganization.Address.Department, Region: n.BuyerOrganization.Address.Region, CountryCode: n.BuyerOrganization.Address.CountryCode, }, } } // Review Organization reviewOrg := &tender.Organization{} if n.ReviewOrganization != nil { reviewOrg = &tender.Organization{ Name: n.ReviewOrganization.Name, CompanyID: n.ReviewOrganization.CompanyID, WebsiteURI: n.ReviewOrganization.WebsiteURI, ContactName: n.ReviewOrganization.ContactName, ContactTelephone: n.ReviewOrganization.ContactTelephone, ContactEmail: n.ReviewOrganization.ContactEmail, ContactFax: n.ReviewOrganization.ContactFax, Role: n.ReviewOrganization.Role, Address: tender.Address{ StreetName: n.ReviewOrganization.Address.StreetName, CityName: n.ReviewOrganization.Address.CityName, PostalZone: n.ReviewOrganization.Address.PostalZone, CountrySubentityCode: n.ReviewOrganization.Address.CountrySubentityCode, Department: n.ReviewOrganization.Address.Department, Region: n.ReviewOrganization.Address.Region, CountryCode: n.ReviewOrganization.Address.CountryCode, }, } } // Organizations organizations := make([]tender.Organization, len(n.Organizations)) for i, org := range n.Organizations { organizations[i] = tender.Organization{ Name: org.Name, CompanyID: org.CompanyID, WebsiteURI: org.WebsiteURI, ContactName: org.ContactName, ContactTelephone: org.ContactTelephone, ContactEmail: org.ContactEmail, ContactFax: org.ContactFax, Role: org.Role, Address: tender.Address{ StreetName: org.Address.StreetName, CityName: org.Address.CityName, PostalZone: org.Address.PostalZone, CountrySubentityCode: org.Address.CountrySubentityCode, Department: org.Address.Department, Region: org.Address.Region, CountryCode: org.Address.CountryCode, }, } } // Selection Criteria selectionCriteria := make([]tender.SelectionCriterion, len(n.SelectionCriteria)) for i, criterion := range n.SelectionCriteria { selectionCriteria[i] = tender.SelectionCriterion{ TypeCode: criterion.TypeCode, Description: criterion.Description, LanguageID: criterion.LanguageID, } } // Winning Tenderer winningTenderer := &tender.Organization{} if n.WinningTenderer != nil { winningTenderer = &tender.Organization{ Name: n.WinningTenderer.Name, CompanyID: n.WinningTenderer.CompanyID, WebsiteURI: n.WinningTenderer.WebsiteURI, ContactName: n.WinningTenderer.ContactName, ContactTelephone: n.WinningTenderer.ContactTelephone, ContactEmail: n.WinningTenderer.ContactEmail, ContactFax: n.WinningTenderer.ContactFax, Role: n.WinningTenderer.Role, Address: tender.Address{ StreetName: n.WinningTenderer.Address.StreetName, CityName: n.WinningTenderer.Address.CityName, PostalZone: n.WinningTenderer.Address.PostalZone, CountrySubentityCode: n.WinningTenderer.Address.CountrySubentityCode, Department: n.WinningTenderer.Address.Department, Region: n.WinningTenderer.Address.Region, CountryCode: n.WinningTenderer.Address.CountryCode, }, } } // Modifications modifications := make([]tender.TenderModification, len(n.Modifications)) for i, modification := range n.Modifications { modifications[i] = tender.TenderModification{ ModificationDate: modification.ModificationDate, ModificationReason: modification.ModificationReason, Description: modification.Description, LanguageID: modification.LanguageID, } } // Awarded Entities awardedEntities := make([]tender.Awarded, len(n.AwardedEntities)) for i, awarded := range n.AwardedEntities { awardedEntities[i] = tender.Awarded{ Name: awarded.Name, Address: awarded.Address, Country: awarded.Country, Amount: awarded.Amount, Currency: awarded.Currency, Share: awarded.Share, AwardDate: awarded.AwardDate, ContractID: awarded.ContractID, TenderID: awarded.TenderID, LotID: awarded.LotID, CompanyID: awarded.CompanyID, OrganizationID: awarded.OrganizationID, } } procurementLots := make([]tender.ProcurementLot, len(n.ProcurementLots)) for i, l := range n.ProcurementLots { procurementLots[i] = tender.ProcurementLot{ LotID: l.LotID, Title: l.Title, Description: l.Description, MainNatureOfContract: l.MainNatureOfContract, MainClassification: l.MainClassification, MainClassificationDescription: l.MainClassificationDescription, AdditionalClassifications: append([]string(nil), l.AdditionalClassifications...), Duration: l.Duration, EstimatedValue: l.EstimatedValue, Currency: l.Currency, } } 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 estimatedVal == 0 && t.EstimatedValue > 0 { estimatedVal = t.EstimatedValue } if awardedVal == 0 && t.AwardedValue > 0 { awardedVal = t.AwardedValue } if currency == "" && t.Currency != "" { currency = t.Currency } 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 } } } if estimatedVal == 0 { if lotValue, lotCurrency := tender.AggregateProcurementLotEstimatedValue(procurementLots); lotValue > 0 { estimatedVal = lotValue if currency == "" { currency = lotCurrency } } } // Title and description stay in the notice language; English translation is handled by the AI service (translation worker). t.Title = title t.Description = description t.NoticePublicationID = n.NoticePublicationID t.ProcurementTypeCode = n.ProcurementTypeCode t.ProcedureCode = n.ProcedureCode t.EstimatedValue = estimatedVal t.Currency = currency t.TenderDeadline = n.TenderDeadline t.SubmissionURL = n.SubmissionURL t.BuyerOrganization = buyerOrg t.ReviewOrganization = reviewOrg t.Organizations = organizations t.SelectionCriteria = selectionCriteria t.WinningTenderer = winningTenderer t.Modifications = modifications t.AwardedEntities = awardedEntities t.ProcurementLots = procurementLots t.NoticeTypeCode = n.NoticeTypeCode t.FormType = n.FormType t.NoticeSubTypeCode = n.NoticeSubTypeCode t.NoticeIdentifier = n.NoticeIdentifier t.NoticeVersion = n.NoticeVersion t.NoticeDispatchAt = n.NoticeDispatchAt t.ESenderDispatchAt = n.ESenderDispatchAt t.NoticeLanguageCode = n.NoticeLanguageCode t.IssueDate = n.IssueDate t.IssueTime = n.IssueTime t.PublicationDate = n.PublicationDate t.SubmissionDeadline = n.SubmissionDeadline t.ApplicationDeadline = n.ApplicationDeadline t.GazetteID = n.GazetteID t.PlaceOfPerformance = n.PlaceOfPerformance t.CountryCode = n.CountryCode t.RegionCode = n.RegionCode t.CityName = n.CityName t.PostalCode = n.PostalCode t.DocumentURI = n.DocumentURI t.BuyerProfileURL = n.BuyerProfileURL t.TenderURL = n.TenderURL t.MainClassification = n.MainClassification t.MainClassificationDescription = n.MainClassificationDescription t.AdditionalClassifications = n.AdditionalClassifications t.OfficialLanguages = n.OfficialLanguages t.Status = chooseTenderStatus(previousStatus, tender.TenderStatus(n.Status)) 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 // Preserve document-scraping/summarization flags on the tender (managed by // downstream workers), but refresh the notice-side processing metadata. t.ProcessingMetadata.ScrapedAt = n.ProcessingMetadata.ScrapedAt t.ProcessingMetadata.ProcessedAt = n.ProcessingMetadata.ProcessedAt t.ProcessingMetadata.ProcessingVersion = n.ProcessingMetadata.ProcessingVersion t.ProcessingMetadata.ParsingErrors = n.ProcessingMetadata.ParsingErrors t.ProcessingMetadata.ValidationErrors = n.ProcessingMetadata.ValidationErrors t.ProcessingMetadata.EnrichmentData = n.ProcessingMetadata.EnrichmentData t.ProcessingMetadata.TranslatedData = n.ProcessingMetadata.TranslatedData t.ProcessingMetadata.TranslatedAt = n.ProcessingMetadata.TranslatedAt t.ProcessingMetadata.Processed = n.ProcessingMetadata.Processed t.CancellationReason = n.CancellationReason t.CancellationDate = n.CancellationDate t.AwardDate = n.AwardDate t.AwardedValue = awardedVal t.ContractNumber = n.ContractNumber t.SuspensionReason = n.SuspensionReason t.SuspensionDate = n.SuspensionDate if preserveTenderID != "" { t.TenderID = preserveTenderID t.ProjectName = preserveProjectName } else { t.TenderID = w.GenerateTenderID(context.Background(), n) t.ProjectName = w.GenerateProjectName(t) } t.RelatedNoticePublicationIDs = mergeRelatedPublicationIDs( preserveRelatedPubIDs, previousNoticePublicationID, n.NoticePublicationID, ) return t, nil } // resolveExistingTender finds the tender that this notice should update. TED ContractNoticeID uniquely // identifies the notice document; try it first so we never open a second tender row for the same notice. // Then procedure-level ContractFolderID (follow-up notices merge), UBL ProcurementProject/ID (per-lot), // then NoticePublicationID. func (w *NoticeWorker) resolveExistingTender(ctx context.Context, n *notice.Notice) *tender.Tender { if strings.TrimSpace(n.ContractNoticeID) != "" { if t, err := w.TenderRepo.GetByContractNoticeID(ctx, n.ContractNoticeID); err == nil && t != nil { return t } } 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 } } return nil } // isStaleNotice reports whether an incoming notice is older than what the existing tender already // reflects, and therefore should not overwrite it. Two cases qualify as stale: // 1. Same TED contract notice id (a re-scrape of the same document) with an older BT-757 NoticeVersion. // 2. A strictly earlier IssueDate than the tender's current IssueDate (an out-of-order publication on // the same procedure, e.g. a stale CN arriving after a CAN has already been ingested). func isStaleNotice(existing *tender.Tender, n *notice.Notice) bool { if existing == nil || existing.ID.IsZero() { return false } if existing.ContractNoticeID != "" && existing.ContractNoticeID == n.ContractNoticeID { if compareNoticeVersions(n.NoticeVersion, existing.NoticeVersion) < 0 { return true } } if existing.IssueDate > 0 && n.IssueDate > 0 && n.IssueDate < existing.IssueDate { return true } return false } // parseNoticeVersion parses the BT-757 numeric notice version (e.g. "01", "02"). Empty or unparsable // strings yield 0, which matches the convention used on the notice-ingest side. func parseNoticeVersion(s string) int { s = strings.TrimSpace(s) if s == "" { return 0 } v, err := strconv.Atoi(s) if err != nil { return 0 } return v } // compareNoticeVersions returns -1/0/1 for incoming vs existing notice versions. func compareNoticeVersions(incoming, existing string) int { vi := parseNoticeVersion(incoming) ve := parseNoticeVersion(existing) switch { case vi > ve: return 1 case vi < ve: return -1 default: return 0 } } // chooseTenderStatus enforces lifecycle precedence so a stale earlier-stage notice cannot downgrade a // tender that has already reached a later stage. The ranks follow the TED lifecycle: // draft < active/published < modified < suspended < expired < closed < awarded < cancelled. // Cancellation is the strongest terminal state so an Awarded tender can still be moved to Cancelled // (a procedure can be cancelled post-award), but a stale Active CN cannot downgrade an Awarded tender. // Equal-or-higher incoming ranks are accepted; strictly lower ranks are rejected. func chooseTenderStatus(existing, incoming tender.TenderStatus) tender.TenderStatus { if existing == "" { return incoming } if incoming == "" { return existing } if tenderStatusRank(existing) > tenderStatusRank(incoming) { return existing } return incoming } func tenderStatusRank(s tender.TenderStatus) int { switch s { case tender.TenderStatusCancelled: return 7 case tender.TenderStatusAwarded: return 6 case tender.TenderStatusClosed: return 5 case tender.TenderStatusExpired: return 4 case tender.TenderStatusSuspended: return 3 case tender.TenderStatusModified: return 2 case tender.TenderStatusActive, tender.TenderStatusPublished: return 1 case tender.TenderStatusDraft: return 0 default: return 1 } } // mergeRelatedPublicationIDs appends previous and current publication ids to the existing history, // preserving order and deduplicating. The latest publication id (passed via current) lives on // Tender.NoticePublicationID; this list tracks every publication that contributed to this tender so // historical TED detail URLs stay resolvable. func mergeRelatedPublicationIDs(existing []string, previous, current string) []string { seen := make(map[string]struct{}, len(existing)+2) out := make([]string, 0, len(existing)+2) add := func(id string) { id = strings.TrimSpace(id) if id == "" { return } if _, ok := seen[id]; ok { return } seen[id] = struct{}{} out = append(out, id) } for _, id := range existing { add(id) } add(previous) add(current) if len(out) == 0 { return nil } 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 { sourceCode := getSourceCode(t) year := getCurrentYear() sequentialNumber := w.getNextSequentialNumber(ctx, sourceCode, year) return fmt.Sprintf("%s%s%d", sourceCode, year, sequentialNumber) } // getSourceCode determines the 3-letter source code based on tender characteristics func getSourceCode(t *notice.Notice) string { // Since all notices come from TED (Tenders Electronic Daily), use PTD for Public Tender return "PTD" } // getCurrentYear returns the last two digits of the current year func getCurrentYear() string { now := time.Now() return fmt.Sprintf("%02d", now.Year()%100) } // GenerateProjectName generates the project name using PBL naming convention: // __ func (w *NoticeWorker) GenerateProjectName(t *tender.Tender) string { clientName := w.getClientName(t) opportunityName := w.getOpportunityName(t) // Clean and format the names (replace spaces with hyphens, remove special chars) clientName = cleanForProjectName(clientName) opportunityName = cleanForProjectName(opportunityName) // If client name and opportunity name are the same, just use one of them if clientName == opportunityName { return fmt.Sprintf("%s_%s_PBL", t.TenderID, clientName) } return fmt.Sprintf("%s_%s-%s_PBL", t.TenderID, clientName, opportunityName) } // getClientName extracts the client name from the tender func (w *NoticeWorker) getClientName(t *tender.Tender) string { if t.BuyerOrganization != nil && t.BuyerOrganization.Name != "" { return t.BuyerOrganization.Name } return "UnknownClient" } // getOpportunityName extracts the opportunity name from the tender title func (w *NoticeWorker) getOpportunityName(t *tender.Tender) string { if t.Title != "" { // Take first few words of the title as opportunity name words := strings.Fields(t.Title) if len(words) > 3 { return strings.Join(words[:3], " ") } return t.Title } return "UnknownOpportunity" } // cleanForProjectName cleans a string for use in project names (removes special chars, replaces spaces with hyphens) func cleanForProjectName(s string) string { // Replace spaces with hyphens s = strings.ReplaceAll(s, " ", "-") // Remove special characters except hyphens and alphanumeric clean := "" for _, r := range s { if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '-' { clean += string(r) } } // Remove multiple consecutive hyphens for strings.Contains(clean, "--") { clean = strings.ReplaceAll(clean, "--", "-") } // Trim hyphens from start and end clean = strings.Trim(clean, "-") return clean } // getNextSequentialNumber atomically allocates the next sequential number for the given // (sourceCode, year) pair via a Mongo counter document. The counter is seeded once per process // per key from the current max tender_id in the tenders collection, then incremented atomically // with $inc so two parallel goroutines (or processes) can never get the same value. func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode, year string) int { key := fmt.Sprintf("%s%s", sourceCode, year) if err := w.ensureTenderIDCounterSeeded(ctx, sourceCode, year, key); err != nil { w.Logger.Error("Failed to seed tender id counter, falling back to live max query", map[string]interface{}{ "error": err.Error(), "key": key, }) return w.queryMaxTenderSeq(ctx, sourceCode, year) + 1 } var res struct { Seq int `bson:"seq"` } err := w.Mongo.GetCollection(tenderIDCountersCollection).FindOneAndUpdate( ctx, bson.M{"_id": key}, bson.M{"$inc": bson.M{"seq": 1}}, options.FindOneAndUpdate().SetUpsert(true).SetReturnDocument(options.After), ).Decode(&res) if err != nil { w.Logger.Error("Failed to increment tender id counter, falling back to live max query", map[string]interface{}{ "error": err.Error(), "key": key, }) return w.queryMaxTenderSeq(ctx, sourceCode, year) + 1 } return res.Seq } // ensureTenderIDCounterSeeded sets the counter to at least the current max sequential number // observed in the tenders collection. It is idempotent (uses $max) and only runs once per key // per process. The first call across all processes pays the aggregation cost; subsequent calls // are no-ops at the process level and (worst case) one extra cheap $max operation at Mongo. func (w *NoticeWorker) ensureTenderIDCounterSeeded(ctx context.Context, sourceCode, year, key string) error { w.tenderIDSeedMu.Lock() defer w.tenderIDSeedMu.Unlock() if w.tenderIDSeeded[key] { return nil } currentMax := w.queryMaxTenderSeq(ctx, sourceCode, year) _, err := w.Mongo.GetCollection(tenderIDCountersCollection).UpdateOne( ctx, bson.M{"_id": key}, bson.M{ "$max": bson.M{"seq": currentMax}, "$setOnInsert": bson.M{"_id": key}, }, options.UpdateOne().SetUpsert(true), ) if err != nil { return fmt.Errorf("seed counter %s: %w", key, err) } w.tenderIDSeeded[key] = true w.Logger.Info("Tender id counter seeded", map[string]interface{}{ "key": key, "seed_value": currentMax, "next_id": fmt.Sprintf("%s%d", key, currentMax+1), }) return nil } // queryMaxTenderSeq returns the highest NNN suffix found in the tenders collection for tender_ids // matching {sourceCode}{year}NNN, or 0 if none exist. Used to seed the counter. func (w *NoticeWorker) queryMaxTenderSeq(ctx context.Context, sourceCode, year string) int { prefix := fmt.Sprintf("%s%s", sourceCode, year) pattern := fmt.Sprintf("^%s\\d+$", prefix) pipeline := mongodriver.Pipeline{ bson.D{{Key: "$match", Value: bson.M{ "tender_id": bson.M{"$regex": pattern}, }}}, bson.D{{Key: "$project", Value: bson.M{ "sequential_num": bson.M{"$substrCP": bson.A{"$tender_id", len(prefix), bson.M{"$strLenCP": "$tender_id"}}}, }}}, bson.D{{Key: "$group", Value: bson.M{ "_id": nil, "max_num": bson.M{"$max": bson.M{"$toInt": "$sequential_num"}}, }}}, } cursor, err := w.Mongo.GetCollection("tenders").Aggregate(ctx, pipeline) if err != nil { w.Logger.Error("Failed to aggregate tenders for sequential numbering", map[string]interface{}{ "error": err.Error(), "source_code": sourceCode, "year": year, }) return 0 } defer cursor.Close(ctx) var result struct { MaxNum int `bson:"max_num"` } if cursor.Next(ctx) { if err := cursor.Decode(&result); err == nil { return result.MaxNum } } return 0 } // persistTender writes the given tender to Mongo. For new tenders (zero ID) it retries on // duplicate-key errors against tender_id by allocating a fresh sequential number; this guards // against a stale counter or external writes that bypass the counter. For existing tenders it // is a straight update. func (w *NoticeWorker) persistTender(ctx context.Context, t *tender.Tender, n *notice.Notice) error { if !t.ID.IsZero() { return w.TenderRepo.Update(ctx, t) } var lastErr error for attempt := 0; attempt < tenderIDMaxRetries; attempt++ { err := w.TenderRepo.Create(ctx, t) if err == nil { return nil } lastErr = err // Another goroutine inserted the same procedure/publication first: merge into the winner row. if isDuplicateKeyError(err) && (isDuplicateTenderBusinessKeyError(err) || isDuplicateContractNoticeIDKeyError(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 } // Re-allocate a tender id and refresh the project name. Reset Mongo's ObjectID so the // repo treats this as a fresh insert on the next attempt. t.ID = bson.NilObjectID t.TenderID = w.GenerateTenderID(ctx, n) t.ProjectName = w.GenerateProjectName(t) w.Logger.Warn("Duplicate tender_id on insert, retrying with new id", map[string]interface{}{ "attempt": attempt + 1, "new_tender_id": t.TenderID, "notice_id": n.ID.Hex(), "contract_folder": n.ContractFolderID, }) } return fmt.Errorf("exhausted %d tender_id retries: %w", tenderIDMaxRetries, lastErr) } // isDuplicateKeyError is true for Mongo duplicate key (E11000) write failures. func isDuplicateKeyError(err error) bool { if err == nil { return false } if strings.Contains(err.Error(), "E11000") { return true } var we mongodriver.WriteException if errors.As(err, &we) { for _, wErr := range we.WriteErrors { if wErr.Code == 11000 { return true } } } return false } func isDuplicateTenderBusinessKeyError(err error) bool { if err == nil { return false } msg := err.Error() return strings.Contains(msg, "notice_publication_id") } func isDuplicateContractNoticeIDKeyError(err error) bool { if err == nil { return false } return strings.Contains(err.Error(), "contract_notice_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 clean := strings.ToUpper(strings.ReplaceAll(str, "-", "")) clean = strings.ReplaceAll(clean, " ", "") if len(clean) >= length { return clean[:length] } // Pad with zeros return clean + strings.Repeat("0", length-len(clean)) } // unixToDateString converts Unix to date string for TenderID generation func UnixToDateString(unix int64) string { if unix == 0 { return "00000000" } return time.Unix(unix, 0).Format("20060102") }