fix e2e: update auth route mocks and config for Better Auth

The J8 test calls /api/auth/session which maps to /auth/session in Better Auth. Adding mock to ensure consistent behavior.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

.github/workflows/ci.yml

@@ -2,9 +2,9 @@ name: CI
 
 on:
   push:
-    branches: [main]
+    branches: [main, dev, uat]
   pull_request:
-    branches: [main]
+    branches: [main, dev, uat]
 
 concurrency:
   group: ci-${{ github.ref }}
@@ -17,6 +17,9 @@ permissions:
 env:
   REGISTRY: ghcr.io
   IMAGE_NAME: cartsnitch/cartsnitch
+  AUTH_IMAGE_NAME: cartsnitch/auth
+  RECEIPTWITNESS_IMAGE_NAME: cartsnitch/receiptwitness
+  API_IMAGE_NAME: cartsnitch/api
 
 jobs:
   lint:
@@ -45,11 +48,62 @@ jobs:
       - name: Run tests
         run: npx vitest run
 
+  audit:
+    runs-on: runners-cartsnitch
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+          cache: npm
+      - run: npm ci
+      - name: Check for vulnerabilities
+        run: npm audit --audit-level=high
+
+  e2e:
+    runs-on: runners-cartsnitch
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+          cache: npm
+      - run: npm ci
+      - run: npx playwright install --with-deps chromium
+      - run: npx playwright test
+
+  lighthouse:
+    runs-on: runners-cartsnitch
+    needs: [test]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+          cache: npm
+      - run: npm ci
+      - run: npm run build
+      - name: Install Chromium for Lighthouse
+        run: |
+          npm install -g playwright
+          npx playwright install --with-deps chromium
+      - name: Start preview server
+        run: |
+          npm run preview &
+          npx wait-on http://localhost:4173/ --timeout 30000
+      - name: Run Lighthouse CI
+        run: |
+          CHROME_PATH=$(find /home/runner/.cache/ms-playwright -name chrome -type f 2>/dev/null | head -1)
+          npm install -g @lhci/cli
+          CHROME_PATH="$CHROME_PATH" lhci autorun --chrome-flags="--headless=new --no-sandbox --disable-gpu --disable-dev-shm-usage"
+
   build-and-push:
     runs-on: runners-cartsnitch
-    needs: [lint, test]
+    if: github.event_name == 'push'
+    needs: [lint, test, e2e]
     outputs:
       calver_tag: ${{ steps.calver.outputs.version }}
+      sha_tag: sha-${{ github.sha }}
     steps:
       - uses: actions/checkout@v4
         with:
@@ -72,8 +126,15 @@ jobs:
           echo "version=$VERSION" >> "$GITHUB_OUTPUT"
           echo "CalVer tag: $VERSION"
 
+      - name: Log in to Docker Hub
+        if: github.event_name == 'push'
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
       - name: Log in to GHCR
-        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+        if: github.event_name == 'push'
         uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
@@ -86,7 +147,7 @@ jobs:
         with:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
           tags: |
-            type=sha,prefix=sha-
+            type=sha,prefix=sha-,format=long
             type=raw,value=${{ steps.calver.outputs.version }},enable=${{ github.ref == 'refs/heads/main' }}
             type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
 
@@ -94,7 +155,7 @@ jobs:
         uses: docker/build-push-action@v6
         with:
           context: .
-          push: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }}
+          push: ${{ github.event_name == 'push' }}
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}
           target: prod
@@ -107,10 +168,186 @@ jobs:
           git tag "v${{ steps.calver.outputs.version }}"
           git push origin "v${{ steps.calver.outputs.version }}"
 
+  build-and-push-auth:
+    runs-on: runners-cartsnitch
+    if: github.event_name == 'push'
+    needs: [lint, test, e2e]
+    outputs:
+      calver_tag: ${{ steps.calver.outputs.version }}
+      sha_tag: sha-${{ github.sha }}
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Generate CalVer tag
+        id: calver
+        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+        run: |
+          DATE_TAG=$(date -u +%Y.%m.%d)
+          EXISTING=$(git tag -l "v${DATE_TAG}*" | sort -V | tail -1)
+          if [ -z "$EXISTING" ]; then
+            VERSION="$DATE_TAG"
+          elif [ "$EXISTING" = "v${DATE_TAG}" ]; then
+            VERSION="${DATE_TAG}.2"
+          else
+            BUILD_NUM=$(echo "$EXISTING" | sed "s/v${DATE_TAG}\.//")
+            VERSION="${DATE_TAG}.$((BUILD_NUM + 1))"
+          fi
+          echo "version=$VERSION" >> "$GITHUB_OUTPUT"
+
+      - name: Log in to Docker Hub
+        if: github.event_name == 'push'
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+      - name: Log in to GHCR
+        if: github.event_name == 'push'
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata (auth)
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.AUTH_IMAGE_NAME }}
+          tags: |
+            type=sha,prefix=sha-,format=long
+            type=raw,value=${{ steps.calver.outputs.version }},enable=${{ github.ref == 'refs/heads/main' }}
+            type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
+
+      - name: Build and push auth Docker image
+        uses: docker/build-push-action@v6
+        with:
+          context: ./auth
+          file: ./auth/Dockerfile
+          push: ${{ github.event_name == 'push' }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+
+  build-and-push-receiptwitness:
+    runs-on: runners-cartsnitch
+    if: github.event_name == 'push'
+    needs: [lint, test]
+    outputs:
+      calver_tag: ${{ steps.calver.outputs.version }}
+      sha_tag: sha-${{ github.sha }}
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Generate CalVer tag
+        id: calver
+        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+        run: |
+          DATE_TAG=$(date -u +%Y.%m.%d)
+          EXISTING=$(git tag -l "v${DATE_TAG}*" | sort -V | tail -1)
+          if [ -z "$EXISTING" ]; then VERSION="$DATE_TAG"
+          elif [ "$EXISTING" = "v${DATE_TAG}" ]; then VERSION="${DATE_TAG}.2"
+          else BUILD_NUM=$(echo "$EXISTING" | sed "s/v${DATE_TAG}\.//"); VERSION="${DATE_TAG}.$((BUILD_NUM + 1))"; fi
+          echo "version=$VERSION" >> "$GITHUB_OUTPUT"
+
+      - name: Log in to Docker Hub
+        if: github.event_name == 'push'
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+      - name: Log in to GHCR
+        if: github.event_name == 'push'
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.RECEIPTWITNESS_IMAGE_NAME }}
+          tags: |
+            type=sha,prefix=sha-,format=long
+            type=raw,value=${{ steps.calver.outputs.version }},enable=${{ github.ref == 'refs/heads/main' }}
+            type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
+
+      - name: Build and push receiptwitness image
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ./receiptwitness/Dockerfile
+          push: ${{ github.event_name == 'push' }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+
+  build-and-push-api:
+    runs-on: runners-cartsnitch
+    if: github.event_name == 'push'
+    needs: [lint, test]
+    outputs:
+      calver_tag: ${{ steps.calver.outputs.version }}
+      sha_tag: sha-${{ github.sha }}
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Generate CalVer tag
+        id: calver
+        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+        run: |
+          DATE_TAG=$(date -u +%Y.%m.%d)
+          EXISTING=$(git tag -l "v${DATE_TAG}*" | sort -V | tail -1)
+          if [ -z "$EXISTING" ]; then VERSION="$DATE_TAG"
+          elif [ "$EXISTING" = "v${DATE_TAG}" ]; then VERSION="${DATE_TAG}.2"
+          else BUILD_NUM=$(echo "$EXISTING" | sed "s/v${DATE_TAG}\.//"); VERSION="${DATE_TAG}.$((BUILD_NUM + 1))"; fi
+          echo "version=$VERSION" >> "$GITHUB_OUTPUT"
+
+      - name: Log in to Docker Hub
+        if: github.event_name == 'push'
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+      - name: Log in to GHCR
+        if: github.event_name == 'push'
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata (API)
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.API_IMAGE_NAME }}
+          tags: |
+            type=sha,prefix=sha-,format=long
+            type=raw,value=${{ steps.calver.outputs.version }},enable=${{ github.ref == 'refs/heads/main' }}
+            type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
+
+      - name: Build and push API Docker image
+        uses: docker/build-push-action@v6
+        with:
+          context: ./api
+          file: ./api/Dockerfile
+          push: ${{ github.event_name == 'push' }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+
   deploy-dev:
     runs-on: runners-cartsnitch
-    needs: [build-and-push]
-    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+    needs: [build-and-push, build-and-push-auth, build-and-push-receiptwitness, build-and-push-api]
+    if: always() && !cancelled() && github.event_name == 'push' && (github.ref == 'refs/heads/dev' || github.ref == 'refs/heads/main')
     steps:
       - name: Generate GitHub App token
         id: app-token
@@ -118,6 +355,8 @@ jobs:
         with:
           app-id: ${{ secrets.CARTSNITCH_APP_ID }}
           private-key: ${{ secrets.CARTSNITCH_APP_PRIVATE_KEY }}
+          owner: ${{ github.repository_owner }}
+          repositories: infra
 
       - name: Checkout infra repo
         uses: actions/checkout@v4
@@ -125,40 +364,178 @@ jobs:
           repository: cartsnitch/infra
           token: ${{ steps.app-token.outputs.token }}
           ref: main
+          path: infra
 
       - name: Install kubectl
         uses: azure/setup-kubectl@v4
 
-      - name: Update dev overlay image tag
-        working-directory: apps/overlays/dev
+      - name: Install kustomize
+        uses: imranismail/setup-kustomize@v2
+
+      - name: Determine image tag for frontend
+        id: frontend_tag
         run: |
-          kustomize edit set image ghcr.io/cartsnitch/cartsnitch:${{ needs.build-and-push.outputs.calver_tag }}
+          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+            echo "tag=${{ needs.build-and-push.outputs.calver_tag }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=${{ needs.build-and-push.outputs.sha_tag }}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Update frontend image tag
+        if: needs.build-and-push.result == 'success'
+        run: |
+          cd infra/apps/overlays/dev
+          kustomize edit set image ghcr.io/cartsnitch/cartsnitch:${{ steps.frontend_tag.outputs.tag }}
+
+      - name: Determine image tag for auth
+        id: auth_tag
+        run: |
+          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+            echo "tag=${{ needs.build-and-push-auth.outputs.calver_tag }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=${{ needs.build-and-push-auth.outputs.sha_tag }}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Update auth image tag
+        if: needs.build-and-push-auth.result == 'success'
+        run: |
+          cd infra/apps/overlays/dev
+          kustomize edit set image ghcr.io/cartsnitch/auth:${{ steps.auth_tag.outputs.tag }}
+
+      - name: Determine image tag for receiptwitness
+        id: receiptwitness_tag
+        run: |
+          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+            echo "tag=${{ needs.build-and-push-receiptwitness.outputs.calver_tag }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=${{ needs.build-and-push-receiptwitness.outputs.sha_tag }}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Update receiptwitness image tag
+        if: needs.build-and-push-receiptwitness.result == 'success'
+        run: |
+          cd infra/apps/overlays/dev
+          kustomize edit set image ghcr.io/cartsnitch/receiptwitness:${{ steps.receiptwitness_tag.outputs.tag }}
+
+      - name: Determine image tag for api
+        id: api_tag
+        run: |
+          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+            echo "tag=${{ needs.build-and-push-api.outputs.calver_tag }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=${{ needs.build-and-push-api.outputs.sha_tag }}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Update api image tag
+        if: needs.build-and-push-api.result == 'success'
+        run: |
+          cd infra/apps/overlays/dev
+          kustomize edit set image ghcr.io/cartsnitch/api:${{ steps.api_tag.outputs.tag }}
 
       - name: Commit and push to infra
         run: |
-          cd apps/overlays/dev
+          cd infra
           git config user.name "cartsnitch-ci[bot]"
           git config user.email "cartsnitch-ci[bot]@users.noreply.github.com"
-          git add kustomization.yaml
-          git commit -m "ci(dev): update cartsnitch image to ${{ needs.build-and-push.outputs.calver_tag }}"
+          git add apps/overlays/dev/kustomization.yaml
+          git commit -m "ci(dev): update cartsnitch, auth, receiptwitness, and api images"
+          git pull --rebase origin main
           git push origin main
 
-  trigger-uat:
+  deploy-uat:
     runs-on: runners-cartsnitch
-    needs: [deploy-dev, build-and-push]
-    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+    needs: [build-and-push, build-and-push-auth, build-and-push-receiptwitness, build-and-push-api]
+    if: always() && !cancelled() && github.event_name == 'push' && (github.ref == 'refs/heads/uat' || github.ref == 'refs/heads/main')
     steps:
-      - name: Create UAT issue in Paperclip
+      - name: Generate GitHub App token
+        id: app-token
+        uses: actions/create-github-app-token@v1
+        with:
+          app-id: ${{ secrets.CARTSNITCH_APP_ID }}
+          private-key: ${{ secrets.CARTSNITCH_APP_PRIVATE_KEY }}
+          owner: ${{ github.repository_owner }}
+          repositories: infra
+
+      - name: Checkout infra repo
+        uses: actions/checkout@v4
+        with:
+          repository: cartsnitch/infra
+          token: ${{ steps.app-token.outputs.token }}
+          ref: main
+          path: infra
+
+      - name: Install kubectl
+        uses: azure/setup-kubectl@v4
+
+      - name: Install kustomize
+        uses: imranismail/setup-kustomize@v2
+
+      - name: Determine image tag for frontend
+        id: frontend_tag
         run: |
-          curl -s -X POST \
-            -H "Authorization: Bearer ${{ secrets.PAPERCLIP_API_KEY }}" \
-            -H "Content-Type: application/json" \
-            "${{ secrets.PAPERCLIP_API_URL }}/api/companies/${{ secrets.PAPERCLIP_COMPANY_ID }}/issues" \
-            --data-raw '{
-              "title": "UAT: cartsnitch ${{ needs.build-and-push.outputs.calver_tag }} deployed to dev",
-              "description": "## UAT Required\n\nService: cartsnitch (frontend)\nImage: ghcr.io/cartsnitch/cartsnitch:${{ needs.build-and-push.outputs.calver_tag }}\nCommit: ${{ github.sha }}\nWorkflow run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}\n\nPlease run full regression against cartsnitch.dev.farh.net",
-              "status": "todo",
-              "priority": "high",
-              "assigneeAgentId": "1fc33bd9-308c-4abf-a355-87d12b6b0064",
-              "projectId": "05f7827d-54df-4ff8-9b27-293ffba6e049"
-            }'
+          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+            echo "tag=${{ needs.build-and-push.outputs.calver_tag }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=${{ needs.build-and-push.outputs.sha_tag }}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Update frontend image tag
+        if: needs.build-and-push.result == 'success'
+        run: |
+          cd infra/apps/overlays/uat
+          kustomize edit set image ghcr.io/cartsnitch/cartsnitch:${{ steps.frontend_tag.outputs.tag }}
+
+      - name: Determine image tag for auth
+        id: auth_tag
+        run: |
+          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+            echo "tag=${{ needs.build-and-push-auth.outputs.calver_tag }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=${{ needs.build-and-push-auth.outputs.sha_tag }}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Update auth image tag
+        if: needs.build-and-push-auth.result == 'success'
+        run: |
+          cd infra/apps/overlays/uat
+          kustomize edit set image ghcr.io/cartsnitch/auth:${{ steps.auth_tag.outputs.tag }}
+
+      - name: Determine image tag for receiptwitness
+        id: receiptwitness_tag
+        run: |
+          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+            echo "tag=${{ needs.build-and-push-receiptwitness.outputs.calver_tag }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=${{ needs.build-and-push-receiptwitness.outputs.sha_tag }}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Update receiptwitness image tag
+        if: needs.build-and-push-receiptwitness.result == 'success'
+        run: |
+          cd infra/apps/overlays/uat
+          kustomize edit set image ghcr.io/cartsnitch/receiptwitness:${{ steps.receiptwitness_tag.outputs.tag }}
+
+      - name: Determine image tag for api
+        id: api_tag
+        run: |
+          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+            echo "tag=${{ needs.build-and-push-api.outputs.calver_tag }}" >> "$GITHUB_OUTPUT"
+          else
+            echo "tag=${{ needs.build-and-push-api.outputs.sha_tag }}" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Update api image tag
+        if: needs.build-and-push-api.result == 'success'
+        run: |
+          cd infra/apps/overlays/uat
+          kustomize edit set image ghcr.io/cartsnitch/api:${{ steps.api_tag.outputs.tag }}
+
+      - name: Commit and push to infra
+        run: |
+          cd infra
+          git config user.name "cartsnitch-ci[bot]"
+          git config user.email "cartsnitch-ci[bot]@users.noreply.github.com"
+          git add apps/overlays/uat/kustomization.yaml
+          git commit -m "ci(uat): update cartsnitch, auth, receiptwitness, and api images"
+          git pull --rebase origin main
+          git push origin main

.gitignore

@@ -11,6 +11,7 @@ node_modules
 dist
 dist-ssr
 *.local
