# Shopify OAuth Connector Implementation Plan


**Goal:** Let a merchant connect a Shopify store to a GalaxiQ tenant from `tests/chat_ui.html`, and store a verified, encrypted offline access token in `galaxiq_master.product_sources`.

**Architecture:** Three new modules behind a `/api/shopify` router. `shopify_oauth.py` holds pure functions (validation, HMAC, URL building, HTTP calls) with no DB or global state, so every security control is unit-testable without a database. `product_sources.py` owns the source table and Fernet encryption, and is source-kind agnostic so a later HTTP-API connector reuses it unchanged. `app/api/shopify.py` wires them into three endpoints. The OAuth `state` nonce lives in Redis via the existing pool.

**Tech Stack:** FastAPI, psycopg2, redis-py (async), httpx, cryptography (Fernet), pytest.

**Spec:** `docs/specs/2026-08-10-shopify-oauth-connector-design.md`

## Global Constraints

- **Never commit, never push, never stage.** The user commits all work themselves and has not yet asked for commits. Every task ends by listing changed files and reporting. Do not run `git commit`, `git push`, or `git add`. Steps in this plan that say "stage changes" mean: run `git status --short` and report the changed files.
- **No assistant attribution.** Commit messages carry no AI-assistant trailers or co-author lines.
- **Comments explain why, not what.** This codebase's norm is a short comment only where the reason for a decision is not evident from the code — see `CRAWL_MAX_PAGES` in `app/core/config.py` and the `save_products` docstring. Do not narrate what a line does, do not label obvious blocks, and do not restate the function name in its docstring. Where this plan's sample code carries a comment that only describes mechanics, drop it; keep the ones that record a reason or a hazard. Delete every comment that would read as noise to a reviewer who knows Python.
- **Match existing conventions exactly:** module-level `logger = logging.getLogger(__name__)`, snake_case, double quotes, 4-space indent, type hints on public function signatures, and the same import ordering as neighbouring modules (stdlib, third-party, `app.*`).
- **API version is pinned to `2026-07` in code**, not in settings, matching Shopify app version `galaxiq-5`.
- **Scopes are exactly `read_products,read_inventory`.**
- **Redirect URI is `{PUBLIC_BASE_URL}/api/shopify/callback`.** `PUBLIC_BASE_URL` is already `http://localhost:8001` in `.env` and matches the URL registered with Shopify. Do not hardcode the host.
- **Omit `grant_options[]` from the authorize URL** — its absence is what yields an offline token. An online token expires with the merchant's admin session.
- **The access token is never logged, never returned by any endpoint, and never sent to the browser.**
- Settings use `pydantic_settings.BaseSettings` in `app/core/config.py` with `extra = "ignore"`.
- Tests run from the repo root: `.venv/bin/python -m pytest`. `tests/unit/conftest.py` seeds required env vars so tests run without `.env`.
- Follow existing codebase conventions: module-level `logger = logging.getLogger(__name__)`, `psycopg2` with `sql.SQL`/`sql.Identifier` for any interpolated identifier, `httpx.AsyncClient` for outbound HTTP.

---

### Task 1: Configuration, dependencies, and test environment

Adds the four settings the connector needs, pins `cryptography`, and seeds test env vars. No behaviour yet — this is the foundation every later task imports.

**Files:**
- Modify: `app/core/config.py`
- Modify: `requirements.txt`
- Modify: `tests/unit/conftest.py`
- Modify: `.env`
- Test: `tests/unit/test_shopify_config.py`

**Interfaces:**
- Consumes: nothing.
- Produces: `settings.SHOPIFY_CLIENT_ID: str`, `settings.SHOPIFY_CLIENT_SECRET: str`, `settings.SOURCE_CREDENTIALS_KEY: str`, `settings.FRONTEND_URL: str`. All later tasks read these from `app.core.config.settings`.

- [ ] **Step 1: Generate a Fernet key and add secrets to `.env`**

Run:

```bash
.venv/bin/python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
```

Append to `.env`, using the generated key and the **rotated** client secret:

```
SHOPIFY_CLIENT_ID=487ae6335e5697ebe1d51e2a932ae240
SHOPIFY_CLIENT_SECRET=<rotated secret from Dev Dashboard>
SOURCE_CREDENTIALS_KEY=<the key printed above>
FRONTEND_URL=http://localhost:5173/chat_ui.html
```

`.env` is already in `.gitignore`. Confirm with `grep -n '^\.env$' .gitignore` before writing.

- [ ] **Step 2: Write the failing test**

Create `tests/unit/test_shopify_config.py`:

```python
from app.core.config import settings


def test_shopify_settings_present():
    assert settings.SHOPIFY_CLIENT_ID
    assert settings.SHOPIFY_CLIENT_SECRET
    assert settings.SOURCE_CREDENTIALS_KEY
    assert settings.FRONTEND_URL.startswith("http")


def test_credentials_key_is_valid_fernet():
    from cryptography.fernet import Fernet
    f = Fernet(settings.SOURCE_CREDENTIALS_KEY.encode())
    assert f.decrypt(f.encrypt(b"probe")) == b"probe"
```

- [ ] **Step 3: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_config.py -v`
Expected: FAIL with `AttributeError: 'Settings' object has no attribute 'SHOPIFY_CLIENT_ID'`

- [ ] **Step 4: Add the settings**

In `app/core/config.py`, inside `class Settings(BaseSettings)`, immediately before `class Config:`:

```python
    # Shopify OAuth connector.
    # CLIENT_ID and CLIENT_SECRET come from the Shopify Dev Dashboard app.
    # The secret signs every callback and webhook HMAC; rotating it invalidates
    # nothing stored, but a leaked one lets anyone forge callbacks.
    SHOPIFY_CLIENT_ID: str = ""
    SHOPIFY_CLIENT_SECRET: str = ""

    # Fernet key for every product source's credentials, not just Shopify.
    # Rotating this makes all stored credentials undecryptable, so merchants
    # would have to reconnect.
    SOURCE_CREDENTIALS_KEY: str = ""

    # Where the OAuth callback sends the merchant when the flow finishes.
    FRONTEND_URL: str = "http://localhost:5173/chat_ui.html"
```

- [ ] **Step 5: Seed test env vars**

Append to `tests/unit/conftest.py`:

```python
os.environ.setdefault("SHOPIFY_CLIENT_ID", "test-client-id")
os.environ.setdefault("SHOPIFY_CLIENT_SECRET", "test-client-secret")
os.environ.setdefault("SOURCE_CREDENTIALS_KEY",
                      "1Xh9nLxJ0mQ8vKcR2sT4uY6wZ8aB0cD2eF4gH6iJ8kM=")
os.environ.setdefault("FRONTEND_URL", "http://localhost:5173/chat_ui.html")
os.environ.setdefault("PUBLIC_BASE_URL", "http://localhost:8001")
```

The `SOURCE_CREDENTIALS_KEY` above is a fixed 32-byte urlsafe-base64 value so tests are deterministic. It is a test fixture, never used in `.env`.

- [ ] **Step 6: Pin cryptography**

Add to `requirements.txt`, after the `redis` line:

```
cryptography
```

- [ ] **Step 7: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_config.py -v`
Expected: PASS, 2 tests

- [ ] **Step 8: Report changed files**

```bash
git status --short
```

Do not commit and do not stage. Report the changed files.

---

### Task 2: Shop domain validation and authorize URL

The shop regex is a security control, not input tidiness. Without it, `?shop=evil.com` makes the server redirect merchants to an attacker's host.

