diff --git a/.env.example b/.env.example index 411d307..e72bcab 100644 --- a/.env.example +++ b/.env.example @@ -11,3 +11,12 @@ SESSION_SECRET=replace-with-at-least-32-random-characters CURSOR_SECRET=replace-with-a-different-32-character-secret FRONTEND_ORIGIN=http://127.0.0.1:5173 COOKIE_SECURE=false +AI_GROWTH_ENABLED=false +AVALAI_API_KEY= +AVALAI_BASE_URL=https://api.avalai.ir +AVALAI_MODEL=gpt-5-mini-2025-08-07 +AI_DAILY_REQUEST_BUDGET=0 +AI_DAILY_TOKEN_BUDGET=0 +AI_MAX_CONCURRENT=4 +AI_CREATE_PER_MINUTE=10 +AI_UNCACHED_SCOPES_PER_DAY=5 diff --git a/cmd/api/main.go b/cmd/api/main.go index a5047c2..2e71edb 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -13,7 +13,9 @@ import ( "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/api" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/config" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/database" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/growthreport" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/identity" + "github.com/redis/go-redis/v9" ) func main() { @@ -35,7 +37,12 @@ func main() { os.Exit(1) } defer pool.Close() - handler := api.NewServer(identity.New(pool), analytics.New(pool, cfg.CursorSecret), logger, cfg.FrontendOrigin, cfg.CookieSecure).Handler() + redisClient := redis.NewClient(&redis.Options{Addr: cfg.RedisAddr, Password: cfg.RedisPassword}) + defer redisClient.Close() + repository := growthreport.NewRepository(pool) + coordinator := growthreport.NewCoordinator(redisClient, growthreport.CoordinatorConfig{CreatePerMinute: cfg.AICreatePerMinute, UncachedPerDay: cfg.AIUncachedScopesPerDay, DailyRequests: cfg.AIDailyRequestBudget, DailyTokens: cfg.AIDailyTokenBudget, MaxConcurrent: cfg.AIMaxConcurrent}) + growthService := growthreport.NewService(repository, coordinator, growthreport.ServiceConfig{Enabled: cfg.AIGrowthEnabled, Model: cfg.AvalAIModel, PromptVersion: "1.0.0", SchemaVersion: "1.0.0"}) + handler := api.NewServer(identity.New(pool), analytics.New(pool, cfg.CursorSecret), logger, cfg.FrontendOrigin, cfg.CookieSecure, growthService).Handler() server := &http.Server{Addr: cfg.HTTPAddr, Handler: handler, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 65 * time.Second, IdleTimeout: 60 * time.Second} go func() { <-ctx.Done() diff --git a/cmd/migrate/main.go b/cmd/migrate/main.go index affafb1..61420b3 100644 --- a/cmd/migrate/main.go +++ b/cmd/migrate/main.go @@ -25,14 +25,16 @@ func main() { os.Exit(1) } defer pool.Close() - migration, err := os.ReadFile("migrations/000001_mvp.sql") - if err != nil { - logger.Error("reading migration failed", "error", err) - os.Exit(1) + for _, path := range []string{"migrations/000001_mvp.sql", "migrations/000002_growth_analyst.sql"} { + migration, err := os.ReadFile(path) + if err != nil { + logger.Error("reading migration failed", "path", path, "error", err) + os.Exit(1) + } + if err := database.Migrate(ctx, pool, string(migration)); err != nil { + logger.Error("migration failed", "path", path, "error", err) + os.Exit(1) + } } - if err := database.Migrate(ctx, pool, string(migration)); err != nil { - logger.Error("migration failed", "error", err) - os.Exit(1) - } - logger.Info("migration complete", "version", 1) + logger.Info("migration complete", "version", 2) } diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 07d855f..28494fd 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -2,12 +2,16 @@ package main import ( "context" + "errors" "log/slog" + "net/http" "os" "os/signal" "time" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/config" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/database" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/growthreport" "github.com/redis/go-redis/v9" ) @@ -18,16 +22,49 @@ func main() { logger.Error("configuration failed", "error", err) os.Exit(1) } - ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) - defer cancel() - client := redis.NewClient(&redis.Options{Addr: cfg.RedisAddr, Password: cfg.RedisPassword}) - defer client.Close() - pingCtx, pingCancel := context.WithTimeout(ctx, 5*time.Second) - defer pingCancel() - if err := client.Ping(pingCtx).Err(); err != nil { - logger.Error("redis connection failed", "error", err) + if err := cfg.ValidateAPI(); err != nil { + logger.Error("worker configuration failed", "error", err) os.Exit(1) } - logger.Info("analytics worker ready", "redis_address", cfg.RedisAddr) - <-ctx.Done() + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) + defer cancel() + pool, err := database.Open(ctx, cfg.DatabaseURL) + if err != nil { + logger.Error("database connection failed", "error", err) + os.Exit(1) + } + defer pool.Close() + redisClient := redis.NewClient(&redis.Options{Addr: cfg.RedisAddr, Password: cfg.RedisPassword}) + defer redisClient.Close() + repository := growthreport.NewRepository(pool) + coordinator := growthreport.NewCoordinator(redisClient, growthreport.CoordinatorConfig{CreatePerMinute: cfg.AICreatePerMinute, UncachedPerDay: cfg.AIUncachedScopesPerDay, DailyRequests: cfg.AIDailyRequestBudget, DailyTokens: cfg.AIDailyTokenBudget, MaxConcurrent: cfg.AIMaxConcurrent}) + avalai := growthreport.NewAvalAIClient(growthreport.AvalAIConfig{BaseURL: cfg.AvalAIBaseURL, APIKey: cfg.AvalAIAPIKey, Model: cfg.AvalAIModel, Timeout: 30 * time.Second}, &http.Client{}) + processor := growthreport.NewProcessor(repository, growthreport.NewPostgreSQLBuilder(pool), avalai, coordinator, cfg.AIGrowthEnabled && cfg.AvalAIAPIKey != "") + logger.Info("growth worker ready", "ai_enabled", cfg.AIGrowthEnabled && cfg.AvalAIAPIKey != "") + poll := time.NewTicker(time.Second) + defer poll.Stop() + cleanup := time.NewTicker(24 * time.Hour) + defer cleanup.Stop() + for { + select { + case <-ctx.Done(): + return + case <-poll.C: + workCtx, workCancel := context.WithTimeout(ctx, 2*time.Minute) + err := processor.ProcessNext(workCtx) + workCancel() + if err != nil && !errors.Is(err, growthreport.ErrNotFound) { + logger.Error("growth report processing failed", "error", err) + } + case <-cleanup.C: + cleanupCtx, cleanupCancel := context.WithTimeout(ctx, 30*time.Second) + deleted, err := repository.CleanupSuperseded(cleanupCtx, time.Now().UTC().AddDate(0, 0, -90)) + cleanupCancel() + if err != nil { + logger.Error("growth cleanup failed", "error", err) + } else if deleted > 0 { + logger.Info("growth cleanup complete", "deleted_reports", deleted) + } + } + } }