package payments

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"github.com/google/uuid"
)

// myNalogLockKey is the advisory-lock key guarding an export run. One run at a time is a hard
// requirement, not a nicety: the tax service accepts no idempotency key, so two overlapping runs
// could file the same income twice.
const myNalogLockKey int64 = 0x6D796E616C6F67 // "mynalog"

// taxableIncome is the predicate selecting ledger rows this export is responsible for: rouble
// funding on the direct rail. Store rails are excluded because the ledger records their price in
// the store's own currency and holds no rouble amount at all.
const taxableIncome = `l.kind = 'fund' AND l.provider = $1 AND l.order_id IS NOT NULL
	AND l.snapshot->>'currency' = $2`

// refundedOrder matches an income whose order has since been refunded.
const refundedOrder = `EXISTS (SELECT 1 FROM payments.ledger x
	WHERE x.order_id = l.order_id AND x.kind = 'refund')`

// incomeColumns is the projection every queue row is read through.
const incomeColumns = `l.ledger_id, l.order_id, l.account_id, l.snapshot, l.created_at,
	COALESCE(r.status, ''), COALESCE(r.receipt_name, ''), COALESCE(r.receipt_uuid, ''),
	COALESCE(r.attempts, 0), COALESCE(r.last_error, '')`

// myNalogQueue reads everything one export run must consider. It is four cheap reads against the
// ledger and the export table; nothing here touches the tax service, which is what lets the engine
// decide it has no work before authenticating.
func (s *Store) myNalogQueue(ctx context.Context, limit int) (MyNalogQueue, error) {
	var q MyNalogQueue
	var err error

	// Owed: never attempted, or attempted and understood to have failed — and not since refunded.
	if q.Incomes, err = s.myNalogIncomes(ctx, `(r.status IS NULL OR r.status = '`+MyNalogFailed+`')
		AND NOT `+refundedOrder, limit); err != nil {
		return MyNalogQueue{}, err
	}
	// Moot: refunded before a receipt was ever issued, so nothing is filed and nothing is annulled.
	if q.NotRequired, err = s.myNalogIncomes(ctx, `(r.status IS NULL OR r.status = '`+MyNalogFailed+`')
		AND `+refundedOrder, limit); err != nil {
		return MyNalogQueue{}, err
	}
	// Stuck: an outcome nobody can infer. These halt the run and wait for a human.
	if q.Attention, err = s.myNalogIncomes(ctx,
		`r.status IN ('`+MyNalogUnknown+`', '`+MyNalogSending+`')`, limit); err != nil {
		return MyNalogQueue{}, err
	}
	if q.Cancels, err = s.myNalogCancels(ctx, limit); err != nil {
		return MyNalogQueue{}, err
	}
	return q, nil
}

