From c4120de9fc639b4e5d0e6998279581bd33c4df1d Mon Sep 17 00:00:00 2001
From: Glenn Thompson <glennt40@gmail.com>
Date: Mon, 8 Dec 2025 08:25:06 +0300
Subject: [PATCH] Add listener statistics feature

- Add database schema for listener snapshots, sessions, and aggregates
- Implement background polling of Icecast admin XML stats
- Add API endpoints for current, daily, and geo stats
- Add listener stats section to admin dashboard with auto-refresh
- GDPR compliant: IP hashing, data retention cleanup
---
 asteroid.asd                           |   1 +
 asteroid.lisp                          |  63 ++++-
 listener-stats.lisp                    | 352 +++++++++++++++++++++++++
 migrations/002-listener-statistics.sql | 127 +++++++++
 static/asteroid.css                    |   6 +
 static/asteroid.lass                   |   6 +-
 template/admin.ctml                    |  89 +++++++
 7 files changed, 637 insertions(+), 7 deletions(-)
 create mode 100644 listener-stats.lisp
 create mode 100644 migrations/002-listener-statistics.sql

diff --git a/asteroid.asd b/asteroid.asd
index 9704836..d60e402 100644
--- a/asteroid.asd
+++ b/asteroid.asd
@@ -58,6 +58,7 @@
                (:file "user-management")
                (:file "playlist-management")
                (:file "stream-control")
+               (:file "listener-stats")
                (:file "auth-routes")
                (:file "frontend-partials")
                (:file "asteroid")))
diff --git a/asteroid.lisp b/asteroid.lisp
index 0f947fe..0926c70 100644
--- a/asteroid.lisp
+++ b/asteroid.lisp
@@ -823,11 +823,16 @@
 (define-api asteroid/user/listening-stats () ()
   "Get user listening statistics"
   (require-authentication)
-  (api-output `(("status" . "success")
-                ("stats" . (("total_listen_time" . 0)
-                            ("tracks_played" . 0)
-                            ("session_count" . 0)
-                            ("favorite_genre" . "Unknown"))))))
+  (let* ((current-user (get-current-user))
+         (user-id (when current-user (dm:id current-user)))
+         (stats (if user-id 
+                    (get-user-listening-stats user-id)
+                    (list :total-listen-time 0 :session-count 0 :tracks-played 0))))
+    (api-output `(("status" . "success")
+                  ("stats" . (("total_listen_time" . ,(getf stats :total-listen-time 0))
+                              ("tracks_played" . ,(getf stats :tracks-played 0))
+                              ("session_count" . ,(getf stats :session-count 0))
+                              ("favorite_genre" . "Unknown")))))))
 
 (define-api asteroid/user/recent-tracks (&optional (limit "3")) ()
   "Get recently played tracks for user"
@@ -1000,6 +1005,39 @@
            `(("error" . "Could not connect to Icecast server"))
            :status 503)))))
 
+;;; Listener Statistics API Endpoints
+
+(define-api asteroid/stats/current () ()
+  "Get current listener count from recent snapshots"
+  (let ((listeners (get-current-listeners)))
+    (api-output `(("status" . "success")
+                  ("listeners" . ,listeners)
+                  ("timestamp" . ,(get-universal-time))))))
+
+(define-api asteroid/stats/daily (&optional (days "30")) ()
+  "Get daily listener statistics (admin only)"
+  (require-role :admin)
+  (let ((stats (get-daily-stats (parse-integer days :junk-allowed t))))
+    (api-output `(("status" . "success")
+                  ("stats" . ,(mapcar (lambda (row)
+                                        `(("date" . ,(first row))
+                                          ("mount" . ,(second row))
+                                          ("unique_listeners" . ,(third row))
+                                          ("peak_concurrent" . ,(fourth row))
+                                          ("total_listen_minutes" . ,(fifth row))
+                                          ("avg_session_minutes" . ,(sixth row))))
+                                      stats))))))
+
+(define-api asteroid/stats/geo (&optional (days "7")) ()
+  "Get geographic distribution of listeners (admin only)"
+  (require-role :admin)
+  (let ((stats (get-geo-stats (parse-integer days :junk-allowed t))))
+    (api-output `(("status" . "success")
+                  ("geo" . ,(mapcar (lambda (row)
+                                      `(("country_code" . ,(first row))
+                                        ("total_listeners" . ,(second row))
+                                        ("total_minutes" . ,(third row))))
+                                    stats))))))
 
 ;; RADIANCE server management functions
 
