import asyncio import json import logging from datetime import date, datetime, timedelta, timezone from email.message import EmailMessage from typing import List import aiosmtplib from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from sqlalchemy import select from config import settings from crypto import decrypt_string from database import AsyncSessionLocal from dns_checker import check_dns_across_resolvers from integrations import get_registrar from models import AppSettings, DnsCheckResult, DnsRecord, Domain, Notification, RegistrarAccount logger = logging.getLogger(__name__) scheduler = AsyncIOScheduler() async def sync_all_domains_job() -> int: """Sync domains across all active registrar accounts.""" logger.info("Starting scheduled job: sync_all_domains") total_synced = 0 now = datetime.now(timezone.utc) async with AsyncSessionLocal() as db: stmt = select(RegistrarAccount).where(RegistrarAccount.is_active == True) accounts = (await db.execute(stmt)).scalars().all() for account in accounts: try: creds = json.loads(decrypt_string(account.credentials_encrypted)) client = get_registrar(account.registrar, creds) domains_data = await client.list_domains() for item in domains_data: domain_name = item.get("domain_name", "").lower().strip() if not domain_name: continue d_stmt = select(Domain).where( Domain.account_id == account.id, Domain.domain_name == domain_name, ) existing = (await db.execute(d_stmt)).scalar_one_or_none() if existing: if item.get("expiry_date"): existing.expiry_date = item["expiry_date"] if "auto_renew" in item: existing.auto_renew = item["auto_renew"] if item.get("status"): existing.status = item["status"] if item.get("registrar_domain_id"): existing.registrar_domain_id = item["registrar_domain_id"] existing.last_synced = now else: new_domain = Domain( account_id=account.id, domain_name=domain_name, expiry_date=item.get("expiry_date"), auto_renew=item.get("auto_renew", False), status=item.get("status", "active"), registrar_domain_id=item.get("registrar_domain_id"), last_synced=now, ) db.add(new_domain) total_synced += 1 except Exception as e: logger.error(f"Error in sync_all_domains for account {account.id}: {e}") await db.commit() logger.info(f"Finished sync_all_domains: {total_synced} domains updated") return total_synced async def snapshot_dns_records_job() -> int: """Fetch current DNS records from registrar APIs and store snapshots.""" logger.info("Starting scheduled job: snapshot_dns_records") total_snapshots = 0 now = datetime.now(timezone.utc) async with AsyncSessionLocal() as db: query = ( select(Domain, RegistrarAccount) .join(RegistrarAccount, Domain.account_id == RegistrarAccount.id) .where(RegistrarAccount.is_active == True) ) rows = (await db.execute(query)).all() for domain, account in rows: try: creds = json.loads(decrypt_string(account.credentials_encrypted)) client = get_registrar(account.registrar, creds) records_data = await client.get_dns_records(domain.domain_name) for r in records_data: rec = DnsRecord( domain_id=domain.id, record_type=r.get("record_type", "A").upper(), name=r.get("name", "@"), value=r.get("value", ""), ttl=r.get("ttl"), priority=r.get("priority"), snapshotted_at=now, ) db.add(rec) total_snapshots += 1 except Exception as e: logger.warning(f"Failed DNS snapshot for {domain.domain_name}: {e}") await db.commit() logger.info(f"Finished snapshot_dns_records: {total_snapshots} records recorded") return total_snapshots async def live_dns_health_check_job() -> int: """Run live DNS resolution checks for all registered domains.""" logger.info("Starting scheduled job: live_dns_health_check") total_checks = 0 now = datetime.now(timezone.utc) async with AsyncSessionLocal() as db: domains = (await db.execute(select(Domain))).scalars().all() for domain in domains: try: results = await check_dns_across_resolvers(domain.domain_name, "A") for res in results: check_obj = DnsCheckResult( domain_id=domain.id, checked_at=now, resolver_used=res["resolver_used"], record_type=res["record_type"], resolved_values=json.dumps(res["resolved_values"]), is_reachable=res["is_reachable"], response_time_ms=res["response_time_ms"], error_message=res["error_message"], ) db.add(check_obj) total_checks += 1 except Exception as e: logger.warning(f"Live DNS check failed for {domain.domain_name}: {e}") await db.commit() logger.info(f"Finished live_dns_health_check: {total_checks} checks recorded") return total_checks async def send_expiry_alerts_job() -> int: """Check for domains expiring soon and send email notifications.""" logger.info("Starting scheduled job: send_expiry_alerts") alerts_sent = 0 today = date.today() now = datetime.now(timezone.utc) async with AsyncSessionLocal() as db: app_settings = await db.get(AppSettings, 1) if not app_settings or not app_settings.smtp_host or not app_settings.alert_emails: logger.info("SMTP or alert emails not configured. Skipping email alerts.") return 0 recipient_list = [e.strip() for e in app_settings.alert_emails.split(",") if e.strip()] if not recipient_list: return 0 alert_days = [int(d.strip()) for d in app_settings.expiry_alert_days.split(",") if d.strip().isdigit()] smtp_pass = decrypt_string(app_settings.smtp_password_encrypted) if app_settings.smtp_password_encrypted else None for days in alert_days: target_date = today + timedelta(days=days) notification_type = f"expiry_{days}d" stmt = ( select(Domain, RegistrarAccount) .join(RegistrarAccount, Domain.account_id == RegistrarAccount.id) .where( Domain.expiry_date == target_date, Domain.auto_renew == False, ) ) rows = (await db.execute(stmt)).all() for domain, account in rows: # Check if notification was already sent today for this domain & type notif_check = select(Notification).where( Notification.domain_id == domain.id, Notification.notification_type == notification_type, ) already_sent = (await db.execute(notif_check)).scalar_one_or_none() if already_sent: continue # Compose email subject = f"⚠️ Domain Expiry Warning: {domain.domain_name} expires in {days} day(s)" body = ( f"Domain Expiry Notice\n\n" f"Domain: {domain.domain_name}\n" f"Registrar: {account.registrar.title()} ({account.name})\n" f"Expiry Date: {domain.expiry_date}\n" f"Auto-Renew: OFF\n" f"Days Remaining: {days}\n\n" f"Please log in to your registrar account or platform to renew this domain.\n" ) for email_addr in recipient_list: msg = EmailMessage() msg["Subject"] = subject msg["From"] = app_settings.smtp_from or app_settings.smtp_user or "noreply@domainplatform.local" msg["To"] = email_addr msg.set_content(body) try: await aiosmtplib.send( msg, hostname=app_settings.smtp_host, port=app_settings.smtp_port, username=app_settings.smtp_user, password=smtp_pass, start_tls=True if app_settings.smtp_port == 587 else False, use_tls=True if app_settings.smtp_port == 465 else False, timeout=10.0, ) notif = Notification( domain_id=domain.id, notification_type=notification_type, sent_at=now, recipient_email=email_addr, ) db.add(notif) alerts_sent += 1 except Exception as e: logger.error(f"Failed to send email alert to {email_addr}: {e}") await db.commit() logger.info(f"Finished send_expiry_alerts: {alerts_sent} emails sent") return alerts_sent def start_scheduler(): """Start in-process scheduler when running in Docker/self-hosted mode.""" if settings.DEPLOY_ENV != "docker": logger.info(f"Running in {settings.DEPLOY_ENV} mode; skipping in-process APScheduler.") return # 1. Sync domains daily at 02:00 scheduler.add_job(sync_all_domains_job, CronTrigger(hour=2, minute=0), id="sync_domains", replace_existing=True) # 2. DNS snapshots daily at 03:00 scheduler.add_job(snapshot_dns_records_job, CronTrigger(hour=3, minute=0), id="snapshot_dns", replace_existing=True) # 3. Expiry email alerts daily at 08:00 scheduler.add_job(send_expiry_alerts_job, CronTrigger(hour=8, minute=0), id="expiry_alerts", replace_existing=True) # 4. Live DNS health check every 6 hours scheduler.add_job(live_dns_health_check_job, CronTrigger(hour="0,6,12,18", minute=30), id="dns_health", replace_existing=True) scheduler.start() logger.info("APScheduler background jobs initialized and running.")