report-listener: add CleanAI intelligence endpoint with server-side limits

Author: borisolver

report-listener/config/config.go

@@ -32,6 +32,11 @@ type Config struct {
 	RabbitExchange                 string
 	RabbitAnalysedReportRoutingKey string
 	RabbitTwitterReplyRoutingKey   string
+
+	// Intelligence chat configuration
+	GeminiAPIKey                string
+	GeminiModel                 string
+	IntelligenceFreeTierMaxTurn int
 }
 
 // Load loads configuration from environment variables
@@ -61,6 +66,11 @@ func Load() *Config {
 		RabbitExchange:                 getEnv("RABBITMQ_EXCHANGE", "cleanapp"),
 		RabbitAnalysedReportRoutingKey: getEnv("RABBITMQ_ANALYSED_REPORT_ROUTING_KEY", "report.analysed"),
 		RabbitTwitterReplyRoutingKey:   getEnv("RABBITMQ_TWITTER_REPLY_ROUTING_KEY", "twitter.reply"),
+
+		// Intelligence defaults
+		GeminiAPIKey:                getEnv("GEMINI_API_KEY", ""),
+		GeminiModel:                 getEnv("GEMINI_MODEL", "gemini-2.5-flash"),
+		IntelligenceFreeTierMaxTurn: getIntEnv("INTELLIGENCE_FREE_MAX_TURNS", 5),
 	}
 
 	return config

report-listener/database/intelligence.go

@@ -0,0 +1,194 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+)
+
+type NamedCount struct {
+	Name  string
+	Count int
+}
+
+type IntelligenceContext struct {
+	OrgID               string
+	ReportsAnalyzed     int
+	ReportsThisMonth    int
+	HighPriorityCount   int
+	MediumPriorityCount int
+	TopClassifications  []NamedCount
+	TopIssues           []NamedCount
+	RecentSummaries     []string
+}
+
+func (d *Database) EnsureIntelligenceTables(ctx context.Context) error {
+	_, err := d.db.ExecContext(ctx, `
+		CREATE TABLE IF NOT EXISTS intelligence_usage (
+			session_id VARCHAR(128) PRIMARY KEY,
+			turns_used INT NOT NULL DEFAULT 0,
+			expires_at TIMESTAMP NOT NULL,
+			created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+			updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
+		)
+	`)
+	if err != nil {
+		return fmt.Errorf("failed to ensure intelligence_usage table: %w", err)
+	}
+	return nil
+}
+
+func (d *Database) GetAndIncrementIntelligenceUsage(ctx context.Context, sessionID string, maxTurns int, ttl time.Duration) (bool, int, error) {
+	if strings.TrimSpace(sessionID) == "" {
+		return false, 0, fmt.Errorf("session_id is required")
+	}
+	if maxTurns <= 0 {
+		return true, 0, nil
+	}
+
+	tx, err := d.db.BeginTx(ctx, nil)
+	if err != nil {
+		return false, 0, fmt.Errorf("failed to start tx: %w", err)
+	}
+	defer tx.Rollback()
+
+	now := time.Now().UTC()
+	var turnsUsed int
+	var expiresAt time.Time
+	hasRow := true
+
+	rowErr := tx.QueryRowContext(ctx, `
+		SELECT turns_used, expires_at
+		FROM intelligence_usage
+		WHERE session_id = ?
+		FOR UPDATE
+	`, sessionID).Scan(&turnsUsed, &expiresAt)
+	if rowErr == sql.ErrNoRows {
+		hasRow = false
+		turnsUsed = 0
+		expiresAt = now.Add(ttl)
+	} else if rowErr != nil {
+		return false, 0, fmt.Errorf("failed to read intelligence usage: %w", rowErr)
+	}
+
+	expired := !hasRow || !expiresAt.After(now)
+	if expired {
+		turnsUsed = 0
+		expiresAt = now.Add(ttl)
+	}
+
+	if turnsUsed >= maxTurns {
+		if err := tx.Commit(); err != nil {
+			return false, turnsUsed, fmt.Errorf("failed to commit usage tx: %w", err)
+		}
+		return false, turnsUsed, nil
+	}
+
+	turnsUsed++
+	if _, err := tx.ExecContext(ctx, `
+		INSERT INTO intelligence_usage (session_id, turns_used, expires_at)
+		VALUES (?, ?, ?)
+		ON DUPLICATE KEY UPDATE
+			turns_used = VALUES(turns_used),
+			expires_at = VALUES(expires_at)
+	`, sessionID, turnsUsed, expiresAt); err != nil {
+		return false, 0, fmt.Errorf("failed to upsert intelligence usage: %w", err)
+	}
+
+	if err := tx.Commit(); err != nil {
+		return false, 0, fmt.Errorf("failed to commit usage tx: %w", err)
+	}
+	return true, turnsUsed, nil
+}
+
+func (d *Database) GetIntelligenceContext(ctx context.Context, orgID string) (*IntelligenceContext, error) {
+	org := strings.ToLower(strings.TrimSpace(orgID))
+	if org == "" {
+		return nil, fmt.Errorf("org_id is required")
+	}
+
+	total, high, medium, err := d.GetBrandPriorityCountsByBrandName(ctx, org)
+	if err != nil {
+		return nil, err
+	}
+
+	result := &IntelligenceContext{
+		OrgID:               org,
+		ReportsAnalyzed:     total,
+		HighPriorityCount:   high,
+		MediumPriorityCount: medium,
+	}
+
+	_ = d.db.QueryRowContext(ctx, `
+		SELECT COUNT(*)
+		FROM reports r
+		INNER JOIN report_analysis ra ON ra.seq = r.seq
+		WHERE ra.brand_name = ?
+		AND ra.is_valid = TRUE
+		AND r.ts >= DATE_FORMAT(UTC_TIMESTAMP(), '%Y-%m-01 00:00:00')
+	`, org).Scan(&result.ReportsThisMonth)
+
+	classRows, err := d.db.QueryContext(ctx, `
+		SELECT COALESCE(classification, 'unknown') AS classification, COUNT(*) AS cnt
+		FROM report_analysis
+		WHERE brand_name = ?
+		AND is_valid = TRUE
+		GROUP BY classification
+		ORDER BY cnt DESC
+		LIMIT 5
+	`, org)
+	if err == nil {
+		defer classRows.Close()
+		for classRows.Next() {
+			var item NamedCount
+			if scanErr := classRows.Scan(&item.Name, &item.Count); scanErr == nil {
+				result.TopClassifications = append(result.TopClassifications, item)
+			}
+		}
+	}
+
+	issueRows, err := d.db.QueryContext(ctx, `
+		SELECT title, COUNT(*) AS cnt
+		FROM report_analysis
+		WHERE brand_name = ?
+		AND is_valid = TRUE
+		AND title IS NOT NULL
+		AND title != ''
+		GROUP BY title
+		ORDER BY cnt DESC
+		LIMIT 5
+	`, org)
+	if err == nil {
+		defer issueRows.Close()
+		for issueRows.Next() {
+			var item NamedCount
+			if scanErr := issueRows.Scan(&item.Name, &item.Count); scanErr == nil {
+				result.TopIssues = append(result.TopIssues, item)
+			}
+		}
+	}
+
+	summaryRows, err := d.db.QueryContext(ctx, `
+		SELECT summary
+		FROM report_analysis
+		WHERE brand_name = ?
+		AND is_valid = TRUE
+		AND summary IS NOT NULL
+		AND summary != ''
+		ORDER BY created_at DESC
+		LIMIT 5
+	`, org)
+	if err == nil {
+		defer summaryRows.Close()
+		for summaryRows.Next() {
+			var summary string
+			if scanErr := summaryRows.Scan(&summary); scanErr == nil {
+				result.RecentSummaries = append(result.RecentSummaries, summary)
+			}
+		}
+	}
+
+	return result, nil
+}

