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

package lkgateway

// depotest.go — режим тестирования DEPO: оператор собирает перевод (клиент и бумаги из DEPO,
// контрагент, исход по каждой бумаге), bj пишет его в трансферные таблицы FTV_M2M* и
// проигрывает события по шагам так, как они шли бы в бою, имитируя ответы НРД (решение
// контрагента, приём и исполнение поручения 16/16.1, отказ НРД, истечение срока). После
// каждого шага ждёт, пока джоба bj2depo обработает запись, и показывает STATE/COMMENT.
// В НРД, контрагенту, QUIK и ЛК ничего не отправляется — только база DEPO.

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

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

const depoTestFile = "/var/lib/bj/.bj/depo-tests.json"

// Исходы строки.
const (
	dtExec    = "exec"    // принято и исполнено
	dtDecline = "decline" // отказ по M2M: наш (зачисление) или контрагента (списание)
	dtNSD     = "nsd"     // поручение отклонено НРД (не сквитовано)
	dtExpire  = "expire"  // поручение не исполнено в срок (30 дней)
)

var depoTestOutcomes = []struct{ Code, Title string }{
	{dtExec, "исполнено (MS010: исполнено)"},
	{dtDecline, "отказ по M2M с кодом"},
	{dtNSD, "поручение отклонено НРД (MS010: не сквитовано)"},
	{dtExpire, "не исполнено за 30 дней (отмена)"},
}

type DepoTestLine struct {
	Ref, ISIN, Qty, Outcome, Code string
	Code2                         string // код бумаги в НРД, ушедший в SEC_CODE
	State, DOL, Comment           string // результат джобы по строке FTV_M2M_LINES
}

type DepoTestStep struct {
	At                time.Time
	Title, Event, Ref string
	State, Comment    string // STATE/COMMENT, которые поставила джоба
	Emulated          string // что «ответил» имитатор НРД (для наглядности)
}

type DepoTestRun struct {
	ID, GUID, Direction, CP, CPName string
	Client, Ser, Num                string
	Lines                           []DepoTestLine
	Steps                           []DepoTestStep
	Created                         time.Time
	Status                          string // running | done | error
	Doc, HeaderState, HeaderComment string
	Errors                          int
}

type depoTestStore struct {
	mu   sync.Mutex
	runs []*DepoTestRun
}

var globalDepoTests = loadDepoTests()

func loadDepoTests() *depoTestStore {
	s := &depoTestStore{}
	if b, err := os.ReadFile(depoTestFile); err == nil {
		_ = json.Unmarshal(b, &s.runs)
	}
	for _, r := range s.runs {
		if r.Status == "running" {
			r.Status = "error"
			r.Steps = append(r.Steps, DepoTestStep{At: time.Now(), Title: "прогон прерван перезапуском bj"})
		}
	}
	return s
}

func (s *depoTestStore) saveLocked() {
	if len(s.runs) > 200 {
		s.runs = s.runs[len(s.runs)-200:]
	}
	b, _ := json.MarshalIndent(s.runs, "", " ")
	tmp := depoTestFile + ".tmp"
	if os.WriteFile(tmp, b, 0o600) == nil {
		_ = os.Rename(tmp, depoTestFile)
	}
}

func (s *depoTestStore) update(id string, fn func(r *DepoTestRun)) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, r := range s.runs {
		if r.ID == id {
			fn(r)
		}
	}
	s.saveLocked()
}

func (s *depoTestStore) list() []DepoTestRun {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]DepoTestRun, 0, len(s.runs))
	for i := len(s.runs) - 1; i >= 0; i-- {
		r := *s.runs[i]
		r.Steps = append([]DepoTestStep(nil), r.Steps...)
		r.Lines = append([]DepoTestLine(nil), r.Lines...)
		out = append(out, r)
	}
	return out
}

// isDepoTestGUID — перевод создан тестовым режимом: по нему не шлём оповещений сотрудникам.
func isDepoTestGUID(guid string) bool {
	g := strings.ToUpper(strings.TrimSpace(guid))
	globalDepoTests.mu.Lock()
	defer globalDepoTests.mu.Unlock()
	for _, r := range globalDepoTests.runs {
		if r.GUID == g {
			return true
		}
	}
	return false
}

func newDepoGUID() string {
	n := time.Now()
	return strings.ToUpper(fmt.Sprintf("%08x-%04x-4000-8000-%012x", n.Unix()&0xFFFFFFFF, n.UnixNano()&0xFFFF, n.UnixNano()&0xFFFFFFFFFFFF))
}

