worker bug fix

cmd/worker/workers/notice.go

@@ -100,14 +100,22 @@ func (w *NoticeWorker) cleanupProcessedNotices() {
 func (w *NoticeWorker) Run() {
 	w.Logger.Info("Notice worker started", map[string]interface{}{})
 
-	w.cleanupProcessedNotices()
-
-	skip := 0
 	processedCount := 0
 	maxToProcess := w.ProcessingLimit
+	batchSize := 10
 
 	for maxToProcess == 0 || processedCount < maxToProcess {
-		notices, _, err := w.NoticeRepo.GetUnProcessedNotices(context.Background(), skip, maxToProcess)
+		// Calculate how many to fetch in this batch
+		fetchLimit := batchSize
+		if maxToProcess > 0 {
+			remaining := maxToProcess - processedCount
+			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{}{
 				"error": err.Error(),
@@ -137,9 +145,19 @@ func (w *NoticeWorker) Run() {
 
 			t, err := w.ToTender(&n)
 			if err != nil {
-				w.Logger.Error("Failed to create tender", map[string]interface{}{
-					"error": err.Error(),
+				w.Logger.Error("Failed to convert notice to tender, skipping notice", map[string]interface{}{
+					"error":     err.Error(),
+					"notice_id": n.ID.Hex(),
 				})
+				// Mark notice as processed so it doesn't block the queue
+				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(),
+					})
+				}
 				continue
 			}
 
@@ -158,6 +176,8 @@ func (w *NoticeWorker) Run() {
 					})
 				}
 			}
+
+			// 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)
@@ -171,28 +191,17 @@ func (w *NoticeWorker) Run() {
 					"notice":          n.ID.Hex(),
 					"processed_count": processedCount,
 				})
-
-				// Clean up: Delete processed notice from MongoDB
-				if err := w.NoticeRepo.Delete(context.Background(), n.ID.Hex()); err != nil {
-					w.Logger.Warn("Failed to delete processed notice", map[string]interface{}{
-						"notice_id": n.ID.Hex(),
-						"error":     err.Error(),
-					})
-				} else {
-					w.Logger.Debug("Successfully deleted processed notice", map[string]interface{}{
-						"notice_id": n.ID.Hex(),
-					})
-				}
 			}
 		}
 
-		skip += maxToProcess
-
 		// Check if we need to exit outer loop (break from inner loop means we hit the limit)
 		if maxToProcess > 0 && processedCount >= maxToProcess {
 			break
 		}
 	}
+
+	// Clean up: delete notices that were marked as processed in previous runs
+	w.cleanupProcessedNotices()
 }
 
 // scrapeDocumentsForNotice calls the scraper SDK to scrape documents for a processed notice
@@ -508,13 +517,13 @@ func (w *NoticeWorker) ToTender(n *notice.Notice) (*tender.Tender, error) {
 }
 
 // GenerateTenderID generates a unique tender ID using the PBL naming convention: SCDYYNNN
-// SCD = Source Code (3 letters), YY = Year (2 digits), NNN = Sequential number (3 digits)
+// 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%03d", sourceCode, year, sequentialNumber)
+	return fmt.Sprintf("%s%s%d", sourceCode, year, sequentialNumber)
 }
 
 // getSourceCode determines the 3-letter source code based on tender characteristics
@@ -593,8 +602,9 @@ func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode,
 	// 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
 
-	// Pattern to match: e.g., "PTD25" followed by 3 digits
-	pattern := fmt.Sprintf("^%s%s\\d{3}$", sourceCode, year)
+	// Pattern to match: e.g., "PTD25" followed by one or more digits
+	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{
@@ -602,7 +612,7 @@ func (w *NoticeWorker) getNextSequentialNumber(ctx context.Context, sourceCode,
 			"tender_id": bson.M{"$regex": pattern},
 		}}},
 		bson.D{{Key: "$project", Value: bson.M{
-			"sequential_num": bson.M{"$substr": bson.A{"$tender_id", 5, 3}}, // Extract last 3 chars (NNN)
+			"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,