feat: add persisted page insight platform
This commit is contained in:
183
internal/insight/builder.go
Normal file
183
internal/insight/builder.go
Normal file
@@ -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
|
||||
}
|
||||
145
internal/insight/model.go
Normal file
145
internal/insight/model.go
Normal file
@@ -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
|
||||
}
|
||||
22
internal/insight/model_test.go
Normal file
22
internal/insight/model_test.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
328
internal/insight/repository.go
Normal file
328
internal/insight/repository.go
Normal file
@@ -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<now()`); err != nil {
|
||||
return claimedJob{}, fmt.Errorf("recovering insight leases: %w", err)
|
||||
}
|
||||
var job claimedJob
|
||||
var terminal *string
|
||||
err = tx.QueryRow(ctx, `SELECT j.id,j.snapshot_id,j.merchant_id,s.terminal_source_key,j.module,j.created_at,j.updated_at
|
||||
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 scheduled_at,created_at FOR UPDATE SKIP LOCKED LIMIT 1)`).Scan(&job.JobID, &job.SnapshotID, &job.MerchantID, &terminal, &job.Module, &job.CreatedAt, &job.UpdatedAt)
|
||||
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
|
||||
}
|
||||
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) 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
|
||||
}
|
||||
114
internal/insight/service.go
Normal file
114
internal/insight/service.go
Normal file
@@ -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)
|
||||
}
|
||||
Reference in New Issue
Block a user