Knowledge Provider - DB Migration to Postgres

.dockerignore

@@ -67,6 +67,7 @@ test_ocr_output*/
 
 # Scripts (not needed at runtime inside the container)
 scripts/
+!scripts/migrate_sqlite_to_postgres.py
 
 # Claude / AI tooling
 .claude/

.env.example

@@ -193,32 +193,12 @@ VITE_KEYCLOAK_IDP_HINT=google
 KEYCLOAK_ADMIN_USERNAME=admin
 KEYCLOAK_ADMIN_PASSWORD=
 
-# --- Master Catalog (Postgres) ---
-# code/name/tool_name/prompt_snippet, synced on DEV ingest (status=dev)
-# and PROD promote (status=live). No DB container is provisioned in docker-compose —
-# point these at your existing Postgres instance (any host/server is fine; this just
-# needs its own database + user on it, e.g. `CREATE DATABASE master_catalog`).
-MASTER_CATALOG_PG_HOST=
-MASTER_CATALOG_PG_PORT=5432
-MASTER_CATALOG_PG_DB=master_catalog
-MASTER_CATALOG_PG_USER=master_catalog
-MASTER_CATALOG_PG_PASSWORD=
-MASTER_CATALOG_PG_SSLMODE=disable
-
-# --- AI layer Redis (bharat-oan-api) ---
-# docs-pipeline pushes the catalog snapshot directly into this Redis instance
-# after every Postgres sync. Leave AI_LAYER_REDIS_HOST empty to disable the
-# push (Postgres sync still happens; only the Redis mirror is skipped).
-AI_LAYER_REDIS_HOST=
-AI_LAYER_REDIS_PORT=6379
-AI_LAYER_REDIS_DB=0
-AI_LAYER_REDIS_PASSWORD=
-MASTER_CATALOG_REDIS_TTL_SECONDS=172800
-
 # --- Publish to Network (Discovery Service) ---
-# After DEV ingest, the publishing_to_network stage POSTs a Beckn catalog/publish
-# envelope announcing the kind of knowledge this provider serves. Required — there
-# is no feature flag to disable the stage. Worker-only; the API and UI never read these.
+# After a document is promoted to PROD, the publishing_to_network stage POSTs a
+# Beckn catalog/publish envelope announcing the kind of knowledge this provider
+# serves. Set ENABLE_PUBLISH_TO_NETWORK=true to enable publishing.
+# DISCOVERY_SERVICE_ENDPOINT and NETWORK_SENDER_ID are required when enabled.
+ENABLE_PUBLISH_TO_NETWORK=false
 DISCOVERY_SERVICE_ENDPOINT=
 NETWORK_SENDER_ID=
 # Publicly reachable base URL of this provider, sent as each catalog's bppUri.
@@ -265,6 +245,14 @@ KC_SMTP_SSL=false
 KC_SMTP_USERNAME=
 KC_SMTP_PASSWORD=
 
+# Shared secret for POST /webhook. Callers must send it in the `x-api-key` header.
+# Required: if unset or empty, /webhook returns 503. Wrong or missing key returns 401.
+# Generate a 32-char hex value with: openssl rand -hex 16
+# Never commit a real value.
+WEBHOOK_API_KEY=your_32_character_hex_key_here
+
+# Max records accepted per POST /webhook request (more returns 413)
+WEBHOOK_MAX_RECORDS=500
 
 # --- Logs to ClickHouse over OTLP (ADR 0007). Empty endpoint = off.
 # Local: http://otel-collector:4318 with `docker compose --profile observability up -d`

.github/workflows/run-tests.yml

@@ -9,6 +9,22 @@ jobs:
   pipeline:
     name: pipeline (lint, test, coverage)
     runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:15-alpine
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: knowledge_provider_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      TEST_DATABASE_URL: postgresql://postgres:postgres@localhost:5432/knowledge_provider_test
     steps:
       - uses: actions/checkout@v4
 

AGENTS.md

@@ -4,13 +4,13 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 
 ## What this is
 