// myNalogIncomes reads taxable incomes narrowed by an extra predicate over the export row, oldest
// first — the oldest income is the one closest to its filing deadline.
func (s *Store) myNalogIncomes(ctx context.Context, extra string, limit int) ([]MyNalogIncome, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT `+incomeColumns+`
		FROM payments.ledger l
		LEFT JOIN payments.mynalog_receipt r ON r.ledger_id = l.ledger_id
		WHERE `+taxableIncome+` AND (`+extra+`)
		ORDER BY l.created_at
		LIMIT $3`, myNalogProvider, string(CurrencyRUB), limit)
	if err != nil {
		return nil, fmt.Errorf("payments: read mynalog queue: %w", err)
	}
	defer rows.Close()

	var out []MyNalogIncome
	for rows.Next() {
		var (
			in       MyNalogIncome
			orderID  uuid.NullUUID
			snapshot sql.NullString
		)
		if err := rows.Scan(&in.LedgerID, &orderID, &in.AccountID, &snapshot, &in.OperationTime,
			&in.Status, &in.ReceiptName, &in.ReceiptUUID, &in.Attempts, &in.LastError); err != nil {
			return nil, fmt.Errorf("payments: scan mynalog queue row: %w", err)
		}
		in.OrderID = orderID.UUID
		in.Amount = moneyFromSnapshot(snapshot.String)
		in.Title, in.Chips = describeSnapshot(snapshot.String)
		out = append(out, in)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("payments: read mynalog queue: %w", err)
	}
	return out, nil
}

// myNalogIncomeByLedger reads one taxable income by its ledger row, reporting false when that row
// is not one this export is responsible for.
func (s *Store) myNalogIncomeByLedger(ctx context.Context, ledgerID uuid.UUID) (MyNalogIncome, bool, error) {
	var (
		in       MyNalogIncome
		orderID  uuid.NullUUID
		snapshot sql.NullString
	)
	err := s.db.QueryRowContext(ctx, `
		SELECT `+incomeColumns+`
		FROM payments.ledger l
		LEFT JOIN payments.mynalog_receipt r ON r.ledger_id = l.ledger_id
		WHERE `+taxableIncome+` AND l.ledger_id = $3`,
		myNalogProvider, string(CurrencyRUB), ledgerID).
		Scan(&in.LedgerID, &orderID, &in.AccountID, &snapshot, &in.OperationTime,
			&in.Status, &in.ReceiptName, &in.ReceiptUUID, &in.Attempts, &in.LastError)
	if err == sql.ErrNoRows {
		return MyNalogIncome{}, false, nil
	}
	if err != nil {
		return MyNalogIncome{}, false, fmt.Errorf("payments: read mynalog income: %w", err)
	}
	in.OrderID = orderID.UUID
	in.Amount = moneyFromSnapshot(snapshot.String)
	in.Title, in.Chips = describeSnapshot(snapshot.String)
	return in, true, nil
}

// myNalogCancels reads receipts whose income has been refunded and which are therefore owed an
// annulment, oldest refund first. A previously failed annulment is included: the receipt still
// stands, so it is still owed.
func (s *Store) myNalogCancels(ctx context.Context, limit int) ([]MyNalogCancel, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT r.ledger_id, x.ledger_id, r.order_id, r.account_id, r.receipt_uuid,
		       r.cancel_status, r.last_error
		FROM payments.mynalog_receipt r
		JOIN payments.ledger x ON x.order_id = r.order_id AND x.kind = 'refund'
		WHERE r.status = $1 AND r.cancel_status IN ($2, $3)
		ORDER BY x.created_at
		LIMIT $4`, MyNalogSent, MyNalogCancelNone, MyNalogCancelFailed, limit)
	if err != nil {
		return nil, fmt.Errorf("payments: read mynalog cancels: %w", err)
	}
	defer rows.Close()

	var out []MyNalogCancel
	for rows.Next() {
		var c MyNalogCancel
		if err := rows.Scan(&c.LedgerID, &c.RefundLedgerID, &c.OrderID, &c.AccountID, &c.ReceiptUUID,
			&c.CancelStatus, &c.LastError); err != nil {
			return nil, fmt.Errorf("payments: scan mynalog cancel: %w", err)
		}
		out = append(out, c)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("payments: read mynalog cancels: %w", err)
	}
	return out, nil
}

// myNalogBacklogBefore counts and sums taxable income received before the cut-off that has neither
// been registered nor been ruled out.
func (s *Store) myNalogBacklogBefore(ctx context.Context, before time.Time) (MyNalogBacklog, error) {
	var (
		out    MyNalogBacklog
		oldest sql.NullTime
	)
	err := s.db.QueryRowContext(ctx, `
		SELECT count(*),
		       COALESCE(SUM((l.snapshot->>'amount_minor')::bigint), 0),
		       MIN(l.created_at)
		FROM payments.ledger l
		LEFT JOIN payments.mynalog_receipt r ON r.ledger_id = l.ledger_id
		WHERE `+taxableIncome+`
		  AND l.created_at < $3
		  AND (r.status IS NULL OR r.status NOT IN ($4, $5))`,
		myNalogProvider, string(CurrencyRUB), before, MyNalogSent, MyNalogNotRequired,
	).Scan(&out.Count, &out.AmountMinor, &oldest)
	if err != nil {
		return MyNalogBacklog{}, fmt.Errorf("payments: read mynalog backlog: %w", err)
	}
	out.Oldest = oldest.Time
	return out, nil
}

