1314 lines
66 KiB
Go
1314 lines
66 KiB
Go
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<now()`)
|
|
var job claimedJob
|
|
err = tx.QueryRow(ctx, `SELECT id,merchant_id,terminal_id,terminal_source_key,profile_id,cadence,period_start,period_end,created_at,updated_at FROM growth_coach_jobs WHERE id=(SELECT id FROM growth_coach_jobs WHERE status='QUEUED' AND scheduled_at<=now() ORDER BY scheduled_at,created_at FOR UPDATE SKIP LOCKED LIMIT 1)`).Scan(&job.JobID, &job.MerchantID, &job.TerminalDBID, &job.TerminalID, &job.ProfileID, &job.Cadence, &job.Start, &job.End, &job.CreatedAt, &job.UpdatedAt)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return claimedJob{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return claimedJob{}, err
|
|
}
|
|
job.LeaseToken, job.State = newID("lease_"), "RUNNING"
|
|
job.EventsURL = "/api/v1/growth-coach/refreshes/" + job.JobID + "/events"
|
|
if _, err = tx.Exec(ctx, `UPDATE growth_coach_jobs SET status='RUNNING',attempts=attempts+1,lease_token=$2,lease_expires_at=now()+interval '3 minutes',updated_at=now() WHERE id=$1`, job.JobID, job.LeaseToken); err != nil {
|
|
return claimedJob{}, err
|
|
}
|
|
if err = tx.Commit(ctx); err != nil {
|
|
return claimedJob{}, 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.running',$4)`, job.MerchantID, job.TerminalDBID, job.JobID, []byte(`{"state":"RUNNING"}`))
|
|
return job, nil
|
|
}
|
|
|
|
func (r *Repository) Evidence(ctx context.Context, job claimedJob) (Profile, MetricEvidence, error) {
|
|
profile, err := r.Profile(ctx, job.MerchantID, job.TerminalID)
|
|
if err != nil {
|
|
return Profile{}, MetricEvidence{}, err
|
|
}
|
|
if profile.Version > 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 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.deadline<current_date),coalesce(count(a.id) FILTER(WHERE a.status='COMPLETED')::float/nullif(count(a.id),0),0) FROM terminals t LEFT JOIN growth_coach_profiles p ON p.merchant_id=t.merchant_id AND p.terminal_id=t.id AND p.is_current LEFT JOIN golden_steps a ON a.merchant_id=t.merchant_id AND a.terminal_id=t.id WHERE t.merchant_id=$1 GROUP BY t.id,t.source_key,p.id ORDER BY t.source_key`, merchantID)
|
|
if err != nil {
|
|
return Dashboard{}, err
|
|
}
|
|
defer rows.Close()
|
|
result := Dashboard{Scope: "ALL_TERMINALS", GenerationState: "IDLE", Plans: []Plan{}, Actions: []Action{}, Terminals: []TerminalRollup{}, DifficultyProposals: []DifficultyProposal{}}
|
|
for rows.Next() {
|
|
var item TerminalRollup
|
|
if err := rows.Scan(&item.TerminalID, &item.OnboardingStatus, &item.ActiveActions, &item.OverdueActions, &item.CompletionRate); err != nil {
|
|
return Dashboard{}, err
|
|
}
|
|
result.Terminals = append(result.Terminals, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (r *Repository) Plans(ctx context.Context, merchantID int64, terminalKey string) ([]Plan, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT id,parent_id,cadence,status,period_start,period_end,daily_brief,difficulty,created_at FROM growth_coach_plans WHERE merchant_id=$1 AND ($2='' OR terminal_source_key=$2) ORDER BY period_start DESC,created_at DESC LIMIT 30`, merchantID, terminalKey)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
plans := []Plan{}
|
|
for rows.Next() {
|
|
var p Plan
|
|
var start, end time.Time
|
|
if err := rows.Scan(&p.PlanID, &p.ParentID, &p.Cadence, &p.Status, &start, &end, &p.DailyBrief, &p.Difficulty, &p.GeneratedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
p.PeriodStart, p.PeriodEnd = start.Format(time.DateOnly), end.Format(time.DateOnly)
|
|
p.Suggestions = []Recommendation{}
|
|
plans = append(plans, p)
|
|
}
|
|
for i := range plans {
|
|
recs, err := r.recommendations(ctx, merchantID, plans[i].PlanID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
plans[i].Suggestions = recs
|
|
}
|
|
return plans, rows.Err()
|
|
}
|
|
|
|
func (r *Repository) recommendations(ctx context.Context, merchantID int64, planID string) ([]Recommendation, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT id,plan_id,rank,status,action_type,title,rationale,expected_impact,effort,evidence_ids,metric_id,metric_direction,action_level,baseline::float8,target::float8,sample_size,deadline,subtasks,event_ids FROM growth_coach_recommendations WHERE merchant_id=$1 AND plan_id=$2 ORDER BY rank`, merchantID, planID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := []Recommendation{}
|
|
for rows.Next() {
|
|
var item Recommendation
|
|
var ids, subtasks, events []byte
|
|
var deadline time.Time
|
|
if err := rows.Scan(&item.RecommendationID, &item.PlanID, &item.Rank, &item.Status, &item.ActionType, &item.Title, &item.Rationale, &item.ExpectedImpact, &item.Effort, &ids, &item.MetricID, &item.MetricDirection, &item.ActionLevel, &item.Baseline, &item.Target, &item.SampleSize, &deadline, &subtasks, &events); err != nil {
|
|
return nil, err
|
|
}
|
|
_ = json.Unmarshal(ids, &item.EvidenceIDs)
|
|
_ = json.Unmarshal(subtasks, &item.Subtasks)
|
|
_ = json.Unmarshal(events, &item.EventIDs)
|
|
item.Deadline = deadline.Format(time.DateOnly)
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (r *Repository) Actions(ctx context.Context, merchantID int64, terminalKey string) ([]Action, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT a.id,a.recommendation_id,a.terminal_source_key,a.action_type,a.title,a.status,a.metric_id,a.metric_direction,a.baseline::float8,a.target::float8,a.current_value::float8,a.sample_size,a.deadline,a.updated_at FROM golden_steps a WHERE a.merchant_id=$1 AND ($2='' OR a.terminal_source_key=$2) ORDER BY CASE WHEN a.status IN('NOT_STARTED','IN_PROGRESS') THEN 0 ELSE 1 END,a.deadline,a.created_at DESC LIMIT 100`, merchantID, terminalKey)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := []Action{}
|
|
for rows.Next() {
|
|
var item Action
|
|
var deadline time.Time
|
|
if err := rows.Scan(&item.ActionID, &item.RecommendationID, &item.TerminalID, &item.ActionType, &item.Title, &item.Status, &item.MetricID, &item.MetricDirection, &item.Baseline, &item.Target, &item.CurrentValue, &item.SampleSize, &deadline, &item.UpdatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
item.Deadline = deadline.Format(time.DateOnly)
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (r *Repository) Goals(ctx context.Context, merchantID int64, terminalKey string) ([]MonthlyGoal, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT id,recommendation_id,terminal_source_key,title,expected_impact,status,metric_id,metric_direction,baseline::float8,target::float8,current_value::float8,progress_bps,sample_size,deadline,event_ids,updated_at FROM monthly_goals WHERE merchant_id=$1 AND ($2='' OR terminal_source_key=$2) ORDER BY CASE WHEN status='ACTIVE' THEN 0 ELSE 1 END,deadline,accepted_at DESC LIMIT 30`, merchantID, terminalKey)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
goals := []MonthlyGoal{}
|
|
for rows.Next() {
|
|
var goal MonthlyGoal
|
|
var deadline time.Time
|
|
var eventIDs []byte
|
|
var progressBPS int
|
|
if err := rows.Scan(&goal.GoalID, &goal.RecommendationID, &goal.TerminalID, &goal.Title, &goal.ExpectedImpact, &goal.Status, &goal.MetricID, &goal.MetricDirection, &goal.Baseline, &goal.Target, &goal.CurrentValue, &progressBPS, &goal.SampleSize, &deadline, &eventIDs, &goal.UpdatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
goal.ProgressPercent = float64(progressBPS) / 100
|
|
goal.Deadline = deadline.Format(time.DateOnly)
|
|
_ = json.Unmarshal(eventIDs, &goal.EventIDs)
|
|
goal.Subtasks = []GoalSubtask{}
|
|
goals = append(goals, goal)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
for index := range goals {
|
|
subtasks, err := r.goalSubtasks(ctx, merchantID, goals[index].GoalID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
goals[index].Subtasks = subtasks
|
|
proposals, err := r.goalDeadlineProposals(ctx, merchantID, goals[index].GoalID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
goals[index].DeadlineProposals = proposals
|
|
}
|
|
return goals, nil
|
|
}
|
|
|
|
func (r *Repository) goalDeadlineProposals(ctx context.Context, merchantID int64, goalID string) ([]DeadlineProposal, error) {
|
|
rows, err := r.pool.Query(ctx, `SELECT id,goal_id,subtask_id,event_id,current_deadline,proposed_deadline,reason,status,decided_at FROM goal_deadline_proposals WHERE merchant_id=$1 AND goal_id=$2 ORDER BY CASE WHEN status='OFFERED' THEN 0 ELSE 1 END,created_at DESC`, merchantID, goalID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := []DeadlineProposal{}
|
|
for rows.Next() {
|
|
var item DeadlineProposal
|
|
var current, proposed time.Time
|
|
if err := rows.Scan(&item.ProposalID, &item.GoalID, &item.SubtaskID, &item.EventID, ¤t, &proposed, &item.Reason, &item.Status, &item.DecidedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
item.CurrentDeadline = current.Format(time.DateOnly)
|
|
item.ProposedDeadline = proposed.Format(time.DateOnly)
|
|
result = append(result, item)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (r *Repository) ProposeEventDeadlines(ctx context.Context, merchantID int64, terminalKey string) (int64, error) {
|
|
result, err := r.pool.Exec(ctx, `INSERT INTO goal_deadline_proposals(id,goal_id,subtask_id,merchant_id,event_id,current_deadline,proposed_deadline,reason,status)
|
|
SELECT 'deadline_'||substr(md5(g.id||s.id||e.event_key||greatest(g.accepted_at::date,e.start_date-e.lead_days)::text),1,24),g.id,s.id,g.merchant_id,e.event_key,s.deadline,
|
|
greatest(g.accepted_at::date,e.start_date-e.lead_days),
|
|
'هماهنگی زمان انجام کار با رویداد اثرگذار «'||e.title_fa||'»','OFFERED'
|
|
FROM monthly_goals g JOIN goal_subtasks s ON s.goal_id=g.id
|
|
JOIN calendar_catalog_versions v ON v.status='ACTIVE'
|
|
JOIN calendar_events e ON e.catalog_version_id=v.id AND e.event_key=s.event_id
|
|
WHERE g.merchant_id=$1 AND g.status='ACTIVE' AND ($2='' OR g.terminal_source_key=$2)
|
|
AND greatest(g.accepted_at::date,e.start_date-e.lead_days)<=g.deadline
|
|
AND greatest(g.accepted_at::date,e.start_date-e.lead_days)<>s.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<current_date GROUP BY p.id,p.period_end ORDER BY p.period_end DESC LIMIT 2`, item.MerchantID, item.TerminalDBID)
|
|
if queryErr != nil {
|
|
return created, queryErr
|
|
}
|
|
for rows.Next() {
|
|
var rate float64
|
|
if rows.Scan(&rate) == nil {
|
|
history.RecentWeeklyRates = append(history.RecentWeeklyRates, rate)
|
|
history.RecentPeriodRates = append(history.RecentPeriodRates, rate)
|
|
}
|
|
}
|
|
rows.Close()
|
|
if len(history.RecentWeeklyRates) == 2 {
|
|
history.RecentWeeklyRates[0], history.RecentWeeklyRates[1] = history.RecentWeeklyRates[1], history.RecentWeeklyRates[0]
|
|
history.RecentPeriodRates[0], history.RecentPeriodRates[1] = history.RecentPeriodRates[1], history.RecentPeriodRates[0]
|
|
}
|
|
_ = r.pool.QueryRow(ctx, `SELECT coalesce(max(least(30,p.period_end::date-s.completed_at::date)),0) FROM growth_coach_plans p JOIN growth_coach_recommendations rec ON rec.plan_id=p.id AND rec.status='ACCEPTED' JOIN golden_steps s ON s.recommendation_id=rec.id WHERE p.merchant_id=$1 AND p.terminal_id=$2 AND p.cadence='MONTHLY' AND s.status='COMPLETED' AND NOT EXISTS(SELECT 1 FROM growth_coach_recommendations r2 LEFT JOIN golden_steps s2 ON s2.recommendation_id=r2.id WHERE r2.plan_id=p.id AND r2.status='ACCEPTED' AND coalesce(s2.status,'')<>'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
|
|
}
|