package tender import ( "context" "time" "tm/pkg/logger" mongopkg "tm/pkg/mongo" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo" ) // TenderRepository interface defines tender data access methods type TenderRepository interface { // Tender CRUD operations Create(ctx context.Context, tender *Tender) error GetByID(ctx context.Context, id string) (*Tender, error) GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Tender, error) GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error) Update(ctx context.Context, tender *Tender) error Delete(ctx context.Context, id string) error List(ctx context.Context, criteria TenderSearchCriteria, limit, offset int) ([]Tender, int64, error) // Bulk operations CreateBatch(ctx context.Context, tenders []Tender) error UpdateBatch(ctx context.Context, tenders []Tender) error // Search and filtering Search(ctx context.Context, criteria TenderSearchCriteria) ([]Tender, error) GetByCountryCode(ctx context.Context, countryCode string, limit, offset int) ([]Tender, error) GetByStatus(ctx context.Context, status TenderStatus, limit, offset int) ([]Tender, error) GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error) GetActiveTenders(ctx context.Context, limit, offset int) ([]Tender, error) // Statistics and aggregations GetStatistics(ctx context.Context) (*TenderStatistics, error) GetTenderCountByCountry(ctx context.Context) (map[string]int64, error) GetTenderCountByType(ctx context.Context) (map[string]int64, error) GetTenderCountByClassification(ctx context.Context) (map[string]int64, error) // Scraping state management GetScrapingState(ctx context.Context) (*ScrapingState, error) SaveScrapingState(ctx context.Context, state *ScrapingState) error // Scraping jobs management CreateScrapingJob(ctx context.Context, job *ScrapingJob) error GetScrapingJob(ctx context.Context, id string) (*ScrapingJob, error) UpdateScrapingJob(ctx context.Context, job *ScrapingJob) error ListScrapingJobs(ctx context.Context, limit, offset int) ([]ScrapingJob, int64, error) GetActiveScrapingJobs(ctx context.Context) ([]ScrapingJob, error) } // tenderRepository implements TenderRepository interface using MongoDB ORM type tenderRepository struct { ormRepo mongopkg.Repository[Tender] stateOrmRepo mongopkg.Repository[ScrapingState] jobOrmRepo mongopkg.Repository[ScrapingJob] mongoManager *mongopkg.ConnectionManager logger logger.Logger } // NewTenderRepository creates a new tender repository func NewTenderRepository(mongoManager *mongopkg.ConnectionManager, logger logger.Logger) TenderRepository { // Create indexes for tenders collection tenderIndexes := []mongopkg.Index{ *mongopkg.CreateUniqueIndex("notice_publication_id_idx", bson.D{{Key: "notice_publication_id", Value: 1}}), *mongopkg.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}), *mongopkg.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}), *mongopkg.NewIndex("source_idx", bson.D{{Key: "source", Value: 1}}), *mongopkg.NewIndex("country_code_idx", bson.D{{Key: "country_code", Value: 1}}), *mongopkg.NewIndex("notice_type_code_idx", bson.D{{Key: "notice_type_code", Value: 1}}), *mongopkg.NewIndex("procurement_type_code_idx", bson.D{{Key: "procurement_type_code", Value: 1}}), *mongopkg.NewIndex("main_classification_idx", bson.D{{Key: "main_classification", Value: 1}}), *mongopkg.NewIndex("publication_date_idx", bson.D{{Key: "publication_date", Value: -1}}), *mongopkg.NewIndex("tender_deadline_idx", bson.D{{Key: "tender_deadline", Value: 1}}), *mongopkg.NewIndex("estimated_value_idx", bson.D{{Key: "estimated_value", Value: -1}}), *mongopkg.NewIndex("created_at_idx", bson.D{{Key: "created_at", Value: -1}}), *mongopkg.NewIndex("updated_at_idx", bson.D{{Key: "updated_at", Value: -1}}), // Compound indexes for common queries *mongopkg.NewIndex("status_country_idx", bson.D{ {Key: "status", Value: 1}, {Key: "country_code", Value: 1}, }), *mongopkg.NewIndex("notice_type_status_idx", bson.D{ {Key: "notice_type_code", Value: 1}, {Key: "status", Value: 1}, }), // Index for CPV matching queries *mongopkg.NewIndex("cpv_matching_idx", bson.D{ {Key: "status", Value: 1}, {Key: "main_classification", Value: 1}, {Key: "additional_classifications", Value: 1}, {Key: "publication_date", Value: -1}, }), // Text index for search *mongopkg.CreateTextIndex("search_idx", "title", "description"), } // Create indexes for scraping jobs collection jobIndexes := []mongopkg.Index{ *mongopkg.NewIndex("type_idx", bson.D{{Key: "type", Value: 1}}), *mongopkg.NewIndex("status_idx", bson.D{{Key: "status", Value: 1}}), *mongopkg.NewIndex("started_at_idx", bson.D{{Key: "started_at", Value: -1}}), *mongopkg.NewIndex("type_status_idx", bson.D{ {Key: "type", Value: 1}, {Key: "status", Value: 1}, }), } // Create indexes if err := mongoManager.CreateIndexes("tenders", tenderIndexes); err != nil { logger.Warn("Failed to create tender indexes", map[string]interface{}{ "error": err.Error(), }) } if err := mongoManager.CreateIndexes("scraping_jobs", jobIndexes); err != nil { logger.Warn("Failed to create scraping job indexes", map[string]interface{}{ "error": err.Error(), }) } // Create ORM repositories tenderOrmRepo := mongopkg.NewRepository[Tender](mongoManager.GetCollection("tenders"), logger) stateOrmRepo := mongopkg.NewRepository[ScrapingState](mongoManager.GetCollection("scraping_state"), logger) jobOrmRepo := mongopkg.NewRepository[ScrapingJob](mongoManager.GetCollection("scraping_jobs"), logger) return &tenderRepository{ ormRepo: tenderOrmRepo, stateOrmRepo: stateOrmRepo, jobOrmRepo: jobOrmRepo, mongoManager: mongoManager, logger: logger, } } // Create creates a new tender func (r *tenderRepository) Create(ctx context.Context, tender *Tender) error { err := r.ormRepo.Create(ctx, tender) if err != nil { r.logger.Error("Failed to create tender", map[string]interface{}{ "error": err.Error(), "notice_publication_id": tender.NoticePublicationID, }) return err } r.logger.Info("Tender created successfully", map[string]interface{}{ "tender_id": tender.ID, "notice_publication_id": tender.NoticePublicationID, }) return nil } // GetByID retrieves a tender by ID func (r *tenderRepository) GetByID(ctx context.Context, id string) (*Tender, error) { tender, err := r.ormRepo.FindByID(ctx, id) if err != nil { r.logger.Error("Failed to get tender by ID", map[string]interface{}{ "tender_id": id, "error": err.Error(), }) return nil, err } return tender, nil } // GetByContractNoticeID retrieves a tender by contract notice ID func (r *tenderRepository) GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Tender, error) { filter := bson.M{"contract_notice_id": contractNoticeID} pagination := mongopkg.Pagination{Limit: 1} result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get tender by contract notice ID", map[string]interface{}{ "contract_notice_id": contractNoticeID, "error": err.Error(), }) return nil, err } if len(result.Items) == 0 { return nil, mongopkg.ErrDocumentNotFound } return &result.Items[0], nil } // GetByNoticePublicationID retrieves a tender by notice publication ID func (r *tenderRepository) GetByNoticePublicationID(ctx context.Context, noticePublicationID string) (*Tender, error) { filter := bson.M{"notice_publication_id": noticePublicationID} pagination := mongopkg.Pagination{Limit: 1} result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get tender by notice publication ID", map[string]interface{}{ "notice_publication_id": noticePublicationID, "error": err.Error(), }) return nil, err } if len(result.Items) == 0 { return nil, mongopkg.ErrDocumentNotFound } return &result.Items[0], nil } // Update updates an existing tender func (r *tenderRepository) Update(ctx context.Context, tender *Tender) error { tender.UpdatedAt = time.Now().Unix() err := r.ormRepo.Update(ctx, tender) if err != nil { r.logger.Error("Failed to update tender", map[string]interface{}{ "tender_id": tender.ID, "error": err.Error(), }) return err } r.logger.Info("Tender updated successfully", map[string]interface{}{ "tender_id": tender.ID, }) return nil } // Delete deletes a tender by ID func (r *tenderRepository) Delete(ctx context.Context, id string) error { err := r.ormRepo.Delete(ctx, id) if err != nil { r.logger.Error("Failed to delete tender", map[string]interface{}{ "tender_id": id, "error": err.Error(), }) return err } r.logger.Info("Tender deleted successfully", map[string]interface{}{ "tender_id": id, }) return nil } // List retrieves tenders with pagination and filtering func (r *tenderRepository) List(ctx context.Context, criteria TenderSearchCriteria, limit, offset int) ([]Tender, int64, error) { filter := r.buildSearchFilter(criteria) pagination := mongopkg.Pagination{ Limit: limit, Skip: offset, SortField: "created_at", SortOrder: -1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to list tenders", map[string]interface{}{ "error": err.Error(), }) return nil, 0, err } return result.Items, result.TotalCount, nil } // CreateBatch creates multiple tenders in a single operation func (r *tenderRepository) CreateBatch(ctx context.Context, tenders []Tender) error { if len(tenders) == 0 { return nil } for i := range tenders { if err := r.ormRepo.Create(ctx, &tenders[i]); err != nil { r.logger.Error("Failed to create tender in batch", map[string]interface{}{ "tender_id": tenders[i].ID, "error": err.Error(), }) return err } } r.logger.Info("Tender batch created successfully", map[string]interface{}{ "count": len(tenders), }) return nil } // UpdateBatch updates multiple tenders func (r *tenderRepository) UpdateBatch(ctx context.Context, tenders []Tender) error { if len(tenders) == 0 { return nil } now := time.Now().Unix() for i := range tenders { tenders[i].UpdatedAt = now if err := r.ormRepo.Update(ctx, &tenders[i]); err != nil { r.logger.Error("Failed to update tender in batch", map[string]interface{}{ "tender_id": tenders[i].ID, "error": err.Error(), }) return err } } r.logger.Info("Tender batch updated successfully", map[string]interface{}{ "count": len(tenders), }) return nil } // Search searches tenders based on criteria func (r *tenderRepository) Search(ctx context.Context, criteria TenderSearchCriteria) ([]Tender, error) { filter := r.buildSearchFilter(criteria) pagination := mongopkg.Pagination{ Limit: 100, SortField: "created_at", SortOrder: -1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to search tenders", map[string]interface{}{ "error": err.Error(), }) return nil, err } return result.Items, nil } // GetByCountryCode retrieves tenders by country code func (r *tenderRepository) GetByCountryCode(ctx context.Context, countryCode string, limit, offset int) ([]Tender, error) { filter := bson.M{"country_code": countryCode} pagination := mongopkg.Pagination{ Limit: limit, Skip: offset, SortField: "created_at", SortOrder: -1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get tenders by country code", map[string]interface{}{ "country_code": countryCode, "error": err.Error(), }) return nil, err } return result.Items, nil } // GetByStatus retrieves tenders by status func (r *tenderRepository) GetByStatus(ctx context.Context, status TenderStatus, limit, offset int) ([]Tender, error) { filter := bson.M{"status": status} pagination := mongopkg.Pagination{ Limit: limit, Skip: offset, SortField: "created_at", SortOrder: -1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get tenders by status", map[string]interface{}{ "status": status, "error": err.Error(), }) return nil, err } return result.Items, nil } // GetExpiredTenders retrieves tenders that are expired func (r *tenderRepository) GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error) { filter := bson.M{ "$or": []bson.M{ {"status": TenderStatusExpired}, { "tender_deadline": bson.M{ "$lt": beforeDate, }, "status": bson.M{"$ne": TenderStatusExpired}, }, }, } pagination := mongopkg.Pagination{ Limit: 1000, // Large limit for maintenance operations SortField: "tender_deadline", SortOrder: 1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get expired tenders", map[string]interface{}{ "before_date": beforeDate, "error": err.Error(), }) return nil, err } return result.Items, nil } // GetActiveTenders retrieves active tenders func (r *tenderRepository) GetActiveTenders(ctx context.Context, limit, offset int) ([]Tender, error) { filter := bson.M{"status": TenderStatusActive} pagination := mongopkg.Pagination{ Limit: limit, Skip: offset, SortField: "tender_deadline", SortOrder: 1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get active tenders", map[string]interface{}{ "error": err.Error(), }) return nil, err } return result.Items, nil } // GetStatistics calculates tender statistics using aggregation func (r *tenderRepository) GetStatistics(ctx context.Context) (*TenderStatistics, error) { stats := &TenderStatistics{ LastUpdated: time.Now().Unix(), } // Get total count total, err := r.ormRepo.Count(ctx, bson.M{}) if err != nil { return nil, err } stats.TotalTenders = total // Get active count active, err := r.ormRepo.Count(ctx, bson.M{"status": TenderStatusActive}) if err != nil { return nil, err } stats.ActiveTenders = active // Get expired count expired, err := r.ormRepo.Count(ctx, bson.M{"status": TenderStatusExpired}) if err != nil { return nil, err } stats.ExpiredTenders = expired // Get statistics by country, type, and classification stats.TendersByCountry, _ = r.GetTenderCountByCountry(ctx) stats.TendersByType, _ = r.GetTenderCountByType(ctx) stats.TendersByClassification, _ = r.GetTenderCountByClassification(ctx) // Calculate value statistics using aggregation pipeline := mongo.Pipeline{ { {Key: "$group", Value: bson.M{ "_id": nil, "total_estimated_value": bson.M{"$sum": "$estimated_value"}, "count": bson.M{"$sum": 1}, }}, }, } result, err := r.ormRepo.Aggregate(ctx, pipeline) if err == nil && len(result) > 0 { if totalValue, ok := result[0]["total_estimated_value"].(float64); ok { stats.TotalEstimatedValue = totalValue if count, ok := result[0]["count"].(int32); ok && count > 0 { stats.AverageEstimatedValue = totalValue / float64(count) } } } return stats, nil } // GetTenderCountByCountry gets tender count grouped by country func (r *tenderRepository) GetTenderCountByCountry(ctx context.Context) (map[string]int64, error) { pipeline := mongo.Pipeline{ { {Key: "$group", Value: bson.M{ "_id": "$country_code", "count": bson.M{"$sum": 1}, }}, }, } results, err := r.ormRepo.Aggregate(ctx, pipeline) if err != nil { return nil, err } counts := make(map[string]int64) for _, result := range results { if country, ok := result["_id"].(string); ok { if count, ok := result["count"].(int32); ok { counts[country] = int64(count) } } } return counts, nil } // GetTenderCountByType gets tender count grouped by type func (r *tenderRepository) GetTenderCountByType(ctx context.Context) (map[string]int64, error) { pipeline := mongo.Pipeline{ { {Key: "$group", Value: bson.M{ "_id": "$notice_type_code", "count": bson.M{"$sum": 1}, }}, }, } results, err := r.ormRepo.Aggregate(ctx, pipeline) if err != nil { return nil, err } counts := make(map[string]int64) for _, result := range results { if noticeType, ok := result["_id"].(string); ok { if count, ok := result["count"].(int32); ok { counts[noticeType] = int64(count) } } } return counts, nil } // GetTenderCountByClassification gets tender count grouped by classification func (r *tenderRepository) GetTenderCountByClassification(ctx context.Context) (map[string]int64, error) { pipeline := mongo.Pipeline{ { {Key: "$group", Value: bson.M{ "_id": "$main_classification", "count": bson.M{"$sum": 1}, }}, }, { {Key: "$limit", Value: 20}, }, } results, err := r.ormRepo.Aggregate(ctx, pipeline) if err != nil { return nil, err } counts := make(map[string]int64) for _, result := range results { if classification, ok := result["_id"].(string); ok { if count, ok := result["count"].(int32); ok { counts[classification] = int64(count) } } } return counts, nil } // GetScrapingState retrieves the current scraping state func (r *tenderRepository) GetScrapingState(ctx context.Context) (*ScrapingState, error) { // Try to find existing state filter := bson.M{} pagination := mongopkg.Pagination{Limit: 1} result, err := r.stateOrmRepo.FindAll(ctx, filter, pagination) if err != nil { return nil, err } if len(result.Items) == 0 { // Return default state if none exists return &ScrapingState{ ID: "default", LastProcessedYear: time.Now().Year(), LastProcessedNumber: 0, LastRunAt: 0, NextRunAt: time.Now().Add(4 * time.Hour).Unix(), IsRunning: false, CreatedAt: time.Now().Unix(), UpdatedAt: time.Now().Unix(), }, nil } return &result.Items[0], nil } // SaveScrapingState saves the scraping state func (r *tenderRepository) SaveScrapingState(ctx context.Context, state *ScrapingState) error { if state.ID == "" { state.ID = "default" } now := time.Now().Unix() if state.CreatedAt == 0 { state.CreatedAt = now } state.UpdatedAt = now // Try to update existing state first existing, err := r.GetScrapingState(ctx) if err == nil && existing.ID != "" { state.ID = existing.ID state.CreatedAt = existing.CreatedAt return r.stateOrmRepo.Update(ctx, state) } // Create new state if none exists return r.stateOrmRepo.Create(ctx, state) } // CreateScrapingJob creates a new scraping job func (r *tenderRepository) CreateScrapingJob(ctx context.Context, job *ScrapingJob) error { if job.ID == "" { job.ID = primitive.NewObjectID().Hex() } now := time.Now().Unix() job.CreatedAt = now job.UpdatedAt = now err := r.jobOrmRepo.Create(ctx, job) if err != nil { r.logger.Error("Failed to create scraping job", map[string]interface{}{ "error": err.Error(), "type": job.Type, }) return err } r.logger.Info("Scraping job created successfully", map[string]interface{}{ "job_id": job.ID, "type": job.Type, }) return nil } // GetScrapingJob retrieves a scraping job by ID func (r *tenderRepository) GetScrapingJob(ctx context.Context, id string) (*ScrapingJob, error) { job, err := r.jobOrmRepo.FindByID(ctx, id) if err != nil { r.logger.Error("Failed to get scraping job", map[string]interface{}{ "job_id": id, "error": err.Error(), }) return nil, err } return job, nil } // UpdateScrapingJob updates a scraping job func (r *tenderRepository) UpdateScrapingJob(ctx context.Context, job *ScrapingJob) error { job.UpdatedAt = time.Now().Unix() err := r.jobOrmRepo.Update(ctx, job) if err != nil { r.logger.Error("Failed to update scraping job", map[string]interface{}{ "job_id": job.ID, "error": err.Error(), }) return err } r.logger.Info("Scraping job updated successfully", map[string]interface{}{ "job_id": job.ID, }) return nil } // ListScrapingJobs retrieves scraping jobs with pagination func (r *tenderRepository) ListScrapingJobs(ctx context.Context, limit, offset int) ([]ScrapingJob, int64, error) { filter := bson.M{} pagination := mongopkg.Pagination{ Limit: limit, Skip: offset, SortField: "started_at", SortOrder: -1, } result, err := r.jobOrmRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to list scraping jobs", map[string]interface{}{ "error": err.Error(), }) return nil, 0, err } return result.Items, result.TotalCount, nil } // GetActiveScrapingJobs retrieves currently active scraping jobs func (r *tenderRepository) GetActiveScrapingJobs(ctx context.Context) ([]ScrapingJob, error) { filter := bson.M{ "status": bson.M{ "$in": []ScrapingJobStatus{ ScrapingJobStatusPending, ScrapingJobStatusRunning, }, }, } pagination := mongopkg.Pagination{ Limit: 100, SortField: "started_at", SortOrder: -1, } result, err := r.jobOrmRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get active scraping jobs", map[string]interface{}{ "error": err.Error(), }) return nil, err } return result.Items, nil } // buildSearchFilter builds MongoDB filter from search criteria func (r *tenderRepository) buildSearchFilter(criteria TenderSearchCriteria) bson.M { filter := bson.M{} if criteria.Query != "" { filter["$text"] = bson.M{"$search": criteria.Query} } if criteria.NoticeType != "" { filter["notice_type_code"] = criteria.NoticeType } if criteria.ProcurementType != "" { filter["procurement_type_code"] = criteria.ProcurementType } if len(criteria.CountryCodes) > 0 { filter["country_code"] = bson.M{"$in": criteria.CountryCodes} } if len(criteria.RegionCodes) > 0 { filter["region_code"] = bson.M{"$in": criteria.RegionCodes} } if len(criteria.Classifications) > 0 { filter["$or"] = []bson.M{ {"main_classification": bson.M{"$in": criteria.Classifications}}, {"additional_classifications": bson.M{"$in": criteria.Classifications}}, } } if criteria.MinEstimatedValue != nil || criteria.MaxEstimatedValue != nil { valueFilter := bson.M{} if criteria.MinEstimatedValue != nil { valueFilter["$gte"] = *criteria.MinEstimatedValue } if criteria.MaxEstimatedValue != nil { valueFilter["$lte"] = *criteria.MaxEstimatedValue } filter["estimated_value"] = valueFilter } if criteria.Currency != "" { filter["currency"] = criteria.Currency } if criteria.DeadlineFrom != nil || criteria.DeadlineTo != nil { deadlineFilter := bson.M{} if criteria.DeadlineFrom != nil { deadlineFilter["$gte"] = *criteria.DeadlineFrom } if criteria.DeadlineTo != nil { deadlineFilter["$lte"] = *criteria.DeadlineTo } filter["tender_deadline"] = deadlineFilter } if criteria.PublicationDateFrom != nil || criteria.PublicationDateTo != nil { pubDateFilter := bson.M{} if criteria.PublicationDateFrom != nil { pubDateFilter["$gte"] = *criteria.PublicationDateFrom } if criteria.PublicationDateTo != nil { pubDateFilter["$lte"] = *criteria.PublicationDateTo } filter["publication_date"] = pubDateFilter } if len(criteria.Status) > 0 { filter["status"] = bson.M{"$in": criteria.Status} } if len(criteria.Source) > 0 { filter["source"] = bson.M{"$in": criteria.Source} } if criteria.BuyerOrganizationID != "" { filter["buyer_organization.id"] = criteria.BuyerOrganizationID } if len(criteria.Languages) > 0 { filter["notice_language_code"] = bson.M{"$in": criteria.Languages} } return filter }