package main

import (
	"database/sql"
	"encoding/json"
	"log"
	"os"
	"time"

	_ "github.com/jackc/pgx/v5/stdlib"
	amqp "github.com/rabbitmq/amqp091-go"
)

type paymentMessage struct {
	PaymentID string    `json:"payment_id"`
	UserID    int       `json:"user_id"`
	LicenseID string    `json:"license_id"`
	Amount    int       `json:"amount"`
	Timestamp time.Time `json:"timestamp"`
}

type license struct {
	ID     string
	Price  int
	Name   string
	Hidden bool
}

func main() {
	rabbitURL := envDefault("RABBITMQ_URL", "amqp://guest:guest@rabbitmq:5672/")
	dbDSN := envDefault("DB_DSN", "postgres://license:license@postgres:5432/licenses?sslmode=disable")

	db := mustOpenDB(dbDSN)
	defer db.Close()
	db.SetMaxOpenConns(50)
	db.SetMaxIdleConns(20)

	if err := initSchema(db); err != nil {
		log.Fatalf("schema: %v", err)
	}

	conn := mustDialRabbit(rabbitURL)
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("amqp channel: %v", err)
	}
	defer ch.Close()

	if err := ch.Qos(200, 0, false); err != nil {
		log.Fatalf("qos: %v", err)
	}

	queueName := "payments"
	_, err = ch.QueueDeclare(queueName, true, false, false, false, nil)
	if err != nil {
		log.Fatalf("queue: %v", err)
	}

	msgs, err := ch.Consume(queueName, "", false, false, false, false, nil)
	if err != nil {
		log.Fatalf("consume: %v", err)
	}

	log.Printf("json worker started")
	sem := make(chan struct{}, 50)
	for msg := range msgs {
		sem <- struct{}{}
		go func(delivery amqp.Delivery) {
			defer func() { <-sem }()
			var payment paymentMessage
			if err := json.Unmarshal(delivery.Body, &payment); err != nil {
				_ = delivery.Ack(false)
				return
			}
			if err := processPayment(db, payment); err != nil {
				log.Printf("process: %v", err)
			}
			_ = delivery.Ack(false)
		}(msg)
	}
}

func envDefault(key, fallback string) string {
	val := os.Getenv(key)
	if val == "" {
		return fallback
	}
	return val
}

func mustOpenDB(dsn string) *sql.DB {
	var lastErr error
	for i := 0; i < 30; i++ {
		db, err := sql.Open("pgx", dsn)
		if err != nil {
			lastErr = err
			time.Sleep(2 * time.Second)
			continue
		}
		if err := db.Ping(); err != nil {
			lastErr = err
			_ = db.Close()
			time.Sleep(2 * time.Second)
			continue
		}
		return db
	}
	log.Fatalf("db: %v", lastErr)
	return nil
}