// depoTestExec — проигрывание сценария. pace — пауза между шагами (имитация задержек НРД).
func depoTestExec(rc *RuntimeConfig, run *DepoTestRun, pace time.Duration, stopOnError bool) {
	id := run.ID
	fail := func(msg string) {
		globalDepoTests.update(id, func(r *DepoTestRun) {
			r.Status = "error"
			r.Steps = append(r.Steps, DepoTestStep{At: time.Now(), Title: msg, State: "ERROR"})
		})
	}
	db, err := depoEventsDB(rc)
	if err != nil {
		fail("нет подключения к DEPO: " + err.Error())
		return
	}
	defer db.Close()

	// 1. шапка и строки
	place := rc.Snapshot().NSD.SectionID
	var lines []depoLine
	for i, l := range run.Lines {
		// SEC_CODE — код бумаги в НРД (как в настоящем пакете M2M): по ISIN+коду джоба находит бумагу в DEPO
		code := depoNSDCode(db, l.ISIN)
		if code == "" {
			code = l.ISIN
			globalDepoTests.update(id, func(r *DepoTestRun) {
				r.Steps = append(r.Steps, DepoTestStep{At: time.Now(), Title: "Бумага " + l.ISIN + ": кода НРД (кодировщик «НДЦ») в DEPO нет — SEC_CODE = ISIN, джоба может не найти бумагу", State: "WARN"})
			})
		}
		globalDepoTests.update(id, func(r *DepoTestRun) { r.Lines[i].Code2 = code })
		lines = append(lines, depoLine{Ref: l.Ref, ISIN: l.ISIN, Code: code, Amount: parseQty(l.Qty), Place: place})
	}
	dir := "IN"
	if run.Direction == "out" {
		dir = "OUT"
	}
	if !depoUpsertOrder(rc, run.GUID, dir, run.CP, run.Client, strings.TrimSpace(run.Ser+" "+run.Num), run.Ser, run.Num, lines, "ТЕСТ DEPO "+run.ID) {
		fail("не удалось записать шапку и строки в FTV_M2M (журнал bj-server)")
		return
	}
	state, doc, comment := depoTestWaitHeader(db, run.GUID)
	globalDepoTests.update(id, func(r *DepoTestRun) {
		r.Doc, r.HeaderState, r.HeaderComment = doc, state, comment
		r.Steps = append(r.Steps, DepoTestStep{At: time.Now(), Title: "Перевод записан в FTV_M2M / FTV_M2M_LINES → документ DEPO", Event: "FTV_M2M", State: state, Comment: pick(comment, "документ № "+doc)})
		if state == "ERROR" {
			r.Errors++
		}
	})
	if state == "ERROR" && stopOnError {
		depoTestFinish(db, id, run.GUID)
		return
	}

	// 2. события по шагам
	type ev struct{ title, typ, codes, params, emu string }
	stepsFor := func(l DepoTestLine) []ev {
		var s []ev
		if run.Direction == "out" {
			s = append(s, ev{"Отправили M2MTR в НРД", "BJ_SENT", "", "", ""})
			if l.Outcome == dtDecline {
				return append(s, ev{"Контрагент отказал (M2MTD)", "BJ_CP_REJECT", l.Code, m2mCodeLabel(l.Code), "НРД: решение контрагента — отказ " + l.Code})
			}
			s = append(s, ev{"Контрагент подтвердил (M2MTD)", "BJ_CP_CONFIRM", "", "", "НРД: решение контрагента — подтверждение"})
		} else {
			s = append(s, ev{"Получили M2MTR от контрагента", "BJ_RECEIVED", "", "", "НРД: входящий M2MTR"})
			if l.Outcome == dtDecline {
				return append(s, ev{"Наше решение: отказ (M2MTD)", "BJ_DECISION_REJECT", l.Code, m2mCodeLabel(l.Code), ""})
			}
			s = append(s, ev{"Наше решение: подтверждение (M2MTD)", "BJ_DECISION_CONFIRM", "", "", ""})
		}
		form := "16/1"
		if run.Direction == "out" {
			form = "16"
		}
		s = append(s, ev{"Подали поручение " + form + " в НРД", "BJ_ORDER_SENT", "", "EMU-" + l.Ref, "НРД: поручение принято к обработке"})
		switch l.Outcome {
		case dtNSD:
			s = append(s, ev{"НРД отклонил поручение (MS010)", "BJ_ORDER_REJECTED", "", "GS116 — поручение не сквитовано со встречным поручением контрагента (имитация НРД)", "НРД: MS010 — отказ, GS116"})
		case dtExpire:
			s = append(s, ev{"Нет ответа НРД в срок (просрочка)", "BJ_ORDER_LATE", "", "", "НРД: MS010 не пришёл"},
				ev{"30 дней истекли — поручение отменено", "BJ_CANCELLED", "", "", "НРД: срок исполнения истёк"})
		default:
			s = append(s, ev{"НРД принял поручение (MS010)", "BJ_ORDER_ACCEPTED", "", "", "НРД: MS010 — принято"},
				ev{"НРД исполнил поручение (MS010)", "BJ_EXECUTED", "", "", "НРД: MS010 — исполнено"})
		}
		return s
	}
	// порядок как в бою: сначала общий этап по всем строкам, затем исходы по строкам
	var queue [][]ev
	maxLen := 0
	for _, l := range run.Lines {
		st := stepsFor(l)
		queue = append(queue, st)
		if len(st) > maxLen {
			maxLen = len(st)
		}
	}
	for k := 0; k < maxLen; k++ {
		for i, l := range run.Lines {
			if k >= len(queue[i]) {
				continue
			}
			e := queue[i][k]
			if pace > 0 {
				time.Sleep(pace)
			}
			depoEventAt(rc, time.Now(), run.GUID, l.Ref, e.typ, e.codes, e.params)
			st, cm := depoTestWaitEvent(db, run.GUID, l.Ref, e.typ)
			globalDepoTests.update(id, func(r *DepoTestRun) {
				r.Steps = append(r.Steps, DepoTestStep{At: time.Now(), Title: e.title, Event: e.typ, Ref: l.Ref + " " + l.ISIN, State: st, Comment: cm, Emulated: e.emu})
				if st == "ERROR" || st == "" {
					r.Errors++
				}
			})
			if st == "ERROR" && stopOnError {
				depoTestFinish(db, id, run.GUID)
				return
			}
		}
	}
	depoTestFinish(db, id, run.GUID)
}

