package lkgateway import ( "context" "encoding/json" "log" "os" "strings" "time" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/m2m" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/nsdadapter/igw" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/nsdxml" ) // pollerStateFile — постоянная память id применённых входящих пакетов ИШ. // Без неё перезапуск bj-server заново вычитывал бы старые ответы НРД и // повторно применял их к заявкам (для ответов с нулевым GUID это приводило // к ложным «отклонениям» по FIFO). Файл переживает перезапуски. const pollerStateFile = "/var/lib/bj/.bj/poller-processed.json" // loadProcessed читает множество обработанных id; второй результат false, // если файла нет (самый первый запуск). func loadProcessed() (map[int]bool, bool) { m := make(map[int]bool) b, err := os.ReadFile(pollerStateFile) if err != nil { return m, false } var ids []int if json.Unmarshal(b, &ids) != nil { return m, false } for _, id := range ids { m[id] = true } return m, true } // saveProcessed атомарно сохраняет множество обработанных id. func saveProcessed(m map[int]bool) { ids := make([]int, 0, len(m)) for id := range m { ids = append(ids, id) } b, err := json.Marshal(ids) if err != nil { return } tmp := pollerStateFile + ".tmp" if os.WriteFile(tmp, b, 0o640) == nil { _ = os.Rename(tmp, pollerStateFile) } } // pollIncoming периодически опрашивает ИШ на входящие пакеты от НРД // (M2MTransferDecision / Response) и применяет их через svc.ApplyDecision. // Замыкает цикл: bj-server отправил заявку → ИШ → НРД → робот ответил → // ИШ забрал ответ во входящие → этот поллер применяет Decision (статус // заявки переходит в confirmed/rejected, срабатывает callback в ЛК). // // Дедупликация по id обработанных пакетов: ИШ возвращает их повторно, // пока мы не подтвердим, поэтому держим множество уже обработанных. func (s *Server) pollIncoming(ctx context.Context) { const interval = 10 * time.Second // синхронно с fetchInterval ИШ (10с) ticker := time.NewTicker(interval) defer ticker.Stop() processed, existed := loadProcessed() if !existed { // Первый запуск: помечаем все уже лежащие во входящих пакеты как // обработанные, чтобы не применять исторический backlog (старые ответы // НРД) к текущим заявкам. Реальные новые ответы придут позже. s.seedProcessed(ctx, processed) saveProcessed(processed) } log.Printf("lk-gateway: поллер входящих ИШ запущен (канал %s, интервал %s, обработанных в памяти %d)", s.igwChannel, interval, len(processed)) for { select { case <-ctx.Done(): return case <-ticker.C: healthBeat("poll-incoming") s.fetchAndApply(ctx, processed) } } } // fetchAndApply — один проход поллера: список входящих → для каждого нового // забираем тело, распаковываем, парсим Decision, применяем. func (s *Server) fetchAndApply(ctx context.Context, processed map[int]bool) { cctx, cancel := context.WithTimeout(ctx, 25*time.Second) defer cancel() // Тип не указываем — ИШ вернёт оба (M2MTD + M2MER). Date=сегодня. pkgs, err := s.igwClient.ListIncoming(cctx, igw.ListFilter{ Channel: s.igwChannel, Date: time.Now(), }) if err != nil { log.Printf("lk-gateway: поллер ListIncoming: %v", err) return } for _, p := range pkgs { if processed[p.ID] { continue } if err := s.applyIncoming(cctx, p); err != nil { log.Printf("lk-gateway: поллер пакет id=%d (%s): %v", p.ID, p.Type, err) continue // не помечаем обработанным — повторим в след. раз } processed[p.ID] = true saveProcessed(processed) // переживает перезапуск log.Printf("lk-gateway: поллер применил входящий пакет id=%d тип=%s", p.ID, p.Type) } } // seedProcessed помечает все текущие входящие пакеты как обработанные // (используется при самом первом запуске поллера). func (s *Server) seedProcessed(ctx context.Context, processed map[int]bool) { cctx, cancel := context.WithTimeout(ctx, 25*time.Second) defer cancel() pkgs, err := s.igwClient.ListIncoming(cctx, igw.ListFilter{Channel: s.igwChannel, Date: time.Now()}) if err != nil { log.Printf("lk-gateway: поллер seed: ListIncoming: %v", err) return } for _, p := range pkgs { processed[p.ID] = true } log.Printf("lk-gateway: поллер первый запуск — засеяно %d существующих пакетов как обработанные", len(pkgs)) } // applyIncoming забирает тело пакета и применяет M2M-ответ к сделке. // Среди входящих от НРД много служебных пакетов (квитанции ЭДО типа C/Z, // конверты) — они не M2M-документы и пропускаются. Реальные ответы — // M2MTransferDecision (решение принимающей стороны) и M2MTransferResponse // (ответ сервиса МОСТ, в т.ч. ошибки M2Mxx). func (s *Server) applyIncoming(ctx context.Context, p igw.Package) error { zipBytes, err := s.igwClient.GetPackage(ctx, p.ID) if err != nil { return err // сетевая ошибка — повторим в след. раз } unpacked, err := igw.UnpackPackage(zipBytes) if err != nil { // Нет основного .xml — служебный пакет (квитанция/конверт ЭДО). // Не ошибка: помечаем обработанным, чтобы не повторять. log.Printf("lk-gateway: поллер пакет id=%d (%s) — служебный (квитанция/конверт), пропуск", p.ID, p.Type) return nil } doc := string(unpacked.DocXML) switch { case strings.Contains(doc, "M2MTransferDecision"): decision, err := igw.ParseDecision(unpacked.DocXML) if err != nil { log.Printf("lk-gateway: поллер Decision id=%d: разбор: %v", p.ID, err) return nil } s.applyDecisionEffects(ctx, decision, "поллер", "") return nil case strings.Contains(doc, "M2MTransferResponse"): resp, err := igw.ParseResponse(unpacked.DocXML) if err != nil { log.Printf("lk-gateway: поллер Response id=%d: разбор: %v", p.ID, err) return nil } // Ответ сервиса МОСТ: статус + код (M2Mxx). Применяем к сделке: // INFO — приём в обработку (статус не меняется), ERROR — отказ сервиса // (напр. M2M14 — отправитель не в справочнике), сделка → Отклонена. // Ответ сохраняется в сделке и виден в карточке заявки. var codes string for _, rr := range resp.Responses { codes += string(rr.Code) + " " } log.Printf("lk-gateway: поллер M2MTransferResponse id=%d: статус=%s коды=[%s] GUID=%s", p.ID, resp.StatusCode, strings.TrimSpace(codes), resp.GUID) if err := s.svc.ApplyServiceResponse(ctx, resp, unpacked.DocXML); err != nil { // Сделка может быть не найдена (ответ на чужой/старый GUID) — // логируем, но помечаем обработанным, чтобы не зациклиться. log.Printf("lk-gateway: поллер ApplyServiceResponse id=%d GUID=%s: %v", p.ID, resp.GUID, err) } // ERROR сервиса МОСТ — отказ: замыкаем на ЛК с человекочитаемой причиной. if strings.EqualFold(string(resp.StatusCode), "ERROR") { reason := "Отклонено сервисом MOEX МОСТ" if c := strings.TrimSpace(codes); c != "" { parts := make([]string, 0) for _, code := range strings.Fields(c) { parts = append(parts, m2mCodeLabel(code)) } reason = strings.Join(parts, "; ") } // Итог для заявки модуля ЛК (lk.transfer_requests): 'declined' — отказ после отправки в НРД. if string(resp.GUID) != "" && string(resp.GUID) != zeroGUID { if e := lkFinalizeByGUID(ctx, s.rc, string(resp.GUID), "declined", "rejected", reason); e != nil { log.Printf("lk-gateway: поллер: итог заявки модуля ЛК GUID=%s: %v", resp.GUID, e) } } OnLKOutcome(ctx, s.rc, string(resp.GUID), "rejected", reason) quikOnServiceError(ctx, s.rc, string(resp.GUID), reason) go depoOnServiceError(s.rc, string(resp.GUID), codes, reason) } return nil case strings.Contains(doc, "M2MTransferRequest"): // Входящий запрос на перевод — другая сторона (робот в сценарии 3333) // выступает ПЕРЕДАЮЩЕЙ. Мы должны ответить M2MTransferDecision. Сохраняем // во «входящие переводы»; решение оператор формирует в UI. var req m2m.M2MTransferRequest // Робот шлёт корень без namespace (префиксы только // на детях) — подставляем default namespace request для корня. if err := nsdxml.UnmarshalDefaultNS(unpacked.DocXML, &req, "http://nsd.ru/schemas/m2m/request"); err != nil { log.Printf("lk-gateway: поллер входящий Request id=%d: разбор: %v", p.ID, err) return nil } it, existed := s.incoming.AddFromRequest(&req, string(nsdxml.DecodeWindows1251(unpacked.DocXML))) if existed { log.Printf("lk-gateway: поллер входящий перевод GUID=%s уже сохранён (id=%s)", it.GUID, it.ID) } else { // Каждый новый GUID — ОТДЕЛЬНЫЙ перевод и требует решения оператора. // Совпадение по «отпечатку» (тот же инвестор и бумаги) НЕ значит дубль: // один и тот же человек может законно перевести 10 акций, затем ещё 10 // (разные GUID = разные отправки). Поэтому авто-ответа нет — только // информируем, если запрос похож на ранее отвеченный. if dup, ok := s.incoming.FindAnsweredDuplicate(it.Fingerprint, it.ID); ok && it.Fingerprint != "" { log.Printf("lk-gateway: поллер входящий GUID=%s похож по составу на ранее отвеченный (id=%s) — "+ "это отдельный перевод, ждёт решения", it.GUID, dup.ID) } else { log.Printf("lk-gateway: поллер НОВЫЙ входящий перевод от %s GUID=%s id=%s — ждёт решения", it.SenderCode, it.GUID, it.ID) } // Уведомляем депозитарий по почте (к вам переведены бумаги — учесть). go notifyDepositoryIncoming(s.rc, it) // DEPO (bj2depo): шапка + строки + BJ_RECEIVED — поручение в DEPO создаётся сразу. go depoOnIncomingReceived(s.rc, it) // Сверка с DEPO — только рекомендация для оператора (эмуляция ответа), // в статус попадает бейдж «сверка: принять / отказ M2Mxx». Не отвечаем. go func(item *IncomingTransfer) { if r := runDepoCheck(item); r.Ran { s.incoming.SetDepo(item.ID, r.Recommend, r.Summary) log.Printf("lk-gateway: сверка DEPO по входящему GUID=%s: %s (не отправлено)", item.GUID, r.Summary) } else { log.Printf("lk-gateway: сверка DEPO по входящему GUID=%s не выполнена: %s", item.GUID, r.Err) } }(it) } return nil case strings.Contains(doc, "M2MTransferHandbook") && !strings.Contains(doc, "M2MTransferHandbookRequest"): // Справочник участников M2M (ответ НРД на #M2MHR). Разбираем и сохраняем // — это карта ИНН → MC-код участника и перечень мест расчётов. var hb m2m.M2MTransferHandbook if err := nsdxml.Unmarshal(unpacked.DocXML, &hb); err != nil { log.Printf("lk-gateway: поллер справочник M2M id=%d: разбор: %v", p.ID, err) return nil } saveHandbook(&hb) go func() { if err := depoWriteHandbook(s.rc); err != nil { log.Printf("depo-events: справочник после получения: %v", err) } }() log.Printf("lk-gateway: поллер СПРАВОЧНИК M2M получен id=%d: участников %d, мест расчётов %d", p.ID, len(hb.Participants.Participants), len(hb.SettlementPlaces.Places)) return nil default: log.Printf("lk-gateway: поллер пакет id=%d (%s) — неизвестный M2M-документ, пропуск", p.ID, p.Type) return nil } } // rejectCodesLabel собирает человекочитаемые причины отказа по кодам M2M. func rejectCodesLabel(codes []string) string { if len(codes) == 0 { return "без кода" } seen := map[string]bool{} var parts []string for _, c := range codes { if c == "" || seen[c] { continue } seen[c] = true parts = append(parts, m2mCodeLabel(c)) } return strings.Join(parts, "; ") } // handbookFile — последний полученный справочник участников M2M (для UI/поиска // ИНН→MC-код контрагента). const handbookFile = "/var/lib/bj/.bj/m2m-handbook.json" // HandbookEntry — плоская запись справочника для сохранения/показа. type HandbookEntry struct { INN string `json:"inn"` Code string `json:"code"` // MC-код участника (депозитарий/брокер) Kind string `json:"kind"` // "depo" | "broker" | "place" Name string `json:"name"` // краткое (рус.) FullName string `json:"full_name,omitempty"` DisplayName string `json:"display_name,omitempty"` EngName string `json:"eng_name,omitempty"` } // saveHandbook раскладывает справочник в плоский список и пишет в файл. func saveHandbook(hb *m2m.M2MTransferHandbook) { out := struct { At string `json:"at"` Participants []HandbookEntry `json:"participants"` Places []HandbookEntry `json:"settlement_places"` }{At: time.Now().Format(time.RFC3339)} for _, p := range hb.Participants.Participants { e := HandbookEntry{INN: string(p.INN), Name: p.Names.Rus.ShortName, FullName: p.Names.Rus.FullName, DisplayName: p.Names.Rus.DisplayName} if p.Names.Eng != nil { e.EngName = p.Names.Eng.DisplayName if e.EngName == "" { e.EngName = p.Names.Eng.FullName } } if e.Name == "" { e.Name = e.DisplayName } if p.DepositoryPlace != nil { e.Code, e.Kind = string(p.DepositoryPlace.ParticipantCode), "depo" } else if p.BrokerPlace != nil { e.Code, e.Kind = string(p.BrokerPlace.ParticipantCode), "broker" } out.Participants = append(out.Participants, e) log.Printf("lk-gateway: справочник M2M: ИНН %s → %s [%s] %s", e.INN, e.Code, e.Kind, e.Name) } for _, pl := range hb.SettlementPlaces.Places { out.Places = append(out.Places, HandbookEntry{INN: string(pl.INN), Name: pl.ShortName, DisplayName: pl.DisplayName, Kind: "place"}) } if raw, err := json.Marshal(out); err == nil { _ = os.WriteFile(handbookFile, raw, 0o600) } }