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

package lkgateway

// Контроль сроков депо-поручений 16 / 16/1 (регламент депозитария, 09.09.2026):
//   - отчёт MS010 ожидается в течение 5 минут после подачи; нет — оповещение
//     «не исполнено в срок»;
//   - далее исполнение ожидается до 30 дней; по истечении — поручение считается
//     отменённым (оповещение, по исходящему снимается блокировка в QUIK).
// Состояние — /var/lib/bj/.bj/order-watch.json (чтобы оповещать один раз).

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"git.zetit.ru/zuevav/Bridge-and-Join-s/internal/m2mcore"
)

const (
	orderWatchFile   = "/var/lib/bj/.bj/order-watch.json"
	orderMS010Within = 5 * time.Minute
	orderExecWithin  = 30 * 24 * time.Hour
)

type OrderWatch struct {
	GUID      string    `json:"guid"`
	Direction string    `json:"direction"` // out (16) | in (16/1)
	OrderAt   time.Time `json:"order_at"`
	MS010At   time.Time `json:"ms010_at,omitempty"`   // все строки получили MS010
	Rejected  bool      `json:"rejected,omitempty"`   // хотя бы одна строка отвергнута
	LateAt    time.Time `json:"late_at,omitempty"`    // когда зафиксировали просрочку 5 мин
	ExpiredAt time.Time `json:"expired_at,omitempty"` // все незакрытые строки истекли (30 дней)
	// Lines — состояние по каждой строке пакета (одно депо-поручение = одна бумага),
	// ключ — contr_d_id. По договорённости с Fancy (18.09.2026) исходы пишутся в DEPO
	// по строкам с REFERENCE и сразу по MS010 этой строки, не дожидаясь остальных:
	// пакет может исполниться частично (Сбер — да, Т-Технологии — отказ).
	Lines map[string]*OrderLineWatch `json:"lines,omitempty"`
}

// OrderLineWatch — одна строка поручения: какая бумага (REFERENCE/ISIN) и её исход.
type OrderLineWatch struct {
	Ref       string    `json:"ref,omitempty"` // REFERENCE строки в FTV_M2M_LINES
	ISIN      string    `json:"isin,omitempty"`
	MS010At   time.Time `json:"ms010_at,omitempty"`
	Rejected  bool      `json:"rejected,omitempty"`
	ExpiredAt time.Time `json:"expired_at,omitempty"`
}

// orderLine — строка поручения для наблюдателя: contr_d_id ↔ бумага (REFERENCE, ISIN).
type orderLine struct {
	ContrDID string
	Ref      string
	ISIN     string
}

// lineDone — строка в финальном состоянии (MS010 получен или срок истёк).
func (l *OrderLineWatch) lineDone() bool {
	return l != nil && (!l.MS010At.IsZero() || !l.ExpiredAt.IsZero())
}

// linesSummary — сколько строк закрыто MS010 / отвергнуто / истекло из общего числа.
// Нужно для частичного исполнения: пакет может быть исполнен не целиком.
func (w *OrderWatch) linesSummary() (total, executed, rejected, expired int) {
	if w == nil {
		return
	}
	for _, l := range w.Lines {
		total++
		switch {
		case !l.MS010At.IsZero() && l.Rejected:
			rejected++
		case !l.MS010At.IsZero():
			executed++
		case !l.ExpiredAt.IsZero():
			expired++
		}
	}
	return
}

