MengMengCode_VoCat_20260809-175859/internal/store/settings.go

package store

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"
)

func DefaultNotificationSensitiveFields(channel string) []string {
	switch strings.ToLower(strings.TrimSpace(channel)) {
	case "telegram":
		return []string{"bot_token"}
	case "email":
		return []string{"password"}
	case "webhook":
		return []string{"secret"}
	case "pushplus":
		return []string{"token"}
	default:
		return nil
	}
}

func (s *Store) UpsertNotificationSetting(
	ctx context.Context,
	value NotificationSetting,
) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin notification setting update: %w", err)
	}
	defer tx.Rollback()
	if err := upsertNotificationSetting(ctx, tx, value); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit notification setting update: %w", err)
	}
	return nil
}

func upsertNotificationSetting(
	ctx context.Context,
	executor contextQueryExecer,
	value NotificationSetting,
) error {
	value.Channel = strings.ToLower(strings.TrimSpace(value.Channel))
	if value.Channel == "" {
		return errors.New("notification channel is required")
	}
	config, err := normalizeJSONObject(value.Config)
	if err != nil {
		return fmt.Errorf("normalize %s notification config: %w", value.Channel, err)
	}

	current, currentErr := notificationSetting(executor.QueryRowContext(
		ctx,
		notificationSettingSelect+` WHERE channel = ?`,
		value.Channel,
	))
	if currentErr != nil && !errors.Is(currentErr, ErrNotFound) {
		return fmt.Errorf("read %s notification setting before update: %w", value.Channel, currentErr)
	}
	fields := uniqueNonemptyStrings(
		DefaultNotificationSensitiveFields(value.Channel),
		value.SensitiveFields,
	)
	if currentErr == nil {
		fields = uniqueNonemptyStrings(fields, current.SensitiveFields)
		config, err = mergeJSONSecrets(config, current.Config, fields)
		if err != nil {
			return fmt.Errorf("preserve %s notification secrets: %w", value.Channel, err)
		}
	}
	fieldsJSON, err := json.Marshal(fields)
	if err != nil {
		return fmt.Errorf("encode notification sensitive fields: %w", err)
	}
	now := time.Now().UTC()
	createdAt := value.CreatedAt
	if createdAt.IsZero() {
		createdAt = now
	}
	updatedAt := value.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = now
	}
	_, err = executor.ExecContext(ctx, `
		INSERT INTO notification_settings (
			channel, enabled, config_json, sensitive_fields_json,
			created_at, updated_at
		) VALUES (?, ?, ?, ?, ?, ?)
		ON CONFLICT(channel) DO UPDATE SET
			enabled = excluded.enabled,
			config_json = excluded.config_json,
			sensitive_fields_json = excluded.sensitive_fields_json,
			updated_at = excluded.updated_at
	`,
		value.Channel, boolInt(value.Enabled), string(config),
		string(fieldsJSON), createdAt.Unix(), updatedAt.Unix(),
	)
	if err != nil {
		return fmt.Errorf("upsert %s notification setting: %w", value.Channel, err)
	}
	return nil
}

// SaveNotificationSettings applies a multi-channel settings form atomically.
func (s *Store) SaveNotificationSettings(
	ctx context.Context,
	values []NotificationSetting,
) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin notification settings batch: %w", err)
	}
	defer tx.Rollback()
	seen := make(map[string]struct{}, len(values))
	for index, value := range values {
		channel := strings.ToLower(strings.TrimSpace(value.Channel))
		if _, duplicate := seen[channel]; duplicate {
			return fmt.Errorf("duplicate notification channel %q", channel)
		}
		if err := upsertNotificationSetting(ctx, tx, value); err != nil {
			return fmt.Errorf("save notification channel %d: %w", index, err)
		}
		seen[channel] = struct{}{}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit notification settings batch: %w", err)
	}
	return nil
}

func (s *Store) NotificationSetting(
	ctx context.Context,
	channel string,
) (NotificationSetting, error) {
	return notificationSetting(s.db.QueryRowContext(
		ctx,
		notificationSettingSelect+` WHERE channel = ?`,
		strings.ToLower(strings.TrimSpace(channel)),
	))
}

