tm_back/cmd/worker/bootstrap/bootstrap.go

package bootstrap

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"
	"tm/cmd/worker/workers"
	"tm/internal/company"
	"tm/internal/company_category"
	"tm/internal/notice"
	notificationDomain "tm/internal/notification"
	"tm/internal/tender"
	ai_summarizer "tm/pkg/ai_summarizer"
	"tm/pkg/config"
	"tm/pkg/logger"
	"tm/pkg/mongo"
	"tm/pkg/notification"
	"tm/pkg/ollama"
	"tm/pkg/redis"
	"tm/pkg/schedule"
)

// aiPipelineDailyRunMu ensures only one AI pipeline daily-run executes at a time (startup catch-up and cron).
var aiPipelineDailyRunMu sync.Mutex

// Init Application Configuration
func InitConfig() (*Config, error) {
	conf, err := config.LoadConfig(".", &Config{})
	if err != nil {
		return nil, fmt.Errorf("failed to load config: %v", err)
	}

	return conf, nil
}

// Init Logger Service
func InitLogger(conf config.LoggingConfig) logger.Logger {
	return logger.NewLogger(&logger.Config{
		Level:  conf.Level,
		Format: conf.Format,
		Output: conf.Output,
		File: logger.FileConfig{
			Path:       conf.File.Path,
			MaxSize:    conf.File.MaxSize,
			MaxBackups: conf.File.MaxBackups,
			MaxAge:     conf.File.MaxAge,
			Compress:   conf.File.Compress,
		},
	})
}

// Init MongoDB Connection Manager
func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionManager {
	// Convert infra.MongoConfig to mongo.ConnectionConfig
	connectionConfig := mongo.ConnectionConfig{
		URI:                    conf.URI,
		Database:               conf.Name,
		MaxPoolSize:            uint64(conf.MaxPoolSize),
		MinPoolSize:            5,                // Default minimum pool size
		MaxConnIdleTime:        30 * time.Minute, // Default idle time
		ConnectTimeout:         conf.Timeout,
		ServerSelectionTimeout: 5 * time.Second, // Default server selection timeout
	}

	// Create MongoDB connection manager
	connectionManager, err := mongo.NewConnectionManager(connectionConfig, log)
	if err != nil {
		log.Error("Failed to initialize MongoDB connection", map[string]interface{}{
			"error": err.Error(),
			"uri":   conf.URI,
			"db":    conf.Name,
		})
		panic(fmt.Sprintf("Failed to initialize MongoDB connection: %v", err))
	}

	log.Info("MongoDB connection manager initialized successfully", map[string]interface{}{
		"database":  conf.Name,
		"uri":       conf.URI,
		"pool_size": conf.MaxPoolSize,
	})

	return connectionManager
}

