478 lines
24 KiB
Go
478 lines
24 KiB
Go
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) DailyProviderUsage(ctx context.Context) (int64, int64, error) {
|
|
var requests, tokens int64
|
|
err := r.pool.QueryRow(ctx, `SELECT coalesce(sum(provider_requests),0),coalesce(sum(input_tokens+output_tokens),0) FROM ai_insight_snapshots WHERE (created_at AT TIME ZONE 'Asia/Tehran')::date=(now() AT TIME ZONE 'Asia/Tehran')::date`).Scan(&requests, &tokens)
|
|
return requests, tokens, err
|
|
}
|
|
|
|
func (r *Repository) Create(ctx context.Context, in CreateRequest) (Job, bool, error) {
|
|
if in.Section == "" {
|
|
in.Section = SectionFull
|
|
}
|
|
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:%s", in.MerchantID, in.TerminalID, in.Module, in.Period.PeriodID)
|
|
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
|
|
var held bool
|
|
if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM ai_generation_holds WHERE merchant_id=$1 AND terminal_id=$2)`, in.MerchantID, id).Scan(&held); err != nil {
|
|
return Job{}, false, fmt.Errorf("checking terminal AI generation hold: %w", err)
|
|
}
|
|
if held {
|
|
return Job{}, false, ErrOnboardingRequired
|
|
}
|
|
}
|
|
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.period_id IS NOT DISTINCT FROM $5::text 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, nilIfEmpty(in.Period.PeriodID)).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)
|
|
}
|
|
}
|
|
refreshSections := []Section{in.Section}
|
|
if in.Section == SectionFull {
|
|
refreshSections = refreshableSections(in.Module)
|
|
}
|
|
if in.ManualRefresh {
|
|
for _, section := range refreshSections {
|
|
var uses int
|
|
var latest *time.Time
|
|
if err := tx.QueryRow(ctx, `SELECT count(*),max(accepted_at) FROM ai_insight_refresh_attempts WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND module=$3 AND period_id IS NOT DISTINCT FROM $4::text AND section_id=$5 AND tehran_refresh_day=$6`, in.MerchantID, nilIfEmpty(in.TerminalID), in.Module, nilIfEmpty(in.Period.PeriodID), section, in.Time.AsOfDate).Scan(&uses, &latest); err != nil {
|
|
return Job{}, false, fmt.Errorf("checking section insight refresh: %w", err)
|
|
}
|
|
if uses >= 3 {
|
|
return Job{}, false, ErrRefreshLimited
|
|
}
|
|
if latest != nil && time.Since(*latest) < time.Minute {
|
|
return Job{}, false, &RefreshCooldownError{NextRefreshAt: latest.Add(time.Minute)}
|
|
}
|
|
}
|
|
}
|
|
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 AND period_id IS NOT DISTINCT FROM $4::text`, in.MerchantID, nilIfEmpty(in.TerminalID), in.Module, nilIfEmpty(in.Period.PeriodID)).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,section_id,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,period_id,period_type,period_start_at,period_end_at,period_label_fa)
|
|
VALUES($1,$2,$3,$4,$5,$6,$7,$8,'QUEUED',$9,$10,$11,$12,$13,$14,$15,$16,$16,'unassigned',$17,$18,$19,$20,$21,$22,$23,$24)`, snapshotID, in.MerchantID, terminalDBID, nilIfEmpty(in.TerminalID), in.Module, in.Section, revision, in.Fingerprint, in.Time.AsOfDate, in.Time.DataThroughDate, in.Time.Freshness, in.Time.DataLagDays, in.Confidence, in.CatalogVersion, AnalyticsVersion, PageInsightVersion, in.Evidence, in.ManualRefresh, nullableDate(in.ManualRefresh, in.Time.AsOfDate), nilIfEmpty(in.Period.PeriodID), nilIfEmpty(in.Period.PeriodType), nullableTime(in.Period.PeriodID != "", in.Period.StartAt), nullableTime(in.Period.PeriodID != "", in.Period.EndAt), nilIfEmpty(in.Period.JalaliLabelFA))
|
|
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,section_id,trigger_reason,status,idempotency_key,request_hash) VALUES($1,$2,$3,$4,$5,$6,'QUEUED',$7,$8)`, jobID, snapshotID, in.MerchantID, in.Module, in.Section, in.TriggerReason, nilIfEmpty(in.IdempotencyKey), nilIfEmpty(in.RequestHash))
|
|
if err != nil {
|
|
return Job{}, false, fmt.Errorf("creating insight job: %w", err)
|
|
}
|
|
if in.ManualRefresh {
|
|
for _, section := range refreshSections {
|
|
if _, err := tx.Exec(ctx, `INSERT INTO ai_insight_refresh_attempts(job_id,merchant_id,terminal_source_key,module,period_id,section_id,tehran_refresh_day) VALUES($1,$2,$3,$4,$5,$6,$7)`, jobID, in.MerchantID, nilIfEmpty(in.TerminalID), in.Module, nilIfEmpty(in.Period.PeriodID), section, in.Time.AsOfDate); err != nil {
|
|
return Job{}, false, fmt.Errorf("recording accepted section refresh: %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
|
|
Period PeriodScope
|
|
LeaseToken string
|
|
BaseCopy Copy
|
|
}
|
|
|
|
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<now()`); err != nil {
|
|
return claimedJob{}, fmt.Errorf("recovering insight leases: %w", err)
|
|
}
|
|
var job claimedJob
|
|
var terminal *string
|
|
var periodID, periodType, periodLabel *string
|
|
var periodStart, periodEnd *time.Time
|
|
err = tx.QueryRow(ctx, `SELECT j.id,j.snapshot_id,j.merchant_id,s.terminal_source_key,j.module,j.section_id,j.created_at,j.updated_at,s.period_id,s.period_type,s.period_start_at,s.period_end_at,s.period_label_fa
|
|
FROM ai_insight_jobs j JOIN ai_insight_snapshots s ON s.id=j.snapshot_id
|
|
WHERE j.id=(SELECT id FROM ai_insight_jobs WHERE status='QUEUED' AND scheduled_at<=now()
|
|
ORDER BY CASE trigger_reason WHEN 'MANUAL' THEN 0 WHEN 'PROFILE' THEN 1 ELSE 2 END,scheduled_at,created_at FOR UPDATE SKIP LOCKED LIMIT 1)`).Scan(&job.JobID, &job.SnapshotID, &job.MerchantID, &terminal, &job.Module, &job.Section, &job.CreatedAt, &job.UpdatedAt, &periodID, &periodType, &periodStart, &periodEnd, &periodLabel)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return claimedJob{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return claimedJob{}, fmt.Errorf("selecting insight job: %w", err)
|
|
}
|
|
if terminal != nil {
|
|
job.TerminalID = *terminal
|
|
}
|
|
if periodID != nil {
|
|
job.Period = PeriodScope{PeriodID: *periodID, PeriodType: *periodType, StartAt: *periodStart, EndAt: *periodEnd, JalaliLabelFA: *periodLabel}
|
|
}
|
|
if job.Section != SectionFull {
|
|
var encoded []byte
|
|
if err := tx.QueryRow(ctx, `SELECT coalesce(ai_response,deterministic_copy) FROM ai_insight_snapshots WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND module=$3 AND period_id IS NOT DISTINCT FROM $4::text AND id<>$5 AND status IN('READY','PARTIAL') ORDER BY (status='READY') DESC,revision DESC LIMIT 1`, job.MerchantID, nilIfEmpty(job.TerminalID), job.Module, nilIfEmpty(job.Period.PeriodID), job.SnapshotID).Scan(&encoded); err == nil {
|
|
_ = json.Unmarshal(encoded, &job.BaseCopy)
|
|
}
|
|
}
|
|
job.State, job.EventsURL, job.LeaseToken = "RUNNING", "/api/v1/analytics/ai-insight-refreshes/"+job.JobID+"/events", newID("lease_")
|
|
if _, err := tx.Exec(ctx, `UPDATE ai_insight_jobs SET status='RUNNING',attempts=attempts+1,lease_token=$2,lease_expires_at=now()+make_interval(secs=>$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) Prioritize(ctx context.Context, merchantID int64, jobID string) error {
|
|
_, err := r.pool.Exec(ctx, `UPDATE ai_insight_jobs SET trigger_reason='MANUAL',scheduled_at=least(scheduled_at,now()),updated_at=now() WHERE id=$1 AND merchant_id=$2 AND status='QUEUED'`, jobID, merchantID)
|
|
if err != nil {
|
|
return fmt.Errorf("prioritizing interactive insight job: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) Complete(ctx context.Context, job claimedJob, completion Completion) error {
|
|
tx, err := r.pool.Begin(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("beginning insight completion: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
lockKey := fmt.Sprintf("ai-insight-complete:%d:%s:%s:%s", job.MerchantID, job.TerminalID, job.Module, job.Period.PeriodID)
|
|
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, lockKey); err != nil {
|
|
return fmt.Errorf("locking insight completion: %w", err)
|
|
}
|
|
if job.Section != SectionFull {
|
|
var latestJSON []byte
|
|
if err := tx.QueryRow(ctx, `SELECT coalesce(ai_response,deterministic_copy) FROM ai_insight_snapshots WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND module=$3 AND period_id IS NOT DISTINCT FROM $4::text AND id<>$5 AND status IN('READY','PARTIAL') ORDER BY completed_at DESC NULLS LAST,revision DESC LIMIT 1`, job.MerchantID, nilIfEmpty(job.TerminalID), job.Module, nilIfEmpty(job.Period.PeriodID), job.SnapshotID).Scan(&latestJSON); err == nil {
|
|
var latest Copy
|
|
if json.Unmarshal(latestJSON, &latest) == nil {
|
|
mergedDeterministic, mergeErr := MergeSection(latest, completion.Analysis.Copy, job.Section)
|
|
if mergeErr != nil {
|
|
return fmt.Errorf("merging latest deterministic insight section: %w", mergeErr)
|
|
}
|
|
completion.Analysis.Copy = mergedDeterministic
|
|
if completion.AICopy != nil {
|
|
merged, mergeErr := MergeSection(latest, *completion.AICopy, job.Section)
|
|
if mergeErr != nil {
|
|
return fmt.Errorf("merging latest AI insight section: %w", mergeErr)
|
|
}
|
|
completion.AICopy = &merged
|
|
}
|
|
}
|
|
} else if !errors.Is(err, pgx.ErrNoRows) {
|
|
return fmt.Errorf("loading latest insight copy at completion: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE ai_insight_snapshots SET revision=(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 AND period_id IS NOT DISTINCT FROM $4::text AND id<>$5) WHERE id=$5`, job.MerchantID, nilIfEmpty(job.TerminalID), job.Module, nilIfEmpty(job.Period.PeriodID), job.SnapshotID); err != nil {
|
|
return fmt.Errorf("ordering insight revision: %w", err)
|
|
}
|
|
}
|
|
deterministicCopy, _ := json.Marshal(completion.Analysis.Copy)
|
|
var aiCopy any
|
|
if completion.AICopy != nil {
|
|
encoded, _ := json.Marshal(completion.AICopy)
|
|
aiCopy = encoded
|
|
}
|
|
_, 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,deterministic_copy=$10,ai_response=$11,failure_code=$12,model=coalesce(nullif($13,''),model),input_tokens=$14,output_tokens=$15,provider_requests=$16,latency_ms=$17,provider_request_id=nullif($18,''),estimated_cost=nullif($19,''),updated_at=now(),completed_at=now() WHERE id=$1 AND EXISTS(SELECT 1 FROM ai_insight_jobs WHERE id=$2 AND lease_token=$20)`, job.SnapshotID, job.JobID, completion.Status, completion.Analysis.Time.AsOfDate, completion.Analysis.Time.DataThroughDate, completion.Analysis.Time.Freshness, completion.Analysis.Time.DataLagDays, completion.Analysis.Confidence, completion.Analysis.Evidence, deterministicCopy, aiCopy, nilIfEmpty(completion.FailureCode), completion.Model, completion.Usage.InputTokens, completion.Usage.OutputTokens, completion.Usage.ProviderRequests, completion.Usage.LatencyMS, completion.Usage.ProviderRequestID, completion.Usage.EstimatedCost, 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": completion.Status}); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (r *Repository) Latest(ctx context.Context, merchantID int64, terminalID string, module Module) (Snapshot, error) {
|
|
return r.latest(ctx, merchantID, terminalID, module, "")
|
|
}
|
|
|
|
func (r *Repository) LatestForPeriod(ctx context.Context, merchantID int64, terminalID string, module Module, periodID string) (Snapshot, error) {
|
|
return r.latest(ctx, merchantID, terminalID, module, periodID)
|
|
}
|
|
|
|
func (r *Repository) latest(ctx context.Context, merchantID int64, terminalID string, module Module, periodID string) (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.period_id IS NOT DISTINCT FROM $4::text AND s.status IN ('READY','PARTIAL') ORDER BY (s.status='READY') DESC,s.revision DESC LIMIT 1`, merchantID, nilIfEmpty(terminalID), module, nilIfEmpty(periodID))
|
|
result, err := scanSnapshot(row)
|
|
if err != nil {
|
|
return Snapshot{}, err
|
|
}
|
|
result.SectionRefresh = make(map[Section]RefreshState)
|
|
for _, section := range refreshableSections(module) {
|
|
result.SectionRefresh[section] = RefreshState{State: "IDLE"}
|
|
}
|
|
if _, err := r.pool.Exec(ctx, `UPDATE ai_insight_jobs j SET trigger_reason='MANUAL',scheduled_at=least(j.scheduled_at,now()),updated_at=now() FROM ai_insight_snapshots s WHERE s.id=j.snapshot_id AND s.merchant_id=$1 AND s.terminal_source_key IS NOT DISTINCT FROM $2::text AND s.module=$3 AND s.period_id IS NOT DISTINCT FROM $4::text AND j.status='QUEUED'`, merchantID, nilIfEmpty(terminalID), module, nilIfEmpty(periodID)); err != nil {
|
|
return Snapshot{}, fmt.Errorf("prioritizing active insight refreshes: %w", err)
|
|
}
|
|
jobs, queryErr := r.pool.Query(ctx, `SELECT j.id,j.status,j.section_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.period_id IS NOT DISTINCT FROM $4::text AND j.status IN ('QUEUED','RUNNING') ORDER BY j.created_at`, merchantID, nilIfEmpty(terminalID), module, nilIfEmpty(periodID))
|
|
if queryErr != nil {
|
|
return Snapshot{}, fmt.Errorf("loading active insight refreshes: %w", queryErr)
|
|
}
|
|
defer jobs.Close()
|
|
for jobs.Next() {
|
|
var jobID, state string
|
|
var section Section
|
|
if err := jobs.Scan(&jobID, &state, §ion); err != nil {
|
|
return Snapshot{}, err
|
|
}
|
|
refresh := RefreshState{State: state, JobID: jobID, EventsURL: "/api/v1/analytics/ai-insight-refreshes/" + jobID + "/events"}
|
|
result.RefreshState = refresh
|
|
if section == SectionFull {
|
|
for _, item := range refreshableSections(module) {
|
|
result.SectionRefresh[item] = refresh
|
|
}
|
|
} else {
|
|
result.SectionRefresh[section] = refresh
|
|
}
|
|
}
|
|
if err := jobs.Err(); err != nil {
|
|
return Snapshot{}, err
|
|
}
|
|
if result.RefreshState.State == "" {
|
|
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.period_id,s.as_of_date,s.data_through_date,s.freshness,s.data_lag_days,s.confidence,coalesce(s.ai_response,s.deterministic_copy),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 periodID *string
|
|
var asOf, through time.Time
|
|
var copyJSON []byte
|
|
err := row.Scan(&s.InsightID, &s.Module, &s.Revision, &s.Status, &terminal, &periodID, &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.Scope.PeriodID = periodID
|
|
s.AsOfDate, s.ForecastStart = asOf.Format(time.DateOnly), asOf.Format(time.DateOnly)
|
|
s.DataThroughDate = through.Format(time.DateOnly)
|
|
_ = json.Unmarshal(copyJSON, &s.Copy)
|
|
var evidence struct {
|
|
CalendarContext []CalendarContext `json:"calendarContext"`
|
|
}
|
|
if err := json.Unmarshal(s.Evidence, &evidence); err == nil && evidence.CalendarContext != nil {
|
|
s.CalendarContext = evidence.CalendarContext
|
|
} else {
|
|
s.CalendarContext = []CalendarContext{}
|
|
}
|
|
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,module,section_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.Module, &job.Section, &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 refreshableSections(module Module) []Section {
|
|
sections := []Section{SectionPast, SectionLatestObserved, SectionFuture7Days, SectionFuture30Days, SectionFindings, SectionOptions}
|
|
if module == ModuleGrowth {
|
|
sections = append(sections, SectionFuture90Days)
|
|
}
|
|
return sections
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func nullableTime(enabled bool, value time.Time) any {
|
|
if !enabled {
|
|
return nil
|
|
}
|
|
return value
|
|
}
|