Marketplace Payouts Backend with Stripe Connect

-- payouts/migrations/1_create_ledger.up.sql
CREATE TABLE sellers (
    id TEXT PRIMARY KEY,
    stripe_account_id TEXT NOT NULL
);

CREATE TABLE payouts (
    id BIGSERIAL PRIMARY KEY,
    seller_id TEXT NOT NULL REFERENCES sellers(id),
    amount_cents BIGINT NOT NULL CHECK (amount_cents > 0),
    status TEXT NOT NULL DEFAULT 'pending',
    stripe_transfer_id TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE seller_entries (
    id BIGSERIAL PRIMARY KEY,
    seller_id TEXT NOT NULL REFERENCES sellers(id),
    order_id TEXT NOT NULL,
    kind TEXT NOT NULL,
    amount_cents BIGINT NOT NULL,
    available_at TIMESTAMPTZ NOT NULL,
    payout_id BIGINT REFERENCES payouts(id),
    UNIQUE (order_id, kind)
);

// payouts/ledger.go
package payouts

import (
	"context"
	"time"

	"encore.dev/cron"
	"encore.dev/pubsub"
	"encore.dev/storage/sqldb"
)

var db = sqldb.NewDatabase("payouts", sqldb.DatabaseConfig{Migrations: "./migrations"})

type PayoutCreated struct {
	PayoutID int64 `json:"payout_id"`
}

var Payouts = pubsub.NewTopic[*PayoutCreated]("payout-created", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

type Sale struct {
	SellerID       string `json:"seller_id"`
	OrderID        string `json:"order_id"`
	AmountCents    int64  `json:"amount_cents"`
	PlatformFeeBps int64  `json:"platform_fee_bps"`
}

// RecordSale credits the seller, held for a seven-day refund window.
//
//encore:api private
func RecordSale(ctx context.Context, s *Sale) error {
	net := s.AmountCents - s.AmountCents*s.PlatformFeeBps/10000
	_, err := db.Exec(ctx, `
		INSERT INTO seller_entries (seller_id, order_id, kind, amount_cents, available_at)
		VALUES ($1, $2, 'sale', $3, $4)
		ON CONFLICT (order_id, kind) DO NOTHING
	`, s.SellerID, s.OrderID, net, time.Now().Add(7*24*time.Hour))
	return err
}

var _ = cron.NewJob("daily-payouts", cron.JobConfig{
	Title:    "Batch available seller balances into payouts",
	Schedule: "0 6 * * *",
	Endpoint: CreatePayouts,
})

//encore:api private
func CreatePayouts(ctx context.Context) error {
	rows, err := db.Query(ctx, `
		SELECT seller_id FROM seller_entries
		WHERE payout_id IS NULL AND available_at <= now()
		GROUP BY seller_id HAVING sum(amount_cents) > 0
	`)
	if err != nil {
		return err
	}
	var sellers []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return err
		}
		sellers = append(sellers, id)
	}
	rows.Close()

	for _, sellerID := range sellers {
		payoutID, err := claimEntries(ctx, sellerID)
		if err != nil {
			return err
		}
		if payoutID == 0 {
			continue
		}
		if _, err := Payouts.Publish(ctx, &PayoutCreated{PayoutID: payoutID}); err != nil {
			return err
		}
	}
	return nil
}

// claimEntries moves a seller's open entries into one payout atomically.
func claimEntries(ctx context.Context, sellerID string) (int64, error) {
	tx, err := db.Begin(ctx)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback()

	// Lock the unclaimed rows so an overlapping run cannot sum the same balance.
	rows, err := tx.Query(ctx, `
		SELECT amount_cents FROM seller_entries
		WHERE seller_id = $1 AND payout_id IS NULL AND available_at <= now()
		FOR UPDATE
	`, sellerID)
	if err != nil {
		return 0, err
	}
	var total int64
	for rows.Next() {
		var amount int64
		if err := rows.Scan(&amount); err != nil {
			rows.Close()
			return 0, err
		}
		total += amount
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, err
	}
	if total <= 0 {
		return 0, nil
	}
	var payoutID int64
	if err := tx.QueryRow(ctx, `
		INSERT INTO payouts (seller_id, amount_cents) VALUES ($1, $2) RETURNING id
	`, sellerID, total).Scan(&payoutID); err != nil {
		return 0, err
	}
	result, err := tx.Exec(ctx, `
		UPDATE seller_entries SET payout_id = $1
		WHERE seller_id = $2 AND payout_id IS NULL AND available_at <= now()
	`, payoutID, sellerID)
	if err != nil {
		return 0, err
	}
	updated, err := result.RowsAffected()
	if err != nil {
		return 0, err
	}
	if updated == 0 {
		return 0, nil
	}
	return payoutID, tx.Commit()
}

// payouts/transfer.go
package payouts

import (
	"context"
	"strconv"
	"time"

	"encore.dev/pubsub"
	"github.com/stripe/stripe-go/v87"
)

var secrets struct {
	StripeSecretKey string
}

var _ = pubsub.NewSubscription(Payouts, "send-transfer", pubsub.SubscriptionConfig[*PayoutCreated]{
	Handler:        sendTransfer,
	MaxConcurrency: 4,
	RetryPolicy:    &pubsub.RetryPolicy{MinBackoff: 30 * time.Second, MaxRetries: 10},
})

func sendTransfer(ctx context.Context, msg *PayoutCreated) error {
	var amount int64
	var status, account string
	err := db.QueryRow(ctx, `
		SELECT p.amount_cents, p.status, s.stripe_account_id
		FROM payouts p JOIN sellers s ON s.id = p.seller_id
		WHERE p.id = $1
	`, msg.PayoutID).Scan(&amount, &status, &account)
	if err != nil {
		return err
	}
	if status == "paid" {
		return nil
	}

	sc := stripe.NewClient(secrets.StripeSecretKey)
	params := &stripe.TransferCreateParams{
		Amount:      stripe.Int64(amount),
		Currency:    stripe.String("usd"),
		Destination: stripe.String(account),
	}
	params.SetIdempotencyKey("payout-" + strconv.FormatInt(msg.PayoutID, 10))
	tr, err := sc.V1Transfers.Create(ctx, params)
	if err != nil {
		return err
	}
	_, err = db.Exec(ctx, `
		UPDATE payouts SET status = 'paid', stripe_transfer_id = $1 WHERE id = $2
	`, tr.ID, msg.PayoutID)
	return err
}