merge tenders which have the same procedure id - worker performance fix

cmd/worker/workers/notice.go

@@ -2,44 +2,76 @@ package workers
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 	"tm/internal/notice"
 	"tm/internal/tender"
-	"tm/pkg/glm"
 	"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
-	GLM                     *glm.SDK
-	ProcessingLimit         int
-	DeleteProcessedNotices  bool
+	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, glmService *glm.SDK, processingLimit int, deleteProcessedNotices bool) *NoticeWorker {
+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,
-		GLM:                    glmService,
 		ProcessingLimit:        processingLimit,
 		DeleteProcessedNotices: deleteProcessedNotices,
+		Concurrency:            concurrency,
+		FetchBatchSize:         fetchBatchSize,
+		tenderIDSeeded:         make(map[string]bool),
 	}
 }
 
@@ -97,23 +129,27 @@ func (w *NoticeWorker) cleanupProcessedNotices() {
 }
 
 func (w *NoticeWorker) Run() {
-	w.Logger.Info("Notice worker started", map[string]interface{}{})
+	w.Logger.Info("Notice worker started", map[string]interface{}{
+		"concurrency":      w.Concurrency,
+		"fetch_batch":      w.FetchBatchSize,
+		"processing_limit": w.ProcessingLimit,
+	})
 
-	processedCount := 0
+	var processedTotal atomic.Int64
 	maxToProcess := w.ProcessingLimit
-	batchSize := 10
 
-	for maxToProcess == 0 || processedCount < maxToProcess {
-		// Calculate how many to fetch in this batch
-		fetchLimit := batchSize
+	for maxToProcess == 0 || int(processedTotal.Load()) < maxToProcess {
+		fetchLimit := w.FetchBatchSize
 		if maxToProcess > 0 {
-			remaining := maxToProcess - processedCount
+			remaining := maxToProcess - int(processedTotal.Load())
+			if remaining < 1 {
+				break
+			}
 			if remaining < fetchLimit {
 				fetchLimit = remaining
 			}
 		}
 
-		// Always skip=0 since processed notices are excluded by the filter
 		notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), fetchLimit, 0)
 		if err != nil {
 			w.Logger.Error("Failed to get notices", map[string]interface{}{
@@ -124,77 +160,75 @@ func (w *NoticeWorker) Run() {
 
 		if len(notices) == 0 {
 			w.Logger.Info("No more unprocessed notices found", map[string]interface{}{
-				"processed_count": processedCount,
+				"processed_count": processedTotal.Load(),
 			})
 			break
 		}
 
-		for _, n := range notices {
-			if maxToProcess > 0 && processedCount >= maxToProcess {
-				w.Logger.Info("Reached maximum processing limit, stopping", map[string]interface{}{
-					"processed_count": processedCount,
-					"max_to_process":  maxToProcess,
-				})
-				break
-			}
-
-			w.Logger.Info("Notice", map[string]interface{}{
-				"notice": n.ID.Hex(),
-			})
-
-			t, err := w.ToTender(&n)
-			if err != nil {
-				w.Logger.Error("Failed to convert notice to tender, skipping notice", map[string]interface{}{
-					"error":     err.Error(),
+		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(),
 				})
-				// Mark notice as processed so it doesn't block the queue
+
+				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 after conversion error", map[string]interface{}{
+					w.Logger.Error("Failed to update notice", map[string]interface{}{
 						"error":     updateErr.Error(),
 						"notice_id": n.ID.Hex(),
 					})
-				}
-				continue
-			}
-
-			if t.ID.IsZero() {
-				err = w.TenderRepo.Create(context.Background(), t)
-				if err != nil {
-					w.Logger.Error("Failed to create tender", map[string]interface{}{
-						"error": err.Error(),
+				} else {
+					processedTotal.Add(1)
+					w.Logger.Debug("Notice processed successfully", map[string]interface{}{
+						"notice":          n.ID.Hex(),
+						"processed_count": processedTotal.Load(),
 					})
 				}
-			} else {
-				err = w.TenderRepo.Update(context.Background(), t)
-				if err != nil {
-					w.Logger.Error("Failed to update tender", map[string]interface{}{
-						"error": err.Error(),
-					})
-				}
-			}
-
-			// Mark notice as processed so it's excluded from future queries
-			n.ProcessingMetadata.Processed = true
-			n.ProcessingMetadata.ProcessedAt = time.Now().Unix()
-			err = w.NoticeRepo.Update(context.Background(), &n)
-			if err != nil {
-				w.Logger.Error("Failed to update notice", map[string]interface{}{
-					"error": err.Error(),
-				})
-			} else {
-				processedCount++
-				w.Logger.Info("Notice processed successfully", map[string]interface{}{
-					"notice":          n.ID.Hex(),
-					"processed_count": processedCount,
-				})
-			}
+				return nil
+			})
+		}
+		if err := g.Wait(); err != nil {
+			w.Logger.Error("Notice worker batch failed", map[string]interface{}{
+				"error": err.Error(),
+			})
 		}
 
-		// Check if we need to exit outer loop (break from inner loop means we hit the limit)
-		if maxToProcess > 0 && processedCount >= maxToProcess {
+		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
 		}
 	}
@@ -208,19 +242,47 @@ func (w *NoticeWorker) Run() {
 }
 
 func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) {
-	t, err := w.TenderRepo.GetByNoticePublicationID(context.Background(), n.NoticePublicationID)
-	if err != nil {
+	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)
 	}
 
-	var preserveTenderID, preserveProjectName string
-	if t != nil && !t.ID.IsZero() {
+	// 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
@@ -374,47 +436,9 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) {
 		}
 	}
 
-	var title string
-	if w.GLM != nil {
-		title, err = w.GLM.Translate(context.Background(), n.Title, n.NoticeLanguageCode, "en")
-		if err != nil {
-			w.Logger.Error("Failed to translate title", map[string]interface{}{
-				"notice_publication_id": n.NoticePublicationID,
-				"country_code":          n.CountryCode,
-				"content":               n.Title,
-				"error":                 err.Error(),
-			})
-			title = n.Title
-		}
-	} else {
-		w.Logger.Warn("GLM service not available, using original title", map[string]interface{}{
-			"notice_publication_id": n.NoticePublicationID,
-			"country_code":          n.CountryCode,
-		})
-		title = n.Title
-	}
-
-	var description string
-	if w.GLM != nil {
-		description, err = w.GLM.Translate(context.Background(), n.Description, n.NoticeLanguageCode, "en")
-		if err != nil {
-			w.Logger.Error("Failed to translate description", map[string]interface{}{
-				"notice_publication_id": n.NoticePublicationID,
-				"country_code":          n.CountryCode,
-				"content":               n.Description,
-				"error":                 err.Error(),
-			})
-			description = n.Description
-		}
-	} else {
-		w.Logger.Warn("GLM service not available, using original description", map[string]interface{}{
-			"notice_publication_id": n.NoticePublicationID,
-			"country_code":          n.CountryCode,
-		})
-		description = n.Description
-	}
-	t.Title = title
-	t.Description = description
+	// Title and description stay in the notice language; English translation is handled by the AI service (translation worker).
+	t.Title = n.Title
+	t.Description = n.Description
 	t.NoticePublicationID = n.NoticePublicationID
 	t.ProcurementTypeCode = n.ProcurementTypeCode
 	t.ProcedureCode = n.ProcedureCode
@@ -455,7 +479,7 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) {
 	t.MainClassification = n.MainClassification
 	t.AdditionalClassifications = n.AdditionalClassifications
 	t.OfficialLanguages = n.OfficialLanguages
-	t.Status = tender.TenderStatus(n.Status)
+	t.Status = chooseTenderStatus(previousStatus, tender.TenderStatus(n.Status))
 	t.Source = tender.TenderSource(n.Source)
 	t.ContractNoticeID = n.ContractNoticeID
 	t.ContractFolderID = n.ContractFolderID
@@ -491,9 +515,151 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) {
 		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) and falling back to NoticePublicationID for older rows or notices without a folder id.
