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

package lkgateway

// Страница «Обмен с DEPO (Fancy)» — /admin/fancy-queue. Показывает реальный
// обмен через трансферные таблицы FTV_M2M / FTV_M2M_LINES / FTV_M2M_EVENTS:
// какие переводы bj положил в DEPO, что ответил job bj2depo (DEPO_ORDER_CREATED /
// DEPO_ERROR), в каком состоянии документ в депозитарии и какие события ещё
// не обработаны. Заменила прежнюю «очередь на проверку крана».

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"net/http"
	"net/url"
	"sort"
	"strings"
	"time"
)

type depoQueueLine struct {
	ISIN, SecCode string
	Amount        string
	State         string
	Comment       string
}

type depoQueueEvent struct {
	Ref, State, Type, Codes, Comment string
	At                               time.Time
}

type depoQueueRow struct {
	GUID        string
	State       string
	Direction   string
	Deponent    string
	DeponentDoc string
	CP          string
	CreatedAt   time.Time
	Doc         string
	DocState    string
	DocAt       time.Time
	Comment     string
	Lines       []depoQueueLine
	Events      []depoQueueEvent
	Pending     int // события со STATE=TODO (job ещё не обработал)
	Errors      []string
	LastEvent   string
	LastAt      time.Time
	Link        *DepoLink
}

func (r depoQueueRow) DirectionRu() string {
	if strings.EqualFold(r.Direction, "OUT") {
		return "↗ списание (мы отдаём)"
	}
	return "↘ зачисление (мы принимаем)"
}

// StageRu — человеческое состояние перевода по совокупности признаков.
func (r depoQueueRow) StageRu() (cls, text string) {
	up := strings.ToUpper(r.State)
	switch {
	case len(r.Errors) > 0 || up == "ERROR":
		return "rejected", "ошибка DEPO"
	case strings.Contains(strings.ToLower(r.DocState), "отмен"):
		return "rejected", "документ отменён в DEPO"
	case strings.Contains(strings.ToLower(r.DocState), "конец"):
		return "confirmed", "документ проведён"
	case r.Doc != "":
		return "sent", "документ создан: " + r.DocState
	case r.Pending > 0:
		return "awaiting_decision", "ждёт job bj2depo"
	case up == "SUCCESS" || up == "DONE":
		return "confirmed", "обработан"
	}
	if up == "" {
		return "", "—"
	}
	return "", up
}

type depoQueueSummary struct {
	Total, Pending, WithDoc, Errors int
	Handbook                        int
	HandbookAt                      time.Time
	HasHandbook                     bool
	LastEventAt                     time.Time
}