**Files:**
- Create: `app/services/shopify_oauth.py`
- Test: `tests/unit/test_shopify_oauth.py`

**Interfaces:**
- Consumes: `settings.SHOPIFY_CLIENT_ID`, `settings.PUBLIC_BASE_URL`.
- Produces:
  - `SHOPIFY_API_VERSION: str` (`"2026-07"`)
  - `SHOPIFY_SCOPES: str` (`"read_products,read_inventory"`)
  - `is_valid_shop_domain(shop: str | None) -> bool`
  - `normalise_shop_domain(raw: str | None) -> str | None`
  - `build_authorize_url(shop: str, state: str) -> str`
  - `redirect_uri() -> str`

- [ ] **Step 1: Write the failing test**

Create `tests/unit/test_shopify_oauth.py`:

```python
import pytest
from urllib.parse import urlparse, parse_qs

from app.services.shopify_oauth import (
    SHOPIFY_API_VERSION, SHOPIFY_SCOPES,
    build_authorize_url, is_valid_shop_domain, normalise_shop_domain, redirect_uri,
)


@pytest.mark.parametrize("shop", [
    "galaxiq-braexaal.myshopify.com",
    "a.myshopify.com",
    "store123.myshopify.com",
])
def test_valid_shop_domains_accepted(shop):
    assert is_valid_shop_domain(shop) is True


@pytest.mark.parametrize("shop", [
    None,
    "",
    "evil.com",
    "mystore.com",
    "evil.com/x.myshopify.com",
    "../../etc",
    "galaxiq.myshopify.com.attacker.com",
    "-leading-hyphen.myshopify.com",
    "has space.myshopify.com",
    "galaxiq.myshopify.com\n",
    "GALAXIQ.MYSHOPIFY.COM.evil.com",
])
def test_invalid_shop_domains_rejected(shop):
    assert is_valid_shop_domain(shop) is False


@pytest.mark.parametrize("raw,expected", [
    ("galaxiq-braexaal", "galaxiq-braexaal.myshopify.com"),
    ("galaxiq-braexaal.myshopify.com", "galaxiq-braexaal.myshopify.com"),
    ("https://galaxiq-braexaal.myshopify.com/admin", "galaxiq-braexaal.myshopify.com"),
    ("  GALAXIQ-BRAEXAAL.MyShopify.com  ", "galaxiq-braexaal.myshopify.com"),
    ("evil.com", None),
    ("", None),
    (None, None),
])
def test_normalise_shop_domain(raw, expected):
    assert normalise_shop_domain(raw) == expected


def test_authorize_url_shape():
    url = build_authorize_url("galaxiq-braexaal.myshopify.com", "nonce123")
    parsed = urlparse(url)
    assert parsed.scheme == "https"
    assert parsed.netloc == "galaxiq-braexaal.myshopify.com"
    assert parsed.path == "/admin/oauth/authorize"

    q = parse_qs(parsed.query)
    assert q["scope"] == [SHOPIFY_SCOPES]
    assert q["state"] == ["nonce123"]
    assert q["redirect_uri"] == [redirect_uri()]
    assert q["client_id"] == ["test-client-id"]


def test_authorize_url_omits_grant_options():
    # Presence of grant_options[] yields an ONLINE token that expires with the
    # merchant's admin session, breaking every scheduled sync.
    url = build_authorize_url("galaxiq-braexaal.myshopify.com", "nonce123")
    assert "grant_options" not in url


def test_constants():
    assert SHOPIFY_API_VERSION == "2026-07"
    assert SHOPIFY_SCOPES == "read_products,read_inventory"


def test_redirect_uri_matches_registered_path():
    assert redirect_uri().endswith("/api/shopify/callback")
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_oauth.py -v`
Expected: FAIL with `ModuleNotFoundError: No module named 'app.services.shopify_oauth'`

- [ ] **Step 3: Write the implementation**

Create `app/services/shopify_oauth.py`:

```python
"""Pure functions for the Shopify OAuth authorization code grant.

No database access and no global state, so every security control here is
unit-testable without infrastructure.
"""
import logging
import re
from urllib.parse import urlencode

from app.core.config import settings

logger = logging.getLogger(__name__)

# Pinned in code rather than settings so config drift cannot silently change
# API behaviour. Must match the app version in the Shopify Dev Dashboard.
SHOPIFY_API_VERSION = "2026-07"
SHOPIFY_SCOPES = "read_products,read_inventory"

# Anchored at both ends. An unanchored pattern would accept
# "evil.com/x.myshopify.com" and turn this service into an open redirect.
SHOP_RE = re.compile(r"\A[a-zA-Z0-9][a-zA-Z0-9-]*\.myshopify\.com\Z")


def is_valid_shop_domain(shop) -> bool:
    """True only for a well-formed <handle>.myshopify.com domain."""
    if not shop or not isinstance(shop, str):
        return False
    return SHOP_RE.match(shop) is not None


def normalise_shop_domain(raw):
    """Turn what a merchant typed into a valid shop domain, or None.

    Merchants routinely enter their custom domain or a full admin URL. A bare
    handle gets the suffix appended; anything that still fails validation
    returns None rather than a guess.
    """
    if not raw or not isinstance(raw, str):
        return None
    d = raw.strip().lower()
    d = re.sub(r"^https?://", "", d)
    d = d.split("/")[0]
    if not d:
        return None
    if not d.endswith(".myshopify.com"):
        d = f"{d}.myshopify.com"
    return d if is_valid_shop_domain(d) else None


def redirect_uri() -> str:
    """The callback URL registered with Shopify. Must match exactly."""
    return f"{settings.PUBLIC_BASE_URL.rstrip('/')}/api/shopify/callback"


def build_authorize_url(shop: str, state: str) -> str:
    """Authorize URL for the offline-token authorization code grant.

    grant_options[] is deliberately omitted: including it as "per-user" yields
    an online token that dies with the merchant's admin session.
    """
    params = {
        "client_id": settings.SHOPIFY_CLIENT_ID,
        "scope": SHOPIFY_SCOPES,
        "redirect_uri": redirect_uri(),
        "state": state,
    }
    return f"https://{shop}/admin/oauth/authorize?{urlencode(params)}"
```

- [ ] **Step 4: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_oauth.py -v`
Expected: PASS, 27 tests

- [ ] **Step 5: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 3: HMAC verification

**This is the highest-risk task in the plan.** Shopify's message construction is subtly different from full percent-encoding, and getting it wrong produces a digest that looks plausible and never matches. Symptom: every real callback returns 403 while all your hand-written tests pass.

The rule: drop `hmac` and `signature`, then within each remaining key and value replace `%` → `%25`, `&` → `%26`, `=` → `%3D` — **`%` first**, or the other replacements get double-escaped. Sort by key, join `key=value` with `&`.

**Files:**
- Modify: `app/services/shopify_oauth.py`
- Modify: `tests/unit/test_shopify_oauth.py`

**Interfaces:**
- Consumes: `settings.SHOPIFY_CLIENT_SECRET`.
- Produces:
  - `build_hmac_message(params: dict) -> str`
  - `verify_hmac(params: dict) -> bool`

- [ ] **Step 1: Write the failing test**

Append to `tests/unit/test_shopify_oauth.py`:

```python
import hashlib
import hmac as hmac_mod

from app.services.shopify_oauth import build_hmac_message, verify_hmac

SECRET = "test-client-secret"


