Architecture Upgrade - LavinMQ Queue Migration, Neon DB Optimization, Calendar Validation & Telemetry

README.md

@@ -25,13 +25,13 @@ ContestSync is a robust, highly optimized web platform and distributed worker sy
 
 | Platform | Scraper/API Type | Fetch Payload Format | Update Frequency | Rate Limit Strategy |
 | :--- | :--- | :--- | :--- | :--- |
-| **LeetCode** | GraphQL Endpoint | JSON Query Payload | Daily Cron Trigger | Dynamic Backoff + Retries |
-| **Codeforces** | REST API Filter | HTTP JSON Response | Daily Cron Trigger | Local Token Bucket Gating |
-| **CodeChef** | REST API Fetcher | Nested JSON Payload | Daily Cron Trigger | Session Connection Pool |
-| **AtCoder** | HTML Scraper | DOM Node Parsing | Daily Cron Trigger | Strict User Agent Gating |
-| **HackerRank** | REST API Fetcher | Flat JSON Document | Daily Cron Trigger | Secure Host Verification |
-| **GeeksforGeeks**| REST API Parser | Raw JSON Payload | Daily Cron Trigger | Bounded Payload Capping |
-| **Naukri Code360**| REST API Parser | Structured JSON Array | Daily Cron Trigger | Dynamic Event Mapping |
+| **LeetCode** | GraphQL Endpoint | JSON Query Payload | On startup, then every 30m | Dynamic Backoff + Retries |
+| **Codeforces** | REST API Filter | HTTP JSON Response | On startup, then every 30m | Local Token Bucket Gating |
+| **CodeChef** | REST API Fetcher | Nested JSON Payload | On startup, then every 30m | Session Connection Pool |
+| **AtCoder** | HTML Scraper | DOM Node Parsing | On startup, then every 30m | Strict User Agent Gating |
+| **HackerRank** | REST API Fetcher | Flat JSON Document | On startup, then every 30m | Secure Host Verification |
+| **GeeksforGeeks**| REST API Parser | Raw JSON Payload | On startup, then every 30m | Bounded Payload Capping |
+| **Naukri Code360**| REST API Parser | Structured JSON Array | On startup, then every 30m | Dynamic Event Mapping |
 
 ---
 
@@ -45,37 +45,70 @@ ContestSync is a robust, highly optimized web platform and distributed worker sy
 
 ## Core SaaS & Enterprise Capabilities
 
-### 1. High-Concurrency Neon DB Read/Write Split Engine
+### 1. High-Concurrency CloudAMQP LavinMQ Queue Engine
+
+* **Ultra-Light & Blazing Fast**: Powered by CloudAMQP LavinMQ (AMQP 0.9.1), consuming extraction tasks (concurrency limit 3) and user sync tasks (concurrency limit 10) with near-zero idle memory footprint (~15MB RAM).
+* **Transparent Fallback**: Automatic, zero-configuration fallback to thread-safe in-memory channels (`chan`) when neither `AMQP_URL` nor `CLOUDAMQP_URL` is configured, or if AMQP connection initialization fails. When operating in-memory, unconsumed in-flight tasks may be lost on process restart.
+* **Persistent Task Queuing**: Manages persistent `extraction-tasks` and `sync-tasks` queues with automatic retry handling.
+
+### 2. High-Concurrency Neon DB Read/Write Split Engine
+
 * **Isolation of Concerns**: Database operations are split into separate Write Primary and multiple Read Replica connection pools.
 * **Atomic Round-Robin Distribution**: Reads are distributed across replicas using lock-free atomic round-robin counters, utilizing the Neon DB pool endpoints for high performance.
 * **Massive Concurrency Support**: Integrates PgBouncer configurations, allowing up to 10,000 concurrent pooled connections per read replica, while direct connections are preserved for write paths and migrations.
 * **In-Memory CA Validation**: Implements in-memory PEM certificate loading for database connections, avoiding disk leakage issues.
 
-### 2. Multi-Tier Distributed Valkey Caching Model
+### 3. Multi-Tier Distributed Valkey Caching Model
 
 | Caching Area | Cache Key Format | Time To Live (TTL) | Eviction Trigger | Storage Format |
 | :--- | :--- | :--- | :--- | :--- |
 | **Contests List** | `cache:contests:<platform>` | 12 Hours | Crawler batch finished | JSON Contest Array |
