e2e9159ac3
- Modified the GetUnProcessedNotices method in the notice repository to include a filter for main_classification set to "72000000". - Changed the sort order from ascending to descending for the created_at field, improving the retrieval of unprocessed notices.
182 lines
4.8 KiB
Go
182 lines
4.8 KiB
Go
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)
|
|
GetUnProcessedNotices(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{}
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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, "main_classification": "72000000"}
|
|
|
|
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
|
|
}
|
|
|
|
// 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
|
|
}
|