Veyra/internal/repository/notification.go

package repository

import (
	"context"
	"encoding/json"
	"fmt"
	"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
	"git.sundynix.cn/Blizzard/Veyra/internal/model"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"time"
)

type NotificationRepository struct{ pool *pgxpool.Pool }

func NewNotificationRepository(pool *pgxpool.Pool) *NotificationRepository {
	return &NotificationRepository{pool}
}
func (r *NotificationRepository) List(ctx context.Context, user string) ([]model.Notification, error) {
	rows, err := r.pool.Query(ctx, `SELECT id,user_id,event_id,plan_id,title,body,read_at,created_at FROM veyra_notification WHERE user_id=$1 ORDER BY created_at DESC,id DESC LIMIT 100`, user)
	if err != nil {
		return nil, mapError(err)
	}
	defer rows.Close()
	list := make([]model.Notification, 0)
	for rows.Next() {
		var n model.Notification
		if err := rows.Scan(&n.ID, &n.UserID, &n.EventID, &n.PlanID, &n.Title, &n.Body, &n.ReadAt, &n.CreatedAt); err != nil {
			return nil, mapError(err)
		}
		list = append(list, n)
	}
	return list, mapError(rows.Err())
}
func (r *NotificationRepository) MarkRead(ctx context.Context, user, id string) error {
	tag, err := r.pool.Exec(ctx, `UPDATE veyra_notification SET read_at=COALESCE(read_at,now()) WHERE id=$1 AND user_id=$2`, id, user)
	if err != nil {
		return mapError(err)
	}
	if tag.RowsAffected() != 1 {
		return apperror.ErrNotFound
	}
	return nil
}
func (r *NotificationRepository) Tasks(ctx context.Context) ([]model.NotificationTask, error) {
	rows, err := r.pool.Query(ctx, `SELECT e.id,e.aggregate_type,e.aggregate_id,e.aggregate_version,e.event_type,e.payload,e.created_at,e.processed_at,(SELECT count(*)::integer FROM veyra_notification n WHERE n.event_id=e.id) FROM veyra_outbox_event e ORDER BY created_at DESC,id DESC LIMIT 100`)
	if err != nil {
		return nil, mapError(err)
	}
	defer rows.Close()
	list := make([]model.NotificationTask, 0)
	for rows.Next() {
		var t model.NotificationTask
		if err := rows.Scan(&t.ID, &t.AggregateType, &t.AggregateID, &t.AggregateVersion, &t.EventType, &t.Payload, &t.CreatedAt, &t.ProcessedAt, &t.RecipientCount); err != nil {
			return nil, mapError(err)
		}
		list = append(list, t)
	}
	return list, mapError(rows.Err())
}

const quoteColumns = `id,symbol,exchange,data_mode,price_minor,observed_at,version,created_at,updated_at,source_id,source_version`

