Design a Subscription Billing System (Stripe style Recurring Billing)

CREATE TABLE products (
    product_id    UUID PRIMARY KEY,
    merchant_id   UUID NOT NULL,
    name          VARCHAR(256) NOT NULL,
    tax_code      VARCHAR(64),
    active        BOOLEAN DEFAULT true,
    created_at    TIMESTAMPTZ NOT NULL
);

CREATE TABLE prices (
    price_id       UUID PRIMARY KEY,
    product_id     UUID NOT NULL REFERENCES products(product_id),
    unit_amount    BIGINT NOT NULL,          -- cents
    currency       VARCHAR(3) NOT NULL,
    billing_interval VARCHAR(16) NOT NULL,   -- 'month', 'year'
    interval_count INT DEFAULT 1,
    trial_period_days INT DEFAULT 0,
    billing_scheme VARCHAR(16) DEFAULT 'per_unit',  -- 'per_unit', 'tiered', 'metered'
    active         BOOLEAN DEFAULT true,
    created_at     TIMESTAMPTZ NOT NULL
);

CREATE TABLE subscriptions (
    subscription_id       UUID PRIMARY KEY,
    merchant_id           UUID NOT NULL,
    customer_id           UUID NOT NULL,
    price_id              UUID NOT NULL REFERENCES prices(price_id),
    status                VARCHAR(20) NOT NULL,  -- trialing, active, past_due, paused, canceled, unpaid
    current_period_start  TIMESTAMPTZ NOT NULL,
    current_period_end    TIMESTAMPTZ NOT NULL,
    trial_end             TIMESTAMPTZ,
    cancel_at_period_end  BOOLEAN DEFAULT false,
    canceled_at           TIMESTAMPTZ,
    billing_anchor_day    SMALLINT,
    billing_anchor_time    TIME NOT NULL DEFAULT '00:00:00',
    billing_timezone       VARCHAR(64) NOT NULL DEFAULT 'UTC',
    billing_shard          SMALLINT NOT NULL, -- hash(subscription_id) % 64
    paused_at               TIMESTAMPTZ,
    resume_at               TIMESTAMPTZ,
    default_payment_method VARCHAR(64),
    event_version         INT NOT NULL DEFAULT 0,  -- optimistic concurrency (event sourcing)
    created_at             TIMESTAMPTZ NOT NULL,
    updated_at             TIMESTAMPTZ NOT NULL
);

CREATE INDEX idx_subscriptions_billing
    ON subscriptions(billing_shard, status, current_period_end, subscription_id);
CREATE INDEX idx_subscriptions_customer ON subscriptions(customer_id);

CREATE TABLE invoices (
    invoice_id        UUID PRIMARY KEY,
    subscription_id   UUID NOT NULL,
    merchant_id       UUID NOT NULL,
    customer_id       UUID NOT NULL,
    status            VARCHAR(20) NOT NULL,  -- draft, open, paid, void, uncollectible
    amount_due        BIGINT NOT NULL,
    amount_paid       BIGINT NOT NULL DEFAULT 0,
    currency          VARCHAR(3) NOT NULL,
    period_start      TIMESTAMPTZ NOT NULL,
    period_end        TIMESTAMPTZ NOT NULL,
    price_snapshot_json JSONB NOT NULL,      -- immutable effective price, quantity, currency, and tax inputs
    effective_payment_attempt_id UUID,       -- populated only for the effective successful payment
    invoice_kind      VARCHAR(32) NOT NULL,  -- renewal, proration, adjustment, manual
    source_event_id   UUID,                  -- source event when applicable, such as a plan change
    idempotency_key   VARCHAR(128) NOT NULL UNIQUE,   -- stable key for the logical invoice operation
    due_date          TIMESTAMPTZ,
    finalized_at      TIMESTAMPTZ,
    paid_at           TIMESTAMPTZ,
    created_at        TIMESTAMPTZ NOT NULL
);

-- A subscription may have multiple legitimate invoices within one billing period,
-- such as a recurring renewal plus one or more proration or adjustment invoices.
-- The logical operation key above provides idempotency without blocking those cases.
CREATE INDEX idx_invoices_subscription ON invoices(subscription_id, period_start);

CREATE TABLE invoice_line_items (
    line_item_id   UUID PRIMARY KEY,
    invoice_id     UUID NOT NULL REFERENCES invoices(invoice_id),
    description    TEXT NOT NULL,
    amount         BIGINT NOT NULL,          -- negative for credits, stored in minor units
    unit_amount    BIGINT NOT NULL,
    quantity       INT NOT NULL DEFAULT 1,
    currency       VARCHAR(3) NOT NULL,
    price_id       UUID,
    tax_code       VARCHAR(64),
    discount_amount BIGINT NOT NULL DEFAULT 0,
    tax_amount     BIGINT NOT NULL DEFAULT 0
);

