Bridge-and-Join-s/internal/lkgateway/lkflow.go

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"
	"git.zetit.ru/zuevav/Bridge-and-Join-s/internal/lkgateway/fsql"
	"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, fsql.QFl01, 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, fsql.Flow01+docCol+fsql.Flow02, 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})
		})
	}
}