diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 87d9872..06bdb00 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -75,7 +75,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v4 with: - go-version: "1.21" + go-version: "1.24" check-latest: true cache: false @@ -91,7 +91,7 @@ jobs: GOARCH: ${{ matrix.arch }} CGO_ENABLED: 0 run: | - go build -ldflags="-s -w" -o ${{ matrix.binary_name }} ./cmd/aproxy + go build -ldflags="-s -w -X main.Version=${{ github.ref_name }}" -o ${{ matrix.binary_name }} ./cmd/aproxy - name: Upload artifact uses: actions/upload-artifact@v4 diff --git a/CLAUDE.md b/CLAUDE.md index 129516a..ed41ebd 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -65,10 +65,10 @@ docker-compose up ### Core Components -- **Scraper** (`pkg/scraper/`): Fetches proxy lists from multiple sources (ProxyScrape, FreeProxyList, Geonode, ProxyListOrg, GitHub) +- **Scraper** (`pkg/scraper/`): Fetches proxy lists from multiple sources (ProxyScrape, FreeProxyList, ProxyListOrg, GitHub) - **Checker** (`pkg/checker/`): Validates proxy health with SQLite caching, intelligent check intervals, and unified logging - **Manager** (`pkg/manager/`): Manages proxy pool with database persistence, in-memory cache, and auto-refresh -- **Database** (`internal/database/`): SQLite-based persistent storage with Jet ORM for type-safe queries +- **Database** (`internal/database/`): SQLite-based persistent storage with sqlc-generated type-safe queries - **Proxy Server** (`pkg/proxy/`): HTTP/HTTPS proxy server with privacy features - **Config** (`internal/config/`): Advanced configuration management with Viper and validation support @@ -121,10 +121,9 @@ AProxy uses **Viper** for advanced configuration management with validation: 3. **Config files**: YAML, JSON, TOML supported (searches `./`, `./config/`, `/etc/aproxy/`) 4. **Defaults**: Sensible defaults for all settings -**Supported Scraper Sources:** +**Supported Scraper Sources:** (all are plain `host:port` / `proto://host:port` text lists) - `proxyscrape`: ProxyScrape API - `freeproxylist`: FreeProxyList scraper -- `geonode`: Geonode API scraper - `proxylistorg`: ProxyListOrg scraper - `github`: GitHub proxy list scraper (proxifly/free-proxy-list) @@ -180,12 +179,13 @@ curl -x http://localhost:8080 \ - `checker.test_url`: URL used to test proxy health (default: `http://icanhazip.com`) **Scraper Configuration Options:** -- `scraper.sources`: List of proxy sources to use (default: `["proxyscrape", "freeproxylist", "geonode", "github"]`) +- `scraper.sources`: List of proxy sources to use (default: `["proxyscrape", "freeproxylist", "github"]`) - `scraper.timeout`: Request timeout for scraping (default: `30s`) - `scraper.user_agent`: User agent string for scraper requests **Logging Configuration:** -- Currently logs to stdout only in JSON format +- Logs to stdout as JSON via `log/slog` (each line carries `component`, and `id` for correlated operations) +- Level via `LOG_LEVEL` env var (`debug`/`info`/`warn`/`error`, default `info`) - Log level can be controlled via command line or environment variables - File-based logging is not yet implemented @@ -215,45 +215,39 @@ The SQLite database includes: - **Indexes**: Optimized for fast lookups by host:port, status, and timestamps - **Automatic cleanup**: Removes old unhealthy proxies based on configuration -## Recent Improvements - -### GitHub Proxy Scraper (v1.2) -- **New GitHub source**: Added scraper for proxifly/free-proxy-list GitHub repository -- **Enhanced source variety**: Now supports 5 different proxy sources for better diversity -- **Configuration validation**: Added `github` to allowed scraper sources in config validation - -### Logging System Improvements (v1.2) -- **Unified logging**: Standardized all checker logging to use internal logger package -- **Reduced verbosity**: Removed verbose individual proxy failure debugging output -- **Consistent log levels**: Proper use of InfoBg/WarnBg throughout checker components -- **Better performance**: Less logging overhead during proxy health checks - -### Configuration System (v1.1) -- **Migrated to Viper**: Replaced manual config parsing with Viper library -- **Added validation**: All config values validated using `go-playground/validator` -- **YAML support**: Config files now use YAML format (JSON/TOML also supported) -- **Better error messages**: Detailed validation errors with field names and constraints - -### Docker Support -- **Production-ready**: Multi-stage Docker build with Alpine Linux -- **Security**: Non-root user, minimal attack surface -- **Health checks**: Proper health endpoint with curl-based Docker healthchecks -- **Persistent volumes**: Database and logs stored in `./data/` for volume mounting -- **Resource limits**: CPU and memory constraints in docker-compose - -### Performance Optimizations -- **Non-blocking architecture**: Server starts immediately, proxy checking happens in background -- **Progressive checking**: Proxies checked in small batches with delays to reduce system load -- **Intelligent caching**: Only checks proxies older than 10 minutes, persists all results including failures -- **Fixed race conditions**: HTTPS CONNECT bidirectional copying now uses channel coordination -- **Batch database updates**: Replaced concurrent individual updates with single transaction batches -- **Removed GitHub sources**: Eliminated high-volume proxy sources to reduce database load -- **Improved caching**: Better SQLite connection pooling and prepared statements +## Extending the Codebase + +### Adding a new proxy source +Most sources are plain text lists, so you don't write a new file — add a row to the `sources` registry in `pkg/scraper/list.go`: +1. Append a `source{name, urls, defaultType}` to the `sources` slice in `pkg/scraper/list.go`. `parseLine` already handles both `proto://host:port` and bare `host:port` lines. +2. Add `` to the `oneof=...` validator tag on `Scraper.Sources` in `internal/config/config.go` (otherwise config validation rejects it). +3. Optionally add it to the `scraper.sources` default list in `setDefaults` (`internal/config/config.go`). + +For a non-text source (custom JSON API, etc.), implement the `Scraper` interface (`pkg/scraper/types.go`) in its own file and append an instance in `NewMultiScraperWithConfig`. `MultiScraper.ScrapeAll` runs all configured sources, dedups, and aggregates; the manager hands the result to the checker. + +### Database queries are sqlc-generated +- The data layer uses **sqlc** (`sqlc.yaml`). SQL lives in `internal/database/schema.sql` (table defs) and `internal/database/query.sql` (named queries). Run `sqlc generate` to regenerate `internal/database/db/` — do not hand-edit those generated files. +- `internal/database/service.go` wraps the generated `db.Queries` with the methods the app calls. The one hand-written query is `GetProxiesByAddresses` (sqlc's sqlite engine can't do `sqlc.slice()` IN-lists). +- ⚠️ The schema lives in **two** places that must stay in sync: `schema.sql` (read by sqlc) and the inline string in `db.go`'s `initSchema()` (run at startup). Change both. + +### Logging conventions +Use the internal `logger` package (`internal/logger/logger.go`), not the standard `log`, for component logging: +- `logger.New("")` creates a component-scoped logger. +- ID-tagged methods (`Info`/`Warn`/`Error`/`Debug`) take a request/operation ID from `logger.GenerateID()` to correlate a multi-step operation's log lines. +- `*Bg` variants (`InfoBg`/`WarnBg`/`ErrorBg`/`DebugBg`) are for background/non-correlated operations — used throughout the checker. + +## Request Flow + +1. `main.go` loads config (Viper) → opens SQLite DB → builds `scraper.ScraperConfig` and `checker.CheckerConfig`. +2. `manager.NewDBManagerWithConfig` wires together the multi-scraper, the DB-backed checker, and the in-memory cache; `mgr.Start(updateInterval)` launches background refresh. +3. `RefreshProxies` (manager) calls `scraper.ScrapeAll` → `dbChecker.CheckProxiesWithCaching` (skips proxies checked within `check_interval`) → persists results to SQLite → reloads the healthy in-memory cache. +4. `proxy.NewServer(mgr, cfg)` serves requests; `GetNextProxy`/`GetRandomProxy` pull from the cache, `ReportProxyFailure` feeds failures back. +5. Server starts immediately on cached proxies — the pool fills progressively in the background (non-blocking startup). ### Key Dependencies - `github.com/spf13/viper`: Configuration management - `github.com/go-playground/validator/v10`: Configuration validation -- `github.com/go-jet/jet/v2`: Type-safe SQL queries +- `sqlc` (build-time, not a runtime dep): generates the type-safe query layer in `internal/database/db/` - `modernc.org/sqlite`: Pure Go SQLite driver ## Development Notes diff --git a/cmd/aproxy/main.go b/cmd/aproxy/main.go index a1bd719..6d57c17 100644 --- a/cmd/aproxy/main.go +++ b/cmd/aproxy/main.go @@ -4,7 +4,6 @@ import ( "context" "flag" "fmt" - "log" "os" "os/signal" "syscall" @@ -12,10 +11,9 @@ import ( "aproxy/internal/config" "aproxy/internal/database" - "aproxy/pkg/checker" + "aproxy/internal/logger" "aproxy/pkg/manager" "aproxy/pkg/proxy" - "aproxy/pkg/scraper" ) var ( @@ -24,9 +22,8 @@ var ( version = flag.Bool("version", false, "Show version") ) -var ( - Version = "1.0.0" -) +// Version is set at build time via -ldflags "-X main.Version=...". +var Version = "dev" const ( Banner = ` @@ -44,7 +41,7 @@ ______ ______ ______ ______ ______ ______ ______ ______ ░ ░ ______ ______ ______ ______ ______ ______ ______ ______ -AProxy - Anonymous Proxy Server v%s +AProxy - Anonymous Proxy Server %s https://github.com/ArnabXD/aproxy ______ ______ ______ ______ ______ ______ ______ ______ @@ -56,15 +53,17 @@ func main() { flag.Parse() if *version { - fmt.Printf("AProxy v%s\n", Version) + fmt.Printf("AProxy %s\n", Version) return } fmt.Printf(Banner, Version) + log := logger.New("main") + if *genConfig { if err := config.SaveConfigTemplate("config.yaml"); err != nil { - log.Fatalf("Failed to generate config: %v", err) + log.Fatal("Failed to generate config: %v", err) } fmt.Println("Default config generated: config.yaml") return @@ -72,68 +71,40 @@ func main() { cfg, err := config.LoadConfig(*configPath) if err != nil { - log.Fatalf("Failed to load config: %v", err) + log.Fatal("Failed to load config: %v", err) } - log.Printf("Starting AProxy v%s", Version) + log.InfoBg("Starting AProxy %s", Version) config.PrintConfig(cfg) // Initialize database db, err := database.NewDB(cfg.Database.Path) if err != nil { - log.Fatalf("Failed to initialize database: %v", err) + log.Fatal("Failed to initialize database: %v", err) } defer db.Close() - // Create configuration objects for checker and scraper - scraperConfig := scraper.ScraperConfig{ - Timeout: cfg.Scraper.Timeout, - UserAgent: cfg.Scraper.UserAgent, - Sources: cfg.Scraper.Sources, - } - - checkerConfig := checker.CheckerConfig{ - TestURL: cfg.Checker.TestURL, - Timeout: cfg.Checker.Timeout, - MaxWorkers: cfg.Checker.MaxWorkers, - UserAgent: cfg.Checker.UserAgent, - } - - // Use database manager with configuration - mgr := manager.NewDBManagerWithConfig(db, scraperConfig, checkerConfig, cfg.Checker.CheckInterval, cfg.Checker.BackgroundEnabled, cfg.Checker.BatchSize, cfg.Checker.BatchDelay) + mgr := manager.NewDBManager(db, cfg) if err := mgr.Start(cfg.Proxy.UpdateInterval); err != nil { - log.Fatalf("Failed to start proxy manager: %v", err) - } - - proxyConfig := &proxy.Config{ - ListenAddr: cfg.Server.ListenAddr, - ReadTimeout: cfg.Server.ReadTimeout, - WriteTimeout: cfg.Server.WriteTimeout, - IdleTimeout: cfg.Server.IdleTimeout, - MaxConnections: cfg.Server.MaxConnections, - EnableHTTPS: cfg.Server.EnableHTTPS, - MaxRetries: cfg.Server.MaxRetries, - StripHeaders: cfg.Server.StripHeaders, - AddHeaders: cfg.Server.AddHeaders, - AuthToken: cfg.Server.AuthToken, + log.Fatal("Failed to start proxy manager: %v", err) } - server := proxy.NewServer(mgr, proxyConfig) + server := proxy.NewServer(mgr, cfg.Server) go func() { if err := server.Start(); err != nil { - log.Printf("Server error: %v", err) + log.ErrorBg("Server error: %v", err) } }() - log.Printf("Proxy server started on %s", cfg.Server.ListenAddr) - log.Println("Press Ctrl+C to stop") + log.InfoBg("Proxy server started on %s", cfg.Server.ListenAddr) + log.InfoBg("Press Ctrl+C to stop") c := make(chan os.Signal, 1) signal.Notify(c, os.Interrupt, syscall.SIGTERM) <-c - log.Println("Shutting down...") + log.InfoBg("Shutting down...") // Stop the manager first to cancel background operations mgr.Stop() @@ -143,8 +114,8 @@ func main() { defer cancel() if err := server.Stop(ctx); err != nil { - log.Printf("Server shutdown error: %v", err) + log.ErrorBg("Server shutdown error: %v", err) } - log.Println("Shutdown complete") + log.InfoBg("Shutdown complete") } diff --git a/config.example.yaml b/config.example.yaml index 16ad069..4a1450c 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -31,7 +31,7 @@ scraper: sources: - "proxyscrape" - "freeproxylist" - - "geonode" + - "github" checker: test_url: "http://icanhazip.com" diff --git a/go.mod b/go.mod index 1da59c0..af4f0e6 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,6 @@ module aproxy go 1.24.1 require ( - github.com/go-jet/jet/v2 v2.13.0 github.com/go-playground/validator/v10 v10.26.0 github.com/spf13/viper v1.20.1 golang.org/x/net v0.38.0 @@ -11,26 +10,24 @@ require ( ) require ( - github.com/davecgh/go-spew v1.1.1 // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/fsnotify/fsnotify v1.8.0 // indirect github.com/gabriel-vasile/mimetype v1.4.8 // indirect github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-viper/mapstructure/v2 v2.3.0 // indirect + github.com/google/go-cmp v0.7.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/ncruces/go-strftime v0.1.9 // indirect github.com/pelletier/go-toml/v2 v2.2.3 // indirect - github.com/pmezard/go-difflib v1.0.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/sagikazarmark/locafero v0.7.0 // indirect github.com/sourcegraph/conc v0.3.0 // indirect github.com/spf13/afero v1.12.0 // indirect github.com/spf13/cast v1.7.1 // indirect github.com/spf13/pflag v1.0.6 // indirect - github.com/stretchr/testify v1.10.0 // indirect github.com/subosito/gotenv v1.6.0 // indirect go.uber.org/atomic v1.9.0 // indirect go.uber.org/multierr v1.9.0 // indirect diff --git a/go.sum b/go.sum index f50dc02..65309bd 100644 --- a/go.sum +++ b/go.sum @@ -9,8 +9,6 @@ github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/ github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= github.com/gabriel-vasile/mimetype v1.4.8 h1:FfZ3gj38NjllZIeJAmMhr+qKL8Wu+nOoI3GqacKw1NM= github.com/gabriel-vasile/mimetype v1.4.8/go.mod h1:ByKUIKGjh1ODkGM1asKUbQZOLGrPjydw3hYPU2YU9t8= -github.com/go-jet/jet/v2 v2.13.0 h1:DcD2IJRGos+4X40IQRV6S6q9onoOfZY/GPdvU6ImZcQ= -github.com/go-jet/jet/v2 v2.13.0/go.mod h1:YhT75U1FoYAxFOObbQliHmXVYQeffkBKWT7ZilZ3zPc= github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA= diff --git a/internal/config/config.go b/internal/config/config.go index 7c57fa9..d0c7e57 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,15 +2,18 @@ package config import ( "fmt" - "log" "os" "strings" "time" + "aproxy/internal/logger" + "github.com/go-playground/validator/v10" "github.com/spf13/viper" ) +var log = logger.New("config") + type Config struct { Server ServerConfig `mapstructure:"server" validate:"required"` Proxy ProxyConfig `mapstructure:"proxy" validate:"required"` @@ -41,7 +44,7 @@ type ProxyConfig struct { type ScraperConfig struct { Timeout time.Duration `mapstructure:"timeout" validate:"required,min=5s,max=2m"` UserAgent string `mapstructure:"user_agent" validate:"required,min=10"` - Sources []string `mapstructure:"sources" validate:"required,min=1,dive,oneof=proxyscrape freeproxylist geonode proxylistorg github"` + Sources []string `mapstructure:"sources" validate:"required,min=1,dive,oneof=proxyscrape freeproxylist proxylistorg github"` } type CheckerConfig struct { @@ -87,7 +90,7 @@ func setDefaults() { // Scraper defaults viper.SetDefault("scraper.timeout", "30s") viper.SetDefault("scraper.user_agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36") - viper.SetDefault("scraper.sources", []string{"proxyscrape", "freeproxylist", "geonode", "github"}) + viper.SetDefault("scraper.sources", []string{"proxyscrape", "freeproxylist", "github"}) // Checker defaults viper.SetDefault("checker.test_url", "http://icanhazip.com") @@ -128,7 +131,7 @@ func LoadConfig(configPath string) (*Config, error) { viper.SetConfigFile(".env") viper.SetConfigType("env") if err := viper.MergeInConfig(); err != nil { - log.Printf("Warning: Failed to load .env file: %v", err) + log.WarnBg("Failed to load .env file: %v", err) } } @@ -142,7 +145,7 @@ func LoadConfig(configPath string) (*Config, error) { return nil, fmt.Errorf("failed to read config file: %w", err) } // Config file not found, use defaults and env vars - log.Println("No config file found, using defaults and environment variables") + log.InfoBg("No config file found, using defaults and environment variables") } // Unmarshal configuration @@ -191,18 +194,18 @@ func SaveConfigTemplate(path string) error { return nil } -// PrintConfig displays the current configuration (for debugging) +// PrintConfig logs the loaded configuration (for debugging). func PrintConfig(config *Config) { - log.Printf("Configuration loaded:") - log.Printf(" Server: %s (HTTPS: %v)", config.Server.ListenAddr, config.Server.EnableHTTPS) + authToken := "[NOT SET]" if config.Server.AuthToken != "" { - log.Printf(" Auth Token: [SET] (length: %d)", len(config.Server.AuthToken)) - } else { - log.Printf(" Auth Token: [NOT SET]") + authToken = fmt.Sprintf("[SET] (length: %d)", len(config.Server.AuthToken)) } - log.Printf(" Database: %s (Max Age: %v)", config.Database.Path, config.Database.MaxAge) - log.Printf(" Proxy Update: %v (Max Failures: %d)", config.Proxy.UpdateInterval, config.Proxy.MaxFailures) - log.Printf(" Checker: %d workers, %v timeout, batch size: %d, batch delay: %v, background: %v", - config.Checker.MaxWorkers, config.Checker.Timeout, config.Checker.BatchSize, config.Checker.BatchDelay, config.Checker.BackgroundEnabled) - log.Printf(" Scraper Sources: %v", config.Scraper.Sources) + log.InfoBg("Configuration loaded: server=%s https=%v auth=%s db=%s maxAge=%v "+ + "proxyUpdate=%v maxFailures=%d checker=%dw/%v batch=%d/%v bg=%v sources=%v", + config.Server.ListenAddr, config.Server.EnableHTTPS, authToken, + config.Database.Path, config.Database.MaxAge, + config.Proxy.UpdateInterval, config.Proxy.MaxFailures, + config.Checker.MaxWorkers, config.Checker.Timeout, + config.Checker.BatchSize, config.Checker.BatchDelay, config.Checker.BackgroundEnabled, + config.Scraper.Sources) } diff --git a/internal/database/db/db.go b/internal/database/db/db.go new file mode 100644 index 0000000..f43598b --- /dev/null +++ b/internal/database/db/db.go @@ -0,0 +1,31 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 + +package db + +import ( + "context" + "database/sql" +) + +type DBTX interface { + ExecContext(context.Context, string, ...interface{}) (sql.Result, error) + PrepareContext(context.Context, string) (*sql.Stmt, error) + QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error) + QueryRowContext(context.Context, string, ...interface{}) *sql.Row +} + +func New(db DBTX) *Queries { + return &Queries{db: db} +} + +type Queries struct { + db DBTX +} + +func (q *Queries) WithTx(tx *sql.Tx) *Queries { + return &Queries{ + db: tx, + } +} diff --git a/internal/database/db/models.go b/internal/database/db/models.go new file mode 100644 index 0000000..95d7a92 --- /dev/null +++ b/internal/database/db/models.go @@ -0,0 +1,25 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 + +package db + +import ( + "time" +) + +type Proxy struct { + ID int64 + Host string + Port int64 + ProxyType string + Country *string + Anonymity *string + Https *bool + Status string + ResponseTimeMs *int64 + FailCount *int64 + FirstSeenAt time.Time + LastCheckedAt *time.Time + LastHealthyAt *time.Time +} diff --git a/internal/database/db/query.sql.go b/internal/database/db/query.sql.go new file mode 100644 index 0000000..0aaf449 --- /dev/null +++ b/internal/database/db/query.sql.go @@ -0,0 +1,227 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: query.sql + +package db + +import ( + "context" + "time" +) + +const cleanupOldProxies = `-- name: CleanupOldProxies :exec +DELETE FROM proxies +WHERE last_healthy_at IS NULL OR last_healthy_at < ? +` + +func (q *Queries) CleanupOldProxies(ctx context.Context, lastHealthyAt *time.Time) error { + _, err := q.db.ExecContext(ctx, cleanupOldProxies, lastHealthyAt) + return err +} + +const countHealthyProxies = `-- name: CountHealthyProxies :one +SELECT COUNT(*) FROM proxies WHERE status = 'healthy' +` + +func (q *Queries) CountHealthyProxies(ctx context.Context) (int64, error) { + row := q.db.QueryRowContext(ctx, countHealthyProxies) + var count int64 + err := row.Scan(&count) + return count, err +} + +const countProxies = `-- name: CountProxies :one +SELECT COUNT(*) FROM proxies +` + +func (q *Queries) CountProxies(ctx context.Context) (int64, error) { + row := q.db.QueryRowContext(ctx, countProxies) + var count int64 + err := row.Scan(&count) + return count, err +} + +const countProxiesByType = `-- name: CountProxiesByType :many +SELECT proxy_type, COUNT(*) AS count FROM proxies GROUP BY proxy_type +` + +type CountProxiesByTypeRow struct { + ProxyType string + Count int64 +} + +func (q *Queries) CountProxiesByType(ctx context.Context) ([]CountProxiesByTypeRow, error) { + rows, err := q.db.QueryContext(ctx, countProxiesByType) + if err != nil { + return nil, err + } + defer rows.Close() + var items []CountProxiesByTypeRow + for rows.Next() { + var i CountProxiesByTypeRow + if err := rows.Scan(&i.ProxyType, &i.Count); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getHealthyProxies = `-- name: GetHealthyProxies :many +SELECT id, host, port, proxy_type, country, anonymity, https, status, response_time_ms, fail_count, first_seen_at, last_checked_at, last_healthy_at FROM proxies +WHERE status = 'healthy' +ORDER BY last_healthy_at DESC +` + +func (q *Queries) GetHealthyProxies(ctx context.Context) ([]Proxy, error) { + rows, err := q.db.QueryContext(ctx, getHealthyProxies) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Proxy + for rows.Next() { + var i Proxy + if err := rows.Scan( + &i.ID, + &i.Host, + &i.Port, + &i.ProxyType, + &i.Country, + &i.Anonymity, + &i.Https, + &i.Status, + &i.ResponseTimeMs, + &i.FailCount, + &i.FirstSeenAt, + &i.LastCheckedAt, + &i.LastHealthyAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getProxyByHostPort = `-- name: GetProxyByHostPort :one +SELECT id, host, port, proxy_type, country, anonymity, https, status, response_time_ms, fail_count, first_seen_at, last_checked_at, last_healthy_at FROM proxies +WHERE host = ? AND port = ? +` + +type GetProxyByHostPortParams struct { + Host string + Port int64 +} + +func (q *Queries) GetProxyByHostPort(ctx context.Context, arg GetProxyByHostPortParams) (Proxy, error) { + row := q.db.QueryRowContext(ctx, getProxyByHostPort, arg.Host, arg.Port) + var i Proxy + err := row.Scan( + &i.ID, + &i.Host, + &i.Port, + &i.ProxyType, + &i.Country, + &i.Anonymity, + &i.Https, + &i.Status, + &i.ResponseTimeMs, + &i.FailCount, + &i.FirstSeenAt, + &i.LastCheckedAt, + &i.LastHealthyAt, + ) + return i, err +} + +const markProxyHealthy = `-- name: MarkProxyHealthy :exec +UPDATE proxies +SET status = ?, last_checked_at = CURRENT_TIMESTAMP, response_time_ms = ?, + last_healthy_at = CURRENT_TIMESTAMP, fail_count = 0 +WHERE id = ? +` + +type MarkProxyHealthyParams struct { + Status string + ResponseTimeMs *int64 + ID int64 +} + +func (q *Queries) MarkProxyHealthy(ctx context.Context, arg MarkProxyHealthyParams) error { + _, err := q.db.ExecContext(ctx, markProxyHealthy, arg.Status, arg.ResponseTimeMs, arg.ID) + return err +} + +const markProxyUnhealthy = `-- name: MarkProxyUnhealthy :exec +UPDATE proxies +SET status = ?, last_checked_at = CURRENT_TIMESTAMP, response_time_ms = ?, + fail_count = fail_count + 1 +WHERE id = ? +` + +type MarkProxyUnhealthyParams struct { + Status string + ResponseTimeMs *int64 + ID int64 +} + +func (q *Queries) MarkProxyUnhealthy(ctx context.Context, arg MarkProxyUnhealthyParams) error { + _, err := q.db.ExecContext(ctx, markProxyUnhealthy, arg.Status, arg.ResponseTimeMs, arg.ID) + return err +} + +const upsertProxy = `-- name: UpsertProxy :one +INSERT INTO proxies (host, port, proxy_type, country, first_seen_at) +VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) +ON CONFLICT(host, port) DO UPDATE SET + proxy_type = excluded.proxy_type, + country = excluded.country +RETURNING id, host, port, proxy_type, country, anonymity, https, status, response_time_ms, fail_count, first_seen_at, last_checked_at, last_healthy_at +` + +type UpsertProxyParams struct { + Host string + Port int64 + ProxyType string + Country *string +} + +func (q *Queries) UpsertProxy(ctx context.Context, arg UpsertProxyParams) (Proxy, error) { + row := q.db.QueryRowContext(ctx, upsertProxy, + arg.Host, + arg.Port, + arg.ProxyType, + arg.Country, + ) + var i Proxy + err := row.Scan( + &i.ID, + &i.Host, + &i.Port, + &i.ProxyType, + &i.Country, + &i.Anonymity, + &i.Https, + &i.Status, + &i.ResponseTimeMs, + &i.FailCount, + &i.FirstSeenAt, + &i.LastCheckedAt, + &i.LastHealthyAt, + ) + return i, err +} diff --git a/internal/database/models/model/proxies.go b/internal/database/models/model/proxies.go deleted file mode 100644 index 01a2b2c..0000000 --- a/internal/database/models/model/proxies.go +++ /dev/null @@ -1,28 +0,0 @@ -// -// Code generated by go-jet DO NOT EDIT. -// -// WARNING: Changes to this file may cause incorrect behavior -// and will be lost if the code is regenerated -// - -package model - -import ( - "time" -) - -type Proxies struct { - ID *int32 `sql:"primary_key"` - Host string - Port int32 - ProxyType string - Country *string - Anonymity *string - HTTPS *bool - Status string - ResponseTimeMs *int32 - FailCount *int32 - FirstSeenAt time.Time - LastCheckedAt *time.Time - LastHealthyAt *time.Time -} diff --git a/internal/database/models/table/proxies.go b/internal/database/models/table/proxies.go deleted file mode 100644 index 0138683..0000000 --- a/internal/database/models/table/proxies.go +++ /dev/null @@ -1,114 +0,0 @@ -// -// Code generated by go-jet DO NOT EDIT. -// -// WARNING: Changes to this file may cause incorrect behavior -// and will be lost if the code is regenerated -// - -package table - -import ( - "github.com/go-jet/jet/v2/sqlite" -) - -var Proxies = newProxiesTable("", "proxies", "") - -type proxiesTable struct { - sqlite.Table - - // Columns - ID sqlite.ColumnInteger - Host sqlite.ColumnString - Port sqlite.ColumnInteger - ProxyType sqlite.ColumnString - Country sqlite.ColumnString - Anonymity sqlite.ColumnString - HTTPS sqlite.ColumnBool - Status sqlite.ColumnString - ResponseTimeMs sqlite.ColumnInteger - FailCount sqlite.ColumnInteger - FirstSeenAt sqlite.ColumnTimestamp - LastCheckedAt sqlite.ColumnTimestamp - LastHealthyAt sqlite.ColumnTimestamp - - AllColumns sqlite.ColumnList - MutableColumns sqlite.ColumnList - DefaultColumns sqlite.ColumnList -} - -type ProxiesTable struct { - proxiesTable - - EXCLUDED proxiesTable -} - -// AS creates new ProxiesTable with assigned alias -func (a ProxiesTable) AS(alias string) *ProxiesTable { - return newProxiesTable(a.SchemaName(), a.TableName(), alias) -} - -// Schema creates new ProxiesTable with assigned schema name -func (a ProxiesTable) FromSchema(schemaName string) *ProxiesTable { - return newProxiesTable(schemaName, a.TableName(), a.Alias()) -} - -// WithPrefix creates new ProxiesTable with assigned table prefix -func (a ProxiesTable) WithPrefix(prefix string) *ProxiesTable { - return newProxiesTable(a.SchemaName(), prefix+a.TableName(), a.TableName()) -} - -// WithSuffix creates new ProxiesTable with assigned table suffix -func (a ProxiesTable) WithSuffix(suffix string) *ProxiesTable { - return newProxiesTable(a.SchemaName(), a.TableName()+suffix, a.TableName()) -} - -func newProxiesTable(schemaName, tableName, alias string) *ProxiesTable { - return &ProxiesTable{ - proxiesTable: newProxiesTableImpl(schemaName, tableName, alias), - EXCLUDED: newProxiesTableImpl("", "excluded", ""), - } -} - -func newProxiesTableImpl(schemaName, tableName, alias string) proxiesTable { - var ( - IDColumn = sqlite.IntegerColumn("id") - HostColumn = sqlite.StringColumn("host") - PortColumn = sqlite.IntegerColumn("port") - ProxyTypeColumn = sqlite.StringColumn("proxy_type") - CountryColumn = sqlite.StringColumn("country") - AnonymityColumn = sqlite.StringColumn("anonymity") - HTTPSColumn = sqlite.BoolColumn("https") - StatusColumn = sqlite.StringColumn("status") - ResponseTimeMsColumn = sqlite.IntegerColumn("response_time_ms") - FailCountColumn = sqlite.IntegerColumn("fail_count") - FirstSeenAtColumn = sqlite.TimestampColumn("first_seen_at") - LastCheckedAtColumn = sqlite.TimestampColumn("last_checked_at") - LastHealthyAtColumn = sqlite.TimestampColumn("last_healthy_at") - allColumns = sqlite.ColumnList{IDColumn, HostColumn, PortColumn, ProxyTypeColumn, CountryColumn, AnonymityColumn, HTTPSColumn, StatusColumn, ResponseTimeMsColumn, FailCountColumn, FirstSeenAtColumn, LastCheckedAtColumn, LastHealthyAtColumn} - mutableColumns = sqlite.ColumnList{HostColumn, PortColumn, ProxyTypeColumn, CountryColumn, AnonymityColumn, HTTPSColumn, StatusColumn, ResponseTimeMsColumn, FailCountColumn, FirstSeenAtColumn, LastCheckedAtColumn, LastHealthyAtColumn} - defaultColumns = sqlite.ColumnList{HTTPSColumn, StatusColumn, FailCountColumn, FirstSeenAtColumn} - ) - - return proxiesTable{ - Table: sqlite.NewTable(schemaName, tableName, alias, allColumns...), - - //Columns - ID: IDColumn, - Host: HostColumn, - Port: PortColumn, - ProxyType: ProxyTypeColumn, - Country: CountryColumn, - Anonymity: AnonymityColumn, - HTTPS: HTTPSColumn, - Status: StatusColumn, - ResponseTimeMs: ResponseTimeMsColumn, - FailCount: FailCountColumn, - FirstSeenAt: FirstSeenAtColumn, - LastCheckedAt: LastCheckedAtColumn, - LastHealthyAt: LastHealthyAtColumn, - - AllColumns: allColumns, - MutableColumns: mutableColumns, - DefaultColumns: defaultColumns, - } -} diff --git a/internal/database/models/table/table_use_schema.go b/internal/database/models/table/table_use_schema.go deleted file mode 100644 index d3dd734..0000000 --- a/internal/database/models/table/table_use_schema.go +++ /dev/null @@ -1,14 +0,0 @@ -// -// Code generated by go-jet DO NOT EDIT. -// -// WARNING: Changes to this file may cause incorrect behavior -// and will be lost if the code is regenerated -// - -package table - -// UseSchema sets a new schema name for all generated table SQL builder types. It is recommended to invoke -// this method only once at the beginning of the program. -func UseSchema(schema string) { - Proxies = Proxies.FromSchema(schema) -} diff --git a/internal/database/query.sql b/internal/database/query.sql new file mode 100644 index 0000000..5dca1c4 --- /dev/null +++ b/internal/database/query.sql @@ -0,0 +1,41 @@ +-- name: UpsertProxy :one +INSERT INTO proxies (host, port, proxy_type, country, first_seen_at) +VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) +ON CONFLICT(host, port) DO UPDATE SET + proxy_type = excluded.proxy_type, + country = excluded.country +RETURNING *; + +-- name: GetHealthyProxies :many +SELECT * FROM proxies +WHERE status = 'healthy' +ORDER BY last_healthy_at DESC; + +-- name: GetProxyByHostPort :one +SELECT * FROM proxies +WHERE host = ? AND port = ?; + +-- name: MarkProxyHealthy :exec +UPDATE proxies +SET status = ?, last_checked_at = CURRENT_TIMESTAMP, response_time_ms = ?, + last_healthy_at = CURRENT_TIMESTAMP, fail_count = 0 +WHERE id = ?; + +-- name: MarkProxyUnhealthy :exec +UPDATE proxies +SET status = ?, last_checked_at = CURRENT_TIMESTAMP, response_time_ms = ?, + fail_count = fail_count + 1 +WHERE id = ?; + +-- name: CleanupOldProxies :exec +DELETE FROM proxies +WHERE last_healthy_at IS NULL OR last_healthy_at < ?; + +-- name: CountProxies :one +SELECT COUNT(*) FROM proxies; + +-- name: CountHealthyProxies :one +SELECT COUNT(*) FROM proxies WHERE status = 'healthy'; + +-- name: CountProxiesByType :many +SELECT proxy_type, COUNT(*) AS count FROM proxies GROUP BY proxy_type; diff --git a/internal/database/schema.sql b/internal/database/schema.sql new file mode 100644 index 0000000..11406ab --- /dev/null +++ b/internal/database/schema.sql @@ -0,0 +1,20 @@ +-- Proxy storage and caching schema. Kept in sync with db.go initSchema(). +CREATE TABLE proxies ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + host TEXT NOT NULL, + port INTEGER NOT NULL, + proxy_type TEXT NOT NULL, + country TEXT, + anonymity TEXT, + https BOOLEAN DEFAULT 0, + + status TEXT NOT NULL DEFAULT 'unknown', + response_time_ms INTEGER, + fail_count INTEGER DEFAULT 0, + + first_seen_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + last_checked_at DATETIME, + last_healthy_at DATETIME, + + UNIQUE(host, port) +); diff --git a/internal/database/service.go b/internal/database/service.go index d28764f..4a53dec 100644 --- a/internal/database/service.go +++ b/internal/database/service.go @@ -4,212 +4,81 @@ import ( "context" "database/sql" "fmt" - "log" "strings" "time" - "aproxy/internal/database/models/model" - "aproxy/internal/database/models/table" + "aproxy/internal/database/db" "aproxy/pkg/scraper" - - . "github.com/go-jet/jet/v2/sqlite" ) -// Service handles database operations for proxies +// Proxy is the stored proxy row (re-exported sqlc model). +type Proxy = db.Proxy + +// Service handles database operations for proxies. type Service struct { + q *db.Queries db *DB } -// NewService creates a new database service -func NewService(db *DB) *Service { - return &Service{db: db} +// NewService creates a new database service. +func NewService(database *DB) *Service { + return &Service{q: db.New(database), db: database} } -// UpsertProxy inserts or updates a proxy in the database -func (s *Service) UpsertProxy(ctx context.Context, proxy scraper.Proxy) (*model.Proxies, error) { +// UpsertProxy inserts a proxy or, on host:port conflict, refreshes its metadata +// (preserving health/timestamp columns). +func (s *Service) UpsertProxy(ctx context.Context, proxy scraper.Proxy) (*Proxy, error) { country := proxy.Country - proxyModel := model.Proxies{ + p, err := s.q.UpsertProxy(ctx, db.UpsertProxyParams{ Host: proxy.Host, - Port: int32(proxy.Port), + Port: int64(proxy.Port), ProxyType: proxy.Type, Country: &country, - HTTPS: nil, // Not available in scraper.Proxy - } - - // Try to insert, if it fails due to unique constraint, update only metadata (preserve health data) - now := time.Now() - stmt := table.Proxies.INSERT( - table.Proxies.Host, - table.Proxies.Port, - table.Proxies.ProxyType, - table.Proxies.Country, - table.Proxies.FirstSeenAt, - ).VALUES( - proxyModel.Host, - proxyModel.Port, - proxyModel.ProxyType, - proxyModel.Country, - String(now.Format("2006-01-02 15:04:05")), - ).ON_CONFLICT(table.Proxies.Host, table.Proxies.Port).DO_UPDATE(SET( - table.Proxies.ProxyType.SET(String(proxyModel.ProxyType)), - table.Proxies.Country.SET(String(*proxyModel.Country)), - // DO NOT update timestamps - preserve existing health check data - )).RETURNING(table.Proxies.AllColumns) - - var result model.Proxies - err := stmt.QueryContext(ctx, s.db, &result) + }) if err != nil { return nil, fmt.Errorf("failed to upsert proxy: %w", err) } - - return &result, nil + return &p, nil } -// GetProxiesNeedingCheck returns proxies that haven't been checked in the last checkInterval -func (s *Service) GetProxiesNeedingCheck(ctx context.Context, checkInterval time.Duration) ([]model.Proxies, error) { - cutoff := time.Now().Add(-checkInterval) - - query := ` - SELECT id, host, port, proxy_type, country, anonymity, https, status, response_time_ms, fail_count, first_seen_at, last_checked_at, last_healthy_at - FROM proxies - WHERE last_checked_at IS NULL OR last_checked_at < ? - ` - - rows, err := s.db.QueryContext(ctx, query, cutoff.Format("2006-01-02 15:04:05")) +// GetHealthyProxies returns all healthy proxies. +func (s *Service) GetHealthyProxies(ctx context.Context) ([]Proxy, error) { + proxies, err := s.q.GetHealthyProxies(ctx) if err != nil { - return nil, fmt.Errorf("failed to get proxies needing check: %w", err) - } - defer rows.Close() - - var proxies []model.Proxies - for rows.Next() { - var p model.Proxies - err := rows.Scan( - &p.ID, &p.Host, &p.Port, &p.ProxyType, &p.Country, &p.Anonymity, - &p.HTTPS, &p.Status, &p.ResponseTimeMs, &p.FailCount, - &p.FirstSeenAt, &p.LastCheckedAt, &p.LastHealthyAt, - ) - if err != nil { - return nil, fmt.Errorf("failed to scan proxy: %w", err) - } - proxies = append(proxies, p) + return nil, fmt.Errorf("failed to get healthy proxies: %w", err) } - return proxies, nil } -// BatchUpdateProxyHealth updates multiple proxy health statuses in a single transaction -func (s *Service) BatchUpdateProxyHealth(ctx context.Context, updates map[int32]CheckResult) error { - if len(updates) == 0 { - return nil - } - - tx, err := s.db.BeginTx(ctx, nil) - if err != nil { - return fmt.Errorf("failed to begin transaction: %w", err) - } - defer tx.Rollback() - - now := time.Now() - nowStr := now.Format("2006-01-02 15:04:05") - - // Prepare statements for healthy and unhealthy updates - healthyQuery := ` - UPDATE proxies - SET status = ?, last_checked_at = ?, response_time_ms = ?, last_healthy_at = ?, fail_count = 0 - WHERE id = ? - ` - unhealthyQuery := ` - UPDATE proxies - SET status = ?, last_checked_at = ?, response_time_ms = ?, fail_count = fail_count + 1 - WHERE id = ? - ` - - healthyStmt, err := tx.Prepare(healthyQuery) - if err != nil { - return fmt.Errorf("failed to prepare healthy statement: %w", err) - } - defer healthyStmt.Close() - - unhealthyStmt, err := tx.Prepare(unhealthyQuery) +// GetProxyByHostPort finds a proxy by host and port, or returns (nil, nil). +func (s *Service) GetProxyByHostPort(ctx context.Context, host string, port int) (*Proxy, error) { + p, err := s.q.GetProxyByHostPort(ctx, db.GetProxyByHostPortParams{Host: host, Port: int64(port)}) if err != nil { - return fmt.Errorf("failed to prepare unhealthy statement: %w", err) - } - defer unhealthyStmt.Close() - - // Execute all updates - for proxyID, result := range updates { - if result.Status == StatusHealthy { - _, err = healthyStmt.Exec( - result.Status.String(), - nowStr, - int32(result.ResponseTime.Milliseconds()), - nowStr, - proxyID, - ) - } else { - _, err = unhealthyStmt.Exec( - result.Status.String(), - nowStr, - int32(result.ResponseTime.Milliseconds()), - proxyID, - ) - } - - if err != nil { - return fmt.Errorf("failed to update proxy %d: %w", proxyID, err) + if err == sql.ErrNoRows { + return nil, nil } + return nil, fmt.Errorf("failed to get proxy: %w", err) } - - // Commit the transaction - if err = tx.Commit(); err != nil { - return fmt.Errorf("failed to commit transaction: %w", err) - } - - log.Printf("Batch updated %d proxy health records", len(updates)) - return nil + return &p, nil } -// GetHealthyProxies returns all healthy proxies -func (s *Service) GetHealthyProxies(ctx context.Context) ([]model.Proxies, error) { - stmt := SELECT( - table.Proxies.AllColumns, - ).FROM( - table.Proxies, - ).WHERE( - table.Proxies.Status.EQ(String("healthy")), - ).ORDER_BY( - table.Proxies.LastHealthyAt.DESC(), - ) - - var proxies []model.Proxies - err := stmt.QueryContext(ctx, s.db, &proxies) - if err != nil { - return nil, fmt.Errorf("failed to get healthy proxies: %w", err) - } - - return proxies, nil -} - -// GetProxiesByAddresses returns existing proxies for the given host:port addresses -func (s *Service) GetProxiesByAddresses(ctx context.Context, addresses []string) (map[string]*model.Proxies, error) { +// GetProxiesByAddresses returns existing proxies for the given host:port keys. +// Hand-written: sqlc's sqlite engine doesn't support sqlc.slice() for IN lists. +func (s *Service) GetProxiesByAddresses(ctx context.Context, addresses []string) (map[string]*Proxy, error) { + result := make(map[string]*Proxy) if len(addresses) == 0 { - return make(map[string]*model.Proxies), nil + return result, nil } - // Build the query with placeholders - query := ` - SELECT id, host, port, proxy_type, country, anonymity, https, status, response_time_ms, fail_count, first_seen_at, last_checked_at, last_healthy_at - FROM proxies - WHERE (host || ':' || port) IN (` - - args := make([]interface{}, len(addresses)) + args := make([]any, len(addresses)) placeholders := make([]string, len(addresses)) for i, addr := range addresses { placeholders[i] = "?" args[i] = addr } - query += strings.Join(placeholders, ",") + ")" + query := `SELECT id, host, port, proxy_type, country, anonymity, https, status, + response_time_ms, fail_count, first_seen_at, last_checked_at, last_healthy_at + FROM proxies WHERE (host || ':' || port) IN (` + strings.Join(placeholders, ",") + ")" rows, err := s.db.QueryContext(ctx, query, args...) if err != nil { @@ -217,99 +86,93 @@ func (s *Service) GetProxiesByAddresses(ctx context.Context, addresses []string) } defer rows.Close() - result := make(map[string]*model.Proxies) for rows.Next() { - var p model.Proxies - err := rows.Scan( + var p Proxy + if err := rows.Scan( &p.ID, &p.Host, &p.Port, &p.ProxyType, &p.Country, &p.Anonymity, - &p.HTTPS, &p.Status, &p.ResponseTimeMs, &p.FailCount, + &p.Https, &p.Status, &p.ResponseTimeMs, &p.FailCount, &p.FirstSeenAt, &p.LastCheckedAt, &p.LastHealthyAt, - ) - if err != nil { + ); err != nil { return nil, fmt.Errorf("failed to scan proxy: %w", err) } - - address := fmt.Sprintf("%s:%d", p.Host, p.Port) - result[address] = &p + result[fmt.Sprintf("%s:%d", p.Host, p.Port)] = &p } - - return result, nil + return result, rows.Err() } -// GetProxyByHostPort finds a proxy by host and port -func (s *Service) GetProxyByHostPort(ctx context.Context, host string, port int) (*model.Proxies, error) { - stmt := SELECT( - table.Proxies.AllColumns, - ).FROM( - table.Proxies, - ).WHERE( - table.Proxies.Host.EQ(String(host)). - AND(table.Proxies.Port.EQ(Int32(int32(port)))), - ) +// BatchUpdateProxyHealth updates many proxies' health in one transaction. +func (s *Service) BatchUpdateProxyHealth(ctx context.Context, updates map[int32]CheckResult) error { + if len(updates) == 0 { + return nil + } - var proxy model.Proxies - err := stmt.QueryContext(ctx, s.db, &proxy) + tx, err := s.db.BeginTx(ctx, nil) if err != nil { - if err == sql.ErrNoRows { - return nil, nil + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer tx.Rollback() + + qtx := s.q.WithTx(tx) + for id, result := range updates { + rt := int64(result.ResponseTime.Milliseconds()) + if result.Status == StatusHealthy { + err = qtx.MarkProxyHealthy(ctx, db.MarkProxyHealthyParams{ + Status: result.Status.String(), ResponseTimeMs: &rt, ID: int64(id), + }) + } else { + err = qtx.MarkProxyUnhealthy(ctx, db.MarkProxyUnhealthyParams{ + Status: result.Status.String(), ResponseTimeMs: &rt, ID: int64(id), + }) + } + if err != nil { + return fmt.Errorf("failed to update proxy %d: %w", id, err) } - return nil, fmt.Errorf("failed to get proxy: %w", err) } - return &proxy, nil + if err = tx.Commit(); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + return nil } -// CleanupOldProxies removes proxies that haven't been healthy for a long time +// CleanupOldProxies removes proxies that haven't been healthy since maxAge ago. func (s *Service) CleanupOldProxies(ctx context.Context, maxAge time.Duration) error { cutoff := time.Now().Add(-maxAge) - - query := `DELETE FROM proxies WHERE last_healthy_at IS NULL OR last_healthy_at < ?` - - _, err := s.db.ExecContext(ctx, query, cutoff.Format("2006-01-02 15:04:05")) - if err != nil { + if err := s.q.CleanupOldProxies(ctx, &cutoff); err != nil { return fmt.Errorf("failed to cleanup old proxies: %w", err) } - return nil } -// GetProxyStats returns statistics about the proxy database +// GetProxyStats returns aggregate statistics about the proxy table. func (s *Service) GetProxyStats(ctx context.Context) (ProxyStats, error) { var stats ProxyStats - // Count total proxies using raw SQL - err := s.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM proxies").Scan(&stats.Total) + total, err := s.q.CountProxies(ctx) if err != nil { return stats, fmt.Errorf("failed to count total proxies: %w", err) } + stats.Total = int(total) - // Count healthy proxies using raw SQL - err = s.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM proxies WHERE status = 'healthy'").Scan(&stats.Healthy) + healthy, err := s.q.CountHealthyProxies(ctx) if err != nil { return stats, fmt.Errorf("failed to count healthy proxies: %w", err) } + stats.Healthy = int(healthy) - // Count by type using raw SQL - rows, err := s.db.QueryContext(ctx, "SELECT proxy_type, COUNT(*) FROM proxies GROUP BY proxy_type") + byType, err := s.q.CountProxiesByType(ctx) if err != nil { - return stats, fmt.Errorf("failed to get proxy types: %w", err) + return stats, fmt.Errorf("failed to count proxies by type: %w", err) } - defer rows.Close() - - stats.ByType = make(map[string]int) - for rows.Next() { - var proxyType string - var count int - if err := rows.Scan(&proxyType, &count); err != nil { - return stats, fmt.Errorf("failed to scan proxy type row: %w", err) - } - stats.ByType[proxyType] = count + stats.ByType = make(map[string]int, len(byType)) + for _, row := range byType { + stats.ByType[row.ProxyType] = int(row.Count) } return stats, nil } -// ProxyStats contains statistics about the proxy database +// ProxyStats contains statistics about the proxy database. type ProxyStats struct { Total int `json:"total"` Healthy int `json:"healthy"` diff --git a/internal/logger/logger.go b/internal/logger/logger.go index 417bf0a..a8a31f7 100644 --- a/internal/logger/logger.go +++ b/internal/logger/logger.go @@ -1,77 +1,64 @@ package logger import ( + "context" "crypto/rand" "encoding/hex" "fmt" - "log" + "log/slog" + "os" ) -// Logger provides structured logging across the application -type Logger struct { - component string -} +// slog handler shared by all component loggers. JSON to stdout, honoring LOG_LEVEL. +var handler = slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: levelFromEnv()}) -// New creates a new logger for a specific component -func New(component string) *Logger { - return &Logger{component: component} +func levelFromEnv() slog.Level { + switch os.Getenv("LOG_LEVEL") { + case "debug", "DEBUG": + return slog.LevelDebug + case "warn", "WARN": + return slog.LevelWarn + case "error", "ERROR": + return slog.LevelError + default: + return slog.LevelInfo + } } -// GenerateID creates a short unique identifier for request/operation tracing -func GenerateID() string { - bytes := make([]byte, 4) - rand.Read(bytes) - return hex.EncodeToString(bytes) -} - -// Log writes a structured log message with fixed-width formatting -func (l *Logger) Log(id, level, message string, args ...interface{}) { - formattedMsg := fmt.Sprintf(message, args...) - log.Printf("[%s] [%-5s] [%-8s] %s", id, level, l.component, formattedMsg) -} - -// Debug logs debug level messages -func (l *Logger) Debug(id, message string, args ...interface{}) { - l.Log(id, "DEBUG", message, args...) -} - -// Info logs info level messages -func (l *Logger) Info(id, message string, args ...interface{}) { - l.Log(id, "INFO", message, args...) -} - -// Warn logs warning level messages -func (l *Logger) Warn(id, message string, args ...interface{}) { - l.Log(id, "WARN", message, args...) +// Logger is a thin component-scoped facade over log/slog. +type Logger struct { + log *slog.Logger } -// Error logs error level messages -func (l *Logger) Error(id, message string, args ...interface{}) { - l.Log(id, "ERROR", message, args...) +// New creates a logger tagged with the given component. +func New(component string) *Logger { + return &Logger{log: slog.New(handler).With("component", component)} } -// LogWithoutID logs without an ID (for background operations) -func (l *Logger) LogWithoutID(level, message string, args ...interface{}) { - formattedMsg := fmt.Sprintf(message, args...) - log.Printf("[xxxxxxxx] [%-5s] [%-8s] %s", level, l.component, formattedMsg) +// GenerateID creates a short unique identifier for request/operation tracing. +func GenerateID() string { + b := make([]byte, 4) + rand.Read(b) + return hex.EncodeToString(b) } -// DebugBg logs debug messages for background operations -func (l *Logger) DebugBg(message string, args ...interface{}) { - l.LogWithoutID("DEBUG", message, args...) +func (l *Logger) at(level slog.Level, id, msg string, args ...any) { + l.log.Log(context.Background(), level, fmt.Sprintf(msg, args...), "id", id) } -// InfoBg logs info messages for background operations -func (l *Logger) InfoBg(message string, args ...interface{}) { - l.LogWithoutID("INFO", message, args...) -} +func (l *Logger) Debug(id, msg string, args ...any) { l.at(slog.LevelDebug, id, msg, args...) } +func (l *Logger) Info(id, msg string, args ...any) { l.at(slog.LevelInfo, id, msg, args...) } +func (l *Logger) Warn(id, msg string, args ...any) { l.at(slog.LevelWarn, id, msg, args...) } +func (l *Logger) Error(id, msg string, args ...any) { l.at(slog.LevelError, id, msg, args...) } -// WarnBg logs warning messages for background operations -func (l *Logger) WarnBg(message string, args ...interface{}) { - l.LogWithoutID("WARN", message, args...) -} +// *Bg variants are for background/non-correlated operations (no trace id). +func (l *Logger) DebugBg(msg string, args ...any) { l.log.Debug(fmt.Sprintf(msg, args...)) } +func (l *Logger) InfoBg(msg string, args ...any) { l.log.Info(fmt.Sprintf(msg, args...)) } +func (l *Logger) WarnBg(msg string, args ...any) { l.log.Warn(fmt.Sprintf(msg, args...)) } +func (l *Logger) ErrorBg(msg string, args ...any) { l.log.Error(fmt.Sprintf(msg, args...)) } -// ErrorBg logs error messages for background operations -func (l *Logger) ErrorBg(message string, args ...interface{}) { - l.LogWithoutID("ERROR", message, args...) +// Fatal logs at error level then exits non-zero, mirroring stdlib log.Fatalf. +func (l *Logger) Fatal(msg string, args ...any) { + l.log.Error(fmt.Sprintf(msg, args...)) + os.Exit(1) } diff --git a/pkg/checker/checker.go b/pkg/checker/checker.go index 799a247..fdbf97e 100644 --- a/pkg/checker/checker.go +++ b/pkg/checker/checker.go @@ -11,6 +11,7 @@ import ( "sync" "time" + "aproxy/internal/config" "aproxy/internal/logger" "aproxy/pkg/scraper" netproxy "golang.org/x/net/proxy" @@ -57,24 +58,7 @@ type Checker struct { logger *logger.Logger } -type CheckerConfig struct { - TestURL string - Timeout time.Duration - MaxWorkers int - UserAgent string -} - -func NewChecker() *Checker { - return &Checker{ - testURL: "http://httpbin.org/ip", - timeout: 20 * time.Second, - maxWorkers: 20, - userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", - logger: logger.New("checker"), - } -} - -func NewCheckerWithConfig(config CheckerConfig) *Checker { +func NewChecker(config config.CheckerConfig) *Checker { return &Checker{ testURL: config.TestURL, timeout: config.Timeout, @@ -84,18 +68,6 @@ func NewCheckerWithConfig(config CheckerConfig) *Checker { } } -func (c *Checker) SetTestURL(url string) { - c.testURL = url -} - -func (c *Checker) SetTimeout(timeout time.Duration) { - c.timeout = timeout -} - -func (c *Checker) SetMaxWorkers(workers int) { - c.maxWorkers = workers -} - func (c *Checker) CheckProxy(ctx context.Context, proxy scraper.Proxy) CheckResult { start := time.Now() result := CheckResult{ @@ -170,27 +142,19 @@ func (c *Checker) CheckProxies(ctx context.Context, proxies []scraper.Proxy) []C } func (c *Checker) testProxy(ctx context.Context, proxy scraper.Proxy) (ProxyStatus, error) { - // Handle SOCKS proxies with specialized testing - if proxy.Type == "socks4" || proxy.Type == "socks5" { - return c.testSOCKSProxy(ctx, proxy) - } - - proxyURL, err := c.buildProxyURL(proxy) + transport, err := c.buildTransport(proxy) if err != nil { return StatusError, err } + return c.runCheck(ctx, transport) +} - // Create a more permissive transport +// buildTransport returns an http.Transport routed through the given proxy, +// for HTTP/HTTPS proxies (via Proxy URL) or SOCKS proxies (via a SOCKS dialer). +func (c *Checker) buildTransport(proxy scraper.Proxy) (*http.Transport, error) { transport := &http.Transport{ - Proxy: http.ProxyURL(proxyURL), - DialContext: (&net.Dialer{ - Timeout: 10 * time.Second, - KeepAlive: 0, // Disable keep-alive for proxy checks - }).DialContext, - TLSClientConfig: &tls.Config{ - InsecureSkipVerify: true, - }, - DisableKeepAlives: true, // Important for proxy testing + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + DisableKeepAlives: true, DisableCompression: true, MaxIdleConns: 0, IdleConnTimeout: 1 * time.Second, @@ -199,20 +163,44 @@ func (c *Checker) testProxy(ctx context.Context, proxy scraper.Proxy) (ProxyStat ExpectContinueTimeout: 1 * time.Second, } + if proxy.Type == "socks4" || proxy.Type == "socks5" { + // ponytail: x/net/proxy has no SOCKS4 dialer, so socks4 is probed via + // the SOCKS5 handshake. A pure-SOCKS4 proxy will fail this and be marked + // unhealthy — acceptable; swap in a socks4 dialer if you need them. + dialer, err := createSOCKSDialer(proxy.Host, proxy.Port) + if err != nil { + return nil, err + } + transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + return dialer.Dial(network, addr) + } + return transport, nil + } + + // HTTP/HTTPS proxy. + proxyURL, err := url.Parse(fmt.Sprintf("http://%s:%d", proxy.Host, proxy.Port)) + if err != nil { + return nil, err + } + transport.Proxy = http.ProxyURL(proxyURL) + transport.DialContext = (&net.Dialer{Timeout: 10 * time.Second, KeepAlive: 0}).DialContext + return transport, nil +} + +// runCheck issues the test request over the transport and classifies the result. +func (c *Checker) runCheck(ctx context.Context, transport *http.Transport) (ProxyStatus, error) { client := &http.Client{ Transport: transport, Timeout: c.timeout, CheckRedirect: func(req *http.Request, via []*http.Request) error { - return http.ErrUseLastResponse // Don't follow redirects + return http.ErrUseLastResponse }, } - // Use a simple, reliable test URL req, err := http.NewRequestWithContext(ctx, "GET", c.testURL, nil) if err != nil { return StatusError, err } - req.Header.Set("User-Agent", c.userAgent) req.Header.Set("Accept", "text/plain, application/json") req.Header.Set("Connection", "close") @@ -222,7 +210,6 @@ func (c *Checker) testProxy(ctx context.Context, proxy scraper.Proxy) (ProxyStat if isTimeoutError(err) { return StatusTimeout, err } - // Check for common connection errors if isConnectionError(err) { return StatusUnhealthy, err } @@ -230,30 +217,12 @@ func (c *Checker) testProxy(ctx context.Context, proxy scraper.Proxy) (ProxyStat } defer resp.Body.Close() - // Accept any 2xx status code if resp.StatusCode >= 200 && resp.StatusCode < 300 { return StatusHealthy, nil } - return StatusUnhealthy, fmt.Errorf("HTTP %d", resp.StatusCode) } -func (c *Checker) buildProxyURL(proxy scraper.Proxy) (*url.URL, error) { - var scheme string - switch proxy.Type { - case "http", "https": - scheme = "http" - case "socks4": - scheme = "socks4" - case "socks5": - scheme = "socks5" - default: - scheme = "http" - } - - return url.Parse(fmt.Sprintf("%s://%s:%d", scheme, proxy.Host, proxy.Port)) -} - func isTimeoutError(err error) bool { if netErr, ok := err.(net.Error); ok { return netErr.Timeout() @@ -282,158 +251,6 @@ func FilterHealthyProxies(results []CheckResult) []scraper.Proxy { return healthy } -func GetHealthyCount(results []CheckResult) int { - count := 0 - for _, result := range results { - if result.Status == StatusHealthy { - count++ - } - } - return count -} - -func GroupByStatus(results []CheckResult) map[ProxyStatus][]CheckResult { - groups := make(map[ProxyStatus][]CheckResult) - for _, result := range results { - groups[result.Status] = append(groups[result.Status], result) - } - return groups -} - -// testSOCKSProxy tests SOCKS4 and SOCKS5 proxies -func (c *Checker) testSOCKSProxy(ctx context.Context, proxy scraper.Proxy) (ProxyStatus, error) { - // For SOCKS proxies, we'll test by establishing a connection and making a simple HTTP request - // This is more complex than HTTP proxies but necessary for proper validation - - if proxy.Type == "socks4" { - return c.testSOCKS4Proxy(ctx, proxy) - } else if proxy.Type == "socks5" { - return c.testSOCKS5Proxy(ctx, proxy) - } - - return StatusError, fmt.Errorf("unsupported SOCKS type: %s", proxy.Type) -} - -// testSOCKS4Proxy tests a SOCKS4 proxy by making a connection -func (c *Checker) testSOCKS4Proxy(ctx context.Context, proxy scraper.Proxy) (ProxyStatus, error) { - // Create a dialer that uses the SOCKS4 proxy (note: we'll use SOCKS5 for SOCKS4 as it's more widely supported) - dialer, err := createSOCKSDialer(proxy.Host, proxy.Port) - if err != nil { - return StatusError, err - } - - // Test by connecting to a simple HTTP endpoint through the SOCKS proxy - transport := &http.Transport{ - DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { - return dialer.Dial(network, addr) - }, - DisableKeepAlives: true, - DisableCompression: true, - MaxIdleConns: 0, - IdleConnTimeout: 1 * time.Second, - TLSHandshakeTimeout: 5 * time.Second, - ResponseHeaderTimeout: 10 * time.Second, - ExpectContinueTimeout: 1 * time.Second, - } - - client := &http.Client{ - Transport: transport, - Timeout: c.timeout, - CheckRedirect: func(req *http.Request, via []*http.Request) error { - return http.ErrUseLastResponse - }, - } - - // Make a simple HTTP request through the SOCKS proxy - req, err := http.NewRequestWithContext(ctx, "GET", c.testURL, nil) - if err != nil { - return StatusError, err - } - - req.Header.Set("User-Agent", c.userAgent) - req.Header.Set("Accept", "text/plain, application/json") - req.Header.Set("Connection", "close") - - resp, err := client.Do(req) - if err != nil { - if isTimeoutError(err) { - return StatusTimeout, err - } - if isConnectionError(err) { - return StatusUnhealthy, err - } - return StatusError, err - } - defer resp.Body.Close() - - // Accept any 2xx status code - if resp.StatusCode >= 200 && resp.StatusCode < 300 { - return StatusHealthy, nil - } - - return StatusUnhealthy, fmt.Errorf("HTTP %d", resp.StatusCode) -} - -// testSOCKS5Proxy tests a SOCKS5 proxy by making a connection -func (c *Checker) testSOCKS5Proxy(ctx context.Context, proxy scraper.Proxy) (ProxyStatus, error) { - // Create a dialer that uses the SOCKS5 proxy - dialer, err := createSOCKSDialer(proxy.Host, proxy.Port) - if err != nil { - return StatusError, err - } - - // Test by connecting to a simple HTTP endpoint through the SOCKS proxy - transport := &http.Transport{ - DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { - return dialer.Dial(network, addr) - }, - DisableKeepAlives: true, - DisableCompression: true, - MaxIdleConns: 0, - IdleConnTimeout: 1 * time.Second, - TLSHandshakeTimeout: 5 * time.Second, - ResponseHeaderTimeout: 10 * time.Second, - ExpectContinueTimeout: 1 * time.Second, - } - - client := &http.Client{ - Transport: transport, - Timeout: c.timeout, - CheckRedirect: func(req *http.Request, via []*http.Request) error { - return http.ErrUseLastResponse - }, - } - - // Make a simple HTTP request through the SOCKS proxy - req, err := http.NewRequestWithContext(ctx, "GET", c.testURL, nil) - if err != nil { - return StatusError, err - } - - req.Header.Set("User-Agent", c.userAgent) - req.Header.Set("Accept", "text/plain, application/json") - req.Header.Set("Connection", "close") - - resp, err := client.Do(req) - if err != nil { - if isTimeoutError(err) { - return StatusTimeout, err - } - if isConnectionError(err) { - return StatusUnhealthy, err - } - return StatusError, err - } - defer resp.Body.Close() - - // Accept any 2xx status code - if resp.StatusCode >= 200 && resp.StatusCode < 300 { - return StatusHealthy, nil - } - - return StatusUnhealthy, fmt.Errorf("HTTP %d", resp.StatusCode) -} - // createSOCKSDialer creates a dialer that uses a SOCKS5 proxy (works for most SOCKS4 too) func createSOCKSDialer(host string, port int) (netproxy.Dialer, error) { // Using golang.org/x/net/proxy package for SOCKS support @@ -444,23 +261,3 @@ func createSOCKSDialer(host string, port int) (netproxy.Dialer, error) { } return dialer, nil } - -// TestSingleProxy tests a single proxy manually (for debugging) -func TestSingleProxy(host string, port int) { - proxy := scraper.Proxy{ - Host: host, - Port: port, - Type: "http", - } - - checker := NewChecker() - ctx := context.Background() - - checker.logger.InfoBg("Testing proxy %s:%d", host, port) - result := checker.CheckProxy(ctx, proxy) - - checker.logger.InfoBg("Result: %s (took %v)", result.Status.String(), result.ResponseTime) - if result.Error != nil { - checker.logger.WarnBg("Error: %v", result.Error) - } -} diff --git a/pkg/checker/db_checker.go b/pkg/checker/db_checker.go index 12318ba..2c7c159 100644 --- a/pkg/checker/db_checker.go +++ b/pkg/checker/db_checker.go @@ -5,12 +5,21 @@ import ( "fmt" "time" + "aproxy/internal/config" "aproxy/internal/database" - "aproxy/internal/database/models/model" "aproxy/internal/logger" "aproxy/pkg/scraper" ) +// checker.ProxyStatus and database.ProxyStatus are cast across the package +// boundary below; these asserts fail the build if their iota values ever drift. +// ponytail: cheaper than merging the two enums, which would need a package +// restructure to break the checker<->database import cycle. +const ( + _ = uint(database.StatusHealthy - database.ProxyStatus(StatusHealthy)) + _ = uint(database.StatusError - database.ProxyStatus(StatusError)) +) + // DBChecker is a checker that uses SQLite for caching proxy health status type DBChecker struct { *Checker @@ -21,26 +30,14 @@ type DBChecker struct { logger *logger.Logger } -// NewDBChecker creates a new database-backed checker -func NewDBChecker(dbService *database.Service, checkInterval time.Duration, batchSize int, batchDelay time.Duration) *DBChecker { - return &DBChecker{ - Checker: NewChecker(), - dbService: dbService, - checkInterval: checkInterval, - batchSize: batchSize, - batchDelay: batchDelay, - logger: logger.New("db-checker"), - } -} - -// NewDBCheckerWithConfig creates a new database-backed checker with configuration -func NewDBCheckerWithConfig(dbService *database.Service, checkerConfig CheckerConfig, checkInterval time.Duration, batchSize int, batchDelay time.Duration) *DBChecker { +// NewDBChecker creates a new database-backed checker with configuration +func NewDBChecker(dbService *database.Service, cfg config.CheckerConfig) *DBChecker { return &DBChecker{ - Checker: NewCheckerWithConfig(checkerConfig), + Checker: NewChecker(cfg), dbService: dbService, - checkInterval: checkInterval, - batchSize: batchSize, - batchDelay: batchDelay, + checkInterval: cfg.CheckInterval, + batchSize: cfg.BatchSize, + batchDelay: cfg.BatchDelay, logger: logger.New("db-checker"), } } @@ -74,7 +71,7 @@ func (c *DBChecker) CheckProxiesWithCaching(ctx context.Context, proxies []scrap // Separate new proxies that need to be inserted vs existing ones var newProxies []scraper.Proxy - var dbProxies []*model.Proxies + var dbProxies []*database.Proxy cutoff := time.Now().Add(-c.checkInterval) for addr, proxy := range proxyByAddr { @@ -99,7 +96,7 @@ func (c *DBChecker) CheckProxiesWithCaching(ctx context.Context, proxies []scrap // Determine which proxies need health checks var proxiesToCheck []scraper.Proxy - var proxiesNeedingCheck []*model.Proxies + var proxiesNeedingCheck []*database.Proxy for _, dbProxy := range dbProxies { needsCheck := false @@ -137,7 +134,7 @@ func (c *DBChecker) CheckProxiesWithCaching(ctx context.Context, proxies []scrap results := c.checkProxiesProgressive(ctx, proxiesToCheck) // Batch update database with all results in a single transaction - proxyMap := make(map[string]*model.Proxies) + proxyMap := make(map[string]*database.Proxy) for _, dbProxy := range proxiesNeedingCheck { addr := fmt.Sprintf("%s:%d", dbProxy.Host, dbProxy.Port) proxyMap[addr] = dbProxy @@ -162,7 +159,7 @@ func (c *DBChecker) CheckProxiesWithCaching(ctx context.Context, proxies []scrap CheckedAt: result.CheckedAt, } - updates[*dbProxy.ID] = dbResult + updates[int32(dbProxy.ID)] = dbResult } // Execute batch update in smaller chunks to avoid timeouts @@ -203,106 +200,62 @@ func (c *DBChecker) CheckProxiesWithCaching(ctx context.Context, proxies []scrap return c.getAllResults(ctx, dbProxies, results) } -// getCachedResults returns cached check results for all proxies -func (c *DBChecker) getCachedResults(ctx context.Context, dbProxies []*model.Proxies) []CheckResult { - var results []CheckResult - - for _, dbProxy := range dbProxies { - proxy := scraper.Proxy{ - Host: dbProxy.Host, - Port: int(dbProxy.Port), - Type: dbProxy.ProxyType, - Country: "", - } - if dbProxy.Country != nil { - proxy.Country = *dbProxy.Country - } - - status := StatusUnknown - switch dbProxy.Status { - case "healthy": - status = StatusHealthy - case "unhealthy": - status = StatusUnhealthy - case "timeout": - status = StatusTimeout - case "error": - status = StatusError - } - - result := CheckResult{ - Proxy: proxy, - Status: status, - } - - if dbProxy.LastCheckedAt != nil { - result.CheckedAt = *dbProxy.LastCheckedAt - } +// dbProxyToResult converts a stored proxy row into a cached CheckResult. +func dbProxyToResult(dbProxy *database.Proxy) CheckResult { + proxy := scraper.Proxy{ + Host: dbProxy.Host, + Port: int(dbProxy.Port), + Type: dbProxy.ProxyType, + } + if dbProxy.Country != nil { + proxy.Country = *dbProxy.Country + } - if dbProxy.ResponseTimeMs != nil { - result.ResponseTime = time.Duration(*dbProxy.ResponseTimeMs) * time.Millisecond - } + status := StatusUnknown + switch dbProxy.Status { + case "healthy": + status = StatusHealthy + case "unhealthy": + status = StatusUnhealthy + case "timeout": + status = StatusTimeout + case "error": + status = StatusError + } - results = append(results, result) + result := CheckResult{Proxy: proxy, Status: status} + if dbProxy.LastCheckedAt != nil { + result.CheckedAt = *dbProxy.LastCheckedAt + } + if dbProxy.ResponseTimeMs != nil { + result.ResponseTime = time.Duration(*dbProxy.ResponseTimeMs) * time.Millisecond } + return result +} +// getCachedResults returns cached check results for all proxies +func (c *DBChecker) getCachedResults(ctx context.Context, dbProxies []*database.Proxy) []CheckResult { + results := make([]CheckResult, 0, len(dbProxies)) + for _, dbProxy := range dbProxies { + results = append(results, dbProxyToResult(dbProxy)) + } return results } // getAllResults combines fresh check results with cached results -func (c *DBChecker) getAllResults(ctx context.Context, dbProxies []*model.Proxies, freshResults []CheckResult) []CheckResult { - // Create a map of fresh results by proxy address +func (c *DBChecker) getAllResults(ctx context.Context, dbProxies []*database.Proxy, freshResults []CheckResult) []CheckResult { freshMap := make(map[string]CheckResult) for _, result := range freshResults { freshMap[result.Proxy.Address()] = result } - var allResults []CheckResult - + allResults := make([]CheckResult, 0, len(dbProxies)) for _, dbProxy := range dbProxies { - proxyAddr := fmt.Sprintf("%s:%d", dbProxy.Host, dbProxy.Port) - - // Use fresh result if available, otherwise use cached result - if freshResult, exists := freshMap[proxyAddr]; exists { - allResults = append(allResults, freshResult) + addr := fmt.Sprintf("%s:%d", dbProxy.Host, dbProxy.Port) + if fresh, ok := freshMap[addr]; ok { + allResults = append(allResults, fresh) } else { - // Create cached result - proxy := scraper.Proxy{ - Host: dbProxy.Host, - Port: int(dbProxy.Port), - Type: dbProxy.ProxyType, - Country: "", - } - if dbProxy.Country != nil { - proxy.Country = *dbProxy.Country - } - - status := StatusUnknown - switch dbProxy.Status { - case "healthy": - status = StatusHealthy - case "unhealthy": - status = StatusUnhealthy - case "timeout": - status = StatusTimeout - case "error": - status = StatusError - } - - result := CheckResult{ - Proxy: proxy, - Status: status, - } - - if dbProxy.LastCheckedAt != nil { - result.CheckedAt = *dbProxy.LastCheckedAt - } - - if dbProxy.ResponseTimeMs != nil { - result.ResponseTime = time.Duration(*dbProxy.ResponseTimeMs) * time.Millisecond - } - - allResults = append(allResults, result) + allResults = append(allResults, dbProxyToResult(dbProxy)) } } @@ -400,8 +353,8 @@ func (c *DBChecker) checkProxiesProgressive(ctx context.Context, proxies []scrap if dbProxies, err := c.dbService.GetProxiesByAddresses(saveCtx, addresses); err == nil { updates := make(map[int32]database.CheckResult) for _, result := range results { - if dbProxy, exists := dbProxies[result.Proxy.Address()]; exists && dbProxy.ID != nil { - updates[*dbProxy.ID] = database.CheckResult{ + if dbProxy, exists := dbProxies[result.Proxy.Address()]; exists { + updates[int32(dbProxy.ID)] = database.CheckResult{ Proxy: result.Proxy, Status: database.ProxyStatus(result.Status), ResponseTime: result.ResponseTime, diff --git a/pkg/manager/db_manager.go b/pkg/manager/db_manager.go index bc68d73..afffc81 100644 --- a/pkg/manager/db_manager.go +++ b/pkg/manager/db_manager.go @@ -7,12 +7,21 @@ import ( "sync" "time" + "aproxy/internal/config" "aproxy/internal/database" "aproxy/internal/logger" "aproxy/pkg/checker" "aproxy/pkg/scraper" ) +// Stats summarizes the in-memory proxy pool. +type Stats struct { + TotalProxies int + HealthyCount int + TypeCount map[string]int + CountryCount map[string]int +} + // DBManager is a manager that uses SQLite for persistent proxy storage type DBManager struct { scraper *scraper.MultiScraper @@ -34,21 +43,21 @@ type DBManager struct { updateInterval time.Duration } -// NewDBManagerWithConfig creates a new database-backed manager with configuration -func NewDBManagerWithConfig(db *database.DB, scraperConfig scraper.ScraperConfig, checkerConfig checker.CheckerConfig, checkInterval time.Duration, backgroundEnabled bool, batchSize int, batchDelay time.Duration) *DBManager { +// NewDBManager creates a new database-backed manager with configuration +func NewDBManager(db *database.DB, cfg *config.Config) *DBManager { ctx, cancel := context.WithCancel(context.Background()) dbService := database.NewService(db) - dbChecker := checker.NewDBCheckerWithConfig(dbService, checkerConfig, checkInterval, batchSize, batchDelay) + dbChecker := checker.NewDBChecker(dbService, cfg.Checker) return &DBManager{ - scraper: scraper.NewMultiScraperWithConfig(scraperConfig), + scraper: scraper.NewMultiScraper(cfg.Scraper), dbChecker: dbChecker, dbService: dbService, ctx: ctx, cancel: cancel, cachedProxies: make([]scraper.Proxy, 0), - backgroundEnabled: backgroundEnabled, + backgroundEnabled: cfg.Checker.BackgroundEnabled, logger: logger.New("manager"), } } diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go deleted file mode 100644 index 0828c38..0000000 --- a/pkg/manager/manager.go +++ /dev/null @@ -1,258 +0,0 @@ -package manager - -import ( - "context" - "fmt" - "log" - "math/rand" - "sync" - "time" - - "aproxy/pkg/checker" - "aproxy/pkg/scraper" -) - -// ProxyManager defines the interface that proxy managers must implement -type ProxyManager interface { - GetNextProxy() (*scraper.Proxy, error) - GetRandomProxy() (*scraper.Proxy, error) - ReportProxyFailure(scraper.Proxy) - GetStats() Stats - Start(updateInterval time.Duration) error - Stop() - RefreshProxies() error -} - -type ProxyPool struct { - proxies []scraper.Proxy - healthStatus map[string]checker.ProxyStatus - lastChecked map[string]time.Time - failCount map[string]int - mu sync.RWMutex - currentIndex int - maxFails int - recheckTime time.Duration -} - -type Manager struct { - pool *ProxyPool - scraper *scraper.MultiScraper - checker *checker.Checker - updateTicker *time.Ticker - ctx context.Context - cancel context.CancelFunc - wg sync.WaitGroup -} - -func NewManager() *Manager { - ctx, cancel := context.WithCancel(context.Background()) - - return &Manager{ - pool: &ProxyPool{ - proxies: make([]scraper.Proxy, 0), - healthStatus: make(map[string]checker.ProxyStatus), - lastChecked: make(map[string]time.Time), - failCount: make(map[string]int), - maxFails: 3, - recheckTime: 5 * time.Minute, - }, - scraper: scraper.NewMultiScraper(), - checker: checker.NewChecker(), - ctx: ctx, - cancel: cancel, - } -} - -func (m *Manager) Start(updateInterval time.Duration) error { - log.Println("Starting proxy manager...") - - if err := m.RefreshProxies(); err != nil { - return fmt.Errorf("initial proxy refresh failed: %w", err) - } - - m.updateTicker = time.NewTicker(updateInterval) - - m.wg.Add(1) - go m.updateLoop() - - log.Printf("Proxy manager started with %d proxies", m.pool.Count()) - return nil -} - -func (m *Manager) Stop() { - log.Println("Stopping proxy manager...") - - if m.updateTicker != nil { - m.updateTicker.Stop() - } - - m.cancel() - m.wg.Wait() - - log.Println("Proxy manager stopped") -} - -func (m *Manager) RefreshProxies() error { - log.Println("Refreshing proxy list...") - - ctx, cancel := context.WithTimeout(m.ctx, 2*time.Minute) - defer cancel() - - proxies, err := m.scraper.ScrapeAll(ctx) - if err != nil { - return fmt.Errorf("failed to scrape proxies: %w", err) - } - - log.Printf("Scraped %d proxies, checking health...", len(proxies)) - - results := m.checker.CheckProxies(ctx, proxies) - healthyProxies := checker.FilterHealthyProxies(results) - - log.Printf("Found %d healthy proxies out of %d checked", len(healthyProxies), len(results)) - - m.pool.mu.Lock() - m.pool.proxies = healthyProxies - m.pool.currentIndex = 0 - - for _, result := range results { - key := result.Proxy.Address() - m.pool.healthStatus[key] = result.Status - m.pool.lastChecked[key] = result.CheckedAt - - if result.Status != checker.StatusHealthy { - m.pool.failCount[key]++ - } else { - m.pool.failCount[key] = 0 - } - } - m.pool.mu.Unlock() - - return nil -} - -func (m *Manager) GetNextProxy() (*scraper.Proxy, error) { - m.pool.mu.Lock() - defer m.pool.mu.Unlock() - - if len(m.pool.proxies) == 0 { - return nil, fmt.Errorf("no healthy proxies available") - } - - proxy := &m.pool.proxies[m.pool.currentIndex] - m.pool.currentIndex = (m.pool.currentIndex + 1) % len(m.pool.proxies) - - return proxy, nil -} - -func (m *Manager) GetRandomProxy() (*scraper.Proxy, error) { - m.pool.mu.RLock() - defer m.pool.mu.RUnlock() - - if len(m.pool.proxies) == 0 { - return nil, fmt.Errorf("no healthy proxies available") - } - - index := rand.Intn(len(m.pool.proxies)) - proxy := &m.pool.proxies[index] - - return proxy, nil -} - -func (m *Manager) ReportProxyFailure(proxy scraper.Proxy) { - m.pool.mu.Lock() - defer m.pool.mu.Unlock() - - key := proxy.Address() - m.pool.failCount[key]++ - m.pool.healthStatus[key] = checker.StatusUnhealthy - - if m.pool.failCount[key] >= m.pool.maxFails { - m.removeProxy(proxy) - log.Printf("Removed failing proxy: %s (failed %d times)", key, m.pool.failCount[key]) - } -} - -func (m *Manager) removeProxy(targetProxy scraper.Proxy) { - targetKey := targetProxy.Address() - newProxies := make([]scraper.Proxy, 0, len(m.pool.proxies)) - - for _, proxy := range m.pool.proxies { - if proxy.Address() != targetKey { - newProxies = append(newProxies, proxy) - } - } - - m.pool.proxies = newProxies - - if m.pool.currentIndex >= len(m.pool.proxies) && len(m.pool.proxies) > 0 { - m.pool.currentIndex = 0 - } -} - -func (m *Manager) GetStats() Stats { - m.pool.mu.RLock() - defer m.pool.mu.RUnlock() - - stats := Stats{ - TotalProxies: len(m.pool.proxies), - HealthyCount: 0, - TypeCount: make(map[string]int), - CountryCount: make(map[string]int), - } - - for _, proxy := range m.pool.proxies { - key := proxy.Address() - if m.pool.healthStatus[key] == checker.StatusHealthy { - stats.HealthyCount++ - } - - stats.TypeCount[proxy.Type]++ - if proxy.Country != "" { - stats.CountryCount[proxy.Country]++ - } - } - - return stats -} - -func (m *Manager) updateLoop() { - defer m.wg.Done() - - for { - select { - case <-m.ctx.Done(): - return - case <-m.updateTicker.C: - if err := m.RefreshProxies(); err != nil { - log.Printf("Failed to refresh proxies: %v", err) - } - } - } -} - -func (p *ProxyPool) Count() int { - p.mu.RLock() - defer p.mu.RUnlock() - return len(p.proxies) -} - -func (p *ProxyPool) HealthyCount() int { - p.mu.RLock() - defer p.mu.RUnlock() - - count := 0 - for _, proxy := range p.proxies { - key := proxy.Address() - if p.healthStatus[key] == checker.StatusHealthy { - count++ - } - } - return count -} - -type Stats struct { - TotalProxies int - HealthyCount int - TypeCount map[string]int - CountryCount map[string]int -} diff --git a/pkg/proxy/server.go b/pkg/proxy/server.go index d2c2bf3..afe5fc7 100644 --- a/pkg/proxy/server.go +++ b/pkg/proxy/server.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "crypto/tls" + "encoding/json" "fmt" "io" "net" @@ -13,6 +14,7 @@ import ( "sync" "time" + "aproxy/internal/config" "aproxy/internal/logger" "aproxy/pkg/manager" "aproxy/pkg/scraper" @@ -21,28 +23,15 @@ import ( ) type Server struct { - manager manager.ProxyManager + manager *manager.DBManager server *http.Server - config *Config + config config.ServerConfig stats *Stats logger *logger.Logger httpLogger *logger.Logger httpsLogger *logger.Logger } -type Config struct { - ListenAddr string - ReadTimeout time.Duration - WriteTimeout time.Duration - IdleTimeout time.Duration - MaxConnections int - EnableHTTPS bool - MaxRetries int - StripHeaders []string - AddHeaders map[string]string - AuthToken string -} - type Stats struct { RequestsHandled int64 BytesTransferred int64 @@ -51,11 +40,7 @@ type Stats struct { mu sync.RWMutex } -func NewServer(mgr manager.ProxyManager, config *Config) *Server { - if config == nil { - config = DefaultConfig() - } - +func NewServer(mgr *manager.DBManager, config config.ServerConfig) *Server { return &Server{ manager: mgr, config: config, @@ -66,28 +51,6 @@ func NewServer(mgr manager.ProxyManager, config *Config) *Server { } } -func DefaultConfig() *Config { - return &Config{ - ListenAddr: ":8080", - ReadTimeout: 30 * time.Second, - WriteTimeout: 30 * time.Second, - IdleTimeout: 60 * time.Second, - MaxConnections: 1000, - EnableHTTPS: true, - MaxRetries: 3, - StripHeaders: []string{ - "X-Forwarded-For", - "X-Real-IP", - "X-Original-IP", - "CF-Connecting-IP", - "True-Client-IP", - }, - AddHeaders: map[string]string{ - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", - }, - } -} - func (s *Server) Start() error { s.server = &http.Server{ Addr: s.config.ListenAddr, @@ -572,26 +535,31 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) { managerStats := s.manager.GetStats() serverStats := s.getStats() - // Try to get database stats if this is a DBManager - dbStatsJSON := `"not_available"` - if dbManager, ok := s.manager.(*manager.DBManager); ok { - if dbStats, err := dbManager.GetDBStats(context.Background()); err == nil { - dbStatsJSON = fmt.Sprintf(`{"total_in_db": %d, "healthy_in_db": %d, "by_type": %s}`, dbStats.Total, dbStats.Healthy, formatMap(dbStats.ByType)) + resp := map[string]any{ + "proxy_stats": map[string]any{ + "cached_proxies": managerStats.TotalProxies, + "cached_healthy": managerStats.HealthyCount, + "proxy_types": managerStats.TypeCount, + "proxy_countries": managerStats.CountryCount, + }, + "server_stats": map[string]any{ + "requests_handled": serverStats.RequestsHandled, + "bytes_transferred": serverStats.BytesTransferred, + "active_connections": serverStats.ActiveConnections, + "failed_requests": serverStats.FailedRequests, + }, + "database_stats": "not_available", + } + if dbStats, err := s.manager.GetDBStats(context.Background()); err == nil { + resp["database_stats"] = map[string]any{ + "total_in_db": dbStats.Total, + "healthy_in_db": dbStats.Healthy, + "by_type": dbStats.ByType, } } w.Header().Set("Content-Type", "application/json") - fmt.Fprintf(w, `{"proxy_stats": {"cached_proxies": %d, "cached_healthy": %d, "proxy_types": %s, "proxy_countries": %s}, "database_stats": %s, "server_stats": {"requests_handled": %d, "bytes_transferred": %d, "active_connections": %d, "failed_requests": %d}}`, - managerStats.TotalProxies, - managerStats.HealthyCount, - formatMap(managerStats.TypeCount), - formatMap(managerStats.CountryCount), - dbStatsJSON, - serverStats.RequestsHandled, - serverStats.BytesTransferred, - serverStats.ActiveConnections, - serverStats.FailedRequests, - ) + json.NewEncoder(w).Encode(resp) } func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { @@ -616,39 +584,26 @@ func (s *Server) handleProxies(w http.ResponseWriter, r *http.Request) { return } - // Get healthy proxies from manager - var proxies []scraper.Proxy - if dbManager, ok := s.manager.(*manager.DBManager); ok { - proxies = dbManager.GetHealthyProxies() - } else { - http.Error(w, "Proxy list not available", http.StatusServiceUnavailable) - return - } + proxies := s.manager.GetHealthyProxies() if len(proxies) == 0 { http.Error(w, "No healthy proxies available", http.StatusServiceUnavailable) return } - // Format proxies as JSON - w.Header().Set("Content-Type", "application/json") - fmt.Fprintf(w, `{"proxies": [`) - for i, proxy := range proxies { - if i > 0 { - fmt.Fprintf(w, `, `) + list := make([]map[string]any, len(proxies)) + for i, p := range proxies { + list[i] = map[string]any{ + "host": p.Host, + "port": p.Port, + "type": p.Type, + "country": p.Country, + "last_seen": p.LastSeen.Format("2006-01-02T15:04:05Z"), } - fmt.Fprintf(w, `{"host": "%s", "port": %d, "type": "%s", "country": "%s", "last_seen": "%s"}`, - proxy.Host, proxy.Port, proxy.Type, proxy.Country, proxy.LastSeen.Format("2006-01-02T15:04:05Z")) } - fmt.Fprintf(w, `], "count": %d}`, len(proxies)) -} -func formatMap(m map[string]int) string { - var parts []string - for k, v := range m { - parts = append(parts, fmt.Sprintf(`"%s": %d`, k, v)) - } - return "{" + strings.Join(parts, ", ") + "}" + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"proxies": list, "count": len(proxies)}) } func (s *Server) getStats() Stats { diff --git a/pkg/scraper/freeproxylist.go b/pkg/scraper/freeproxylist.go deleted file mode 100644 index a9f92c0..0000000 --- a/pkg/scraper/freeproxylist.go +++ /dev/null @@ -1,132 +0,0 @@ -package scraper - -import ( - "aproxy/internal/logger" - "bufio" - "context" - "fmt" - "io" - "net/http" - "net/url" - "strconv" - "strings" - "time" -) - -type FreeProxyListScraper struct { - client *http.Client - userAgent string - logger *logger.Logger -} - -func NewFreeProxyListScraper() *FreeProxyListScraper { - return &FreeProxyListScraper{ - client: &http.Client{ - Timeout: 30 * time.Second, - }, - userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", - logger: logger.New("freeproxylist"), - } -} - -func NewFreeProxyListScraperWithConfig(config ScraperConfig) *FreeProxyListScraper { - return &FreeProxyListScraper{ - client: &http.Client{ - Timeout: config.Timeout, - }, - userAgent: config.UserAgent, - logger: logger.New("freeproxylist"), - } -} - -func (f *FreeProxyListScraper) Name() string { - return "freeproxylist" -} - -func (f *FreeProxyListScraper) Scrape(ctx context.Context) ([]Proxy, error) { - urls := []string{ - "https://www.proxy-list.download/api/v1/get?type=http", - "https://www.proxy-list.download/api/v1/get?type=https", - "https://www.proxy-list.download/api/v1/get?type=socks4", - "https://www.proxy-list.download/api/v1/get?type=socks5", - } - - var allProxies []Proxy - for _, apiURL := range urls { - proxies, err := f.scrapeURL(ctx, apiURL) - if err != nil { - continue - } - allProxies = append(allProxies, proxies...) - } - - return allProxies, nil -} - -func (f *FreeProxyListScraper) scrapeURL(ctx context.Context, apiURL string) ([]Proxy, error) { - req, err := http.NewRequestWithContext(ctx, "GET", apiURL, nil) - if err != nil { - return nil, err - } - - req.Header.Set("User-Agent", f.userAgent) - - resp, err := f.client.Do(req) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("HTTP %d", resp.StatusCode) - } - - return f.parseProxies(resp.Body, getTypeFromURL(apiURL)) -} - -func (f *FreeProxyListScraper) parseProxies(reader io.Reader, proxyType string) ([]Proxy, error) { - var proxies []Proxy - scanner := bufio.NewScanner(reader) - - for scanner.Scan() { - line := strings.TrimSpace(scanner.Text()) - if line == "" { - continue - } - - parts := strings.Split(line, ":") - if len(parts) != 2 { - continue - } - - port, err := strconv.Atoi(parts[1]) - if err != nil { - continue - } - - proxy := Proxy{ - Host: parts[0], - Port: port, - Type: proxyType, - LastSeen: time.Now(), - } - - proxies = append(proxies, proxy) - } - - return proxies, scanner.Err() -} - -func getTypeFromURL(apiURL string) string { - u, err := url.Parse(apiURL) - if err != nil { - return "http" - } - - query := u.Query() - proxyType := query.Get("type") - if proxyType == "" { - return "http" - } - return proxyType -} \ No newline at end of file diff --git a/pkg/scraper/geonode.go b/pkg/scraper/geonode.go deleted file mode 100644 index e733306..0000000 --- a/pkg/scraper/geonode.go +++ /dev/null @@ -1,124 +0,0 @@ -package scraper - -import ( - "aproxy/internal/logger" - "context" - "encoding/json" - "fmt" - "net/http" - "strconv" - "time" -) - -type GeonodeAPIScraper struct { - client *http.Client - userAgent string - logger *logger.Logger -} - -type GeonodeResponse struct { - Data []GeonodeProxy `json:"data"` - Total int `json:"total"` - Page int `json:"page"` - Limit int `json:"limit"` -} - -type GeonodeProxy struct { - ID string `json:"_id"` - IP string `json:"ip"` - Port string `json:"port"` - Protocols []string `json:"protocols"` - Country string `json:"country"` - AnonymityLevel string `json:"anonymityLevel"` - Latency float64 `json:"latency"` - Speed int `json:"speed"` - UpTime float64 `json:"upTime"` - LastChecked int64 `json:"lastChecked"` - CreatedAt string `json:"created_at"` - UpdatedAt string `json:"updated_at"` -} - -func NewGeonodeAPIScraper() *GeonodeAPIScraper { - return &GeonodeAPIScraper{ - client: &http.Client{ - Timeout: 60 * time.Second, - }, - userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", - logger: logger.New("geonode"), - } -} - -func NewGeonodeAPIScraperWithConfig(config ScraperConfig) *GeonodeAPIScraper { - return &GeonodeAPIScraper{ - client: &http.Client{ - Timeout: config.Timeout, - }, - userAgent: config.UserAgent, - logger: logger.New("geonode"), - } -} - -func (g *GeonodeAPIScraper) Name() string { - return "geonode-api" -} - -func (g *GeonodeAPIScraper) Scrape(ctx context.Context) ([]Proxy, error) { - apiURL := "https://proxylist.geonode.com/api/proxy-list?limit=500" - - req, err := http.NewRequestWithContext(ctx, "GET", apiURL, nil) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - - req.Header.Set("User-Agent", g.userAgent) - req.Header.Set("Accept", "application/json") - - resp, err := g.client.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to fetch data: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("HTTP %d", resp.StatusCode) - } - - var geonodeResp GeonodeResponse - if err := json.NewDecoder(resp.Body).Decode(&geonodeResp); err != nil { - return nil, fmt.Errorf("failed to parse JSON: %w", err) - } - - var proxies []Proxy - httpCount := 0 - socksCount := 0 - - for _, geoProxy := range geonodeResp.Data { - port, err := strconv.Atoi(geoProxy.Port) - if err != nil { - continue - } - - for _, protocol := range geoProxy.Protocols { - proxy := Proxy{ - Host: geoProxy.IP, - Port: port, - Type: protocol, - Country: geoProxy.Country, - LastSeen: time.Now(), - } - - proxies = append(proxies, proxy) - - if protocol == "http" || protocol == "https" { - httpCount++ - } else if protocol == "socks4" || protocol == "socks5" { - socksCount++ - } - } - } - - g.logger.InfoBg("Geonode API collected: %d HTTP/HTTPS, %d SOCKS from %d total proxies", - httpCount, socksCount, len(geonodeResp.Data)) - - return proxies, nil -} \ No newline at end of file diff --git a/pkg/scraper/github.go b/pkg/scraper/github.go deleted file mode 100644 index f5aa7c9..0000000 --- a/pkg/scraper/github.go +++ /dev/null @@ -1,124 +0,0 @@ -package scraper - -import ( - "aproxy/internal/logger" - "bufio" - "context" - "fmt" - "io" - "net/http" - "strconv" - "strings" - "time" -) - -type GitHubProxyScraper struct { - client *http.Client - userAgent string - logger *logger.Logger -} - -func NewGitHubProxyScraper() *GitHubProxyScraper { - return &GitHubProxyScraper{ - client: &http.Client{ - Timeout: 30 * time.Second, - }, - userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", - logger: logger.New("github"), - } -} - -func NewGitHubProxyScraperWithConfig(config ScraperConfig) *GitHubProxyScraper { - return &GitHubProxyScraper{ - client: &http.Client{ - Timeout: config.Timeout, - }, - userAgent: config.UserAgent, - logger: logger.New("github"), - } -} - -func (g *GitHubProxyScraper) Name() string { - return "github" -} - -func (g *GitHubProxyScraper) Scrape(ctx context.Context) ([]Proxy, error) { - url := "https://raw.githubusercontent.com/proxifly/free-proxy-list/refs/heads/main/proxies/all/data.txt" - - proxies, err := g.scrapeTextURL(ctx, url) - if err != nil { - return nil, fmt.Errorf("GitHub proxy scrape failed: %w", err) - } - - httpCount := 0 - for _, proxy := range proxies { - if proxy.Type == "http" || proxy.Type == "https" { - httpCount++ - } - } - - g.logger.InfoBg("GitHub collected: %d HTTP/HTTPS proxies", httpCount) - return proxies, nil -} - -func (g *GitHubProxyScraper) scrapeTextURL(ctx context.Context, url string) ([]Proxy, error) { - req, err := http.NewRequestWithContext(ctx, "GET", url, nil) - if err != nil { - return nil, err - } - - req.Header.Set("User-Agent", g.userAgent) - - resp, err := g.client.Do(req) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("HTTP %d", resp.StatusCode) - } - - return g.parseProtocolProxies(resp.Body) -} - -func (g *GitHubProxyScraper) parseProtocolProxies(reader io.Reader) ([]Proxy, error) { - var proxies []Proxy - scanner := bufio.NewScanner(reader) - - for scanner.Scan() { - line := strings.TrimSpace(scanner.Text()) - if line == "" { - continue - } - - parts := strings.Split(line, "://") - if len(parts) != 2 { - continue - } - - protocol := parts[0] - hostPort := parts[1] - - hostPortParts := strings.Split(hostPort, ":") - if len(hostPortParts) != 2 { - continue - } - - port, err := strconv.Atoi(hostPortParts[1]) - if err != nil { - continue - } - - proxy := Proxy{ - Host: hostPortParts[0], - Port: port, - Type: protocol, - LastSeen: time.Now(), - } - - proxies = append(proxies, proxy) - } - - return proxies, scanner.Err() -} \ No newline at end of file diff --git a/pkg/scraper/list.go b/pkg/scraper/list.go new file mode 100644 index 0000000..9f7c59c --- /dev/null +++ b/pkg/scraper/list.go @@ -0,0 +1,140 @@ +package scraper + +import ( + "aproxy/internal/config" + "aproxy/internal/logger" + "bufio" + "context" + "fmt" + "net/http" + "strconv" + "strings" + "time" +) + +// source describes a plain-text proxy list: one or more URLs returning lines of +// either "proto://host:port" or "host:port". When a line has no protocol prefix, +// defaultType is used. +type source struct { + name string + urls []string + defaultType string // type for bare "host:port" lines +} + +// sources is the registry of text-list proxy providers. Add a row to add a source. +var sources = []source{ + { + name: "proxyscrape", + urls: []string{"https://api.proxyscrape.com/v4/free-proxy-list/get?request=get_proxies&proxy_format=protocolipport&format=text"}, + defaultType: "http", + }, + { + name: "freeproxylist", + urls: []string{ + "https://www.proxy-list.download/api/v1/get?type=http", + "https://www.proxy-list.download/api/v1/get?type=https", + "https://www.proxy-list.download/api/v1/get?type=socks4", + "https://www.proxy-list.download/api/v1/get?type=socks5", + }, + defaultType: "http", // these lists are bare host:port; type is implied by the URL but we can't see it per-line, so default http. ponytail: type fidelity lost here, acceptable—checker probes the real protocol anyway. + }, + { + name: "github", + urls: []string{"https://raw.githubusercontent.com/proxifly/free-proxy-list/refs/heads/main/proxies/all/data.txt"}, + defaultType: "http", + }, + { + name: "proxylistorg", + urls: []string{ + "https://raw.githubusercontent.com/clarketm/proxy-list/master/proxy-list-raw.txt", + "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/http.txt", + }, + defaultType: "http", + }, +} + +// listScraper fetches one source's URLs and parses host:port lines. +type listScraper struct { + src source + client *http.Client + userAgent string + logger *logger.Logger +} + +func newListScraper(src source, config config.ScraperConfig) *listScraper { + return &listScraper{ + src: src, + client: &http.Client{Timeout: config.Timeout}, + userAgent: config.UserAgent, + logger: logger.New(src.name), + } +} + +func (s *listScraper) Name() string { return s.src.name } + +func (s *listScraper) Scrape(ctx context.Context) ([]Proxy, error) { + var all []Proxy + for _, u := range s.src.urls { + proxies, err := s.fetch(ctx, u) + if err != nil { + s.logger.WarnBg("fetch %s failed: %v", u, err) + continue + } + all = append(all, proxies...) + } + s.logger.InfoBg("collected %d proxies", len(all)) + return all, nil +} + +func (s *listScraper) fetch(ctx context.Context, url string) ([]Proxy, error) { + req, err := http.NewRequestWithContext(ctx, "GET", url, nil) + if err != nil { + return nil, err + } + req.Header.Set("User-Agent", s.userAgent) + + resp, err := s.client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("HTTP %d", resp.StatusCode) + } + + var proxies []Proxy + scanner := bufio.NewScanner(resp.Body) + for scanner.Scan() { + if p, ok := parseLine(scanner.Text(), s.src.defaultType); ok { + proxies = append(proxies, p) + } + } + return proxies, scanner.Err() +} + +// parseLine parses "proto://host:port" or "host:port". Returns ok=false for +// blanks, comments, or malformed lines. +func parseLine(line, defaultType string) (Proxy, bool) { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + return Proxy{}, false + } + + typ := defaultType + if scheme, rest, found := strings.Cut(line, "://"); found { + typ = scheme + line = rest + } + + host, portStr, found := strings.Cut(line, ":") + if !found { + return Proxy{}, false + } + port, err := strconv.Atoi(portStr) + if err != nil { + return Proxy{}, false + } + + return Proxy{Host: host, Port: port, Type: typ, LastSeen: time.Now()}, true +} diff --git a/pkg/scraper/list_test.go b/pkg/scraper/list_test.go new file mode 100644 index 0000000..7b8fa1e --- /dev/null +++ b/pkg/scraper/list_test.go @@ -0,0 +1,31 @@ +package scraper + +import "testing" + +func TestParseLine(t *testing.T) { + cases := []struct { + in string + defaultType string + want Proxy // zero Host means expect ok=false + }{ + {"1.2.3.4:8080", "http", Proxy{Host: "1.2.3.4", Port: 8080, Type: "http"}}, + {"socks5://9.9.9.9:1080", "http", Proxy{Host: "9.9.9.9", Port: 1080, Type: "socks5"}}, + {" 5.6.7.8:3128 ", "https", Proxy{Host: "5.6.7.8", Port: 3128, Type: "https"}}, + {"# comment", "http", Proxy{}}, + {"", "http", Proxy{}}, + {"garbage-no-port", "http", Proxy{}}, + {"1.2.3.4:notaport", "http", Proxy{}}, + } + + for _, c := range cases { + got, ok := parseLine(c.in, c.defaultType) + wantOK := c.want.Host != "" + if ok != wantOK { + t.Errorf("parseLine(%q): ok=%v, want %v", c.in, ok, wantOK) + continue + } + if ok && (got.Host != c.want.Host || got.Port != c.want.Port || got.Type != c.want.Type) { + t.Errorf("parseLine(%q) = %+v, want %+v", c.in, got, c.want) + } + } +} diff --git a/pkg/scraper/proxylistorg.go b/pkg/scraper/proxylistorg.go deleted file mode 100644 index 19cda59..0000000 --- a/pkg/scraper/proxylistorg.go +++ /dev/null @@ -1,115 +0,0 @@ -package scraper - -import ( - "aproxy/internal/logger" - "bufio" - "context" - "fmt" - "io" - "net/http" - "strconv" - "strings" - "time" -) - -type ProxyListOrgScraper struct { - client *http.Client - userAgent string - logger *logger.Logger -} - -func NewProxyListOrgScraper() *ProxyListOrgScraper { - return &ProxyListOrgScraper{ - client: &http.Client{ - Timeout: 30 * time.Second, - }, - userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", - logger: logger.New("proxylistorg"), - } -} - -func NewProxyListOrgScraperWithConfig(config ScraperConfig) *ProxyListOrgScraper { - return &ProxyListOrgScraper{ - client: &http.Client{ - Timeout: config.Timeout, - }, - userAgent: config.UserAgent, - logger: logger.New("proxylistorg"), - } -} - -func (p *ProxyListOrgScraper) Name() string { - return "proxylistorg" -} - -func (p *ProxyListOrgScraper) Scrape(ctx context.Context) ([]Proxy, error) { - urls := []string{ - "https://raw.githubusercontent.com/clarketm/proxy-list/master/proxy-list-raw.txt", - "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/http.txt", - } - - var allProxies []Proxy - for _, apiURL := range urls { - proxies, err := p.scrapeURL(ctx, apiURL) - if err != nil { - continue - } - allProxies = append(allProxies, proxies...) - } - - return allProxies, nil -} - -func (p *ProxyListOrgScraper) scrapeURL(ctx context.Context, apiURL string) ([]Proxy, error) { - req, err := http.NewRequestWithContext(ctx, "GET", apiURL, nil) - if err != nil { - return nil, err - } - - req.Header.Set("User-Agent", p.userAgent) - - resp, err := p.client.Do(req) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("HTTP %d", resp.StatusCode) - } - - return p.parseProxies(resp.Body, "http") -} - -func (p *ProxyListOrgScraper) parseProxies(reader io.Reader, proxyType string) ([]Proxy, error) { - var proxies []Proxy - scanner := bufio.NewScanner(reader) - - for scanner.Scan() { - line := strings.TrimSpace(scanner.Text()) - if line == "" || strings.HasPrefix(line, "#") { - continue - } - - parts := strings.Split(line, ":") - if len(parts) != 2 { - continue - } - - port, err := strconv.Atoi(parts[1]) - if err != nil { - continue - } - - proxy := Proxy{ - Host: parts[0], - Port: port, - Type: proxyType, - LastSeen: time.Now(), - } - - proxies = append(proxies, proxy) - } - - return proxies, scanner.Err() -} \ No newline at end of file diff --git a/pkg/scraper/proxyscrape.go b/pkg/scraper/proxyscrape.go deleted file mode 100644 index ee49203..0000000 --- a/pkg/scraper/proxyscrape.go +++ /dev/null @@ -1,134 +0,0 @@ -package scraper - -import ( - "aproxy/internal/logger" - "bufio" - "context" - "fmt" - "io" - "net/http" - "strconv" - "strings" - "time" -) - -type ProxyScrapeAPI struct { - client *http.Client - userAgent string - logger *logger.Logger -} - -func NewProxyScrapeAPI() *ProxyScrapeAPI { - return &ProxyScrapeAPI{ - client: &http.Client{ - Timeout: 30 * time.Second, - }, - userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", - logger: logger.New("proxyscrape"), - } -} - -func NewProxyScrapeAPIWithConfig(config ScraperConfig) *ProxyScrapeAPI { - return &ProxyScrapeAPI{ - client: &http.Client{ - Timeout: config.Timeout, - }, - userAgent: config.UserAgent, - logger: logger.New("proxyscrape"), - } -} - -func (p *ProxyScrapeAPI) Name() string { - return "proxyscrape" -} - -func (p *ProxyScrapeAPI) Scrape(ctx context.Context) ([]Proxy, error) { - urls := []string{ - "https://api.proxyscrape.com/v4/free-proxy-list/get?request=get_proxies&proxy_format=protocolipport&format=text", - } - - var allProxies []Proxy - for _, apiURL := range urls { - proxies, err := p.scrapeTextURL(ctx, apiURL) - if err != nil { - p.logger.WarnBg("ProxyScrape API failed: %v", err) - continue - } - - httpCount := 0 - socksCount := 0 - for _, proxy := range proxies { - allProxies = append(allProxies, proxy) - if proxy.Type == "http" || proxy.Type == "https" { - httpCount++ - } else if proxy.Type == "socks4" || proxy.Type == "socks5" { - socksCount++ - } - } - p.logger.InfoBg("ProxyScrape collected: %d HTTP/HTTPS, %d SOCKS", httpCount, socksCount) - } - - return allProxies, nil -} - -func (p *ProxyScrapeAPI) scrapeTextURL(ctx context.Context, apiURL string) ([]Proxy, error) { - req, err := http.NewRequestWithContext(ctx, "GET", apiURL, nil) - if err != nil { - return nil, err - } - - req.Header.Set("User-Agent", p.userAgent) - - resp, err := p.client.Do(req) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("HTTP %d", resp.StatusCode) - } - - return p.parseProtocolProxies(resp.Body) -} - -func (p *ProxyScrapeAPI) parseProtocolProxies(reader io.Reader) ([]Proxy, error) { - var proxies []Proxy - scanner := bufio.NewScanner(reader) - - for scanner.Scan() { - line := strings.TrimSpace(scanner.Text()) - if line == "" { - continue - } - - parts := strings.Split(line, "://") - if len(parts) != 2 { - continue - } - - protocol := parts[0] - hostPort := parts[1] - - hostPortParts := strings.Split(hostPort, ":") - if len(hostPortParts) != 2 { - continue - } - - port, err := strconv.Atoi(hostPortParts[1]) - if err != nil { - continue - } - - proxy := Proxy{ - Host: hostPortParts[0], - Port: port, - Type: protocol, - LastSeen: time.Now(), - } - - proxies = append(proxies, proxy) - } - - return proxies, scanner.Err() -} \ No newline at end of file diff --git a/pkg/scraper/scraper.go b/pkg/scraper/scraper.go index 522e06d..9154be1 100644 --- a/pkg/scraper/scraper.go +++ b/pkg/scraper/scraper.go @@ -1,6 +1,7 @@ package scraper import ( + "aproxy/internal/config" "aproxy/internal/logger" "context" ) @@ -10,42 +11,16 @@ type MultiScraper struct { logger *logger.Logger } -func NewMultiScraper() *MultiScraper { - return &MultiScraper{ - scrapers: []Scraper{ - NewProxyScrapeAPI(), - NewFreeProxyListScraper(), - NewGeonodeAPIScraper(), - NewGitHubProxyScraper(), - }, - logger: logger.New("multiscraper"), +func NewMultiScraper(config config.ScraperConfig) *MultiScraper { + enabled := make(map[string]bool, len(config.Sources)) + for _, s := range config.Sources { + enabled[s] = true } -} -func NewMultiScraperWithConfig(config ScraperConfig) *MultiScraper { var scrapers []Scraper - - for _, source := range config.Sources { - switch source { - case "proxyscrape": - scrapers = append(scrapers, NewProxyScrapeAPIWithConfig(config)) - case "freeproxylist": - scrapers = append(scrapers, NewFreeProxyListScraperWithConfig(config)) - case "geonode": - scrapers = append(scrapers, NewGeonodeAPIScraperWithConfig(config)) - case "proxylistorg": - scrapers = append(scrapers, NewProxyListOrgScraperWithConfig(config)) - case "github": - scrapers = append(scrapers, NewGitHubProxyScraperWithConfig(config)) - } - } - - if len(scrapers) == 0 { - scrapers = []Scraper{ - NewProxyScrapeAPIWithConfig(config), - NewFreeProxyListScraperWithConfig(config), - NewGeonodeAPIScraperWithConfig(config), - NewGitHubProxyScraperWithConfig(config), + for _, src := range sources { + if len(enabled) == 0 || enabled[src.name] { + scrapers = append(scrapers, newListScraper(src, config)) } } @@ -58,7 +33,6 @@ func NewMultiScraperWithConfig(config ScraperConfig) *MultiScraper { func (m *MultiScraper) ScrapeAll(ctx context.Context) ([]Proxy, error) { var allProxies []Proxy seen := make(map[string]bool) - totalUnique := 0 for _, scraper := range m.scrapers { proxies, err := scraper.Scrape(ctx) @@ -67,20 +41,18 @@ func (m *MultiScraper) ScrapeAll(ctx context.Context) ([]Proxy, error) { continue } - uniqueCount := 0 + unique := 0 for _, proxy := range proxies { key := proxy.Address() if !seen[key] { seen[key] = true allProxies = append(allProxies, proxy) - uniqueCount++ + unique++ } } - - m.logger.InfoBg("Scraper %s: %d total, %d unique", scraper.Name(), len(proxies), uniqueCount) - totalUnique += uniqueCount + m.logger.InfoBg("Scraper %s: %d total, %d unique", scraper.Name(), len(proxies), unique) } - m.logger.InfoBg("Total unique proxies collected: %d", totalUnique) + m.logger.InfoBg("Total unique proxies collected: %d", len(allProxies)) return allProxies, nil -} \ No newline at end of file +} diff --git a/pkg/scraper/types.go b/pkg/scraper/types.go index 4c3f797..20a7186 100644 --- a/pkg/scraper/types.go +++ b/pkg/scraper/types.go @@ -22,9 +22,3 @@ type Scraper interface { Name() string Scrape(ctx context.Context) ([]Proxy, error) } - -type ScraperConfig struct { - Timeout time.Duration - UserAgent string - Sources []string -} diff --git a/sqlc.yaml b/sqlc.yaml new file mode 100644 index 0000000..4001fc7 --- /dev/null +++ b/sqlc.yaml @@ -0,0 +1,19 @@ +version: "2" +sql: + - engine: "sqlite" + schema: "internal/database/schema.sql" + queries: "internal/database/query.sql" + gen: + go: + package: "db" + out: "internal/database/db" + emit_json_tags: false + emit_pointers_for_null_types: true + overrides: + - db_type: "DATETIME" + go_type: "time.Time" + - db_type: "DATETIME" + nullable: true + go_type: + type: "time.Time" + pointer: true