Ilia Denisov
4d4452596d
CI / changes (pull_request) Successful in 2s
CI / unit (pull_request) Successful in 11s
CI / integration (pull_request) Successful in 23s
CI / ui (pull_request) Successful in 1m16s
CI / conformance (pull_request) Successful in 11s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m50s
Catalog titles already name the pack ("50 фишек"), so appending the chip count
produced "50 фишек — 50 фишек, 79.00 ₽". The letter now carries the title
exactly as the storefront shows it, then the price.
The tax receipt is unaffected: it states the quantity in the wording the tax
service wants, which is a separate surface with separate requirements.
576 lines
23 KiB
Go
576 lines
23 KiB
Go
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
|
|
}
|