// 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,
		"notice_concurrency":         config.Worker.NoticeWorkerConcurrency,
		"notice_fetch_batch":         config.Worker.NoticeFetchBatchSize,
		"notice_processing_limit":    config.Worker.NoticeProcessingLimit,
		"notice_batch_pause":         config.Worker.NoticeBatchPause.String(),
		"notice_fetch_error_backoff": config.Worker.NoticeFetchErrorBackoff.String(),
		"translation_enabled":        config.Worker.TranslationEnabled,
		"translation_interval":       config.Worker.TranslationInterval,
		"notification_interval":      config.Worker.NotificationInterval,
		"ai_pipeline_daily_enabled":  config.Worker.AIPipelineAutoEnabled,
		"ai_pipeline_daily_interval": config.Worker.AIPipelineAutoInterval,
	})
	// Initialize repositories
	noticeRepo := notice.NewRepository(mongoManager, 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,
		)
		if configurer, ok := recommendationRefresher.(interface {
			SetPageCacheLanguages([]string)
		}); ok {
			configurer.SetPageCacheLanguages(tender.ParsePageCacheLanguagesConfig(config.AISummarizer.RecommendationPageCacheLanguages))
		}
		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)

	// Schedule Notice Worker job (no blocking run at startup; first tick runs on schedule)
	workerInterval := config.Worker.Interval
	if workerInterval == "" {
		workerInterval = "* 10 * * * *" // Default: every hour at minute 10
		appLogger.Warn("WORKER_INTERVAL not set, using default schedule", map[string]interface{}{
			"interval": workerInterval,
		})
	}

	scheduler.AddJob(schedule.Job{
		Name: "Notice Worker Job",
		Func: func() {
			worker := workers.NewNoticeWorker(
				mongoManager,
				appLogger,
				&notify,
				noticeRepo,
				tenderRepo,
				config.Worker.NoticeProcessingLimit,
				config.Worker.NoticeWorkerConcurrency,
				config.Worker.NoticeFetchBatchSize,
				config.Worker.DeleteProcessedNotices,
				config.Worker.NoticeBatchPause,
				config.Worker.NoticeFetchErrorBackoff,
			)
			worker.Run()
		},
		Expr: workerInterval,
	})

	// Initialize translation backfill worker for configured languages
	if !config.Worker.TranslationEnabled {
		appLogger.Info("Tender translation worker disabled by configuration", map[string]interface{}{
			"translation_enabled": false,
		})
	} else if aiClient != nil {
		targetLanguages := parseTranslationLanguages(config.AISummarizer)
		translationSuccessCounter := mongo.NewCounter(mongoManager)
		scheduler.AddJob(schedule.Job{
			Name: "Tender Translation Worker Job",
			Func: func() {
				worker := workers.NewTranslationWorker(
					mongoManager,
					appLogger,
					tenderRepo,
					aiClient,
					aiStorage,
					translationSuccessCounter,
					targetLanguages,
					config.Worker.TranslationBatchSize,
				)
				worker.Run()
			},
			Expr: config.Worker.TranslationInterval,
		})
		appLogger.Info("Scheduled tender translation worker", map[string]interface{}{
			"target_languages": targetLanguages,
			"interval":         config.Worker.TranslationInterval,
			"batch_size":       config.Worker.TranslationBatchSize,
		})

		go func() {
			appLogger.Info("Running startup catch-up for tender translations", map[string]interface{}{
				"target_languages": targetLanguages,
			})
			worker := workers.NewTranslationWorker(
				mongoManager,
				appLogger,
				tenderRepo,
				aiClient,
				aiStorage,
				translationSuccessCounter,
				targetLanguages,
				config.Worker.TranslationBatchSize,
			)
			worker.Run()
		}()
	} else {
		appLogger.Warn("AI summarizer client not available, tender translation worker is disabled", map[string]interface{}{})
	}

	notificationInterval := config.Worker.NotificationInterval
	if notificationInterval == "" {
		notificationInterval = "0 * * * * *"
		appLogger.Warn("WORKER_NOTIFICATION_INTERVAL not set, using default schedule", map[string]interface{}{
			"interval": notificationInterval,
		})
	}

	scheduler.AddJob(schedule.Job{
		Name: "Scheduled Notification Worker Job",
		Func: func() {
			worker := workers.NewNotificationWorker(notificationRepo, appLogger)
			worker.Run()
		},
		Expr: notificationInterval,
	})
	appLogger.Info("Scheduled notification delivery worker", map[string]interface{}{
		"interval": notificationInterval,
	})

	if !config.Worker.AIPipelineAutoEnabled {
		appLogger.Info("AI pipeline daily-run worker disabled by configuration", map[string]interface{}{
			"ai_pipeline_daily_enabled": false,
		})
	} else if aiClient != nil {
		dailyJobTracker := schedule.NewDailyJobTracker(mongoManager, appLogger)

		runAIPipelineDaily := func(trigger string) {
			aiPipelineDailyRunMu.Lock()
			defer aiPipelineDailyRunMu.Unlock()

			ctx := context.Background()
			today := time.Now().Local()

			completed, checkErr := dailyJobTracker.IsCompleted(ctx, schedule.AIPipelineDailyJobName, today)
			if checkErr != nil {
				appLogger.Error("Failed to check AI pipeline daily-run completion status", map[string]interface{}{
					"trigger": trigger,
					"error":   checkErr.Error(),
					"date":    today.Format("02/01/2006"),
				})
			} else if completed {
				appLogger.Info("AI pipeline daily-run skipped: already completed today", map[string]interface{}{
					"trigger": trigger,
					"date":    today.Format("02/01/2006"),
				})
				return
			}

			if trigger == "startup" {
				appLogger.Info("Running startup catch-up for today's AI pipeline daily-run", map[string]interface{}{
					"date": today.Format("02/01/2006"),
				})
			} else {
				appLogger.Info("Running scheduled AI pipeline daily-run", map[string]interface{}{
					"date": today.Format("02/01/2006"),
				})
			}

			worker := workers.NewAIPipelineDailyWorker(appLogger, aiClient)
			if err := worker.Run(); err != nil {
				if errors.Is(err, ai_summarizer.ErrPipelineJobRunning) {
					appLogger.Info("AI pipeline daily-run deferred: job already running", map[string]interface{}{
						"trigger": trigger,
						"date":    today.Format("02/01/2006"),
					})
				} else {
					appLogger.Error("AI pipeline daily-run failed", map[string]interface{}{
						"trigger": trigger,
						"date":    today.Format("02/01/2006"),
						"error":   err.Error(),
					})
				}
				return
			}

			if markErr := dailyJobTracker.MarkCompleted(ctx, schedule.AIPipelineDailyJobName, today); markErr != nil {
				appLogger.Error("Failed to mark AI pipeline daily-run as completed", map[string]interface{}{
					"trigger": trigger,
					"date":    today.Format("02/01/2006"),
					"error":   markErr.Error(),
				})
			}

			appLogger.Info("AI pipeline daily-run completed successfully", map[string]interface{}{
				"trigger": trigger,
				"date":    today.Format("02/01/2006"),
			})

			if recommendationRefresher != nil {
				recommendationRefresher.ScheduleRefreshCachedAIRecommendationsAfterPipeline()
			}
		}

		scheduler.AddJob(schedule.Job{
			Name: "AI Pipeline Daily Run Job",
			Func: func() { runAIPipelineDaily("scheduled") },
			Expr: config.Worker.AIPipelineAutoInterval,
		})
		appLogger.Info("Scheduled AI pipeline daily-run worker", map[string]interface{}{
			"interval": config.Worker.AIPipelineAutoInterval,
		})

		go func() {
			runAIPipelineDaily("startup")
		}()
	} else {
		appLogger.Warn("AI summarizer client not available, AI pipeline daily-run worker is disabled", map[string]interface{}{})
	}

	// After a server restart, process any unprocessed notices (including today's) without blocking startup.
	go func() {
		appLogger.Info("Running startup catch-up for unprocessed notices", map[string]interface{}{})
		w := workers.NewNoticeWorker(
			mongoManager,
			appLogger,
			&notify,
			noticeRepo,
			tenderRepo,
			config.Worker.NoticeProcessingLimit,
			config.Worker.NoticeWorkerConcurrency,
			config.Worker.NoticeFetchBatchSize,
			config.Worker.DeleteProcessedNotices,
			config.Worker.NoticeBatchPause,
			config.Worker.NoticeFetchErrorBackoff,
		)
		w.Run()
	}()

	// Start the cron scheduler so all recurring jobs actually run
	scheduler.Start()
	appLogger.Info("Cron scheduler started with all worker jobs", map[string]interface{}{})

	return scheduler
}

