Mazyar
1100 lines
36 KiB
Go
1100 lines
36 KiB
Go
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"
|
|
)
|
|
|
|
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 8).
|
|
Concurrency int
|
|
// FetchBatchSize is how many unprocessed notices to load from MongoDB per batch (default 50).
|
|
FetchBatchSize int
|
|
|
|
// 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) *NoticeWorker {
|
|
if processingLimit < 0 {
|
|
processingLimit = 5 // Default limit
|
|
}
|
|
if concurrency < 1 {
|
|
concurrency = 8
|
|
}
|
|
if fetchBatchSize < 1 {
|
|
fetchBatchSize = 50
|
|
}
|
|
return &NoticeWorker{
|
|
Mongo: mongo,
|
|
Logger: logger,
|
|
Notify: notify,
|
|
NoticeRepo: noticeRepo,
|
|
TenderRepo: tenderRepo,
|
|
ProcessingLimit: processingLimit,
|
|
DeleteProcessedNotices: deleteProcessedNotices,
|
|
Concurrency: concurrency,
|
|
FetchBatchSize: fetchBatchSize,
|
|
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() {
|
|
w.Logger.Info("Notice worker started", map[string]interface{}{
|
|
"concurrency": w.Concurrency,
|
|
"fetch_batch": w.FetchBatchSize,
|
|
"processing_limit": w.ProcessingLimit,
|
|
})
|
|
|
|
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(),
|
|
})
|
|
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 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,
|
|
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 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 = 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.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
|
|
t.ContentXML = n.ContentXML
|
|
// 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, preferring the procedure-level
|
|
// ContractFolderID (so follow-up notices like a Result for an earlier Competition notice merge into the
|
|
// 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
|
|
}
|
|
}
|
|
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:
|
|
// <InternalProjectNumber>_<Client-OpportunityName>_<PartnerCode>
|
|
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) || 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 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")
|
|
}
|