summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--.env.example4
-rw-r--r--cmd/admin-cli/main.go2
-rw-r--r--cmd/server/main.go139
-rw-r--r--handlers/admin.go54
-rw-r--r--handlers/php_proxy.go18
-rw-r--r--handlers/proxy.go13
-rw-r--r--middleware/ratelimit.go13
-rw-r--r--types/types.go9
-rw-r--r--worker/worker.go62
9 files changed, 300 insertions, 14 deletions
diff --git a/.env.example b/.env.example
index 0d60870..70e44e4 100644
--- a/.env.example
+++ b/.env.example
@@ -1,3 +1,7 @@
+# --- Server Settings ---
+# The port for the proxy server to listen on
+LISTEN_PORT=8080
+
# --- Required ---
# The URL of the backend service you want to protect
TARGET_URL=http://localhost:8081
diff --git a/cmd/admin-cli/main.go b/cmd/admin-cli/main.go
index fdcfe2e..91439c4 100644
--- a/cmd/admin-cli/main.go
+++ b/cmd/admin-cli/main.go
@@ -59,4 +59,4 @@ func main() {
}
fmt.Printf("Admin user '%s' created successfully.\n", *username)
-}
+} \ No newline at end of file
diff --git a/cmd/server/main.go b/cmd/server/main.go
index 15ba62b..7d2fa25 100644
--- a/cmd/server/main.go
+++ b/cmd/server/main.go
@@ -6,9 +6,13 @@ import (
"auth-proxy/handlers"
"auth-proxy/logging"
"auth-proxy/middleware"
+ "auth-proxy/models"
+ "auth-proxy/types"
+ "auth-proxy/worker"
"net/http"
"os"
"strings"
+ "time"
"github.com/joho/godotenv"
)
@@ -29,6 +33,11 @@ func main() {
database.InitDB()
defer database.CloseDB()
+ // Initialize and start the worker pool
+ pool := worker.NewWorkerPool()
+ defer worker.Close(pool)
+ handlers.SetWorkerPool(pool)
+
// Get target URL for proxy
targetURL := os.Getenv("TARGET_URL")
if targetURL == "" {
@@ -147,10 +156,130 @@ func main() {
proxyHandler.ServeHTTP(w, r)
})
+ // Get server port
+ port := os.Getenv("LISTEN_PORT")
+ if port == "" {
+ port = "8080"
+ }
+ listenAddr := ":" + port
+
// --- Start Server ---
- logging.AppLog.Info("Server starting on :8080")
- if err := http.ListenAndServe(":8080", finalHandler); err != nil {
- logging.AppLog.Error("Server failed to start", "error", err)
- os.Exit(1)
+ logging.AppLog.Info("Server starting on " + listenAddr)
+
+ // Start periodic background tasks
+
+ go startBackgroundTasks(pool, rateLimiter)
+
+
+
+ if err := http.ListenAndServe(listenAddr, finalHandler); err != nil {
+
+ logging.AppLog.Error("Server failed to start", "error", err)
+
+ os.Exit(1)
+
+ }
+
}
-}
+
+
+
+ // startBackgroundTasks launches periodic jobs in the background.
+
+ func startBackgroundTasks(pool *types.WorkerPool, limiter *types.RateLimiter) {
+
+ // Ticker for expired token cleanup (every hour)
+
+ tokenTicker := time.NewTicker(1 * time.Hour)
+
+ defer tokenTicker.Stop()
+
+
+
+ // Ticker for rate limiter cleanup (every 10 minutes)
+
+ rateLimiterTicker := time.NewTicker(24 * time.Hour)
+
+ defer rateLimiterTicker.Stop()
+
+
+
+ logging.AppLog.Info("Background task scheduler started")
+
+
+
+ // Run initial cleanup tasks at startup
+
+ submitTokenCleanup(pool)
+
+ submitRateLimiterCleanup(pool, limiter)
+
+
+
+ // Run tasks on their respective tickers
+
+ for {
+
+ select {
+
+ case <-tokenTicker.C:
+
+ submitTokenCleanup(pool)
+
+ case <-rateLimiterTicker.C:
+
+ submitRateLimiterCleanup(pool, limiter)
+
+ }
+
+ }
+
+ }
+
+
+
+ // submitTokenCleanup submits the expired token deletion task to the worker pool.
+
+ func submitTokenCleanup(pool *types.WorkerPool) {
+
+ worker.Submit(pool, func() {
+
+ logging.AppLog.Info("Running background task: deleting expired tokens")
+
+ rowsAffected, err := models.DeleteExpiredTokens()
+
+ if err != nil {
+
+ logging.AppLog.Error("Failed to delete expired tokens", "error", err)
+
+ } else {
+
+ if rowsAffected > 0 {
+
+ logging.AppLog.Info("Finished deleting expired tokens", "deleted_count", rowsAffected)
+
+ }
+
+ }
+
+ })
+
+ }
+
+
+
+ // submitRateLimiterCleanup submits the rate limiter cleanup task to the worker pool.
+
+ func submitRateLimiterCleanup(pool *types.WorkerPool, limiter *types.RateLimiter) {
+
+ worker.Submit(pool, func() {
+
+ logging.AppLog.Info("Running background task: cleaning up rate limiter")
+
+ middleware.Cleanup(limiter)
+
+ })
+
+ }
+
+
diff --git a/handlers/admin.go b/handlers/admin.go
index 44c36a5..61519fa 100644
--- a/handlers/admin.go
+++ b/handlers/admin.go
@@ -5,11 +5,22 @@ import (
"auth-proxy/logging"
"auth-proxy/models"
"auth-proxy/types"
+ "auth-proxy/worker"
"encoding/json"
+ "fmt"
"net/http"
"strconv"
+ "sync"
+ "time"
)
+var workerPool *types.WorkerPool
+
+// SetWorkerPool sets the package-level worker pool instance.
+func SetWorkerPool(pool *types.WorkerPool) {
+ workerPool = pool
+}
+
// AdminPageHandler serves the admin dashboard page.
func AdminPageHandler(w http.ResponseWriter, r *http.Request) {
// Fetch all users to render on the page
@@ -24,12 +35,20 @@ func AdminPageHandler(w http.ResponseWriter, r *http.Request) {
"Paths": config.Paths,
"Users": users,
}
- if err := templates.ExecuteTemplate(w, "admin.html", data); err != nil {
+ if err := templates.ExecuteTemplate(w, "admin.html", data); err != nil {
logging.AppLog.Error("Failed to execute admin template", "error", err)
}
}
-// GetUsersHandler returns a list of all users.
+// EnrichedUser is a wrapper for User with additional computed fields.
+// Note: We use pointers to the original User to avoid copying large structs.
+type EnrichedUser struct {
+ *types.User
+ TimeSinceLogin string `json:"time_since_login"`
+}
+
+// GetUsersHandler returns a list of all users with enriched information.
+// It uses a worker pool to calculate extra information for each user in parallel.
func GetUsersHandler(w http.ResponseWriter, r *http.Request) {
ip := logging.GetClientIP(r)
users, err := models.GetAllUsers()
@@ -39,8 +58,37 @@ func GetUsersHandler(w http.ResponseWriter, r *http.Request) {
return
}
+ enrichedUsers := make([]EnrichedUser, len(users))
+ var wg sync.WaitGroup
+
+ for i, user := range users {
+ wg.Add(1)
+ // Create copies of loop variables to avoid race conditions in the closure
+ index := i
+ userCopy := user
+
+ task := func() {
+ defer wg.Done()
+ // Simulate some work, e.g., calculating a value or fetching related data.
+ time.Sleep(5 * time.Millisecond)
+
+ timeSinceLogin := "N/A"
+ if userCopy.LastLoginAt.Valid {
+ timeSinceLogin = fmt.Sprintf("%.0fm ago", time.Since(userCopy.LastLoginAt.Time).Minutes())
+ }
+
+ enrichedUsers[index] = EnrichedUser{
+ User: &userCopy,
+ TimeSinceLogin: timeSinceLogin,
+ }
+ }
+ worker.Submit(workerPool, task)
+ }
+
+ wg.Wait() // Wait for all tasks to complete
+
w.Header().Set("Content-Type", "application/json")
- if err := json.NewEncoder(w).Encode(users); err != nil {
+ if err := json.NewEncoder(w).Encode(enrichedUsers); err != nil {
logging.AppLog.Error("Failed to encode users to JSON", "error", err, "ip", ip)
}
}
diff --git a/handlers/php_proxy.go b/handlers/php_proxy.go
index 98b62ba..d30f363 100644
--- a/handlers/php_proxy.go
+++ b/handlers/php_proxy.go
@@ -3,6 +3,7 @@ package handlers
import (
"auth-proxy/config"
"auth-proxy/logging"
+ "auth-proxy/worker"
"bytes"
"io"
"io/ioutil"
@@ -28,6 +29,11 @@ func NewPhpProxyHandler(scriptName string) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
fcgi, err := fcgiclient.Dial("unix", phpFpmSocket)
if err != nil {
+ errStr := err.Error()
+ ip := logging.GetClientIP(r)
+ worker.Submit(workerPool, func() {
+ logging.AppLog.Error("Failed to connect to php-fpm", "error", errStr, "ip", ip)
+ })
http.Error(w, "Failed to connect to php-fpm: "+err.Error(), http.StatusInternalServerError)
return
}
@@ -73,7 +79,10 @@ func NewPhpProxyHandler(scriptName string) http.Handler {
env["AUTH_ASSETS_PATH"] = config.Paths.Assets
- logging.AppLog.Debug("Proxying to PHP-FPM", "script_filename", scriptFilename, "doc_root", finalDocRoot)
+ // Asynchronously log the proxying action to avoid blocking the request.
+ worker.Submit(workerPool, func() {
+ logging.AppLog.Debug("Proxying to PHP-FPM", "script_filename", scriptFilename, "doc_root", finalDocRoot)
+ })
env["REQUEST_METHOD"] = r.Method
env["SERVER_SOFTWARE"] = "go / auth-proxy"
@@ -99,7 +108,12 @@ func NewPhpProxyHandler(scriptName string) http.Handler {
resp, err := fcgi.Request(env, body)
if err != nil {
- http.Error(w, "FastCGI request failed: "+err.Error(), http.StatusInternalServerError)
+ ip := logging.GetClientIP(r)
+ errStr := err.Error()
+ worker.Submit(workerPool, func() {
+ logging.AppLog.Error("FastCGI request failed", "error", errStr, "ip", ip)
+ })
+ http.Error(w, "FastCGI request failed: "+err.Error(), http.StatusInternalServerError)
return
}
diff --git a/handlers/proxy.go b/handlers/proxy.go
index 602fc78..d1f6def 100644
--- a/handlers/proxy.go
+++ b/handlers/proxy.go
@@ -2,6 +2,7 @@ package handlers
import (
"auth-proxy/logging"
+ "auth-proxy/worker"
"net/http"
"net/http/httputil"
"net/url"
@@ -22,15 +23,21 @@ func NewProxy(targetURL string) http.Handler {
director(req) // Run the default director
req.Host = target.Host // Set the Host header to the target's host
- // Log the proxying action
+ // Asynchronously log the proxying action to avoid blocking the request.
+ urlStr := req.URL.String()
ip := logging.GetClientIP(req)
- logging.AppLog.Info("Proxying request", "url", req.URL.String(), "ip", ip)
+ worker.Submit(workerPool, func() {
+ logging.AppLog.Info("Proxying request", "url", urlStr, "ip", ip)
+ })
}
// Custom error handler for logging
proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
ip := logging.GetClientIP(r)
- logging.AppLog.Error("Proxy error", "error", err, "ip", ip)
+ // Asynchronously log the proxy error to avoid blocking the response.
+ worker.Submit(workerPool, func() {
+ logging.AppLog.Error("Proxy error", "error", err, "ip", ip)
+ })
http.Error(w, "Proxy Error", http.StatusBadGateway)
}
diff --git a/middleware/ratelimit.go b/middleware/ratelimit.go
index fcb38cc..944f863 100644
--- a/middleware/ratelimit.go
+++ b/middleware/ratelimit.go
@@ -67,3 +67,16 @@ func RateLimitMiddleware(l *types.RateLimiter) func(http.Handler) http.Handler {
})
}
}
+
+// Cleanup iterates through the rate limiter's request map and removes entries
+// for IPs that have no recent requests.
+func Cleanup(l *types.RateLimiter) {
+ l.Mutex.Lock()
+ defer l.Mutex.Unlock()
+
+ for ip, requests := range l.Requests {
+ if len(requests) == 0 {
+ delete(l.Requests, ip)
+ }
+ }
+}
diff --git a/types/types.go b/types/types.go
index 14c4db0..a91fbcd 100644
--- a/types/types.go
+++ b/types/types.go
@@ -72,3 +72,12 @@ type RateLimiter struct {
Max int
Window time.Duration
}
+
+// WorkerPool manages a pool of goroutines to execute tasks concurrently.
+type WorkerPool struct {
+ Workers int
+ TaskQueue chan func()
+ ResultChan chan interface{}
+ Wg sync.WaitGroup
+ Active int64
+}
diff --git a/worker/worker.go b/worker/worker.go
new file mode 100644
index 0000000..ccb9939
--- /dev/null
+++ b/worker/worker.go
@@ -0,0 +1,62 @@
+package worker
+
+import (
+ "auth-proxy/types"
+ "runtime"
+ "sync/atomic"
+)
+
+// NewWorkerPool creates and starts a new worker pool.
+// The number of workers is determined by the number of CPU cores, with a minimum of 2 and a max of 16.
+func NewWorkerPool() *types.WorkerPool {
+ numWorkers := runtime.NumCPU()
+ if numWorkers > 16 {
+ numWorkers = 16
+ }
+ if numWorkers < 2 {
+ numWorkers = 2
+ }
+
+ pool := &types.WorkerPool{
+ Workers: numWorkers,
+ TaskQueue: make(chan func(), numWorkers*2),
+ ResultChan: make(chan interface{}, numWorkers*2),
+ Active: 0,
+ }
+
+ for i := 0; i < numWorkers; i++ {
+ pool.Wg.Add(1)
+ go worker(pool)
+ }
+
+ return pool
+}
+
+// worker is a single worker goroutine that processes tasks from the TaskQueue.
+func worker(p *types.WorkerPool) {
+ defer p.Wg.Done()
+ for task := range p.TaskQueue {
+ atomic.AddInt64(&p.Active, 1)
+ task()
+ atomic.AddInt64(&p.Active, -1)
+ }
+}
+
+// Submit adds a task to the worker pool's queue.
+// This is a blocking operation if the queue is full.
+func Submit(p *types.WorkerPool, task func()) {
+ p.TaskQueue <- task
+}
+
+// ActiveWorkers returns the number of currently active (processing) workers.
+func ActiveWorkers(p *types.WorkerPool) int64 {
+ return atomic.LoadInt64(&p.Active)
+}
+
+// Close gracefully shuts down the worker pool.
+// It closes the task queue and waits for all workers to finish their current tasks.
+func Close(p *types.WorkerPool) {
+ close(p.TaskQueue)
+ p.Wg.Wait()
+ close(p.ResultChan)
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-15 14:03:39 +0000