package notice import ( "context" "time" "tm/pkg/logger" orm "tm/pkg/mongo" "go.mongodb.org/mongo-driver/v2/bson" ) // NoticeRepository interface defines notice data access methods type Repository interface { Import(ctx context.Context, notice *Notice) error Update(ctx context.Context, notice *Notice) error BulkImport(ctx context.Context, notices []Notice) error GetByID(ctx context.Context, id string) (*Notice, error) GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Notice, error) FindNoticesWithContentXML(ctx context.Context, limit, skip int) ([]Notice, error) GetUnProcessedNotices(ctx context.Context, ProcessingLimit int, skip int) ([]Notice, int64, error) GetProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error) Delete(ctx context.Context, id string) error } func collectionName() string { return "notices" } // noticeRepository implements NoticeRepository interface using MongoDB ORM type noticeRepository struct { ormRepo orm.Repository[Notice] mongoManager *orm.ConnectionManager logger logger.Logger } // NewRepository creates a new notice repository func NewRepository(mongoManager *orm.ConnectionManager, logger logger.Logger) Repository { // Create indexes for notices collection noticeIndexes := []orm.Index{ // processing_metadata.processed *orm.NewIndex("processing_metadata_processed_idx", bson.D{{Key: "processing_metadata.processed", Value: 1}}), *orm.NewIndex("main_classification_idx", bson.D{{Key: "main_classification", Value: 1}}), *orm.NewIndex("contract_notice_id_idx", bson.D{{Key: "contract_notice_id", Value: 1}}), // One row per TED ContractNoticeID when present (parallel scrape races). *orm.CreateUniqueIndex("contract_notice_id_unique", bson.D{{Key: "contract_notice_id", Value: 1}}). WithPartialFilterExpression(orm.NonEmptyStringPartialFilter("contract_notice_id")), } // Create indexes if err := mongoManager.CreateIndexes(collectionName(), noticeIndexes); err != nil { logger.Warn("Failed to create notice indexes", map[string]interface{}{ "error": err.Error(), }) } // Create ORM repositories noticeOrmRepo := orm.NewRepository[Notice](mongoManager.GetCollection(collectionName()), logger) return ¬iceRepository{ ormRepo: noticeOrmRepo, mongoManager: mongoManager, logger: logger, } } // Import imports a new notice func (r *noticeRepository) Import(ctx context.Context, notice *Notice) error { now := time.Now().Unix() notice.SetCreatedAt(now) err := r.ormRepo.Create(ctx, notice) if err != nil { r.logger.Error("Failed to create notice", map[string]interface{}{ "error": err.Error(), "notice_publication_id": notice.NoticePublicationID, }) return err } r.logger.Info("Notice created successfully", map[string]interface{}{ "notice_id": notice.ID, "notice_publication_id": notice.NoticePublicationID, }) return nil } // BulkImport imports a new notice func (r *noticeRepository) BulkImport(ctx context.Context, notices []Notice) error { err := r.ormRepo.CreateMany(ctx, notices) if err != nil { r.logger.Error("Failed to bulk import notices", map[string]interface{}{ "error": err.Error(), }) return err } return nil } // Update updates an existing notice func (r *noticeRepository) Update(ctx context.Context, notice *Notice) error { notice.UpdatedAt = time.Now().Unix() err := r.ormRepo.Update(ctx, notice) if err != nil { r.logger.Error("Failed to update notice", map[string]interface{}{ "notice_id": notice.ID, "error": err.Error(), }) return err } return nil } // GetByID retrieves a notice by ID func (r *noticeRepository) GetByID(ctx context.Context, id string) (*Notice, error) { notice, err := r.ormRepo.FindByID(ctx, id) if err != nil { r.logger.Error("Failed to get notice by ID", map[string]interface{}{ "notice_id": id, "error": err.Error(), }) return nil, err } return notice, nil } // GetByContractNoticeID retrieves a notice by contract notice ID func (r *noticeRepository) GetByContractNoticeID(ctx context.Context, contractNoticeID string) (*Notice, error) { filter := bson.M{"contract_notice_id": contractNoticeID} pagination := orm.Pagination{Limit: 1} result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get notice by contract notice ID", map[string]interface{}{ "contract_notice_id": contractNoticeID, "error": err.Error(), }) return nil, err } if len(result.Items) == 0 { return nil, orm.ErrDocumentNotFound } return &result.Items[0], nil } // FindNoticesWithContentXML returns notices that have non-empty stored TED XML (for backfill / remapping). func (r *noticeRepository) FindNoticesWithContentXML(ctx context.Context, limit, skip int) ([]Notice, error) { filter := bson.M{"content_xml": bson.M{"$exists": true, "$ne": ""}} pagination := orm.Pagination{ Limit: limit, Skip: skip, SortField: "_id", SortOrder: 1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to list notices with content_xml", map[string]interface{}{ "error": err.Error(), }) return nil, err } return result.Items, nil } // GetUnProcessedNotices retrieves unprocessed notices func (r *noticeRepository) GetUnProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error) { filter := bson.M{"processing_metadata.processed": false} pagination := orm.Pagination{ Limit: limit, Skip: skip, SortField: "created_at", SortOrder: -1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get unprocessed notices", map[string]interface{}{ "error": err.Error(), }) return nil, 0, err } return result.Items, result.TotalCount, nil } // GetProcessedNotices retrieves notices that have been successfully processed func (r *noticeRepository) GetProcessedNotices(ctx context.Context, limit int, skip int) ([]Notice, int64, error) { filter := bson.M{"processing_metadata.processed": true} pagination := orm.Pagination{ Limit: limit, Skip: skip, SortField: "updated_at", SortOrder: -1, } result, err := r.ormRepo.FindAll(ctx, filter, pagination) if err != nil { r.logger.Error("Failed to get processed notices", map[string]interface{}{ "error": err.Error(), }) return nil, 0, err } return result.Items, result.TotalCount, nil } // Delete deletes a notice by ID func (r *noticeRepository) Delete(ctx context.Context, id string) error { err := r.ormRepo.Delete(ctx, id) if err != nil { r.logger.Error("Failed to delete notice", map[string]interface{}{ "notice_id": id, "error": err.Error(), }) return err } r.logger.Info("Notice deleted successfully", map[string]interface{}{ "notice_id": id, }) return nil }