package coach import ( "context" "crypto/rand" "encoding/hex" "encoding/json" "errors" "fmt" "math" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) type Repository struct{ pool *pgxpool.Pool } func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} } type scheduledProfile struct { MerchantID, TerminalDBID int64 Profile Profile } func (r *Repository) CurrentProfiles(ctx context.Context) ([]scheduledProfile, error) { rows, err := r.pool.Query(ctx, `SELECT merchant_id,terminal_id,id,terminal_source_key,version,difficulty,business_description,niches,sales_channels,primary_goal,constraints_text,created_at FROM growth_coach_profiles WHERE is_current ORDER BY merchant_id,terminal_id`) if err != nil { return nil, err } defer rows.Close() result := []scheduledProfile{} for rows.Next() { var item scheduledProfile var niches, channels []byte if err := rows.Scan(&item.MerchantID, &item.TerminalDBID, &item.Profile.ProfileID, &item.Profile.TerminalID, &item.Profile.Version, &item.Profile.Difficulty, &item.Profile.BusinessDescription, &niches, &channels, &item.Profile.PrimaryGoal, &item.Profile.Constraints, &item.Profile.CreatedAt); err != nil { return nil, err } _ = json.Unmarshal(niches, &item.Profile.Niches) _ = json.Unmarshal(channels, &item.Profile.SalesChannels) result = append(result, item) } return result, rows.Err() } func (r *Repository) terminal(ctx context.Context, merchantID int64, sourceKey string) (int64, error) { var id int64 err := r.pool.QueryRow(ctx, `SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2`, merchantID, sourceKey).Scan(&id) if errors.Is(err, pgx.ErrNoRows) { return 0, ErrUnknownTerminal } if err != nil { return 0, fmt.Errorf("validating coach terminal: %w", err) } return id, nil } func (r *Repository) Profile(ctx context.Context, merchantID int64, terminalKey string) (Profile, error) { var profile Profile var niches, channels, changed []byte err := r.pool.QueryRow(ctx, `SELECT id,terminal_source_key,version,difficulty,business_description,niches,sales_channels,primary_goal,constraints_text,created_at,changed_fields FROM growth_coach_profiles WHERE merchant_id=$1 AND terminal_source_key=$2 AND is_current`, merchantID, terminalKey). Scan(&profile.ProfileID, &profile.TerminalID, &profile.Version, &profile.Difficulty, &profile.BusinessDescription, &niches, &channels, &profile.PrimaryGoal, &profile.Constraints, &profile.CreatedAt, &changed) if errors.Is(err, pgx.ErrNoRows) { return Profile{}, ErrNotFound } if err != nil { return Profile{}, fmt.Errorf("loading coach profile: %w", err) } _ = json.Unmarshal(niches, &profile.Niches) _ = json.Unmarshal(channels, &profile.SalesChannels) _ = json.Unmarshal(changed, &profile.ChangedFields) return profile, nil } func (r *Repository) SaveProfile(ctx context.Context, merchantID, userID int64, terminalKey, idempotencyKey, requestHash string, input ProfileInput) (Profile, bool, error) { tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable}) if err != nil { return Profile{}, false, err } defer func() { _ = tx.Rollback(ctx) }() var terminalID int64 err = tx.QueryRow(ctx, `SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2 FOR SHARE`, merchantID, terminalKey).Scan(&terminalID) if errors.Is(err, pgx.ErrNoRows) { return Profile{}, false, ErrUnknownTerminal } if err != nil { return Profile{}, false, fmt.Errorf("validating profile terminal: %w", err) } if idempotencyKey != "" { var existingHash string var response []byte err = tx.QueryRow(ctx, `SELECT request_hash,response FROM growth_coach_idempotency WHERE merchant_id=$1 AND idempotency_key=$2`, merchantID, idempotencyKey).Scan(&existingHash, &response) if err == nil { if existingHash != requestHash { return Profile{}, false, ErrIdempotencyConflict } var profile Profile _ = json.Unmarshal(response, &profile) return profile, true, nil } if !errors.Is(err, pgx.ErrNoRows) { return Profile{}, false, err } } if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, terminalID); err != nil { return Profile{}, false, err } var previous Profile var previousNiches, previousChannels, previousChanged []byte previousErr := tx.QueryRow(ctx, `SELECT id,terminal_source_key,version,difficulty,business_description,niches,sales_channels,primary_goal,constraints_text,created_at,changed_fields FROM growth_coach_profiles WHERE merchant_id=$1 AND terminal_id=$2 AND is_current FOR UPDATE`, merchantID, terminalID).Scan(&previous.ProfileID, &previous.TerminalID, &previous.Version, &previous.Difficulty, &previous.BusinessDescription, &previousNiches, &previousChannels, &previous.PrimaryGoal, &previous.Constraints, &previous.CreatedAt, &previousChanged) if previousErr == nil { _ = json.Unmarshal(previousNiches, &previous.Niches) _ = json.Unmarshal(previousChannels, &previous.SalesChannels) _ = json.Unmarshal(previousChanged, &previous.ChangedFields) if len(ChangedProfileFields(previous.ProfileInput, input)) == 0 { encoded, _ := json.Marshal(previous) if idempotencyKey != "" { if _, err := tx.Exec(ctx, `INSERT INTO growth_coach_idempotency(merchant_id,idempotency_key,request_hash,response) VALUES($1,$2,$3,$4)`, merchantID, idempotencyKey, requestHash, encoded); err != nil { return Profile{}, false, err } } if err := tx.Commit(ctx); err != nil { return Profile{}, false, err } return previous, true, nil } } else if !errors.Is(previousErr, pgx.ErrNoRows) { return Profile{}, false, previousErr } var version, difficulty int if err := tx.QueryRow(ctx, `SELECT coalesce(max(version),0)+1,coalesce(max(difficulty) FILTER(WHERE is_current),2) FROM growth_coach_profiles WHERE merchant_id=$1 AND terminal_id=$2`, merchantID, terminalID).Scan(&version, &difficulty); err != nil { return Profile{}, false, err } if _, err := tx.Exec(ctx, `UPDATE growth_coach_profiles SET is_current=false WHERE merchant_id=$1 AND terminal_id=$2 AND is_current`, merchantID, terminalID); err != nil { return Profile{}, false, err } changedFields := []string{"businessDescription", "niches", "salesChannels", "primaryGoal", "constraints"} if previousErr == nil { changedFields = ChangedProfileFields(previous.ProfileInput, input) } profile := Profile{ProfileID: newID("profile_"), TerminalID: terminalKey, Version: version, Difficulty: difficulty, ProfileInput: input, CreatedAt: time.Now().UTC(), ChangedFields: changedFields} niches, _ := json.Marshal(input.Niches) channels, _ := json.Marshal(input.SalesChannels) changedJSON, _ := json.Marshal(changedFields) err = tx.QueryRow(ctx, `INSERT INTO growth_coach_profiles(id,merchant_id,terminal_id,terminal_source_key,version,business_description,niches,sales_channels,primary_goal,constraints_text,difficulty,actor_user_id,changed_fields) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) RETURNING created_at`, profile.ProfileID, merchantID, terminalID, terminalKey, version, input.BusinessDescription, niches, channels, input.PrimaryGoal, input.Constraints, difficulty, userID, changedJSON).Scan(&profile.CreatedAt) if err != nil { return Profile{}, false, fmt.Errorf("saving coach profile: %w", err) } encoded, _ := json.Marshal(profile) if idempotencyKey != "" { if _, err := tx.Exec(ctx, `INSERT INTO growth_coach_idempotency(merchant_id,idempotency_key,request_hash,response) VALUES($1,$2,$3,$4)`, merchantID, idempotencyKey, requestHash, encoded); err != nil { return Profile{}, false, err } } if err := audit(ctx, tx, merchantID, terminalID, &userID, "PROFILE", profile.ProfileID, "profile.created", map[string]any{"version": version}); err != nil { return Profile{}, false, err } if previousErr == nil { if _, err := tx.Exec(ctx, `UPDATE growth_coach_recommendations SET status='EXPIRED' WHERE merchant_id=$1 AND terminal_id=$2 AND status='OFFERED' AND plan_id IN(SELECT id FROM growth_coach_plans WHERE profile_id<>$3)`, merchantID, terminalID, profile.ProfileID); err != nil { return Profile{}, false, err } } if _, err := tx.Exec(ctx, `DELETE FROM ai_generation_holds WHERE merchant_id=$1 AND terminal_id=$2`, merchantID, terminalID); err != nil { return Profile{}, false, err } if err := tx.Commit(ctx); err != nil { return Profile{}, false, err } return profile, false, nil } type enqueueInput struct { MerchantID int64 TerminalID int64 TerminalKey, ProfileID, Cadence, Reason, Fingerprint, IdempotencyKey, RequestHash string Start, End time.Time } func (r *Repository) Enqueue(ctx context.Context, in enqueueInput) (Job, bool, error) { job := Job{JobID: newID("coachjob_"), State: "QUEUED", EventsURL: "", CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC()} err := r.pool.QueryRow(ctx, `INSERT INTO growth_coach_jobs(id,merchant_id,terminal_id,terminal_source_key,profile_id,cadence,trigger_reason,period_start,period_end,fingerprint,idempotency_key,request_hash,status) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,nullif($11,''),nullif($12,''),'QUEUED') ON CONFLICT(fingerprint) DO NOTHING RETURNING created_at,updated_at`, job.JobID, in.MerchantID, in.TerminalID, in.TerminalKey, in.ProfileID, in.Cadence, in.Reason, in.Start, in.End, in.Fingerprint, in.IdempotencyKey, in.RequestHash).Scan(&job.CreatedAt, &job.UpdatedAt) if errors.Is(err, pgx.ErrNoRows) { err = r.pool.QueryRow(ctx, `SELECT id,status,created_at,updated_at FROM growth_coach_jobs WHERE fingerprint=$1`, in.Fingerprint).Scan(&job.JobID, &job.State, &job.CreatedAt, &job.UpdatedAt) if err != nil { return Job{}, false, err } job.EventsURL = "/api/v1/growth-coach/refreshes/" + job.JobID + "/events" return job, true, nil } if err != nil { return Job{}, false, fmt.Errorf("enqueueing coach generation: %w", err) } job.EventsURL = "/api/v1/growth-coach/refreshes/" + job.JobID + "/events" _, _ = r.pool.Exec(ctx, `INSERT INTO growth_coach_events(merchant_id,terminal_id,entity_type,entity_id,event_type,payload) VALUES($1,$2,'JOB',$3,'job.queued',$4)`, in.MerchantID, in.TerminalID, job.JobID, []byte(`{"state":"QUEUED"}`)) return job, false, nil } func (r *Repository) EnqueueManual(ctx context.Context, in enqueueInput) (Job, bool, error) { tx, err := r.pool.Begin(ctx) if err != nil { return Job{}, false, fmt.Errorf("beginning manual coach refresh: %w", err) } defer func() { _ = tx.Rollback(ctx) }() lockKey := fmt.Sprintf("growth-coach-refresh:%d:%d:%s:%s", in.MerchantID, in.TerminalID, in.Cadence, in.Start.Format(time.DateOnly)) if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, lockKey); err != nil { return Job{}, false, fmt.Errorf("locking manual coach refresh: %w", err) } if in.IdempotencyKey != "" { var existing Job var existingHash string err := tx.QueryRow(ctx, `SELECT id,status,created_at,updated_at,coalesce(request_hash,'') FROM growth_coach_jobs WHERE merchant_id=$1 AND idempotency_key=$2`, in.MerchantID, in.IdempotencyKey). Scan(&existing.JobID, &existing.State, &existing.CreatedAt, &existing.UpdatedAt, &existingHash) if err == nil { if existingHash != in.RequestHash { return Job{}, false, ErrIdempotencyConflict } existing.EventsURL = "/api/v1/growth-coach/refreshes/" + existing.JobID + "/events" return existing, true, nil } if !errors.Is(err, pgx.ErrNoRows) { return Job{}, false, fmt.Errorf("checking manual coach refresh replay: %w", err) } } var weeklyUses int if in.Cadence == "WEEKLY" { if err := tx.QueryRow(ctx, `SELECT count(*) FROM growth_coach_jobs WHERE merchant_id=$1 AND terminal_id=$2 AND cadence='WEEKLY' AND trigger_reason='MANUAL' AND period_start=$3`, in.MerchantID, in.TerminalID, in.Start).Scan(&weeklyUses); err != nil { return Job{}, false, fmt.Errorf("counting weekly coach refreshes: %w", err) } } if !ManualRefreshAllowed(in.Cadence, weeklyUses) { return Job{}, false, ErrRefreshLimited } job := Job{JobID: newID("coachjob_"), State: "QUEUED", CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC()} if err := tx.QueryRow(ctx, `INSERT INTO growth_coach_jobs(id,merchant_id,terminal_id,terminal_source_key,profile_id,cadence,trigger_reason,period_start,period_end,fingerprint,idempotency_key,request_hash,status) VALUES($1,$2,$3,$4,$5,$6,'MANUAL',$7,$8,$9,nullif($10,''),nullif($11,''),'QUEUED') RETURNING created_at,updated_at`, job.JobID, in.MerchantID, in.TerminalID, in.TerminalKey, in.ProfileID, in.Cadence, in.Start, in.End, in.Fingerprint, in.IdempotencyKey, in.RequestHash).Scan(&job.CreatedAt, &job.UpdatedAt); err != nil { return Job{}, false, fmt.Errorf("enqueueing manual coach refresh: %w", err) } if _, err := tx.Exec(ctx, `INSERT INTO growth_coach_events(merchant_id,terminal_id,entity_type,entity_id,event_type,payload) VALUES($1,$2,'JOB',$3,'job.queued',$4)`, in.MerchantID, in.TerminalID, job.JobID, []byte(`{"state":"QUEUED"}`)); err != nil { return Job{}, false, fmt.Errorf("recording manual coach refresh event: %w", err) } if err := tx.Commit(ctx); err != nil { return Job{}, false, fmt.Errorf("committing manual coach refresh: %w", err) } job.EventsURL = "/api/v1/growth-coach/refreshes/" + job.JobID + "/events" return job, false, nil } type claimedJob struct { Job MerchantID, TerminalDBID int64 TerminalID, ProfileID, Cadence, LeaseToken string Start, End time.Time } func (r *Repository) Claim(ctx context.Context) (claimedJob, error) { tx, err := r.pool.Begin(ctx) if err != nil { return claimedJob{}, err } defer func() { _ = tx.Rollback(ctx) }() _, _ = tx.Exec(ctx, `UPDATE growth_coach_jobs SET status='QUEUED',lease_token=NULL,lease_expires_at=NULL WHERE status='RUNNING' AND lease_expires_at 1 { var previous ProfileInput var niches, channels []byte if err := r.pool.QueryRow(ctx, `SELECT business_description,niches,sales_channels,primary_goal,constraints_text FROM growth_coach_profiles WHERE merchant_id=$1 AND terminal_id=$2 AND version=$3`, job.MerchantID, job.TerminalDBID, profile.Version-1).Scan(&previous.BusinessDescription, &niches, &channels, &previous.PrimaryGoal, &previous.Constraints); err == nil { _ = json.Unmarshal(niches, &previous.Niches) _ = json.Unmarshal(channels, &previous.SalesChannels) profile.PreviousInput = &previous } } var evidence MetricEvidence evidence.Values = map[string]float64{} evidence.EvidenceIDs = []string{"payment-health", "payment-volume", "operational-efficiency"} if err := r.pool.QueryRow(ctx, `SELECT category_title FROM merchants WHERE id=$1`, job.MerchantID).Scan(&evidence.Industry); err != nil { return Profile{}, MetricEvidence{}, fmt.Errorf("loading coach industry: %w", err) } eventTerms := append([]string{evidence.Industry}, profile.Niches...) eventTerms = append(eventTerms, profile.SalesChannels...) eventRows, eventErr := r.pool.Query(ctx, `SELECT e.event_key,e.title_fa,e.start_date,e.end_date, CASE WHEN historical.paid>0 THEN 'اثر این رویداد در سابقه پرداخت همین پایانه دیده شده است' WHEN e.merchant_categories ?| $1::text[] THEN 'مرتبط با صنعت و حوزه کسب‌وکار' WHEN e.behavioral_channels ?| $1::text[] THEN 'مرتبط با کانال فروش' ELSE 'رویداد تجاری یا عملیاتی اثرگذار' END FROM calendar_events e JOIN calendar_catalog_versions v ON v.id=e.catalog_version_id AND v.status='ACTIVE' LEFT JOIN LATERAL(SELECT count(*) paid FROM calendar_events prior JOIN payment_sessions ps ON ps.merchant_id=$2 AND ps.terminal_id=$3 AND ps.status IN('Paid','Verified') AND (ps.created_at AT TIME ZONE 'Asia/Tehran')::date BETWEEN prior.start_date AND prior.end_date WHERE prior.catalog_version_id=e.catalog_version_id AND prior.id<>e.id AND prior.end_date=e.start_date-400 AND ((e.recurrence_rule IS NOT NULL AND prior.recurrence_rule=e.recurrence_rule) OR regexp_replace(prior.event_key,'-[0-9]+$','')=regexp_replace(e.event_key,'-[0-9]+$',''))) historical ON true WHERE e.end_date+e.lag_days>=current_date AND e.start_date-e.lead_days<=current_date+90 AND (e.event_type IN('COMMERCIAL','SEASONAL','EDUCATION','ECONOMIC','EMERGENCY_CLOSURE') OR e.merchant_categories ?| $1::text[] OR e.behavioral_channels ?| $1::text[]) AND (NOT e.official_holiday OR e.relevance_bps>=7000 OR e.merchant_categories ?| $1::text[] OR e.behavioral_channels ?| $1::text[]) ORDER BY (e.merchant_categories ?| $1::text[]) DESC,(e.behavioral_channels ?| $1::text[]) DESC,(historical.paid>0) DESC,(e.relevance_bps::bigint*e.confidence_bps) DESC,e.start_date LIMIT 8`, eventTerms, job.MerchantID, job.TerminalDBID) if eventErr != nil { return Profile{}, MetricEvidence{}, fmt.Errorf("loading personalized coach events: %w", eventErr) } for eventRows.Next() { var item CoachEvent var start, end time.Time if err := eventRows.Scan(&item.EventID, &item.Title, &start, &end, &item.Reason); err != nil { eventRows.Close() return Profile{}, MetricEvidence{}, err } item.StartDate, item.EndDate = start.Format(time.DateOnly), end.Format(time.DateOnly) evidence.RelevantEvents = append(evidence.RelevantEvents, item) } eventRows.Close() if err := eventRows.Err(); err != nil { return Profile{}, MetricEvidence{}, err } var paid, amount, fees int64 var latency float64 var through *time.Time err = r.pool.QueryRow(ctx, ` WITH terminal_data AS ( SELECT max(created_at AT TIME ZONE 'Asia/Tehran')::date AS data_through FROM payment_sessions WHERE merchant_id=$1 AND terminal_id=$2 ) SELECT count(ps.id), count(ps.id) FILTER(WHERE ps.status 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)) FILTER(WHERE ps.id IS NOT NULL),0), terminal_data.data_through FROM terminal_data LEFT JOIN payment_sessions ps ON ps.merchant_id=$1 AND ps.terminal_id=$2 AND ps.created_at >= ((terminal_data.data_through-29)::timestamp AT TIME ZONE 'Asia/Tehran') AND ps.created_at < ((terminal_data.data_through+1)::timestamp AT TIME ZONE 'Asia/Tehran') 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 GROUP BY terminal_data.data_through`, job.MerchantID, job.TerminalDBID).Scan(&evidence.Sessions, &paid, &amount, &fees, &latency, &through) if err != nil { return Profile{}, MetricEvidence{}, fmt.Errorf("loading coach evidence: %w", err) } evidence.Values["paid_count"] = float64(paid) evidence.Values["paid_amount"] = float64(amount) evidence.Values["fee_amount"] = float64(fees) evidence.Values["avg_latency"] = latency if evidence.Sessions > 0 { evidence.Values["success_rate"] = float64(paid) / float64(evidence.Sessions) * 100 } if through != nil { evidence.DataThrough = through.Format(time.DateOnly) } evidence.RecentDecisions = []CoachMemoryItem{} decisionRows, decisionErr := r.pool.Query(ctx, `SELECT title,status,coalesce(decision_reason,'') FROM growth_coach_recommendations WHERE merchant_id=$1 AND terminal_id=$2 AND status IN('ACCEPTED','REJECTED') ORDER BY decided_at DESC LIMIT 30`, job.MerchantID, job.TerminalDBID) if decisionErr != nil { return Profile{}, MetricEvidence{}, decisionErr } for decisionRows.Next() { var item CoachMemoryItem if err := decisionRows.Scan(&item.Title, &item.Status, &item.Reason); err != nil { decisionRows.Close() return Profile{}, MetricEvidence{}, err } evidence.RecentDecisions = append(evidence.RecentDecisions, item) } decisionRows.Close() evidence.RecentOutcomes = []CoachMemoryItem{} outcomeRows, outcomeErr := r.pool.Query(ctx, `SELECT title,status,'' FROM golden_steps WHERE merchant_id=$1 AND terminal_id=$2 ORDER BY updated_at DESC LIMIT 30`, job.MerchantID, job.TerminalDBID) if outcomeErr != nil { return Profile{}, MetricEvidence{}, outcomeErr } for outcomeRows.Next() { var item CoachMemoryItem if err := outcomeRows.Scan(&item.Title, &item.Status, &item.Reason); err != nil { outcomeRows.Close() return Profile{}, MetricEvidence{}, err } evidence.RecentOutcomes = append(evidence.RecentOutcomes, item) } outcomeRows.Close() _ = r.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_coach_recommendations rec JOIN growth_coach_plans p ON p.id=rec.plan_id WHERE rec.merchant_id=$1 AND rec.terminal_id=$2 AND rec.status='ACCEPTED' AND p.cadence='MONTHLY' AND p.period_end>=current_date)`, job.MerchantID, job.TerminalDBID).Scan(&evidence.HasAcceptedMonthlyGoal) return profile, evidence, nil } func (r *Repository) Complete(ctx context.Context, job claimedJob, generation Generation, evidence MetricEvidence, usage ProviderUsage, model string) error { tx, err := r.pool.Begin(ctx) if err != nil { return err } defer func() { _ = tx.Rollback(ctx) }() var claimedID string err = tx.QueryRow(ctx, `SELECT id FROM growth_coach_jobs WHERE id=$1 AND lease_token=$2 AND status='RUNNING' FOR UPDATE`, job.JobID, job.LeaseToken).Scan(&claimedID) if errors.Is(err, pgx.ErrNoRows) { return ErrNotFound } if err != nil { return fmt.Errorf("validating coach job lease: %w", err) } parentID := any(nil) if job.Cadence == "WEEKLY" { var id string if tx.QueryRow(ctx, `SELECT id FROM growth_coach_plans WHERE merchant_id=$1 AND terminal_id=$2 AND cadence='MONTHLY' AND status='READY' AND period_start<=$3 AND period_end>=$3 ORDER BY created_at DESC LIMIT 1`, job.MerchantID, job.TerminalDBID, job.Start).Scan(&id) == nil { parentID = id } } planID := newID("plan_") brief := any(nil) if generation.DailyBrief != "" { brief = generation.DailyBrief } _, err = tx.Exec(ctx, `INSERT INTO growth_coach_plans(id,merchant_id,terminal_id,terminal_source_key,profile_id,parent_id,cadence,status,period_start,period_end,daily_brief,difficulty,prompt_version,schema_version,model,provider_request_id,input_tokens,output_tokens,latency_ms,provider_requests,estimated_cost,completed_at) SELECT $1,$2,$3,$4,$5,$6,$7,'READY',$8,$9,$10,difficulty,'coach_v1',$11,$12,nullif($13,''),$14,$15,$16,$17,nullif($18,''),now() FROM growth_coach_profiles WHERE id=$5`, planID, job.MerchantID, job.TerminalDBID, job.TerminalID, job.ProfileID, parentID, job.Cadence, job.Start, job.End, brief, schemaFor(job.Cadence), model, usage.ProviderRequestID, usage.InputTokens, usage.OutputTokens, usage.LatencyMS, usage.ProviderRequests, usage.EstimatedCost) if err != nil { return fmt.Errorf("creating coach plan: %w", err) } for index, suggestion := range generation.Suggestions { if index >= 5 { break } var baseline, target any sample := evidence.Sessions if suggestion.ActionType == "METRIC" { if suggestion.MetricID == nil || suggestion.ActionLevel == nil { return errors.New("growth coach: invalid metric suggestion") } value, ok := evidence.Values[*suggestion.MetricID] if !ok { return errors.New("growth coach: unsupported metric") } calculated, calcErr := CalculateTarget(*suggestion.MetricID, *suggestion.ActionLevel, value, sample) if calcErr != nil { return calcErr } baseline, target = value, calculated } evidenceJSON, _ := json.Marshal(suggestion.EvidenceIDs) subtasksJSON, _ := json.Marshal(suggestion.Subtasks) eventsJSON, _ := json.Marshal(suggestion.EventIDs) _, err = tx.Exec(ctx, `INSERT INTO growth_coach_recommendations(id,plan_id,merchant_id,terminal_id,rank,status,action_type,title,rationale,expected_impact,effort,evidence_ids,metric_id,metric_direction,action_level,baseline,target,sample_size,deadline,subtasks,event_ids) VALUES($1,$2,$3,$4,$5,'OFFERED',$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20)`, newID("rec_"), planID, job.MerchantID, job.TerminalDBID, index+1, suggestion.ActionType, suggestion.Title, suggestion.Rationale, suggestion.ExpectedImpact, suggestion.Effort, evidenceJSON, suggestion.MetricID, suggestion.MetricDirection, suggestion.ActionLevel, baseline, target, sample, job.End, subtasksJSON, eventsJSON) if err != nil { return err } } command, err := tx.Exec(ctx, `UPDATE growth_coach_jobs SET status='COMPLETED',lease_token=NULL,lease_expires_at=NULL,updated_at=now(),completed_at=now() WHERE id=$1 AND lease_token=$2 AND status='RUNNING'`, job.JobID, job.LeaseToken) if err != nil { return err } if command.RowsAffected() != 1 { return ErrNotFound } if err := audit(ctx, tx, job.MerchantID, job.TerminalDBID, nil, "JOB", job.JobID, "job.completed", map[string]any{"state": "COMPLETED", "planId": planID}); err != nil { return err } return tx.Commit(ctx) } func (r *Repository) Fail(ctx context.Context, job claimedJob, code string) error { command, err := r.pool.Exec(ctx, `UPDATE growth_coach_jobs SET status='FAILED',failure_code=$3,lease_token=NULL,lease_expires_at=NULL,updated_at=now(),completed_at=now() WHERE id=$1 AND lease_token=$2 AND status='RUNNING'`, job.JobID, job.LeaseToken, code) if err != nil { return err } if command.RowsAffected() != 1 { return ErrNotFound } payload, err := json.Marshal(map[string]any{"state": "FAILED", "failureCode": code}) if err != nil { return err } _, err = r.pool.Exec(ctx, `INSERT INTO growth_coach_events(merchant_id,terminal_id,entity_type,entity_id,event_type,payload) VALUES($1,$2,'JOB',$3,'job.failed',$4)`, job.MerchantID, job.TerminalDBID, job.JobID, payload) return err } func (r *Repository) Dashboard(ctx context.Context, merchantID int64, terminalKey string) (Dashboard, error) { if terminalKey == "" { return r.rollup(ctx, merchantID) } if _, err := r.terminal(ctx, merchantID, terminalKey); err != nil { return Dashboard{}, err } profile, err := r.Profile(ctx, merchantID, terminalKey) if errors.Is(err, ErrNotFound) { return Dashboard{Scope: "TERMINAL", TerminalID: &terminalKey, GenerationState: "ONBOARDING_REQUIRED", Plans: []Plan{}, Actions: []Action{}, DifficultyProposals: []DifficultyProposal{}}, nil } if err != nil { return Dashboard{}, err } plans, err := r.Plans(ctx, merchantID, terminalKey) if err != nil { return Dashboard{}, err } actions, err := r.Actions(ctx, merchantID, terminalKey) if err != nil { return Dashboard{}, err } state := "IDLE" var running bool _ = r.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_coach_jobs WHERE merchant_id=$1 AND terminal_source_key=$2 AND status IN('QUEUED','RUNNING'))`, merchantID, terminalKey).Scan(&running) if running { state = "GENERATING" } else { var failed bool _ = r.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_coach_jobs WHERE merchant_id=$1 AND terminal_source_key=$2 AND status='FAILED' AND completed_at>=coalesce((SELECT max(completed_at) FROM growth_coach_plans WHERE merchant_id=$1 AND terminal_source_key=$2),'-infinity'::timestamptz))`, merchantID, terminalKey).Scan(&failed) if failed { state = "UNAVAILABLE" } } var through *time.Time _ = r.pool.QueryRow(ctx, `SELECT max(created_at AT TIME ZONE 'Asia/Tehran')::date FROM payment_sessions WHERE merchant_id=$1`, merchantID).Scan(&through) var date *string if through != nil { value := through.Format(time.DateOnly) date = &value } proposals := []DifficultyProposal{} proposalRows, proposalErr := r.pool.Query(ctx, `SELECT id,from_level,to_level,reason,status FROM growth_coach_difficulty_proposals WHERE merchant_id=$1 AND terminal_id=(SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2) AND status='OFFERED' ORDER BY created_at DESC`, merchantID, terminalKey) if proposalErr != nil { return Dashboard{}, proposalErr } defer proposalRows.Close() for proposalRows.Next() { var proposal DifficultyProposal if err := proposalRows.Scan(&proposal.ProposalID, &proposal.FromLevel, &proposal.ToLevel, &proposal.Reason, &proposal.Status); err != nil { return Dashboard{}, err } proposals = append(proposals, proposal) } return Dashboard{Scope: "TERMINAL", TerminalID: &terminalKey, DataThrough: date, GenerationState: state, Profile: &profile, Plans: plans, Actions: actions, Terminals: []TerminalRollup{}, DifficultyProposals: proposals}, nil } func (r *Repository) rollup(ctx context.Context, merchantID int64) (Dashboard, error) { rows, err := r.pool.Query(ctx, `SELECT t.source_key,CASE WHEN p.id IS NULL THEN 'REQUIRED' ELSE 'COMPLETE' END,count(a.id) FILTER(WHERE a.status IN('NOT_STARTED','IN_PROGRESS')),count(a.id) FILTER(WHERE a.status IN('NOT_STARTED','IN_PROGRESS') AND a.deadlines.deadline ON CONFLICT DO NOTHING`, merchantID, terminalKey) if err != nil { return 0, err } return result.RowsAffected(), nil } func (r *Repository) DecideDeadlineProposal(ctx context.Context, merchantID, userID int64, proposalID, key, requestHash, decision string) (DeadlineProposal, error) { tx, err := r.pool.Begin(ctx) if err != nil { return DeadlineProposal{}, err } defer func() { _ = tx.Rollback(ctx) }() if err := lockMutation(ctx, tx, merchantID, key); err != nil { return DeadlineProposal{}, err } var cached DeadlineProposal if found, err := readMutation(ctx, tx, merchantID, key, requestHash, &cached); err != nil { return DeadlineProposal{}, err } else if found { return cached, nil } var item DeadlineProposal var current, proposed, accepted, goalDeadline time.Time err = tx.QueryRow(ctx, `SELECT p.id,p.goal_id,p.subtask_id,p.event_id,p.current_deadline,p.proposed_deadline,p.reason,p.status,p.decided_at,g.accepted_at::date,g.deadline FROM goal_deadline_proposals p JOIN monthly_goals g ON g.id=p.goal_id WHERE p.merchant_id=$1 AND p.id=$2 FOR UPDATE`, merchantID, proposalID).Scan(&item.ProposalID, &item.GoalID, &item.SubtaskID, &item.EventID, ¤t, &proposed, &item.Reason, &item.Status, &item.DecidedAt, &accepted, &goalDeadline) if errors.Is(err, pgx.ErrNoRows) { return DeadlineProposal{}, ErrNotFound } if err != nil { return DeadlineProposal{}, err } if item.Status != "OFFERED" || proposed.Before(accepted) || proposed.After(goalDeadline) { return DeadlineProposal{}, ErrInvalidTransition } next := "REJECTED" if decision == "ACCEPT" { next = "ACCEPTED" if _, err := tx.Exec(ctx, `UPDATE goal_subtasks SET deadline=$3,updated_at=now() WHERE merchant_id=$1 AND id=$2`, merchantID, item.SubtaskID, proposed); err != nil { return DeadlineProposal{}, err } if _, err := tx.Exec(ctx, `UPDATE goal_deadline_proposals SET status='EXPIRED',decided_at=now() WHERE merchant_id=$1 AND subtask_id=$2 AND id<>$3 AND status='OFFERED'`, merchantID, item.SubtaskID, item.ProposalID); err != nil { return DeadlineProposal{}, err } } if _, err := tx.Exec(ctx, `UPDATE goal_deadline_proposals SET status=$3,decided_by=$4,decided_at=now() WHERE merchant_id=$1 AND id=$2`, merchantID, proposalID, next, userID); err != nil { return DeadlineProposal{}, err } item.Status, item.CurrentDeadline, item.ProposedDeadline = next, current.Format(time.DateOnly), proposed.Format(time.DateOnly) now := time.Now().UTC() item.DecidedAt = &now if err := storeMutation(ctx, tx, merchantID, key, requestHash, item); err != nil { return DeadlineProposal{}, err } if err := tx.Commit(ctx); err != nil { return DeadlineProposal{}, err } return item, nil } func (r *Repository) goalSubtasks(ctx context.Context, merchantID int64, goalID string) ([]GoalSubtask, error) { rows, err := r.pool.Query(ctx, `SELECT id,position,title,status,deadline,event_id,completed_at FROM goal_subtasks WHERE merchant_id=$1 AND goal_id=$2 ORDER BY position`, merchantID, goalID) if err != nil { return nil, err } defer rows.Close() result := []GoalSubtask{} for rows.Next() { var item GoalSubtask var deadline time.Time if err := rows.Scan(&item.SubtaskID, &item.Position, &item.Title, &item.Status, &deadline, &item.EventID, &item.CompletedAt); err != nil { return nil, err } item.Deadline = deadline.Format(time.DateOnly) result = append(result, item) } return result, rows.Err() } func (r *Repository) TransitionSubtask(ctx context.Context, merchantID, userID int64, goalID, subtaskID, key, requestHash, transition string) (GoalSubtask, error) { tx, err := r.pool.Begin(ctx) if err != nil { return GoalSubtask{}, err } defer func() { _ = tx.Rollback(ctx) }() if err := lockMutation(ctx, tx, merchantID, key); err != nil { return GoalSubtask{}, err } var cached GoalSubtask if found, err := readMutation(ctx, tx, merchantID, key, requestHash, &cached); err != nil { return GoalSubtask{}, err } else if found { return cached, nil } var item GoalSubtask var deadline time.Time err = tx.QueryRow(ctx, `SELECT s.id,s.position,s.title,s.status,s.deadline,s.event_id,s.completed_at FROM goal_subtasks s JOIN monthly_goals g ON g.id=s.goal_id WHERE s.merchant_id=$1 AND s.goal_id=$2 AND s.id=$3 FOR UPDATE`, merchantID, goalID, subtaskID).Scan(&item.SubtaskID, &item.Position, &item.Title, &item.Status, &deadline, &item.EventID, &item.CompletedAt) if errors.Is(err, pgx.ErrNoRows) { return GoalSubtask{}, ErrNotFound } if err != nil { return GoalSubtask{}, err } next := "" if transition == "COMPLETE" && item.Status == "TODO" { next = "DONE" } if transition == "REOPEN" && item.Status == "DONE" { next = "TODO" } if next == "" { return GoalSubtask{}, ErrInvalidTransition } if _, err := tx.Exec(ctx, `UPDATE goal_subtasks SET status=$4::text,completed_by=CASE WHEN $4::text='DONE' THEN $5::bigint ELSE NULL::bigint END,completed_at=CASE WHEN $4::text='DONE' THEN now() ELSE NULL END,updated_at=now() WHERE merchant_id=$1 AND goal_id=$2 AND id=$3`, merchantID, goalID, subtaskID, next, userID); err != nil { return GoalSubtask{}, err } item.Status, item.Deadline = next, deadline.Format(time.DateOnly) if next == "DONE" { now := time.Now().UTC() item.CompletedAt = &now } else { item.CompletedAt = nil } if err := storeMutation(ctx, tx, merchantID, key, requestHash, item); err != nil { return GoalSubtask{}, err } if err := tx.Commit(ctx); err != nil { return GoalSubtask{}, err } return item, nil } func (r *Repository) DecideRecommendation(ctx context.Context, merchantID, userID int64, id, key, requestHash string, input DecisionInput) (*Action, error) { tx, err := r.pool.Begin(ctx) if err != nil { return nil, err } defer func() { _ = tx.Rollback(ctx) }() if err := lockMutation(ctx, tx, merchantID, key); err != nil { return nil, err } var cached *Action if found, err := readMutation(ctx, tx, merchantID, key, requestHash, &cached); err != nil { return nil, err } else if found { return cached, nil } var terminalID int64 var terminalKey, actionType, title, status, profileID, expectedImpact string var metricID, direction *string var baseline, target *float64 var sample int64 var deadline, periodStart, periodEnd time.Time var subtasksJSON, eventIDsJSON []byte err = tx.QueryRow(ctx, `SELECT r.terminal_id,t.source_key,r.action_type,r.title,r.status,r.metric_id,r.metric_direction,r.baseline::float8,r.target::float8,r.sample_size,r.deadline,p.profile_id,r.expected_impact,r.subtasks,r.event_ids,p.period_start,p.period_end FROM growth_coach_recommendations r JOIN terminals t ON t.id=r.terminal_id JOIN growth_coach_plans p ON p.id=r.plan_id WHERE r.merchant_id=$1 AND r.id=$2 FOR UPDATE`, merchantID, id).Scan(&terminalID, &terminalKey, &actionType, &title, &status, &metricID, &direction, &baseline, &target, &sample, &deadline, &profileID, &expectedImpact, &subtasksJSON, &eventIDsJSON, &periodStart, &periodEnd) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound } if err != nil { return nil, err } if status != "OFFERED" { return nil, ErrInvalidTransition } decision := input.Decision if decision != "ACCEPTED" && decision != "REJECTED" { return nil, ErrInvalidTransition } if decision == "ACCEPTED" && actionType == "METRIC" { if metricID == nil || direction == nil || baseline == nil || target == nil || *baseline == *target { return nil, ErrInvalidTransition } var generated []GenerationSubtask if err := json.Unmarshal(subtasksJSON, &generated); err != nil || len(generated) < 3 || len(generated) > 7 { return nil, ErrInvalidTransition } if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, fmt.Sprintf("monthly-goals:%d", terminalID)); err != nil { return nil, err } var active int if err := tx.QueryRow(ctx, `SELECT count(*) FROM monthly_goals WHERE merchant_id=$1 AND terminal_id=$2 AND status='ACTIVE'`, merchantID, terminalID).Scan(&active); err != nil { return nil, err } if active >= 3 { return nil, ErrInvalidTransition } } _, err = tx.Exec(ctx, `UPDATE growth_coach_recommendations SET status=$3,decision_reason=nullif($4,''),decision_note=nullif($5,''),decided_by=$6,decided_at=now() WHERE merchant_id=$1 AND id=$2`, merchantID, id, decision, input.Reason, input.Note, userID) if err != nil { return nil, err } if err := audit(ctx, tx, merchantID, terminalID, &userID, "RECOMMENDATION", id, "recommendation.decided", map[string]any{"decision": decision, "reason": input.Reason}); err != nil { return nil, err } if decision == "REJECTED" { if err := storeMutation(ctx, tx, merchantID, key, requestHash, cached); err != nil { return nil, err } if err = tx.Commit(ctx); err != nil { return nil, err } return nil, nil } action := Action{ActionID: newID("step_"), RecommendationID: id, TerminalID: terminalKey, ActionType: actionType, Title: title, Status: "NOT_STARTED", MetricID: metricID, MetricDirection: direction, Baseline: baseline, Target: target, SampleSize: sample, Deadline: deadline.Format(time.DateOnly), UpdatedAt: time.Now().UTC()} err = tx.QueryRow(ctx, `INSERT INTO golden_steps(id,recommendation_id,merchant_id,terminal_id,terminal_source_key,action_type,title,status,metric_id,metric_direction,baseline,target,current_value,sample_size,deadline,actor_user_id) VALUES($1,$2,$3,$4,$5,$6,$7,'NOT_STARTED',$8,$9,$10,$11,$10,$12,$13,$14) RETURNING updated_at`, action.ActionID, id, merchantID, terminalID, terminalKey, actionType, title, metricID, direction, baseline, target, sample, deadline, userID).Scan(&action.UpdatedAt) if err != nil { return nil, err } if actionType == "METRIC" && metricID != nil && direction != nil && baseline != nil && target != nil && *baseline != *target { var watermark string if err := tx.QueryRow(ctx, `SELECT coalesce((SELECT source_sha256 FROM import_runs WHERE status='COMPLETED' ORDER BY completed_at DESC LIMIT 1),'no-data')`).Scan(&watermark); err != nil { return nil, err } goalID := newID("goal_") if _, err := tx.Exec(ctx, `INSERT INTO monthly_goals(id,recommendation_id,merchant_id,terminal_id,terminal_source_key,profile_id,title,expected_impact,status,metric_id,metric_direction,baseline,target,current_value,sample_size,baseline_watermark,progress_bps,event_ids,deadline,actor_user_id) VALUES($1,$2,$3,$4,$5,$6,$7,$8,'ACTIVE',$9,$10,$11,$12,$11,$13,$14,0,$15,$16,$17)`, goalID, id, merchantID, terminalID, terminalKey, profileID, title, expectedImpact, *metricID, *direction, *baseline, *target, sample, watermark, eventIDsJSON, deadline, userID); err != nil { return nil, err } var generated []GenerationSubtask if err := json.Unmarshal(subtasksJSON, &generated); err != nil || len(generated) < 3 || len(generated) > 7 { return nil, errors.New("growth coach: accepted goal has invalid subtasks") } for index, item := range generated { subtaskDeadline := periodEnd if item.Deadline != nil { if parsed, parseErr := time.Parse(time.DateOnly, *item.Deadline); parseErr == nil && !parsed.Before(periodStart) && !parsed.After(periodEnd) { subtaskDeadline = parsed } } if _, err := tx.Exec(ctx, `INSERT INTO goal_subtasks(id,goal_id,merchant_id,position,title,status,deadline,event_id) VALUES($1,$2,$3,$4,$5,'TODO',$6,$7)`, newID("subtask_"), goalID, merchantID, index+1, item.Title, subtaskDeadline, item.EventID); err != nil { return nil, err } } } if err := storeMutation(ctx, tx, merchantID, key, requestHash, &action); err != nil { return nil, err } if err = tx.Commit(ctx); err != nil { return nil, err } return &action, nil } func (r *Repository) Transition(ctx context.Context, merchantID, userID int64, id, key, requestHash, transition string) (Action, error) { tx, err := r.pool.Begin(ctx) if err != nil { return Action{}, err } defer func() { _ = tx.Rollback(ctx) }() if err := lockMutation(ctx, tx, merchantID, key); err != nil { return Action{}, err } var cached Action if found, err := readMutation(ctx, tx, merchantID, key, requestHash, &cached); err != nil { return Action{}, err } else if found { return cached, nil } var action Action var terminalDBID int64 var deadline time.Time err = tx.QueryRow(ctx, `SELECT id,recommendation_id,terminal_id,terminal_source_key,action_type,title,status,metric_id,metric_direction,baseline::float8,target::float8,current_value::float8,sample_size,deadline,updated_at FROM golden_steps WHERE merchant_id=$1 AND id=$2 FOR UPDATE`, merchantID, id).Scan(&action.ActionID, &action.RecommendationID, &terminalDBID, &action.TerminalID, &action.ActionType, &action.Title, &action.Status, &action.MetricID, &action.MetricDirection, &action.Baseline, &action.Target, &action.CurrentValue, &action.SampleSize, &deadline, &action.UpdatedAt) if errors.Is(err, pgx.ErrNoRows) { return Action{}, ErrNotFound } if err != nil { return Action{}, err } next, err := ValidTransition(action.ActionType, action.Status, transition) if err != nil { return Action{}, err } if err := audit(ctx, tx, merchantID, terminalDBID, &userID, "ACTION", id, "action.transitioned", map[string]any{"transition": transition, "status": next}); err != nil { return Action{}, err } _, err = tx.Exec(ctx, `UPDATE golden_steps SET status=$3,started_at=CASE WHEN $3='IN_PROGRESS' THEN coalesce(started_at,now()) ELSE started_at END,completed_at=CASE WHEN $3='COMPLETED' THEN now() ELSE completed_at END,actor_user_id=$4,updated_at=now() WHERE merchant_id=$1 AND id=$2`, merchantID, id, next, userID) if err != nil { return Action{}, err } action.Status = next action.Deadline = deadline.Format(time.DateOnly) action.UpdatedAt = time.Now().UTC() if err := storeMutation(ctx, tx, merchantID, key, requestHash, action); err != nil { return Action{}, err } if err = tx.Commit(ctx); err != nil { return Action{}, err } return action, nil } func (r *Repository) GetJob(ctx context.Context, merchantID int64, id string) (Job, error) { var job Job err := r.pool.QueryRow(ctx, `SELECT id,status,created_at,updated_at FROM growth_coach_jobs WHERE merchant_id=$1 AND id=$2`, merchantID, id).Scan(&job.JobID, &job.State, &job.CreatedAt, &job.UpdatedAt) if errors.Is(err, pgx.ErrNoRows) { return Job{}, ErrNotFound } job.EventsURL = "/api/v1/growth-coach/refreshes/" + id + "/events" return job, err } func (r *Repository) EventsAfter(ctx context.Context, merchantID int64, id string, after int64) ([]Event, error) { if _, err := r.GetJob(ctx, merchantID, id); err != nil { return nil, err } rows, err := r.pool.Query(ctx, `SELECT id,event_type,payload,occurred_at FROM growth_coach_events WHERE merchant_id=$1 AND entity_type='JOB' AND entity_id=$2 AND id>$3 AND occurred_at>=now()-interval '1 hour' ORDER BY id LIMIT 100`, merchantID, id, after) if err != nil { return nil, err } defer rows.Close() result := []Event{} for rows.Next() { var event Event if err := rows.Scan(&event.ID, &event.Type, &event.Payload, &event.OccurredAt); err != nil { return nil, err } result = append(result, event) } return result, rows.Err() } func (r *Repository) EvaluateMetrics(ctx context.Context, watermark string) (int, error) { rows, err := r.pool.Query(ctx, `SELECT id,merchant_id,terminal_id,metric_id,metric_direction,target::float8,deadline FROM golden_steps WHERE action_type='METRIC' AND status='IN_PROGRESS' AND baseline IS NOT NULL AND target IS NOT NULL AND target<>baseline AND NOT EXISTS(SELECT 1 FROM growth_coach_progress_snapshots p WHERE p.action_id=golden_steps.id AND p.watermark=$1) ORDER BY id`, watermark) if err != nil { return 0, err } defer rows.Close() type pending struct { id string merchantID, terminalID int64 metric, direction string target float64 deadline time.Time } items := []pending{} for rows.Next() { var item pending if err := rows.Scan(&item.id, &item.merchantID, &item.terminalID, &item.metric, &item.direction, &item.target, &item.deadline); err != nil { return 0, err } items = append(items, item) } if err := rows.Err(); err != nil { return 0, err } completed := 0 for _, item := range items { var sessions, paid, amount, fees int64 var latency float64 err := r.pool.QueryRow(ctx, `SELECT count(*),count(*) FILTER(WHERE ps.status 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) 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 ps.terminal_id=$2 AND ps.created_at>=now()-interval '30 days'`, item.merchantID, item.terminalID).Scan(&sessions, &paid, &amount, &fees, &latency) if err != nil { return completed, err } values := map[string]float64{"paid_count": float64(paid), "paid_amount": float64(amount), "fee_amount": float64(fees), "avg_latency": latency} if sessions > 0 { values["success_rate"] = float64(paid) / float64(sessions) * 100 } value, ok := values[item.metric] if !ok { continue } tx, err := r.pool.Begin(ctx) if err != nil { return completed, err } _, err = tx.Exec(ctx, `INSERT INTO growth_coach_progress_snapshots(action_id,watermark,value,sample_size) VALUES($1,$2,$3,$4) ON CONFLICT(action_id,watermark) DO UPDATE SET value=excluded.value,sample_size=excluded.sample_size,evaluated_at=now()`, item.id, watermark, value, sessions) if err == nil { reached := MetricTargetReached(item.direction, value, item.target) status := "IN_PROGRESS" if reached { status = "COMPLETED" completed++ } else if time.Now().After(item.deadline.AddDate(0, 0, 1)) { status = "MISSED" } _, err = tx.Exec(ctx, `UPDATE golden_steps SET current_value=$2,sample_size=$3,status=$4,started_at=coalesce(started_at,now()),completed_at=CASE WHEN $4='COMPLETED' THEN now() ELSE completed_at END,updated_at=now() WHERE id=$1`, item.id, value, sessions, status) } if err != nil { _ = tx.Rollback(ctx) return completed, err } if err = tx.Commit(ctx); err != nil { return completed, err } } goalCompleted, err := r.evaluateGoals(ctx, watermark) return completed + goalCompleted, err } func (r *Repository) evaluateGoals(ctx context.Context, watermark string) (int, error) { rows, err := r.pool.Query(ctx, `SELECT id,merchant_id,terminal_id,metric_id,metric_direction,baseline::float8,target::float8,deadline FROM monthly_goals WHERE status='ACTIVE' AND baseline_watermark IS DISTINCT FROM $1 AND latest_watermark IS DISTINCT FROM $1 ORDER BY id`, watermark) if err != nil { return 0, err } type pendingGoal struct { id string merchantID, terminalID int64 metric, direction string baseline, target float64 deadline time.Time } items := []pendingGoal{} for rows.Next() { var item pendingGoal if err := rows.Scan(&item.id, &item.merchantID, &item.terminalID, &item.metric, &item.direction, &item.baseline, &item.target, &item.deadline); err != nil { rows.Close() return 0, err } items = append(items, item) } rows.Close() if err := rows.Err(); err != nil { return 0, err } completed := 0 for _, item := range items { var sessions, paid, amount, fees int64 var latency float64 err := r.pool.QueryRow(ctx, `WITH latest AS (SELECT max(created_at) through FROM payment_sessions WHERE merchant_id=$1 AND terminal_id=$2) SELECT count(*),count(*) FILTER(WHERE ps.status 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) FROM payment_sessions ps CROSS JOIN latest 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 ps.terminal_id=$2 AND ps.created_at>=latest.through-interval '29 days'`, item.merchantID, item.terminalID).Scan(&sessions, &paid, &amount, &fees, &latency) if err != nil { return completed, err } if sessions < 100 { continue } values := map[string]float64{"paid_count": float64(paid), "paid_amount": float64(amount), "fee_amount": float64(fees), "avg_latency": latency, "success_rate": float64(paid) / float64(sessions) * 100} current, ok := values[item.metric] if !ok { continue } progress, actionable := GoalProgress(item.direction, item.baseline, item.target, current) if !actionable { continue } status := "ACTIVE" if MetricTargetReached(item.direction, current, item.target) { status = "COMPLETED" completed++ } else if time.Now().After(item.deadline.AddDate(0, 0, 1)) { status = "MISSED" } if _, err := r.pool.Exec(ctx, `UPDATE monthly_goals SET current_value=$2,sample_size=$3,latest_watermark=$4,progress_bps=$5,status=$6,completed_at=CASE WHEN $6='COMPLETED' THEN now() ELSE NULL END,updated_at=now() WHERE id=$1 AND status='ACTIVE'`, item.id, current, sessions, watermark, int(math.Round(progress*100)), status); err != nil { return completed, err } } return completed, nil } func (r *Repository) CurrentWatermark(ctx context.Context) (string, error) { var value string err := r.pool.QueryRow(ctx, `SELECT source_sha256 FROM import_runs WHERE status='COMPLETED' ORDER BY completed_at DESC LIMIT 1`).Scan(&value) if errors.Is(err, pgx.ErrNoRows) { return "no-data", nil } return value, err } 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 growth_coach_plans 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) DecideDifficulty(ctx context.Context, merchantID, userID int64, id, key, requestHash, decision string) (DifficultyProposal, error) { tx, err := r.pool.Begin(ctx) if err != nil { return DifficultyProposal{}, err } defer func() { _ = tx.Rollback(ctx) }() if err := lockMutation(ctx, tx, merchantID, key); err != nil { return DifficultyProposal{}, err } var cached DifficultyProposal if found, err := readMutation(ctx, tx, merchantID, key, requestHash, &cached); err != nil { return DifficultyProposal{}, err } else if found { return cached, nil } var proposal DifficultyProposal var terminalID int64 err = tx.QueryRow(ctx, `SELECT id,terminal_id,from_level,to_level,reason,status FROM growth_coach_difficulty_proposals WHERE merchant_id=$1 AND id=$2 FOR UPDATE`, merchantID, id).Scan(&proposal.ProposalID, &terminalID, &proposal.FromLevel, &proposal.ToLevel, &proposal.Reason, &proposal.Status) if errors.Is(err, pgx.ErrNoRows) { return DifficultyProposal{}, ErrNotFound } if err != nil { return DifficultyProposal{}, err } if proposal.Status != "OFFERED" || (decision != "ACCEPTED" && decision != "REJECTED") { return DifficultyProposal{}, ErrInvalidTransition } _, err = tx.Exec(ctx, `UPDATE growth_coach_difficulty_proposals SET status=$3,decided_by=$4,decided_at=now() WHERE merchant_id=$1 AND id=$2`, merchantID, id, decision, userID) if err != nil { return DifficultyProposal{}, err } if decision == "ACCEPTED" { _, err = tx.Exec(ctx, `UPDATE growth_coach_profiles SET difficulty=$3 WHERE merchant_id=$1 AND terminal_id=$2 AND is_current`, merchantID, terminalID, proposal.ToLevel) if err != nil { return DifficultyProposal{}, err } } proposal.Status = decision if err := audit(ctx, tx, merchantID, terminalID, &userID, "DIFFICULTY", id, "difficulty.decided", map[string]any{"decision": decision, "toLevel": proposal.ToLevel}); err != nil { return DifficultyProposal{}, err } if err := storeMutation(ctx, tx, merchantID, key, requestHash, proposal); err != nil { return DifficultyProposal{}, err } if err = tx.Commit(ctx); err != nil { return DifficultyProposal{}, err } return proposal, nil } func (r *Repository) EvaluateDifficulty(ctx context.Context) (int, error) { profiles, err := r.CurrentProfiles(ctx) if err != nil { return 0, err } created := 0 for _, item := range profiles { var offered bool if err := r.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_coach_difficulty_proposals WHERE merchant_id=$1 AND terminal_id=$2 AND status='OFFERED')`, item.MerchantID, item.TerminalDBID).Scan(&offered); err != nil { return created, err } if offered { continue } history := DifficultyHistory{Current: item.Profile.Difficulty} _ = r.pool.QueryRow(ctx, `SELECT count(*) FROM growth_coach_recommendations WHERE merchant_id=$1 AND terminal_id=$2 AND status='REJECTED' AND decision_reason='خیلی سخت' AND decided_at>=now()-interval '90 days'`, item.MerchantID, item.TerminalDBID).Scan(&history.HardRejections) rows, queryErr := r.pool.Query(ctx, `SELECT coalesce(count(s.id) FILTER(WHERE s.status='COMPLETED' AND s.completed_at::date<=p.period_end)::float/nullif(count(s.id),0),0) FROM growth_coach_plans p LEFT JOIN growth_coach_recommendations rec ON rec.plan_id=p.id AND rec.status='ACCEPTED' LEFT JOIN golden_steps s ON s.recommendation_id=rec.id WHERE p.merchant_id=$1 AND p.terminal_id=$2 AND p.cadence='WEEKLY' AND p.period_end'COMPLETED')`, item.MerchantID, item.TerminalDBID).Scan(&history.MonthlyCompletedDaysEarly) to, reason, ok := ProposeDifficulty(history) if !ok { continue } _, err = r.pool.Exec(ctx, `INSERT INTO growth_coach_difficulty_proposals(id,merchant_id,terminal_id,from_level,to_level,reason,status) VALUES($1,$2,$3,$4,$5,$6,'OFFERED')`, newID("difficulty_"), item.MerchantID, item.TerminalDBID, item.Profile.Difficulty, to, reason) if err != nil { return created, err } created++ } return created, nil } func schemaFor(cadence string) string { switch cadence { case "MONTHLY": return "monthly_plan_v1" case "WEEKLY": return "weekly_plan_v1" default: return "daily_brief_v1" } } func newID(prefix string) string { data := make([]byte, 12) if _, err := rand.Read(data); err != nil { panic("crypto/rand unavailable") } return prefix + hex.EncodeToString(data) } func audit(ctx context.Context, tx pgx.Tx, merchantID, terminalID int64, userID *int64, entityType, entityID, eventType string, payload any) error { encoded, _ := json.Marshal(payload) _, err := tx.Exec(ctx, `INSERT INTO growth_coach_events(merchant_id,terminal_id,actor_user_id,entity_type,entity_id,event_type,payload) VALUES($1,$2,$3,$4,$5,$6,$7)`, merchantID, terminalID, userID, entityType, entityID, eventType, encoded) return err } func lockMutation(ctx context.Context, tx pgx.Tx, merchantID int64, key string) error { _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, fmt.Sprintf("coach:%d:%s", merchantID, key)) return err } func readMutation(ctx context.Context, tx pgx.Tx, merchantID int64, key, requestHash string, target any) (bool, error) { var existingHash string var response []byte err := tx.QueryRow(ctx, `SELECT request_hash,response FROM growth_coach_idempotency WHERE merchant_id=$1 AND idempotency_key=$2`, merchantID, key).Scan(&existingHash, &response) if errors.Is(err, pgx.ErrNoRows) { return false, nil } if err != nil { return false, err } if existingHash != requestHash { return false, ErrIdempotencyConflict } if err := json.Unmarshal(response, target); err != nil { return false, err } return true, nil } func storeMutation(ctx context.Context, tx pgx.Tx, merchantID int64, key, requestHash string, response any) error { encoded, err := json.Marshal(response) if err != nil { return err } _, err = tx.Exec(ctx, `INSERT INTO growth_coach_idempotency(merchant_id,idempotency_key,request_hash,response) VALUES($1,$2,$3,$4)`, merchantID, key, requestHash, encoded) return err }