feat(bj): модуль ЛК на PostgreSQL, обмен с DEPO, боевой ЛК с имитатором НРД, QUIK, НПР
- bj_exchange и «кран» Fancy убраны: данные модуля ЛК в PostgreSQL (схема lk) - трансферные таблицы FTV_M2M: создание при установке, события, справочник участников - сверка с DEPO: счёт, ФИО, дата рождения, ИИС по договору, квал-инвестор из опции DEPO - боевой ЛК ЕСИА Финанс при тестовом ИШ: журнал, отказ через DEPO, статусы в ЛК - онлайн-лицензирование в bj, план тестирования M2M для контрагента (интерактивный HTML) - модули QUIK МНП, экспорт QUIK, НПР/УДС; документы и тестовые материалы Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
zuevav
@@ -0,0 +1,387 @@
|
||||
package lkgateway
|
||||
|
||||
// Путь заявки ЛК ЕСИА Финанс через шлюз bj (режим имитации НРД): шаги для карточки в
|
||||
// журнале исходящих заявок и замыкание отказа через Fancy.
|
||||
//
|
||||
// заявка ЛК → проверка шлюза → отказ: сделка «Отклонена», в DEPO FTV_M2M (OUT) + строки
|
||||
// + BJ_GATEWAY_REJECT по каждой строке → обработчик DEPO пометил события SUCCESS → ЛК «Отклонена»
|
||||
//
|
||||
// Статус в ЛК ставится только после успешной обработки в Fancy. Если обработчик ответил
|
||||
// ERROR, заявка ждёт: оповещение, шаг в карточке, после возврата записей в очередь путь
|
||||
// продолжается сам. Шаги хранятся по GUID перевода в lk-flow.json и показываются в
|
||||
// таймлайне карточки (buildClaimTimeline).
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.zetit.ru/zuevav/Bridge-and-Join-s/internal/m2m"
|
||||
"git.zetit.ru/zuevav/Bridge-and-Join-s/internal/m2mcore"
|
||||
)
|
||||
|
||||
const lkFlowFile = "/var/lib/bj/.bj/lk-flow.json"
|
||||
|
||||
// lkFlowDepoAlertAfter — через сколько без ответа Fancy поднимать оповещение (ждать продолжаем).
|
||||
const lkFlowDepoAlertAfter = 10 * time.Minute
|
||||
|
||||
// LKFlowStep — шаг пути заявки (Kind — класс шага таймлайна: db | crane | reject | gateway | response).
|
||||
type LKFlowStep struct {
|
||||
At time.Time `json:"at"`
|
||||
Kind string `json:"kind"`
|
||||
Title string `json:"title"`
|
||||
Detail string `json:"detail,omitempty"`
|
||||
}
|
||||
|
||||
// LKFlow — путь одной заявки ЛК через шлюз.
|
||||
type LKFlow struct {
|
||||
GUID string `json:"guid"`
|
||||
ClaimID int `json:"claim_id"`
|
||||
Stage string `json:"stage,omitempty"` // depo_pending | lk_pending | done | sent
|
||||
Code string `json:"code,omitempty"`
|
||||
Reason string `json:"reason,omitempty"` // подробно: журнал bj и Fancy
|
||||
Short string `json:"short,omitempty"` // коротко: комментарий клиенту в ЛК
|
||||
DepoAt time.Time `json:"depo_at,omitempty"` // когда отказ записан в DEPO
|
||||
DepoAlert bool `json:"depo_alert,omitempty"`
|
||||
DepoFail string `json:"depo_fail,omitempty"` // последняя ошибка обработчика DEPO (уже показана)
|
||||
LKReason string `json:"lk_reason,omitempty"`
|
||||
LKAlert bool `json:"lk_alert,omitempty"`
|
||||
Steps []LKFlowStep `json:"steps"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
type lkFlowStore struct {
|
||||
mu sync.Mutex
|
||||
items map[string]*LKFlow
|
||||
}
|
||||
|
||||
var globalLKFlows = loadLKFlows()
|
||||
|
||||
func loadLKFlows() *lkFlowStore {
|
||||
s := &lkFlowStore{items: map[string]*LKFlow{}}
|
||||
if b, err := os.ReadFile(lkFlowFile); err == nil {
|
||||
var list []*LKFlow
|
||||
if json.Unmarshal(b, &list) == nil {
|
||||
for _, f := range list {
|
||||
s.items[strings.ToUpper(f.GUID)] = f
|
||||
}
|
||||
}
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *lkFlowStore) saveLocked() {
|
||||
list := make([]*LKFlow, 0, len(s.items))
|
||||
for _, f := range s.items {
|
||||
list = append(list, f)
|
||||
}
|
||||
sort.Slice(list, func(i, j int) bool { return list[i].UpdatedAt.Before(list[j].UpdatedAt) })
|
||||
b, _ := json.MarshalIndent(list, "", " ")
|
||||
tmp := lkFlowFile + ".tmp"
|
||||
if os.WriteFile(tmp, b, 0o600) == nil {
|
||||
_ = os.Rename(tmp, lkFlowFile)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *lkFlowStore) Get(guid string) *LKFlow {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
f, ok := s.items[strings.ToUpper(guid)]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
cp := *f
|
||||
cp.Steps = append([]LKFlowStep(nil), f.Steps...)
|
||||
return &cp
|
||||
}
|
||||
|
||||
func (s *lkFlowStore) update(guid string, fn func(f *LKFlow)) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
k := strings.ToUpper(guid)
|
||||
f, ok := s.items[k]
|
||||
if !ok {
|
||||
f = &LKFlow{GUID: guid}
|
||||
s.items[k] = f
|
||||
}
|
||||
fn(f)
|
||||
f.UpdatedAt = time.Now()
|
||||
s.saveLocked()
|
||||
}
|
||||
|
||||
// step — добавить шаг пути (создаёт запись, если её нет).
|
||||
func (s *lkFlowStore) step(guid string, claimID int, kind, title, detail string) {
|
||||
s.update(guid, func(f *LKFlow) {
|
||||
if claimID != 0 {
|
||||
f.ClaimID = claimID
|
||||
}
|
||||
f.Steps = append(f.Steps, LKFlowStep{At: time.Now(), Kind: kind, Title: title, Detail: detail})
|
||||
})
|
||||
}
|
||||
|
||||
// stepIfExists — шаг только для заявок, путь которых уже ведётся.
|
||||
func (s *lkFlowStore) stepIfExists(guid, kind, title, detail string) {
|
||||
s.mu.Lock()
|
||||
_, ok := s.items[strings.ToUpper(guid)]
|
||||
s.mu.Unlock()
|
||||
if ok {
|
||||
s.step(guid, 0, kind, title, detail)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *lkFlowStore) pending() []*LKFlow {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var out []*LKFlow
|
||||
for _, f := range s.items {
|
||||
if f.Stage == "depo_pending" || f.Stage == "lk_pending" {
|
||||
cp := *f
|
||||
out = append(out, &cp)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// lkGatewayReject — отказ шлюза bj: сделка в журнале «Отклонена» с кодом, отказ уходит в
|
||||
// Fancy (DEPO). Статус в ЛК поставит lkFlowLoop, когда обработчик DEPO успешно обработает отказ.
|
||||
// short — короткий комментарий для клиента в ЛК, reason — подробный для журнала и Fancy.
|
||||
func lkGatewayReject(ctx context.Context, rc *RuntimeConfig, svc *Service, id int, investorID string, req *m2m.M2MTransferRequest, code, short, reason string) {
|
||||
guid := string(req.Header.GUID)
|
||||
deal, err := svc.Repo().GetByGUID(ctx, req.Header.GUID)
|
||||
if err != nil || deal == nil {
|
||||
deal = nil
|
||||
if nd, e := m2mcore.NewDeal(req.Header.GUID, investorID, nil); e == nil {
|
||||
nd.Request = req
|
||||
if saved, e := svc.Repo().Create(ctx, nd); e == nil {
|
||||
deal = saved
|
||||
} else {
|
||||
log.Printf("lk-gateway: ЛК #%d: сделка для журнала не создана: %v", id, e)
|
||||
}
|
||||
}
|
||||
}
|
||||
if deal != nil && !m2mcore.IsTerminal(deal.State) {
|
||||
if deal.Request == nil {
|
||||
deal.Request = req
|
||||
}
|
||||
if e := deal.RejectManually(ctx, "bj-gateway", code, reason); e != nil {
|
||||
log.Printf("lk-gateway: ЛК #%d: отказ по сделке %s: %v", id, shortGUID(guid), e)
|
||||
}
|
||||
if e := svc.Repo().Update(ctx, deal); e != nil {
|
||||
log.Printf("lk-gateway: ЛК #%d: сохранение отказа: %v", id, e)
|
||||
}
|
||||
}
|
||||
lkMarkSent(id)
|
||||
lkRecordOrigin(guid, id)
|
||||
globalLKFlows.update(guid, func(f *LKFlow) {
|
||||
f.ClaimID, f.Stage, f.Code, f.Reason, f.Short = id, "depo_pending", code, reason, short
|
||||
})
|
||||
lkFlowWriteDepo(rc, guid, req)
|
||||
}
|
||||
|
||||
// lkFlowWriteDepo — отказ шлюза в Fancy: шапка FTV_M2M (OUT) + строки + BJ_GATEWAY_REJECT
|
||||
// по каждой строке. Идемпотентно (шапка по GUID, события по GUID+REFERENCE+EVENT_TYPE).
|
||||
func lkFlowWriteDepo(rc *RuntimeConfig, guid string, req *m2m.M2MTransferRequest) bool {
|
||||
f := globalLKFlows.Get(guid)
|
||||
if f == nil || req == nil {
|
||||
return false
|
||||
}
|
||||
inv := req.Data.InvestorInformation
|
||||
ser := ""
|
||||
if inv.IdentityDocument.DocumentSeries != nil {
|
||||
ser = string(*inv.IdentityDocument.DocumentSeries)
|
||||
}
|
||||
num := string(inv.IdentityDocument.DocumentNumber)
|
||||
place := rc.Snapshot().NSD.SectionID
|
||||
var lines []depoLine
|
||||
for _, s := range req.Data.TransferredSecurities.Securities {
|
||||
lines = append(lines, depoLine{Ref: string(s.ReferenceID), ISIN: quikSecISIN(s), Code: string(s.SecurityCode), Amount: quikSecQty(s.Quantity), Place: place})
|
||||
}
|
||||
inn := string(req.Data.ReceivingDepository.INN)
|
||||
name := strings.TrimSpace(strings.Join([]string{inv.LastName, inv.FirstName, inv.MiddleName}, " "))
|
||||
comment := fmt.Sprintf("исходящий перевод M2M из ЛК #%d, получатель ИНН %s — отклонён шлюзом bj до отправки в НРД", f.ClaimID, inn)
|
||||
if !depoUpsertOrder(rc, guid, "OUT", handbookCodeByINN(inn), name, strings.TrimSpace(ser+" "+num), ser, num, lines, comment) {
|
||||
return false
|
||||
}
|
||||
for _, ln := range lines {
|
||||
depoEvent(rc, guid, ln.Ref, "BJ_GATEWAY_REJECT", f.Code, f.Reason)
|
||||
}
|
||||
globalLKFlows.update(guid, func(x *LKFlow) {
|
||||
if !x.DepoAt.IsZero() {
|
||||
return
|
||||
}
|
||||
x.DepoAt = time.Now()
|
||||
x.Steps = append(x.Steps, LKFlowStep{At: x.DepoAt, Kind: "db", Title: "Отказ передан в Fancy (DEPO)",
|
||||
Detail: fmt.Sprintf("FTV_M2M (OUT) и событие BJ_GATEWAY_REJECT по строкам: %d. Ждём обработку в DEPO.", len(lines))})
|
||||
})
|
||||
return true
|
||||
}
|
||||
|
||||
// lkDepoAnswer — ответ Fancy на отказ по GUID. done — все BJ_GATEWAY_REJECT обработаны успешно;
|
||||
// failed — текст ошибки обработчика (ERROR по событию, поручению или строке); events — сколько
|
||||
// событий отказа записано.
|
||||
func lkDepoAnswer(ctx context.Context, rc *RuntimeConfig, guid, ourComment string) (done bool, failed string, events int, text string, err error) {
|
||||
db, err := depoEventsDB(rc)
|
||||
if err != nil {
|
||||
return false, "", 0, "", err
|
||||
}
|
||||
defer db.Close()
|
||||
g := strings.ToUpper(guid)
|
||||
rows, err := db.QueryContext(ctx, "SELECT ISNULL(STATE,''), ISNULL(COMMENT,'') FROM dbo.FTV_M2M_EVENTS WHERE GUID=@g AND EVENT_TYPE='BJ_GATEWAY_REJECT'", sql.Named("g", g))
|
||||
if err != nil {
|
||||
return false, "", 0, "", err
|
||||
}
|
||||
todo := 0
|
||||
var states, comments, errs []string
|
||||
for rows.Next() {
|
||||
var st, cm string
|
||||
if rows.Scan(&st, &cm) != nil {
|
||||
continue
|
||||
}
|
||||
events++
|
||||
cm = strings.TrimSpace(strings.TrimPrefix(cm, ourComment))
|
||||
switch strings.ToUpper(st) {
|
||||
case "", "TODO":
|
||||
todo++
|
||||
continue
|
||||
case "ERROR":
|
||||
if cm != "" && !containsStr(errs, cm) {
|
||||
errs = append(errs, cm)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !containsStr(states, st) {
|
||||
states = append(states, st)
|
||||
}
|
||||
if cm != "" && !containsStr(comments, cm) {
|
||||
comments = append(comments, cm)
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
|
||||
docCol := "NULL"
|
||||
if depoHasColumn(ctx, db, "FTV_M2M", "DOC") {
|
||||
docCol = "CAST(m.DOC AS varchar(32))"
|
||||
}
|
||||
var doc, mState, mComment, lineErr string
|
||||
_ = db.QueryRowContext(ctx, `SELECT ISNULL(`+docCol+`,''), ISNULL(m.STATE,''), ISNULL(m.COMMENT,''),
|
||||
ISNULL((SELECT TOP 1 ISNULL(l.COMMENT,'') FROM dbo.FTV_M2M_LINES l WHERE l.GUID=m.GUID AND ISNULL(l.STATE,'')='ERROR'),'')
|
||||
FROM dbo.FTV_M2M m WHERE m.GUID=@g`, sql.Named("g", g)).Scan(&doc, &mState, &mComment, &lineErr)
|
||||
if strings.EqualFold(mState, "ERROR") && mComment != "" && !containsStr(errs, mComment) {
|
||||
errs = append(errs, mComment)
|
||||
}
|
||||
if lineErr != "" && !containsStr(errs, lineErr) {
|
||||
errs = append(errs, lineErr)
|
||||
}
|
||||
|
||||
done = events > 0 && todo == 0 && len(errs) == 0
|
||||
failed = strings.Join(errs, "; ")
|
||||
var parts []string
|
||||
if done {
|
||||
parts = append(parts, "событие отказа обработано в DEPO ("+strings.Join(states, ", ")+")")
|
||||
}
|
||||
if doc != "" && doc != "0" {
|
||||
parts = append(parts, "документ DEPO №"+doc)
|
||||
}
|
||||
if len(comments) > 0 {
|
||||
parts = append(parts, "комментарий DEPO: "+strings.Join(comments, "; "))
|
||||
}
|
||||
return done, failed, events, strings.Join(parts, "; "), nil
|
||||
}
|
||||
|
||||
// lkFlowLoop — доводит отказы шлюза до ЛК: ждёт успешной обработки в Fancy и ставит «Отклонена».
|
||||
func (s *Server) lkFlowLoop(ctx context.Context) {
|
||||
t := time.NewTicker(15 * time.Second)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
}
|
||||
for _, f := range globalLKFlows.pending() {
|
||||
lkFlowAdvance(ctx, s.rc, s.svc, f)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func lkFlowAdvance(ctx context.Context, rc *RuntimeConfig, svc *Service, f *LKFlow) {
|
||||
cctx, cancel := context.WithTimeout(ctx, 40*time.Second)
|
||||
defer cancel()
|
||||
if f.Stage == "depo_pending" {
|
||||
var (
|
||||
done bool
|
||||
failed, text string
|
||||
events int
|
||||
err error
|
||||
)
|
||||
if !f.DepoAt.IsZero() {
|
||||
done, failed, events, text, err = lkDepoAnswer(cctx, rc, f.GUID, f.Reason)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
if f.DepoAt.IsZero() || events == 0 { // в DEPO ещё не записано (база была недоступна) — повторяем
|
||||
if deal, e := svc.Repo().GetByGUID(cctx, m2m.UUID(f.GUID)); e == nil && deal != nil && deal.Request != nil {
|
||||
lkFlowWriteDepo(rc, f.GUID, deal.Request)
|
||||
}
|
||||
return
|
||||
}
|
||||
if failed != "" { // обработчик DEPO ответил ошибкой — в ЛК не пишем, ждём возврата в очередь
|
||||
if failed != f.DepoFail {
|
||||
globalLKFlows.update(f.GUID, func(x *LKFlow) {
|
||||
x.DepoFail = failed
|
||||
x.Steps = append(x.Steps, LKFlowStep{At: time.Now(), Kind: "reject", Title: "Fancy не смог обработать отказ",
|
||||
Detail: failed + ". Статус в ЛК не меняется: устраните причину и верните записи в очередь (Обмен с DEPO)."})
|
||||
})
|
||||
raiseSystemAlert(rc, fmt.Sprintf("Fancy: ошибка обработки отказа по заявке ЛК #%d", f.ClaimID),
|
||||
fmt.Sprintf("Перевод %s: %s. Статус в ЛК не изменён. Устраните причину и нажмите «Вернуть в очередь» на странице «Обмен с DEPO».", shortGUID(f.GUID), failed))
|
||||
log.Printf("lk-gateway: ЛК #%d (%s): обработчик DEPO вернул ошибку — %s", f.ClaimID, shortGUID(f.GUID), failed)
|
||||
}
|
||||
return
|
||||
}
|
||||
if !done {
|
||||
if !f.DepoAlert && time.Since(f.DepoAt) > lkFlowDepoAlertAfter {
|
||||
globalLKFlows.update(f.GUID, func(x *LKFlow) { x.DepoAlert = true })
|
||||
raiseSystemAlert(rc, fmt.Sprintf("Fancy не обработал отказ по заявке ЛК #%d", f.ClaimID),
|
||||
fmt.Sprintf("Перевод %s: событие BJ_GATEWAY_REJECT дольше %s ждёт обработки. Статус в ЛК не изменён — проверьте обработку DEPO.", shortGUID(f.GUID), lkFlowDepoAlertAfter))
|
||||
}
|
||||
return
|
||||
}
|
||||
// клиенту — только короткий итог проверки; подробности остаются в журнале bj и в Fancy
|
||||
lkReason := f.Short
|
||||
if lkReason == "" {
|
||||
lkReason, _, _ = strings.Cut(f.Reason, ". Причина:")
|
||||
}
|
||||
globalLKFlows.update(f.GUID, func(x *LKFlow) {
|
||||
x.Stage, x.LKReason, x.DepoFail = "lk_pending", lkReason, ""
|
||||
x.Steps = append(x.Steps, LKFlowStep{At: time.Now(), Kind: "response", Title: "Ответ Fancy (DEPO): отказ обработан", Detail: text})
|
||||
})
|
||||
log.Printf("lk-gateway: ЛК #%d (%s): Fancy обработал отказ — %s", f.ClaimID, shortGUID(f.GUID), text)
|
||||
f.Stage, f.LKReason = "lk_pending", lkReason
|
||||
}
|
||||
if f.Stage == "lk_pending" {
|
||||
ep, ok := lkActiveEndpoint(rc)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
pc, pb := lkPatchRejected(cctx, ep, f.ClaimID, f.LKReason)
|
||||
if pc >= 400 || pc == 0 {
|
||||
if !f.LKAlert {
|
||||
globalLKFlows.update(f.GUID, func(x *LKFlow) { x.LKAlert = true })
|
||||
raiseSystemAlert(rc, "ЛК: не удалось отклонить заявку", fmt.Sprintf("Заявка ЛК #%d: Fancy отказ обработал, но статус в ЛК не изменён: HTTP %d %s. bj повторяет каждые 15 с.", f.ClaimID, pc, pb))
|
||||
}
|
||||
return
|
||||
}
|
||||
globalLKFlows.update(f.GUID, func(x *LKFlow) {
|
||||
x.Stage = "done"
|
||||
x.Steps = append(x.Steps, LKFlowStep{At: time.Now(), Kind: "reject", Title: "Статус в ЛК: «Отклонена»", Detail: f.LKReason})
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user