"""Durable, race-safe background task engine for SCpanel v1.3.6."""
from __future__ import annotations
import json, sqlite3, threading, time, traceback, uuid, shutil, subprocess
from pathlib import Path
DATA=Path('/var/lib/scpanel'); DB=DATA/'scpanel.db'; _lock=threading.RLock(); _threads={}; _cancel_flags=set()

def _conn():
    DATA.mkdir(parents=True,exist_ok=True); c=sqlite3.connect(DB,timeout=30); c.row_factory=sqlite3.Row; return c

def init(c=None):
    own=c is None; c=c or _conn(); c.execute('''CREATE TABLE IF NOT EXISTS tasks(id TEXT PRIMARY KEY,kind TEXT NOT NULL,app_key TEXT,operation TEXT NOT NULL,requested_version TEXT DEFAULT '',status TEXT NOT NULL,progress INTEGER NOT NULL DEFAULT 0,step TEXT,message TEXT,output TEXT,error TEXT,actor TEXT,ip TEXT,attempts INTEGER NOT NULL DEFAULT 0,max_attempts INTEGER NOT NULL DEFAULT 1,created_at TEXT DEFAULT CURRENT_TIMESTAMP,started_at TEXT,finished_at TEXT)''')
    cols={r[1] for r in c.execute('PRAGMA table_info(tasks)')}
    if 'requested_version' not in cols:c.execute("ALTER TABLE tasks ADD COLUMN requested_version TEXT DEFAULT ''")
    c.execute('CREATE INDEX IF NOT EXISTS idx_tasks_created ON tasks(created_at DESC)'); c.commit()
    if own:c.close()

def _get(tid):
    c=_conn(); r=c.execute('SELECT * FROM tasks WHERE id=?',(tid,)).fetchone(); c.close(); return dict(r) if r else None

def _update(tid,**fields):
    if not fields:return
    c=_conn(); sets=[];vals=[]
    for k,v in fields.items():sets.append(f'{k}=?');vals.append(v)
    vals.append(tid);c.execute(f'UPDATE tasks SET {", ".join(sets)} WHERE id=?',vals);c.commit();c.close()

def _cancelled(tid):
    if tid in _cancel_flags:return True
    t=_get(tid); return bool(t and t.get('status')=='cancel_requested')

def _worker(tid):
    t=_get(tid)
    if not t:return
    # Atomically claim queued task so two Gunicorn processes cannot execute it.
    c=_conn(); now=time.strftime('%Y-%m-%d %H:%M:%S'); cur=c.execute("UPDATE tasks SET status='running',progress=5,step='Running',started_at=?,attempts=attempts+1 WHERE id=? AND status='queued'",(now,tid)); c.commit(); c.close()
    if cur.rowcount!=1:
        with _lock:_threads.pop(tid,None)
        return
    try:
        if t['kind']=='website' and t['operation']=='ssl':
            result=_website_ssl(t['app_key'],lambda p,step:_progress(tid,p,step),lambda:_cancelled(tid))
        elif t['kind']=='ssl':
            result=_ssl_operation(t['app_key'],t['operation'],lambda p,step:_progress(tid,p,step),lambda:_cancelled(tid))
        elif t['kind']=='maintenance':
            result=_maintenance(t['operation'],lambda p,step:_progress(tid,p,step),lambda:_cancelled(tid))
        else:
            from app.providers.service_engine import action
            if _cancelled(tid):raise RuntimeError('Operation cancelled before execution.')
            result=action(t['app_key'],t['operation'],progress=lambda p,step:_progress(tid,p,step),requested_version=t.get('requested_version',''),cancel_check=lambda:_cancelled(tid))
        if _cancelled(tid):_update(tid,status='cancelled',progress=min(99,int(_get(tid).get('progress',0))),step='Cancelled',message='Cancellation requested; operation stopped at the next safe checkpoint.',finished_at=time.strftime('%Y-%m-%d %H:%M:%S'))
        else:_update(tid,progress=98,step='Verifying result');_update(tid,status='completed',progress=100,step='Completed',message=result.get('message',f"{t['operation']} completed") if isinstance(result,dict) else str(result),output=json.dumps(result,default=str),finished_at=time.strftime('%Y-%m-%d %H:%M:%S'))
    except Exception as exc:
        if _cancelled(tid):_update(tid,status='cancelled',step='Cancelled',message=str(exc),error=traceback.format_exc(limit=8),finished_at=time.strftime('%Y-%m-%d %H:%M:%S'))
        else:_update(tid,status='failed',progress=100,step='Failed',message=str(exc),error=traceback.format_exc(limit=8),output=traceback.format_exc(limit=8),finished_at=time.strftime('%Y-%m-%d %H:%M:%S'))
    finally:
        with _lock:_threads.pop(tid,None);_cancel_flags.discard(tid)

def _progress(tid,p,step):
    if _cancelled(tid):raise RuntimeError('Operation cancellation requested.')
    _update(tid,progress=max(5,min(95,int(p))),step=step)