def _sign(params):
    """Produce the digest Shopify would send for these params."""
    msg = build_hmac_message(params)
    return hmac_mod.new(SECRET.encode(), msg.encode(), hashlib.sha256).hexdigest()


def test_message_excludes_hmac_and_signature_and_sorts():
    msg = build_hmac_message({
        "shop": "s.myshopify.com",
        "code": "abc",
        "hmac": "deadbeef",
        "signature": "ignored",
        "timestamp": "1700000000",
    })
    assert msg == "code=abc&shop=s.myshopify.com&timestamp=1700000000"


def test_message_escapes_percent_first_then_amp_and_equals():
    # % must be escaped before & and =, or "%" becomes "%2526".
    msg = build_hmac_message({"a": "100%", "b": "x&y", "c": "k=v"})
    assert msg == "a=100%25&b=x%26y&c=k%3Dv"


def test_valid_hmac_accepted():
    params = {
        "code": "0907a61c0c8d55e99db179b68161bc00",
        "shop": "galaxiq-braexaal.myshopify.com",
        "state": "nonce123",
        "timestamp": "1786337376",
    }
    params["hmac"] = _sign(params)
    assert verify_hmac(params) is True


def test_tampered_parameter_rejected():
    params = {
        "code": "0907a61c0c8d55e99db179b68161bc00",
        "shop": "galaxiq-braexaal.myshopify.com",
        "state": "nonce123",
        "timestamp": "1786337376",
    }
    params["hmac"] = _sign(params)
    params["shop"] = "attacker.myshopify.com"   # one field changed
    assert verify_hmac(params) is False


def test_single_character_tamper_rejected():
    params = {"code": "abc", "shop": "s.myshopify.com", "timestamp": "1"}
    params["hmac"] = _sign(params)
    params["code"] = "abd"
    assert verify_hmac(params) is False


def test_missing_hmac_rejected():
    assert verify_hmac({"shop": "s.myshopify.com"}) is False


def test_short_hmac_returns_false_and_does_not_raise():
    # compare_digest raises on differing lengths; that must not 500 the route.
    assert verify_hmac({"shop": "s.myshopify.com", "hmac": "ab"}) is False


def test_non_hex_hmac_returns_false():
    assert verify_hmac({"shop": "s.myshopify.com", "hmac": "zz" * 32}) is False
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_oauth.py -k hmac -v`
Expected: FAIL with `ImportError: cannot import name 'build_hmac_message'`

- [ ] **Step 3: Write the implementation**

Append to `app/services/shopify_oauth.py`:

```python
import hashlib
import hmac

_HMAC_EXCLUDED = ("hmac", "signature")


def _escape(value: str) -> str:
    """Shopify's HMAC escaping. Order matters: % before & and =."""
    return (
        str(value)
        .replace("%", "%25")
        .replace("&", "%26")
        .replace("=", "%3D")
    )


def build_hmac_message(params: dict) -> str:
    """Canonical message Shopify signs: sorted key=value pairs joined by &."""
    pairs = [
        f"{_escape(k)}={_escape(v)}"
        for k, v in sorted(params.items())
        if k not in _HMAC_EXCLUDED
    ]
    return "&".join(pairs)


