// Package mynalogsync registers the direct rail's rouble income with the professional-income tax // service and annuls a receipt when the money is returned. It is the orchestration layer between // the payments domain, which knows what is owed, and the mynalog client, which knows how to say it. // // One engine serves both ways of running it. The admin console calls RunBatch with a short budget // so the page does not hang; the background worker calls the same RunBatch with a long one. There // is deliberately no second implementation, because the delicate part — deciding what a failed // registration means — must not exist twice. // // Two rules shape the whole design, and both come from the tax API accepting no idempotency key: // // - Exactly one run at a time, enforced by an advisory lock. Two concurrent runs could file the // same income twice, and an over-filed income costs real money to unpick. // - An outcome that cannot be established stops everything. Where a normal integration would // retry, this one stops and asks a human, because the two ways of being wrong are "the buyer's // income was never declared" and "it was declared twice", and no automatic choice between them // is safe. package mynalogsync import ( "context" "errors" "fmt" "net/http" "time" "github.com/google/uuid" "go.uber.org/zap" "scrabble/backend/internal/mynalog" "scrabble/backend/internal/payments" ) // Pacing and sizing of a run. None of these are configurable at runtime: they are the shape of the // rail, not an operational preference. const ( // WorkerInterval is how often the automatic export wakes. A tick with an empty queue costs one // local query and no traffic at all to the tax service. WorkerInterval = 15 * time.Minute // WorkerBudget bounds an automatic run to comfortably less than the interval, so a long queue // simply continues on the next tick instead of overlapping with it. WorkerBudget = 13 * time.Minute // ConsoleBudget bounds a run started from the admin console, so the operator's request returns // while they are still looking at it. ConsoleBudget = 100 * time.Second // defaultPace is the gap between two calls to the tax service. It is the whole of our politeness // towards an undocumented API: everything else about a run is bounded by time, not by count. defaultPace = 2 * time.Second // queueLimit caps how much of the queue one run reads. queueLimit = 500 // probeWindow is how far back the post-failure probe searches for our own receipt. A day is // ample: the income being probed was attempted moments ago. probeWindow = 24 * time.Hour // permanentStreakLimit is how many consecutive unfixable rejections take the rail out of // service. A changed API rejects everything, and the alternative to stopping is thousands of // futile requests overnight. permanentStreakLimit = 3 // downtimeAlertAfter is how long the tax service may be unreachable before it is worth a letter. // Below that it is noise: the service is often briefly unavailable and the queue simply waits. downtimeAlertAfter = 24 * time.Hour ) // Alert kinds. They double as the deduplication key: a persistent fault produces one letter a day // per kind rather than one per run. const ( alertPaused = "paused" alertUnknown = "unknown" alertLimit = "limit" alertDown = "down" alertSignIn = "signin" // The three filing-deadline kinds are distinct so that an escalation is never suppressed by the // previous, gentler letter still being within its deduplication window. alertBacklog = "backlog" alertBacklogDue = "backlog-due" alertBacklogOverdue = "backlog-overdue" ) // ErrNotSignedIn is returned when the rail has no usable session. Only an operator can fix it, by // signing in again with the tax cabinet password. var ErrNotSignedIn = errors.New("mynalogsync: not signed in to the tax cabinet") // ErrBusy is returned when another run already holds the export lock. var ErrBusy = errors.New("mynalogsync: an export is already running") // Alerter delivers an operator alert. kind is the deduplication key, not part of the message. type Alerter interface { Alert(ctx context.Context, kind, subject, body string) error } // Config is the deployment-time shape of the rail. An empty Key disables it entirely: without // somewhere safe to keep the refresh token there is no way to stay signed in. type Config struct { Key []byte BaseURL string Location *time.Location // Pace overrides the gap between calls. It exists for tests, which would otherwise take // minutes; production leaves it zero. Pace time.Duration } // Exporter registers income with the tax service on behalf of one taxpayer. type Exporter struct { payments *payments.Service alert Alerter log *zap.Logger cfg Config http *http.Client now func() time.Time } // New builds an Exporter. It returns nil when cfg carries no encryption key, which is how the rail // stays dormant on a deployment that has not configured it. func New(svc *payments.Service, cfg Config, alert Alerter, log *zap.Logger) *Exporter { if svc == nil || len(cfg.Key) == 0 { return nil } if cfg.Location == nil { cfg.Location = time.UTC } if cfg.Pace <= 0 { cfg.Pace = defaultPace } return &Exporter{ payments: svc, alert: alert, log: log, cfg: cfg, // One client shared across a run so connections pool; the ceiling is the API client's own, // and per-request cancellation still rides the context. http: &http.Client{Timeout: mynalog.RequestTimeout}, now: func() time.Time { return time.Now().UTC() }, } } // Summary is what one run did. Stopped explains an early exit in words an operator can act on; it // is empty when the run simply ran out of work. type Summary struct { Registered int Cancelled int NotRequired int Failed int Unknown int Busy bool Paused bool Stopped string } // clientFor builds a tax-cabinet client presenting the given device. func (e *Exporter) clientFor(deviceID string) *mynalog.Client { return mynalog.NewClient(e.cfg.BaseURL, deviceID, e.http) } // Location is the taxpayer's time zone. Every moment sent to the tax service is rendered in it, // because the offset decides which tax period an income falls into. func (e *Exporter) Location() *time.Location { return e.cfg.Location } // ReceiptURL is where a registered receipt can be read, against the configured service root. func (e *Exporter) ReceiptURL(inn, receiptUUID string) string { return mynalog.ReceiptURL(e.cfg.BaseURL, inn, receiptUUID) } // ReceiptName renders exactly the service description an income would be filed under. The console // shows it, the manual-entry export lists it, and the recovery probe searches for it — all three // must agree, so they all come from here. func (e *Exporter) ReceiptName(in payments.MyNalogIncome) string { if in.ReceiptName != "" { return in.ReceiptName // already frozen by an earlier attempt; the probe depends on it } return mynalog.IncomeName(in.Chips, in.OrderID, in.Title) } // SignIn exchanges the cabinet password for a session and stores it. Only the refresh token is // kept, sealed with the configured key; the password is not persisted anywhere, which is why this // is an interactive action rather than a configuration value. func (e *Exporter) SignIn(ctx context.Context, login, password string) error { deviceID := mynalog.DeviceIDForLogin(login) sess, err := e.clientFor(deviceID).AuthByPassword(ctx, login, password, e.now()) if err != nil { return err } if sess.RefreshToken == "" { return fmt.Errorf("mynalogsync: the tax cabinet issued no refresh token: %w", mynalog.ErrPermanent) } sealed, err := mynalog.Seal(e.cfg.Key, sess.RefreshToken) if err != nil { return err } return e.payments.SaveMyNalogSession(ctx, sess.INN, deviceID, sealed) } // SignOut forgets the stored session, which also disarms the automatic export. func (e *Exporter) SignOut(ctx context.Context) error { return e.payments.DropMyNalogSession(ctx) } // session renews the stored session and returns a client bound to it. A rotated refresh token is // persisted immediately: losing one would strand the next renewal and force a password login. func (e *Exporter) session(ctx context.Context) (mynalog.Session, *mynalog.Client, error) { stored, ok, err := e.payments.MyNalogSession(ctx) if err != nil { return mynalog.Session{}, nil, err } if !ok { return mynalog.Session{}, nil, ErrNotSignedIn } refresh, err := mynalog.Open(e.cfg.Key, stored.RefreshTokenEnc) if err != nil { // The key was rotated or lost. Nothing is broken beyond repair; the operator signs in again. return mynalog.Session{}, nil, fmt.Errorf("%w: %v", ErrNotSignedIn, err) } client := e.clientFor(stored.DeviceID) sess, err := client.AuthByRefresh(ctx, refresh, stored.INN, e.now()) if err != nil { return mynalog.Session{}, nil, err } if sess.RefreshToken != refresh { sealed, err := mynalog.Seal(e.cfg.Key, sess.RefreshToken) if err != nil { return mynalog.Session{}, nil, err } if err := e.payments.UpdateMyNalogRefreshToken(ctx, sealed); err != nil { return mynalog.Session{}, nil, err } } return sess, client, nil } // RunBatch performs one export run within budget and reports what it did. // // The order is deliberate. Local bookkeeping happens first, so progress is made even when the tax // service is unreachable. Annulments go before registrations, because an annulment removes income // from the tax base and is the more time-sensitive of the two. And nothing authenticates until the // queue is known to be non-empty, so an idle installation is invisible to the tax service. func (e *Exporter) RunBatch(ctx context.Context, budget time.Duration) (Summary, error) { release, ok, err := e.payments.TryLockMyNalog(ctx) if err != nil { return Summary{}, err } if !ok { return Summary{Busy: true}, nil } defer release() var sum Summary deadline := e.now().Add(budget) // A row still marked as being sent belongs to a run that died mid-call. Under the lock nothing // is genuinely in flight, so it becomes an unresolved outcome rather than a silent retry. if stranded, err := e.payments.PromoteStrandedSending(ctx); err != nil { return sum, err } else if stranded > 0 { e.log.Warn("mynalog export found stranded rows", zap.Int("count", stranded)) } queue, err := e.payments.MyNalogQueue(ctx, queueLimit) if err != nil { return sum, err } // Refunded before ever being filed: nothing to declare and nothing to annul. Purely local. for _, in := range queue.NotRequired { if err := e.payments.MarkMyNalogNotRequired(ctx, in, "the money was returned before the income was ever filed"); err != nil { return sum, err } sum.NotRequired++ } stored, ok, err := e.payments.MyNalogSession(ctx) if err != nil { return sum, err } if !ok { if len(queue.Incomes) > 0 || len(queue.Cancels) > 0 { return sum, ErrNotSignedIn } return sum, nil } if stored.Paused() { sum.Paused = true sum.Stopped = stored.PauseReason return sum, nil } if len(queue.Incomes) == 0 && len(queue.Cancels) == 0 { return sum, nil } sess, client, err := e.session(ctx) if err != nil { return e.handleSessionFailure(ctx, sum, err) } run := &runner{e: e, client: client, sess: sess} sum, err = run.cancels(ctx, queue.Cancels, deadline, sum) if err != nil { return sum, err } // An unresolved outcome from an earlier run blocks new filings. It does not block annulments — // those only ever remove income — but it does mean the operator must look before more income is // declared, which is the only way the backlog of unknowns stays at zero. if len(queue.Attention) > 0 { sum.Unknown = len(queue.Attention) sum.Stopped = "there are incomes with an unresolved outcome; resolve them before filing more" e.raise(ctx, alertUnknown, "«Мой налог»: есть чеки с неизвестным исходом", fmt.Sprintf("Выгрузка остановлена: %d приход(ов) в состоянии «неизвестно». "+ "Откройте раздел «Мой налог» в админ-консоли и разберите их — проверьте повторно "+ "или введите УИД вручную. Пока они не разобраны, новые чеки не отправляются.", len(queue.Attention))) return sum, nil } return run.incomes(ctx, queue.Incomes, deadline, sum) } // runner carries the authenticated session through one run so that a token which ages out mid-queue // can be renewed in place instead of derailing the income it happened to land on. type runner struct { e *Exporter client *mynalog.Client sess mynalog.Session reauthed bool } // reauth renews the session once per run. Twice would mean something other than expiry is wrong, // and looping on it would hammer the authentication endpoint. func (r *runner) reauth(ctx context.Context) error { if r.reauthed { return errors.New("mynalogsync: the session was already renewed once this run") } sess, client, err := r.e.session(ctx) if err != nil { return err } r.sess, r.client, r.reauthed = sess, client, true return nil } // call runs op, renewing the session and retrying once when the access token turns out to have aged // out mid-run. // // The retry is safe precisely because a rejected token is refused before the request is processed: // no receipt can have been created, so this is the one failure that may be repeated without a probe. // Treating it as an unknown outcome instead would halt the queue every hour for no reason. func (r *runner) call(ctx context.Context, op func() error) error { err := op() if !errors.Is(err, mynalog.ErrAuthExpired) || r.reauthed { return err } if authErr := r.reauth(ctx); authErr != nil { return fmt.Errorf("%w: renewing the session failed: %v", mynalog.ErrPermanent, authErr) } return op() } // handleSessionFailure turns a failed sign-in into an alert and, where the fault is unfixable by // repetition, takes the rail out of service. func (e *Exporter) handleSessionFailure(ctx context.Context, sum Summary, err error) (Summary, error) { switch { case errors.Is(err, ErrNotSignedIn), errors.Is(err, mynalog.ErrPermanent): reason := "сессия «Мой налог» недействительна — нужно войти заново" if pauseErr := e.payments.PauseMyNalog(ctx, reason); pauseErr != nil { return sum, pauseErr } sum.Paused = true sum.Stopped = reason e.raise(ctx, alertSignIn, "«Мой налог»: требуется вход", "Автоматическая выгрузка остановлена: сохранённая сессия больше не действует. "+ "Откройте раздел «Мой налог» в админ-консоли и войдите заново по логину и паролю.") return sum, nil default: sum.Stopped = "сервис «Мой налог» недоступен" e.noteDowntime(ctx) return sum, nil } } // cancels annuls receipts whose income was refunded. func (r *runner) cancels(ctx context.Context, cancels []payments.MyNalogCancel, deadline time.Time, sum Summary) (Summary, error) { e := r.e for _, c := range cancels { if e.now().After(deadline) { sum.Stopped = "время прогона вышло; остаток уйдёт следующим прогоном" return sum, nil } err := r.call(ctx, func() error { return r.client.CancelIncome(ctx, r.sess, c.ReceiptUUID, e.now().In(e.cfg.Location)) }) switch { case err == nil: if err := e.payments.MarkMyNalogCancelled(ctx, c.LedgerID, c.RefundLedgerID); err != nil { return sum, err } _ = e.payments.NoteMyNalogOK(ctx) sum.Cancelled++ case errors.Is(err, mynalog.ErrThrottled): sum.Stopped = "сервис «Мой налог» просит снизить темп" return sum, nil case errors.Is(err, mynalog.ErrPermanent): // The receipt stands and repeating will not change that; record it and let the operator // see it rather than hammering the same call. if markErr := e.payments.MarkMyNalogCancelFailed(ctx, c.LedgerID, c.RefundLedgerID, err.Error()); markErr != nil { return sum, markErr } sum.Failed++ default: sum.Stopped = "сервис «Мой налог» недоступен" e.noteDowntime(ctx) return sum, nil } if err := e.pause(ctx); err != nil { return sum, nil } } return sum, nil } // incomes files income, one at a time, until the queue empties or the budget runs out. func (r *runner) incomes(ctx context.Context, incomes []payments.MyNalogIncome, deadline time.Time, sum Summary) (Summary, error) { e := r.e streak := 0 for _, in := range incomes { if e.now().After(deadline) { sum.Stopped = "время прогона вышло; остаток уйдёт следующим прогоном" return sum, nil } name := e.ReceiptName(in) // Frozen before the call, and once per income: the name is the probe's only handle, and the // attempt counter should count incomes tried, not tokens renewed. if err := e.payments.BeginMyNalogSend(ctx, in, name); err != nil { return sum, err } var receipt string err := r.call(ctx, func() error { var callErr error receipt, callErr = r.client.AddIncome(ctx, r.sess, mynalog.IncomeRequest{ Name: name, Amount: in.Amount.Major(), OperationTime: in.OperationTime.In(e.cfg.Location), RequestTime: e.now().In(e.cfg.Location), }) return callErr }) switch { case err == nil: if err := e.payments.MarkMyNalogSent(ctx, in.LedgerID, receipt, false); err != nil { return sum, err } _ = e.payments.NoteMyNalogOK(ctx) sum.Registered++ streak = 0 case errors.Is(err, mynalog.ErrThrottled): // Nothing was filed; the row goes back into the queue for the next run. if markErr := e.payments.MarkMyNalogOutcome(ctx, in.LedgerID, payments.MyNalogFailed, err.Error()); markErr != nil { return sum, markErr } sum.Stopped = "сервис «Мой налог» просит снизить темп" return sum, nil case errors.Is(err, mynalog.ErrPermanent): if markErr := e.payments.MarkMyNalogOutcome(ctx, in.LedgerID, payments.MyNalogFailed, err.Error()); markErr != nil { return sum, markErr } sum.Failed++ streak++ if errors.Is(err, mynalog.ErrIncomeLimit) { return e.stop(ctx, sum, alertLimit, "превышен годовой лимит дохода по НПД", "«Мой налог» отклонил чек: превышен годовой лимит дохода. Это уже не техническая "+ "ошибка — режим НПД больше не применяется. Выгрузка остановлена.") } if streak >= permanentStreakLimit { return e.stop(ctx, sum, alertPaused, "«Мой налог» отклоняет чеки подряд", fmt.Sprintf("Подряд отклонено %d чеков — похоже, изменился формат API или данные. "+ "Выгрузка поставлена на паузу, чтобы не долбить сервис. Последняя ошибка: %v", streak, err)) } default: // The outcome is genuinely unknown: ask the tax service whether our receipt exists. if e.resolveByProbe(ctx, r.client, r.sess, in, name) { _ = e.payments.NoteMyNalogOK(ctx) sum.Registered++ streak = 0 break } if markErr := e.payments.MarkMyNalogOutcome(ctx, in.LedgerID, payments.MyNalogUnknown, err.Error()); markErr != nil { return sum, markErr } sum.Unknown++ sum.Stopped = "исход отправки не удалось установить; дальше — только вручную" e.raise(ctx, alertUnknown, "«Мой налог»: исход отправки неизвестен", fmt.Sprintf("Чек на сумму %s не удалось подтвердить: сервис не ответил, а поиск по "+ "названию «%s» ничего не нашёл. Выгрузка остановлена, чтобы не задвоить доход. "+ "В админ-консоли, раздел «Мой налог»: «проверить ещё раз», либо введите УИД "+ "вручную, либо верните приход в очередь. Ошибка: %v", in.Amount, name, err)) return sum, nil } if err := e.pause(ctx); err != nil { return sum, nil } } return sum, nil } // resolveByProbe asks the tax service whether the receipt we just failed to confirm exists after // all, recording it when it does. It reports whether the income ended up filed. // // This is the whole recovery story for a non-idempotent write: without it a timeout would leave // every ambiguous income to be sorted out by hand. func (e *Exporter) resolveByProbe(ctx context.Context, client *mynalog.Client, sess mynalog.Session, in payments.MyNalogIncome, name string) bool { from := in.OperationTime.In(e.cfg.Location).Add(-probeWindow) to := e.now().In(e.cfg.Location).Add(time.Minute) receipt, found, err := client.FindIncome(ctx, sess, name, from, to) if err != nil || !found { return false } if err := e.payments.MarkMyNalogSent(ctx, in.LedgerID, receipt, false); err != nil { e.log.Error("mynalog probe found the receipt but it could not be recorded", zap.String("ledger", in.LedgerID.String()), zap.Error(err)) return false } e.log.Info("mynalog receipt recovered by probe after an ambiguous call", zap.String("ledger", in.LedgerID.String()), zap.String("receipt", receipt)) return true } // stop takes the rail out of service, alerts, and ends the run. func (e *Exporter) stop(ctx context.Context, sum Summary, kind, reason, body string) (Summary, error) { if err := e.payments.PauseMyNalog(ctx, reason); err != nil { return sum, err } sum.Paused = true sum.Stopped = reason e.raise(ctx, kind, "«Мой налог»: выгрузка остановлена", body) return sum, nil } // pause waits out the gap between two calls, returning the context's error if it ends first. func (e *Exporter) pause(ctx context.Context) error { timer := time.NewTimer(e.cfg.Pace) defer timer.Stop() select { case <-ctx.Done(): return ctx.Err() case <-timer.C: return nil } } // noteDowntime alerts only once the tax service has been unreachable long enough to be worth a // letter. Below that threshold an outage is normal and the queue simply waits for it to pass. func (e *Exporter) noteDowntime(ctx context.Context) { stored, ok, err := e.payments.MyNalogSession(ctx) if err != nil || !ok { return } since := stored.UpdatedAt if stored.LastOKAt != nil { since = *stored.LastOKAt } if e.now().Sub(since) < downtimeAlertAfter { return } e.raise(ctx, alertDown, "«Мой налог»: сервис недоступен больше суток", fmt.Sprintf("Последняя успешная отправка была %s. Очередь ждёт; если сервис не поднимется, "+ "выгрузите чеки вручную из раздела «Мой налог» в админ-консоли.", since.In(e.cfg.Location).Format("2006-01-02 15:04"))) } // raise sends an operator alert unless one of the same kind went out within the last day. A // persistent fault should be one letter a day, not one per run. func (e *Exporter) raise(ctx context.Context, kind, subject, body string) { if e.alert == nil { return } stored, ok, err := e.payments.MyNalogSession(ctx) if err == nil && ok && stored.LastAlertKind == kind && stored.LastAlertAt != nil && e.now().Sub(*stored.LastAlertAt) < 24*time.Hour { return } if err := e.alert.Alert(ctx, kind, subject, body); err != nil { e.log.Error("mynalog alert could not be delivered", zap.String("kind", kind), zap.Error(err)) return } if err := e.payments.NoteMyNalogAlert(ctx, kind); err != nil { e.log.Error("mynalog alert could not be recorded", zap.String("kind", kind), zap.Error(err)) } } // Recheck probes the tax service once for an income whose outcome was never established, and // reports whether the receipt turned out to exist. func (e *Exporter) Recheck(ctx context.Context, ledgerID uuid.UUID) (bool, error) { in, ok, err := e.payments.MyNalogIncomeAt(ctx, ledgerID) if err != nil { return false, err } if !ok { return false, fmt.Errorf("mynalogsync: no such income") } release, locked, err := e.payments.TryLockMyNalog(ctx) if err != nil { return false, err } if !locked { return false, ErrBusy } defer release() sess, client, err := e.session(ctx) if err != nil { return false, err } return e.resolveByProbe(ctx, client, sess, in, e.ReceiptName(in)), nil } // RecordManual records an income the operator filed by hand in the tax cabinet, along with the // receipt identifier it was given. Such a row is thereafter treated like any other: in particular a // later refund annuls its receipt automatically, which is the whole reason the identifier is asked // for rather than merely ticking the income off. func (e *Exporter) RecordManual(ctx context.Context, ledgerID uuid.UUID, receiptUUID string) error { in, ok, err := e.payments.MyNalogIncomeAt(ctx, ledgerID) if err != nil { return err } if !ok { return fmt.Errorf("mynalogsync: no such income") } return e.payments.MarkMyNalogManual(ctx, in, e.ReceiptName(in), receiptUUID) } // Requeue puts an income whose outcome was unresolved back into the queue. It is the operator // asserting what the software may not assume: that the receipt really was never created. func (e *Exporter) Requeue(ctx context.Context, ledgerID uuid.UUID) error { return e.payments.MarkMyNalogOutcome(ctx, ledgerID, payments.MyNalogFailed, "returned to the queue by an operator") } // Skip rules an income out of the export for a reason the operator gives. func (e *Exporter) Skip(ctx context.Context, ledgerID uuid.UUID, reason string) error { in, ok, err := e.payments.MyNalogIncomeAt(ctx, ledgerID) if err != nil { return err } if !ok { return fmt.Errorf("mynalogsync: no such income") } return e.payments.MarkMyNalogNotRequired(ctx, in, reason) }