// promoteStrandedSending reclassifies rows left mid-call by a process that died. The caller holds
// the export lock, so nothing is genuinely in flight; and because the registration call is not
// idempotent, the safe reading of "we started and do not know how it ended" is MyNalogUnknown,
// never "try again".
func (s *Store) promoteStrandedSending(ctx context.Context, now time.Time) (int, error) {
	res, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_receipt
		SET status = $1, last_error = $2, updated_at = $3
		WHERE status = $4`,
		MyNalogUnknown, "the export stopped mid-call; the outcome was never established",
		now, MyNalogSending)
	if err != nil {
		return 0, fmt.Errorf("payments: promote stranded mynalog rows: %w", err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("payments: promote stranded mynalog rows: %w", err)
	}
	return int(n), nil
}

// beginMyNalogSend writes the intent to register, freezing the receipt name. It commits before the
// request leaves so that a crash mid-call is visible afterwards.
func (s *Store) beginMyNalogSend(ctx context.Context, in MyNalogIncome, receiptName string, now time.Time) error {
	_, err := s.db.ExecContext(ctx, `
		INSERT INTO payments.mynalog_receipt (ledger_id, order_id, account_id, status, receipt_name,
			operation_time, amount_minor, currency, attempts, created_at, updated_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 1, $9, $9)
		ON CONFLICT (ledger_id) DO UPDATE SET
			status = EXCLUDED.status,
			receipt_name = EXCLUDED.receipt_name,
			attempts = payments.mynalog_receipt.attempts + 1,
			last_error = '',
			updated_at = EXCLUDED.updated_at`,
		in.LedgerID, in.OrderID, in.AccountID, MyNalogSending, receiptName,
		in.OperationTime, in.Amount.Minor(), string(in.Amount.Currency()), now)
	if err != nil {
		return fmt.Errorf("payments: begin mynalog send: %w", err)
	}
	return nil
}

// markMyNalogSent records the receipt the tax service issued.
func (s *Store) markMyNalogSent(ctx context.Context, ledgerID uuid.UUID, receiptUUID string, manual bool, now time.Time) error {
	res, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_receipt
		SET status = $1, receipt_uuid = $2, entered_manually = $3, sent_at = $4,
		    last_error = '', updated_at = $4
		WHERE ledger_id = $5`,
		MyNalogSent, receiptUUID, manual, now, ledgerID)
	if err != nil {
		return fmt.Errorf("payments: mark mynalog sent: %w", err)
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return fmt.Errorf("payments: mark mynalog sent: no export row for ledger %s", ledgerID)
	}
	return nil
}

// markMyNalogManual records an income an operator filed by hand, creating the row when the export
// never attempted it. Such a row is deliberately indistinguishable from an automatic one apart from
// the flag, so a later refund still annuls the receipt without further human involvement.
func (s *Store) markMyNalogManual(ctx context.Context, in MyNalogIncome, receiptName, receiptUUID string, now time.Time) error {
	_, err := s.db.ExecContext(ctx, `
		INSERT INTO payments.mynalog_receipt (ledger_id, order_id, account_id, status, receipt_uuid,
			receipt_name, operation_time, amount_minor, currency, entered_manually, sent_at,
			created_at, updated_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, true, $10, $10, $10)
		ON CONFLICT (ledger_id) DO UPDATE SET
			status = EXCLUDED.status,
			receipt_uuid = EXCLUDED.receipt_uuid,
			entered_manually = true,
			sent_at = EXCLUDED.sent_at,
			last_error = '',
			updated_at = EXCLUDED.updated_at`,
		in.LedgerID, in.OrderID, in.AccountID, MyNalogSent, receiptUUID, receiptName,
		in.OperationTime, in.Amount.Minor(), string(in.Amount.Currency()), now)
	if err != nil {
		return fmt.Errorf("payments: mark mynalog manual: %w", err)
	}
	return nil
}

