renjue
fe8ea784d0
CI / docker (push) Successful in 2m54s
Co-authored-by: Cursor <cursoragent@cursor.com>
509 lines
16 KiB
Go
509 lines
16 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
)
|
|
|
|
func (s *Store) ListSubscriptions(ctx context.Context) ([]Subscription, error) {
|
|
rows, err := s.db.QueryContext(ctx, `SELECT id,name,url,type,interval_sec,enabled,last_fetch_at,last_error,node_count,created_at,updated_at FROM subscriptions ORDER BY id`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []Subscription
|
|
for rows.Next() {
|
|
var sub Subscription
|
|
var enabled, nodeCount int
|
|
var lastFetch, lastErr sql.NullString
|
|
var created, updated string
|
|
if err := rows.Scan(&sub.ID, &sub.Name, &sub.URL, &sub.Type, &sub.IntervalSec, &enabled, &lastFetch, &lastErr, &nodeCount, &created, &updated); err != nil {
|
|
return nil, err
|
|
}
|
|
sub.Enabled = boolFromInt(enabled)
|
|
sub.NodeCount = nodeCount
|
|
sub.LastFetchAt = parseTimePtr(lastFetch)
|
|
sub.LastError = lastErr.String
|
|
sub.CreatedAt = parseTime(created)
|
|
sub.UpdatedAt = parseTime(updated)
|
|
out = append(out, sub)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetSubscription(ctx context.Context, id int64) (*Subscription, error) {
|
|
var sub Subscription
|
|
var enabled, nodeCount int
|
|
var lastFetch, lastErr sql.NullString
|
|
var created, updated string
|
|
err := s.db.QueryRowContext(ctx, `SELECT id,name,url,type,interval_sec,enabled,last_fetch_at,last_error,node_count,created_at,updated_at FROM subscriptions WHERE id=?`, id).
|
|
Scan(&sub.ID, &sub.Name, &sub.URL, &sub.Type, &sub.IntervalSec, &enabled, &lastFetch, &lastErr, &nodeCount, &created, &updated)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sub.Enabled = boolFromInt(enabled)
|
|
sub.NodeCount = nodeCount
|
|
sub.LastFetchAt = parseTimePtr(lastFetch)
|
|
sub.LastError = lastErr.String
|
|
sub.CreatedAt = parseTime(created)
|
|
sub.UpdatedAt = parseTime(updated)
|
|
return &sub, nil
|
|
}
|
|
|
|
func (s *Store) CreateSubscription(ctx context.Context, sub *Subscription) error {
|
|
res, err := s.db.ExecContext(ctx, `INSERT INTO subscriptions(name,url,type,interval_sec,enabled) VALUES(?,?,?,?,?)`,
|
|
sub.Name, sub.URL, sub.Type, sub.IntervalSec, intFromBool(sub.Enabled))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id, _ := res.LastInsertId()
|
|
sub.ID = id
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) UpdateSubscription(ctx context.Context, sub *Subscription) error {
|
|
_, err := s.db.ExecContext(ctx, `UPDATE subscriptions SET name=?,url=?,type=?,interval_sec=?,enabled=?,updated_at=datetime('now') WHERE id=?`,
|
|
sub.Name, sub.URL, sub.Type, sub.IntervalSec, intFromBool(sub.Enabled), sub.ID)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) DeleteSubscription(ctx context.Context, id int64) error {
|
|
_, err := s.db.ExecContext(ctx, `DELETE FROM subscriptions WHERE id=?`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) UpdateSubscriptionFetch(ctx context.Context, id int64, nodeCount int, lastError string) error {
|
|
_, err := s.db.ExecContext(ctx, `UPDATE subscriptions SET last_fetch_at=datetime('now'), last_error=?, node_count=?, updated_at=datetime('now') WHERE id=?`,
|
|
lastError, nodeCount, id)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) proxyNodesWhere(source string, subscriptionID *int64, country string) (string, []any) {
|
|
query := ` FROM proxy_nodes WHERE 1=1`
|
|
args := []any{}
|
|
if source != "" {
|
|
query += ` AND source = ?`
|
|
args = append(args, source)
|
|
}
|
|
if subscriptionID != nil {
|
|
query += ` AND subscription_id = ?`
|
|
args = append(args, *subscriptionID)
|
|
}
|
|
if country != "" {
|
|
query += ` AND country = ?`
|
|
args = append(args, country)
|
|
}
|
|
return query, args
|
|
}
|
|
|
|
func (s *Store) CountProxyNodes(ctx context.Context, source string, subscriptionID *int64, country string) (int, error) {
|
|
where, args := s.proxyNodesWhere(source, subscriptionID, country)
|
|
var n int
|
|
err := s.db.QueryRowContext(ctx, `SELECT COUNT(*)`+where, args...).Scan(&n)
|
|
return n, err
|
|
}
|
|
|
|
func (s *Store) ListProxyNodes(ctx context.Context, source string, subscriptionID *int64) ([]ProxyNode, error) {
|
|
return s.ListProxyNodesPage(ctx, source, subscriptionID, "", 0, 0)
|
|
}
|
|
|
|
func (s *Store) ListProxyNodesPage(ctx context.Context, source string, subscriptionID *int64, country string, limit, offset int) ([]ProxyNode, error) {
|
|
where, args := s.proxyNodesWhere(source, subscriptionID, country)
|
|
query := `SELECT id,subscription_id,name,runtime_name,type,raw_config,source,enabled,country,latency_ms,alive,checked_at,created_at,updated_at` + where + ` ORDER BY id`
|
|
if limit > 0 {
|
|
query += ` LIMIT ? OFFSET ?`
|
|
args = append(args, limit, offset)
|
|
}
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
return scanProxyNodes(rows)
|
|
}
|
|
|
|
func (s *Store) GetProxyNode(ctx context.Context, id int64) (*ProxyNode, error) {
|
|
row := s.db.QueryRowContext(ctx, `SELECT id,subscription_id,name,runtime_name,type,raw_config,source,enabled,country,latency_ms,alive,checked_at,created_at,updated_at FROM proxy_nodes WHERE id=?`, id)
|
|
return scanProxyNode(row)
|
|
}
|
|
|
|
func (s *Store) CreateProxyNode(ctx context.Context, n *ProxyNode) error {
|
|
res, err := s.db.ExecContext(ctx, `INSERT INTO proxy_nodes(subscription_id,name,runtime_name,type,raw_config,source,enabled,country) VALUES(?,?,?,?,?,?,?,?)`,
|
|
n.SubscriptionID, n.Name, n.RuntimeName, n.Type, n.RawConfig, n.Source, intFromBool(n.Enabled), n.Country)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id, _ := res.LastInsertId()
|
|
n.ID = id
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) UpdateProxyNode(ctx context.Context, n *ProxyNode) error {
|
|
_, err := s.db.ExecContext(ctx, `UPDATE proxy_nodes SET name=?,runtime_name=?,type=?,raw_config=?,enabled=?,country=?,updated_at=datetime('now') WHERE id=?`,
|
|
n.Name, n.RuntimeName, n.Type, n.RawConfig, intFromBool(n.Enabled), n.Country, n.ID)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) DeleteProxyNode(ctx context.Context, id int64) error {
|
|
_, err := s.db.ExecContext(ctx, `DELETE FROM proxy_nodes WHERE id=?`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) ReplaceManualProxies(ctx context.Context, nodes []ProxyNode) error {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM proxy_nodes WHERE source='manual'`); err != nil {
|
|
return err
|
|
}
|
|
for _, n := range nodes {
|
|
n.Source = "manual"
|
|
if _, err := tx.ExecContext(ctx,
|
|
`INSERT INTO proxy_nodes(subscription_id,name,runtime_name,type,raw_config,source,enabled,country) VALUES(?,?,?,?,?,?,?,?)`,
|
|
n.SubscriptionID, n.Name, n.RuntimeName, n.Type, n.RawConfig, "manual", intFromBool(n.Enabled), n.Country); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *Store) AppendManualProxies(ctx context.Context, nodes []ProxyNode) error {
|
|
for i := range nodes {
|
|
nodes[i].Source = "manual"
|
|
if err := s.CreateProxyNode(ctx, &nodes[i]); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) UpdateProxyNodeHealth(ctx context.Context, id int64, latency int, alive bool) error {
|
|
_, err := s.db.ExecContext(ctx, `UPDATE proxy_nodes SET latency_ms=?, alive=?, checked_at=datetime('now'), updated_at=datetime('now') WHERE id=?`,
|
|
latency, intFromBool(alive), id)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) ReplaceSubscriptionData(ctx context.Context, subID int64, nodes []ProxyNode, groups []ProxyGroup, rules []Rule) error {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM proxy_nodes WHERE subscription_id=?`, subID); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM proxy_groups WHERE subscription_id=?`, subID); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM rules WHERE subscription_id=?`, subID); err != nil {
|
|
return err
|
|
}
|
|
|
|
for _, n := range nodes {
|
|
n.SubscriptionID = &subID
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO proxy_nodes(subscription_id,name,runtime_name,type,raw_config,source,enabled,country) VALUES(?,?,?,?,?,?,?,?)`,
|
|
subID, n.Name, n.RuntimeName, n.Type, n.RawConfig, "subscription", intFromBool(n.Enabled), n.Country); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
for _, g := range groups {
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO proxy_groups(name,runtime_name,type,proxies,source,subscription_id,enabled) VALUES(?,?,?,?,?,?,?)`,
|
|
g.Name, g.RuntimeName, g.Type, g.Proxies, "subscription", subID, intFromBool(g.Enabled)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
basePriority := 10000
|
|
for i, r := range rules {
|
|
r.Priority = basePriority + i
|
|
r.SubscriptionID = &subID
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO rules(priority,rule_type,payload,target,source,subscription_id,enabled) VALUES(?,?,?,?,?,?,?)`,
|
|
r.Priority, r.RuleType, r.Payload, r.Target, "subscription", subID, intFromBool(r.Enabled)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func scanProxyNodes(rows *sql.Rows) ([]ProxyNode, error) {
|
|
var out []ProxyNode
|
|
for rows.Next() {
|
|
n, err := scanProxyNodeFromRows(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, *n)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func scanProxyNode(row *sql.Row) (*ProxyNode, error) {
|
|
// reuse via a mini adapter - sql.Row doesn't share with Rows
|
|
var n ProxyNode
|
|
var subID sql.NullInt64
|
|
var enabled, alive int
|
|
var checked sql.NullString
|
|
var created, updated string
|
|
err := row.Scan(&n.ID, &subID, &n.Name, &n.RuntimeName, &n.Type, &n.RawConfig, &n.Source, &enabled, &n.Country, &n.LatencyMs, &alive, &checked, &created, &updated)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if subID.Valid {
|
|
n.SubscriptionID = &subID.Int64
|
|
}
|
|
n.Enabled = boolFromInt(enabled)
|
|
n.Alive = boolFromInt(alive)
|
|
n.CheckedAt = parseTimePtr(checked)
|
|
n.CreatedAt = parseTime(created)
|
|
n.UpdatedAt = parseTime(updated)
|
|
return &n, nil
|
|
}
|
|
|
|
func scanProxyNodeFromRows(rows *sql.Rows) (*ProxyNode, error) {
|
|
var n ProxyNode
|
|
var subID sql.NullInt64
|
|
var enabled, alive int
|
|
var checked sql.NullString
|
|
var created, updated string
|
|
if err := rows.Scan(&n.ID, &subID, &n.Name, &n.RuntimeName, &n.Type, &n.RawConfig, &n.Source, &enabled, &n.Country, &n.LatencyMs, &alive, &checked, &created, &updated); err != nil {
|
|
return nil, err
|
|
}
|
|
if subID.Valid {
|
|
n.SubscriptionID = &subID.Int64
|
|
}
|
|
n.Enabled = boolFromInt(enabled)
|
|
n.Alive = boolFromInt(alive)
|
|
n.CheckedAt = parseTimePtr(checked)
|
|
n.CreatedAt = parseTime(created)
|
|
n.UpdatedAt = parseTime(updated)
|
|
return &n, nil
|
|
}
|
|
|
|
func (s *Store) ListRules(ctx context.Context, source string) ([]Rule, error) {
|
|
query := `SELECT id,priority,rule_type,payload,target,source,subscription_id,enabled,created_at,updated_at FROM rules WHERE 1=1`
|
|
args := []any{}
|
|
if source != "" {
|
|
query += ` AND source = ?`
|
|
args = append(args, source)
|
|
}
|
|
query += ` ORDER BY priority ASC, id ASC`
|
|
rows, err := s.db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []Rule
|
|
for rows.Next() {
|
|
var r Rule
|
|
var subID sql.NullInt64
|
|
var enabled int
|
|
var created, updated string
|
|
if err := rows.Scan(&r.ID, &r.Priority, &r.RuleType, &r.Payload, &r.Target, &r.Source, &subID, &enabled, &created, &updated); err != nil {
|
|
return nil, err
|
|
}
|
|
if subID.Valid {
|
|
r.SubscriptionID = &subID.Int64
|
|
}
|
|
r.Enabled = boolFromInt(enabled)
|
|
r.CreatedAt = parseTime(created)
|
|
r.UpdatedAt = parseTime(updated)
|
|
out = append(out, r)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) GetRule(ctx context.Context, id int64) (*Rule, error) {
|
|
var r Rule
|
|
var subID sql.NullInt64
|
|
var enabled int
|
|
var created, updated string
|
|
err := s.db.QueryRowContext(ctx, `SELECT id,priority,rule_type,payload,target,source,subscription_id,enabled,created_at,updated_at FROM rules WHERE id=?`, id).
|
|
Scan(&r.ID, &r.Priority, &r.RuleType, &r.Payload, &r.Target, &r.Source, &subID, &enabled, &created, &updated)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if subID.Valid {
|
|
r.SubscriptionID = &subID.Int64
|
|
}
|
|
r.Enabled = boolFromInt(enabled)
|
|
r.CreatedAt = parseTime(created)
|
|
r.UpdatedAt = parseTime(updated)
|
|
return &r, nil
|
|
}
|
|
|
|
func (s *Store) CreateRule(ctx context.Context, r *Rule) error {
|
|
res, err := s.db.ExecContext(ctx, `INSERT INTO rules(priority,rule_type,payload,target,source,subscription_id,enabled) VALUES(?,?,?,?,?,?,?)`,
|
|
r.Priority, r.RuleType, r.Payload, r.Target, r.Source, r.SubscriptionID, intFromBool(r.Enabled))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id, _ := res.LastInsertId()
|
|
r.ID = id
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) UpdateRule(ctx context.Context, r *Rule) error {
|
|
_, err := s.db.ExecContext(ctx, `UPDATE rules SET priority=?,rule_type=?,payload=?,target=?,enabled=?,updated_at=datetime('now') WHERE id=?`,
|
|
r.Priority, r.RuleType, r.Payload, r.Target, intFromBool(r.Enabled), r.ID)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) DeleteRule(ctx context.Context, id int64) error {
|
|
_, err := s.db.ExecContext(ctx, `DELETE FROM rules WHERE id=? AND source='manual'`, id)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) ReplaceManualRules(ctx context.Context, rules []Rule) error {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
if _, err := tx.ExecContext(ctx, `DELETE FROM rules WHERE source='manual'`); err != nil {
|
|
return err
|
|
}
|
|
for _, r := range rules {
|
|
r.Source = "manual"
|
|
if _, err := tx.ExecContext(ctx,
|
|
`INSERT INTO rules(priority,rule_type,payload,target,source,subscription_id,enabled) VALUES(?,?,?,?,?,?,?)`,
|
|
r.Priority, r.RuleType, r.Payload, r.Target, "manual", nil, intFromBool(r.Enabled)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *Store) AppendManualRules(ctx context.Context, rules []Rule) error {
|
|
for i := range rules {
|
|
rules[i].Source = "manual"
|
|
if err := s.CreateRule(ctx, &rules[i]); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) ListProxyGroups(ctx context.Context) ([]ProxyGroup, error) {
|
|
rows, err := s.db.QueryContext(ctx, `SELECT id,name,runtime_name,type,proxies,source,subscription_id,enabled,created_at,updated_at FROM proxy_groups ORDER BY id`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []ProxyGroup
|
|
for rows.Next() {
|
|
var g ProxyGroup
|
|
var subID sql.NullInt64
|
|
var enabled int
|
|
var created, updated string
|
|
if err := rows.Scan(&g.ID, &g.Name, &g.RuntimeName, &g.Type, &g.Proxies, &g.Source, &subID, &enabled, &created, &updated); err != nil {
|
|
return nil, err
|
|
}
|
|
if subID.Valid {
|
|
g.SubscriptionID = &subID.Int64
|
|
}
|
|
g.Enabled = boolFromInt(enabled)
|
|
g.CreatedAt = parseTime(created)
|
|
g.UpdatedAt = parseTime(updated)
|
|
out = append(out, g)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) CountEnabledProxies(ctx context.Context) (total, alive int, err error) {
|
|
err = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM proxy_nodes WHERE enabled=1`).Scan(&total)
|
|
if err != nil {
|
|
return
|
|
}
|
|
err = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM proxy_nodes WHERE enabled=1 AND alive=1`).Scan(&alive)
|
|
return
|
|
}
|
|
|
|
func (s *Store) NextManualRulePriority(ctx context.Context) (int, error) {
|
|
var max sql.NullInt64
|
|
err := s.db.QueryRowContext(ctx, `SELECT MAX(priority) FROM rules WHERE source='manual'`).Scan(&max)
|
|
if err != nil {
|
|
return 100, err
|
|
}
|
|
if !max.Valid || max.Int64 >= 9999 {
|
|
return 100, nil
|
|
}
|
|
return int(max.Int64) + 10, nil
|
|
}
|
|
|
|
func (s *Store) ListMergedRules(ctx context.Context, defaultPolicy string) ([]MergedRule, error) {
|
|
rules, err := s.ListRules(ctx, "")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return buildMergedRules(rules, defaultPolicy), nil
|
|
}
|
|
|
|
func (s *Store) CountMergedRules(ctx context.Context) (int, error) {
|
|
var n int
|
|
err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM rules WHERE enabled=1`).Scan(&n)
|
|
return n, err
|
|
}
|
|
|
|
func (s *Store) ListMergedRulesPage(ctx context.Context, limit, offset int) ([]MergedRule, error) {
|
|
if limit <= 0 {
|
|
limit = 50
|
|
}
|
|
if limit > 500 {
|
|
limit = 500
|
|
}
|
|
if offset < 0 {
|
|
offset = 0
|
|
}
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT priority,rule_type,payload,target,source FROM rules
|
|
WHERE enabled=1
|
|
ORDER BY priority ASC, id ASC
|
|
LIMIT ? OFFSET ?`, limit, offset)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []MergedRule
|
|
for rows.Next() {
|
|
var r Rule
|
|
if err := rows.Scan(&r.Priority, &r.RuleType, &r.Payload, &r.Target, &r.Source); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, MergedRule{
|
|
Priority: r.Priority,
|
|
Rule: formatRuleLine(r),
|
|
Source: r.Source,
|
|
})
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func buildMergedRules(rules []Rule, defaultPolicy string) []MergedRule {
|
|
var out []MergedRule
|
|
for _, r := range rules {
|
|
if !r.Enabled {
|
|
continue
|
|
}
|
|
out = append(out, MergedRule{
|
|
Priority: r.Priority,
|
|
Rule: formatRuleLine(r),
|
|
Source: r.Source,
|
|
})
|
|
}
|
|
if defaultPolicy == "" {
|
|
defaultPolicy = "DIRECT"
|
|
}
|
|
out = append(out, MergedRule{Priority: 999999, Rule: "MATCH," + defaultPolicy, Source: "system"})
|
|
return out
|
|
}
|
|
|
|
func formatRuleLine(r Rule) string {
|
|
if r.Payload == "" {
|
|
return fmt.Sprintf("%s,%s", r.RuleType, r.Target)
|
|
}
|
|
return fmt.Sprintf("%s,%s,%s", r.RuleType, r.Payload, r.Target)
|
|
}
|