// InitAISummarizerClient initializes the AI summarizer HTTP client used by worker jobs.
func InitAISummarizerClient(conf AISummarizerConfig, mongoManager *mongo.ConnectionManager, log logger.Logger) *ai_summarizer.Client {
	if conf.APIBaseURL == "" {
		log.Warn("AI Summarizer API base URL not configured, translation worker will be unavailable", map[string]interface{}{})
		return nil
	}

	translationCounter := mongo.NewCounter(mongoManager)
	cfg := &ai_summarizer.Config{
		APIBaseURL:              conf.APIBaseURL,
		APITimeout:              conf.APITimeout,
		APIRetryCount:           conf.APIRetryCount,
		APIRetryDelay:           conf.APIRetryDelay,
		OnSuccessfulTranslation: mongo.AITranslationSuccessCallback(translationCounter),
	}

	client, err := ai_summarizer.NewClient(cfg, log)
	if err != nil {
		log.Error("Failed to initialize AI Summarizer client for worker", map[string]interface{}{
			"error": err.Error(),
		})
		return nil
	}

	log.Info("AI Summarizer client initialized for worker", map[string]interface{}{
		"base_url": conf.APIBaseURL,
		"timeout":  conf.APITimeout,
	})

	return client
}

// InitAISummarizerStorage initializes the AI summarizer MinIO storage client for workers.
func InitAISummarizerStorage(conf AISummarizerConfig, log logger.Logger) *ai_summarizer.StorageClient {
	if conf.MinioEndpoint == "" {
		log.Warn("AI Summarizer MinIO endpoint not configured, translation sync from storage will be unavailable", map[string]interface{}{})
		return nil
	}

	cfg := &ai_summarizer.Config{
		APIBaseURL:     conf.APIBaseURL,
		APITimeout:     conf.APITimeout,
		APIRetryCount:  conf.APIRetryCount,
		APIRetryDelay:  conf.APIRetryDelay,
		MinioEndpoint:  conf.MinioEndpoint,
		MinioAccessKey: conf.MinioAccessKey,
		MinioSecretKey: conf.MinioSecretKey,
		MinioUseSSL:    conf.MinioUseSSL,
		MinioRegion:    conf.MinioRegion,
		MinioBucket:    conf.MinioBucket,
	}

	storage, err := ai_summarizer.NewStorageClient(cfg, log)
	if err != nil {
		log.Error("Failed to initialize AI Summarizer storage client for worker", map[string]interface{}{
			"error": err.Error(),
		})
		return nil
	}

	log.Info("AI Summarizer storage client initialized for worker", map[string]interface{}{
		"endpoint": conf.MinioEndpoint,
		"bucket":   conf.MinioBucket,
	})

	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 == "" {
		raw = conf.DefaultLanguage
	}
	parts := strings.Split(raw, ",")
	langs := make([]string, 0, len(parts))
	seen := make(map[string]struct{}, len(parts))
	for _, part := range parts {
		lang := strings.ToLower(strings.TrimSpace(part))
		if lang == "" {
			continue
		}
		if _, dup := seen[lang]; dup {
			continue
		}
		seen[lang] = struct{}{}
		langs = append(langs, lang)
	}
	if len(langs) == 0 {
		return []string{"en"}
	}
	return langs
}