CREATE TABLE payment_attempts (
    payment_attempt_id      UUID PRIMARY KEY,
    invoice_id              UUID NOT NULL REFERENCES invoices(invoice_id),
    attempt_number          INT NOT NULL,
    gateway_idempotency_key VARCHAR(128) NOT NULL UNIQUE,
    provider_payment_id     TEXT,
    provider_status         VARCHAR(64),
    status                   VARCHAR(20) NOT NULL, -- pending, unknown, succeeded, failed
    is_effective            BOOLEAN NOT NULL DEFAULT false,
    failure_code             VARCHAR(64),
    created_at               TIMESTAMPTZ NOT NULL,
    completed_at             TIMESTAMPTZ,
    UNIQUE (invoice_id, attempt_number)
);

-- Provider duplicate captures can both be recorded as succeeded, but only one
-- attempt may be marked effective for a given invoice.
CREATE UNIQUE INDEX idx_one_effective_success_per_invoice
    ON payment_attempts(invoice_id)
    WHERE status = 'succeeded' AND is_effective = true;

-- Keep the invoice's effective payment linked to a payment attempt for the same invoice.
CREATE UNIQUE INDEX idx_payment_attempt_invoice_pair
    ON payment_attempts(payment_attempt_id, invoice_id);

ALTER TABLE invoices
    ADD CONSTRAINT fk_invoice_effective_payment_attempt
    FOREIGN KEY (invoice_id, effective_payment_attempt_id)
    REFERENCES payment_attempts(invoice_id, payment_attempt_id);

CREATE TABLE gateway_events (
    provider_event_id TEXT PRIMARY KEY,
    event_type        VARCHAR(64) NOT NULL,
    payment_attempt_id UUID REFERENCES payment_attempts(payment_attempt_id),
    invoice_id        UUID REFERENCES invoices(invoice_id),
    payload_json           JSONB NOT NULL,
    received_at            TIMESTAMPTZ NOT NULL,
    processing_started_at  TIMESTAMPTZ,
    lease_expires_at       TIMESTAMPTZ,
    processed_at            TIMESTAMPTZ,
    status             VARCHAR(20) NOT NULL -- received, processing, processed, ignored
);

CREATE INDEX idx_gateway_events_processing ON gateway_events(status, received_at, processing_started_at);

CREATE TABLE payment_refunds (
    refund_id                UUID PRIMARY KEY,
    invoice_id               UUID NOT NULL REFERENCES invoices(invoice_id),
    payment_attempt_id       UUID NOT NULL REFERENCES payment_attempts(payment_attempt_id),
    gateway_refund_id        TEXT,
    refund_idempotency_key   VARCHAR(128) NOT NULL UNIQUE,
    amount                   BIGINT NOT NULL,
    currency                 VARCHAR(3) NOT NULL,
    reason                   VARCHAR(64) NOT NULL, -- duplicate_capture, customer_request, billing_correction
    status                   VARCHAR(20) NOT NULL, -- pending, succeeded, failed
    created_at               TIMESTAMPTZ NOT NULL,
    completed_at             TIMESTAMPTZ
);

CREATE INDEX idx_payment_refunds_invoice ON payment_refunds(invoice_id, created_at);

CREATE TABLE dunning_attempts (
    dunning_attempt_id UUID PRIMARY KEY,
    invoice_id         UUID NOT NULL REFERENCES invoices(invoice_id),
    attempt_number     INT NOT NULL,
    payment_attempt_id  UUID REFERENCES payment_attempts(payment_attempt_id),
    scheduled_at       TIMESTAMPTZ NOT NULL,
    started_at         TIMESTAMPTZ,
    status             VARCHAR(20) NOT NULL, -- scheduled, running, succeeded, failed, skipped
    failure_code       VARCHAR(64),
    created_at         TIMESTAMPTZ NOT NULL,
    UNIQUE (invoice_id, attempt_number)
);

CREATE TABLE usage_records (
    usage_record_id  UUID PRIMARY KEY,
    subscription_id  UUID NOT NULL REFERENCES subscriptions(subscription_id),
    metric_name      VARCHAR(128) NOT NULL,
    quantity         NUMERIC(20,6) NOT NULL,
    occurred_at      TIMESTAMPTZ NOT NULL,
    ingested_at      TIMESTAMPTZ NOT NULL,
    idempotency_key  VARCHAR(128) NOT NULL UNIQUE
);

CREATE INDEX idx_usage_records_period
    ON usage_records(subscription_id, metric_name, occurred_at);

CREATE TABLE usage_aggregates (
    subscription_id UUID NOT NULL REFERENCES subscriptions(subscription_id),
    metric_name     VARCHAR(128) NOT NULL,
    period_start    TIMESTAMPTZ NOT NULL,
    period_end      TIMESTAMPTZ NOT NULL,
    quantity        NUMERIC(20,6) NOT NULL,
    cutoff_at       TIMESTAMPTZ,
    finalized_at    TIMESTAMPTZ,
    PRIMARY KEY (subscription_id, metric_name, period_start)
);