// depoNSDCode — код бумаги в НРД из DEPO (кодировщик «НДЦ»).
func depoNSDCode(db *sql.DB, isin string) string {
	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
	defer cancel()
	var c string
	_ = db.QueryRowContext(ctx, fsql.DT04, sql.Named("i", strings.ToUpper(strings.TrimSpace(isin)))).Scan(&c)
	return strings.TrimSpace(c)
}

func depoTestFinish(db *sql.DB, id, guid string) {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	res := map[string]DepoTestLine{}
	if rows, err := db.QueryContext(ctx, fsql.DT03, sql.Named("g", guid)); err == nil {
		for rows.Next() {
			var l DepoTestLine
			if rows.Scan(&l.Ref, &l.ISIN, &l.State, &l.DOL, &l.Comment) == nil {
				res[l.Ref] = l
			}
		}
		rows.Close()
	}
	var st, doc, cm string
	_ = db.QueryRowContext(ctx, fsql.DT01, sql.Named("g", guid)).Scan(&st, &doc, &cm)
	globalDepoTests.update(id, func(r *DepoTestRun) {
		for i := range r.Lines {
			if x, ok := res[r.Lines[i].Ref]; ok {
				r.Lines[i].State, r.Lines[i].DOL, r.Lines[i].Comment = x.State, x.DOL, x.Comment
			}
		}
		r.HeaderState, r.Doc, r.HeaderComment = st, doc, cm
		if r.Status == "running" {
			r.Status = "done"
		}
	})
	log.Printf("depo-test: прогон %s завершён (GUID %s)", id, shortGUID(guid))
}

func depoTestWaitHeader(db *sql.DB, guid string) (state, doc, comment string) {
	deadline := time.Now().Add(90 * time.Second)
	for time.Now().Before(deadline) {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		err := db.QueryRowContext(ctx, fsql.DT01, sql.Named("g", guid)).Scan(&state, &doc, &comment)
		cancel()
		if err == nil && state != "" && state != "TODO" && state != "WORK" {
			return
		}
		time.Sleep(2 * time.Second)
	}
	return pick(state, "TODO"), doc, "джоба не обработала шапку за 90 с — проверьте, что обработка DEPO запущена"
}

func depoTestWaitEvent(db *sql.DB, guid, ref, typ string) (state, comment string) {
	deadline := time.Now().Add(90 * time.Second)
	for time.Now().Before(deadline) {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		err := db.QueryRowContext(ctx, fsql.DT02, sql.Named("g", guid), sql.Named("r", ref), sql.Named("t", typ)).Scan(&state, &comment)
		cancel()
		if err == nil && state != "" && state != "TODO" && state != "WORK" {
			return
		}
		time.Sleep(2 * time.Second)
	}
	return pick(state, "TODO"), "джоба не обработала событие за 90 с"
}