func (s *Store) ListNotificationSettings(ctx context.Context) ([]NotificationSetting, error) {
	rows, err := s.db.QueryContext(ctx, notificationSettingSelect+` ORDER BY channel`)
	if err != nil {
		return nil, fmt.Errorf("list notification settings: %w", err)
	}
	defer rows.Close()
	values := make([]NotificationSetting, 0)
	for rows.Next() {
		value, err := notificationSetting(rows)
		if err != nil {
			return nil, fmt.Errorf("scan notification setting: %w", err)
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate notification settings: %w", err)
	}
	return values, nil
}

func (s *Store) DeleteNotificationSetting(ctx context.Context, channel string) error {
	result, err := s.db.ExecContext(
		ctx,
		`DELETE FROM notification_settings WHERE channel = ?`,
		strings.ToLower(strings.TrimSpace(channel)),
	)
	if err != nil {
		return fmt.Errorf("delete notification setting %q: %w", channel, err)
	}
	return requireAffected(result)
}

const notificationSettingSelect = `
	SELECT channel, enabled, config_json, sensitive_fields_json,
		created_at, updated_at
	FROM notification_settings`

func notificationSetting(row rowScanner) (NotificationSetting, error) {
	var value NotificationSetting
	var enabled int
	var config, fields string
	var createdAt, updatedAt int64
	err := row.Scan(
		&value.Channel, &enabled, &config, &fields, &createdAt, &updatedAt,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return NotificationSetting{}, ErrNotFound
	}
	if err != nil {
		return NotificationSetting{}, err
	}
	if err := json.Unmarshal([]byte(fields), &value.SensitiveFields); err != nil {
		return NotificationSetting{}, fmt.Errorf("decode sensitive fields: %w", err)
	}
	value.Enabled = enabled != 0
	value.Config = []byte(config)
	value.CreatedAt = time.Unix(createdAt, 0).UTC()
	value.UpdatedAt = time.Unix(updatedAt, 0).UTC()
	return value, nil
}

func uniqueNonemptyStrings(groups ...[]string) []string {
	seen := make(map[string]struct{})
	for _, group := range groups {
		for _, item := range group {
			item = strings.TrimSpace(item)
			if item != "" {
				seen[item] = struct{}{}
			}
		}
	}
	result := make([]string, 0, len(seen))
	for item := range seen {
		result = append(result, item)
	}
	sort.Strings(result)
	return result
}

func (s *Store) UpsertAppSetting(ctx context.Context, value AppSetting) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin app setting update: %w", err)
	}
	defer tx.Rollback()
	if err := upsertAppSetting(ctx, tx, value); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit app setting update: %w", err)
	}
	return nil
}

func upsertAppSetting(
	ctx context.Context,
	executor contextQueryExecer,
	value AppSetting,
) error {
	value.Key = strings.TrimSpace(value.Key)
	if value.Key == "" {
		return errors.New("app setting key is required")
	}
	normalized, err := normalizeJSONValue(value.Value)
	if err != nil {
		return fmt.Errorf("normalize app setting %q: %w", value.Key, err)
	}
	if value.Sensitive && maskedJSONValue(normalized) {
		current, currentErr := appSetting(executor.QueryRowContext(
			ctx,
			appSettingSelect+` WHERE key = ?`,
			value.Key,
		))
		switch {
		case currentErr == nil:
			normalized = current.Value
		case errors.Is(currentErr, ErrNotFound):
			return fmt.Errorf("new sensitive app setting %q requires a value", value.Key)
		default:
			return fmt.Errorf("read app setting before update: %w", currentErr)
		}
	}
	updatedAt := value.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = time.Now().UTC()
	}
	_, err = executor.ExecContext(ctx, `
		INSERT INTO app_settings (key, value_json, sensitive, updated_at)
		VALUES (?, ?, ?, ?)
		ON CONFLICT(key) DO UPDATE SET
			value_json = excluded.value_json,
			sensitive = excluded.sensitive,
			updated_at = excluded.updated_at
	`, value.Key, string(normalized), boolInt(value.Sensitive), updatedAt.Unix())
	if err != nil {
		return fmt.Errorf("upsert app setting %q: %w", value.Key, err)
	}
	return nil
}

func (s *Store) AppSetting(ctx context.Context, key string) (AppSetting, error) {
	return appSetting(s.db.QueryRowContext(
		ctx,
		appSettingSelect+` WHERE key = ?`,
		strings.TrimSpace(key),
	))
}