def _website_ssl(website_id,progress,cancel_check):
    from app.app import db, website_cert, _write_website_config, run_system
    try: wid=int(str(website_id).split(':')[-1])
    except Exception: raise RuntimeError('Invalid website task target.')
    c=db(); w=c.execute('SELECT * FROM websites WHERE id=?',(wid,)).fetchone()
    if not w:c.close();raise RuntimeError('Website not found.')
    a=c.execute('SELECT * FROM accounts WHERE id=?',(w['account_id'],)).fetchone(); c.close()
    if not a:raise RuntimeError('Hosting account not found.')
    if not shutil.which('certbot'):raise RuntimeError('Certbot is not installed. Install it from App Store first.')
    domains=[w['domain']]+[x for x in (w['aliases'] or '').split() if x]
    cmd=['certbot','certonly','--webroot','-w',w['document_root'],'--non-interactive','--agree-tos','--register-unsafely-without-email']
    for d in domains:cmd += ['-d',d]
    progress(15,"Requesting Let's Encrypt certificate")
    r=subprocess.run(cmd,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True,timeout=900)
    if r.returncode:raise RuntimeError('Certbot failed: '+r.stdout[-1800:])
    progress(75,'Installing certificate into SCpanel SSL store')
    import shutil as _sh
    crt_src=Path('/etc/letsencrypt/live')/w['domain']/ 'fullchain.pem'; key_src=Path('/etc/letsencrypt/live')/w['domain']/ 'privkey.pem'
    crt,key=website_cert(w['domain']); crt.parent.mkdir(parents=True,exist_ok=True); _sh.copy2(crt_src,crt); _sh.copy2(key_src,key)
    progress(90,'Regenerating website configuration'); _write_website_config(w,a); progress(100,'SSL complete')
    return {'message':'SSL certificate issued, installed and website configuration safely reloaded.','domain':w['domain'],'certificate':str(crt)}

def _ssl_operation(target,operation,progress,cancel_check):
    """Issue/renew/revoke/delete a managed website certificate."""
    from app.app import db, website_cert, _write_website_config, run_system, _sync_ssl_record
    try: wid=int(str(target).split(':')[-1])
    except Exception: raise RuntimeError('Invalid SSL website target.')
    c=db(); w=c.execute('SELECT * FROM websites WHERE id=?',(wid,)).fetchone(); a=None
    if w: a=c.execute('SELECT * FROM accounts WHERE id=?',(w['account_id'],)).fetchone()
    c.close()
    if not w: raise RuntimeError('Website not found.')
    domain=w['domain']; domains=[domain]+[x for x in (w['aliases'] or '').split() if x]
    crt,key=website_cert(domain)
    if operation in {'issue','renew'}:
        if not shutil.which('certbot'): raise RuntimeError('Certbot is not installed. Install it from App Store first.')
        progress(10,"Preparing Let's Encrypt request")
        if operation=='issue':
            cmd=['certbot','certonly','--webroot','-w',w['document_root'],'--non-interactive','--agree-tos','--register-unsafely-without-email']
            for d in domains: cmd += ['-d',d]
        else:
            cmd=['certbot','renew','--cert-name',domain,'--non-interactive','--deploy-hook','/usr/local/sbin/scpanel-certbot-deploy-hook']
        progress(20,'Running Certbot')
        r=subprocess.run(cmd,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True,timeout=900)
        if r.returncode: raise RuntimeError('Certbot failed: '+r.stdout[-2200:])
        progress(75,'Synchronizing certificate into SCpanel SSL store')
        src_crt=Path('/etc/letsencrypt/live')/domain/'fullchain.pem'; src_key=Path('/etc/letsencrypt/live')/domain/'privkey.pem'
        if not src_crt.exists() or not src_key.exists(): raise RuntimeError('Certbot completed but certificate files were not found.')
        crt.parent.mkdir(parents=True,exist_ok=True); shutil.copy2(src_crt,crt); shutil.copy2(src_key,key); os.chmod(key,0o600)
        progress(88,'Validating and reloading web-server configuration'); _write_website_config(w,a); _sync_ssl_record(wid)
        c=db(); c.execute("UPDATE ssl_certificates SET status='valid',last_renewed_at=CURRENT_TIMESTAMP,revoked_at='',updated_at=CURRENT_TIMESTAMP WHERE website_id=?",(wid,)); c.commit(); c.close()
        return {'message':f"Let's Encrypt certificate {operation}ed for {domain}.",'domain':domain,'certificate':str(crt)}
    if operation=='revoke':
        if not shutil.which('certbot'): raise RuntimeError('Certbot is not installed. Install it from App Store first.')
        if not crt.exists(): raise RuntimeError('No managed certificate exists to revoke.')
        progress(20,'Revoking certificate with Let’s Encrypt')
        r=subprocess.run(['certbot','revoke','--cert-path',str(crt),'--non-interactive','--reason','cessation-of-operation'],stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True,timeout=300)
        if r.returncode: raise RuntimeError('Certbot revoke failed: '+r.stdout[-1800:])
        progress(75,'Disabling HTTPS configuration');
        try:
            c=db();c.execute('UPDATE websites SET https_enforce=0,updated_at=CURRENT_TIMESTAMP WHERE id=?',(wid,));c.execute("UPDATE ssl_certificates SET status='revoked',force_https=0,revoked_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE website_id=?",(wid,));c.commit();c.close()
            w2=dict(w);w2['https_enforce']=0;_write_website_config(w2,a)
        except Exception: pass
        return {'message':f'Certificate revoked for {domain}.','domain':domain}
    if operation=='delete':
        progress(15,'Removing certificate from Certbot')
        if shutil.which('certbot'):
            r=subprocess.run(['certbot','delete','--cert-name',domain,'--non-interactive'],stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True,timeout=300)
            if r.returncode and 'not found' not in r.stdout.lower(): raise RuntimeError('Certbot delete failed: '+r.stdout[-1800:])
        progress(65,'Removing SCpanel certificate files')
        crt.unlink(missing_ok=True);key.unlink(missing_ok=True)
        c=db();c.execute('UPDATE websites SET https_enforce=0,updated_at=CURRENT_TIMESTAMP WHERE id=?',(wid,));c.execute("UPDATE ssl_certificates SET status='missing',force_https=0,issued_at='',expires_at='',revoked_at='',updated_at=CURRENT_TIMESTAMP WHERE website_id=?",(wid,));c.commit();c.close()
        w2=dict(w);w2['https_enforce']=0;_write_website_config(w2,a)
        return {'message':f'Certificate deleted for {domain}.','domain':domain}
    raise RuntimeError('Unsupported SSL operation.')

