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

package lkgateway

// Здоровье обработки трансферных таблиц DEPO: пользователь, от имени которого работает
// процедура обработки (EXECUTE AS), записи, помеченные обработчиком ERROR, и возврат их
// в очередь после устранения причины. У каждого клиента свой набор пользователей в базе
// DEPO, поэтому имя настраивается (Fancy.DepoJobRunAs) и проверяется при настройке.

import (
	"context"
	"database/sql"
	"fmt"
	"git.zetit.ru/zuevav/Bridge-and-Join-s/internal/lkgateway/fsql"
	"log"
	"regexp"
	"strings"
	"time"
)

var depoRunAsRe = regexp.MustCompile(`^[A-Za-z0-9_.\-\\]{1,128}$`)

// depoJobRunAs — пользователь базы DEPO, от имени которого работает процедура обработки.
func depoJobRunAs(f FancyMSSQLSettings) string {
	if u := strings.TrimSpace(f.DepoJobRunAs); u != "" {
		return u
	}
	return "SYSDBA"
}

// depoRunAsCheck — есть ли в базе DEPO пользователь для EXECUTE AS. USER_ID видит пользователя,
// даже если логину bj закрыт просмотр списка пользователей базы.
func depoRunAsCheck(ctx context.Context, db *sql.DB, f FancyMSSQLSettings) (bool, string) {
	name := depoJobRunAs(f)
	var id sql.NullInt64
	if err := db.QueryRowContext(ctx, "SELECT USER_ID(@n)", sql.Named("n", name)).Scan(&id); err != nil {
		return false, "не удалось проверить пользователя " + name + ": " + quikErrorDetail(err)
	}
	if !id.Valid {
		return false, fmt.Sprintf("в базе %s нет пользователя %s, от имени которого работает процедура обработки: создайте его или укажите другого пользователя в настройках (Обмен Fancy → «Где запускается обработка»)", f.DepoDB(), name)
	}
	return true, fmt.Sprintf("пользователь %s для процедуры обработки в базе есть", name)
}

// depoJobFailure — запись трансферных таблиц, которую обработчик пометил ERROR.
type depoJobFailure struct {
	Table, GUID, Ref, Type, Comment string
	ID                              int64
	At                              time.Time
}

func (f depoJobFailure) TableRu() string {
	if f.Table == "FTV_M2M" {
		return "поручение"
	}
	return "событие"
}

// depoJobFailures — записи с ошибкой обработки за days дней, новые первыми.
func depoJobFailures(ctx context.Context, db *sql.DB, days, limit int) ([]depoJobFailure, error) {
	rows, err := db.QueryContext(ctx, fsql.Hlth01, sql.Named("n", limit), sql.Named("d", days))
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []depoJobFailure
	for rows.Next() {
		var f depoJobFailure
		var at sql.NullTime
		if err := rows.Scan(&f.Table, &f.ID, &f.GUID, &f.Ref, &f.Type, &at, &f.Comment); err != nil {
			return out, err
		}
		if at.Valid {
			f.At = at.Time
		}
		f.GUID = strings.ToUpper(f.GUID)
		out = append(out, f)
	}
	return out, rows.Err()
}

// depoRequeueFailures — вернуть записи с ошибкой (за days дней) в очередь обработчика. Текст
// ошибки в COMMENT заменяется: у событий отказа шлюза — исходной причиной из пути заявки.
func depoRequeueFailures(ctx context.Context, db *sql.DB, days int) (orders, events int64, err error) {
	fails, err := depoJobFailures(ctx, db, days, 500)
	if err != nil {
		return 0, 0, err
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return 0, 0, err
	}
	for _, f := range fails {
		comment := "повторная обработка из bj"
		q := fsql.QHl01
		if f.Table == "FTV_M2M_EVENTS" {
			q = fsql.QHl02
			if lf := globalLKFlows.Get(f.GUID); f.Type == "BJ_GATEWAY_REJECT" && lf != nil && lf.Reason != "" {
				comment = lf.Reason
			}
		}
		res, e := tx.ExecContext(ctx, q, sql.Named("c", comment), sql.Named("id", f.ID))
		if e != nil {
			tx.Rollback()
			return 0, 0, e
		}
		n, _ := res.RowsAffected()
		if f.Table == "FTV_M2M" {
			orders += n
			// строки поручения, помеченные ошибкой, тоже возвращаем
			if _, e := tx.ExecContext(ctx, fsql.QHl03, sql.Named("g", f.GUID)); e != nil {
				tx.Rollback()
				return 0, 0, e
			}
		} else {
			events += n
		}
	}
	if err := tx.Commit(); err != nil {
		return 0, 0, err
	}
	return orders, events, nil
}

