jason/qrknit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: admin observability — platform overview, audit log, purge, notifications

Browse Source 
- GET /api/admin/overview: platform totals (users, orgs, links, clicks),
  a 30-day click series, top links, and per-user quota consumption with
  near-limit flags (org members show pooled usage).
- Audit logging wired into all admin mutations: user create/update/delete,
  password resets, org create/update/delete, purges. Browsable via
  GET /api/admin/audit and a new Audit Log section in the admin UI.
- POST /api/admin/purge hard-deletes soft-deleted links older than N days,
  freeing their short codes and click history.
- Contact-form submissions now fire a WEBHOOK_URL notification.
- Fix: deleting a user who owned links or tags failed with a foreign-key
  500 — their links/tags are now deliberately orphaned (kept in the DB,
  unowned) and their API keys removed, matching the UI's wording.
- Admin UI: Platform Overview stat cards, quota-usage table with ⚠ NEAR
  LIMIT badges, audit table, and a purge action.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jason Stedwell
2026-07-03 01:40:29 -05:00
parent a321ff6171
commit 45be86bb36
2 changed files with 216 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+122 -2
View File
@@ -1906,6 +1906,8 @@ def admin_create_user():
)
user = conn.execute('SELECT id, username, is_admin, plan, created_at FROM users WHERE username=?',
(username,)).fetchone()
log_audit(conn, 'user.create', target=username,
details=f'plan={plan} admin={bool(is_admin)}')
return jsonify(dict(user)), 201
@@ -1942,6 +1944,8 @@ def admin_update_user(user_id):
'o.name as org_name FROM users u LEFT JOIN organizations o ON u.org_id=o.id '
'WHERE u.id=?', (user_id,)
).fetchone()
log_audit(conn, 'user.update', target=user['username'],
details=', '.join(f'{k}={v}' for k, v in updates.items()))
return jsonify(dict(user))
@@ -1954,7 +1958,13 @@ def admin_delete_user(user_id):
user = conn.execute('SELECT * FROM users WHERE id=?', (user_id,)).fetchone()
if not user:
return jsonify({'error': 'Not found'}), 404
# Orphan the user's links/tags (they stay in the DB, unowned) and drop
# their API keys — otherwise the FK constraints reject the delete.
conn.execute('UPDATE links SET user_id=NULL WHERE user_id=?', (user_id,))
conn.execute('UPDATE tags SET user_id=NULL WHERE user_id=?', (user_id,))
conn.execute('DELETE FROM api_keys WHERE user_id=?', (user_id,))
conn.execute('DELETE FROM users WHERE id=?', (user_id,))
log_audit(conn, 'user.delete', target=user['username'])
return jsonify({'success': True})
@@ -1966,10 +1976,12 @@ def admin_change_password(user_id):
if not password:
return jsonify({'error': 'Password required'}), 400
with get_db() as conn:
if not conn.execute('SELECT 1 FROM users WHERE id=?', (user_id,)).fetchone():
user = conn.execute('SELECT username FROM users WHERE id=?', (user_id,)).fetchone()
if not user:
return jsonify({'error': 'Not found'}), 404
conn.execute('UPDATE users SET password_hash=? WHERE id=?',
(generate_password_hash(password), user_id))
log_audit(conn, 'user.password_reset', target=user['username'])
return jsonify({'success': True})
@@ -1992,6 +2004,7 @@ def submit_contact():
'INSERT INTO messages (name, email, subject, body, created_at) VALUES (?,?,?,?,?)',
(name, email, subject, body, now)
)
notify_webhook(f'📨 New {APP_NAME} contact message from {name} <{email}>: {subject}')
return jsonify({'success': True})
@@ -2059,6 +2072,7 @@ def admin_create_organization():
org = conn.execute(
'SELECT id, name, plan, created_at FROM organizations WHERE name=?', (name,)
).fetchone()
log_audit(conn, 'org.create', target=name, details=f'plan={plan}')
return jsonify(dict(org)), 201
@@ -2089,6 +2103,8 @@ def admin_update_organization(org_id):
'FROM organizations o LEFT JOIN users u ON u.org_id=o.id WHERE o.id=?',
(org_id,)
).fetchone()
log_audit(conn, 'org.update', target=org['name'],
details=', '.join(f'{k}={v}' for k, v in updates.items()))
return jsonify(dict(org))
@@ -2096,14 +2112,118 @@ def admin_update_organization(org_id):
@admin_required
def admin_delete_organization(org_id):
with get_db() as conn:
if not conn.execute('SELECT 1 FROM organizations WHERE id=?', (org_id,)).fetchone():
org = conn.execute('SELECT name FROM organizations WHERE id=?', (org_id,)).fetchone()
if not org:
return jsonify({'error': 'Not found'}), 404
# Remove org_id from all members before deleting
conn.execute('UPDATE users SET org_id=NULL WHERE org_id=?', (org_id,))
conn.execute('DELETE FROM organizations WHERE id=?', (org_id,))
log_audit(conn, 'org.delete', target=org['name'])
return jsonify({'success': True})
# ─────────────────────────────────────────────
# Admin — Platform overview, audit log, data lifecycle
# ─────────────────────────────────────────────
@app.route('/api/admin/overview')
@admin_required
def admin_overview():
"""Bird's-eye platform view: totals, 30-day click series, per-user quota
consumption with near-limit flags, and top links platform-wide."""
now = datetime.now(timezone.utc).replace(tzinfo=None)
since_30d = (now - timedelta(days=30)).isoformat()
since_7d = (now - timedelta(days=7)).isoformat()
today = now.strftime('%Y-%m-%d')
with get_db() as conn:
totals = {
'users': conn.execute('SELECT COUNT(*) FROM users').fetchone()[0],
'organizations':conn.execute('SELECT COUNT(*) FROM organizations').fetchone()[0],
'active_links': conn.execute('SELECT COUNT(*) FROM links WHERE is_active=1').fetchone()[0],
'deleted_links':conn.execute('SELECT COUNT(*) FROM links WHERE is_active=0').fetchone()[0],
'total_clicks': conn.execute('SELECT COALESCE(SUM(clicks),0) FROM links').fetchone()[0],
'clicks_today': conn.execute('SELECT COUNT(*) FROM clicks WHERE clicked_at>=?', (today,)).fetchone()[0],
'clicks_7d': conn.execute('SELECT COUNT(*) FROM clicks WHERE clicked_at>=?', (since_7d,)).fetchone()[0],
'clicks_30d': conn.execute('SELECT COUNT(*) FROM clicks WHERE clicked_at>=?', (since_30d,)).fetchone()[0],
}
daily_rows = conn.execute("""
SELECT substr(clicked_at,1,10) as day, COUNT(*) as count
FROM clicks WHERE clicked_at>=? GROUP BY day ORDER BY day
""", (since_30d,)).fetchall()
daily_map = {r['day']: r['count'] for r in daily_rows}
daily = []
for i in range(30):
d = (now - timedelta(days=29-i)).strftime('%Y-%m-%d')
daily.append({'date': d, 'clicks': daily_map.get(d, 0)})
top_links = [dict(r) for r in conn.execute("""
SELECT l.code, l.title, l.long_url, l.clicks, u.username as created_by
FROM links l LEFT JOIN users u ON l.user_id=u.id
WHERE l.is_active=1 ORDER BY l.clicks DESC LIMIT 10
""").fetchall()]
# Per-user quota consumption (org members show pooled usage)
users = conn.execute(
'SELECT u.id, u.username, u.is_admin, o.name as org_name '
'FROM users u LEFT JOIN organizations o ON u.org_id=o.id ORDER BY u.username'
).fetchall()
usage = []
for u in users:
plan = get_user_plan(conn, u['id'])
limits = PLAN_LIMITS[plan]
links_used = get_user_active_link_count(conn, u['id'])
clicks_used = get_user_monthly_clicks(conn, u['id'])
def _pct(used, limit):
return round(used / limit * 100, 1) if limit else None
links_pct = _pct(links_used, limits['max_links'])
clicks_pct = _pct(clicks_used, limits['monthly_clicks'])
usage.append({
'id': u['id'], 'username': u['username'], 'is_admin': bool(u['is_admin']),
'org_name': u['org_name'], 'plan': plan,
'links_used': links_used, 'links_limit': limits['max_links'], 'links_pct': links_pct,
'clicks_used': clicks_used,'clicks_limit': limits['monthly_clicks'], 'clicks_pct': clicks_pct,
'near_limit': bool((links_pct and links_pct >= 80) or (clicks_pct and clicks_pct >= 80)),
})
return jsonify({'totals': totals, 'daily': daily, 'top_links': top_links, 'usage': usage})
@app.route('/api/admin/audit')
@admin_required
def admin_audit_log():
limit = min(int(request.args.get('limit', 200)), 1000)
with get_db() as conn:
rows = conn.execute(
'SELECT * FROM audit_log ORDER BY id DESC LIMIT ?', (limit,)
).fetchall()
return jsonify({'entries': [dict(r) for r in rows]})
@app.route('/api/admin/purge', methods=['POST'])
@admin_required
def admin_purge_links():
"""Hard-delete soft-deleted links (and their clicks/tags) older than N days.
Frees their short codes for reuse and reclaims click storage."""
data = request.get_json(silent=True) or {}
days = max(0, int(data.get('days', 30)))
cutoff = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days)).isoformat()
with get_db() as conn:
rows = conn.execute(
'SELECT id FROM links WHERE is_active=0 AND (deleted_at IS NULL OR deleted_at < ?)',
(cutoff,)
).fetchall()
ids = [r['id'] for r in rows]
if ids:
ph = ','.join('?' * len(ids))
conn.execute(f'DELETE FROM clicks WHERE link_id IN ({ph})', ids)
conn.execute(f'DELETE FROM link_tags WHERE link_id IN ({ph})', ids)
conn.execute(f'DELETE FROM links WHERE id IN ({ph})', ids)
log_audit(conn, 'links.purge', details=f'{len(ids)} soft-deleted links purged (deleted >{days}d ago)')
return jsonify({'purged': len(ids)})
# ─────────────────────────────────────────────
# SEO — robots.txt & sitemap.xml
# ─────────────────────────────────────────────