func depoQueueLoad(ctx context.Context, rc *RuntimeConfig, limit int) ([]depoQueueRow, depoQueueSummary, error) {
	var sum depoQueueSummary
	db, err := depoEventsDB(rc)
	if err != nil {
		return nil, sum, err
	}
	ctx, cancel := context.WithTimeout(ctx, 20*time.Second)
	defer cancel()
	if !depoTableExists(ctx, db, "FTV_M2M") {
		return nil, sum, errNoDepoTables
	}
	docCol := "NULL"
	if depoHasColumn(ctx, db, "FTV_M2M", "DOC") {
		docCol = "CAST(m.DOC AS varchar(32))"
	}
	timeCol := "NULL"
	for _, c := range []string{"TimeNew", "TIME_NEW", "CREATED_AT"} {
		if depoHasColumn(ctx, db, "FTV_M2M", c) {
			timeCol = "m." + c
			break
		}
	}
	rows, err := db.QueryContext(ctx, `SELECT TOP (@n) CAST(m.GUID AS varchar(36)), ISNULL(m.STATE,''), ISNULL(m.DIRECTION,''), ISNULL(m.DEPONENT_NAME,''), ISNULL(m.DEPONENT_DOC,''), ISNULL(m.CONTRAGENT_M2M_CODE,''), `+timeCol+`, ISNULL(`+docCol+`,''), ISNULL(m.COMMENT,''),
		ISNULL((SELECT TOP 1 ISNULL(st.NAME,'') FROM OD_DOCS d LEFT JOIN OD_DOC_STATES st ON st.ID=d.STATE AND st.D_CAT=0 WHERE CAST(d.ID AS varchar(32))=`+docCol+`),''),
		(SELECT TOP 1 d.PROC_DATE FROM OD_DOCS d WHERE CAST(d.ID AS varchar(32))=`+docCol+`)
		FROM dbo.FTV_M2M m ORDER BY `+strings.Replace(timeCol, "NULL", "m.GUID", 1)+` DESC`, sql.Named("n", limit))
	if err != nil {
		return nil, sum, err
	}
	var out []depoQueueRow
	idx := map[string]int{}
	for rows.Next() {
		var r depoQueueRow
		var created, docAt sql.NullTime
		if err := rows.Scan(&r.GUID, &r.State, &r.Direction, &r.Deponent, &r.DeponentDoc, &r.CP, &created, &r.Doc, &r.Comment, &r.DocState, &docAt); err != nil {
			rows.Close()
			return nil, sum, err
		}
		if created.Valid {
			r.CreatedAt = created.Time
		}
		if docAt.Valid {
			r.DocAt = docAt.Time
		}
		r.GUID = strings.ToUpper(r.GUID)
		r.Link = globalDepoLinks.Get(r.GUID)
		idx[r.GUID] = len(out)
		out = append(out, r)
	}
	rows.Close()
	if len(out) == 0 {
		return out, sum, nil
	}
	// строки
	if depoTableExists(ctx, db, "FTV_M2M_LINES") {
		lr, err := db.QueryContext(ctx, `SELECT CAST(l.GUID AS varchar(36)), ISNULL(l.ISIN,''), ISNULL(l.SEC_CODE,''), ISNULL(CAST(l.AMOUNT AS varchar(40)),''), ISNULL(l.STATE,''), ISNULL(l.COMMENT,'') FROM dbo.FTV_M2M_LINES l WHERE l.GUID IN (SELECT TOP (@n) GUID FROM dbo.FTV_M2M ORDER BY `+strings.Replace(strings.TrimPrefix(timeCol, "m."), "NULL", "GUID", 1)+` DESC)`, sql.Named("n", limit))
		if err == nil {
			for lr.Next() {
				var g string
				var l depoQueueLine
				if lr.Scan(&g, &l.ISIN, &l.SecCode, &l.Amount, &l.State, &l.Comment) == nil {
					if i, ok := idx[strings.ToUpper(g)]; ok {
						l.Amount = strings.TrimRight(strings.TrimRight(l.Amount, "0"), ".")
						out[i].Lines = append(out[i].Lines, l)
						if strings.EqualFold(l.State, "ERROR") && l.Comment != "" {
							out[i].Errors = append(out[i].Errors, l.Comment)
						}
					}
				}
			}
			lr.Close()
		}
	}
	// события
	if depoTableExists(ctx, db, "FTV_M2M_EVENTS") {
		er, err := db.QueryContext(ctx, `SELECT CAST(e.GUID AS varchar(36)), ISNULL(e.REFERENCE,''), ISNULL(e.STATE,''), ISNULL(e.EVENT_TYPE,''), e.EVENT_DATE, ISNULL(e.M2M_REJECT_CODE,''), ISNULL(e.COMMENT,'') FROM dbo.FTV_M2M_EVENTS e WHERE e.GUID IN (SELECT TOP (@n) GUID FROM dbo.FTV_M2M ORDER BY `+strings.Replace(strings.TrimPrefix(timeCol, "m."), "NULL", "GUID", 1)+` DESC) ORDER BY e.EVENT_DATE`, sql.Named("n", limit))
		if err == nil {
			for er.Next() {
				var g string
				var e depoQueueEvent
				var at sql.NullTime
				if er.Scan(&g, &e.Ref, &e.State, &e.Type, &at, &e.Codes, &e.Comment) == nil {
					if at.Valid {
						e.At = at.Time
					}
					if i, ok := idx[strings.ToUpper(g)]; ok {
						out[i].Events = append(out[i].Events, e)
						if strings.EqualFold(e.State, "TODO") {
							out[i].Pending++
						}
						if e.Type == "DEPO_ERROR" && e.Comment != "" {
							out[i].Errors = append(out[i].Errors, e.Comment)
						}
						if e.At.After(out[i].LastAt) {
							out[i].LastAt, out[i].LastEvent = e.At, e.Type
						}
						if e.At.After(sum.LastEventAt) {
							sum.LastEventAt = e.At
						}
					}
				}
			}
			er.Close()
		}
	}
	if depoTableExists(ctx, db, "FTV_M2M_HANDBOOK") {
		sum.HasHandbook = true
		_ = db.QueryRowContext(ctx, "SELECT count(*) FROM dbo.FTV_M2M_HANDBOOK").Scan(&sum.Handbook)
		if depoHasColumn(ctx, db, "FTV_M2M_HANDBOOK", "HANDBOOK_AT") {
			var t sql.NullTime
			if db.QueryRowContext(ctx, "SELECT max(HANDBOOK_AT) FROM dbo.FTV_M2M_HANDBOOK").Scan(&t) == nil && t.Valid {
				sum.HandbookAt = t.Time
			}
		}
	}
	for i := range out {
		sum.Total++
		if out[i].Pending > 0 {
			sum.Pending++
		}
		if out[i].Doc != "" {
			sum.WithDoc++
		}
		if len(out[i].Errors) > 0 {
			sum.Errors++
		}
	}
	sort.SliceStable(out, func(i, j int) bool {
		ai, aj := out[i].CreatedAt, out[j].CreatedAt
		if ai.IsZero() {
			ai = out[i].LastAt
		}
		if aj.IsZero() {
			aj = out[j].LastAt
		}
		return ai.After(aj)
	})
	return out, sum, nil
}

