zarinpal-analytics-backend/internal/growthreport/orchestrator.go

package growthreport

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"git.amiirkhl.ir/starcoach-hackathon/zarinpal-analytics-backend/internal/growth"
)

var ErrDisabled = errors.New("growth report: disabled")

const MaxProviderRequests int64 = 4

type ServiceConfig struct {
	Enabled                             bool
	Model, PromptVersion, SchemaVersion string
}
type Service struct {
	repository  *Repository
	coordinator *Coordinator
	config      ServiceConfig
	now         func() time.Time
}

func NewService(repository *Repository, coordinator *Coordinator, config ServiceConfig) *Service {
	return &Service{repository: repository, coordinator: coordinator, config: config, now: time.Now}
}

func (s *Service) Create(ctx context.Context, userID, merchantID int64, request CreateRequest, idempotencyKey string) (Record, bool, error) {
	if !s.config.Enabled {
		return Record{}, false, ErrDisabled
	}
	now := s.now().UTC()
	if err := s.coordinator.AllowCreate(ctx, userID, now); err != nil {
		return Record{}, false, err
	}
	watermark, start, err := s.repository.DatasetMetadata(ctx)
	if err != nil {
		return Record{}, false, err
	}
	fingerprint := s.fingerprint(merchantID, request.TerminalID, watermark, start)
	var refreshDay *time.Time
	if request.Refresh {
		day := tehranMidnight(now)
		refreshDay = &day
	}
	record, cached, err := s.repository.CreateOrReuse(ctx, CreateRecord{ID: newReportID(), MerchantID: merchantID, TerminalID: request.TerminalID, Fingerprint: fingerprint, DatasetWatermark: watermark, ForecastStartDate: &start, CatalogVersion: growth.CatalogVersion, AnalyticsVersion: growth.AnalyticsVersion, PromptVersion: s.config.PromptVersion, SchemaVersion: s.config.SchemaVersion, Model: s.config.Model, ManualRefresh: request.Refresh, RefreshDay: refreshDay, IdempotencyKey: idempotencyKey, RequestHash: RequestHash(merchantID, request.TerminalID, request.Refresh)}, func() error {
		return s.coordinator.AllowUncached(ctx, merchantID, now)
	})
	if err != nil {
		if errors.Is(err, ErrRefreshLimited) {
			next, quotaErr := s.repository.NextRefreshAt(ctx, merchantID, now)
			if quotaErr != nil {
				return Record{}, false, quotaErr
			}
			return Record{}, false, &RefreshLimitError{NextRefreshAt: next}
		}
		return Record{}, false, err
	}
	record.Cached = cached
	record.NextRefreshAt, err = s.repository.NextRefreshAt(ctx, merchantID, now)
	return record, cached, err
}
func (s *Service) Get(ctx context.Context, merchantID int64, id string) (Record, error) {
	record, err := s.repository.Get(ctx, merchantID, id)
	if err == nil {
		record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, s.now())
	}
	return record, err
}
func (s *Service) Latest(ctx context.Context, merchantID int64, terminalID string) (Record, error) {
	record, err := s.repository.Latest(ctx, merchantID, terminalID)
	if err == nil {
		record.Cached = true
		watermark, start, metadataErr := s.repository.DatasetMetadata(ctx)
		if metadataErr != nil {
			record.Stale = true
		} else {
			record.Stale = record.Fingerprint != s.fingerprint(merchantID, terminalID, watermark, start)
		}
		record.NextRefreshAt, _ = s.repository.NextRefreshAt(ctx, merchantID, s.now())
	}
	return record, err
}

func (s *Service) fingerprint(merchantID int64, terminalID, watermark string, start time.Time) string {
	return growth.Fingerprint(growth.FingerprintInput{MerchantID: merchantID, TerminalID: terminalID, DatasetWatermark: watermark, ForecastStart: start, HorizonDays: 90, CatalogVersion: growth.CatalogVersion, AnalyticsVersion: growth.AnalyticsVersion, PromptVersion: s.config.PromptVersion, SchemaVersion: s.config.SchemaVersion, Model: s.config.Model, ModelConfigVersion: growth.DefaultModelConfigVersion})
}

type Processor struct {
	repository  *Repository
	builder     Builder
	avalai      *AvalAIClient
	coordinator *Coordinator
	aiEnabled   bool
	now         func() time.Time
}