func initSchema(db *sql.DB) error {
	queries := []string{
		`CREATE TABLE IF NOT EXISTS users (
			id INTEGER PRIMARY KEY,
			username TEXT UNIQUE,
			password_hash TEXT,
			balance INTEGER NOT NULL
		)`,
		`CREATE TABLE IF NOT EXISTS licenses (id TEXT PRIMARY KEY, name TEXT NOT NULL, price INTEGER NOT NULL, hidden BOOLEAN NOT NULL)`,
		`CREATE TABLE IF NOT EXISTS user_licenses (user_id INTEGER NOT NULL, license_id TEXT NOT NULL, PRIMARY KEY (user_id, license_id))`,
		`CREATE TABLE IF NOT EXISTS user_payments (id BIGSERIAL PRIMARY KEY, user_id INTEGER NOT NULL, license_id TEXT NOT NULL, amount INTEGER NOT NULL, created_at TIMESTAMPTZ NOT NULL)`,
		`CREATE TABLE IF NOT EXISTS payments (
			payment_id TEXT PRIMARY KEY,
			user_id INTEGER NOT NULL,
			license_id TEXT NOT NULL,
			amount INTEGER NOT NULL,
			status TEXT NOT NULL,
			created_at TIMESTAMPTZ NOT NULL,
			processed_at TIMESTAMPTZ
		)`,
		`CREATE SEQUENCE IF NOT EXISTS users_id_seq`,
		`ALTER TABLE users ALTER COLUMN id SET DEFAULT nextval('users_id_seq')`,
	}
	for _, q := range queries {
		if _, err := db.Exec(q); err != nil {
			return err
		}
	}
	if _, err := db.Exec(`SELECT setval('users_id_seq', COALESCE((SELECT MAX(id) FROM users), 0) + 1, false)`); err != nil {
		return err
	}
	if _, err := db.Exec(`INSERT INTO users (id, balance) VALUES (1, 100) ON CONFLICT (id) DO NOTHING`); err != nil {
		return err
	}
	seeds := []license{
		{ID: "license-1", Name: "Полная лицензий", Price: 100, Hidden: false},
		{ID: "license-2", Name: "Без смены ника", Price: 130, Hidden: false},
		{ID: "license-3", Name: "Полная с флагом на плаще", Price: 200, Hidden: false},
		{ID: "license-4", Name: "Без гарантии", Price: 70, Hidden: false},
		{ID: "license-5", Name: "Без смены скина", Price: 90, Hidden: false},
		{ID: "license-6", Name: "Доступ на сервера СП", Price: 160, Hidden: false},
	}
	for _, lic := range seeds {
		_, err := db.Exec(
			`INSERT INTO licenses (id, name, price, hidden) VALUES ($1, $2, $3, $4) ON CONFLICT (id) DO NOTHING`,
			lic.ID, lic.Name, lic.Price, lic.Hidden,
		)
		if err != nil {
			return err
		}
	}
	return nil
}

func getBalance(db *sql.DB, userID int) (int, error) {
	var balance int
	err := db.QueryRow(`SELECT balance FROM users WHERE id = $1`, userID).Scan(&balance)
	return balance, err
}

func processPayment(db *sql.DB, payment paymentMessage) error {
	res, err := db.Exec(
		`UPDATE users SET balance = balance - $1 WHERE id = $2 AND balance >= $1`,
		payment.Amount, payment.UserID,
	)
	if err != nil {
		return err
	}
	affected, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if affected == 0 {
		_, _ = db.Exec(
			`UPDATE payments SET status = $1, processed_at = $2 WHERE payment_id = $3`,
			"failed", time.Now().UTC(), payment.PaymentID,
		)
		return nil
	}
	if _, err := db.Exec(
		`INSERT INTO user_payments (user_id, license_id, amount, created_at) VALUES ($1, $2, $3, $4)`,
		payment.UserID, payment.LicenseID, payment.Amount, time.Now().UTC(),
	); err != nil {
		return err
	}

	var price int
	if err := db.QueryRow(`SELECT price FROM licenses WHERE id = $1`, payment.LicenseID).Scan(&price); err != nil {
		return err
	}
	var total int
	if err := db.QueryRow(
		`SELECT COALESCE(SUM(amount), 0) FROM user_payments WHERE user_id = $1 AND license_id = $2`,
		payment.UserID, payment.LicenseID,
	).Scan(&total); err != nil {
		return err
	}
	if total < price {
		return nil
	}
	_, err = db.Exec(`INSERT INTO user_licenses (user_id, license_id) VALUES ($1, $2) ON CONFLICT DO NOTHING`, payment.UserID, payment.LicenseID)
	if err != nil {
		return err
	}
	_, err = db.Exec(
		`UPDATE payments SET status = $1, processed_at = $2 WHERE payment_id = $3`,
		"processed", time.Now().UTC(), payment.PaymentID,
	)
	return err
}

func mustDialRabbit(url string) *amqp.Connection {
	var lastErr error
	for i := 0; i < 30; i++ {
		conn, err := amqp.Dial(url)
		if err == nil {
			return conn
		}
		lastErr = err
		time.Sleep(2 * time.Second)
	}
	log.Fatalf("rabbitmq: %v", lastErr)
	return nil
}