def verify_hmac(params: dict) -> bool:
    """Timing-safe verification of Shopify's signature over the query params."""
    received = params.get("hmac")
    if not received:
        return False

    digest = hmac.new(
        settings.SHOPIFY_CLIENT_SECRET.encode("utf-8"),
        build_hmac_message(params).encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()

    try:
        # compare_digest raises TypeError/ValueError on length or type mismatch.
        return hmac.compare_digest(digest, received)
    except Exception:
        return False
```

- [ ] **Step 4: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_oauth.py -v`
Expected: PASS, all tests

- [ ] **Step 5: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 4: Token exchange and shop verification

Exchanges the authorization code for a token, then immediately proves the token works by querying the shop. A stored token that has never made a successful call is worse than no token — the failure surfaces during a sync instead of during the connect flow the merchant is watching.

**Files:**
- Modify: `app/services/shopify_oauth.py`
- Modify: `tests/unit/test_shopify_oauth.py`

**Interfaces:**
- Consumes: `SHOPIFY_API_VERSION`, `settings.SHOPIFY_CLIENT_ID`, `settings.SHOPIFY_CLIENT_SECRET`.
- Produces:
  - `async exchange_code(shop: str, code: str, *, client=None) -> dict` returning `{"access_token": str, "scope": str}`; raises `TokenExchangeError`.
  - `async fetch_shop_info(shop: str, token: str, *, client=None) -> dict` returning `{"name": str, "currency_code": str, "primary_domain": str}`; raises `ShopQueryError`.
  - `class TokenExchangeError(Exception)`, `class ShopQueryError(Exception)`.

- [ ] **Step 1: Write the failing test**

Append to `tests/unit/test_shopify_oauth.py`:

```python
import httpx
import pytest

from app.services.shopify_oauth import (
    ShopQueryError, TokenExchangeError, exchange_code, fetch_shop_info,
)


def _client(handler):
    return httpx.AsyncClient(transport=httpx.MockTransport(handler))


@pytest.mark.asyncio
async def test_exchange_code_returns_token_and_scope():
    seen = {}

    def handler(request):
        seen["url"] = str(request.url)
        seen["body"] = json.loads(request.content)
        return httpx.Response(200, json={
            "access_token": "shpat_realtoken", "scope": "read_products,read_inventory",
        })

    async with _client(handler) as c:
        out = await exchange_code("s.myshopify.com", "the-code", client=c)

    assert out == {"access_token": "shpat_realtoken",
                   "scope": "read_products,read_inventory"}
    assert seen["url"] == "https://s.myshopify.com/admin/oauth/access_token"
    assert seen["body"]["code"] == "the-code"
    assert seen["body"]["client_id"] == "test-client-id"
    assert seen["body"]["client_secret"] == "test-client-secret"


@pytest.mark.asyncio
async def test_exchange_code_raises_on_non_200():
    def handler(request):
        return httpx.Response(400, text="Oauth error invalid_request")

    async with _client(handler) as c:
        with pytest.raises(TokenExchangeError):
            await exchange_code("s.myshopify.com", "bad-code", client=c)


@pytest.mark.asyncio
async def test_exchange_code_raises_when_token_missing():
    def handler(request):
        return httpx.Response(200, json={"scope": "read_products"})

    async with _client(handler) as c:
        with pytest.raises(TokenExchangeError):
            await exchange_code("s.myshopify.com", "c", client=c)


@pytest.mark.asyncio
async def test_fetch_shop_info_parses_graphql():
    seen = {}

    def handler(request):
        seen["url"] = str(request.url)
        seen["token"] = request.headers.get("X-Shopify-Access-Token")
        return httpx.Response(200, json={"data": {"shop": {
            "name": "Galaxiq dev",
            "currencyCode": "USD",
            "primaryDomain": {"url": "https://galaxiq-braexaal.myshopify.com"},
        }}})

    async with _client(handler) as c:
        out = await fetch_shop_info("s.myshopify.com", "shpat_x", client=c)

    assert out == {"name": "Galaxiq dev", "currency_code": "USD",
                   "primary_domain": "https://galaxiq-braexaal.myshopify.com"}
    assert seen["url"] == "https://s.myshopify.com/admin/api/2026-07/graphql.json"
    assert seen["token"] == "shpat_x"


@pytest.mark.asyncio
async def test_fetch_shop_info_raises_on_401():
    def handler(request):
        return httpx.Response(401, text="Invalid API key or access token")

    async with _client(handler) as c:
        with pytest.raises(ShopQueryError):
            await fetch_shop_info("s.myshopify.com", "bad", client=c)


@pytest.mark.asyncio
async def test_fetch_shop_info_raises_on_graphql_errors():
    def handler(request):
        return httpx.Response(200, json={"errors": [{"message": "Access denied"}]})

    async with _client(handler) as c:
        with pytest.raises(ShopQueryError):
            await fetch_shop_info("s.myshopify.com", "t", client=c)
```

Add `import json` to the top of the test file if not already present.

- [ ] **Step 2: Confirm async test support**

Run: `.venv/bin/python -c "import pytest_asyncio; print(pytest_asyncio.__version__)"`

If it fails, add `pytest-asyncio` to `requirements.txt`, install it with `.venv/bin/pip install pytest-asyncio`, and create `pytest.ini` at the repo root:

```ini
[pytest]
asyncio_mode = auto
```

With `asyncio_mode = auto` the `@pytest.mark.asyncio` decorators are harmless but unnecessary; leave them for explicitness.

- [ ] **Step 3: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_oauth.py -k "exchange or shop_info" -v`
Expected: FAIL with `ImportError: cannot import name 'exchange_code'`

- [ ] **Step 4: Write the implementation**

Append to `app/services/shopify_oauth.py`:

```python
import contextlib

import httpx

_SHOP_QUERY = "{ shop { name currencyCode primaryDomain { url } } }"


class TokenExchangeError(Exception):
    """Shopify refused to exchange the authorization code for a token."""


class ShopQueryError(Exception):
    """The new token could not perform a real Admin API call."""


@contextlib.asynccontextmanager
async def _http(client):
    """Use the caller's client when injected (tests), else own one."""
    if client is not None:
        yield client
    else:
        async with httpx.AsyncClient(timeout=20.0) as owned:
            yield owned


async def exchange_code(shop: str, code: str, *, client=None) -> dict:
    """Trade the authorization code for a permanent offline access token."""
    async with _http(client) as c:
        resp = await c.post(
            f"https://{shop}/admin/oauth/access_token",
            headers={"Content-Type": "application/json"},
            json={
                "client_id": settings.SHOPIFY_CLIENT_ID,
                "client_secret": settings.SHOPIFY_CLIENT_SECRET,
                "code": code,
            },
        )

    if resp.status_code != 200:
        # Never log the body: it can echo the code, and the code is a credential.
        logger.error("Shopify token exchange failed for %s: HTTP %s",
                     shop, resp.status_code)
        raise TokenExchangeError(f"HTTP {resp.status_code}")

    data = resp.json()
    token = data.get("access_token")
    if not token:
        logger.error("Shopify token exchange for %s returned no access_token", shop)
        raise TokenExchangeError("no access_token in response")

    return {"access_token": token, "scope": data.get("scope", "")}


async def fetch_shop_info(shop: str, token: str, *, client=None) -> dict:
    """Prove the token works, and collect what the catalog sync will need.

    Currency and primary domain are required later for price normalisation and
    product URL construction, and are free to fetch at install time.
    """
    async with _http(client) as c:
        resp = await c.post(
            f"https://{shop}/admin/api/{SHOPIFY_API_VERSION}/graphql.json",
            headers={
                "X-Shopify-Access-Token": token,
                "Content-Type": "application/json",
            },
            json={"query": _SHOP_QUERY},
        )

    if resp.status_code != 200:
        logger.error("Shop query failed for %s: HTTP %s", shop, resp.status_code)
        raise ShopQueryError(f"HTTP {resp.status_code}")

    body = resp.json()
    if body.get("errors"):
        logger.error("Shop query returned GraphQL errors for %s: %s",
                     shop, body["errors"])
        raise ShopQueryError(str(body["errors"]))

    shop_node = (body.get("data") or {}).get("shop") or {}
    if not shop_node:
        raise ShopQueryError("empty shop node")

    return {
        "name": shop_node.get("name"),
        "currency_code": shop_node.get("currencyCode"),
        "primary_domain": (shop_node.get("primaryDomain") or {}).get("url"),
    }
```

- [ ] **Step 5: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_oauth.py -v`
Expected: PASS, all tests

- [ ] **Step 6: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 5: Product source storage

One table for every ingestion source, so the later HTTP-API connector needs no migration. Credentials are an encrypted JSON object rather than a bare string, which is what makes the envelope scheme-agnostic.

**Files:**
- Create: `app/services/product_sources.py`
- Test: `tests/unit/test_product_sources.py`

**Interfaces:**
- Consumes: `settings.SOURCE_CREDENTIALS_KEY`, `database.get_master_db_connection`.
- Produces:
  - `encrypt_credentials(creds: dict) -> str`
  - `decrypt_credentials(blob: str) -> dict`
  - `ensure_table() -> None`
  - `upsert_source(tenant_id, kind, external_ref, config: dict, credentials: dict) -> None`
  - `get_sources(tenant_id: str) -> list[dict]` — never includes credentials
  - `get_credentials(kind: str, external_ref: str) -> dict | None`

- [ ] **Step 1: Write the failing test**

Create `tests/unit/test_product_sources.py`:

```python
import pytest

from app.services.product_sources import decrypt_credentials, encrypt_credentials


def test_encrypt_round_trips():
    creds = {"access_token": "shpat_secret_value"}
    blob = encrypt_credentials(creds)
    assert decrypt_credentials(blob) == creds


def test_ciphertext_does_not_contain_plaintext():
    blob = encrypt_credentials({"access_token": "shpat_secret_value"})
    assert "shpat_secret_value" not in blob
    assert "access_token" not in blob


def test_encryption_is_non_deterministic():
    # Fernet embeds a random IV; identical input must not produce identical
    # ciphertext, or stored tokens become comparable across tenants.
    creds = {"access_token": "same"}
    assert encrypt_credentials(creds) != encrypt_credentials(creds)


def test_scheme_agnostic_envelope():
    creds = {"api_key": "k", "client_id": "c", "client_secret": "s"}
    assert decrypt_credentials(encrypt_credentials(creds)) == creds


def test_tampered_ciphertext_rejected():
    from cryptography.fernet import InvalidToken
    blob = encrypt_credentials({"access_token": "x"})
    tampered = blob[:-4] + ("AAAA" if not blob.endswith("AAAA") else "BBBB")
    with pytest.raises(InvalidToken):
        decrypt_credentials(tampered)
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_product_sources.py -v`
Expected: FAIL with `ModuleNotFoundError: No module named 'app.services.product_sources'`

- [ ] **Step 3: Write the implementation**

Create `app/services/product_sources.py`:

```python
"""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.
"""
import json
import logging

from cryptography.fernet import Fernet
from psycopg2.extras import Json, RealDictCursor

from app.core.config import settings
from app.services.database import get_master_db_connection

logger = logging.getLogger(__name__)

KIND_SHOPIFY = "shopify"
KIND_HTTP_API = "http_api"


def _fernet() -> Fernet:
    return Fernet(settings.SOURCE_CREDENTIALS_KEY.encode())


def encrypt_credentials(creds: dict) -> str:
    """Encrypt a credentials object. A dict, not a string, so the same envelope
    holds a Shopify access token or an API key plus secret."""
    return _fernet().encrypt(json.dumps(creds, sort_keys=True).encode()).decode()


def decrypt_credentials(blob: str) -> dict:
    """Raises cryptography.fernet.InvalidToken if tampered or wrongly keyed."""
    return json.loads(_fernet().decrypt(blob.encode()).decode())


def ensure_table() -> None:
    conn = get_master_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("""
                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)
                )
            """)
            cur.execute(
                "CREATE INDEX IF NOT EXISTS product_sources_tenant_idx "
                "ON product_sources (tenant_id)"
            )
        conn.commit()
    finally:
        conn.close()


def upsert_source(tenant_id: str, kind: str, external_ref: str,
                  config: dict, credentials: dict) -> None:
    """Insert or replace a source. Reconnecting updates the existing row, so a
    reinstall can never create a duplicate tenant row."""
    ensure_table()
    conn = get_master_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("""
                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
            """, (tenant_id, kind, external_ref, Json(config),
                  encrypt_credentials(credentials)))
        conn.commit()
        logger.info("Stored %s source %s for tenant %s", kind, external_ref, tenant_id)
    except Exception:
        conn.rollback()
        logger.error("Could not store %s source for tenant %s", kind, tenant_id,
                     exc_info=True)
        raise
    finally:
        conn.close()


def get_sources(tenant_id: str) -> list:
    """Every active source for a tenant. Credentials are never included."""
    ensure_table()
    conn = get_master_db_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute("""
                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
            """, (tenant_id,))
            return [dict(r) for r in cur.fetchall()]
    finally:
        conn.close()


def get_credentials(kind: str, external_ref: str):
    """Decrypted credentials for one source, or None. Callers must not log
    the return value."""
    ensure_table()
    conn = get_master_db_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT credentials_encrypted FROM product_sources "
                "WHERE kind = %s AND external_ref = %s AND status = 'active'",
                (kind, external_ref),
            )
            row = cur.fetchone()
            return decrypt_credentials(row[0]) if row else None
    finally:
        conn.close()
```

- [ ] **Step 4: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/test_product_sources.py -v`
Expected: PASS, 5 tests

These tests cover encryption only — they need no database. The DB functions are exercised in Task 10 against the live Postgres.

- [ ] **Step 5: Verify the table creates against the real master DB**

Run:

```bash
.venv/bin/python -c "
from app.services.product_sources import ensure_table
ensure_table(); print('product_sources ready')
"
```

Expected: `product_sources ready`. If it fails on connection, confirm `MASTER_DB_NAME=galaxiq_master` exists.

- [ ] **Step 6: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 6: Install endpoint

**Files:**
- Create: `app/api/shopify.py`
- Test: `tests/unit/test_shopify_routes.py`

**Interfaces:**
- Consumes: everything from Tasks 2–5, plus `redis_service.get_redis_client`.
- Produces: `router: APIRouter` with prefix `/api/shopify`; `STATE_PREFIX = "shopify:oauth:"`; `STATE_TTL_SECONDS = 600`.

- [ ] **Step 1: Write the failing test**

Create `tests/unit/test_shopify_routes.py`:

```python
import json

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.api import shopify as shopify_api


class FakeRedis:
    """Minimal stand-in for the async redis client used by the routes."""

    def __init__(self):
        self.store = {}

    async def setex(self, key, ttl, value):
        self.store[key] = value

    async def get(self, key):
        return self.store.get(key)

    async def delete(self, key):
        self.store.pop(key, None)

    async def __aenter__(self):
        return self

    async def __aexit__(self, *a):
        return False


@pytest.fixture
def fake_redis(monkeypatch):
    r = FakeRedis()
    monkeypatch.setattr(shopify_api, "get_redis_client", lambda: r)
    return r


@pytest.fixture
def client():
    app = FastAPI()
    app.include_router(shopify_api.router)
    return TestClient(app)


def test_install_redirects_to_shopify(client, fake_redis):
    resp = client.get("/api/shopify/install",
                      params={"shop": "galaxiq-braexaal.myshopify.com",
                              "tenant_id": "org_test"},
                      follow_redirects=False)

    assert resp.status_code == 307
    loc = resp.headers["location"]
    assert loc.startswith("https://galaxiq-braexaal.myshopify.com/admin/oauth/authorize")
    assert "grant_options" not in loc


def test_install_stores_state_bound_to_shop_and_tenant(client, fake_redis):
    client.get("/api/shopify/install",
               params={"shop": "galaxiq-braexaal.myshopify.com",
                       "tenant_id": "org_test"},
               follow_redirects=False)

    assert len(fake_redis.store) == 1
    key, raw = next(iter(fake_redis.store.items()))
    assert key.startswith("shopify:oauth:")
    payload = json.loads(raw)
    assert payload == {"shop": "galaxiq-braexaal.myshopify.com",
                       "tenant_id": "org_test"}


def test_install_accepts_bare_handle(client, fake_redis):
    resp = client.get("/api/shopify/install",
                      params={"shop": "galaxiq-braexaal", "tenant_id": "org_test"},
                      follow_redirects=False)
    assert resp.status_code == 307
    assert "galaxiq-braexaal.myshopify.com" in resp.headers["location"]


@pytest.mark.parametrize("shop", ["evil.com", "../../etc", "", "mystore.com"])
def test_install_rejects_bad_shop_domain(client, fake_redis, shop):
    resp = client.get("/api/shopify/install",
                      params={"shop": shop, "tenant_id": "org_test"},
                      follow_redirects=False)
    assert resp.status_code == 400
    assert fake_redis.store == {}


def test_install_requires_tenant_id(client, fake_redis):
    resp = client.get("/api/shopify/install",
                      params={"shop": "galaxiq-braexaal.myshopify.com"},
                      follow_redirects=False)
    assert resp.status_code == 422


def test_state_is_unique_per_call(client, fake_redis):
    for _ in range(2):
        client.get("/api/shopify/install",
                   params={"shop": "galaxiq-braexaal.myshopify.com",
                           "tenant_id": "org_test"},
                   follow_redirects=False)
    assert len(fake_redis.store) == 2
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_routes.py -v`
Expected: FAIL with `ModuleNotFoundError: No module named 'app.api.shopify'`

- [ ] **Step 3: Write the implementation**

Create `app/api/shopify.py`:

```python
"""Shopify OAuth endpoints.

Mounted at /api/shopify to match the redirect URL registered in the Shopify Dev
Dashboard (app version galaxiq-5). Changing this prefix breaks the callback.
"""
import json
import logging
import secrets

from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import RedirectResponse

from app.services.redis_service import get_redis_client
from app.services.shopify_oauth import build_authorize_url, normalise_shop_domain

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/shopify", tags=["shopify"])

STATE_PREFIX = "shopify:oauth:"
STATE_TTL_SECONDS = 600


@router.get("/install")
async def install(
    shop: str = Query(..., description="<handle>.myshopify.com"),
    tenant_id: str = Query(..., description="GalaxiQ tenant to bind the shop to"),
):
    """Start the authorization code grant.

    TODO(auth): tenant_id arrives from the query string because this codebase
    has no session. That lets anyone bind a shop to any tenant. It must come
    from an authenticated session before a single external merchant connects.
    """
    domain = normalise_shop_domain(shop)
    if not domain:
        # Rejecting here is what prevents this route becoming an open redirect.
        logger.warning("Rejected install for invalid shop domain: %r", shop)
        raise HTTPException(status_code=400, detail="Invalid shop domain")

    state = secrets.token_hex(24)
    payload = json.dumps({"shop": domain, "tenant_id": tenant_id})

    client = get_redis_client()
    async with client:
        await client.setex(f"{STATE_PREFIX}{state}", STATE_TTL_SECONDS, payload)

    logger.info("Starting Shopify install for %s (tenant %s)", domain, tenant_id)
    return RedirectResponse(build_authorize_url(domain, state))
```

- [ ] **Step 4: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_routes.py -v`
Expected: PASS, 9 tests

- [ ] **Step 5: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 7: Callback endpoint

Runs the four security controls in order, exchanges the code, verifies the token, persists, and redirects to the chat UI.

**Files:**
- Modify: `app/api/shopify.py`
- Modify: `tests/unit/test_shopify_routes.py`

**Interfaces:**
- Consumes: `verify_hmac`, `exchange_code`, `fetch_shop_info`, `TokenExchangeError`, `ShopQueryError`, `product_sources.upsert_source`, `product_sources.KIND_SHOPIFY`.
- Produces: `GET /api/shopify/callback`; `_error_redirect(reason: str, status: int) -> RedirectResponse`.

- [ ] **Step 1: Write the failing test**

Append to `tests/unit/test_shopify_routes.py`:

```python
from urllib.parse import parse_qs, urlparse

from app.services.shopify_oauth import ShopQueryError, TokenExchangeError


def _seed_state(fake_redis, state="s-nonce", shop="galaxiq-braexaal.myshopify.com",
                tenant="org_test"):
    fake_redis.store[f"shopify:oauth:{state}"] = json.dumps(
        {"shop": shop, "tenant_id": tenant})
    return state


@pytest.fixture
def happy_path(monkeypatch):
    """Stub the network and the database; the route logic is what is under test."""
    saved = {}

    async def fake_exchange(shop, code, **kw):
        return {"access_token": "shpat_realtoken",
                "scope": "read_products,read_inventory"}

    async def fake_shop_info(shop, token, **kw):
        return {"name": "Galaxiq dev", "currency_code": "USD",
                "primary_domain": "https://galaxiq-braexaal.myshopify.com"}

    def fake_upsert(tenant_id, kind, external_ref, config, credentials):
        saved.update(tenant_id=tenant_id, kind=kind, external_ref=external_ref,
                     config=config, credentials=credentials)

    monkeypatch.setattr(shopify_api, "exchange_code", fake_exchange)
    monkeypatch.setattr(shopify_api, "fetch_shop_info", fake_shop_info)
    monkeypatch.setattr(shopify_api, "upsert_source", fake_upsert)
    monkeypatch.setattr(shopify_api, "verify_hmac", lambda params: True)
    return saved


def _callback(client, **params):
    return client.get("/api/shopify/callback", params=params, follow_redirects=False)


def test_callback_success_persists_and_redirects(client, fake_redis, happy_path):
    state = _seed_state(fake_redis)
    resp = _callback(client, shop="galaxiq-braexaal.myshopify.com",
                     code="the-code", state=state, hmac="x")

    assert resp.status_code == 307
    q = parse_qs(urlparse(resp.headers["location"]).query)
    assert q["connected"] == ["shopify"]
    assert q["shop"] == ["galaxiq-braexaal.myshopify.com"]

    assert happy_path["tenant_id"] == "org_test"
    assert happy_path["kind"] == "shopify"
    assert happy_path["external_ref"] == "galaxiq-braexaal.myshopify.com"
    assert happy_path["credentials"] == {"access_token": "shpat_realtoken"}
    assert happy_path["config"]["currency_code"] == "USD"
    assert happy_path["config"]["scopes"] == "read_products,read_inventory"
    assert happy_path["config"]["api_version"] == "2026-07"


def test_callback_consumes_state_once(client, fake_redis, happy_path):
    state = _seed_state(fake_redis)
    first = _callback(client, shop="galaxiq-braexaal.myshopify.com",
                      code="c", state=state, hmac="x")
    assert first.status_code == 307

    replay = _callback(client, shop="galaxiq-braexaal.myshopify.com",
                       code="c", state=state, hmac="x")
    assert replay.status_code == 403
    assert "invalid_state" in replay.headers["location"]


def test_callback_rejects_unknown_state(client, fake_redis, happy_path):
    resp = _callback(client, shop="galaxiq-braexaal.myshopify.com",
                     code="c", state="never-issued", hmac="x")
    assert resp.status_code == 403
    assert "invalid_state" in resp.headers["location"]


def test_callback_rejects_shop_mismatch(client, fake_redis, happy_path):
    state = _seed_state(fake_redis, shop="other-store.myshopify.com")
    resp = _callback(client, shop="galaxiq-braexaal.myshopify.com",
                     code="c", state=state, hmac="x")
    assert resp.status_code == 403
    assert "shop_mismatch" in resp.headers["location"]


def test_callback_rejects_bad_shop_domain(client, fake_redis, happy_path):
    resp = _callback(client, shop="evil.com", code="c", state="s", hmac="x")
    assert resp.status_code == 400
    assert "invalid_shop" in resp.headers["location"]


def test_callback_rejects_bad_hmac(client, fake_redis, happy_path, monkeypatch):
    monkeypatch.setattr(shopify_api, "verify_hmac", lambda params: False)
    state = _seed_state(fake_redis)
    resp = _callback(client, shop="galaxiq-braexaal.myshopify.com",
                     code="c", state=state, hmac="forged")
    assert resp.status_code == 403
    assert "hmac_failed" in resp.headers["location"]


def test_callback_state_consumed_before_hmac_check(client, fake_redis, happy_path,
                                                   monkeypatch):
    # A failed attempt must still burn the nonce, or an attacker can retry
    # signatures against a live state value.
    monkeypatch.setattr(shopify_api, "verify_hmac", lambda params: False)
    state = _seed_state(fake_redis)
    _callback(client, shop="galaxiq-braexaal.myshopify.com",
              code="c", state=state, hmac="forged")
    assert fake_redis.store == {}


def test_callback_token_exchange_failure(client, fake_redis, happy_path, monkeypatch):
    async def boom(shop, code, **kw):
        raise TokenExchangeError("HTTP 400")

    monkeypatch.setattr(shopify_api, "exchange_code", boom)
    state = _seed_state(fake_redis)
    resp = _callback(client, shop="galaxiq-braexaal.myshopify.com",
                     code="bad", state=state, hmac="x")
    assert resp.status_code == 502
    assert "token_exchange_failed" in resp.headers["location"]


def test_callback_persists_nothing_when_verification_fails(client, fake_redis,
                                                           happy_path, monkeypatch):
    async def boom(shop, token, **kw):
        raise ShopQueryError("HTTP 401")

    monkeypatch.setattr(shopify_api, "fetch_shop_info", boom)
    state = _seed_state(fake_redis)
    resp = _callback(client, shop="galaxiq-braexaal.myshopify.com",
                     code="c", state=state, hmac="x")

    assert resp.status_code == 502
    assert "token_verification_failed" in resp.headers["location"]
    assert happy_path == {}   # nothing written
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_routes.py -k callback -v`
Expected: FAIL — 404 on `/api/shopify/callback`

- [ ] **Step 3: Write the implementation**

Update the imports at the top of `app/api/shopify.py`:

```python
from app.core.config import settings
from app.services.product_sources import KIND_SHOPIFY, upsert_source
from app.services.redis_service import get_redis_client
from app.services.shopify_oauth import (
    SHOPIFY_API_VERSION, ShopQueryError, TokenExchangeError, build_authorize_url,
    exchange_code, fetch_shop_info, normalise_shop_domain, verify_hmac,
)
```

Append to `app/api/shopify.py`:

```python
def _error_redirect(reason: str, status: int) -> RedirectResponse:
    """Send the merchant back to the UI with a readable reason.

    Rendering an error page instead would leave the flow untestable from the
    browser, which is where it actually runs.
    """
    sep = "&" if "?" in settings.FRONTEND_URL else "?"
    return RedirectResponse(
        f"{settings.FRONTEND_URL}{sep}shopify_error={reason}",
        status_code=status,
    )


@router.get("/callback")
async def callback(
    shop: str = Query(None),
    code: str = Query(None),
    state: str = Query(None),
    hmac: str = Query(None),
    timestamp: str = Query(None),
    host: str = Query(None),
):
    # 1. Shop domain. Never trust it twice.
    domain = normalise_shop_domain(shop)
    if not domain:
        logger.warning("Callback with invalid shop domain: %r", shop)
        return _error_redirect("invalid_shop", 400)

    # 2. State nonce: read and delete before anything else, so a replay cannot
    #    succeed even inside the TTL and a failed attempt burns the nonce.
    client = get_redis_client()
    async with client:
        raw = await client.get(f"{STATE_PREFIX}{state}") if state else None
        if raw:
            await client.delete(f"{STATE_PREFIX}{state}")

    if not raw:
        logger.warning("Callback with unknown or expired state for %s", domain)
        return _error_redirect("invalid_state", 403)

    stored = json.loads(raw)
    if stored.get("shop") != domain:
        logger.warning("Callback shop mismatch: state had %s, callback had %s",
                       stored.get("shop"), domain)
        return _error_redirect("shop_mismatch", 403)

    # 3. HMAC over every query parameter Shopify sent.
    params = {k: v for k, v in {
        "shop": shop, "code": code, "state": state, "hmac": hmac,
        "timestamp": timestamp, "host": host,
    }.items() if v is not None}

    if not verify_hmac(params):
        logger.warning("Callback HMAC verification failed for %s", domain)
        return _error_redirect("hmac_failed", 403)

    # 4. Exchange the code for an offline token.
    try:
        token_data = await exchange_code(domain, code)
    except TokenExchangeError:
        return _error_redirect("token_exchange_failed", 502)

    # 5. Prove the token works before storing it, and collect what the catalog
    #    sync needs. An unverified token fails later, during a sync nobody is
    #    watching, instead of now.
    try:
        info = await fetch_shop_info(domain, token_data["access_token"])
    except ShopQueryError:
        return _error_redirect("token_verification_failed", 502)

    upsert_source(
        tenant_id=stored["tenant_id"],
        kind=KIND_SHOPIFY,
        external_ref=domain,
        config={
            "shop_domain": domain,
            "shop_name": info["name"],
            "scopes": token_data["scope"],
            "api_version": SHOPIFY_API_VERSION,
            "currency_code": info["currency_code"],
            "primary_domain": info["primary_domain"],
        },
        credentials={"access_token": token_data["access_token"]},
    )

    logger.info("Connected Shopify store %s to tenant %s", domain,
                stored["tenant_id"])
    sep = "&" if "?" in settings.FRONTEND_URL else "?"
    return RedirectResponse(
        f"{settings.FRONTEND_URL}{sep}connected=shopify&shop={domain}",
        status_code=307,
    )
```

- [ ] **Step 4: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_routes.py -v`
Expected: PASS, all tests

- [ ] **Step 5: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 8: Status endpoint and app wiring

**Files:**
- Modify: `app/api/shopify.py`
- Modify: `app/main.py`
- Modify: `tests/unit/test_shopify_routes.py`

**Interfaces:**
- Consumes: `product_sources.get_sources`.
- Produces: `GET /api/shopify/status?tenant_id=…` returning `{"sources": [...]}`; the router mounted on the real app.

- [ ] **Step 1: Write the failing test**

Append to `tests/unit/test_shopify_routes.py`:

```python
def test_status_lists_sources_without_credentials(client, monkeypatch):
    monkeypatch.setattr(shopify_api, "get_sources", lambda tenant_id: [{
        "kind": "shopify",
        "external_ref": "galaxiq-braexaal.myshopify.com",
        "config": {"currency_code": "USD", "scopes": "read_products,read_inventory"},
        "status": "active",
        "connected_at": None,
        "last_synced_at": None,
    }])

    resp = client.get("/api/shopify/status", params={"tenant_id": "org_test"})
    assert resp.status_code == 200

    body = resp.json()
    assert body["sources"][0]["external_ref"] == "galaxiq-braexaal.myshopify.com"
    assert "credentials" not in json.dumps(body)
    assert "access_token" not in json.dumps(body)


def test_status_empty_when_not_connected(client, monkeypatch):
    monkeypatch.setattr(shopify_api, "get_sources", lambda tenant_id: [])
    resp = client.get("/api/shopify/status", params={"tenant_id": "org_test"})
    assert resp.status_code == 200
    assert resp.json() == {"sources": []}


def test_status_requires_tenant_id(client):
    assert client.get("/api/shopify/status").status_code == 422
```

- [ ] **Step 2: Run test to verify it fails**

Run: `.venv/bin/python -m pytest tests/unit/test_shopify_routes.py -k status -v`
Expected: FAIL — 404 on `/api/shopify/status`

- [ ] **Step 3: Write the implementation**

Add to the imports in `app/api/shopify.py`:

```python
from app.services.product_sources import KIND_SHOPIFY, get_sources, upsert_source
```

Append to `app/api/shopify.py`:

```python
@router.get("/status")
async def status(tenant_id: str = Query(...)):
    """Connected sources for a tenant. Never returns credentials.

    Returns a list because a tenant may hold several sources once the HTTP API
    connector lands.
    """
    return {"sources": get_sources(tenant_id)}
```

- [ ] **Step 4: Wire the router into the app**

In `app/main.py`, after `from app.api.websocket import router as ws_router`:

```python
from app.api.shopify import router as shopify_router
```

And after `app.include_router(ws_router)`:

```python
app.include_router(shopify_router)
```

- [ ] **Step 5: Run test to verify it passes**

Run: `.venv/bin/python -m pytest tests/unit/ -v`
Expected: PASS — the whole unit suite, including pre-existing tests

- [ ] **Step 6: Verify the routes are live**

Restart uvicorn, then run:

```bash
curl -s http://localhost:8001/openapi.json \
  | .venv/bin/python -c "import sys,json; print('\n'.join(k for k in json.load(sys.stdin)['paths'] if 'shopify' in k))"
```

Expected:

```
/api/shopify/install
/api/shopify/callback
/api/shopify/status
```

- [ ] **Step 7: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 9: Connect UI in the test chat

**Files:**
- Modify: `tests/chat_ui.html`

**Interfaces:**
- Consumes: `GET /api/shopify/status`, `GET /api/shopify/install`.
- Produces: no code interface; verified by eye and in Task 10.

- [ ] **Step 1: Add the connect bar markup**

In `tests/chat_ui.html`, immediately after `</header>` and before `<div id="log"></div>`:

```html
<div id="shopifyBar" style="display:flex;gap:8px;align-items:center;flex-wrap:wrap;
     padding:10px 14px;border-bottom:1px solid #24262b;font-size:13px">
  <strong style="font-size:13px">Shopify</strong>
  <span id="shopifyStatus" style="color:var(--muted)">checking…</span>
  <span class="spacer" style="flex:1"></span>
  <input id="shopInput" placeholder="your-store.myshopify.com" autocomplete="off"
         style="width:260px;padding:6px 8px;border-radius:6px;border:1px solid #2a2d33;
                background:#16181c;color:inherit">
  <button id="shopConnect">Connect Shopify</button>
</div>
<div id="shopifyHelp" style="padding:0 14px 10px;font-size:11.5px;color:var(--muted)">
  Find it in Shopify admin → Settings → Domains — the <code>.myshopify.com</code> one,
  not your custom domain.
</div>
```

- [ ] **Step 2: Add the behaviour**

At the end of the `<script>` block in `tests/chat_ui.html`, before `</script>`:

```javascript
// ---- Shopify connect -------------------------------------------------------
const ERRORS = {
  invalid_shop: "That is not a valid .myshopify.com domain.",
  invalid_state: "The connect link expired. Please try again.",
  shop_mismatch: "The store did not match the one you started with.",
  hmac_failed: "Shopify's signature did not verify. Please try again.",
  token_exchange_failed: "Shopify refused the authorisation code.",
  token_verification_failed: "Got a token, but it could not call the Shopify API.",
};

function setShopifyStatus(text, colour) {
  const el = document.getElementById("shopifyStatus");
  el.textContent = text;
  el.style.color = colour || "var(--muted)";
}

async function refreshShopifyStatus() {
  try {
    const r = await fetch(`${API}/api/shopify/status?tenant_id=${encodeURIComponent(TENANT)}`);
    const { sources } = await r.json();
    const shop = (sources || []).find(s => s.kind === "shopify");
    if (shop) {
      const cur = shop.config?.currency_code ? ` · ${shop.config.currency_code}` : "";
      setShopifyStatus(`connected: ${shop.external_ref}${cur}`, "#4ade80");
      document.getElementById("shopInput").value = shop.external_ref;
    } else {
      setShopifyStatus("not connected");
    }
  } catch {
    setShopifyStatus("status unavailable", "#f87171");
  }
}

document.getElementById("shopConnect").onclick = () => {
  // Merchants routinely type their custom domain. Accept a bare handle and add
  // the suffix; the server validates again regardless.
  let d = document.getElementById("shopInput").value.trim().toLowerCase()
    .replace(/^https?:\/\//, "").replace(/\/.*$/, "");
  if (!d) { setShopifyStatus("enter your store domain first", "#f87171"); return; }
  if (!d.endsWith(".myshopify.com")) d += ".myshopify.com";
  window.location.href =
    `${API}/api/shopify/install?shop=${encodeURIComponent(d)}` +
    `&tenant_id=${encodeURIComponent(TENANT)}`;
};

(function handleShopifyReturn() {
  const q = new URLSearchParams(location.search);
  if (q.get("connected") === "shopify") {
    setShopifyStatus(`connected: ${q.get("shop") || ""}`, "#4ade80");
  } else if (q.get("shopify_error")) {
    const code = q.get("shopify_error");
    setShopifyStatus(ERRORS[code] || `error: ${code}`, "#f87171");
  }
  if (q.has("connected") || q.has("shopify_error")) {
    // Strip the params so a refresh does not re-show a stale result.
    history.replaceState({}, "", location.pathname);
  }
})();

refreshShopifyStatus();
```

- [ ] **Step 3: Verify by eye**

Open `http://localhost:5173/chat_ui.html`. Expected: a Shopify bar under the header reading `not connected`, with an input and a Connect button. The existing chat must still work — send a message and confirm a reply.

- [ ] **Step 4: Report changed files**

```bash
git status --short
```

Do not commit and do not stage.

---

### Task 10: Live end-to-end verification

The only step that proves `SHOPIFY_CLIENT_SECRET` is correct. It cannot be validated earlier: Shopify's token endpoint checks the authorization code before the secret, so a bogus-code request returns an identical error whether the secret is right or wrong.

**Files:** none — verification only.

**Prerequisites:** uvicorn running on `:8001`, static server on `:5173`, Redis running, Shopify app version `galaxiq-5` active with redirect URL `http://localhost:8001/api/shopify/callback`.

- [ ] **Step 1: Confirm the environment**

```bash
redis-cli ping
curl -s -o /dev/null -w "8001 %{http_code}\n" http://localhost:8001/health
curl -s -o /dev/null -w "5173 %{http_code}\n" http://localhost:5173/chat_ui.html
```

Expected: `PONG`, `8001 200`, `5173 200`.

- [ ] **Step 2: Connect end to end**

Open `http://localhost:5173/chat_ui.html`, enter `galaxiq-braexaal.myshopify.com`, click Connect Shopify, approve in Shopify.

Expected: return to the chat UI with a green `connected: galaxiq-braexaal.myshopify.com · USD`.

If it fails with `hmac_failed`, the message construction in Task 3 is wrong — capture the full callback query string from the browser address bar and use it as a fixture before changing any code.

- [ ] **Step 3: Confirm the stored row**

```bash
.venv/bin/python -c "
from app.services.product_sources import get_sources, get_credentials
s = get_sources('org_8c32bf3e-6a18-4739-9b1c-94c0cf11125f')
print(s)
c = get_credentials('shopify', 'galaxiq-braexaal.myshopify.com')
print('token prefix:', c['access_token'][:6])
"
```

Expected: one source with `currency_code` populated, and `token prefix: shpat_`.

A populated `currency_code` is the proof the token performed a real Admin API call, and therefore that the client secret is correct.

- [ ] **Step 4: Confirm the token is not stored in plaintext**

```bash
psql -h localhost -U postgres -d galaxiq_master -c \
  "SELECT left(credentials_encrypted, 20) FROM product_sources;"
```

Expected: a Fernet blob beginning `gAAAAA`, with no `shpat_` visible anywhere.

- [ ] **Step 5: Confirm the security controls reject bad input**

```bash
BASE=http://localhost:8001/api/shopify
echo -n "garbage shop:  "; curl -s -o /dev/null -w "%{http_code}\n" "$BASE/install?shop=evil.com&tenant_id=org_test"
echo -n "path traversal: "; curl -s -o /dev/null -w "%{http_code}\n" "$BASE/install?shop=../../etc&tenant_id=org_test"
echo -n "bad state:     "; curl -s -o /dev/null -w "%{http_code}\n" "$BASE/callback?shop=galaxiq-braexaal.myshopify.com&code=x&state=forged&hmac=y"
```

Expected: `400`, `400`, `403`.

- [ ] **Step 6: Confirm reconnect does not duplicate**

Repeat Step 2, then:

```bash
psql -h localhost -U postgres -d galaxiq_master -c \
  "SELECT count(*) FROM product_sources WHERE external_ref='galaxiq-braexaal.myshopify.com';"
```

Expected: `1`.

- [ ] **Step 7: Run the full unit suite**

Run: `.venv/bin/python -m pytest tests/unit/ -v`
Expected: all pass, no regressions in pre-existing tests.

- [ ] **Step 8: Report**

Summarise for the user: staged files, test counts, the live result, and confirmation that the client secret is validated. Remind them to commit, and that `# TODO(auth)` in `install()` must be closed before any external merchant connects.

---

## Follow-on work (not in this plan)

- **B:** product fetch, `strategist_products` reshape (pricing, stock, status, taxonomy), category resolution, deterministic normalisation, and the HTTP-API connector including its URL template requirement.
- **C:** taxonomy handling, LLM enrichment, quality scoring, golden fixtures.
- **Before any external merchant:** close `TODO(auth)`, choose a distribution method on a *separate production app*, and move off `localhost` to a public HTTPS domain.