Mazyar
b388af3518
- Eliminated the `ContentXML` field from the `Tender` struct in the entity definition, streamlining the data model. - Removed the `FindTendersWithContentXML` method from the `TenderRepository` interface, as it is no longer necessary. - Updated the `ToTender` method in the `NoticeWorker` to reflect these changes, ensuring consistency across the codebase. This update simplifies the tender data structure and repository interface, improving maintainability and clarity in the tender management system.
1161 lines
38 KiB
Go
1161 lines
38 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"
|
|
)
|
|
|
|
// 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:
|
|
// <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) || 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")
|
|
}
|