From a8e939da4000633d7a3359f317ebe7606a9e1d2e Mon Sep 17 00:00:00 2001 From: ryan Date: Sat, 27 Jun 2026 13:06:54 +0800 Subject: [PATCH] feat: rule and message_log store CRUD --- internal/store/message_log.go | 72 ++++++++++++++++ internal/store/rule.go | 156 ++++++++++++++++++++++++++++++++++ 2 files changed, 228 insertions(+) create mode 100644 internal/store/message_log.go create mode 100644 internal/store/rule.go diff --git a/internal/store/message_log.go b/internal/store/message_log.go new file mode 100644 index 0000000..74bee67 --- /dev/null +++ b/internal/store/message_log.go @@ -0,0 +1,72 @@ +package store + +import ( + "context" + "fmt" + + "aiaa-notification-service/internal/model" +) + +type MessageLogFilter struct { + Source string `form:"source"` + Event string `form:"event"` + Status string `form:"status"` + Page int `form:"page"` + PageSize int `form:"page_size"` +} + +func (s *Store) CreateMessageLog(ctx context.Context, ml *model.MessageLog) error { + query := `INSERT INTO message_log (rule_id, channel_id, source, event, payload, content, status, retry_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` + result, err := s.DB.ExecContext(ctx, query, ml.RuleID, ml.ChannelID, ml.Source, ml.Event, ml.Payload, ml.Content, ml.Status, ml.RetryCount) + if err != nil { + return fmt.Errorf("create message_log: %w", err) + } + id, _ := result.LastInsertId() + ml.ID = id + return nil +} + +func (s *Store) UpdateMessageLog(ctx context.Context, id int64, status string, response, errMsg *string) error { + query := `UPDATE message_log SET status=?, response=?, error_msg=? WHERE id=?` + _, err := s.DB.ExecContext(ctx, query, status, response, errMsg, id) + return err +} + +func (s *Store) ListMessageLogs(ctx context.Context, filter MessageLogFilter) ([]model.MessageLog, int, error) { + where := "WHERE 1=1" + args := []interface{}{} + if filter.Source != "" { + where += " AND source = ?" + args = append(args, filter.Source) + } + if filter.Event != "" { + where += " AND event = ?" + args = append(args, filter.Event) + } + if filter.Status != "" { + where += " AND status = ?" + args = append(args, filter.Status) + } + + var count int + countQuery := "SELECT COUNT(*) FROM message_log " + where + if err := s.DB.GetContext(ctx, &count, countQuery, args...); err != nil { + return nil, 0, err + } + + if filter.Page <= 0 { + filter.Page = 1 + } + if filter.PageSize <= 0 { + filter.PageSize = 20 + } + offset := (filter.Page - 1) * filter.PageSize + + var logs []model.MessageLog + query := "SELECT * FROM message_log " + where + " ORDER BY id DESC LIMIT ? OFFSET ?" + args = append(args, filter.PageSize, offset) + if err := s.DB.SelectContext(ctx, &logs, query, args...); err != nil { + return nil, 0, err + } + return logs, count, nil +} diff --git a/internal/store/rule.go b/internal/store/rule.go new file mode 100644 index 0000000..e3c9e9c --- /dev/null +++ b/internal/store/rule.go @@ -0,0 +1,156 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + + "aiaa-notification-service/internal/model" +) + +func (s *Store) CreateRule(ctx context.Context, r *model.Rule, channelIDs []int) error { + tx, err := s.DB.BeginTxx(ctx, nil) + if err != nil { + return fmt.Errorf("begin tx: %w", err) + } + defer tx.Rollback() + + query := `INSERT INTO rule (source_id, event, template_id, conditions, enabled) VALUES (?, ?, ?, ?, ?)` + condsJSON, _ := marshalJSON(r.Conditions) + result, err := tx.ExecContext(ctx, query, r.SourceID, r.Event, r.TemplateID, condsJSON, r.Enabled) + if err != nil { + return fmt.Errorf("create rule: %w", err) + } + id, _ := result.LastInsertId() + r.ID = int(id) + + for _, chID := range channelIDs { + _, err := tx.ExecContext(ctx, `INSERT INTO rule_channel (rule_id, channel_id, enabled) VALUES (?, ?, 1)`, r.ID, chID) + if err != nil { + return fmt.Errorf("add rule_channel: %w", err) + } + } + return tx.Commit() +} + +func (s *Store) GetRule(ctx context.Context, id int) (*model.Rule, error) { + var r model.Rule + var condsBytes []byte + row := s.DB.QueryRowContext(ctx, `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM rule WHERE id = ?`, id) + if err := row.Scan(&r.ID, &r.SourceID, &r.Event, &r.TemplateID, &condsBytes, &r.Enabled, &r.CreatedAt, &r.UpdatedAt); err != nil { + return nil, fmt.Errorf("get rule %d: %w", id, err) + } + if len(condsBytes) > 0 && string(condsBytes) != "null" { + raw := json.RawMessage(condsBytes) + r.Conditions = &raw + } + return &r, nil +} + +func (s *Store) GetRuleBySourceEvent(ctx context.Context, sourceID int, event string) (*model.Rule, error) { + var r model.Rule + var condsBytes []byte + query := `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM rule WHERE source_id = ? AND event = ? AND enabled = 1` + row := s.DB.QueryRowContext(ctx, query, sourceID, event) + if err := row.Scan(&r.ID, &r.SourceID, &r.Event, &r.TemplateID, &condsBytes, &r.Enabled, &r.CreatedAt, &r.UpdatedAt); err != nil { + return nil, fmt.Errorf("get rule by source+event: %w", err) + } + if len(condsBytes) > 0 && string(condsBytes) != "null" { + raw := json.RawMessage(condsBytes) + r.Conditions = &raw + } + return &r, nil +} + +func (s *Store) ListRules(ctx context.Context) ([]model.Rule, error) { + rows, err := s.DB.QueryContext(ctx, `SELECT id, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM rule ORDER BY id`) + if err != nil { + return nil, fmt.Errorf("list rules: %w", err) + } + defer rows.Close() + return scanRules(rows) +} + +func (s *Store) UpdateRule(ctx context.Context, id int, r *model.Rule, channelIDs []int) error { + tx, err := s.DB.BeginTxx(ctx, nil) + if err != nil { + return fmt.Errorf("begin tx: %w", err) + } + defer tx.Rollback() + + condsJSON, _ := marshalJSON(r.Conditions) + _, err = tx.ExecContext(ctx, `UPDATE rule SET source_id=?, event=?, template_id=?, conditions=?, enabled=? WHERE id=?`, + r.SourceID, r.Event, r.TemplateID, condsJSON, r.Enabled, id) + if err != nil { + return fmt.Errorf("update rule: %w", err) + } + + if channelIDs != nil { + _, _ = tx.ExecContext(ctx, `DELETE FROM rule_channel WHERE rule_id = ?`, id) + for _, chID := range channelIDs { + _, err := tx.ExecContext(ctx, `INSERT INTO rule_channel (rule_id, channel_id, enabled) VALUES (?, ?, 1)`, id, chID) + if err != nil { + return fmt.Errorf("add rule_channel: %w", err) + } + } + } + return tx.Commit() +} + +func (s *Store) DeleteRule(ctx context.Context, id int) error { + _, err := s.DB.ExecContext(ctx, `DELETE FROM rule WHERE id = ?`, id) + return err +} + +func (s *Store) SetRuleEnabled(ctx context.Context, id int, enabled bool) error { + v := 0 + if enabled { + v = 1 + } + _, err := s.DB.ExecContext(ctx, `UPDATE rule SET enabled = ? WHERE id = ?`, v, id) + return err +} + +func (s *Store) GetRuleChannels(ctx context.Context, ruleID int) ([]model.RuleChannel, error) { + var rcs []model.RuleChannel + err := s.DB.SelectContext(ctx, &rcs, `SELECT id, rule_id, channel_id, enabled FROM rule_channel WHERE rule_id = ? AND enabled = 1`, ruleID) + if err != nil { + return nil, fmt.Errorf("get rule channels: %w", err) + } + return rcs, nil +} + +func (s *Store) SetRuleChannelEnabled(ctx context.Context, ruleID, channelID int, enabled bool) error { + v := 0 + if enabled { + v = 1 + } + _, err := s.DB.ExecContext(ctx, `UPDATE rule_channel SET enabled = ? WHERE rule_id = ? AND channel_id = ?`, v, ruleID, channelID) + return err +} + +// helpers +func marshalJSON(v *json.RawMessage) (interface{}, error) { + if v == nil { + return nil, nil + } + return json.Marshal(v) +} + +func scanRules(rows *sql.Rows) ([]model.Rule, error) { + var rules []model.Rule + for rows.Next() { + var r model.Rule + var condsBytes []byte + if err := rows.Scan(&r.ID, &r.SourceID, &r.Event, &r.TemplateID, &condsBytes, &r.Enabled, &r.CreatedAt, &r.UpdatedAt); err != nil { + return nil, err + } + if len(condsBytes) > 0 && string(condsBytes) != "null" { + raw := json.RawMessage(condsBytes) + r.Conditions = &raw + } + rules = append(rules, r) + } + return rules, rows.Err() +}