Knowledge Provider - DB Migration to Postgres by ravismula · Pull Request #43 · OpenAgriNet/knowledge-provider

ravismula · 2026-11-03T11:46:46Z

Summary

This replaces the SQLite store with Postgres as the canonical metadata store. It also introduces groups, catalogs and code-level documents. Documents now carry an explicit document type (scheme or advisory).

What changed

Database

Postgres replaces SQLite. Migrations live in migrations/ and run at API and worker startup.
The new schema adds groups, catalogs and code-level documents (code, name, aliases). A document's group is now derived through its catalog.
scripts/migrate_sqlite_to_postgres.py moves existing SQLite data into the new schema, and the legacy scripts are retired.

API

/upload now requires the document type (kind: scheme or advisory) and takes an optional code. The old tenancy naming became group.
Replacing a document is guarded, and vectors still used by another live document are never deleted.
Document type is read-only after upload. The display name is stored as documents.name.

Advisory indexing

Webhook advisory records are indexed into Qdrant through Temporal workflows, with a startup backfill.
The document type is written as the Qdrant payload type.

UI

The ingest page has a required Document type select, using group/type/code/name.

Stop syncing documents into the Postgres master_catalog table and pushing a scheme-list snapshot to Redis. bharat-oan-api no longer reads that snapshot, and nothing else read the table. Removed the module and its tests, the sync after DEV ingest and PROD promote, the cleanup on document disable (and the master_catalog_* fields in the disable response), the related settings, docs and the psycopg2-binary and redis dependencies. The scheme catalog and its /catalog/v1/* pull API are unchanged. Add PUBLISH_TO_NETWORK (default off) to control the Beckn Discovery Service publish after PROD promotion, matching docs-pipeline which has no network publish. The setting is checked inside the publish activity so Temporal replay is unaffected. When off, the activity records a skipped artifact without building the Discovery Service client, so DISCOVERY_SERVICE_ENDPOINT and NETWORK_SENDER_ID are required only when the toggle is on. /pipeline/stages and the UI stepper hide the publishing_to_network stage while it is off. Documented in ENV.md and ADR 0005.

…work publish config options

Accepts a JSON array of records and upserts them into a new SQLite webhook_records table keyed on (source, external_id). Auth via x-api-key checked against WEBHOOK_API_KEY (401 on mismatch, 503 when unset). Body capped by WEBHOOK_MAX_RECORDS (413). Store only; no ingestion. Ported from OpenAgriNet/docs-pipeline#165.

Add a postgres service to docker-compose, psycopg[binary,pool] and test dependencies, DATABASE_URL and DB_POOL_* settings, a Postgres service in the test workflow, and document the new variables in ENV.md. uv.lock also picks up unrelated lock refreshes that were already pending.

0001_initial_schema.sql creates users, catalog_entries, documents, workflows, workflow_artifacts, pages, chunks, audit_logs, search_settings, email_otps and webhook_records, plus the document_current_workflow and catalog views. pipeline/migrate.py applies versioned SQL files under an advisory lock. docs/POSTGRES_SCHEMA.md describes the schema; docs/SQLITE_SCHEMA.md is kept as the legacy reference.

Rewrite pipeline/db.py on a psycopg connection pool, splitting documents from workflows. Functions keep their names and flat dict shapes and accept the document UUID or any legacy identifier (doc-<md5>, temporal workflow id, fingerprint, filename). Artifact and job ids are now UUIDs. Scheme catalog is served from catalog_entries and the catalog view. Authenticated callers are upserted into users. Tests run against a real Postgres (TEST_DATABASE_URL or testcontainers).

…ipts migrate_sqlite_to_postgres.py imports an existing documents.db (dry-run, row-count report, idempotent). Remove start_manifest_translation_batch.py and drop_legacy_marqo_columns.py, whose tables and columns no longer exist. Adapt the backup, failed-workflow, OCR-batch and validity e2e scripts to Postgres.

Port of docs-pipeline PR #171 onto the Postgres-backed knowledge-provider. - POST /webhook starts one AdvisoryIndexWorkflow per new or changed record; chunks (document_kind=advisory) go to the Qdrant collection - admin endpoints: /webhook/advisory/{reindex,promote,status} - AdvisoryPromoteWorkflow copies to the PROD collection (PROD_VECTOR_DB_*) - migration 0002 adds webhook_records.indexed_hash / promoted_hash - advisory payload fields on Qdrant points; embedding token counter - ADVISORY_BACKFILL_ON_START queues unindexed rows at worker startup - migration-runner tests use version 0099 as their probe

POST /webhook/advisory/promote without record_ids promoted every published record, building chunks from the current payload. A record edited after its last dev index could reach PROD without being indexed in dev. The default now selects only rows where indexed_hash = payload_hash.

…uments Rewrite migration 0001 in place (no Postgres holds v1 yet): - add groups (tenant node, keycloak id filled on login sync) and users.group_id - replace catalog_entries with catalogs, one row per (group, schema type) - documents carry catalog_id, code, name and aliases; period, uploader, index_status and flags live in metadata; counts are derived from pages/chunks - audit_logs snapshot group_id instead of instance - enums become plain TEXT, validated in pipeline/models.py - catalog view becomes catalog_status (one row per document) db.py gets group/catalog get-or-create, find_document_by_code, reset_document_for_reupload and DuplicateDocumentCode. The flat document dict and db params are renamed to group/kind/code/name/aliases. instances.py becomes group_names.py. Scheme/advisory map to schema types in network_constants.py (placeholder values). Bootstrap catalog seeding is removed.

- /upload, /documents and /documents/batch take kind (scheme|advisory) and an optional code; an existing code updates that document with the new file - group, kind, code, name and aliases replace instance, document_kind, scheme_code, scheme_name and scheme_aliases in requests and responses - tenancy helpers renamed to group; users.group_id set from the token tenants - demo flag, X-Include-Demo, POST /catalog/v1/bootstrap and the never-classified reingest guard are removed - /catalog/v1 keys, Qdrant payload keys and Temporal arguments keep their names

Build groups and catalogs from legacy instance codes, fold period, uploader and index state into documents.metadata, skip kind=document and demo documents with counts in the report. Deploy migration plan is still to be written.

Rename to group/kind/code/name/aliases, cover upload with kind, re-upload of an existing code, catalog_status, group snapshots in audit_logs and the new unique/foreign-key constraints in tests/test_schema_v2.py.

NewDocumentView sends group and kind. The classification panel offers scheme and advisory only and no longer blocks ingestion on a missing type. Instance helpers and badge are renamed to group.

Replace guard (api): an upload now needs upload rights in the target group, not just somewhere. Replacing an existing code also needs review rights in the document's group, is refused while the document is disabled or a run is in progress, and needs approval rights when the document is live in PROD. Orphaned vectors: a re-upload changes the fingerprint, which is the document's key in the vector index. reset_document_for_reupload now records the old key in metadata.superseded_doc_ids for each index that holds it. The next DEV ingest and PROD promote delete those points after the new content is indexed (a failed delete keeps the key for the next run). Disable and purge now delete the current and superseded keys, and the PROD cleanup follows the PROD index entry, so a document promoted as a scheme is cleaned up even if its kind changed.

The vector index key is the file fingerprint, so two documents holding the same file share points. Replace cleanup, disable and purge now skip any key that another document uses as its fingerprint or index entry doc_id. Replace cleanup forgets such a key without deleting it. A per-document key would remove the sharing but needs a reindex, so it is left for a separate change.

The shared-key guard counted any document with the same fingerprint, including disabled or never-indexed ones, and forgot a skipped key so its points could stay in the index for good. It now counts only another live document that is indexed under the key in the same environment (dev or prod), and a skipped key stays on metadata.superseded_doc_ids so a later ingest, disable or purge can still remove it.

/upload, /documents and /documents/batch now require kind (scheme|advisory) and accept group and code. The upload helpers take DOC_KIND/PIPELINE_GROUP (shell, PowerShell) or --kind/--group (Python), the validity check calls upsert_document and _prepare_records with group/kind/code, list_failed_workflows drops the removed include_demo, and ENV.md drops the bootstrap endpoint and lists the new PATCH body.

The native select did not match the console theme or the Select used on the Documents filters. Switch to the shared Select component.

Video returns as a third type with a placeholder catalog schema name (openagrinet:VideoResource), offered on the upload page. Custom types stay removed. The type decides which catalog a document lives in and is chosen at upload, so the document page now shows it read-only and no longer sends kind in the metadata PATCH. Tests that used video as the example of an unknown kind now use podcast.

… upload A group row now gets one catalog per supported type (scheme, advisory, video) in the same transaction, named "<GROUP> <kind> catalog". Startup runs an idempotent sync so a type added to the kind map reaches existing groups. Upload keeps get-or-create as a fallback for a missing catalog. The kind is fixed by the catalog chosen at upload, so it is removed from the scheme-metadata PATCH and from update_document_fields.

type is now scheme/advisory/video (or the kind slug) instead of scheme/document. Existing points need a backfill before readers that filter on type=advisory are deployed.