-| **User Preferences** | `cache:user:<userID>` | 24 Hours | Google OAuth callback / Preferences save / Account deletion | JSON Profile Struct |
+| **User Preferences** | `cache:user:v2:<userID>` | 24 Hours | Google OAuth callback / Preferences save / Account deletion | JSON Profile Struct |
+| **Calendar Validation**| `user:cal_val:<userID>` | 15 Min (Success/Credential) / 1 Min (Operational) | Google OAuth login / Token Revocation | JSON Status Struct |
 | **Synced Events** | `cache:synced_events:<userID>` | 24 Hours | End of SyncUser run (if new sync recorded) | JSON String Array |
 | **Platforms List**| `cache:platforms` | 24 Hours | Static Compile (None) | Pre-serialized JSON |
-| **IP Rate Limiter** | `limit:<ip_address>:<window>` | Variable (Dynamic) | Auto-Expires on window end | Numeric string counter |
+| **IP Rate Limiter** | `ratelimit:<route_pattern>:<ip>` | Variable (Dynamic) | Auto-Expires on window end | Numeric string counter |
 | **User Sessions** | `session:<session_id>` | 7 Days | Account logout | Encoded Gorilla Session |
 
-### 3. Real-Time Contest Overwrite & Obsolete Contest Pruning
+### 4. Cached Calendar Scope Validation & 24h Ungranted User Pruning
+
+* **Cached Scope Verification**: `GET /auth/calendar/validate` checks user Google Calendar access and returns cached validation status from Valkey (`user:cal_val:<userID>`, 15m for valid/credential failure, 1m for operational failure), invalidated on OAuth login or token revoke.
+* **Frontend Warning State**: Displays explicit **Google Calendar Access Missing** alerts for missing/revoked tokens and **Temporary Calendar Service Issues** for API outages.
+* **24-Hour Automated Cleanup**: Background cron task automatically deletes accounts with ungranted tokens (`refresh_token IS NULL`) after 24 hours to prevent orphan DB records.
+
+### 5. Scheduled Contest Overwrite & Obsolete Contest Pruning
+
 * **Dynamic Pruning**: When scrapers run, a cleanup operation is executed: `DELETE FROM contests WHERE platform = $1 AND start_time > NOW() AND id != ALL($2)`. This deletes upcoming contests that were rescheduled or cancelled on the host platform.
 * **Boundary Integrity**: The prune query is locked to upcoming contests (`start_time > NOW()`). This preserves past contests and their synced events, preventing redundant event syncs or double-syncing.
-* **Safe Overwrite**: If a scrape returns zero upcoming contests, all future scheduled entries for that platform are deleted.
+* **Safe Overwrite**: If a scrape returns zero upcoming contests, future scheduled entries for that platform are preserved to protect against API blocks or temporary empty responses.
+
+---
 
-### 4. Security Controls Matrix (Zero-Trust Security Gating)
+## Detailed Environment Configurations
 
-| Threat Vector | Security Countermeasure | Implementation Layer | Target Boundary |
+| Environment Variable | Description | Default Value | Example Value |
 | :--- | :--- | :--- | :--- |
