feat(payments): report income to «Мой налог»

The direct rail runs on НПД, where the provider neither files with the tax service nor issues a receipt — so nobody was doing it. This registers each rouble purchase, annuls its receipt on a refund, and hands the buyer the receipt by email. Two properties of the (unofficial) lknpd API shape the design. Registering an income takes no idempotency key, so an error does not mean nothing happened: the service name is frozen before the call and carries a marker from the tail of the order id, and after a failure the taxpayer's income list is searched for that exact name. Found means filed; not found halts the queue for a human, because declaring an income twice is as wrong as not declaring it. And faults are classified rather than logged: a token is renewed silently, a throttle backs off, an outage retries, but three unfixable rejections take the rail out of service — a changed format must not become thousands of requests overnight. The console button and the worker share one RunBatch. Automatic mode is armed from the console, not from configuration, so the operator can watch a run go through by hand first. A daily watchdog runs whether or not it is armed, since the case it exists for is the export being off. An idle queue issues no call at all — not even an authentication. No payment path changed: the purchase letter rides the existing payment-event outbox on its own cursor, the receipt and annulment letters ride the export row. Decisions D53-D60.
2026-07-28 15:40:36 +02:00
parent c13f5cdb1e
commit e3c2e80a0a
35 changed files with 4926 additions and 13 deletions
@@ -0,0 +1,575 @@
+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, ''), COALESCE((e.payload->>'chips')::int, 0), 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.Chips, &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, m.Chips = 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
+}