+.env
 
 # Editor directories and files
 .vscode/*

CLAUDE.md

@@ -12,6 +12,7 @@ CartSnitch is a self-hosted grocery price intelligence platform. This repo (`car
 | Directory | Service | Purpose |
 |-----------|---------|---------|
 | `/` (root) | Frontend | React PWA, mobile-first (this directory) |
+| `auth/` | Auth | Better-Auth Node.js service (session management, email/password, OAuth) |
 | `api/` | API Gateway | Frontend-facing REST API |
 | `common/` | Common | Shared Python models, schemas, Alembic migrations |
 | `receiptwitness/` | ReceiptWitness | Purchase data ingestion via retailer scrapers |
@@ -166,9 +167,13 @@ frontend/
 
 All data comes from the CartSnitch API gateway (`cartsnitch/api`). Base URL configured via environment variable `VITE_API_URL`.
 
-- JWT auth: store access token in memory (not localStorage), refresh token in httpOnly cookie if possible, or secure storage.
+- **Authentication via Better-Auth** (`auth/` service). Sessions are managed via httpOnly cookies — no tokens in localStorage or memory.
+  - Auth service URL configured via `VITE_AUTH_URL` (default: `http://localhost:3001`)
+  - Frontend uses `better-auth/react` client for sign-in, sign-up, sign-out, and `useSession()` hook
+  - API gateway validates sessions by querying the shared `sessions` table in Postgres
+  - Both cookie-based and Bearer token auth are supported (cookies for web, Bearer for API clients)
 - TanStack Query handles caching, background refetching, and optimistic updates.
-- API client should handle 401 responses by attempting token refresh before retrying.
+- API client sends `credentials: 'include'` on all requests to forward session cookies.
 
 ## Development Workflow
 

README.md

@@ -1,45 +1 @@
-# CartSnitch Monorepo
-
-CartSnitch is a self-hosted grocery price intelligence platform. This repo consolidates the core services and the flagship frontend PWA.
-
-## Services
-
-| Directory | Service | Purpose |
-|-----------|---------|---------|
-| `/` (root) | **Frontend** | React 18 PWA — mobile-first price intelligence UI |
-| `api/` | **API Gateway** | FastAPI — frontend-facing REST API |
-| `common/` | **Common** | Shared Python models, schemas, Alembic migrations |
-| `receiptwitness/` | **ReceiptWitness** | Purchase ingestion via retailer scrapers |
-
-## Quick Start
-
-### Frontend (root)
-
-```bash
-npm install
-npm run dev        # http://localhost:5173
-npm run build      # production build
-npm run test       # unit tests (Vitest)
-```
-
-### Python Services
-
-Each Python service uses [uv](https://github.com/astral-sh/uv) and has its own `pyproject.toml`:
-
-```bash
-cd api             # or common / receiptwitness
-uv sync
-uv run pytest
-```
-
-## Development Workflow
-
-- **Never push directly to main.** Always open a PR from a feature branch.
-- Branch naming: `feature/<description>` or `fix/<description>`
-- Conventional commits: `feat:`, `fix:`, `refactor:`, `docs:`, `chore:`
-
-## Architecture
-
-For full details see [CLAUDE.md](./CLAUDE.md) or the per-service `CLAUDE.md` in each subdirectory.
-
-CartSnitch is a polyrepo-style monorepo: each service can be built and deployed independently, but sharing code between `common/` and the other Python services is done via local path dependencies in `pyproject.toml`.
+# CartSnitch

api/Dockerfile

@@ -12,10 +12,14 @@ RUN pip install --no-cache-dir --prefix=/install .
 
 FROM python:3.12-slim AS prod
 
+RUN apt-get update && apt-get install -y --no-install-recommends libpq5 && rm -rf /var/lib/apt/lists/*
+
 WORKDIR /app
 RUN adduser --system --group --uid 1000 app
 COPY --from=build /install /usr/local
 COPY src/ ./src/
+COPY alembic.ini ./
+COPY alembic/ ./alembic/
 
 USER 1000
 EXPOSE 8000
@@ -23,4 +27,4 @@ EXPOSE 8000
 HEALTHCHECK --interval=30s --timeout=3s \
     CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"
 
-CMD ["uvicorn", "cartsnitch_api.main:app", "--host", "0.0.0.0", "--port", "8000"]
+CMD ["sh", "-c", "python -m alembic upgrade head && uvicorn cartsnitch_api.main:app --host 0.0.0.0 --port 8000"]

api/alembic/env.py

@@ -18,7 +18,7 @@ if not db_url:
         "CARTSNITCH_DATABASE_URL_SYNC must be set. "
         "Example: postgresql://user:pass@localhost:5432/cartsnitch"
     )
-config.set_main_option("sqlalchemy.url", db_url)
+config.set_main_option("sqlalchemy.url", db_url.replace("%", "%%"))
 
 target_metadata = Base.metadata
 
@@ -31,6 +31,7 @@ def run_migrations_offline() -> None:
         target_metadata=target_metadata,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
+        version_table_column_width=128,
     )
     with context.begin_transaction():
         context.run_migrations()
@@ -44,9 +45,20 @@ def run_migrations_online() -> None:
         poolclass=pool.NullPool,
     )
     with connectable.connect() as connection:
-        context.configure(connection=connection, target_metadata=target_metadata)
+        context.configure(connection=connection, target_metadata=target_metadata, version_table_column_width=128)
         with context.begin_transaction():
             context.run_migrations()
+        # Create any tables defined in models but not yet created by migrations.
+        # This bootstraps fresh databases that have no legacy schema.
+        # checkfirst=True ensures this is a no-op on existing databases.
+        try:
+            Base.metadata.create_all(bind=connection, checkfirst=True)
+            connection.commit()
+        except Exception as exc:
+            import logging
+            logging.getLogger("alembic.env").warning(
+                "create_all failed (non-fatal, migrations should handle table creation): %s", exc
+            )
 
 
 if context.is_offline_mode():

api/alembic/versions/001_encrypt_session_data.py

@@ -33,6 +33,21 @@ def _is_fernet_token(value: str) -> bool:
 
 
 def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    # Fresh DB — table created by Base.metadata.create_all with correct TEXT type
+    if not inspector.has_table("user_store_accounts"):
+        return
+
+    # Already migrated? Skip if session_data is already TEXT (not JSON)
+    cols = {c["name"]: c for c in inspector.get_columns("user_store_accounts")}
+    if "session_data" not in cols:
+        return
+    col_type = str(cols["session_data"]["type"]).lower()
+    if "text" in col_type and "json" not in col_type:
+        return  # already TEXT — nothing to do
+
     # Change column type from JSON to TEXT to hold Fernet ciphertext
     op.alter_column(
         "user_store_accounts",
@@ -43,7 +58,6 @@ def upgrade() -> None:
         postgresql_using="session_data::text",
     )
 
-    conn = op.get_bind()
     rows = conn.execute(
         text("SELECT id, session_data FROM user_store_accounts WHERE session_data IS NOT NULL")
     ).fetchall()

api/alembic/versions/002_better_auth_tables.py

@@ -0,0 +1,114 @@
+"""Add Better-Auth tables and extend users table.
+
+Creates sessions, accounts, and verifications tables for Better-Auth.
+Adds email_verified and image columns to existing users table.
+Migrates password hashes from users.hashed_password to accounts.password.
+
+Revision ID: 002_better_auth_tables
+Revises: 001_encrypt_session_data
+Create Date: 2026-03-28
+"""
+
+import sqlalchemy as sa
+from sqlalchemy import text
+
+from alembic import op
+
+revision = "002_better_auth_tables"
+down_revision = "001_encrypt_session_data"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    # --- Extend users table for Better-Auth compatibility ---
+    # Guard: on a fresh DB Base.metadata.create_all (called in env.py after migrations)
+    # creates the users table with all columns, so migration 002 must not re-run add_column.
+    if inspector.has_table("users"):
+        existing_user_cols = [c["name"] for c in inspector.get_columns("users")]
+        if "email_verified" not in existing_user_cols:
+            op.add_column("users", sa.Column("email_verified", sa.Boolean(), nullable=False, server_default="false"))
+        if "image" not in existing_user_cols:
+            op.add_column("users", sa.Column("image", sa.Text(), nullable=True))
+
+    # --- Create sessions table ---
+    if not inspector.has_table("sessions"):
+        op.create_table(
+            "sessions",
+            sa.Column("id", sa.Text(), nullable=False),
+            sa.Column("token", sa.Text(), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("ip_address", sa.Text(), nullable=True),
+            sa.Column("user_agent", sa.Text(), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+        )
+        op.create_index("ix_sessions_token", "sessions", ["token"], unique=True)
+        op.create_index("ix_sessions_user_id", "sessions", ["user_id"])
+
+    # --- Create accounts table ---
+    if not inspector.has_table("accounts"):
+        op.create_table(
+            "accounts",
+            sa.Column("id", sa.Text(), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("account_id", sa.Text(), nullable=False),
+            sa.Column("provider_id", sa.Text(), nullable=False),
+            sa.Column("access_token", sa.Text(), nullable=True),
+            sa.Column("refresh_token", sa.Text(), nullable=True),
+            sa.Column("access_token_expires_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("refresh_token_expires_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("scope", sa.Text(), nullable=True),
+            sa.Column("id_token", sa.Text(), nullable=True),
+            sa.Column("password", sa.Text(), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+        )
+        op.create_index("ix_accounts_user_id", "accounts", ["user_id"])
+
+    # --- Create verifications table ---
+    if not inspector.has_table("verifications"):
+        op.create_table(
+            "verifications",
+            sa.Column("id", sa.Text(), nullable=False),
+            sa.Column("identifier", sa.Text(), nullable=False),
+            sa.Column("value", sa.Text(), nullable=False),
+            sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+        )
+
+    # --- Migrate existing password hashes to accounts table ---
+    # Only run on existing (non-fresh) DBs that already have users table with data
+    if inspector.has_table("users"):
+        users = conn.execute(
+            text("SELECT id, hashed_password FROM users WHERE hashed_password IS NOT NULL")
+        ).fetchall()
+
+        for user_id, hashed_password in users:
+            user_id_str = str(user_id)
+            conn.execute(
+                text(
+                    "INSERT INTO accounts (id, user_id, account_id, provider_id, password, created_at, updated_at) "
+                    "VALUES (gen_random_uuid()::text, :user_id, :account_id, 'credential', :password, now(), now())"
+                ),
+                {"user_id": user_id_str, "account_id": user_id_str, "password": hashed_password},
+            )
+
+
+def downgrade() -> None:
+    op.execute(text("DROP INDEX IF EXISTS ix_accounts_user_id"))
+    op.execute(text("DROP TABLE IF EXISTS verifications"))
+    op.execute(text("DROP TABLE IF EXISTS accounts"))
+    op.execute(text("DROP INDEX IF EXISTS ix_sessions_user_id"))
+    op.execute(text("DROP INDEX IF EXISTS ix_sessions_token"))
+    op.execute(text("DROP TABLE IF EXISTS sessions"))
+    op.execute(text("ALTER TABLE users DROP COLUMN IF EXISTS image"))
+    op.execute(text("ALTER TABLE users DROP COLUMN IF EXISTS email_verified"))

api/alembic/versions/003_make_users_hashed_password_nullable.py

@@ -0,0 +1,43 @@
+"""Make users.hashed_password nullable.
+
+Better-Auth inserts users without hashed_password (passwords live in the
+accounts table). This column is now purely optional.
+
+Revision ID: 003_make_users_hashed_password_nullable
+Revises: 002_better_auth_tables
+Create Date: 2026-03-30
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "003_make_users_hashed_password_nullable"
+down_revision = "002_better_auth_tables"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    # Fresh DB — nothing to alter
+    if not inspector.has_table("users"):
+        return
+
+    cols = {c["name"]: c for c in inspector.get_columns("users")}
+    if "hashed_password" in cols and not cols["hashed_password"]["nullable"]:
+        op.alter_column("users", "hashed_password", existing_type=sa.String(255), nullable=True)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    if not inspector.has_table("users"):
+        return
+
+    cols = {c["name"]: c for c in inspector.get_columns("users")}
+    if "hashed_password" in cols and cols["hashed_password"]["nullable"]:
+        op.alter_column("users", "hashed_password", existing_type=sa.String(255), nullable=False)

api/alembic/versions/004_fix_user_id_text.py

@@ -0,0 +1,136 @@
+"""Fix users.id UUID->text type mismatch for Better-Auth compatibility.
+
+Better-Auth generates nanoid-style text IDs (e.g. pGud2ln2WAFHC0KYjBVKR4Rc7mM8OcTI),
+but the users table was using PostgreSQL uuid type. When Better-Auth tries to INSERT
+a new user, Postgres throws:
+  ERROR: invalid input syntax for type uuid: "pGud2ln2WAFHC0KYjBVKR4Rc7mM8OcTI"
+
+The sessions, accounts, and verifications tables already use text IDs — only users,
+user_store_accounts.user_id, and purchases.user_id needed fixing.
+
+Revision ID: 004_fix_user_id_text
+Revises: 003_make_users_hashed_password_nullable
+Create Date: 2026-03-31
+"""
+
+import sqlalchemy as sa
+from sqlalchemy import text
+
+from alembic import op
+
+revision = "004_fix_user_id_text"
+down_revision = "003_make_users_hashed_password_nullable"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    # Fresh DB — no tables yet, nothing to convert
+    if not inspector.has_table("users"):
+        return
+
+    # Check if already TEXT (Base.metadata.create_all uses TEXT for fresh DB)
+    users_cols = {c["name"]: c for c in inspector.get_columns("users")}
+    if "id" in users_cols:
+        id_type = str(users_cols["id"]["type"]).lower()
+        if "text" in id_type and "uuid" not in id_type:
+            return  # already TEXT — nothing to do
+
+    # Step 1: Drop existing FK constraints (ignore if they don't exist)
+    op.execute(text("ALTER TABLE user_store_accounts DROP CONSTRAINT IF EXISTS user_store_accounts_user_id_fkey"))
+    op.execute(text("ALTER TABLE purchases DROP CONSTRAINT IF EXISTS purchases_user_id_fkey"))
+
+    # Step 2: Alter users.id from uuid to text
+    op.alter_column(
+        "users",
+        "id",
+        type_=sa.Text(),
+        existing_type=sa.UUID(),
+        postgresql_using="id::text",
+    )
+
+    # Step 3: Alter user_store_accounts.user_id from uuid to text
+    op.alter_column(
+        "user_store_accounts",
+        "user_id",
+        type_=sa.Text(),
+        existing_type=sa.UUID(),
+        postgresql_using="user_id::text",
+    )
+
+    # Step 4: Alter purchases.user_id from uuid to text
+    op.alter_column(
+        "purchases",
+        "user_id",
+        type_=sa.Text(),
+        existing_type=sa.UUID(),
+        postgresql_using="user_id::text",
+    )
+
+    # Step 5: Re-add FK constraints
+    op.execute(
+        text(
+            "ALTER TABLE user_store_accounts "
+            "ADD CONSTRAINT user_store_accounts_user_id_fkey "
+            "FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE"
+        )
+    )
+    op.execute(
+        text(
+            "ALTER TABLE purchases "
+            "ADD CONSTRAINT purchases_user_id_fkey "
+            "FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE"
+        )
+    )
+
+
+def downgrade() -> None:
+    # Drop FK constraints
+    op.execute(text("ALTER TABLE user_store_accounts DROP CONSTRAINT IF EXISTS user_store_accounts_user_id_fkey"))
+    op.execute(text("ALTER TABLE purchases DROP CONSTRAINT IF EXISTS purchases_user_id_fkey"))
+
+    # Revert users.id from text to uuid
+    op.alter_column(
+        "users",
+        "id",
+        type_=sa.UUID(),
+        existing_type=sa.Text(),
+        postgresql_using="id::uuid",
+    )
+
+    # Revert user_store_accounts.user_id from text to uuid
+    op.alter_column(
+        "user_store_accounts",
+        "user_id",
+        type_=sa.UUID(),
+        existing_type=sa.Text(),
+        postgresql_using="user_id::uuid",
+    )
+
+    # Revert purchases.user_id from text to uuid
+    op.alter_column(
+        "purchases",
+        "user_id",
+        type_=sa.UUID(),
+        existing_type=sa.Text(),
+        postgresql_using="user_id::uuid",
+    )
+
+    # Re-add FK constraints (PostgreSQL will auto-name them)
+    op.execute(
+        text(
+            "ALTER TABLE user_store_accounts "
+            "ADD CONSTRAINT user_store_accounts_user_id_fkey "
+            "FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE"
+        )
+    )
+    op.execute(
+        text(
+            "ALTER TABLE purchases "
+            "ADD CONSTRAINT purchases_user_id_fkey "
+            "FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE"
+        )
+    )

api/alembic/versions/005_add_email_inbound_token.py

@@ -0,0 +1,57 @@
+"""Add email_inbound_token to users.
+
+Revision ID: 005_add_email_inbound_token
+Revises: 004_fix_user_id_text
+Create Date: 2026-04-02
+"""
+
+import secrets
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "005_add_email_inbound_token"
+down_revision = "004_fix_user_id_text"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    # Guard: on a fresh DB Base.metadata.create_all creates users table with the column already present
+    if not inspector.has_table("users"):
+        return
+    existing_cols = [c["name"] for c in inspector.get_columns("users")]
+    if "email_inbound_token" in existing_cols:
+        return
+
+    # Add column nullable first so existing rows can be backfilled
+    op.add_column(
+        "users",
+        sa.Column("email_inbound_token", sa.String(22), nullable=True),
+    )
+
+    # Backfill existing users with unique tokens
+    result = conn.execute(sa.text("SELECT id FROM users WHERE email_inbound_token IS NULL"))
+    for (user_id,) in result:
+        token = secrets.token_urlsafe(16)
+        conn.execute(
+            sa.text("UPDATE users SET email_inbound_token = :token WHERE id = :id"),
+            {"token": token, "id": user_id},
+        )
+
+    # Now enforce non-null and unique
+    op.alter_column("users", "email_inbound_token", nullable=False)
+    op.create_index(
+        "ix_users_email_inbound_token",
+        "users",
+        ["email_inbound_token"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_users_email_inbound_token", table_name="users")
+    op.drop_column("users", "email_inbound_token")

api/alembic/versions/006_email_inbound_token_server_default.py

@@ -0,0 +1,42 @@
+"""Add server_default to users.email_inbound_token.
+
+Revision ID: 006_email_inbound_token_server_default
+Revises: 005_add_email_inbound_token
+Create Date: 2026-04-04
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "006_email_inbound_token_server_default"
+down_revision = "005_add_email_inbound_token"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    # Guard: on a fresh DB Base.metadata.create_all already sets the server_default
+    if not inspector.has_table("users"):
+        return
+    cols = {c["name"]: c for c in inspector.get_columns("users")}
+    if "email_inbound_token" not in cols:
+        return
+    if cols["email_inbound_token"].get("default") is not None:
+        return
+    op.alter_column(
+        "users",
+        "email_inbound_token",
+        server_default=sa.text(
+            "replace(replace(trim(trailing '=' from encode(gen_random_bytes(16), 'base64')), '+', '-'), '/', '_')"
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.alter_column(
+        "users",
+        "email_inbound_token",
+        server_default=None,
+    )

api/alembic/versions/007_bootstrap_users_table.py

@@ -0,0 +1,47 @@
+"""Bootstrap users table on fresh databases.
+
+On fresh databases, migrations 001-006 skip users-table operations because
+the table does not exist yet. Base.metadata.create_all() in env.py is meant
+to handle this, but if it fails (import errors, etc.) the table is never
+created. This migration creates the users table with raw SQL as a safety net.
+
+Revision ID: 007_bootstrap_users_table
+Revises: 006_email_inbound_token_server_default
+Create Date: 2026-04-04
+"""
+
+import sqlalchemy as sa
+from sqlalchemy import text
+
+from alembic import op
+
+revision = "007_bootstrap_users_table"
+down_revision = "006_email_inbound_token_server_default"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if inspector.has_table("users"):
+        return  # Table already exists (non-fresh DB or create_all already ran)
+
+    conn.execute(text("""
+        CREATE TABLE users (
+            id TEXT PRIMARY KEY,
+            email VARCHAR(255) NOT NULL UNIQUE,
+            hashed_password VARCHAR(255),
+            display_name VARCHAR(100),
+            email_verified BOOLEAN NOT NULL DEFAULT false,
+            image TEXT,
+            email_inbound_token VARCHAR(22) NOT NULL UNIQUE
+                DEFAULT replace(replace(trim(trailing '=' from encode(gen_random_bytes(16), 'base64')), '+', '-'), '/', '_'),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )
+    """))
+
+
+def downgrade() -> None:
+    op.execute(text("DROP TABLE IF EXISTS users"))

api/alembic/versions/008_create_domain_tables.py

@@ -0,0 +1,210 @@
+"""Create domain tables (stores, purchases, coupons, etc.).
+
+Revision ID: 008_create_domain_tables
+Revises: 007_bootstrap_users_table
+Create Date: 2026-04-04
+"""
+
+import sqlalchemy as sa
+from sqlalchemy import text
+
+from alembic import op
+
+revision = "008_create_domain_tables"
+down_revision = "007_bootstrap_users_table"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    # 1. stores
+    if not inspector.has_table("stores"):
+        op.create_table(
+            "stores",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("name", sa.String(100), nullable=False),
+            sa.Column("slug", sa.String(20), nullable=False, unique=True),
+            sa.Column("logo_url", sa.String(500), nullable=True),
+            sa.Column("website_url", sa.String(500), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        )
+
+    # 2. store_locations
+    if not inspector.has_table("store_locations"):
+        op.create_table(
+            "store_locations",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("store_id", sa.Uuid(), sa.ForeignKey("stores.id"), nullable=False),
+            sa.Column("address", sa.String(300), nullable=False),
+            sa.Column("city", sa.String(100), nullable=False),
+            sa.Column("state", sa.String(2), nullable=False),
+            sa.Column("zip", sa.String(10), nullable=False),
+            sa.Column("lat", sa.Float(), nullable=True),
+            sa.Column("lng", sa.Float(), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        )
+
+    # 3. normalized_products
+    if not inspector.has_table("normalized_products"):
+        op.create_table(
+            "normalized_products",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("canonical_name", sa.String(300), nullable=False),
+            sa.Column("category", sa.String(50), nullable=True),
+            sa.Column("subcategory", sa.String(100), nullable=True),
+            sa.Column("brand", sa.String(200), nullable=True),
+            sa.Column("size", sa.String(50), nullable=True),
+            sa.Column("size_unit", sa.String(10), nullable=True),
+            sa.Column("upc_variants", sa.JSON(), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        )
+
+    # 4. purchases
+    if not inspector.has_table("purchases"):
+        op.create_table(
+            "purchases",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("user_id", sa.Text(), sa.ForeignKey("users.id"), nullable=False),
+            sa.Column("store_id", sa.Uuid(), sa.ForeignKey("stores.id"), nullable=False),
+            sa.Column("store_location_id", sa.Uuid(), sa.ForeignKey("store_locations.id"), nullable=True),
+            sa.Column("receipt_id", sa.String(200), nullable=False),
+            sa.Column("purchase_date", sa.Date(), nullable=False),
+            sa.Column("total", sa.Numeric(10, 2), nullable=False),
+            sa.Column("subtotal", sa.Numeric(10, 2), nullable=True),
+            sa.Column("tax", sa.Numeric(10, 2), nullable=True),
+            sa.Column("savings_total", sa.Numeric(10, 2), nullable=True),
+            sa.Column("source_url", sa.String(500), nullable=True),
+            sa.Column("raw_data", sa.JSON(), nullable=True),
+            sa.Column("ingested_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.UniqueConstraint("user_id", "store_id", "receipt_id", name="uq_purchase_receipt"),
+            sa.Index("ix_purchases_user_store", "user_id", "store_id"),
+        )
+
+    # 5. purchase_items
+    if not inspector.has_table("purchase_items"):
+        op.create_table(
+            "purchase_items",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("purchase_id", sa.Uuid(), sa.ForeignKey("purchases.id"), nullable=False),
+            sa.Column("product_name_raw", sa.String(300), nullable=False),
+            sa.Column("upc", sa.String(20), nullable=True),
+            sa.Column("quantity", sa.Numeric(10, 3), nullable=False),
+            sa.Column("unit_price", sa.Numeric(10, 2), nullable=False),
+            sa.Column("extended_price", sa.Numeric(10, 2), nullable=False),
+            sa.Column("regular_price", sa.Numeric(10, 2), nullable=True),
+            sa.Column("sale_price", sa.Numeric(10, 2), nullable=True),
+            sa.Column("coupon_discount", sa.Numeric(10, 2), nullable=True),
+            sa.Column("loyalty_discount", sa.Numeric(10, 2), nullable=True),
+            sa.Column("category_raw", sa.String(100), nullable=True),
+            sa.Column("normalized_product_id", sa.Uuid(), sa.ForeignKey("normalized_products.id"), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        )
+
+    # 6. coupons
+    if not inspector.has_table("coupons"):
+        op.create_table(
+            "coupons",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("store_id", sa.Uuid(), sa.ForeignKey("stores.id"), nullable=False),
+            sa.Column("normalized_product_id", sa.Uuid(), sa.ForeignKey("normalized_products.id"), nullable=True),
+            sa.Column("title", sa.String(300), nullable=False),
+            sa.Column("description", sa.String(1000), nullable=True),
+            sa.Column("discount_type", sa.String(20), nullable=False),
+            sa.Column("discount_value", sa.Numeric(10, 2), nullable=True),
+            sa.Column("min_purchase", sa.Numeric(10, 2), nullable=True),
+            sa.Column("valid_from", sa.Date(), nullable=True),
+            sa.Column("valid_to", sa.Date(), nullable=True),
+            sa.Column("requires_clip", sa.Boolean(), server_default=text("false"), nullable=False),
+            sa.Column("coupon_code", sa.String(100), nullable=True),
+            sa.Column("source_url", sa.String(500), nullable=True),
+            sa.Column("scraped_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        )
+
+    # 7. price_history
+    if not inspector.has_table("price_history"):
+        op.create_table(
+            "price_history",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("normalized_product_id", sa.Uuid(), sa.ForeignKey("normalized_products.id"), nullable=False),
+            sa.Column("store_id", sa.Uuid(), sa.ForeignKey("stores.id"), nullable=False),
+            sa.Column("observed_date", sa.Date(), nullable=False),
+            sa.Column("regular_price", sa.Numeric(10, 2), nullable=False),
+            sa.Column("sale_price", sa.Numeric(10, 2), nullable=True),
+            sa.Column("loyalty_price", sa.Numeric(10, 2), nullable=True),
+            sa.Column("coupon_price", sa.Numeric(10, 2), nullable=True),
+            sa.Column("source", sa.String(20), nullable=False),
+            sa.Column("purchase_item_id", sa.Uuid(), sa.ForeignKey("purchase_items.id"), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Index("ix_price_history_product_store_date", "normalized_product_id", "store_id", "observed_date"),
+        )
+
+    # 8. shrinkflation_events
+    if not inspector.has_table("shrinkflation_events"):
+        op.create_table(
+            "shrinkflation_events",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("normalized_product_id", sa.Uuid(), sa.ForeignKey("normalized_products.id"), nullable=False),
+            sa.Column("detected_date", sa.Date(), nullable=False),
+            sa.Column("old_size", sa.String(50), nullable=False),
+            sa.Column("new_size", sa.String(50), nullable=False),
+            sa.Column("old_unit", sa.String(10), nullable=True),
+            sa.Column("new_unit", sa.String(10), nullable=True),
+            sa.Column("price_at_old_size", sa.Numeric(10, 2), nullable=True),
+            sa.Column("price_at_new_size", sa.Numeric(10, 2), nullable=True),
+            sa.Column("confidence", sa.Numeric(3, 2), server_default=text("1.00"), nullable=False),
+            sa.Column("notes", sa.String(1000), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        )
+
+    # 9. user_store_accounts
+    if not inspector.has_table("user_store_accounts"):
+        op.create_table(
+            "user_store_accounts",
+            sa.Column("id", sa.Uuid(), server_default=text("gen_random_uuid()"), primary_key=True),
+            sa.Column("user_id", sa.Text(), sa.ForeignKey("users.id"), nullable=False),
+            sa.Column("store_id", sa.Uuid(), sa.ForeignKey("stores.id"), nullable=False),
+            sa.Column("session_data", sa.JSON(), nullable=True),
+            sa.Column("session_expires_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("last_sync_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("status", sa.String(20), server_default=text("'active'"), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.UniqueConstraint("user_id", "store_id", name="uq_user_store_account"),
+        )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+
+    if inspector.has_table("user_store_accounts"):
+        op.drop_table("user_store_accounts")
+    if inspector.has_table("shrinkflation_events"):
+        op.drop_table("shrinkflation_events")
+    if inspector.has_table("price_history"):
+        op.drop_table("price_history")
+    if inspector.has_table("coupons"):
+        op.drop_table("coupons")
+    if inspector.has_table("purchase_items"):
+        op.drop_table("purchase_items")
+    if inspector.has_table("purchases"):
+        op.drop_table("purchases")
+    if inspector.has_table("normalized_products"):
+        op.drop_table("normalized_products")
+    if inspector.has_table("store_locations"):
+        op.drop_table("store_locations")
+    if inspector.has_table("stores"):
+        op.drop_table("stores")

api/src/cartsnitch_api/auth/dependencies.py

@@ -1,34 +1,92 @@
-"""FastAPI dependency injection for authentication."""
+"""FastAPI dependency injection for authentication.
 
-from uuid import UUID
+Validates Better-Auth session tokens from cookies or Bearer header.
+Sessions are verified by querying the shared sessions table directly.
+"""
 
-from fastapi import Depends, Header, HTTPException, status
+from datetime import UTC, datetime
+from fastapi import Cookie, Depends, Header, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
 
-from cartsnitch_api.auth.jwt import decode_token
 from cartsnitch_api.config import settings
+from cartsnitch_api.database import get_db
 
-bearer_scheme = HTTPBearer()
+# Keep Bearer scheme as optional — Better-Auth primarily uses cookies,
+# but we support Bearer tokens for service-to-service or mobile clients.
+bearer_scheme = HTTPBearer(auto_error=False)
+
+# Better-Auth session cookie name
+SESSION_COOKIE_NAME = "better-auth.session_token"
+# Secure prefix used by better-auth on HTTPS deployments
+SECURE_SESSION_COOKIE_NAME = "__Secure-better-auth.session_token"
+
+
+async def _validate_session_token(token: str, db: AsyncSession) -> str:
+    """Validate a Better-Auth session token against the sessions table.
+
+    Better-Auth stores the raw token in the DB. The cookie/Bearer header
+    carries the same raw token, so we compare directly.
+    """
+    result = await db.execute(
+        text("SELECT user_id, expires_at FROM sessions WHERE token = :token"),
+        {"token": token},
+    )
+    row = result.first()
+
+    if not row:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid session token",
+        )
+
+    user_id, expires_at = row
+    if expires_at.tzinfo is None:
+        # Treat naive datetimes as UTC
+        expires_at = expires_at.replace(tzinfo=UTC)
+
+    if expires_at < datetime.now(UTC):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Session expired",
+        )
+
+    return str(user_id)
 
 
 async def get_current_user(
-    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
-) -> UUID:
-    try:
-        payload = decode_token(credentials.credentials)
-    except ValueError:
+    request: Request,
+    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
+    db: AsyncSession = Depends(get_db),
+) -> str:
+    """Extract and validate the session token from cookie or Authorization header.
+
+    Checks in order:
+    1. Better-Auth session cookie (primary — web clients)
+    2. Bearer token in Authorization header (fallback — API clients)
+    """
+    token: str | None = None
+
+    # 1. Check session cookie — prefer __Secure- variant (HTTPS) over plain (HTTP dev)
+    cookie_token = request.cookies.get(SECURE_SESSION_COOKIE_NAME) or request.cookies.get(SESSION_COOKIE_NAME)
+    if cookie_token:
+        # Better-Auth cookie format is "token.sessionId" — extract just the token part
+        token = cookie_token.split(".")[0] if "." in cookie_token else cookie_token
+
+    # 2. Fall back to Bearer header
+    if not token and credentials:
+        # Callers might pass the compound value here too
+        raw = credentials.credentials
+        token = raw.split(".")[0] if "." in raw else raw
+
+    if not token:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid or expired token",
-        ) from None
+            detail="Authentication required",
+        )
 
-    if payload.get("type") != "access":
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid token type",
-        ) from None
-
-    return UUID(payload["sub"])
+    return await _validate_session_token(token, db)
 
 
 async def verify_service_key(x_service_key: str = Header()) -> None:

api/src/cartsnitch_api/auth/routes.py

@@ -1,17 +1,19 @@
-"""Auth routes: register, login, refresh, me, update, delete."""
+"""Auth routes: user profile management.
 
-from uuid import UUID
+Registration, login, refresh, and session management are handled by
+the Better-Auth service (auth/). This router provides user profile
+endpoints that query our own user data from the shared database.
+"""
 
 from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from cartsnitch_api.auth.dependencies import get_current_user
 from cartsnitch_api.database import get_db
+from cartsnitch_api.models import User
 from cartsnitch_api.schemas import (
-    LoginRequest,
-    RefreshRequest,
-    RegisterRequest,
-    TokenResponse,
     UpdateUserRequest,
     UserResponse,
 )
@@ -20,40 +22,9 @@ from cartsnitch_api.services.auth import AuthService
 router = APIRouter(prefix="/auth", tags=["auth"])
 
 
-@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
-async def register(body: RegisterRequest, db: AsyncSession = Depends(get_db)):
-    svc = AuthService(db)
-    try:
-        return await svc.register(body.email, body.password, body.display_name)
-    except ValueError as e:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) from e
-
-
-@router.post("/login", response_model=TokenResponse)
-async def login(body: LoginRequest, db: AsyncSession = Depends(get_db)):
-    svc = AuthService(db)
-    try:
-        return await svc.login(body.email, body.password)
-    except ValueError:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password"
-        ) from None
-
-
-@router.post("/refresh", response_model=TokenResponse)
-async def refresh(body: RefreshRequest, db: AsyncSession = Depends(get_db)):
-    svc = AuthService(db)
-    try:
-        return await svc.refresh(body.refresh_token)
-    except ValueError:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token"
-        ) from None
-
-
 @router.get("/me", response_model=UserResponse)
 async def get_me(
-    user_id: UUID = Depends(get_current_user),
+    user_id: str = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
     svc = AuthService(db)
@@ -68,7 +39,7 @@ async def get_me(
 @router.patch("/me", response_model=UserResponse)
 async def update_me(
     body: UpdateUserRequest,
-    user_id: UUID = Depends(get_current_user),
+    user_id: str = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
     svc = AuthService(db)
@@ -84,7 +55,7 @@ async def update_me(
 
 @router.delete("/me", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_me(
-    user_id: UUID = Depends(get_current_user),
+    user_id: str = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
     svc = AuthService(db)

api/src/cartsnitch_api/cache.py

@@ -1,26 +1,51 @@
 """Redis/DragonflyDB caching helpers."""
 
+import redis.asyncio as redis
+
 from cartsnitch_api.config import settings
 
 
 class CacheClient:
-    """Stub for Redis/DragonflyDB caching.
+    """Redis/DragonflyDB caching with connection pooling.
 
     Will be used for expensive queries: price trends, product comparisons.
     Cache invalidation via Redis pub/sub events from other services.
     """
 
     def __init__(self) -> None:
-        self.url = settings.redis_url
+        self._pool: redis.ConnectionPool | None = None
+        self._client: redis.Redis | None = None
+
+    async def initialize(self) -> None:
+        """Initialize the Redis connection pool."""
+        self._pool = redis.ConnectionPool.from_url(
+            settings.redis_url,
+            max_connections=20,
+            decode_responses=True,
+        )
+        self._client = redis.Redis(connection_pool=self._pool)
+
+    async def close(self) -> None:
+        """Close the Redis connection pool."""
+        if self._client:
+            await self._client.aclose()
+        if self._pool:
+            await self._pool.aclose()
 
     async def get(self, key: str) -> str | None:
-        # TODO: implement with redis-py async
-        return None
+        if not self._client:
+            return None
+        return await self._client.get(key)
 
     async def set(self, key: str, value: str, ttl_seconds: int = 300) -> None:
-        # TODO: implement with redis-py async
-        pass
+        if not self._client:
+            return
+        await self._client.set(key, value, ex=ttl_seconds)
 
     async def delete(self, key: str) -> None:
-        # TODO: implement with redis-py async
-        pass
+        if not self._client:
+            return
+        await self._client.delete(key)
+
+
+cache_client = CacheClient()

api/src/cartsnitch_api/config.py

@@ -1,23 +1,27 @@
 import base64
 
-from pydantic import model_validator
+from pydantic import AliasChoices, Field, model_validator
 from pydantic_settings import BaseSettings
 
 
 class Settings(BaseSettings):
     model_config = {"env_prefix": "CARTSNITCH_"}
 
-    database_url: str = "postgresql+asyncpg://cartsnitch:cartsnitch@localhost:5432/cartsnitch"
+    database_url: str = Field(
+        default="postgresql+asyncpg://cartsnitch:cartsnitch@localhost:5432/cartsnitch",
+        validation_alias=AliasChoices("CARTSNITCH_DATABASE_URL", "DATABASE_URL"),
+    )
     redis_url: str = "redis://localhost:6379/0"
 
-    jwt_secret_key: str = "change-me-in-production"
+    jwt_secret_key: str
     jwt_algorithm: str = "HS256"
     jwt_access_token_expire_minutes: int = 15
     jwt_refresh_token_expire_days: int = 7
 
-    service_key: str = "change-me-in-production"
-    # Valid Fernet key for local dev — MUST be overridden in production
-    fernet_key: str = "7reF42nmTwbdN21PBoubGp7h_FU8qSimstmlaMLoRK8="
+    service_key: str
+    fernet_key: str
+
+    auth_service_url: str = "http://auth:3001"
 
     cors_origins: list[str] = ["http://localhost:3000", "https://cartsnitch.com"]
 
@@ -30,9 +34,26 @@ class Settings(BaseSettings):
     rate_limit_window_seconds: int = 60
     rate_limit_enabled: bool = True
 
+    _PLACEHOLDER_VALUES = {"change-me-in-production"}
+
     @model_validator(mode="after")
-    def validate_fernet_key(self):
-        """Validate fernet_key is a valid 32-byte url-safe base64 key at startup."""
+    def validate_secrets(self):
+        if not self.jwt_secret_key or self.jwt_secret_key in self._PLACEHOLDER_VALUES:
+            raise ValueError(
+                "CARTSNITCH_JWT_SECRET_KEY must be set to a secure value. "
+                'Generate one with: python -c "import secrets; print(secrets.token_urlsafe(32))"'
+            )
+        if not self.service_key or self.service_key in self._PLACEHOLDER_VALUES:
+            raise ValueError(
+                "CARTSNITCH_SERVICE_KEY must be set to a secure value. "
+                'Generate one with: python -c "import secrets; print(secrets.token_urlsafe(32))"'
+            )
+        if not self.fernet_key or self.fernet_key in self._PLACEHOLDER_VALUES:
+            raise ValueError(
+                "CARTSNITCH_FERNET_KEY must be set to a valid Fernet key. "
+                "Generate one with: python -c "
+                "'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())'"
+            )
         try:
             decoded = base64.urlsafe_b64decode(self.fernet_key.encode())
             if len(decoded) != 32:
@@ -47,5 +68,12 @@ class Settings(BaseSettings):
             ) from None
         return self
 
+    @model_validator(mode="after")
+    def normalize_database_url(self):
+        """Normalize postgresql:// → postgresql+asyncpg:// for the asyncpg driver."""
+        if self.database_url.startswith("postgresql://"):
+            self.database_url = self.database_url.replace("postgresql://", "postgresql+asyncpg://", 1)
+        return self
+
 
 settings = Settings()

api/src/cartsnitch_api/database.py

@@ -6,7 +6,14 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 
 from cartsnitch_api.config import settings
 
-engine = create_async_engine(settings.database_url, echo=False)
+engine = create_async_engine(
+    settings.database_url,
+    echo=False,
+    pool_size=10,
+    max_overflow=20,
+    pool_pre_ping=True,
+    pool_recycle=3600,
+)
 async_session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
 
@@ -14,3 +21,8 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]:
     """FastAPI dependency that yields an async DB session."""
     async with async_session_factory() as session:
         yield session
+
+
+async def dispose_engine() -> None:
+    """Dispose the database engine, closing all pooled connections."""
+    await engine.dispose()

api/src/cartsnitch_api/main.py

@@ -2,9 +2,11 @@
 
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI
+from fastapi import APIRouter, FastAPI
 
 from cartsnitch_api.auth.routes import router as auth_router
+from cartsnitch_api.cache import cache_client
+from cartsnitch_api.database import dispose_engine
 from cartsnitch_api.middleware.cors import add_cors_middleware
 from cartsnitch_api.middleware.error_handler import add_error_handlers, add_error_monitor_middleware
 from cartsnitch_api.middleware.rate_limit import add_rate_limit_middleware
@@ -18,13 +20,15 @@ from cartsnitch_api.routes.purchases import router as purchases_router
 from cartsnitch_api.routes.scraping import router as scraping_router
 from cartsnitch_api.routes.shopping import router as shopping_router
 from cartsnitch_api.routes.stores import router as stores_router
+from cartsnitch_api.routes.user import router as user_router
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # TODO: initialize DB session pool, Redis connection, service clients
+    await cache_client.initialize()
     yield
-    # TODO: cleanup connections
+    await cache_client.close()
+    await dispose_engine()
 
 
 def create_app() -> FastAPI:
@@ -46,15 +50,20 @@ def create_app() -> FastAPI:
     # Routers
     app.include_router(health_router)
     app.include_router(auth_router)
-    app.include_router(stores_router)
-    app.include_router(purchases_router)
-    app.include_router(products_router)
-    app.include_router(prices_router)
-    app.include_router(coupons_router)
-    app.include_router(shopping_router)
-    app.include_router(alerts_router)
-    app.include_router(scraping_router)
-    app.include_router(public_router)
+
+    # Data endpoints mounted under /api/v1
+    v1_router = APIRouter(prefix="/api/v1")
+    v1_router.include_router(user_router)
+    v1_router.include_router(stores_router)
+    v1_router.include_router(purchases_router)
+    v1_router.include_router(products_router)
+    v1_router.include_router(prices_router)
+    v1_router.include_router(coupons_router)
+    v1_router.include_router(shopping_router)
+    v1_router.include_router(alerts_router)
+    v1_router.include_router(scraping_router)
+    v1_router.include_router(public_router)
+    app.include_router(v1_router)
 
     return app
 

api/src/cartsnitch_api/middleware/cors.py

@@ -11,6 +11,6 @@ def add_cors_middleware(app: FastAPI) -> None:
         CORSMiddleware,
         allow_origins=settings.cors_origins,
         allow_credentials=True,
-        allow_methods=["*"],
-        allow_headers=["*"],
+        allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"],
+        allow_headers=["Content-Type", "Authorization", "Accept", "Origin", "X-Requested-With"],
     )

api/src/cartsnitch_api/middleware/rate_limit.py

@@ -4,6 +4,7 @@ Uses in-memory sliding window as fallback, Redis/DragonflyDB when available.
 Per-IP limiting on public endpoints, per-token limiting on authenticated endpoints.
 """
 
+import hashlib
 import time
 from collections import defaultdict
 from threading import Lock
@@ -71,8 +72,8 @@ def _get_rate_limit_key(request: Request) -> tuple[str, _SlidingWindowCounter]:
     auth_header = request.headers.get("authorization", "")
     if auth_header.startswith("Bearer "):
         token = auth_header[7:]
-        # Use last 16 chars of token as key to avoid storing full tokens
-        return f"token:{token[-16:]}", _auth_limiter
+        token_hash = hashlib.sha256(token.encode()).hexdigest()
+        return f"token:{token_hash}", _auth_limiter
 
     # Fallback to IP for unauthenticated non-public endpoints
     return f"ip:{_get_client_ip(request)}", _public_limiter

api/src/cartsnitch_api/models/purchase.py

@@ -32,8 +32,8 @@ class Purchase(UUIDPrimaryKeyMixin, TimestampMixin, Base):
 
     __tablename__ = "purchases"
 
-    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"), nullable=False)
-    store_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("stores.id"), nullable=False)
+    user_id: Mapped[str] = mapped_column(ForeignKey("users.id"), nullable=False)
+    store_id: Mapped[str] = mapped_column(ForeignKey("stores.id"), nullable=False)
     store_location_id: Mapped[uuid.UUID | None] = mapped_column(ForeignKey("store_locations.id"))
     receipt_id: Mapped[str] = mapped_column(String(200), nullable=False)
     purchase_date: Mapped[date] = mapped_column(Date, nullable=False)

api/src/cartsnitch_api/models/user.py

@@ -1,10 +1,11 @@
 """User and UserStoreAccount models."""
 
-import uuid
+import secrets
 from datetime import datetime
 from typing import TYPE_CHECKING
 
-from sqlalchemy import DateTime, ForeignKey, String, UniqueConstraint
+import sqlalchemy as sa
+from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from cartsnitch_api.constants import AccountStatus
@@ -16,14 +17,28 @@ if TYPE_CHECKING:
     from cartsnitch_api.models.store import Store
 
 
-class User(UUIDPrimaryKeyMixin, TimestampMixin, Base):
+class User(TimestampMixin, Base):
     """Application user."""
 
     __tablename__ = "users"
 
+    id: Mapped[str] = mapped_column(Text, primary_key=True)
     email: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
-    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
+    hashed_password: Mapped[str | None] = mapped_column(String(255), nullable=True)
     display_name: Mapped[str | None] = mapped_column(String(100))
+    email_verified: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, server_default="false"
+    )
+    image: Mapped[str | None] = mapped_column(Text, nullable=True)
+    email_inbound_token: Mapped[str] = mapped_column(
+        String(22),
+        nullable=False,
+        unique=True,
+        default=lambda: secrets.token_urlsafe(16),
+        server_default=sa.text(
+            "replace(replace(trim(trailing '=' from encode(gen_random_bytes(16), 'base64')), '+', '-'), '/', '_')"
+        ),
+    )
 
     # Relationships
     store_accounts: Mapped[list["UserStoreAccount"]] = relationship(back_populates="user")
@@ -36,8 +51,8 @@ class UserStoreAccount(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     __tablename__ = "user_store_accounts"
     __table_args__ = (UniqueConstraint("user_id", "store_id", name="uq_user_store_account"),)
 
-    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"), nullable=False)
-    store_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("stores.id"), nullable=False)
+    user_id: Mapped[str] = mapped_column(ForeignKey("users.id"), nullable=False)
+    store_id: Mapped[str] = mapped_column(ForeignKey("stores.id"), nullable=False)
     session_data: Mapped[dict | None] = mapped_column(EncryptedJSON)
     session_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     last_sync_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

api/src/cartsnitch_api/routes/user.py

@@ -0,0 +1,32 @@
+"""User routes: per-user account endpoints (email-in address, etc.)."""
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from cartsnitch_api.auth.dependencies import get_current_user
+from cartsnitch_api.database import get_db
+from cartsnitch_api.schemas import EmailInAddressResponse
+from cartsnitch_api.services.auth import AuthService
+
+router = APIRouter(tags=["user"])
+
+
+@router.get("/me/email-in-address", response_model=EmailInAddressResponse)
+async def get_email_in_address(
+    user_id: str = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    svc = AuthService(db)
+    try:
+        email_address = await svc.get_email_in_address(user_id)
+        return EmailInAddressResponse(
+            email_address=email_address,
+            instructions=(
+                "Forward your digital receipt emails to this address. "
+                "We currently support Meijer, Kroger, and Target receipt emails."
+            ),
+        )
+    except LookupError:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
+        ) from None

api/src/cartsnitch_api/schemas.py

@@ -6,28 +6,8 @@ from uuid import UUID
 from pydantic import BaseModel, EmailStr, Field
 
 # ---------- Auth ----------
-
-
-class RegisterRequest(BaseModel):
-    email: EmailStr
-    password: str = Field(min_length=8, max_length=128)
-    display_name: str = Field(min_length=1, max_length=100)
-
-
-class LoginRequest(BaseModel):
-    email: EmailStr
-    password: str
-
-
-class RefreshRequest(BaseModel):
-    refresh_token: str
-
-
-class TokenResponse(BaseModel):
-    access_token: str
-    refresh_token: str
-    token_type: str = "bearer"
-    expires_in: int
+# Registration, login, and session management are handled by Better-Auth (auth/ service).
+# These schemas are for the profile management endpoints only.
 
 
 class UpdateUserRequest(BaseModel):
@@ -36,12 +16,17 @@ class UpdateUserRequest(BaseModel):
 
 
 class UserResponse(BaseModel):
-    id: UUID
+    id: str
     email: str
     display_name: str
     created_at: datetime
 
 
+class EmailInAddressResponse(BaseModel):
+    email_address: str
+    instructions: str
+
+
 # ---------- Stores ----------
 
 

api/src/cartsnitch_api/services/auth.py

@@ -1,68 +1,19 @@
-"""Auth service — user registration, login, token management."""
+"""Auth service — user profile management.
 
-from uuid import UUID
+Registration, login, token management, and session handling are now
+handled by the Better-Auth service (auth/). This service provides
+user lookup and profile update operations for the API gateway.
+"""
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from cartsnitch_api.auth.jwt import create_access_token, create_refresh_token, decode_token
-from cartsnitch_api.auth.passwords import hash_password, verify_password
-from cartsnitch_api.config import settings
-
 
 class AuthService:
     def __init__(self, db: AsyncSession) -> None:
         self.db = db
 
-    async def register(self, email: str, password: str, display_name: str) -> dict:
-        from cartsnitch_api.models import User
-
-        existing = await self.db.execute(select(User).where(User.email == email))
-        if existing.scalar_one_or_none():
-            raise ValueError("Email already registered")
-
-        user = User(
-            email=email,
-            hashed_password=hash_password(password),
-            display_name=display_name,
-        )
-        self.db.add(user)
-        await self.db.commit()
-        await self.db.refresh(user)
-
-        return self._make_token_response(user.id)
-
-    async def login(self, email: str, password: str) -> dict:
-        from cartsnitch_api.models import User
-
-        result = await self.db.execute(select(User).where(User.email == email))
-        user = result.scalar_one_or_none()
-        if not user or not verify_password(password, user.hashed_password):
-            raise ValueError("Invalid email or password")
-
-        return self._make_token_response(user.id)
-
-    async def refresh(self, refresh_token: str) -> dict:
-        from cartsnitch_api.models import User
-
-        try:
-            payload = decode_token(refresh_token)
-        except ValueError:
-            raise ValueError("Invalid refresh token") from None
-
-        if payload.get("type") != "refresh":
-            raise ValueError("Invalid token type") from None
-
-        user_id = UUID(payload["sub"])
-
-        # Verify the user still exists before issuing new tokens
-        result = await self.db.execute(select(User).where(User.id == user_id))
-        if not result.scalar_one_or_none():
-            raise ValueError("User no longer exists")
-
-        return self._make_token_response(user_id)
-
-    async def get_user(self, user_id: UUID) -> dict:
+    async def get_user(self, user_id: str) -> dict:
         from cartsnitch_api.models import User
 
         result = await self.db.execute(select(User).where(User.id == user_id))
@@ -77,7 +28,7 @@ class AuthService:
             "created_at": user.created_at,
         }
 
-    async def update_user(self, user_id: UUID, **fields) -> dict:
+    async def update_user(self, user_id: str, **fields) -> dict:
         from cartsnitch_api.models import User
 
         result = await self.db.execute(select(User).where(User.id == user_id))
@@ -105,7 +56,7 @@ class AuthService:
             "created_at": user.created_at,
         }
 
-    async def delete_user(self, user_id: UUID) -> None:
+    async def delete_user(self, user_id: str) -> None:
         from cartsnitch_api.models import User
 
         result = await self.db.execute(select(User).where(User.id == user_id))
@@ -116,10 +67,13 @@ class AuthService:
         await self.db.delete(user)
         await self.db.commit()
 
-    def _make_token_response(self, user_id: UUID) -> dict:
-        return {
-            "access_token": create_access_token(user_id),
-            "refresh_token": create_refresh_token(user_id),
-            "token_type": "bearer",
-            "expires_in": settings.jwt_access_token_expire_minutes * 60,
-        }
+    async def get_email_in_address(self, user_id: str) -> str:
+        """Return the per-user email-in address for receipt forwarding."""
+        from cartsnitch_api.models import User
+
+        result = await self.db.execute(select(User).where(User.id == user_id))
+        user = result.scalar_one_or_none()
+        if not user:
+            raise LookupError("User not found")
+
+        return f"receipts+{user.email_inbound_token}@receipts.cartsnitch.com"

api/tests/conftest.py

@@ -1,8 +1,16 @@
-"""Shared test fixtures with in-memory SQLite database."""
+"""Shared test fixtures with in-memory SQLite database.
+
+Session-based auth: tests create users and sessions directly in the DB,
+matching the Better-Auth session validation flow.
+"""
+
+import secrets
+import uuid
+from datetime import UTC, datetime, timedelta
 
 import pytest
 from httpx import ASGITransport, AsyncClient
-from sqlalchemy import create_engine, event
+from sqlalchemy import create_engine, event, text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import sessionmaker
 
@@ -11,6 +19,25 @@ from cartsnitch_api.database import get_db
 from cartsnitch_api.main import create_app
 from cartsnitch_api.models import Base
 
+TEST_JWT_SECRET = secrets.token_urlsafe(32)
+TEST_SERVICE_KEY = secrets.token_urlsafe(32)
+TEST_FERNET_KEY = "7reF42nmTwbdN21PBoubGp7h_FU8qSimstmlaMLoRK8="
+
+
+@pytest.fixture(autouse=True)
+def setup_test_settings():
+    original_jwt = cartsnitch_settings.jwt_secret_key
+    original_service = cartsnitch_settings.service_key
+    original_fernet = cartsnitch_settings.fernet_key
+    cartsnitch_settings.jwt_secret_key = TEST_JWT_SECRET
+    cartsnitch_settings.service_key = TEST_SERVICE_KEY
+    cartsnitch_settings.fernet_key = TEST_FERNET_KEY
+    yield
+    cartsnitch_settings.jwt_secret_key = original_jwt
+    cartsnitch_settings.service_key = original_service
+    cartsnitch_settings.fernet_key = original_fernet
+
+
 TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
 
 
@@ -51,6 +78,52 @@ async def db_engine():
 
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
+        # Create Better-Auth tables (not managed by SQLAlchemy models)
+        await conn.execute(
+            text("""
+            CREATE TABLE IF NOT EXISTS sessions (
+                id TEXT PRIMARY KEY,
+                token TEXT NOT NULL UNIQUE,
+                user_id TEXT NOT NULL,
+                expires_at TIMESTAMP NOT NULL,
+                ip_address TEXT,
+                user_agent TEXT,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
+                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
+            )
+        """)
+        )
+        await conn.execute(
+            text("""
+            CREATE TABLE IF NOT EXISTS accounts (
+                id TEXT PRIMARY KEY,
+                user_id TEXT NOT NULL,
+                account_id TEXT NOT NULL,
+                provider_id TEXT NOT NULL,
+                access_token TEXT,
+                refresh_token TEXT,
+                access_token_expires_at TIMESTAMP,
+                refresh_token_expires_at TIMESTAMP,
+                scope TEXT,
+                id_token TEXT,
+                password TEXT,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
+                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
+            )
+        """)
+        )
+        await conn.execute(
+            text("""
+            CREATE TABLE IF NOT EXISTS verifications (
+                id TEXT PRIMARY KEY,
+                identifier TEXT NOT NULL,
+                value TEXT NOT NULL,
+                expires_at TIMESTAMP NOT NULL,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
+                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
+            )
+        """)
+        )
 
     yield engine
 
@@ -85,17 +158,58 @@ async def client(db_engine):
     app.dependency_overrides.clear()
 
 
+async def _create_test_user_and_session(
+    client: AsyncClient, db_engine, **user_overrides
+) -> tuple[dict, str]:
+    """Create a test user and a valid session directly in the DB.
+
+    Returns (user_dict, session_token).  Better-Auth stores the raw token
+    in the DB, so we insert it as-is.
+    """
+    user_id = str(uuid.uuid4())
+    email = user_overrides.get("email", "test@example.com")
+    display_name = user_overrides.get("display_name", "Test User")
+    session_token = secrets.token_urlsafe(32)
+    session_id = str(uuid.uuid4())
+    now = datetime.now(UTC).isoformat()
+    expires = (datetime.now(UTC) + timedelta(days=7)).isoformat()
+
+    async with db_engine.begin() as conn:
+        await conn.execute(
+            text(
+                "INSERT INTO users (id, email, hashed_password, display_name, email_verified, created_at, updated_at) "
+                "VALUES (:id, :email, :hashed_password, :display_name, :email_verified, :created_at, :updated_at)"
+            ),
+            {
+                "id": user_id,
+                "email": email,
+                "hashed_password": "not-used-with-better-auth",
+                "display_name": display_name,
+                "email_verified": False,
+                "created_at": now,
+                "updated_at": now,
+            },
+        )
+        await conn.execute(
+            text(
+                "INSERT INTO sessions (id, token, user_id, expires_at, created_at, updated_at) "
+                "VALUES (:id, :token, :user_id, :expires_at, :created_at, :updated_at)"
+            ),
+            {
+                "id": session_id,
+                "token": session_token,
+                "user_id": user_id,
+                "expires_at": expires,
+                "created_at": now,
+                "updated_at": now,
+            },
+        )
+
+    return {"id": user_id, "email": email, "display_name": display_name}, session_token
+
+
 @pytest.fixture
-async def auth_headers(client):
-    """Register a test user and return auth headers."""
-    resp = await client.post(
-        "/auth/register",
-        json={
-            "email": "test@example.com",
-            "password": "testpass123",
-            "display_name": "Test User",
-        },
-    )
-    assert resp.status_code == 201
-    token = resp.json()["access_token"]
-    return {"Authorization": f"Bearer {token}"}
+async def auth_headers(client, db_engine):
+    """Create a test user with a valid session and return auth headers."""
+    _, session_token = await _create_test_user_and_session(client, db_engine)
+    return {"Cookie": f"better-auth.session_token={session_token}"}

api/tests/test_auth/test_auth_endpoints.py

@@ -1,146 +1,13 @@
-"""Integration tests for auth endpoints."""
+"""Integration tests for auth profile endpoints.
+
+Registration, login, and session management are handled by the Better-Auth
+service. These tests cover the profile endpoints (GET/PATCH/DELETE /auth/me)
+which validate sessions via the shared sessions table.
+"""
 
 import pytest
 
 
-@pytest.mark.asyncio
-async def test_register_success(client):
-    resp = await client.post(
-        "/auth/register",
-        json={
-            "email": "new@example.com",
-            "password": "securepass123",
-            "display_name": "New User",
-        },
-    )
-    assert resp.status_code == 201
-    data = resp.json()
-    assert "access_token" in data
-    assert "refresh_token" in data
-    assert data["token_type"] == "bearer"
-    assert data["expires_in"] == 900  # 15 min * 60
-
-
-@pytest.mark.asyncio
-async def test_register_duplicate_email(client):
-    await client.post(
-        "/auth/register",
-        json={
-            "email": "dupe@example.com",
-            "password": "securepass123",
-            "display_name": "User One",
-        },
-    )
-    resp = await client.post(
-        "/auth/register",
-        json={
-            "email": "dupe@example.com",
-            "password": "securepass456",
-            "display_name": "User Two",
-        },
-    )
-    assert resp.status_code == 409
-
-
-@pytest.mark.asyncio
-async def test_register_short_password(client):
-    resp = await client.post(
-        "/auth/register",
-        json={
-            "email": "short@example.com",
-            "password": "short",
-            "display_name": "Short Pass",
-        },
-    )
-    assert resp.status_code == 422
-
-
-@pytest.mark.asyncio
-async def test_login_success(client):
-    await client.post(
-        "/auth/register",
-        json={
-            "email": "login@example.com",
-            "password": "securepass123",
-            "display_name": "Login User",
-        },
-    )
-    resp = await client.post(
-        "/auth/login",
-        json={
-            "email": "login@example.com",
-            "password": "securepass123",
-        },
-    )
-    assert resp.status_code == 200
-    assert "access_token" in resp.json()
-
-
-@pytest.mark.asyncio
-async def test_login_wrong_password(client):
-    await client.post(
-        "/auth/register",
-        json={
-            "email": "wrong@example.com",
-            "password": "securepass123",
-            "display_name": "Wrong Pass",
-        },
-    )
-    resp = await client.post(
-        "/auth/login",
-        json={
-            "email": "wrong@example.com",
-            "password": "badpassword1",
-        },
-    )
-    assert resp.status_code == 401
-
-
-@pytest.mark.asyncio
-async def test_login_nonexistent_user(client):
-    resp = await client.post(
-        "/auth/login",
-        json={
-            "email": "ghost@example.com",
-            "password": "doesntmatter",
-        },
-    )
-    assert resp.status_code == 401
-
-
-@pytest.mark.asyncio
-async def test_refresh_token(client):
-    reg = await client.post(
-        "/auth/register",
-        json={
-            "email": "refresh@example.com",
-            "password": "securepass123",
-            "display_name": "Refresh User",
-        },
-    )
-    refresh_token = reg.json()["refresh_token"]
-
-    resp = await client.post(
-        "/auth/refresh",
-        json={
-            "refresh_token": refresh_token,
-        },
-    )
-    assert resp.status_code == 200
-    assert "access_token" in resp.json()
-
-
-@pytest.mark.asyncio
-async def test_refresh_with_invalid_token(client):
-    resp = await client.post(
-        "/auth/refresh",
-        json={
-            "refresh_token": "invalid.token.here",
-        },
-    )
-    assert resp.status_code == 401
-
-
 @pytest.mark.asyncio
 async def test_get_me(client, auth_headers):
     resp = await client.get("/auth/me", headers=auth_headers)
@@ -155,7 +22,32 @@ async def test_get_me(client, auth_headers):
 @pytest.mark.asyncio
 async def test_get_me_unauthorized(client):
     resp = await client.get("/auth/me")
-    assert resp.status_code in (401, 403)  # No auth header
+    assert resp.status_code in (401, 403)
+
+
+@pytest.mark.asyncio
+async def test_get_me_invalid_session(client):
+    resp = await client.get(
+        "/auth/me",
+        headers={"Cookie": "better-auth.session_token=invalid-token"},
+    )
+    assert resp.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_get_me_with_bearer_token(client, db_engine):
+    """Session tokens can also be passed as Bearer tokens for API clients."""
+    from tests.conftest import _create_test_user_and_session
+
+    _, session_token = await _create_test_user_and_session(
+        client, db_engine, email="bearer@example.com", display_name="Bearer User"
+    )
+    resp = await client.get(
+        "/auth/me",
+        headers={"Authorization": f"Bearer {session_token}"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["email"] == "bearer@example.com"
 
 
 @pytest.mark.asyncio
@@ -163,9 +55,7 @@ async def test_update_me(client, auth_headers):
     resp = await client.patch(
         "/auth/me",
         headers=auth_headers,
-        json={
-            "display_name": "Updated Name",
-        },
+        json={"display_name": "Updated Name"},
     )
     assert resp.status_code == 200
     assert resp.json()["display_name"] == "Updated Name"
@@ -176,34 +66,108 @@ async def test_delete_me(client, auth_headers):
     resp = await client.delete("/auth/me", headers=auth_headers)
     assert resp.status_code == 204
 
-    # Verify user is gone (token still valid but user deleted)
+    # Session is still valid but user is gone
     resp = await client.get("/auth/me", headers=auth_headers)
     assert resp.status_code == 404
 
 
 @pytest.mark.asyncio
-async def test_refresh_after_delete_fails(client):
-    """Refresh token for a deleted user must be rejected."""
-    reg = await client.post(
-        "/auth/register",
-        json={
-            "email": "ghost@example.com",
-            "password": "securepass123",
-            "display_name": "Ghost User",
-        },
+async def test_get_me_compound_cookie(client, db_engine):
+    """Compound cookie value (token.sessionId) must be parsed to extract the token part."""
+    from tests.conftest import _create_test_user_and_session
+
+    _, session_token = await _create_test_user_and_session(
+        client, db_engine, email="compound@example.com", display_name="Compound User"
     )
-    tokens = reg.json()
-    headers = {"Authorization": f"Bearer {tokens['access_token']}"}
+    compound = f"{session_token}.B0atkJCFxK1rZlwWPMK97nVO2LnyDun7"
+    resp = await client.get(
+        "/auth/me",
+        headers={"Cookie": f"better-auth.session_token={compound}"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["email"] == "compound@example.com"
 
-    # Delete the user
-    resp = await client.delete("/auth/me", headers=headers)
-    assert resp.status_code == 204
 
-    # Refresh token should now fail
-    resp = await client.post(
-        "/auth/refresh",
-        json={
-            "refresh_token": tokens["refresh_token"],
-        },
+@pytest.mark.asyncio
+async def test_get_me_raw_token_cookie(client, db_engine):
+    """Raw token (no dot) in cookie must still work — regression guard."""
+    from tests.conftest import _create_test_user_and_session
+
+    _, session_token = await _create_test_user_and_session(
+        client, db_engine, email="rawcookie@example.com", display_name="Raw Cookie User"
+    )
+    resp = await client.get(
+        "/auth/me",
+        headers={"Cookie": f"better-auth.session_token={session_token}"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["email"] == "rawcookie@example.com"
+
+
+@pytest.mark.asyncio
+async def test_get_me_compound_bearer(client, db_engine):
+    """Compound Bearer token (token.sessionId) must be parsed to extract the token part."""
+    from tests.conftest import _create_test_user_and_session
+
+    _, session_token = await _create_test_user_and_session(
+        client, db_engine, email="compoundbearer@example.com", display_name="Compound Bearer User"
+    )
+    compound = f"{session_token}.B0atkJCFxK1rZlwWPMK97nVO2LnyDun7"
+    resp = await client.get(
+        "/auth/me",
+        headers={"Authorization": f"Bearer {compound}"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["email"] == "compoundbearer@example.com"
+
+
+@pytest.mark.asyncio
+async def test_expired_session_rejected(client, db_engine):
+    """Expired sessions must be rejected."""
+    import secrets
+    import uuid
+    from datetime import UTC, datetime, timedelta
+
+    from sqlalchemy import text
+
+    user_id = str(uuid.uuid4())
+    session_token = secrets.token_urlsafe(32)
+    now = datetime.now(UTC).isoformat()
+    expired = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+
+    async with db_engine.begin() as conn:
+        await conn.execute(
+            text(
+                "INSERT INTO users (id, email, hashed_password, display_name, email_verified, created_at, updated_at) "
+                "VALUES (:id, :email, :hp, :dn, :ev, :ca, :ua)"
+            ),
+            {
+                "id": user_id,
+                "email": "expired@example.com",
+                "hp": "unused",
+                "dn": "Expired User",
+                "ev": False,
+                "ca": now,
+                "ua": now,
+            },
+        )
+        await conn.execute(
+            text(
+                "INSERT INTO sessions (id, token, user_id, expires_at, created_at, updated_at) "
+                "VALUES (:id, :token, :uid, :ea, :ca, :ua)"
+            ),
+            {
+                "id": str(uuid.uuid4()),
+                "token": session_token,
+                "uid": user_id,
+                "ea": expired,
+                "ca": now,
+                "ua": now,
+            },
+        )
+
+    resp = await client.get(
+        "/auth/me",
+        headers={"Cookie": f"better-auth.session_token={session_token}"},
     )
     assert resp.status_code == 401

api/tests/test_config.py

@@ -0,0 +1,48 @@
+"""Tests for Settings config, specifically the database_url env var fallback."""
+
+import os
+
+from cartsnitch_api.config import Settings
+
+
+def test_database_url_prefers_cartsnitch_prefix():
+    """CARTSNITCH_DATABASE_URL takes precedence over DATABASE_URL."""
+    env = {
+        "CARTSNITCH_DATABASE_URL": "postgresql+asyncpg://user1:pass1@host1:5432/db1",
+        "DATABASE_URL": "postgresql://user2:pass2@host2:5432/db2",
+    }
+    settings = Settings(**env)
+    assert settings.database_url == "postgresql+asyncpg://user1:pass1@host1:5432/db1"
+
+
+def test_database_url_falls_back_to_database_url():
+    """When CARTSNITCH_DATABASE_URL is absent, DATABASE_URL is accepted."""
+    env = {
+        "DATABASE_URL": "postgresql://user:pass@dbhost:5432/mydb",
+    }
+    settings = Settings(**env)
+    assert settings.database_url == "postgresql+asyncpg://user:pass@dbhost:5432/mydb"
+
+
+def test_database_url_normalizes_plain_postgresql_prefix():
+    """DATABASE_URL with plain postgresql:// is normalized to postgresql+asyncpg://."""
+    env = {
+        "DATABASE_URL": "postgresql://cartsnitch:cartsnitch@localhost:5432/cartsnitch",
+    }
+    settings = Settings(**env)
+    assert settings.database_url == "postgresql+asyncpg://cartsnitch:cartsnitch@localhost:5432/cartsnitch"
+
+
+def test_database_url_preserves_asyncpg_prefix():
+    """CARTSNITCH_DATABASE_URL with postgresql+asyncpg:// is left unchanged."""
+    env = {
+        "CARTSNITCH_DATABASE_URL": "postgresql+asyncpg://cartsnitch:cartsnitch@localhost:5432/cartsnitch",
+    }
+    settings = Settings(**env)
+    assert settings.database_url == "postgresql+asyncpg://cartsnitch:cartsnitch@localhost:5432/cartsnitch"
+
+
+def test_database_url_default():
+    """When neither env var is set, the hardcoded default is used."""
+    settings = Settings()
+    assert settings.database_url == "postgresql+asyncpg://cartsnitch:cartsnitch@localhost:5432/cartsnitch"

api/tests/test_e2e/conftest.py

@@ -10,9 +10,9 @@ from decimal import Decimal
 from uuid import UUID
 
 import pytest
+from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from cartsnitch_api.auth.jwt import decode_token
 from cartsnitch_api.models import (
     Coupon,
     NormalizedProduct,
@@ -126,10 +126,16 @@ async def seed_data(db_engine, auth_headers):
         session.add_all(prices)
         await session.flush()
 
-        # -- Purchases (need the user_id from the registered test user) --
-        token = auth_headers["Authorization"].split(" ")[1]
-        payload = decode_token(token)
-        user_id = UUID(payload["sub"])
+        # -- Get the user_id from the session token in auth_headers --
+        cookie_str = auth_headers.get("Cookie", "")
+        session_token = cookie_str.split("=", 1)[1] if "=" in cookie_str else ""
+
+        result = await session.execute(
+            text("SELECT user_id FROM sessions WHERE token = :token"),
+            {"token": session_token},
+        )
+        row = result.first()
+        user_id = UUID(row[0])
 
         purchase1 = Purchase(
             user_id=user_id,

api/tests/test_e2e/test_auth_validation.py

@@ -1,132 +1,103 @@
-"""E2E: Auth and token validation flows."""
+"""E2E: Auth and session validation flows.
 
-import asyncio
+Registration and login are handled by the Better-Auth service.
+These tests validate session token handling at the API gateway level.
+"""
 
 import pytest
 
-
-@pytest.mark.asyncio
-class TestAuthRegistrationLogin:
-    """Full registration → login → token refresh → profile flow."""
-
-    async def test_full_auth_lifecycle(self, client, db_engine):
-        """Register → login → get profile → refresh → get profile again."""
-        # Register
-        reg = await client.post(
-            "/auth/register",
-            json={
-                "email": "lifecycle@example.com",
-                "password": "securepass123",
-                "display_name": "Lifecycle User",
-            },
-        )
-        assert reg.status_code == 201
-        tokens = reg.json()
-        assert "access_token" in tokens
-        assert "refresh_token" in tokens
-        assert tokens["token_type"] == "bearer"
-        assert tokens["expires_in"] > 0
-
-        headers = {"Authorization": f"Bearer {tokens['access_token']}"}
-
-        # Get profile with access token
-        me = await client.get("/auth/me", headers=headers)
-        assert me.status_code == 200
-        assert me.json()["email"] == "lifecycle@example.com"
-        assert me.json()["display_name"] == "Lifecycle User"
-
-        # Sleep 1s so the new token has a different exp than the registration token
-        await asyncio.sleep(1)
-
-        # Login with same credentials
-        login = await client.post(
-            "/auth/login",
-            json={"email": "lifecycle@example.com", "password": "securepass123"},
-        )
-        assert login.status_code == 200
-        login_tokens = login.json()
-        assert login_tokens["access_token"] != tokens["access_token"]
-
-        # Refresh token
-        refresh = await client.post(
-            "/auth/refresh",
-            json={"refresh_token": tokens["refresh_token"]},
-        )
-        assert refresh.status_code == 200
-        new_tokens = refresh.json()
-        assert new_tokens["access_token"] != tokens["access_token"]
-
-        # Use refreshed token to access profile
-        new_headers = {"Authorization": f"Bearer {new_tokens['access_token']}"}
-        me2 = await client.get("/auth/me", headers=new_headers)
-        assert me2.status_code == 200
-        assert me2.json()["email"] == "lifecycle@example.com"
+from tests.conftest import _create_test_user_and_session
 
 
 @pytest.mark.asyncio
-class TestTokenValidation:
-    """Token edge cases and error responses."""
+class TestSessionValidation:
+    """Session edge cases and error responses."""
 
-    async def test_expired_token_rejected(self, client, db_engine):
-        """Manually craft an expired token and verify rejection."""
-        import uuid
-        from datetime import UTC, datetime, timedelta
-
-        from jose import jwt
-
-        from cartsnitch_api.config import settings
-
-        payload = {
-            "sub": str(uuid.uuid4()),
-            "exp": datetime.now(UTC) - timedelta(minutes=5),
-            "type": "access",
-        }
-        token = jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
-        resp = await client.get("/auth/me", headers={"Authorization": f"Bearer {token}"})
+    async def test_invalid_session_token_rejected(self, client, db_engine):
+        resp = await client.get(
+            "/auth/me",
+            headers={"Cookie": "better-auth.session_token=not-a-real-token"},
+        )
         assert resp.status_code == 401
 
-    async def test_invalid_token_rejected(self, client, db_engine):
-        resp = await client.get("/auth/me", headers={"Authorization": "Bearer not-a-real-token"})
-        assert resp.status_code == 401
-
-    async def test_missing_auth_header(self, client, db_engine):
+    async def test_missing_auth(self, client, db_engine):
         resp = await client.get("/auth/me")
         assert resp.status_code in (401, 403)
 
-    async def test_refresh_token_cannot_access_endpoints(self, client, db_engine):
-        """A refresh token should not work as an access token."""
-        reg = await client.post(
-            "/auth/register",
-            json={
-                "email": "refresh-test@example.com",
-                "password": "securepass123",
-                "display_name": "Refresh Test",
-            },
+    async def test_bearer_token_also_works(self, client, db_engine):
+        """Session tokens passed as Bearer tokens should also be accepted."""
+        _, session_token = await _create_test_user_and_session(
+            client, db_engine, email="bearer@e2e.com", display_name="Bearer E2E"
         )
-        refresh_token = reg.json()["refresh_token"]
-        resp = await client.get("/auth/me", headers={"Authorization": f"Bearer {refresh_token}"})
-        assert resp.status_code == 401
-
-    async def test_deleted_user_token_invalid(self, client, db_engine):
-        """After deleting an account, tokens should no longer work."""
-        reg = await client.post(
-            "/auth/register",
-            json={
-                "email": "delete-me@example.com",
-                "password": "securepass123",
-                "display_name": "Delete Me",
-            },
+        resp = await client.get(
+            "/auth/me",
+            headers={"Authorization": f"Bearer {session_token}"},
         )
-        tokens = reg.json()
-        headers = {"Authorization": f"Bearer {tokens['access_token']}"}
+        assert resp.status_code == 200
+        assert resp.json()["email"] == "bearer@e2e.com"
+
+    async def test_deleted_user_session_returns_not_found(self, client, db_engine):
+        """After deleting a user, their session should result in 404 for profile."""
+        _, session_token = await _create_test_user_and_session(
+            client, db_engine, email="delete-me@e2e.com", display_name="Delete Me"
+        )
+        headers = {"Cookie": f"better-auth.session_token={session_token}"}
 
-        # Delete account
         delete_resp = await client.delete("/auth/me", headers=headers)
         assert delete_resp.status_code == 204
 
-        # Profile should fail
         me = await client.get("/auth/me", headers=headers)
-        assert me.status_code in (401, 404)
+        assert me.status_code == 404
+
+    async def test_expired_session_rejected(self, client, db_engine):
+        """Expired sessions must be rejected."""
+        import secrets
+        import uuid
+        from datetime import UTC, datetime, timedelta
+
+        from sqlalchemy import text
+
+        user_id = str(uuid.uuid4())
+        session_token = secrets.token_urlsafe(32)
+        now = datetime.now(UTC).isoformat()
+        expired = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+
+        async with db_engine.begin() as conn:
+            await conn.execute(
+                text(
+                    "INSERT INTO users (id, email, hashed_password, display_name, email_verified, created_at, updated_at) "
+                    "VALUES (:id, :email, :hp, :dn, :ev, :ca, :ua)"
+                ),
+                {
+                    "id": user_id,
+                    "email": "expired@e2e.com",
+                    "hp": "unused",
+                    "dn": "Expired User",
+                    "ev": False,
+                    "ca": now,
+                    "ua": now,
+                },
+            )
+            await conn.execute(
+                text(
+                    "INSERT INTO sessions (id, token, user_id, expires_at, created_at, updated_at) "
+                    "VALUES (:id, :token, :uid, :ea, :ca, :ua)"
+                ),
+                {
+                    "id": str(uuid.uuid4()),
+                    "token": session_token,
+                    "uid": user_id,
+                    "ea": expired,
+                    "ca": now,
+                    "ua": now,
+                },
+            )
+
+        resp = await client.get(
+            "/auth/me",
+            headers={"Cookie": f"better-auth.session_token={session_token}"},
+        )
+        assert resp.status_code == 401
 
 
 @pytest.mark.asyncio
@@ -154,60 +125,38 @@ class TestAuthProtectedEndpoints:
 class TestCrossUserDataIsolation:
     """Verify that users cannot access other users' data."""
 
-    async def test_user_b_cannot_access_user_a_purchases(self, client, seed_data):
-        """Register a second user and verify they cannot see User A's purchases."""
-        # User A's purchase (from seed_data)
+    async def test_user_b_cannot_access_user_a_purchases(self, client, db_engine, seed_data):
+        """A second user cannot see User A's purchases."""
         purchase_id = str(seed_data["purchases"]["meijer_trip"].id)
 
-        # Register User B
-        reg = await client.post(
-            "/auth/register",
-            json={
-                "email": "userb@example.com",
-                "password": "securepass123",
-                "display_name": "User B",
-            },
+        _, session_token = await _create_test_user_and_session(
+            client, db_engine, email="userb@e2e.com", display_name="User B"
         )
-        assert reg.status_code == 201
-        user_b_headers = {"Authorization": f"Bearer {reg.json()['access_token']}"}
+        user_b_headers = {"Cookie": f"better-auth.session_token={session_token}"}
 
-        # User B tries to access User A's specific purchase
         resp = await client.get(f"/purchases/{purchase_id}", headers=user_b_headers)
         assert resp.status_code in (403, 404), (
             "User B should not be able to access User A's purchase"
         )
 
-    async def test_user_b_purchase_list_is_empty(self, client, seed_data):
-        """A new user should see no purchases (not User A's purchases)."""
-        reg = await client.post(
-            "/auth/register",
-            json={
-                "email": "userc@example.com",
-                "password": "securepass123",
-                "display_name": "User C",
-            },
+    async def test_user_b_purchase_list_is_empty(self, client, db_engine, seed_data):
+        """A new user should see no purchases."""
+        _, session_token = await _create_test_user_and_session(
+            client, db_engine, email="userc@e2e.com", display_name="User C"
         )
-        assert reg.status_code == 201
-        user_c_headers = {"Authorization": f"Bearer {reg.json()['access_token']}"}
+        user_c_headers = {"Cookie": f"better-auth.session_token={session_token}"}
 
         resp = await client.get("/purchases", headers=user_c_headers)
         assert resp.status_code == 200
         assert len(resp.json()) == 0, "New user should have no purchases"
 
-    async def test_user_b_stores_isolated(self, client, seed_data):
+    async def test_user_b_stores_isolated(self, client, db_engine, seed_data):
         """User B's connected stores should be independent from User A."""
-        reg = await client.post(
-            "/auth/register",
-            json={
-                "email": "userd@example.com",
-                "password": "securepass123",
-                "display_name": "User D",
-            },
+        _, session_token = await _create_test_user_and_session(
+            client, db_engine, email="userd@e2e.com", display_name="User D"
         )
-        assert reg.status_code == 201
-        user_d_headers = {"Authorization": f"Bearer {reg.json()['access_token']}"}
+        user_d_headers = {"Cookie": f"better-auth.session_token={session_token}"}
 
-        # User D should have no connected stores
         resp = await client.get("/me/stores", headers=user_d_headers)
         assert resp.status_code == 200
         assert len(resp.json()) == 0, "New user should have no connected stores"

api/tests/test_e2e/test_email_in_address.py

@@ -0,0 +1,61 @@
+"""Tests for GET /api/v1/me/email-in-address endpoint."""
+
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+async def test_get_email_in_address_authenticated(client: AsyncClient, auth_headers: dict):
+    """Authenticated user gets their email-in address."""
+    response = await client.get(
+        "/api/v1/me/email-in-address",
+        headers=auth_headers,
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    assert "email_address" in data
+    assert data["email_address"].startswith("receipts+")
+    assert data["email_address"].endswith("@receipts.cartsnitch.com")
+    assert len(data["email_address"]) > len("receipts+@receipts.cartsnitch.com")
+    assert "instructions" in data
+    assert "Meijer" in data["instructions"]
+    assert "Kroger" in data["instructions"]
+    assert "Target" in data["instructions"]
+
+
+@pytest.mark.asyncio
+async def test_get_email_in_address_unauthenticated(client: AsyncClient):
+    """Unauthenticated request returns 401."""
+    response = await client.get("/api/v1/me/email-in-address")
+    assert response.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_get_email_in_address_invalid_token(client: AsyncClient):
+    """Invalid JWT token returns 401."""
+    response = await client.get(
+        "/api/v1/me/email-in-address",
+        headers={"Authorization": "Bearer invalid-token-xyz"},
+    )
+    assert response.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_email_address_format(client: AsyncClient, auth_headers: dict):
+    """Email address format is receipts+{22-char-urlsafe-token}@receipts.cartsnitch.com."""
+    response = await client.get(
+        "/api/v1/me/email-in-address",
+        headers=auth_headers,
+    )
+
+    assert response.status_code == 200
+    data = response.json()
+    email = data["email_address"]
+    # Format: receipts+<22-char-urlsafe-token>@receipts.cartsnitch.com
+    assert email.startswith("receipts+")
+    assert email.endswith("@receipts.cartsnitch.com")
+    # token_urlsafe(16) produces 22 chars
+    middle = email[len("receipts+") : -len("@receipts.cartsnitch.com")]
+    assert len(middle) == 22
+    assert "@" not in middle

api/tests/test_middleware/test_rate_limit.py

@@ -1,8 +1,10 @@
 """Tests for rate limiting middleware."""
 
+from unittest.mock import MagicMock
+
 import pytest
 
-from cartsnitch_api.middleware.rate_limit import _SlidingWindowCounter
+from cartsnitch_api.middleware.rate_limit import _SlidingWindowCounter, _get_rate_limit_key
 
 
 class TestSlidingWindowCounter:
@@ -53,3 +55,32 @@ async def test_health_skips_rate_limit(client):
     resp = await client.get("/health")
     assert resp.status_code == 200
     assert "x-ratelimit-limit" not in resp.headers
+
+
+class TestGetRateLimitKey:
+    def _make_request(self, auth_header: str = "") -> MagicMock:
+        req = MagicMock()
+        req.url.path = "/purchases"
+        req.headers = {"authorization": auth_header} if auth_header else {}
+        return req
+
+    def test_distinct_tokens_produce_distinct_keys(self):
+        req1 = self._make_request("Bearer token_alpha_12345")
+        req2 = self._make_request("Bearer token_beta_67890")
+        key1, _ = _get_rate_limit_key(req1)
+        key2, _ = _get_rate_limit_key(req2)
+        assert key1 != key2
+
+    def test_same_token_produces_same_key(self):
+        req1 = self._make_request("Bearer same_token_value_abc")
+        req2 = self._make_request("Bearer same_token_value_abc")
+        key1, _ = _get_rate_limit_key(req1)
+        key2, _ = _get_rate_limit_key(req2)
+        assert key1 == key2
+
+    def test_key_does_not_contain_raw_token_suffix(self):
+        raw_token = "my_secret_jwt_token_xyz"
+        req = self._make_request(f"Bearer {raw_token}")
+        key, _ = _get_rate_limit_key(req)
+        assert raw_token[-16:] not in key
+        assert raw_token not in key

api/tests/test_openapi.py

@@ -6,13 +6,14 @@ from httpx import ASGITransport, AsyncClient
 from cartsnitch_api.main import app
 
 EXPECTED_ROUTES = [
-    # Auth (6)
+    # Auth (7)
     ("post", "/auth/register"),
     ("post", "/auth/login"),
     ("post", "/auth/refresh"),
     ("get", "/auth/me"),
     ("patch", "/auth/me"),
     ("delete", "/auth/me"),
+    ("get", "/auth/me/email-in-address"),
     # Stores (4)
     ("get", "/stores"),
     ("get", "/me/stores"),
@@ -89,4 +90,4 @@ async def test_route_count():
                 if method in ("get", "post", "put", "delete", "patch"):
                     count += 1
 
-        assert count == 33, f"Expected 33 routes, found {count}"
+        assert count == 34, f"Expected 34 routes, found {count}"

api/tests/test_routes/test_purchases.py

@@ -1,26 +1,25 @@
 """Integration tests for purchase endpoints."""
 
+import secrets
 import uuid
-from datetime import date
+from datetime import UTC, date, datetime, timedelta
 from decimal import Decimal
 
 import pytest
+from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from cartsnitch_api.auth.jwt import create_access_token
 from cartsnitch_api.models import Purchase, PurchaseItem, Store, User
 
 
 @pytest.fixture
 async def purchase_data(db_engine):
-    """Seed a user, store, purchase, and items."""
+    """Seed a user, store, purchase, items, and a valid session."""
     factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
     async with factory() as session:
-        from cartsnitch_api.auth.passwords import hash_password
-
         user = User(
             email="buyer@example.com",
-            hashed_password=hash_password("testpass123"),
+            hashed_password="not-used-with-better-auth",
             display_name="Buyer",
         )
         store = Store(name="Kroger", slug="kroger")
@@ -50,13 +49,33 @@ async def purchase_data(db_engine):
         session.add(item)
         await session.commit()
 
-        token = create_access_token(user.id)
-        return {
-            "user": user,
-            "store": store,
-            "purchase": purchase,
-            "headers": {"Authorization": f"Bearer {token}"},
-        }
+    # Create a session token directly in the sessions table
+    session_token = secrets.token_urlsafe(32)
+    now = datetime.now(UTC).isoformat()
+    expires = (datetime.now(UTC) + timedelta(days=7)).isoformat()
+
+    async with db_engine.begin() as conn:
+        await conn.execute(
+            text(
+                "INSERT INTO sessions (id, token, user_id, expires_at, created_at, updated_at) "
+                "VALUES (:id, :token, :user_id, :expires_at, :created_at, :updated_at)"
+            ),
+            {
+                "id": str(uuid.uuid4()),
+                "token": session_token,
+                "user_id": str(user.id),
+                "expires_at": expires,
+                "created_at": now,
+                "updated_at": now,
+            },
+        )
+
+    return {
+        "user": user,
+        "store": store,
+        "purchase": purchase,
+        "headers": {"Cookie": f"better-auth.session_token={session_token}"},
+    }
 
 
 @pytest.mark.asyncio

auth/.env.example

@@ -0,0 +1,11 @@
+# Required: Generate with `openssl rand -base64 32`
+BETTER_AUTH_SECRET=change-me-in-production-min-32-chars!!
+
+# Base URL of the auth service
+BETTER_AUTH_URL=http://localhost:3001
+
+# Shared PostgreSQL database
+DATABASE_URL=postgresql://cartsnitch:cartsnitch@localhost:5432/cartsnitch
+
+# Port the auth service listens on
+PORT=3001

auth/Dockerfile

@@ -0,0 +1,17 @@
+FROM node:22-alpine AS builder
+WORKDIR /app
+COPY package.json package-lock.json* ./
+RUN npm ci
+COPY tsconfig.json ./
+COPY src/ src/
+RUN npm run build
+
+FROM node:22-alpine
+WORKDIR /app
+ENV NODE_ENV=production
+COPY package.json package-lock.json* ./
+RUN npm ci --omit=dev
+COPY --from=builder /app/dist/ dist/
+USER 101
+EXPOSE 3001
+CMD ["node", "dist/index.js"]

auth/package.json

@@ -0,0 +1,24 @@
+{
+  "name": "@cartsnitch/auth",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "dev": "tsx watch src/index.ts",
+    "build": "tsc",
+    "start": "node dist/index.js",
+    "generate": "npx @better-auth/cli generate"
+  },
+  "dependencies": {
+    "better-auth": "^1.2.0",
+    "pg": "^8.13.0",
+    "bcrypt": "^5.1.1"
+  },
+  "devDependencies": {
+    "@types/node": "^22.0.0",
+    "@types/pg": "^8.11.0",
+    "@types/bcrypt": "^5.0.2",
+    "tsx": "^4.19.0",
+    "typescript": "^5.7.0"
+  }
+}

auth/src/auth.ts

@@ -0,0 +1,106 @@
+import { betterAuth } from "better-auth";
+import bcrypt from "bcrypt";
+import pg from "pg";
+
+const { Pool } = pg;
+
+const secret = process.env.BETTER_AUTH_SECRET;
+if (!secret) {
+  throw new Error("BETTER_AUTH_SECRET environment variable is required");
+}
+
+const databaseUrl = process.env.DATABASE_URL;
+if (!databaseUrl) {
+  console.warn(
+    "WARNING: DATABASE_URL is not set — using default localhost connection. " +
+    "Set DATABASE_URL for production deployments."
+  );
+}
+
+const pool = new Pool({
+  connectionString: databaseUrl ?? "postgresql://cartsnitch:cartsnitch@localhost:5432/cartsnitch",
+});
+
+export const auth = betterAuth({
+  database: pool,
+  basePath: "/auth",
+  secret,
+  baseURL: process.env.BETTER_AUTH_URL ?? "http://localhost:3001",
+
+  emailAndPassword: {
+    enabled: true,
+    minPasswordLength: 8,
+    maxPasswordLength: 128,
+    password: {
+      hash: async (password: string) => {
+        return bcrypt.hash(password, 10);
+      },
+      verify: async (data: { hash: string; password: string }) => {
+        return bcrypt.compare(data.password, data.hash);
+      },
+    },
+  },
+
+  session: {
+    modelName: "sessions",
+    fields: {
+      userId: "user_id",
+      expiresAt: "expires_at",
+      ipAddress: "ip_address",
+      userAgent: "user_agent",
+      createdAt: "created_at",
+      updatedAt: "updated_at",
+    },
+    expiresIn: 60 * 60 * 24 * 7, // 7 days
+    updateAge: 60 * 60 * 24, // refresh after 1 day
+    cookieCache: {
+      enabled: true,
+      maxAge: 5 * 60, // 5-minute cookie cache
+    },
+  },
+
+  user: {
+    modelName: "users",
+    fields: {
+      name: "display_name",
+      emailVerified: "email_verified",
+      image: "image",
+      createdAt: "created_at",
+      updatedAt: "updated_at",
+    },
+  },
+
+  account: {
+    modelName: "accounts",
+    fields: {
+      userId: "user_id",
+      accountId: "account_id",
+      providerId: "provider_id",
+      accessToken: "access_token",
+      refreshToken: "refresh_token",
+      accessTokenExpiresAt: "access_token_expires_at",
+      refreshTokenExpiresAt: "refresh_token_expires_at",
+      idToken: "id_token",
+      createdAt: "created_at",
+      updatedAt: "updated_at",
+    },
+  },
+
+  verification: {
+    modelName: "verifications",
+    fields: {
+      expiresAt: "expires_at",
+      createdAt: "created_at",
+      updatedAt: "updated_at",
+    },
+  },
+
+  trustedOrigins: [
+    "http://localhost:3000",
+    "http://localhost:5173",
+    "https://cartsnitch.com",
+    "https://cartsnitch.farh.net",
+    "https://cartsnitch.dev.farh.net",
+    "https://cartsnitch.uat.farh.net",
+  ],
+});

auth/src/index.ts

@@ -0,0 +1,23 @@
+import { createServer } from "node:http";
+import { toNodeHandler } from "better-auth/node";
+import { auth } from "./auth.js";
+
+const port = parseInt(process.env.PORT ?? "3001", 10);
+
+const handler = toNodeHandler(auth);
+
+const server = createServer(async (req, res) => {
+  // Health check
+  if (req.url === "/health" && req.method === "GET") {
+    res.writeHead(200, { "Content-Type": "application/json" });
+    res.end(JSON.stringify({ status: "ok" }));
+    return;
+  }
+
+  // All /auth/* routes handled by Better-Auth
+  await handler(req, res);
+});
+
+server.listen(port, "0.0.0.0", () => {
+  console.log(`CartSnitch auth service listening on port ${port}`);
+});

auth/tsconfig.json

@@ -0,0 +1,16 @@
+{
+  "compilerOptions": {
+    "target": "ES2022",
+    "module": "ESNext",
+    "moduleResolution": "bundler",
+    "outDir": "dist",
+    "rootDir": "src",
+    "strict": true,
+    "esModuleInterop": true,
+    "skipLibCheck": true,
+    "declaration": true,
+    "resolveJsonModule": true
+  },
+  "include": ["src"],
+  "exclude": ["node_modules", "dist"]
+}

common/README.md

@@ -0,0 +1,28 @@
+# CartSnitch Common
+
+Shared models, schemas, and utilities for CartSnitch services.
+
+## Test Users
+
+The following users are seeded by `cartsnitch-seed` and can be used for local development and UAT.
+
+| Email | Password | Display Name | Notes |
+|---|---|---|---|
+| `uat@cartsnitch.com` | `CartSnitch-UAT-2026!` | UAT Tester | Primary UAT account. Use for regression testing in the CartSnitch frontend. Created by the seed runner via Better-Auth's bcrypt path — credentials work against the live auth service. Idempotent; re-running the seed skips this user if it already exists. |
+
+### Running the Seed
+
+```bash
+# Install with seed dependencies
+pip install -e "cartsnitch-common[seed]"
+
+# Run (requires CARTSNITCH_DATABASE_URL_SYNC)
+CARTSNITCH_DATABASE_URL_SYNC=postgresql://user:pass@localhost:5432/cartsnitch \
+  cartsnitch-seed
+```
+
+### Architecture
+
+- **Models** live in `src/cartsnitch_common/models/`
+- **Alembic migrations** run via the `api` service (`api/alembic/`)
+- **Seed runner** runs via `cartsnitch-seed` (installed as a package entry point)

common/alembic/env.py

@@ -14,7 +14,7 @@ if config.config_file_name is not None:
 
 db_url = os.environ.get("CARTSNITCH_DATABASE_URL_SYNC")
 if db_url:
-    config.set_main_option("sqlalchemy.url", db_url)
+    config.set_main_option("sqlalchemy.url", db_url.replace("%", "%%"))
 
 target_metadata = Base.metadata
 

common/alembic/versions/001_add_email_inbound_token.py

@@ -0,0 +1,37 @@
+"""Add email_inbound_token to users.
+
+Revision ID: 001_add_email_inbound_token
+Revises:
+Create Date: 2026-04-02
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "001_add_email_inbound_token"
+down_revision: str | None = None
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("users", sa.Column("email_inbound_token", sa.String(22), nullable=True))
+    op.create_unique_constraint("uq_users_email_inbound_token", "users", ["email_inbound_token"])
+
+    # Backfill existing users with generated tokens (PostgreSQL)
+    op.execute(
+        "UPDATE users SET email_inbound_token = "
+        "substring(replace(gen_random_uuid()::text, '-', ''), 1, 22) "
+        "WHERE email_inbound_token IS NULL"
+    )
+
+    # Alter to non-nullable
+    op.alter_column("users", "email_inbound_token", nullable=False)
+
+
+def downgrade() -> None:
+    op.drop_constraint("uq_users_email_inbound_token", "users", type_="unique")
+    op.drop_column("users", "email_inbound_token")

common/pyproject.toml

@@ -27,6 +27,7 @@ dev = [
 ]
 seed = [
     "faker>=33.0,<34.0",
+    "bcrypt>=4.0,<6.0",
 ]
 
 [project.scripts]

common/src/cartsnitch_common/models/user.py

@@ -1,10 +1,11 @@
 """User and UserStoreAccount models."""
 
+import secrets
 import uuid
 from datetime import datetime
 from typing import TYPE_CHECKING
 
-from sqlalchemy import JSON, DateTime, ForeignKey, String, UniqueConstraint
+from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, String, Text, UniqueConstraint, text
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from cartsnitch_common.constants import AccountStatus
@@ -21,8 +22,19 @@ class User(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     __tablename__ = "users"
 
     email: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
-    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
+    email_inbound_token: Mapped[str] = mapped_column(
+        String(22),
+        nullable=False,
+        unique=True,
+        default=lambda: secrets.token_urlsafe(16),
+        server_default=text(
+            "replace(replace(trim(trailing '=' from encode(gen_random_bytes(16), 'base64')), '+', '-'), '/', '_')"
+        ),
+    )
+    hashed_password: Mapped[str | None] = mapped_column(String(255), nullable=True)
     display_name: Mapped[str | None] = mapped_column(String(100))
+    email_verified: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default="false")
+    image: Mapped[str | None] = mapped_column(Text, nullable=True)
 
     # Relationships
     store_accounts: Mapped[list["UserStoreAccount"]] = relationship(back_populates="user")

common/src/cartsnitch_common/schemas/user.py

@@ -20,6 +20,7 @@ class UserRead(BaseModel):
     id: uuid.UUID
     email: str
     display_name: str | None
+    email_inbound_token: str
     created_at: datetime
     updated_at: datetime
 

common/src/cartsnitch_common/seed/runner.py

@@ -2,8 +2,10 @@
 
 import random
 import time
+import uuid
 from typing import Any
 
+import bcrypt
 from faker import Faker
 from sqlalchemy import text
 from sqlalchemy.orm import Session
@@ -184,6 +186,65 @@ def run_seed(
 
         session.commit()
 
+        _seed_uat_user(session)
+
     elapsed = time.monotonic() - t0
     _log("")
     _log(f"Seed complete in {elapsed:.1f}s")
+
+
+# ---------------------------------------------------------------------------
+# UAT seed user
+# ---------------------------------------------------------------------------
+
+UAT_EMAIL = "uat@cartsnitch.com"
+UAT_PASSWORD = "CartSnitch-UAT-2026!"
+UAT_DISPLAY_NAME = "UAT Tester"
+UAT_USER_ID = uuid.UUID("00000000-0000-0000-0000-000000000001")
+
+
+def _seed_uat_user(session: Session) -> None:
+    """Insert or verify the dedicated UAT test user.
+
+    The user is created via Better-Auth's bcrypt hashing path so credentials
+    work against the live auth service. Idempotent — skips if the user already
+    exists.
+    """
+    existing = session.execute(
+        text("SELECT id FROM users WHERE email = :email"),
+        {"email": UAT_EMAIL},
+    ).fetchone()
+
+    if existing is not None:
+        _log(f"UAT user {UAT_EMAIL} already exists — skipping")
+        return
+
+    password_hash = bcrypt.hashpw(UAT_PASSWORD.encode(), bcrypt.gensalt()).decode()
+
+    session.execute(
+        text(
+            "INSERT INTO users (id, email, hashed_password, display_name, email_verified, created_at, updated_at) "
+            "VALUES (:id, :email, :hashed_password, :display_name, true, now(), now())"
+        ),
+        {
+            "id": str(UAT_USER_ID),
+            "email": UAT_EMAIL,
+            "hashed_password": password_hash,
+            "display_name": UAT_DISPLAY_NAME,
+        },
+    )
+
+    session.execute(
+        text(
+            "INSERT INTO accounts (id, user_id, account_id, provider_id, password, created_at, updated_at) "
+            "VALUES (gen_random_uuid()::text, :user_id, :account_id, 'credential', :password, now(), now())"
+        ),
+        {
+            "user_id": str(UAT_USER_ID),
+            "account_id": str(UAT_USER_ID),
+            "password": password_hash,
+        },
+    )
+
+    session.commit()
+    _log(f"UAT user {UAT_EMAIL} created")

common/tests/test_models.py

@@ -147,6 +147,40 @@ class TestStoreLocationModel:
         assert loc.lat == pytest.approx(42.2808)
 
 
+class TestUserModel:
+    def test_email_inbound_token_auto_populated(self, session):
+        user = User(
+            id=uuid.uuid4(),
+            email="token_test@example.com",
+            hashed_password="hashed",
+            created_at=datetime.now(UTC),
+            updated_at=datetime.now(UTC),
+        )
+        session.add(user)
+        session.commit()
+        assert user.email_inbound_token is not None
+        assert len(user.email_inbound_token) == 22
+
+    def test_email_inbound_token_unique(self, session):
+        user1 = User(
+            id=uuid.uuid4(),
+            email="user1@example.com",
+            hashed_password="hashed",
+            created_at=datetime.now(UTC),
+            updated_at=datetime.now(UTC),
+        )
+        user2 = User(
+            id=uuid.uuid4(),
+            email="user2@example.com",
+            hashed_password="hashed",
+            created_at=datetime.now(UTC),
+            updated_at=datetime.now(UTC),
+        )
+        session.add_all([user1, user2])
+        session.commit()
+        assert user1.email_inbound_token != user2.email_inbound_token
+
+
 class TestUserStoreAccountModel:
     def test_account_status_enum(self, session):
         user = User(

docs/uat-runbook.md

@@ -0,0 +1,151 @@
+# CartSnitch UAT Runbook v1
+
+**Version:** 1.0
+**Author:** Savannah Savings, CTO
+**Date:** 2026-03-30
+**Effective:** Immediately upon Phase 1 completion
+
+---
+
+## 1. Defect Severity Classification
+
+Every defect discovered during UAT **must** be classified by severity and priority before triage.
+
+### Severity Levels
+
+| Severity | Definition | Examples |
+|----------|-----------|----------|
+| **S1 — Critical** | Blocks all users from completing a core journey. System is down, data is lost, or security is breached. | Login page crashes for all users; purchase data deleted; auth tokens exposed in response |
+| **S2 — High** | Blocks a major user flow for a significant portion of users. Core feature is broken but workarounds may exist. | Registration fails for email addresses with `+` character; price alerts never trigger; store comparison shows wrong prices |
+| **S3 — Medium** | Feature is degraded but usable. User can complete the journey with friction. | Date formatting shows raw ISO string instead of friendly date; slow page load (>5s) on product detail; search results not sorted correctly |
+| **S4 — Low** | Cosmetic issue, minor UI inconsistency, or edge case with minimal user impact. | Button text truncated on narrow screens; extra whitespace in footer; tooltip shows on hover but not on focus |
+
+### Priority Levels
+
+Priority determines **when** the defect must be fixed. Priority is set by the CTO based on severity, business impact, and sprint capacity.
+
+| Priority | SLA | When to Use |
+|----------|-----|------------|
+| **P0 — Fix Now** | Triage within 1 hour, fix deployed within 4 hours | S1 defects, any security vulnerability, data integrity issues |
+| **P1 — Fix This Sprint** | Triage within 4 hours, fix in current sprint | S2 defects blocking upcoming release, S1 defects with viable workaround |
+| **P2 — Fix Next Sprint** | Triage within 24 hours, scheduled for next sprint | S3 defects, S2 defects with easy workarounds |
+| **P3 — Backlog** | Triage within 48 hours, prioritized against backlog | S4 defects, minor improvements, nice-to-haves |
+
+### Defect Report Template
+
+Every defect filed during UAT must include:
+
+```
+**Title:** [Short description]
+**Severity:** S1/S2/S3/S4
+**Priority:** P0/P1/P2/P3 (set by CTO at triage)
+**Journey:** [Which user journey — J1 through J10]
+**Environment:** [Dev / Prod, deployed image tag]
+**Steps to Reproduce:**
+1. Navigate to ...
+2. Click ...
+3. Enter ...
+**Expected Result:** ...
+**Actual Result:** ...
+**Screenshots/Logs:** [Attach or link]
+**Browser/Device:** [e.g., Chromium 124, mobile viewport 390x844]
+```
+
+---
+
+## 2. UAT Entry Criteria
+
+UAT **must not begin** until ALL of the following are satisfied. Checkout Charlie verifies these before opening the UAT gate.
+
+| # | Criterion | Verified By |
+|---|-----------|------------|
+| E1 | CI pipeline passes on the merged commit (lint, type-check, unit tests, build) | GitHub Actions (automated) |
+| E2 | Docker image is built and pushed to GHCR with a CalVer tag | GitHub Actions (automated) |
+| E3 | Dev environment is deployed and accessible at `cartsnitch.dev.farh.net` | Flux reconciliation + health check |
+| E4 | All Playwright E2E tests pass in CI | GitHub Actions (automated) |
+| E5 | No open S1/S2 defects from previous UAT cycle | Checkout Charlie (manual check) |
+| E6 | PR has been reviewed and approved by QA (Checkout Charlie) and CTO (Savannah Savings) | GitHub PR approvals |
+| E7 | PR has been merged to main by CEO (Coupon Carl) | GitHub merge event |
+| E8 | Acceptance criteria for the feature/change are documented in the Paperclip issue | Checkout Charlie (manual check) |
+
+**If any entry criterion is not met**, UAT is blocked. Checkout Charlie must comment on the Paperclip issue specifying which criteria failed and assign back to the responsible party.
+
+---
+
+## 3. UAT Exit Criteria
+
+UAT is **complete** only when ALL of the following are satisfied. Rollback Rhonda verifies these before signing off.
+
+| # | Criterion | Verified By |
+|---|-----------|------------|
+| X1 | All 10 critical user journeys (J1-J10) have been executed | Rollback Rhonda (full regression) |
+| X2 | Zero open S1 (Critical) defects | Defect tracker |
+| X3 | Zero open S2 (High) defects, OR CTO has granted a documented exception | Defect tracker + CTO sign-off |
+| X4 | All S3/S4 defects are logged and triaged (not necessarily fixed) | Defect tracker |
+| X5 | 100% test execution rate -- every test case was run, none skipped | Rollback Rhonda's UAT report |
+| X6 | Accessibility scan (axe-core) reports zero critical violations | Automated in E2E suite |
+| X7 | Lighthouse performance score >= 50, accessibility score >= 90 | Lighthouse CI |
+| X8 | Written sign-off from Rollback Rhonda confirming all criteria met | Paperclip comment on issue |
+
+**If any exit criterion is not met**, the release is blocked. Rollback Rhonda must:
+1. File defects for all failures using the Defect Report Template above.
+2. Comment on the Paperclip issue specifying which exit criteria failed.
+3. Assign back to CTO for triage and redistribution.
+
+---
+
+## 4. UAT Execution Procedure
+
+### 4.1 Pre-UAT (Checkout Charlie)
+
+1. Verify all entry criteria (E1-E8) are met.
+2. Comment on the Paperclip issue: "UAT gate open -- all entry criteria verified."
+3. Assign to Rollback Rhonda with status todo.
+
+### 4.2 UAT Execution (Rollback Rhonda)
+
+1. **Full regression run** -- execute ALL 10 user journeys against cartsnitch.dev.farh.net. No partial runs. No exceptions.
+2. For each journey, verify:
+   - All interactive elements respond correctly (buttons, forms, links, toggles)
+   - State transitions are correct (auth state, data mutations, navigation)
+   - Error states are handled gracefully (invalid input, network failures)
+   - Accessibility scan passes (axe-core integrated in Playwright)
+3. Log results for each journey: PASS / FAIL with details.
+4. File defects immediately for any failures.
+5. Complete the UAT report with execution results.
+
+### 4.3 Post-UAT Sign-Off
+
+1. If all exit criteria (X1-X8) are met:
+   - Rollback Rhonda posts sign-off comment: "UAT PASSED -- all exit criteria met."
+   - Production promotion is automated via Flux on UAT pass.
+2. If any exit criterion fails:
+   - Rollback Rhonda posts failure comment with specific failures.
+   - CTO triages defects and redistributes to engineers.
+   - After fixes are merged, UAT restarts from 4.1 (full cycle).
+
+---
+
+## 5. Critical User Journeys Reference
+
+| ID | Journey | Key Interactions |
+|----|---------|-----------------|
+| J1 | Registration -> Login -> Dashboard | Form submission, auth state, redirect |
+| J2 | Login -> Browse Products -> View Detail -> Price Chart | Search, navigation, data visualization |
+| J3 | Login -> Purchases -> Purchase Detail -> Product Link | List navigation, detail view, cross-linking |
+| J4 | Login -> Connect Store Account -> Verify Connection | OAuth flow, external integration |
+| J5 | Login -> Create Price Alert -> View -> Delete Alert | CRUD operations, confirmation dialogs |
+| J6 | Login -> Browse Coupons -> Copy Code | Clipboard interaction, toast feedback |
+| J7 | Login -> Settings -> Toggle Preferences -> Sign Out | Checkbox toggles, theme switch, session termination |
+| J8 | Login -> Store Comparison -> Compare Prices | Data comparison, sorting, price display |
+| J9 | Forgot Password Flow | Email input, validation, redirect |
+| J10 | Unauth Access -> Redirect to Login | Route protection, redirect behavior |
+
+---
+
+## 6. Revision History
+
+| Version | Date | Author | Changes |
+|---------|------|--------|---------|
+| 1.0 | 2026-03-30 | Savannah Savings | Initial runbook -- defect taxonomy, entry/exit criteria, execution procedure |
+

e2e/fixtures.ts

@@ -0,0 +1,99 @@
+import { test as base, expect, type Page } from "@playwright/test";
+import AxeBuilder from "@axe-core/playwright";
+
+export const test = base.extend<{ axeCheck: void }>({
+  axeCheck: [async ({ page }, use) => {
+    await use();
+    const results = await new AxeBuilder({ page }).analyze();
+    expect(results.violations).toEqual([]);
+  }, { auto: true }],
+});
+
+export { expect } from "@playwright/test";
+
+const MOCK_USER_ID = "mock_user_123";
+const MOCK_SESSION_ID = "mock_session_456";
+
+function mockAuthRoutes(page: Page, authenticated = false) {
+  page.route(/.*\/auth\/sign-up\/email.*/, async (route) => {
+    await route.fulfill({
+      status: 200,
+      contentType: "application/json",
+      body: JSON.stringify({
+        token: null,
+        user: {
+          id: MOCK_USER_ID,
+          email: "mock@cartsnitch.test",
+          name: "Mock User",
+          emailVerified: true,
+          createdAt: new Date().toISOString(),
+          updatedAt: new Date().toISOString(),
+        },
+      }),
+    });
+  });
+
+  page.route(/.*\/auth\/sign-in\/email.*/, async (route) => {
+    await route.fulfill({
+      status: 200,
+      contentType: "application/json",
+      body: JSON.stringify({
+        redirect: false,
+        token: "mock_token_123",
+        user: {
+          id: MOCK_USER_ID,
+          email: "mock@cartsnitch.test",
+          name: "Mock User",
+          emailVerified: true,
+          createdAt: new Date().toISOString(),
+          updatedAt: new Date().toISOString(),
+        },
+      }),
+    });
+  });
+
+  page.route(/.*\/auth\/get-session.*/, async (route) => {
+    if (authenticated) {
+      await route.fulfill({
+        status: 200,
+        contentType: "application/json",
+        body: JSON.stringify({
+          session: {
+            id: MOCK_SESSION_ID,
+            expiresAt: new Date(Date.now() + 7 * 24 * 60 * 60 * 1000).toISOString(),
+            createdAt: new Date().toISOString(),
+            updatedAt: new Date().toISOString(),
+            ipAddress: null,
+            userAgent: null,
+          },
+          user: {
+            id: MOCK_USER_ID,
+            email: "mock@cartsnitch.test",
+            name: "Mock User",
+            emailVerified: true,
+            createdAt: new Date().toISOString(),
+            updatedAt: new Date().toISOString(),
+          },
+        }),
+      });
+    } else {
+      await route.fulfill({
+        status: 401,
+        contentType: "application/json",
+        body: JSON.stringify({ error: "Unauthorized" }),
+      });
+    }
+  });
+}
+
+export function mockSessionPending(page: Page) {
+  page.route(/.*\/auth\/session.*/, async (route) => {
+    await route.fulfill({
+      status: 401,
+      contentType: "application/json",
+      body: JSON.stringify({ error: "Unauthorized" }),
+    });
+  });
+}
+
+export { mockAuthRoutes };

