"""Remove donation

Revision ID: 6de0c2b1262f
Revises: ebc55e7c3d7d
Create Date: 2024-11-06 14:05:37.475810

"""

import concurrent.futures
import random
import time
from collections.abc import Sequence
from typing import Any, cast

import sqlalchemy as sa
import stripe as stripe_lib
from alembic import op
from sqlalchemy.dialects import postgresql

from polar.config import settings

# Polar Custom Imports

# revision identifiers, used by Alembic.
revision = "6de0c2b1262f"
down_revision = "ebc55e7c3d7d"
branch_labels: tuple[str] | None = None
depends_on: tuple[str] | None = None


stripe_client = stripe_lib.StripeClient(
    settings.STRIPE_SECRET_KEY,
    http_client=stripe_lib.HTTPXClient(allow_sync_methods=True),
)


def create_stripe_product(
    result: tuple[str, str, str], retry: int = 1
) -> tuple[str, str, str]:
    (organization_id, organization_slug, product_id) = result
    try:
        stripe_product = stripe_client.products.create(
            {
                "name": f"{organization_slug} - Donation",
                "default_price_data": {
                    "currency": "usd",
                    "custom_unit_amount": {
                        "enabled": True,
                        "preset": 2000,
                    },
                },
                "metadata": {
                    "product_id": str(product_id),
                    "organization_id": str(organization_id),
                    "organization_name": organization_slug,
                },
            }
        )
    except stripe_lib.RateLimitError:
        time.sleep(retry + random.random())
        return create_stripe_product(result, retry=retry + 1)
    return organization_id, stripe_product.id, cast(str, stripe_product.default_price)


