feat: rule and message_log store CRUD
This commit is contained in:
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
Reference in New Issue
Block a user