e2e/journeys/j1-registration-login.spec.ts

@@ -0,0 +1,43 @@
+import { test, expect } from '@playwright/test';
+import { mockAuthRoutes } from '../fixtures';
+
+const uniqueEmail = () => `betty+e2e-${Date.now()}@cartsnitch.test`;
+
+test.describe('J1: Registration and Login', () => {
+  test('can register a new account and see check your email screen', async ({ page }) => {
+    mockAuthRoutes(page, true);
+    await page.goto('/register');
+    await page.fill('[placeholder="Full Name"]', 'Betty Tester');
+    await page.fill('[placeholder="Email"]', uniqueEmail());
+    await page.fill('[placeholder="Password (min. 8 characters)"]', 'TestPass123!');
+    await page.click('button[type="submit"]');
+
+    await expect(page.getByRole('heading', { name: /check your email/i })).toBeVisible();
+  });
+
+  test('shows validation error when registration fields are empty', async ({ page }) => {
+    await page.goto('/register');
+    await page.click('button[type="submit"]');
+
+    await expect(page.locator('.bg-red-50')).toContainText('Please fill in all fields');
+  });
+
+  test('can navigate from register to login', async ({ page }) => {
+    await page.goto('/register');
+    await page.getByRole('link', { name: /sign in/i }).click();
+
+    await expect(page).toHaveURL(/\/login/);
+    await expect(page.getByRole('heading', { name: /cartsnitch/i })).toBeVisible();
+  });
+
+  test('can sign in with credentials and land on dashboard', async ({ page }) => {
+    mockAuthRoutes(page, true);
+    await page.goto('/login');
+    await page.fill('[placeholder="Email"]', 'test@cartsnitch.test');
+    await page.fill('[placeholder="Password"]', 'TestPass123!');
+    await page.click('button[type="submit"]');
+
+    await expect(page).toHaveURL('http://localhost:5173/');
+  });
+
+});

