package growthreport

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var (
	ErrNotFound            = errors.New("growth report: not found")
	ErrUnknownTerminal     = errors.New("growth report: unknown terminal")
	ErrIdempotencyConflict = errors.New("growth report: idempotency conflict")
	ErrRefreshLimited      = errors.New("growth report: refresh limited")
)

type Repository struct{ pool *pgxpool.Pool }

func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} }

type CreateRecord struct {
	ID                string
	MerchantID        int64
	TerminalID        string
	Fingerprint       string
	DatasetWatermark  string
	ForecastStartDate *time.Time
	CatalogVersion    string
	AnalyticsVersion  string
	PromptVersion     string
	SchemaVersion     string
	Model             string
	ManualRefresh     bool
	RefreshDay        *time.Time
	IdempotencyKey    string
	RequestHash       string
}

func (r *Repository) CreateOrReuse(ctx context.Context, in CreateRecord) (Record, bool, error) {
	tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable})
	if err != nil {
		return Record{}, false, fmt.Errorf("beginning report transaction: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()
	lockKey := fmt.Sprintf("growth-revision:%d:%s", in.MerchantID, in.TerminalID)
	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, lockKey); err != nil {
		return Record{}, false, fmt.Errorf("locking growth report request: %w", err)
	}

	var terminalDBID *int64
	if in.TerminalID != "" {
		var id int64
		if err := tx.QueryRow(ctx, `SELECT id FROM terminals WHERE merchant_id=$1 AND source_key=$2`, in.MerchantID, in.TerminalID).Scan(&id); errors.Is(err, pgx.ErrNoRows) {
			return Record{}, false, ErrUnknownTerminal
		} else if err != nil {
			return Record{}, false, fmt.Errorf("validating report terminal: %w", err)
		}
		terminalDBID = &id
	}
	if in.IdempotencyKey != "" {
		var requestHash, reportID string
		err := tx.QueryRow(ctx, `SELECT request_hash,report_id FROM growth_report_idempotency WHERE merchant_id=$1 AND idempotency_key=$2`, in.MerchantID, in.IdempotencyKey).Scan(&requestHash, &reportID)
		if err == nil {
			if requestHash != in.RequestHash {
				return Record{}, false, ErrIdempotencyConflict
			}
			record, err := getWith(ctx, tx, in.MerchantID, reportID)
			if err != nil {
				return Record{}, false, err
			}
			return record, true, nil
		}
		if !errors.Is(err, pgx.ErrNoRows) {
			return Record{}, false, fmt.Errorf("reading report idempotency: %w", err)
		}
	}
	if !in.ManualRefresh {
		record, err := currentWith(ctx, tx, in.MerchantID, in.Fingerprint)
		if err == nil {
			return record, true, nil
		}
		if !errors.Is(err, ErrNotFound) {
			return Record{}, false, err
		}
	}
	if in.ManualRefresh && in.RefreshDay != nil {
		var exists bool
		if err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM growth_reports WHERE merchant_id=$1 AND manual_refresh=true AND tehran_refresh_day=$2)`, in.MerchantID, in.RefreshDay).Scan(&exists); err != nil {
			return Record{}, false, fmt.Errorf("checking daily refresh: %w", err)
		}
		if exists {
			return Record{}, false, ErrRefreshLimited
		}
	}
	var revision int
	if err := tx.QueryRow(ctx, `SELECT coalesce(max(revision),0)+1 FROM growth_reports WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text`, in.MerchantID, nilIfEmpty(in.TerminalID)).Scan(&revision); err != nil {
		return Record{}, false, fmt.Errorf("calculating growth report revision: %w", err)
	}
	_, err = tx.Exec(ctx, `INSERT INTO growth_reports(id,merchant_id,terminal_id,terminal_source_key,fingerprint,revision,status,forecast_start_date,dataset_watermark,catalog_version,analytics_version,prompt_version,schema_version,model,manual_refresh,tehran_refresh_day)
		VALUES($1,$2,$3,$4,$5,$6,'QUEUED',$7,$8,$9,$10,$11,$12,$13,$14,$15)`, in.ID, in.MerchantID, terminalDBID, nilIfEmpty(in.TerminalID), in.Fingerprint, revision, in.ForecastStartDate, in.DatasetWatermark, in.CatalogVersion, in.AnalyticsVersion, in.PromptVersion, in.SchemaVersion, in.Model, in.ManualRefresh, in.RefreshDay)
	if err != nil {
		if in.ManualRefresh && isUniqueViolation(err) {
			return Record{}, false, ErrRefreshLimited
		}
		return Record{}, false, fmt.Errorf("creating growth report: %w", err)
	}
	if in.IdempotencyKey != "" {
		if _, err := tx.Exec(ctx, `INSERT INTO growth_report_idempotency(merchant_id,idempotency_key,request_hash,report_id) VALUES($1,$2,$3,$4)`, in.MerchantID, in.IdempotencyKey, in.RequestHash, in.ID); err != nil {
			return Record{}, false, fmt.Errorf("creating report idempotency: %w", err)
		}
	}
	record, err := getWith(ctx, tx, in.MerchantID, in.ID)
	if err != nil {
		return Record{}, false, err
	}
	if err := tx.Commit(ctx); err != nil {
		return Record{}, false, fmt.Errorf("committing growth report: %w", err)
	}
	return record, false, nil
}

func (r *Repository) Get(ctx context.Context, merchantID int64, id string) (Record, error) {
	return getWith(ctx, r.pool, merchantID, id)
}

func (r *Repository) Latest(ctx context.Context, merchantID int64, terminalID string) (Record, error) {
	return scanRecord(r.pool.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND terminal_source_key IS NOT DISTINCT FROM $2::text AND status IN ('READY','PARTIAL') ORDER BY created_at DESC LIMIT 1`, merchantID, nilIfEmpty(terminalID)))
}