func (s *Store) ListAppSettings(ctx context.Context) ([]AppSetting, error) {
	rows, err := s.db.QueryContext(ctx, appSettingSelect+` ORDER BY key`)
	if err != nil {
		return nil, fmt.Errorf("list app settings: %w", err)
	}
	defer rows.Close()
	values := make([]AppSetting, 0)
	for rows.Next() {
		value, err := appSetting(rows)
		if err != nil {
			return nil, fmt.Errorf("scan app setting: %w", err)
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate app settings: %w", err)
	}
	return values, nil
}

func (s *Store) DeleteAppSetting(ctx context.Context, key string) error {
	result, err := s.db.ExecContext(ctx, `DELETE FROM app_settings WHERE key = ?`, key)
	if err != nil {
		return fmt.Errorf("delete app setting %q: %w", key, err)
	}
	return requireAffected(result)
}

const appSettingSelect = `
	SELECT key, value_json, sensitive, updated_at
	FROM app_settings`

func appSetting(row rowScanner) (AppSetting, error) {
	var value AppSetting
	var sensitive int
	var raw string
	var updatedAt int64
	err := row.Scan(&value.Key, &raw, &sensitive, &updatedAt)
	if errors.Is(err, sql.ErrNoRows) {
		return AppSetting{}, ErrNotFound
	}
	if err != nil {
		return AppSetting{}, err
	}
	value.Value = []byte(raw)
	value.Sensitive = sensitive != 0
	value.UpdatedAt = time.Unix(updatedAt, 0).UTC()
	return value, nil
}

func maskedJSONValue(value json.RawMessage) bool {
	if bytes.Equal(bytes.TrimSpace(value), []byte(`null`)) {
		return true
	}
	var text string
	if json.Unmarshal(value, &text) == nil {
		return text == "" || text == SecretMask
	}
	return false
}

func (s *Store) UpsertCardPolicy(ctx context.Context, value CardPolicy) error {
	value.ICCID = strings.TrimSpace(value.ICCID)
	if value.ICCID == "" {
		return errors.New("card policy ICCID is required")
	}
	value.IPVersion = strings.ToUpper(strings.TrimSpace(value.IPVersion))
	switch value.IPVersion {
	case "", "IP", "IPV6", "IPV4V6":
	default:
		return fmt.Errorf("unsupported card policy IP version %q", value.IPVersion)
	}
	if value.VoWiFiEnabled && value.AirplaneEnabled {
		return errors.New("VoWiFi and airplane mode cannot both be enabled")
	}
	now := time.Now().UTC()
	createdAt := value.CreatedAt
	if createdAt.IsZero() {
		createdAt = now
	}
	updatedAt := value.UpdatedAt
	if updatedAt.IsZero() {
		updatedAt = now
	}
	_, err := s.db.ExecContext(ctx, `
		INSERT INTO card_policies (
			iccid, network_enabled, vowifi_enabled, airplane_enabled,
			apn, ip_version, source, created_at, updated_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(iccid) DO UPDATE SET
			network_enabled = excluded.network_enabled,
			vowifi_enabled = excluded.vowifi_enabled,
			airplane_enabled = excluded.airplane_enabled,
			apn = excluded.apn,
			ip_version = excluded.ip_version,
			source = excluded.source,
			updated_at = excluded.updated_at
	`,
		value.ICCID, boolInt(value.NetworkEnabled), boolInt(value.VoWiFiEnabled),
		boolInt(value.AirplaneEnabled), value.APN, value.IPVersion,
		value.Source, createdAt.Unix(), updatedAt.Unix(),
	)
	if err != nil {
		return fmt.Errorf("upsert card policy %q: %w", value.ICCID, err)
	}
	return nil
}

func (s *Store) CardPolicy(ctx context.Context, iccid string) (CardPolicy, error) {
	return cardPolicy(s.db.QueryRowContext(
		ctx,
		cardPolicySelect+` WHERE iccid = ?`,
		strings.TrimSpace(iccid),
	))
}

func (s *Store) ListCardPolicies(ctx context.Context) ([]CardPolicy, error) {
	rows, err := s.db.QueryContext(ctx, cardPolicySelect+` ORDER BY iccid`)
	if err != nil {
		return nil, fmt.Errorf("list card policies: %w", err)
	}
	defer rows.Close()
	values := make([]CardPolicy, 0)
	for rows.Next() {
		value, err := cardPolicy(rows)
		if err != nil {
			return nil, fmt.Errorf("scan card policy: %w", err)
		}
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate card policies: %w", err)
	}
	return values, nil
}

func (s *Store) DeleteCardPolicy(ctx context.Context, iccid string) error {
	result, err := s.db.ExecContext(ctx, `DELETE FROM card_policies WHERE iccid = ?`, iccid)
	if err != nil {
		return fmt.Errorf("delete card policy %q: %w", iccid, err)
	}
	return requireAffected(result)
}

const cardPolicySelect = `
	SELECT iccid, network_enabled, vowifi_enabled, airplane_enabled,
		apn, ip_version, source, created_at, updated_at
	FROM card_policies`

func cardPolicy(row rowScanner) (CardPolicy, error) {
	var value CardPolicy
	var networkEnabled, vowifiEnabled, airplaneEnabled int
	var createdAt, updatedAt int64
	err := row.Scan(
		&value.ICCID, &networkEnabled, &vowifiEnabled, &airplaneEnabled,
		&value.APN, &value.IPVersion, &value.Source, &createdAt, &updatedAt,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return CardPolicy{}, ErrNotFound
	}
	if err != nil {
		return CardPolicy{}, err
	}
	value.NetworkEnabled = networkEnabled != 0
	value.VoWiFiEnabled = vowifiEnabled != 0
	value.AirplaneEnabled = airplaneEnabled != 0
	value.CreatedAt = time.Unix(createdAt, 0).UTC()
	value.UpdatedAt = time.Unix(updatedAt, 0).UTC()
	return value, nil
}

func (s *Store) UpsertTrafficBucket(ctx context.Context, value TrafficBucket) error {
	return s.writeTrafficBucket(ctx, value, false)
}

// AddTrafficBucket atomically accumulates counters for concurrent collectors.
func (s *Store) AddTrafficBucket(ctx context.Context, value TrafficBucket) error {
	return s.writeTrafficBucket(ctx, value, true)
}

func (s *Store) writeTrafficBucket(
	ctx context.Context,
	value TrafficBucket,
	accumulate bool,
) error {
	value.DeviceID = strings.TrimSpace(value.DeviceID)
	value.Bucket = strings.TrimSpace(value.Bucket)
	if value.DeviceID == "" || value.Bucket == "" {
		return errors.New("traffic bucket device id and bucket are required")
	}
	if value.PeriodStart.IsZero() {
		return errors.New("traffic bucket period start is required")
	}
	if value.RXBytes < 0 || value.TXBytes < 0 {
		return errors.New("traffic byte counters cannot be negative")
	}
	update := `
		rx_bytes = excluded.rx_bytes,
		tx_bytes = excluded.tx_bytes`
	if accumulate {
		update = `
			rx_bytes = traffic_buckets.rx_bytes + excluded.rx_bytes,
			tx_bytes = traffic_buckets.tx_bytes + excluded.tx_bytes`
	}
	_, err := s.db.ExecContext(ctx, `
		INSERT INTO traffic_buckets (
			device_id, bucket, period_start, rx_bytes, tx_bytes
		) VALUES (?, ?, ?, ?, ?)
		ON CONFLICT(device_id, bucket, period_start) DO UPDATE SET`+update,
		value.DeviceID, value.Bucket, value.PeriodStart.UTC().Unix(),
		value.RXBytes, value.TXBytes,
	)
	if err != nil {
		return fmt.Errorf("write traffic bucket: %w", err)
	}
	return nil
}

func (s *Store) ListTrafficBuckets(
	ctx context.Context,
	filter TrafficFilter,
) ([]TrafficBucket, error) {
	clauses := make([]string, 0, 4)
	args := make([]any, 0, 5)
	if filter.DeviceID != "" {
		clauses = append(clauses, `device_id = ?`)
		args = append(args, filter.DeviceID)
	}
	if filter.Bucket != "" {
		clauses = append(clauses, `bucket = ?`)
		args = append(args, filter.Bucket)
	}
	if !filter.Since.IsZero() {
		clauses = append(clauses, `period_start >= ?`)
		args = append(args, filter.Since.UTC().Unix())
	}
	if !filter.Until.IsZero() {
		clauses = append(clauses, `period_start < ?`)
		args = append(args, filter.Until.UTC().Unix())
	}
	query := `
		SELECT device_id, bucket, period_start, rx_bytes, tx_bytes
		FROM traffic_buckets`
	if len(clauses) > 0 {
		query += ` WHERE ` + strings.Join(clauses, ` AND `)
	}
	query += ` ORDER BY period_start ASC, device_id LIMIT ?`
	args = append(args, normalizedLimit(filter.Limit))
	rows, err := s.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("list traffic buckets: %w", err)
	}
	defer rows.Close()
	values := make([]TrafficBucket, 0)
	for rows.Next() {
		var value TrafficBucket
		var periodStart int64
		if err := rows.Scan(
			&value.DeviceID, &value.Bucket, &periodStart,
			&value.RXBytes, &value.TXBytes,
		); err != nil {
			return nil, fmt.Errorf("scan traffic bucket: %w", err)
		}
		value.PeriodStart = time.Unix(periodStart, 0).UTC()
		values = append(values, value)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate traffic buckets: %w", err)
	}
	return values, nil
}

func (s *Store) DeleteTrafficBefore(ctx context.Context, before time.Time) (int64, error) {
	result, err := s.db.ExecContext(
		ctx,
		`DELETE FROM traffic_buckets WHERE period_start < ?`,
		before.UTC().Unix(),
	)
	if err != nil {
		return 0, fmt.Errorf("delete old traffic buckets: %w", err)
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return 0, fmt.Errorf("read deleted traffic bucket count: %w", err)
	}
	return affected, nil
}