123 lines
4.2 KiB
Go
123 lines
4.2 KiB
Go
// Package db 本机 MySQL 审计库（sql/hy_transit_schema.sql）。
|
||
//
|
||
// 与云端 xk_hy_transit_* 的分工：
|
||
// - 云端表：运营在 xk-admin 查看拉取/回调状态（权威）
|
||
// - 本机表：外网机离线排错、重试统计，sync 过程双写 hy_push_record / hy_push_log
|
||
//
|
||
// 不替代云端 batch/record；BatchCreate/Pull/Callback 仍以 xk-api 为准。
|
||
package db
|
||
|
||
import (
|
||
"database/sql"
|
||
"encoding/json"
|
||
"time"
|
||
|
||
_ "github.com/go-sql-driver/mysql"
|
||
)
|
||
|
||
// Store 本机中转库连接。
|
||
type Store struct {
|
||
db *sql.DB
|
||
}
|
||
|
||
// Open 连接 MySQL 并 Ping。
|
||
func Open(dsn string) (*Store, error) {
|
||
db, err := sql.Open("mysql", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := db.Ping(); err != nil {
|
||
return nil, err
|
||
}
|
||
db.SetMaxOpenConns(10)
|
||
return &Store{db: db}, nil
|
||
}
|
||
|
||
// Close 关闭连接。
|
||
func (s *Store) Close() error {
|
||
return s.db.Close()
|
||
}
|
||
|
||
// UpsertJob 写入/更新 hy_sync_job（按 anchor_date + step 维度汇总一次任务）。
|
||
func (s *Store) UpsertJob(anchorDate, step, status string, total, success, failed int, errMsg string) (int64, error) {
|
||
res, err := s.db.Exec(`
|
||
INSERT INTO hy_sync_job (anchor_date, step, status, total_count, success_count, failed_count, error_message, started_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, NOW())
|
||
ON DUPLICATE KEY UPDATE
|
||
status=VALUES(status),
|
||
total_count=VALUES(total_count),
|
||
success_count=VALUES(success_count),
|
||
failed_count=VALUES(failed_count),
|
||
error_message=VALUES(error_message),
|
||
updated_at=NOW()`,
|
||
anchorDate, step, status, total, success, failed, errMsg,
|
||
)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return res.LastInsertId()
|
||
}
|
||
|
||
// PushRecord 本机推送记录行（查询用）。
|
||
type PushRecord struct {
|
||
ID int64
|
||
BizKey string
|
||
PushStatus string
|
||
ValidationErrors []string
|
||
}
|
||
|
||
// SaveRecord 写入 hy_push_record；有 validationErrors 时 push_status=skipped。
|
||
func (s *Store) SaveRecord(jobID int64, anchorDate, method, serviceMethod, bussID, bizKey, payloadHash, payloadJSON string, validationErrors []string) error {
|
||
ve, _ := json.Marshal(validationErrors)
|
||
status := "pending"
|
||
if len(validationErrors) > 0 {
|
||
status = "skipped"
|
||
}
|
||
_, err := s.db.Exec(`
|
||
INSERT INTO hy_push_record (job_id, anchor_date, method, service_method, buss_id, biz_key, payload_hash, payload_json, validation_errors, push_status)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
payload_json=VALUES(payload_json),
|
||
validation_errors=VALUES(validation_errors),
|
||
push_status=IF(VALUES(validation_errors) IS NOT NULL AND JSON_LENGTH(VALUES(validation_errors))>0, 'skipped', push_status),
|
||
updated_at=NOW()`,
|
||
jobID, anchorDate, method, serviceMethod, bussID, bizKey, payloadHash, payloadJSON, string(ve), status,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// UpdateRecordStatus 更新本机推送结果。
|
||
func (s *Store) UpdateRecordStatus(bizKey, status, lastError string) error {
|
||
_, err := s.db.Exec(`UPDATE hy_push_record SET push_status=?, last_error=?, updated_at=NOW() WHERE biz_key=?`, status, lastError, bizKey)
|
||
return err
|
||
}
|
||
|
||
// IncRetry 增加重试计数。
|
||
func (s *Store) IncRetry(bizKey string) error {
|
||
_, err := s.db.Exec(`UPDATE hy_push_record SET retry_count=retry_count+1, updated_at=NOW() WHERE biz_key=?`, bizKey)
|
||
return err
|
||
}
|
||
|
||
// SaveLog 写入单次 HTTP 上报日志 hy_push_log。
|
||
func (s *Store) SaveLog(recordID int64, httpCode, msgCode int, msg, traceID string, bodyLen int, plainJSON, headersJSON, responseBody string, durationMs int) error {
|
||
_, err := s.db.Exec(`
|
||
INSERT INTO hy_push_log (record_id, http_code, msg_code, msg, trace_id, request_body_len, request_plain_json, request_headers_json, response_body, duration_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
recordID, httpCode, msgCode, msg, traceID, bodyLen, plainJSON, headersJSON, responseBody, durationMs,
|
||
)
|
||
return err
|
||
}
|
||
|
||
// GetRecordID 按 biz_key 查本机记录 id。
|
||
func (s *Store) GetRecordID(bizKey string) (int64, error) {
|
||
var id int64
|
||
err := s.db.QueryRow(`SELECT id FROM hy_push_record WHERE biz_key=?`, bizKey).Scan(&id)
|
||
return id, err
|
||
}
|
||
|
||
// FinishJob 标记 hy_sync_job 结束。
|
||
func (s *Store) FinishJob(anchorDate, step, status string) error {
|
||
_, err := s.db.Exec(`UPDATE hy_sync_job SET status=?, finished_at=? WHERE anchor_date=? AND step=?`, status, time.Now(), anchorDate, step)
|
||
return err
|
||
}
|