var errNoDepoTables = errNoDepo("в базе DEPO нет таблицы FTV_M2M — обмен ещё не настроен программистом Fancy")

type errNoDepo string

func (e errNoDepo) Error() string { return string(e) }

// fancyQueue — GET /admin/fancy-queue: обмен с DEPO через FTV_M2M*.
func (h *setupHandlers) fancyQueue(w http.ResponseWriter, r *http.Request) {
	s := h.rc.Snapshot().Fancy
	var rows []depoQueueRow
	var sum depoQueueSummary
	var loadErr string
	if s.Server == "" {
		loadErr = "Подключение к базе DEPO (Fancy) не настроено — Настройка → Обмен Fancy"
	} else if rs, sm, err := depoQueueLoad(r.Context(), h.rc, 100); err != nil {
		loadErr = "не удалось прочитать трансферные таблицы: " + err.Error()
	} else {
		rows, sum = rs, sm
	}
	// ошибки обработчика и пользователь для EXECUTE AS — видны сразу на странице
	var fails []depoJobFailure
	runAsOK, runAsMsg := true, ""
	if s.Server != "" && loadErr == "" {
		if db, err := depoEventsDB(h.rc); err == nil {
			ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
			fails, _ = depoJobFailures(ctx, db, 7, 50)
			runAsOK, runAsMsg = depoRunAsCheck(ctx, db, s)
			cancel()
			db.Close()
		}
	}
	data := struct {
		page
		Rows     []depoQueueRow
		Sum      depoQueueSummary
		Error    string
		Flash    string
		DB       string
		Fancy    FancyMSSQLSettings
		JobMode  string
		Failures []depoJobFailure
		RunAs    string
		RunAsOK  bool
		RunAsMsg string
	}{nowPage("Обмен с DEPO", "fancy"), rows, sum, loadErr, r.URL.Query().Get("flash"), s.DepoDB(), s, depoJobModeRu(s.DepoJobMode), fails, depoJobRunAs(s), runAsOK, runAsMsg}
	render(w, h.tpl.a.fancyQueue, data)
}

// fancyQueueRequeue — POST /admin/fancy-queue/requeue: вернуть записи с ошибкой обработчика в очередь.
func (h *setupHandlers) fancyQueueRequeue(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method", http.StatusMethodNotAllowed)
		return
	}
	msg := ""
	db, err := depoEventsDB(h.rc)
	if err == nil {
		ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second)
		var o, e int64
		o, e, err = depoRequeueFailures(ctx, db, 7)
		cancel()
		db.Close()
		if err == nil {
			msg = fmt.Sprintf("Возвращено в очередь обработчика: поручений %d, событий %d. Обновите страницу через 10–20 секунд.", o, e)
			log.Printf("depo-events: записи с ошибкой возвращены в очередь: поручений %d, событий %d", o, e)
		}
	}
	if err != nil {
		msg = "Не удалось вернуть записи в очередь: " + quikErrorDetail(err)
	}
	http.Redirect(w, r, "/admin/fancy-queue?flash="+url.QueryEscape(msg), http.StatusSeeOther)
}

func depoJobModeRu(m string) string {
	switch m {
	case "bj":
		return "процедуру вызывает bj"
	case "sql-auto":
		return "SQL Agent (создан из bj)"
	case "sql-manual":
		return "SQL Agent (настроен вручную)"
	}
	return "не выбран"
}