func NewProcessor(repository *Repository, builder Builder, avalai *AvalAIClient, coordinator *Coordinator, aiEnabled bool) *Processor {
	return &Processor{repository: repository, builder: builder, avalai: avalai, coordinator: coordinator, aiEnabled: aiEnabled, now: time.Now}
}
func (p *Processor) ProcessNext(ctx context.Context) error {
	record, err := p.repository.Claim(ctx)
	if err != nil {
		return err
	}
	terminal := ""
	if record.Scope.TerminalID != nil {
		terminal = *record.Scope.TerminalID
	}
	analysis, err := p.builder.Build(ctx, record.MerchantID, terminal)
	if err != nil {
		_ = p.repository.CompleteClaimed(ctx, record.ReportID, record.ClaimToken, StatusFailed, json.RawMessage(`{}`), nil, Usage{}, "ANALYTICS_FAILED")
		return err
	}
	if err := p.repository.StoreEvidence(ctx, record.ReportID, record.ClaimToken, analysis.Encoded); err != nil {
		return err
	}
	copy := analysis.Fallback
	status := StatusPartial
	failure := "AI_DISABLED_OR_UNAVAILABLE"
	if analysis.ShouldCallAI && p.aiEnabled && p.avalai != nil {
		reservedTokens := EstimatedTokenReservation(analysis.Encoded)
		release, enterErr := p.coordinator.Enter(ctx)
		if enterErr == nil {
			defer release()
			if err := p.coordinator.ReserveBudget(ctx, MaxProviderRequests, reservedTokens, p.now()); err == nil {
				generated, usage, aiErr := p.avalai.GenerateWithTools(ctx, analysis.Encoded, analysis.KnownEvidence, NewEvidenceTools(analysis))
				usage.ReservedTokens = reservedTokens
				if aiErr == nil {
					slog.Info("growth AI generation complete", "report_id", record.ReportID, "provider_requests", usage.ProviderRequests, "reserved_tokens", usage.ReservedTokens, "input_tokens", usage.InputTokens, "output_tokens", usage.OutputTokens, "latency_ms", usage.LatencyMS)
					return p.repository.CompleteClaimed(ctx, record.ReportID, record.ClaimToken, StatusReady, analysis.Encoded, &generated, usage, "")
				}
				slog.Warn("growth AI output rejected", "report_id", record.ReportID, "reason", "AI_REJECTED", "provider_requests", usage.ProviderRequests, "reserved_tokens", usage.ReservedTokens, "input_tokens", usage.InputTokens, "output_tokens", usage.OutputTokens, "latency_ms", usage.LatencyMS)
				failure = "AI_REJECTED"
				return p.repository.CompleteClaimed(ctx, record.ReportID, record.ClaimToken, status, analysis.Encoded, &copy, usage, failure)
			} else {
				failure = "AI_BUDGET_EXHAUSTED"
				slog.Warn("growth AI skipped", "report_id", record.ReportID, "reason", failure)
			}
		} else {
			failure = "AI_CONCURRENCY_UNAVAILABLE"
			slog.Warn("growth AI skipped", "report_id", record.ReportID, "reason", failure)
		}
	}
	if !analysis.ShouldCallAI || !p.aiEnabled {
		slog.Info("growth deterministic fallback used", "report_id", record.ReportID, "reason", failure)
	}
	return p.repository.CompleteClaimed(ctx, record.ReportID, record.ClaimToken, status, analysis.Encoded, &copy, Usage{}, failure)
}

// EstimatedTokenReservation reserves for two complete attempts, each with two
// provider turns, before any network call. Bytes are deliberately counted as
// tokens and the fixed allowance covers tool schemas, instructions, tool
// outputs, and every turn's 1,600-token output ceiling.
func EstimatedTokenReservation(evidence json.RawMessage) int64 {
	return int64(12*len(evidence) + 32000)
}

func newReportID() string {
	data := make([]byte, 12)
	if _, err := rand.Read(data); err != nil {
		return fmt.Sprintf("growth_%d", time.Now().UnixNano())
	}
	return "growth_" + hex.EncodeToString(data)
}
func tehranMidnight(now time.Time) time.Time {
	location, err := time.LoadLocation("Asia/Tehran")
	if err != nil {
		location = time.FixedZone("Asia/Tehran", 12600)
	}
	local := now.In(location)
	return time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, location)
}