// Init Notification Service
func InitNotificationService(conf config.NotificationConfig, log logger.Logger) notification.SDK {
	notificationConfig := &notification.Config{
		BaseURL:       conf.BaseURL,
		Timeout:       conf.Timeout,
		RetryAttempts: conf.RetryAttempts,
		RetryDelay:    conf.RetryDelay,
		EnableLogging: conf.EnableLogging,
		UserAgent:     conf.UserAgent,
	}

	notificationSDK, err := notification.New(notificationConfig, log)
	if err != nil {
		log.Error("Failed to initialize Notification SDK", map[string]interface{}{
			"error": err.Error(),
		})
		fmt.Printf("Failed to initialize Notification SDK: %v\n", err.Error())
	}

	log.Info("Notification SDK initialized successfully", map[string]interface{}{
		"base_url":       conf.BaseURL,
		"timeout":        conf.Timeout,
		"retry_attempts": conf.RetryAttempts,
		"retry_delay":    conf.RetryDelay,
		"enable_logging": conf.EnableLogging,
		"user_agent":     conf.UserAgent,
	})

	return *notificationSDK
}

// Init Ollama Service
func InitOllamaService(conf config.OllamaConfig, log logger.Logger) *ollama.SDK {
	ollamaSDK, err := ollama.New(&ollama.Config{
		BaseURL:            conf.BaseURL,
		Timeout:            conf.Timeout,
		RetryAttempts:      conf.RetryAttempts,
		RetryDelay:         conf.RetryDelay,
		EnableLogging:      conf.EnableLogging,
		UserAgent:          conf.UserAgent,
		DefaultModel:       conf.DefaultModel,
		DefaultTemperature: conf.DefaultTemperature,
		DefaultTopP:        conf.DefaultTopP,
		DefaultMaxTokens:   conf.DefaultMaxTokens,
		EnableStreaming:    conf.EnableStreaming,
	}, log)
	if err != nil {
		log.Error("Failed to initialize Ollama SDK", map[string]interface{}{
			"error": err.Error(),
		})
	}

	return ollamaSDK
}