j" TdZddlZddlZddlmZddlmZmZddlm Z ddl m Z eje ZdZdZd Zd efd Zd ed efd Zded efdZddZdededededed df dZded eefdZdeded edzfdZdededed dfdZdededed dfdZded efdZdeded dfdZ y)a{Storage for every product ingestion source, keyed by (kind, external_ref). Lives in the master database rather than a per-tenant schema for two reasons. Shopify webhooks arrive keyed only by shop domain with no tenant, so per-tenant storage would mean scanning every schema to route one. And the UNIQUE constraint enforces one-shop-one-tenant globally, which per-schema cannot. N)Fernet)JsonRealDictCursor)settings)get_master_db_connectionshopifyhttp_apicrawlreturncPttjjS)N)rrSOURCE_CREDENTIALS_KEYencode;/var/www/html/strategist-ai/app/services/catalog/sources.py_fernetrs (1188: ;;rcredsctjtj|dj j S)aOEncrypt a credentials object. A dict rather than a bare string, so the same envelope holds a Shopify access token or an API key plus client secret without a schema change. sort_keys makes the plaintext stable, which matters for tests but not for security -- Fernet's IV already makes ciphertext non-deterministic. T) sort_keys)rencryptjsondumpsrdecode)rs rencrypt_credentialsrs4 9  TZZ>EEG H O O QQrblobctjtj|j j S)zMDecrypt a credentials blob. Raises InvalidToken if tampered or wrongly keyed.)rloadsrdecryptrr)rs rdecrypt_credentialsr*s- ::gi'' 6==? @@rc~t} |j5}|jd|jdddd|j |jy#1swY+xYw#t$r)|j t jddwxYw#|jwxYw)Na CREATE TABLE IF NOT EXISTS product_sources ( id SERIAL PRIMARY KEY, tenant_id TEXT NOT NULL, kind TEXT NOT NULL, external_ref TEXT NOT NULL, config JSONB NOT NULL DEFAULT '{}', credentials_encrypted TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active', connected_at TIMESTAMPTZ NOT NULL DEFAULT now(), disconnected_at TIMESTAMPTZ, last_synced_at TIMESTAMPTZ, UNIQUE (kind, external_ref) ) zTCREATE INDEX IF NOT EXISTS product_sources_tenant_idx ON product_sources (tenant_id)z&Could not ensure product_sources tableTexc_info rcursorexecutecommit Exceptionrollbackloggererrorclose)conncurs r ensure_tabler./s # %D [[] c KK  KK1 ! ( 5  *   = M  s-A5#A)A5)A2.A552B''B**B< tenant_idkind external_refconfig credentialsc tt} |j5}|jd|||t |t |fddd|j tjd||| |jy#1swYCxYw#t$r+|jtjd||dwxYw#|jwxYw)zInsert a source, or update it in place on (kind, external_ref) conflict. The ON CONFLICT target is what makes a reinstall update the existing row instead of creating a second row under a different tenant. a INSERT INTO product_sources (tenant_id, kind, external_ref, config, credentials_encrypted, status, connected_at, disconnected_at) VALUES (%s, %s, %s, %s, %s, 'active', now(), NULL) ON CONFLICT (kind, external_ref) DO UPDATE SET tenant_id = EXCLUDED.tenant_id, config = EXCLUDED.config, credentials_encrypted = EXCLUDED.credentials_encrypted, status = 'active', connected_at = now(), disconnected_at = NULL Nz!Stored %s source %s for tenant %sz'Could not store %s source for tenant %sTr!) r.rr$r%rrr&r)infor'r(r*r+)r/r0r1r2r3r,r-s r upsert_sourcer6OsN # %D [[] 5c KK T<f%k24 5 5  7|YW / 5 5"   >i"  $  s.B*B0BBB4CCC'cptt} |jt5}|j d|f|j Dcgc] }t |c}cddd|jScc}w#1swYnxYw |jy#|jwxYw)z=Every active source for a tenant. Never includes credentials.)cursor_factorya SELECT kind, external_ref, config, status, connected_at, last_synced_at FROM product_sources WHERE tenant_id = %s AND status = 'active' ORDER BY connected_at DESC N)r.rr$rr%fetchalldictr+)r/r,r-rs r get_sourcesr<ssN # %D  [[[ 7 53 KK   &)\\^4DG4 5 5 5 5 5 5  s4B#&BB$B& B#BB B##B5cRtt} |j5}|jd||f|j }|rt |dndcddd|j S#1swYnxYw |j y#|j wxYw)zsDecrypted credentials for one active source, or None if not found. Callers must not log the return value. zmSELECT credentials_encrypted FROM product_sources WHERE kind = %s AND external_ref = %s AND status = 'active'rN)r.rr$r%fetchonerr+)r0r1r,r-rows rget_credentialsr@s N # %D  [[] @c KKN|$  ,,.C25&s1v.4 @ @  @ @ @  s"B6A6 B6A?;BB&czt} |j5}|jdt|||fddd|j  |jy#1swY+xYw#t $r+|j tjd||dwxYw#|jwxYw)zePersist a config change (e.g. first-sync field inference) without touching credentials or status.zLUPDATE product_sources SET config = %s WHERE kind = %s AND external_ref = %sNz!Could not update config for %s:%sTr!) rr$r%rr&r'r(r)r*r+)r0r1r2r,r-s rupdate_source_configrBs $ %D [[] c KK8ft\2         8$ "  $  s-A1A%A1%A.*A114B%%B((B:statuscht} |j5}|jd|||fddd|j |jy#1swY+xYw#t$r+|j t jd||dwxYw#|jwxYw)zA source whose credentials stopped working must stop being retried silently; the dashboard needs to show that it needs reconnecting.zLUPDATE product_sources SET status = %s WHERE kind = %s AND external_ref = %sNzCould not set status for %s:%sTr!r#)r0r1rCr,r-s rmark_source_statusrEs $ %D [[] c KK8|,         5t\"  $  -A(AA(A%!A((4BBB1ctt} |j5}|jd|f|j}ddd|j t jd|||jS#1swYBxYw#t$r*|jt jd|dwxYw#|jwxYw)aRevoke every active source a tenant has connected. Pairs with wiping the whole catalogue: without this, a source stays connected after its products are deleted, and the next build silently re-crawls it and repopulates exactly what was just removed.zqUPDATE product_sources SET status = 'revoked', disconnected_at = now() WHERE tenant_id = %s AND status = 'active'Nz'Disconnected %d source(s) for tenant %sz*Could not disconnect sources for tenant %sTr!) r.rr$r%rowcountr&r)r5r+r'r(r*)r/r,r-counts rdisconnect_all_sourcesrJs N # %D [[] !c KK=   LLE  !  =uiP  ! !   A9W[ \  s.B B0BB B3CC Ccht} |j5}|jd|||fddd|j |jy#1swY+xYw#t$r+|j t jd||dwxYw#|jwxYw)NzTUPDATE product_sources SET last_synced_at = %s WHERE kind = %s AND external_ref = %sz&Could not set last_synced_at for %s:%sTr!r#)r0r1whenr,r-s rtouch_last_syncedrMs # %D [[] c KK8t\*         =t\"  $  rF)r N)!__doc__rloggingcryptography.fernetrpsycopg2.extrasrrapp.core.configrapp.services.infra.databaser getLogger__name__r) KIND_SHOPIFY KIND_HTTP_API KIND_CRAWLrr:strrrr.r6listr<r@rBrEintrJrMrrrr\sV &0$@   8 $   <<RtRRAcAdA @!S!!3!!.2!7;!H34:$#STD[(s#t*SST*cc4CsTr