// markMyNalogOutcome records a non-final outcome and why.
func (s *Store) markMyNalogOutcome(ctx context.Context, ledgerID uuid.UUID, status, reason string, now time.Time) error {
	res, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_receipt
		SET status = $1, last_error = $2, updated_at = $3
		WHERE ledger_id = $4`, status, reason, now, ledgerID)
	if err != nil {
		return fmt.Errorf("payments: mark mynalog outcome: %w", err)
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return fmt.Errorf("payments: mark mynalog outcome: no export row for ledger %s", ledgerID)
	}
	return nil
}

// markMyNalogNotRequired records that an income needs no receipt, because the buyer had the money
// back before one was issued.
func (s *Store) markMyNalogNotRequired(ctx context.Context, in MyNalogIncome, reason string, now time.Time) error {
	_, err := s.db.ExecContext(ctx, `
		INSERT INTO payments.mynalog_receipt (ledger_id, order_id, account_id, status, receipt_name,
			operation_time, amount_minor, currency, last_error, created_at, updated_at)
		VALUES ($1, $2, $3, $4, '', $5, $6, $7, $8, $9, $9)
		ON CONFLICT (ledger_id) DO UPDATE SET
			status = EXCLUDED.status,
			last_error = EXCLUDED.last_error,
			updated_at = EXCLUDED.updated_at`,
		in.LedgerID, in.OrderID, in.AccountID, MyNalogNotRequired,
		in.OperationTime, in.Amount.Minor(), string(in.Amount.Currency()), reason, now)
	if err != nil {
		return fmt.Errorf("payments: mark mynalog not required: %w", err)
	}
	return nil
}

// markMyNalogCancel records the outcome of annulling a receipt.
func (s *Store) markMyNalogCancel(ctx context.Context, ledgerID, refundLedgerID uuid.UUID, status, reason string, now time.Time) error {
	var cancelledAt any
	if status == MyNalogCancelled {
		cancelledAt = now
	}
	res, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_receipt
		SET cancel_status = $1, cancel_refund_ledger_id = $2, cancelled_at = $3,
		    last_error = $4, updated_at = $5
		WHERE ledger_id = $6`, status, refundLedgerID, cancelledAt, reason, now, ledgerID)
	if err != nil {
		return fmt.Errorf("payments: mark mynalog cancel: %w", err)
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return fmt.Errorf("payments: mark mynalog cancel: no export row for ledger %s", ledgerID)
	}
	return nil
}

// saveMyNalogSession stores the session. Signing in again clears a self-imposed pause — that is the
// operator telling us the cause has been dealt with — while the automatic mode is left as it was,
// so a routine re-login does not silently arm or disarm it.
func (s *Store) saveMyNalogSession(ctx context.Context, inn, deviceID string, refreshTokenEnc []byte, now time.Time) error {
	_, err := s.db.ExecContext(ctx, `
		INSERT INTO payments.mynalog_session (only_row, inn, device_id, refresh_token_enc, updated_at)
		VALUES (true, $1, $2, $3, $4)
		ON CONFLICT (only_row) DO UPDATE SET
			inn = EXCLUDED.inn,
			device_id = EXCLUDED.device_id,
			refresh_token_enc = EXCLUDED.refresh_token_enc,
			paused_at = NULL,
			pause_reason = '',
			updated_at = EXCLUDED.updated_at`,
		inn, deviceID, refreshTokenEnc, now)
	if err != nil {
		return fmt.Errorf("payments: save mynalog session: %w", err)
	}
	return nil
}

// updateMyNalogRefreshToken replaces the stored token after the service rotated it.
func (s *Store) updateMyNalogRefreshToken(ctx context.Context, refreshTokenEnc []byte, now time.Time) error {
	_, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_session SET refresh_token_enc = $1, updated_at = $2
		WHERE only_row`, refreshTokenEnc, now)
	if err != nil {
		return fmt.Errorf("payments: update mynalog refresh token: %w", err)
	}
	return nil
}

// myNalogSession reads the stored session.
func (s *Store) myNalogSession(ctx context.Context) (MyNalogSession, bool, error) {
	var (
		out                             MyNalogSession
		pausedAt, lastOKAt, lastAlertAt sql.NullTime
	)
	err := s.db.QueryRowContext(ctx, `
		SELECT inn, device_id, refresh_token_enc, auto_enabled, paused_at, pause_reason,
		       last_ok_at, last_alert_at, last_alert_kind, updated_at
		FROM payments.mynalog_session WHERE only_row`).
		Scan(&out.INN, &out.DeviceID, &out.RefreshTokenEnc, &out.AutoEnabled, &pausedAt,
			&out.PauseReason, &lastOKAt, &lastAlertAt, &out.LastAlertKind, &out.UpdatedAt)
	if err == sql.ErrNoRows {
		return MyNalogSession{}, false, nil
	}
	if err != nil {
		return MyNalogSession{}, false, fmt.Errorf("payments: read mynalog session: %w", err)
	}
	if pausedAt.Valid {
		out.PausedAt = &pausedAt.Time
	}
	if lastOKAt.Valid {
		out.LastOKAt = &lastOKAt.Time
	}
	if lastAlertAt.Valid {
		out.LastAlertAt = &lastAlertAt.Time
	}
	return out, true, nil
}

// dropMyNalogSession signs the rail out.
func (s *Store) dropMyNalogSession(ctx context.Context) error {
	if _, err := s.db.ExecContext(ctx, `DELETE FROM payments.mynalog_session WHERE only_row`); err != nil {
		return fmt.Errorf("payments: drop mynalog session: %w", err)
	}
	return nil
}

// setMyNalogAuto arms or disarms the automatic export.
func (s *Store) setMyNalogAuto(ctx context.Context, enabled bool, now time.Time) error {
	res, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_session SET auto_enabled = $1, updated_at = $2
		WHERE only_row`, enabled, now)
	if err != nil {
		return fmt.Errorf("payments: set mynalog auto: %w", err)
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return fmt.Errorf("payments: set mynalog auto: sign in first")
	}
	return nil
}