def _maintenance(operation,progress,cancel_check):
    if operation=='services':
        from app.providers.service_engine import catalog
        progress(50,'Rebuilding service discovery cache'); catalog(force=True); return {'message':'Service discovery cache rebuilt.'}
    if operation=='tasks':
        progress(50,'Recovering task queue'); recover_pending(); return {'message':'Task queue recovery completed.'}
    raise RuntimeError('Unsupported maintenance operation.')

def recover_pending():
    init();c=_conn();c.execute("UPDATE tasks SET status='failed',progress=100,step='Interrupted',message='Task interrupted by SCpanel restart; retry required.',error='worker process restarted',finished_at=CURRENT_TIMESTAMP WHERE status='running'");queued=[r['id'] for r in c.execute("SELECT id FROM tasks WHERE status='queued' ORDER BY created_at ASC LIMIT 50").fetchall()];c.commit();c.close()
    for tid in queued:_start(tid)

def _start(tid):
    with _lock:
        if tid in _threads:return
        t=threading.Thread(target=_worker,args=(tid,),daemon=True,name='scpanel-task-'+tid[:8]);_threads[tid]=t;t.start()

def cancel(tid,actor='system',ip='-'):
    t=_get(tid)
    if not t:raise RuntimeError('Task not found.')
    if t['status']=='queued':_update(tid,status='cancelled',progress=0,step='Cancelled',message='Cancelled before execution.',finished_at=time.strftime('%Y-%m-%d %H:%M:%S'),actor=actor,ip=ip);return _get(tid)
    if t['status']=='running':_cancel_flags.add(tid);_update(tid,status='cancel_requested',step='Stopping safely…',message='Stop requested; current OS command will finish or reach its next safe checkpoint.',actor=actor,ip=ip);return _get(tid)
    raise RuntimeError('Only queued or running tasks can be cancelled.')

def create(kind,app_key,operation,actor='system',ip='-',max_attempts=2,requested_version=''):
    init();tid=uuid.uuid4().hex;c=_conn();c.execute('INSERT INTO tasks(id,kind,app_key,operation,requested_version,status,progress,step,actor,ip,max_attempts) VALUES(?,?,?,?,?,?,?,?,?,?,?)',(tid,kind,app_key,operation,requested_version or '', 'queued',0,'Queued',actor,ip,max(1,int(max_attempts))));c.commit();c.close();_start(tid);return _get(tid)

def retry(tid,actor='system',ip='-'):
    t=_get(tid)
    if not t:raise RuntimeError('Task not found.')
    if t['status'] not in ('failed','cancelled'):raise RuntimeError('Only failed/cancelled tasks can be retried.')
    if int(t['attempts'])>=int(t['max_attempts']):raise RuntimeError('Retry limit reached for this task.')
    _update(tid,status='queued',progress=0,step='Queued',message='',error='',output='',finished_at=None,actor=actor,ip=ip);_start(tid);return _get(tid)

def live_output(tid):
    t=_get(tid)
    if not t:return ''
    try:
        if t.get('app_key'):
            p=Path('/var/log/scpanel/appstore')/(t['app_key']+'.log')
            if p.exists():return p.read_text(errors='replace')[-12000:]
    except Exception:pass
    return t.get('output') or t.get('error') or ''

def get(tid):return _get(tid)
def list_tasks(limit=100):
    c=_conn();rows=c.execute('SELECT * FROM tasks ORDER BY created_at DESC LIMIT ?',(max(1,min(int(limit),500)),)).fetchall();c.close();return [dict(r) for r in rows]