# New Knowledge Base

## Bedrock fallback inference profile
- **Primary** `INFERENCE_PROFILE_ARN` (e.g. `eu.anthropic.claude-sonnet-4-6`); optional **`BEDROCK_FALLBACK_INFERENCE_PROFILE_ARN`** (e.g. `eu.anthropic.claude-sonnet-4-5`). On **`ThrottlingException`** / **`TooManyRequestsException`** after primary retries, the Lambda invokes **once** with the fallback **`modelId`**. **`BedrockUsage`** logs include **`model_id`**. Separate quotas per profile — not a substitute for Service Quotas increases if both pools are exhausted.

## Bedrock dry-run test API
- **`POST .../analyze-image-test`** (Lambda **`test_app.lambda_handler`**, same **`CodeUri: app/`** as **`BedrockImageAnalysis`**). Bedrock via **`invoke_bedrock_test_model`** only — **does not** call **`analyze_image_with_bedrock`**. **No** DB writes, **no** S3 artifacts. Default model: **Qwen** (`qwen.*` **Converse**); override with body **`model_id`** or env **`BEDROCK_TEST_MODEL_ID`**.
- **`qwen.*`** → **Converse**; other ids → **InvokeModel** (Anthropic Messages). Response **`test_invoke_path`** distinguishes paths. Harness prompt in **`test_app`** differs from production **`app.py`** prompt length/detail.
- **Qwen Converse** may return **`Failed to buffer the request body: length limit exceeded`** when **two** ~3.8 MiB images + prompt exceed the model gateway cap — use **`BEDROCK_TEST_CONVERSE_MAX_IMAGE_BYTES`** (default **1 500 000**) on the test Lambda.

## Bedrock batch Step Functions (product-centric)
- **Purpose:** Same **SAST `fixed_at` today** universe as **`ListBedrockRetryTasks`**, but **group by `product_id`**: one **`BedrockBatchAnalyze`** invocation loads the **product `image_path` once** and attaches **several shelf images** (different `task_id` / `product_images` rows). Reduces duplicate reference-image and instruction tokens versus **`BedrockRetry`** (one **`BedrockImageAnalysis`** call per **task** with **`task_id`** only).
- **List shape:** **`ListBedrockBatchGroups`** returns **`groups`**: each element `{ product_id, items: [{ image_id } | { task_id }] }` with **`items`** length capped by **`BedrockBatchMaxShelfImages`** (template env **`BATCH_MAX_SHELF_IMAGES`**). Rows without **`product_id`** are omitted from batch listing (use **`BedrockRetry`**).
- **Orchestration:** **`BedrockBatchWorkflow`** mirrors retry pagination (**`has_more`**, **`next_offset`**, **`run_date`**); **Map** **`ResultPath: null`**; **`BedrockBatchMapMaxConcurrency`**.
- **Output:** Model returns **JSON** (`results[]` per `shelf_index`); Lambda maps to **`update_analysis_result`** fields. Missing rows → **`genai_error`** on **`product_images`** when applicable.

## Bedrock analysis retry Step Functions
- **List filter:** **`ListBedrockRetryTasks`** reads **`task` only** (no **`product_images`**); returns **`{ task_id }`** per incomplete task with non-empty **`task.image_url`** in the SAST **`fixed_at`** window.
- **Pattern:** A small **list** Lambda queries Supabase (same anon key + SSM as the worker) and returns **pages** of `{ items, has_more, next_offset, run_date, total_count, ... }`. Step Functions loops List → Map until `has_more` is false. **Map** uses **`ResultPath: null`** so the workflow does not retain one JSON blob per Bedrock invocation (avoids **~256 KB** state limit / `States.DataLimitExceeded`). **`MaxConcurrency`** comes from `BedrockRetryMapMaxConcurrency`; stay under Bedrock RPM/TPM (Service Quotas).
- **Worker reuse:** The Bedrock Lambda accepts **direct** payloads (`image_id` / `task_id` at the root of the event) in addition to API Gateway `body` JSON.
- **Date window:** “Today” for `fixed_at` is the **Africa/Johannesburg** calendar day, expressed as a UTC `[start, end)` range in the list Lambda (`zoneinfo`). Queries use **`fixed_at IS NOT NULL`** explicitly plus **`fixed_at`** range.
- **Backfill:** Step Functions **Start execution** input **`{ "run_date": "YYYY-MM-DD" }`** seeds the first **List** call for that SAST day (`$$.Execution.Input.run_date`); empty **`{}`** keeps “today” SAST.
- **RLS:** If the list returns zero rows but the DB has matches, check Supabase RLS on `task` / `product_images` for the anon role (same issue as other Lambdas that use the anon key).

