"""Reads and writes the platform's integration tables. Those tables belong to the Node application and are managed by its Sequelize migrations. Every access is funnelled through this module so the cross-service coupling sits in one reviewable place, and so a schema change over there breaks one file rather than several. """ import logging import time import uuid as uuid_lib from psycopg2.extras import Json, RealDictCursor from app.services.infra.database import get_master_db_connection logger = logging.getLogger(__name__) TENANT_PREFIX = "org_" JOB_NAME = "integration_oauth_connect" # verify_hmac calls get_platform_config on every OAuth callback, and a dead # master DB costs up to ~34s (3 retries x 10s timeout + 2 x 2s backoff) per # lookup. Caching -- including the miss case -- means that cost is paid once # per TTL window instead of once per callback. 300s bounds how long a rotated # client secret takes to apply without a process restart, which matters # because easy rotation is the reason credentials moved into this table. _CONFIG_CACHE_TTL_SECONDS = 300 _config_cache = {} class TenantHasNoUserError(Exception): """integrations.user is NOT NULL, and this tenant has no owning user.""" def strip_tenant_prefix(tenant_id) -> str: if not tenant_id or not isinstance(tenant_id, str): raise ValueError("tenant_id must be a non-empty string") if not tenant_id.startswith(TENANT_PREFIX): raise ValueError(f"tenant_id {tenant_id!r} is missing the org_ prefix") raw = tenant_id[len(TENANT_PREFIX):] # Parsed rather than pattern-matched: the value lands in a uuid column with a # foreign key, so a malformed one must fail here, not mid-callback. return str(uuid_lib.UUID(raw)) def get_platform_config(platform: str): cached = _config_cache.get(platform) if cached is not None: value, expires_at = cached if time.monotonic() < expires_at: return value conn = None try: conn = get_master_db_connection() with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( 'SELECT "clientId", "clientSecret", "redirectUri", scope ' 'FROM platform_configs WHERE platform = %s AND "isActive" = true', (platform,)) row = cur.fetchone() value = None if not row else { "client_id": row["clientId"], "client_secret": row["clientSecret"], "redirect_uri": row["redirectUri"], "scope": row["scope"]} except Exception: # A dead master DB must still only cost one ~34s connection attempt # per TTL window: cache the miss so the next callback returns fast # instead of hanging again, then let this first failure propagate so # the caller's existing fallback-and-log path still runs. _config_cache[platform] = (None, time.monotonic() + _CONFIG_CACHE_TTL_SECONDS) raise finally: if conn is not None: conn.close() _config_cache[platform] = (value, time.monotonic() + _CONFIG_CACHE_TTL_SECONDS) return value def resolve_tenant_user(tenant_uuid: str): conn = get_master_db_connection() try: with conn.cursor() as cur: cur.execute('SELECT "userId" FROM "Tenants" WHERE id = %s', (tenant_uuid,)) row = cur.fetchone() return row[0] if row and row[0] else None finally: conn.close() def upsert_integration(tenant_id: str, provider: str, auth_type: str, access_token_ciphertext: str, metadata: dict) -> str: tenant_uuid = strip_tenant_prefix(tenant_id) user_id = resolve_tenant_user(tenant_uuid) if not user_id: raise TenantHasNoUserError( f"Tenant {tenant_uuid} has no userId; integrations.user is NOT NULL") payload = dict(metadata or {}) # Records that accessToken holds ciphertext, not a raw token, so a future # reader does not send it to Shopify verbatim. payload["token_encoding"] = "fernet" conn = get_master_db_connection() try: with conn.cursor() as cur: cur.execute( 'SELECT id FROM integrations WHERE "tenantId" = %s AND provider = %s', (tenant_uuid, provider)) existing = cur.fetchone() if existing: cur.execute( 'UPDATE integrations SET "accessToken" = %s, "authType" = %s, ' 'metadata = %s, status = %s, "updatedAt" = now() WHERE id = %s', (access_token_ciphertext, auth_type, Json(payload), "ACTIVE", existing[0])) integration_id = str(existing[0]) else: integration_id = str(uuid_lib.uuid4()) cur.execute( 'INSERT INTO integrations (id, "user", provider, "authType", ' '"accessToken", metadata, status, "tenantId", "createdAt", ' '"updatedAt") VALUES (%s,%s,%s,%s,%s,%s,%s,%s, now(), now())', (integration_id, user_id, provider, auth_type, access_token_ciphertext, Json(payload), "ACTIVE", tenant_uuid)) conn.commit() logger.info("Recorded %s integration for tenant %s", provider, tenant_uuid) return integration_id except Exception: conn.rollback() logger.error("Could not record %s integration for tenant %s", provider, tenant_uuid, exc_info=True) raise finally: conn.close() def set_integration_status(tenant_id: str, provider: str, status: str) -> None: tenant_uuid = strip_tenant_prefix(tenant_id) conn = get_master_db_connection() try: with conn.cursor() as cur: cur.execute( 'UPDATE integrations SET status = %s, "updatedAt" = now() ' 'WHERE "tenantId" = %s AND provider = %s', (status, tenant_uuid, provider)) conn.commit() except Exception: conn.rollback() logger.error("Could not set %s status for tenant %s", provider, tenant_uuid, exc_info=True) raise finally: conn.close() def get_integration(tenant_id: str, provider: str): """Never returns accessToken -- callers that need the token read it from product_sources, where it is stored with the rest of the sync state.""" tenant_uuid = strip_tenant_prefix(tenant_id) conn = get_master_db_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( 'SELECT id, provider, "authType", status, metadata, "createdAt", ' '"updatedAt" FROM integrations WHERE "tenantId" = %s AND provider = %s', (tenant_uuid, provider)) row = cur.fetchone() return dict(row) if row else None finally: conn.close() def track_connect(tenant_id: str, platform: str, delivery_status: str, outcome: str, error_message=None, extra=None) -> None: """Best-effort audit. A tracker failure must never fail a connection that otherwise succeeded, so this logs and returns rather than raising.""" try: tenant_uuid = strip_tenant_prefix(tenant_id) except ValueError: logger.warning("Cannot track connect for malformed tenant %r", tenant_id) return metadata = {"jobName": JOB_NAME, "providerRaw": platform, "connectStep": delivery_status} metadata.update(extra or {}) # Connection acquisition sits inside the try too: get_master_db_connection # can itself raise (a dead master DB is the likeliest failure of the # three), and this function must never raise regardless of which step # fails. conn starts as None so the finally cannot NameError if the # connection was never obtained. conn = None try: conn = get_master_db_connection() with conn.cursor() as cur: cur.execute( 'INSERT INTO integration_connect_tracker ("tenantId", platform, ' 'outcome, "deliveryStatus", "errorMessage", metadata, "createdAt", ' '"updatedAt") VALUES (%s,%s,%s,%s,%s,%s, now(), now())', (tenant_uuid, platform, outcome, delivery_status, error_message, Json(metadata))) conn.commit() except Exception: if conn is not None: conn.rollback() logger.error("Could not write connect tracker row for %s", platform, exc_info=True) finally: if conn is not None: conn.close()