+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.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
+}
+
 // 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 {
@@ -575,16 +741,80 @@ func cleanForProjectName(s string) string {
 	return clean
 }
 
-// getNextSequentialNumber returns the next sequential number for the given source code and year
+// 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 {
-	// Query tenders with tender IDs that match the pattern: {sourceCode}{year}*
-	// We need to find the highest NNN value for this source code and year
+	key := fmt.Sprintf("%s%s", sourceCode, year)
 
-	// Pattern to match: e.g., "PTD25" followed by one or more digits
+	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)
 
-	// Create aggregation pipeline to find the highest sequential number
 	pipeline := mongodriver.Pipeline{
 		bson.D{{Key: "$match", Value: bson.M{
 			"tender_id": bson.M{"$regex": pattern},
@@ -598,7 +828,6 @@ func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode,
 		}}},
 	}
 
-	// Execute aggregation
 	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{}{
@@ -606,22 +835,80 @@ func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode,
 			"source_code": sourceCode,
 			"year":        year,
 		})
-		return 1 // Return 1 as fallback
+		return 0
 	}
 	defer cursor.Close(ctx)
 
-	// Parse result
 	var result struct {
 		MaxNum int `bson:"max_num"`
 	}
-
 	if cursor.Next(ctx) {
 		if err := cursor.Decode(&result); err == nil {
-			return result.MaxNum + 1
+			return result.MaxNum
 		}
 	}
+	return 0
+}
 
-	return 1 // No existing tenders found, start with 1
+// 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
+		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)
+}
+
+// isDuplicateTenderIDError returns true if err is the Mongo "E11000 duplicate key" error on the
+// tender_id_1 index. We retry only that specific case; other write errors propagate.
+func isDuplicateTenderIDError(err error) bool {
+	if err == nil {
+		return false
+	}
+	// Cheap, driver-version-agnostic check: look for the error code and the specific index.
+	msg := err.Error()
+	if !strings.Contains(msg, "E11000") {
+		return false
+	}
+	if strings.Contains(msg, "tender_id_1") || strings.Contains(msg, "tender_id:") {
+		return true
+	}
+	var we mongodriver.WriteException
+	if errors.As(err, &we) {
+		for _, wErr := range we.WriteErrors {
+			if wErr.Code == 11000 {
+				return true
+			}
+		}
+	}
+	return false
 }
 
 // padOrTruncate pads a string with zeros or truncates it to the specified length