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,