## Step Functions + S3 for Lambda timeout avoidance
- When a single Lambda would time out (e.g. 15 min) doing many repeated operations (e.g. 34k DB inserts), split the work:
  1. **Orchestrator Lambda:** Do the fast part (fetch + transform), write payloads to S3 in chunks (e.g. 5k records per JSON file), then start a Step Function with a list of S3 keys.
  2. **Worker Lambda(s):** Invoked by the Step Function Map state; each reads one S3 object and performs the heavy operation (e.g. one batch upsert). Each run stays well under the Lambda timeout.
  3. **Finalize Lambda:** Run after the Map (all batches done); do post-work (e.g. update aggregates, trigger next job, send notification).
- Use a short S3 lifecycle (e.g. 1 day) on the staging prefix to avoid storing sensitive data long-term.
- CreateTasks uses this pattern: CreateTasks → S3 batches → Step Function (Map: InsertTasksBatch, then CreateTasksFinalize).

## Supabase Python client options
- The version of `supabase-py` used in our Lambdas does not support `ClientOptions(http_client=...)`.
- Attempting to pass `http_client` raises: `__init__() got an unexpected keyword argument 'http_client'`.
- Recommended pattern:
  - Initialize with `create_client(url, key)`.
  - For long-running operations, adjust database timeouts via SQL, e.g.:
    - `supabase.postgrest.rpc("raw_sql", {"query": "SET statement_timeout = '0';"})`

## Supabase: no `.truncate()` on table queries
- The `supabase-py` client does not provide a `.truncate()` method on `table('...')` query builders.
- To "truncate" a table from Python, either:
  - Issue a delete-all with a catch-all filter, e.g. `supabase.table('store_processing_queue').delete().neq('id', 0).execute()`; or
  - Expose a Postgres function that performs `TRUNCATE` and invoke it via `supabase.rpc('your_truncate_fn').execute()`.
- Calling `.truncate()` directly on a `SyncRequestBuilder` raises: `AttributeError: 'SyncRequestBuilder' object has no attribute 'truncate'`.

## Lambda error-handling guard
- When constructing user-facing error messages, initialize variables like `file_name` and `record_id` early (defaults) to avoid `UnboundLocalError` when setup fails before they are assigned.

## Lambda Layers for shared Python deps
- For faster deployments, move heavy/common dependencies into a SAM Layer:
  - Define `CommonPythonLayer` with `BuildMethod: python3.9` and `ContentUri: lambda/datafy_new/lambda_layer/`.
  - Put a `requirements.txt` in that folder and any shared modules under `python/` (e.g., `email_service.py`).
  - Attach the layer to functions via `Layers: [!Ref CommonPythonLayer]`.
- After changing the layer requirements, rebuild with `sam build --use-container --clean` and redeploy.

## Supabase RPC duplicate-key (23505) handling
- Context: `insert_distinct_products` inserts into a table with a unique constraint across `(category, brand, product, variant, size)`.
- Problem: Reprocessing weekly files can attempt to insert products that already exist, raising Postgres `23505 duplicate key`.
- Resolution: In `lambda/datafy_new/read_files/app.py` (and `app copy.py`), catch the RPC error and ignore when it matches code `23505` or the duplicate-key message, allowing the workflow to continue to subsequent updates.
- Rationale: Duplicate products are expected and benign; failing the whole import is undesirable.

## Supabase HTTP client timeout vs DB statement timeout
- Even with `SET statement_timeout = '0'` to disable Postgres server-side timeouts, client-side HTTP timeouts can still fail long RPCs.
- We increase the httpx read timeout on the underlying PostgREST session to 300s during client initialization for `Read Files`.
- This combination avoids `httpx.ReadTimeout` while ensuring DB does not cancel long operations.

## Cron-gated email behaviour for Lambdas
- For Lambdas that can be triggered both via API Gateway and EventBridge (cron), prefer **cron-gated** notifications so manual/API runs do not spam recipients.
- Pattern used in `create_combined_weekly_report/app.py`:
  - Inspect the raw event for `source == "aws.events"` (typical scheduled events) and/or an explicit payload flag such as `is_cron` / `send_email`.
  - Only send email when this flag is present/true; otherwise, generate the artefact (e.g., report) and return its URL without emailing.