// registerDepoTest — /admin/depo-test: форма, список прогонов; POST /admin/depo-test/run — запуск.
func registerDepoTest(mux *http.ServeMux, a *admin) {
	mux.HandleFunc("/admin/depo-test", func(w http.ResponseWriter, r *http.Request) {
		clients, secs := depoSimLists()
		var parts []HandbookEntry
		for _, p := range handbookParticipants() {
			if p.Code != "" {
				parts = append(parts, p)
			}
		}
		sort.Slice(parts, func(i, j int) bool { return parts[i].Name < parts[j].Name })
		var codes []struct{ Code, Title string }
		for i := 1; i <= 9; i++ {
			c := fmt.Sprintf("M2M%02d", i)
			codes = append(codes, struct{ Code, Title string }{c, c + " — " + m2mCodeLabel(c)})
		}
		runs := globalDepoTests.list()
		running := false
		for _, x := range runs {
			if x.Status == "running" {
				running = true
			}
		}
		fs := globalRC.Snapshot().Fancy
		data := struct {
			page
			Clients  []depoSimClient
			Secs     []depoSimSec
			Parts    []HandbookEntry
			Codes    []struct{ Code, Title string }
			Outcomes []struct{ Code, Title string }
			Runs     []DepoTestRun
			Running  bool
			DB       string
			JobMode  string
			Flash    string
			Rows     []string
		}{nowPage("Тестирование DEPO", "depotest"), clients, secs, parts, codes,
			append([]struct{ Code, Title string }(nil), depoTestOutcomes...), runs, running,
			fs.Server + " / " + fs.DepoDB(), fs.DepoJobMode, r.URL.Query().Get("flash"), []string{"1", "2", "3", "4", "5"}}
		render(w, a.depoTest, data)
	})
	mux.HandleFunc("/admin/depo-test/run", func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			http.Error(w, "method", http.StatusMethodNotAllowed)
			return
		}
		back := func(msg string) {
			http.Redirect(w, r, "/admin/depo-test?flash="+urlQ(msg), http.StatusSeeOther)
		}
		dir := r.FormValue("direction")
		if dir != "out" {
			dir = "in"
		}
		cp := strings.TrimSpace(r.FormValue("cp"))
		client := strings.TrimSpace(r.FormValue("client")) // «ФИО|серия|номер»
		p := strings.Split(client, "|")
		if len(p) != 3 || p[0] == "" || p[2] == "" {
			back("Выберите клиента из DEPO")
			return
		}
		if cp == "" {
			back("Выберите контрагента")
			return
		}
		run := &DepoTestRun{ID: time.Now().Format("0102-150405"), GUID: newDepoGUID(), Direction: dir, CP: cp, CPName: handbookNameByCode(cp),
			Client: p[0], Ser: p[1], Num: p[2], Created: time.Now(), Status: "running"}
		base := time.Now().UnixNano() / 1e6 % 1e10
		for i := 1; i <= 5; i++ {
			isin := strings.ToUpper(strings.TrimSpace(r.FormValue("isin" + strconv.Itoa(i))))
			if isin == "" {
				continue
			}
			l := DepoTestLine{Ref: fmt.Sprintf("M2MT%010d%02d", base, i), ISIN: isin, Qty: pick(strings.TrimSpace(r.FormValue("qty"+strconv.Itoa(i))), "1"),
				Outcome: r.FormValue("out" + strconv.Itoa(i)), Code: r.FormValue("code" + strconv.Itoa(i))}
			switch l.Outcome {
			case dtExec, dtNSD, dtExpire:
				l.Code = ""
			case dtDecline:
				if l.Code == "" {
					l.Code = "M2M05"
				}
			default:
				l.Outcome = dtExec
			}
			run.Lines = append(run.Lines, l)
		}
		if len(run.Lines) == 0 {
			back("Добавьте хотя бы одну бумагу")
			return
		}
		pace, _ := strconv.Atoi(r.FormValue("pace"))
		stop := r.FormValue("stop") == "1"
		globalDepoTests.mu.Lock()
		globalDepoTests.runs = append(globalDepoTests.runs, run)
		globalDepoTests.saveLocked()
		globalDepoTests.mu.Unlock()
		cpy := *run
		cpy.Lines = append([]DepoTestLine(nil), run.Lines...)
		go depoTestExec(globalRC, &cpy, time.Duration(pace)*time.Second, stop)
		log.Printf("depo-test: запущен прогон %s (%s, %s, %d бумаг) пользователем %s", run.ID, dir, cp, len(run.Lines), func() string {
			if u := currentUser(w); u != nil {
				return u.Login
			}
			return "-"
		}())
		http.Redirect(w, r, "/admin/depo-test#run-"+run.ID, http.StatusSeeOther)
	})
}