From 0e4fadaf292fd39853f0de41c05b9a4f4ccb7362 Mon Sep 17 00:00:00 2001 From: Mazyar Date: Wed, 8 Jul 2026 00:45:53 +0330 Subject: [PATCH] Enhance company recommendation caching and pipeline integration - Updated the company service to include a new method for scheduling the refresh of cached AI recommendations after the AI pipeline execution. - Introduced a new interface for managing cached recommendation refreshes, improving the separation of concerns within the service layer. - Enhanced the worker initialization to include Redis client support, allowing for better management of recommendation caching. - Added functionality to list company IDs with existing recommendation caches, ensuring efficient updates post-pipeline runs. - Implemented unit tests to validate the new recommendation refresh logic and ensure proper handling of various scenarios. This update significantly improves the handling of AI recommendations by integrating caching mechanisms with the AI pipeline, enhancing overall system performance and responsiveness. --- cmd/web/main.go | 4 +- cmd/worker/bootstrap/bootstrap.go | 60 ++++- cmd/worker/bootstrap/config.go | 30 +-- cmd/worker/main.go | 12 +- internal/ai_pipeline/service.go | 39 +++- .../recommendation_pipeline_refresh.go | 221 ++++++++++++++++++ .../recommendation_pipeline_refresh_test.go | 30 +++ internal/company/repository.go | 44 ++++ internal/company/service.go | 6 + 9 files changed, 421 insertions(+), 25 deletions(-) create mode 100644 internal/company/recommendation_pipeline_refresh.go create mode 100644 internal/company/recommendation_pipeline_refresh_test.go diff --git a/cmd/web/main.go b/cmd/web/main.go index a442f91..6f68b65 100644 --- a/cmd/web/main.go +++ b/cmd/web/main.go @@ -240,7 +240,7 @@ func main() { userService := user.NewService(userRepository, logger, userAuthService, userValidator, notificationSDK, redisClient, auditLogger) categoryService := company_category.NewService(categoryRepository, logger) - companyService := company.NewService(companyRepository, categoryService, fileStoreService, aiRecommendationClient, redisClient, conf.AISummarizer.RecommendationCacheTTL, logger) + companyService := company.NewService(companyRepository, categoryService, fileStoreService, aiRecommendationClient, aiPipelineClient, redisClient, conf.AISummarizer.RecommendationCacheTTL, logger) customerService := customer.NewService(customerRepository, logger, customerAuthService, companyService, redisClient, notificationSDK, customerValidator, auditLogger) tenderService := tender.NewService(tenderRepository, companyService, customerService, tenderApprovalRepository, logger, aiSummarizerClient, aiSummarizerStorage, conf.AISummarizer.DefaultLanguage) feedbackService := feedback.NewService(feedbackRepo, tenderService, companyService, logger) @@ -254,7 +254,7 @@ func main() { documentScraperService := document_scraper.NewService(tenderRepository, aiPipelineClient, logger) dashboardRepository := dashboard.NewRepository(mongoManager, logger, procedureDocumentsLister) dashboardService := dashboard.NewService(dashboardRepository, logger, redisClient) - aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger, tenderService) + aiPipelineService := ai_pipeline.NewService(aiPipelineClient, logger, tenderService, companyService) auditLogRepository := auditlog.NewRepository(auditStore) auditLogService := auditlog.NewService(auditLogRepository, logger) logger.Info("Services initialized successfully", map[string]interface{}{ diff --git a/cmd/worker/bootstrap/bootstrap.go b/cmd/worker/bootstrap/bootstrap.go index ed6081a..6c70d08 100644 --- a/cmd/worker/bootstrap/bootstrap.go +++ b/cmd/worker/bootstrap/bootstrap.go @@ -8,6 +8,8 @@ import ( "sync" "time" "tm/cmd/worker/workers" + "tm/internal/company" + "tm/internal/company_category" "tm/internal/notice" notificationDomain "tm/internal/notification" "tm/internal/tender" @@ -17,6 +19,7 @@ import ( "tm/pkg/mongo" "tm/pkg/notification" "tm/pkg/ollama" + "tm/pkg/redis" "tm/pkg/schedule" ) @@ -82,8 +85,8 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa return connectionManager } -// Init Worker -func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient) *schedule.CronScheduler { +// InitWorker +func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient, redisClient redis.Client) *schedule.CronScheduler { // Debug: Log worker config appLogger.Info("Worker configuration", map[string]interface{}{ "worker_interval": config.Worker.Interval, @@ -103,6 +106,26 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger tenderRepo := tender.NewRepository(mongoManager, appLogger) notificationRepo := notificationDomain.NewRepository(mongoManager, appLogger) + var recommendationRefresher company.Service + if config.Worker.RecommendationRefreshAfterPipelineEnabled && aiClient != nil && redisClient != nil { + companyRepo := company.NewRepository(mongoManager, appLogger) + categoryRepo := company_category.NewRepository(mongoManager, appLogger) + categoryService := company_category.NewService(categoryRepo, appLogger) + recommendationRefresher = company.NewService( + companyRepo, + categoryService, + nil, + aiClient, + aiClient, + redisClient, + config.AISummarizer.RecommendationCacheTTL, + appLogger, + ) + appLogger.Info("Company recommendation refresh after pipeline enabled", map[string]interface{}{}) + } else if config.Worker.RecommendationRefreshAfterPipelineEnabled { + appLogger.Warn("Company recommendation refresh after pipeline disabled: AI client or Redis unavailable", map[string]interface{}{}) + } + // Create a single shared cron scheduler for all recurring jobs scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo) @@ -275,6 +298,10 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger "trigger": trigger, "date": today.Format("02/01/2006"), }) + + if recommendationRefresher != nil { + recommendationRefresher.ScheduleRefreshCachedAIRecommendationsAfterPipeline() + } } scheduler.AddJob(schedule.Job{ @@ -387,6 +414,35 @@ func InitAISummarizerStorage(conf AISummarizerConfig, log logger.Logger) *ai_sum return storage } +// InitRedis initializes the Redis client used by worker jobs. +func InitRedis(conf config.RedisConfig, log logger.Logger) redis.Client { + connectionConfig := &redis.Config{ + Host: conf.Host, + Port: conf.Port, + Password: conf.Password, + DB: conf.DB, + PoolSize: conf.PoolSize, + } + + redisClient, err := redis.NewClient(connectionConfig, log) + if err != nil { + log.Error("Failed to initialize Redis connection for worker", map[string]interface{}{ + "error": err.Error(), + "host": conf.Host, + "port": conf.Port, + }) + panic(fmt.Sprintf("Failed to initialize Redis connection: %v", err)) + } + + log.Info("Redis connection initialized for worker", map[string]interface{}{ + "host": conf.Host, + "port": conf.Port, + "pool_size": conf.PoolSize, + }) + + return redisClient +} + func parseTranslationLanguages(conf AISummarizerConfig) []string { raw := strings.TrimSpace(conf.TranslationLanguages) if raw == "" { diff --git a/cmd/worker/bootstrap/config.go b/cmd/worker/bootstrap/config.go index 18606b8..b5f5ac3 100644 --- a/cmd/worker/bootstrap/config.go +++ b/cmd/worker/bootstrap/config.go @@ -8,6 +8,7 @@ import ( // Config defines the worker application configuration type Config struct { Database config.DatabaseConfig + Redis config.RedisConfig Logging config.LoggingConfig Notification config.NotificationConfig Ollama config.OllamaConfig @@ -15,7 +16,7 @@ type Config struct { Scraper config.ScraperConfig MinIO config.MinIOConfig AISummarizer AISummarizerConfig - AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"` + AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"` } type WorkerConfig struct { @@ -40,20 +41,23 @@ type WorkerConfig struct { AIPipelineAutoEnabled bool `env:"WORKER_AI_PIPELINE_AUTO_ENABLED" envDefault:"true"` // AIPipelineAutoInterval runs daily-run once per day by default at 11:00 UTC (after TED scrape at 10:00). AIPipelineAutoInterval string `env:"WORKER_AI_PIPELINE_AUTO_INTERVAL" envDefault:"0 0 11 * * *"` + // RecommendationRefreshAfterPipelineEnabled refreshes cached company recommendations after the daily AI pipeline run. + RecommendationRefreshAfterPipelineEnabled bool `env:"WORKER_RECOMMENDATION_REFRESH" envDefault:"true"` } // AISummarizerConfig holds configuration for the external AI summarizer service. type AISummarizerConfig struct { - APIBaseURL string `env:"AI_SUMMARIZER_API_BASE_URL" envDefault:""` - APITimeout time.Duration `env:"AI_SUMMARIZER_API_TIMEOUT" envDefault:"120s"` - APIRetryCount int `env:"AI_SUMMARIZER_API_RETRY_COUNT" envDefault:"2"` - APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"` - DefaultLanguage string `env:"AI_SUMMARIZER_DEFAULT_LANGUAGE" envDefault:"en"` - TranslationLanguages string `env:"AI_SUMMARIZER_TRANSLATION_LANGUAGES" envDefault:"en"` - MinioEndpoint string `env:"AI_SUMMARIZER_MINIO_ENDPOINT" envDefault:""` - MinioAccessKey string `env:"AI_SUMMARIZER_MINIO_ACCESS_KEY" envDefault:""` - MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""` - MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"` - MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"` - MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens"` + APIBaseURL string `env:"AI_SUMMARIZER_API_BASE_URL" envDefault:""` + APITimeout time.Duration `env:"AI_SUMMARIZER_API_TIMEOUT" envDefault:"120s"` + APIRetryCount int `env:"AI_SUMMARIZER_API_RETRY_COUNT" envDefault:"2"` + APIRetryDelay time.Duration `env:"AI_SUMMARIZER_API_RETRY_DELAY" envDefault:"3s"` + DefaultLanguage string `env:"AI_SUMMARIZER_DEFAULT_LANGUAGE" envDefault:"en"` + TranslationLanguages string `env:"AI_SUMMARIZER_TRANSLATION_LANGUAGES" envDefault:"en"` + MinioEndpoint string `env:"AI_SUMMARIZER_MINIO_ENDPOINT" envDefault:""` + MinioAccessKey string `env:"AI_SUMMARIZER_MINIO_ACCESS_KEY" envDefault:""` + MinioSecretKey string `env:"AI_SUMMARIZER_MINIO_SECRET_KEY" envDefault:""` + MinioUseSSL bool `env:"AI_SUMMARIZER_MINIO_USE_SSL" envDefault:"false"` + MinioRegion string `env:"AI_SUMMARIZER_MINIO_REGION" envDefault:"us-east-1"` + MinioBucket string `env:"AI_SUMMARIZER_MINIO_BUCKET" envDefault:"opplens"` + RecommendationCacheTTL time.Duration `env:"AI_RECOMMENDATION_CACHE_TTL" envDefault:"0"` } diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 5c0f69a..ac3e940 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -40,9 +40,19 @@ func main() { // Initialize AI summarizer client (translation worker) aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, mongoManager, appLogger) aiSummarizerStorage := bootstrap.InitAISummarizerStorage(config.AISummarizer, appLogger) + redisClient := bootstrap.InitRedis(config.Redis, appLogger) + defer func() { + if redisClient != nil { + if err := redisClient.Close(); err != nil { + appLogger.Error("Failed to close Redis connection", map[string]interface{}{ + "error": err.Error(), + }) + } + } + }() // Initialize Worker - scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, aiSummarizerClient, aiSummarizerStorage) + scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, aiSummarizerClient, aiSummarizerStorage, redisClient) // Set up signal handling for graceful shutdown signalChan := make(chan os.Signal, 1) diff --git a/internal/ai_pipeline/service.go b/internal/ai_pipeline/service.go index 0cfd5ae..5ea3ab8 100644 --- a/internal/ai_pipeline/service.go +++ b/internal/ai_pipeline/service.go @@ -48,23 +48,35 @@ type Service interface { GetMinioStats(ctx context.Context) (ai_summarizer.PipelineMinioStatsResponse, error) } +// CachedRecommendationRefresher schedules a background refresh of company recommendation caches. +type CachedRecommendationRefresher interface { + ScheduleRefreshCachedAIRecommendationsAfterPipeline() +} + // ScrapedDocumentMetadataSyncer persists MinIO scraped-document metadata onto tender records. type ScrapedDocumentMetadataSyncer interface { SyncScrapedDocumentsFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) error } type service struct { - client Client - logger logger.Logger - metadataSyncer ScrapedDocumentMetadataSyncer + client Client + logger logger.Logger + metadataSyncer ScrapedDocumentMetadataSyncer + recommendationRefresher CachedRecommendationRefresher } // NewService creates an AI pipeline admin service. -func NewService(client Client, log logger.Logger, metadataSyncer ScrapedDocumentMetadataSyncer) Service { +func NewService( + client Client, + log logger.Logger, + metadataSyncer ScrapedDocumentMetadataSyncer, + recommendationRefresher CachedRecommendationRefresher, +) Service { return &service{ - client: client, - logger: log, - metadataSyncer: metadataSyncer, + client: client, + logger: log, + metadataSyncer: metadataSyncer, + recommendationRefresher: recommendationRefresher, } } @@ -219,6 +231,8 @@ func (s *service) Sync(ctx context.Context) (*ai_summarizer.PipelineSyncResponse return nil, fmt.Errorf("pipeline sync failed: %w", err) } + s.scheduleRecommendationRefreshAfterPipeline() + return resp, nil } @@ -322,6 +336,8 @@ func (s *service) DailyRun(ctx context.Context) (*ai_summarizer.PipelineStartedR return nil, fmt.Errorf("pipeline daily-run failed: %w", err) } + s.scheduleRecommendationRefreshAfterPipeline() + return resp, nil } @@ -340,6 +356,8 @@ func (s *service) Auto(ctx context.Context) (*ai_summarizer.PipelineStartedRespo return nil, fmt.Errorf("pipeline auto failed: %w", err) } + s.scheduleRecommendationRefreshAfterPipeline() + return resp, nil } @@ -425,6 +443,13 @@ func (s *service) GetMinioStats(ctx context.Context) (ai_summarizer.PipelineMini return resp, nil } +func (s *service) scheduleRecommendationRefreshAfterPipeline() { + if s.recommendationRefresher == nil { + return + } + s.recommendationRefresher.ScheduleRefreshCachedAIRecommendationsAfterPipeline() +} + func validateTenderBatch(forms []TenderRefForm) ([]ai_summarizer.TenderRef, error) { if len(forms) == 0 { return nil, ErrEmptyTenderBatch diff --git a/internal/company/recommendation_pipeline_refresh.go b/internal/company/recommendation_pipeline_refresh.go new file mode 100644 index 0000000..b1222c4 --- /dev/null +++ b/internal/company/recommendation_pipeline_refresh.go @@ -0,0 +1,221 @@ +package company + +import ( + "context" + "strings" + "sync/atomic" + "time" + + "golang.org/x/sync/errgroup" + + "tm/pkg/ai_summarizer" +) + +// AIPipelineStatusClient reports pipeline job status from the AI service. +type AIPipelineStatusClient interface { + GetPipelineLastRun(ctx context.Context) (*ai_summarizer.PipelineReportResponse, error) +} + +const ( + recommendationPipelineWaitInitialDelay = 5 * time.Minute + recommendationPipelineWaitMaxAttempts = 24 + recommendationRefreshConcurrency = 3 +) + +// ScheduleRefreshCachedAIRecommendationsAfterPipeline waits for the AI daily pipeline to finish, +// then re-fetches ranked tenders for every company that already has a recommendation cache. +func (s *companyService) ScheduleRefreshCachedAIRecommendationsAfterPipeline() { + if s.aiRecommendationClient == nil { + return + } + + go func() { + ctx := context.Background() + s.logger.Info("Scheduling AI recommendation cache refresh after pipeline sync", map[string]interface{}{}) + + if err := s.refreshCachedAIRecommendationsAfterPipeline(ctx); err != nil { + s.logger.Error("AI recommendation cache refresh after pipeline failed", map[string]interface{}{ + "error": err.Error(), + }) + return + } + + s.logger.Info("AI recommendation cache refresh after pipeline completed", map[string]interface{}{}) + }() +} + +func (s *companyService) refreshCachedAIRecommendationsAfterPipeline(ctx context.Context) error { + if err := s.waitForPipelineDailyRunCompletion(ctx); err != nil { + return err + } + + companyIDs, err := s.listCompanyIDsWithRecommendationCache(ctx) + if err != nil { + return err + } + if len(companyIDs) == 0 { + s.logger.Info("No companies with recommendation cache to refresh after pipeline", map[string]interface{}{}) + return nil + } + + s.logger.Info("Refreshing AI recommendation caches after pipeline", map[string]interface{}{ + "company_count": len(companyIDs), + }) + + group, groupCtx := errgroup.WithContext(ctx) + group.SetLimit(recommendationRefreshConcurrency) + + var refreshed atomic.Int32 + var failed atomic.Int32 + for _, companyID := range companyIDs { + companyID := companyID + group.Go(func() error { + if err := s.refreshAIRecommendationsCache(groupCtx, companyID); err != nil { + s.logger.Error("Failed to refresh AI recommendation cache for company", map[string]interface{}{ + "company_id": companyID, + "error": err.Error(), + }) + failed.Add(1) + return nil + } + refreshed.Add(1) + return nil + }) + } + + if err := group.Wait(); err != nil { + return err + } + + s.logger.Info("AI recommendation cache refresh summary", map[string]interface{}{ + "company_count": len(companyIDs), + "refreshed": refreshed.Load(), + "failed": failed.Load(), + }) + return nil +} + +func (s *companyService) waitForPipelineDailyRunCompletion(ctx context.Context) error { + if s.aiPipelineStatusClient == nil { + s.logger.Info("AI pipeline status client not configured; waiting before recommendation refresh", map[string]interface{}{ + "delay_sec": int(recommendationPipelineWaitInitialDelay.Seconds()), + }) + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(recommendationPipelineWaitInitialDelay): + } + return nil + } + + delay := recommendationPipelineWaitInitialDelay + for attempt := 1; attempt <= recommendationPipelineWaitMaxAttempts; attempt++ { + if attempt == 1 { + s.logger.Info("Waiting before first AI pipeline status check", map[string]interface{}{ + "delay_sec": int(delay.Seconds()), + }) + } else { + s.logger.Info("Retrying AI pipeline status check", map[string]interface{}{ + "attempt": attempt, + "delay_sec": int(delay.Seconds()), + }) + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(delay): + } + + if attempt > 1 { + delay *= 2 + if delay > 30*time.Minute { + delay = 30 * time.Minute + } + } + + report, err := s.aiPipelineStatusClient.GetPipelineLastRun(ctx) + if err != nil { + s.logger.Warn("Failed to read AI pipeline last-run status", map[string]interface{}{ + "attempt": attempt, + "error": err.Error(), + }) + if attempt == recommendationPipelineWaitMaxAttempts { + s.logger.Warn("Proceeding with recommendation refresh without pipeline status", map[string]interface{}{}) + return nil + } + continue + } + + if !isPipelineDailyRunInProgress(report) { + s.logger.Info("AI pipeline daily-run finished; refreshing recommendation caches", map[string]interface{}{ + "attempt": attempt, + "status": strings.TrimSpace(report.Status), + }) + return nil + } + + if attempt == recommendationPipelineWaitMaxAttempts { + s.logger.Warn("AI pipeline still in progress after max wait; refreshing recommendation caches anyway", map[string]interface{}{ + "attempts": attempt, + "status": strings.TrimSpace(report.Status), + }) + return nil + } + } + + return nil +} + +func isPipelineDailyRunInProgress(report *ai_summarizer.PipelineReportResponse) bool { + if report == nil { + return true + } + status := strings.ToLower(strings.TrimSpace(report.Status)) + switch status { + case "", "running", "in_progress", "started", "processing", "pending", "queued": + return true + default: + return false + } +} + +func (s *companyService) listCompanyIDsWithRecommendationCache(ctx context.Context) ([]string, error) { + allIDs, err := s.repository.ListIDs(ctx) + if err != nil { + s.logger.Error("Failed to list companies for recommendation refresh", map[string]interface{}{ + "error": err.Error(), + }) + return nil, err + } + + cached := make([]string, 0, len(allIDs)) + for _, companyID := range allIDs { + if _, ok := s.getCachedAIRecommendations(ctx, companyID); ok { + cached = append(cached, companyID) + } + } + return cached, nil +} + +// refreshAIRecommendationsCache updates Redis when the AI service returns a non-empty ranked list. +// Empty or failed fetches leave the existing cache untouched. +func (s *companyService) refreshAIRecommendationsCache(ctx context.Context, companyID string) error { + responses, err := s.fetchAIRecommendations(ctx, companyID) + if err != nil { + return err + } + if len(responses) == 0 { + s.logger.Info("Skipping recommendation cache update after pipeline: AI returned empty list", map[string]interface{}{ + "company_id": companyID, + }) + return nil + } + + s.cacheAIRecommendations(ctx, companyID, responses) + s.logger.Info("AI recommendation cache refreshed after pipeline", map[string]interface{}{ + "company_id": companyID, + "count": len(responses), + }) + return nil +} diff --git a/internal/company/recommendation_pipeline_refresh_test.go b/internal/company/recommendation_pipeline_refresh_test.go new file mode 100644 index 0000000..d5d7ba5 --- /dev/null +++ b/internal/company/recommendation_pipeline_refresh_test.go @@ -0,0 +1,30 @@ +package company + +import ( + "testing" + + "tm/pkg/ai_summarizer" +) + +func TestIsPipelineDailyRunInProgress(t *testing.T) { + tests := []struct { + name string + report *ai_summarizer.PipelineReportResponse + want bool + }{ + {name: "nil report", report: nil, want: true}, + {name: "empty status", report: &ai_summarizer.PipelineReportResponse{}, want: true}, + {name: "running", report: &ai_summarizer.PipelineReportResponse{Status: "running"}, want: true}, + {name: "in progress", report: &ai_summarizer.PipelineReportResponse{Status: "in_progress"}, want: true}, + {name: "completed", report: &ai_summarizer.PipelineReportResponse{Status: "completed"}, want: false}, + {name: "failed", report: &ai_summarizer.PipelineReportResponse{Status: "failed"}, want: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isPipelineDailyRunInProgress(tt.report); got != tt.want { + t.Fatalf("isPipelineDailyRunInProgress() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/internal/company/repository.go b/internal/company/repository.go index 5a81dfb..4733a96 100644 --- a/internal/company/repository.go +++ b/internal/company/repository.go @@ -24,6 +24,7 @@ type Repository interface { GetByRegistrationNumber(ctx context.Context, registrationNumber string) (*Company, error) GetByTaxID(ctx context.Context, taxID string) (*Company, error) Search(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*orm.PaginatedResult[Company], error) + ListIDs(ctx context.Context) ([]string, error) Delete(ctx context.Context, id string) error UpdateStatus(ctx context.Context, id string, status CompanyStatus) error UpdateVerification(ctx context.Context, id string, isVerified, isCompliant bool, complianceNotes *string) error @@ -383,6 +384,49 @@ func (r *companyRepository) Search(ctx context.Context, form *SearchForm, pagina return result, nil } +// ListIDs returns every company document ID. +func (r *companyRepository) ListIDs(ctx context.Context) ([]string, error) { + const pageSize = 500 + ids := make([]string, 0) + skip := 0 + + for { + result, err := r.ormRepo.FindAll(ctx, bson.M{}, orm.Pagination{ + Limit: pageSize, + Skip: skip, + SortField: "_id", + SortOrder: 1, + SkipCount: true, + Projection: bson.M{"_id": 1}, + }) + if err != nil { + r.logger.Error("Failed to list company IDs", map[string]interface{}{ + "error": err.Error(), + "skip": skip, + }) + return nil, err + } + + if len(result.Items) == 0 { + break + } + + for i := range result.Items { + id := strings.TrimSpace(result.Items[i].GetID()) + if id != "" { + ids = append(ids, id) + } + } + + if len(result.Items) < pageSize { + break + } + skip += pageSize + } + + return ids, nil +} + // UpdateStatus updates company status func (r *companyRepository) UpdateStatus(ctx context.Context, id string, status CompanyStatus) error { // Get company first diff --git a/internal/company/service.go b/internal/company/service.go index 7fd2105..0c43ce8 100644 --- a/internal/company/service.go +++ b/internal/company/service.go @@ -48,6 +48,9 @@ type Service interface { // GetAIRecommendations returns ranked tender recommendations from the AI team GetAIRecommendations(ctx context.Context, companyID string) ([]RecommendedTenderResponse, error) GetMergedAIRecommendations(ctx context.Context, companyIDs []string) ([]RecommendedTenderResponse, error) + + // ScheduleRefreshCachedAIRecommendationsAfterPipeline re-fetches cached company recommendations after tender sync. + ScheduleRefreshCachedAIRecommendationsAfterPipeline() } // companyService implements the Service interface @@ -56,6 +59,7 @@ type companyService struct { categoryService company_category.Service fileStore filestore.FileStoreService aiRecommendationClient AIRecommendationClient + aiPipelineStatusClient AIPipelineStatusClient redisClient redis.Client recommendationCacheTTL time.Duration logger logger.Logger @@ -67,6 +71,7 @@ func NewService( categoryService company_category.Service, fileStore filestore.FileStoreService, aiRecommendationClient AIRecommendationClient, + aiPipelineStatusClient AIPipelineStatusClient, redisClient redis.Client, recommendationCacheTTL time.Duration, logger logger.Logger, @@ -76,6 +81,7 @@ func NewService( categoryService: categoryService, fileStore: fileStore, aiRecommendationClient: aiRecommendationClient, + aiPipelineStatusClient: aiPipelineStatusClient, redisClient: redisClient, recommendationCacheTTL: recommendationCacheTTL, logger: logger,