122 lines
9.7 KiB
Python
122 lines
9.7 KiB
Python
"""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)
|
|
if mode == 'active' and (runtime['mode'] != 'active' or (not internal_allowed and str(context.get('customer_id')) not in [str(value) for value in 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
|