def create_stripe_products(
    results: Sequence[tuple[str, str, str]],
) -> dict[str, tuple[str, str]]:
    stripe_products = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=16) as executor:
        futures = [executor.submit(create_stripe_product, result) for result in results]
        for future in concurrent.futures.as_completed(futures):
            (organization_id, stripe_product_id, stripe_price_id) = future.result()
            stripe_products[organization_id] = (stripe_product_id, stripe_price_id)
    return stripe_products


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column(
        "products",
        sa.Column(
            "user_metadata", postgresql.JSONB(astext_type=sa.Text()), nullable=True
        ),
    )
    op.execute("UPDATE products SET user_metadata = '{}' WHERE user_metadata IS NULL")
    op.alter_column("products", "user_metadata", nullable=False)

    connection = op.get_bind()
    # Create anonymous donation users
    users_results = connection.execute(
        sa.text("""
        SELECT email, MIN(created_at) as created_at
        FROM donations
        WHERE donations.by_user_id IS NULL AND LOWER(email) NOT IN (SELECT LOWER(email) FROM users)
        GROUP BY email
        """)
    )
    for user_result in users_results:
        email, created_at = user_result._tuple()
        connection.execute(
            sa.text("""
            INSERT INTO users (id, created_at, email, email_verified, accepted_terms_of_service)
            VALUES (uuid_generate_v4(), :created_at, :email, FALSE, FALSE)
            RETURNING id
            """).bindparams(
                email=email,
                created_at=created_at,
            )
        )

    results = connection.execute(
        sa.text("""
            SELECT organizations.id, organizations.slug, uuid_generate_v4()
            FROM organizations
            WHERE donations_enabled = TRUE
        """)
    )
    organizations = [result._tuple() for result in results]
    stripe_products = create_stripe_products(organizations)

    for organization_id, _, product_id in organizations:
        stripe_product_id, stripe_price_id = stripe_products[organization_id]
        # Create Message custom field
        custom_field_result = connection.execute(
            sa.text("""
            INSERT INTO custom_fields (id, created_at, type, slug, name, properties, user_metadata, organization_id)
            VALUES (uuid_generate_v4(), NOW(), 'text', 'message', 'Message', '{"textarea": true}', '{}', :organization_id)
            RETURNING id
            """).bindparams(
                organization_id=organization_id,
            )
        )
        row = custom_field_result.fetchone()
        assert row is not None
        custom_field_id = row[0]

        # Create Product
        product_result = connection.execute(
            sa.text("""
            INSERT INTO products (id, created_at, name, is_archived, stripe_product_id, organization_id, user_metadata)
            VALUES (:product_id, NOW(), 'Donation', FALSE, :stripe_product_id, :organization_id, jsonb_build_object('donation_product', TRUE))
            RETURNING id
            """).bindparams(
                product_id=product_id,
                stripe_product_id=stripe_product_id,
                organization_id=organization_id,
            )
        )
        row = product_result.fetchone()
        assert row is not None
        product_id = row[0]

        # Create Price
        price_result = connection.execute(
            sa.text("""
            INSERT INTO product_prices (id, created_at, type, amount_type, price_currency, stripe_price_id, is_archived, preset_amount, product_id)
            VALUES (uuid_generate_v4(), NOW(), 'one_time', 'custom', 'usd', :stripe_price_id, FALSE, 2000, :product_id)
            RETURNING id
            """).bindparams(
                product_id=product_id,
                stripe_price_id=stripe_price_id,
            )
        )
        row = price_result.fetchone()
        assert row is not None
        price_id = row[0]

        # Attach Custom Field to Product
        connection.execute(
            sa.text("""
            INSERT INTO product_custom_fields (id, created_at, product_id, custom_field_id, "order", required)
            VALUES (uuid_generate_v4(), NOW(), :product_id, :custom_field_id, 0, FALSE)
            """).bindparams(
                product_id=product_id,
                custom_field_id=custom_field_id,
            )
        )

        # Migrate User Donations records to Orders
        connection.execute(
            sa.text("""
            INSERT INTO orders (id, created_at, amount, tax_amount, currency, billing_reason, user_id, product_id, product_price_id, user_metadata, custom_field_data)
            SELECT
                uuid_generate_v4(),
                created_at,
                amount,
                0,
                'usd',
                'purchase',
                by_user_id,
                :product_id,
                :price_id,
                jsonb_strip_nulls(jsonb_build_object('donation_id', donations.id::text, 'donation_issue_id', donations.issue_id::text, 'charge_id', donations.charge_id)),
                jsonb_strip_nulls(jsonb_build_object('message', message))
            FROM donations
            WHERE to_organization_id = :organization_id AND by_user_id IS NOT NULL
            """).bindparams(
                organization_id=organization_id,
                product_id=product_id,
                price_id=price_id,
            )
        )

        # Migrate Anonymous Donations records to Orders
        connection.execute(
            sa.text("""
            INSERT INTO orders (id, created_at, amount, tax_amount, currency, billing_reason, user_id, product_id, product_price_id, user_metadata, custom_field_data)
            SELECT
                uuid_generate_v4(),
                donations.created_at,
                amount,
                0,
                'usd',
                'purchase',
                users.id,
                :product_id,
                :price_id,
                jsonb_strip_nulls(jsonb_build_object('donation_id', donations.id::text, 'donation_issue_id', donations.issue_id::text, 'charge_id', donations.charge_id)),
                jsonb_strip_nulls(jsonb_build_object('message', message))
            FROM donations
            JOIN users ON LOWER(users.email) = LOWER(donations.email)
            WHERE to_organization_id = :organization_id AND by_user_id IS NULL
            """).bindparams(
                organization_id=organization_id,
                product_id=product_id,
                price_id=price_id,
            )
        )

    # Relink held balances donations to orders
    op.execute(
        """
        UPDATE held_balances
        SET order_id = orders.id
        FROM orders
        WHERE held_balances.donation_id = (orders.user_metadata->>'donation_id')::uuid
        """
    )
    # Relink transactions donations to orders
    op.execute(
        """
        UPDATE transactions
        SET order_id = orders.id
        FROM orders
        WHERE transactions.donation_id = (orders.user_metadata->>'donation_id')::uuid
        """
    )
    # Remove "donation.created" event from webhook endpoints
    op.execute(
        """
        UPDATE webhook_endpoints
        SET events = events - 'donation.created'
        WHERE events ? 'donation.created'
        """
    )
    # Remove MaintainerDonationReceived notifications
    op.execute(
        """
        DELETE FROM notifications
        WHERE type = 'MaintainerDonationReceived'
        """
    )

    # op.drop_index("ix_donations_by_organization_id", table_name="donations")
    # op.drop_index("ix_donations_by_user_id", table_name="donations")
    # op.drop_index("ix_donations_charge_id", table_name="donations")
    # op.drop_index("ix_donations_created_at", table_name="donations")
    # op.drop_index("ix_donations_deleted_at", table_name="donations")
    # op.drop_index("ix_donations_email", table_name="donations")
    # op.drop_index("ix_donations_modified_at", table_name="donations")
    # op.drop_index("ix_donations_on_behalf_of_organization_id", table_name="donations")
    # op.drop_index("ix_donations_payment_id", table_name="donations")
    # op.drop_index("ix_held_balances_donation_id", table_name="held_balances")
    # op.drop_constraint(
    #     "held_balances_donation_id_fkey", "held_balances", type_="foreignkey"
    # )
    # op.drop_column("held_balances", "donation_id")
    # op.drop_column("organizations", "donations_enabled")

    # op.drop_index("ix_transactions_donation_id", table_name="transactions")
    # op.drop_constraint(
    #     "transactions_donation_id_fkey", "transactions", type_="foreignkey"
    # )
    # op.drop_column("transactions", "donation_id")

    # op.drop_table("donations")
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "donations",
        sa.Column("to_organization_id", sa.UUID(), autoincrement=False, nullable=False),
        sa.Column("payment_id", sa.VARCHAR(), autoincrement=False, nullable=False),
        sa.Column("email", sa.VARCHAR(), autoincrement=False, nullable=False),
        sa.Column("amount", sa.BIGINT(), autoincrement=False, nullable=False),
        sa.Column("amount_received", sa.BIGINT(), autoincrement=False, nullable=False),
        sa.Column("by_user_id", sa.UUID(), autoincrement=False, nullable=True),
        sa.Column("by_organization_id", sa.UUID(), autoincrement=False, nullable=True),
        sa.Column(
            "on_behalf_of_organization_id",
            sa.UUID(),
            autoincrement=False,
            nullable=True,
        ),
        sa.Column("created_by_user_id", sa.UUID(), autoincrement=False, nullable=True),
        sa.Column("id", sa.UUID(), autoincrement=False, nullable=False),
        sa.Column(
            "created_at",
            postgresql.TIMESTAMP(timezone=True),
            autoincrement=False,
            nullable=False,
        ),
        sa.Column(
            "modified_at",
            postgresql.TIMESTAMP(timezone=True),
            autoincrement=False,
            nullable=True,
        ),
        sa.Column(
            "deleted_at",
            postgresql.TIMESTAMP(timezone=True),
            autoincrement=False,
            nullable=True,
        ),
        sa.Column("charge_id", sa.VARCHAR(), autoincrement=False, nullable=False),
        sa.Column("message", sa.VARCHAR(), autoincrement=False, nullable=True),
        sa.Column("issue_id", sa.UUID(), autoincrement=False, nullable=True),
        sa.Column(
            "currency", sa.VARCHAR(length=3), autoincrement=False, nullable=False
        ),
        sa.ForeignKeyConstraint(
            ["by_organization_id"],
            ["organizations.id"],
            name="donations_by_organization_id_fkey",
        ),
        sa.ForeignKeyConstraint(
            ["by_user_id"], ["users.id"], name="donations_by_user_id_fkey"
        ),
        sa.ForeignKeyConstraint(
            ["created_by_user_id"],
            ["users.id"],
            name="donations_created_by_user_id_fkey",
        ),
        sa.ForeignKeyConstraint(
            ["issue_id"], ["issues.id"], name="donations_issue_id_fkey"
        ),
        sa.ForeignKeyConstraint(
            ["on_behalf_of_organization_id"],
            ["organizations.id"],
            name="donations_on_behalf_of_organization_id_fkey",
        ),
        sa.ForeignKeyConstraint(
            ["to_organization_id"],
            ["organizations.id"],
            name="donations_to_organization_id_fkey",
        ),
        sa.PrimaryKeyConstraint("id", name="donations_pkey"),
    )
    op.create_index(
        "ix_donations_payment_id", "donations", ["payment_id"], unique=False
    )
    op.create_index(
        "ix_donations_on_behalf_of_organization_id",
        "donations",
        ["on_behalf_of_organization_id"],
        unique=False,
    )
    op.create_index(
        "ix_donations_modified_at", "donations", ["modified_at"], unique=False
    )
    op.create_index("ix_donations_email", "donations", ["email"], unique=False)
    op.create_index(
        "ix_donations_deleted_at", "donations", ["deleted_at"], unique=False
    )
    op.create_index(
        "ix_donations_created_at", "donations", ["created_at"], unique=False
    )
    op.create_index("ix_donations_charge_id", "donations", ["charge_id"], unique=False)
    op.create_index(
        "ix_donations_by_user_id", "donations", ["by_user_id"], unique=False
    )
    op.create_index(
        "ix_donations_by_organization_id",
        "donations",
        ["by_organization_id"],
        unique=False,
    )

    op.add_column(
        "transactions",
        sa.Column("donation_id", sa.UUID(), autoincrement=False, nullable=True),
    )
    op.create_foreign_key(
        "transactions_donation_id_fkey",
        "transactions",
        "donations",
        ["donation_id"],
        ["id"],
        ondelete="SET NULL",
    )
    op.create_index(
        "ix_transactions_donation_id", "transactions", ["donation_id"], unique=False
    )
    op.drop_column("products", "user_metadata")
    op.add_column(
        "organizations",
        sa.Column(
            "donations_enabled", sa.BOOLEAN(), autoincrement=False, nullable=True
        ),
    )
    op.execute("UPDATE organizations SET donations_enabled = FALSE")
    op.alter_column("organizations", "donations_enabled", nullable=False)

    op.add_column(
        "held_balances",
        sa.Column("donation_id", sa.UUID(), autoincrement=False, nullable=True),
    )
    op.create_foreign_key(
        "held_balances_donation_id_fkey",
        "held_balances",
        "donations",
        ["donation_id"],
        ["id"],
        ondelete="SET NULL",
    )
    op.create_index(
        "ix_held_balances_donation_id", "held_balances", ["donation_id"], unique=False
    )
    # ### end Alembic commands ###
