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

package lkgateway

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

import (
	"context"
	"database/sql"
	"fmt"
	"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, `SELECT TOP (@n) * FROM (
		SELECT 'FTV_M2M' AS T, ID, CAST(GUID AS varchar(36)) AS G, '' AS R, ISNULL(DIRECTION,'') AS K, ISNULL(TimeEnd, TimeNew) AS A, ISNULL(COMMENT,'') AS C
		  FROM dbo.FTV_M2M WHERE STATE='ERROR' AND ISNULL(TimeEnd, TimeNew) > DATEADD(day, -@d, GETDATE())
		UNION ALL
		SELECT 'FTV_M2M_EVENTS', ID, CAST(GUID AS varchar(36)), ISNULL(REFERENCE,''), EVENT_TYPE, ISNULL(TimeEnd, TimeNew), ISNULL(COMMENT,'')
		  FROM dbo.FTV_M2M_EVENTS WHERE STATE='ERROR' AND EVENT_TYPE LIKE 'BJ[_]%' AND ISNULL(TimeEnd, TimeNew) > DATEADD(day, -@d, GETDATE())
		) x ORDER BY A DESC`, 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 := "UPDATE dbo.FTV_M2M SET STATE='TODO', TimeBeg=NULL, TimeEnd=NULL, COMMENT=@c WHERE ID=@id AND STATE='ERROR'"
		if f.Table == "FTV_M2M_EVENTS" {
			q = "UPDATE dbo.FTV_M2M_EVENTS SET STATE='TODO', TimeBeg=NULL, TimeEnd=NULL, COMMENT=@c WHERE ID=@id AND STATE='ERROR'"
			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, "UPDATE dbo.FTV_M2M_LINES SET STATE='TODO', TimeBeg=NULL, TimeEnd=NULL WHERE GUID=@g AND STATE='ERROR'", 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

// depoAlertEventFailures — оповещение о новых ошибках обработки событий BJ_* (поручения с
// ошибкой оповещает depoPollOrders).
func depoAlertEventFailures(ctx context.Context, db *sql.DB, rc *RuntimeConfig) {
	var id sql.NullInt64
	var typ, guid, comment string
	var n int
	err := db.QueryRowContext(ctx, `SELECT COUNT(*) OVER (), ID, EVENT_TYPE, CAST(GUID AS varchar(36)), ISNULL(COMMENT,'')
		FROM dbo.FTV_M2M_EVENTS WHERE STATE='ERROR' AND EVENT_TYPE LIKE 'BJ[_]%' AND ID > @last AND ISNULL(TimeEnd, TimeNew) > DATEADD(day, -1, GETDATE())
		ORDER BY ID DESC OFFSET 0 ROWS FETCH NEXT 1 ROWS ONLY`, sql.Named("last", depoFailAlertedID)).Scan(&n, &id, &typ, &guid, &comment)
	if err != nil || !id.Valid {
		return
	}
	depoFailAlertedID = id.Int64
	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))
}