// Состояние по-русски для карточек.
func (w *OrderWatch) StatusRu() string {
	if w == nil {
		return ""
	}
	// Частичное исполнение: часть строк закрыта, часть ещё ждёт / отвергнута / истекла.
	if total, executed, rejected, expired := w.linesSummary(); total > 1 && executed > 0 && executed < total {
		s := fmt.Sprintf("исполнено частично: %d из %d бумаг", executed, total)
		if rejected > 0 {
			s += fmt.Sprintf(", отвергнуто %d", rejected)
		}
		if expired > 0 {
			s += fmt.Sprintf(", истёк срок по %d", expired)
		}
		if executed+rejected+expired < total {
			s += fmt.Sprintf(", ожидаем MS010 по %d", total-executed-rejected-expired)
		}
		return s
	}
	switch {
	case !w.MS010At.IsZero() && w.Rejected:
		return "НРД отверг поручение (MS010 " + w.MS010At.Local().Format("02.01 15:04") + ")"
	case !w.MS010At.IsZero():
		late := ""
		if w.MS010At.Sub(w.OrderAt) > orderMS010Within {
			late = fmt.Sprintf(", с опозданием %s", w.MS010At.Sub(w.OrderAt).Round(time.Minute))
		}
		return "исполнено: MS010 получен " + w.MS010At.Local().Format("02.01.2006 15:04") + late
	case !w.ExpiredAt.IsZero():
		return "СРОК 30 ДНЕЙ ИСТЁК " + w.ExpiredAt.Local().Format("02.01.2006") + " — поручение считается отменённым"
	case !w.LateAt.IsZero():
		return "НЕ ИСПОЛНЕНО В СРОК 5 МИН (подано " + w.OrderAt.Local().Format("02.01 15:04") + "); ожидаем исполнения до " + w.OrderAt.Add(orderExecWithin).Local().Format("02.01.2006")
	default:
		return "ожидаем MS010 до " + w.OrderAt.Add(orderMS010Within).Local().Format("15:04")
	}
}

func (w *OrderWatch) Kind() string {
	if w != nil {
		// частично исполнено — предупреждение, а не «отказ» и не «ок»
		if total, executed, _, _ := w.linesSummary(); total > 1 && executed > 0 && executed < total {
			return "warn"
		}
	}
	switch {
	case w == nil:
		return ""
	case !w.MS010At.IsZero() && w.Rejected:
		return "bad"
	case !w.MS010At.IsZero():
		return "ok"
	case !w.ExpiredAt.IsZero():
		return "bad"
	case !w.LateAt.IsZero():
		return "warn"
	}
	return "wait"
}

type orderWatchStore struct {
	mu    sync.Mutex
	items map[string]*OrderWatch
}

var globalOrderWatch = loadOrderWatch()

func loadOrderWatch() *orderWatchStore {
	s := &orderWatchStore{items: map[string]*OrderWatch{}}
	if b, err := os.ReadFile(orderWatchFile); err == nil {
		var list []*OrderWatch
		if json.Unmarshal(b, &list) == nil {
			for _, w := range list {
				s.items[w.GUID] = w
			}
		}
	}
	return s
}

func (s *orderWatchStore) saveLocked() {
	list := make([]*OrderWatch, 0, len(s.items))
	for _, w := range s.items {
		list = append(list, w)
	}
	b, _ := json.MarshalIndent(list, "", "  ")
	_ = os.MkdirAll(filepath.Dir(orderWatchFile), 0o755)
	tmp := orderWatchFile + ".tmp"
	if os.WriteFile(tmp, b, 0o640) == nil {
		_ = os.Rename(tmp, orderWatchFile)
	}
}

// Get — состояние контроля по GUID (nil, если поручение не подавалось).
func (s *orderWatchStore) Get(guid string) *OrderWatch {
	s.mu.Lock()
	defer s.mu.Unlock()
	if w, ok := s.items[guid]; ok {
		cp := *w
		return &cp
	}
	return nil
}

func (s *orderWatchStore) update(guid string, fn func(w *OrderWatch)) *OrderWatch {
	s.mu.Lock()
	defer s.mu.Unlock()
	w, ok := s.items[guid]
	if !ok {
		w = &OrderWatch{GUID: guid}
		s.items[guid] = w
	}
	fn(w)
	s.saveLocked()
	cp := *w
	return &cp
}

var orderWatchStarted = time.Now()

// outgoingLineRefsCache — REFERENCE по ISIN для исходящей сделки: берётся из
// M2MTR-запроса в репозитории; после отправки не меняется, поэтому кэшируем по GUID.
var outgoingLineRefsCache = struct {
	mu sync.Mutex
	m  map[string]map[string]string // guid → ISIN → ref
}{m: map[string]map[string]string{}}