-| **Token Exposure** | AES-256-GCM Cryptographic Envelope | `internal/auth` | Google OAuth Refresh Tokens |
-| **CSRF Exploit** | Session-Bound Verification Gating | `internal/api` | State-Modifying Endpoints (`POST`/`DELETE`) |
-| **Session Fixation** | Complete Session ID Regeneration | `internal/api` | Google Callback Post-Authorization Handshake |
-| **Memory Exhaustion** | Body Limit Constraints & Monitored Headers | `internal/api` | Capped Request Payload (1MB) / Headers (256B) |
-| **IP Spoofing** | Environment-Gated Trust Gating | `internal/api` | `TRUST_PROXY` Toggle validation |
+| `POSTGRES_DB` | Connection URL for Primary Write Postgres Database | None (Required) | `postgres://user:pass@host:port/db?sslmode=require` |
+| `POSTGRES_READ_DB` | Comma-separated Connection URLs for Read Replica Databases | None | `postgres://user:pass@rep1:port/db,postgres://user:pass@rep2:port/db` |
+| `CONNECTION_LIMIT` | Maximum connections allowed in Primary Write pool | `800` | `20` |
+| `CONNECTION_POOL_LIMIT`| Maximum connections allowed per Read Replica Pool | `10000` | `100` |
+| `VALKEY_URI` | Connection URI string for Valkey instance | None | `rediss://default:password@host:port` |
+| `CLOUDAMQP_URL` / `AMQP_URL` | AMQP 0.9.1 connection URL for CloudAMQP LavinMQ (fallback: in-memory) | None | `amqps://user:pass@lemming.rmq.cloudamqp.com/vhost` |
+| `GOOGLE_CLIENT_ID` | OAuth 2.0 Web Application Client ID from Google Cloud Console | None | `abc-123.apps.googleusercontent.com` |
+| `GOOGLE_CLIENT_SECRET` | OAuth 2.0 Web Application Client Secret from Google Cloud Console | None | `sec_code_xyz` |
+| `GOOGLE_REDIRECT_URL` | Redirect Callback URL registered in Google API credentials | None | `http://localhost:8080/auth/google/callback` |
+| `SESSION_SECRET` | 32-byte (64 hex characters) key for Gorilla Cookie / Valkey Sessions | None | `<64-hex-character-secret>` (generate via `openssl rand -hex 32`) |
+| `ENCRYPTION_KEY` | 32-byte (64 hex characters) key for AES Refresh Token encryption | None | `<64-hex-character-secret>` (generate via `openssl rand -hex 32`) |
+| `ADMIN_PASSWORD` | Standard password for admin panel authentication | None | `adm_pwd_secure` |
+| `TRUST_PROXY` | Gated verification trust toggle for client IP forwarding | `false` | `true` |
+| `TELEGRAM_PROXY_URL` | Outbound Proxy target for forwarding warnings/errors | None | `https://tg-proxy.myorg.workers.dev` |
+| `PROXY_SECRET_KEY` | Token key for proxy HTTP authorization | None | `sec_key_abc` |
+| `TELEGRAM_GROUP_ID` | Group Identifier target for slog alerts | None | `-1002938475` |
+| `TELEGRAM_GROUP_TOPIC_ID`| Topic Thread ID inside Telemetry Group | None | `12` |
+| `FROM_ENV` | Application node identifier string for diagnostics | None | `Server-Node-Staging` |
+| `PORT` | API Web Server HTTP listener port | `8080` | `7860` |
+| `WORKER_PORT` | Background Worker Micro-Health server HTTP port | `8081` | `8082` |
+| `ENV` | System execution environment gating flag | `production` | `development` |
 
 ---
 
@@ -127,8 +160,8 @@ sequenceDiagram
     actor User as Competitive Programmer
     participant Browser as Web Browser
     participant Server as Server Binary (cmd/server)
-    participant Queue as Queue Broker (Kafka / Local)
-    participant Worker as Background Worker (cmd/worker)
+    participant Queue as Queue Broker (LavinMQ / In-Memory)
+    participant Worker as Background Worker (cmd/worker / cmd/server)
     participant Valkey as Valkey Cache
     participant DB_Read as Neon DB Replicas (Round-Robin)
     participant DB_Write as Neon DB Primary
@@ -204,7 +237,7 @@ sequenceDiagram
 │   ├── auth  ─────────── Cryptographic AES-256-GCM Encryption/Decryption Modules
 │   ├── db  ───────────── Database Initializer, Pool Splits, Replica Round-Robin counters
 │   ├── extractor  ────── Platform-Specific HTML/JSON Fetchers & Active Parsers
-│   ├── queue  ────────── Distributed Kafka / Local Fallback Queue Broker
+│   ├── queue  ────────── CloudAMQP LavinMQ / In-Memory Fallback Queue Broker
 │   ├── observability  ── Asynchronous Telegram Warnings and Error Diagnostics Telemetry
 │   ├── scheduler  ────── robfig/cron background trigger orchestration
 │   └── sync  ─────────── Synchronization Engine & Deterministic Google Calendar Sync