-A review-driven document ingestion pipeline (product name "Bharat Vistaar Docs Pipeline"): Temporal workflows + FastAPI + SQLite + MinIO + Qdrant. Documents move through explicit stages (OCR → review → translate → review → chunk → review → ingest DEV → Super Admin promote to PROD), with every major output persisted as an inspectable artifact rather than hidden.
+A review-driven document ingestion pipeline (product name "Bharat Vistaar Docs Pipeline"): Temporal workflows + FastAPI + Postgres + MinIO + Qdrant. Documents move through explicit stages (OCR → review → translate → review → chunk → review → ingest DEV → Super Admin promote to PROD), with every major output persisted as an inspectable artifact rather than hidden.
 
 Full narrative docs (read these before deep architectural changes, don't re-derive):
 - `README.md` — what the system does, API surface, supported inputs, auth model
 - `docs/SYSTEM_DESIGN.md` — C4 diagrams, containers, DEV/PROD indexing, stage/activity table
 - `docs/architecture-pipeline-rbac.md` — login flow, roles, DEV vs Super-Admin-PROD gate
-- `docs/MASTER_SCHEME_CATALOG_AND_AI_SYNC_DESIGN.md` — master catalog → Redis sync to AI layer (bharat-oan-api)
+- `docs/MASTER_SCHEME_CATALOG_AND_AI_SYNC_DESIGN.md` — original scheme catalog design (historical: the Postgres master catalog and Redis push it describes were removed; the pull API `/catalog/v1/*` remains)
 - `ENV.md` — full annotated environment variable reference (`.env.example` for a working template)
 
 ## Commands
@@ -69,7 +69,7 @@ Core modules (all under `pipeline/`):
 - `workflows.py` — Temporal workflow definitions (`DocumentPipelineWorkflow` is the main one; also `ReingestionWorkflow`, `PromoteToProdWorkflow`, `TranslationOnlyWorkflow`, `OcrOnlyWorkflow`, `ChunkingOnlyWorkflow`)
 - `activities.py` — Temporal activities invoked by the workflows (OCR, translate, chunk, ingest, state updates)
 - `worker.py` — Temporal worker process; registers workflows/activities on task queue `ocr-pipeline`
-- `db.py` — SQLite access layer; SQLite is the **canonical** metadata/review-state store (documents, pages, chunks, jobs, artifacts, audit log), not just a cache
+- `db.py` — Postgres access layer (psycopg3 sync connection pool); Postgres is the **canonical** metadata/review-state store (documents, workflows, pages, chunks, artifacts, audit log, settings), not just a cache. Documents and jobs come back as flat dicts keyed by `workflow_id` (the document's main Temporal workflow id); every function also accepts the document UUID. Schema: `migrations/` (applied at api/worker startup by `migrate.py`), described in `docs/POSTGRES_SCHEMA.md`
 - `document_repository.py` — `DocumentRepository`, a domain layer over `db.py` for document reads; `db.py` stays one-function-per-query with no domain knowledge
 - `network_constants.py` — fixed values sent on the network (catalog/resource ids, topics, languages, Beckn version). The file v2 edits when the AI layer starts deriving them
 - `catalog_builder.py` — pure builder mapping a document's knowledge kind (`advisory`/`scheme`) to the single `OnDemand` Beckn catalog announced for that kind; takes the resolved provider and prefixes catalog/resource ids for a Group; returns `None` for any other kind. No env, no I/O
@@ -79,8 +79,8 @@ Core modules (all under `pipeline/`):
 - `models.py` — Pydantic models, including `DocumentStage` enum and `PIPELINE_STAGES` (the stepper-UI stage list)
 - `config.py` — `Config` dataclass reading env vars, with defaults
 - `observability.py` — bridges root-logger `INFO`+ lines to OTLP when `OTEL_EXPORTER_OTLP_ENDPOINT` is set (ClickHouse/HyperDX); no-op otherwise. See ADR 0007
-- `instances.py` — instance/tenant (state) code ↔ human name mapping; `bv` = Bharat Vistaar platform-wide, others are state codes matching Keycloak group paths (`/states/MH/...` → `mh`)
-- `master_catalog_pg.py`, `scheme_catalog.py` — Postgres-backed master scheme catalog, synced to Redis for the AI layer (bharat-oan-api) to consume
+- `instances.py` — tenant instance code ↔ human name mapping; `bv` = Bharat Vistaar platform-wide, others are state codes matching Keycloak group paths (`/states/MH/...` → `mh`). A tenant code is an `instances` row (name = code); a document's `instance` is derived through `documents.catalog_id → catalogs.instance_id`, there is no `documents.instance` column
+- `scheme_catalog.py` — scheme catalog behind the pull-based `/catalog/v1/*` API, served from the computed `catalog_status` view (one row per document; `catalogs` rows are `(instance, type)`, `documents` carry `code`/`name`/`aliases`)
 - `ocr/`, `translation/`, `chunking/`, `vector_store/` — pluggable provider packages, each with a `base.py` protocol/ABC and a `service.py` (or equivalent) that dispatches to the configured provider
 - `auth/` — Keycloak JWT validation, group-based tenancy/permissions, email-OTP flow, admin user management
 
@@ -106,7 +106,7 @@ registered → ocr_processing → ocr_review → translation_processing → tran
 
 ### Storage responsibilities (don't blur these)
 
-- **SQLite** — canonical metadata/content/review state (documents, pages, chunks, jobs, artifacts, audit, search settings). Source of truth for anything editable.
+- **Postgres** — canonical metadata/content/review state (documents, workflows, pages, chunks, artifacts, audit, search settings). Source of truth for anything editable.
 - **MinIO** — blob storage for original uploads, normalized files, and stage artifacts (OCR/translation/chunk exports, vector-index payload snapshots).
 - **Temporal** — orchestration/retries/review-gate signaling only; not a content store.
 - **Qdrant** — downstream search projections of approved chunk state; never the place to edit content.
@@ -121,4 +121,4 @@ React 18 + Vite + Tailwind v4 + Radix UI operator console (not a plain upload fo
 
 ### Scripts (`scripts/`)
 
-Operational tools, not one-off hacks — e.g. `list_failed_workflows.py`, `terminate_stuck_workflows.py`, `keycloak_bootstrap_docs_pipeline.py`, `drop_legacy_marqo_columns.py`, `backfill_provider_id.py` (sets `provider_id` on existing DEV/PROD chunks; dry run unless `--apply`). Check here before writing a new maintenance script — an equivalent likely exists.
+Operational tools, not one-off hacks — e.g. `list_failed_workflows.py`, `terminate_stuck_workflows.py`, `keycloak_bootstrap_docs_pipeline.py`, `drop_legacy_marqo_columns.py`, `backfill_provider_id.py` (sets `provider_id` on existing DEV/PROD chunks; dry run unless `--apply`), `migrate_sqlite_to_postgres.py`. Check here before writing a new maintenance script — an equivalent likely exists.

CONTEXT.md

@@ -12,16 +12,12 @@ _Avoid_: "publish" alone — ambiguous with Publish to Network.
 The pipeline stage that runs only after a document's PROD promotion succeeds, and POSTs a Beckn-shaped `publish` envelope to an external Discovery Service. Skipped entirely when PROD is disabled (`DISABLE_PROD_SETTING=true`) — a document never reaches the network without first promoting to PROD. The envelope it sends is kind-level (see Knowledge Kind), not document-level: it does not make any single document searchable, only re-announces that the document's provider (its Group Provider, or the Shared Provider) serves that kind at all.
 _Avoid_: "publish" alone — ambiguous with Publish to Dev.
 
-**Master Catalog**:
-The global, Postgres-backed registry of prompt/tool metadata (any content type: scheme, advisory, etc.), synced to Redis so the AI layer can pick up entries without a redeploy.
-_Avoid_: "catalog" alone — ambiguous with Scheme Catalog and the Network Catalog Envelope.
-
 **Scheme Catalog**:
-The SQLite-backed, scheme-only registry that aggregates one or more documents sharing a `scheme_code`. Backs the pull-based `GET /catalog/v1/*` API that other BAPs poll.
-_Avoid_: "catalog" alone — ambiguous with Master Catalog and the Network Catalog Envelope.
+The scheme-only registry (`catalog_entries` rows plus the computed `catalog` view) that aggregates one or more documents sharing a `scheme_code`. Backs the pull-based `GET /catalog/v1/*` API that other BAPs poll.
+_Avoid_: "catalog" alone — ambiguous with the Network Catalog Envelope.
 
 **Network Catalog Envelope**:
-The Beckn wire-protocol `message.catalogs` object sent by Publish to Network to the Discovery Service. Structurally and semantically unrelated to the Master Catalog and Scheme Catalog tables — do not conflate the three.
+The Beckn wire-protocol `message.catalogs` object sent by Publish to Network to the Discovery Service. Structurally and semantically unrelated to the Scheme Catalog tables — do not conflate the two.
 _Avoid_: "catalog" alone.
 
 **Group**:

Dockerfile

@@ -29,8 +29,10 @@ RUN --mount=type=cache,target=/root/.cache/uv \
 
 ENV PATH="/app/.venv/bin:$PATH"
 
-# Copy pipeline code
+# Copy pipeline code and the SQL migrations the api/worker apply at startup
 COPY pipeline/ ./pipeline/
+COPY migrations/ ./migrations/
+COPY scripts/migrate_sqlite_to_postgres.py ./scripts/migrate_sqlite_to_postgres.py
 
 # Catch imports that work locally but fail on the shipped Python 3.10 image.
 RUN python -c "import pipeline.api"

ENV.md

@@ -64,7 +64,11 @@ Root `.env` (see `.env.example`). Required by FastAPI (`pipeline/api/`), Tempora
 | `LOG_LEVEL` | `INFO` | Worker log verbosity (`DEBUG`/`INFO`/`WARNING`/`ERROR`); the `temporalio` SDK logger stays at `INFO` regardless, to avoid its own noisy debug output |
 | `MINIO_ENDPOINT` | `localhost:9000` | MinIO API host:port |
 | `MINIO_BUCKET` | `documents` | Object storage bucket |
-| `DOCUMENT_DB_PATH` | `/data/documents.db` | SQLite path (use `./data/documents.db` for local non-Docker) |
+| `DATABASE_URL` | `postgresql://postgres:postgres@localhost:5432/knowledge_provider` | Application Postgres (documents, workflows, pages, chunks, audit, settings). The compose `postgres` service builds it from `DB_USERNAME` / `DB_PASSWORD` / `DB_NAME` (defaults `docs` / `docs` / `knowledge_provider`). Pending SQL migrations in `migrations/` are applied by the api and worker at startup |
+| `DB_POOL_MIN_SIZE` | `1` | Connection pool minimum per process |
+| `DB_POOL_MAX_SIZE` | `10` | Connection pool maximum per process (api and worker each hold their own pool) |
+| `DB_POOL_TIMEOUT_SECONDS` | `30` | Seconds to wait for a free pooled connection |
+| `DOCUMENT_DB_PATH` | `/data/documents.db` | Only read by `scripts/migrate_sqlite_to_postgres.py`: the legacy SQLite file to import |
 
 ### Logs to ClickHouse (OTLP)
 
@@ -89,6 +93,18 @@ Off unless `OTEL_EXPORTER_OTLP_ENDPOINT` is set. When it is set, API and worker
 | `DOCS_PIPELINE_UI_URL` | `http://localhost:3000` | Public UI base for links |
 | `DEFAULT_INSTANCE` | `default` | Default tenant / instance id |
 
+### Webhook (`POST /webhook`)
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `WEBHOOK_MAX_RECORDS` | `500` | Max records per `/webhook` request; more returns 413 |
+| `WEBHOOK_API_KEY` | *(empty — required)* | 32-char hex secret. Callers send it in the `x-api-key` header. Missing/invalid header → 401. Unset/empty → 503 (fails closed, ignores `AUTH_DISABLED`). Generate with `openssl rand -hex 16`. |
+| `ADVISORY_BACKFILL_ON_START` | `false` | When `true`, the worker queues an index workflow for every unindexed `webhook_records` row once at startup (backfill of existing records). Leave off normally; `POST /webhook/advisory/reindex` does the same on demand. |
+
+Each saved record whose `payload_hash` differs from its `indexed_hash` also starts one Temporal workflow `advisory-<record id>` that writes advisory chunks (`document_kind=advisory`) to the Qdrant collection `VECTOR_DB_COLLECTION_NAME`. Admin endpoints (Keycloak admin): `POST /webhook/advisory/reindex`, `POST /webhook/advisory/promote` (copies to `PROD_VECTOR_DB_COLLECTION_NAME` (on `PROD_VECTOR_DB_URL`)), `GET /webhook/advisory/status`.
+
+Body is a JSON array of objects (max `WEBHOOK_MAX_RECORDS`); stored in the `webhook_records` table, upserted on `(source, external_id)` where `external_id` is `contentID` (else `id`, `slug`, payload hash).
+
 ### Auth / Keycloak (backend JWT validation)
 
 | Variable | Default | Purpose |
@@ -237,24 +253,7 @@ Exposes `/catalog/v1/*` so **bharat-oan-api** and **bharat-provider-backend** ca
 | `PROD_SCHEME_VECTOR_DB_API_KEY` | `PROD_VECTOR_DB_API_KEY` | Scheme collection API key |
 | `PROD_SCHEME_VECTOR_DB_COLLECTION_NAME` | `schemes-index` | Must **not** equal documents collection |
 
-**Endpoints:** `GET /catalog/v1/snapshot`, `/version`, `/schemes`, `/tool-prompt`; `POST /catalog/v1/rebuild`, `/bootstrap`; `PATCH /documents/{id}/scheme-metadata`.
-
-### Master Catalog — Postgres + Redis push (dev preview channel)
-
-Separate from the SQLite/Qdrant catalog above (which tracks PROD vector publication and is scheme-specific). This one is a generic Postgres table (`master_catalog`: code, content_type, name, tool_name, doc_id, prompt_snippet, status) — not scheme-only, since ingested documents can be schemes, advisories, or other kinds that later grow their own code/name/tool metadata. docs-pipeline writes to it on **DEV ingest complete** (`status=dev`) and **PROD promote complete** (`status=live`), then pushes directly into the AI layer's Redis so bharat-oan-api can test an entry's prompt/tool routing in the dev chatbot without an AI-layer redeploy.
-
-| Variable | Default | Purpose |
-|----------|---------|---------|
-| `MASTER_CATALOG_PG_HOST` | *(empty — required)* | Host of your existing Postgres instance. No DB container is provisioned by docker-compose for this — bring your own server and create the database on it (e.g. `CREATE DATABASE master_catalog`). Tables are created automatically on first sync. |
-| `MASTER_CATALOG_PG_PORT` | `5432` | Postgres port |
-| `MASTER_CATALOG_PG_DB` | `master_catalog` | Database name |
-| `MASTER_CATALOG_PG_USER` / `MASTER_CATALOG_PG_PASSWORD` | `master_catalog` / *(empty)* | Credentials |
-| `MASTER_CATALOG_PG_SSLMODE` | `disable` | Use `require` for managed prod Postgres |
-| `AI_LAYER_REDIS_HOST` | *(empty)* | bharat-oan-api's Redis host. Empty = Postgres sync still happens, Redis push is skipped |
-| `AI_LAYER_REDIS_PORT` / `AI_LAYER_REDIS_DB` / `AI_LAYER_REDIS_PASSWORD` | `6379` / `0` / *(empty)* | AI layer Redis connection |
-| `MASTER_CATALOG_REDIS_TTL_SECONDS` | `172800` (48h) | Snapshot key TTL — a dead-man's switch, not a freshness mechanism (writes are push-driven) |
-
-**Redis keys:** `master-catalog:dev:snapshot` (dev + live entries — what the dev chatbot reads), `master-catalog:live:snapshot` (live only — what prod reads). Plain JSON via raw `redis-py`, not routed through bharat-oan-api's `aiocache` layer, since it's an external write contract rather than an internal cache value.
+**Endpoints:** `GET /catalog/v1/snapshot`, `/version`, `/schemes`, `/tool-prompt`; `POST /catalog/v1/rebuild`; `PATCH /documents/{id}/scheme-metadata` (body: `kind`, `code`, `name`, `aliases`, `tool_routing`, `catalog_visible`, `network_visible`, `valid_from`, `valid_to`). The catalog is computed from documents (one row per document), so there is no bootstrap endpoint. `/upload`, `/documents` and `/documents/batch` take a required `kind` (`scheme` or `advisory`) and an optional `group` and `code`.
 
 ### Publish to Network
 
@@ -285,7 +284,7 @@ Each instance has one provider for both its advisory and scheme catalogs: id and
 |---------|----|-----|--------|
 | `VITE_*` proxies & Keycloak browser login | ✅ | — | — |
 | JWT validation (`AUTH_*`, `KEYCLOAK_*`) | — | ✅ | — |
-| Temporal / MinIO / SQLite | — | ✅ | ✅ |
+| Temporal / MinIO / Postgres (`DATABASE_URL`) | — | ✅ | ✅ |
 | OCR / translation / chunking / domain tags | — | config / status | ✅ runs jobs |
 | Qdrant (`VECTOR_DB_*`) | — | ✅ | ✅ |
 | Publish to Network (`DISCOVERY_SERVICE_ENDPOINT`, `NETWORK_SENDER_ID`) | — | — | ✅ runs the publish activity |
@@ -299,7 +298,7 @@ Each instance has one provider for both its advisory and scheme catalogs: id and
 
 ```bash
 set -a && source .env && set +a
-export DOCUMENT_DB_PATH="$(pwd)/data/documents.db"
+export DATABASE_URL="postgresql://docs:docs@localhost:5433/knowledge_provider"   # the compose `postgres` service (`docker compose up -d postgres`)
 export ALLOWED_FILE_PATHS="$(pwd)/books,$(pwd)/data/documents"
 export CORS_ORIGINS="http://localhost:3000,http://localhost:3001"
 export PYTHONPATH="$(pwd)"