@@ -1012,11 +1050,24 @@
   ;; (unless (radiance:environment)
   ;;   (setf (radiance:environment) "asteroid"))
   
-  (radiance:startup))
+  (radiance:startup)
+  
+  ;; Start listener statistics polling
+  (handler-case
+      (progn
+        (format t "Starting listener statistics polling...~%")
+        (start-stats-polling))
+    (error (e)
+      (format t "Warning: Could not start stats polling: ~a~%" e))))
 
 (defun stop-server ()
   "Stop the Asteroid Radio RADIANCE server"
   (format t "Stopping Asteroid Radio server...~%")
+  ;; Stop listener statistics polling
+  (handler-case
+      (stop-stats-polling)
+    (error (e)
+      (format t "Warning: Error stopping stats polling: ~a~%" e)))
   (radiance:shutdown)
   (format t "Server stopped.~%"))
 
diff --git a/listener-stats.lisp b/listener-stats.lisp
new file mode 100644
index 0000000..cf5b3e0
--- /dev/null
+++ b/listener-stats.lisp
@@ -0,0 +1,352 @@
+;;;; listener-stats.lisp - Listener Statistics Collection Service
+;;;; Polls Icecast for listener data and stores with GDPR compliance
+
+(in-package #:asteroid)
+
+;;; Use postmodern for direct SQL queries
+;;; Connection params from environment or defaults matching config/radiance-postgres.lisp
+(defun get-db-connection-params ()
+  "Get database connection parameters"
+  (list (or (uiop:getenv "ASTEROID_DB_NAME") "asteroid")
+        (or (uiop:getenv "ASTEROID_DB_USER") "asteroid")
+        (or (uiop:getenv "ASTEROID_DB_PASSWORD") "asteroid_db_2025")
+        "localhost"
+        :port 5432))
+
+(defmacro with-db (&body body)
+  "Execute body with database connection"
+  `(postmodern:with-connection (get-db-connection-params)
+     ,@body))
+
+;;; Configuration
+(defvar *stats-polling-interval* 60
+  "Seconds between Icecast polls")
+
+(defvar *stats-polling-thread* nil
+  "Background thread for polling")
+
+(defvar *stats-polling-active* nil
+  "Flag to control polling loop")
+
+(defvar *icecast-stats-url* "http://localhost:8000/admin/stats"
+  "Icecast admin stats endpoint (XML)")
+
+(defvar *icecast-admin-user* "admin"
+  "Icecast admin username")
+
+(defvar *icecast-admin-pass* "asteroid_admin_2024"
+  "Icecast admin password")
+
+(defvar *geoip-api-url* "http://ip-api.com/json/~a?fields=status,countryCode,city,regionName"
+  "GeoIP lookup API (free tier: 45 req/min)")
+
+(defvar *session-retention-days* 30
+  "Days to retain individual session data")
+
+;;; Active listener tracking (in-memory)
+(defvar *active-listeners* (make-hash-table :test 'equal)
+  "Hash table tracking active listeners by IP hash")
+
+;;; Utility Functions
+
+(defun hash-ip-address (ip-address)
+  "Hash an IP address using SHA256 for privacy-safe storage"
+  (let ((digest (ironclad:make-digest :sha256)))
+    (ironclad:update-digest digest (babel:string-to-octets ip-address))
+    (ironclad:byte-array-to-hex-string (ironclad:produce-digest digest))))
+
+(defun generate-session-id ()
+  "Generate a unique session ID"
+  (let ((digest (ironclad:make-digest :sha256)))
+    (ironclad:update-digest digest 
+                            (babel:string-to-octets 
+                             (format nil "~a-~a" (get-universal-time) (random 1000000))))
+    (subseq (ironclad:byte-array-to-hex-string (ironclad:produce-digest digest)) 0 32)))
+
+;;; GeoIP Lookup
+
+(defun lookup-geoip (ip-address)
+  "Look up geographic location for an IP address.
+   Returns plist with :country-code :city :region or NIL on failure.
+   Note: Does not store the raw IP - only uses it for lookup."
+  (handler-case
+      (let* ((url (format nil *geoip-api-url* ip-address))
+             (response (drakma:http-request url :want-stream nil))
+             (data (cl-json:decode-json-from-string response)))
+        (when (string= (cdr (assoc :status data)) "success")
+          (list :country-code (cdr (assoc :country-code data))
+                :city (cdr (assoc :city data))
+                :region (cdr (assoc :region-name data)))))
+    (error (e)
+      (log:debug "GeoIP lookup failed for ~a: ~a" ip-address e)
+      nil)))
+
+;;; Icecast Polling
+
+(defun extract-xml-value (xml tag)
+  "Extract value between XML tags. Simple regex-based extraction."
+  (let ((pattern (format nil "<~a>([^<]*)</~a>" tag tag)))
+    (multiple-value-bind (match groups)
+        (cl-ppcre:scan-to-strings pattern xml)
+      (when match
+        (aref groups 0)))))
+
+(defun extract-xml-sources (xml)
+  "Extract all source blocks from Icecast XML"
+  (let ((sources nil)
+        (pattern "<source mount=\"([^\"]+)\">(.*?)</source>"))
+    (cl-ppcre:do-register-groups (mount content) (pattern xml)
+      (let ((listeners (extract-xml-value content "listeners"))
+            (listener-peak (extract-xml-value content "listener_peak"))
+            (server-name (extract-xml-value content "server_name")))
+        (push (list :mount mount
+                    :server-name server-name
+                    :listeners (if listeners (parse-integer listeners :junk-allowed t) 0)
+                    :listener-peak (if listener-peak (parse-integer listener-peak :junk-allowed t) 0))
+              sources)))
+    (nreverse sources)))
+
+(defun fetch-icecast-stats ()
+  "Fetch current statistics from Icecast admin XML endpoint"
+  (handler-case
+      (let ((response (drakma:http-request *icecast-stats-url* 
+                                           :want-stream nil
+                                           :connection-timeout 5
+                                           :basic-authorization (list *icecast-admin-user* 
+                                                                      *icecast-admin-pass*))))
+        ;; Response is XML, return as string for parsing
+        (if (stringp response)
+            response
+            (babel:octets-to-string response :encoding :utf-8)))
+    (error (e)
+      (log:warn "Failed to fetch Icecast stats: ~a" e)
+      nil)))
+
+(defun parse-icecast-sources (xml-string)
+  "Parse Icecast XML stats and extract source/mount information.
+   Returns list of plists with mount info."
+  (when xml-string
+    (extract-xml-sources xml-string)))
+
+;;; Database Operations
+
+(defun store-listener-snapshot (mount listener-count)
+  "Store a listener count snapshot"
+  (handler-case
+      (with-db
+        (postmodern:query 
+         (:insert-into 'listener_snapshots :set 'mount mount 'listener_count listener-count)))
+    (error (e)
+      (log:error "Failed to store snapshot: ~a" e))))
+
+(defun store-listener-session (session-id ip-hash mount &key country-code city region user-agent user-id)
+  "Create a new listener session record"
+  (handler-case
+      (with-db
+        (postmodern:query 
+         (:insert-into 'listener_sessions 
+          :set 'session_id session-id 
+               'ip_hash ip-hash 
+               'mount mount
+               'country_code country-code
+               'city city
+               'region region
+               'user_agent user-agent
+               'user_id user-id)))
+    (error (e)
+      (log:error "Failed to store session: ~a" e))))
+
+(defun end-listener-session (session-id)
+  "Mark a listener session as ended and calculate duration"
+  (handler-case
+      (with-db
+        (postmodern:execute 
+         (format nil "UPDATE listener_sessions 
+                      SET session_end = NOW(),
+                          duration_seconds = EXTRACT(EPOCH FROM (NOW() - session_start))::INTEGER
+                      WHERE session_id = '~a' AND session_end IS NULL" session-id)))
+    (error (e)
+      (log:error "Failed to end session: ~a" e))))
+
+(defun cleanup-old-sessions ()
+  "Remove session data older than retention period (GDPR compliance)"
+  (handler-case
+      (with-db
+        (let ((result (postmodern:query 
+                       (format nil "SELECT cleanup_old_listener_data(~a)" *session-retention-days*))))
+          (log:info "Session cleanup completed: ~a records removed" (caar result))))
+    (error (e)
+      (log:error "Session cleanup failed: ~a" e))))
+
+;;; Statistics Aggregation
+;;; Note: Complex aggregation queries use raw SQL via postmodern:execute
+
+(defun aggregate-daily-stats (date)
+  "Compute daily aggregates from session data"
+  (handler-case
+      (with-db
+        (postmodern:execute 
+         (format nil "INSERT INTO listener_daily_stats 
+          (date, mount, unique_listeners, peak_concurrent, total_listen_minutes, avg_session_minutes)
+          SELECT 
+            '~a'::date,
+            mount,
+            COUNT(DISTINCT ip_hash),
+            (SELECT COALESCE(MAX(listener_count), 0) FROM listener_snapshots 
+             WHERE timestamp::date = '~a'::date AND listener_snapshots.mount = listener_sessions.mount),
+            COALESCE(SUM(duration_seconds) / 60, 0),
+            COALESCE(AVG(duration_seconds) / 60.0, 0)
+          FROM listener_sessions
+          WHERE session_start::date = '~a'::date
+          GROUP BY mount
+          ON CONFLICT (date, mount) DO UPDATE SET
+            unique_listeners = EXCLUDED.unique_listeners,
+            peak_concurrent = EXCLUDED.peak_concurrent,
+            total_listen_minutes = EXCLUDED.total_listen_minutes,
+            avg_session_minutes = EXCLUDED.avg_session_minutes" date date date)))
+    (error (e)
+      (log:error "Failed to aggregate daily stats: ~a" e))))
+
+(defun aggregate-hourly-stats (date hour)
+  "Compute hourly aggregates"
+  (handler-case
+      (with-db
+        (postmodern:execute
+         (format nil "INSERT INTO listener_hourly_stats (date, hour, mount, unique_listeners, peak_concurrent)
+          SELECT 
+            '~a'::date,
+            ~a,
+            mount,
+            COUNT(DISTINCT ip_hash),
+            COALESCE(MAX(listener_count), 0)
+          FROM listener_sessions ls
+          LEFT JOIN listener_snapshots lsn ON lsn.mount = ls.mount 
+            AND DATE_TRUNC('hour', lsn.timestamp) = DATE_TRUNC('hour', ls.session_start)
+          WHERE session_start::date = '~a'::date 
+            AND EXTRACT(HOUR FROM session_start) = ~a
+          GROUP BY mount
+          ON CONFLICT (date, hour, mount) DO UPDATE SET
+            unique_listeners = EXCLUDED.unique_listeners,
+            peak_concurrent = EXCLUDED.peak_concurrent" date hour date hour)))
+    (error (e)
+      (log:error "Failed to aggregate hourly stats: ~a" e))))
+
+;;; Query Functions (for API endpoints)
+
+(defun get-current-listeners ()
+  "Get current listener count from most recent snapshot"
+  (handler-case
+      (with-db
+        (let ((result (postmodern:query 
+                       "SELECT mount, listener_count, timestamp 
+                        FROM listener_snapshots 
+                        WHERE timestamp > NOW() - INTERVAL '5 minutes'
+                        ORDER BY timestamp DESC")))
+          (mapcar (lambda (row)
+                    (list :mount (first row)
+                          :listeners (second row)
+                          :timestamp (third row)))
+                  result)))
+    (error (e)
+      (log:error "Failed to get current listeners: ~a" e)
+      nil)))
+
+(defun get-daily-stats (&optional (days 30))
+  "Get daily statistics for the last N days"
+  (handler-case
+      (with-db
+        (postmodern:query 
+         (format nil "SELECT date, mount, unique_listeners, peak_concurrent, total_listen_minutes, avg_session_minutes
+          FROM listener_daily_stats 
+          WHERE date > NOW() - INTERVAL '~a days'
+          ORDER BY date DESC" days)))
+    (error (e)
+      (log:error "Failed to get daily stats: ~a" e)
+      nil)))
+
+(defun get-geo-stats (&optional (days 7))
+  "Get geographic distribution for the last N days"
+  (handler-case
+      (with-db
+        (postmodern:query
+         (format nil "SELECT country_code, SUM(listener_count) as total_listeners, SUM(listen_minutes) as total_minutes
+          FROM listener_geo_stats
+          WHERE date > NOW() - INTERVAL '~a days'
+          GROUP BY country_code
+          ORDER BY total_listeners DESC
+          LIMIT 20" days)))
+    (error (e)
+      (log:error "Failed to get geo stats: ~a" e)
+      nil)))
+
+(defun get-user-listening-stats (user-id)
+  "Get listening statistics for a specific user"
+  (handler-case
+      (with-db
+        (let ((total-time (caar (postmodern:query 
+                                 (format nil "SELECT COALESCE(SUM(duration_seconds), 0) 
+                                  FROM listener_sessions WHERE user_id = ~a" user-id))))
+              (session-count (caar (postmodern:query
+                                    (format nil "SELECT COUNT(*) FROM listener_sessions WHERE user_id = ~a" user-id))))
+              (track-count (caar (postmodern:query
+                                  (format nil "SELECT COUNT(*) FROM user_listening_history WHERE user_id = ~a" user-id)))))
+          (list :total-listen-time (or total-time 0)
+                :session-count (or session-count 0)
+                :tracks-played (or track-count 0))))
+    (error (e)
+      (log:error "Failed to get user stats: ~a" e)
+      (list :total-listen-time 0 :session-count 0 :tracks-played 0))))
+
+;;; Polling Service
+
+(defun poll-and-store-stats ()
+  "Single poll iteration: fetch stats and store"
+  (let ((stats (fetch-icecast-stats)))
+    (when stats
+      (let ((sources (parse-icecast-sources stats)))
+        (dolist (source sources)
+          (let ((mount (getf source :mount))
+                (listeners (getf source :listeners)))
+            (when mount
+              (store-listener-snapshot mount listeners)
+              (log:debug "Stored snapshot: ~a = ~a listeners" mount listeners))))))))
+
+(defun stats-polling-loop ()
+  "Main polling loop - runs in background thread"
+  (log:info "Listener statistics polling started (interval: ~as)" *stats-polling-interval*)
+  (loop while *stats-polling-active*
+        do (handler-case
+               (poll-and-store-stats)
+             (error (e)
+               (log:error "Polling error: ~a" e)))
+           (sleep *stats-polling-interval*))
+  (log:info "Listener statistics polling stopped"))
+
+(defun start-stats-polling ()
+  "Start the background statistics polling thread"
+  (when *stats-polling-thread*
+    (stop-stats-polling))
+  (setf *stats-polling-active* t)
+  (setf *stats-polling-thread*
+        (bt:make-thread #'stats-polling-loop :name "stats-poller"))
+  (log:info "Stats polling thread started"))
+
+(defun stop-stats-polling ()
+  "Stop the background statistics polling thread"
+  (setf *stats-polling-active* nil)
+  (when (and *stats-polling-thread* (bt:thread-alive-p *stats-polling-thread*))
+    (bt:join-thread *stats-polling-thread* :timeout 5))
+  (setf *stats-polling-thread* nil)
+  (log:info "Stats polling thread stopped"))
+
+;;; Initialization
+
+(defun init-listener-stats ()
+  "Initialize the listener statistics system"
+  (log:info "Initializing listener statistics system...")
+  (start-stats-polling))
+
+(defun shutdown-listener-stats ()
+  "Shutdown the listener statistics system"
+  (log:info "Shutting down listener statistics system...")
+  (stop-stats-polling))
diff --git a/migrations/002-listener-statistics.sql b/migrations/002-listener-statistics.sql
new file mode 100644
index 0000000..d03ef9b
--- /dev/null
+++ b/migrations/002-listener-statistics.sql
@@ -0,0 +1,127 @@
+-- Migration: Listener Statistics Tables
+-- Version: 002
+-- Date: 2025-12-08
+-- Description: Add tables for tracking listener statistics with GDPR compliance
+
+-- Listener snapshots: periodic counts from Icecast polling
+CREATE TABLE IF NOT EXISTS listener_snapshots (
+    _id SERIAL PRIMARY KEY,
+    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    mount VARCHAR(100) NOT NULL,
+    listener_count INTEGER NOT NULL DEFAULT 0
+);
+
+CREATE INDEX IF NOT EXISTS idx_snapshots_timestamp ON listener_snapshots(timestamp);
+CREATE INDEX IF NOT EXISTS idx_snapshots_mount ON listener_snapshots(mount);
+
+-- Listener sessions: individual connection records (privacy-safe)
+CREATE TABLE IF NOT EXISTS listener_sessions (
+    _id SERIAL PRIMARY KEY,
+    session_id VARCHAR(64) UNIQUE NOT NULL,
+    session_start TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    session_end TIMESTAMP,
+    ip_hash VARCHAR(64) NOT NULL,  -- SHA256 hash, not reversible
+    country_code VARCHAR(2),
+    city VARCHAR(100),
+    region VARCHAR(100),
+    user_agent TEXT,
+    mount VARCHAR(100) NOT NULL,
+    duration_seconds INTEGER,
+    user_id INTEGER REFERENCES "USERS"(_id) ON DELETE SET NULL  -- Optional link to registered user
+);
+
+CREATE INDEX IF NOT EXISTS idx_sessions_start ON listener_sessions(session_start);
+CREATE INDEX IF NOT EXISTS idx_sessions_ip_hash ON listener_sessions(ip_hash);
+CREATE INDEX IF NOT EXISTS idx_sessions_country ON listener_sessions(country_code);
+CREATE INDEX IF NOT EXISTS idx_sessions_user ON listener_sessions(user_id);
+
+-- Daily aggregated statistics (for efficient dashboard queries)
+CREATE TABLE IF NOT EXISTS listener_daily_stats (
+    _id SERIAL PRIMARY KEY,
+    date DATE NOT NULL,
+    mount VARCHAR(100) NOT NULL,
+    unique_listeners INTEGER DEFAULT 0,
+    peak_concurrent INTEGER DEFAULT 0,
+    total_listen_minutes INTEGER DEFAULT 0,
+    new_listeners INTEGER DEFAULT 0,
+    returning_listeners INTEGER DEFAULT 0,
+    avg_session_minutes DECIMAL(10,2),
+    UNIQUE(date, mount)
+);
+
+CREATE INDEX IF NOT EXISTS idx_daily_stats_date ON listener_daily_stats(date);
+
+-- Hourly breakdown for time-of-day analysis
+CREATE TABLE IF NOT EXISTS listener_hourly_stats (
+    _id SERIAL PRIMARY KEY,
+    date DATE NOT NULL,
+    hour INTEGER NOT NULL CHECK (hour >= 0 AND hour <= 23),
+    mount VARCHAR(100) NOT NULL,
+    unique_listeners INTEGER DEFAULT 0,
+    peak_concurrent INTEGER DEFAULT 0,
+    UNIQUE(date, hour, mount)
+);
+
+CREATE INDEX IF NOT EXISTS idx_hourly_stats_date ON listener_hourly_stats(date);
+
+-- Geographic aggregates
+CREATE TABLE IF NOT EXISTS listener_geo_stats (
+    _id SERIAL PRIMARY KEY,
+    date DATE NOT NULL,
+    country_code VARCHAR(2) NOT NULL,
+    city VARCHAR(100),
+    listener_count INTEGER DEFAULT 0,
+    listen_minutes INTEGER DEFAULT 0,
+    UNIQUE(date, country_code, city)
+);
+
+CREATE INDEX IF NOT EXISTS idx_geo_stats_date ON listener_geo_stats(date);
+CREATE INDEX IF NOT EXISTS idx_geo_stats_country ON listener_geo_stats(country_code);
+
+-- User listening history (for registered users only)
+CREATE TABLE IF NOT EXISTS user_listening_history (
+    _id SERIAL PRIMARY KEY,
+    user_id INTEGER NOT NULL REFERENCES "USERS"(_id) ON DELETE CASCADE,
+    track_title VARCHAR(500),
+    track_artist VARCHAR(500),
+    listened_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    duration_seconds INTEGER
+);
+
+CREATE INDEX IF NOT EXISTS idx_user_history_user ON user_listening_history(user_id);
+CREATE INDEX IF NOT EXISTS idx_user_history_listened ON user_listening_history(listened_at);
+
+-- Data retention: function to clean old session data (GDPR compliance)
+CREATE OR REPLACE FUNCTION cleanup_old_listener_data(retention_days INTEGER DEFAULT 30)
+RETURNS INTEGER AS $$
+DECLARE
+    deleted_count INTEGER;
+BEGIN
+    -- Delete individual sessions older than retention period
+    DELETE FROM listener_sessions 
+    WHERE session_start < NOW() - (retention_days || ' days')::INTERVAL;
+    
+    GET DIAGNOSTICS deleted_count = ROW_COUNT;
+    
+    RAISE NOTICE 'Cleaned up % listener session records older than % days', deleted_count, retention_days;
+    
+    RETURN deleted_count;
+END;
+$$ LANGUAGE plpgsql;
+
+-- Grant permissions
+GRANT ALL PRIVILEGES ON listener_snapshots TO asteroid;
+GRANT ALL PRIVILEGES ON listener_sessions TO asteroid;
+GRANT ALL PRIVILEGES ON listener_daily_stats TO asteroid;
+GRANT ALL PRIVILEGES ON listener_hourly_stats TO asteroid;
+GRANT ALL PRIVILEGES ON listener_geo_stats TO asteroid;
+GRANT ALL PRIVILEGES ON user_listening_history TO asteroid;
+GRANT ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA public TO asteroid;
+
+-- Success message
+DO $$
+BEGIN
+    RAISE NOTICE 'Listener statistics tables created successfully!';
+    RAISE NOTICE 'Tables: listener_snapshots, listener_sessions, listener_daily_stats, listener_hourly_stats, listener_geo_stats, user_listening_history';
+    RAISE NOTICE 'GDPR: IP addresses are hashed, cleanup function available';
+END $$;
diff --git a/static/asteroid.css b/static/asteroid.css
index 2b7e936..f1b4190 100644
--- a/static/asteroid.css
+++ b/static/asteroid.css
@@ -1224,6 +1224,12 @@ body .stat-card .stat-label{
     margin-top: 0.5rem;
 }
 
+body .stat-card .stat-detail{
+    color: #888;
+    font-size: 0.75rem;
+    margin-top: 0.25rem;
+}
+
 body.persistent-player-container{
     margin: 0;
     padding: 10px;;
diff --git a/static/asteroid.lass b/static/asteroid.lass
index 189319e..582943c 100644
--- a/static/asteroid.lass
+++ b/static/asteroid.lass
@@ -978,7 +978,11 @@
 
      (.stat-label :color "#ccc"
                   :font-size 0.875rem
-                  :margin-top 0.5rem))
+                  :margin-top 0.5rem)
+     
+     (.stat-detail :color "#888"
+                   :font-size 0.75rem
+                   :margin-top 0.25rem))
 
     ;; Center alignment for player page
     ;; (body.player-page
diff --git a/template/admin.ctml b/template/admin.ctml
index 8c655db..345e40f 100644
--- a/template/admin.ctml
+++ b/template/admin.ctml
@@ -42,6 +42,41 @@
       </div>
     </div>
 
+    <!-- Listener Statistics -->
+    <div class="admin-section">
+      <h2>📊 Listener Statistics</h2>
+      <div class="admin-grid" id="listener-stats-grid">
+        <div class="status-card">
+          <h3>🎵 asteroid.mp3</h3>
+          <p class="stat-number" id="listeners-mp3">0</p>
+          <p class="stat-label">Current Listeners</p>
+          <p class="stat-detail">Peak: <span id="peak-mp3">0</span></p>
+        </div>
+        <div class="status-card">
+          <h3>🎧 asteroid.aac</h3>
+          <p class="stat-number" id="listeners-aac">0</p>
+          <p class="stat-label">Current Listeners</p>
+          <p class="stat-detail">Peak: <span id="peak-aac">0</span></p>
+        </div>
+        <div class="status-card">
+          <h3>📱 asteroid-low.mp3</h3>
+          <p class="stat-number" id="listeners-low">0</p>
+          <p class="stat-label">Current Listeners</p>
+          <p class="stat-detail">Peak: <span id="peak-low">0</span></p>
+        </div>
+        <div class="status-card">
+          <h3>📈 Total Listeners</h3>
+          <p class="stat-number" id="listeners-total">0</p>
+          <p class="stat-label">All Streams</p>
+          <p class="stat-detail">Updated: <span id="stats-updated">--</span></p>
+        </div>
+      </div>
+      <div class="admin-controls" style="margin-top: 15px;">
+        <button id="refresh-stats" class="btn btn-secondary" onclick="refreshListenerStats()">🔄 Refresh Stats</button>
+        <span id="stats-status" style="margin-left: 15px;"></span>
+      </div>
+    </div>
+
     <!-- Music Library Management -->
     <div class="admin-section">
       <h2>Music Library Management</h2>
@@ -192,6 +227,60 @@
   </div>
 
   <script>
+    // Listener Statistics
+    function refreshListenerStats() {
+      const statusEl = document.getElementById('stats-status');
+      statusEl.textContent = 'Loading...';
+      
+      fetch('/api/asteroid/stats/current')
+        .then(response => response.json())
+        .then(result => {
+          const data = result.data || result;
+          if (data.status === 'success' && data.listeners) {
+            // Process listener data - get most recent for each mount
+            const mounts = {};
+            data.listeners.forEach(item => {
+              // item is [mount, "/asteroid.mp3", listeners, 1, timestamp, 123456]
+              const mount = item[1];
+              const listeners = item[3];
+              if (!mounts[mount] || item[5] > mounts[mount].timestamp) {
+                mounts[mount] = { listeners: listeners, timestamp: item[5] };
+              }
+            });
+            
+            // Update UI
+            const mp3 = mounts['/asteroid.mp3']?.listeners || 0;
+            const aac = mounts['/asteroid.aac']?.listeners || 0;
+            const low = mounts['/asteroid-low.mp3']?.listeners || 0;
+            
+            document.getElementById('listeners-mp3').textContent = mp3;
+            document.getElementById('listeners-aac').textContent = aac;
+            document.getElementById('listeners-low').textContent = low;
+            document.getElementById('listeners-total').textContent = mp3 + aac + low;
+            
+            const now = new Date();
+            document.getElementById('stats-updated').textContent = 
+              now.toLocaleTimeString();
+            
+            statusEl.textContent = '';
+          } else {
+            statusEl.textContent = 'No data available';
+          }
+        })
+        .catch(error => {
+          console.error('Error fetching stats:', error);
+          statusEl.textContent = 'Error loading stats';
+        });
+    }
+    
+    // Auto-refresh stats every 30 seconds
+    setInterval(refreshListenerStats, 30000);
+    
+    // Initial load
+    document.addEventListener('DOMContentLoaded', function() {
+      refreshListenerStats();
+    });
+
     // Admin password reset handler
     function resetUserPassword(event) {
       event.preventDefault();