e2e/journeys/j8-unauth-access.spec.ts

@@ -0,0 +1,47 @@
+import { test, expect } from '@playwright/test';
+import { mockAuthRoutes } from '../fixtures';
+
+test.describe('J8: Unauthenticated Access', () => {
+  test('redirects /dashboard (/) to /login when not authenticated', async ({ page }) => {
+    mockAuthRoutes(page, false);
+    await page.context().clearCookies();
+    await page.goto('/');
+
+    await expect(page).toHaveURL(/\/login/);
+    await expect(page.getByRole('heading', { name: /cartsnitch/i })).toBeVisible();
+  });
+
+  test('redirects /purchases to /login when not authenticated', async ({ page }) => {
+    mockAuthRoutes(page, false);
+    await page.context().clearCookies();
+    await page.goto('/purchases');
+
+    await expect(page).toHaveURL(/\/login/);
+    await expect(page.getByRole('heading', { name: /cartsnitch/i })).toBeVisible();
+  });
+
+  test('redirects /products to /login when not authenticated', async ({ page }) => {
+    mockAuthRoutes(page, false);
+    await page.context().clearCookies();
+    await page.goto('/products');
+
+    await expect(page).toHaveURL(/\/login/);
+    await expect(page.getByRole('heading', { name: /cartsnitch/i })).toBeVisible();
+  });
+
+  test('redirects /coupons to /login when not authenticated', async ({ page }) => {
+    mockAuthRoutes(page, false);
+    await page.context().clearCookies();
+    await page.goto('/coupons');
+
+    await expect(page).toHaveURL(/\/login/);
+    await expect(page.getByRole('heading', { name: /cartsnitch/i })).toBeVisible();
+  });
+
+  test('shows loading spinner while auth session is pending', async ({ page }) => {
+    mockAuthRoutes(page, false);
+    await page.context().clearCookies();
+    await page.goto('/purchases');
+    await expect(page).toHaveURL(/\/login/, { timeout: 10_000 });
+  });
+});

