feat: migrate workbench services

This commit is contained in:
richjohnrambo
2026-10-10 23:34:06 +08:00
parent c13fb72849
commit 81e3cf1d03
88 changed files with 7358 additions and 15333 deletions
+175
View File
@@ -0,0 +1,175 @@
package store
import (
"context"
"database/sql"
"errors"
"strconv"
"strings"
"time"
"github.com/go-sql-driver/mysql"
"github.com/jmoiron/sqlx"
)
const (
memberStatusActive = 1
memberCategoryReal = 1
)
type AccountRegistration struct {
TenantID int64
MemberNo string
Username string
MobileCipher string
MobileHash string
MobileMask string
PasswordHash string
ReferralCode string
InviteCode string
}
type Account struct {
TenantID int64 `db:"tenant_id"`
MemberID int64 `db:"member_id"`
MemberNo string `db:"member_no"`
Username string `db:"username"`
MobileMask string `db:"mobile_mask"`
MobileHash string `db:"mobile_hash"`
PasswordHash string `db:"password_hash"`
InviteCode string `db:"invite_code"`
MemberStatus int `db:"member_status"`
MemberCategory int `db:"member_category"`
}
func (s *Store) RegisterAccount(ctx context.Context, input AccountRegistration) (*Account, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
parentID, teamLevel, tree, err := referralParent(ctx, tx, input.TenantID, input.ReferralCode)
if err != nil {
return nil, err
}
levelID, err := defaultMemberLevel(ctx, tx, input.TenantID)
if err != nil {
return nil, err
}
result, err := tx.ExecContext(ctx, `INSERT INTO hg_app_member
(tenant_id,member_no,username,mobile_cipher,mobile_hash,mobile_mask,password_hash,
level_id,parent_id,level,tree,invite_code,member_status,risk_status,auto_dispatch,
credit_score,is_test_account,created_by,updated_by)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,1,1,2,100,2,0,0)`,
input.TenantID, input.MemberNo, input.Username, input.MobileCipher, input.MobileHash,
input.MobileMask, input.PasswordHash, levelID, parentID, teamLevel, tree, input.InviteCode)
if err != nil {
if isDuplicate(err) {
return nil, ErrAccountExists
}
return nil, err
}
memberID, err := result.LastInsertId()
if err != nil {
return nil, err
}
if _, err = tx.ExecContext(ctx, `INSERT INTO hg_app_member_profile
(tenant_id,member_id,created_by,updated_by) VALUES(?,?,0,0)`, input.TenantID, memberID); err != nil {
return nil, err
}
if _, err = tx.ExecContext(ctx, `INSERT INTO hg_wallet_account
(tenant_id,member_id,currency,available_amount,frozen_amount,total_income,total_expense,account_status,version)
VALUES(?,?,'CNY',0,0,0,0,1,0)`, input.TenantID, memberID); err != nil {
return nil, err
}
if err = tx.Commit(); err != nil {
return nil, err
}
return &Account{TenantID: input.TenantID, MemberID: memberID, MemberNo: input.MemberNo,
Username: input.Username, MobileMask: input.MobileMask, MobileHash: input.MobileHash,
InviteCode: input.InviteCode, MemberStatus: memberStatusActive, MemberCategory: memberCategoryReal}, nil
}
func (s *Store) AccountByMobileHash(ctx context.Context, tenantID int64, mobileHash string) (*Account, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
return accountByMobileHash(ctx, s.db, tenantID, mobileHash)
}
func (s *Store) RecordLogin(ctx context.Context, tenantID, memberID int64) error {
_, err := s.db.ExecContext(ctx, `UPDATE hg_app_member SET last_login_at=?,updated_at=?
WHERE tenant_id=? AND id=? AND deleted_at IS NULL`, time.Now().UTC(), time.Now().UTC(), tenantID, memberID)
return err
}
type queryer interface {
GetContext(context.Context, any, string, ...any) error
}
func accountByMobileHash(ctx context.Context, db queryer, tenantID int64, mobileHash string) (*Account, error) {
var row Account
err := db.GetContext(ctx, &row, `SELECT tenant_id, id AS member_id, member_no, username, mobile_mask,
mobile_hash, password_hash, invite_code, member_status, risk_status AS member_category
FROM hg_app_member WHERE tenant_id=? AND mobile_hash=? AND deleted_at IS NULL LIMIT 1`, tenantID, mobileHash)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrInvalidCredentials
}
if err != nil {
return nil, err
}
if row.MemberStatus != memberStatusActive {
return nil, ErrInvalidCredentials
}
return &row, nil
}
func referralParent(ctx context.Context, tx *sqlx.Tx, tenantID int64, code string) (int64, int, string, error) {
code = strings.TrimSpace(code)
if code == "" {
return 0, 1, "", nil
}
var parent struct {
ID int64 `db:"id"`
Level int `db:"level"`
Tree string `db:"tree"`
}
err := tx.GetContext(ctx, &parent, `SELECT id,level,tree FROM hg_app_member
WHERE tenant_id=? AND invite_code=? AND member_status=1
AND deleted_at IS NULL LIMIT 1 FOR UPDATE`, tenantID, code)
if errors.Is(err, sql.ErrNoRows) {
return 0, 0, "", ErrReferralInvalid
}
if err != nil {
return 0, 0, "", err
}
tree := strings.Trim(strings.TrimSpace(parent.Tree)+","+formatInt64(parent.ID), ",")
if len(tree) > 512 {
return 0, 0, "", ErrReferralInvalid
}
return parent.ID, parent.Level + 1, tree, nil
}
func defaultMemberLevel(ctx context.Context, tx *sqlx.Tx, tenantID int64) (int64, error) {
var id int64
err := tx.GetContext(ctx, &id, `SELECT id FROM hg_app_member_level
WHERE tenant_id=? AND status=1 AND deleted_at IS NULL ORDER BY sort ASC,id ASC LIMIT 1`, tenantID)
if errors.Is(err, sql.ErrNoRows) {
return 0, nil
}
return id, err
}
func isDuplicate(err error) bool {
var mysqlErr *mysql.MySQLError
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1062
}
func formatInt64(value int64) string {
return strconv.FormatInt(value, 10)
}
-306
View File
@@ -1,306 +0,0 @@
package store
import (
"context"
"database/sql"
"strings"
"time"
)
type AdminWorkbenchModule struct {
ID int64 `db:"id"`
Code string `db:"code"`
Name string `db:"name"`
IconFileID string `db:"icon_file_id"`
Route string `db:"route"`
SortOrder int `db:"sort_order"`
UnreadKey string `db:"unread_key"`
Enabled int `db:"enabled"`
Permission string `db:"permission"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type AdminWorkbenchModuleInput struct {
ID int64
Code string
Name string
IconFileID string
Route string
SortOrder int
UnreadKey string
Enabled int
Permission string
}
type AdminWorkbenchReorderItem struct {
Code string
SortOrder int
}
type AdminMailboxConfig struct {
ID int64 `db:"id"`
Enabled int `db:"enabled"`
AllowAnonymous int `db:"allow_anonymous"`
RequiredFields string `db:"required_fields"`
MaxLength int `db:"max_length"`
AttachmentPolicy string `db:"attachment_policy"`
NoticeTitle string `db:"notice_title"`
NoticeContent string `db:"notice_content"`
NoticeUpdatedAt time.Time `db:"notice_updated_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type AdminFeedbackConfig struct {
ID int64 `db:"id"`
Categories string `db:"categories"`
MaxLength int `db:"max_length"`
AttachmentPolicy string `db:"attachment_policy"`
AllowContact int `db:"allow_contact"`
DiagnosticsNotice string `db:"diagnostics_notice"`
UpdatedAt time.Time `db:"updated_at"`
}
func (s *Store) AdminWorkbenchModules(ctx context.Context, page, perPage int, code, name string, enabled *int) ([]AdminWorkbenchModule, int, error) {
if page <= 0 {
page = 1
}
if perPage <= 0 {
perPage = 20
}
if perPage > 200 {
perPage = 200
}
where := []string{"tenant_id=?"}
args := []any{s.tenant()}
if code = strings.TrimSpace(code); code != "" {
where = append(where, "code LIKE ?")
args = append(args, "%"+code+"%")
}
if name = strings.TrimSpace(name); name != "" {
where = append(where, "name LIKE ?")
args = append(args, "%"+name+"%")
}
if enabled != nil {
where = append(where, "enabled=?")
args = append(args, *enabled)
}
whereSQL := strings.Join(where, " AND ")
var total int
if err := s.db.GetContext(ctx, &total, "SELECT COUNT(*) FROM im_workbench_apps WHERE "+whereSQL, args...); err != nil {
return nil, 0, err
}
queryArgs := append([]any{}, args...)
queryArgs = append(queryArgs, (page-1)*perPage, perPage)
var items []AdminWorkbenchModule
err := s.db.SelectContext(ctx, &items, `SELECT id,code,name,IFNULL(icon_file_id,'') AS icon_file_id,route,sort_order,IFNULL(unread_key,'') AS unread_key,enabled+0 AS enabled,IFNULL(permission,'enabled') AS permission,created_at,updated_at
FROM im_workbench_apps WHERE `+whereSQL+` ORDER BY sort_order ASC,id ASC LIMIT ?,?`, queryArgs...)
return items, total, err
}
func (s *Store) UpsertAdminWorkbenchModule(ctx context.Context, in AdminWorkbenchModuleInput) (*AdminWorkbenchModule, error) {
now := time.Now().UTC().Round(time.Millisecond)
in.Code = strings.TrimSpace(in.Code)
in.Name = strings.TrimSpace(in.Name)
in.IconFileID = strings.TrimSpace(in.IconFileID)
in.Route = strings.TrimSpace(in.Route)
in.UnreadKey = strings.TrimSpace(in.UnreadKey)
in.Permission = strings.TrimSpace(in.Permission)
if in.Permission == "" {
in.Permission = "enabled"
}
if in.ID > 0 {
res, err := s.db.ExecContext(ctx, `UPDATE im_workbench_apps SET code=?,name=?,icon_file_id=?,route=?,sort_order=?,unread_key=?,enabled=?,permission=?,updated_at=? WHERE tenant_id=? AND id=?`,
in.Code, in.Name, in.IconFileID, in.Route, in.SortOrder, in.UnreadKey, in.Enabled, in.Permission, now, s.tenant(), in.ID)
if err != nil {
return nil, err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return nil, ErrNotFound
}
return s.adminWorkbenchModuleByID(ctx, in.ID)
}
res, err := s.db.ExecContext(ctx, `INSERT INTO im_workbench_apps(tenant_id,code,name,icon_file_id,route,sort_order,unread_key,enabled,permission,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?)`, s.tenant(), in.Code, in.Name, in.IconFileID, in.Route, in.SortOrder, in.UnreadKey, in.Enabled, in.Permission, now, now)
if err != nil {
return nil, err
}
id, _ := res.LastInsertId()
return s.adminWorkbenchModuleByID(ctx, id)
}
func (s *Store) DisableAdminWorkbenchModule(ctx context.Context, id int64, code string) error {
now := time.Now().UTC().Round(time.Millisecond)
var res sql.Result
var err error
if id > 0 {
res, err = s.db.ExecContext(ctx, `UPDATE im_workbench_apps SET enabled=0,updated_at=? WHERE tenant_id=? AND id=?`, now, s.tenant(), id)
} else {
res, err = s.db.ExecContext(ctx, `UPDATE im_workbench_apps SET enabled=0,updated_at=? WHERE tenant_id=? AND code=?`, now, s.tenant(), strings.TrimSpace(code))
}
if err != nil {
return err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return ErrNotFound
}
return nil
}
func (s *Store) ReorderAdminWorkbenchModules(ctx context.Context, items []AdminWorkbenchReorderItem) error {
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
now := time.Now().UTC().Round(time.Millisecond)
for _, item := range items {
if _, err = tx.ExecContext(ctx, `UPDATE im_workbench_apps SET sort_order=?,updated_at=? WHERE tenant_id=? AND code=?`, item.SortOrder, now, s.tenant(), strings.TrimSpace(item.Code)); err != nil {
return err
}
}
return tx.Commit()
}
func (s *Store) AdminOrganizationDepartments(ctx context.Context, page, perPage int, parentID, name, state string) ([]OrganizationDepartment, int, error) {
page, perPage = normalizeAdminPage(page, perPage)
where := []string{"tenant_id=?"}
args := []any{s.tenant()}
if parentID = strings.TrimSpace(parentID); parentID != "" {
where = append(where, "parent_id=?")
args = append(args, parentID)
}
if name = strings.TrimSpace(name); name != "" {
where = append(where, "name LIKE ?")
args = append(args, "%"+name+"%")
}
if state = strings.TrimSpace(state); state != "" {
where = append(where, "state=?")
args = append(args, state)
}
whereSQL := strings.Join(where, " AND ")
var total int
if err := s.db.GetContext(ctx, &total, "SELECT COUNT(*) FROM im_org_departments WHERE "+whereSQL, args...); err != nil {
return nil, 0, err
}
queryArgs := append([]any{}, args...)
queryArgs = append(queryArgs, (page-1)*perPage, perPage)
var items []OrganizationDepartment
err := s.db.SelectContext(ctx, &items, `SELECT d.department_id,d.name,d.parent_id,
(SELECT COUNT(*) FROM im_org_departments c WHERE c.tenant_id=d.tenant_id AND c.parent_id=d.department_id AND c.state='active') AS child_count,
(SELECT COUNT(*) FROM im_org_members m WHERE m.tenant_id=d.tenant_id AND m.department_id=d.department_id AND m.employment_state='active') AS member_count,
d.version
FROM im_org_departments d WHERE `+whereSQL+` ORDER BY d.parent_id ASC,d.version DESC,d.department_id ASC LIMIT ?,?`, queryArgs...)
return items, total, err
}
func (s *Store) AdminOrganizationMembers(ctx context.Context, page, perPage int, departmentID, keyword, state string, userID int64) ([]OrganizationMember, int, error) {
page, perPage = normalizeAdminPage(page, perPage)
where := []string{"tenant_id=?"}
args := []any{s.tenant()}
if departmentID = strings.TrimSpace(departmentID); departmentID != "" {
where = append(where, "department_id=?")
args = append(args, departmentID)
}
if keyword = strings.TrimSpace(keyword); keyword != "" {
where = append(where, "(display_name LIKE ? OR position LIKE ? OR mobile LIKE ? OR email LIKE ?)")
like := "%" + keyword + "%"
args = append(args, like, like, like, like)
}
if state = strings.TrimSpace(state); state != "" {
where = append(where, "employment_state=?")
args = append(args, state)
}
if userID > 0 {
where = append(where, "user_id=?")
args = append(args, userID)
}
whereSQL := strings.Join(where, " AND ")
var total int
if err := s.db.GetContext(ctx, &total, "SELECT COUNT(*) FROM im_org_members WHERE "+whereSQL, args...); err != nil {
return nil, 0, err
}
queryArgs := append([]any{}, args...)
queryArgs = append(queryArgs, (page-1)*perPage, perPage)
var items []OrganizationMember
err := s.db.SelectContext(ctx, &items, `SELECT user_id,department_id,display_name,position,IFNULL(mobile,'') AS mobile,IFNULL(email,'') AS email,
IFNULL(avatar_file_id,'') AS avatar_file_id,employment_state AS status,contactable,version
FROM im_org_members WHERE `+whereSQL+` ORDER BY department_id ASC,user_id ASC LIMIT ?,?`, queryArgs...)
return items, total, err
}
func (s *Store) AdminMailboxConfig(ctx context.Context) (*AdminMailboxConfig, error) {
var cfg AdminMailboxConfig
if err := s.db.GetContext(ctx, &cfg, `SELECT id,enabled+0 AS enabled,allow_anonymous+0 AS allow_anonymous,CAST(required_fields AS CHAR) AS required_fields,max_length,CAST(attachment_policy AS CHAR) AS attachment_policy,notice_title,notice_content,notice_updated_at,updated_at FROM im_mailbox_config WHERE tenant_id=? LIMIT 1`, s.tenant()); err != nil {
if errorsIsNoRows(err) {
now := time.Now().UTC().Round(time.Millisecond)
return &AdminMailboxConfig{Enabled: 1, AllowAnonymous: 1, RequiredFields: `["subject","content"]`, MaxLength: 5000, AttachmentPolicy: `{"max_files":5,"purpose":"executive_mailbox"}`, NoticeTitle: "总经理信箱", NoticeContent: "请如实填写诉求。", NoticeUpdatedAt: now, UpdatedAt: now}, nil
}
return nil, err
}
return &cfg, nil
}
func (s *Store) UpdateAdminMailboxConfig(ctx context.Context, cfg AdminMailboxConfig) error {
now := time.Now().UTC().Round(time.Millisecond)
_, err := s.db.ExecContext(ctx, `INSERT INTO im_mailbox_config(tenant_id,enabled,allow_anonymous,required_fields,max_length,attachment_policy,notice_title,notice_content,notice_updated_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE enabled=VALUES(enabled),allow_anonymous=VALUES(allow_anonymous),required_fields=VALUES(required_fields),max_length=VALUES(max_length),attachment_policy=VALUES(attachment_policy),notice_title=VALUES(notice_title),notice_content=VALUES(notice_content),notice_updated_at=VALUES(notice_updated_at),updated_at=VALUES(updated_at)`,
s.tenant(), cfg.Enabled, cfg.AllowAnonymous, defaultString(cfg.RequiredFields, `["subject","content"]`), cfg.MaxLength, defaultString(cfg.AttachmentPolicy, `{"max_files":5,"purpose":"executive_mailbox"}`), cfg.NoticeTitle, cfg.NoticeContent, now, now, now)
return err
}
func (s *Store) AdminFeedbackConfig(ctx context.Context) (*AdminFeedbackConfig, error) {
var cfg AdminFeedbackConfig
if err := s.db.GetContext(ctx, &cfg, `SELECT id,CAST(categories AS CHAR) AS categories,max_length,CAST(attachment_policy AS CHAR) AS attachment_policy,allow_contact+0 AS allow_contact,diagnostics_notice,updated_at FROM im_feedback_config WHERE tenant_id=? LIMIT 1`, s.tenant()); err != nil {
if errorsIsNoRows(err) {
return &AdminFeedbackConfig{Categories: `["bug","suggestion","complaint","other"]`, MaxLength: 2000, AttachmentPolicy: `{"max_files":5,"purpose":"feedback"}`, AllowContact: 1, UpdatedAt: time.Now().UTC().Round(time.Millisecond)}, nil
}
return nil, err
}
return &cfg, nil
}
func (s *Store) UpdateAdminFeedbackConfig(ctx context.Context, cfg AdminFeedbackConfig) error {
now := time.Now().UTC().Round(time.Millisecond)
_, err := s.db.ExecContext(ctx, `INSERT INTO im_feedback_config(tenant_id,categories,max_length,attachment_policy,allow_contact,diagnostics_notice,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?)
ON DUPLICATE KEY UPDATE categories=VALUES(categories),max_length=VALUES(max_length),attachment_policy=VALUES(attachment_policy),allow_contact=VALUES(allow_contact),diagnostics_notice=VALUES(diagnostics_notice),updated_at=VALUES(updated_at)`,
s.tenant(), defaultString(cfg.Categories, `["bug","suggestion","complaint","other"]`), cfg.MaxLength, defaultString(cfg.AttachmentPolicy, `{"max_files":5,"purpose":"feedback"}`), cfg.AllowContact, cfg.DiagnosticsNotice, now, now)
return err
}
func (s *Store) adminWorkbenchModuleByID(ctx context.Context, id int64) (*AdminWorkbenchModule, error) {
var item AdminWorkbenchModule
if err := s.db.GetContext(ctx, &item, `SELECT id,code,name,IFNULL(icon_file_id,'') AS icon_file_id,route,sort_order,IFNULL(unread_key,'') AS unread_key,enabled+0 AS enabled,IFNULL(permission,'enabled') AS permission,created_at,updated_at FROM im_workbench_apps WHERE tenant_id=? AND id=?`, s.tenant(), id); err != nil {
if errorsIsNoRows(err) {
return nil, ErrNotFound
}
return nil, err
}
return &item, nil
}
func errorsIsNoRows(err error) bool {
return err == sql.ErrNoRows
}
func defaultString(value string, fallback string) string {
if strings.TrimSpace(value) == "" {
return fallback
}
return value
}
func normalizeAdminPage(page, perPage int) (int, int) {
if page <= 0 {
page = 1
}
if perPage <= 0 {
perPage = 20
}
if perPage > 200 {
perPage = 200
}
return page, perPage
}
-603
View File
@@ -1,603 +0,0 @@
package store
import (
"context"
"database/sql"
"encoding/json"
"strconv"
"strings"
"time"
)
type AICapability struct {
Feature string `db:"feature"`
Enabled bool `db:"enabled"`
Model string `db:"model"`
DailyLimit int `db:"daily_limit"`
ConcurrencyLimit int `db:"concurrency_limit"`
MaxInputChars int `db:"max_input_chars"`
MaxFileSizeMB int `db:"max_file_size_mb"`
ConfigJSON string `db:"config_json"`
}
type AITask struct {
TaskID string `db:"task_id"`
TenantID int64 `db:"tenant_id"`
TenantCode string `db:"tenant_code"`
UserID string `db:"user_id"`
Feature string `db:"feature"`
Status string `db:"status"`
Progress int `db:"progress"`
SourceType string `db:"source_type"`
SourceRefID string `db:"source_ref_id"`
InputHash string `db:"input_hash"`
InputSnapshot sql.NullString `db:"input_snapshot"`
PromptVersion string `db:"prompt_version"`
Model string `db:"model"`
ResultRefID string `db:"result_ref_id"`
ResultJSON sql.NullString `db:"result_json"`
ErrorReason string `db:"error_reason"`
ErrorMessage string `db:"error_message"`
RetryCount int `db:"retry_count"`
StartedAt *time.Time `db:"started_at"`
FinishedAt *time.Time `db:"finished_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type AIRecord struct {
RecordID string `db:"record_id"`
TenantCode string `db:"tenant_code"`
UserID string `db:"user_id"`
RecordType string `db:"record_type"`
Title string `db:"title"`
Summary string `db:"summary"`
Status string `db:"status"`
DetailID string `db:"detail_id"`
TaskID string `db:"task_id"`
FileName string `db:"file_name"`
FileType string `db:"file_type"`
FileSize int64 `db:"file_size"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
DeletedAt *time.Time `db:"deleted_at"`
}
type AIDocument struct {
RecordID string `db:"record_id"`
UserID string `db:"user_id"`
DocumentType string `db:"document_type"`
Title string `db:"title"`
ContentMarkdown string `db:"content_markdown"`
ContentPlain string `db:"content_plain"`
SectionsJSON string `db:"sections_json"`
SourceExcerpt string `db:"source_excerpt"`
PromptVersion string `db:"prompt_version"`
Model string `db:"model"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type AITableAnalysis struct {
RecordID string `db:"record_id"`
UserID string `db:"user_id"`
FileID string `db:"file_id"`
FileName string `db:"file_name"`
FileType string `db:"file_type"`
FileSize int64 `db:"file_size"`
Topic string `db:"topic"`
MetricsJSON string `db:"metrics_json"`
PreviewJSON string `db:"preview_json"`
AlertsJSON string `db:"alerts_json"`
Interpretation string `db:"interpretation"`
ExportFileID string `db:"export_file_id"`
PromptVersion string `db:"prompt_version"`
Model string `db:"model"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type AIMeetingSummary struct {
SummaryID string `db:"summary_id"`
MeetingID string `db:"meeting_id"`
UserID string `db:"user_id"`
TaskID string `db:"task_id"`
ConclusionsJSON string `db:"conclusions_json"`
DecisionsJSON string `db:"decisions_json"`
ActionItemsJSON string `db:"action_items_json"`
SourceFileIDsJSON string `db:"source_file_ids_json"`
PromptVersion string `db:"prompt_version"`
Model string `db:"model"`
EditedBy string `db:"edited_by"`
EditedAt *time.Time `db:"edited_at"`
Version int `db:"version"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type AIUsageLog struct {
TenantCode string
UserID string
Feature string
TaskID string
Provider string
Model string
ProviderRequestID string
PromptTokens int
CompletionTokens int
TotalTokens int
LatencyMS int
Status string
ErrorReason string
InputHash string
OutputHash string
}
type AIFileMetadata struct {
MIMEType string
FileSize int64
FileETag string
Location string
}
type CompletedFileUpload struct {
ID int64
UserID int64
MIMEType string
Size int64
ETag string
Location string
CreatedAt time.Time
}
func (s *Store) AICapabilities(ctx context.Context) ([]AICapability, error) {
var items []AICapability
err := s.db.SelectContext(ctx, &items, `SELECT feature,enabled,model,daily_limit,concurrency_limit,
max_input_chars,max_file_size_mb,COALESCE(CAST(config_json AS CHAR), '{}') AS config_json
FROM wb_ai_capability_config WHERE tenant_id=? ORDER BY feature ASC`, s.tenant())
if err != nil {
return nil, err
}
if len(items) > 0 {
return items, nil
}
err = s.db.SelectContext(ctx, &items, `SELECT feature,enabled,model,daily_limit,concurrency_limit,
max_input_chars,max_file_size_mb,COALESCE(CAST(config_json AS CHAR), '{}') AS config_json
FROM wb_ai_capability_config WHERE tenant_id=0 ORDER BY feature ASC`)
return items, err
}
func (s *Store) AIFileMetadata(ctx context.Context, fileID string) AIFileMetadata {
meta := s.fileUploadMetadata(ctx, strings.TrimSpace(fileID))
return AIFileMetadata{
MIMEType: meta.MIMEType,
FileSize: meta.FileSize,
FileETag: meta.FileETag,
Location: meta.Location,
}
}
func (s *Store) CreateAITask(ctx context.Context, task AITask, result any) error {
now := time.Now().UTC().Round(time.Millisecond)
resultJSON, err := marshalNullable(result)
if err != nil {
return err
}
var inputSnapshot any
if task.InputSnapshot.Valid {
inputSnapshot = task.InputSnapshot.String
}
_, err = s.db.ExecContext(ctx, `INSERT INTO wb_ai_task
(task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id,input_hash,
input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count,
started_at,finished_at,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0,?,?,?,?)`,
task.TaskID, s.tenant(), task.TenantCode, task.UserID, task.Feature, task.Status, task.Progress,
task.SourceType, task.SourceRefID, task.InputHash, inputSnapshot, task.PromptVersion, task.Model, task.ResultRefID,
resultJSON, task.ErrorReason, task.ErrorMessage, task.StartedAt, task.FinishedAt, now, now)
return err
}
func (s *Store) ClaimQueuedAITask(ctx context.Context) (*AITask, error) {
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
var task AITask
err = tx.GetContext(ctx, &task, `SELECT task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id,
input_hash,input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count,
started_at,finished_at,created_at,updated_at
FROM wb_ai_task
WHERE status='queued' AND feature IN ('smart_document','smart_table','meeting_summary')
ORDER BY created_at ASC,id ASC LIMIT 1 FOR UPDATE SKIP LOCKED`)
if err == sql.ErrNoRows {
return nil, ErrNotFound
}
if err != nil {
return nil, err
}
now := time.Now().UTC().Round(time.Millisecond)
res, err := tx.ExecContext(ctx, `UPDATE wb_ai_task SET status='processing',progress=?,started_at=COALESCE(started_at,?),updated_at=?
WHERE tenant_id=? AND task_id=? AND status='queued'`, maxInt(task.Progress, 10), now, now, task.TenantID, task.TaskID)
if err != nil {
return nil, err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return nil, ErrNotFound
}
if err := tx.Commit(); err != nil {
return nil, err
}
task.Status = "processing"
task.Progress = maxInt(task.Progress, 10)
task.StartedAt = &now
task.UpdatedAt = now
return &task, nil
}
func (s *Store) UpdateAITaskResult(ctx context.Context, taskID, status string, progress int, resultRefID string, result any, reason, message string) error {
now := time.Now().UTC().Round(time.Millisecond)
resultJSON, err := marshalNullable(result)
if err != nil {
return err
}
res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_task SET status=?,progress=?,result_ref_id=?,result_json=?,
error_reason=?,error_message=?,finished_at=?,updated_at=? WHERE tenant_id=? AND task_id=?`,
status, progress, resultRefID, resultJSON, reason, message, now, now, s.tenant(), taskID)
if err != nil {
return err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return ErrNotFound
}
return nil
}
func (s *Store) FinishAITaskResult(ctx context.Context, taskID, status string, progress int, resultRefID string, result any, reason, message string) (bool, error) {
now := time.Now().UTC().Round(time.Millisecond)
resultJSON, err := marshalNullable(result)
if err != nil {
return false, err
}
res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_task SET status=?,progress=?,result_ref_id=?,result_json=?,
error_reason=?,error_message=?,finished_at=?,updated_at=? WHERE tenant_id=? AND task_id=? AND status='processing'`,
status, progress, resultRefID, resultJSON, reason, message, now, now, s.tenant(), taskID)
if err != nil {
return false, err
}
rows, err := res.RowsAffected()
if err != nil {
return false, err
}
return rows > 0, nil
}
func (s *Store) CancelAITask(ctx context.Context, userID int64, taskID string) error {
now := time.Now().UTC().Round(time.Millisecond)
res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_task SET status='canceled',progress=100,finished_at=?,updated_at=?
WHERE tenant_id=? AND user_id=? AND task_id=? AND status IN ('queued','processing')`,
now, now, s.tenant(), formatInt64(userID), taskID)
if err != nil {
return err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return ErrNotFound
}
return nil
}
func (s *Store) AITask(ctx context.Context, userID int64, taskID string) (*AITask, error) {
var task AITask
err := s.db.GetContext(ctx, &task, `SELECT task_id,tenant_id,tenant_code,user_id,feature,status,progress,source_type,source_ref_id,
input_hash,input_snapshot,prompt_version,model,result_ref_id,result_json,error_reason,error_message,retry_count,
started_at,finished_at,created_at,updated_at
FROM wb_ai_task WHERE tenant_id=? AND user_id=? AND task_id=?`, s.tenant(), formatInt64(userID), taskID)
if err == sql.ErrNoRows {
return nil, ErrNotFound
}
if err != nil {
return nil, err
}
return &task, nil
}
func (s *Store) InsertAIUsageLog(ctx context.Context, log AIUsageLog) error {
now := time.Now().UTC().Round(time.Millisecond)
_, err := s.db.ExecContext(ctx, `INSERT INTO wb_ai_usage_log
(tenant_id,tenant_code,user_id,feature,task_id,provider,model,provider_request_id,prompt_tokens,
completion_tokens,total_tokens,latency_ms,status,error_reason,input_hash,output_hash,created_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
s.tenant(), log.TenantCode, log.UserID, log.Feature, log.TaskID, log.Provider, log.Model, log.ProviderRequestID,
log.PromptTokens, log.CompletionTokens, log.TotalTokens, log.LatencyMS, log.Status, log.ErrorReason,
log.InputHash, log.OutputHash, now)
return err
}
func (s *Store) SaveAIDocument(ctx context.Context, tenantCode string, userID int64, record AIRecord, doc AIDocument) error {
now := time.Now().UTC().Round(time.Millisecond)
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
_, err = tx.ExecContext(ctx, `INSERT INTO wb_ai_record
(record_id,tenant_id,tenant_code,user_id,record_type,title,summary,status,detail_id,task_id,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
record.RecordID, s.tenant(), tenantCode, formatInt64(userID), record.RecordType, record.Title, record.Summary,
record.Status, doc.RecordID, record.TaskID, now, now)
if err != nil {
return err
}
_, err = tx.ExecContext(ctx, `INSERT INTO wb_ai_document
(record_id,tenant_id,user_id,document_type,title,content_markdown,content_plain,sections_json,source_excerpt,
prompt_version,model,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)`,
doc.RecordID, s.tenant(), formatInt64(userID), doc.DocumentType, doc.Title, doc.ContentMarkdown, doc.ContentPlain,
doc.SectionsJSON, doc.SourceExcerpt, doc.PromptVersion, doc.Model, now, now)
if err != nil {
return err
}
return tx.Commit()
}
func (s *Store) AIDocuments(ctx context.Context, userID int64, queryText, cursor string, limit int) ([]AIRecord, string, error) {
return s.aiRecords(ctx, userID, "smart_document", queryText, cursor, limit)
}
func (s *Store) AIDocument(ctx context.Context, userID int64, recordID string) (*AIDocument, error) {
var doc AIDocument
err := s.db.GetContext(ctx, &doc, `SELECT d.record_id,d.user_id,d.document_type,d.title,d.content_markdown,
COALESCE(d.content_plain,'') AS content_plain,COALESCE(CAST(d.sections_json AS CHAR), '[]') AS sections_json,
d.source_excerpt,d.prompt_version,d.model,d.created_at,d.updated_at
FROM wb_ai_document d
JOIN wb_ai_record r ON r.tenant_id=d.tenant_id AND r.record_id=d.record_id
WHERE d.tenant_id=? AND d.user_id=? AND d.record_id=? AND r.deleted_at IS NULL`,
s.tenant(), formatInt64(userID), recordID)
if err == sql.ErrNoRows {
return nil, ErrNotFound
}
if err != nil {
return nil, err
}
return &doc, nil
}
func (s *Store) SaveAITableAnalysis(ctx context.Context, tenantCode string, userID int64, record AIRecord, analysis AITableAnalysis) error {
now := time.Now().UTC().Round(time.Millisecond)
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
_, err = tx.ExecContext(ctx, `INSERT INTO wb_ai_record
(record_id,tenant_id,tenant_code,user_id,record_type,title,summary,status,detail_id,task_id,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)`,
record.RecordID, s.tenant(), tenantCode, formatInt64(userID), record.RecordType, record.Title, record.Summary,
record.Status, analysis.RecordID, record.TaskID, now, now)
if err != nil {
return err
}
_, err = tx.ExecContext(ctx, `INSERT INTO wb_ai_table_analysis
(record_id,tenant_id,user_id,file_id,file_name,file_type,file_size,topic,metrics_json,preview_json,alerts_json,
interpretation,export_file_id,prompt_version,model,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
analysis.RecordID, s.tenant(), formatInt64(userID), analysis.FileID, analysis.FileName, analysis.FileType, analysis.FileSize,
analysis.Topic, analysis.MetricsJSON, analysis.PreviewJSON, analysis.AlertsJSON, analysis.Interpretation, analysis.ExportFileID,
analysis.PromptVersion, analysis.Model, now, now)
if err != nil {
return err
}
return tx.Commit()
}
func (s *Store) SaveCompletedFileUpload(ctx context.Context, file CompletedFileUpload) error {
now := file.CreatedAt.UTC().Round(time.Millisecond)
if now.IsZero() {
now = time.Now().UTC().Round(time.Millisecond)
}
db := s.db
if s.core != nil && s.core.db != nil {
db = s.core.db
}
_, err := db.ExecContext(ctx, `INSERT INTO im_fileuploads
(tenant_id,id,createdat,updatedat,userid,status,mimetype,size,etag,location)
VALUES(?,?,?,?,?,?,?,?,?,?)`,
s.tenant(), file.ID, now, now, formatInt64(file.UserID), 1, file.MIMEType, file.Size, file.ETag, file.Location)
return err
}
func (s *Store) UpdateAITableAnalysisExportFileID(ctx context.Context, userID int64, recordID, fileID string) error {
now := time.Now().UTC().Round(time.Millisecond)
res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_table_analysis SET export_file_id=?,updated_at=?
WHERE tenant_id=? AND user_id=? AND record_id=?`,
fileID, now, s.tenant(), formatInt64(userID), recordID)
if err != nil {
return err
}
affected, err := res.RowsAffected()
if err != nil {
return err
}
if affected == 0 {
return ErrNotFound
}
return nil
}
func (s *Store) AITableAnalyses(ctx context.Context, userID int64, queryText, cursor string, limit int) ([]AIRecord, string, error) {
where := `r.tenant_id=? AND r.user_id=? AND r.record_type='smart_table' AND r.deleted_at IS NULL`
args := []any{s.tenant(), formatInt64(userID)}
if queryText != "" {
where += ` AND (r.title LIKE ? OR r.summary LIKE ? OR a.file_name LIKE ?)`
like := "%" + queryText + "%"
args = append(args, like, like, like)
}
if cursor != "" {
where += ` AND r.record_id<?`
args = append(args, cursor)
}
args = append(args, limit+1)
var items []AIRecord
err := s.db.SelectContext(ctx, &items, `SELECT r.record_id,r.tenant_code,r.user_id,r.record_type,r.title,r.summary,r.status,r.detail_id,r.task_id,
COALESCE(a.file_name, '') AS file_name,COALESCE(a.file_type, '') AS file_type,COALESCE(a.file_size, 0) AS file_size,
r.created_at,r.updated_at,r.deleted_at
FROM wb_ai_record r
LEFT JOIN wb_ai_table_analysis a ON a.tenant_id=r.tenant_id AND a.record_id=r.detail_id
WHERE `+where+` ORDER BY r.created_at DESC,r.record_id DESC LIMIT ?`, args...)
if err != nil {
return nil, "", err
}
next := ""
if len(items) > limit {
next = items[limit-1].RecordID
items = items[:limit]
}
return items, next, nil
}
func (s *Store) AIRecords(ctx context.Context, userID int64, recordType, queryText, cursor string, limit int) ([]AIRecord, string, error) {
return s.aiRecords(ctx, userID, recordType, queryText, cursor, limit)
}
func (s *Store) DeleteAIRecord(ctx context.Context, userID int64, recordID string) error {
now := time.Now().UTC().Round(time.Millisecond)
res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_record SET status='deleted',deleted_at=?,updated_at=?
WHERE tenant_id=? AND user_id=? AND record_id=? AND deleted_at IS NULL`,
now, now, s.tenant(), formatInt64(userID), recordID)
if err != nil {
return err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return ErrNotFound
}
return nil
}
func (s *Store) AITableAnalysis(ctx context.Context, userID int64, recordID string) (*AITableAnalysis, error) {
var item AITableAnalysis
err := s.db.GetContext(ctx, &item, `SELECT a.record_id,a.user_id,a.file_id,a.file_name,a.file_type,a.file_size,a.topic,
COALESCE(CAST(a.metrics_json AS CHAR), '[]') AS metrics_json,
COALESCE(CAST(a.preview_json AS CHAR), '{}') AS preview_json,
COALESCE(CAST(a.alerts_json AS CHAR), '[]') AS alerts_json,
a.interpretation,a.export_file_id,a.prompt_version,a.model,a.created_at,a.updated_at
FROM wb_ai_table_analysis a
JOIN wb_ai_record r ON r.tenant_id=a.tenant_id AND r.record_id=a.record_id
WHERE a.tenant_id=? AND a.user_id=? AND a.record_id=? AND r.deleted_at IS NULL`,
s.tenant(), formatInt64(userID), recordID)
if err == sql.ErrNoRows {
return nil, ErrNotFound
}
if err != nil {
return nil, err
}
return &item, nil
}
func (s *Store) UpsertAIMeetingSummary(ctx context.Context, tenantCode string, userID int64, summary AIMeetingSummary) error {
now := time.Now().UTC().Round(time.Millisecond)
_, err := s.db.ExecContext(ctx, `INSERT INTO wb_ai_meeting_summary
(summary_id,tenant_id,tenant_code,meeting_id,user_id,task_id,conclusions_json,decisions_json,action_items_json,
source_file_ids_json,prompt_version,model,version,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,1,?,?)
ON DUPLICATE KEY UPDATE task_id=VALUES(task_id),conclusions_json=VALUES(conclusions_json),
decisions_json=VALUES(decisions_json),action_items_json=VALUES(action_items_json),
source_file_ids_json=VALUES(source_file_ids_json),prompt_version=VALUES(prompt_version),model=VALUES(model),
version=version+1,updated_at=VALUES(updated_at)`,
summary.SummaryID, s.tenant(), tenantCode, summary.MeetingID, formatInt64(userID), summary.TaskID,
summary.ConclusionsJSON, summary.DecisionsJSON, summary.ActionItemsJSON, summary.SourceFileIDsJSON,
summary.PromptVersion, summary.Model, now, now)
return err
}
func (s *Store) AIMeetingSummary(ctx context.Context, userID int64, meetingID string) (*AIMeetingSummary, error) {
var item AIMeetingSummary
err := s.db.GetContext(ctx, &item, `SELECT summary_id,meeting_id,user_id,task_id,
COALESCE(CAST(conclusions_json AS CHAR), '[]') AS conclusions_json,
COALESCE(CAST(decisions_json AS CHAR), '[]') AS decisions_json,
COALESCE(CAST(action_items_json AS CHAR), '[]') AS action_items_json,
COALESCE(CAST(source_file_ids_json AS CHAR), '[]') AS source_file_ids_json,
prompt_version,model,edited_by,edited_at,version,created_at,updated_at
FROM wb_ai_meeting_summary WHERE tenant_id=? AND meeting_id=? LIMIT 1`, s.tenant(), meetingID)
if err == sql.ErrNoRows {
return nil, ErrNotFound
}
if err != nil {
return nil, err
}
return &item, nil
}
func (s *Store) UpdateAIMeetingSummary(ctx context.Context, userID int64, meetingID, summaryID, conclusionsJSON, decisionsJSON, actionItemsJSON string) (*AIMeetingSummary, error) {
meeting, err := s.Meeting(ctx, userID, meetingID)
if err != nil {
return nil, err
}
if meeting.OrganizerID != userID && meeting.RecorderID != userID {
return nil, ErrForbidden
}
now := time.Now().UTC().Round(time.Millisecond)
res, err := s.db.ExecContext(ctx, `UPDATE wb_ai_meeting_summary SET conclusions_json=?,decisions_json=?,action_items_json=?,
edited_by=?,edited_at=?,version=version+1,updated_at=? WHERE tenant_id=? AND meeting_id=? AND summary_id=?`,
conclusionsJSON, decisionsJSON, actionItemsJSON, formatInt64(userID), now, now, s.tenant(), meetingID, summaryID)
if err != nil {
return nil, err
}
if rows, _ := res.RowsAffected(); rows == 0 {
return nil, ErrNotFound
}
return s.AIMeetingSummary(ctx, userID, meetingID)
}
func (s *Store) aiRecords(ctx context.Context, userID int64, recordType, queryText, cursor string, limit int) ([]AIRecord, string, error) {
where := `tenant_id=? AND user_id=? AND deleted_at IS NULL`
args := []any{s.tenant(), formatInt64(userID)}
if recordType != "" {
where += ` AND record_type=?`
args = append(args, recordType)
}
if queryText != "" {
where += ` AND (title LIKE ? OR summary LIKE ?)`
like := "%" + queryText + "%"
args = append(args, like, like)
}
if cursor != "" {
where += ` AND record_id<?`
args = append(args, cursor)
}
args = append(args, limit+1)
var items []AIRecord
err := s.db.SelectContext(ctx, &items, `SELECT record_id,tenant_code,user_id,record_type,title,summary,status,detail_id,task_id,
created_at,updated_at,deleted_at FROM wb_ai_record WHERE `+where+` ORDER BY created_at DESC,record_id DESC LIMIT ?`, args...)
if err != nil {
return nil, "", err
}
next := ""
if len(items) > limit {
next = items[limit-1].RecordID
items = items[:limit]
}
return items, next, nil
}
func marshalNullable(value any) (any, error) {
if value == nil {
return nil, nil
}
data, err := json.Marshal(value)
if err != nil {
return nil, err
}
return string(data), nil
}
func formatInt64(value int64) string {
return strconv.FormatInt(value, 10)
}
func maxInt(a, b int) int {
if a > b {
return a
}
return b
}
-116
View File
@@ -1,116 +0,0 @@
package store
import (
"context"
"strings"
"git.118im.com/im_team/im_workbench/internal/config"
"github.com/jmoiron/sqlx"
)
// CoreStore provides read-only access to shared IM data.
type CoreStore struct {
db *sqlx.DB
}
func coreStoreOwnsDB(primaryDSN, coreDSN string) bool {
return strings.TrimSpace(primaryDSN) != strings.TrimSpace(coreDSN)
}
func openCoreDSN(cfg config.MySQLConfig) (*CoreStore, error) {
db, err := openMySQLDB(cfg)
if err != nil {
return nil, err
}
return &CoreStore{db: db}, nil
}
func (s *CoreStore) Close() error {
if s == nil || s.db == nil {
return nil
}
return s.db.Close()
}
func (s *CoreStore) fileUploadMetadata(ctx context.Context, tenantID int64, fileID string) fileUploadMetadata {
var meta fileUploadMetadata
decodedID, storageName := parseUploadFileID(fileID)
locationCandidates := uploadLocationCandidates(fileID, storageName)
if decodedID <= 0 && len(locationCandidates) == 0 {
return meta
}
if decodedID > 0 {
err := s.db.GetContext(ctx, &meta, `SELECT IFNULL(mimetype,'') AS mimetype,IFNULL(size,0) AS size,IFNULL(etag,'') AS etag,IFNULL(location,'') AS location,IFNULL(original_name,'') AS original_name FROM im_fileuploads WHERE tenant_id=? AND id=? LIMIT 1`, tenantID, decodedID)
if err == nil {
return meta
}
}
for _, candidate := range locationCandidates {
err := s.db.GetContext(ctx, &meta, `SELECT IFNULL(mimetype,'') AS mimetype,IFNULL(size,0) AS size,IFNULL(etag,'') AS etag,IFNULL(location,'') AS location,IFNULL(original_name,'') AS original_name
FROM im_fileuploads WHERE tenant_id=? AND (location=? OR location LIKE ?) ORDER BY createdat DESC LIMIT 1`, tenantID, candidate, "%/"+candidate)
if err == nil {
return meta
}
}
return meta
}
func (s *CoreStore) populateUserSummaries(ctx context.Context, tenantID int64, ids []int64, result map[int64]UserSummary) error {
query, args, err := sqlx.In(`SELECT user_id,nickname AS display_name,IFNULL(avatar_file_id,'') AS avatar_file_id
FROM im_user_profiles WHERE tenant_id=? AND user_id IN (?) ORDER BY user_id ASC`, tenantID, ids)
if err != nil {
return err
}
var profileRows []struct {
UserID int64 `db:"user_id"`
DisplayName string `db:"display_name"`
AvatarFileID string `db:"avatar_file_id"`
}
if err := s.db.SelectContext(ctx, &profileRows, s.db.Rebind(query), args...); err != nil && !isMissingTable(err) {
return err
}
for _, row := range profileRows {
summary := result[row.UserID]
if summary.UserID == 0 {
summary.UserID = row.UserID
}
if summary.DisplayName == "" {
summary.DisplayName = strings.TrimSpace(row.DisplayName)
}
if summary.AvatarFileID == "" {
summary.AvatarFileID = row.AvatarFileID
}
result[row.UserID] = summary
}
query, args, err = sqlx.In(`SELECT id AS user_id,
COALESCE(
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.display_name')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.fn')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.nickname')),''),
NULLIF(JSON_UNQUOTE(JSON_EXTRACT(public,'$.name')),''),
''
) AS display_name
FROM im_users WHERE tenant_id=? AND id IN (?)`, tenantID, ids)
if err != nil {
return err
}
var userRows []struct {
UserID int64 `db:"user_id"`
DisplayName string `db:"display_name"`
}
if err := s.db.SelectContext(ctx, &userRows, s.db.Rebind(query), args...); err != nil && !isMissingTable(err) {
return err
}
for _, row := range userRows {
summary := result[row.UserID]
if summary.UserID == 0 {
summary.UserID = row.UserID
}
if summary.DisplayName == "" {
summary.DisplayName = strings.TrimSpace(row.DisplayName)
}
result[row.UserID] = summary
}
return nil
}
-23
View File
@@ -1,23 +0,0 @@
package store
import "testing"
func TestCoreStoreOwnsDBOnlyForDifferentDSN(t *testing.T) {
tests := []struct {
name string
primaryDSN string
coreDSN string
wantOwnDB bool
}{
{name: "same", primaryDSN: "root@tcp(db:3306)/meeting_im", coreDSN: "root@tcp(db:3306)/meeting_im", wantOwnDB: false},
{name: "same with whitespace", primaryDSN: " root@tcp(db:3306)/meeting_im ", coreDSN: "root@tcp(db:3306)/meeting_im", wantOwnDB: false},
{name: "different", primaryDSN: "root@tcp(db:3306)/workbench", coreDSN: "root@tcp(db:3306)/meeting_im", wantOwnDB: true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := coreStoreOwnsDB(tt.primaryDSN, tt.coreDSN); got != tt.wantOwnDB {
t.Fatalf("coreStoreOwnsDB() = %t, want %t", got, tt.wantOwnDB)
}
})
}
}
+28 -4
View File
@@ -3,8 +3,32 @@ package store
import "errors"
var (
ErrNotFound = errors.New("resource not found")
ErrForbidden = errors.New("forbidden")
ErrConflict = errors.New("conflict")
ErrPolicy = errors.New("policy violation")
ErrAccountExists = errors.New("account already exists")
ErrAccountNotFound = errors.New("account not found")
ErrAccountRestricted = errors.New("account restricted")
ErrAttachmentInvalid = errors.New("attachment invalid")
ErrConcurrentUpdate = errors.New("concurrent update")
ErrInvalidCredentials = errors.New("invalid credentials")
ErrReferralInvalid = errors.New("referral code invalid")
ErrResourceNotFound = errors.New("resource not found")
ErrTaskHallNotFound = errors.New("task hall not found")
ErrTaskDisabled = errors.New("task feature disabled")
ErrOutsideBusinessHours = errors.New("outside business hours")
ErrInsufficientBalance = errors.New("insufficient balance")
ErrDailyLimitReached = errors.New("daily limit reached")
ErrLevelNotEligible = errors.New("member level not eligible")
ErrPendingOrderExists = errors.New("pending order exists")
ErrPaymentAccountExists = errors.New("payment account already exists")
ErrPaymentAccountInvalid = errors.New("payment account invalid")
ErrWithdrawalDisabled = errors.New("withdrawal disabled")
ErrWithdrawalChannel = errors.New("withdrawal channel unavailable")
ErrWithdrawalAmount = errors.New("withdrawal amount invalid")
ErrWithdrawalLimit = errors.New("withdrawal daily limit reached")
ErrWithdrawalNotFound = errors.New("withdrawal order not found")
ErrIdempotencyConflict = errors.New("idempotency key reused with different request")
ErrNoTaskAvailable = errors.New("no task available")
ErrOrderNotFound = errors.New("task order not found")
ErrOrderState = errors.New("task order state invalid")
ErrOrderExpired = errors.New("task order expired")
ErrEvidenceInvalid = errors.New("task evidence invalid")
)
+78
View File
@@ -0,0 +1,78 @@
package store
import (
"context"
"database/sql"
"errors"
)
type ManualProduct struct {
ID int64 `db:"id"`
ProductNo string `db:"product_no"`
CategoryID int64 `db:"category_id"`
CategoryCode string `db:"category_code"`
CategoryName string `db:"category_name"`
PaymentType string `db:"payment_type"`
Name string `db:"name"`
ShopName string `db:"shop_name"`
PriceAmount int64 `db:"price_amount"`
Currency string `db:"currency"`
CoverURL string `db:"cover_url"`
DetailHTML string `db:"detail_html"`
IsRecommended bool `db:"is_recommended"`
ClaimedCount int64 `db:"claimed_count"`
CreatedAt string `db:"created_at"`
UpdatedAt string `db:"updated_at"`
}
type ManualProductPage struct {
Items []ManualProduct
Total int64
}
func (s *Store) ManualProducts(ctx context.Context, tenantID, categoryID int64, offset, limit int) (*ManualProductPage, error) {
where := ` WHERE p.tenant_id=? AND p.product_type=2 AND p.publish_status=3
AND p.deleted_at IS NULL AND c.status=1 AND c.deleted_at IS NULL
AND (p.start_at IS NULL OR p.start_at<=UTC_TIMESTAMP(3))
AND (p.end_at IS NULL OR p.end_at>UTC_TIMESTAMP(3))`
args := []any{tenantID}
if categoryID > 0 {
where += ` AND p.category_id=?`
args = append(args, categoryID)
}
var total int64
if err := s.db.GetContext(ctx, &total, `SELECT COUNT(*) FROM hg_product p
JOIN hg_product_category c ON c.tenant_id=p.tenant_id AND c.id=p.category_id`+where, args...); err != nil {
return nil, err
}
args = append(args, limit, offset)
items := make([]ManualProduct, 0)
err := s.db.SelectContext(ctx, &items, manualProductSelectSQL()+where+
` ORDER BY p.is_recommended ASC,p.sort ASC,p.id ASC LIMIT ? OFFSET ?`, args...)
return &ManualProductPage{Items: items, Total: total}, err
}
func (s *Store) ManualProduct(ctx context.Context, tenantID, productID int64) (*ManualProduct, error) {
var row ManualProduct
err := s.db.GetContext(ctx, &row, manualProductSelectSQL()+` WHERE p.tenant_id=? AND p.id=?
AND p.product_type=2 AND p.publish_status=3 AND p.deleted_at IS NULL
AND c.status=1 AND c.deleted_at IS NULL
AND (p.start_at IS NULL OR p.start_at<=UTC_TIMESTAMP(3))
AND (p.end_at IS NULL OR p.end_at>UTC_TIMESTAMP(3)) LIMIT 1`, tenantID, productID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrResourceNotFound
}
return &row, err
}
func manualProductSelectSQL() string {
return `SELECT p.id,p.product_no,p.category_id,c.category_code,c.name AS category_name,
p.payment_type,p.name,p.shop_name,p.price_amount,p.currency,p.cover_url,
COALESCE(p.detail_html,'') AS detail_html,(p.is_recommended=1) AS is_recommended,
(SELECT COUNT(*) FROM hg_trade_order o WHERE o.tenant_id=p.tenant_id AND o.product_id=p.id
AND o.order_type=3 AND o.order_status NOT IN (7,8)) AS claimed_count,
DATE_FORMAT(p.created_at,'%Y-%m-%dT%H:%i:%sZ') AS created_at,
DATE_FORMAT(p.updated_at,'%Y-%m-%dT%H:%i:%sZ') AS updated_at
FROM hg_product p JOIN hg_product_category c
ON c.tenant_id=p.tenant_id AND c.id=p.category_id`
}
-94
View File
@@ -1,94 +0,0 @@
package store
import "testing"
func TestNormalizeMeetingNoteFileMetadataUsesUploadMetadata(t *testing.T) {
gotName, gotMime, gotSize, gotETag := normalizeMeetingNoteFileMetadata(MeetingNoteFileInput{
FileID: "lxzvtUzicEo",
}, fileUploadMetadata{
MIMEType: "application/pdf",
FileSize: 12345,
FileETag: "etag-1",
Location: "uploads/wontuse",
OriginalName: "会议纪要.pdf",
}, "lxzvtUzicEo")
if gotName != "会议纪要.pdf" {
t.Fatalf("name=%q, want original upload name", gotName)
}
if gotMime != "application/pdf" || gotSize != 12345 || gotETag != "etag-1" {
t.Fatalf("metadata mime=%q size=%d etag=%q", gotMime, gotSize, gotETag)
}
}
func TestNormalizeMeetingNoteFileMetadataFallsBackToLocation(t *testing.T) {
gotName, gotMime, gotSize, gotETag := normalizeMeetingNoteFileMetadata(MeetingNoteFileInput{
FileID: "lxzvtUzicEo",
}, fileUploadMetadata{
MIMEType: "application/pdf",
FileSize: 12345,
FileETag: "etag-1",
Location: "uploads/lxzvtUzicEo.pdf",
}, "lxzvtUzicEo")
if gotName != "lxzvtUzicEo.pdf" {
t.Fatalf("name=%q, want location file name", gotName)
}
if gotMime != "application/pdf" || gotSize != 12345 || gotETag != "etag-1" {
t.Fatalf("metadata mime=%q size=%d etag=%q", gotMime, gotSize, gotETag)
}
}
func TestNormalizeMeetingNoteFileMetadataKeepsRequestMetadata(t *testing.T) {
size := int64(88)
gotName, gotMime, gotSize, gotETag := normalizeMeetingNoteFileMetadata(MeetingNoteFileInput{
FileID: "file_1",
FileName: "../会议纪要.docx",
MIMEType: "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
FileSize: &size,
FileETag: "request-etag",
}, fileUploadMetadata{
MIMEType: "application/pdf",
FileSize: 12345,
FileETag: "upload-etag",
Location: "uploads/file_1.pdf",
}, "file_1")
if gotName != "会议纪要.docx" {
t.Fatalf("name=%q, want original file name", gotName)
}
if gotMime != "application/vnd.openxmlformats-officedocument.wordprocessingml.document" || gotSize != 88 || gotETag != "request-etag" {
t.Fatalf("metadata mime=%q size=%d etag=%q", gotMime, gotSize, gotETag)
}
}
func TestParseUploadFileIDAcceptsExtension(t *testing.T) {
decodedID, storageName := parseUploadFileID("lxzvtUzicEo.pdf")
if decodedID <= 0 {
t.Fatalf("decodedID=%d, want positive", decodedID)
}
if storageName == "" {
t.Fatal("storageName is empty")
}
}
func TestParseUploadFileIDAcceptsOSSURL(t *testing.T) {
decodedID, storageName := parseUploadFileID("https://oss.example.com/files/lxzvtUzicEo.pdf?Expires=1")
if decodedID <= 0 {
t.Fatalf("decodedID=%d, want positive", decodedID)
}
if storageName == "" {
t.Fatal("storageName is empty")
}
}
func TestParseUploadFileIDAcceptsObjectKey(t *testing.T) {
decodedID, storageName := parseUploadFileID("tenants/1/2026/09/s4oo7nkm4jyeu")
if decodedID != 0 {
t.Fatalf("decodedID=%d, want 0 for object key", decodedID)
}
if storageName != "s4oo7nkm4jyeu" {
t.Fatalf("storageName=%q, want s4oo7nkm4jyeu", storageName)
}
candidates := uploadLocationCandidates("tenants/1/2026/09/s4oo7nkm4jyeu", storageName)
if len(candidates) < 2 || candidates[0] != "s4oo7nkm4jyeu" || candidates[1] != "tenants/1/2026/09/s4oo7nkm4jyeu" {
t.Fatalf("candidates=%#v", candidates)
}
}
+3 -1329
View File
File diff suppressed because it is too large Load Diff
+153
View File
@@ -0,0 +1,153 @@
package store
import (
"context"
"database/sql"
"encoding/json"
"errors"
)
const publishStatusPublished = 3
type Banner struct {
ID int64 `db:"id"`
ImageURL string `db:"image_url"`
AltText string `db:"alt_text"`
TargetType int `db:"target_type"`
TargetValue string `db:"target_value"`
Sort int `db:"sort"`
}
type Content struct {
ContentKey string `db:"content_key"`
ContentType string `db:"content_type"`
Title string `db:"title"`
Summary string `db:"summary"`
BodyHTML string `db:"body_html"`
BodyText string `db:"body_text"`
CoverURL string `db:"cover_url"`
Version int `db:"version"`
PublishedAt string `db:"published_at"`
AssetIDs sql.NullString `db:"asset_ids"`
}
type MemberLevel struct {
ID int64 `db:"id"`
Code string `db:"level_code"`
Name string `db:"name"`
IconURL string `db:"icon_url"`
UpgradeAmount int64 `db:"upgrade_amount"`
GiftAmount int64 `db:"gift_amount"`
DailyOrderLimit int `db:"daily_order_limit"`
CommissionRate int `db:"commission_rate"`
DailyWithdrawLimit int `db:"daily_withdraw_limit"`
RuleJSON json.RawMessage `db:"rule_json"`
Version int `db:"version"`
Sort int `db:"sort"`
}
type MemberNotice struct {
ID int64 `db:"id"`
Title string `db:"title"`
Content string `db:"content"`
SourceType int `db:"source_type"`
StartAt string `db:"start_at"`
EndAt string `db:"end_at"`
}
type TaskSummary struct {
TodayCount int `db:"today_count" json:"today_count"`
TodayCommissionAmount int64 `db:"today_commission_amount" json:"today_commission_amount"`
YesterdayCommissionAmount int64 `db:"yesterday_commission_amount" json:"yesterday_commission_amount"`
}
func (s *Store) CurrentSiteConfig(ctx context.Context, tenantID int64) (json.RawMessage, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
var raw []byte
err := s.db.GetContext(ctx, &raw, `SELECT config_json FROM hg_site_config_version
WHERE tenant_id=? AND publish_status=? AND (effective_at IS NULL OR effective_at<=UTC_TIMESTAMP(3))
ORDER BY COALESCE(effective_at,published_at,created_at) DESC,id DESC LIMIT 1`, tenantID, publishStatusPublished)
if errors.Is(err, sql.ErrNoRows) {
return json.RawMessage(`{}`), nil
}
return json.RawMessage(raw), err
}
func (s *Store) ActiveBanners(ctx context.Context, tenantID int64, position string) ([]Banner, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
rows := make([]Banner, 0)
err := s.db.SelectContext(ctx, &rows, `SELECT id,image_url,alt_text,target_type,COALESCE(target_value,'') AS target_value,sort
FROM hg_banner WHERE tenant_id=? AND position=? AND publish_status=?
AND (start_at IS NULL OR start_at<=UTC_TIMESTAMP(3)) AND (end_at IS NULL OR end_at>UTC_TIMESTAMP(3))
AND deleted_at IS NULL ORDER BY sort ASC,id ASC`, tenantID, position, publishStatusPublished)
return rows, err
}
func (s *Store) PublishedContent(ctx context.Context, tenantID int64, key string) (*Content, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
var row Content
err := s.db.GetContext(ctx, &row, `SELECT c.content_key,c.content_type,v.title,COALESCE(v.summary,'') AS summary,
v.body_html,COALESCE(v.body_text,'') AS body_text,COALESCE(v.cover_url,'') AS cover_url,
v.version_no AS version,COALESCE(DATE_FORMAT(c.published_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS published_at,v.asset_ids
FROM hg_content c JOIN hg_content_version v
ON v.tenant_id=c.tenant_id AND v.id=c.published_version_id
WHERE c.tenant_id=? AND c.content_key=? AND c.publish_status=? AND v.publish_status=?
AND (v.effective_at IS NULL OR v.effective_at<=UTC_TIMESTAMP(3))
AND c.deleted_at IS NULL LIMIT 1`, tenantID, key, publishStatusPublished, publishStatusPublished)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrResourceNotFound
}
return &row, err
}
func (s *Store) ActiveMemberLevels(ctx context.Context, tenantID int64) ([]MemberLevel, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
rows := make([]MemberLevel, 0)
err := s.db.SelectContext(ctx, &rows, `SELECT l.id,l.level_code,l.name,l.icon_url,l.sort,
v.upgrade_amount,v.gift_amount,v.daily_order_limit,v.commission_rate,v.daily_withdraw_limit,
COALESCE(v.rule_json,JSON_OBJECT()) AS rule_json,v.version_no AS version
FROM hg_app_member_level l JOIN hg_app_member_level_version v
ON v.tenant_id=l.tenant_id AND v.id=l.current_version_id
WHERE l.tenant_id=? AND l.status=1 AND v.publish_status=?
AND (v.effective_at IS NULL OR v.effective_at<=UTC_TIMESTAMP(3))
AND l.deleted_at IS NULL ORDER BY l.sort ASC,l.id ASC`, tenantID, publishStatusPublished)
return rows, err
}
func (s *Store) ActiveMemberNotices(ctx context.Context, tenantID, memberID int64, limit int) ([]MemberNotice, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
rows := make([]MemberNotice, 0)
err := s.db.SelectContext(ctx, &rows, `SELECT id,title,content,source_type,
DATE_FORMAT(start_at,'%Y-%m-%dT%H:%i:%sZ') AS start_at,
DATE_FORMAT(end_at,'%Y-%m-%dT%H:%i:%sZ') AS end_at
FROM hg_app_member_notice WHERE tenant_id=? AND member_id IN (0,?) AND notice_status=1
AND start_at<=UTC_TIMESTAMP(3) AND end_at>UTC_TIMESTAMP(3)
ORDER BY start_at DESC,id DESC LIMIT ?`, tenantID, memberID, limit)
return rows, err
}
func (s *Store) MemberTaskSummary(ctx context.Context, tenantID, memberID int64) (*TaskSummary, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
var row TaskSummary
err := s.db.GetContext(ctx, &row, `SELECT
(SELECT COUNT(*) FROM hg_trade_order o WHERE o.tenant_id=? AND o.member_id=?
AND o.created_at>=UTC_DATE() AND o.created_at<UTC_DATE()+INTERVAL 1 DAY) AS today_count,
(SELECT COALESCE(SUM(t.commission_amount),0) FROM hg_task_submission t WHERE t.tenant_id=? AND t.member_id=?
AND t.submission_status=4 AND t.settled_at>=UTC_DATE() AND t.settled_at<UTC_DATE()+INTERVAL 1 DAY) AS today_commission_amount,
(SELECT COALESCE(SUM(y.commission_amount),0) FROM hg_task_submission y WHERE y.tenant_id=? AND y.member_id=?
AND y.submission_status=4 AND y.settled_at>=UTC_DATE()-INTERVAL 1 DAY AND y.settled_at<UTC_DATE()) AS yesterday_commission_amount`,
tenantID, memberID, tenantID, memberID, tenantID, memberID)
return &row, err
}
+161
View File
@@ -0,0 +1,161 @@
package store
import (
"context"
"database/sql"
"errors"
)
type MemberOverview struct {
TenantID int64 `db:"tenant_id"`
MemberID int64 `db:"member_id"`
MemberNo string `db:"member_no"`
Username string `db:"username"`
MobileMask string `db:"mobile_mask"`
InviteCode string `db:"invite_code"`
MemberStatus int `db:"member_status"`
MemberCategory int `db:"member_category"`
CreditScore int `db:"credit_score"`
AvatarAttachmentID int64 `db:"avatar_attachment_id"`
AvatarURL string `db:"avatar_url"`
LevelID int64 `db:"level_id"`
LevelCode string `db:"level_code"`
LevelName string `db:"level_name"`
LevelIconURL string `db:"level_icon_url"`
CommissionRate int `db:"commission_rate"`
AvailableAmount int64 `db:"available_amount"`
FrozenAmount int64 `db:"frozen_amount"`
TotalIncome int64 `db:"total_income"`
TotalExpense int64 `db:"total_expense"`
Currency string `db:"currency"`
WalletStatus int `db:"wallet_status"`
WalletVersion int `db:"wallet_version"`
}
type AccountSecurity struct {
PasswordHash string `db:"password_hash"`
TradePasswordHash string `db:"trade_password_hash"`
MemberStatus int `db:"member_status"`
}
type Avatar struct {
AttachmentID int64 `db:"attachment_id"`
URL string `db:"url"`
}
func (s *Store) MemberOverview(ctx context.Context, tenantID, memberID int64) (*MemberOverview, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
var row MemberOverview
err := s.db.GetContext(ctx, &row, `SELECT m.tenant_id,m.id AS member_id,m.member_no,m.username,m.mobile_mask,
m.invite_code,m.member_status,m.risk_status AS member_category,m.credit_score,
COALESCE(p.avatar_attachment_id,0) AS avatar_attachment_id,COALESCE(a.file_url,'') AS avatar_url,
COALESCE(l.id,0) AS level_id,COALESCE(l.level_code,'') AS level_code,
COALESCE(l.name,'') AS level_name,COALESCE(l.icon_url,'') AS level_icon_url,
COALESCE(lv.commission_rate,0) AS commission_rate,
COALESCE(w.available_amount,0) AS available_amount,COALESCE(w.frozen_amount,0) AS frozen_amount,
COALESCE(w.total_income,0) AS total_income,COALESCE(w.total_expense,0) AS total_expense,
COALESCE(w.currency,'CNY') AS currency,COALESCE(w.account_status,1) AS wallet_status,
COALESCE(w.version,0) AS wallet_version
FROM hg_app_member m
LEFT JOIN hg_app_member_profile p ON p.tenant_id=m.tenant_id AND p.member_id=m.id
LEFT JOIN hg_sys_attachment a ON a.id=p.avatar_attachment_id AND a.status=1
LEFT JOIN hg_app_member_level l ON l.tenant_id=m.tenant_id AND l.id=m.level_id AND l.deleted_at IS NULL
LEFT JOIN hg_app_member_level_version lv ON lv.tenant_id=l.tenant_id AND lv.id=l.current_version_id
LEFT JOIN hg_wallet_account w ON w.tenant_id=m.tenant_id AND w.member_id=m.id AND w.currency='CNY'
WHERE m.tenant_id=? AND m.id=? AND m.member_status=1 AND m.deleted_at IS NULL LIMIT 1`, tenantID, memberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountNotFound
}
return &row, err
}
func (s *Store) UpdateAvatar(ctx context.Context, tenantID, memberID, attachmentID int64) (*Avatar, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
var activeMemberID int64
err = tx.GetContext(ctx, &activeMemberID, `SELECT id FROM hg_app_member
WHERE tenant_id=? AND id=? AND member_status=1 AND deleted_at IS NULL LIMIT 1 FOR UPDATE`, tenantID, memberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountNotFound
}
if err != nil {
return nil, err
}
var avatar Avatar
err = tx.GetContext(ctx, &avatar, `SELECT id AS attachment_id,COALESCE(file_url,'') AS url FROM hg_sys_attachment
WHERE id=? AND app_id='workbench' AND member_id=? AND status=1
AND mime_type LIKE 'image/%' LIMIT 1 FOR UPDATE`, attachmentID, memberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAttachmentInvalid
}
if err != nil {
return nil, err
}
_, err = tx.ExecContext(ctx, `INSERT INTO hg_app_member_profile
(tenant_id,member_id,avatar_attachment_id,created_by,updated_by)
VALUES(?,?,?,0,0) ON DUPLICATE KEY UPDATE avatar_attachment_id=VALUES(avatar_attachment_id),updated_at=CURRENT_TIMESTAMP(3)`,
tenantID, memberID, attachmentID)
if err != nil {
return nil, err
}
if err := tx.Commit(); err != nil {
return nil, err
}
return &avatar, nil
}
func (s *Store) AccountSecurity(ctx context.Context, tenantID, memberID int64) (*AccountSecurity, error) {
if s == nil || s.db == nil {
return nil, errors.New("store is unavailable")
}
var row AccountSecurity
err := s.db.GetContext(ctx, &row, `SELECT password_hash,trade_password_hash,member_status
FROM hg_app_member WHERE tenant_id=? AND id=? AND member_status=1 AND deleted_at IS NULL LIMIT 1`, tenantID, memberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountNotFound
}
return &row, err
}
func (s *Store) UpdateLoginPassword(ctx context.Context, tenantID, memberID int64, oldHash, newHash string) error {
result, err := s.db.ExecContext(ctx, `UPDATE hg_app_member SET password_hash=?,updated_at=CURRENT_TIMESTAMP(3)
WHERE tenant_id=? AND id=? AND password_hash=? AND member_status=1 AND deleted_at IS NULL`,
newHash, tenantID, memberID, oldHash)
if err != nil {
return err
}
affected, err := result.RowsAffected()
if err != nil {
return err
}
if affected != 1 {
return ErrConcurrentUpdate
}
return nil
}
func (s *Store) UpdateTradePassword(ctx context.Context, tenantID, memberID int64, oldHash, newHash string) error {
result, err := s.db.ExecContext(ctx, `UPDATE hg_app_member SET trade_password_hash=?,updated_at=CURRENT_TIMESTAMP(3)
WHERE tenant_id=? AND id=? AND trade_password_hash=? AND member_status=1 AND deleted_at IS NULL`,
newHash, tenantID, memberID, oldHash)
if err != nil {
return err
}
affected, err := result.RowsAffected()
if err != nil {
return err
}
if affected != 1 {
return ErrConcurrentUpdate
}
return nil
}
+480
View File
@@ -0,0 +1,480 @@
package store
import (
"context"
"database/sql"
"encoding/json"
"errors"
"time"
"github.com/jmoiron/sqlx"
)
const (
orderStatusMatching = 1
orderStatusPending = 2
orderStatusProcessing = 3
orderStatusCompleted = 4
orderStatusExpired = 5
orderStatusFrozen = 6
orderStatusCancelled = 7
orderStatusFailed = 8
)
type TaskHall struct {
ID int64 `db:"id"`
Code string `db:"hall_code"`
Name string `db:"name"`
IconURL string `db:"icon_url"`
Summary string `db:"summary"`
CommissionRate int `db:"commission_rate"`
RateUnit string `db:"rate_unit"`
ClientRoute string `db:"client_route"`
}
type TaskEligibility struct {
MemberLevelID int64 `db:"member_level_id"`
MemberStatus int `db:"member_status"`
HallEligible bool `db:"hall_eligible"`
AvailableAmount int64 `db:"available_amount"`
WalletStatus int `db:"wallet_status"`
DailyLimit int `db:"daily_limit"`
TodayCount int `db:"today_count"`
PendingOrderID int64 `db:"pending_order_id"`
}
type ClaimTaskInput struct {
TenantID int64
MemberID int64
HallID int64
MinimumAmount int64
DeadlineAt time.Time
OrderNo string
}
type TaskOrder struct {
ID int64 `db:"id"`
OrderNo string `db:"order_no"`
TaskID int64 `db:"task_id"`
ProductID int64 `db:"product_id"`
ProductName string `db:"product_name"`
CoverURL string `db:"cover_url"`
Quantity int `db:"quantity"`
UnitPriceAmount int64 `db:"unit_price_amount"`
TotalAmount int64 `db:"total_amount"`
CommissionAmount int64 `db:"commission_amount"`
Currency string `db:"currency"`
MemberLevelID int64 `db:"member_level_id"`
RuleSnapshot json.RawMessage `db:"rule_snapshot"`
Message string `db:"message"`
OrderStatus int `db:"order_status"`
FreezeStatus int `db:"freeze_status"`
DeadlineAt string `db:"deadline_at"`
MatchedAt string `db:"matched_at"`
CompletedAt string `db:"completed_at"`
CreatedAt string `db:"created_at"`
UpdatedAt string `db:"updated_at"`
TaskTitle string `db:"task_title"`
TaskDescription string `db:"task_description"`
TaskDetailHTML string `db:"task_detail_html"`
SubmissionNo string `db:"submission_no"`
SubmissionStatus int `db:"submission_status"`
EvidenceText string `db:"evidence_text"`
EvidenceIDs json.RawMessage `db:"evidence_attachment_ids"`
ReviewReasonCode string `db:"review_reason_code"`
ReviewReason string `db:"review_reason"`
AllowResubmit int `db:"allow_resubmit"`
SubmittedAt string `db:"submitted_at"`
ReviewedAt string `db:"reviewed_at"`
SettledAt string `db:"settled_at"`
}
type TaskOrderPage struct {
Items []TaskOrder
Total int64
}
type TaskStats struct {
AvailableAmount int64 `db:"available_amount"`
FrozenAmount int64 `db:"frozen_amount"`
TodayCommissionAmount int64 `db:"today_commission_amount"`
YesterdayCommissionAmount int64 `db:"yesterday_commission_amount"`
TodayClaimCount int `db:"today_claim_count"`
TodayCompletedCount int `db:"today_completed_count"`
}
type TaskSubmissionInput struct {
TenantID int64
MemberID int64
OrderID int64
SubmissionNo string
EvidenceText string
AttachmentIDs []int64
}
type TaskSubmissionResult struct {
SubmissionNo string `db:"submission_no"`
OrderStatus int
SubmittedAt time.Time
}
func (s *Store) TaskHalls(ctx context.Context, tenantID, memberID int64) ([]TaskHall, error) {
rows := make([]TaskHall, 0)
err := s.db.SelectContext(ctx, &rows, `SELECT h.id,h.hall_code,h.name,h.icon_url,h.summary,
h.commission_rate,h.rate_unit,h.client_route FROM hg_task_hall h
JOIN hg_app_member m ON m.tenant_id=h.tenant_id AND m.id=? AND m.member_status=1 AND m.deleted_at IS NULL
JOIN hg_task_hall_level hl ON hl.tenant_id=h.tenant_id AND hl.hall_id=h.id AND hl.level_id=m.level_id
WHERE h.tenant_id=? AND h.publish_status=3 AND h.deleted_at IS NULL ORDER BY h.sort,h.id`, memberID, tenantID)
return rows, err
}
func (s *Store) TaskEligibility(ctx context.Context, tenantID, memberID, hallID int64) (*TaskEligibility, error) {
var row TaskEligibility
err := s.db.GetContext(ctx, &row, `SELECT m.level_id AS member_level_id,m.member_status,
EXISTS(SELECT 1 FROM hg_task_hall h JOIN hg_task_hall_level hl
ON hl.tenant_id=h.tenant_id AND hl.hall_id=h.id AND hl.level_id=m.level_id
WHERE h.tenant_id=m.tenant_id AND h.id=? AND h.publish_status=3 AND h.deleted_at IS NULL) AS hall_eligible,
COALESCE(w.available_amount,0) AS available_amount,COALESCE(w.account_status,0) AS wallet_status,
COALESCE(lv.daily_order_limit,0) AS daily_limit,
(SELECT COUNT(*) FROM hg_trade_order o WHERE o.tenant_id=m.tenant_id AND o.member_id=m.id
AND o.order_type=2 AND o.created_at>=UTC_DATE() AND o.created_at<UTC_DATE()+INTERVAL 1 DAY) AS today_count,
COALESCE((SELECT o2.id FROM hg_trade_order o2 WHERE o2.tenant_id=m.tenant_id AND o2.member_id=m.id
AND o2.order_type=2 AND o2.order_status IN (1,2,3,6) ORDER BY o2.id DESC LIMIT 1),0) AS pending_order_id
FROM hg_app_member m
LEFT JOIN hg_app_member_level l ON l.tenant_id=m.tenant_id AND l.id=m.level_id AND l.status=1 AND l.deleted_at IS NULL
LEFT JOIN hg_app_member_level_version lv ON lv.tenant_id=l.tenant_id AND lv.id=l.current_version_id AND lv.publish_status=3
LEFT JOIN hg_wallet_account w ON w.tenant_id=m.tenant_id AND w.member_id=m.id AND w.currency='CNY'
WHERE m.tenant_id=? AND m.id=? AND m.deleted_at IS NULL LIMIT 1`, hallID, tenantID, memberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountNotFound
}
return &row, err
}
func (s *Store) ClaimTask(ctx context.Context, input ClaimTaskInput) (*TaskOrder, error) {
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
var member struct {
LevelID int64 `db:"level_id"`
}
err = tx.GetContext(ctx, &member, `SELECT level_id FROM hg_app_member
WHERE tenant_id=? AND id=? AND member_status=1 AND deleted_at IS NULL LIMIT 1 FOR UPDATE`, input.TenantID, input.MemberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountNotFound
}
if err != nil {
return nil, err
}
var hallCount int
err = tx.GetContext(ctx, &hallCount, `SELECT COUNT(*) FROM hg_task_hall h
JOIN hg_task_hall_level hl ON hl.tenant_id=h.tenant_id AND hl.hall_id=h.id AND hl.level_id=?
WHERE h.tenant_id=? AND h.id=? AND h.publish_status=3 AND h.deleted_at IS NULL`, member.LevelID, input.TenantID, input.HallID)
if err != nil {
return nil, err
}
if hallCount == 0 {
return nil, ErrLevelNotEligible
}
var wallet struct {
AvailableAmount int64 `db:"available_amount"`
Status int `db:"account_status"`
}
err = tx.GetContext(ctx, &wallet, `SELECT available_amount,account_status FROM hg_wallet_account
WHERE tenant_id=? AND member_id=? AND currency='CNY' LIMIT 1 FOR UPDATE`, input.TenantID, input.MemberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrInsufficientBalance
}
if err != nil {
return nil, err
}
if wallet.Status != 1 {
return nil, ErrAccountRestricted
}
if wallet.AvailableAmount < input.MinimumAmount {
return nil, ErrInsufficientBalance
}
var pending int64
err = tx.GetContext(ctx, &pending, `SELECT id FROM hg_trade_order
WHERE tenant_id=? AND member_id=? AND order_type=2 AND order_status IN (1,2,3,6)
ORDER BY id DESC LIMIT 1 FOR UPDATE`, input.TenantID, input.MemberID)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return nil, err
}
if pending > 0 {
return nil, ErrPendingOrderExists
}
var dailyLimit int
err = tx.GetContext(ctx, &dailyLimit, `SELECT COALESCE(v.daily_order_limit,0) FROM hg_app_member_level l
LEFT JOIN hg_app_member_level_version v ON v.tenant_id=l.tenant_id AND v.id=l.current_version_id AND v.publish_status=3
WHERE l.tenant_id=? AND l.id=? AND l.status=1 AND l.deleted_at IS NULL LIMIT 1`, input.TenantID, member.LevelID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrLevelNotEligible
}
if err != nil {
return nil, err
}
if dailyLimit > 0 {
var todayCount int
if err := tx.GetContext(ctx, &todayCount, `SELECT COUNT(*) FROM hg_trade_order WHERE tenant_id=? AND member_id=?
AND order_type=2 AND created_at>=UTC_DATE() AND created_at<UTC_DATE()+INTERVAL 1 DAY`, input.TenantID, input.MemberID); err != nil {
return nil, err
}
if todayCount >= dailyLimit {
return nil, ErrDailyLimitReached
}
}
var task struct {
ID int64 `db:"id"`
Title string `db:"title"`
CoverURL string `db:"cover_url"`
CommissionAmount int64 `db:"commission_amount"`
CommissionRate int `db:"commission_rate"`
PerMemberLimit int `db:"per_member_limit"`
}
err = tx.GetContext(ctx, &task, `SELECT t.id,t.title,t.cover_url,t.commission_amount,t.commission_rate,t.per_member_limit
FROM hg_task t JOIN hg_task_level tl ON tl.tenant_id=t.tenant_id AND tl.task_id=t.id AND tl.level_id=?
WHERE t.tenant_id=? AND t.hall_id=? AND t.publish_status=3 AND t.deleted_at IS NULL
AND (t.start_at IS NULL OR t.start_at<=UTC_TIMESTAMP(3)) AND (t.end_at IS NULL OR t.end_at>UTC_TIMESTAMP(3))
AND (t.daily_limit=0 OR (SELECT COUNT(*) FROM hg_trade_order od WHERE od.tenant_id=t.tenant_id AND od.task_id=t.id
AND od.order_type=2 AND od.created_at>=UTC_DATE() AND od.created_at<UTC_DATE()+INTERVAL 1 DAY)<t.daily_limit)
AND (t.per_member_limit=0 OR (SELECT COUNT(*) FROM hg_trade_order om WHERE om.tenant_id=t.tenant_id
AND om.task_id=t.id AND om.member_id=? AND om.order_type=2 AND om.created_at>=UTC_DATE() AND om.created_at<UTC_DATE()+INTERVAL 1 DAY)<t.per_member_limit)
ORDER BY t.sort,t.id LIMIT 1 FOR UPDATE`, member.LevelID, input.TenantID, input.HallID, input.MemberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrNoTaskAvailable
}
if err != nil {
return nil, err
}
rules, err := json.Marshal(map[string]any{
"commission_amount": task.CommissionAmount,
"commission_rate": task.CommissionRate,
"deadline_at": input.DeadlineAt.UTC().Format(time.RFC3339),
"member_level_id": member.LevelID,
})
if err != nil {
return nil, err
}
result, err := tx.ExecContext(ctx, `INSERT INTO hg_trade_order
(tenant_id,order_no,member_id,task_id,product_id,product_name,order_type,quantity,
unit_price_amount,total_amount,commission_amount,currency,member_level_id,rule_snapshot,
message,order_status,freeze_status,created_by,updated_by)
VALUES(?,?,?,?,0,?,2,1,0,0,?,'CNY',?,?,?,1,1,0,0)`,
input.TenantID, input.OrderNo, input.MemberID, task.ID, task.Title, task.CommissionAmount,
member.LevelID, rules, "等待系统匹配")
if err != nil {
if isDuplicate(err) {
return nil, ErrPendingOrderExists
}
return nil, err
}
orderID, err := result.LastInsertId()
if err != nil {
return nil, err
}
if err := tx.Commit(); err != nil {
return nil, err
}
return &TaskOrder{ID: orderID, OrderNo: input.OrderNo, TaskID: task.ID, ProductName: task.Title,
CoverURL: task.CoverURL, Quantity: 1, CommissionAmount: task.CommissionAmount, Currency: "CNY",
MemberLevelID: member.LevelID, RuleSnapshot: rules, Message: "等待系统匹配",
OrderStatus: orderStatusMatching, FreezeStatus: 1, DeadlineAt: input.DeadlineAt.UTC().Format(time.RFC3339),
CreatedAt: time.Now().UTC().Format(time.RFC3339)}, nil
}
func (s *Store) TaskOrders(ctx context.Context, tenantID, memberID int64, status, offset, limit int) (*TaskOrderPage, error) {
where := ` WHERE o.tenant_id=? AND o.member_id=? AND o.order_type=2`
args := []any{tenantID, memberID}
if status > 0 {
where += ` AND o.order_status=?`
args = append(args, status)
}
var total int64
if err := s.db.GetContext(ctx, &total, `SELECT COUNT(*) FROM hg_trade_order o`+where, args...); err != nil {
return nil, err
}
args = append(args, limit, offset)
rows := make([]TaskOrder, 0)
err := s.db.SelectContext(ctx, &rows, orderSelectSQL()+where+` ORDER BY o.created_at DESC,o.id DESC LIMIT ? OFFSET ?`, args...)
return &TaskOrderPage{Items: rows, Total: total}, err
}
func (s *Store) TaskOrder(ctx context.Context, tenantID, memberID, orderID int64) (*TaskOrder, error) {
var row TaskOrder
err := s.db.GetContext(ctx, &row, orderSelectSQL()+` WHERE o.tenant_id=? AND o.member_id=? AND o.order_type=2 AND o.id=? LIMIT 1`, tenantID, memberID, orderID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrOrderNotFound
}
return &row, err
}
func orderSelectSQL() string {
return `SELECT o.id,o.order_no,o.task_id,o.product_id,COALESCE(NULLIF(o.product_name,''),t.title,'') AS product_name,
COALESCE(NULLIF(p.cover_url,''),t.cover_url,'') AS cover_url,o.quantity,o.unit_price_amount,
o.total_amount,o.commission_amount,o.currency,o.member_level_id,
COALESCE(o.rule_snapshot,JSON_OBJECT()) AS rule_snapshot,o.message,o.order_status,o.freeze_status,
COALESCE(JSON_UNQUOTE(JSON_EXTRACT(o.rule_snapshot,'$.deadline_at')),'') AS deadline_at,
COALESCE(DATE_FORMAT(o.matched_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS matched_at,
COALESCE(DATE_FORMAT(o.completed_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS completed_at,
DATE_FORMAT(o.created_at,'%Y-%m-%dT%H:%i:%sZ') AS created_at,
DATE_FORMAT(o.updated_at,'%Y-%m-%dT%H:%i:%sZ') AS updated_at,
COALESCE(t.title,'') AS task_title,COALESCE(t.description,'') AS task_description,
COALESCE(t.detail_html,'') AS task_detail_html,COALESCE(s.submission_no,'') AS submission_no,
COALESCE(s.submission_status,0) AS submission_status,COALESCE(s.evidence_text,'') AS evidence_text,
COALESCE(s.evidence_attachment_ids,JSON_ARRAY()) AS evidence_attachment_ids,
COALESCE(s.review_reason_code,'') AS review_reason_code,COALESCE(s.review_reason,'') AS review_reason,
COALESCE(s.allow_resubmit,2) AS allow_resubmit,
COALESCE(DATE_FORMAT(s.created_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS submitted_at,
COALESCE(DATE_FORMAT(s.reviewed_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS reviewed_at,
COALESCE(DATE_FORMAT(s.settled_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS settled_at
FROM hg_trade_order o LEFT JOIN hg_task t ON t.tenant_id=o.tenant_id AND t.id=o.task_id
LEFT JOIN hg_product p ON p.tenant_id=o.tenant_id AND p.id=o.product_id
LEFT JOIN hg_task_submission s ON s.tenant_id=o.tenant_id AND s.trade_order_id=o.id`
}
func (s *Store) TaskStats(ctx context.Context, tenantID, memberID int64) (*TaskStats, error) {
var row TaskStats
err := s.db.GetContext(ctx, &row, `SELECT
COALESCE(w.available_amount,0) AS available_amount,COALESCE(w.frozen_amount,0) AS frozen_amount,
(SELECT COALESCE(SUM(ts.commission_amount),0) FROM hg_task_submission ts WHERE ts.tenant_id=? AND ts.member_id=?
AND ts.submission_status=4 AND ts.settled_at>=UTC_DATE() AND ts.settled_at<UTC_DATE()+INTERVAL 1 DAY) AS today_commission_amount,
(SELECT COALESCE(SUM(ys.commission_amount),0) FROM hg_task_submission ys WHERE ys.tenant_id=? AND ys.member_id=?
AND ys.submission_status=4 AND ys.settled_at>=UTC_DATE()-INTERVAL 1 DAY AND ys.settled_at<UTC_DATE()) AS yesterday_commission_amount,
(SELECT COUNT(*) FROM hg_trade_order tc WHERE tc.tenant_id=? AND tc.member_id=?
AND tc.order_type=2 AND tc.created_at>=UTC_DATE() AND tc.created_at<UTC_DATE()+INTERVAL 1 DAY) AS today_claim_count,
(SELECT COUNT(*) FROM hg_trade_order cc WHERE cc.tenant_id=? AND cc.member_id=? AND cc.order_type=2 AND cc.order_status=4
AND cc.completed_at>=UTC_DATE() AND cc.completed_at<UTC_DATE()+INTERVAL 1 DAY) AS today_completed_count
FROM hg_app_member m LEFT JOIN hg_wallet_account w ON w.tenant_id=m.tenant_id AND w.member_id=m.id AND w.currency='CNY'
WHERE m.tenant_id=? AND m.id=? AND m.member_status=1 AND m.deleted_at IS NULL LIMIT 1`,
tenantID, memberID, tenantID, memberID, tenantID, memberID, tenantID, memberID, tenantID, memberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountNotFound
}
return &row, err
}
func (s *Store) SubmitTask(ctx context.Context, input TaskSubmissionInput) (*TaskSubmissionResult, error) {
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
var order struct {
TaskID int64 `db:"task_id"`
MemberLevelID int64 `db:"member_level_id"`
CommissionAmount int64 `db:"commission_amount"`
OrderStatus int `db:"order_status"`
DeadlineAt string `db:"deadline_at"`
}
err = tx.GetContext(ctx, &order, `SELECT task_id,member_level_id,commission_amount,order_status,
COALESCE(JSON_UNQUOTE(JSON_EXTRACT(rule_snapshot,'$.deadline_at')),'') AS deadline_at
FROM hg_trade_order WHERE tenant_id=? AND member_id=? AND order_type=2 AND id=? LIMIT 1 FOR UPDATE`,
input.TenantID, input.MemberID, input.OrderID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrOrderNotFound
}
if err != nil {
return nil, err
}
if order.OrderStatus != orderStatusPending {
return nil, ErrOrderState
}
if order.DeadlineAt != "" {
if deadline, parseErr := time.Parse(time.RFC3339, order.DeadlineAt); parseErr != nil || !time.Now().UTC().Before(deadline) {
if _, updateErr := tx.ExecContext(ctx, `UPDATE hg_trade_order SET order_status=5,message='任务已超时',updated_at=UTC_TIMESTAMP(3)
WHERE tenant_id=? AND id=? AND order_status=2`, input.TenantID, input.OrderID); updateErr != nil {
return nil, updateErr
}
if commitErr := tx.Commit(); commitErr != nil {
return nil, commitErr
}
return nil, ErrOrderExpired
}
}
if err := validateEvidenceAttachments(ctx, tx, input.TenantID, input.MemberID, input.AttachmentIDs); err != nil {
return nil, err
}
attachments, err := json.Marshal(input.AttachmentIDs)
if err != nil {
return nil, err
}
now := time.Now().UTC()
_, err = tx.ExecContext(ctx, `INSERT INTO hg_task_submission
(tenant_id,submission_no,trade_order_id,task_id,member_id,member_level_id,evidence_text,
evidence_attachment_ids,submission_status,commission_amount,success_rate,allow_resubmit,created_by,updated_by)
VALUES(?,?,?,?,?,?,?,?,1,?,0,2,0,0)`, input.TenantID, input.SubmissionNo, input.OrderID,
order.TaskID, input.MemberID, order.MemberLevelID, input.EvidenceText, attachments, order.CommissionAmount)
if err != nil {
if isDuplicate(err) {
return nil, ErrOrderState
}
return nil, err
}
result, err := tx.ExecContext(ctx, `UPDATE hg_trade_order SET order_status=3,message='已提交,等待审核',updated_at=UTC_TIMESTAMP(3)
WHERE tenant_id=? AND id=? AND member_id=? AND order_status=2`, input.TenantID, input.OrderID, input.MemberID)
if err != nil {
return nil, err
}
if affected, _ := result.RowsAffected(); affected != 1 {
return nil, ErrOrderState
}
if err := tx.Commit(); err != nil {
return nil, err
}
return &TaskSubmissionResult{SubmissionNo: input.SubmissionNo, OrderStatus: orderStatusProcessing, SubmittedAt: now}, nil
}
func validateEvidenceAttachments(ctx context.Context, tx *sqlx.Tx, tenantID, memberID int64, ids []int64) error {
if len(ids) == 0 {
return nil
}
query, args, err := sqlx.In(`SELECT COUNT(*) FROM hg_sys_attachment WHERE id IN (?)
AND tenant_id=? AND app_id='workbench' AND member_id=? AND status=1
AND (mime_type LIKE 'image/%' OR mime_type='application/pdf')`, ids, tenantID, memberID)
if err != nil {
return err
}
var count int
if err := tx.GetContext(ctx, &count, query, args...); err != nil {
return err
}
if count != len(ids) {
return ErrEvidenceInvalid
}
return nil
}
func (s *Store) OrderLedgers(ctx context.Context, tenantID, memberID int64, businessNumbers []string) ([]map[string]any, error) {
if len(businessNumbers) == 0 {
return []map[string]any{}, nil
}
query, args, err := sqlx.In(`SELECT id,ledger_no,biz_type,biz_no,entry_type,direction,amount,before_amount,after_amount,currency,remark,
DATE_FORMAT(occurred_at,'%Y-%m-%dT%H:%i:%sZ') AS occurred_at FROM hg_wallet_ledger
WHERE tenant_id=? AND member_id=? AND biz_no IN (?) ORDER BY occurred_at,id`, tenantID, memberID, businessNumbers)
if err != nil {
return nil, err
}
rows, err := s.db.QueryxContext(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
items := make([]map[string]any, 0)
for rows.Next() {
item := map[string]any{}
if err := rows.MapScan(item); err != nil {
return nil, err
}
for key, value := range item {
if bytes, ok := value.([]byte); ok {
item[key] = string(bytes)
}
}
items = append(items, item)
}
return items, rows.Err()
}
-115
View File
@@ -1,115 +0,0 @@
package store
import (
"context"
"time"
)
// TenantInitializationResult reports how many missing defaults were created.
// Existing tenant customizations are deliberately left untouched.
type TenantInitializationResult struct {
WorkbenchApps int64
AttendanceRules int64
AttendanceShifts int64
MailboxConfigs int64
FeedbackConfigs int64
}
// InitializeTenantDefaults creates the minimum workbench data required by a tenant.
// Tenant-scoped unique indexes make this operation safe to retry.
func (s *Store) InitializeTenantDefaults(ctx context.Context) (*TenantInitializationResult, error) {
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback() }()
now := time.Now().UTC().Round(time.Millisecond)
tenantID := s.tenant()
result := &TenantInitializationResult{}
apps := []struct {
code, name, route string
sort int
}{
{code: "attendance", name: "打卡", route: "/attendance", sort: 10},
{code: "monthly_reports", name: "月报", route: "/reports/monthly", sort: 20},
{code: "daily_reports", name: "日报", route: "/reports/daily", sort: 30},
{code: "organization", name: "部门框架", route: "/organization", sort: 40},
{code: "executive_mailbox", name: "举报信箱", route: "/executive-mailbox", sort: 50},
{code: "meetings", name: "会议通知", route: "/meetings", sort: 60},
{code: "meeting_notes", name: "会议笔记", route: "/meeting-notes", sort: 70},
}
for _, app := range apps {
res, execErr := tx.ExecContext(ctx, `INSERT INTO im_workbench_apps
(tenant_id,code,name,icon_file_id,route,sort_order,unread_key,enabled,permission,created_at,updated_at)
SELECT ?,?,?,?,?,?,?,?,?,?,?
WHERE NOT EXISTS (
SELECT 1 FROM im_workbench_apps WHERE tenant_id=? AND code=?
)`,
tenantID, app.code, app.name, "", app.route, app.sort, "", 1, "enabled", now, now,
tenantID, app.code)
if execErr != nil {
return nil, execErr
}
rows, _ := res.RowsAffected()
result.WorkbenchApps += rows
}
res, err := tx.ExecContext(ctx, `INSERT INTO im_attendance_rules
(tenant_id,rule_id,name,workday,location_required,wifi_required,allow_outside_checkin,appeal_allowed,effective_from,effective_to,state,created_at,updated_at)
SELECT ?,?,?,?,?,?,?,?,CURDATE(),NULL,'active',?,?
WHERE NOT EXISTS (
SELECT 1 FROM im_attendance_rules WHERE tenant_id=? AND rule_id=?
)`,
tenantID, "att_rule_default", "默认考勤规则", "1,2,3,4,5", 0, 0, 1, 1, now, now,
tenantID, "att_rule_default")
if err != nil {
return nil, err
}
result.AttendanceRules, _ = res.RowsAffected()
res, err = tx.ExecContext(ctx, `INSERT INTO im_attendance_shifts
(tenant_id,shift_id,rule_id,name,work_start,work_end,checkin_start,checkout_end,state,sort_order,created_at,updated_at)
SELECT ?,?,?,?,?,?,?,?,?,?,?,?
WHERE NOT EXISTS (
SELECT 1 FROM im_attendance_shifts WHERE tenant_id=? AND shift_id=?
)`,
tenantID, "shift_default", "att_rule_default", "默认班次", "09:00:00", "18:00:00", "08:00:00", "19:00:00", "active", 10, now, now,
tenantID, "shift_default")
if err != nil {
return nil, err
}
result.AttendanceShifts, _ = res.RowsAffected()
res, err = tx.ExecContext(ctx, `INSERT INTO im_mailbox_config
(tenant_id,enabled,allow_anonymous,required_fields,max_length,attachment_policy,notice_title,notice_content,notice_updated_at,created_at,updated_at)
SELECT ?,1,1,?,5000,?,'总经理信箱','请如实填写诉求。',?,?,?
WHERE NOT EXISTS (
SELECT 1 FROM im_mailbox_config WHERE tenant_id=?
)`,
tenantID, `["subject","content"]`, `{"max_files":5,"purpose":"executive_mailbox"}`, now, now, now,
tenantID)
if err != nil {
return nil, err
}
result.MailboxConfigs, _ = res.RowsAffected()
res, err = tx.ExecContext(ctx, `INSERT INTO im_feedback_config
(tenant_id,categories,max_length,attachment_policy,allow_contact,diagnostics_notice,created_at,updated_at)
SELECT ?,?,2000,?,1,'',?,?
WHERE NOT EXISTS (
SELECT 1 FROM im_feedback_config WHERE tenant_id=?
)`,
tenantID, `["bug","suggestion","complaint","other"]`, `{"max_files":5,"purpose":"feedback"}`, now, now,
tenantID)
if err != nil {
return nil, err
}
result.FeedbackConfigs, _ = res.RowsAffected()
if err = tx.Commit(); err != nil {
return nil, err
}
return result, nil
}
-234
View File
@@ -1,234 +0,0 @@
package store
import "time"
type WorkbenchModule struct {
Code string `db:"code"`
Name string `db:"name"`
IconFileID string `db:"icon_file_id"`
Route string `db:"route"`
SortOrder int `db:"sort_order"`
UnreadCount int `db:"unread_count"`
Enabled bool `db:"enabled"`
Permission string `db:"permission"`
}
type OrganizationDepartment struct {
DepartmentID string `db:"department_id"`
Name string `db:"name"`
ParentID string `db:"parent_id"`
ChildCount int `db:"child_count"`
MemberCount int `db:"member_count"`
Version int `db:"version"`
}
type OrganizationMember struct {
UserID int64 `db:"user_id"`
DepartmentID string `db:"department_id"`
DisplayName string `db:"display_name"`
Position string `db:"position"`
Mobile string `db:"mobile"`
Email string `db:"email"`
AvatarFileID string `db:"avatar_file_id"`
Status string `db:"status"`
Contactable bool `db:"contactable"`
Version int `db:"version"`
}
type UserSummary struct {
UserID int64
DisplayName string
AvatarFileID string
}
type AttendanceRule struct {
RuleID string `db:"rule_id"`
Name string `db:"name"`
Workday string `db:"workday"`
LocationRequired bool `db:"location_required"`
WifiRequired bool `db:"wifi_required"`
AllowOutsideCheckin bool `db:"allow_outside_checkin"`
AppealAllowed bool `db:"appeal_allowed"`
AppealDeadline *time.Time `db:"appeal_deadline"`
}
type AttendanceShift struct {
ShiftID string `db:"shift_id"`
RuleID string `db:"rule_id"`
Name string `db:"name"`
WorkStart string `db:"work_start"`
WorkEnd string `db:"work_end"`
CheckinStart string `db:"checkin_start"`
CheckoutEnd string `db:"checkout_end"`
}
type AttendanceCheckin struct {
CheckinID string `db:"checkin_id"`
Action string `db:"action"`
ClientTime time.Time `db:"client_time"`
ServerTime time.Time `db:"server_time"`
Result string `db:"result"`
Location string `db:"location"`
Wifi string `db:"wifi"`
Device string `db:"device"`
}
type AttendanceRecord struct {
RecordID string `db:"record_id"`
UserID int64 `db:"user_id"`
Date time.Time `db:"record_date"`
RuleID string `db:"rule_id"`
ShiftID string `db:"shift_id"`
Status string `db:"status"`
Exceptions string `db:"exceptions"`
AppealStatus string `db:"appeal_status"`
Version int `db:"version"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
Checkins []AttendanceCheckin
}
type AttendanceConfig struct {
Date time.Time
Timezone string
Rule AttendanceRule
Shifts []AttendanceShift
TodayRecord *AttendanceRecord
}
type AttendanceStats struct {
ExpectedCount int
CheckedCount int
ExceptionCount int
NotCheckedCount int
}
type AttendanceAppeal struct {
AppealID string `db:"appeal_id"`
RecordID string `db:"record_id"`
UserID int64 `db:"user_id"`
AppealType string `db:"appeal_type"`
Reason string `db:"reason"`
TargetAction string `db:"target_action"`
TargetTime *time.Time `db:"target_time"`
AttachmentFileIDs string `db:"attachment_file_ids"`
Status string `db:"status"`
Version int `db:"version"`
SubmittedAt time.Time `db:"submitted_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type Meeting struct {
MeetingID string `db:"meeting_id"`
OrganizerID int64 `db:"organizer_id"`
RecorderID int64 `db:"recorder_user_id"`
Title string `db:"title"`
StartAt time.Time `db:"start_at"`
EndAt time.Time `db:"end_at"`
Location string `db:"location"`
State string `db:"state"`
ReminderMinutes string `db:"reminder_minutes"`
IdempotencyKey string `db:"idempotency_key"`
Version int `db:"version"`
CanceledAt *time.Time `db:"canceled_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
Organizer UserSummary
Recorder UserSummary
Participants []MeetingParticipant
}
type MeetingParticipant struct {
MeetingID string `db:"meeting_id"`
UserID int64 `db:"user_id"`
Response string `db:"response"`
ResponseAt *time.Time `db:"response_at"`
User UserSummary
}
type MeetingNoteFile struct {
NoteFileID string `db:"note_file_id"`
MeetingID string `db:"meeting_id"`
FileID string `db:"file_id"`
UploaderID int64 `db:"uploader_id"`
Uploader UserSummary
FileName string `db:"file_name"`
MIMEType string `db:"mime_type"`
FileSize int64 `db:"file_size"`
FileETag string `db:"file_etag"`
Status string `db:"status"`
PreviewStatus string `db:"preview_status"`
PreviewMode string `db:"preview_mode"`
PreviewFileID string `db:"preview_file_id"`
Version int `db:"version"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type MeetingNoteFileInput struct {
FileID string
FileName string
MIMEType string
FileSize *int64
FileETag string
}
type DailyReport struct {
ReportID string `db:"report_id"`
AuthorID int64 `db:"author_id"`
ReportDate time.Time `db:"report_date"`
Content string `db:"content"`
Attachments string `db:"attachments"`
Status string `db:"status"`
ReadState string `db:"read_state"`
Version int `db:"version"`
SubmittedAt *time.Time `db:"submitted_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type MonthlyReport struct {
ReportID string `db:"report_id"`
AuthorID int64 `db:"author_id"`
Month string `db:"report_month"`
WorkSummary string `db:"work_summary"`
NextPlan string `db:"next_plan"`
Attachments string `db:"attachments"`
Status string `db:"status"`
ReadState string `db:"read_state"`
Version int `db:"version"`
SubmittedAt *time.Time `db:"submitted_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
type ReportStats struct {
ExpectedCount int
SubmittedCount int
MissingCount int
DepartmentID string
}
type MailboxConfig struct {
Enabled bool `db:"enabled"`
AllowAnonymous bool `db:"allow_anonymous"`
RequiredFields string `db:"required_fields"`
MaxLength int `db:"max_length"`
AttachmentPolicy string `db:"attachment_policy"`
NoticeTitle string `db:"notice_title"`
NoticeContent string `db:"notice_content"`
Version int `db:"version"`
}
type MailboxSubmission struct {
SubmissionID string `db:"submission_id"`
SenderID int64 `db:"sender_id"`
Subject string `db:"subject"`
Content string `db:"content"`
Anonymous bool `db:"anonymous"`
AttachmentFileIDs string `db:"attachment_file_ids"`
NoticeVersion string `db:"notice_version"`
Status string `db:"status"`
SubmittedAt time.Time `db:"submitted_at"`
}
+477
View File
@@ -0,0 +1,477 @@
package store
import (
"context"
"database/sql"
"encoding/json"
"errors"
"github.com/jmoiron/sqlx"
)
type PaymentAccount struct {
ID int64 `db:"id"`
AccountType int `db:"account_type"`
BankCode string `db:"bank_code"`
BankName string `db:"bank_name"`
HolderCipher string `db:"holder_cipher"`
HolderMask string `db:"holder_mask"`
MobileCipher string `db:"mobile_cipher"`
MobileMask string `db:"mobile_mask"`
AccountNoCipher string `db:"account_no_cipher"`
AccountNoHash string `db:"account_no_hash"`
AccountNoMask string `db:"account_no_mask"`
QRAttachmentID int64 `db:"qr_attachment_id"`
VerifyStatus int `db:"verify_status"`
IsDefault int `db:"is_default"`
CreatedAt string `db:"created_at"`
UpdatedAt string `db:"updated_at"`
}
type PaymentAccountCreation struct {
TenantID int64
MemberID int64
ExpectedTradePasswordHash string
AccountType int
BankCode string
BankName string
HolderCipher string
HolderMask string
MobileCipher string
MobileMask string
AccountNoCipher string
AccountNoHash string
AccountNoMask string
QRAttachmentID int64
}
type WithdrawalChannel struct {
ID int64 `db:"id"`
Code string `db:"channel_code"`
Name string `db:"name"`
ChannelType int `db:"channel_type"`
MinAmount int64 `db:"min_amount"`
MaxAmount int64 `db:"max_amount"`
FeeType int `db:"fee_type"`
FeeFixedAmount int64 `db:"fee_fixed_amount"`
FeeRate int `db:"fee_rate"`
}
type WithdrawalWallet struct {
AccountID int64 `db:"account_id"`
AvailableAmount int64 `db:"available_amount"`
FrozenAmount int64 `db:"frozen_amount"`
AccountStatus int `db:"account_status"`
DailyLimit int `db:"daily_limit"`
TodayCount int `db:"today_count"`
}
type WithdrawalCreation struct {
TenantID int64
MemberID int64
ExpectedTradePasswordHash string
WithdrawalNo string
PaymentAccountID int64
ExpectedAccountType int
ExpectedBankCode string
ExpectedBankName string
ExpectedHolderCipher string
ExpectedMobileCipher string
ExpectedAccountNoCipher string
ExpectedAccountNoHash string
ExpectedQRAttachmentID int64
Channel WithdrawalChannel
RequestAmount int64
FeeAmount int64
ActualAmount int64
Currency string
AccountSnapshotCipher string
AccountMask string
BankName string
HolderMask string
AvailableLedgerNo string
FrozenLedgerNo string
}
type WithdrawalOrder struct {
ID int64 `db:"id"`
WithdrawalNo string `db:"withdrawal_no"`
PaymentAccountID int64 `db:"payment_account_id"`
ChannelID int64 `db:"channel_id"`
ChannelName string `db:"channel_name"`
RequestAmount int64 `db:"request_amount"`
FeeAmount int64 `db:"fee_amount"`
ActualAmount int64 `db:"actual_amount"`
Currency string `db:"currency"`
AccountMask string `db:"account_mask"`
BankName string `db:"bank_name"`
HolderMask string `db:"holder_mask"`
RiskResult json.RawMessage `db:"risk_result"`
Status int `db:"withdrawal_status"`
ReviewReasonCode string `db:"review_reason_code"`
ReviewReason string `db:"review_reason"`
ExternalNo string `db:"external_no"`
ReviewedAt string `db:"reviewed_at"`
ProcessedAt string `db:"processed_at"`
CompletedAt string `db:"completed_at"`
CreatedAt string `db:"created_at"`
UpdatedAt string `db:"updated_at"`
Replayed bool `db:"-"`
}
type WithdrawalPage struct {
Items []WithdrawalOrder
Total int64
}
func (s *Store) PaymentAccounts(ctx context.Context, tenantID, memberID int64) ([]PaymentAccount, error) {
items := make([]PaymentAccount, 0)
err := s.db.SelectContext(ctx, &items, `SELECT id,account_type,bank_code,bank_name,holder_mask,mobile_mask,
account_no_mask,qr_attachment_id,verify_status,is_default,
DATE_FORMAT(created_at,'%Y-%m-%dT%H:%i:%sZ') AS created_at,
DATE_FORMAT(updated_at,'%Y-%m-%dT%H:%i:%sZ') AS updated_at
FROM hg_app_member_payment_account WHERE tenant_id=? AND member_id=? AND deleted_at IS NULL
ORDER BY is_default ASC,id DESC`, tenantID, memberID)
return items, err
}
func (s *Store) CreatePaymentAccount(ctx context.Context, input PaymentAccountCreation) (*PaymentAccount, error) {
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
var memberID int64
err = tx.GetContext(ctx, &memberID, `SELECT id FROM hg_app_member WHERE tenant_id=? AND id=?
AND trade_password_hash=? AND member_status=1 AND deleted_at IS NULL LIMIT 1 FOR UPDATE`,
input.TenantID, input.MemberID, input.ExpectedTradePasswordHash)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrConcurrentUpdate
}
if err != nil {
return nil, err
}
if input.QRAttachmentID > 0 {
var attachmentCount int
err = tx.GetContext(ctx, &attachmentCount, `SELECT COUNT(*) FROM hg_sys_attachment
WHERE tenant_id=? AND id=? AND app_id='workbench' AND member_id=? AND status=1 AND mime_type LIKE 'image/%'`,
input.TenantID, input.QRAttachmentID, input.MemberID)
if err != nil {
return nil, err
}
if attachmentCount != 1 {
return nil, ErrAttachmentInvalid
}
}
var accountCount int
if err = tx.GetContext(ctx, &accountCount, `SELECT COUNT(*) FROM hg_app_member_payment_account
WHERE tenant_id=? AND member_id=? AND deleted_at IS NULL`, input.TenantID, input.MemberID); err != nil {
return nil, err
}
isDefault := 2
if accountCount == 0 {
isDefault = 1
}
result, err := tx.ExecContext(ctx, `INSERT INTO hg_app_member_payment_account
(tenant_id,member_id,account_type,bank_code,bank_name,holder_cipher,holder_mask,mobile_cipher,
mobile_mask,account_no_cipher,account_no_hash,account_no_mask,qr_attachment_id,verify_status,
is_default,created_by,updated_by)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,1,?,?,?)`, input.TenantID, input.MemberID, input.AccountType,
input.BankCode, input.BankName, input.HolderCipher, input.HolderMask, nullableString(input.MobileCipher),
input.MobileMask, input.AccountNoCipher, input.AccountNoHash, input.AccountNoMask, input.QRAttachmentID,
isDefault, input.MemberID, input.MemberID)
if err != nil {
if isDuplicate(err) {
return nil, ErrPaymentAccountExists
}
return nil, err
}
id, err := result.LastInsertId()
if err != nil {
return nil, err
}
if err = tx.Commit(); err != nil {
return nil, err
}
return &PaymentAccount{ID: id, AccountType: input.AccountType, BankCode: input.BankCode,
BankName: input.BankName, HolderMask: input.HolderMask, MobileMask: input.MobileMask,
AccountNoMask: input.AccountNoMask, QRAttachmentID: input.QRAttachmentID,
VerifyStatus: 1, IsDefault: isDefault}, nil
}
func nullableString(value string) any {
if value == "" {
return nil
}
return value
}
func (s *Store) PaymentAccountForWithdrawal(ctx context.Context, tenantID, memberID, accountID int64) (*PaymentAccount, error) {
var row PaymentAccount
err := s.db.GetContext(ctx, &row, `SELECT id,account_type,bank_code,bank_name,holder_cipher,holder_mask,
COALESCE(mobile_cipher,'') AS mobile_cipher,mobile_mask,account_no_cipher,account_no_hash,
account_no_mask,qr_attachment_id,verify_status,is_default,
DATE_FORMAT(created_at,'%Y-%m-%dT%H:%i:%sZ') AS created_at,
DATE_FORMAT(updated_at,'%Y-%m-%dT%H:%i:%sZ') AS updated_at
FROM hg_app_member_payment_account WHERE tenant_id=? AND member_id=? AND id=? AND deleted_at IS NULL LIMIT 1`,
tenantID, memberID, accountID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrPaymentAccountInvalid
}
return &row, err
}
func (s *Store) ActiveWithdrawalChannel(ctx context.Context, tenantID int64, channelTypes []int) (*WithdrawalChannel, error) {
if len(channelTypes) == 0 {
return nil, ErrWithdrawalChannel
}
query, args, err := sqlx.In(`SELECT id,channel_code,name,channel_type,min_amount,max_amount,
fee_type,fee_fixed_amount,fee_rate FROM hg_payment_channel
WHERE tenant_id=? AND channel_type IN (?) AND direction IN (2,3) AND status=1 AND deleted_at IS NULL
ORDER BY CASE WHEN channel_type=? THEN 0 ELSE 1 END,sort,id LIMIT 1`, tenantID, channelTypes, channelTypes[0])
if err != nil {
return nil, err
}
var row WithdrawalChannel
err = s.db.GetContext(ctx, &row, query, args...)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrWithdrawalChannel
}
return &row, err
}
func (s *Store) WithdrawalWallet(ctx context.Context, tenantID, memberID int64) (*WithdrawalWallet, error) {
var row WithdrawalWallet
err := s.db.GetContext(ctx, &row, `SELECT w.id AS account_id,w.available_amount,w.frozen_amount,w.account_status,
COALESCE(lv.daily_withdraw_limit,0) AS daily_limit,
(SELECT COUNT(*) FROM hg_withdrawal_order o WHERE o.tenant_id=w.tenant_id AND o.member_id=w.member_id
AND o.created_at>=UTC_DATE() AND o.created_at<UTC_DATE()+INTERVAL 1 DAY) AS today_count
FROM hg_wallet_account w JOIN hg_app_member m ON m.tenant_id=w.tenant_id AND m.id=w.member_id
LEFT JOIN hg_app_member_level l ON l.tenant_id=m.tenant_id AND l.id=m.level_id AND l.status=1 AND l.deleted_at IS NULL
LEFT JOIN hg_app_member_level_version lv ON lv.tenant_id=l.tenant_id AND lv.id=l.current_version_id AND lv.publish_status=3
WHERE w.tenant_id=? AND w.member_id=? AND w.currency='CNY' AND m.member_status=1 AND m.deleted_at IS NULL LIMIT 1`,
tenantID, memberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountRestricted
}
return &row, err
}
func (s *Store) CreateWithdrawal(ctx context.Context, input WithdrawalCreation) (*WithdrawalOrder, error) {
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return nil, err
}
defer tx.Rollback()
var memberID int64
err = tx.GetContext(ctx, &memberID, `SELECT id FROM hg_app_member WHERE tenant_id=? AND id=?
AND trade_password_hash=? AND member_status=1 AND deleted_at IS NULL LIMIT 1 FOR UPDATE`,
input.TenantID, input.MemberID, input.ExpectedTradePasswordHash)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrConcurrentUpdate
}
if err != nil {
return nil, err
}
if existing, findErr := withdrawalByNo(ctx, tx, input.TenantID, input.MemberID, input.WithdrawalNo); findErr == nil {
if existing.PaymentAccountID != input.PaymentAccountID || existing.RequestAmount != input.RequestAmount || existing.Currency != input.Currency {
return nil, ErrIdempotencyConflict
}
existing.Replayed = true
if err := tx.Commit(); err != nil {
return nil, err
}
return existing, nil
} else if !errors.Is(findErr, ErrWithdrawalNotFound) {
return nil, findErr
}
var account PaymentAccount
err = tx.GetContext(ctx, &account, `SELECT id,account_type,bank_code,bank_name,holder_cipher,holder_mask,
COALESCE(mobile_cipher,'') AS mobile_cipher,mobile_mask,account_no_cipher,account_no_hash,
account_no_mask,qr_attachment_id,verify_status,is_default,'' AS created_at,'' AS updated_at
FROM hg_app_member_payment_account WHERE tenant_id=? AND member_id=? AND id=?
AND verify_status=2 AND deleted_at IS NULL LIMIT 1 FOR UPDATE`, input.TenantID, input.MemberID, input.PaymentAccountID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrPaymentAccountInvalid
}
if err != nil {
return nil, err
}
if account.AccountType != input.ExpectedAccountType || account.BankCode != input.ExpectedBankCode ||
account.BankName != input.ExpectedBankName || account.HolderCipher != input.ExpectedHolderCipher ||
account.MobileCipher != input.ExpectedMobileCipher || account.AccountNoCipher != input.ExpectedAccountNoCipher ||
account.AccountNoHash != input.ExpectedAccountNoHash || account.QRAttachmentID != input.ExpectedQRAttachmentID {
return nil, ErrConcurrentUpdate
}
var channel WithdrawalChannel
err = tx.GetContext(ctx, &channel, `SELECT id,channel_code,name,channel_type,min_amount,max_amount,
fee_type,fee_fixed_amount,fee_rate FROM hg_payment_channel WHERE tenant_id=? AND id=?
AND direction IN (2,3) AND status=1 AND deleted_at IS NULL LIMIT 1 FOR UPDATE`, input.TenantID, input.Channel.ID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrWithdrawalChannel
}
if err != nil {
return nil, err
}
if channel != input.Channel {
return nil, ErrConcurrentUpdate
}
if input.RequestAmount < maxInt64(1, channel.MinAmount) || (channel.MaxAmount > 0 && input.RequestAmount > channel.MaxAmount) {
return nil, ErrWithdrawalAmount
}
var wallet WithdrawalWallet
err = tx.GetContext(ctx, &wallet, `SELECT w.id AS account_id,w.available_amount,w.frozen_amount,w.account_status,
COALESCE(lv.daily_withdraw_limit,0) AS daily_limit,
(SELECT COUNT(*) FROM hg_withdrawal_order o WHERE o.tenant_id=w.tenant_id AND o.member_id=w.member_id
AND o.created_at>=UTC_DATE() AND o.created_at<UTC_DATE()+INTERVAL 1 DAY) AS today_count
FROM hg_wallet_account w JOIN hg_app_member m ON m.tenant_id=w.tenant_id AND m.id=w.member_id
LEFT JOIN hg_app_member_level l ON l.tenant_id=m.tenant_id AND l.id=m.level_id AND l.status=1 AND l.deleted_at IS NULL
LEFT JOIN hg_app_member_level_version lv ON lv.tenant_id=l.tenant_id AND lv.id=l.current_version_id AND lv.publish_status=3
WHERE w.tenant_id=? AND w.member_id=? AND w.currency=? LIMIT 1 FOR UPDATE`, input.TenantID, input.MemberID, input.Currency)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountRestricted
}
if err != nil {
return nil, err
}
if wallet.AccountStatus != 1 {
return nil, ErrAccountRestricted
}
if wallet.DailyLimit > 0 && wallet.TodayCount >= wallet.DailyLimit {
return nil, ErrWithdrawalLimit
}
if wallet.AvailableAmount < input.RequestAmount {
return nil, ErrInsufficientBalance
}
riskResult := json.RawMessage(`{"result":"passed","mode":"sandbox"}`)
result, err := tx.ExecContext(ctx, `INSERT INTO hg_withdrawal_order
(tenant_id,withdrawal_no,member_id,payment_account_id,channel_id,request_amount,fee_amount,
actual_amount,currency,account_snapshot_cipher,account_mask,bank_name,holder_mask,risk_result,
withdrawal_status,created_by,updated_by)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,1,?,?)`, input.TenantID, input.WithdrawalNo, input.MemberID,
input.PaymentAccountID, channel.ID, input.RequestAmount, input.FeeAmount, input.ActualAmount,
input.Currency, input.AccountSnapshotCipher, input.AccountMask, input.BankName, input.HolderMask,
riskResult, input.MemberID, input.MemberID)
if err != nil {
if isDuplicate(err) {
return nil, ErrIdempotencyConflict
}
return nil, err
}
orderID, err := result.LastInsertId()
if err != nil {
return nil, err
}
beforeAvailable := wallet.AvailableAmount
afterAvailable := beforeAvailable - input.RequestAmount
beforeFrozen := wallet.FrozenAmount
afterFrozen := beforeFrozen + input.RequestAmount
update, err := tx.ExecContext(ctx, `UPDATE hg_wallet_account SET available_amount=?,frozen_amount=?,version=version+1,
updated_at=UTC_TIMESTAMP(3) WHERE id=? AND tenant_id=? AND available_amount=? AND frozen_amount=? AND account_status=1`,
afterAvailable, afterFrozen, wallet.AccountID, input.TenantID, beforeAvailable, beforeFrozen)
if err != nil {
return nil, err
}
if affected, _ := update.RowsAffected(); affected != 1 {
return nil, ErrConcurrentUpdate
}
availableLedgerID, err := insertWithdrawalLedger(ctx, tx, input, wallet.AccountID, input.AvailableLedgerNo,
"available", 2, beforeAvailable, afterAvailable, "available", "取出申请冻结可用余额")
if err != nil {
return nil, err
}
_ = availableLedgerID
frozenLedgerID, err := insertWithdrawalLedger(ctx, tx, input, wallet.AccountID, input.FrozenLedgerNo,
"frozen", 1, beforeFrozen, afterFrozen, "frozen", "取出申请增加冻结余额")
if err != nil {
return nil, err
}
if _, err = tx.ExecContext(ctx, `UPDATE hg_withdrawal_order SET freeze_ledger_id=? WHERE tenant_id=? AND id=?`,
frozenLedgerID, input.TenantID, orderID); err != nil {
return nil, err
}
if err = tx.Commit(); err != nil {
return nil, err
}
return &WithdrawalOrder{ID: orderID, WithdrawalNo: input.WithdrawalNo,
PaymentAccountID: input.PaymentAccountID, ChannelID: channel.ID, ChannelName: channel.Name,
RequestAmount: input.RequestAmount, FeeAmount: input.FeeAmount, ActualAmount: input.ActualAmount,
Currency: input.Currency, AccountMask: input.AccountMask, BankName: input.BankName,
HolderMask: input.HolderMask, RiskResult: riskResult, Status: 1}, nil
}
func insertWithdrawalLedger(ctx context.Context, tx *sqlx.Tx, input WithdrawalCreation, accountID int64,
ledgerNo, entryType string, direction int, beforeAmount, afterAmount int64, suffix, remark string) (int64, error) {
result, err := tx.ExecContext(ctx, `INSERT INTO hg_wallet_ledger
(tenant_id,ledger_no,account_id,member_id,biz_type,biz_no,entry_type,direction,amount,
before_amount,after_amount,currency,reversal_of,idempotency_key,remark,operator_id,occurred_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,0,?,?,?,UTC_TIMESTAMP(3))`, input.TenantID, ledgerNo, accountID,
input.MemberID, "withdraw_freeze", input.WithdrawalNo, entryType, direction, input.RequestAmount,
beforeAmount, afterAmount, input.Currency, "withdrawal:"+input.WithdrawalNo+":"+suffix, remark, input.MemberID)
if err != nil {
return 0, err
}
return result.LastInsertId()
}
func (s *Store) WithdrawalOrders(ctx context.Context, tenantID, memberID int64, status, offset, limit int) (*WithdrawalPage, error) {
where := ` WHERE o.tenant_id=? AND o.member_id=?`
args := []any{tenantID, memberID}
if status > 0 {
where += ` AND o.withdrawal_status=?`
args = append(args, status)
}
var total int64
if err := s.db.GetContext(ctx, &total, `SELECT COUNT(*) FROM hg_withdrawal_order o`+where, args...); err != nil {
return nil, err
}
args = append(args, limit, offset)
items := make([]WithdrawalOrder, 0)
err := s.db.SelectContext(ctx, &items, withdrawalSelectSQL()+where+
` ORDER BY o.created_at DESC,o.id DESC LIMIT ? OFFSET ?`, args...)
return &WithdrawalPage{Items: items, Total: total}, err
}
func (s *Store) WithdrawalOrder(ctx context.Context, tenantID, memberID, orderID int64) (*WithdrawalOrder, error) {
var row WithdrawalOrder
err := s.db.GetContext(ctx, &row, withdrawalSelectSQL()+
` WHERE o.tenant_id=? AND o.member_id=? AND o.id=? LIMIT 1`, tenantID, memberID, orderID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrWithdrawalNotFound
}
return &row, err
}
func withdrawalByNo(ctx context.Context, db queryer, tenantID, memberID int64, no string) (*WithdrawalOrder, error) {
query := withdrawalSelectSQL() + ` WHERE o.tenant_id=? AND o.member_id=? AND o.withdrawal_no=? LIMIT 1`
var row WithdrawalOrder
err := db.GetContext(ctx, &row, query, tenantID, memberID, no)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrWithdrawalNotFound
}
return &row, err
}
func withdrawalSelectSQL() string {
return `SELECT o.id,o.withdrawal_no,o.payment_account_id,o.channel_id,COALESCE(c.name,'') AS channel_name,
o.request_amount,o.fee_amount,o.actual_amount,o.currency,o.account_mask,o.bank_name,o.holder_mask,
COALESCE(o.risk_result,JSON_OBJECT()) AS risk_result,o.withdrawal_status,o.review_reason_code,
o.review_reason,o.external_no,
COALESCE(DATE_FORMAT(o.reviewed_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS reviewed_at,
COALESCE(DATE_FORMAT(o.processed_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS processed_at,
COALESCE(DATE_FORMAT(o.completed_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS completed_at,
DATE_FORMAT(o.created_at,'%Y-%m-%dT%H:%i:%sZ') AS created_at,
DATE_FORMAT(o.updated_at,'%Y-%m-%dT%H:%i:%sZ') AS updated_at
FROM hg_withdrawal_order o LEFT JOIN hg_payment_channel c ON c.tenant_id=o.tenant_id AND c.id=o.channel_id`
}
func maxInt64(left, right int64) int64 {
if left > right {
return left
}
return right
}