bmc_hub/app/workflows/worker.py

"""Durable processing; financial effects are deliberately approval-only."""
import logging
import json
from psycopg2.extras import Json
from .engine import evaluate
from .service import transaction, load_time_context, load_case_context, json_safe, is_incoming_mail

logger = logging.getLogger(__name__)

def process_pending_events(limit=20, mode='shadow', customer_ids=None):
    if mode not in ('shadow','active'): raise ValueError('Invalid workflow mode')
    processed = 0
    for _ in range(min(max(int(limit),0),100)):
        event = None
        try:
            with transaction() as cursor:
                cursor.execute('SELECT * FROM bmc_workflow_runtime WHERE id=1')
                runtime = cursor.fetchone()
                if not runtime or not runtime['enabled']: break
                cursor.execute("SELECT * FROM bmc_workflow_events WHERE status='pending' AND available_at<=now() ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1")
                event = cursor.fetchone()
                if not event: break
                event = dict(event)
                if event['trigger']=='mail.received' and not is_incoming_mail(event['context']):
                    cursor.execute("UPDATE bmc_workflow_events SET status='superseded',error='Outgoing mail is not a mail.received trigger' WHERE id=%s",(event['id'],))
                    continue
                if len(event.get('chain') or []) >= 20:
                    cursor.execute("UPDATE bmc_workflow_events SET status='failed',error='Loop limit exceeded' WHERE id=%s",(event['id'],))
                    continue
                cursor.execute("UPDATE bmc_workflow_events SET status='processing',attempts=attempts+1 WHERE id=%s",(event['id'],))
                if event['entity_type'] in ('time','time_entry'):
                    context = load_time_context(event['entity_id'],cursor)
                    if event['trigger'] not in ('time.created','time.updated'):
                        event['revision']=context.get('workflow_revision',0)
                    if int(context.get('workflow_revision',0)) != int(event['revision']):
                        cursor.execute("UPDATE bmc_workflow_events SET status='superseded' WHERE id=%s",(event['id'],))
                        continue
                    from app.workflows.classification import enrich_time_context
                    from app.core.config import settings
                    if getattr(settings,'WORKFLOW_AI_ENABLED',False):
                        from app.tags.backend.ai_jobs import enqueue_ai_job
                        enqueue_ai_job(cursor,context)
                    context = enrich_time_context(cursor,context)
                elif event['entity_type'] == 'case':
                    context = load_case_context(event['entity_id'],cursor)
                    from .classification import enrich_case_context
                    context=enrich_case_context(cursor,context)
                elif event['entity_type'] in ('task','reminder') and event['context'].get('sag_id'):
                    context=load_case_context(event['context']['sag_id'],cursor)
                    context.update(event['context'])
                    context['case_id']=event['context']['sag_id']
                    context['entity_type']=event['entity_type']
                else:
                    context = dict(event['context'])
                    if event['entity_type']=='customer': context['customer_id']=str(event['entity_id'])
                    if event['entity_type']=='mail': context['description']=context.get('body_text') or context.get('body_html') or ''
                    if event['entity_type'] in ('mail','email'):
                        from .classification import enrich_mail_context
                        context=enrich_mail_context(cursor,context)
                if event['context'].get('target_workflow_id'):
                    context['target_workflow_id']=event['context']['target_workflow_id']
                cursor.execute('SELECT w.*,v.id AS version_id,v.definition FROM bmc_workflows w JOIN bmc_workflow_versions v ON v.id=w.active_version_id AND v.workflow_id=w.id WHERE w.enabled=TRUE')
                result = evaluate(cursor.fetchall(),context,event['trigger'])
                result['input'] = context
                effective_mode = mode
                allowed = runtime['customer_ids'] if customer_ids is None else customer_ids
                internal_allowed=event['entity_type'] in ('time','time_entry') and context.get('customer_id') is None and not context.get('customer_mapping_missing') and runtime.get('internal_enabled',False)
                from .integration import customer_enabled
                customer_allowed=context.get('customer_id') is not None and customer_enabled(context['customer_id'],cursor,fallback_ids=allowed)
                if mode == 'active' and (runtime['mode'] != 'active' or (not internal_allowed and not customer_allowed)):
                    effective_mode = 'shadow'
                result['mode'] = effective_mode
                result['billing'] = result.get('decision',{})
                cursor.execute('INSERT INTO bmc_workflow_executions(event_id,mode,result) VALUES(%s,%s,%s) ON CONFLICT(event_id,mode) DO NOTHING',(event['id'],effective_mode,Json(json_safe(result))))
                if effective_mode == 'active' and event['entity_type'] in ('time','time_entry'):
                    cursor.execute('SELECT * FROM tmodule_times WHERE id=%s FOR UPDATE',(event['entity_id'],))
                    time = cursor.fetchone()
                    if int(time.get('workflow_revision',0)) != int(event['revision']):
                        cursor.execute("UPDATE bmc_workflow_events SET status='superseded' WHERE id=%s",(event['id'],))
                        continue
                    if any(action['type']=='suggest_tag' for action in result['actions']):
                        result.update(status='manual_review',method=None,reason='pending_tag_approval')
                    cursor.execute('SELECT * FROM bmc_billing_decisions WHERE time_id=%s AND is_current FOR UPDATE',(event['entity_id'],))
                    previous = cursor.fetchone()
                    cursor.execute('SELECT time_id FROM bmc_workflow_settlements WHERE time_id=%s',(event['entity_id'],))
                    settled = cursor.fetchone()
                    if not settled and time['status'] not in ('approved','billed') and not (previous and (previous['manual_override'] or previous['status']=='settled')):
                        cursor.execute('UPDATE bmc_billing_decisions SET is_current=FALSE WHERE time_id=%s AND is_current',(event['entity_id'],))
                        cursor.execute('INSERT INTO bmc_billing_decisions(time_id,revision,event_id,status,method,details) VALUES(%s,%s,%s,%s,%s,%s)',(event['entity_id'],event['revision'],event['id'],result['status'],result['method'],Json(json_safe(result))))
                if effective_mode == 'active':
                    chain = list(event.get('chain') or [])
                    matched_ids = list(dict.fromkeys(a['source_workflow_id'] for a in result['actions'] if a.get('source_workflow_id')))
                    repeated = [action for action in result['actions'] if action.get('source_workflow_id') in chain and action['type'] in ('create_task','create_reminder','notify','start_workflow','set_field','assign_user','assign_group')]
                    if repeated:
                        raise ValueError('Workflow action repeats in its causal chain')
                    cursor.execute("SELECT set_config('bmc.workflow_chain',%s,TRUE)",(json.dumps(chain+matched_ids),))
                    from .classification import apply_keyword_matches
                    if event['entity_type'] in ('time','time_entry'):
                        apply_keyword_matches(cursor,context)
                    elif event['entity_type'] in ('case','mail','email'):
                        from .classification import apply_entity_keyword_matches
                        apply_entity_keyword_matches(cursor,'email' if event['entity_type']=='mail' else event['entity_type'],event['entity_id'],context)
                # Every local effect is deduplicated and committed with its event.
                for index, action in enumerate(result['actions']):
                    cursor.execute('INSERT INTO bmc_workflow_effects(event_id,action_key,payload) VALUES(%s,%s,%s) ON CONFLICT(action_key) DO NOTHING RETURNING id',(event['id'],f"{event['id']}:{index}",Json(action)))
                    inserted=cursor.fetchone()
                    if inserted and effective_mode=='active' and action['type']=='set_billing' and event['entity_type'] not in ('time','time_entry'):
                        cursor.execute("UPDATE bmc_workflow_effects SET status='skipped',result=%s WHERE action_key=%s",(Json({'reason':'billing_requires_time_entry'}),f"{event['id']}:{index}"))
                    if inserted and action['type']=='send_to_review':
                        cursor.execute("UPDATE bmc_workflow_effects SET status='done' WHERE action_key=%s",(f"{event['id']}:{index}",))
                    if inserted and effective_mode == 'active' and action['type'] not in ('set_billing','send_to_review'):
                        from .actions import execute_local_action
                        execute_local_action(cursor,event,action)
                        cursor.execute("UPDATE bmc_workflow_effects SET status='done' WHERE action_key=%s",(f"{event['id']}:{index}",))
                cursor.execute("UPDATE bmc_workflow_events SET status='done',error=NULL WHERE id=%s",(event['id'],))
                processed += 1
        except Exception as exc:
            logger.exception('Workflow processing failed')
            if event:
                with transaction() as cursor:
                    cursor.execute("UPDATE bmc_workflow_events SET attempts=attempts+1,status=CASE WHEN attempts>=4 THEN 'failed' ELSE 'pending' END,available_at=now()+interval '30 seconds',error=%s WHERE id=%s",(str(exc)[:1000],event['id']))
            else: raise
    return processed