mazyar/tm_back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Implement AI analysis trigger endpoint and refactor related components' (#38) from AI-Refactor into develop

Browse Source 
Reviewed-on: https://repo.ravanertebat.com/TM/tm_back/pulls/38
Reviewed-by: Hadi Barzegar <barzagarhadi@gmail.com>
This commit is contained in:
m.nazemi
2026-06-08 23:50:36 +03:30
parent de8de9b5c5 9cd22c24b6
commit 35892d5bc0
18 changed files with 595 additions and 1222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/web/router/routes.go
+1
View File
@@ -121,6 +121,7 @@ func RegisterAdminRoutes(e *echo.Echo, userHandler *user.Handler, companyHandler
tendersGP.GET("/:id/documents/download", tenderHandler.DownloadDocuments)
tendersGP.GET("/:id/ai-summary", tenderHandler.GetAISummary)
tendersGP.POST("/:id/ai-summarize", tenderHandler.TriggerAISummarize)
tendersGP.POST("/:id/ai-analyze", tenderHandler.TriggerAIAnalyze)
tendersGP.POST("/:id/ai-translate", tenderHandler.TriggerAITranslate)
}
cmd/worker/bootstrap/bootstrap.go
+6 -113
View File
@@ -10,11 +10,9 @@ import (
ai_summarizer "tm/pkg/ai_summarizer"
"tm/pkg/config"
"tm/pkg/logger"
"tm/pkg/minio"
"tm/pkg/mongo"
"tm/pkg/notification"
"tm/pkg/ollama"
"tm/pkg/queue"
"tm/pkg/schedule"
)
@@ -78,7 +76,7 @@ func InitMongoDB(conf config.MongoConfig, log logger.Logger) *mongo.ConnectionMa
}
// Init Worker
func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, minioService *minio.Service, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient) *schedule.CronScheduler {
func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger logger.Logger, notify notification.SDK, aiClient *ai_summarizer.Client, aiStorage *ai_summarizer.StorageClient) *schedule.CronScheduler {
// Debug: Log worker config
appLogger.Info("Worker configuration", map[string]interface{}{
"worker_interval": config.Worker.Interval,
@@ -87,8 +85,6 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
"notice_processing_limit": config.Worker.NoticeProcessingLimit,
"notice_batch_pause": config.Worker.NoticeBatchPause.String(),
"notice_fetch_error_backoff": config.Worker.NoticeFetchErrorBackoff.String(),
"queue_enabled": config.Queue.Enabled,
"queue_mode": config.Queue.Mode,
})
// Initialize repositories
noticeRepo := notice.NewRepository(mongoManager, appLogger)
@@ -97,70 +93,18 @@ func InitWorker(config Config, mongoManager *mongo.ConnectionManager, appLogger
// Create a single shared cron scheduler for all recurring jobs
scheduler := schedule.NewCronScheduler(appLogger, mongoManager, noticeRepo)
var queueService queue.Queue
if config.Queue.Enabled {
var err error
queueConfig := queue.QueueConfig{
RedisURL: config.Queue.RedisURL,
MaxRetries: config.Queue.MaxRetries,
RetryBackoff: config.Queue.RetryBackoff,
JobTimeout: config.Queue.JobTimeout,
PoolSize: config.Queue.PoolSize,
IsDLQEnabled: config.Queue.IsDLQEnabled,
MaxQueueLength: 0, // Unlimited
}
queueService, err = queue.NewRedisQueue(queueConfig, appLogger)
if err != nil {
appLogger.Error("Failed to initialize queue service", map[string]interface{}{
"error": err.Error(),
})
appLogger.Warn("Queue-based scraping is disabled, falling back to scheduled scraping", map[string]interface{}{})
config.Queue.Enabled = false
} else {
appLogger.Info("Queue service initialized successfully", map[string]interface{}{
"mode": config.Queue.Mode,
})
}
}
// Start Queue Producer Worker (enqueues unscraped tenders periodically)
if config.Queue.Enabled && (config.Queue.Mode == "producer" || config.Queue.Mode == "both") {
appLogger.Info("Scheduling queue producer worker", map[string]interface{}{
"batch_size": config.Queue.ProducerBatch,
"interval": config.Queue.ProducerInterval,
})
scheduler.AddJob(schedule.Job{
Name: "Queue Producer Worker Job",
Func: func() {
producer := workers.NewQueueProducerWorker(
mongoManager,
appLogger,
tenderRepo,
queueService,
config.Queue.ProducerBatch,
)
producer.Run()
},
Expr: config.Queue.ProducerInterval,
})
}
// Document summarization via external AI service (requires MinIO + AI_SUMMARIZER_API_BASE_URL)
if minioService != nil && aiClient != nil {
// Document summarization via external AI service (requires AI_SUMMARIZER_API_BASE_URL)
if aiClient != nil {
scheduler.AddJob(schedule.Job{
Name: "Document Summarization Worker Job",
Func: func() {
worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, minioService, aiClient)
worker := workers.NewDocumentSummarizationWorker(mongoManager, appLogger, tenderRepo, aiClient, aiStorage)
worker.Run()
},
Expr: "0 11 * * * *", // Run at 11 AM every day (1 hour after document scraping)
Expr: "0 11 * * * *", // Run at 11 AM every day (after AI pipeline scrape)
})
} else if minioService != nil {
appLogger.Warn("Document summarization worker not scheduled: AI summarizer API is not configured", map[string]interface{}{})
} else {
appLogger.Warn("MinIO service not available, document summarization will be skipped", map[string]interface{}{})
appLogger.Warn("Document summarization worker not scheduled: AI summarizer API is not configured", map[string]interface{}{})
}
// Schedule Notice Worker job (no blocking run at startup; first tick runs on schedule)
@@ -395,54 +339,3 @@ func InitOllamaService(conf config.OllamaConfig, log logger.Logger) *ollama.SDK
return ollamaSDK
}
// Init MinIO Service
func InitMinIOService(conf config.MinIOConfig, log logger.Logger) *minio.Service {
// Convert bootstrap config to minio config
minioConfig := minio.Config{
Endpoint: conf.Endpoint,
AccessKeyID: conf.AccessKeyID,
SecretAccessKey: conf.SecretAccessKey,
UseSSL: conf.UseSSL,
Region: conf.Region,
DefaultBucket: conf.DefaultBucket,
HierarchicalBucket: conf.HierarchicalBucket,
}
// Validate config
if err := minioConfig.Validate(); err != nil {
log.Warn("Invalid MinIO configuration, MinIO features will be disabled", map[string]interface{}{
"error": err.Error(),
})
return nil // Return nil instead of panicking
}
// Create connection manager
connManager, err := minio.NewConnectionManager(minioConfig, log)
if err != nil {
log.Error("Failed to create MinIO connection manager", map[string]interface{}{
"error": err.Error(),
})
panic(fmt.Sprintf("Failed to create MinIO connection manager: %v", err))
}
// Connect to MinIO
if err := connManager.Connect(); err != nil {
log.Warn("Failed to connect to MinIO, MinIO features will be disabled", map[string]interface{}{
"endpoint": minioConfig.Endpoint,
"error": err.Error(),
})
return nil // Return nil instead of panicking
}
// Create service
service := minio.NewService(connManager, minioConfig, log)
log.Info("MinIO service initialized successfully", map[string]interface{}{
"endpoint": conf.Endpoint,
"default_bucket": conf.DefaultBucket,
"hierarchical_bucket": conf.HierarchicalBucket,
})
return service
}
cmd/worker/bootstrap/config.go
-16
View File
@@ -15,25 +15,9 @@ type Config struct {
Scraper config.ScraperConfig
MinIO config.MinIOConfig
AISummarizer AISummarizerConfig
Queue QueueConfig `env:"QUEUE_" envPrefix:"QUEUE_"`
AlertMail string `env:"ALERT_MAIL" envDefault:"alerts@opplens.com"`
}
// QueueConfig defines the queue configuration
type QueueConfig struct {
Enabled bool `env:"QUEUE_ENABLED" envDefault:"true"`
RedisURL string `env:"QUEUE_REDIS_URL" envDefault:"redis://localhost:6379"`
Mode string `env:"QUEUE_MODE" envDefault:"consumer"` // "producer", "consumer", or "both"
MaxRetries int `env:"QUEUE_MAX_RETRIES" envDefault:"3"`
RetryBackoff time.Duration `env:"QUEUE_RETRY_BACKOFF" envDefault:"10s"`
JobTimeout time.Duration `env:"QUEUE_JOB_TIMEOUT" envDefault:"5m"`
PoolSize int `env:"QUEUE_POOL_SIZE" envDefault:"10"`
IsDLQEnabled bool `env:"QUEUE_DLQ_ENABLED" envDefault:"true"`
Concurrency int `env:"QUEUE_CONCURRENCY" envDefault:"5"`
ProducerBatch int `env:"QUEUE_PRODUCER_BATCH" envDefault:"100"`
ProducerInterval string `env:"QUEUE_PRODUCER_INTERVAL" envDefault:"0 */6 * * * *"` // Every 6 hours
}
type WorkerConfig struct {
Interval string `env:"WORKER_INTERVAL" envDefault:"* 10 * * * *"`
NoticeProcessingLimit int `env:"WORKER_NOTICE_PROCESSING_LIMIT" envDefault:"0"`
cmd/worker/main.go
+1 -7
View File
@@ -37,18 +37,12 @@ func main() {
// Initialize notification service
notificationService := bootstrap.InitNotificationService(config.Notification, appLogger)
// Initialize MinIO service
minioService := bootstrap.InitMinIOService(config.MinIO, appLogger)
if minioService != nil {
appLogger.Info("MinIO service initialized successfully", map[string]interface{}{})
}
// Initialize AI summarizer client (translation + document summarization workers)
aiSummarizerClient := bootstrap.InitAISummarizerClient(config.AISummarizer, mongoManager, appLogger)
aiSummarizerStorage := bootstrap.InitAISummarizerStorage(config.AISummarizer, appLogger)
// Initialize Worker
scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, minioService, aiSummarizerClient, aiSummarizerStorage)
scheduler := bootstrap.InitWorker(*config, mongoManager, appLogger, notificationService, aiSummarizerClient, aiSummarizerStorage)
// Set up signal handling for graceful shutdown
signalChan := make(chan os.Signal, 1)
cmd/worker/workers/queue_producer.go
-165
View File
@@ -1,165 +0,0 @@
package workers
import (
"context"
"time"
"tm/internal/tender"
"tm/pkg/logger"
"tm/pkg/mongo"
"tm/pkg/queue"
)
// QueueProducerWorker handles enqueuing unscraped tenders to the job queue
type QueueProducerWorker struct {
Mongo *mongo.ConnectionManager
Logger logger.Logger
TenderRepo tender.TenderRepository
Queue queue.Queue
BatchSize int
}
// NewQueueProducerWorker creates a new queue producer worker
func NewQueueProducerWorker(
mongo *mongo.ConnectionManager,
logger logger.Logger,
tenderRepo tender.TenderRepository,
queue queue.Queue,
batchSize int,
) *QueueProducerWorker {
if batchSize <= 0 {
batchSize = 100 // Default batch size
}
logger.Info("Creating new queue producer worker", map[string]interface{}{
"batch_size": batchSize,
})
return &QueueProducerWorker{
Mongo: mongo,
Logger: logger,
TenderRepo: tenderRepo,
Queue: queue,
BatchSize: batchSize,
}
}
// Run continuously processes and enqueues unscraped tenders
func (w *QueueProducerWorker) Run() {
w.Logger.Info("Queue producer worker started", map[string]interface{}{})
defer func() {
w.Logger.Info("Queue producer worker stopped", map[string]interface{}{})
}()
batchIndex := 0
hasMore := true
for hasMore {
w.Logger.Info("Queue producer fetching batch of tenders", map[string]interface{}{
"batch_index": batchIndex,
"batch_size": w.BatchSize,
})
// Get batch of unscraped tenders
tenders, totalCount, err := w.TenderRepo.GetUnScrapedTenders(
context.Background(),
w.BatchSize,
batchIndex*w.BatchSize,
)
if err != nil {
w.Logger.Error("Failed to fetch unscraped tenders", map[string]interface{}{
"batch_index": batchIndex,
"error": err.Error(),
})
// Wait before retrying
time.Sleep(30 * time.Second)
continue
}
w.Logger.Info("Fetched tender batch", map[string]interface{}{
"batch_size": len(tenders),
"total_count": totalCount,
"batch_index": batchIndex,
})
if len(tenders) == 0 {
// No more tenders in this batch
hasMore = false
w.Logger.Info("No more unscraped tenders found", map[string]interface{}{
"total_processed": totalCount,
})
break
}
// Enqueue each tender
successCount := 0
failCount := 0
for _, t := range tenders {
if err := w.enqueueTender(&t); err != nil {
w.Logger.Error("Failed to enqueue tender", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
"error": err.Error(),
})
failCount++
} else {
successCount++
}
}
w.Logger.Info("Batch enqueuing completed", map[string]interface{}{
"success_count": successCount,
"fail_count": failCount,
"batch_index": batchIndex,
})
// Move to next batch
batchIndex++
// Check if there are more tenders
if int64(batchIndex*w.BatchSize) >= totalCount {
hasMore = false
}
// Avoid tight loop - brief pause between batches
time.Sleep(1 * time.Second)
}
}
// enqueueTender adds a single tender to the queue
func (w *QueueProducerWorker) enqueueTender(t *tender.Tender) error {
// Skip if missing required fields
if t.NoticePublicationID == "" || t.DocumentURI == "" {
w.Logger.Warn("Skipping tender - missing required fields", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
"document_uri": t.DocumentURI,
})
return nil
}
// Create job
job := queue.NewDocumentScraperJob(
t.NoticePublicationID,
t.DocumentURI,
t.ID.Hex(),
queue.DefaultMaxRetries,
)
// Enqueue job
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := w.Queue.Enqueue(ctx, job); err != nil {
return err
}
return nil
}
// RunOnce processes all unscraped tenders in a single run (for scheduled use)
func (w *QueueProducerWorker) RunOnce() {
w.Run()
}
cmd/worker/workers/summarizer.go
+83 -179
View File
@@ -2,35 +2,39 @@ package workers
import (
"context"
"path/filepath"
"strings"
"time"
"tm/internal/tender"
ai_summarizer "tm/pkg/ai_summarizer"
"tm/pkg/logger"
"tm/pkg/minio"
"tm/pkg/mongo"
)
const presignedDocumentURLSeconds int64 = 7200
// DocumentSummarizationWorker asks the external AI service to summarize scraped tender documents.
// DocumentSummarizationWorker triggers AI summarization for scraped tenders.
// The AI service owns MinIO artefacts; this worker marks Mongo processing flags
// after the pipeline or on-demand summarize API completes.
type DocumentSummarizationWorker struct {
Mongo *mongo.ConnectionManager
Logger logger.Logger
TenderRepo tender.TenderRepository
MinioSDK *minio.Service
AIClient *ai_summarizer.Client
AIStorage *ai_summarizer.StorageClient
}
// NewDocumentSummarizationWorker creates a document summarization worker.
func NewDocumentSummarizationWorker(mongo *mongo.ConnectionManager, logger logger.Logger, tenderRepo tender.TenderRepository, minioSDK *minio.Service, aiClient *ai_summarizer.Client) *DocumentSummarizationWorker {
func NewDocumentSummarizationWorker(
mongo *mongo.ConnectionManager,
logger logger.Logger,
tenderRepo tender.TenderRepository,
aiClient *ai_summarizer.Client,
aiStorage *ai_summarizer.StorageClient,
) *DocumentSummarizationWorker {
return &DocumentSummarizationWorker{
Mongo: mongo,
Logger: logger,
TenderRepo: tenderRepo,
MinioSDK: minioSDK,
AIClient: aiClient,
AIStorage: aiStorage,
}
}
@@ -42,28 +46,31 @@ func (w *DocumentSummarizationWorker) Run() {
w.Logger.Warn("AI summarizer client not configured, skipping document summarization run", map[string]interface{}{})
return
}
if w.MinioSDK == nil {
w.Logger.Warn("MinIO not configured, skipping document summarization run", map[string]interface{}{})
return
if _, err := w.AIClient.TriggerPipelineSummarize(context.Background()); err != nil {
w.Logger.Warn("Failed to trigger AI pipeline summarize (continuing with per-tender sync)", map[string]interface{}{
"error": err.Error(),
})
}
limit := 5
skip := 0
for {
tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, 0)
tenders, totalCount, err := w.TenderRepo.GetUnSummarizedTenders(context.Background(), limit, skip)
if err != nil {
w.Logger.Error("Failed to get un-summarized tenders", map[string]interface{}{
"error": err.Error(),
})
continue
return
}
w.Logger.Info("Found tenders for document summarization", map[string]interface{}{
"count": len(tenders),
"total_count": totalCount,
"skip": skip,
})
if len(tenders) == 0 {
w.Logger.Info("No more tenders to process for document summarization", map[string]interface{}{})
break
}
@@ -72,78 +79,85 @@ func (w *DocumentSummarizationWorker) Run() {
w.Logger.Info("Processing tender for document summarization", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
"document_count": len(t.ScrapedDocuments),
})
w.summarizeDocumentsForTender(t)
w.summarizeTender(t)
}
skip += len(tenders)
if int64(skip) >= totalCount {
break
}
}
w.Logger.Info("Document summarization worker completed", map[string]interface{}{})
}
func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tender) {
if len(t.ScrapedDocuments) == 0 {
w.Logger.Warn("No scraped documents found for tender", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
func (w *DocumentSummarizationWorker) summarizeTender(t *tender.Tender) {
if strings.TrimSpace(t.NoticePublicationID) == "" {
w.Logger.Warn("Skipping tender summarization: missing notice_publication_id", map[string]interface{}{
"tender_id": t.ID.Hex(),
})
t.ProcessingMetadata.DocumentsSummarized = true
t.ProcessingMetadata.DocumentsSummarizedAt = time.Now().Unix()
_ = w.updateTenderSummarizationStatus(t)
w.markTenderSummarized(t)
return
}
if strings.TrimSpace(t.ContractFolderID) == "" {
w.Logger.Warn("Skipping tender summarization: missing contract_folder_id", map[string]interface{}{
"tender_id": t.ID.Hex(),
})
w.markTenderSummarized(t)
return
}
summaries := make([]tender.DocumentSummary, 0, len(t.ScrapedDocuments))
summarizedAt := time.Now().Unix()
ctx := context.Background()
if w.isSummaryReadyInStorage(ctx, t) {
w.markTenderSummarized(t)
return
}
for _, doc := range t.ScrapedDocuments {
summaryText, model, errStr := w.summarizeDocumentViaAI(context.Background(), t, doc)
if errStr != "" {
w.Logger.Error("Failed to summarize document", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
"document_name": doc.Filename,
"object_name": doc.ObjectName,
"error": errStr,
})
summaries = append(summaries, tender.DocumentSummary{
DocumentName: doc.Filename,
ObjectName: doc.ObjectName,
BucketName: doc.BucketName,
Summary: "",
SummaryLanguage: "en",
DocumentType: w.getDocumentType(doc.Filename),
SummarizedAt: summarizedAt,
SummaryModel: model,
Error: errStr,
})
continue
}
summaries = append(summaries, tender.DocumentSummary{
DocumentName: doc.Filename,
ObjectName: doc.ObjectName,
BucketName: doc.BucketName,
Summary: summaryText,
SummaryLanguage: "en",
DocumentType: w.getDocumentType(doc.Filename),
SummarizedAt: summarizedAt,
SummaryModel: model,
})
w.Logger.Info("Document summarized successfully", map[string]interface{}{
req := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID)
if _, err := w.AIClient.FetchSummaryOnDemand(ctx, req); err != nil {
w.Logger.Error("Failed to summarize tender via AI service", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
"contract_folder_id": t.ContractFolderID,
"error": err.Error(),
})
return
}
if !w.isSummaryReadyInStorage(ctx, t) {
w.Logger.Warn("AI summarize completed but storage is not ready yet", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
"document_name": doc.Filename,
"summary_length": len(summaryText),
})
}
t.DocumentSummaries = summaries
t.ProcessingMetadata.DocumentsSummarized = true
t.ProcessingMetadata.DocumentsSummarizedAt = summarizedAt
w.markTenderSummarized(t)
}
if err := w.updateTenderSummarizationStatus(t); err != nil {
func (w *DocumentSummarizationWorker) isSummaryReadyInStorage(ctx context.Context, t *tender.Tender) bool {
if w.AIStorage == nil {
return false
}
ready, err := w.AIStorage.IsSummaryReady(ctx, t.ContractFolderID, t.NoticePublicationID)
if err != nil {
w.Logger.Debug("Could not check summarization status in storage", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
"error": err.Error(),
})
return false
}
return ready
}
func (w *DocumentSummarizationWorker) markTenderSummarized(t *tender.Tender) {
now := time.Now().Unix()
t.ProcessingMetadata.DocumentsSummarized = true
t.ProcessingMetadata.DocumentsSummarizedAt = now
t.UpdatedAt = now
if err := w.TenderRepo.Update(context.Background(), t); err != nil {
w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
@@ -152,118 +166,8 @@ func (w *DocumentSummarizationWorker) summarizeDocumentsForTender(t *tender.Tend
return
}
w.Logger.Info("Document summarization completed for tender", map[string]interface{}{
w.Logger.Info("Tender marked as summarized", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_publication_id": t.NoticePublicationID,
"summarized_count": len(summaries),
"total_documents": len(t.ScrapedDocuments),
})
}
func (w *DocumentSummarizationWorker) summarizeDocumentViaAI(ctx context.Context, t *tender.Tender, doc tender.ScrapedDocument) (summaryText, model, errStr string) {
docURL, err := w.MinioSDK.GetPresignedURL(ctx, doc.BucketName, doc.ObjectName, presignedDocumentURLSeconds)
if err != nil {
return "", "", err.Error()
}
if strings.TrimSpace(t.NoticePublicationID) == "" {
return "", "", "tender has no notice_publication_id for AI summarize request"
}
if strings.TrimSpace(t.ContractFolderID) == "" {
return "", "", "tender has no contract_folder_id for AI summarize request"
}
req := ai_summarizer.NewSummarizeRequest(ai_summarizer.SummarizeRequestInput{
ContractFolderID: t.ContractFolderID,
NoticePublicationID: t.NoticePublicationID,
DocumentURL: docURL,
Title: t.Title,
Description: t.Description,
Country: t.CountryCode,
Currency: t.Currency,
SubmissionDeadline: t.SubmissionDeadline,
EstimatedValue: t.EstimatedValue,
AuthorityName: authorityNameFromTender(t),
})
resp, err := w.AIClient.FetchSummaryOnDemand(ctx, req)
if err != nil {
return "", "", err.Error()
}
if resp == nil {
return "", "", "AI summarize returned nil response"
}
summaryText, model = pickDocumentSummary(resp.Summaries, doc.Filename)
if summaryText == "" && len(resp.Summaries) > 0 {
// Fall back to first summary if filename did not match
summaryText = resp.Summaries[0].Summary
model = resp.Summaries[0].SummaryModel
}
if summaryText == "" {
return "", model, "AI summarize returned no summary text for document"
}
return summaryText, model, ""
}
func authorityNameFromTender(t *tender.Tender) string {
if t == nil || t.BuyerOrganization == nil {
return ""
}
return strings.TrimSpace(t.BuyerOrganization.Name)
}
func pickDocumentSummary(items []ai_summarizer.DocumentSummary, filename string) (text, model string) {
filename = strings.TrimSpace(filename)
for _, s := range items {
if strings.EqualFold(strings.TrimSpace(s.DocumentName), filename) {
return strings.TrimSpace(s.Summary), strings.TrimSpace(s.SummaryModel)
}
}
for _, s := range items {
if strings.TrimSpace(s.Summary) != "" {
return strings.TrimSpace(s.Summary), strings.TrimSpace(s.SummaryModel)
}
}
return "", ""
}
func (w *DocumentSummarizationWorker) getDocumentType(filename string) string {
ext := strings.ToLower(filepath.Ext(filename))
switch ext {
case ".xlsx":
return "xlsx"
case ".pdf":
return "pdf"
case ".docx":
return "docx"
case ".doc":
return "doc"
case ".txt":
return "txt"
case ".html", ".htm":
return "html"
default:
return "unknown"
}
}
func (w *DocumentSummarizationWorker) updateTenderSummarizationStatus(tender *tender.Tender) error {
err := w.TenderRepo.Update(context.Background(), tender)
if err != nil {
w.Logger.Error("Failed to update tender summarization status", map[string]interface{}{
"tender_id": tender.ID.Hex(),
"error": err.Error(),
})
return err
}
w.Logger.Debug("Updated tender summarization status", map[string]interface{}{
"tender_id": tender.ID.Hex(),
"documents_summarized": tender.ProcessingMetadata.DocumentsSummarized,
"documents_summarized_at": tender.ProcessingMetadata.DocumentsSummarizedAt,
"summary_count": len(tender.DocumentSummaries),
})
return nil
}
cmd/worker/workers/translation.go
+42 -103
View File
@@ -2,7 +2,6 @@ package workers
import (
"context"
"errors"
"strings"
"tm/internal/tender"
"tm/pkg/ai_summarizer"
@@ -10,17 +9,18 @@ import (
"tm/pkg/mongo"
)
// TranslationWorker backfills missing tender translations using the AI pipeline
// and on-demand translate API. Results are persisted by the AI service in MinIO only.
// TranslationWorker syncs translation completion from the AI pipeline MinIO storage.
// Batch translation is triggered via POST /pipeline/translate; this worker only
// verifies which tenders now have translations available in storage.
type TranslationWorker struct {
Mongo *mongo.ConnectionManager
Logger logger.Logger
TenderRepo tender.TenderRepository
AIClient *ai_summarizer.Client
AIStorage *ai_summarizer.StorageClient
Mongo *mongo.ConnectionManager
Logger logger.Logger
TenderRepo tender.TenderRepository
AIClient *ai_summarizer.Client
AIStorage *ai_summarizer.StorageClient
TranslationSuccessCounter *mongo.Counter
TargetLanguages []string
BatchSize int
TargetLanguages []string
BatchSize int
}
// NewTranslationWorker creates a translation worker for the given target languages.
@@ -66,7 +66,7 @@ func NewTranslationWorker(
}
}
// Run starts backfilling translations for tenders missing target-language content.
// Run triggers the AI translation pipeline and logs tenders with ready translations in MinIO.
func (w *TranslationWorker) Run() {
if w.AIClient == nil {
w.Logger.Warn("Translation worker skipped: AI client is not configured", map[string]interface{}{
@@ -83,14 +83,15 @@ func (w *TranslationWorker) Run() {
startCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob)
if _, err := w.AIClient.TriggerPipelineTranslate(context.Background(), w.TargetLanguages); err != nil {
w.Logger.Warn("Failed to trigger AI pipeline translate (continuing with per-tender sync)", map[string]interface{}{
w.Logger.Warn("Failed to trigger AI pipeline translate", map[string]interface{}{
"target_languages": w.TargetLanguages,
"error": err.Error(),
})
}
readyCount := 0
for _, language := range w.TargetLanguages {
w.runForLanguage(language)
readyCount += w.syncLanguageFromStorage(language)
}
endCount := w.getSuccessfulTranslationCount(mongo.AITranslationSuccessCounterKeyDailyJob)
@@ -101,10 +102,11 @@ func (w *TranslationWorker) Run() {
}
w.Logger.Info("Translation worker completed", map[string]interface{}{
"target_languages": w.TargetLanguages,
"successful_ai_translation_requests_run": runCount,
"successful_ai_translation_requests_daily_job": endCount,
"successful_ai_translation_requests_total": totalCount,
"target_languages": w.TargetLanguages,
"translations_ready_in_storage": readyCount,
"successful_ai_translation_requests_run": runCount,
"successful_ai_translation_requests_daily_job": endCount,
"successful_ai_translation_requests_total": totalCount,
})
}
@@ -123,19 +125,27 @@ func (w *TranslationWorker) getSuccessfulTranslationCount(key string) int64 {
return count
}
func (w *TranslationWorker) runForLanguage(language string) {
func (w *TranslationWorker) syncLanguageFromStorage(language string) int {
if w.AIStorage == nil {
w.Logger.Warn("Translation storage client not configured; skipping MinIO sync", map[string]interface{}{
"target_language": language,
})
return 0
}
readyCount := 0
skip := 0
for {
tenders, totalCount, err := w.TenderRepo.GetUnTranslatedTenders(context.Background(), language, w.BatchSize, skip)
if err != nil {
w.Logger.Error("Failed to fetch tenders for translation backfill", map[string]interface{}{
w.Logger.Error("Failed to fetch tenders for translation sync", map[string]interface{}{
"target_language": language,
"error": err.Error(),
})
return
return readyCount
}
w.Logger.Info("Fetched tenders for translation backfill", map[string]interface{}{
w.Logger.Info("Fetched tenders for translation sync", map[string]interface{}{
"target_language": language,
"batch_count": len(tenders),
"total_count": totalCount,
@@ -143,63 +153,29 @@ func (w *TranslationWorker) runForLanguage(language string) {
})
if len(tenders) == 0 {
return
return readyCount
}
for _, t := range tenders {
w.translateTender(&t, language)
if w.hasTranslationInStorage(&t, language) {
readyCount++
w.Logger.Debug("Translation available in MinIO", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_id": t.NoticePublicationID,
"target_language": language,
})
}
}
skip += len(tenders)
if int64(skip) >= totalCount {
return
return readyCount
}
}
}
func (w *TranslationWorker) translateTender(t *tender.Tender, language string) {
if t.NoticePublicationID == "" {
return
}
if strings.TrimSpace(t.ContractFolderID) == "" {
w.Logger.Debug("Skipping tender translation: missing contract_folder_id", map[string]interface{}{
"tender_id": t.ID.Hex(),
})
return
}
if w.hasTranslationInStorage(t, language) {
w.Logger.Debug("Skipping tender translation: already available in MinIO", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_id": t.NoticePublicationID,
"target_language": language,
})
return
}
title, description, source, err := w.resolveTranslation(t, language)
if err != nil {
w.Logger.Error("Failed to translate tender", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_id": t.NoticePublicationID,
"target_language": language,
"translation_source": source,
"translation_error": err.Error(),
})
return
}
w.Logger.Info("Tender translated successfully", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_id": t.NoticePublicationID,
"target_language": language,
"translation_source": source,
"title_len": len(title),
"description_len": len(description),
})
}
func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language string) bool {
if w.AIStorage == nil {
if strings.TrimSpace(t.ContractFolderID) == "" || t.NoticePublicationID == "" {
return false
}
stored, err := w.AIStorage.GetTranslationFromStorage(
@@ -210,40 +186,3 @@ func (w *TranslationWorker) hasTranslationInStorage(t *tender.Tender, language s
)
return err == nil && strings.TrimSpace(stored.Title) != ""
}
func (w *TranslationWorker) resolveTranslation(t *tender.Tender, language string) (title, description, source string, err error) {
if w.AIStorage != nil {
stored, storageErr := w.AIStorage.GetTranslationFromStorage(
context.Background(),
t.ContractFolderID,
t.NoticePublicationID,
language,
)
if storageErr == nil {
return stored.Title, stored.Description, "storage", nil
}
if !errors.Is(storageErr, ai_summarizer.ErrTranslationNotReady) &&
!errors.Is(storageErr, ai_summarizer.ErrObjectNotFound) &&
!errors.Is(storageErr, ai_summarizer.ErrNoTranslation) {
w.Logger.Debug("Translation not available from storage yet", map[string]interface{}{
"tender_id": t.ID.Hex(),
"notice_id": t.NoticePublicationID,
"target_language": language,
"error": storageErr.Error(),
})
}
}
resp, apiErr := w.AIClient.FetchTranslationOnDemand(context.Background(), ai_summarizer.TranslateRequest{
ContractFolderID: t.ContractFolderID,
NoticePublicationID: t.NoticePublicationID,
Title: t.Title,
Description: t.Description,
Language: language,
RequestSource: ai_summarizer.TranslationRequestSourceDailyJob,
})
if apiErr != nil {
return "", "", "api", apiErr
}
return resp.TranslatedTitle, resp.TranslatedDescription, "api", nil
}
internal/tender/errors.go
+13
View File
@@ -0,0 +1,13 @@
package tender
import "errors"
// Sentinel errors for AI trigger operations and HTTP mapping.
var (
ErrEmptyTenderID = errors.New("tender: id is required")
ErrInvalidTenderID = errors.New("tender: invalid id")
ErrTenderNotFound = errors.New("tender: not found")
ErrTenderMissingNoticeID = errors.New("tender: missing notice publication id")
ErrTenderMissingContractFolderID = errors.New("tender: missing contract folder id")
ErrAINotConfigured = errors.New("tender: ai service not configured")
)
internal/tender/form.go
+20
View File
@@ -573,3 +573,23 @@ type AIDocumentSummary struct {
SummaryModel string `json:"summary_model"`
Error string `json:"error,omitempty"`
}
// AIAnalyzeResponse represents the response for on-demand agentic analysis.
type AIAnalyzeResponse struct {
ContractFolderID string `json:"contract_folder_id"`
NoticePublicationID string `json:"notice_publication_id"`
Summary string `json:"summary"`
Documents []AIAnalyzeDocument `json:"documents"`
ToolCallsLog []string `json:"tool_calls_log"`
Iterations int `json:"iterations"`
Model string `json:"model"`
}
// AIAnalyzeDocument describes one document referenced during analysis.
type AIAnalyzeDocument struct {
Filename string `json:"filename"`
SizeBytes int64 `json:"size_bytes"`
TimesRead int `json:"times_read"`
TimesQueried int `json:"times_queried"`
Description string `json:"description"`
}
internal/tender/handler.go
+48 -2
View File
@@ -398,7 +398,7 @@ func (h *TenderHandler) GetPublicTenderDetails(c echo.Context) error {
// GetDocumentSummaries retrieves document summaries for a tender
// @Summary Get document summaries for a tender
// @Description Retrieve AI-generated summaries of documents for a specific tender
// @Description Retrieve per-document AI summaries from the AI service MinIO storage for a specific tender
// @Tags Tenders
// @Produce json
// @Param id path string true "Tender ID"
@@ -589,12 +589,58 @@ func (h *TenderHandler) TriggerAISummarize(c echo.Context) error {
"tender_id": id,
"error": err.Error(),
})
return response.InternalServerError(c, err.Error())
return mapAITriggerHTTPError(c, err, "AI summarization")
}
return response.Success(c, result, "AI summarization completed successfully")
}
// TriggerAIAnalyze triggers on-demand agentic analysis for a tender
// @Summary Trigger AI analysis
// @Description Trigger on-demand agentic document analysis for a tender via the external AI service
// @Tags Admin-Tenders
// @Produce json
// @Param id path string true "Tender ID"
// @Success 200 {object} response.APIResponse{data=AIAnalyzeResponse}
// @Failure 400 {object} response.APIResponse
// @Failure 404 {object} response.APIResponse
// @Failure 500 {object} response.APIResponse
// @Router /admin/v1/tenders/{id}/ai-analyze [post]
// @Security BearerAuth
func (h *TenderHandler) TriggerAIAnalyze(c echo.Context) error {
id := c.Param("id")
if id == "" {
return response.BadRequest(c, "Tender ID is required", "ID parameter cannot be empty")
}
result, err := h.service.TriggerAIAnalyze(c.Request().Context(), id)
if err != nil {
h.logger.Error("Failed to trigger AI analysis", map[string]interface{}{
"tender_id": id,
"error": err.Error(),
})
return mapAITriggerHTTPError(c, err, "AI analysis")
}
return response.Success(c, result, "AI analysis completed successfully")
}
// mapAITriggerHTTPError maps AI trigger service errors to safe HTTP responses.
func mapAITriggerHTTPError(c echo.Context, err error, action string) error {
switch {
case errors.Is(err, ErrEmptyTenderID), errors.Is(err, ErrInvalidTenderID):
return response.BadRequest(c, "Invalid request", "Invalid tender ID")
case errors.Is(err, ErrTenderNotFound), errors.Is(err, orm.ErrDocumentNotFound):
return response.NotFound(c, "Tender not found")
case errors.Is(err, ErrTenderMissingNoticeID), errors.Is(err, ErrTenderMissingContractFolderID):
return response.BadRequest(c, "Tender not eligible", "Tender is missing required identifiers for AI processing")
case errors.Is(err, ErrAINotConfigured):
return response.InternalServerError(c, "AI service is not available")
default:
return response.InternalServerError(c, action+" failed")
}
}
// TriggerAITranslate triggers on-demand AI translation for a tender
// @Summary Trigger AI translation
// @Description Trigger on-demand AI translation for tender title and description (cached in MinIO by the AI service)
internal/tender/repository.go
-31
View File
@@ -31,7 +31,6 @@ type TenderRepository interface {
GetTenderCountByClassification(ctx context.Context) (map[string]int64, error)
Search(ctx context.Context, form *SearchForm, pagination *response.Pagination) (*SearchPageResult, error)
GetExpiredTenders(ctx context.Context, beforeDate int64) ([]Tender, error)
GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error)
GetPendingDocumentScrapeTenders(ctx context.Context, countryCodes []string, minDeadline int64, limit, skip int) ([]Tender, bool, error)
GetUnSummarizedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error)
GetUnTranslatedTenders(ctx context.Context, language string, limit, skip int) ([]Tender, int64, error)
@@ -549,36 +548,6 @@ func (r *tenderRepository) GetExpiredTenders(ctx context.Context, beforeDate int
return result.Items, nil
}
// GetUnScrapedTenders retrieves tenders that haven't been processed for document scraping
func (r *tenderRepository) GetUnScrapedTenders(ctx context.Context, limit, skip int) ([]Tender, int64, error) {
filter := bson.M{
"$or": []bson.M{
{"processing_metadata.documents_scraped": bson.M{"$exists": false}},
{"processing_metadata.documents_scraped": false},
{"processing_metadata.documents_scraped": nil},
},
}
pagination := orm.Pagination{
Limit: limit,
Skip: skip,
SortField: "created_at",
SortOrder: 1, // Oldest first
}
result, err := r.ormRepo.FindAll(ctx, filter, pagination)
if err != nil {
r.logger.Error("Failed to get un-scraped tenders", map[string]interface{}{
"limit": limit,
"skip": skip,
"error": err.Error(),
})
return nil, 0, err
}
return result.Items, result.TotalCount, nil
}
// GetPendingDocumentScrapeTenders returns unscraped tenders for the given countries with an active deadline.
func (r *tenderRepository) GetPendingDocumentScrapeTenders(ctx context.Context, countryCodes []string, minDeadline int64, limit, skip int) ([]Tender, bool, error) {
if len(countryCodes) == 0 {
internal/tender/service.go
+171 -39
View File
@@ -15,13 +15,15 @@ import (
"tm/internal/customer"
"tm/pkg/ai_summarizer"
"tm/pkg/logger"
orm "tm/pkg/mongo"
"tm/pkg/response"
)
// AISummarizerClient defines the interface for on-demand AI summarization
// AISummarizerClient defines the interface for on-demand AI operations.
type AISummarizerClient interface {
FetchSummaryOnDemand(ctx context.Context, reqBody ai_summarizer.SummarizeRequest) (*ai_summarizer.SummarizeResponse, error)
FetchTranslationOnDemand(ctx context.Context, reqBody ai_summarizer.TranslateRequest) (*ai_summarizer.TranslateResponse, error)
FetchAnalyzeOnDemand(ctx context.Context, reqBody ai_summarizer.AnalyzeRequest) (*ai_summarizer.AnalyzeResponse, error)
TriggerPipelineTranslate(ctx context.Context, languages []string) (*ai_summarizer.PipelineTranslateResponse, error)
}
@@ -30,6 +32,7 @@ type AISummarizerClient interface {
// PROC_<contractFolderID>/<noticePublicationID>/ in the shared bucket.
type AISummarizerStorage interface {
GetSummaryFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) (string, error)
GetDocumentSummariesFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) ([]ai_summarizer.DocumentSummary, error)
GetTranslationFromStorage(ctx context.Context, contractFolderID, noticePublicationID, language string) (ai_summarizer.StoredTranslation, error)
ListDocuments(ctx context.Context, contractFolderID, noticePublicationID string) ([]ai_summarizer.StoredDocument, error)
DownloadDocument(ctx context.Context, objectName string) (io.ReadCloser, error)
@@ -60,6 +63,7 @@ type Service interface {
// AI summarization operations
GetAISummary(ctx context.Context, id string) (*AISummaryResponse, error)
TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error)
TriggerAIAnalyze(ctx context.Context, id string) (*AIAnalyzeResponse, error)
TriggerAITranslate(ctx context.Context, id, language string) (*AITranslateResponse, error)
GetAllAISummaries(ctx context.Context) ([]ai_summarizer.TenderSummaryItem, error)
// ListTendersWithScrapedDocuments lists tenders that have documents in MinIO. When syncMongoMetadata is true,
@@ -451,14 +455,13 @@ func (s *tenderService) Delete(ctx context.Context, id string) error {
return nil
}
// GetDocumentSummaries retrieves document summaries for a specific tender
// GetDocumentSummaries retrieves per-document summaries from the AI service MinIO storage.
func (s *tenderService) GetDocumentSummaries(ctx context.Context, id string) ([]DocumentSummary, error) {
if id == "" {
return nil, fmt.Errorf("tender ID cannot be empty")
}
// Get tender from repository
tender, err := s.repository.GetByID(ctx, id)
t, err := s.repository.GetByID(ctx, id)
if err != nil {
s.logger.Error("Failed to get tender for document summaries", map[string]interface{}{
"tender_id": id,
@@ -466,22 +469,60 @@ func (s *tenderService) GetDocumentSummaries(ctx context.Context, id string) ([]
})
return nil, fmt.Errorf("failed to get tender: %w", err)
}
if tender == nil {
if t == nil {
return nil, fmt.Errorf("tender not found")
}
// Return document summaries (empty slice if none exist)
if tender.DocumentSummaries == nil {
if strings.TrimSpace(t.NoticePublicationID) == "" || strings.TrimSpace(t.ContractFolderID) == "" {
return []DocumentSummary{}, nil
}
if s.aiSummarizerStorage == nil {
return []DocumentSummary{}, nil
}
stored, err := s.aiSummarizerStorage.GetDocumentSummariesFromStorage(ctx, t.ContractFolderID, t.NoticePublicationID)
if errors.Is(err, ai_summarizer.ErrSummaryNotReady) {
return []DocumentSummary{}, nil
}
if errors.Is(err, ai_summarizer.ErrObjectNotFound) {
return []DocumentSummary{}, nil
}
if err != nil {
if errors.Is(err, ai_summarizer.ErrMinIOUnavailable) {
return nil, fmt.Errorf("failed to get document summaries: MinIO connection unavailable: %w", err)
}
s.logger.Error("Failed to get document summaries from storage", map[string]interface{}{
"tender_id": id,
"error": err.Error(),
})
return nil, fmt.Errorf("failed to get document summaries: %w", err)
}
summaries := mapAIDocumentSummaries(stored)
s.logger.Info("Retrieved document summaries for tender", map[string]interface{}{
"tender_id": id,
"summaries_count": len(tender.DocumentSummaries),
"summaries_count": len(summaries),
"source": "storage",
})
return tender.DocumentSummaries, nil
return summaries, nil
}
func mapAIDocumentSummaries(items []ai_summarizer.DocumentSummary) []DocumentSummary {
if len(items) == 0 {
return []DocumentSummary{}
}
out := make([]DocumentSummary, 0, len(items))
for _, item := range items {
out = append(out, DocumentSummary{
DocumentName: item.DocumentName,
Summary: item.Summary,
SummaryLanguage: "en",
DocumentType: item.DocumentType,
SummaryModel: item.SummaryModel,
Error: item.Error,
})
}
return out
}
// GetDocuments retrieves scraped document metadata for a tender.
@@ -1147,11 +1188,11 @@ func (s *tenderService) GetAISummary(ctx context.Context, id string) (*AISummary
// It calls the external AI service HTTP API to generate summaries.
func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AISummarizeResponse, error) {
if id == "" {
return nil, fmt.Errorf("tender ID cannot be empty")
return nil, ErrEmptyTenderID
}
if s.aiSummarizerClient == nil {
return nil, fmt.Errorf("AI summarizer service is not configured")
return nil, ErrAINotConfigured
}
// Get tender to build the request
@@ -1161,37 +1202,27 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS
"tender_id": id,
"error": err.Error(),
})
if errors.Is(err, orm.ErrDocumentNotFound) {
return nil, ErrTenderNotFound
}
if strings.Contains(err.Error(), "invalid id format") {
return nil, ErrInvalidTenderID
}
return nil, fmt.Errorf("failed to get tender: %w", err)
}
if t == nil {
return nil, fmt.Errorf("tender not found")
return nil, ErrTenderNotFound
}
if t.NoticePublicationID == "" {
return nil, fmt.Errorf("tender has no notice publication ID")
return nil, ErrTenderMissingNoticeID
}
if strings.TrimSpace(t.ContractFolderID) == "" {
return nil, fmt.Errorf("tender has no contract folder ID")
return nil, ErrTenderMissingContractFolderID
}
authorityName := ""
if t.BuyerOrganization != nil {
authorityName = t.BuyerOrganization.Name
}
reqBody := ai_summarizer.NewSummarizeRequest(ai_summarizer.SummarizeRequestInput{
ContractFolderID: t.ContractFolderID,
NoticePublicationID: t.NoticePublicationID,
DocumentURL: t.DocumentURI,
Title: t.Title,
Description: t.Description,
Country: t.CountryCode,
Currency: t.Currency,
SubmissionDeadline: t.SubmissionDeadline,
EstimatedValue: t.EstimatedValue,
AuthorityName: authorityName,
})
reqBody := ai_summarizer.NewSummarizeRequest(t.ContractFolderID, t.NoticePublicationID)
s.logger.Info("Triggering on-demand AI summarization", map[string]interface{}{
"tender_id": id,
@@ -1247,6 +1278,95 @@ func (s *tenderService) TriggerAISummarize(ctx context.Context, id string) (*AIS
}, nil
}
// TriggerAIAnalyze triggers on-demand agentic analysis for a tender.
func (s *tenderService) TriggerAIAnalyze(ctx context.Context, id string) (*AIAnalyzeResponse, error) {
if id == "" {
return nil, ErrEmptyTenderID
}
if s.aiSummarizerClient == nil {
return nil, ErrAINotConfigured
}
t, err := s.repository.GetByID(ctx, id)
if err != nil {
s.logger.Error("Failed to get tender for on-demand analysis", map[string]interface{}{
"tender_id": id,
"error": err.Error(),
})
if errors.Is(err, orm.ErrDocumentNotFound) {
return nil, ErrTenderNotFound
}
if strings.Contains(err.Error(), "invalid id format") {
return nil, ErrInvalidTenderID
}
return nil, fmt.Errorf("failed to get tender: %w", err)
}
if t == nil {
return nil, ErrTenderNotFound
}
if t.NoticePublicationID == "" {
return nil, ErrTenderMissingNoticeID
}
if strings.TrimSpace(t.ContractFolderID) == "" {
return nil, ErrTenderMissingContractFolderID
}
reqBody := ai_summarizer.NewAnalyzeRequest(t.ContractFolderID, t.NoticePublicationID)
s.logger.Info("Triggering on-demand AI analysis", map[string]interface{}{
"tender_id": id,
"contract_folder_id": t.ContractFolderID,
"notice_publication_id": t.NoticePublicationID,
})
resp, err := s.aiSummarizerClient.FetchAnalyzeOnDemand(ctx, reqBody)
if err != nil {
s.logger.Error("On-demand AI analysis failed", map[string]interface{}{
"tender_id": id,
"contract_folder_id": t.ContractFolderID,
"notice_publication_id": t.NoticePublicationID,
"error": err.Error(),
})
return nil, fmt.Errorf("AI analysis request failed: %w", err)
}
documents := make([]AIAnalyzeDocument, 0, len(resp.Documents))
for _, doc := range resp.Documents {
documents = append(documents, AIAnalyzeDocument{
Filename: doc.Filename,
SizeBytes: doc.SizeBytes,
TimesRead: doc.TimesRead,
TimesQueried: doc.TimesQueried,
Description: doc.Description,
})
}
noticePublicationID := strings.TrimSpace(resp.NoticePublicationID)
if noticePublicationID == "" {
noticePublicationID = t.NoticePublicationID
}
contractFolderID := strings.TrimSpace(resp.ContractFolderID)
if contractFolderID == "" {
contractFolderID = t.ContractFolderID
}
s.logger.Info("On-demand AI analysis completed", map[string]interface{}{
"tender_id": id,
"contract_folder_id": contractFolderID,
"notice_publication_id": noticePublicationID,
"iterations": resp.Iterations,
})
return &AIAnalyzeResponse{
ContractFolderID: contractFolderID,
NoticePublicationID: noticePublicationID,
Summary: resp.Summary,
Documents: documents,
ToolCallsLog: resp.ToolCallsLog,
Iterations: resp.Iterations,
Model: resp.Model,
}, nil
}
// GetAllAISummaries retrieves all AI-generated summaries from the MinIO storage.
// It lists all tender.json files in the bucket and returns summaries for tenders
// where the AI pipeline has completed processing.
@@ -1510,11 +1630,12 @@ func (s *tenderService) buildAITranslateResponse(t *Tender, language, title, des
func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, language, requestSource string) (string, string, bool, error) {
if s.aiSummarizerStorage != nil {
stored, err := s.aiSummarizerStorage.GetTranslationFromStorage(ctx, t.ContractFolderID, t.NoticePublicationID, language)
if err == nil {
stored, _, err := s.lookupTranslationForTender(ctx, t, language)
if err == nil && strings.TrimSpace(stored.Title) != "" {
return stored.Title, stored.Description, false, nil
}
if !errors.Is(err, ai_summarizer.ErrTranslationNotReady) &&
if err != nil &&
!errors.Is(err, ai_summarizer.ErrTranslationNotReady) &&
!errors.Is(err, ai_summarizer.ErrObjectNotFound) &&
!errors.Is(err, ai_summarizer.ErrNoTranslation) {
s.logger.Debug("Translation not available from storage", map[string]interface{}{
@@ -1528,15 +1649,26 @@ func (s *tenderService) resolveTranslation(ctx context.Context, t *Tender, langu
resp, err := s.aiSummarizerClient.FetchTranslationOnDemand(ctx, ai_summarizer.TranslateRequest{
ContractFolderID: t.ContractFolderID,
NoticePublicationID: t.NoticePublicationID,
Title: t.Title,
Description: t.Description,
Language: language,
RequestSource: requestSource,
})
if err != nil {
return "", "", false, err
}
return resp.TranslatedTitle, resp.TranslatedDescription, true, nil
title := strings.TrimSpace(resp.TranslatedTitle)
description := resp.TranslatedDescription
if s.aiSummarizerStorage != nil {
if stored, _, storageErr := s.lookupTranslationForTender(ctx, t, language); storageErr == nil && strings.TrimSpace(stored.Title) != "" {
title = stored.Title
description = stored.Description
}
}
if title == "" {
title = strings.TrimSpace(resp.TranslatedTitle)
}
return title, description, true, nil
}
func (s *tenderService) pickResponseLanguage(language *string) string {
pkg/ai_summarizer/client.go
+92
View File
@@ -169,6 +169,98 @@ func (c *Client) TriggerPipelineTranslate(ctx context.Context, languages []strin
return &result, nil
}
// FetchAnalyzeOnDemand calls POST /ai/analyze for one tender.
func (c *Client) FetchAnalyzeOnDemand(ctx context.Context, reqBody AnalyzeRequest) (*AnalyzeResponse, error) {
jsonBody, err := json.Marshal(reqBody)
if err != nil {
return nil, fmt.Errorf("ai_summarizer: failed to marshal analyze request body: %w", err)
}
url := c.config.APIBaseURL + "/ai/analyze"
httpResp, bodyBytes, err := c.doRawPost(ctx, url, jsonBody)
if err != nil {
return nil, err
}
defer httpResp.Body.Close()
if httpResp.StatusCode == http.StatusNotFound {
return nil, fmt.Errorf("%w: status 404, body: %s", ErrObjectNotFound, string(bodyBytes))
}
if httpResp.StatusCode >= 400 {
return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes))
}
var result AnalyzeResponse
if err := json.Unmarshal(bodyBytes, &result); err != nil {
return nil, fmt.Errorf("ai_summarizer: failed to decode analyze response JSON: %w", err)
}
c.logger.Info("AI analyze request succeeded", map[string]interface{}{
"contract_folder_id": result.ContractFolderID,
"notice_publication_id": result.NoticePublicationID,
"iterations": result.Iterations,
})
return &result, nil
}
// TriggerPipelineSummarize calls POST /pipeline/summarize.
func (c *Client) TriggerPipelineSummarize(ctx context.Context) (*PipelineActionResponse, error) {
return c.triggerPipelineAction(ctx, "/pipeline/summarize", "pipeline summarize")
}
func (c *Client) triggerPipelineAction(ctx context.Context, path, label string) (*PipelineActionResponse, error) {
url := c.config.APIBaseURL + path
httpResp, bodyBytes, err := c.doRawPost(ctx, url, nil)
if err != nil {
return nil, err
}
defer httpResp.Body.Close()
if httpResp.StatusCode >= 400 {
return nil, fmt.Errorf("%w: status %d, body: %s", ErrAPINonSuccess, httpResp.StatusCode, string(bodyBytes))
}
var result PipelineActionResponse
if len(bodyBytes) > 0 {
if err := json.Unmarshal(bodyBytes, &result); err != nil {
return nil, fmt.Errorf("ai_summarizer: failed to decode %s response: %w", label, err)
}
}
c.logger.Info("AI pipeline action triggered", map[string]interface{}{
"path": path,
"status": result.Status,
})
return &result, nil
}
func (c *Client) doRawPost(ctx context.Context, url string, jsonBody []byte) (*http.Response, []byte, error) {
var body io.Reader
if len(jsonBody) > 0 {
body = bytes.NewReader(jsonBody)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, body)
if err != nil {
return nil, nil, fmt.Errorf("ai_summarizer: failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
httpResp, err := c.httpClient.Do(req)
if err != nil {
return nil, nil, fmt.Errorf("ai_summarizer: request error: %w", err)
}
bodyBytes, err := io.ReadAll(httpResp.Body)
if err != nil {
httpResp.Body.Close()
return nil, nil, fmt.Errorf("ai_summarizer: failed to read response body: %w", err)
}
return httpResp, bodyBytes, nil
}
// doPost performs a single POST request and parses the response.
func (c *Client) doPost(ctx context.Context, url string, jsonBody []byte) (*SummarizeResponse, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonBody))
pkg/ai_summarizer/entities.go
+75 -65
View File
@@ -2,58 +2,20 @@ package ai_summarizer
import (
"strings"
"time"
)
// SummarizeRequest represents the request payload sent to the AI service
// POST /ai/summarize endpoint for on-demand summarization.
// SummarizeRequest is the payload for POST /ai/summarize.
type SummarizeRequest struct {
ContractFolderID string `json:"contract_folder_id"`
NoticePublicationID string `json:"notice_publication_id"`
DocumentURL string `json:"document_url,omitempty"`
Title string `json:"title,omitempty"`
Description string `json:"description,omitempty"`
Country string `json:"country,omitempty"`
EstimatedValue *float64 `json:"estimated_value,omitempty"`
Currency string `json:"currency,omitempty"`
SubmissionDeadline string `json:"submission_deadline,omitempty"`
AuthorityName string `json:"authority_name,omitempty"`
}
// SummarizeRequestInput collects fields used to build SummarizeRequest.
type SummarizeRequestInput struct {
ContractFolderID string
NoticePublicationID string
DocumentURL string
Title string
Description string
Country string
Currency string
SubmissionDeadline int64 // Unix seconds; omitted from request when zero
EstimatedValue float64
AuthorityName string
ContractFolderID string `json:"contract_folder_id"`
NoticePublicationID string `json:"notice_publication_id"`
}
// NewSummarizeRequest builds a SummarizeRequest for POST /ai/summarize.
func NewSummarizeRequest(in SummarizeRequestInput) SummarizeRequest {
req := SummarizeRequest{
ContractFolderID: strings.TrimSpace(in.ContractFolderID),
NoticePublicationID: strings.TrimSpace(in.NoticePublicationID),
DocumentURL: strings.TrimSpace(in.DocumentURL),
Title: in.Title,
Description: in.Description,
Country: in.Country,
Currency: in.Currency,
AuthorityName: strings.TrimSpace(in.AuthorityName),
func NewSummarizeRequest(contractFolderID, noticePublicationID string) SummarizeRequest {
return SummarizeRequest{
ContractFolderID: strings.TrimSpace(contractFolderID),
NoticePublicationID: strings.TrimSpace(noticePublicationID),
}
if in.EstimatedValue > 0 {
v := in.EstimatedValue
req.EstimatedValue = &v
}
if in.SubmissionDeadline > 0 {
req.SubmissionDeadline = time.Unix(in.SubmissionDeadline, 0).UTC().Format(time.RFC3339)
}
return req
}
// SummarizeResponse represents the response from the AI summarization API.
@@ -63,6 +25,8 @@ type SummarizeResponse struct {
NoticeID string `json:"notice_id"` // legacy field name
Summaries []DocumentSummary `json:"summaries"`
OverallSummary *string `json:"overall_summary"`
RollupS float64 `json:"rollup_s"`
TotalS float64 `json:"total_s"`
}
// ResolvedNoticePublicationID returns the notice publication ID from the response.
@@ -82,7 +46,7 @@ type DocumentSummary struct {
DocumentType string `json:"document_type"`
Summary string `json:"summary"`
SummaryModel string `json:"summary_model"`
Error string `json:"error"` // empty string when no error
Error string `json:"error"` // empty when no error
}
// ProcedureDocumentsSummary reports scraped documents stored under one procedure folder in MinIO.
@@ -101,13 +65,10 @@ type StoredDocument struct {
DocumentType string `json:"document_type,omitempty"`
}
// TranslateRequest represents the request payload sent to the AI service
// POST /ai/translate endpoint.
// TranslateRequest is the payload for POST /ai/translate.
type TranslateRequest struct {
ContractFolderID string `json:"contract_folder_id"`
NoticePublicationID string `json:"notice_publication_id"`
Title string `json:"title"`
Description string `json:"description"`
Language string `json:"language"`
RequestSource string `json:"-"` // internal: daily_job | manual_trigger
}
@@ -126,6 +87,41 @@ type TranslateResponse struct {
TranslatedDescription string `json:"translated_description"`
}
// AnalyzeRequest is the payload for POST /ai/analyze.
type AnalyzeRequest struct {
ContractFolderID string `json:"contract_folder_id"`
NoticePublicationID string `json:"notice_publication_id"`
}
// NewAnalyzeRequest builds an AnalyzeRequest for POST /ai/analyze.
func NewAnalyzeRequest(contractFolderID, noticePublicationID string) AnalyzeRequest {
return AnalyzeRequest{
ContractFolderID: strings.TrimSpace(contractFolderID),
NoticePublicationID: strings.TrimSpace(noticePublicationID),
}
}
// AnalyzeDocumentStats describes one document touched during agentic analysis.
type AnalyzeDocumentStats struct {
Filename string `json:"filename"`
SizeBytes int64 `json:"size_bytes"`
TimesRead int `json:"times_read"`
TimesQueried int `json:"times_queried"`
Description string `json:"description"`
}
// AnalyzeResponse is the payload from POST /ai/analyze.
type AnalyzeResponse struct {
ContractFolderID string `json:"contract_folder_id"`
NoticePublicationID string `json:"notice_publication_id"`
Summary string `json:"summary"`
Documents []AnalyzeDocumentStats `json:"documents"`
Scratchpad map[string]interface{} `json:"scratchpad"`
ToolCallsLog []string `json:"tool_calls_log"`
Iterations int `json:"iterations"`
Model string `json:"model"`
}
// PipelineTranslateRequest represents the request payload for POST /pipeline/translate.
type PipelineTranslateRequest struct {
Languages []string `json:"languages"`
@@ -137,6 +133,11 @@ type PipelineTranslateResponse struct {
Languages []string `json:"languages"`
}
// PipelineActionResponse is returned by fire-and-forget pipeline endpoints.
type PipelineActionResponse struct {
Status string `json:"status"`
}
// StoredTranslation is one language entry inside tender.json translations map.
type StoredTranslation struct {
Title string `json:"title"`
@@ -173,6 +174,7 @@ type TenderJSON struct {
Summary string `json:"summary"`
OverallSummary *string `json:"overall_summary"`
OverallSummaryCamel *string `json:"overallSummary"`
Summaries []DocumentSummary `json:"summaries"`
SummarizeStatus SummarizeStatus `json:"summarize_status"`
SummarizeStatusCamel SummarizeStatus `json:"summarizeStatus"`
Translations map[string]StoredTranslation `json:"translations"`
@@ -225,23 +227,31 @@ func (t *TenderJSON) translationsDone() []string {
return t.TranslationStatusCamel.languagesDone()
}
// translationForLanguage returns stored translation text when the pipeline marked
// the language done and non-empty title exists.
// storedTranslationText returns translation text from the translations map when
// title is non-empty. It does not require translation_status.done — on-demand
// writes may persist translations before updating status.
func (t *TenderJSON) storedTranslationText(language string) (StoredTranslation, bool) {
if t == nil || t.Translations == nil {
return StoredTranslation{}, false
}
lang := strings.ToLower(strings.TrimSpace(language))
for key, entry := range t.Translations {
if strings.ToLower(strings.TrimSpace(key)) != lang {
continue
}
if strings.TrimSpace(entry.Title) == "" {
continue
}
return entry, true
}
return StoredTranslation{}, false
}
// translationForLanguage returns stored translation when the pipeline marked the
// language done and non-empty title exists.
func (t *TenderJSON) translationForLanguage(language string) (StoredTranslation, bool) {
if t == nil || !languageInDone(t.translationsDone(), language) {
return StoredTranslation{}, false
}
lang := strings.ToLower(strings.TrimSpace(language))
if t.Translations != nil {
for key, entry := range t.Translations {
if strings.ToLower(strings.TrimSpace(key)) != lang {
continue
}
if strings.TrimSpace(entry.Title) == "" {
continue
}
return entry, true
}
}
return StoredTranslation{}, false
return t.storedTranslationText(language)
}
pkg/ai_summarizer/storage.go
+43 -7
View File
@@ -326,8 +326,9 @@ func (s *StorageClient) GetSummaryFromStorage(ctx context.Context, contractFolde
return s.getSummaryFromObjectKey(ctx, prefix+"tender.json", contractFolderID, noticePublicationID)
}
// GetTranslationFromStorage reads tender.json and returns the translation for a language
// when translation_status.done includes that language.
// GetTranslationFromStorage reads tender.json and returns the translation for a language.
// It prefers entries marked done in translation_status; if missing, it falls back to
// translations[<lang>] when title is non-empty (on-demand API may lag status updates).
func (s *StorageClient) GetTranslationFromStorage(ctx context.Context, contractFolderID, noticePublicationID, language string) (StoredTranslation, error) {
if strings.TrimSpace(contractFolderID) == "" {
return StoredTranslation{}, fmt.Errorf("ai_summarizer: contractFolderID is required")
@@ -353,14 +354,16 @@ func (s *StorageClient) getTranslationFromObjectKey(ctx context.Context, objectK
return StoredTranslation{}, err
}
if entry, ok := tenderJSON.translationForLanguage(language); ok {
return entry, nil
}
if entry, ok := tenderJSON.storedTranslationText(language); ok {
return entry, nil
}
if !languageInDone(tenderJSON.translationsDone(), language) {
return StoredTranslation{}, ErrTranslationNotReady
}
entry, ok := tenderJSON.translationForLanguage(language)
if !ok {
return StoredTranslation{}, ErrNoTranslation
}
return entry, nil
return StoredTranslation{}, ErrNoTranslation
}
// getSummaryFromObjectKey reads tender.json at objectKey and returns summary text when ready.
@@ -432,6 +435,39 @@ func (s *StorageClient) getSummaryFromObjectKey(ctx context.Context, objectKey,
return summaryText, nil
}
// GetDocumentSummariesFromStorage reads per-document summaries from tender.json
// when summarize_status.done is true.
func (s *StorageClient) GetDocumentSummariesFromStorage(ctx context.Context, contractFolderID, noticePublicationID string) ([]DocumentSummary, error) {
if strings.TrimSpace(contractFolderID) == "" {
return nil, fmt.Errorf("ai_summarizer: contractFolderID is required")
}
if strings.TrimSpace(noticePublicationID) == "" {
return nil, fmt.Errorf("ai_summarizer: noticePublicationID is required")
}
prefix, err := s.resolveNoticeStoragePrefix(ctx, contractFolderID, noticePublicationID)
if err != nil {
return nil, err
}
tenderJSON, err := s.readTenderJSONAtKey(ctx, prefix+"tender.json")
if err != nil {
return nil, err
}
done, _ := tenderJSON.resolved()
if !done {
return nil, ErrSummaryNotReady
}
if len(tenderJSON.Summaries) == 0 {
return []DocumentSummary{}, nil
}
out := make([]DocumentSummary, len(tenderJSON.Summaries))
copy(out, tenderJSON.Summaries)
return out, nil
}
// IsSummaryReady is a convenience method that checks whether the tender.json
// exists and the summarize_status.done flag is true, without returning the
// actual summary text.
pkg/queue/config.go
-53
View File
@@ -1,53 +0,0 @@
package queue
import (
"time"
)
/*
Queue Configuration and Constants
This package provides a Redis-based job queue system for managing
document scraping tasks in a distributed manner.
*/
const (
// Queue names
DocumentScraperQueue = "scraper:document:queue"
DocumentScraperDLQ = "scraper:document:dlq" // Dead Letter Queue
// Job status keys
JobStatusPrefix = "job:"
QueueMetricsKey = "queue:metrics"
// Default timeouts
DefaultJobTimeout = 5 * time.Minute
DefaultMaxRetries = 3
DefaultRetryBackoff = 10 * time.Second
MaxRetryBackoff = 5 * time.Minute // Cap retry backoff at 5 minutes
InProgressExpiration = 10 * time.Minute // Job must complete within this time
)
// QueueConfig represents configuration for the queue service
type QueueConfig struct {
RedisURL string
MaxRetries int
RetryBackoff time.Duration
JobTimeout time.Duration
PoolSize int
IsDLQEnabled bool
MaxQueueLength int // 0 means unlimited
}
// DefaultConfig returns default queue configuration
func DefaultConfig() QueueConfig {
return QueueConfig{
RedisURL: "redis://localhost:6379",
MaxRetries: DefaultMaxRetries,
RetryBackoff: DefaultRetryBackoff,
JobTimeout: DefaultJobTimeout,
PoolSize: 10,
IsDLQEnabled: true,
MaxQueueLength: 0, // Unlimited by default
}
}
pkg/queue/models.go
-85
View File
@@ -1,85 +0,0 @@
package queue
import (
"encoding/json"
"time"
)
// JobStatus represents the status of a queue job
type JobStatus string
const (
StatusPending JobStatus = "pending"
StatusInProgress JobStatus = "in_progress"
StatusCompleted JobStatus = "completed"
StatusFailed JobStatus = "failed"
StatusRetrying JobStatus = "retrying"
)
// DocumentScraperJob represents a document scraping task in the queue
type DocumentScraperJob struct {
ID string `json:"id"` // Unique job ID (typically tender ID)
NoticeID string `json:"notice_id"` // Notice publication ID
NoticeURL string `json:"notice_url"` // Document URL to scrape
TenderID string `json:"tender_id,omitempty"` // Associated tender ID from MongoDB
CreatedAt int64 `json:"created_at"`
UpdatedAt int64 `json:"updated_at"`
Attempts int `json:"attempts"`
MaxRetries int `json:"max_retries"`
Status string `json:"status"`
Error string `json:"error,omitempty"`
NextRetryTime int64 `json:"next_retry_time,omitempty"`
CompletedAt int64 `json:"completed_at,omitempty"`
ProcessingStartedAt int64 `json:"processing_started_at,omitempty"`
}
// NewDocumentScraperJob creates a new document scraper job
func NewDocumentScraperJob(noticeID, noticeURL, tenderID string, maxRetries int) *DocumentScraperJob {
now := time.Now().Unix()
return &DocumentScraperJob{
ID: noticeID,
NoticeID: noticeID,
NoticeURL: noticeURL,
TenderID: tenderID,
CreatedAt: now,
UpdatedAt: now,
Attempts: 0,
MaxRetries: maxRetries,
Status: string(StatusPending),
}
}
// ToJSON marshals the job to JSON
func (j *DocumentScraperJob) ToJSON() (string, error) {
data, err := json.Marshal(j)
return string(data), err
}
// FromJSON unmarshals JSON to DocumentScraperJob
func (j *DocumentScraperJob) FromJSON(data string) error {
return json.Unmarshal([]byte(data), j)
}
// QueueMetrics represents queue performance metrics
type QueueMetrics struct {
TotalJobs int64 `json:"total_jobs"`
PendingJobs int64 `json:"pending_jobs"`
InProgressJobs int64 `json:"in_progress_jobs"`
CompletedJobs int64 `json:"completed_jobs"`
FailedJobs int64 `json:"failed_jobs"`
AverageWaitTime time.Duration `json:"average_wait_time"`
AverageProcessTime time.Duration `json:"average_process_time"`
UpdatedAt int64 `json:"updated_at"`
}
// JobResult represents the result of a completed job
type JobResult struct {
JobID string `json:"job_id"`
Status string `json:"status"`
Success bool `json:"success"`
Error string `json:"error,omitempty"`
CompletedAt int64 `json:"completed_at"`
Duration int64 `json:"duration_ms"`
UploadedCount int `json:"uploaded_count"`
Details map[string]interface{} `json:"details,omitempty"`
}
pkg/queue/queue.go
-357
View File
@@ -1,357 +0,0 @@
package queue
import (
"context"
"fmt"
"time"
"tm/pkg/logger"
"github.com/redis/go-redis/v9"
)
// Queue interface defines the queue operations
type Queue interface {
// Job operations
Enqueue(ctx context.Context, job *DocumentScraperJob) error
Dequeue(ctx context.Context) (*DocumentScraperJob, error)
AckJob(ctx context.Context, jobID string) error
NackJob(ctx context.Context, jobID string, err error) error
GetJobStatus(ctx context.Context, jobID string) (JobStatus, error)
// Queue inspection
GetMetrics(ctx context.Context) (*QueueMetrics, error)
GetQueueSize(ctx context.Context) (int64, error)
GetDLQSize(ctx context.Context) (int64, error)
GetInProgressJobs(ctx context.Context) ([]DocumentScraperJob, error)
// Cleanup operations
PurgeQueue(ctx context.Context) error
PurgeDLQ(ctx context.Context) error
Close() error
}
// RedisQueue implements Queue interface using Redis
type RedisQueue struct {
client *redis.Client
config QueueConfig
logger logger.Logger
}
// NewRedisQueue creates a new Redis-based queue
func NewRedisQueue(config QueueConfig, logger logger.Logger) (Queue, error) {
opt, err := redis.ParseURL(config.RedisURL)
if err != nil {
logger.Error("Failed to parse Redis URL", map[string]interface{}{
"url": config.RedisURL,
"error": err.Error(),
})
return nil, fmt.Errorf("failed to parse redis url: %w", err)
}
opt.PoolSize = config.PoolSize
client := redis.NewClient(opt)
// Test connection
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := client.Ping(ctx).Err(); err != nil {
logger.Error("Failed to connect to Redis", map[string]interface{}{
"error": err.Error(),
})
return nil, fmt.Errorf("failed to connect to redis: %w", err)
}
logger.Info("Redis queue initialized successfully", map[string]interface{}{
"pool_size": config.PoolSize,
})
return &RedisQueue{
client: client,
config: config,
logger: logger,
}, nil
}
// Enqueue adds a job to the queue
func (q *RedisQueue) Enqueue(ctx context.Context, job *DocumentScraperJob) error {
if q.config.MaxQueueLength > 0 {
size, err := q.GetQueueSize(ctx)
if err != nil {
q.logger.Warn("Failed to check queue size", map[string]interface{}{
"error": err.Error(),
})
} else if size >= int64(q.config.MaxQueueLength) {
return fmt.Errorf("queue is full: %d/%d", size, q.config.MaxQueueLength)
}
}
job.CreatedAt = time.Now().Unix()
job.UpdatedAt = time.Now().Unix()
job.Status = string(StatusPending)
jsonData, err := job.ToJSON()
if err != nil {
q.logger.Error("Failed to marshal job", map[string]interface{}{
"job_id": job.ID,
"error": err.Error(),
})
return err
}
// Push to queue
if err := q.client.RPush(ctx, DocumentScraperQueue, jsonData).Err(); err != nil {
q.logger.Error("Failed to enqueue job", map[string]interface{}{
"job_id": job.ID,
"error": err.Error(),
})
return err
}
// Store job metadata with TTL
jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, job.ID)
if err := q.client.Set(ctx, jobKey, string(StatusPending), 24*time.Hour).Err(); err != nil {
q.logger.Warn("Failed to set job status", map[string]interface{}{
"job_id": job.ID,
"error": err.Error(),
})
}
q.logger.Info("Job enqueued successfully", map[string]interface{}{
"job_id": job.ID,
"notice_id": job.NoticeID,
})
return nil
}
// Dequeue retrieves and locks the next job from the queue
func (q *RedisQueue) Dequeue(ctx context.Context) (*DocumentScraperJob, error) {
// Use BLPOP to get next job with timeout
result, err := q.client.BLPop(ctx, 30*time.Second, DocumentScraperQueue).Result()
if err != nil {
if err == redis.Nil {
return nil, nil // Queue is empty
}
q.logger.Error("Failed to dequeue job", map[string]interface{}{
"error": err.Error(),
})
return nil, err
}
if len(result) < 2 {
return nil, fmt.Errorf("invalid dequeue result")
}
jobData := result[1]
// Parse job
var job DocumentScraperJob
if err := job.FromJSON(jobData); err != nil {
q.logger.Error("Failed to parse job", map[string]interface{}{
"error": err.Error(),
"data": jobData,
})
return nil, err
}
// Mark as in progress with TTL
job.Status = string(StatusInProgress)
job.Attempts++
job.ProcessingStartedAt = time.Now().Unix()
job.UpdatedAt = time.Now().Unix()
jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, job.ID)
if err := q.client.Set(ctx, jobKey, string(StatusInProgress), InProgressExpiration).Err(); err != nil {
q.logger.Warn("Failed to update job status", map[string]interface{}{
"job_id": job.ID,
"error": err.Error(),
})
}
q.logger.Info("Job dequeued successfully", map[string]interface{}{
"job_id": job.ID,
"attempt": job.Attempts,
"max_retry": job.MaxRetries,
})
return &job, nil
}
// AckJob marks a job as completed
func (q *RedisQueue) AckJob(ctx context.Context, jobID string) error {
now := time.Now().Unix()
jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID)
// Update status with TTL for record keeping
if err := q.client.Set(ctx, jobKey, string(StatusCompleted), 48*time.Hour).Err(); err != nil {
q.logger.Warn("Failed to ack job", map[string]interface{}{
"job_id": jobID,
"error": err.Error(),
})
}
// Store completion timestamp
completedKey := fmt.Sprintf("%s:completed:%s", DocumentScraperQueue, jobID)
if err := q.client.Set(ctx, completedKey, now, 48*time.Hour).Err(); err != nil {
q.logger.Warn("Failed to store completion timestamp", map[string]interface{}{
"job_id": jobID,
"error": err.Error(),
})
}
q.logger.Info("Job acknowledged", map[string]interface{}{
"job_id": jobID,
})
return nil
}
// NackJob marks a job as failed or for retry
func (q *RedisQueue) NackJob(ctx context.Context, jobID string, jobErr error) error {
jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID)
// Get the job from Redis to check retry attempts
_, err := q.client.Get(ctx, jobKey).Result()
if err != nil && err != redis.Nil {
q.logger.Warn("Failed to get job for nack", map[string]interface{}{
"job_id": jobID,
"error": err.Error(),
})
}
// Check if we should retry by trying to get the job back from processing
// For now, mark as failed
status := StatusFailed
duration := 48 * time.Hour
if err := q.client.Set(ctx, jobKey, string(status), duration).Err(); err != nil {
q.logger.Warn("Failed to update job status on nack", map[string]interface{}{
"job_id": jobID,
"error": err.Error(),
})
}
// Store error message
errorKey := fmt.Sprintf("%s:error:%s", DocumentScraperQueue, jobID)
if err := q.client.Set(ctx, errorKey, jobErr.Error(), duration).Err(); err != nil {
q.logger.Warn("Failed to store error message", map[string]interface{}{
"job_id": jobID,
"error": err.Error(),
})
}
// Optionally move to DLQ
if q.config.IsDLQEnabled {
jobData, _ := q.client.Get(ctx, JobStatusPrefix+jobID).Result()
if jobData != "" {
if err := q.client.RPush(ctx, DocumentScraperDLQ, jobData).Err(); err != nil {
q.logger.Warn("Failed to add job to DLQ", map[string]interface{}{
"job_id": jobID,
"error": err.Error(),
})
}
}
}
q.logger.Info("Job nacked", map[string]interface{}{
"job_id": jobID,
"error": jobErr.Error(),
})
return nil
}
// GetJobStatus retrieves the status of a job
func (q *RedisQueue) GetJobStatus(ctx context.Context, jobID string) (JobStatus, error) {
jobKey := fmt.Sprintf("%s%s", JobStatusPrefix, jobID)
status, err := q.client.Get(ctx, jobKey).Result()
if err != nil {
if err == redis.Nil {
return StatusPending, nil
}
return StatusPending, err
}
return JobStatus(status), nil
}
// GetMetrics returns queue metrics
func (q *RedisQueue) GetMetrics(ctx context.Context) (*QueueMetrics, error) {
metrics := &QueueMetrics{
UpdatedAt: time.Now().Unix(),
}
// Get queue size
size, err := q.GetQueueSize(ctx)
if err != nil {
q.logger.Warn("Failed to get queue size for metrics", map[string]interface{}{
"error": err.Error(),
})
}
metrics.PendingJobs = size
// Get DLQ size
dlqSize, err := q.GetDLQSize(ctx)
if err != nil {
q.logger.Warn("Failed to get DLQ size for metrics", map[string]interface{}{
"error": err.Error(),
})
}
metrics.FailedJobs = dlqSize
return metrics, nil
}
// GetQueueSize returns the number of items in the queue
func (q *RedisQueue) GetQueueSize(ctx context.Context) (int64, error) {
size, err := q.client.LLen(ctx, DocumentScraperQueue).Result()
if err != nil {
return 0, err
}
return size, nil
}
// GetDLQSize returns the number of items in the dead letter queue
func (q *RedisQueue) GetDLQSize(ctx context.Context) (int64, error) {
size, err := q.client.LLen(ctx, DocumentScraperDLQ).Result()
if err != nil {
return 0, err
}
return size, nil
}
// GetInProgressJobs returns jobs currently being processed
func (q *RedisQueue) GetInProgressJobs(ctx context.Context) ([]DocumentScraperJob, error) {
// This is a simplified implementation
// In production, you'd track this separately
return nil, nil
}
// PurgeQueue removes all jobs from the queue
func (q *RedisQueue) PurgeQueue(ctx context.Context) error {
if err := q.client.Del(ctx, DocumentScraperQueue).Err(); err != nil {
q.logger.Error("Failed to purge queue", map[string]interface{}{
"error": err.Error(),
})
return err
}
q.logger.Info("Queue purged successfully", map[string]interface{}{})
return nil
}
// PurgeDLQ removes all jobs from the dead letter queue
func (q *RedisQueue) PurgeDLQ(ctx context.Context) error {
if err := q.client.Del(ctx, DocumentScraperDLQ).Err(); err != nil {
q.logger.Error("Failed to purge DLQ", map[string]interface{}{
"error": err.Error(),
})
return err
}
q.logger.Info("DLQ purged successfully", map[string]interface{}{})
return nil
}
// Close closes the Redis connection
func (q *RedisQueue) Close() error {
return q.client.Close()
}