func scanQuote(row pgx.Row) (model.QuoteSnapshot, error) {
	var q model.QuoteSnapshot
	err := row.Scan(&q.ID, &q.Symbol, &q.Exchange, &q.DataMode, &q.PriceMinor, &q.ObservedAt, &q.Version, &q.CreatedAt, &q.UpdatedAt, &q.SourceID, &q.SourceVersion)
	q.Price = model.FormatMoney(q.PriceMinor)
	return q, mapError(err)
}
func (r *NotificationRepository) Quotes(ctx context.Context) ([]model.QuoteSnapshot, error) {
	rows, err := r.pool.Query(ctx, `SELECT `+quoteColumns+` FROM veyra_quote_snapshot WHERE data_mode='simulation' ORDER BY updated_at DESC,id DESC LIMIT 100`)
	if err != nil {
		return nil, mapError(err)
	}
	defer rows.Close()
	list := make([]model.QuoteSnapshot, 0)
	for rows.Next() {
		q, err := scanQuote(rows)
		if err != nil {
			return nil, err
		}
		list = append(list, q)
	}
	return list, mapError(rows.Err())
}
func (r *NotificationRepository) SaveQuote(ctx context.Context, q model.QuoteSnapshot, actor, requestID string) (model.QuoteSnapshot, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return q, mapError(err)
	}
	defer tx.Rollback(context.Background())
	q, err = scanQuote(tx.QueryRow(ctx, `INSERT INTO veyra_quote_snapshot(id,symbol,exchange,data_mode,price_minor,observed_at,version) VALUES($1,$2,$3,'simulation',$4,$5,1) ON CONFLICT(exchange,symbol,data_mode) DO UPDATE SET price_minor=EXCLUDED.price_minor,observed_at=EXCLUDED.observed_at,version=veyra_quote_snapshot.version+1,updated_at=now() WHERE EXCLUDED.observed_at>veyra_quote_snapshot.observed_at RETURNING `+quoteColumns, uuid.NewString(), q.Symbol, q.Exchange, q.PriceMinor, q.ObservedAt))
	if err == apperror.ErrNotFound {
		return q, apperror.ErrConflict
	}
	if err != nil {
		return q, err
	}
	payload, err := json.Marshal(q)
	if err != nil {
		return q, apperror.ErrUnavailable
	}
	_, err = tx.Exec(ctx, `INSERT INTO veyra_outbox_event(id,aggregate_type,aggregate_id,aggregate_version,event_type,payload) VALUES($1,'quote_snapshot',$2,$3,'quote.observed',$4)`, uuid.NewString(), q.ID, q.Version, payload)
	if err != nil {
		return q, mapError(err)
	}
	if err := appendAudit(ctx, tx, actor, "SIMULATION_QUOTE_RECORDED", "quote_snapshot", q.ID, requestID, map[string]any{"symbol": q.Symbol, "exchange": q.Exchange, "version": q.Version}); err != nil {
		return q, err
	}
	return q, mapError(tx.Commit(ctx))
}
func insertNotification(ctx context.Context, tx pgx.Tx, user, event string, plan *string, title, body string) error {
	_, err := tx.Exec(ctx, `INSERT INTO veyra_notification(id,user_id,event_id,plan_id,title,body) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT DO NOTHING`, uuid.NewString(), user, event, plan, title, body)
	return mapError(err)
}
func processQuote(ctx context.Context, tx pgx.Tx, event model.OutboxEvent) error {
	var payload model.QuoteSnapshot
	if json.Unmarshal(event.Payload, &payload) != nil {
		return apperror.ErrUnavailable
	}
	// A stale or superseded snapshot cannot generate a delayed buy reminder.
	current, err := scanQuote(tx.QueryRow(ctx, `SELECT `+quoteColumns+` FROM veyra_quote_snapshot WHERE id=$1 FOR SHARE`, event.AggregateID))
	if err != nil {
		return err
	}
	if current.Version != event.AggregateVersion || time.Since(current.ObservedAt) > time.Minute || current.ObservedAt.After(time.Now().Add(time.Minute)) {
		return nil
	}
	if current.DataMode == "public" {
		if current.SourceID == nil || current.SourceVersion == nil {
			return apperror.ErrUnavailable
		}
		var version int64
		var enabled bool
		if err := tx.QueryRow(ctx, `SELECT version,enabled FROM veyra_provider_config WHERE id=$1 FOR SHARE`, *current.SourceID).Scan(&version, &enabled); err != nil {
			return mapError(err)
		}
		if !enabled || version != *current.SourceVersion {
			return nil
		}
	}
	rows, err := tx.Query(ctx, `SELECT p.id,p.candidate_revision_id,a.user_id FROM veyra_trade_plan p JOIN veyra_candidate_revision r ON r.id=p.candidate_revision_id JOIN veyra_account a ON a.id=p.account_id JOIN veyra_user u ON u.id=a.user_id WHERE p.status IN ('ACTIVE','PART_FILLED') AND u.status='ACTIVE' AND r.snapshot->>'symbol'=$1 AND r.snapshot->>'exchange'=$2 ORDER BY r.candidate_id,p.id`, current.Symbol, current.Exchange)
	if err != nil {
		return mapError(err)
	}
	type target struct{ plan, revision, user string }
	targets := make([]target, 0)
	for rows.Next() {
		var t target
		if err := rows.Scan(&t.plan, &t.revision, &t.user); err != nil {
			rows.Close()
			return mapError(err)
		}
		targets = append(targets, t)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return mapError(err)
	}
	for _, target := range targets {
		publication, err := lockCandidate(ctx, tx, target.revision)
		if err != nil {
			return err
		}
		if !publication.IsActivePublication {
			continue
		}
		var status string
		var valid time.Time
		if err := tx.QueryRow(ctx, `SELECT status,valid_until FROM veyra_trade_plan WHERE id=$1 FOR UPDATE`, target.plan).Scan(&status, &valid); err != nil {
			return mapError(err)
		}
		if (status != "ACTIVE" && status != "PART_FILLED") || !valid.After(time.Now()) {
			continue
		}
		low, e1 := model.ParseMoney(publication.Snapshot.ReferencePlan.EntryLow)
		high, e2 := model.ParseMoney(publication.Snapshot.ReferencePlan.EntryHigh)
		if e1 != nil || e2 != nil {
			return apperror.ErrUnavailable
		}
		if current.PriceMinor < low || current.PriceMinor > high {
			continue
		}
		tag, err := tx.Exec(ctx, `INSERT INTO veyra_plan_trigger(id,plan_id,quote_snapshot_id,quote_version) VALUES($1,$2,$3,$4) ON CONFLICT(plan_id) DO NOTHING`, uuid.NewString(), target.plan, current.ID, current.Version)
		if err != nil {
			return mapError(err)
		}
		if tag.RowsAffected() == 0 {
			continue
		}
		label := "模拟价格"
		if current.DataMode == "public" {
			label = "公开行情价格"
		}
		body := fmt.Sprintf("%s ¥%s 进入 %s 的参考区间。观察时间 %s。请打开计划核对当前状态；此提醒不代表成交。", label, current.Price, publication.Snapshot.Name, current.ObservedAt.Format(time.RFC3339))
		if err := insertNotification(ctx, tx, target.user, event.ID, &target.plan, "模拟计划条件已满足", body); err != nil {
			return err
		}
	}
	return nil
}
func processCandidateInvalidation(ctx context.Context, tx pgx.Tx, event model.OutboxEvent) error {
	rows, err := tx.Query(ctx, `SELECT p.id,a.user_id FROM veyra_trade_plan p JOIN veyra_account a ON a.id=p.account_id WHERE p.candidate_revision_id=$1 AND p.status IN ('ACTIVE','PART_FILLED') ORDER BY p.id`, event.AggregateID)
	if err != nil {
		return mapError(err)
	}
	type target struct{ plan, user string }
	targets := make([]target, 0)
	for rows.Next() {
		var t target
		if err := rows.Scan(&t.plan, &t.user); err != nil {
			rows.Close()
			return mapError(err)
		}
		targets = append(targets, t)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return mapError(err)
	}
	for _, t := range targets {
		tag, err := tx.Exec(ctx, `UPDATE veyra_trade_plan SET status='INVALIDATED',reserved_minor=0,version=version+1,updated_at=now() WHERE id=$1 AND status IN ('ACTIVE','PART_FILLED')`, t.plan)
		if err != nil {
			return mapError(err)
		}
		reason := "对应候选已撤回或被新版替代，保留预算已释放。已记录成交与持仓保留。"
		if err := insertNotification(ctx, tx, t.user, event.ID, &t.plan, "计划已失效", reason); err != nil {
			return err
		}
		if tag.RowsAffected() > 0 {
			if err := appendAudit(ctx, tx, t.user, "PLAN_INVALIDATED", "trade_plan", t.plan, event.ID, map[string]any{"cause": event.EventType}); err != nil {
				return err
			}
		}
	}
	return nil
}
func (r *NotificationRepository) ProcessNext(ctx context.Context) (bool, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return false, mapError(err)
	}
	defer tx.Rollback(context.Background())
	var event model.OutboxEvent
	err = tx.QueryRow(ctx, `SELECT id,aggregate_type,aggregate_id,aggregate_version,event_type,payload,created_at,processed_at FROM veyra_outbox_event WHERE processed_at IS NULL ORDER BY created_at,id LIMIT 1 FOR UPDATE SKIP LOCKED`).Scan(&event.ID, &event.AggregateType, &event.AggregateID, &event.AggregateVersion, &event.EventType, &event.Payload, &event.CreatedAt, &event.ProcessedAt)
	if err == pgx.ErrNoRows {
		rows, err := tx.Query(ctx, `SELECT DISTINCT a.user_id FROM veyra_trade_plan p JOIN veyra_account a ON a.id=p.account_id JOIN veyra_candidate_revision r ON r.id=p.candidate_revision_id JOIN veyra_candidate c ON c.id=r.candidate_id WHERE p.status IN ('ACTIVE','PART_FILLED') AND (p.valid_until<=now() OR r.valid_until<=now() OR r.status<>'PUBLISHED' OR c.published_revision_id IS DISTINCT FROM r.id OR NOT (`+screeningEligibility+`)) LIMIT 100`)
		if err != nil {
			return false, mapError(err)
		}
		users := make([]string, 0)
		for rows.Next() {
			var user string
			if err := rows.Scan(&user); err != nil {
				rows.Close()
				return false, mapError(err)
			}
			users = append(users, user)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			return false, mapError(err)
		}
		for _, user := range users {
			if err := invalidatePlans(ctx, tx, user); err != nil {
				return false, err
			}
		}
		return len(users) > 0, mapError(tx.Commit(ctx))
	}
	if err != nil {
		return false, mapError(err)
	}
	switch event.EventType {
	case "quote.observed":
		err = processQuote(ctx, tx, event)
	case "candidate.withdrawn", "candidate.superseded":
		err = processCandidateInvalidation(ctx, tx, event)
	case "plan.invalidated":
		var user string
		var plan string
		err = tx.QueryRow(ctx, `SELECT a.user_id,p.id FROM veyra_trade_plan p JOIN veyra_account a ON a.id=p.account_id WHERE p.id=$1`, event.AggregateID).Scan(&user, &plan)
		if err == nil {
			err = insertNotification(ctx, tx, user, event.ID, &plan, "计划已失效", "对应候选、策略版本已失效或计划已过期，保留预算已释放。已记录成交与持仓保留。")
		}
	case "candidate.published", "position.changed":
	default:
		return false, apperror.ErrInvalidInput
	}
	if err != nil {
		return false, err
	}
	if _, err := tx.Exec(ctx, `UPDATE veyra_outbox_event SET processed_at=now() WHERE id=$1`, event.ID); err != nil {
		return false, mapError(err)
	}
	return true, mapError(tx.Commit(ctx))
}