// outgoingLineRefs — строки исходящего поручения 16: contr_d_id ↔ ISIN ↔ REFERENCE.
// Один черновик MF010 = одна бумага; REFERENCE берём из бумаг M2MTR по ISIN.
func outgoingLineRefs(ctx context.Context, svc *Service, od *OutgoingDeal) []orderLine {
	if od == nil {
		return nil
	}
	outgoingLineRefsCache.mu.Lock()
	byISIN, ok := outgoingLineRefsCache.m[od.GUID]
	outgoingLineRefsCache.mu.Unlock()
	if !ok && svc != nil {
		byISIN = map[string]string{}
		if deals, err := svc.Repo().List(ctx, m2mcore.Filter{Limit: 500}); err == nil {
			for _, d := range deals {
				if d == nil || d.Request == nil || !strings.EqualFold(string(d.GUID), od.GUID) {
					continue
				}
				for _, sec := range d.Request.Data.TransferredSecurities.Securities {
					byISIN[strings.ToUpper(quikSecISIN(sec))] = string(sec.ReferenceID)
				}
				break
			}
		}
		if len(byISIN) > 0 {
			outgoingLineRefsCache.mu.Lock()
			outgoingLineRefsCache.m[od.GUID] = byISIN
			outgoingLineRefsCache.mu.Unlock()
		}
	}
	var out []orderLine
	for _, sec := range od.Secs {
		if sec.ContrDID == "" {
			continue
		}
		isin := strings.ToUpper(sec.ISIN)
		out = append(out, orderLine{ContrDID: sec.ContrDID, Ref: byISIN[isin], ISIN: isin})
	}
	return out
}

// orderWatchCheck — один объект (перевод): подано когда, ответы НРД ПО КАЖДОЙ СТРОКЕ →
// фиксируем MS010/просрочку/истечение и пишем строчные события в DEPO сразу по
// MS010 этой бумаги, не дожидаясь остальных (частичное исполнение пакета).
func orderWatchCheck(ctx context.Context, rc *RuntimeConfig, guid, direction string, orderAt time.Time, lines []orderLine, label string) {
	if guid == "" || orderAt.IsZero() || len(lines) == 0 {
		return
	}
	w := globalOrderWatch.Get(guid)
	if w == nil {
		w = globalOrderWatch.update(guid, func(w *OrderWatch) { w.Direction, w.OrderAt = direction, orderAt })
	}
	if !w.MS010At.IsZero() || !w.ExpiredAt.IsZero() {
		return // финальное состояние уже зафиксировано по всем строкам
	}
	now := time.Now()
	fresh := orderAt.After(orderWatchStarted) // по старым поручениям при первом запуске не шумим

	// 1) исходы по строкам: у каждой бумаги свой MS010
	for _, ln := range lines {
		st := w.Lines[ln.ContrDID]
		if st.lineDone() {
			continue
		}
		resps := findOrderResponses(ln.ContrDID)
		done, rejected, text := orderRespVerdict(resps)
		if !done {
			continue
		}
		at := orderRespAt(resps)
		globalOrderWatch.update(guid, func(w *OrderWatch) {
			if w.Lines == nil {
				w.Lines = map[string]*OrderLineWatch{}
			}
			w.Lines[ln.ContrDID] = &OrderLineWatch{Ref: ln.Ref, ISIN: ln.ISIN, MS010At: at, Rejected: rejected}
			if rejected {
				w.Rejected = true
			}
		})
		if rejected {
			go depoOnOrderLineResult(rc, guid, ln.Ref, "rejected", text)
		} else {
			go depoOnOrderLineResult(rc, guid, ln.Ref, "accepted", "")
		}
	}
	w = globalOrderWatch.Get(guid)

	// 2) все строки закрыты MS010 → финальное состояние объекта
	if allLinesMS010(w, lines) {
		latest := time.Time{}
		for _, ln := range lines {
			if st := w.Lines[ln.ContrDID]; st != nil && st.MS010At.After(latest) {
				latest = st.MS010At
			}
		}
		globalOrderWatch.update(guid, func(w *OrderWatch) { w.MS010At = latest })
		return
	}

	// 3) просрочка 5 минут — один сигнал на объект, событие по каждой незакрытой строке
	if w.LateAt.IsZero() && now.Sub(orderAt) > orderMS010Within {
		globalOrderWatch.update(guid, func(w *OrderWatch) { w.LateAt = now })
		for _, ln := range lines {
			if !w.Lines[ln.ContrDID].lineDone() {
				go depoOnOrderLineResult(rc, guid, ln.Ref, "late", "")
			}
		}
		if fresh {
			raiseSystemAlert(rc, "НРД: депо-поручение не исполнено в срок 5 минут",
				fmt.Sprintf("%s (перевод %s): подано %s, отчёт MS010 не получен. Ожидаем исполнения до %s, после этого поручение считается отменённым.",
					label, shortGUID(guid), orderAt.Local().Format("02.01.2006 15:04"), orderAt.Add(orderExecWithin).Local().Format("02.01.2006")))
		}
	}

	// 4) 30 дней — по каждой незакрытой строке BJ_CANCELLED с REFERENCE; QUIK-блокировка снимается по ISIN
	if now.Sub(orderAt) > orderExecWithin {
		expiredAny := false
		for _, ln := range lines {
			if w.Lines[ln.ContrDID].lineDone() {
				continue
			}
			expiredAny = true
			globalOrderWatch.update(guid, func(w *OrderWatch) {
				if w.Lines == nil {
					w.Lines = map[string]*OrderLineWatch{}
				}
				w.Lines[ln.ContrDID] = &OrderLineWatch{Ref: ln.Ref, ISIN: ln.ISIN, ExpiredAt: now}
			})
			go depoOnOrderLineResult(rc, guid, ln.Ref, "expired", "")
			if direction == "out" {
				for _, l := range globalQuikLinks.ForGUID(guid) {
					if l.State == "registered" && (ln.ISIN == "" || strings.EqualFold(l.ISIN, ln.ISIN)) {
						_, _ = QuikOutDeny(ctx, rc, guid, l.ISIN, "Истёк срок исполнения поручения в НРД (30 дней) — перевод отменён")
					}
				}
			}
		}
		if expiredAny {
			globalOrderWatch.update(guid, func(w *OrderWatch) { w.ExpiredAt = now })
			raiseSystemAlert(rc, "НРД: истёк срок 30 дней по депо-поручению — считается отменённым",
				fmt.Sprintf("%s (перевод %s): подано %s, отчёт MS010 так и не получен. Поручение считается отменённым; проверьте перевод в НРД и позицию клиента.",
					label, shortGUID(guid), orderAt.Local().Format("02.01.2006 15:04")))
		}
	}
}