// depoDismissFailures — снять записи с ошибкой (за days дней) с очереди: STATE='SKIPPED'.
// Обработчик их больше не берёт, и из списка ошибок они пропадают. Для устаревших/ненужных
// записей, которые не нужно переобрабатывать.
func depoDismissFailures(ctx context.Context, db *sql.DB, days int) (orders, events int64, err error) {
	fails, err := depoJobFailures(ctx, db, days, 500)
	if err != nil {
		return 0, 0, err
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return 0, 0, err
	}
	const comment = "снято с очереди администратором (bj)"
	for _, f := range fails {
		q := fsql.QHl04
		if f.Table == "FTV_M2M_EVENTS" {
			q = fsql.QHl05
		}
		res, e := tx.ExecContext(ctx, q, sql.Named("c", comment), sql.Named("id", f.ID))
		if e != nil {
			tx.Rollback()
			return 0, 0, e
		}
		n, _ := res.RowsAffected()
		if f.Table == "FTV_M2M" {
			orders += n
			if _, e := tx.ExecContext(ctx, fsql.QHl06, sql.Named("g", f.GUID)); e != nil {
				tx.Rollback()
				return 0, 0, e
			}
		} else {
			events += n
		}
	}
	if err := tx.Commit(); err != nil {
		return 0, 0, err
	}
	return orders, events, nil
}

var (
	depoFailAlertedID int64
	depoFailSeeded    bool // водяной знак засеян текущим максимумом на старте процесса
)

// depoAlertEventFailures — оповещение о новых ошибках обработки событий BJ_* (поручения с
// ошибкой оповещает depoPollOrders).
//
// Водяной знак depoFailAlertedID — в памяти процесса. Чтобы при каждом РЕСТАРТЕ bj не
// поднимать повторно алерты про уже известные (и, как правило, уже разобранные) ошибки,
// на первом проходе засеиваем знак текущим максимумом ID событий BJ_*: алертим только про
// НОВЫЕ ошибки, появившиеся уже после запуска. (События BJ_* пишет сам bj, пока он лежит —
// новых не появляется, поэтому ничего не пропускаем.)
func depoAlertEventFailures(ctx context.Context, db *sql.DB, rc *RuntimeConfig) {
	if !depoFailSeeded {
		var mx sql.NullInt64
		if e := db.QueryRowContext(ctx, fsql.Hlth03).Scan(&mx); e == nil && mx.Int64 > depoFailAlertedID {
			depoFailAlertedID = mx.Int64
		}
		depoFailSeeded = true
	}
	var id sql.NullInt64
	var typ, guid, comment string
	var n int
	err := db.QueryRowContext(ctx, fsql.Hlth02, sql.Named("last", depoFailAlertedID)).Scan(&n, &id, &typ, &guid, &comment)
	if err != nil || !id.Valid {
		return
	}
	depoFailAlertedID = id.Int64
	if isDepoTestGUID(guid) {
		return // тестовый прогон DEPO: результат виден на странице теста, сотрудников не оповещаем
	}
	log.Printf("depo-events: обработчик пометил ERROR событий: %d, последнее %s по %s: %s", n, typ, shortGUID(guid), comment)
	raiseSystemAlert(rc, fmt.Sprintf("DEPO: обработчик не смог обработать события (%d)", n),
		fmt.Sprintf("Последнее: %s по переводу %s — %s. Устраните причину и верните записи в очередь: Обмен с DEPO → «Вернуть в очередь».", typ, shortGUID(guid), comment))
}