CREATE TABLE api_idempotency_keys (
    idempotency_key    VARCHAR(128) NOT NULL,
    merchant_id        UUID NOT NULL,
    request_hash       TEXT NOT NULL,
    status             VARCHAR(20) NOT NULL, -- processing, completed, failed
    resource_type      VARCHAR(32),          -- subscription, invoice, payment_attempt
    resource_id        UUID,
    response_json      JSONB,
    processing_started_at TIMESTAMPTZ,
    lease_expires_at   TIMESTAMPTZ,
    created_at         TIMESTAMPTZ NOT NULL,
    expires_at         TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (merchant_id, idempotency_key)
);

-- A processing record is reclaimed only after lease_expires_at. The request hash must
-- match a retry, otherwise return an idempotency conflict. A completed record stores the
-- original response so retries reproduce the same result. Internal billing operations also
-- retain their durable business idempotency keys beyond this request replay window.

CREATE TABLE invoice_events (
    event_id          UUID PRIMARY KEY,
    invoice_id        UUID NOT NULL,
    event_type        VARCHAR(64) NOT NULL,
    event_data        JSONB NOT NULL,
    version           INT NOT NULL,          -- monotonic per invoice
    created_at        TIMESTAMPTZ NOT NULL,
    UNIQUE (invoice_id, version)
);

CREATE TABLE subscription_events (
    event_id          UUID PRIMARY KEY,
    subscription_id   UUID NOT NULL,
    event_type        VARCHAR(64) NOT NULL,
    event_data        JSONB NOT NULL,
    version           INT NOT NULL,          -- monotonic per subscription
    created_at        TIMESTAMPTZ NOT NULL,
    UNIQUE (subscription_id, version)
);

CREATE TABLE billing_scheduler_leases (
    lease_name             VARCHAR(64) PRIMARY KEY,
    scheduler_epoch        BIGINT NOT NULL,
    owner_region           VARCHAR(64) NOT NULL,
    owner_id               VARCHAR(128) NOT NULL,
    lease_expires_at       TIMESTAMPTZ NOT NULL,
    updated_at             TIMESTAMPTZ NOT NULL
);

CREATE TABLE billing_scheduler_checkpoints (
    billing_run_id         UUID NOT NULL,
    shard_id               INT NOT NULL,
    scheduler_epoch        BIGINT NOT NULL, -- fencing epoch for the active scheduler leader
    cursor_subscription_id UUID,
    updated_at             TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (billing_run_id, shard_id)
);

CREATE TABLE billing_outbox (
    outbox_id         UUID PRIMARY KEY,
    aggregate_type    VARCHAR(32) NOT NULL, -- subscription or invoice
    aggregate_id      UUID NOT NULL,
    event_id          UUID NOT NULL UNIQUE,
    topic             VARCHAR(128) NOT NULL,
    payload_json      JSONB NOT NULL,
    created_at        TIMESTAMPTZ NOT NULL,
    published_at      TIMESTAMPTZ,
    attempt_count     INT NOT NULL DEFAULT 0,
    next_attempt_at   TIMESTAMPTZ
);

CREATE INDEX idx_billing_outbox_pending ON billing_outbox(published_at, next_attempt_at);

CREATE TABLE merchant_webhook_endpoints (
    endpoint_id       UUID PRIMARY KEY,
    merchant_id       UUID NOT NULL,
    callback_url      TEXT NOT NULL,
    secret_ref        VARCHAR(128) NOT NULL,
    active            BOOLEAN NOT NULL DEFAULT true,
    created_at        TIMESTAMPTZ NOT NULL
);

CREATE INDEX idx_webhook_endpoints_merchant ON merchant_webhook_endpoints(merchant_id, active);

CREATE TABLE webhook_endpoint_sequences (
    endpoint_id       UUID PRIMARY KEY REFERENCES merchant_webhook_endpoints(endpoint_id),
    next_sequence     BIGINT NOT NULL
);

CREATE TABLE webhook_deliveries (
    event_id          UUID NOT NULL,              -- globally unique source event ID
    merchant_id       UUID NOT NULL,
    endpoint_id       UUID NOT NULL REFERENCES merchant_webhook_endpoints(endpoint_id),
    delivery_sequence BIGINT NOT NULL,            -- allocated transactionally per endpoint
    status             VARCHAR(20) NOT NULL,      -- pending, delivered, dead_lettered
    attempt_count     INT NOT NULL DEFAULT 0,
    next_attempt_at   TIMESTAMPTZ,
    last_error        TEXT,
    delivered_at      TIMESTAMPTZ,
    PRIMARY KEY (event_id, endpoint_id),
    UNIQUE (endpoint_id, delivery_sequence)
);

CREATE INDEX idx_webhook_deliveries_due ON webhook_deliveries(endpoint_id, status, next_attempt_at);