package insight import ( "context" "crypto/rand" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "strings" "time" ) type Service struct { repository repositoryStore builder Builder generator copyGenerator now func() time.Time dailyRequests int64 dailyTokens int64 } func (s *Service) WithBudget(requests, tokens int64) *Service { s.dailyRequests = requests s.dailyTokens = tokens return s } type copyGenerator interface { Generate(context.Context, Module, Section, json.RawMessage, Copy) (Copy, ProviderUsage, error) Model() string } type repositoryStore interface { DatasetMetadata(context.Context) (string, time.Time, error) ActiveCatalogVersion(context.Context) (string, error) MerchantScopes(context.Context) ([]MerchantScope, error) Create(context.Context, CreateRequest) (Job, bool, error) Latest(context.Context, int64, string, Module) (Snapshot, error) GetJob(context.Context, int64, string) (Job, error) EventsAfter(context.Context, int64, string, int64) ([]Event, error) Claim(context.Context) (claimedJob, error) Complete(context.Context, claimedJob, Completion) error } type periodRepositoryStore interface { LatestForPeriod(context.Context, int64, string, Module, string) (Snapshot, error) } type interactiveRepositoryStore interface { Prioritize(context.Context, int64, string) error } type periodBuilder interface { BuildForPeriod(context.Context, int64, string, Module, PeriodScope) (Analysis, error) } func NewService(repository repositoryStore, builder Builder, generators ...copyGenerator) *Service { service := &Service{repository: repository, builder: builder, now: time.Now} if len(generators) > 0 { service.generator = generators[0] } return service } 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 } } result, err := s.repository.Latest(ctx, merchantID, terminalID, module) if err == nil { s.prioritizeActiveJob(ctx, merchantID, result) } return result, err } func (s *Service) LatestForPeriod(ctx context.Context, merchantID int64, terminalID string, module Module, period PeriodScope) (Snapshot, error) { if period.PeriodID == "" { return s.Latest(ctx, merchantID, terminalID, module) } if terminalID != "" { if builder, ok := s.builder.(periodBuilder); ok { if _, err := builder.BuildForPeriod(ctx, merchantID, terminalID, module, period); err != nil { return Snapshot{}, err } } else if _, err := s.builder.Build(ctx, merchantID, terminalID, module); err != nil { return Snapshot{}, err } } repository, ok := s.repository.(periodRepositoryStore) if !ok { return Snapshot{}, ErrNotFound } result, err := repository.LatestForPeriod(ctx, merchantID, terminalID, module, period.PeriodID) if err == nil { s.prioritizeActiveJob(ctx, merchantID, result) } return result, err } func (s *Service) prioritizeActiveJob(ctx context.Context, merchantID int64, snapshot Snapshot) { if snapshot.RefreshState.State != "QUEUED" || snapshot.RefreshState.JobID == "" { return } if repository, ok := s.repository.(interactiveRepositoryStore); ok { _ = repository.Prioritize(ctx, merchantID, snapshot.RefreshState.JobID) } } 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, SectionFull, "MANUAL", true, idempotencyKey, PeriodScope{}) } func (s *Service) RefreshForPeriod(ctx context.Context, merchantID int64, terminalID string, module Module, period PeriodScope, idempotencyKey string) (Job, bool, error) { return s.enqueue(ctx, merchantID, terminalID, module, SectionFull, "MANUAL", true, idempotencyKey, period) } func (s *Service) RefreshSection(ctx context.Context, merchantID int64, terminalID string, module Module, section Section, idempotencyKey string) (Job, bool, error) { if _, ok := ParseSection(module, string(section)); !ok { return Job{}, false, errors.New("ai insight: invalid section") } return s.enqueue(ctx, merchantID, terminalID, module, section, "MANUAL", true, idempotencyKey, PeriodScope{}) } func (s *Service) RefreshSectionForPeriod(ctx context.Context, merchantID int64, terminalID string, module Module, section Section, period PeriodScope, idempotencyKey string) (Job, bool, error) { if _, ok := ParseSection(module, string(section)); !ok { return Job{}, false, errors.New("ai insight: invalid section") } return s.enqueue(ctx, merchantID, terminalID, module, section, "MANUAL", true, idempotencyKey, period) } 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, SectionFull, reason, false, "", PeriodScope{}) } func (s *Service) EnqueueAll(ctx context.Context, reason string) (int, error) { scopes, err := s.repository.MerchantScopes(ctx) if err != nil { return 0, err } watermark, dataThrough, err := s.repository.DatasetMetadata(ctx) if err != nil { return 0, err } catalog, err := s.repository.ActiveCatalogVersion(ctx) if err != nil { return 0, err } timeContext := BuildTimeContext(s.now(), dataThrough) modules := []Module{ModuleOverview, ModulePayments, ModuleFees, ModulePeers, ModuleGrowth} created := 0 for _, scope := range scopes { for _, module := range modules { if err := ctx.Err(); err != nil { return created, err } if _, cached, err := s.enqueuePrepared(ctx, scope.MerchantID, scope.TerminalID, module, SectionFull, reason, false, "", watermark, catalog, timeContext, PeriodScope{}); errors.Is(err, ErrOnboardingRequired) { continue } else if 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, section Section, reason string, manual bool, key string, period PeriodScope) (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) return s.enqueuePrepared(ctx, merchantID, terminalID, module, section, reason, manual, key, watermark, catalog, tc, period) } func (s *Service) enqueuePrepared(ctx context.Context, merchantID int64, terminalID string, module Module, section Section, reason string, manual bool, key, watermark, catalog string, tc TimeContext, period PeriodScope) (Job, bool, error) { if section == "" { section = SectionFull } fingerprintSalt := "" triggerReason := reason if strings.HasPrefix(reason, "PROFILE:") { triggerReason = "PROFILE" fingerprintSalt = strings.TrimPrefix(reason, "PROFILE:") } fingerprint := hash(fmt.Sprintf("%d|%s|%s|%s|%s|%s|%s|%s|%s|%s|%s", merchantID, terminalID, module, section, period.PeriodID, tc.AsOfDate, watermark, catalog, AnalyticsVersion, PageInsightVersion, fingerprintSalt)) requestHash := hash(fmt.Sprintf("%d|%s|%s|%s|%s|%t", merchantID, terminalID, module, section, period.PeriodID, manual)) return s.repository.Create(ctx, CreateRequest{MerchantID: merchantID, TerminalID: terminalID, Module: module, Section: section, TriggerReason: triggerReason, ManualRefresh: manual, IdempotencyKey: key, RequestHash: requestHash, Fingerprint: fingerprint, Time: tc, Confidence: confidenceFor(tc.Freshness), CatalogVersion: catalog, Evidence: []byte(`{}`), Period: period}) } 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 } var analysis Analysis if job.Period.PeriodID != "" { if builder, ok := s.builder.(periodBuilder); ok { analysis, err = builder.BuildForPeriod(ctx, job.MerchantID, job.TerminalID, job.Module, job.Period) } else { analysis, err = s.builder.Build(ctx, job.MerchantID, job.TerminalID, job.Module) } } else { 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) } completion := Completion{Analysis: analysis, Status: "PARTIAL", FailureCode: "PROVIDER_UNAVAILABLE"} if s.generator != nil { if usageRepository, ok := s.repository.(interface { DailyProviderUsage(context.Context) (int64, int64, error) }); ok && (s.dailyRequests > 0 || s.dailyTokens > 0) { requests, tokens, usageErr := usageRepository.DailyProviderUsage(ctx) if usageErr != nil { return usageErr } if (s.dailyRequests > 0 && requests >= s.dailyRequests) || (s.dailyTokens > 0 && tokens >= s.dailyTokens) { completion.FailureCode = "BUDGET_EXHAUSTED" return s.repository.Complete(ctx, job, completion) } } copy, usage, generateErr := s.generator.Generate(ctx, job.Module, job.Section, analysis.Evidence, analysis.Copy) completion.Usage, completion.Model = usage, s.generator.Model() if generateErr == nil { base := analysis.Copy if job.Section != SectionFull && job.BaseCopy.Past.Body != "" { base = job.BaseCopy } merged, mergeErr := MergeSection(base, copy, job.Section) if mergeErr == nil { completion.AICopy, completion.Status, completion.FailureCode = &merged, "READY", "" } else { completion.FailureCode = "PROVIDER_OR_VALIDATION_FAILED" } } else { completion.FailureCode = "PROVIDER_OR_VALIDATION_FAILED" } } return s.repository.Complete(ctx, job, completion) } 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) }