// allLinesMS010 — по всем строкам получен MS010.
func allLinesMS010(w *OrderWatch, lines []orderLine) bool {
	if w == nil || len(lines) == 0 {
		return false
	}
	for _, ln := range lines {
		st := w.Lines[ln.ContrDID]
		if st == nil || st.MS010At.IsZero() {
			return false
		}
	}
	return true
}

func orderWatchTick(ctx context.Context, rc *RuntimeConfig, svc *Service) {
	if globalOutgoingStore != nil {
		for _, od := range scanForm16Drafts(globalOutgoingStore) {
			if !od.Submitted {
				continue
			}
			orderWatchCheck(ctx, rc, od.GUID, "out", od.OrderAt, outgoingLineRefs(ctx, svc, od), "Поручение 16 (списание)")
		}
	}
	if globalIncomingStore != nil {
		for _, it := range globalIncomingStore.List() {
			if !it.OrderSubmitted {
				continue
			}
			orderWatchCheck(ctx, rc, it.GUID, "in", it.OrderAt, incomingOrderLines(it), "Поручение 16/1 (зачисление)")
		}
	}
}

// incomingOrderLines — строки поданных 16/1. Старые записи (до строчных событий) знали
// только первый contr_d_id — для них одна строка без REFERENCE.
func incomingOrderLines(it *IncomingTransfer) []orderLine {
	if it == nil {
		return nil
	}
	if len(it.OrderLines) > 0 {
		out := make([]orderLine, 0, len(it.OrderLines))
		for _, l := range it.OrderLines {
			out = append(out, orderLine{ContrDID: l.ContrDID, Ref: l.Ref, ISIN: l.ISIN})
		}
		return out
	}
	if it.OrderContrDID != "" {
		return []orderLine{{ContrDID: it.OrderContrDID}}
	}
	return nil
}

// orderWatchLoop — фоновый контроль сроков (раз в минуту).
func (s *Server) orderWatchLoop(ctx context.Context) {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
		func() {
			defer func() {
				if r := recover(); r != nil {
					log.Printf("order-watch: panic: %v", r)
				}
			}()
			orderWatchTick(ctx, s.rc, s.svc)
		}()
		healthBeat("order-watch")
	}
}