func (r *Repository) Claim(ctx context.Context) (Record, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return Record{}, fmt.Errorf("beginning report claim: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()
	row := tx.QueryRow(ctx, reportSelect+` WHERE id=(SELECT id FROM growth_reports WHERE status='QUEUED' ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1)`)
	record, err := scanRecord(row)
	if errors.Is(err, ErrNotFound) {
		return Record{}, ErrNotFound
	}
	if err != nil {
		return Record{}, err
	}
	if _, err := tx.Exec(ctx, `UPDATE growth_reports SET status='ANALYZING',updated_at=now() WHERE id=$1`, record.ReportID); err != nil {
		return Record{}, fmt.Errorf("claiming growth report: %w", err)
	}
	record.Status = StatusAnalyzing
	record.UpdatedAt = time.Now().UTC()
	if err := tx.Commit(ctx); err != nil {
		return Record{}, fmt.Errorf("committing report claim: %w", err)
	}
	return record, nil
}

func (r *Repository) StoreEvidence(ctx context.Context, reportID string, evidence json.RawMessage) error {
	command, err := r.pool.Exec(ctx, `UPDATE growth_reports SET deterministic_evidence=$2,status='GENERATING',updated_at=now() WHERE id=$1 AND status='ANALYZING'`, reportID, evidence)
	if err != nil {
		return fmt.Errorf("storing deterministic evidence: %w", err)
	}
	if command.RowsAffected() != 1 {
		return ErrNotFound
	}
	return nil
}

func (r *Repository) Complete(ctx context.Context, reportID string, status Status, evidence json.RawMessage, copy *AICopy, usage Usage, failureCode string) error {
	if status != StatusReady && status != StatusPartial && status != StatusFailed {
		return errors.New("growth report: invalid terminal status")
	}
	var ai any
	if copy != nil {
		ai = copy
	}
	_, err := r.pool.Exec(ctx, `UPDATE growth_reports SET status=$2,deterministic_evidence=$3,ai_response=$4,failure_code=$5,input_tokens=$6,output_tokens=$7,latency_ms=$8,updated_at=now(),completed_at=now() WHERE id=$1`, reportID, status, evidence, ai, nilIfEmpty(failureCode), usage.InputTokens, usage.OutputTokens, usage.LatencyMS)
	if err != nil {
		return fmt.Errorf("completing growth report: %w", err)
	}
	return nil
}

const reportSelect = `SELECT id,revision,merchant_id,terminal_id,terminal_source_key,fingerprint,status,horizon_days,forecast_start_date,dataset_watermark,catalog_version,analytics_version,prompt_version,schema_version,model,deterministic_evidence,ai_response,failure_code,input_tokens,output_tokens,latency_ms,manual_refresh,tehran_refresh_day,created_at,updated_at FROM growth_reports`

type rowScanner interface{ Scan(...any) error }
type queryer interface {
	QueryRow(context.Context, string, ...any) pgx.Row
}

func getWith(ctx context.Context, q queryer, merchantID int64, id string) (Record, error) {
	return scanRecord(q.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND id=$2`, merchantID, id))
}
func currentWith(ctx context.Context, q queryer, merchantID int64, fingerprint string) (Record, error) {
	return scanRecord(q.QueryRow(ctx, reportSelect+` WHERE merchant_id=$1 AND fingerprint=$2 AND manual_refresh=false AND status IN ('QUEUED','ANALYZING','GENERATING','READY','PARTIAL') ORDER BY created_at DESC LIMIT 1`, merchantID, fingerprint))
}

func scanRecord(row rowScanner) (Record, error) {
	var r Record
	var status string
	var evidence []byte
	var ai []byte
	var terminalSource *string
	err := row.Scan(&r.ReportID, &r.Revision, &r.MerchantID, &r.TerminalDBID, &terminalSource, &r.Fingerprint, &status, &r.HorizonDays, &r.ForecastStartDate, &r.DatasetWatermark, &r.CatalogVersion, &r.AnalyticsVersion, &r.PromptVersion, &r.SchemaVersion, &r.Model, &evidence, &ai, &r.FailureCode, &r.Usage.InputTokens, &r.Usage.OutputTokens, &r.Usage.LatencyMS, &r.ManualRefresh, &r.TehranRefreshDay, &r.CreatedAt, &r.UpdatedAt)
	if errors.Is(err, pgx.ErrNoRows) {
		return Record{}, ErrNotFound
	}
	if err != nil {
		return Record{}, fmt.Errorf("scanning growth report: %w", err)
	}
	r.Status = Status(status)
	if terminalSource != nil {
		r.Scope.TerminalID = terminalSource
	}
	r.DeterministicEvidence = json.RawMessage(evidence)
	r.EventsURL = "/api/v1/analytics/growth-reports/" + r.ReportID + "/events"
	if len(ai) > 0 {
		var copy AICopy
		if json.Unmarshal(ai, &copy) == nil {
			r.AICopy = &copy
		}
	}
	return r, nil
}

func nilIfEmpty(value string) any {
	if value == "" {
		return nil
	}
	return value
}

func isUniqueViolation(err error) bool {
	type sqlState interface{ SQLState() string }
	var state sqlState
	return errors.As(err, &state) && state.SQLState() == "23505"
}

func (r *Repository) NextRefreshAt(ctx context.Context, merchantID int64, now time.Time) (time.Time, error) {
	var last *time.Time
	if err := r.pool.QueryRow(ctx, `SELECT max(created_at) FROM growth_reports WHERE merchant_id=$1 AND manual_refresh=true`, merchantID).Scan(&last); err != nil {
		return time.Time{}, fmt.Errorf("loading refresh quota: %w", err)
	}
	if last == nil {
		return now.UTC(), nil
	}
	return NextTehranRefresh(*last), nil
}

func (r *Repository) DatasetMetadata(ctx context.Context) (string, time.Time, error) {
	var sha string
	var latest time.Time
	err := r.pool.QueryRow(ctx, `SELECT ir.source_sha256,max(ps.created_at AT TIME ZONE 'Asia/Tehran')::date
		FROM payment_sessions ps JOIN import_runs ir ON ir.id=ps.import_run_id WHERE ir.status='COMPLETED'
		GROUP BY ir.id,ir.source_sha256 ORDER BY ir.completed_at DESC LIMIT 1`).Scan(&sha, &latest)
	if err != nil {
		return "", time.Time{}, fmt.Errorf("loading growth dataset metadata: %w", err)
	}
	return sha + ":" + latest.Format(time.DateOnly), latest.AddDate(0, 0, 1), nil
}

func (r *Repository) CleanupSuperseded(ctx context.Context, cutoff time.Time) (int64, error) {
	command, err := r.pool.Exec(ctx, `DELETE FROM growth_reports old WHERE old.created_at<$1 AND EXISTS(SELECT 1 FROM growth_reports newer WHERE newer.merchant_id=old.merchant_id AND newer.terminal_source_key IS NOT DISTINCT FROM old.terminal_source_key AND newer.id<>old.id AND newer.revision>old.revision AND newer.status IN ('READY','PARTIAL'))`, cutoff)
	if err != nil {
		return 0, fmt.Errorf("cleaning superseded growth reports: %w", err)
	}
	return command.RowsAffected(), nil
}