report-listener/handlers/handlers.go

@@ -13,7 +13,9 @@ import (
 	"sync"
 	"time"
 
+	"report-listener/config"
 	"report-listener/database"
+	"report-listener/intelligence"
 	"report-listener/models"
 	"report-listener/rabbitmq"
 	brandutil "report-listener/utils"
@@ -50,18 +52,30 @@ type Handlers struct {
 	db                *database.Database
 	rabbitmqPublisher *rabbitmq.Publisher
 	rabbitmqReplier   *rabbitmq.Publisher
+	cfg               *config.Config
+	geminiClient      *intelligence.Client
 
 	brandCountsMu    sync.RWMutex
 	brandCountsCache map[string]brandCountsCacheEntry
 }
 
 // NewHandlers creates a new handlers instance
-func NewHandlers(hub *ws.Hub, db *database.Database, pub *rabbitmq.Publisher, replyPub *rabbitmq.Publisher) *Handlers {
+func NewHandlers(
+	cfg *config.Config,
+	hub *ws.Hub,
+	db *database.Database,
+	pub *rabbitmq.Publisher,
+	replyPub *rabbitmq.Publisher,
+) *Handlers {
+	geminiClient := intelligence.NewClient(cfg.GeminiAPIKey, cfg.GeminiModel)
+
 	return &Handlers{
 		hub:               hub,
 		db:                db,
 		rabbitmqPublisher: pub,
 		rabbitmqReplier:   replyPub,
+		cfg:               cfg,
+		geminiClient:      geminiClient,
 		brandCountsCache:  make(map[string]brandCountsCacheEntry),
 	}
 }