- To wire this up from EventBridge, configure the target input as constant JSON (e.g. `{"customer_id": 123, "is_cron": true}`) so only the scheduled job sets the flag.

## Optional S3 artifacts for Lambda (add/remove without code change)
- To store run artifacts (e.g. image, prompt, response) in S3 from a Lambda, use an **env-driven toggle**: e.g. `GENAI_ARTIFACTS_S3_BUCKET` set from a SAM parameter (e.g. `ArtifactsBucketName`). When the parameter is empty, the Lambda skips S3 uploads; when set, it uploads to a known prefix (e.g. `artifacts/{id}/{timestamp}/`). Use a **conditional IAM policy** in the template (`!If [HasArtifactsBucket, { ... }, !Ref AWS::NoValue]`) so the Lambda only gets `s3:PutObject` when a bucket is configured. To “remove” the feature, redeploy with the parameter cleared; no code change required.

## PIL image orientation: EXIF must be applied explicitly
- **PIL `Image.open()` does NOT apply EXIF Orientation** by default. Phones and cameras store images in sensor orientation and set EXIF tag 274 (Orientation) to indicate how to display them. Browsers and image viewers apply this; PIL does not. If you open an image and save/resize without correction, you send the raw orientation to downstream systems (e.g. Bedrock), which can cause inconsistent results (wrong shelf numbering, left/right inversion, etc.). Use `ImageOps.exif_transpose(img)` immediately after `Image.open()` to apply EXIF orientation. It is a no-op when EXIF is absent or orientation is 1 (normal). Place it before `.convert('RGB')` or any other processing.

## Bedrock shelf-verification: size verification
- When product.size is specified (e.g. 100G, 200G, 430g), the prompt injects a SIZE MUST MATCH block. The model must verify the size marking on shelf packaging; same brand/product/variant in a different size (e.g. 200G box when target is 100G) must NOT be counted. If size is not legible on the shelf item, exclude. When Size=N/A, size verification is skipped. This prevents counting the wrong pack size (e.g. Hinds Medium Curry Powder 200G when the target is 100G).

## Bedrock shelf-verification: reducing position/count inconsistencies
- Vision models can misreport shelf row (e.g. wrong shelf number) or give a range instead of exact facings. To reduce this:
  - Define shelf numbering explicitly: "First (topmost) visible row = Shelf 1, next = Shelf 2, … Do not use 0-based numbering or skip rows."
  - Add a mandatory DOUBLE-CHECK step: (1) re-count rows from top to confirm shelf number(s), (2) count each unit in the front row and report that exact number (no range; e.g. if 3 visible, report 3), (3) confirm left/center/right.
  - For FACINGS_COUNT: require "count each visible unit one by one and report the EXACT number—do not report a range; output a single integer." Do not instruct "prefer lower" so the model aims for the actual count.
  - Allow POSITION to list multiple locations when the product appears on more than one shelf (e.g. "Shelf 1, center-left; Shelf 5, right").

## Bedrock cost tracking via CloudWatch Logs Insights
- Bedrock InvokeModel (Messages API) response body includes `usage: { input_tokens, output_tokens }`. Log this once per invocation (e.g. `BedrockUsage: {"input_tokens": N, "output_tokens": M}`) so CloudWatch has token data. Without logging it, CloudWatch only has application logs, not token counts. After deploy, use Logs Insights: filter `@message like /BedrockUsage:/`, parse the two numbers with a regex, then `stats sum(input_tokens), sum(output_tokens)` for the period. Cost = (total_input/1000)*$0.015 + (total_output/1000)*$0.075 (Claude Sonnet 4). For projections (e.g. +3000 users × 10 calls/user/day): get current invocations and total tokens for a period, compute avg tokens per invocation, then extra_cost = extra_invocations * (avg_input/1000*0.015 + avg_output/1000*0.075). See BEDROCK_SETUP.md §4.1–4.2.

## Supabase RPC pagination pattern for large reports
- For large exports (e.g., combined weekly report), use **offset/limit-style pagination** to avoid missing rows and to control memory usage.
- Pattern used in `create_combined_weekly_report/app.py`:
  - Database function signature: `get_combined_week_report_step_by_step(p_customer_id integer, page_offset integer, limit_count integer)`.
  - Start with `page_offset = 0` and `limit_count = 10000`.
  - In a loop, call the RPC with these parameters, append results, and increment `page_offset` by `limit_count` until:
    - The returned batch size is `< limit_count`, or
    - The result set is empty.
- This guarantees that **all rows** are retrieved in predictable chunks before building the final report file.