e2e/smoke.spec.ts

@@ -0,0 +1,9 @@
+import { test, expect, mockAuthRoutes } from './fixtures';
+
+test('app loads', async ({ page }) => {
+  mockAuthRoutes(page, false);
+  await page.goto('/');
+  // Unauthenticated users are redirected to /login
+  await expect(page).toHaveURL(/\/login/);
+  await expect(page.getByRole('heading', { name: /CartSnitch/i })).toBeVisible();
+});

lighthouserc.json

@@ -0,0 +1,24 @@
+{
+  "ci": {
+    "collect": {
+      "staticDistDir": "./dist",
+      "url": ["http://localhost:4173/"],
+      "numberOfRuns": 1,
+      "settings": {
+        "chromeFlags": ["--headless=new", "--no-sandbox", "--disable-gpu", "--disable-dev-shm-usage"],
+        "skipAudits": ["bf-cache"],
+        "disableFullPageScreenshot": true
+      }
+    },
+    "assert": {
+      "assertions": {
+        "categories:performance": ["warn", { "minScore": 0.7 }],
+        "categories:accessibility": ["error", { "minScore": 0.9 }],
+        "categories:best-practices": ["warn", { "minScore": 0.8 }]
+      }
+    },
+    "upload": {
+      "target": "temporary-public-storage"
+    }
+  }
+}

nginx.conf

@@ -9,6 +9,12 @@ server {
     gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xml+rss text/javascript image/svg+xml;
     gzip_min_length 256;
 
+    # Security headers
+    add_header X-Frame-Options "SAMEORIGIN" always;
+    add_header X-Content-Type-Options "nosniff" always;
+    add_header Referrer-Policy "strict-origin-when-cross-origin" always;
+    add_header Content-Security-Policy "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self'; connect-src 'self' https://*.cartsnitch.com https://*.farh.net; frame-ancestors 'self'" always;
+
     # Health endpoint for K8s probes
     location /health {
         access_log off;

package.json

@@ -9,10 +9,13 @@
     "lint": "eslint .",
     "preview": "vite preview",
     "test": "NODE_ENV=test vitest run",
-    "test:watch": "NODE_ENV=test vitest"
+    "test:watch": "NODE_ENV=test vitest",
+    "test:e2e": "npx playwright test"
   },
   "dependencies": {
     "@tanstack/react-query": "^5.0.0",
+    "better-auth": "^1.2.0",
+    "picomatch": "4.0.4",
     "react": "^18.3.1",
     "react-dom": "^18.3.1",
     "react-router-dom": "^7.0.0",
@@ -20,24 +23,36 @@
     "zustand": "^5.0.0"
   },
   "devDependencies": {
+    "@axe-core/playwright": "^4.10.0",
     "@eslint/js": "^9.39.4",
+    "@playwright/test": "^1.58.2",
     "@tailwindcss/vite": "^4.0.0",
     "@testing-library/jest-dom": "^6.6.3",
     "@testing-library/react": "^16.3.2",
     "@types/node": "^24.12.0",
     "@types/react": "^18.3.28",
     "@types/react-dom": "^18.3.7",
-    "@vitejs/plugin-react": "^4.5.2",
+    "@vitejs/plugin-react": "^4.7.0",
     "eslint": "^9.39.4",
     "eslint-plugin-react-hooks": "^7.0.1",
     "eslint-plugin-react-refresh": "^0.5.2",
     "globals": "^17.4.0",
     "jsdom": "^25.0.1",
+    "msw": "^2.12.14",
+    "playwright": "^1.58.2",
     "tailwindcss": "^4.0.0",
     "typescript": "^5.7.3",
     "typescript-eslint": "^8.56.1",
     "vite": "^6.3.5",
     "vite-plugin-pwa": "^0.21.2",
     "vitest": "^3.2.4"
+  },
+  "overrides": {
+    "@rollup/pluginutils": "5.3.0",
+    "flatted": "^3.4.2",
+    "serialize-javascript": "7.0.5",
+    "brace-expansion": ">=1.1.13",
+    "lodash": ">=4.17.24",
+    "minimatch": "^10.2.4"
   }
 }

playwright.config.ts

@@ -0,0 +1,22 @@
+import { defineConfig, devices } from '@playwright/test';
+
+export default defineConfig({
+  testDir: './e2e',
+  projects: [
+    {
+      name: 'chromium',
+      use: { ...devices['Desktop Chrome'] },
+    },
+  ],
+  webServer: {
+    command: 'npm run dev',
+    url: 'http://localhost:5173',
+    reuseExistingServer: !process.env.CI,
+    env: {
+      VITE_MOCK_AUTH: 'true',
+    },
+  },
+  use: {
+    baseURL: 'http://localhost:5173',
+  },
+});

public/robots.txt

@@ -0,0 +1,4 @@
+User-agent: *
+Allow: /
+
+Sitemap: https://cartsnitch.com/sitemap.xml

receiptwitness/.github/workflows/ci.yml

@@ -1,168 +0,0 @@
-name: CI
-
-on:
-  push:
-    branches: [main]
-  pull_request:
-    branches: [main]
-
-concurrency:
-  group: ci-${{ github.ref }}
-  cancel-in-progress: true
-
-permissions:
-  contents: write
-  packages: write
-
-env:
-  REGISTRY: ghcr.io
-  IMAGE_NAME: cartsnitch/receiptwitness
-
-jobs:
-  lint:
-    runs-on: runners-cartsnitch
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-python@v5
-        with:
-          python-version: "3.12"
-          cache: pip
-      - name: Install cartsnitch-common from GitHub
-        run: pip install "cartsnitch-common @ git+https://github.com/cartsnitch/common.git@76685ed0384103228cd670b477b967e7752ebe6b"
-      - run: pip install ruff
-      - name: Ruff lint
-        run: ruff check .
-      - name: Ruff format check
-        run: ruff format --check .
-
-  typecheck:
-    runs-on: runners-cartsnitch
-    continue-on-error: true
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-python@v5
-        with:
-          python-version: "3.12"
-          cache: pip
-      - name: Install cartsnitch-common from GitHub
-        run: pip install "cartsnitch-common @ git+https://github.com/cartsnitch/common.git@76685ed0384103228cd670b477b967e7752ebe6b"
-      - run: pip install -e ".[dev]" mypy
-      - name: Type check
-        run: mypy src/receiptwitness
-
-  test:
-    runs-on: runners-cartsnitch
-    services:
-      postgres:
-        image: postgres:15-alpine
-        credentials:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-        env:
-          POSTGRES_USER: cartsnitch
-          POSTGRES_PASSWORD: cartsnitch_test
-          POSTGRES_DB: cartsnitch_test
-        ports:
-          - 5432:5432
-        options: >-
-          --health-cmd pg_isready
-          --health-interval 10s
-          --health-timeout 5s
-          --health-retries 5
-      redis:
-        image: redis:7-alpine
-        credentials:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-        ports:
-          - 6379:6379
-        options: >-
-          --health-cmd "redis-cli ping"
-          --health-interval 10s
-          --health-timeout 5s
-          --health-retries 5
-    env:
-      DATABASE_URL: postgresql://cartsnitch:cartsnitch_test@localhost:5432/cartsnitch_test
-      REDIS_URL: redis://localhost:6379/0
-      ENCRYPTION_KEY: dGVzdC1lbmNyeXB0aW9uLWtleS0xMjM0NTY3ODk=
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-python@v5
-        with:
-          python-version: "3.12"
-          cache: pip
-      - name: Install cartsnitch-common from GitHub
-        run: pip install "cartsnitch-common @ git+https://github.com/cartsnitch/common.git@76685ed0384103228cd670b477b967e7752ebe6b"
-      - run: pip install -e ".[dev]"
-      - name: Install Playwright browsers
-        run: playwright install chromium --with-deps
-      - name: Run tests
-        run: pytest --tb=short -q
-
-  build-and-push:
-    runs-on: runners-cartsnitch
-    needs: [lint, test]
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-
-      - name: Generate CalVer tag
-        id: calver
-        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
-        run: |
-          DATE_TAG=$(date -u +%Y.%m.%d)
-          EXISTING=$(git tag -l "v${DATE_TAG}*" | sort -V | tail -1)
-          if [ -z "$EXISTING" ]; then
-            VERSION="$DATE_TAG"
-          elif [ "$EXISTING" = "v${DATE_TAG}" ]; then
-            VERSION="${DATE_TAG}.2"
-          else
-            BUILD_NUM=$(echo "$EXISTING" | sed "s/v${DATE_TAG}\.//")
-            VERSION="${DATE_TAG}.$((BUILD_NUM + 1))"
-          fi
-          echo "version=$VERSION" >> "$GITHUB_OUTPUT"
-          echo "CalVer tag: $VERSION"
-
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
-
-      - name: Log in to Docker Hub
-        uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-
-      - name: Log in to GHCR
-        uses: docker/login-action@v3
-        with:
-          registry: ${{ env.REGISTRY }}
-          username: ${{ github.actor }}
-          password: ${{ secrets.GITHUB_TOKEN }}
-
-      - name: Extract metadata
-        id: meta
-        uses: docker/metadata-action@v5
-        with:
-          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
-          tags: |
-            type=sha,prefix=sha-
-            type=raw,value=${{ steps.calver.outputs.version }},enable=${{ github.ref == 'refs/heads/main' }}
-            type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
-
-      - name: Build and push Docker image
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          push: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }}
-          tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
-          target: prod
-          cache-from: type=gha
-          cache-to: type=gha,mode=max
-
-      - name: Create git tag
-        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
-        run: |
-          git tag "v${{ steps.calver.outputs.version }}"
-          git push origin "v${{ steps.calver.outputs.version }}"

receiptwitness/Dockerfile

@@ -3,24 +3,21 @@ FROM python:3.12-slim AS build
 
 WORKDIR /app
 