@@ -224,17 +257,14 @@ sequenceDiagram
 | `CONNECTION_LIMIT` | Maximum connections allowed in Primary Write pool | `800` | `20` |
 | `CONNECTION_POOL_LIMIT`| Maximum connections allowed per Read Replica Pool | `10000` | `100` |
 | `VALKEY_URI` | Connection URI string for Valkey instance | None | `rediss://default:password@host:port` |
+| `CLOUDAMQP_URL` / `AMQP_URL` | AMQP 0.9.1 connection URL for CloudAMQP LavinMQ (fallback: in-memory) | None | `amqps://user:pass@lemming.rmq.cloudamqp.com/vhost` |
 | `GOOGLE_CLIENT_ID` | OAuth 2.0 Web Application Client ID from Google Cloud Console | None | `abc-123.apps.googleusercontent.com` |
 | `GOOGLE_CLIENT_SECRET` | OAuth 2.0 Web Application Client Secret from Google Cloud Console | None | `sec_code_xyz` |
 | `GOOGLE_REDIRECT_URL` | Redirect Callback URL registered in Google API credentials | None | `http://localhost:8080/auth/google/callback` |
-| `SESSION_SECRET` | 32-byte hex-encoded key for Gorilla Cookie / Valkey Sessions | None | `d3b07384d113edec49eaa6238ad5ff00` |
-| `ENCRYPTION_KEY` | 32-byte hex-encoded key for AES Refresh Token encryption | None | `2d9bb20065718dfdc0237af8ad3ff49a` |
+| `SESSION_SECRET` | 32-byte (64 hex characters) key for Gorilla Cookie / Valkey Sessions | None | `<64-hex-character-secret>` (generate via `openssl rand -hex 32`) |
+| `ENCRYPTION_KEY` | 32-byte (64 hex characters) key for AES Refresh Token encryption | None | `<64-hex-character-secret>` (generate via `openssl rand -hex 32`) |
 | `ADMIN_PASSWORD` | Standard password for admin panel authentication | None | `adm_pwd_secure` |
 | `TRUST_PROXY` | Gated verification trust toggle for client IP forwarding | `false` | `true` |
-| `KAFKA_HOST` | Host Address of Apache Kafka broker | None | `kafka.service.consul` |
-| `KAFKA_PORT` | Port number of Apache Kafka broker | `9092` | `9094` |
-| `KAFKA_PARTITIONS` | Dynamic partition count generated for topics | `2` | `4` |
-| `KAFKA_REPLICATION_FACTOR`| Dynamic replication count set on topic creation | `1` | `2` |
 | `TELEGRAM_PROXY_URL` | Outbound Proxy target for forwarding warnings/errors | None | `https://tg-proxy.myorg.workers.dev` |
 | `PROXY_SECRET_KEY` | Token key for proxy HTTP authorization | None | `sec_key_abc` |
 | `TELEGRAM_GROUP_ID` | Group Identifier target for slog alerts | None | `-1002938475` |

cmd/server/main.go

@@ -57,6 +57,11 @@ func main() {
 		defer tgManager.Drain()
 	}
 
+	otelMetrics := observability.InitOTel("contestsync-server")
+	if otelMetrics != nil && otelMetrics.Shutdown != nil {
+		defer otelMetrics.Shutdown(context.Background())
+	}
+
 	if len(cfg.EncryptionKey) != 32 {
 		log.Fatal("ENCRYPTION_KEY must be exactly 32 bytes")
 	}
@@ -120,12 +125,19 @@ func main() {
 		EncryptionKey: cfg.EncryptionKey,
 		Valkey:        valkeyClient,
 	}
+	if tgManager != nil {
+		syncer.OnTelegramEvent = tgManager.TriggerSystemEvent
+	}
 
 	q, err := queue.New(cfg, pool.WriteDB(), syncer)
 	if err != nil {
-		log.Fatalf("failed to initialize kafka queue: %v", err)
+		log.Fatalf("failed to initialize queue: %v", err)
 	}
 	defer q.Close()
