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)) }