diff --git a/cmd/api/main.go b/cmd/api/main.go index 2e71edb..230e1ab 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -15,6 +15,7 @@ import ( "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" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/insight" "github.com/redis/go-redis/v9" ) @@ -42,7 +43,9 @@ func main() { 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() + insightRepository := insight.NewRepository(pool) + insightService := insight.NewService(insightRepository, insight.NewPostgreSQLBuilder(pool)) + handler := api.NewServer(identity.New(pool), analytics.New(pool, cfg.CursorSecret), logger, cfg.FrontendOrigin, cfg.CookieSecure, growthService).WithInsights(insightService).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/calendar/main.go b/cmd/calendar/main.go new file mode 100644 index 0000000..573c1a0 --- /dev/null +++ b/cmd/calendar/main.go @@ -0,0 +1,92 @@ +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "log/slog" + "os" + "os/signal" + + "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/eventcatalog" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/insight" +) + +func main() { + action := flag.String("action", "", "import, approve, or activate") + filePath := flag.String("file", "", "reviewed JSON catalog for import") + version := flag.String("version", "", "catalog version for approve or activate") + actor := flag.String("actor", "", "audited reviewer or operator identity") + currentJalaliYear := flag.Int("current-jalali-year", 0, "current Jalali year used for coverage validation") + flag.Parse() + + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) + defer cancel() + cfg, err := config.Load() + if err != nil { + fail(logger, "configuration failed", err) + } + pool, err := database.Open(ctx, cfg.DatabaseURL) + if err != nil { + fail(logger, "database connection failed", err) + } + defer pool.Close() + repository := eventcatalog.NewRepository(pool) + + switch *action { + case "import": + if *filePath == "" || *currentJalaliYear == 0 { + fail(logger, "invalid import arguments", fmt.Errorf("file and current-jalali-year are required")) + } + file, err := os.Open(*filePath) + if err != nil { + fail(logger, "opening calendar catalog failed", err) + } + defer file.Close() + var input eventcatalog.Import + decoder := json.NewDecoder(file) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&input); err != nil { + fail(logger, "decoding calendar catalog failed", err) + } + if err := eventcatalog.ValidateImport(input, *currentJalaliYear); err != nil { + fail(logger, "calendar validation failed", err) + } + if err := repository.ImportDraft(ctx, input); err != nil { + fail(logger, "calendar import failed", err) + } + logger.Info("calendar draft imported", "version", input.Version, "events", len(input.Events)) + case "approve": + if *version == "" || *actor == "" { + fail(logger, "invalid approval arguments", fmt.Errorf("version and actor are required")) + } + if err := repository.Approve(ctx, *version, *actor); err != nil { + fail(logger, "calendar approval failed", err) + } + logger.Info("calendar approved", "version", *version, "reviewer", *actor) + case "activate": + if *version == "" || *actor == "" { + fail(logger, "invalid activation arguments", fmt.Errorf("version and actor are required")) + } + if err := repository.Activate(ctx, *version, *actor); err != nil { + fail(logger, "calendar activation failed", err) + } + service := insight.NewService(insight.NewRepository(pool), insight.NewPostgreSQLBuilder(pool)) + created, err := service.EnqueueAll(ctx, "CALENDAR") + if err != nil { + fail(logger, "calendar refresh scheduling failed", err) + } + logger.Info("calendar activated", "version", *version, "actor", *actor, "created_jobs", created) + default: + fail(logger, "invalid action", fmt.Errorf("action must be import, approve, or activate")) + } +} + +func fail(logger *slog.Logger, message string, err error) { + logger.Error(message, "error", err) + os.Exit(1) +} diff --git a/cmd/importer/main.go b/cmd/importer/main.go index b401564..9dda77e 100644 --- a/cmd/importer/main.go +++ b/cmd/importer/main.go @@ -9,6 +9,7 @@ import ( "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/importer" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/insight" ) func main() { @@ -33,4 +34,13 @@ func main() { } logger.Info("import complete", "noop", result.IsNoop, "source_sha256", result.SourceSHA256, "source_rows", result.SourceRows, "session_rows", result.SessionRows, "attempt_rows", result.AttemptRows) + if !result.IsNoop { + service := insight.NewService(insight.NewRepository(pool), insight.NewPostgreSQLBuilder(pool)) + created, err := service.EnqueueAll(ctx, "IMPORT") + if err != nil { + logger.Error("post-import insight scheduling failed", "error", err) + os.Exit(1) + } + logger.Info("post-import insight scheduling complete", "created_jobs", created) + } } diff --git a/cmd/migrate/main.go b/cmd/migrate/main.go index 977e41a..28855e1 100644 --- a/cmd/migrate/main.go +++ b/cmd/migrate/main.go @@ -25,7 +25,7 @@ func main() { os.Exit(1) } defer pool.Close() - for _, path := range []string{"migrations/000001_mvp.sql", "migrations/000002_growth_analyst.sql", "migrations/000003_ai_summary_refresh.sql"} { + for _, path := range []string{"migrations/000001_mvp.sql", "migrations/000002_growth_analyst.sql", "migrations/000003_ai_summary_refresh.sql", "migrations/000004_ai_page_insights.sql"} { migration, err := os.ReadFile(path) if err != nil { logger.Error("reading migration failed", "path", path, "error", err) @@ -36,5 +36,5 @@ func main() { os.Exit(1) } } - logger.Info("migration complete", "version", 3) + logger.Info("migration complete", "version", 4) } diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 28494fd..66d867c 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -12,6 +12,7 @@ import ( "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/insight" "github.com/redis/go-redis/v9" ) @@ -40,11 +41,19 @@ func main() { 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 != "") + insightService := insight.NewService(insight.NewRepository(pool), insight.NewPostgreSQLBuilder(pool)) 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() + nightly := time.NewTicker(time.Hour) + defer nightly.Stop() + if created, err := insightService.EnqueueAll(ctx, "NIGHTLY"); err != nil { + logger.Error("initial insight scheduling failed", "error", err) + } else { + logger.Info("initial insight scheduling complete", "created_jobs", created) + } for { select { case <-ctx.Done(): @@ -56,6 +65,12 @@ func main() { if err != nil && !errors.Is(err, growthreport.ErrNotFound) { logger.Error("growth report processing failed", "error", err) } + insightCtx, insightCancel := context.WithTimeout(ctx, 2*time.Minute) + err = insightService.ProcessNext(insightCtx) + insightCancel() + if err != nil && !errors.Is(err, insight.ErrNotFound) { + logger.Error("page insight 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)) @@ -65,6 +80,15 @@ func main() { } else if deleted > 0 { logger.Info("growth cleanup complete", "deleted_reports", deleted) } + case <-nightly.C: + scheduleCtx, scheduleCancel := context.WithTimeout(ctx, 5*time.Minute) + created, err := insightService.EnqueueAll(scheduleCtx, "NIGHTLY") + scheduleCancel() + if err != nil { + logger.Error("nightly insight scheduling failed", "error", err) + } else if created > 0 { + logger.Info("nightly insight scheduling complete", "created_jobs", created) + } } } } diff --git a/docs/openapi.yaml b/docs/openapi.yaml index cf64215..8174313 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -638,6 +638,67 @@ paths: '401': {$ref: '#/components/responses/Unauthenticated'} '404': {$ref: '#/components/responses/ResourceNotFound'} + /analytics/ai-insights/{module}: + get: + tags: [AI page insights] + operationId: getLatestAIInsight + summary: Return the latest persisted insight without waiting for generation + parameters: + - {name: module, in: path, required: true, schema: {$ref: '#/components/schemas/AIInsightModule'}} + - {name: terminalId, in: query, required: false, schema: {type: string, maxLength: 128}} + - $ref: '#/components/parameters/RequestIdHeader' + responses: + '200': {description: Latest ready or deterministic-partial snapshot., content: {application/json: {schema: {$ref: '#/components/schemas/AIInsight'}}}} + '401': {$ref: '#/components/responses/Unauthenticated'} + '404': {$ref: '#/components/responses/ResourceNotFound'} + '422': {$ref: '#/components/responses/ValidationFailed'} + '503': {$ref: '#/components/responses/AnalyticsUnavailable'} + /analytics/ai-insights/{module}/refreshes: + post: + tags: [AI page insights] + operationId: refreshAIInsight + summary: Enqueue an independent insight refresh + parameters: + - {name: module, in: path, required: true, schema: {$ref: '#/components/schemas/AIInsightModule'}} + - $ref: '#/components/parameters/IdempotencyKeyHeader' + - $ref: '#/components/parameters/CsrfTokenHeader' + - $ref: '#/components/parameters/OriginHeader' + requestBody: + required: true + content: {application/json: {schema: {$ref: '#/components/schemas/AIInsightRefreshRequest'}}} + responses: + '200': {description: Idempotent job reused., content: {application/json: {schema: {$ref: '#/components/schemas/AIInsightJob'}}}} + '202': {description: Refresh accepted., content: {application/json: {schema: {$ref: '#/components/schemas/AIInsightJob'}}}} + '401': {$ref: '#/components/responses/Unauthenticated'} + '403': {$ref: '#/components/responses/Forbidden'} + '404': {$ref: '#/components/responses/ResourceNotFound'} + '409': {$ref: '#/components/responses/Conflict'} + '422': {$ref: '#/components/responses/ValidationFailed'} + '429': {$ref: '#/components/responses/RateLimited'} + '503': {$ref: '#/components/responses/AnalyticsUnavailable'} + /analytics/ai-insight-refreshes/{jobId}: + get: + tags: [AI page insights] + operationId: getAIInsightRefresh + parameters: + - {name: jobId, in: path, required: true, schema: {type: string, pattern: '^insightjob_[a-f0-9]{24}$'}} + responses: + '200': {description: Tenant-scoped refresh job., content: {application/json: {schema: {$ref: '#/components/schemas/AIInsightJob'}}}} + '401': {$ref: '#/components/responses/Unauthenticated'} + '404': {$ref: '#/components/responses/ResourceNotFound'} + /analytics/ai-insight-refreshes/{jobId}/events: + get: + tags: [AI page insights] + operationId: streamAIInsightRefreshEvents + summary: Replay up to one hour of durable job events and continue as SSE + parameters: + - {name: jobId, in: path, required: true, schema: {type: string, pattern: '^insightjob_[a-f0-9]{24}$'}} + - {name: Last-Event-ID, in: header, required: false, schema: {type: integer, format: int64}} + responses: + '200': {description: Durable job events and heartbeats., content: {text/event-stream: {schema: {type: string}}}} + '401': {$ref: '#/components/responses/Unauthenticated'} + '404': {$ref: '#/components/responses/ResourceNotFound'} + components: securitySchemes: merchantSession: @@ -1174,6 +1235,83 @@ components: createdAt: {type: string, format: date-time} updatedAt: {type: string, format: date-time} eventsUrl: {type: string, format: uri-reference} + AIInsightModule: + type: string + enum: [overview, payments, fees, peers, transactions, growth] + AIInsightNarrative: + type: object + additionalProperties: false + required: [headline, body, evidenceIds] + properties: + headline: {type: string} + body: {type: string} + evidenceIds: {type: array, items: {type: string}} + AIInsightFinding: + type: object + additionalProperties: false + required: [evidenceId, title, summary, scoreBps, confidence] + properties: + evidenceId: {type: string} + title: {type: string} + summary: {type: string} + scoreBps: {type: integer, minimum: 0, maximum: 10000} + confidence: {type: string, enum: [HIGH, MEDIUM, LOW, INSUFFICIENT_DATA]} + AIInsightOption: + type: object + additionalProperties: false + required: [title, description, evidenceIds] + properties: + title: {type: string} + description: {type: string} + evidenceIds: {type: array, items: {type: string}} + AIInsightCopy: + type: object + additionalProperties: false + required: [past, latestObserved, future7Days, future30Days, findings, options] + properties: + past: {$ref: '#/components/schemas/AIInsightNarrative'} + latestObserved: {$ref: '#/components/schemas/AIInsightNarrative'} + future7Days: {$ref: '#/components/schemas/AIInsightNarrative'} + future30Days: {$ref: '#/components/schemas/AIInsightNarrative'} + future90Days: {$ref: '#/components/schemas/AIInsightNarrative'} + findings: {type: array, maxItems: 3, items: {$ref: '#/components/schemas/AIInsightFinding'}} + options: {type: array, items: {$ref: '#/components/schemas/AIInsightOption'}} + AIInsight: + type: object + additionalProperties: false + required: [insightId, module, revision, status, scope, asOfDate, dataThroughDate, forecastStartDate, freshness, dataLagDays, confidence, copy, evidence, calendarContext, refreshState, generatedAt] + properties: + insightId: {type: string} + module: {$ref: '#/components/schemas/AIInsightModule'} + revision: {type: integer, minimum: 1} + status: {type: string, enum: [READY, PARTIAL]} + scope: {type: object, properties: {terminalId: {type: [string, 'null']}}} + asOfDate: {type: string, format: date} + dataThroughDate: {type: string, format: date} + forecastStartDate: {type: string, format: date} + freshness: {type: string, enum: [FRESH, LAGGING, STALE]} + dataLagDays: {type: integer, minimum: 0} + confidence: {type: string, enum: [HIGH, MEDIUM, LOW, INSUFFICIENT_DATA]} + copy: {$ref: '#/components/schemas/AIInsightCopy'} + evidence: {type: object, additionalProperties: true} + calendarContext: {type: array, items: {type: object, additionalProperties: true}} + refreshState: {type: object, additionalProperties: true} + generatedAt: {type: string, format: date-time} + AIInsightRefreshRequest: + type: object + additionalProperties: false + properties: {terminalId: {type: string, maxLength: 128}} + AIInsightJob: + type: object + additionalProperties: false + required: [jobId, snapshotId, state, eventsUrl, createdAt, updatedAt] + properties: + jobId: {type: string} + snapshotId: {type: string} + state: {type: string, enum: [QUEUED, RUNNING, COMPLETED, FAILED]} + eventsUrl: {type: string, format: uri-reference} + createdAt: {type: string, format: date-time} + updatedAt: {type: string, format: date-time} LoginRequest: type: object additionalProperties: false diff --git a/internal/api/insights.go b/internal/api/insights.go new file mode 100644 index 0000000..f6c7468 --- /dev/null +++ b/internal/api/insights.go @@ -0,0 +1,185 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "regexp" + "strconv" + "time" + + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/insight" +) + +var insightJobIDPattern = regexp.MustCompile(`^insightjob_[a-f0-9]{24}$`) + +type insightService interface { + Latest(context.Context, int64, string, insight.Module) (insight.Snapshot, error) + Refresh(context.Context, int64, string, insight.Module, string) (insight.Job, bool, error) + GetJob(context.Context, int64, string) (insight.Job, error) + EventsAfter(context.Context, int64, string, int64) ([]insight.Event, error) +} + +func (s *Server) latestAIInsight(w http.ResponseWriter, r *http.Request) { + module, ok := insight.ParseModule(r.PathValue("module")) + if !ok || s.insights == nil { + s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") + return + } + terminalID, ok := insightTerminalQuery(r) + if !ok { + s.problem(w, r, 422, "VALIDATION_FAILED", "Only terminalId is accepted") + return + } + result, err := s.insights.Latest(r.Context(), sessionFrom(r.Context()).MerchantID, terminalID, module) + if errors.Is(err, insight.ErrNotFound) || errors.Is(err, insight.ErrUnknownTerminal) { + s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") + return + } + if err != nil { + s.logger.Error("loading AI insight failed", "error", err, "module", module) + s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable") + return + } + s.writeJSON(w, 200, result) +} + +func (s *Server) refreshAIInsight(w http.ResponseWriter, r *http.Request) { + module, ok := insight.ParseModule(r.PathValue("module")) + if !ok || s.insights == nil { + s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") + return + } + key := r.Header.Get("Idempotency-Key") + if !validIdempotencyKey(key) { + s.problem(w, r, 422, "VALIDATION_FAILED", "A valid Idempotency-Key is required") + return + } + var input struct { + TerminalID string `json:"terminalId"` + } + if err := decodeStrictJSON(w, r, &input); err != nil || len(input.TerminalID) > 128 { + s.problem(w, r, 422, "VALIDATION_FAILED", "Invalid insight refresh request") + return + } + job, cached, err := s.insights.Refresh(r.Context(), sessionFrom(r.Context()).MerchantID, input.TerminalID, module, key) + switch { + case errors.Is(err, insight.ErrUnknownTerminal): + s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") + return + case errors.Is(err, insight.ErrIdempotencyConflict): + s.problem(w, r, 409, "IDEMPOTENCY_CONFLICT", "Idempotency key was already used for another request") + return + case errors.Is(err, insight.ErrRefreshLimited): + next := nextTehranDay(time.Now()) + w.Header().Set("Retry-After", strconv.FormatInt(max(1, int64(time.Until(next).Seconds())), 10)) + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(429) + _ = json.NewEncoder(w).Encode(map[string]any{"type": "https://zarinpal.amiirkhl.ir/docs/errors/refresh-limited", "title": "Too Many Requests", "status": 429, "code": "REFRESH_LIMITED", "detail": "Daily refresh is not available yet", "instance": r.URL.Path, "requestId": w.Header().Get("X-Request-ID"), "nextRefreshAt": next}) + return + case err != nil: + s.logger.Error("creating AI insight refresh failed", "error", err) + s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable") + return + } + status := http.StatusAccepted + if cached { + status = http.StatusOK + } + w.Header().Set("Location", "/api/v1/analytics/ai-insight-refreshes/"+job.JobID) + s.writeJSON(w, status, job) +} + +func (s *Server) getAIInsightJob(w http.ResponseWriter, r *http.Request) { + if s.insights == nil || !insightJobIDPattern.MatchString(r.PathValue("jobId")) { + s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") + return + } + job, err := s.insights.GetJob(r.Context(), sessionFrom(r.Context()).MerchantID, r.PathValue("jobId")) + if errors.Is(err, insight.ErrNotFound) { + s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") + return + } + if err != nil { + s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable") + return + } + s.writeJSON(w, 200, job) +} + +func (s *Server) aiInsightEvents(w http.ResponseWriter, r *http.Request) { + if s.insights == nil || !insightJobIDPattern.MatchString(r.PathValue("jobId")) { + s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") + return + } + merchantID, jobID := sessionFrom(r.Context()).MerchantID, r.PathValue("jobId") + if _, err := s.insights.GetJob(r.Context(), merchantID, jobID); errors.Is(err, insight.ErrNotFound) { + s.problem(w, r, 404, "RESOURCE_NOT_FOUND", "Resource not found") + return + } else if err != nil { + s.problem(w, r, 503, "ANALYTICS_UNAVAILABLE", "AI insight unavailable") + return + } + flusher, ok := w.(http.Flusher) + if !ok { + s.problem(w, r, 500, "INTERNAL_ERROR", "Streaming unavailable") + return + } + w.Header().Set("Content-Type", "text/event-stream; charset=utf-8") + w.Header().Set("Cache-Control", "no-cache, no-transform") + w.Header().Set("X-Accel-Buffering", "no") + last, _ := strconv.ParseInt(r.Header.Get("Last-Event-ID"), 10, 64) + ctx, cancel := context.WithTimeout(r.Context(), time.Hour) + defer cancel() + poll, heartbeat := time.NewTicker(2*time.Second), time.NewTicker(15*time.Second) + defer poll.Stop() + defer heartbeat.Stop() + for { + events, err := s.insights.EventsAfter(ctx, merchantID, jobID, last) + if err != nil { + return + } + for _, event := range events { + payload, _ := json.Marshal(event) + _, _ = fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", event.ID, event.Type, payload) + last = event.ID + flusher.Flush() + } + job, err := s.insights.GetJob(ctx, merchantID, jobID) + if err != nil { + return + } + if job.State == "COMPLETED" || job.State == "FAILED" { + return + } + select { + case <-ctx.Done(): + return + case <-poll.C: + case <-heartbeat.C: + _, _ = fmt.Fprint(w, ": heartbeat\n\n") + flusher.Flush() + } + } +} + +func insightTerminalQuery(r *http.Request) (string, bool) { + for key := range r.URL.Query() { + if key != "terminalId" { + return "", false + } + } + value := r.URL.Query().Get("terminalId") + return value, len(value) <= 128 +} + +func nextTehranDay(now time.Time) time.Time { + location, err := time.LoadLocation("Asia/Tehran") + if err != nil { + location = time.FixedZone("Asia/Tehran", 3*60*60+30*60) + } + local := now.In(location) + return time.Date(local.Year(), local.Month(), local.Day()+1, 0, 0, 0, 0, location) +} diff --git a/internal/api/server.go b/internal/api/server.go index c065fbf..081f97d 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -19,6 +19,7 @@ import ( "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/analytics" "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/identity" + "git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/insight" ) const maxExportBytes = 50 << 20 @@ -37,6 +38,7 @@ type Server struct { identity *identity.Service analytics *analytics.Service growth growthReportService + insights insightService logger *slog.Logger frontendOrigin string cookieSecure bool @@ -45,6 +47,11 @@ type Server struct { growthSSEPollInterval time.Duration } +func (s *Server) WithInsights(service *insight.Service) *Server { + s.insights = service + return s +} + type sessionKey struct{} type tokenKey struct{} @@ -74,6 +81,10 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/v1/analytics/growth-reports/latest", s.authenticate(http.HandlerFunc(s.latestGrowthReport))) mux.Handle("GET /api/v1/analytics/growth-reports/{reportId}", s.authenticate(http.HandlerFunc(s.getGrowthReport))) mux.Handle("GET /api/v1/analytics/growth-reports/{reportId}/events", s.authenticate(http.HandlerFunc(s.growthReportEvents))) + mux.Handle("GET /api/v1/analytics/ai-insights/{module}", s.authenticate(http.HandlerFunc(s.latestAIInsight))) + mux.Handle("POST /api/v1/analytics/ai-insights/{module}/refreshes", s.authenticate(s.csrf(http.HandlerFunc(s.refreshAIInsight)))) + mux.Handle("GET /api/v1/analytics/ai-insight-refreshes/{jobId}", s.authenticate(http.HandlerFunc(s.getAIInsightJob))) + mux.Handle("GET /api/v1/analytics/ai-insight-refreshes/{jobId}/events", s.authenticate(http.HandlerFunc(s.aiInsightEvents))) return s.boundary(mux) } diff --git a/internal/eventcatalog/catalog.go b/internal/eventcatalog/catalog.go index dbecf49..4b06102 100644 --- a/internal/eventcatalog/catalog.go +++ b/internal/eventcatalog/catalog.go @@ -104,6 +104,18 @@ func ValidateEvents(events []Event) error { if event.LeadDays < 0 || event.LagDays < 0 || event.RelevanceBPS < 0 || event.RelevanceBPS > 10_000 || event.ConfidenceBPS < 0 || event.ConfidenceBPS > 10_000 { return fmt.Errorf("event catalog: event %q has invalid impact metadata", event.ID) } + if event.LeadDays > 90 || event.LagDays > 90 { + return fmt.Errorf("event catalog: event %q lead or lag window exceeds 90 days", event.ID) + } + if event.Recurrence != "" { + resolved, err := ResolveRecurrence(event.Recurrence, start.Year(), time.UTC) + if err != nil { + return fmt.Errorf("event catalog: event %q recurrence: %w", event.ID, err) + } + if !resolved.Equal(start) || !end.Equal(start) { + return fmt.Errorf("event catalog: event %q does not match its resolved recurrence", event.ID) + } + } } return nil } diff --git a/internal/eventcatalog/import.go b/internal/eventcatalog/import.go new file mode 100644 index 0000000..97df1a3 --- /dev/null +++ b/internal/eventcatalog/import.go @@ -0,0 +1,134 @@ +package eventcatalog + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type Import struct { + Version string `json:"version"` + CoverageStartJalaliYear int `json:"coverageStartJalaliYear"` + CoverageEndJalaliYear int `json:"coverageEndJalaliYear"` + SourceURI string `json:"sourceUri"` + SourceVersion string `json:"sourceVersion"` + SourceChecksum string `json:"sourceChecksum"` + ImportedBy string `json:"importedBy"` + Events []Event `json:"events"` +} + +func ValidateImport(input Import, currentJalaliYear int) error { + if strings.TrimSpace(input.Version) == "" || strings.TrimSpace(input.SourceURI) == "" || strings.TrimSpace(input.SourceVersion) == "" || strings.TrimSpace(input.SourceChecksum) == "" || strings.TrimSpace(input.ImportedBy) == "" { + return errors.New("event catalog: import metadata is incomplete") + } + if input.CoverageStartJalaliYear > currentJalaliYear-3 || input.CoverageEndJalaliYear < currentJalaliYear+2 { + return fmt.Errorf("event catalog: coverage must include Jalali years %d through %d", currentJalaliYear-3, currentJalaliYear+2) + } + if len(input.Events) == 0 { + return errors.New("event catalog: import contains no events") + } + return ValidateEvents(input.Events) +} + +func ValidateApproval(importedBy, reviewedBy string) error { + if strings.TrimSpace(reviewedBy) == "" { + return errors.New("event catalog: reviewer is required") + } + if strings.EqualFold(strings.TrimSpace(importedBy), strings.TrimSpace(reviewedBy)) { + return errors.New("event catalog: importer cannot approve the same catalog") + } + return nil +} + +type Repository struct{ pool *pgxpool.Pool } + +func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} } + +func (r *Repository) ImportDraft(ctx context.Context, input Import) error { + tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return fmt.Errorf("starting calendar import: %w", err) + } + defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }() + var versionID int64 + err = tx.QueryRow(ctx, `INSERT INTO calendar_catalog_versions(version,status,coverage_start_jalali_year,coverage_end_jalali_year,source_uri,source_version,source_checksum,imported_by) + VALUES($1,'DRAFT',$2,$3,$4,$5,$6,$7) RETURNING id`, input.Version, input.CoverageStartJalaliYear, input.CoverageEndJalaliYear, input.SourceURI, input.SourceVersion, input.SourceChecksum, input.ImportedBy).Scan(&versionID) + if err != nil { + return fmt.Errorf("creating calendar catalog version: %w", err) + } + for _, event := range input.Events { + categories, err := json.Marshal(event.MerchantCategories) + if err != nil { + return fmt.Errorf("encoding calendar event categories: %w", err) + } + channels, err := json.Marshal(event.BehavioralChannels) + if err != nil { + return fmt.Errorf("encoding calendar event channels: %w", err) + } + _, err = tx.Exec(ctx, `INSERT INTO calendar_events(catalog_version_id,event_key,title_fa,title_en,start_date,end_date,recurrence_rule,event_type,tone,commercial_posture,official_holiday,source_uri,source_version,source_checksum,inclusion_rationale,geographic_scope,merchant_categories,behavioral_channels,lead_days,lag_days,relevance_bps,confidence_bps) + VALUES($1,$2,$3,nullif($4,''),$5,$6,nullif($7,''),$8,$9,$10,$11,$12,$13,nullif($14,''),$15,$16,$17,$18,$19,$20,$21,$22)`, versionID, event.ID, event.TitleFA, event.TitleEN, event.StartDate, event.EndDate, event.Recurrence, event.Type, event.Tone, event.CommercialPosture, event.OfficialHoliday, event.SourceURI, event.SourceVersion, event.SourceChecksum, event.InclusionRationale, event.GeographicScope, categories, channels, event.LeadDays, event.LagDays, event.RelevanceBPS, event.ConfidenceBPS) + if err != nil { + return fmt.Errorf("inserting calendar event %q: %w", event.ID, err) + } + } + if _, err := tx.Exec(ctx, `INSERT INTO calendar_catalog_audit(catalog_version_id,action,actor) VALUES($1,'IMPORTED',$2)`, versionID, input.ImportedBy); err != nil { + return fmt.Errorf("auditing calendar import: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("committing calendar import: %w", err) + } + return nil +} + +func (r *Repository) Approve(ctx context.Context, version, reviewer string) error { + tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return fmt.Errorf("starting calendar approval: %w", err) + } + defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }() + var id int64 + var importedBy string + if err := tx.QueryRow(ctx, `SELECT id,imported_by FROM calendar_catalog_versions WHERE version=$1 AND status='DRAFT' FOR UPDATE`, version).Scan(&id, &importedBy); err != nil { + return fmt.Errorf("loading draft calendar catalog: %w", err) + } + if err := ValidateApproval(importedBy, reviewer); err != nil { + return err + } + if _, err := tx.Exec(ctx, `UPDATE calendar_catalog_versions SET status='APPROVED',reviewed_by=$2,reviewed_at=now() WHERE id=$1`, id, reviewer); err != nil { + return fmt.Errorf("approving calendar catalog: %w", err) + } + if _, err := tx.Exec(ctx, `INSERT INTO calendar_catalog_audit(catalog_version_id,action,actor) VALUES($1,'APPROVED',$2)`, id, reviewer); err != nil { + return fmt.Errorf("auditing calendar approval: %w", err) + } + return tx.Commit(ctx) +} + +func (r *Repository) Activate(ctx context.Context, version, actor string) error { + tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return fmt.Errorf("starting calendar activation: %w", err) + } + defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }() + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended('calendar-catalog-activation',0))`); err != nil { + return fmt.Errorf("locking calendar activation: %w", err) + } + var id int64 + if err := tx.QueryRow(ctx, `SELECT id FROM calendar_catalog_versions WHERE version=$1 AND status='APPROVED' FOR UPDATE`, version).Scan(&id); err != nil { + return fmt.Errorf("loading approved calendar catalog: %w", err) + } + if _, err := tx.Exec(ctx, `UPDATE calendar_catalog_versions SET status='RETIRED' WHERE status='ACTIVE'`); err != nil { + return fmt.Errorf("retiring active calendar catalog: %w", err) + } + if _, err := tx.Exec(ctx, `UPDATE calendar_catalog_versions SET status='ACTIVE',activated_at=now() WHERE id=$1`, id); err != nil { + return fmt.Errorf("activating calendar catalog: %w", err) + } + if _, err := tx.Exec(ctx, `INSERT INTO calendar_catalog_audit(catalog_version_id,action,actor) VALUES($1,'ACTIVATED',$2)`, id, actor); err != nil { + return fmt.Errorf("auditing calendar activation: %w", err) + } + return tx.Commit(ctx) +} diff --git a/internal/eventcatalog/import_test.go b/internal/eventcatalog/import_test.go new file mode 100644 index 0000000..780b3de --- /dev/null +++ b/internal/eventcatalog/import_test.go @@ -0,0 +1,28 @@ +package eventcatalog + +import "testing" + +func TestValidateImportRequiresThreePastAndTwoFutureJalaliYears(t *testing.T) { + t.Parallel() + + input := Import{Version: "1402-1407-v1", CoverageStartJalaliYear: 1402, CoverageEndJalaliYear: 1407, SourceURI: "https://calendar.ut.ac.ir/", SourceVersion: "reviewed", SourceChecksum: "abc", ImportedBy: "operator-a", Events: []Event{{ID: "nowruz", TitleFA: "نوروز", StartDate: "2026-03-21", EndDate: "2026-03-24", Type: TypeCultural, Tone: ToneCelebratory, CommercialPosture: PosturePromote, SourceURI: "https://calendar.ut.ac.ir/", SourceVersion: "1405", InclusionRationale: "تقویم رسمی", GeographicScope: "IR", RelevanceBPS: 10000, ConfidenceBPS: 10000}}} + if err := ValidateImport(input, 1405); err != nil { + t.Fatal(err) + } + + input.CoverageStartJalaliYear = 1403 + if err := ValidateImport(input, 1405); err == nil { + t.Fatal("expected insufficient past coverage") + } +} + +func TestValidateApprovalRequiresIndependentReviewer(t *testing.T) { + t.Parallel() + + if err := ValidateApproval("operator-a", "operator-a"); err == nil { + t.Fatal("expected independent reviewer error") + } + if err := ValidateApproval("operator-a", "operator-b"); err != nil { + t.Fatal(err) + } +} diff --git a/internal/eventcatalog/recurrence_validation_test.go b/internal/eventcatalog/recurrence_validation_test.go new file mode 100644 index 0000000..549debc --- /dev/null +++ b/internal/eventcatalog/recurrence_validation_test.go @@ -0,0 +1,14 @@ +package eventcatalog + +import "testing" + +func TestValidateEventsRejectsIncorrectBlackFridayDate(t *testing.T) { + event := Event{ID: "black-friday-2026", TitleFA: "جمعه سیاه", StartDate: "2026-11-26", EndDate: "2026-11-26", Recurrence: RecurrenceBlackFriday, Type: TypeCommercial, Tone: ToneNeutral, CommercialPosture: PostureCaution, SourceURI: "https://example.test/black-friday", SourceVersion: "2026", InclusionRationale: "رفتار تجاری مرتبط", GeographicScope: "IR", RelevanceBPS: 8000, ConfidenceBPS: 8000} + if err := ValidateEvents([]Event{event}); err == nil { + t.Fatal("expected recurrence mismatch") + } + event.StartDate, event.EndDate = "2026-11-27", "2026-11-27" + if err := ValidateEvents([]Event{event}); err != nil { + t.Fatal(err) + } +} diff --git a/internal/insight/builder.go b/internal/insight/builder.go new file mode 100644 index 0000000..1a2c0b9 --- /dev/null +++ b/internal/insight/builder.go @@ -0,0 +1,183 @@ +package insight + +import ( + "context" + "encoding/json" + "fmt" + "math" + "sort" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +const AnalyticsVersion = "2.0.0" + +type Analysis struct { + Time TimeContext + Confidence Confidence + Evidence json.RawMessage + Copy Copy + Calendar []CalendarContext +} + +type Builder interface { + Build(context.Context, int64, string, Module) (Analysis, error) +} + +type PostgreSQLBuilder struct { + pool *pgxpool.Pool + now func() time.Time +} + +func NewPostgreSQLBuilder(pool *pgxpool.Pool) *PostgreSQLBuilder { + return &PostgreSQLBuilder{pool: pool, now: time.Now} +} + +type aggregate struct { + Sessions, Paid, Failed, Amount, Fees int64 + AverageLatencyMS int64 +} + +func (b *PostgreSQLBuilder) Build(ctx context.Context, merchantID int64, terminalID string, module Module) (Analysis, error) { + if terminalID != "" { + var exists bool + if err := b.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM terminals WHERE merchant_id=$1 AND source_key=$2)`, merchantID, terminalID).Scan(&exists); err != nil { + return Analysis{}, fmt.Errorf("validating insight terminal: %w", err) + } + if !exists { + return Analysis{}, ErrUnknownTerminal + } + } + var dataThrough time.Time + if err := b.pool.QueryRow(ctx, `SELECT max(created_at AT TIME ZONE 'Asia/Tehran')::date FROM payment_sessions`).Scan(&dataThrough); err != nil { + return Analysis{}, fmt.Errorf("loading insight watermark: %w", err) + } + timeContext := BuildTimeContext(b.now(), dataThrough) + latest, err := b.aggregate(ctx, merchantID, terminalID, dataThrough.AddDate(0, 0, -29), dataThrough) + if err != nil { + return Analysis{}, err + } + previous, err := b.aggregate(ctx, merchantID, terminalID, dataThrough.AddDate(-1, 0, -29), dataThrough.AddDate(-1, 0, 0)) + if err != nil { + return Analysis{}, err + } + confidence := ConfidenceMedium + if timeContext.Freshness == FreshnessStale { + confidence = ConfidenceLow + } + comparisonAvailable := previous.Sessions > 0 + if !comparisonAvailable || latest.Sessions == 0 { + confidence = ConfidenceInsufficientData + } + evidenceMap := map[string]any{ + "schemaVersion": "1.0.0", "module": module, "comparisonAvailable": comparisonAvailable, + "latest30Days": latest, "previousYearEquivalent": previous, + "time": timeContext, "privacy": map[string]any{"aggregateOnly": true, "peerSuppressionApplied": module == ModulePeers}, + } + forecastStart, err := time.Parse(time.DateOnly, timeContext.ForecastStartDate) + if err != nil { + return Analysis{}, fmt.Errorf("parsing forecast start: %w", err) + } + calendar, err := b.calendar(ctx, forecastStart, horizon(module)) + if err != nil { + return Analysis{}, err + } + evidenceMap["calendarContext"] = calendar + encoded, err := json.Marshal(evidenceMap) + if err != nil { + return Analysis{}, fmt.Errorf("encoding insight evidence: %w", err) + } + copy := deterministicCopy(module, latest, previous, comparisonAvailable, timeContext, confidence) + return Analysis{Time: timeContext, Confidence: confidence, Evidence: encoded, Copy: copy, Calendar: calendar}, nil +} + +func (b *PostgreSQLBuilder) aggregate(ctx context.Context, merchantID int64, terminalID string, start, end time.Time) (aggregate, error) { + var value aggregate + err := b.pool.QueryRow(ctx, `SELECT count(*),count(*) FILTER(WHERE ps.status IN ('Paid','Verified')),count(*) FILTER(WHERE ps.status NOT IN ('Paid','Verified')), + coalesce(sum(ps.amount_rial) FILTER(WHERE ps.status IN ('Paid','Verified')),0),coalesce(sum(ps.fee_amount_rial),0), + coalesce(avg(coalesce(pa.verify_time_ms,pa.init_time_ms,0)),0)::bigint + FROM payment_sessions ps LEFT JOIN LATERAL (SELECT init_time_ms,verify_time_ms FROM payment_attempts WHERE session_id=ps.id ORDER BY sequence DESC LIMIT 1) pa ON true + WHERE ps.merchant_id=$1 AND ($2='' OR ps.terminal_id=(SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2)) + AND (ps.created_at AT TIME ZONE 'Asia/Tehran')::date BETWEEN $3 AND $4`, merchantID, terminalID, start, end). + Scan(&value.Sessions, &value.Paid, &value.Failed, &value.Amount, &value.Fees, &value.AverageLatencyMS) + if err != nil { + return aggregate{}, fmt.Errorf("loading aggregate insight evidence: %w", err) + } + return value, nil +} + +func (b *PostgreSQLBuilder) calendar(ctx context.Context, start time.Time, days int) ([]CalendarContext, error) { + rows, err := b.pool.Query(ctx, `SELECT e.event_key,e.title_fa,e.start_date,e.end_date,e.tone,e.commercial_posture,e.official_holiday + FROM calendar_events e JOIN calendar_catalog_versions v ON v.id=e.catalog_version_id AND v.status='ACTIVE' + WHERE e.end_date+e.lag_days >= $1 AND e.start_date-e.lead_days <= $2 ORDER BY e.start_date,e.event_key`, start, start.AddDate(0, 0, days-1)) + if err != nil { + return nil, fmt.Errorf("loading active calendar context: %w", err) + } + defer rows.Close() + var result []CalendarContext + for rows.Next() { + var item CalendarContext + var eventStart, eventEnd time.Time + if err := rows.Scan(&item.EventKey, &item.TitleFA, &eventStart, &eventEnd, &item.Tone, &item.CommercialPosture, &item.OfficialHoliday); err != nil { + return nil, fmt.Errorf("scanning calendar context: %w", err) + } + item.StartDate, item.EndDate = eventStart.Format(time.DateOnly), eventEnd.Format(time.DateOnly) + result = append(result, item) + } + return result, rows.Err() +} + +func horizon(module Module) int { + if module == ModuleGrowth { + return 90 + } + return 30 +} + +func deterministicCopy(module Module, latest, previous aggregate, comparison bool, tc TimeContext, confidence Confidence) Copy { + label := map[Module]string{ModuleOverview: "وضعیت کلی کسب‌وکار", ModulePayments: "عملکرد پرداخت", ModuleFees: "کارمزدها", ModulePeers: "مقایسه هم‌گروه", ModuleTransactions: "الگوی تراکنش‌ها", ModuleGrowth: "رشد"}[module] + change := 0.0 + if previous.Paid > 0 { + change = float64(latest.Paid-previous.Paid) / float64(previous.Paid) * 100 + } + pastBody := "برای مقایسه هم‌دوره سال قبل داده کافی وجود ندارد؛ نتیجه تاریخی ساخته نشده است." + if comparison { + pastBody = fmt.Sprintf("در ۳۰ روز کامل اخیر، تعداد پرداخت‌های موفق نسبت به بازه هم‌ارز سال قبل %.1f درصد تغییر کرده است.", change) + } + latestBody := fmt.Sprintf("آخرین وضعیت مشاهده‌شده تا %s شامل %d نشست و %d پرداخت موفق است. این تاریخ امروز نیست.", tc.DataThroughDate, latest.Sessions, latest.Paid) + uncertainty := "چون داده با تاریخ واقعی امروز فاصله دارد، دامنه عدم‌قطعیت گسترده و سطح اطمینان پایین نگه داشته شده است." + if tc.Freshness != FreshnessStale { + uncertainty = "برآورد با داده نسبتاً تازه ساخته شده، اما همچنان یک سناریوی تحلیلی است و قطعیت علّی ندارد." + } + findings := []Finding{ + {EvidenceID: "latest-volume", Title: "حجم مشاهده‌شده", Summary: latestBody, ScoreBPS: score(latest.Amount, confidence), Confidence: confidence}, + {EvidenceID: "payment-health", Title: "سلامت پرداخت", Summary: fmt.Sprintf("%d مورد ناموفق در برابر %d پرداخت موفق ثبت شده است.", latest.Failed, latest.Paid), ScoreBPS: score(latest.Failed, confidence), Confidence: confidence}, + } + if module == ModuleFees { + findings = append(findings, Finding{EvidenceID: "fee-driver", Title: "محرک کارمزد", Summary: fmt.Sprintf("کارمزد ثبت‌شده در ۳۰ روز آخر %d ریال است.", latest.Fees), ScoreBPS: score(latest.Fees, confidence), Confidence: confidence}) + } + sort.SliceStable(findings, func(i, j int) bool { return findings[i].ScoreBPS > findings[j].ScoreBPS }) + if len(findings) > 3 { + findings = findings[:3] + } + copy := Copy{ + Past: Narrative{Headline: "گذشتهٔ قابل مقایسه", Body: pastBody, EvidenceID: []string{"previous-year-equivalent"}}, + LatestObserved: Narrative{Headline: "آخرین وضعیت مشاهده‌شده", Body: latestBody, EvidenceID: []string{"latest-volume", "payment-health"}}, + Future7Days: Narrative{Headline: "چشم‌انداز ۷ روزه", Body: uncertainty + " سناریو از تاریخ واقعی امروز آغاز می‌شود.", EvidenceID: []string{"latest-volume"}}, + Future30Days: Narrative{Headline: "چشم‌انداز ۳۰ روزه", Body: uncertainty + " رویدادهای تقویمی فقط وقتی جهت مالی می‌گیرند که شواهد پذیرنده آن را پشتیبانی کند.", EvidenceID: []string{"latest-volume"}}, + Findings: findings, + Options: []Option{{Title: "بررسی جزئیات", Description: "قبل از اقدام، روند روزانه و خوشه‌های خطا را کنار این جمع‌بندی بررسی کنید.", EvidenceID: []string{"payment-health"}}}, + } + if module == ModuleGrowth { + copy.Future90Days = &Narrative{Headline: "چشم‌انداز ۹۰ روزه", Body: uncertainty + " هم‌پوشانی رویدادها مانع نسبت‌دادن اثر به یک علت واحد می‌شود.", EvidenceID: []string{"latest-volume"}} + } + copy.LatestObserved.Headline = label + " — آخرین وضعیت مشاهده‌شده" + return copy +} + +func score(impact int64, confidence Confidence) int { + base := int(math.Min(8000, math.Log10(float64(max(int64(1), impact))+1)*1200)) + weight := map[Confidence]int{ConfidenceHigh: 10000, ConfidenceMedium: 7500, ConfidenceLow: 4500, ConfidenceInsufficientData: 2000}[confidence] + return base * weight / 10000 +} diff --git a/internal/insight/model.go b/internal/insight/model.go new file mode 100644 index 0000000..9e8c1d2 --- /dev/null +++ b/internal/insight/model.go @@ -0,0 +1,145 @@ +package insight + +import ( + "encoding/json" + "errors" + "time" +) + +type Module string + +const ( + ModuleOverview Module = "overview" + ModulePayments Module = "payments" + ModuleFees Module = "fees" + ModulePeers Module = "peers" + ModuleTransactions Module = "transactions" + ModuleGrowth Module = "growth" +) + +var ErrNotFound = errors.New("ai insight: not found") +var ErrUnknownTerminal = errors.New("ai insight: unknown terminal") +var ErrRefreshLimited = errors.New("ai insight: refresh limited") +var ErrIdempotencyConflict = errors.New("ai insight: idempotency conflict") + +func ParseModule(value string) (Module, bool) { + module := Module(value) + switch module { + case ModuleOverview, ModulePayments, ModuleFees, ModulePeers, ModuleTransactions, ModuleGrowth: + return module, true + default: + return "", false + } +} + +type Confidence string + +const ( + ConfidenceHigh Confidence = "HIGH" + ConfidenceMedium Confidence = "MEDIUM" + ConfidenceLow Confidence = "LOW" + ConfidenceInsufficientData Confidence = "INSUFFICIENT_DATA" +) + +type Narrative struct { + Headline string `json:"headline"` + Body string `json:"body"` + EvidenceID []string `json:"evidenceIds"` +} + +type Finding struct { + EvidenceID string `json:"evidenceId"` + Title string `json:"title"` + Summary string `json:"summary"` + ScoreBPS int `json:"scoreBps"` + Confidence Confidence `json:"confidence"` +} + +type Option struct { + Title string `json:"title"` + Description string `json:"description"` + EvidenceID []string `json:"evidenceIds"` +} + +type CalendarContext struct { + EventKey string `json:"eventKey"` + TitleFA string `json:"titleFa"` + StartDate string `json:"startDate"` + EndDate string `json:"endDate"` + Tone string `json:"tone"` + CommercialPosture string `json:"commercialPosture"` + OfficialHoliday bool `json:"officialHoliday"` + EvidenceSupported bool `json:"evidenceSupported"` + OverlapConfounded bool `json:"overlapConfounded"` +} + +type Copy struct { + Past Narrative `json:"past"` + LatestObserved Narrative `json:"latestObserved"` + Future7Days Narrative `json:"future7Days"` + Future30Days Narrative `json:"future30Days"` + Future90Days *Narrative `json:"future90Days,omitempty"` + Findings []Finding `json:"findings"` + Options []Option `json:"options"` +} + +type Scope struct { + TerminalID *string `json:"terminalId"` +} + +type RefreshState struct { + State string `json:"state"` + JobID string `json:"jobId,omitempty"` + EventsURL string `json:"eventsUrl,omitempty"` + NextRefreshAt *time.Time `json:"nextRefreshAt,omitempty"` +} + +type Snapshot struct { + InsightID string `json:"insightId"` + Module Module `json:"module"` + Revision int `json:"revision"` + Status string `json:"status"` + Scope Scope `json:"scope"` + AsOfDate string `json:"asOfDate"` + DataThroughDate string `json:"dataThroughDate"` + ForecastStart string `json:"forecastStartDate"` + Freshness Freshness `json:"freshness"` + DataLagDays int `json:"dataLagDays"` + Confidence Confidence `json:"confidence"` + Copy Copy `json:"copy"` + Evidence json.RawMessage `json:"evidence"` + CalendarContext []CalendarContext `json:"calendarContext"` + RefreshState RefreshState `json:"refreshState"` + GeneratedAt time.Time `json:"generatedAt"` +} + +type Job struct { + JobID string `json:"jobId"` + SnapshotID string `json:"snapshotId"` + State string `json:"state"` + EventsURL string `json:"eventsUrl"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type Event struct { + ID int64 `json:"eventId"` + Type string `json:"type"` + Payload json.RawMessage `json:"payload"` + OccurredAt time.Time `json:"occurredAt"` +} + +type CreateRequest struct { + MerchantID int64 + TerminalID string + Module Module + TriggerReason string + ManualRefresh bool + IdempotencyKey string + RequestHash string + Fingerprint string + Time TimeContext + Confidence Confidence + CatalogVersion string + Evidence json.RawMessage +} diff --git a/internal/insight/model_test.go b/internal/insight/model_test.go new file mode 100644 index 0000000..c0e03fb --- /dev/null +++ b/internal/insight/model_test.go @@ -0,0 +1,22 @@ +package insight + +import "testing" + +func TestParseModuleRejectsUnscopedValues(t *testing.T) { + for _, value := range []string{"overview", "payments", "fees", "peers", "transactions", "growth"} { + if _, ok := ParseModule(value); !ok { + t.Fatalf("expected %s to be valid", value) + } + } + for _, value := range []string{"", "transaction-detail", "merchant_2", "OVERVIEW"} { + if _, ok := ParseModule(value); ok { + t.Fatalf("expected %s to be invalid", value) + } + } +} + +func TestStaleAlwaysCapsConfidence(t *testing.T) { + if got := confidenceFor(FreshnessStale); got != ConfidenceLow { + t.Fatalf("got %s", got) + } +} diff --git a/internal/insight/repository.go b/internal/insight/repository.go new file mode 100644 index 0000000..b107cca --- /dev/null +++ b/internal/insight/repository.go @@ -0,0 +1,328 @@ +package insight + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +const leaseDuration = 3 * time.Minute + +type Repository struct{ pool *pgxpool.Pool } + +type MerchantScope struct { + MerchantID int64 + TerminalID string +} + +func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} } + +func (r *Repository) MerchantScopes(ctx context.Context) ([]MerchantScope, error) { + rows, err := r.pool.Query(ctx, `SELECT m.id,NULL::text FROM merchants m UNION ALL SELECT t.merchant_id,t.source_key FROM terminals t ORDER BY 1,2 NULLS FIRST`) + if err != nil { + return nil, fmt.Errorf("loading insight scopes: %w", err) + } + defer rows.Close() + var result []MerchantScope + for rows.Next() { + var scope MerchantScope + var terminal *string + if err := rows.Scan(&scope.MerchantID, &terminal); err != nil { + return nil, fmt.Errorf("scanning insight scope: %w", err) + } + if terminal != nil { + scope.TerminalID = *terminal + } + result = append(result, scope) + } + return result, rows.Err() +} + +func (r *Repository) DatasetMetadata(ctx context.Context) (string, time.Time, error) { + var watermark string + var through time.Time + err := r.pool.QueryRow(ctx, `SELECT ir.source_sha256,max(ps.created_at AT TIME ZONE 'Asia/Tehran')::date + FROM payment_sessions ps JOIN import_runs ir ON ir.id=ps.import_run_id WHERE ir.status='COMPLETED' + GROUP BY ir.id,ir.source_sha256 ORDER BY ir.completed_at DESC LIMIT 1`).Scan(&watermark, &through) + if err != nil { + return "", time.Time{}, fmt.Errorf("loading insight dataset metadata: %w", err) + } + return watermark + ":" + through.Format(time.DateOnly), through, nil +} + +func (r *Repository) ActiveCatalogVersion(ctx context.Context) (string, error) { + var version string + err := r.pool.QueryRow(ctx, `SELECT version FROM calendar_catalog_versions WHERE status='ACTIVE'`).Scan(&version) + if errors.Is(err, pgx.ErrNoRows) { + return "calendar-unavailable", nil + } + if err != nil { + return "", fmt.Errorf("loading active calendar version: %w", err) + } + return version, nil +} + +func (r *Repository) Create(ctx context.Context, in CreateRequest) (Job, bool, error) { + tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable}) + if err != nil { + return Job{}, false, fmt.Errorf("beginning insight request: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + lockKey := fmt.Sprintf("ai-insight:%d:%s:%s", in.MerchantID, in.TerminalID, in.Module) + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, lockKey); err != nil { + return Job{}, false, fmt.Errorf("locking insight request: %w", err) + } + var terminalDBID *int64 + if in.TerminalID != "" { + var id int64 + err := tx.QueryRow(ctx, `SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2`, in.MerchantID, in.TerminalID).Scan(&id) + if errors.Is(err, pgx.ErrNoRows) { + return Job{}, false, ErrUnknownTerminal + } + if err != nil { + return Job{}, false, fmt.Errorf("validating insight terminal: %w", err) + } + terminalDBID = &id + } + if in.IdempotencyKey != "" { + var hash string + var jobID string + err := tx.QueryRow(ctx, `SELECT request_hash,id FROM ai_insight_jobs WHERE merchant_id=$1 AND idempotency_key=$2`, in.MerchantID, in.IdempotencyKey).Scan(&hash, &jobID) + if err == nil { + if hash != in.RequestHash { + return Job{}, false, ErrIdempotencyConflict + } + job, err := getJobWith(ctx, tx, in.MerchantID, jobID) + return job, true, err + } + if !errors.Is(err, pgx.ErrNoRows) { + return Job{}, false, fmt.Errorf("reading insight idempotency: %w", err) + } + } + if !in.ManualRefresh { + var jobID string + err := tx.QueryRow(ctx, `SELECT j.id FROM ai_insight_jobs j JOIN ai_insight_snapshots s ON s.id=j.snapshot_id + WHERE s.merchant_id=$1 AND s.terminal_source_key IS NOT DISTINCT FROM $2::text AND s.module=$3 AND s.fingerprint=$4 + AND s.manual_refresh=false AND s.status IN ('QUEUED','GENERATING','READY','PARTIAL') ORDER BY s.created_at DESC LIMIT 1`, in.MerchantID, nilIfEmpty(in.TerminalID), in.Module, in.Fingerprint).Scan(&jobID) + if err == nil { + job, err := getJobWith(ctx, tx, in.MerchantID, jobID) + return job, true, err + } + if !errors.Is(err, pgx.ErrNoRows) { + return Job{}, false, fmt.Errorf("checking duplicate insight: %w", err) + } + } + if in.ManualRefresh { + var used bool + if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM ai_insight_snapshots WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND module=$3 AND manual_refresh=true AND tehran_refresh_day=$4)`, in.MerchantID, nilIfEmpty(in.TerminalID), in.Module, in.Time.AsOfDate).Scan(&used); err != nil { + return Job{}, false, fmt.Errorf("checking daily insight refresh: %w", err) + } + if used { + return Job{}, false, ErrRefreshLimited + } + } + var revision int + if err := tx.QueryRow(ctx, `SELECT coalesce(max(revision),0)+1 FROM ai_insight_snapshots WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND module=$3`, in.MerchantID, nilIfEmpty(in.TerminalID), in.Module).Scan(&revision); err != nil { + return Job{}, false, fmt.Errorf("calculating insight revision: %w", err) + } + snapshotID, jobID := newID("insight_"), newID("insightjob_") + _, err = tx.Exec(ctx, `INSERT INTO ai_insight_snapshots(id,merchant_id,terminal_id,terminal_source_key,module,revision,fingerprint,status,as_of_date,data_through_date,freshness,data_lag_days,confidence,catalog_version,analytics_version,prompt_version,schema_version,model,deterministic_evidence,manual_refresh,tehran_refresh_day) + VALUES($1,$2,$3,$4,$5,$6,$7,'QUEUED',$8,$9,$10,$11,$12,$13,$14,'1.0.0','1.0.0','deterministic-fallback',$15,$16,$17)`, snapshotID, in.MerchantID, terminalDBID, nilIfEmpty(in.TerminalID), in.Module, revision, in.Fingerprint, in.Time.AsOfDate, in.Time.DataThroughDate, in.Time.Freshness, in.Time.DataLagDays, in.Confidence, in.CatalogVersion, AnalyticsVersion, in.Evidence, in.ManualRefresh, nullableDate(in.ManualRefresh, in.Time.AsOfDate)) + if err != nil { + return Job{}, false, fmt.Errorf("creating insight snapshot: %w", err) + } + _, err = tx.Exec(ctx, `INSERT INTO ai_insight_jobs(id,snapshot_id,merchant_id,module,trigger_reason,status,idempotency_key,request_hash) VALUES($1,$2,$3,$4,$5,'QUEUED',$6,$7)`, jobID, snapshotID, in.MerchantID, in.Module, in.TriggerReason, nilIfEmpty(in.IdempotencyKey), nilIfEmpty(in.RequestHash)) + if err != nil { + return Job{}, false, fmt.Errorf("creating insight job: %w", err) + } + if err := insertEvent(ctx, tx, jobID, in.MerchantID, "job.queued", map[string]any{"state": "QUEUED", "snapshotId": snapshotID}); err != nil { + return Job{}, false, err + } + job, err := getJobWith(ctx, tx, in.MerchantID, jobID) + if err != nil { + return Job{}, false, err + } + if err := tx.Commit(ctx); err != nil { + return Job{}, false, fmt.Errorf("committing insight request: %w", err) + } + return job, false, nil +} + +type claimedJob struct { + Job + MerchantID int64 + TerminalID string + Module Module + LeaseToken string +} + +func (r *Repository) Claim(ctx context.Context) (claimedJob, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return claimedJob{}, fmt.Errorf("beginning insight claim: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + if _, err := tx.Exec(ctx, `UPDATE ai_insight_jobs SET status='QUEUED',lease_token=NULL,lease_expires_at=NULL,updated_at=now() WHERE status='RUNNING' AND lease_expires_at$3),updated_at=now() WHERE id=$1`, job.JobID, job.LeaseToken, int(leaseDuration/time.Second)); err != nil { + return claimedJob{}, fmt.Errorf("claiming insight job: %w", err) + } + if _, err := tx.Exec(ctx, `UPDATE ai_insight_snapshots SET status='GENERATING',updated_at=now() WHERE id=$1`, job.SnapshotID); err != nil { + return claimedJob{}, fmt.Errorf("marking insight generating: %w", err) + } + if err := insertEvent(ctx, tx, job.JobID, job.MerchantID, "job.running", map[string]any{"state": "RUNNING"}); err != nil { + return claimedJob{}, err + } + if err := tx.Commit(ctx); err != nil { + return claimedJob{}, fmt.Errorf("committing insight claim: %w", err) + } + return job, nil +} + +func (r *Repository) Complete(ctx context.Context, job claimedJob, analysis Analysis, status, failureCode string) error { + tx, err := r.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("beginning insight completion: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + copyJSON, _ := json.Marshal(analysis.Copy) + _, err = tx.Exec(ctx, `UPDATE ai_insight_snapshots SET status=$3,as_of_date=$4,data_through_date=$5,freshness=$6,data_lag_days=$7,confidence=$8,deterministic_evidence=$9,ai_response=$10,failure_code=$11,updated_at=now(),completed_at=now() WHERE id=$1 AND EXISTS(SELECT 1 FROM ai_insight_jobs WHERE id=$2 AND lease_token=$12)`, job.SnapshotID, job.JobID, status, analysis.Time.AsOfDate, analysis.Time.DataThroughDate, analysis.Time.Freshness, analysis.Time.DataLagDays, analysis.Confidence, analysis.Evidence, copyJSON, nilIfEmpty(failureCode), job.LeaseToken) + if err != nil { + return fmt.Errorf("completing insight snapshot: %w", err) + } + _, err = tx.Exec(ctx, `UPDATE ai_insight_jobs SET status='COMPLETED',lease_token=NULL,lease_expires_at=NULL,updated_at=now(),completed_at=now() WHERE id=$1 AND lease_token=$2`, job.JobID, job.LeaseToken) + if err != nil { + return fmt.Errorf("completing insight job: %w", err) + } + if err := insertEvent(ctx, tx, job.JobID, job.MerchantID, "job.completed", map[string]any{"state": "COMPLETED", "snapshotStatus": status}); err != nil { + return err + } + return tx.Commit(ctx) +} + +func (r *Repository) Latest(ctx context.Context, merchantID int64, terminalID string, module Module) (Snapshot, error) { + row := r.pool.QueryRow(ctx, snapshotSelect+` WHERE s.merchant_id=$1 AND s.terminal_source_key IS NOT DISTINCT FROM $2::text AND s.module=$3 AND s.status IN ('READY','PARTIAL') ORDER BY s.revision DESC LIMIT 1`, merchantID, nilIfEmpty(terminalID), module) + result, err := scanSnapshot(row) + if err != nil { + return Snapshot{}, err + } + var jobID, state *string + _ = r.pool.QueryRow(ctx, `SELECT j.id,j.status FROM ai_insight_jobs j JOIN ai_insight_snapshots s ON s.id=j.snapshot_id WHERE s.merchant_id=$1 AND s.terminal_source_key IS NOT DISTINCT FROM $2::text AND s.module=$3 AND j.status IN ('QUEUED','RUNNING') ORDER BY j.created_at DESC LIMIT 1`, merchantID, nilIfEmpty(terminalID), module).Scan(&jobID, &state) + if jobID != nil { + result.RefreshState = RefreshState{State: *state, JobID: *jobID, EventsURL: "/api/v1/analytics/ai-insight-refreshes/" + *jobID + "/events"} + } else { + result.RefreshState.State = "IDLE" + } + return result, nil +} + +func (r *Repository) GetJob(ctx context.Context, merchantID int64, jobID string) (Job, error) { + return getJobWith(ctx, r.pool, merchantID, jobID) +} + +func (r *Repository) EventsAfter(ctx context.Context, merchantID int64, jobID string, after int64) ([]Event, error) { + if _, err := r.GetJob(ctx, merchantID, jobID); err != nil { + return nil, err + } + rows, err := r.pool.Query(ctx, `SELECT id,event_type,payload,occurred_at FROM ai_insight_events WHERE merchant_id=$1 AND job_id=$2 AND id>$3 AND occurred_at>=now()-interval '1 hour' ORDER BY id LIMIT 100`, merchantID, jobID, after) + if err != nil { + return nil, fmt.Errorf("loading insight events: %w", err) + } + defer rows.Close() + var events []Event + for rows.Next() { + var event Event + if err := rows.Scan(&event.ID, &event.Type, &event.Payload, &event.OccurredAt); err != nil { + return nil, fmt.Errorf("scanning insight event: %w", err) + } + events = append(events, event) + } + return events, rows.Err() +} + +const snapshotSelect = `SELECT s.id,s.module,s.revision,s.status,s.terminal_source_key,s.as_of_date,s.data_through_date,s.freshness,s.data_lag_days,s.confidence,s.ai_response,s.deterministic_evidence,s.created_at FROM ai_insight_snapshots s` + +type rowScanner interface{ Scan(...any) error } +type queryer interface { + QueryRow(context.Context, string, ...any) pgx.Row +} + +func scanSnapshot(row rowScanner) (Snapshot, error) { + var s Snapshot + var terminal *string + var asOf, through time.Time + var copyJSON []byte + err := row.Scan(&s.InsightID, &s.Module, &s.Revision, &s.Status, &terminal, &asOf, &through, &s.Freshness, &s.DataLagDays, &s.Confidence, ©JSON, &s.Evidence, &s.GeneratedAt) + if errors.Is(err, pgx.ErrNoRows) { + return Snapshot{}, ErrNotFound + } + if err != nil { + return Snapshot{}, fmt.Errorf("scanning insight snapshot: %w", err) + } + if terminal != nil { + s.Scope.TerminalID = terminal + } + s.AsOfDate, s.ForecastStart = asOf.Format(time.DateOnly), asOf.Format(time.DateOnly) + s.DataThroughDate = through.Format(time.DateOnly) + _ = json.Unmarshal(copyJSON, &s.Copy) + return s, nil +} + +func getJobWith(ctx context.Context, q queryer, merchantID int64, jobID string) (Job, error) { + var job Job + err := q.QueryRow(ctx, `SELECT id,snapshot_id,status,created_at,updated_at FROM ai_insight_jobs WHERE merchant_id=$1 AND id=$2`, merchantID, jobID).Scan(&job.JobID, &job.SnapshotID, &job.State, &job.CreatedAt, &job.UpdatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return Job{}, ErrNotFound + } + if err != nil { + return Job{}, fmt.Errorf("loading insight job: %w", err) + } + job.EventsURL = "/api/v1/analytics/ai-insight-refreshes/" + job.JobID + "/events" + return job, nil +} + +func insertEvent(ctx context.Context, tx pgx.Tx, jobID string, merchantID int64, eventType string, payload any) error { + encoded, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("encoding insight event: %w", err) + } + if _, err := tx.Exec(ctx, `INSERT INTO ai_insight_events(job_id,merchant_id,event_type,payload) VALUES($1,$2,$3,$4)`, jobID, merchantID, eventType, encoded); err != nil { + return fmt.Errorf("storing insight event: %w", err) + } + return nil +} + +func nilIfEmpty(value string) any { + if value == "" { + return nil + } + return value +} + +func nullableDate(enabled bool, value string) any { + if !enabled { + return nil + } + return value +} diff --git a/internal/insight/service.go b/internal/insight/service.go new file mode 100644 index 0000000..2bef805 --- /dev/null +++ b/internal/insight/service.go @@ -0,0 +1,114 @@ +package insight + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "fmt" + "time" +) + +type Service struct { + repository *Repository + builder Builder + now func() time.Time +} + +func NewService(repository *Repository, builder Builder) *Service { + return &Service{repository: repository, builder: builder, now: time.Now} +} + +func (s *Service) Latest(ctx context.Context, merchantID int64, terminalID string, module Module) (Snapshot, error) { + if terminalID != "" { + // Validation is deliberately tenant-scoped and returns the same public result as a missing terminal. + if _, err := s.builder.Build(ctx, merchantID, terminalID, module); err != nil { + return Snapshot{}, err + } + } + return s.repository.Latest(ctx, merchantID, terminalID, module) +} + +func (s *Service) Refresh(ctx context.Context, merchantID int64, terminalID string, module Module, idempotencyKey string) (Job, bool, error) { + return s.enqueue(ctx, merchantID, terminalID, module, "MANUAL", true, idempotencyKey) +} + +func (s *Service) Enqueue(ctx context.Context, merchantID int64, terminalID string, module Module, reason string) (Job, bool, error) { + return s.enqueue(ctx, merchantID, terminalID, module, reason, false, "") +} + +func (s *Service) EnqueueAll(ctx context.Context, reason string) (int, error) { + scopes, err := s.repository.MerchantScopes(ctx) + if err != nil { + return 0, err + } + modules := []Module{ModuleOverview, ModulePayments, ModuleFees, ModulePeers, ModuleTransactions, ModuleGrowth} + created := 0 + for _, scope := range scopes { + for _, module := range modules { + if _, cached, err := s.Enqueue(ctx, scope.MerchantID, scope.TerminalID, module, reason); err != nil { + return created, err + } else if !cached { + created++ + } + } + } + return created, nil +} + +func (s *Service) enqueue(ctx context.Context, merchantID int64, terminalID string, module Module, reason string, manual bool, key string) (Job, bool, error) { + watermark, dataThrough, err := s.repository.DatasetMetadata(ctx) + if err != nil { + return Job{}, false, err + } + catalog, err := s.repository.ActiveCatalogVersion(ctx) + if err != nil { + return Job{}, false, err + } + tc := BuildTimeContext(s.now(), dataThrough) + fingerprint := hash(fmt.Sprintf("%d|%s|%s|%s|%s|%s|%s|%s", merchantID, terminalID, module, tc.AsOfDate, watermark, catalog, AnalyticsVersion, "1.0.0")) + requestHash := hash(fmt.Sprintf("%d|%s|%s|%t", merchantID, terminalID, module, manual)) + return s.repository.Create(ctx, CreateRequest{MerchantID: merchantID, TerminalID: terminalID, Module: module, TriggerReason: reason, ManualRefresh: manual, IdempotencyKey: key, RequestHash: requestHash, Fingerprint: fingerprint, Time: tc, Confidence: confidenceFor(tc.Freshness), CatalogVersion: catalog, Evidence: []byte(`{}`)}) +} + +func (s *Service) GetJob(ctx context.Context, merchantID int64, jobID string) (Job, error) { + return s.repository.GetJob(ctx, merchantID, jobID) +} + +func (s *Service) EventsAfter(ctx context.Context, merchantID int64, jobID string, after int64) ([]Event, error) { + return s.repository.EventsAfter(ctx, merchantID, jobID, after) +} + +func (s *Service) ProcessNext(ctx context.Context) error { + job, err := s.repository.Claim(ctx) + if err != nil { + return err + } + analysis, err := s.builder.Build(ctx, job.MerchantID, job.TerminalID, job.Module) + if err != nil { + return fmt.Errorf("building %s insight: %w", job.Module, err) + } + // The deterministic copy is always publishable. A provider can replace it only after + // evidence-reference validation; until then PARTIAL is explicit and last-good remains visible. + return s.repository.Complete(ctx, job, analysis, "PARTIAL", "DETERMINISTIC_FALLBACK") +} + +func confidenceFor(f Freshness) Confidence { + if f == FreshnessStale { + return ConfidenceLow + } + return ConfidenceMedium +} + +func hash(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:]) +} + +func newID(prefix string) string { + value := make([]byte, 12) + if _, err := rand.Read(value); err != nil { + panic("crypto/rand unavailable") + } + return prefix + hex.EncodeToString(value) +} diff --git a/migrations/000004_ai_page_insights.sql b/migrations/000004_ai_page_insights.sql new file mode 100644 index 0000000..56574b8 --- /dev/null +++ b/migrations/000004_ai_page_insights.sql @@ -0,0 +1,141 @@ +CREATE TABLE IF NOT EXISTS calendar_catalog_versions ( + id bigserial PRIMARY KEY, + version text NOT NULL UNIQUE, + status text NOT NULL CHECK (status IN ('DRAFT','APPROVED','ACTIVE','RETIRED')), + coverage_start_jalali_year integer NOT NULL, + coverage_end_jalali_year integer NOT NULL, + source_uri text NOT NULL, + source_version text NOT NULL, + source_checksum text NOT NULL, + imported_by text NOT NULL, + reviewed_by text, + imported_at timestamptz NOT NULL DEFAULT now(), + reviewed_at timestamptz, + activated_at timestamptz, + CHECK (coverage_end_jalali_year >= coverage_start_jalali_year) +); +-- statement-break +CREATE UNIQUE INDEX IF NOT EXISTS calendar_catalog_one_active_idx + ON calendar_catalog_versions ((status)) WHERE status='ACTIVE'; +-- statement-break +CREATE TABLE IF NOT EXISTS calendar_events ( + id bigserial PRIMARY KEY, + catalog_version_id bigint NOT NULL REFERENCES calendar_catalog_versions(id) ON DELETE CASCADE, + event_key text NOT NULL, + title_fa text NOT NULL, + title_en text, + start_date date NOT NULL, + end_date date NOT NULL, + recurrence_rule text, + event_type text NOT NULL CHECK (event_type IN ('PUBLIC_HOLIDAY','RELIGIOUS','NATIONAL','CULTURAL','COMMERCIAL','SEASONAL','EDUCATION','ECONOMIC','EMERGENCY_CLOSURE')), + tone text NOT NULL CHECK (tone IN ('CELEBRATORY','JOYFUL','NEUTRAL','SOLEMN','MOURNING','MEMORIAL')), + commercial_posture text NOT NULL CHECK (commercial_posture IN ('PROMOTE','CAUTION','NO_PROMOTION')), + official_holiday boolean NOT NULL DEFAULT false, + source_uri text NOT NULL, + source_version text NOT NULL, + source_checksum text, + inclusion_rationale text NOT NULL, + geographic_scope text NOT NULL DEFAULT 'IR', + merchant_categories jsonb NOT NULL DEFAULT '[]'::jsonb, + behavioral_channels jsonb NOT NULL DEFAULT '[]'::jsonb, + lead_days integer NOT NULL DEFAULT 0 CHECK (lead_days BETWEEN 0 AND 90), + lag_days integer NOT NULL DEFAULT 0 CHECK (lag_days BETWEEN 0 AND 90), + relevance_bps integer NOT NULL CHECK (relevance_bps BETWEEN 0 AND 10000), + confidence_bps integer NOT NULL CHECK (confidence_bps BETWEEN 0 AND 10000), + created_at timestamptz NOT NULL DEFAULT now(), + CHECK (end_date >= start_date), + CHECK (tone NOT IN ('MOURNING','MEMORIAL') OR commercial_posture <> 'PROMOTE'), + UNIQUE (catalog_version_id,event_key) +); +-- statement-break +CREATE INDEX IF NOT EXISTS calendar_events_active_range_idx + ON calendar_events (start_date,end_date,catalog_version_id); +-- statement-break +CREATE TABLE IF NOT EXISTS calendar_catalog_audit ( + id bigserial PRIMARY KEY, + catalog_version_id bigint NOT NULL REFERENCES calendar_catalog_versions(id) ON DELETE CASCADE, + action text NOT NULL CHECK (action IN ('IMPORTED','APPROVED','ACTIVATED','RETIRED')), + actor text NOT NULL, + occurred_at timestamptz NOT NULL DEFAULT now() +); +-- statement-break +CREATE TABLE IF NOT EXISTS ai_insight_snapshots ( + id text PRIMARY KEY, + merchant_id bigint NOT NULL REFERENCES merchants(id), + terminal_id bigint REFERENCES terminals(id), + terminal_source_key text, + module text NOT NULL CHECK (module IN ('overview','payments','fees','peers','transactions','growth')), + revision integer NOT NULL CHECK (revision > 0), + fingerprint text NOT NULL, + status text NOT NULL CHECK (status IN ('QUEUED','GENERATING','READY','PARTIAL','FAILED')), + as_of_date date NOT NULL, + data_through_date date NOT NULL, + freshness text NOT NULL CHECK (freshness IN ('FRESH','LAGGING','STALE')), + data_lag_days integer NOT NULL CHECK (data_lag_days >= 0), + confidence text NOT NULL CHECK (confidence IN ('HIGH','MEDIUM','LOW','INSUFFICIENT_DATA')), + catalog_version text NOT NULL, + analytics_version text NOT NULL, + prompt_version text NOT NULL, + schema_version text NOT NULL, + model text NOT NULL, + deterministic_evidence jsonb NOT NULL, + ai_response jsonb, + failure_code text, + input_tokens bigint NOT NULL DEFAULT 0 CHECK (input_tokens >= 0), + output_tokens bigint NOT NULL DEFAULT 0 CHECK (output_tokens >= 0), + provider_requests bigint NOT NULL DEFAULT 0 CHECK (provider_requests >= 0), + latency_ms bigint NOT NULL DEFAULT 0 CHECK (latency_ms >= 0), + estimated_cost text, + manual_refresh boolean NOT NULL DEFAULT false, + tehran_refresh_day date, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + completed_at timestamptz +); +-- statement-break +CREATE UNIQUE INDEX IF NOT EXISTS ai_insight_current_fingerprint_idx + ON ai_insight_snapshots (merchant_id,coalesce(terminal_source_key,''),module,fingerprint) + WHERE manual_refresh=false AND status IN ('QUEUED','GENERATING','READY','PARTIAL'); +-- statement-break +CREATE UNIQUE INDEX IF NOT EXISTS ai_insight_manual_refresh_idx + ON ai_insight_snapshots (merchant_id,coalesce(terminal_source_key,''),module,tehran_refresh_day) + WHERE manual_refresh=true; +-- statement-break +CREATE INDEX IF NOT EXISTS ai_insight_latest_scope_idx + ON ai_insight_snapshots (merchant_id,terminal_source_key,module,revision DESC,created_at DESC); +-- statement-break +CREATE TABLE IF NOT EXISTS ai_insight_jobs ( + id text PRIMARY KEY, + snapshot_id text NOT NULL REFERENCES ai_insight_snapshots(id) ON DELETE CASCADE, + merchant_id bigint NOT NULL REFERENCES merchants(id), + module text NOT NULL, + trigger_reason text NOT NULL CHECK (trigger_reason IN ('BACKFILL','NIGHTLY','IMPORT','CALENDAR','ANOMALY','MANUAL')), + status text NOT NULL CHECK (status IN ('QUEUED','RUNNING','COMPLETED','FAILED')), + idempotency_key text, + request_hash text, + attempts integer NOT NULL DEFAULT 0 CHECK (attempts >= 0), + scheduled_at timestamptz NOT NULL DEFAULT now(), + lease_token text, + lease_expires_at timestamptz, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + completed_at timestamptz, + UNIQUE (merchant_id,idempotency_key) +); +-- statement-break +CREATE INDEX IF NOT EXISTS ai_insight_jobs_pending_idx + ON ai_insight_jobs (scheduled_at,created_at) WHERE status IN ('QUEUED','RUNNING'); +-- statement-break +CREATE TABLE IF NOT EXISTS ai_insight_events ( + id bigserial PRIMARY KEY, + job_id text NOT NULL REFERENCES ai_insight_jobs(id) ON DELETE CASCADE, + merchant_id bigint NOT NULL REFERENCES merchants(id), + event_type text NOT NULL, + payload jsonb NOT NULL, + occurred_at timestamptz NOT NULL DEFAULT now() +); +-- statement-break +CREATE INDEX IF NOT EXISTS ai_insight_events_replay_idx + ON ai_insight_events (job_id,id) INCLUDE (occurred_at); +-- statement-break +INSERT INTO schema_migrations(version) VALUES (4) ON CONFLICT DO NOTHING;