+	q.OTel = otelMetrics
+	if tgManager != nil {
+		q.OnTelegramEvent = tgManager.TriggerSystemEvent
+	}
 	q.StartConsumers(shutdownCtx, cfg)
 
 	handlers := &api.Handlers{
@@ -166,6 +178,8 @@ func main() {
 
 	r.HandleFunc("/health", adminHandlers.HealthCheck)
 
+	r.Get("/feed/ical", handlers.ServeICalFeed)
+
 	r.Get("/auth/google", handlers.GoogleLogin)
 	r.Get("/auth/google/callback", handlers.GoogleCallback)
 
@@ -181,6 +195,11 @@ func main() {
 		r.Get("/me", handlers.Me)
 		r.Get("/platforms", handlers.GetPlatforms)
 
+		r.Group(func(r chi.Router) {
+			r.Use(api.RateLimitMiddleware(valkeyClient, 60, time.Minute))
+			r.Get("/auth/calendar/validate", handlers.ValidateCalendarAccess)
+		})
+
 		r.Group(func(r chi.Router) {
 			r.Use(api.CSRFMiddleware(sessionStore))
 			r.Post("/preferences", handlers.SavePreferences)

cmd/worker/main.go

@@ -53,6 +53,11 @@ func main() {
 		defer tgManager.Drain()
 	}
 
+	otelMetrics := observability.InitOTel("contestsync-worker")
+	if otelMetrics != nil && otelMetrics.Shutdown != nil {
+		defer otelMetrics.Shutdown(context.Background())
+	}
+
 	if len(cfg.EncryptionKey) != 32 {
 		log.Fatal("ENCRYPTION_KEY must be exactly 32 bytes")
 	}
@@ -91,12 +96,18 @@ func main() {
 		EncryptionKey: cfg.EncryptionKey,
 		Valkey:        valkeyClient,
 	}
+	if tgManager != nil {
+		syncer.OnTelegramEvent = tgManager.TriggerSystemEvent
+	}
 
 	q, err := queue.New(cfg, pool.WriteDB(), syncer)
 	if err != nil {
-		log.Fatalf("failed to initialize kafka queue: %v", err)
+		log.Fatalf("failed to initialize queue: %v", err)
 	}
 	defer q.Close()
+	if tgManager != nil {
+		q.OnTelegramEvent = tgManager.TriggerSystemEvent
+	}
 	q.StartConsumers(shutdownCtx, cfg)
 
 	mux := http.NewServeMux()

config/config.go

@@ -8,49 +8,29 @@ import (
 )
 
 type Config struct {
-	DatabaseURL            string
-	ReadDatabaseURLs       string
-	ConnectionLimit        int
-	ConnectionPoolLimit    int
-	GoogleClientID         string
-	GoogleClientSecret     string
-	GoogleRedirectURL      string
-	SessionSecret          []byte
-	EncryptionKey          []byte
-	Port                   string
-	WorkerPort             string
-	Env                    string
-	AdminPassword          string
-	KafkaHost              string
-	KafkaPort              string
-	KafkaAccessKey         []byte
-	KafkaAccessCert        []byte
-	KafkaCACert            []byte
-	KafkaPartitions        int
-	KafkaReplicationFactor int
-	ValkeyURI              string
-	TelegramProxyURL       string
-	ProxySecretKey         string
-	TelegramGroupID        string
-	TelegramGroupTopicID   string
-	From                   string
+	DatabaseURL          string
+	ReadDatabaseURLs     string
+	ConnectionLimit      int
+	ConnectionPoolLimit  int
+	GoogleClientID       string
+	GoogleClientSecret   string
+	GoogleRedirectURL    string
+	SessionSecret        []byte
+	EncryptionKey        []byte
+	Port                 string
+	WorkerPort           string
+	Env                  string
+	AdminPassword        string
+	AMQPURL              string
+	ValkeyURI            string
+	TelegramProxyURL     string
+	ProxySecretKey       string
+	TelegramGroupID      string
+	TelegramGroupTopicID string
+	From                 string
 }
 
 func Load() *Config {
-	kafkaAccessKey := []byte(os.Getenv("KAFKA_ACCESS_KEY"))
-	kafkaAccessCert := []byte(os.Getenv("KAFKA_ACCESS_CERTIFICATE"))
-	kafkaCACert := []byte(os.Getenv("KAFKA_CA_CERTIFICATE"))
-
-	kafkaPartitions, err := strconv.Atoi(os.Getenv("KAFKA_PARTITIONS"))
-	if err != nil || kafkaPartitions <= 0 {
-		kafkaPartitions = 2
-	}
-
-	kafkaRepFactor, err := strconv.Atoi(os.Getenv("KAFKA_REPLICATION_FACTOR"))
-	if err != nil || kafkaRepFactor <= 0 {
-		kafkaRepFactor = 1
-	}
-
 	connLimit, _ := strconv.Atoi(os.Getenv("CONNECTION_LIMIT"))
 	if connLimit == 0 {
 		connLimit = 800
@@ -93,32 +73,31 @@ func Load() *Config {
 		secretKey = os.Getenv("ProxySecretKey")
 	}
 
+	amqpURL := os.Getenv("CLOUDAMQP_URL")
+	if amqpURL == "" {
+		amqpURL = os.Getenv("AMQP_URL")
+	}
+
 	return &Config{
-		DatabaseURL:            os.Getenv("POSTGRES_DB"),
-		ReadDatabaseURLs:       os.Getenv("POSTGRES_READ_DB"),
-		ConnectionLimit:        connLimit,
-		ConnectionPoolLimit:    connPoolLimit,
-		GoogleClientID:         os.Getenv("GOOGLE_CLIENT_ID"),
-		GoogleClientSecret:     os.Getenv("GOOGLE_CLIENT_SECRET"),
-		GoogleRedirectURL:      os.Getenv("GOOGLE_REDIRECT_URL"),
-		SessionSecret:          sessionSecret,
-		EncryptionKey:          encryptionKey,
-		Port:                   port,
-		WorkerPort:             workerPort,
-		Env:                    env,
-		AdminPassword:          os.Getenv("ADMIN_PASSWORD"),
-		KafkaHost:              os.Getenv("KAFKA_HOST"),
-		KafkaPort:              os.Getenv("KAFKA_PORT"),
-		KafkaAccessKey:         kafkaAccessKey,
-		KafkaAccessCert:        kafkaAccessCert,
-		KafkaCACert:            kafkaCACert,
-		KafkaPartitions:        kafkaPartitions,
-		KafkaReplicationFactor: kafkaRepFactor,
-		ValkeyURI:              os.Getenv("VALKEY_URI"),
-		TelegramProxyURL:       os.Getenv("TG_PROXY_URL"),
-		ProxySecretKey:         secretKey,
-		TelegramGroupID:        os.Getenv("TG_GROUP_ID"),
-		TelegramGroupTopicID:   os.Getenv("TG_GROUP_TOPIC_ID"),
-		From:                   os.Getenv("FROM"),
+		DatabaseURL:          os.Getenv("POSTGRES_DB"),
+		ReadDatabaseURLs:     os.Getenv("POSTGRES_READ_DB"),
+		ConnectionLimit:      connLimit,
+		ConnectionPoolLimit:  connPoolLimit,
+		GoogleClientID:       os.Getenv("GOOGLE_CLIENT_ID"),
+		GoogleClientSecret:   os.Getenv("GOOGLE_CLIENT_SECRET"),
+		GoogleRedirectURL:    os.Getenv("GOOGLE_REDIRECT_URL"),
+		SessionSecret:        sessionSecret,
+		EncryptionKey:        encryptionKey,
+		Port:                 port,
+		WorkerPort:           workerPort,
+		Env:                  env,
+		AdminPassword:        os.Getenv("ADMIN_PASSWORD"),
+		AMQPURL:              amqpURL,
+		ValkeyURI:            os.Getenv("VALKEY_URI"),
+		TelegramProxyURL:     os.Getenv("TG_PROXY_URL"),
+		ProxySecretKey:       secretKey,
+		TelegramGroupID:      os.Getenv("TG_GROUP_ID"),
+		TelegramGroupTopicID: os.Getenv("TG_GROUP_TOPIC_ID"),
+		From:                 os.Getenv("FROM"),
 	}
 }