-# git is required to install cartsnitch-common from GitHub; build-essential and
-# libpq-dev are needed to compile any C-extension wheels (e.g. psycopg2 fallback)
+# build-essential and libpq-dev are needed to compile any C-extension wheels
+# (e.g. psycopg2 fallback).  No git needed — common/ is copied from the repo root.
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    git \
     libpq-dev \
     build-essential \
     && rm -rf /var/lib/apt/lists/*
 
-COPY pyproject.toml ./
-COPY src/ ./src/
+# Build context is the repo root.  These paths are relative to the root.
+COPY receiptwitness/pyproject.toml ./
+COPY receiptwitness/src/ ./src/
+COPY common/ ./common/
 
-# cartsnitch-common is not on PyPI — install it directly from GitHub, then
-# install the rest of the package dependencies in a single resolver pass so
-# pip can satisfy the cartsnitch-common>=0.1.0 constraint declared in
-# pyproject.toml without hitting PyPI for it.
-RUN pip install --no-cache-dir --prefix=/install \
-    "cartsnitch-common @ git+https://github.com/cartsnitch/common.git@76685ed0384103228cd670b477b967e7752ebe6b" \
-    .
+# Install from the local common/ (cartsnitch-common>=0.1.0 in pyproject.toml
+# will be satisfied by the local package) then install receiptwitness itself.
+RUN pip install --no-cache-dir --prefix=/install ./common/ .
 
 # Stage 2: Production image with Playwright + Chromium
 FROM python:3.12-slim AS prod
@@ -51,7 +48,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 RUN adduser --system --group --uid 1000 app
 
 COPY --from=build /install /usr/local
-COPY src/ ./src/
+COPY receiptwitness/src/ ./src/
 
 # Install Playwright Chromium browser (runs as root; /opt/playwright is world-readable)
 RUN PLAYWRIGHT_BROWSERS_PATH=/opt/playwright playwright install chromium

receiptwitness/pyproject.toml

@@ -14,11 +14,13 @@ dependencies = [
     "cryptography>=42.0,<44.0",
     "fastapi>=0.115,<1.0",
     "uvicorn[standard]>=0.30,<1.0",
+    "beautifulsoup4>=4.12,<5.0",
     "redis>=5.0,<6.0",
     "pydantic>=2.0,<3.0",
     "pydantic-settings>=2.0,<3.0",
     "sqlalchemy[asyncio]>=2.0,<3.0",
     "asyncpg>=0.29,<1.0",
+    "resend>=2.0",
 ]
 
 [project.optional-dependencies]
@@ -27,6 +29,9 @@ dev = [
     "pytest-asyncio>=0.23",
     "ruff>=0.3",
     "pytest-cov>=5.0",
+    "fakeredis[aioredis]>=2.20",
+    "httpx>=0.27",
+    "python-multipart>=0.0.9",
 ]
 
 [tool.hatch.build.targets.wheel]

receiptwitness/src/receiptwitness/api/routes.py

@@ -1,9 +1,65 @@
 """Internal API routes for triggering scrapes and checking status."""
 
-from fastapi import APIRouter
+import hashlib
+import hmac
+import re
+import time
+
+from fastapi import APIRouter, HTTPException, Request
+
+from receiptwitness.config import settings
+from receiptwitness.queue.email import EmailJob, enqueue_email, get_redis
 
 router = APIRouter()
 
+TOKEN_PATTERN = re.compile(r"receipts\+([A-Za-z0-9_-]+)@")
+
+
+def verify_mailgun_signature(token: str, timestamp: str, signature: str) -> bool:
+    """Verify Mailgun webhook signature."""
+    try:
+        ts = int(timestamp)
+    except (ValueError, TypeError):
+        return False
+    if abs(time.time() - ts) > 300:  # 5 min freshness
+        return False
+    key = settings.mailgun_webhook_signing_key.encode()
+    hmac_digest = hmac.new(key, f"{timestamp}{token}".encode(), hashlib.sha256).hexdigest()
+    return hmac.compare_digest(signature, hmac_digest)
+
+
+@router.post("/inbound/email")
+async def receive_inbound_email(request: Request):
+    form = await request.form()
+    # 1. Verify Mailgun signature
+    token = str(form.get("token", ""))
+    timestamp = str(form.get("timestamp", ""))
+    signature = str(form.get("signature", ""))
+    if not verify_mailgun_signature(token, timestamp, signature):
+        raise HTTPException(status_code=406, detail="Invalid signature")
+    # 2. Extract account token from recipient
+    recipient = str(form.get("recipient", ""))
+    match = TOKEN_PATTERN.search(recipient)
+    if not match:
+        raise HTTPException(status_code=406, detail="Invalid recipient")
+    account_token = match.group(1)
+    # 3. Enqueue — worker resolves token -> user_id
+    body_html_val = form.get("body-html")
+    body_plain_val = form.get("body-plain")
+    job = EmailJob(
+        user_id=account_token,
+        sender=str(form.get("sender", "")),
+        recipient=recipient,
+        subject=str(form.get("subject", "")),
+        body_html=str(body_html_val) if body_html_val is not None else None,
+        body_plain=str(body_plain_val) if body_plain_val is not None else None,
+        received_at=str(form.get("timestamp", "")),
+        message_id=str(form.get("Message-Id", "")),
+    )
+    client = await get_redis()
+    await enqueue_email(client, job)
+    return {"status": "queued"}
+
 
 @router.get("/health")
 async def health():

receiptwitness/src/receiptwitness/config.py

@@ -1,8 +1,12 @@
 """Service-specific configuration for ReceiptWitness."""
 
+from pydantic import model_validator
 from pydantic_settings import BaseSettings
 
 
+_PLACEHOLDER_VALUES = {"change-me-in-production"}
+
+
 class ReceiptWitnessSettings(BaseSettings):
     model_config = {"env_prefix": "RW_"}
 
@@ -22,5 +26,42 @@ class ReceiptWitnessSettings(BaseSettings):
     headless: bool = True
     browser_timeout_ms: int = 60000
 
+    # Email notifications (Resend)
+    resend_api_key: str = ""
+    notification_email_from: str = "notifications@cartsnitch.com"
+    notifications_enabled: bool = False
 
-settings = ReceiptWitnessSettings()
+    # Mailgun inbound email webhook
+    mailgun_webhook_signing_key: str = ""
+
+    @model_validator(mode="after")
+    def validate_required_vars(self):
+        errors = []
+        if not self.session_encryption_key or self.session_encryption_key in _PLACEHOLDER_VALUES:
+            errors.append(
+                "RW_SESSION_ENCRYPTION_KEY must be set to a secure value. "
+                'Generate one with: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"'
+            )
+        if self.notifications_enabled and not self.resend_api_key:
+            errors.append(
+                "RW_RESEND_API_KEY must be set when RW_NOTIFICATIONS_ENABLED=true. "
+                "Get an API key from https://resend.com/api-keys"
+            )
+        if errors:
+            raise ValueError(
+                "ReceiptWitness startup failed — missing required config:\n"
+                + "\n".join(f"  - {e}" for e in errors)
+            )
+        return self
+
+
+class _LazySettings:
+    _instance: ReceiptWitnessSettings | None = None
+
+    def __getattr__(self, name: str):
+        if _LazySettings._instance is None:
+            _LazySettings._instance = ReceiptWitnessSettings()
+        return getattr(_LazySettings._instance, name)
+
+
+settings = _LazySettings()

receiptwitness/src/receiptwitness/events.py

@@ -2,12 +2,17 @@
 
 import json
 import logging
+import uuid
 from datetime import UTC, datetime
 from decimal import Decimal
 
 import redis.asyncio as aioredis
+from cartsnitch_common.database import get_async_session_factory
+from cartsnitch_common.models.user import User
+from sqlalchemy import select
 
 from receiptwitness.config import settings
+from receiptwitness.notifications.email import send_receipt_notification
 
 logger = logging.getLogger(__name__)
 
@@ -39,6 +44,36 @@ async def get_redis_client() -> aioredis.Redis:
     return aioredis.Redis(connection_pool=_get_pool())
 
 
+async def _send_notification_for_event(payload: dict) -> None:
+    """Look up user email and send receipt notification. Silently skips on error."""
+    try:
+        user_uuid = uuid.UUID(payload["user_id"])
+    except (ValueError, KeyError):
+        logger.warning("Invalid user_id in event payload: %s", payload.get("user_id"))
+        return
+
+    try:
+        session_factory = get_async_session_factory(settings.database_url)
+        async with session_factory() as session:
+            result = await session.execute(select(User.email).where(User.id == user_uuid))
+            row = result.scalar_one_or_none()
+            if not row:
+                logger.warning("User %s not found for notification", user_uuid)
+                return
+            user_email = row
+    except Exception:
+        logger.exception("Failed to look up user email for notification")
+        return
+
+    await send_receipt_notification(
+        user_email=user_email,
+        store_name=payload["store_slug"],
+        item_count=payload["item_count"],
+        total=payload["total"],
+        purchase_date=payload["purchase_date"],
+    )
+
+
 async def publish_receipt_ingested(
     user_id: str,
     store_slug: str,
@@ -48,18 +83,19 @@ async def publish_receipt_ingested(
     total: Decimal | float,
 ) -> None:
     """Publish a cartsnitch.receipts.ingested event after successful ingestion."""
+    payload = {
+        "user_id": user_id,
+        "store_slug": store_slug,
+        "purchase_id": purchase_id,
+        "purchase_date": purchase_date,
+        "item_count": item_count,
+        "total": float(total) if isinstance(total, Decimal) else total,
+    }
     event = {
         "event_type": CHANNEL_RECEIPTS_INGESTED,
         "timestamp": datetime.now(UTC).isoformat(),
         "service": "receiptwitness",
-        "payload": {
-            "user_id": user_id,
-            "store_slug": store_slug,
-            "purchase_id": purchase_id,
-            "purchase_date": purchase_date,
-            "item_count": item_count,
-            "total": float(total) if isinstance(total, Decimal) else total,
-        },
+        "payload": payload,
     }
 
     try:
@@ -73,3 +109,5 @@ async def publish_receipt_ingested(
     except aioredis.ConnectionError:
         logger.error("Failed to publish event — Redis/DragonflyDB connection error")
         raise
+    else:
+        await _send_notification_for_event(payload)

receiptwitness/src/receiptwitness/notifications/email.py

@@ -0,0 +1,45 @@
+"""Email notifications via Resend."""
+
+import asyncio
+import html
+import logging
+
+import resend
+
+from receiptwitness.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+async def send_receipt_notification(
+    user_email: str,
+    store_name: str,
+    item_count: int,
+    total: float,
+    purchase_date: str,
+) -> None:
+    """Send receipt ingestion confirmation email via Resend."""
+    if not settings.notifications_enabled or not settings.resend_api_key:
+        logger.debug("Notifications disabled — skipping email send")
+        return
+
+    resend.api_key = settings.resend_api_key
+    store_name_safe = html.escape(store_name)
+    purchase_date_safe = html.escape(purchase_date)
+    try:
+        await asyncio.to_thread(
+            resend.Emails.send,
+            {
+                "from": settings.notification_email_from,
+                "to": [user_email],
+                "subject": f"Receipt processed: {store_name} - ${total:.2f}",
+                "html": (
+                    f"<p>Your receipt from <strong>{store_name_safe}</strong> on "
+                    f"{purchase_date_safe} has been processed.</p>"
+                    f"<p>{item_count} items, total: ${total:.2f}</p>"
+                ),
+            },
+        )
+        logger.info("Receipt notification sent to %s", user_email)
+    except Exception:
+        logger.exception("Failed to send receipt notification to %s", user_email)

receiptwitness/src/receiptwitness/parsers/email/__init__.py

@@ -0,0 +1 @@
+"""Email receipt parsers for retailer email receipts."""

receiptwitness/src/receiptwitness/parsers/email/base.py

@@ -0,0 +1,32 @@
+"""Base interface for email receipt parsers."""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+
+
+@dataclass
+class EmailReceipt:
+    """Raw email data before parsing."""
+
+    sender: str
+    recipient: str
+    subject: str
+    body_html: str | None = None
+    body_plain: str | None = None
+    received_at: str | None = None
+    raw_headers: dict = field(default_factory=dict)
+
+
+class BaseEmailParser(ABC):
+    """All retailer email parsers implement this interface."""
+
+    @abstractmethod
+    def can_parse(self, email: EmailReceipt) -> bool:
+        """Return True if this parser handles this email."""
+        ...
+
+    @abstractmethod
+    def parse(self, email: EmailReceipt) -> dict:
+        """Parse email into a dict matching PurchaseCreate schema fields.
+        Must include an items list matching PurchaseItemCreate fields."""
+        ...

receiptwitness/src/receiptwitness/parsers/email/detector.py

@@ -0,0 +1,25 @@
+"""Detect which retailer sent a receipt email."""
+
+import re
+
+from receiptwitness.parsers.email.base import EmailReceipt
+
+RETAILER_PATTERNS: dict[str, list[str]] = {
+    "meijer": [r"@meijer\.com$", r"@email\.meijer\.com$"],
+    "kroger": [r"@kroger\.com$", r"@email\.kroger\.com$"],
+    "target": [r"@target\.com$", r"@email\.target\.com$"],
+}
+
+
+def detect_retailer(email: EmailReceipt) -> str | None:
+    """Return retailer slug or None if unrecognized."""
+    sender = email.sender.lower().strip()
+    # Extract email from "Name <email>" format
+    match = re.search(r"<([^>]+)>", sender)
+    if match:
+        sender = match.group(1)
+    for retailer, patterns in RETAILER_PATTERNS.items():
+        for pattern in patterns:
+            if re.search(pattern, sender):
+                return retailer
+    return None

receiptwitness/src/receiptwitness/parsers/email/kroger.py

@@ -0,0 +1,157 @@
+"""Kroger email receipt parser."""
+
+import logging
+import re
+from datetime import datetime
+from decimal import Decimal, InvalidOperation
+
+from bs4 import BeautifulSoup
+
+from receiptwitness.parsers.email.base import BaseEmailParser, EmailReceipt
+
+logger = logging.getLogger(__name__)
+
+
+def _to_decimal(value: str | float | int | None, default: Decimal = Decimal("0")) -> Decimal:
+    """Safely convert a value to Decimal."""
+    if value is None:
+        return default
+    try:
+        return Decimal(str(value).replace("$", "").replace(",", "").strip())
+    except (InvalidOperation, ValueError):
+        return default
+
+
+def _extract_total(body: str) -> Decimal:
+    """Extract the transaction total from email body."""
+    patterns = [
+        r"Total[:\s]*\$?([0-9,]+\.[0-9]{2})",
+        r"Amount[:\s]*\$?([0-9,]+\.[0-9]{2})",
+        r"Grand\s+Total[:\s]*\$?([0-9,]+\.[0-9]{2})",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, body, re.IGNORECASE)
+        if match:
+            return _to_decimal(match.group(1))
+    return Decimal("0")
+
+
+def _extract_receipt_id(body: str) -> str | None:
+    """Extract receipt ID / transaction ID from HTML body.
+
+    Strips HTML tags first so that whitespace between delimiters and values
+    (e.g. from ``</strong> KR-2026-0315-4829`` -> `` KR-2026-0315-4829``)
+    is normalized and the pattern can match cleanly.
+    """
+    stripped = re.sub(r"<[^>]+>", "", body)
+    patterns = [
+        r"Receipt\s*#[:\s]*([A-Z0-9-]+)",
+        r"Transaction\s*#[:\s]*([A-Z0-9-]+)",
+        r"Order\s*#[:\s]*([A-Z0-9-]+)",
+        r"Confirmation\s*#[:\s]*([A-Z0-9-]+)",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, stripped, re.IGNORECASE)
+        if match:
+            return match.group(1)
+    return None
+
+
+def _extract_date(body: str) -> str:
+    """Extract purchase date from email body. Returns ISO date string or empty string."""
+    patterns = [
+        r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})",
+        r"([A-Z][a-z]{2}\s+\d{1,2},?\s+\d{4})",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, body)
+        if match:
+            raw = match.group(1)
+            try:
+                dt = datetime.strptime(raw.replace(",", ""), "%b %d %Y")
+                return dt.strftime("%Y-%m-%d")
+            except ValueError:
+                pass
+            try:
+                for fmt in ("%m/%d/%Y", "%m/%d/%y", "%d/%m/%Y", "%d/%m/%y"):
+                    try:
+                        dt = datetime.strptime(raw, fmt)
+                        return dt.strftime("%Y-%m-%d")
+                    except ValueError:
+                        continue
+            except Exception:
+                pass
+    return ""
+
+
+def _extract_items_soup(body: str) -> list[dict]:
+    """Extract line items from HTML email body using BeautifulSoup."""
+    items = []
+    try:
+        soup = BeautifulSoup(body, "html.parser")
+        text = soup.get_text(separator="\n", strip=True)
+        # Strip HTML tags from raw body to normalize whitespace
+        stripped = re.sub(r"<[^>]+>", " ", body)
+        stripped = re.sub(r"\s+", " ", stripped)
+        skip_prefixes = (
+            "Subtotal",
+            "Tax",
+            "Total",
+            "Kroger",
+            "Target",
+            "Date",
+            "Receipt",
+            "Order",
+            "Transaction",
+            "Confirmation",
+            "Thank",
+            "Questions",
+            "Keep",
+            "Receipt",
+        )
+        for line in text.split("\n"):
+            line = line.strip()
+            if not line or line.startswith(skip_prefixes):
+                continue
+            # Match lines like "Product Name $9.99"
+            match = re.match(r"(.+?)\s+\$([0-9]+\.[0-9]{2})\s*$", line)
+            if match:
+                name = match.group(1).strip()
+                price = _to_decimal(match.group(2))
+                if len(name) > 2 and price > 0:
+                    items.append(
+                        {
+                            "product_name_raw": name,
+                            "quantity": Decimal("1"),
+                            "unit_price": price,
+                            "extended_price": price,
+                        }
+                    )
+    except Exception:
+        pass
+    return items[:20]
+
+
+class KrogerEmailParser(BaseEmailParser):
+    """Parse Kroger email receipts (digital receipts via kroger.com)."""
+
+    KROGER_KEYWORDS = ("kroger", "kroger.com", "plus")
+
+    def can_parse(self, email: EmailReceipt) -> bool:
+        sender = (email.sender or "").lower()
+        body = (email.body_html or email.body_plain or "").lower()
+        return any(kw in sender or kw in body for kw in self.KROGER_KEYWORDS)
+
+    def parse(self, email: EmailReceipt) -> dict:
+        body = (email.body_html or email.body_plain or "").strip()
+        total = _extract_total(body)
+        receipt_id = _extract_receipt_id(body) or ""
+        purchase_date = _extract_date(body)
+        items = _extract_items_soup(body)
+
+        return {
+            "receipt_id": receipt_id,
+            "purchase_date": purchase_date,
+            "total": total,
+            "items": items,
+        }

receiptwitness/src/receiptwitness/parsers/email/meijer.py

@@ -0,0 +1,259 @@
+"""Parse Meijer digital receipt emails into structured purchase data."""
+
+import re
+from decimal import Decimal, InvalidOperation
+
+from bs4 import BeautifulSoup
+from bs4.element import Tag
+
+from receiptwitness.parsers.email.base import BaseEmailParser, EmailReceipt
+
+
+def _to_decimal(value, default: str = "0") -> Decimal:
+    """Safely convert a value to Decimal."""
+    if value is None:
+        return Decimal(default)
+    try:
+        return Decimal(str(value).replace("$", "").replace(",", "").strip())
+    except (InvalidOperation, ValueError, TypeError):
+        return Decimal(default)
+
+
+def _extract_receipt_id(soup: BeautifulSoup, subject: str | None) -> str | None:
+    """Extract receipt/transaction ID from subject or body."""
+    if subject:
+        match = re.search(r"TXN[-\s]\d{4}[-\s]\d{4}[-\s]\d+", subject)
+        if match:
+            return match.group(0).replace(" ", "-")
+    # Fallback: look in body
+    text = soup.get_text()
+    match = re.search(r"TXN[-\s]\d{4}[-\s]\d{4}[-\s]\d+", text)
+    if match:
+        return match.group(0).replace(" ", "-")
+    return None
+
+
+def _extract_purchase_date(soup: BeautifulSoup, subject: str | None) -> str | None:
+    """Extract purchase date from subject or body."""
+    text = soup.get_text()
+
+    # Try ISO format first: YYYY-MM-DD
+    match = re.search(r"(\d{4})-(\d{2})-(\d{2})", text)
+    if match:
+        return f"{match.group(1)}-{match.group(2)}-{match.group(3)}"
+
+    # Try written format: March 15, 2026
+    match = re.search(r"([A-Za-z]+)\s+(\d{1,2}),?\s+(\d{4})", text)
+    if match:
+        month_str = match.group(1).lower()
+        day = match.group(2)
+        year = match.group(3)
+        month_map = {
+            "january": "01",
+            "february": "02",
+            "march": "03",
+            "april": "04",
+            "may": "05",
+            "june": "06",
+            "july": "07",
+            "august": "08",
+            "september": "09",
+            "october": "10",
+            "november": "11",
+            "december": "12",
+        }
+        month = month_map.get(month_str)
+        if month:
+            return f"{year}-{month}-{day.zfill(2)}"
+
+    # MM/DD/YYYY
+    match = re.search(r"(\d{1,2})/(\d{1,2})/(\d{4})", text)
+    if match:
+        return f"{match.group(3)}-{match.group(1).zfill(2)}-{match.group(2).zfill(2)}"
+
+    return None
+
+
+def _extract_store_info(soup: BeautifulSoup) -> dict:
+    """Extract store name and number from the email body."""
+    store_info: dict = {}
+
+    # Look for store number in header
+    store_num_match = re.search(r"Meijer\s+Store\s+#?(\d+)", soup.get_text(), re.IGNORECASE)
+    if store_num_match:
+        store_info["store_number"] = store_num_match.group(1)
+
+    return store_info
+
+
+def _extract_items(table: Tag | None) -> list[dict]:
+    """Extract line items from the items table."""
+    items: list[dict] = []
+    if not table:
+        return items
+
+    rows = table.find_all("tr")
+    for row in rows:
+        cells = row.find_all("td")
+        if len(cells) < 3:
+            continue
+
+        name_cell = cells[0].get_text(strip=True)
+        qty_cell = cells[1].get_text(strip=True)
+        price_cell = cells[2].get_text(strip=True)
+
+        if not name_cell or name_cell.lower() in ("item", "description"):
+            continue
+
+        # Skip subtotal/tax/total/savings rows
+        if any(
+            label in name_cell.lower()
+            for label in ("subtotal", "tax", "total", "savings", "grand total")
+        ):
+            continue
+
+        try:
+            quantity = Decimal(qty_cell)
+        except (InvalidOperation, ValueError, TypeError):
+            quantity = Decimal("1")
+
+        price_str = price_cell.replace("$", "").replace(",", "").strip()
+        try:
+            unit_price = Decimal(price_str)
+        except (InvalidOperation, ValueError, TypeError):
+            unit_price = Decimal("0")
+
+        extended_price = unit_price  # Default to unit price; no qty column in fixture
+
+        items.append(
+            {
+                "product_name_raw": name_cell,
+                "quantity": quantity,
+                "unit_price": unit_price,
+                "extended_price": extended_price,
+            }
+        )
+
+    return items
+
+
+def _extract_totals_plain(text: str) -> dict:
+    """Extract totals from plain text (no HTML)."""
+    totals: dict = {
+        "subtotal": None,
+        "tax": None,
+        "total": None,
+        "savings_total": None,
+    }
+
+    match = re.search(r"\bSubtotal\b[:\s$]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+    if match:
+        totals["subtotal"] = _to_decimal(match.group(1))
+
+    match = re.search(r"\bTax\b[:\s$]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+    if match:
+        totals["tax"] = _to_decimal(match.group(1))
+
+    grand_total_match = re.search(r"Grand\s+Total\b[:\s$]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+    if grand_total_match:
+        totals["total"] = _to_decimal(grand_total_match.group(1))
+
+    savings_match = re.search(r"\bSavings\b[:\s$\-]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+    if savings_match:
+        totals["savings_total"] = _to_decimal(savings_match.group(1))
+
+    if totals["total"] is None:
+        total_match = re.search(r"\bTotal\b[:\s$]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+        if total_match:
+            totals["total"] = _to_decimal(total_match.group(1))
+
+    return totals
+
+
+def _extract_totals(soup: BeautifulSoup) -> dict:
+    """Extract subtotal, tax, total, and savings from the totals section."""
+    text = soup.get_text()
+
+    totals: dict = {
+        "subtotal": None,
+        "tax": None,
+        "total": None,
+        "savings_total": None,
+    }
+
+    # Subtotal — use word boundary to avoid matching "Subtotal" with "Total"
+    match = re.search(r"\bSubtotal\b[:\s$]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+    if match:
+        totals["subtotal"] = _to_decimal(match.group(1))
+
+    # Tax
+    match = re.search(r"\bTax\b[:\s$]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+    if match:
+        totals["tax"] = _to_decimal(match.group(1))
+
+    # Grand Total (before plain "Total" to avoid matching "Subtotal")
+    grand_total_match = re.search(r"Grand\s+Total\b[:\s$]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+    if grand_total_match:
+        totals["total"] = _to_decimal(grand_total_match.group(1))
+
+    # Savings — allow any combination of whitespace/$- around the number
+    savings_match = re.search(r"\bSavings\b[:\s$\-]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+    if savings_match:
+        totals["savings_total"] = _to_decimal(savings_match.group(1))
+
+    # Plain "Total" only if Grand Total wasn't found
+    if totals["total"] is None:
+        total_match = re.search(r"\bTotal\b[:\s$]*([0-9,]+\.?\d*)", text, re.IGNORECASE)
+        if total_match:
+            totals["total"] = _to_decimal(total_match.group(1))
+
+    return totals
+
+
+class MeijerEmailParser(BaseEmailParser):
+    """Parse Meijer digital receipt emails forwarded by users."""
+
+    def can_parse(self, email: EmailReceipt) -> bool:
+        sender = email.sender.lower().strip()
+        # Extract email from "Name <email>" format
+        match = re.search(r"<([^>]+)>", sender)
+        if match:
+            sender = match.group(1)
+        return "meijer" in sender
+
+    def parse(self, email: EmailReceipt) -> dict:
+        body_html = email.body_html
+        body_plain = email.body_plain or ""
+        body = body_html or body_plain
+        soup = BeautifulSoup(body, "html.parser")
+
+        receipt_id = _extract_receipt_id(soup, email.subject)
+        purchase_date = _extract_purchase_date(soup, email.subject)
+        _ = _extract_store_info(soup)
+
+        # Find the items table — look for one with Item/Qty/Price headers
+        table = None
+        for tbl in soup.find_all("table"):
+            headers = tbl.find_all("th")
+            header_texts = [h.get_text(strip=True).lower() for h in headers]
+            if any("item" in h or "qty" in h or "price" in h for h in header_texts):
+                table = tbl
+                break
+
+        items = _extract_items(table)
+
+        # Extract totals from HTML; fall back to plain text if no HTML
+        if body_html:
+            totals = _extract_totals(soup)
+        else:
+            totals = _extract_totals_plain(body_plain)
+
+        return {
+            "receipt_id": receipt_id or "",
+            "purchase_date": purchase_date or "",
+            "total": totals["total"] or Decimal("0"),
+            "subtotal": totals["subtotal"],
+            "tax": totals["tax"],
+            "savings_total": totals["savings_total"],
+            "items": items,
+        }

receiptwitness/src/receiptwitness/parsers/email/target.py

@@ -0,0 +1,156 @@
+"""Target email receipt parser."""
+
+import logging
+import re
+from datetime import datetime
+from decimal import Decimal, InvalidOperation
+
+from bs4 import BeautifulSoup
+
+from receiptwitness.parsers.email.base import BaseEmailParser, EmailReceipt
+
+logger = logging.getLogger(__name__)
+
+
+def _to_decimal(value: str | float | int | None, default: Decimal = Decimal("0")) -> Decimal:
+    """Safely convert a value to Decimal."""
+    if value is None:
+        return default
+    try:
+        return Decimal(str(value).replace("$", "").replace(",", "").strip())
+    except (InvalidOperation, ValueError):
+        return default
+
+
+def _extract_total(body: str) -> Decimal:
+    """Extract the transaction total from email body."""
+    patterns = [
+        r"Total[:\s]*\$?([0-9,]+\.[0-9]{2})",
+        r"Amount[:\s]*\$?([0-9,]+\.[0-9]{2})",
+        r"Grand\s+Total[:\s]*\$?([0-9,]+\.[0-9]{2})",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, body, re.IGNORECASE)
+        if match:
+            return _to_decimal(match.group(1))
+    return Decimal("0")
+
+
+def _extract_receipt_id(body: str) -> str | None:
+    """Extract receipt ID / transaction ID from HTML body.
+
+    Strips HTML tags first so that whitespace between delimiters and values
+    (e.g. from ``</strong> TGT-2026-0318-9124`` -> `` TGT-2026-0318-9124``)
+    is normalized and the pattern can match cleanly.
+    """
+    stripped = re.sub(r"<[^>]+>", "", body)
+    patterns = [
+        r"Receipt\s*#[:\s]*([A-Z0-9-]+)",
+        r"Order\s*#[:\s]*([A-Z0-9-]+)",
+        r"Confirmation\s*#[:\s]*([A-Z0-9-]+)",
+        r"Target\s+Order\s*#[:\s]*([A-Z0-9-]+)",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, stripped, re.IGNORECASE)
+        if match:
+            return match.group(1)
+    return None
+
+
+def _extract_date(body: str) -> str:
+    """Extract purchase date from email body. Returns ISO date string or empty string."""
+    patterns = [
+        r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})",
+        r"([A-Z][a-z]{2}\s+\d{1,2},?\s+\d{4})",
+    ]
+    for pattern in patterns:
+        match = re.search(pattern, body)
+        if match:
+            raw = match.group(1)
+            try:
+                dt = datetime.strptime(raw.replace(",", ""), "%b %d %Y")
+                return dt.strftime("%Y-%m-%d")
+            except ValueError:
+                pass
+            try:
+                for fmt in ("%m/%d/%Y", "%m/%d/%y", "%d/%m/%Y", "%d/%m/%y"):
+                    try:
+                        dt = datetime.strptime(raw, fmt)
+                        return dt.strftime("%Y-%m-%d")
+                    except ValueError:
+                        continue
+            except Exception:
+                pass
+    return ""
+
+
+def _extract_items_soup(body: str) -> list[dict]:
+    """Extract line items from HTML email body using BeautifulSoup."""
+    items = []
+    try:
+        soup = BeautifulSoup(body, "html.parser")
+        text = soup.get_text(separator="\n", strip=True)
+        for line in text.split("\n"):
+            line = line.strip()
+            if not line or line.startswith(
+                (
+                    "Subtotal",
+                    "Tax",
+                    "Total",
+                    "Target",
+                    "Kroger",
+                    "Date",
+                    "Receipt",
+                    "Order",
+                    "Transaction",
+                    "Confirmation",
+                    "Thank",
+                    "Questions",
+                    "Keep",
+                    "Receipt",
+                    "Store",
+                )
+            ):
+                continue
+            # Match lines like "Product Name $9.99"
+            match = re.match(r"(.+?)\s+\$([0-9]+\.[0-9]{2})\s*$", line)
+            if match:
+                name = match.group(1).strip()
+                price = _to_decimal(match.group(2))
+                if len(name) > 2 and price > 0:
+                    items.append(
+                        {
+                            "product_name_raw": name,
+                            "quantity": Decimal("1"),
+                            "unit_price": price,
+                            "extended_price": price,
+                        }
+                    )
+    except Exception:
+        pass
+    return items[:20]
+
+
+class TargetEmailParser(BaseEmailParser):
+    """Parse Target email receipts (Circle order confirmations)."""
+
+    TARGET_KEYWORDS = ("target.com", "targetnow", "circle", "target")
+
+    def can_parse(self, email: EmailReceipt) -> bool:
+        sender = (email.sender or "").lower()
+        body = (email.body_html or email.body_plain or "").lower()
+        return any(kw in sender or kw in body for kw in self.TARGET_KEYWORDS)
+
+    def parse(self, email: EmailReceipt) -> dict:
+        body = (email.body_html or email.body_plain or "").strip()
+        total = _extract_total(body)
+        receipt_id = _extract_receipt_id(body) or ""
+        purchase_date = _extract_date(body)
+        items = _extract_items_soup(body)
+
+        return {
+            "receipt_id": receipt_id,
+            "purchase_date": purchase_date,
+            "total": total,
+            "items": items,
+        }

receiptwitness/src/receiptwitness/queue/__init__.py

@@ -0,0 +1 @@
+"""DragonflyDB Streams queue for email receipt processing."""

receiptwitness/src/receiptwitness/queue/email.py

@@ -0,0 +1,77 @@
+"""DragonflyDB Streams queue for email receipt processing."""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import asdict, dataclass
+from typing import cast
+
+import redis.asyncio as aioredis
+
+from receiptwitness.config import settings
+
+logger = logging.getLogger(__name__)
+
+STREAM_KEY = "email:receipts"
+CONSUMER_GROUP = "email-workers"
+
+
+@dataclass
+class EmailJob:
+    """Payload for an email receipt processing job."""
+
+    user_id: str
+    sender: str
+    recipient: str
+    subject: str
+    body_html: str | None
+    body_plain: str | None
+    received_at: str
+    message_id: str  # from email provider, for dedup
+
+
+async def get_redis() -> aioredis.Redis:
+    """Get async Redis/DragonflyDB client."""
+    return cast(aioredis.Redis, aioredis.from_url(settings.redis_url, decode_responses=True))
+
+
+async def ensure_consumer_group(client: aioredis.Redis) -> None:
+    """Create consumer group if it does not exist."""
+    try:
+        await client.xgroup_create(STREAM_KEY, CONSUMER_GROUP, id="0", mkstream=True)
+    except aioredis.ResponseError as e:
+        if "BUSYGROUP" not in str(e):
+            raise
+
+
+async def enqueue_email(client: aioredis.Redis, job: EmailJob) -> str:
+    """Add email job to the stream. Returns the stream message ID."""
+    payload: dict[str, str | bytes | int | float] = {"data": json.dumps(asdict(job))}
+    msg_id: str = cast(str, await client.xadd(STREAM_KEY, payload))  # type: ignore[arg-type]  # redis-py StreamCommands.xadd expects broader FieldT union; runtime behavior is correct
+    logger.info("Enqueued email job %s for user %s", msg_id, job.user_id)
+    return msg_id
+
+
+async def consume_emails(
+    client: aioredis.Redis,
+    consumer_name: str,
+    count: int = 1,
+    block_ms: int = 5000,
+) -> list[tuple[str, EmailJob]]:
+    """Read pending messages from the stream. Returns list of (msg_id, EmailJob)."""
+    await ensure_consumer_group(client)
+    messages = await client.xreadgroup(
+        CONSUMER_GROUP, consumer_name, {STREAM_KEY: ">"}, count=count, block=block_ms
+    )
+    results = []
+    for _stream, entries in messages:
+        for msg_id, fields in entries:
+            job = EmailJob(**json.loads(fields["data"]))
+            results.append((msg_id, job))
+    return results
+
+
+async def ack_email(client: aioredis.Redis, msg_id: str) -> None:
+    """Acknowledge a processed message."""
+    await client.xack(STREAM_KEY, CONSUMER_GROUP, msg_id)

receiptwitness/src/receiptwitness/worker/__init__.py

@@ -0,0 +1 @@
+"""Async email receipt worker consuming from DragonflyDB Streams."""

receiptwitness/src/receiptwitness/worker/email_worker.py

@@ -0,0 +1,104 @@
+"""Async worker that consumes email receipt jobs from DragonflyDB Streams."""
+
+import asyncio
+import logging
+
+from cartsnitch_common.database import get_async_session_factory
+from cartsnitch_common.models.user import User
+from sqlalchemy import select
+
+from receiptwitness.config import settings
+from receiptwitness.events import publish_receipt_ingested
+from receiptwitness.parsers.email.base import BaseEmailParser, EmailReceipt
+from receiptwitness.parsers.email.detector import detect_retailer
+from receiptwitness.parsers.email.kroger import KrogerEmailParser
+from receiptwitness.parsers.email.meijer import MeijerEmailParser
+from receiptwitness.parsers.email.target import TargetEmailParser
+from receiptwitness.queue.email import ack_email, consume_emails, get_redis
+
+logger = logging.getLogger(__name__)
+
+CONSUMER_NAME = "worker-1"
+
+# Registry of available email parsers
+PARSERS: dict[str, BaseEmailParser] = {
+    "meijer": MeijerEmailParser(),
+    "kroger": KrogerEmailParser(),
+    "target": TargetEmailParser(),
+}
+
+
+async def resolve_user(token: str) -> str | None:
+    """Look up user_id from email_inbound_token."""
+    session_factory = get_async_session_factory(settings.database_url)
+    async with session_factory() as session:
+        result = await session.execute(select(User.id).where(User.email_inbound_token == token))
+        row = result.scalar_one_or_none()
+        return str(row) if row else None
+
+
+async def process_job(msg_id: str, job) -> bool:
+    """Process a single email job. Returns True on success."""
+    # 1. Resolve user from token
+    user_id = await resolve_user(job.user_id)  # user_id field holds token
+    if not user_id:
+        logger.warning("Unknown token %s, dropping message %s", job.user_id, msg_id)
+        return True  # ack to avoid infinite retry
+
+    # 2. Build EmailReceipt
+    email = EmailReceipt(
+        sender=job.sender,
+        recipient=job.recipient,
+        subject=job.subject,
+        body_html=job.body_html,
+        body_plain=job.body_plain,
+        received_at=job.received_at,
+    )
+
+    # 3. Detect retailer
+    retailer = detect_retailer(email)
+    if not retailer or retailer not in PARSERS:
+        logger.warning(
+            "Unrecognized retailer from %s, archiving msg %s",
+            job.sender,
+            msg_id,
+        )
+        return True  # ack — no parser available
+
+    # 4. Parse
+    parser = PARSERS[retailer]
+    parsed = parser.parse(email)
+
+    # 5. Publish event
+    await publish_receipt_ingested(
+        user_id=user_id,
+        store_slug=retailer,
+        purchase_id=parsed.get("receipt_id", msg_id),
+        purchase_date=parsed.get("purchase_date", ""),
+        item_count=len(parsed.get("items", [])),
+        total=parsed.get("total", 0),
+    )
+    return True
+
+
+async def run_worker() -> None:
+    """Main worker loop — consume and process email jobs."""
+    client = await get_redis()
+    logger.info("Email worker started, consuming from email:receipts")
+    while True:
+        try:
+            jobs = await consume_emails(client, CONSUMER_NAME, count=5, block_ms=5000)
+            for msg_id, job in jobs:
+                try:
+                    success = await process_job(msg_id, job)
+                    if success:
+                        await ack_email(client, msg_id)
+                except Exception:
+                    logger.exception("Failed to process email job %s", msg_id)
+        except Exception:
+            logger.exception("Worker loop error, retrying in 5s")
+            await asyncio.sleep(5)
+
+
+if __name__ == "__main__":
+    asyncio.run(run_worker())

receiptwitness/tests/conftest.py

@@ -1,12 +1,16 @@
 """Shared test fixtures."""
 
 import json
+import os
 from pathlib import Path
 
 import pytest
 
 FIXTURES_DIR = Path(__file__).parent / "fixtures"
 
+os.environ.setdefault("RW_SESSION_ENCRYPTION_KEY", "test-secret-key-for-unit-tests-only-32bytes!")
+os.environ.setdefault("RW_MAILGUN_WEBHOOK_SIGNING_KEY", "test-mailgun-signing-key")
+
 
 @pytest.fixture
 def meijer_receipt_data() -> dict:

receiptwitness/tests/fixtures/kroger_email_receipt.html

@@ -0,0 +1,45 @@
+<!DOCTYPE html>
+<html>
+<head>
+  <meta charset="UTF-8">
+  <title>Kroger Digital Receipt</title>
+</head>
+<body style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px; color: #333;">
+  <div style="background-color: #0057a8; color: white; padding: 20px; text-align: center;">
+    <img src="https://www.kroger.com/email-logo.png" alt="Kroger" style="height: 40px;">
+    <h1 style="margin: 10px 0; font-size: 24px;">Your Digital Receipt</h1>
+    <p style="margin: 0;">Kroger Plus Member</p>
+  </div>
+
+  <div style="padding: 20px; background-color: #f5f5f5;">
+    <h2 style="color: #0057a8; margin-top: 0;">Kroger #882 - Downtown</h2>
+    <p style="margin: 5px 0;">123 Main Street<br>Anytown, OH 45202</p>
+    <p style="margin: 5px 0;"><strong>Date:</strong> 03/15/2026</p>
+    <p style="margin: 5px 0;"><strong>Receipt #:</strong> KR-2026-0315-4829</p>
+    <p style="margin: 5px 0;"><strong>Transaction #:</strong> TXN-789123456</p>
+  </div>
+
+  <div style="padding: 20px;">
+    <h3>Items Purchased</h3>
+    <p>Whole Milk 1 Gallon $3.99</p>
+    <p>Sourdough Bread $4.49</p>
+    <p>Free Range Eggs 12ct $5.99</p>
+    <p>Baby Spinach 5oz $4.29</p>
+  </div>
+
+  <div style="padding: 20px; background-color: #e8f4e8; border-left: 4px solid #0057a8;">
+    <p style="margin: 5px 0;"><strong>Subtotal:</strong> $18.76</p>
+    <p style="margin: 5px 0;"><strong>Tax:</strong> $1.24</p>
+    <p style="margin: 5px 0; color: #0057a8; font-weight: bold; font-size: 18px;">Total: $20.00</p>
+  </div>
+
+  <div style="padding: 15px; margin-top: 15px; background-color: #fff8e1; border-left: 4px solid #ffc107;">
+    <p style="margin: 0; font-size: 14px; color: #666;">Kroger Plus Savings: <strong>$3.25</strong> saved on this order.</p>
+  </div>
+
+  <div style="padding: 20px; text-align: center; color: #999; font-size: 12px; margin-top: 20px;">
+    <p>Thank you for shopping at Kroger!</p>
+    <p>Keep your receipt for returns within 90 days.</p>
+  </div>
+</body>
+</html>

receiptwitness/tests/fixtures/meijer_email_receipt.html

@@ -0,0 +1,127 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+  <meta charset="UTF-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+  <title>Meijer Digital Receipt</title>
+  <style>
+    body { font-family: Arial, sans-serif; background: #f4f4f4; margin: 0; padding: 20px; }
+    .receipt-container { background: #ffffff; max-width: 600px; margin: 0 auto; padding: 30px; border: 1px solid #dddddd; }
+    .header { background: #003399; color: #ffffff; padding: 20px; text-align: center; margin: -30px -30px 20px -30px; }
+    .header h1 { margin: 0; font-size: 24px; }
+    .store-info { text-align: center; margin-bottom: 20px; border-bottom: 2px dashed #cccccc; padding-bottom: 15px; }
+    .store-info h2 { margin: 0; font-size: 18px; color: #003399; }
+    .receipt-meta { display: flex; justify-content: space-between; font-size: 14px; color: #555555; margin-bottom: 20px; }
+    table { width: 100%; border-collapse: collapse; margin-bottom: 20px; }
+    th { background: #f0f0f0; text-align: left; padding: 8px 10px; font-size: 13px; color: #333333; }
+    td { padding: 8px 10px; border-bottom: 1px solid #eeeeee; font-size: 14px; }
+    .item-name { font-weight: bold; }
+    .totals { margin-left: auto; width: 250px; }
+    .totals-row { display: flex; justify-content: space-between; padding: 6px 0; font-size: 14px; }
+    .totals-row.grand-total { font-weight: bold; font-size: 16px; border-top: 2px solid #333333; padding-top: 10px; margin-top: 4px; }
+    .savings { color: #cc0000; }
+    .footer { text-align: center; font-size: 12px; color: #888888; margin-top: 20px; padding-top: 15px; border-top: 1px solid #dddddd; }
+  </style>
+</head>
+<body>
+  <div class="receipt-container">
+    <div class="header">
+      <h1>MEIJER</h1>
+      <p style="margin: 5px 0 0; font-size: 14px;">Digital Receipt</p>
+    </div>
+
+    <div class="store-info">
+      <h2>Meijer Store #42</h2>
+      <p style="margin: 5px 0 0; font-size: 13px; color: #666;">1555 Lake Drive SE, Grand Rapids, MI 49506</p>
+    </div>
+
+    <div class="receipt-meta">
+      <div>
+        <strong>Date:</strong> March 15, 2026<br />
+        <strong>Time:</strong> 2:34 PM
+      </div>
+      <div style="text-align: right;">
+        <strong>Transaction #</strong><br />
+        TXN-2026-0315-0042
+      </div>
+    </div>
+
+    <table>
+      <thead>
+        <tr>
+          <th>Item</th>
+          <th style="text-align: center;">Qty</th>
+          <th style="text-align: right;">Price</th>
+        </tr>
+      </thead>
+      <tbody>
+        <tr>
+          <td class="item-name">ORGANIC BANANAS</td>
+          <td style="text-align: center;">1</td>
+          <td style="text-align: right;">$0.69</td>
+        </tr>
+        <tr>
+          <td class="item-name">WHOLE MILK 1 GAL</td>
+          <td style="text-align: center;">1</td>
+          <td style="text-align: right;">$4.29</td>
+        </tr>
+        <tr>
+          <td class="item-name">MEIJER WHOLE GRAIN OAT CEREAL 18OZ</td>
+          <td style="text-align: center;">1</td>
+          <td style="text-align: right;">$4.99</td>
+        </tr>
+        <tr>
+          <td class="item-name">FRESH BROCCOLI CROWN</td>
+          <td style="text-align: center;">1</td>
+          <td style="text-align: right;">$2.49</td>
+        </tr>
+        <tr>
+          <td class="item-name">GROUND BEEF 85/15 1LB</td>
+          <td style="text-align: center;">1</td>
+          <td style="text-align: right;">$6.99</td>
+        </tr>
+        <tr>
+          <td class="item-name">SOURDOUGH BREAD</td>
+          <td style="text-align: center;">1</td>
+          <td style="text-align: right;">$3.99</td>
+        </tr>
+        <tr>
+          <td class="item-name">MEIJER BABY SPINACH 5OZ</td>
+          <td style="text-align: center;">1</td>
+          <td style="text-align: right;">$4.49</td>
+        </tr>
+        <tr>
+          <td class="item-name">LARGE EGGS DOZEN</td>
+          <td style="text-align: center;">1</td>
+          <td style="text-align: right;">$3.29</td>
+        </tr>
+      </tbody>
+    </table>
+
+    <div class="totals">
+      <div class="totals-row">
+        <span>Subtotal</span>
+        <span>$31.22</span>
+      </div>
+      <div class="totals-row">
+        <span>Tax</span>
+        <span>$2.19</span>
+      </div>
+      <div class="totals-row savings">
+        <span>Total Savings</span>
+        <span>-$3.40</span>
+      </div>
+      <div class="totals-row grand-total">
+        <span>Total</span>
+        <span>$33.41</span>
+      </div>
+    </div>
+
+    <div class="footer">
+      <p>Thank you for shopping at Meijer!</p>
+      <p>Keep your receipt for your records.<br />
+      Questions? Call 1-800-927-8699 or visit meijer.com</p>
+    </div>
+  </div>
+</body>
+</html>

receiptwitness/tests/fixtures/target_email_receipt.html

@@ -0,0 +1,44 @@
+<!DOCTYPE html>
+<html>
+<head>
+  <meta charset="UTF-8">
+  <title>Target Order Confirmation</title>
+</head>
+<body style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px; color: #333;">
+  <div style="background-color: #cc0000; color: white; padding: 20px; text-align: center;">
+    <img src="https://assets.target.com/email-logo.png" alt="Target" style="height: 40px;">
+    <h1 style="margin: 10px 0; font-size: 24px;">Order Confirmation</h1>
+    <p style="margin: 0;">Thanks for shopping Target Circle!</p>
+  </div>
+
+  <div style="padding: 20px; background-color: #f5f5f5;">
+    <h2 style="color: #cc0000; margin-top: 0;">Target Store #1247 - Riverside</h2>
+    <p style="margin: 5px 0;">4500 River Road<br>Columbus, OH 43220</p>
+    <p style="margin: 5px 0;"><strong>Date:</strong> 03/18/2026</p>
+    <p style="margin: 5px 0;"><strong>Order #:</strong> TGT-2026-0318-9124</p>
+    <p style="margin: 5px 0;"><strong>Confirmation #:</strong> CNF-44772819</p>
+  </div>
+
+  <div style="padding: 20px;">
+    <h3>Items Purchased</h3>
+    <p>Good & Gather Whole Milk 1 Gal $3.89</p>
+    <p>Arborio Rice 2lb bag $6.49</p>
+    <p>Parmesan Wedge 8oz $7.99</p>
+  </div>
+
+  <div style="padding: 20px; background-color: #fff8e1; border-left: 4px solid #cc0000;">
+    <p style="margin: 5px 0;"><strong>Subtotal:</strong> $18.37</p>
+    <p style="margin: 5px 0;"><strong>Tax:</strong> $1.45</p>
+    <p style="margin: 5px 0; color: #cc0000; font-weight: bold; font-size: 18px;">Total: $19.82</p>
+  </div>
+
+  <div style="padding: 15px; margin-top: 15px; background-color: #e8f4e8; border-left: 4px solid #4caf50;">
+    <p style="margin: 0; font-size: 14px; color: #333;">Target Circle offer saved you <strong>$0.30</strong> on this order.</p>
+  </div>
+
+  <div style="padding: 20px; text-align: center; color: #999; font-size: 12px; margin-top: 20px;">
+    <p>Questions? Call Target Guest Services at 1-800-591-3869.</p>
+    <p>Receipt valid for returns within 30 days.</p>
+  </div>
+</body>
+</html>

receiptwitness/tests/test_api/__init__.py

@@ -0,0 +1 @@
+"""Tests for the ReceiptWitness API routes."""

receiptwitness/tests/test_api/test_webhook.py

@@ -0,0 +1,125 @@
+"""Tests for the /inbound/email webhook endpoint."""
+
+import hashlib
+import hmac
+import time
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+from receiptwitness.main import app
+
+
+@pytest.fixture
+def client():
+    return TestClient(app)
+
+
+@pytest.fixture
+def mock_redis():
+    redis_mock = AsyncMock()
+    with patch("receiptwitness.api.routes.get_redis", return_value=redis_mock):
+        enqueue_patcher = patch("receiptwitness.api.routes.enqueue_email", new_callable=AsyncMock)
+        with enqueue_patcher as mock_enqueue:
+            yield {"redis": redis_mock, "enqueue": mock_enqueue}
+
+
+def make_signature(signing_key: str, token: str, timestamp: str) -> str:
+    return hmac.new(
+        signing_key.encode(),
+        f"{timestamp}{token}".encode(),
+        hashlib.sha256,
+    ).hexdigest()
+
+
+def valid_form(signing_key: str = "test-secret"):
+    ts = str(int(time.time()))
+    token = "test-token"
+    sig = make_signature(signing_key, token, ts)
+    return {
+        "token": token,
+        "timestamp": ts,
+        "signature": sig,
+        "sender": "sender@example.com",
+        "recipient": "receipts+user123@example.com",
+        "subject": "Your Meijer Receipt",
+        "body-html": "<p>Thank you for shopping at Meijer</p>",
+        "body-plain": "Thank you for shopping at Meijer",
+        "Message-Id": "<msg-001@example.com>",
+    }
+
+
+def test_valid_webhook(client, mock_redis):
+    with patch("receiptwitness.api.routes.settings") as mock_settings:
+        mock_settings.mailgun_webhook_signing_key = "test-secret"
+        response = client.post("/inbound/email", data=valid_form())
+    assert response.status_code == 200
+    assert response.json() == {"status": "queued"}
+    mock_redis["enqueue"].assert_awaited_once()
+
+
+def test_invalid_signature(client, mock_redis):
+    with patch("receiptwitness.api.routes.settings") as mock_settings:
+        mock_settings.mailgun_webhook_signing_key = "test-secret"
+        form = valid_form()
+        form["signature"] = "wrong-signature"
+        response = client.post("/inbound/email", data=form)
+    assert response.status_code == 406
+    assert response.json()["detail"] == "Invalid signature"
+    mock_redis["enqueue"].assert_not_awaited()
+
+
+def test_invalid_recipient_no_plus(client, mock_redis):
+    with patch("receiptwitness.api.routes.settings") as mock_settings:
+        mock_settings.mailgun_webhook_signing_key = "test-secret"
+        form = valid_form()
+        form["recipient"] = "receipts@example.com"  # no plus-address
+        response = client.post("/inbound/email", data=form)
+    assert response.status_code == 406
+    assert response.json()["detail"] == "Invalid recipient"
+    mock_redis["enqueue"].assert_not_awaited()
+
+
+def test_stale_timestamp(client, mock_redis):
+    with patch("receiptwitness.api.routes.settings") as mock_settings:
+        mock_settings.mailgun_webhook_signing_key = "test-secret"
+        ts = str(int(time.time()) - 600)  # 10 min old
+        token = "test-token"
+        sig = make_signature("test-secret", token, ts)
+        form = {
+            "token": token,
+            "timestamp": ts,
+            "signature": sig,
+            "sender": "sender@example.com",
+            "recipient": "receipts+user123@example.com",
+            "subject": "Receipt",
+        }
+        response = client.post("/inbound/email", data=form)
+    assert response.status_code == 406
+    assert response.json()["detail"] == "Invalid signature"
+    mock_redis["enqueue"].assert_not_awaited()
+
+
+def test_invalid_timestamp_returns_406(client, mock_redis):
+    """Empty timestamp should return 406, not 500."""
+    with patch("receiptwitness.api.routes.settings") as mock_settings:
+        mock_settings.mailgun_webhook_signing_key = "test-secret"
+        form = {
+            "token": "test-token",
+            "timestamp": "",
+            "signature": "any-sig",
+            "sender": "sender@example.com",
+            "recipient": "receipts+user123@example.com",
+            "subject": "Receipt",
+        }
+        response = client.post("/inbound/email", data=form)
+    assert response.status_code == 406
+    assert response.json()["detail"] == "Invalid signature"
+    mock_redis["enqueue"].assert_not_awaited()
+
+
+def test_get_inbound_email_returns_405(client):
+    """GET /inbound/email is not allowed."""
+    response = client.get("/inbound/email")
+    assert response.status_code == 405

receiptwitness/tests/test_config.py

@@ -0,0 +1,46 @@
+import pytest
+from receiptwitness.config import ReceiptWitnessSettings
+
+
+def test_valid_config():
+    s = ReceiptWitnessSettings(
+        session_encryption_key="7reF42nmTwbdN21PBoubGp7h_FU8qSimstmlaMLoRK8="
+    )
+    assert s.session_encryption_key
+
+
+def test_missing_session_encryption_key_raises():
+    with pytest.raises(ValueError, match="RW_SESSION_ENCRYPTION_KEY"):
+        ReceiptWitnessSettings(session_encryption_key="")
+
+
+def test_placeholder_session_encryption_key_raises():
+    with pytest.raises(ValueError, match="RW_SESSION_ENCRYPTION_KEY"):
+        ReceiptWitnessSettings(session_encryption_key="change-me-in-production")
+
+
+def test_notifications_enabled_without_resend_key_raises():
+    with pytest.raises(ValueError, match="RW_RESEND_API_KEY"):
+        ReceiptWitnessSettings(
+            session_encryption_key="7reF42nmTwbdN21PBoubGp7h_FU8qSimstmlaMLoRK8=",
+            notifications_enabled=True,
+            resend_api_key="",
+        )
+
+
+def test_notifications_disabled_without_resend_key_ok():
+    s = ReceiptWitnessSettings(
+        session_encryption_key="7reF42nmTwbdN21PBoubGp7h_FU8qSimstmlaMLoRK8=",
+        notifications_enabled=False,
+        resend_api_key="",
+    )
+    assert s.notifications_enabled is False
+
+
+def test_notifications_enabled_with_resend_key_ok():
+    s = ReceiptWitnessSettings(
+        session_encryption_key="7reF42nmTwbdN21PBoubGp7h_FU8qSimstmlaMLoRK8=",
+        notifications_enabled=True,
+        resend_api_key="re_test_1234567890",
+    )
+    assert s.resend_api_key == "re_test_1234567890"

receiptwitness/tests/test_notifications/test_email.py

@@ -0,0 +1,84 @@
+"""Tests for email notifications."""
+
+from unittest.mock import patch
+
+import pytest
+
+
+class TestSendReceiptNotification:
+    @pytest.fixture
+    def mock_resend(self):
+        with patch("receiptwitness.notifications.email.resend") as mock:
+            yield mock
+
+    @pytest.mark.asyncio
+    async def test_sends_email_with_correct_params(self, mock_resend):
+        from receiptwitness.notifications.email import send_receipt_notification
+
+        with (
+            patch("receiptwitness.notifications.email.settings") as mock_settings,
+            patch(
+                "receiptwitness.notifications.email.asyncio.to_thread",
+                new=lambda fn, *args, **kwargs: fn(*args, **kwargs),
+            ),
+        ):
+            mock_settings.notifications_enabled = True
+            mock_settings.resend_api_key = "re_testkey_123"
+            mock_settings.notification_email_from = "noreply@test.com"
+
+            await send_receipt_notification(
+                user_email="user@example.com",
+                store_name="Meijer",
+                item_count=5,
+                total=42.99,
+                purchase_date="2026-03-28",
+            )
+
+        mock_resend.Emails.send.assert_called_once_with(
+            {
+                "from": "noreply@test.com",
+                "to": ["user@example.com"],
+                "subject": "Receipt processed: Meijer - $42.99",
+                "html": (
+                    "<p>Your receipt from <strong>Meijer</strong> on "
+                    "2026-03-28 has been processed.</p>"
+                    "<p>5 items, total: $42.99</p>"
+                ),
+            }
+        )
+
+    @pytest.mark.asyncio
+    async def test_skips_when_disabled(self, mock_resend):
+        from receiptwitness.notifications.email import send_receipt_notification
+
+        with patch("receiptwitness.notifications.email.settings") as mock_settings:
+            mock_settings.notifications_enabled = False
+            mock_settings.resend_api_key = "re_testkey_123"
+
+            await send_receipt_notification(
+                user_email="user@example.com",
+                store_name="Meijer",
+                item_count=5,
+                total=42.99,
+                purchase_date="2026-03-28",
+            )
+
+        mock_resend.Emails.send.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_skips_when_api_key_empty(self, mock_resend):
+        from receiptwitness.notifications.email import send_receipt_notification
+
+        with patch("receiptwitness.notifications.email.settings") as mock_settings:
+            mock_settings.notifications_enabled = True
+            mock_settings.resend_api_key = ""
+
+            await send_receipt_notification(
+                user_email="user@example.com",
+                store_name="Meijer",
+                item_count=5,
+                total=42.99,
+                purchase_date="2026-03-28",
+            )
+
+        mock_resend.Emails.send.assert_not_called()

receiptwitness/tests/test_parsers/test_email/test_detector.py

@@ -0,0 +1,49 @@
+"""Tests for retailer detector."""
+
+from receiptwitness.parsers.email.base import EmailReceipt
+from receiptwitness.parsers.email.detector import detect_retailer
+
+
+def test_detect_meijer():
+    email = EmailReceipt(
+        sender="receipts@meijer.com",
+        recipient="user@example.com",
+        subject="Your Receipt",
+    )
+    assert detect_retailer(email) == "meijer"
+
+
+def test_detect_kroger():
+    email = EmailReceipt(
+        sender="noreply@email.kroger.com",
+        recipient="user@example.com",
+        subject="Your Receipt",
+    )
+    assert detect_retailer(email) == "kroger"
+
+
+def test_detect_target():
+    email = EmailReceipt(
+        sender="Target <receipts@target.com>",
+        recipient="user@example.com",
+        subject="Your Receipt",
+    )
+    assert detect_retailer(email) == "target"
+
+
+def test_detect_unknown():
+    email = EmailReceipt(
+        sender="noreply@walmart.com",
+        recipient="user@example.com",
+        subject="Your Receipt",
+    )
+    assert detect_retailer(email) is None
+
+
+def test_detect_case_insensitive():
+    email = EmailReceipt(
+        sender="Receipts@MEIJER.COM",
+        recipient="user@example.com",
+        subject="Your Receipt",
+    )
+    assert detect_retailer(email) == "meijer"

receiptwitness/tests/test_parsers/test_email/test_kroger_email_parser.py

@@ -0,0 +1,93 @@
+"""Tests for KrogerEmailParser."""
+
+from pathlib import Path
+
+from receiptwitness.parsers.email.base import EmailReceipt
+from receiptwitness.parsers.email.kroger import KrogerEmailParser
+
+FIXTURE_PATH = Path(__file__).parent.parent.parent / "fixtures" / "kroger_email_receipt.html"
+
+
+class TestKrogerEmailParser:
+    """Tests for KrogerEmailParser."""
+
+    def setup_method(self) -> None:
+        self.parser = KrogerEmailParser()
+        self.fixture_html = FIXTURE_PATH.read_text()
+
+    def test_can_parse_kroger_sender(self) -> None:
+        email = EmailReceipt(
+            sender="noreply@email.kroger.com",
+            recipient="user@example.com",
+            subject="Your Kroger Receipt",
+            body_html=self.fixture_html,
+        )
+        assert self.parser.can_parse(email) is True
+
+    def test_can_parse_kroger_in_body(self) -> None:
+        email = EmailReceipt(
+            sender="someone@unknown.com",
+            recipient="user@example.com",
+            subject="Your Receipt",
+            body_html="<html><body>Kroger digital receipt</body></html>",
+        )
+        assert self.parser.can_parse(email) is True
+
+    def test_cannot_parse_unrelated(self) -> None:
+        email = EmailReceipt(
+            sender="noreply@walmart.com",
+            recipient="user@example.com",
+            subject="Your Receipt",
+            body_html="<html><body>Walmart receipt</body></html>",
+        )
+        assert self.parser.can_parse(email) is False
+
+    def test_parse_items(self) -> None:
+        email = EmailReceipt(
+            sender="noreply@kroger.com",
+            recipient="user@example.com",
+            subject="Your Kroger Receipt",
+            body_html=self.fixture_html,
+        )
+        result = self.parser.parse(email)
+        items = result.get("items", [])
+        assert len(items) >= 3
+        product_names = [item["product_name_raw"] for item in items]
+        assert any("Whole Milk" in name for name in product_names)
+        assert any("Sourdough" in name for name in product_names)
+        for item in items:
+            assert "unit_price" in item
+            assert "extended_price" in item
+
+    def test_parse_totals(self) -> None:
+        email = EmailReceipt(
+            sender="noreply@kroger.com",
+            recipient="user@example.com",
+            subject="Your Kroger Receipt",
+            body_html=self.fixture_html,
+        )
+        result = self.parser.parse(email)
+        total = result.get("total", 0)
+        assert total > 0
+
+    def test_parse_receipt_id(self) -> None:
+        email = EmailReceipt(
+            sender="noreply@kroger.com",
+            recipient="user@example.com",
+            subject="Your Kroger Receipt",
+            body_html=self.fixture_html,
+        )
+        result = self.parser.parse(email)
+        receipt_id = result.get("receipt_id", "")
+        assert "KR-2026" in receipt_id or "TXN" in receipt_id
+
+    def test_parse_date(self) -> None:
+        email = EmailReceipt(
+            sender="noreply@kroger.com",
+            recipient="user@example.com",
+            subject="Your Kroger Receipt",
+            body_html=self.fixture_html,
+        )
+        result = self.parser.parse(email)
+        purchase_date = result.get("purchase_date", "")
+        assert purchase_date == "2026-03-15"

receiptwitness/tests/test_parsers/test_email/test_meijer_parser.py

@@ -0,0 +1,182 @@
+"""Tests for the Meijer email receipt parser."""
+
+import os
+from decimal import Decimal
+
+import pytest
+
+from receiptwitness.parsers.email.base import EmailReceipt
+from receiptwitness.parsers.email.meijer import MeijerEmailParser
+
+FIXTURE_PATH = os.path.join(
+    os.path.dirname(__file__), "..", "..", "fixtures", "meijer_email_receipt.html"
+)
+
+
+def load_fixture() -> str:
+    with open(FIXTURE_PATH) as f:
+        return f.read()
+
+
+@pytest.fixture
+def meijer_email() -> EmailReceipt:
+    html = load_fixture()
+    return EmailReceipt(
+        sender="Meijer Receipts <receipts@email.meijer.com>",
+        recipient="shopper@example.com",
+        subject="Your Meijer Receipt — Transaction #TXN-2026-0315-0042",
+        body_html=html,
+        body_plain=None,
+        received_at="2026-03-15T14:34:00Z",
+    )
+
+
+@pytest.fixture
+def kroger_email() -> EmailReceipt:
+    return EmailReceipt(
+        sender="Kroger <noreply@email.kroger.com>",
+        recipient="shopper@example.com",
+        subject="Your Kroger Receipt",
+        body_html="<html><body>Kroger receipt</body></html>",
+    )
+
+
+class TestCanParse:
+    def test_can_parse_meijer(self, meijer_email: EmailReceipt):
+        parser = MeijerEmailParser()
+        assert parser.can_parse(meijer_email) is True
+
+    def test_cannot_parse_kroger(self, kroger_email: EmailReceipt):
+        parser = MeijerEmailParser()
+        assert parser.can_parse(kroger_email) is False
+
+    def test_can_parse_meijer_plain_sender(self):
+        email = EmailReceipt(
+            sender="receipts@meijer.com",
+            recipient="shopper@example.com",
+            subject="Receipt",
+            body_html="<html></html>",
+        )
+        parser = MeijerEmailParser()
+        assert parser.can_parse(email) is True
+
+    def test_cannot_parse_non_meijer(self):
+        email = EmailReceipt(
+            sender=" Target <no-reply@target.com>",
+            recipient="shopper@example.com",
+            subject="Target Receipt",
+            body_html="<html></html>",
+        )
+        parser = MeijerEmailParser()
+        assert parser.can_parse(email) is False
+
+
+class TestParseMeijerReceipt:
+    def test_receipt_id_extracted(self, meijer_email: EmailReceipt):
+        parser = MeijerEmailParser()
+        result = parser.parse(meijer_email)
+        assert result["receipt_id"] == "TXN-2026-0315-0042"
+
+    def test_purchase_date_extracted(self, meijer_email: EmailReceipt):
+        parser = MeijerEmailParser()
+        result = parser.parse(meijer_email)
+        assert result["purchase_date"] == "2026-03-15"
+
+    def test_items_extracted(self, meijer_email: EmailReceipt):
+        parser = MeijerEmailParser()
+        result = parser.parse(meijer_email)
+        items = result["items"]
+        assert len(items) == 8
+
+        names = [item["product_name_raw"] for item in items]
+        assert "ORGANIC BANANAS" in names
+        assert "WHOLE MILK 1 GAL" in names
+        assert "GROUND BEEF 85/15 1LB" in names
+
+    def test_item_quantities(self, meijer_email: EmailReceipt):
+        parser = MeijerEmailParser()
+        result = parser.parse(meijer_email)
+        # Find ORGANIC BANANAS
+        bananas = next(i for i in result["items"] if "BANANAS" in i["product_name_raw"])
+        assert bananas["quantity"] == Decimal("1")
+
+    def test_item_prices(self, meijer_email: EmailReceipt):
+        parser = MeijerEmailParser()
+        result = parser.parse(meijer_email)
+        # Find ORGANIC BANANAS
+        bananas = next(i for i in result["items"] if "BANANAS" in i["product_name_raw"])
+        assert bananas["unit_price"] == Decimal("0.69")
+        assert bananas["extended_price"] == Decimal("0.69")
+
+    def test_totals(self, meijer_email: EmailReceipt):
+        parser = MeijerEmailParser()
+        result = parser.parse(meijer_email)
+        assert result["total"] == Decimal("33.41")
+        assert result["subtotal"] == Decimal("31.22")
+        assert result["tax"] == Decimal("2.19")
+        assert result["savings_total"] == Decimal("3.40")
+
+
+class TestParseHandlesMissingFields:
+    def test_missing_body_html_falls_back_to_plain(self):
+        email = EmailReceipt(
+            sender="receipts@email.meijer.com",
+            recipient="shopper@example.com",
+            subject="Your Meijer Receipt",
+            body_html=None,
+            body_plain="TXN-1234 | March 15, 2026 | Total: $10.00",
+        )
+        parser = MeijerEmailParser()
+        result = parser.parse(email)
+        # Should not raise, returns minimal result
+        assert result["receipt_id"] == ""
+        assert result["purchase_date"] == "2026-03-15"
+        assert result["total"] == Decimal("10.00")
+
+    def test_empty_email(self):
+        email = EmailReceipt(
+            sender="receipts@email.meijer.com",
+            recipient="shopper@example.com",
+            subject="Receipt",
+            body_html="",
+            body_plain="",
+        )
+        parser = MeijerEmailParser()
+        result = parser.parse(email)
+        assert result["receipt_id"] == ""
+        assert result["purchase_date"] == ""
+        assert result["total"] == Decimal("0")
+        assert result["items"] == []
+
+    def test_missing_subject_date_from_body(self):
+        html = """
+        <html>
+          <body>
+            <p>Thank you for shopping on April 1, 2026</p>
+            <p>Total: $15.00</p>
+          </body>
+        </html>
+        """
+        email = EmailReceipt(
+            sender="receipts@email.meijer.com",
+            recipient="shopper@example.com",
+            subject=None,
+            body_html=html,
+        )
+        parser = MeijerEmailParser()
+        result = parser.parse(email)
+        assert result["purchase_date"] == "2026-04-01"
+
+    def test_missing_totals_defaults_to_zero(self):
+        html = "<html><body><p>Just an email with no totals</p></body></html>"
+        email = EmailReceipt(
+            sender="receipts@email.meijer.com",
+            recipient="shopper@example.com",
+            subject="Receipt",
+            body_html=html,
+        )
+        parser = MeijerEmailParser()
+        result = parser.parse(email)
+        assert result["total"] == Decimal("0")
+        assert result["subtotal"] is None
+        assert result["tax"] is None

receiptwitness/tests/test_parsers/test_email/test_target_email_parser.py

@@ -0,0 +1,93 @@
+"""Tests for TargetEmailParser."""
+
+from pathlib import Path
+
+from receiptwitness.parsers.email.base import EmailReceipt
+from receiptwitness.parsers.email.target import TargetEmailParser
+
+FIXTURE_PATH = Path(__file__).parent.parent.parent / "fixtures" / "target_email_receipt.html"
+
+
+class TestTargetEmailParser:
+    """Tests for TargetEmailParser."""
+
+    def setup_method(self) -> None:
+        self.parser = TargetEmailParser()
+        self.fixture_html = FIXTURE_PATH.read_text()
+
+    def test_can_parse_target_sender(self) -> None:
+        email = EmailReceipt(
+            sender="receipts@target.com",
+            recipient="user@example.com",
+            subject="Your Target Order Confirmation",
+            body_html=self.fixture_html,
+        )
+        assert self.parser.can_parse(email) is True
+
+    def test_can_parse_circle_in_body(self) -> None:
+        email = EmailReceipt(
+            sender="someone@unknown.com",
+            recipient="user@example.com",
+            subject="Your Receipt",
+            body_html="<html><body>Target Circle savings offer</body></html>",
+        )
+        assert self.parser.can_parse(email) is True
+
+    def test_cannot_parse_unrelated(self) -> None:
+        email = EmailReceipt(
+            sender="noreply@walmart.com",
+            recipient="user@example.com",
+            subject="Your Receipt",
+            body_html="<html><body>Walmart receipt</body></html>",
+        )
+        assert self.parser.can_parse(email) is False
+
+    def test_parse_items(self) -> None:
+        email = EmailReceipt(
+            sender="orders@target.com",
+            recipient="user@example.com",
+            subject="Your Target Order",
+            body_html=self.fixture_html,
+        )
+        result = self.parser.parse(email)
+        items = result.get("items", [])
+        assert len(items) >= 3
+        product_names = [item["product_name_raw"] for item in items]
+        assert any("Whole Milk" in name for name in product_names)
+        assert any("Arborio" in name for name in product_names)
+        for item in items:
+            assert "unit_price" in item
+            assert "extended_price" in item
+
+    def test_parse_totals(self) -> None:
+        email = EmailReceipt(
+            sender="orders@target.com",
+            recipient="user@example.com",
+            subject="Your Target Order",
+            body_html=self.fixture_html,
+        )
+        result = self.parser.parse(email)
+        total = result.get("total", 0)
+        assert total > 0
+
+    def test_parse_receipt_id(self) -> None:
+        email = EmailReceipt(
+            sender="orders@target.com",
+            recipient="user@example.com",
+            subject="Your Target Order",
+            body_html=self.fixture_html,
+        )
+        result = self.parser.parse(email)
+        receipt_id = result.get("receipt_id", "")
+        assert "TGT-2026" in receipt_id or "CNF" in receipt_id
+
+    def test_parse_date(self) -> None:
+        email = EmailReceipt(
+            sender="orders@target.com",
+            recipient="user@example.com",
+            subject="Your Target Order",
+            body_html=self.fixture_html,
+        )
+        result = self.parser.parse(email)
+        purchase_date = result.get("purchase_date", "")
+        assert purchase_date == "2026-03-18"

receiptwitness/tests/test_queue/test_email_queue.py

@@ -0,0 +1,79 @@
+"""Tests for email queue using DragonflyDB Streams."""
+
+import pytest
+from fakeredis import aioredis as fake_aioredis
+
+from receiptwitness.queue.email import (
+    CONSUMER_GROUP,
+    STREAM_KEY,
+    EmailJob,
+    ack_email,
+    consume_emails,
+    enqueue_email,
+    ensure_consumer_group,
+)
+
+
+@pytest.fixture
+async def fake_client():
+    """Yield a fake async Redis client."""
+    client = fake_aioredis.FakeRedis(decode_responses=True)
+    yield client
+    await client.aclose()
+
+
+@pytest.fixture
+def sample_job():
+    """Sample EmailJob for testing."""
+    return EmailJob(
+        user_id="user-123",
+        sender="no-reply@kroger.com",
+        recipient="user@example.com",
+        subject="Kroger Receipt",
+        body_html="<html><body>Receipt</body></html>",
+        body_plain="Receipt",
+        received_at="2026-04-01T12:00:00Z",
+        message_id="msg-abc-123",
+    )
+
+
+@pytest.mark.asyncio
+async def test_enqueue_and_consume(fake_client, sample_job):
+    """Enqueue a job, consume it, verify fields match."""
+    msg_id = await enqueue_email(fake_client, sample_job)
+    assert msg_id is not None
+
+    consumed = await consume_emails(fake_client, "test-worker", count=1, block_ms=100)
+    assert len(consumed) == 1
+    consumed_id, consumed_job = consumed[0]
+    assert consumed_id == msg_id
+    assert consumed_job.user_id == sample_job.user_id
+    assert consumed_job.sender == sample_job.sender
+    assert consumed_job.recipient == sample_job.recipient
+    assert consumed_job.subject == sample_job.subject
+    assert consumed_job.message_id == sample_job.message_id
+
+
+@pytest.mark.asyncio
+async def test_ack_removes_from_pending(fake_client, sample_job):
+    """After ack, message is no longer pending."""
+    msg_id = await enqueue_email(fake_client, sample_job)
+
+    # Consume the message (moves it to pending)
+    consumed = await consume_emails(fake_client, "test-worker", count=1, block_ms=100)
+    assert len(consumed) == 1
+
+    # Acknowledge it
+    await ack_email(fake_client, msg_id)
+
+    # Check pending count for this consumer group
+    pending = await fake_client.xpending(STREAM_KEY, CONSUMER_GROUP)
+    assert pending is None or pending["pending"] == 0
+
+
+@pytest.mark.asyncio
+async def test_ensure_consumer_group_idempotent(fake_client):
+    """Calling ensure_consumer_group twice does not error."""
+    await ensure_consumer_group(fake_client)
+    # Calling again should not raise
+    await ensure_consumer_group(fake_client)

receiptwitness/tests/test_worker/test_email_worker.py

@@ -0,0 +1,188 @@
+"""Tests for email_worker."""
+
+from decimal import Decimal
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fakeredis import aioredis as fake_aioredis
+
+from receiptwitness.parsers.email.base import EmailReceipt
+from receiptwitness.queue.email import (
+    EmailJob,
+)
+from receiptwitness.worker.email_worker import (
+    process_job,
+    resolve_user,
+)
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def fake_redis():
+    """Fake async Redis client for queue testing."""
+    client = fake_aioredis.FakeRedis(decode_responses=True)
+    yield client
+    await client.aclose()
+
+
+@pytest.fixture
+def sample_email_job():
+    """Sample EmailJob matching DragonflyDB queue schema."""
+    return EmailJob(
+        user_id="token-abc-123",
+        sender="no-reply@meijer.com",
+        recipient="user@example.com",
+        subject="Your Meijer Receipt",
+        body_html="<html><body>Total: $42.00</body></html>",
+        body_plain="Total: $42.00",
+        received_at="2026-04-01T12:00:00Z",
+        message_id="msg-xyz-789",
+    )
+
+
+@pytest.fixture
+def sample_email():
+    """Sample EmailReceipt for parser testing."""
+    return EmailReceipt(
+        sender="no-reply@meijer.com",
+        recipient="user@example.com",
+        subject="Your Meijer Receipt",
+        body_html="<html><body>Total: $42.00<br/>Receipt #12345</body></html>",
+        body_plain="Total: $42.00",
+        received_at="2026-04-01T12:00:00Z",
+    )
+
+
+# ---------------------------------------------------------------------------
+# resolve_user tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_resolve_user_valid_token():
+    """Valid token returns user_id string."""
+    mock_session = AsyncMock()
+    mock_result = MagicMock()
+    mock_result.scalar_one_or_none.return_value = "user-uuid-42"
+    mock_session.execute.return_value = mock_result
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=None)
+
+    factory = MagicMock(return_value=mock_session)
+
+    with patch(
+        "receiptwitness.worker.email_worker.get_async_session_factory",
+        return_value=factory,
+    ):
+        user_id = await resolve_user("token-abc-123")
+
+    assert user_id == "user-uuid-42"
+    factory.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_resolve_user_invalid_token():
+    """Invalid token returns None."""
+    mock_session = AsyncMock()
+    mock_result = MagicMock()
+    mock_result.scalar_one_or_none.return_value = None
+    mock_session.execute.return_value = mock_result
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=None)
+
+    factory = MagicMock(return_value=mock_session)
+
+    with patch(
+        "receiptwitness.worker.email_worker.get_async_session_factory",
+        return_value=factory,
+    ):
+        user_id = await resolve_user("bad-token")
+
+    assert user_id is None
+
+
+# ---------------------------------------------------------------------------
+# process_job tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_process_job_unknown_retailer(sample_email_job):
+    """Unknown retailer logs warning and returns True (ack, no retry)."""
+    unknown_job = EmailJob(
+        user_id="token-abc-123",
+        sender="no-reply@unknownretailer.com",
+        recipient="user@example.com",
+        subject="Receipt",
+        body_html="<html></html>",
+        body_plain="",
+        received_at="2026-04-01T12:00:00Z",
+        message_id="msg-xyz-789",
+    )
+
+    with (
+        patch(
+            "receiptwitness.worker.email_worker.resolve_user",
+            return_value="user-uuid-42",
+        ),
+        patch(
+            "receiptwitness.worker.email_worker.publish_receipt_ingested",
+            new_callable=AsyncMock,
+        ) as mock_publish,
+    ):
+        result = await process_job("msg-id-1", unknown_job)
+
+    assert result is True
+    mock_publish.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_process_job_success(sample_email_job, sample_email):
+    """Known retailer: full pipeline runs — parse, normalize, publish event."""
+    parsed_data = {
+        "receipt_id": "RCP-999",
+        "purchase_date": "2026-04-01",
+        "total": Decimal("42.00"),
+        "items": [
+            {
+                "product_name_raw": "ORGANIC BANANAS",
+                "quantity": Decimal("1"),
+                "unit_price": Decimal("0.69"),
+                "extended_price": Decimal("0.69"),
+            },
+        ],
+    }
+
+    mock_parser = MagicMock()
+    mock_parser.parse.return_value = parsed_data
+
+    with (
+        patch(
+            "receiptwitness.worker.email_worker.resolve_user",
+            return_value="user-uuid-42",
+        ),
+        patch.dict(
+            "receiptwitness.worker.email_worker.PARSERS",
+            {"meijer": mock_parser},
+            clear=False,
+        ),
+        patch(
+            "receiptwitness.worker.email_worker.publish_receipt_ingested",
+            new_callable=AsyncMock,
+        ) as mock_publish,
+    ):
+        result = await process_job("msg-id-1", sample_email_job)
+
+    assert result is True
+    mock_parser.parse.assert_called_once()
+    mock_publish.assert_called_once_with(
+        user_id="user-uuid-42",
+        store_slug="meijer",
+        purchase_id="RCP-999",
+        purchase_date="2026-04-01",
+        item_count=1,
+        total=Decimal("42.00"),
+    )

scripts/seed-dev-job.yaml

@@ -0,0 +1,61 @@
+# seed-dev-job.yaml
+# K8s Job to run the CartSnitch seed runner against the dev database.
+#
+# Usage:
+#   kubectl apply -f seed-dev-job.yaml -n cartsnitch-dev
+#
+# To view logs:
+#   kubectl logs -n cartsnitch-dev job/seed-dev -f
+#
+# To re-run after fixing issues:
+#   kubectl delete -f seed-dev-job.yaml -n cartsnitch-dev && kubectl apply -f seed-dev-job.yaml -n cartsnitch-dev
+#
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: seed-dev
+  namespace: cartsnitch-dev
+  labels:
+    app: cartsnitch
+    component: seed
+    environment: dev
+  annotations:
+    description: "Runs cartsnitch-common seed runner to populate dev database with realistic test data."
+spec:
+  # Prevent retries — a failed seed run should be investigated, not auto-repeated.
+  backoffLimit: 0
+  # Do not run concurrently; sequential runs are safer for truncate+reseed.
+  concurrencyPolicy: Forbid
+  template:
+    metadata:
+      labels:
+        app: cartsnitch
+        component: seed
+        environment: dev
+    spec:
+      restartPolicy: Never
+      containers:
+        - name: seed
+          # Use slim Python image with the cartsnitch-common package installed from git.
+          # The common repo is public; no additional secret is needed for the pip install.
+          image: python:3.12-slim
+          command:
+            - sh
+            - -c
+            - |
+              pip install --no-cache-dir "cartsnitch-common @ git+https://github.com/cartsnitch/common.git@main" && \
+              python -m cartsnitch_common.seed --database-url "$${DATABASE_URL}"
+          env:
+            - name: DATABASE_URL
+              valueFrom:
+                secretKeyRef:
+                  name: cartsnitch-secrets
+                  key: database-url-pg
+                  optional: false
+          resources:
+            requests:
+              cpu: 100m
+              memory: 256Mi
+            limits:
+              cpu: 500m
+              memory: 512Mi

scripts/seed-dev.sh

@@ -0,0 +1,104 @@
+#!/usr/bin/env bash
+# =============================================================================
+# seed-dev.sh — Run the CartSnitch seed runner against the dev database.
+#
+# Usage:
+#   ./seed-dev.sh              Run full seed against dev
+#   ./seed-dev.sh --dry-run    Show planned record counts without writing
+#   ./seed-dev.sh --help       Show this help
+#
+# Prerequisites:
+#   - kubectl configured for the cartsnitch-dev cluster
+#   - Namespace cartsnitch-dev exists (CNPG Postgres must be running)
+#
+# What it does:
+#   1. Starts a background port-forward to cartsnitch-pg-rw:5432
+#   2. Waits for the tunnel to be ready
+#   3. Runs python -m cartsnitch_common.seed with --database-url pointing
+#      to localhost:<forwarded-port>/cartsnitch
+#   4. Cleans up the port-forward on exit (normal, interrupt, or error)
+# =============================================================================
+
+set -euo pipefail
+
+# --- Config -------------------------------------------------------------------
+readonly NAMESPACE="cartsnitch-dev"
+readonly SVC_NAME="cartsnitch-pg-rw"
+readonly LOCAL_PORT="5433"          # use a non-privileged port to avoid conflicts
+readonly DB_NAME="cartsnitch"
+readonly PG_USER="cartsnitch"
+# Retrieve password from the CNPG credentials secret
+readonly PG_PASSWORD="$(
+  kubectl get secret cartsnitch-pg-credentials \
+    -n "$NAMESPACE" \
+    -o jsonpath='{.data.password}' \
+  | base64 -d
+)"
+readonly DB_URL="postgresql://${PG_USER}:${PG_PASSWORD}@localhost:${LOCAL_PORT}/${DB_NAME}"
+
+# --- Helpers ------------------------------------------------------------------
+log()  { echo "[seed-dev] $*"; }
+fail() { log "ERROR: $*" >&2; exit 1; }
+
+# Cleanup port-forward and exit.
+cleanup() {
+  if [[ -n "${PF_PID:-}" ]]; then
+    log "Stopping port-forward (PID $PF_PID)..."
+    kill "$PF_PID" 2>/dev/null || true
+    wait "$PF_PID" 2>/dev/null || true
+  fi
+}
+trap cleanup EXIT
+
+# --- Args ---------------------------------------------------------------------
+DRY_RUN=""
+HELP_FLAG=""
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --dry-run)  DRY_RUN="--dry-run"; shift ;;
+    --help)     HELP_FLAG="1"; shift ;;
+    *)          fail "Unknown argument: $1";;
+  esac
+done
+
+if [[ -n "$HELP_FLAG" ]]; then
+  sed -n '3,/^# ---/p' "$0" | head -n -1 | sed 's/^# //'
+  echo ""
+  echo "Additional arguments are passed through to the seed runner."
+  echo "Common seed-runner options:"
+  echo "  --dry-run          Show planned record counts without writing"
+  echo "  --seed N           Set random seed (default: 42)"
+  exit 0
+fi
+
+# --- Prerequisites ------------------------------------------------------------
+if ! command -v kubectl &>/dev/null; then
+  fail "kubectl not found — must be installed and configured."
+fi
+
+# --- Port-forward -------------------------------------------------------------
+log "Starting port-forward ${SVC_NAME}:5432 -> localhost:${LOCAL_PORT} ..."
+kubectl port-forward \
+  -n "$NAMESPACE" \
+  svc/"$SVC_NAME" \
+  "${LOCAL_PORT}:5432" \
+  &>/dev/null &
+PF_PID=$!
+
+# Give the tunnel a moment to establish
+sleep 2
+
+# Verify the tunnel is up
+if ! kill -0 "$PF_PID" 2>/dev/null; then
+  fail "Port-forward failed to start."
+fi
+log "Port-forward active (PID $PF_PID) on localhost:${LOCAL_PORT}"
+
+# --- Seed --------------------------------------------------------------------
+log "Running seed against dev database..."
+set -x
+python -m cartsnitch_common.seed --database-url "$DB_URL" $DRY_RUN
+set +x
+
+log "Done."