package importer

import (
	"compress/gzip"
	"context"
	"crypto/sha256"
	"encoding/csv"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"os"
	"strings"

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

var expectedHeader = []string{
	"session_key", "try_seq", "terminal_key", "merchant_key", "category_id",
	"category_title", "amount", "adjusted_fee", "session_status", "try_status",
	"switch_response_code", "psp_code", "issuer_bank_code", "payer_card_key",
	"verify_type", "init_time_ms", "verify_time_ms", "created_at", "try_created_at",
	"verified_at", "settled_at", "expire_in",
}

type Result struct {
	IsNoop       bool
	SourceSHA256 string
	SourceRows   int64
	SessionRows  int64
	AttemptRows  int64
}

func Run(ctx context.Context, pool *pgxpool.Pool, path string) (Result, error) {
	checksum, err := checksumFile(path)
	if err != nil {
		return Result{}, err
	}
	if err := validateHeader(path); err != nil {
		return Result{}, err
	}

	var completed bool
	err = pool.QueryRow(ctx, `SELECT EXISTS (
		SELECT 1 FROM import_runs WHERE source_sha256=$1 AND status='COMPLETED'
	)`, checksum).Scan(&completed)
	if err != nil {
		return Result{}, fmt.Errorf("checking import manifest: %w", err)
	}
	if completed {
		return Result{IsNoop: true, SourceSHA256: checksum}, nil
	}

	tx, err := pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return Result{}, fmt.Errorf("starting import transaction: %w", err)
	}
	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(943271)`); err != nil {
		return Result{}, fmt.Errorf("locking importer: %w", err)
	}

	var runID int64
	err = tx.QueryRow(ctx, `INSERT INTO import_runs(source_sha256,source_path,status)
		VALUES ($1,$2,'RUNNING') RETURNING id`, checksum, path).Scan(&runID)
	if err != nil {
		return Result{}, fmt.Errorf("creating import manifest: %w", err)
	}
	if err := createStaging(ctx, tx); err != nil {
		return Result{}, err
	}

	file, err := os.Open(path)
	if err != nil {
		return Result{}, fmt.Errorf("opening import archive: %w", err)
	}
	defer file.Close()
	gzipReader, err := gzip.NewReader(file)
	if err != nil {
		return Result{}, fmt.Errorf("opening gzip stream: %w", err)
	}
	defer gzipReader.Close()

	copySQL := `COPY import_staging (` + strings.Join(expectedHeader, ",") + `) FROM STDIN WITH (FORMAT csv, HEADER true)`
	copyResult, err := tx.Conn().PgConn().CopyFrom(ctx, gzipReader, copySQL)
	if err != nil {
		return Result{}, fmt.Errorf("copying archive into staging: %w", err)
	}
	result := Result{SourceSHA256: checksum, SourceRows: copyResult.RowsAffected()}
	if err := promote(ctx, tx, runID, &result); err != nil {
		return Result{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return Result{}, fmt.Errorf("committing import: %w", err)
	}
	return result, nil
}

func checksumFile(path string) (string, error) {
	file, err := os.Open(path)
	if err != nil {
		return "", fmt.Errorf("opening source for checksum: %w", err)
	}
	defer file.Close()
	hash := sha256.New()
	if _, err := io.Copy(hash, file); err != nil {
		return "", fmt.Errorf("checksumming source: %w", err)
	}
	return hex.EncodeToString(hash.Sum(nil)), nil
}

func validateHeader(path string) error {
	file, err := os.Open(path)
	if err != nil {
		return fmt.Errorf("opening source for preflight: %w", err)
	}
	defer file.Close()
	gzipReader, err := gzip.NewReader(file)
	if err != nil {
		return fmt.Errorf("opening source gzip for preflight: %w", err)
	}
	defer gzipReader.Close()
	header, err := csv.NewReader(gzipReader).Read()
	if err != nil {
		return fmt.Errorf("reading source header: %w", err)
	}
	if len(header) != len(expectedHeader) {
		return fmt.Errorf("source header has %d columns, expected %d", len(header), len(expectedHeader))
	}
	for i := range expectedHeader {
		if header[i] != expectedHeader[i] {
			return fmt.Errorf("source header column %d is %q, expected %q", i+1, header[i], expectedHeader[i])
		}
	}
	return nil
}

func createStaging(ctx context.Context, tx pgx.Tx) error {
	columns := make([]string, 0, len(expectedHeader))
	for _, column := range expectedHeader {
		columns = append(columns, pgx.Identifier{column}.Sanitize()+" text")
	}
	statement := `CREATE TEMP TABLE import_staging (` + strings.Join(columns, ",") + `) ON COMMIT DROP`
	if _, err := tx.Exec(ctx, statement); err != nil {
		return fmt.Errorf("creating import staging table: %w", err)
	}
	return nil
}

func promote(ctx context.Context, tx pgx.Tx, runID int64, result *Result) error {
	queries := []string{
		`INSERT INTO merchants(source_key,category_id,category_title,display_name)
		 SELECT DISTINCT merchant_key,category_id::bigint,category_title,'کسب‌وکار '||merchant_key
		 FROM import_staging ON CONFLICT (source_key) DO NOTHING`,
		`INSERT INTO terminals(merchant_id,source_key,verify_type)
		 SELECT DISTINCT m.id,s.terminal_key,s.verify_type FROM import_staging s
		 JOIN merchants m ON m.source_key=s.merchant_key
		 ON CONFLICT (merchant_id,source_key) DO NOTHING`,
		`INSERT INTO payment_sessions(import_run_id,merchant_id,terminal_id,source_key,status,amount_rial,fee_amount_rial,payer_card_key,created_at,expires_at,verified_at,settled_at)
		 SELECT DISTINCT ON (m.id,s.session_key) $1,m.id,t.id,s.session_key,s.session_status,s.amount::bigint,s.adjusted_fee::bigint,
		 nullif(s.payer_card_key,''),s.created_at::timestamp AT TIME ZONE 'Asia/Tehran',
		 nullif(s.expire_in,'')::timestamp AT TIME ZONE 'Asia/Tehran',nullif(s.verified_at,'')::timestamp AT TIME ZONE 'Asia/Tehran',
		 nullif(s.settled_at,'')::timestamp AT TIME ZONE 'Asia/Tehran'
		 FROM import_staging s JOIN merchants m ON m.source_key=s.merchant_key
		 JOIN terminals t ON t.merchant_id=m.id AND t.source_key=s.terminal_key
		 ORDER BY m.id,s.session_key,s.try_seq::integer DESC
		 ON CONFLICT (merchant_id,source_key) DO NOTHING`,
		`INSERT INTO payment_attempts(session_id,sequence,status,created_at,init_time_ms,verify_time_ms)
		 SELECT ps.id,s.try_seq::integer,s.try_status,nullif(s.try_created_at,'')::timestamp AT TIME ZONE 'Asia/Tehran',
		 nullif(s.init_time_ms,'')::integer,nullif(s.verify_time_ms,'')::integer
		 FROM import_staging s JOIN merchants m ON m.source_key=s.merchant_key
		 JOIN payment_sessions ps ON ps.merchant_id=m.id AND ps.source_key=s.session_key
		 WHERE s.try_seq::integer > 0 AND s.try_status <> 'NoAttempt'
		 ON CONFLICT (session_id,sequence) DO NOTHING`,
	}
	for i, query := range queries {
		args := []any{}
		if i == 2 {
			args = append(args, runID)
		}
		if _, err := tx.Exec(ctx, query, args...); err != nil {
			return fmt.Errorf("promoting import phase %d: %w", i+1, err)
		}
	}
	if err := tx.QueryRow(ctx, `SELECT count(*) FROM payment_sessions WHERE import_run_id=$1`, runID).Scan(&result.SessionRows); err != nil {
		return fmt.Errorf("counting imported sessions: %w", err)
	}
	if err := tx.QueryRow(ctx, `SELECT count(*) FROM payment_attempts pa JOIN payment_sessions ps ON ps.id=pa.session_id WHERE ps.import_run_id=$1`, runID).Scan(&result.AttemptRows); err != nil {
		return fmt.Errorf("counting imported attempts: %w", err)
	}
	if result.SessionRows == 0 || result.AttemptRows > result.SourceRows {
		return errors.New("import reconciliation failed")
	}
	_, err := tx.Exec(ctx, `UPDATE import_runs SET status='COMPLETED',source_rows=$2,session_rows=$3,attempt_rows=$4,completed_at=now() WHERE id=$1`, runID, result.SourceRows, result.SessionRows, result.AttemptRows)
	if err != nil {
		return fmt.Errorf("completing import manifest: %w", err)
	}
	return nil
}