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}) }) } }