import hashlib
from contextlib import closing
import hmac
import json
import sqlite3
import tempfile
import threading
import unittest
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from payments import PaymentService, Rejected, CrashAfterCredit, create_database

SECRET = b"public-demo-secret-not-a-real-credential"


def request(event_id="psp-42", account_id="wallet-a", amount_minor=1000):
    raw = json.dumps(dict(event_id=event_id, account_id=account_id,
                          amount_minor=amount_minor, currency="USD")).encode()
    return raw, hmac.new(SECRET, raw, hashlib.sha256).hexdigest()


class PaymentReviewTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.path = Path(self.directory.name) / "ledger.db"
        create_database(self.path)
        self.service = PaymentService(self.path, SECRET)

    def balance(self, tenant="alpha", account_id="wallet-a"):
        with closing(sqlite3.connect(self.path)) as db:
            return db.execute("SELECT balance_minor FROM accounts WHERE tenant=? AND account_id=?",
                              (tenant, account_id)).fetchone()[0]

    def count(self):
        with closing(sqlite3.connect(self.path)) as db:
            return db.execute("SELECT COUNT(*) FROM ledger").fetchone()[0]

    def test_before_crash_and_retry_double_credits(self):
        raw, sig = request()
        with self.assertRaises(CrashAfterCredit):
            self.service.callback_before("alpha", raw, sig, crash=True)
        self.assertEqual(self.balance(), 11000)
        self.assertEqual(self.count(), 0)
        self.service.callback_before("alpha", raw, sig)
        self.assertEqual(self.balance(), 12000)  # Only one 1000-unit event existed.
        self.assertEqual(self.count(), 1)

    def test_after_crash_rolls_back_and_retry_credits_once(self):
        raw, sig = request()
        with self.assertRaises(CrashAfterCredit):
            self.service.callback_after("alpha", raw, sig, crash=True)
        self.assertEqual(self.balance(), 10000)
        self.assertEqual(self.count(), 0)
        # New service instance uses persisted state, not in-memory deduplication.
        fresh = PaymentService(self.path, SECRET)
        self.assertEqual(fresh.callback_after("alpha", raw, sig), "credited")
        self.assertEqual(fresh.callback_after("alpha", raw, sig), "duplicate")
        self.assertEqual(self.balance(), 11000)
        self.assertEqual(self.count(), 1)

    def test_after_eight_concurrent_callbacks_credit_once(self):
        raw, sig = request()
        with ThreadPoolExecutor(max_workers=8) as pool:
            results = list(pool.map(lambda _: self.service.callback_after("alpha", raw, sig), range(8)))
        self.assertEqual(results.count("credited"), 1)
        self.assertEqual(results.count("duplicate"), 7)
        self.assertEqual(self.balance(), 11000)
        self.assertEqual(self.count(), 1)

    def test_after_reused_event_with_changed_amount_rejected(self):
        self.service.callback_after("alpha", *request())
        with self.assertRaises(Rejected):
            self.service.callback_after("alpha", *request(amount_minor=2000))
        self.assertEqual(self.balance(), 11000)

    def test_invalid_signature_has_no_effect(self):
        raw, _ = request()
        with self.assertRaises(Rejected):
            self.service.callback_after("alpha", raw, "0" * 64)
        self.assertEqual(self.balance(), 10000)
        self.assertEqual(self.count(), 0)

    def test_before_concurrent_withdrawals_overdraw_balance(self):
        barrier = threading.Barrier(2)
        with ThreadPoolExecutor(max_workers=2) as pool:
            results = list(pool.map(lambda i: self.service.withdraw_before(
                "alpha", "wallet-a", 8000, f"withdraw-{i}", barrier=barrier), range(2)))
        self.assertEqual(results, ["reserved", "reserved"])
        self.assertEqual(self.balance(), -6000)

    def test_after_concurrent_withdrawals_allow_only_one(self):
        def withdraw(i):
            try:
                return self.service.withdraw_after("alpha", "wallet-a", 8000, f"withdraw-{i}")
            except Rejected:
                return "rejected"
        with ThreadPoolExecutor(max_workers=2) as pool:
            results = list(pool.map(withdraw, range(2)))
        self.assertCountEqual(results, ["reserved", "rejected"])
        self.assertEqual(self.balance(), 2000)
        self.assertEqual(self.count(), 1)

    def test_after_withdrawal_retry_is_idempotent_and_bound(self):
        self.service.withdraw_after("alpha", "wallet-a", 1000, "withdraw-1")
        self.assertEqual(self.service.withdraw_after("alpha", "wallet-a", 1000, "withdraw-1"), "duplicate")
        with self.assertRaises(Rejected):
            self.service.withdraw_after("alpha", "wallet-a", 2000, "withdraw-1")
        self.assertEqual(self.balance(), 9000)

    def test_before_foreign_statement_is_visible(self):
        self.service.callback_after("beta", *request(account_id="wallet-b"))
        rows = self.service.statement_before("alpha", "wallet-b")
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]["tenant"], "beta")

    def test_after_foreign_statement_and_callback_are_denied(self):
        self.service.callback_after("beta", *request(account_id="wallet-b"))
        with self.assertRaises(Rejected):
            self.service.statement_after("alpha", "wallet-b")
        with self.assertRaises(Rejected):
            self.service.callback_after("alpha", *request("other", "wallet-b"))
        self.assertEqual(len(self.service.statement_after("beta", "wallet-b")), 1)
        self.assertEqual(self.balance("beta", "wallet-b"), 51000)


if __name__ == "__main__":
    unittest.main(verbosity=2)