j% dZddlZddlZddlZddlZddlmZmZmZddlZddl m Z ddl m Z m Z ddlmZej eZdZdZd ZGd d eZd#d Zded efdZdeded efdZdeded efdZdededed dfdZd$dededed dfdZ d$dededed dfdZ!defdZ"defdZ#deded e$fdZ%d%deded e$fd Z&ded dfd!Z'ded dfd"Z(y)&aJob state store for long-running catalogue jobs (sync, enrich, pair, build). These jobs used to hold an HTTP connection open for minutes. This module lets a job run in the background while a client polls `get(job_id)` for progress. State lives in Postgres rather than Redis. Redis was the first choice -- job state is ephemeral and a TTL is free there -- but the deployment could not reach it reliably, and a control plane that fails whenever its own store is unreachable is worse than one that shares a database already proven to work. Postgres also gives two things Redis did not: a job survives a restart, and the "one job per tenant per kind" rule becomes a unique index the database enforces rather than a lock the application has to remember to release. N)datetime timedeltatimezone)errors)JsonRealDictCursor)get_master_db_connectioni,iQ)queuedrunningc(eZdZdZdeffd ZxZS)JobAlreadyRunningzRaised when a tenant already has a job of this kind in flight. Two concurrent runs of the same kind (e.g. two `pair` jobs) would both rewrite the same downstream table, so `create` refuses the second one. job_idc8||_t| d|y)NzJob already running: )rsuper__init__)selfr __class__s 6/var/www/html/strategist-ai/app/services/infra/jobs.pyrzJobAlreadyRunning.__init__)s  09:)__name__ __module__ __qualname____doc__strr __classcell__)rs@rr r "s ;s;;rr returnct} |j5}|jd|jd|jdddd|j |jy#1swY+xYw#t$r)|j t jddwxYw#|jwxYw)NaU CREATE TABLE IF NOT EXISTS catalog_jobs ( job_id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, kind TEXT NOT NULL, status TEXT NOT NULL, percent INT NOT NULL DEFAULT 0, step TEXT NOT NULL DEFAULT '', result JSONB, error TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ) z CREATE UNIQUE INDEX IF NOT EXISTS catalog_jobs_active_idx ON catalog_jobs (tenant_id, kind) WHERE status IN ('queued', 'running') z_CREATE INDEX IF NOT EXISTS catalog_jobs_tenant_idx ON catalog_jobs (tenant_id, created_at DESC)z#Could not ensure catalog_jobs tableTexc_info) r cursorexecutecommit Exceptionrollbackloggererrorclose)conncurs r ensure_tabler*.s # %D" [[] @c KK  $ KK  KK? @1 @6 C @ @8   :T J  s.B4A:B:B?B2B88B;;C rowc B|d}|d}tjtj|z j }|dk(r |t kDrd}|d|d|d|t |d|d xsd |d |d xsd|d j|jd S)Nstatus updated_atr lostr tenant_idkindpercentstepresultr& created_at) rr0r1r-r2r3r5r&r6r.)rnowrutc total_secondsSTALE_AFTER_SECONDSint timestamp)r+r-r.ages r _row_to_stater>Us ]F\"J << % 2 A A CCs%88h-%F s9~&F !rh-W%,'113 **,  rr0r1ctdtjj}t j t jttz }t} |jt5}|jd|||f|jdt j t jttz f |jd|||f ddd|j) |j-|S#tj $rw|j#|jt5}|jd||f|j%}dddn #1swYnxYwt'r |d|wxYw#1swYxYw#t&$rt*$r|j#wxYw#|j-wxYw) Njob_)secondscursor_factoryz UPDATE catalog_jobs SET status = 'lost', updated_at = now() WHERE tenant_id = %s AND kind = %s AND status IN ('queued', 'running') AND updated_at < %s z.DELETE FROM catalog_jobs WHERE created_at < %sz INSERT INTO catalog_jobs (job_id, tenant_id, kind, status) VALUES (%s, %s, %s, 'queued') z SELECT job_id FROM catalog_jobs WHERE tenant_id = %s AND kind = %s AND status IN ('queued', 'running') LIMIT 1 r)r*uuiduuid4hexrr7rr8rr:r r rr!JOB_TTL_SECONDSrUniqueViolationr$fetchoner r"r#r')r0r1r stale_beforer(r)retryexistings r_createrMmsNDJJL$$% &F<< - BU0VVL # %D( [[[ 7 T3 KK T<0  2 KK@hll+i.PPR T T i.0# T@  M))) T [[[?05MM# $T* , %~~/H000(h(:SSFSS T+ T TB     s[+F7AF+D3F79F(%F< F(F F((F++F40F77$GGG0chKtjjfdd{S7w)NctSN)rM)r1r0srzcreate..s')T2Jranyio to_threadrun_sync)r0r1s``rcreaterVs#))*JK KK K '202rr2r3cZ t} |j5}|jdt|||fddd|j |j y#1swY*xYw#|j wxYw#t $rtjd|dYywxYw)a]Best-effort progress report. Never raises. Deliberately synchronous: the job runs in a worker thread, so it can write its own progress rather than handing a coroutine back to the event loop. This runs alongside work that is already spending real money on model calls -- a database hiccup here must not abort an enrichment run. z UPDATE catalog_jobs SET status = 'running', percent = %s, step = %s, updated_at = now() WHERE job_id = %s Nz!Progress update failed for job %sTr) r r r!r;r"r'r#r%warning)rr2r3r(r)s rtickrZsS')  3#  'lD&1 3 3 KKM JJL 3 3 JJL S:FTRSs> BA1A%A1B%A.*A11BB!B*)B*r5r&c t} |j5}|jd|rdnd|| t|nd||fddd|j |jy#1swY*xYw#t $r+|j tjd|dYPwxYw#|jwxYw)Na< UPDATE catalog_jobs SET status = %s, percent = CASE WHEN %s IS NULL THEN 100 ELSE percent END, result = COALESCE(%s, result), error = %s, updated_at = now() WHERE job_id = %s faileddonezCould not finish job %sTr) r r r!rr"r#r$r%r&r')rr5r&r(r)s r_finishr^s # %D [[] Pc KK$("("4$v,$vO P P % P P G  . FG s:A:)A.A:.A73A::1B.+B1-B..B11CcnKtjjfdd{y7w)NctSrP)r^)r&rr5srrQzfinish..s7665+IrrR)rr5r&s```rfinishras // " "#I JJJs (535c.t} |jt5}|jd|f|j }ddd|j rt t|SdS#1swY1xYw#|j wxYw)NrBz,SELECT * FROM catalog_jobs WHERE job_id = %s)r r rr!rIr'r>dict)rr(r)r+s r_getrds} # %D [[[ 7 !3 KKF R,,.C ! '*=c #44  ! ! s"B$A6B6A?;BBcdKtjjfdd{S7w)NctSrP)rdrsrrQzget..s $v,rrRrgs`rgetrhs#))*>? ?? ?s &0.0limitcNt} |jt5}|jd||f|j }ddd|j Dcgc]}t t|c}S#1swY;xYw#|j wxYwcc}w)NrBz SELECT * FROM catalog_jobs WHERE tenant_id = %s ORDER BY created_at DESC LIMIT %s )r r rr!fetchallr'r>rc)r0rir(r)rowsr+s r_recentrms # %D [[[ 7 "3 KKU# %<<>D  " 04 5M$s) $ 55 " " 5s(B %BB #B"B B BchKtjjfdd{S7w)NctSrP)rm)rir0srrQzrecent..s')U2KrrR)r0ris``rrecentrps#))*KL LL LrWct} |j5}|jd|fddd|j|j y#1swY*xYw#|j wxYw)Nz-DELETE FROM catalog_jobs WHERE tenant_id = %s)r r r!r"r')r0r(r)s r _clear_tenantrrsb # %D [[] Wc KKG) V W   W W s!A%AA%A"A%%A7cfKtjjfdd{y7w)zTest-only teardown.ctSrP)rrr0srrQzclear_tenant..s =+CrNrRrus`r clear_tenantrvs // " "#C DDDs &1/1)rN)NN)))rjsonloggingtimerDrrrrSpsycopg2rpsycopg2.extrasrrapp.services.infra.databaser getLoggerrr%r:rGACTIVE_STATUSESr#r r*rcr>rrMrVr;rZr^rardrhlistrmrprrrvrrrs   22 0@   8 $' ; ;$Nt00s0#0#0fLCLsLsLSSsS#S$S4CSD0KKdK#KK55@c@ 6s 63 64 6MCMMTMSTE#E$Er