// pauseMyNalog takes the rail out of service with a reason.
func (s *Store) pauseMyNalog(ctx context.Context, reason string, now time.Time) error {
	if _, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_session SET paused_at = $1, pause_reason = $2, updated_at = $1
		WHERE only_row AND paused_at IS NULL`, now, reason); err != nil {
		return fmt.Errorf("payments: pause mynalog: %w", err)
	}
	return nil
}

// resumeMyNalog clears a self-imposed pause.
func (s *Store) resumeMyNalog(ctx context.Context, now time.Time) error {
	if _, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_session SET paused_at = NULL, pause_reason = '', updated_at = $1
		WHERE only_row`, now); err != nil {
		return fmt.Errorf("payments: resume mynalog: %w", err)
	}
	return nil
}

// noteMyNalogOK records a successful exchange with the tax service.
func (s *Store) noteMyNalogOK(ctx context.Context, now time.Time) error {
	if _, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_session SET last_ok_at = $1, updated_at = $1
		WHERE only_row`, now); err != nil {
		return fmt.Errorf("payments: note mynalog ok: %w", err)
	}
	return nil
}

// noteMyNalogAlert records that an alert of the given kind went out.
func (s *Store) noteMyNalogAlert(ctx context.Context, kind string, now time.Time) error {
	if _, err := s.db.ExecContext(ctx, `
		UPDATE payments.mynalog_session SET last_alert_at = $1, last_alert_kind = $2, updated_at = $1
		WHERE only_row`, now, kind); err != nil {
		return fmt.Errorf("payments: note mynalog alert: %w", err)
	}
	return nil
}

// tryLockMyNalog takes the advisory lock guarding an export run.
//
// The lock is held on one pinned connection, because a session-level advisory lock belongs to the
// connection that took it: releasing it from another pooled connection would silently fail and
// leave the rail locked until the process restarted.
func (s *Store) tryLockMyNalog(ctx context.Context) (func(), bool, error) {
	conn, err := s.db.Conn(ctx)
	if err != nil {
		return nil, false, fmt.Errorf("payments: lock mynalog: %w", err)
	}
	var locked bool
	if err := conn.QueryRowContext(ctx, `SELECT pg_try_advisory_lock($1)`, myNalogLockKey).Scan(&locked); err != nil {
		_ = conn.Close()
		return nil, false, fmt.Errorf("payments: lock mynalog: %w", err)
	}
	if !locked {
		_ = conn.Close()
		return nil, false, nil
	}
	release := func() {
		// The caller's context is typically done by now (that is what ended the run), and the lock
		// outlives the connection's return to the pool, so the unlock must not inherit that
		// cancellation.
		_, _ = conn.ExecContext(context.WithoutCancel(ctx), `SELECT pg_advisory_unlock($1)`, myNalogLockKey)
		_ = conn.Close()
	}
	return release, true, nil
}

// pendingPurchaseMails reads credited purchases whose buyer has not been written to yet. Every
// undecided succeeded event is returned, including store-rail ones the caller will merely stamp:
// the cursor tracks the decision, so leaving a row unexamined would mean revisiting it forever.
func (s *Store) pendingPurchaseMails(ctx context.Context, limit int) ([]PurchaseMail, error) {
	rows, err := s.db.QueryContext(ctx, `
		SELECT e.event_id, e.account_id, e.order_id, COALESCE(o.origin, ''),
		       COALESCE(o.expected_amount, 0), COALESCE(o.currency, ''),
		       COALESCE(p.title, ''), e.created_at
		FROM payments.payment_events e
		LEFT JOIN payments.orders o ON o.order_id = e.order_id
		LEFT JOIN payments.product p ON p.product_id = o.product_id
		WHERE e.type = 'succeeded' AND e.mailed_at IS NULL
		ORDER BY e.created_at
		LIMIT $1`, limit)
	if err != nil {
		return nil, fmt.Errorf("payments: read purchase mail queue: %w", err)
	}
	defer rows.Close()

	var out []PurchaseMail
	for rows.Next() {
		var (
			m       PurchaseMail
			orderID uuid.NullUUID
		)
		if err := rows.Scan(&m.EventID, &m.AccountID, &orderID, &m.Origin, &m.AmountMinor,
			&m.Currency, &m.Title, &m.CreatedAt); err != nil {
			return nil, fmt.Errorf("payments: scan purchase mail: %w", err)
		}
		m.OrderID = orderID.UUID
		out = append(out, m)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("payments: read purchase mail queue: %w", err)
	}
	return out, nil
}

// markPurchaseMailed stamps the mail decision for an event.
func (s *Store) markPurchaseMailed(ctx context.Context, eventID uuid.UUID, now time.Time) error {
	if _, err := s.db.ExecContext(ctx, `
		UPDATE payments.payment_events SET mailed_at = $1 WHERE event_id = $2`,
		now, eventID); err != nil {
		return fmt.Errorf("payments: mark purchase mailed: %w", err)
	}
	return nil
}

// pendingMyNalogMails reads receipts owed a letter: the annulment letter when cancelled is true,
// the receipt letter otherwise. The product title and pack size come from the ledger snapshot the
// income was filed from, so the letter describes what the buyer actually bought.
func (s *Store) pendingMyNalogMails(ctx context.Context, limit int, cancelled bool) ([]ReceiptMail, error) {
	// Both variants bind the same two parameters, so neither leaves a placeholder unreferenced. A
	// cancelled receipt is always a sent one, so repeating the status test costs nothing.
	where := `r.status = $1 AND r.receipt_mail_sent_at IS NULL`
	if cancelled {
		where = `r.status = $1 AND r.cancel_status = '` + MyNalogCancelled +
			`' AND r.cancel_mail_sent_at IS NULL`
	}
	rows, err := s.db.QueryContext(ctx, `
		SELECT r.ledger_id, r.account_id, r.receipt_uuid, l.snapshot,
		       r.amount_minor, r.currency, r.operation_time
		FROM payments.mynalog_receipt r
		JOIN payments.ledger l ON l.ledger_id = r.ledger_id
		WHERE `+where+`
		ORDER BY r.updated_at
		LIMIT $2`, MyNalogSent, limit)
	if err != nil {
		return nil, fmt.Errorf("payments: read receipt mail queue: %w", err)
	}
	defer rows.Close()

	var out []ReceiptMail
	for rows.Next() {
		var (
			m        ReceiptMail
			snapshot sql.NullString
		)
		if err := rows.Scan(&m.LedgerID, &m.AccountID, &m.ReceiptUUID, &snapshot,
			&m.AmountMinor, &m.Currency, &m.OperationTime); err != nil {
			return nil, fmt.Errorf("payments: scan receipt mail: %w", err)
		}
		m.Title, _ = describeSnapshot(snapshot.String)
		out = append(out, m)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("payments: read receipt mail queue: %w", err)
	}
	return out, nil
}

// markMyNalogMailed stamps the letter decision for an income.
func (s *Store) markMyNalogMailed(ctx context.Context, ledgerID uuid.UUID, cancelled bool, now time.Time) error {
	column := "receipt_mail_sent_at"
	if cancelled {
		column = "cancel_mail_sent_at"
	}
	if _, err := s.db.ExecContext(ctx,
		`UPDATE payments.mynalog_receipt SET `+column+` = $1, updated_at = $1 WHERE ledger_id = $2`,
		now, ledgerID); err != nil {
		return fmt.Errorf("payments: mark receipt mailed: %w", err)
	}
	return nil
}